imp(): move rate limiting to grouper

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,6 +3,19 @@
 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

workers/grouper/src/index.ts

@@ -22,10 +22,12 @@
 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 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 @@
    */
   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 @@
    */
   private handledTasksCount = 0;
 
+  /**
+   * Create grouper worker instance
+   */
+  constructor() {
+    super();
+    this.projectLimitsCache = new ProjectLimitsCache(this.accountsDb);
+  }
+
   /**
    * Start consuming messages
    */
@@ -131,6 +156,18 @@
     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(() => {
+      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 +192,11 @@
       this.cacheCleanupInterval = null;
     }
 
+    if (this.projectLimitsRefreshInterval) {
+      clearInterval(this.projectLimitsRefreshInterval);
+      this.projectLimitsRefreshInterval = null;
+    }
+
     this.memoryMonitor.logShutdown(this.handledTasksCount);
     await super.finish();
     this.prepareCache();
@@ -170,6 +212,16 @@
    */
   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 +232,30 @@
     }
   }
 
+  /**
+   * 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 +377,7 @@
           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.
    *

workers/grouper/src/projectLimitsCache.ts

@@ -0,0 +1,170 @@
+import type { Db } from 'mongodb';
+import { DatabaseController } from '../../../lib/db/controller';
+import createLogger from '../../../lib/logger';
+
+/**
+ * Rate limit settings stored in MongoDB (plans, workspaces, projects).
+ */
+interface RateLimitSettingsDocument {
+  N?: number;
+  T?: number;
+}
+
+interface ProjectDocument {
+  _id: { toString(): string };
+  workspaceId: { toString(): string };
+  rateLimitSettings?: RateLimitSettingsDocument;
+}
+
+interface PlanDocument {
+  _id: unknown;
+  rateLimitSettings?: RateLimitSettingsDocument;
+}
+
+interface WorkspaceDocument {
+  _id: { toString(): string };
+  rateLimitSettings?: RateLimitSettingsDocument;
+  plan: PlanDocument;
+}
+
+/**
+ * Resolved rate limit for a project.
+ */
+export interface ProjectRateLimits {
+  eventsLimit: number;
+  eventsPeriod: number;
+}
+
+const CONTEXT_TIMEOUT_MS = 5000;
+
+/**
+ * In-memory cache of project rate limits loaded from MongoDB accounts database.
+ * Priority: project → workspace → plan.
+ */
+export default class ProjectLimitsCache {
+  private projectLimits = new Map<string, ProjectRateLimits>();
+
+  private readonly logger = createLogger();
+
+  /**
+   * @param accountsDb - accounts database controller
+   */
+  constructor(private readonly accountsDb: DatabaseController) {}
+
+  /**
+   * Returns cached rate limits for a project.
+   *
+   * @param projectId - project id
+   */
+  public getProjectLimits(projectId: string): ProjectRateLimits | undefined {
+    return this.projectLimits.get(projectId);
+  }
+
+  /**
+   * Reload all project limits from MongoDB.
+   */
+  public async refresh(): Promise<void> {
+    const db = await this.accountsDb.connect();
+    const projectLimitsTmp = new Map<string, ProjectRateLimits>();
+
+    const workspaceMap = await this.loadWorkspacesWithPlans(db);
+
+    const projects = await db.collection<ProjectDocument>('projects')
+      .find({}, { maxTimeMS: CONTEXT_TIMEOUT_MS })
+      .toArray();
+
+    for (const project of projects) {
+      const projectId = project._id.toString();
+      let finalLimits: ProjectRateLimits = {
+        eventsLimit: 0,
+        eventsPeriod: 0,
+      };
+
+      const workspace = workspaceMap.get(project.workspaceId.toString());
+
+      if (workspace) {
+        finalLimits = {
+          eventsLimit: toNumber(workspace.plan?.rateLimitSettings?.N),
+          eventsPeriod: toNumber(workspace.plan?.rateLimitSettings?.T),
+        };
+
+        const workspaceLimit = toNumber(workspace.rateLimitSettings?.N);
+        const workspacePeriod = toNumber(workspace.rateLimitSettings?.T);
+
+        if (workspaceLimit > 0) {
+          finalLimits.eventsLimit = workspaceLimit;
+        }
+        if (workspacePeriod > 0) {
+          finalLimits.eventsPeriod = workspacePeriod;
+        }
+      }
+
+      const projectLimit = toNumber(project.rateLimitSettings?.N);
+      const projectPeriod = toNumber(project.rateLimitSettings?.T);
+
+      if (projectLimit > 0) {
+        finalLimits.eventsLimit = projectLimit;
+      }
+      if (projectPeriod > 0) {
+        finalLimits.eventsPeriod = projectPeriod;
+      }
+
+      projectLimitsTmp.set(projectId, finalLimits);
+    }
+
+    this.projectLimits = projectLimitsTmp;
+    this.logger.debug(`Project limits cache refreshed with ${projectLimitsTmp.size} projects`);
+  }
+
+  /**
+   * Load workspaces joined with tariff plans.
+   *
+   * @param db - accounts database
+   */
+  private async loadWorkspacesWithPlans(db: Db): Promise<Map<string, WorkspaceDocument>> {
+    const workspaces = await db.collection<WorkspaceDocument>('workspaces')
+      .aggregate<WorkspaceDocument>([
+        {
+          $lookup: {
+            from: 'plans',
+            localField: 'tariffPlanId',
+            foreignField: '_id',
+            as: 'plan',
+          },
+        },
+        { $unwind: '$plan' },
+      ], { maxTimeMS: CONTEXT_TIMEOUT_MS })
+      .toArray();
+
+    const workspaceMap = new Map<string, WorkspaceDocument>();
+
+    for (const workspace of workspaces) {
+      workspaceMap.set(workspace._id.toString(), workspace);
+    }
+
+    return workspaceMap;
+  }
+}
+
+/**
+ * Coerce MongoDB numeric values to number.
+ *
+ * @param value - raw value from document
+ */
+function toNumber(value: unknown): number {
+  if (typeof value === 'number') {
+    return value;
+  }
+
+  if (typeof value === 'object' && value !== null && 'toNumber' in value && typeof (value as { toNumber: () => number }).toNumber === 'function') {
+    return (value as { toNumber: () => number }).toNumber();
+  }
+
+  if (typeof value === 'string') {
+    const parsed = Number(value);
+
+    return Number.isFinite(parsed) ? parsed : 0;
+  }
+
+  return 0;
+}