imp(): move rate limiting to grouper

.env.test

@@ -36,8 +36,8 @@ ACCOUNTING_API_ENDPOINT=http://localhost:3999/graphql
 # Notify URL for sending Telegram notifications
 REPORT_NOTIFY_URL=http://mock.com/
 
-# Url for connecting to Redis
-REDIS_URL=redis://localhost:6379
+# REDIS_URL is set automatically by jest.setup.redis-mock.js (redis-memory-server).
+# Set REDIS_URL in the environment to use an external Redis instance instead.
 
 # Disable memoization in tests
 MEMOIZATION_TTL=-1

workers/grouper/.env.sample

@@ -1,2 +1,8 @@
 # Url for connecting to Redis
 REDIS_URL=redis://redis:6379
+
+# How often to refresh project rate limits from MongoDB (seconds)
+PROJECTS_LIMITS_UPDATE_PERIOD=3600
+
+# Redis hash key for per-project rate limit counters
+REDIS_RATE_LIMITS_KEY=rate_limits

workers/grouper/README.md

@@ -3,11 +3,31 @@
 Language-workers adds tasks for Group Worker in this format.
 Group Worker gets these tasks (events from language-workers) and saves it to the DB
 
+## Rate limiting
+
+Per-project rate limits are enforced before events are saved to MongoDB. Limits are loaded from the accounts database (`rateLimitSettings` on plans, workspaces, and projects) and tracked in Redis hash `rate_limits` (`timestamp:count` per project).
+
+Environment variables:
+
+| Variable | Description | Default |
+|----------|-------------|---------|
+| `PROJECTS_LIMITS_UPDATE_PERIOD` | Cache refresh interval (seconds) | `3600` |
+| `REDIS_RATE_LIMITS_KEY` | Redis hash key for counters | `rate_limits` |
+
+When the limit is exceeded, the event is dropped (message acked, no DB write) and `events-rate-limited` TimeSeries metrics are recorded.
+
 ## How to run
 
 1. Make sure you are in Workers root directory
-3. `yarn install`
-4. `yarn run-grouper`
-
+2. `nvm use` (requires Node 24, see `.nvmrc`)
+3. Start dependencies: Redis and MongoDB (e.g. `docker compose up -d redis` from repo root)
+4. `yarn install`
+5. `yarn run-grouper`
 
+## Tests
 
+```bash
+nvm use
+yarn build
+yarn test:grouper
+```

workers/grouper/src/index.ts

@@ -22,10 +22,12 @@ import { MS_IN_SEC } from '../../../lib/utils/consts';
 import TimeMs from '../../../lib/utils/time';
 import DataFilter from './data-filter';
 import RedisHelper from './redisHelper';
+import ProjectLimitsCache from './projectLimitsCache';
 import { computeDelta } from './utils/repetitionDiff';
 import { bucketTimestampMs } from './utils/bucketTimestamp';
 import { rightTrim } from '../../../lib/utils/string';
 import { hasValue } from '../../../lib/utils/hasValue';
+import { positiveIntEnv } from '../../../lib/utils/positiveIntEnv';
 import GrouperMetrics from './metrics/grouperMetrics';
 import GrouperMemoryMonitor from './metrics/memoryMonitor';
 import SlowHandleDiagnostics, { SlowHandleSession } from './metrics/slowHandleDiagnostics';
@@ -63,6 +65,11 @@ const DAILY_METRICS_RETENTION_DAYS = 90;
  */
 const MAX_CODE_LINE_LENGTH = 140;
 
+/**
+ * Default interval for refreshing project limits cache (in seconds)
+ */
+const DEFAULT_PROJECTS_LIMITS_UPDATE_PERIOD_SECONDS = 3600;
+
 /**
  * Worker for handling Javascript events
  */
@@ -92,6 +99,16 @@ export default class GrouperWorker extends Worker {
    */
   private redis = new RedisHelper();
 
+  /**
+   * Cached project rate limits loaded from accounts MongoDB
+   */
+  private projectLimitsCache: ProjectLimitsCache;
+
+  /**
+   * Interval for periodic project limits cache refresh
+   */
+  private projectLimitsRefreshInterval: NodeJS.Timeout | null = null;
+
   /**
    * Prometheus metrics facade.
    */
@@ -117,6 +134,14 @@ export default class GrouperWorker extends Worker {
    */
   private handledTasksCount = 0;
 
+  /**
+   * Create grouper worker instance
+   */
+  constructor() {
+    super();
+    this.projectLimitsCache = new ProjectLimitsCache(this.accountsDb);
+  }
+
   /**
    * Start consuming messages
    */
@@ -131,6 +156,21 @@ export default class GrouperWorker extends Worker {
     await this.redis.initialize();
     console.log('redis initialized');
 
+    await this.projectLimitsCache.refresh();
+
+    const limitsUpdatePeriodSeconds = positiveIntEnv(
+      process.env.PROJECTS_LIMITS_UPDATE_PERIOD,
+      DEFAULT_PROJECTS_LIMITS_UPDATE_PERIOD_SECONDS,
+    );
+
+    this.projectLimitsRefreshInterval = setInterval(() => {
+      /* eslint-disable-next-line no-void */
+      void this.projectLimitsCache.refresh()
+        .catch((error) => {
+          this.logger.error('Failed to refresh project limits cache', error);
+        });
+    }, limitsUpdatePeriodSeconds * MS_IN_SEC);
+
     /**
      * Start periodic cache cleanup to prevent memory leaks from unbounded cache growth
      * Runs every 30 seconds to clear old cache entries
@@ -155,6 +195,11 @@ export default class GrouperWorker extends Worker {
       this.cacheCleanupInterval = null;
     }
 
+    if (this.projectLimitsRefreshInterval) {
+      clearInterval(this.projectLimitsRefreshInterval);
+      this.projectLimitsRefreshInterval = null;
+    }
+
     this.memoryMonitor.logShutdown(this.handledTasksCount);
     await super.finish();
     this.prepareCache();
@@ -170,6 +215,16 @@ export default class GrouperWorker extends Worker {
    */
   public async handle(task: GroupWorkerTask<ErrorsCatcherType>): Promise<void> {
     try {
+      const withinLimit = await this.checkRateLimit(task.projectId);
+
+      if (!withinLimit) {
+        this.grouperMetrics.incrementRateLimitedTotal();
+        await this.recordProjectMetrics(task.projectId, 'events-rate-limited');
+        this.logger.info(`[rate-limit] project=${task.projectId} title="${task.payload?.title}" dropped`);
+
+        return;
+      }
+
       await this.grouperMetrics.observeHandleDuration(async () => {
         await this.handleInternal(task);
       });
@@ -180,6 +235,30 @@ export default class GrouperWorker extends Worker {
     }
   }
 
+  /**
+   * Check and update per-project rate limit in Redis.
+   *
+   * @param projectId - project id
+   */
+  private async checkRateLimit(projectId: string): Promise<boolean> {
+    const limits = this.projectLimitsCache.getProjectLimits(projectId);
+
+    if (!limits) {
+      this.logger.warn(`Project ${projectId} is not in the projects limits cache`);
+    }
+
+    const eventsLimit = limits?.eventsLimit ?? 0;
+    const eventsPeriod = limits?.eventsPeriod ?? 0;
+
+    try {
+      return await this.redis.updateRateLimit(projectId, eventsLimit, eventsPeriod);
+    } catch (error) {
+      this.logger.error(`Failed to update rate limit for project ${projectId}`, error);
+
+      return false;
+    }
+  }
+
   /**
    * Internal task handling function
    *
@@ -301,7 +380,7 @@ export default class GrouperWorker extends Worker {
           this.grouperMetrics.incrementDuplicateRetriesTotal();
           this.logger.info(`[saveEvent] project=${task.projectId} title="${task.payload.title}" duplicate key, retrying as repetition`);
 
-          await this.handle(task);
+          await this.handleInternal(task);
 
           return;
         }

workers/grouper/src/metrics/grouperMetrics.ts

@@ -135,6 +135,15 @@ export default class GrouperMetrics {
     })
   );
 
+  private readonly rateLimitedTotal = getOrCreateMetric(
+    'hawk_grouper_events_rate_limited_total',
+    () => new client.Counter({
+      name: 'hawk_grouper_events_rate_limited_total',
+      help: 'Number of events dropped due to rate limiting',
+      registers: [ register ],
+    })
+  );
+
   /**
    * Measure top-level handle() duration.
    *
@@ -207,6 +216,13 @@ export default class GrouperMetrics {
     this.duplicateRetriesTotal.inc();
   }
 
+  /**
+   * Increment events dropped by rate limiting.
+   */
+  public incrementRateLimitedTotal(): void {
+    this.rateLimitedTotal.inc();
+  }
+
   /**
    * Measure Mongo operation duration.
    *