improve(queue): how we handle incoming events and session ends

packages/db/src/buffers/base-buffer.ts

@@ -67,8 +67,6 @@ export class BaseBuffer {
           lockId,
         });
       }
-    } else {
-      this.logger.warn('Failed to acquire lock. Skipping flush.', { lockId });
     }
   }
 }

packages/db/src/buffers/session-buffer.ts

@@ -58,14 +58,17 @@ export class SessionBuffer extends BaseBuffer {
       if (event.origin) {
         newSession.exit_origin = event.origin;
       }
-      newSession.duration =
+      const duration =
         new Date(newSession.ended_at).getTime() -
         new Date(newSession.created_at).getTime();
-      if (newSession.duration < 0) {
+      if (duration > 0) {
+        newSession.duration = duration;
+      } else {
         this.logger.warn('Session duration is negative', {
+          duration,
+          event,
           session: newSession,
         });
-        newSession.duration = 0;
       }
       newSession.properties = toDots({
         ...(event.properties || {}),
@@ -73,7 +76,7 @@ export class SessionBuffer extends BaseBuffer {
       });
       // newSession.revenue += event.properties?.__revenue ?? 0;
 
-      if (event.name === 'screen_view') {
+      if (event.name === 'screen_view' && event.path) {
         newSession.screen_views.push(event.path);
         newSession.screen_view_count += 1;
       } else {
@@ -161,8 +164,6 @@ export class SessionBuffer extends BaseBuffer {
       const sessions = await this.getSession(event);
       const [newSession] = sessions;
 
-      console.log(`Adding sessions ${sessions.length}`);
-
       const multi = this.redis.multi();
       multi.set(
         `session:${newSession.id}`,

packages/db/src/services/event.service.ts

@@ -1,8 +1,8 @@
-import { path, assocPath, last, mergeDeepRight, pick, uniq } from 'ramda';
+import { path, assocPath, last, mergeDeepRight } from 'ramda';
 import { escape } from 'sqlstring';
 import { v4 as uuid } from 'uuid';
 
-import { toDots } from '@openpanel/common';
+import { DateTime, toDots } from '@openpanel/common';
 import { cacheable, getCache } from '@openpanel/redis';
 import type { IChartEventFilter } from '@openpanel/validation';
 
@@ -19,13 +19,8 @@ import type { EventMeta, Prisma } from '../prisma-client';
 import { db } from '../prisma-client';
 import { createSqlBuilder } from '../sql-builder';
 import { getEventFiltersWhereClause } from './chart.service';
-import type { IClickhouseProfile, IServiceProfile } from './profile.service';
-import {
-  getProfileById,
-  getProfiles,
-  transformProfile,
-  upsertProfile,
-} from './profile.service';
+import type { IServiceProfile } from './profile.service';
+import { getProfileById, getProfiles, upsertProfile } from './profile.service';
 
 export type IImportedEvent = Omit<
   IClickhouseEvent,
@@ -293,6 +288,42 @@ export async function createEvent(payload: IServiceCreateEventPayload) {
     payload.profileId = payload.deviceId;
   }
 
+  const event: IClickhouseEvent = {
+    id: uuid(),
+    name: payload.name,
+    device_id: payload.deviceId,
+    profile_id: payload.profileId ? String(payload.profileId) : '',
+    project_id: payload.projectId,
+    session_id: payload.sessionId,
+    properties: toDots(payload.properties),
+    path: payload.path ?? '',
+    origin: payload.origin ?? '',
+    created_at: DateTime.fromJSDate(payload.createdAt)
+      .setZone('UTC')
+      .toFormat('yyyy-MM-dd HH:mm:ss.SSS'),
+    country: payload.country ?? '',
+    city: payload.city ?? '',
+    region: payload.region ?? '',
+    longitude: payload.longitude ?? null,
+    latitude: payload.latitude ?? null,
+    os: payload.os ?? '',
+    os_version: payload.osVersion ?? '',
+    browser: payload.browser ?? '',
+    browser_version: payload.browserVersion ?? '',
+    device: payload.device ?? '',
+    brand: payload.brand ?? '',
+    model: payload.model ?? '',
+    duration: payload.duration,
+    referrer: payload.referrer ?? '',
+    referrer_name: payload.referrerName ?? '',
+    referrer_type: payload.referrerType ?? '',
+    imported_at: null,
+    sdk_name: payload.sdkName ?? '',
+    sdk_version: payload.sdkVersion ?? '',
+  };
+
+  await Promise.all([sessionBuffer.add(event), eventBuffer.add(event)]);
+
   if (payload.profileId) {
     const profile = {
       id: String(payload.profileId),
@@ -326,40 +357,6 @@ export async function createEvent(payload: IServiceCreateEventPayload) {
     }
   }
 
-  const event: IClickhouseEvent = {
-    id: uuid(),
-    name: payload.name,
-    device_id: payload.deviceId,
-    profile_id: payload.profileId ? String(payload.profileId) : '',
-    project_id: payload.projectId,
-    session_id: payload.sessionId,
-    properties: toDots(payload.properties),
-    path: payload.path ?? '',
-    origin: payload.origin ?? '',
-    created_at: formatClickhouseDate(payload.createdAt),
-    country: payload.country ?? '',
-    city: payload.city ?? '',
-    region: payload.region ?? '',
-    longitude: payload.longitude ?? null,
-    latitude: payload.latitude ?? null,
-    os: payload.os ?? '',
-    os_version: payload.osVersion ?? '',
-    browser: payload.browser ?? '',
-    browser_version: payload.browserVersion ?? '',
-    device: payload.device ?? '',
-    brand: payload.brand ?? '',
-    model: payload.model ?? '',
-    duration: payload.duration,
-    referrer: payload.referrer ?? '',
-    referrer_name: payload.referrerName ?? '',
-    referrer_type: payload.referrerType ?? '',
-    imported_at: null,
-    sdk_name: payload.sdkName ?? '',
-    sdk_version: payload.sdkVersion ?? '',
-  };
-
-  await Promise.all([sessionBuffer.add(event), eventBuffer.add(event)]);
-
   return {
     document: event,
   };

packages/queue/src/queues.ts

@@ -22,14 +22,18 @@ export interface EventsQueuePayloadIncomingEvent {
     headers: Record<string, string | undefined>;
     currentDeviceId: string;
     previousDeviceId: string;
-    priority: boolean;
   };
 }
 export interface EventsQueuePayloadCreateEvent {
   type: 'createEvent';
   payload: Omit<IServiceEvent, 'id'>;
 }
-type SessionEndRequired = 'sessionId' | 'deviceId' | 'profileId' | 'projectId';
+type SessionEndRequired =
+  | 'sessionId'
+  | 'deviceId'
+  | 'profileId'
+  | 'projectId'
+  | 'createdAt';
 export interface EventsQueuePayloadCreateSessionEnd {
   type: 'createSessionEnd';
   payload: Partial<Omit<IServiceEvent, SessionEndRequired>> &