wip

Author: lindesvard

apps/worker/src/boot-workers.ts

@@ -33,7 +33,6 @@ export async function bootWorkers() {
   const eventsGroupWorker = new GroupWorker<
     EventsQueuePayloadIncomingEvent['payload']
   >({
-    // redis: getRedisGroupQueue(),
     queue: eventsGroupQueue,
     handler: async (job) => {
       logger.info('processing event (group queue)', {

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

@@ -9,7 +9,9 @@ import type { IClickhouseSession } from '../services/session.service';
 import { BaseBuffer } from './base-buffer';
 
 export class SessionBuffer extends BaseBuffer {
-  private batchSize = 0;
+  private batchSize = process.env.SESSION_BUFFER_BATCH_SIZE
+    ? Number.parseInt(process.env.SESSION_BUFFER_BATCH_SIZE, 10)
+    : 1000;
 
   private readonly redisKey = 'session-buffer';
   private redis: Redis;
@@ -62,11 +64,11 @@ export class SessionBuffer extends BaseBuffer {
       if (duration > 0) {
         newSession.duration = duration;
       } else {
-        // this.logger.warn('Session duration is negative', {
-        //   duration,
-        //   event,
-        //   session: newSession,
-        // });
+        this.logger.warn('Session duration is negative', {
+          duration,
+          event,
+          session: newSession,
+        });
       }
       newSession.properties = toDots({
         ...(event.properties || {}),

packages/queue/src/queues.ts

@@ -109,7 +109,7 @@ export const eventsGroupQueue = new GroupQueue<
 >({
   namespace: 'group_events',
   redis: getRedisGroupQueue(),
-  orderingDelayMs: 5_000,
+  orderingDelayMs: 2_000,
   keepCompleted: 1000,
   keepFailed: Number.MAX_SAFE_INTEGER,
 });