feat(grouper): add counters to the grouper worker (#521)

e11sy · Copilot · neSpecc · web-flow · commit ae01c55e4270 · 2026-03-04T00:54:39.000+03:00
* chore(grouper): add counters to the grouper worker

* chore(): eslint fix

* chore(): clean up

* chore(grouper): remove redundant rate-limit increment logic

* chore(grouper): remove redundant mocks

* chore(): eslint fix

* chore(): change metric type

* Update workers/grouper/src/index.ts

Co-authored-by: Copilot &lt;175728472+Copilot@users.noreply.github.com&gt;

* imp(): use lua for create if not exists, to avoid race-cond

* add logs to sentry worker

* chore(): remove redundant import

---------

Co-authored-by: Copilot &lt;175728472+Copilot@users.noreply.github.com&gt;
Co-authored-by: Peter Savchenko &lt;specc.dev@gmail.com&gt;
diff --git a/workers/grouper/src/index.ts b/workers/grouper/src/index.ts
@@ -306,6 +306,74 @@ export default class GrouperWorker extends Worker {
         });
       }
     }
+
+    await this.recordProjectMetrics(task.projectId, 'events-accepted');
+  }
+
+  /**
+   * Build RedisTimeSeries key for project metrics.
+   *
+   * @param projectId - id of the project
+   * @param metricType - metric type identifier
+   * @param granularity - time granularity
+   */
+  private getTimeSeriesKey(
+    projectId: string,
+    metricType: string,
+    granularity: 'minutely' | 'hourly' | 'daily'
+  ): string {
+    return `ts:project-${metricType}:${projectId}:${granularity}`;
+  }
+
+  /**
+   * Record project metrics to Redis TimeSeries.
+   *
+   * @param projectId - id of the project
+   * @param metricType - metric type identifier
+   */
+  private async recordProjectMetrics(projectId: string, metricType: string): Promise<void> {
+    const minutelyKey = this.getTimeSeriesKey(projectId, metricType, 'minutely');
+    const hourlyKey = this.getTimeSeriesKey(projectId, metricType, 'hourly');
+    const dailyKey = this.getTimeSeriesKey(projectId, metricType, 'daily');
+
+    const labels: Record<string, string> = {
+      type: 'error',
+      status: metricType,
+      project: projectId,
+    };
+
+    const series = [
+      {
+        key: minutelyKey,
+        label: 'minutely',
+        retentionMs: TimeMs.DAY,
+      },
+      {
+        key: hourlyKey,
+        label: 'hourly',
+        retentionMs: TimeMs.WEEK,
+      },
+      {
+        key: dailyKey,
+        label: 'daily',
+        // eslint-disable-next-line @typescript-eslint/no-magic-numbers
+        retentionMs: 90 * TimeMs.DAY,
+      },
+    ];
+
+    const operations = series.map(({ key, label, retentionMs }) => ({
+      label,
+      promise: this.redis.safeTsAdd(key, 1, labels, retentionMs),
+    }));
+
+    const results = await Promise.allSettled(operations.map((op) => op.promise));
+
+    results.forEach((result, index) => {
+      if (result.status === 'rejected') {
+        const { label } = operations[index];
+        this.logger.error(`Failed to add ${label} TS for ${metricType}`, result.reason);
+      }
+    });
   }
 
   /**
diff --git a/workers/grouper/src/redisHelper.ts b/workers/grouper/src/redisHelper.ts
@@ -110,6 +110,176 @@ export default class RedisHelper {
     return result === null;
   }
 
+  /**
+   * Creates a RedisTimeSeries key if it doesn't exist.
+   *
+   * @param key - time series key
+   * @param labels - labels to attach to the time series
+   * @param retentionMs - optional retention in milliseconds
+   */
+  public async tsCreateIfNotExists(
+    key: string,
+    labels: Record<string, string>,
+    retentionMs = 0
+  ): Promise<void> {
+    const script = `
+      if redis.call('EXISTS', KEYS[1]) == 1 then
+        return 0
+      end
+
+      redis.call('TS.CREATE', KEYS[1], unpack(ARGV))
+      return 1
+    `;
+
+    const args: string[] = [];
+
+    if (retentionMs > 0) {
+      args.push('RETENTION', Math.floor(retentionMs).toString());
+    }
+
+    args.push(...this.buildLabelArguments(labels));
+
+    await this.redisClient.eval(
+      script,
+      {
+        keys: [key],
+        arguments: args,
+      }
+    );
+  }
+
+  /**
+   * Increments a RedisTimeSeries key with labels and timestamp.
+   *
+   * @param key - time series key
+   * @param value - value to increment by
+   * @param timestampMs - timestamp in milliseconds, defaults to current time
+   * @param labels - labels to attach to the sample
+   */
+  public async tsIncrBy(
+    key: string,
+    value: number,
+    timestampMs = 0,
+    labels: Record<string, string> = {}
+  ): Promise<void> {
+    const labelArgs = this.buildLabelArguments(labels);
+    const timestamp = timestampMs === 0 ? Date.now() : timestampMs;
+
+    const args: string[] = [
+      'TS.INCRBY',
+      key,
+      value.toString(),
+      'TIMESTAMP',
+      Math.floor(timestamp).toString(),
+      ...labelArgs,
+    ];
+
+    await this.redisClient.sendCommand(args);
+  }
+
+  /**
+   * Ensures that a RedisTimeSeries key exists and increments it safely.
+   *
+   * @param key - time series key
+   * @param value - value to increment by
+   * @param labels - labels to attach to the time series
+   * @param retentionMs - optional retention in milliseconds
+   */
+  public async safeTsIncrBy(
+    key: string,
+    value: number,
+    labels: Record<string, string>,
+    retentionMs = 0
+  ): Promise<void> {
+    const timestamp = Date.now();
+
+    try {
+      await this.tsIncrBy(key, value, timestamp, labels);
+    } catch (error) {
+      if (error instanceof Error && error.message.includes('TSDB: key does not exist')) {
+        this.logger.warn(`TS key ${key} does not exist, creating it...`);
+        await this.tsCreateIfNotExists(key, labels, retentionMs);
+        await this.tsIncrBy(key, value, timestamp, labels);
+      } else {
+        throw error;
+      }
+    }
+  }
+
+  /**
+   * Adds a sample to a RedisTimeSeries key.
+   *
+   * @param key - time series key
+   * @param value - value to add
+   * @param timestampMs - timestamp in milliseconds, defaults to current time
+   * @param labels - labels to attach to the sample
+   */
+  public async tsAdd(
+    key: string,
+    value: number,
+    timestampMs = 0,
+    labels: Record<string, string> = {}
+  ): Promise<void> {
+    const labelArgs = this.buildLabelArguments(labels);
+    const timestamp = timestampMs === 0 ? Date.now() : timestampMs;
+
+    const args: string[] = [
+      'TS.ADD',
+      key,
+      Math.floor(timestamp).toString(),
+      value.toString(),
+      'ON_DUPLICATE',
+      'SUM',
+      ...labelArgs,
+    ];
+
+    await this.redisClient.sendCommand(args);
+  }
+
+  /**
+   * Ensures that a RedisTimeSeries key exists and adds a sample safely.
+   *
+   * @param key - time series key
+   * @param value - value to add
+   * @param labels - labels to attach to the time series
+   * @param retentionMs - optional retention in milliseconds
+   */
+  public async safeTsAdd(
+    key: string,
+    value: number,
+    labels: Record<string, string>,
+    retentionMs = 0
+  ): Promise<void> {
+    const timestamp = Date.now();
+
+    try {
+      await this.tsAdd(key, value, timestamp, labels);
+    } catch (error) {
+      if (error instanceof Error && error.message.includes('TSDB: key does not exist')) {
+        this.logger.warn(`TS key ${key} does not exist, creating it...`);
+        await this.tsCreateIfNotExists(key, labels, retentionMs);
+        await this.tsAdd(key, value, timestamp, labels);
+      } else {
+        throw error;
+      }
+    }
+  }
+
+  /**
+   * Build label arguments for RedisTimeSeries commands
+   *
+   * @param labels - labels to attach to the time series
+   */
+  private buildLabelArguments(labels: Record<string, string>): string[] {
+    const labelArgs: string[] = [ 'LABELS' ];
+
+    for (const [labelKey, labelValue] of Object.entries(labels)) {
+      labelArgs.push(labelKey, labelValue);
+    }
+
+    return labelArgs;
+  }
+
   /**
    * Creates callback function for Redis operations
    *
@@ -130,4 +300,4 @@ export default class RedisHelper {
       resolve(resp !== 'OK');
     };
   }
-}
+}
diff --git a/workers/grouper/tests/index.test.ts b/workers/grouper/tests/index.test.ts
@@ -7,6 +7,7 @@ import type { Collection } from 'mongodb';
 import { MongoClient } from 'mongodb';
 import type { ErrorsCatcherType, EventAddons, EventData } from '@hawk.so/types';
 import { MS_IN_SEC } from '../../../lib/utils/consts';
+import TimeMs from '../../../lib/utils/time';
 import * as mongodb from 'mongodb';
 import { patch } from '@n1ru4l/json-patch-plus';
 
@@ -115,7 +116,6 @@ describe('GrouperWorker', () => {
   let projectsCollection: Collection;
   let redisClient: RedisClientType;
   let worker: GrouperWorker;
-
   beforeAll(async () => {
     worker = new GrouperWorker();
 
@@ -743,6 +743,84 @@ describe('GrouperWorker', () => {
     });
   });
 
+  describe('Events-accepted metrics', () => {
+    test('writes minutely, hourly, and daily samples after handling an event', async () => {
+      const safeTsAddSpy = jest.spyOn((worker as any).redis, 'safeTsAdd');
+
+      try {
+        await worker.handle(generateTask());
+
+        expect(safeTsAddSpy).toHaveBeenCalledTimes(3);
+
+        const expectedLabels = {
+          type: 'error',
+          status: 'events-accepted',
+          project: projectIdMock,
+        };
+
+        expect(safeTsAddSpy).toHaveBeenNthCalledWith(
+          1,
+          `ts:project-events-accepted:${projectIdMock}:minutely`,
+          1,
+          expectedLabels,
+          TimeMs.DAY
+        );
+        expect(safeTsAddSpy).toHaveBeenNthCalledWith(
+          2,
+          `ts:project-events-accepted:${projectIdMock}:hourly`,
+          1,
+          expectedLabels,
+          TimeMs.WEEK
+        );
+        expect(safeTsAddSpy).toHaveBeenNthCalledWith(
+          3,
+          `ts:project-events-accepted:${projectIdMock}:daily`,
+          1,
+          expectedLabels,
+          90 * TimeMs.DAY
+        );
+      } finally {
+        safeTsAddSpy.mockRestore();
+      }
+    });
+
+    test('logs when a time-series write fails but continues processing', async () => {
+      const safeTsAddSpy = jest.spyOn((worker as any).redis, 'safeTsAdd');
+      const loggerErrorSpy = jest.spyOn((worker as any).logger, 'error').mockImplementation(() => undefined);
+      const failure = new Error('TS failure');
+
+      safeTsAddSpy
+        .mockImplementationOnce(() => Promise.resolve())
+        .mockImplementationOnce(async () => {
+          throw failure;
+        })
+        .mockImplementationOnce(() => Promise.resolve());
+
+      try {
+        await worker.handle(generateTask());
+
+        expect(loggerErrorSpy).toHaveBeenCalledWith('Failed to add hourly TS for events-accepted', failure);
+        expect(await eventsCollection.find().count()).toBe(1);
+      } finally {
+        safeTsAddSpy.mockRestore();
+        loggerErrorSpy.mockRestore();
+      }
+    });
+
+    test('records metrics exactly once per handled event', async () => {
+      const recordMetricsSpy = jest.spyOn(worker as any, 'recordProjectMetrics');
+
+      try {
+        await worker.handle(generateTask());
+
+        expect(recordMetricsSpy).toHaveBeenCalledTimes(1);
+        expect(recordMetricsSpy).toHaveBeenCalledWith(projectIdMock, 'events-accepted');
+      } finally {
+        recordMetricsSpy.mockRestore();
+      }
+    });
+  });
+
   afterAll(async () => {
     await redisClient.quit();
     await worker.finish();
diff --git a/workers/sentry/src/index.ts b/workers/sentry/src/index.ts
@@ -41,7 +41,6 @@ export default class SentryEventWorker extends Worker {
 
       if (items.length === 0) {
         this.logger.warn('Received envelope with no items');
-
         return;
       }
 
@@ -50,7 +49,6 @@ export default class SentryEventWorker extends Worker {
 
       for (const item of items) {
         const result = await this.handleEnvelopeItem(headers, item, event.projectId);
-
         if (result === 'processed') {
           processedCount++;
         } else if (result === 'skipped') {
@@ -249,15 +247,16 @@ export default class SentryEventWorker extends Worker {
       if (!taskSent) {
         /**
          * If addTask returns false, the message was not queued (queue full or channel closed)
+         * This is a critical error that should be logged and thrown
          */
         const error = new Error(`Failed to queue event to ${workerName} worker. Queue may be full or channel closed.`);
-
         this.logger.error(error.message);
         this.logger.info('👇 Here is the event that failed to queue:');
         this.logger.json(hawkEvent);
         throw error;
       }
 
+      this.logger.verbose(`Successfully queued event to ${workerName} worker`);
       return 'processed';
     } catch (error) {
       this.logger.error('Error handling envelope item:', error);
