fix: optimize event buffer (#278)

* fix: how we fetch profiles in the buffer * perf: optimize event buffer * remove unused file * fix * wip * wip: try groupmq 2 * try simplified event buffer with duration calculation on the fly instead
2026-03-16 13:29:40 +01:00
parent 4736f8509d
commit 4483e464d1
46 changed files with 887 additions and 1841 deletions
--- a/apps/worker/package.json
+++ b/apps/worker/package.json
@@ -16,11 +16,11 @@
    "@openpanel/common": "workspace:*",
    "@openpanel/db": "workspace:*",
    "@openpanel/email": "workspace:*",
+    "@openpanel/importer": "workspace:*",
    "@openpanel/integrations": "workspace:^",
    "@openpanel/js-runtime": "workspace:*",
    "@openpanel/json": "workspace:*",
    "@openpanel/logger": "workspace:*",
-    "@openpanel/importer": "workspace:*",
    "@openpanel/payments": "workspace:*",
    "@openpanel/queue": "workspace:*",
    "@openpanel/redis": "workspace:*",
--- a/apps/worker/src/boot-workers.ts
+++ b/apps/worker/src/boot-workers.ts
@@ -1,10 +1,9 @@
-import type { Queue, WorkerOptions } from 'bullmq';
-import { Worker } from 'bullmq';
-
+import { performance } from 'node:perf_hooks';
+import { setTimeout as sleep } from 'node:timers/promises';
 import {
+  cronQueue,
  EVENTS_GROUP_QUEUES_SHARDS,
  type EventsQueuePayloadIncomingEvent,
-  cronQueue,
  eventsGroupQueues,
  gscQueue,
  importQueue,
@@ -15,14 +14,12 @@ import {
  sessionsQueue,
 } from '@openpanel/queue';
 import { getRedisQueue } from '@openpanel/redis';
-
-import { performance } from 'node:perf_hooks';
-import { setTimeout as sleep } from 'node:timers/promises';
+import type { Queue, WorkerOptions } from 'bullmq';
+import { Worker } from 'bullmq';
 import { Worker as GroupWorker } from 'groupmq';
-
 import { cronJob } from './jobs/cron';
-import { gscJob } from './jobs/gsc';
 import { incomingEvent } from './jobs/events.incoming-event';
+import { gscJob } from './jobs/gsc';
 import { importJob } from './jobs/import';
 import { insightsProjectJob } from './jobs/insights';
 import { miscJob } from './jobs/misc';
@@ -95,7 +92,7 @@ function getConcurrencyFor(queueName: string, defaultValue = 1): number {
  return defaultValue;
 }

-export async function bootWorkers() {
+export function bootWorkers() {
  const enabledQueues = getEnabledQueues();

  const workers: (Worker | GroupWorker<any>)[] = [];
@@ -119,12 +116,14 @@ export async function bootWorkers() {

  for (const index of eventQueuesToStart) {
    const queue = eventsGroupQueues[index];
-    if (!queue) continue;
+    if (!queue) {
+      continue;
+    }

    const queueName = `events_${index}`;
    const concurrency = getConcurrencyFor(
      queueName,
-      Number.parseInt(process.env.EVENT_JOB_CONCURRENCY || '10', 10),
+      Number.parseInt(process.env.EVENT_JOB_CONCURRENCY || '10', 10)
    );

    const worker = new GroupWorker<EventsQueuePayloadIncomingEvent['payload']>({
@@ -132,7 +131,7 @@ export async function bootWorkers() {
      concurrency,
      logger: process.env.NODE_ENV === 'production' ? queueLogger : undefined,
      blockingTimeoutSec: Number.parseFloat(
-        process.env.EVENT_BLOCKING_TIMEOUT_SEC || '1',
+        process.env.EVENT_BLOCKING_TIMEOUT_SEC || '1'
      ),
      handler: async (job) => {
        return await incomingEvent(job.data);
@@ -172,7 +171,7 @@ export async function bootWorkers() {
    const notificationWorker = new Worker(
      notificationQueue.name,
      notificationJob,
-      { ...workerOptions, concurrency },
+      { ...workerOptions, concurrency }
    );
    workers.push(notificationWorker);
    logger.info('Started worker for notification', { concurrency });
@@ -224,7 +223,7 @@ export async function bootWorkers() {

  if (workers.length === 0) {
    logger.warn(
-      'No workers started. Check ENABLED_QUEUES environment variable.',
+      'No workers started. Check ENABLED_QUEUES environment variable.'
    );
  }

@@ -254,7 +253,7 @@ export async function bootWorkers() {
          const elapsed = job.finishedOn - job.processedOn;
          eventsGroupJobDuration.observe(
            { name: worker.name, status: 'failed' },
-            elapsed,
+            elapsed
          );
        }
        logger.error('job failed', {
@@ -267,23 +266,6 @@ export async function bootWorkers() {
      }
    });

-    (worker as Worker).on('completed', (job) => {
-      if (job) {
-        if (job.processedOn && job.finishedOn) {
-          const elapsed = job.finishedOn - job.processedOn;
-          logger.info('job completed', {
-            jobId: job.id,
-            worker: worker.name,
-            elapsed,
-          });
-          eventsGroupJobDuration.observe(
-            { name: worker.name, status: 'success' },
-            elapsed,
-          );
-        }
-      }
-    });
-
    (worker as Worker).on('ioredis:close', () => {
      logger.error('worker closed due to ioredis:close', {
        worker: worker.name,
@@ -293,7 +275,7 @@ export async function bootWorkers() {

  async function exitHandler(
    eventName: string,
-    evtOrExitCodeOrError: number | string | Error,
+    evtOrExitCodeOrError: number | string | Error
  ) {
    // Log the actual error details for unhandled rejections/exceptions
    if (evtOrExitCodeOrError instanceof Error) {
@@ -339,7 +321,7 @@ export async function bootWorkers() {
      process.on(evt, (code) => {
        exitHandler(evt, code);
      });
-    },
+    }
  );

  return workers;
--- a/apps/worker/src/jobs/cron.salt.ts
+++ b/apps/worker/src/jobs/cron.salt.ts
@@ -33,7 +33,7 @@ async function generateNewSalt() {
    return created;
  });

-  getSalts.clear();
+  await getSalts.clear();

  return newSalt;
 }
--- a/apps/worker/src/jobs/events.incoming-event.ts
+++ b/apps/worker/src/jobs/events.incoming-event.ts
@@ -1,9 +1,5 @@
 import { getTime, isSameDomain, parsePath } from '@openpanel/common';
-import {
-  getReferrerWithQuery,
-  parseReferrer,
-  parseUserAgent,
-} from '@openpanel/common/server';
+import { getReferrerWithQuery, parseReferrer } from '@openpanel/common/server';
 import type { IServiceCreateEventPayload, IServiceEvent } from '@openpanel/db';
 import {
  checkNotificationRulesForEvent,
@@ -14,10 +10,12 @@ import {
 } from '@openpanel/db';
 import type { ILogger } from '@openpanel/logger';
 import type { EventsQueuePayloadIncomingEvent } from '@openpanel/queue';
-import { getLock } from '@openpanel/redis';
 import { anyPass, isEmpty, isNil, mergeDeepRight, omit, reject } from 'ramda';
 import { logger as baseLogger } from '@/utils/logger';
-import { createSessionEndJob, getSessionEnd } from '@/utils/session-handler';
+import {
+  createSessionEndJob,
+  extendSessionEndJob,
+} from '@/utils/session-handler';

 const GLOBAL_PROPERTIES = ['__path', '__referrer', '__timestamp', '__revenue'];

@@ -93,7 +91,8 @@ export async function incomingEvent(
    projectId,
    deviceId,
    sessionId,
-    uaInfo: _uaInfo,
+    uaInfo,
+    session,
  } = jobPayload;
  const properties = body.properties ?? {};
  const reqId = headers['request-id'] ?? 'unknown';
@@ -121,16 +120,15 @@ export async function incomingEvent(
    ? null
    : parseReferrer(getProperty('__referrer'));
  const utmReferrer = getReferrerWithQuery(query);
-  const userAgent = headers['user-agent'];
  const sdkName = headers['openpanel-sdk-name'];
  const sdkVersion = headers['openpanel-sdk-version'];
-  // TODO: Remove both user-agent and parseUserAgent
-  const uaInfo = _uaInfo ?? parseUserAgent(userAgent, properties);

-  const baseEvent = {
+  const baseEvent: IServiceCreateEventPayload = {
    name: body.name,
    profileId,
    projectId,
+    deviceId,
+    sessionId,
    properties: omit(GLOBAL_PROPERTIES, {
      ...properties,
      __hash: hash,
@@ -149,7 +147,7 @@ export async function incomingEvent(
    origin,
    referrer: referrer?.url || '',
    referrerName: utmReferrer?.name || referrer?.name || referrer?.url,
-    referrerType: referrer?.type || utmReferrer?.type || '',
+    referrerType: utmReferrer?.type || referrer?.type || '',
    os: uaInfo.os,
    osVersion: uaInfo.osVersion,
    browser: uaInfo.browser,
@@ -161,16 +159,17 @@ export async function incomingEvent(
      body.name === 'revenue' && '__revenue' in properties
        ? parseRevenue(properties.__revenue)
        : undefined,
-  } as const;
+  };

  // if timestamp is from the past we dont want to create a new session
  if (uaInfo.isServer || isTimestampFromThePast) {
-    const session = profileId
-      ? await sessionBuffer.getExistingSession({
-          profileId,
-          projectId,
-        })
-      : null;
+    const session =
+      profileId && !isTimestampFromThePast
+        ? await sessionBuffer.getExistingSession({
+            profileId,
+            projectId,
+          })
+        : null;

    const payload = {
      ...baseEvent,
@@ -198,82 +197,48 @@ export async function incomingEvent(
    return createEventAndNotify(payload as IServiceEvent, logger, projectId);
  }

-  const sessionEnd = await getSessionEnd({
-    projectId,
-    deviceId,
-    profileId,
-  });
-  const activeSession = sessionEnd
-    ? await sessionBuffer.getExistingSession({
-        sessionId: sessionEnd.sessionId,
-      })
-    : null;
-
  const payload: IServiceCreateEventPayload = merge(baseEvent, {
-    deviceId: sessionEnd?.deviceId ?? deviceId,
-    sessionId: sessionEnd?.sessionId ?? sessionId,
-    referrer: sessionEnd?.referrer ?? baseEvent.referrer,
-    referrerName: sessionEnd?.referrerName ?? baseEvent.referrerName,
-    referrerType: sessionEnd?.referrerType ?? baseEvent.referrerType,
-    // if the path is not set, use the last screen view path
-    path: baseEvent.path || activeSession?.exit_path || '',
-    origin: baseEvent.origin || activeSession?.exit_origin || '',
+    referrer: session?.referrer ?? baseEvent.referrer,
+    referrerName: session?.referrerName ?? baseEvent.referrerName,
+    referrerType: session?.referrerType ?? baseEvent.referrerType,
  } as Partial<IServiceCreateEventPayload>) as IServiceCreateEventPayload;

-  // If the triggering event is filtered, do not create session_start or the event (issue #2)
  const isExcluded = await isEventExcludedByProjectFilter(payload, projectId);
  if (isExcluded) {
    logger.info(
      'Skipping session_start and event (excluded by project filter)',
+      { event: payload.name, projectId }
+    );
+    return null;
+  }
+
+  if (session) {
+    await extendSessionEndJob({
+      projectId,
+      deviceId,
+    }).catch((error) => {
+      logger.error('Error finding and extending session end job', { error });
+      throw error;
+    });
+  } else {
+    await createEventAndNotify(
      {
-        event: payload.name,
-        projectId,
-      }
-    );
-    return null;
-  }
+        ...payload,
+        name: 'session_start',
+        createdAt: new Date(getTime(payload.createdAt) - 100),
+      },
+      logger,
+      projectId
+    ).catch((error) => {
+      logger.error('Error creating session start event', { event: payload });
+      throw error;
+    });

-  if (!sessionEnd) {
-    const locked = await getLock(
-      `session_start:${projectId}:${sessionId}`,
-      '1',
-      1000
-    );
-    if (locked) {
-      logger.info('Creating session start event', { event: payload });
-      await createEventAndNotify(
-        {
-          ...payload,
-          name: 'session_start',
-          createdAt: new Date(getTime(payload.createdAt) - 100),
-        },
-        logger,
-        projectId
-      ).catch((error) => {
-        logger.error('Error creating session start event', { event: payload });
-        throw error;
-      });
-    } else {
-      logger.info('Session start already claimed by another worker', {
-        event: payload,
-      });
-    }
-  }
-
-  const event = await createEventAndNotify(payload, logger, projectId);
-
-  if (!event) {
-    // Skip creating session end when event was excluded
-    return null;
-  }
-
-  if (!sessionEnd) {
-    logger.info('Creating session end job', { event: payload });
    await createSessionEndJob({ payload }).catch((error) => {
      logger.error('Error creating session end job', { event: payload });
      throw error;
    });
  }

-  return event;
+  return createEventAndNotify(payload, logger, projectId);
 }
--- a/apps/worker/src/jobs/events.incoming-events.test.ts
+++ b/apps/worker/src/jobs/events.incoming-events.test.ts
@@ -186,6 +186,11 @@ describe('incomingEvent', () => {
      projectId,
      deviceId,
      sessionId: 'session-123',
+      session: {
+        referrer: '',
+        referrerName: '',
+        referrerType: '',
+      },
    };

    const changeDelay = vi.fn();
--- a/apps/worker/src/metrics.ts
+++ b/apps/worker/src/metrics.ts
@@ -1,5 +1,3 @@
-import client from 'prom-client';
-
 import {
  botBuffer,
  eventBuffer,
@@ -8,6 +6,7 @@ import {
  sessionBuffer,
 } from '@openpanel/db';
 import { cronQueue, eventsGroupQueues, sessionsQueue } from '@openpanel/queue';
+import client from 'prom-client';

 const Registry = client.Registry;

@@ -20,7 +19,7 @@ export const eventsGroupJobDuration = new client.Histogram({
  name: 'job_duration_ms',
  help: 'Duration of job processing (in ms)',
  labelNames: ['name', 'status'],
-  buckets: [10, 25, 50, 100, 250, 500, 750, 1000, 2000, 5000, 10000, 30000], // 10ms to 30s
+  buckets: [10, 25, 50, 100, 250, 500, 750, 1000, 2000, 5000, 10_000, 30_000], // 10ms to 30s
 });

 register.registerMetric(eventsGroupJobDuration);
@@ -28,57 +27,61 @@ register.registerMetric(eventsGroupJobDuration);
 queues.forEach((queue) => {
  register.registerMetric(
    new client.Gauge({
-      name: `${queue.name.replace(/[\{\}]/g, '')}_active_count`,
+      name: `${queue.name.replace(/[{}]/g, '')}_active_count`,
      help: 'Active count',
      async collect() {
        const metric = await queue.getActiveCount();
        this.set(metric);
      },
-    }),
+    })
  );

  register.registerMetric(
    new client.Gauge({
-      name: `${queue.name.replace(/[\{\}]/g, '')}_delayed_count`,
+      name: `${queue.name.replace(/[{}]/g, '')}_delayed_count`,
      help: 'Delayed count',
      async collect() {
-        const metric = await queue.getDelayedCount();
-        this.set(metric);
+        if ('getDelayedCount' in queue) {
+          const metric = await queue.getDelayedCount();
+          this.set(metric);
+        } else {
+          this.set(0);
+        }
      },
-    }),
+    })
  );

  register.registerMetric(
    new client.Gauge({
-      name: `${queue.name.replace(/[\{\}]/g, '')}_failed_count`,
+      name: `${queue.name.replace(/[{}]/g, '')}_failed_count`,
      help: 'Failed count',
      async collect() {
        const metric = await queue.getFailedCount();
        this.set(metric);
      },
-    }),
+    })
  );

  register.registerMetric(
    new client.Gauge({
-      name: `${queue.name.replace(/[\{\}]/g, '')}_completed_count`,
+      name: `${queue.name.replace(/[{}]/g, '')}_completed_count`,
      help: 'Completed count',
      async collect() {
        const metric = await queue.getCompletedCount();
        this.set(metric);
      },
-    }),
+    })
  );

  register.registerMetric(
    new client.Gauge({
-      name: `${queue.name.replace(/[\{\}]/g, '')}_waiting_count`,
+      name: `${queue.name.replace(/[{}]/g, '')}_waiting_count`,
      help: 'Waiting count',
      async collect() {
        const metric = await queue.getWaitingCount();
        this.set(metric);
      },
-    }),
+    })
  );
 });

@@ -90,7 +93,7 @@ register.registerMetric(
      const metric = await eventBuffer.getBufferSize();
      this.set(metric);
    },
-  }),
+  })
 );

 register.registerMetric(
@@ -101,7 +104,7 @@ register.registerMetric(
      const metric = await profileBuffer.getBufferSize();
      this.set(metric);
    },
-  }),
+  })
 );

 register.registerMetric(
@@ -112,7 +115,7 @@ register.registerMetric(
      const metric = await botBuffer.getBufferSize();
      this.set(metric);
    },
-  }),
+  })
 );

 register.registerMetric(
@@ -123,7 +126,7 @@ register.registerMetric(
      const metric = await sessionBuffer.getBufferSize();
      this.set(metric);
    },
-  }),
+  })
 );

 register.registerMetric(
@@ -134,5 +137,5 @@ register.registerMetric(
      const metric = await replayBuffer.getBufferSize();
      this.set(metric);
    },
-  }),
+  })
 );
--- a/apps/worker/src/utils/session-handler.ts
+++ b/apps/worker/src/utils/session-handler.ts
@@ -1,13 +1,39 @@
 import type { IServiceCreateEventPayload } from '@openpanel/db';
-import {
-  type EventsQueuePayloadCreateSessionEnd,
-  sessionsQueue,
-} from '@openpanel/queue';
-import type { Job } from 'bullmq';
-import { logger } from './logger';
+import { sessionsQueue } from '@openpanel/queue';

 export const SESSION_TIMEOUT = 1000 * 60 * 30;

+const CHANGE_DELAY_THROTTLE_MS = process.env.CHANGE_DELAY_THROTTLE_MS
+  ? Number.parseInt(process.env.CHANGE_DELAY_THROTTLE_MS, 10)
+  : 60_000; // 1 minute
+
+const CHANGE_DELAY_THROTTLE_MAP = new Map<string, number>();
+
+export async function extendSessionEndJob({
+  projectId,
+  deviceId,
+}: {
+  projectId: string;
+  deviceId: string;
+}) {
+  const last = CHANGE_DELAY_THROTTLE_MAP.get(`${projectId}:${deviceId}`) ?? 0;
+  const isThrottled = Date.now() - last < CHANGE_DELAY_THROTTLE_MS;
+
+  if (isThrottled) {
+    return;
+  }
+
+  const jobId = getSessionEndJobId(projectId, deviceId);
+  const job = await sessionsQueue.getJob(jobId);
+
+  if (!job) {
+    return;
+  }
+
+  await job.changeDelay(SESSION_TIMEOUT);
+  CHANGE_DELAY_THROTTLE_MAP.set(`${projectId}:${deviceId}`, Date.now());
+}
+
 const getSessionEndJobId = (projectId: string, deviceId: string) =>
  `sessionEnd:${projectId}:${deviceId}`;

@@ -33,106 +59,3 @@ export function createSessionEndJob({
    }
  );
 }
-
-export async function getSessionEnd({
-  projectId,
-  deviceId,
-  profileId,
-}: {
-  projectId: string;
-  deviceId: string;
-  profileId: string;
-}) {
-  const sessionEnd = await getSessionEndJob({
-    projectId,
-    deviceId,
-  });
-
-  if (sessionEnd) {
-    const existingSessionIsAnonymous =
-      sessionEnd.job.data.payload.profileId ===
-      sessionEnd.job.data.payload.deviceId;
-
-    const eventIsIdentified =
-      profileId && sessionEnd.job.data.payload.profileId !== profileId;
-
-    if (existingSessionIsAnonymous && eventIsIdentified) {
-      await sessionEnd.job.updateData({
-        ...sessionEnd.job.data,
-        payload: {
-          ...sessionEnd.job.data.payload,
-          profileId,
-        },
-      });
-    }
-
-    await sessionEnd.job.changeDelay(SESSION_TIMEOUT);
-    return sessionEnd.job.data.payload;
-  }
-
-  return null;
-}
-
-export async function getSessionEndJob(args: {
-  projectId: string;
-  deviceId: string;
-  retryCount?: number;
-}): Promise<{
-  deviceId: string;
-  job: Job<EventsQueuePayloadCreateSessionEnd>;
-} | null> {
-  const { retryCount = 0 } = args;
-
-  if (retryCount >= 6) {
-    throw new Error('Failed to get session end');
-  }
-
-  async function handleJobStates(
-    job: Job<EventsQueuePayloadCreateSessionEnd>,
-    deviceId: string
-  ): Promise<{
-    deviceId: string;
-    job: Job<EventsQueuePayloadCreateSessionEnd>;
-  } | null> {
-    const state = await job.getState();
-    if (state !== 'delayed') {
-      logger.debug(`[session-handler] Session end job is in "${state}" state`, {
-        state,
-        retryCount,
-        jobTimestamp: new Date(job.timestamp).toISOString(),
-        jobDelta: Date.now() - job.timestamp,
-        jobId: job.id,
-        payload: job.data.payload,
-      });
-    }
-
-    if (state === 'delayed' || state === 'waiting') {
-      return { deviceId, job };
-    }
-
-    if (state === 'active') {
-      await new Promise((resolve) => setTimeout(resolve, 100));
-      return getSessionEndJob({
-        ...args,
-        retryCount: retryCount + 1,
-      });
-    }
-
-    if (state === 'completed') {
-      await job.remove();
-    }
-
-    return null;
-  }
-
-  // Check current device job
-  const currentJob = await sessionsQueue.getJob(
-    getSessionEndJobId(args.projectId, args.deviceId)
-  );
-  if (currentJob) {
-    return await handleJobStates(currentJob, args.deviceId);
-  }
-
-  // Create session
-  return null;
-}