speed up worker

apps/worker/src/jobs/events.incoming-event.ts

@@ -13,6 +13,7 @@ import { createEvent, getEvents } from '@openpanel/db';
 import { findJobByPrefix } from '@openpanel/queue';
 import { eventsQueue } from '@openpanel/queue/src/queues';
 import type { EventsQueuePayloadIncomingEvent } from '@openpanel/queue/src/queues';
+import { redis } from '@openpanel/redis';
 
 const GLOBAL_PROPERTIES = ['__path', '__referrer'];
 const SESSION_TIMEOUT = 1000 * 60 * 30;
@@ -89,18 +90,22 @@ export async function incomingEvent(job: Job<EventsQueuePayloadIncomingEvent>) {
     return createEvent(payload);
   }
 
-  const [sessionEndJobCurrentDeviceId, sessionEndJobPreviousDeviceId] =
+  const [sessionEndKeys, eventsKeys] = await Promise.all([
-    await Promise.all([
+    redis.keys(`bull:events:sessionEnd:${projectId}:*`),
-      findJobByPrefix(
+    redis.keys(`bull:events:event:${projectId}:*`),
-        eventsQueue,
-        `sessionEnd:${projectId}:${currentDeviceId}:`
-      ),
-      findJobByPrefix(
-        eventsQueue,
-        `sessionEnd:${projectId}:${previousDeviceId}:`
-      ),
   ]);
 
+  const sessionEndJobCurrentDeviceId = await findJobByPrefix(
+    eventsQueue,
+    sessionEndKeys,
+    `sessionEnd:${projectId}:${currentDeviceId}:`
+  );
+  const sessionEndJobPreviousDeviceId = await findJobByPrefix(
+    eventsQueue,
+    sessionEndKeys,
+    `sessionEnd:${projectId}:${previousDeviceId}:`
+  );
+
   const createSessionStart =
     !sessionEndJobCurrentDeviceId && !sessionEndJobPreviousDeviceId;
 
@@ -130,12 +135,15 @@ export async function incomingEvent(job: Job<EventsQueuePayloadIncomingEvent>) {
     );
   }
 
-  const [[sessionStartEvent], prevEventJob] = await Promise.all([
+  const prevEventJob = await findJobByPrefix(
-    getEvents(
+    eventsQueue,
+    eventsKeys,
+    `event:${projectId}:${deviceId}:`
+  );
+
+  const [sessionStartEvent] = await getEvents(
     `SELECT * FROM events WHERE name = 'session_start' AND device_id = ${escape(deviceId)} AND project_id = ${escape(projectId)} ORDER BY created_at DESC LIMIT 1`
-    ),
+  );
-    findJobByPrefix(eventsQueue, `event:${projectId}:${deviceId}:`),
-  ]);
 
   const payload: Omit<IServiceCreateEventPayload, 'id'> = {
     name: body.name,

packages/queue/src/utils.ts

@@ -2,23 +2,38 @@ import type { Queue } from 'bullmq';
 
 export async function findJobByPrefix<T>(
   queue: Queue<T, any, string>,
+  keys: string[],
   matcher: string
 ) {
-  const delayed = await queue.getJobs('delayed');
+  const filtered = keys.filter((key) => key.includes(matcher));
-  const filtered = delayed.filter((job) =>
-    job?.opts?.jobId?.startsWith(matcher)
-  );
   const getTime = (val?: string) => {
     if (!val) return null;
     const match = val.match(/:(\d+)$/);
     return match?.[1] ? parseInt(match[1], 10) : null;
   };
+
   filtered.sort((a, b) => {
-    const aTime = getTime(a?.opts?.jobId);
+    const aTime = getTime(a);
-    const bTime = getTime(b?.opts?.jobId);
+    const bTime = getTime(b);
     if (aTime === null) return 1;
     if (bTime === null) return -1;
     return aTime - bTime;
   });
-  return filtered[0];
+
+  async function getJob(index: number) {
+    if (index >= filtered.length) return null;
+
+    const key = filtered[index]?.replace(/^bull:events:/, '');
+    // return new Promise((resolve) => )
+    if (key) {
+      const job = await queue.getJob(key);
+      if ((await job?.getState()) === 'delayed') {
+        return job;
+      }
+    }
+
+    return getJob(index + 1);
+  }
+
+  return getJob(0);
 }