fix comments

Author: lindesvard

apps/api/src/hooks/duplicate.hook.ts

@@ -8,12 +8,19 @@ export async function duplicateHook(
   }>,
   reply: FastifyReply,
 ) {
-  const isDuplicate = await isDuplicatedEvent({
-    ip: req.clientIp ?? '',
-    origin: req.headers.origin ?? '',
-    payload: req.body,
-    projectId: (req.headers['openpanel-client-id'] as string) || '',
-  });
+  const ip = req.clientIp;
+  const origin = req.headers.origin;
+  const clientId = req.headers['openpanel-client-id'];
+  const shouldCheck = ip && origin && clientId;
+
+  const isDuplicate = shouldCheck
+    ? await isDuplicatedEvent({
+        ip,
+        origin,
+        payload: req.body,
+        projectId: clientId as string,
+      })
+    : false;
 
   if (isDuplicate) {
     return reply.status(200).send('Duplicate event');

apps/api/src/routes/track.router.ts

@@ -2,11 +2,13 @@ import { handler } from '@/controllers/track.controller';
 import type { FastifyPluginCallback } from 'fastify';
 
 import { clientHook } from '@/hooks/client.hook';
+import { duplicateHook } from '@/hooks/duplicate.hook';
 import { isBotHook } from '@/hooks/is-bot.hook';
 
 const trackRouter: FastifyPluginCallback = async (fastify) => {
-  fastify.addHook('preHandler', clientHook);
   fastify.addHook('preHandler', isBotHook);
+  fastify.addHook('preValidation', duplicateHook);
+  fastify.addHook('preHandler', clientHook);
 
   fastify.route({
     method: 'POST',

apps/api/src/utils/auth.ts

@@ -137,7 +137,7 @@ export async function validateSdkRequest(
 
   if (client.secret && clientSecret) {
     const isVerified = await getCache(
-      `client:auth:${clientId}:${clientSecret.slice(0, 5)}`,
+      `client:auth:${clientId}:${Buffer.from(clientSecret).toString('base64')}`,
       60 * 5,
       async () => await verifyPassword(clientSecret, client.secret!),
       true,

apps/worker/scripts/cleanup-old-event-buffer-keys.ts

@@ -104,8 +104,12 @@ async function cleanupOldEventBufferKeys(): Promise<CleanupStats> {
           const events = await redis.lrange(sessionKey, 0, -1);
 
           if (events.length > 0) {
-            // Move events to new queue
-            await redis.rpush(newQueueKey, ...events);
+            // Move events to new queue in safe batches to avoid exceeding V8 arg limits
+            const chunkSize = 1000;
+            for (let offset = 0; offset < events.length; offset = chunkSize) {
+              const chunk = events.slice(offset, offset + chunkSize);
+              await redis.rpush(newQueueKey, ...chunk);
+            }
             // Update buffer counter
             await redis.incrby('event_buffer:total_count', events.length);
             totalEventsMigrated += events.length;

apps/worker/src/boot-cron.ts

@@ -58,7 +58,7 @@ export async function bootCron() {
     // TODO: Switch to getJobSchedulers
     const repeatableJobs = await cronQueue.getRepeatableJobs();
     for (const repeatableJob of repeatableJobs) {
-      cronQueue.removeRepeatableByKey(repeatableJob.key);
+      await cronQueue.removeRepeatableByKey(repeatableJob.key);
     }
 
     // Add repeatable jobs

apps/worker/src/boot-workers.ts

@@ -19,7 +19,7 @@ import { setTimeout as sleep } from 'node:timers/promises';
 import { Worker as GroupWorker } from 'groupmq';
 
 import { cronJob } from './jobs/cron';
-import { incomingEventPure } from './jobs/events.incoming-event';
+import { incomingEvent } from './jobs/events.incoming-event';
 import { importJob } from './jobs/import';
 import { miscJob } from './jobs/misc';
 import { notificationJob } from './jobs/notification';
@@ -122,7 +122,7 @@ export async function bootWorkers() {
         process.env.EVENT_BLOCKING_TIMEOUT_SEC || '1',
       ),
       handler: async (job) => {
-        return await incomingEventPure(job.data);
+        return await incomingEvent(job.data);
       },
     });
 
@@ -184,7 +184,7 @@ export async function bootWorkers() {
       concurrency,
     });
     workers.push(importWorker);
-    logger.info('Started worker for misc', { concurrency });
+    logger.info('Started worker for import', { concurrency });
   }
 
   if (workers.length === 0) {

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

@@ -45,18 +45,8 @@ async function createEventAndNotify(
 }
 
 export async function incomingEvent(
-  job: Job<EventsQueuePayloadIncomingEvent>,
-  token?: string,
-) {
-  return incomingEventPure(job.data.payload, job, token);
-}
-
-export async function incomingEventPure(
   jobPayload: EventsQueuePayloadIncomingEvent['payload'],
-  job?: Job<EventsQueuePayloadIncomingEvent>,
-  token?: string,
 ) {
-  await getRedisCache().incr('queue:counter');
   const {
     geo,
     event: body,