merge: resolve conflicts with origin/master for grouper metrics

Author: Kuchizu

docker-compose.dev.yml

@@ -158,6 +158,18 @@ services:
       - ./:/usr/src/app
       - workers-deps:/usr/src/app/node_modules
 
+  hawk-worker-webhook:
+    build:
+      dockerfile: "dev.Dockerfile"
+      context: .
+    env_file:
+      - .env
+    restart: unless-stopped
+    entrypoint: yarn run-webhook
+    volumes:
+      - ./:/usr/src/app
+      - workers-deps:/usr/src/app/node_modules
+
 volumes:
   workers-deps:
 

package.json

@@ -1,7 +1,7 @@
 {
     "name": "hawk.workers",
     "private": true,
-    "version": "0.1.2",
+    "version": "0.1.3",
     "description": "Hawk workers",
     "repository": "[email protected]:codex-team/hawk.workers.git",
     "license": "BUSL-1.1",
@@ -34,6 +34,7 @@
         "test:notifier": "jest workers/notifier",
         "test:js": "jest workers/javascript",
         "test:task-manager": "jest workers/task-manager",
+        "test:webhook": "jest workers/webhook",
         "test:clear": "jest --clearCache",
         "run-default": "yarn worker hawk-worker-default",
         "run-sentry": "yarn worker hawk-worker-sentry",
@@ -49,13 +50,14 @@
         "run-email": "yarn worker hawk-worker-email",
         "run-telegram": "yarn worker hawk-worker-telegram",
         "run-limiter": "yarn worker hawk-worker-limiter",
-        "run-task-manager": "yarn worker hawk-worker-task-manager"
+        "run-task-manager": "yarn worker hawk-worker-task-manager",
+        "run-webhook": "yarn worker hawk-worker-webhook"
     },
     "dependencies": {
         "@babel/parser": "^7.26.9",
         "@babel/traverse": "7.26.9",
         "@hawk.so/nodejs": "^3.1.1",
-        "@hawk.so/types": "^0.5.7",
+        "@hawk.so/types": "^0.5.9",
         "@types/amqplib": "^0.8.2",
         "@types/jest": "^29.5.14",
         "@types/mongodb": "^3.5.15",

workers/email/tests/provider.test.ts

@@ -10,7 +10,7 @@ nodemailerMock.createTransport = jest.fn(() => ({
 
 jest.mock('nodemailer', () => nodemailerMock);
 
-import { DecodedGroupedEvent, ProjectDBScheme } from '@hawk.so/types';
+import { DecodedGroupedEvent, ProjectDBScheme, UserDBScheme } from '@hawk.so/types';
 import '../src/env';
 import EmailProvider from '../src/provider';
 import Templates from '../src/templates/names';
@@ -187,6 +187,11 @@ describe('EmailProvider', () => {
           password: '$argon2i$v=19$m=4096,t=3,p=1$QOo3u8uEor0t+nqCLpEW3g$aTCDEaHht9ro9VZDD1yZGpaTi+g1OWsfHbYd5TQBRPs',
           name: 'Hahahawk',
         },
+        assignee: {
+          _id: new ObjectId('5ec3ffe769c0030022f88f25'),
+          email: '[email protected]',
+          name: 'Assignee User',
+        } as UserDBScheme,
         project: {
           _id: new ObjectId('5d206f7f9aaf7c0071d64596'),
           token: 'project-token',

workers/grouper/src/data-filter.ts

@@ -1,6 +1,11 @@
 import type { EventAddons, EventData } from '@hawk.so/types';
 import { unsafeFields } from '../../../lib/utils/unsafeFields';
 
+/**
+ * Maximum depth for object traversal to prevent excessive memory allocations
+ */
+const MAX_TRAVERSAL_DEPTH = 20;
+
 /**
  * Recursively iterate through object and call function on each key
  *
@@ -18,7 +23,12 @@ function forAll(obj: Record<string, unknown>, callback: (path: string[], key: st
       if (!(typeof value === 'object' && !Array.isArray(value))) {
         callback(path, key, current);
       } else {
-        visit(value, [...path, key]);
+        /**
+         * Limit path depth to prevent excessive memory allocations from deep nesting
+         * This reduces GC pressure and memory usage for deeply nested objects
+         */
+        const newPath = path.length < MAX_TRAVERSAL_DEPTH ? path.concat(key) : path;
+        visit(value, newPath);
       }
     }
   };

workers/grouper/src/index.ts

@@ -19,11 +19,16 @@ import type { RepetitionDBScheme } from '../types/repetition';
 import { DatabaseReadWriteError, DiffCalculationError, ValidationError } from '../../../lib/workerErrors';
 import { decodeUnsafeFields, encodeUnsafeFields } from '../../../lib/utils/unsafeFields';
 import { MS_IN_SEC } from '../../../lib/utils/consts';
+import TimeMs from '../../../lib/utils/time';
 import DataFilter from './data-filter';
 import RedisHelper from './redisHelper';
 import { computeDelta } from './utils/repetitionDiff';
 import { rightTrim } from '../../../lib/utils/string';
 import { hasValue } from '../../../lib/utils/hasValue';
+
+/**
+ * eslint does not count decorators as a variable usage
+ */
 /* eslint-disable-next-line no-unused-vars */
 import { memoize } from '../../../lib/memoize';
 import { register, client } from '../../../lib/metrics';
@@ -32,7 +37,12 @@ import { register, client } from '../../../lib/metrics';
  * eslint does not count decorators as a variable usage
  */
 /* eslint-disable-next-line no-unused-vars */
-const MEMOIZATION_TTL = Number(process.env.MEMOIZATION_TTL ?? 0);
+const MEMOIZATION_TTL = 600_000;
+
+/**
+ * Cache cleanup interval in minutes
+ */
+const CACHE_CLEANUP_INTERVAL_MINUTES = 5;
 
 /**
  * Error code of MongoDB key duplication error
@@ -122,6 +132,11 @@ export default class GrouperWorker extends Worker {
     registers: [register],
   });
 
+  /**
+   * Interval for periodic cache cleanup to prevent memory leaks from unbounded cache growth
+   */
+  private cacheCleanupInterval: NodeJS.Timeout | null = null;
+
   /**
    * Start consuming messages
    */
@@ -135,13 +150,30 @@ export default class GrouperWorker extends Worker {
 
     await this.redis.initialize();
     console.log('redis initialized');
+
+    /**
+     * Start periodic cache cleanup to prevent memory leaks from unbounded cache growth
+     * Runs every 5 minutes to clear old cache entries
+     */
+    this.cacheCleanupInterval = setInterval(() => {
+      this.clearCache();
+    }, CACHE_CLEANUP_INTERVAL_MINUTES * TimeMs.MINUTE);
+
     await super.start();
   }
 
   /**
    * Finish everything
    */
   public async finish(): Promise<void> {
+    /**
+     * Clear cache cleanup interval to prevent resource leaks
+     */
+    if (this.cacheCleanupInterval) {
+      clearInterval(this.cacheCleanupInterval);
+      this.cacheCleanupInterval = null;
+    }
+
     await super.finish();
     this.prepareCache();
     await this.eventsDb.close();
@@ -198,7 +230,7 @@ export default class GrouperWorker extends Worker {
     const similarEvent = await this.findSimilarEvent(task.projectId, task.payload.title);
 
     if (similarEvent) {
-      this.logger.info(`similar event: ${JSON.stringify(similarEvent)}`);
+      this.logger.info(`[handle] similar event found, groupHash=${similarEvent.groupHash} totalCount=${similarEvent.totalCount}`);
 
       /**
        * Override group hash with found event's group hash
@@ -331,6 +363,12 @@ export default class GrouperWorker extends Worker {
       } as RepetitionDBScheme;
 
       repetitionId = await this.saveRepetition(task.projectId, newRepetition);
+
+      /**
+       * Clear the large event payload references to allow garbage collection
+       * This prevents memory leaks from retaining full event objects after delta is computed
+       */
+      delta = undefined;
     }
 
     /**
@@ -432,7 +470,7 @@ export default class GrouperWorker extends Worker {
    * @param projectId - where to find
    * @param title - title of the event to find similar one
    */
-  @memoize({ max: 200, ttl: MEMOIZATION_TTL, strategy: 'hash', skipCache: [undefined] })
+  @memoize({ max: 50, ttl: MEMOIZATION_TTL, strategy: 'hash', skipCache: [undefined] })
   private async findSimilarEvent(projectId: string, title: string): Promise<GroupedEventDBScheme | undefined> {
     /**
      * If no match by Levenshtein, try matching by patterns
@@ -446,9 +484,7 @@ export default class GrouperWorker extends Worker {
         try {
           const originalEvent = await this.findFirstEventByPattern(matchingPattern.pattern, projectId);
 
-          const originalEventSize = Buffer.byteLength(JSON.stringify(originalEvent));
-
-          this.logger.info(`[findSimilarEvent] found by pattern, originalEventSize=${originalEventSize}b`);
+          this.logger.info(`[findSimilarEvent] found by pattern, groupHash=${originalEvent?.groupHash} title="${originalEvent?.payload?.title}"`);
 
           if (originalEvent) {
             return originalEvent;

workers/grouper/tests/data-filter.test.ts

@@ -327,5 +327,39 @@ describe('GrouperWorker', () => {
       expect(event.context['secret']).toBe('[filtered]');
       expect(event.context['auth']).toBe('[filtered]');
     });
+
+    test('should handle deeply nested objects (>20 levels) without excessive memory allocations', () => {
+      // Create an object nested deeper than the cap (>20 levels)
+      let deeplyNested: any = { value: 'leaf', secret: 'should-be-filtered' };
+
+      for (let i = 0; i < 25; i++) {
+        deeplyNested = { [`level${i}`]: deeplyNested, password: `sensitive${i}` };
+      }
+
+      const event = generateEvent({
+        context: deeplyNested,
+      });
+
+      // This should not throw or cause memory issues
+      dataFilter.processEvent(event);
+
+      // Verify that filtering still works at various depths
+      expect(event.context['password']).toBe('[filtered]');
+
+      // Navigate to a mid-level and check filtering
+      let current = event.context['level24'] as any;
+      for (let i = 24; i > 15; i--) {
+        expect(current['password']).toBe('[filtered]');
+        current = current[`level${i - 1}`];
+      }
+
+      // At the leaf level, the secret should still be filtered
+      // (though path tracking may be capped, filtering should still work)
+      let leaf = event.context;
+      for (let i = 24; i >= 0; i--) {
+        leaf = leaf[`level${i}`] as any;
+      }
+      expect(leaf['secret']).toBe('[filtered]');
+    });
   });
 });

workers/notifier/types/channel.ts

@@ -6,6 +6,7 @@ export enum ChannelType {
   Telegram = 'telegram',
   Slack = 'slack',
   Loop = 'loop',
+  Webhook = 'webhook',
 }
 
 /**

workers/sender/src/index.ts

@@ -253,6 +253,7 @@ export default abstract class SenderWorker extends Worker {
         project,
         event,
         whoAssigned,
+        assignee,
         daysRepeated,
       },
     } as AssigneeNotification);

workers/sender/types/template-variables/assignee.ts

@@ -21,6 +21,11 @@ export interface AssigneeTemplateVariables extends CommonTemplateVariables {
    */
   whoAssigned: UserDBScheme;
 
+  /**
+   * User who was assigned to resolve the event
+   */
+  assignee: UserDBScheme;
+
   /**
    * Number of event repetitions
    */

workers/webhook/package.json

@@ -0,0 +1,11 @@
+{
+  "name": "hawk-worker-webhook",
+  "version": "1.0.0",
+  "description": "Webhook sender worker — delivers event notifications as JSON POST requests",
+  "main": "src/index.ts",
+  "license": "MIT",
+  "workerType": "sender/webhook",
+  "scripts": {
+    "test": "jest"
+  }
+}