add funnels

apps/sdk-api/package.json

@@ -3,6 +3,7 @@
   "version": "0.0.1",
   "scripts": {
     "dev": "dotenv -e ../../.env -c -v WATCH=1 tsup",
+    "testing": "API_PORT=3333 pnpm dev",
     "start": "node dist/index.js",
     "build": "rm -rf dist && tsup",
     "lint": "eslint .",
@@ -21,7 +22,8 @@
     "pino": "^8.17.2",
     "ramda": "^0.29.1",
     "sharp": "^0.33.2",
-    "ua-parser-js": "^1.0.37"
+    "ua-parser-js": "^1.0.37",
+    "uuid": "^9.0.1"
   },
   "devDependencies": {
     "@mixan/eslint-config": "workspace:*",
@@ -30,6 +32,7 @@
     "@mixan/types": "workspace:*",
     "@types/ramda": "^0.29.6",
     "@types/ua-parser-js": "^0.7.39",
+    "@types/uuid": "^9.0.8",
     "@types/ws": "^8.5.10",
     "eslint": "^8.48.0",
     "prettier": "^3.0.3",

apps/sdk-api/src/controllers/event.controller.ts

@@ -4,10 +4,11 @@ import { getReferrerWithQuery, parseReferrer } from '@/utils/parseReferrer';
 import { isUserAgentSet, parseUserAgent } from '@/utils/parseUserAgent';
 import type { FastifyReply, FastifyRequest } from 'fastify';
 import { omit } from 'ramda';
+import { v4 as uuid } from 'uuid';
 
 import { generateDeviceId, getTime, toISOString } from '@mixan/common';
 import type { IServiceCreateEventPayload } from '@mixan/db';
-import { createBotEvent, getEvents, getSalts } from '@mixan/db';
+import { createBotEvent, createEvent, getEvents, getSalts } from '@mixan/db';
 import type { JobsOptions } from '@mixan/queue';
 import { eventsQueue, findJobByPrefix } from '@mixan/queue';
 import type { PostEventPayload } from '@mixan/types';
@@ -108,6 +109,7 @@ export async function postEvent(
       payload: {
         name: body.name,
         deviceId: event?.deviceId || '',
+        sessionId: event?.sessionId || '',
         profileId,
         projectId,
         properties: body.properties ?? {},
@@ -145,11 +147,16 @@ export async function postEvent(
     return reply.status(200).send('');
   }
 
-  const [geo, eventsJobs] = await Promise.all([
+  const [geo, eventsJobs, events] = await Promise.all([
     parseIp(ip),
     eventsQueue.getJobs(['delayed']),
+    getEvents(
+      `SELECT * FROM events WHERE name = 'session_start' AND profile_id = '${profileId}' AND project_id = '${projectId}' ORDER BY created_at DESC LIMIT 1`
+    ),
   ]);
 
+  const sessionStartEvent = events[0];
+
   // find session_end job
   const sessionEndJobCurrentDeviceId = findJobByPrefix(
     eventsJobs,
@@ -197,6 +204,7 @@ export async function postEvent(
     deviceId,
     profileId,
     projectId,
+    sessionId: createSessionStart ? uuid() : sessionStartEvent?.sessionId ?? '',
     properties: Object.assign({}, omit(['path', 'referrer'], body.properties), {
       hash,
       query,
@@ -246,14 +254,12 @@ export async function postEvent(
   }
 
   if (createSessionStart) {
-    eventsQueue.add('event', {
-      type: 'createEvent',
-      payload: {
-        ...payload,
-        name: 'session_start',
-        // @ts-expect-error
-        createdAt: toISOString(getTime(payload.createdAt) - 10),
-      },
+    // We do not need to queue session_start
+    await createEvent({
+      ...payload,
+      name: 'session_start',
+      // @ts-expect-error
+      createdAt: toISOString(getTime(payload.createdAt) - 10),
     });
   }