fix(telemetry): send to custom.io

packages/telemetry/package.json

@@ -10,6 +10,7 @@
   },
   "devDependencies": {
     "@cloudflare/workers-types": "^4.20231218.0",
+    "@customerio/cdp-analytics-node": "^0.2.0",
     "typescript": "^5.5.2",
     "wrangler": "^3.60.3"
   }

packages/telemetry/src/index.ts

@@ -1,5 +1,7 @@
+import { Analytics } from '@customerio/cdp-analytics-node';
+
 export default {
-  async fetch(request): Promise<Response> {
+  async fetch(request, env: Env): Promise<Response> {
     // https://github.com/mixpanel/tracking-proxy/blob/master/nginx.conf
     const headers = new Headers(request.headers);
     if (request.method !== 'POST' && request.method !== 'OPTIONS') {
@@ -19,7 +21,43 @@ export default {
     headers.set('Host', 'api-js.mixpanel.com');
     const url = new URL(request.url);
     url.host = 'api-js.mixpanel.com';
-    const req = new Request(url, request);
+    const req = new Request(url, request.clone());
+    try {
+      const body = await request.text();
+      const requestData = decodeURIComponent(body).split('data=');
+      const payload = requestData[1];
+      if (payload) {
+        const events = JSON.parse(atob(payload));
+        const eventsToSend = events.filter((payload: any) => !payload.event?.includes?.('page_view') && payload.properties?.$user_id);
+        if (eventsToSend.length) {
+          const sendQueue: Promise<void>[] = [];
+          const analytics = new Analytics({
+            writeKey: env.CUSTOM_IO_KEY,
+            flushInterval: 10,
+            maxEventsInBatch: eventsToSend.length,
+          });
+          for (const event of eventsToSend) {
+            sendQueue.push(
+              new Promise((resolve) => {
+                analytics.track(
+                  {
+                    userId: event.properties.$user_id,
+                    event: event.event,
+                    properties: event.properties,
+                  },
+                  () => {
+                    resolve();
+                  },
+                );
+              }),
+            );
+          }
+          await Promise.all(sendQueue);
+        }
+      }
+    } catch (e) {
+      console.error(`event can not be parsed`, e);
+    }
     const res = await fetch(req, {
       headers,
     });

packages/telemetry/worker-configuration.d.ts

@@ -1,4 +1,5 @@
 // Generated by Wrangler
 // After adding bindings to `wrangler.toml`, regenerate this interface via `npm run cf-typegen`
 interface Env {
+  CUSTOM_IO_KEY: string;
 }