refactor: rate limit store

.env.example

@@ -100,10 +100,6 @@ ROLLBAR_SERVER_TOKEN=''
 ALGOLIA_APPLICATION_ID=''
 ALGOLIA_API_KEY=''
 
-# HONEYCOMBE - server analytics, only set to true on hosted server
-HONEYCOMB_ENABLED=false
-HONEYCOMB_API_KEY=''
-
 # Nx 18 enables using plugins to infer targets by default
 # This is disabled for existing workspaces to maintain compatibility
 # For more info, see: https://nx.dev/concepts/inferred-tasks

apps/api/src/main.ts

@@ -1,5 +1,5 @@
-import { ClusterMemoryStorePrimary } from '@express-rate-limit/cluster-memory-store';
 import '@jetstream/api-config'; // this gets imported first to ensure as some items require early initialization
+
 import { createRateLimit, ENV, getExceptionLog, httpLogger, logger, pgPool } from '@jetstream/api-config';
 import '@jetstream/auth/types';
 import { HTTP, SESSION_EXP_DAYS } from '@jetstream/shared/constants';
@@ -68,9 +68,6 @@ if (ENV.NODE_ENV === 'production' && !ENV.CI && cluster.isPrimary) {
 
   setupPrimary();
 
-  const rateLimiterStore = new ClusterMemoryStorePrimary();
-  rateLimiterStore.init();
-
   for (let i = 0; i < CPU_COUNT; i++) {
     cluster.fork();
   }

libs/api-config/src/lib/api-rate-limit.config.ts

@@ -1,16 +1,12 @@
-import { ClusterMemoryStoreWorker } from '@express-rate-limit/cluster-memory-store';
 import { HTTP } from '@jetstream/shared/constants';
-import cluster from 'cluster';
-import { MemoryStore, Options, rateLimit } from 'express-rate-limit';
+import type { ClientRateLimitInfo, IncrementResponse, Store } from 'express-rate-limit';
+import { Options, rateLimit } from 'express-rate-limit';
+import { prisma } from './api-db-config';
+import { getExceptionLog, logger } from './api-logger';
 
 export function createRateLimit(prefix: string, options: Partial<Options>) {
   return rateLimit({
-    // cluster.isPrimary will be true on development and production master process
-    store: cluster.isPrimary
-      ? new MemoryStore()
-      : new ClusterMemoryStoreWorker({
-          prefix,
-        }),
+    store: new PrismaRateLimitStore({ prefix }),
     windowMs: 1000 * 60 * 1, // 1 minute
     max: 50, // limit each IP to 50 requests per windowMs
     standardHeaders: true, // Return rate limit info in the `RateLimit-*` headers
@@ -29,3 +25,230 @@ export function createRateLimit(prefix: string, options: Partial<Options>) {
     ...options,
   });
 }
+
+export interface PrismaStoreOptions {
+  /**
+   * Optional field to differentiate hit counts when multiple rate-limits are in use
+   */
+  prefix?: string;
+
+  /**
+   * How often to clean up expired entries (in milliseconds)
+   * @default 60_000 (1 minute)
+   */
+  cleanupIntervalMs?: number;
+}
+
+/**
+ * A Prisma-backed Store implementation for express-rate-limit.
+ * Stores rate limit data in a postgres table with optimized queries for performance.
+ *
+ * @public
+ */
+export class PrismaRateLimitStore implements Store {
+  localKeys = false;
+  prefix: string;
+  windowMs!: number;
+
+  /**
+   * Interval for cleanup of expired entries
+   */
+  private cleanupIntervalMs: number;
+
+  /**
+   * Timer reference for cleanup interval
+   */
+  private cleanupTimer?: NodeJS.Timeout;
+
+  constructor(options: PrismaStoreOptions = {}) {
+    this.prefix = options.prefix ?? 'rl:';
+    this.cleanupIntervalMs = options.cleanupIntervalMs ?? 60_000;
+  }
+
+  init(options: Options): void {
+    this.windowMs = options.windowMs;
+
+    // Start cleanup interval to remove expired entries
+    this.startCleanup();
+  }
+
+  /**
+   * Method to prefix the keys with the given text.
+   *
+   * @param key {string} - The key.
+   *
+   * @returns {string} - The prefixed key.
+   *
+   * @private
+   */
+  private prefixKey(key: string): string {
+    return `${this.prefix}${key}`;
+  }
+
+  async get(key: string): Promise<ClientRateLimitInfo | undefined> {
+    const prefixedKey = this.prefixKey(key);
+
+    try {
+      const row = await prisma.rateLimit.findUnique({
+        where: { key: prefixedKey },
+      });
+
+      if (!row) {
+        return undefined;
+      }
+
+      const now = new Date();
+      const resetTime = row.resetTime;
+
+      // If the reset time has passed, this entry is expired
+      if (resetTime <= now) {
+        return undefined;
+      }
+
+      return {
+        totalHits: row.hits,
+        resetTime,
+      };
+    } catch (error) {
+      logger.error(getExceptionLog(error), `[RATE_LIMIT][GET] Error fetching rate limit for key: ${prefixedKey}`);
+      throw error;
+    }
+  }
+
+  async increment(key: string): Promise<IncrementResponse> {
+    const prefixedKey = this.prefixKey(key);
+    const now = new Date();
+    const resetTime = new Date(now.getTime() + this.windowMs);
+
+    try {
+      // Use an atomic upsert (INSERT ... ON CONFLICT) for optimal performance
+      // If the key exists and hasn't expired, increment the counter
+      // If the key doesn't exist or has expired, create/reset with count of 1
+      const row = await prisma.rateLimit.upsert({
+        where: { key: prefixedKey },
+        select: { hits: true, resetTime: true },
+        update: {
+          hits: { increment: 1 },
+          resetTime,
+          updatedAt: now,
+        },
+        create: {
+          key: prefixedKey,
+          hits: 1,
+          resetTime,
+          createdAt: now,
+          updatedAt: now,
+        },
+      });
+
+      return {
+        totalHits: row.hits,
+        resetTime: row.resetTime,
+      };
+    } catch (error) {
+      logger.error(getExceptionLog(error), `[RATE_LIMIT][INCREMENT] Error incrementing rate limit for key: ${prefixedKey}`);
+      throw error;
+    }
+  }
+
+  async decrement(key: string): Promise<void> {
+    const prefixedKey = this.prefixKey(key);
+    const now = new Date();
+
+    try {
+      // Only decrement if the entry exists and hasn't expired
+      // Don't let hits go below 0
+      await prisma.rateLimit.updateMany({
+        where: { key: prefixedKey, resetTime: { gt: now } },
+        data: {
+          hits: { decrement: 1 },
+          updatedAt: now,
+        },
+      });
+    } catch (error) {
+      logger.error(getExceptionLog(error), `[RATE_LIMIT][DECREMENT] Error decrementing rate limit for key: ${prefixedKey}`);
+      throw error;
+    }
+  }
+
+  async resetKey(key: string): Promise<void> {
+    const prefixedKey = this.prefixKey(key);
+
+    try {
+      await prisma.rateLimit.delete({
+        where: { key: prefixedKey },
+      });
+    } catch (error) {
+      logger.error(getExceptionLog(error), `[RATE_LIMIT][RESET_KEY] Error resetting rate limit for key: ${prefixedKey}`);
+      throw error;
+    }
+  }
+
+  async resetAll(): Promise<void> {
+    try {
+      // Only reset entries with the current prefix to avoid affecting other rate limiters
+      await prisma.rateLimit.deleteMany({
+        where: {
+          key: { startsWith: this.prefix },
+        },
+      });
+    } catch (error) {
+      logger.error(getExceptionLog(error), `[RATE_LIMIT][RESET_ALL] Error resetting all rate limits for prefix: ${this.prefix}`);
+      throw error;
+    }
+  }
+
+  /**
+   * Starts the cleanup interval to periodically remove expired entries.
+   *
+   * @private
+   */
+  private startCleanup(): void {
+    // Clear any existing timer
+    if (this.cleanupTimer) {
+      clearInterval(this.cleanupTimer);
+    }
+
+    // Run cleanup periodically
+    this.cleanupTimer = setInterval(() => {
+      this.cleanup().catch((error) => {
+        logger.error(getExceptionLog(error), '[RATE_LIMIT][CLEANUP] Error during cleanup interval');
+      });
+    }, this.cleanupIntervalMs);
+
+    // Don't prevent the process from exiting
+    this.cleanupTimer.unref();
+  }
+
+  /**
+   * Removes expired entries from the database to keep the table size manageable.
+   *
+   * @private
+   */
+  private async cleanup(): Promise<void> {
+    const now = new Date();
+
+    try {
+      await prisma.rateLimit.deleteMany({
+        where: {
+          resetTime: { lte: now },
+        },
+      });
+    } catch (error) {
+      logger.error(getExceptionLog(error), '[RATE_LIMIT][CLEANUP] Error removing expired entries');
+      // Don't throw - cleanup failures shouldn't affect rate limiting
+    }
+  }
+
+  /**
+   * Stops the cleanup interval. Call this when shutting down the application.
+   *
+   * @public
+   */
+  shutdown(): void {
+    if (this.cleanupTimer) {
+      clearInterval(this.cleanupTimer);
+      this.cleanupTimer = undefined;
+    }
+  }
+}

package.json

@@ -266,7 +266,6 @@
     "@casl/react": "^5.0.0",
     "@emotion/react": "11.14.0",
     "@emotion/styled": "11.14.1",
-    "@express-rate-limit/cluster-memory-store": "^0.3.1",
     "@floating-ui/react": "^0.27.16",
     "@fullhuman/postcss-purgecss": "^2.2.0",
     "@heroicons/react": "^2.2.0",

prisma/migrations/20260104191734_add_rate_limit_store/migration.sql

@@ -0,0 +1,13 @@
+-- CreateTable
+CREATE TABLE "rate_limit" (
+    "key" VARCHAR(255) NOT NULL,
+    "hits" INTEGER NOT NULL DEFAULT 0,
+    "resetTime" TIMESTAMP(6) NOT NULL,
+    "createdAt" TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    "updatedAt" TIMESTAMP(6) NOT NULL,
+
+    CONSTRAINT "rate_limit_pkey" PRIMARY KEY ("key")
+);
+
+-- CreateIndex
+CREATE INDEX "rate_limit_resetTime_idx" ON "rate_limit"("resetTime");

prisma/schema.prisma

@@ -548,3 +548,15 @@ model AuditLog {
   @@index([resource, resourceId, createdAt])
   @@map("audit_log")
 }
+
+model RateLimit {
+  key       String   @id @db.VarChar(255)
+  hits      Int      @default(0)
+  resetTime DateTime @db.Timestamp(6)
+  createdAt DateTime @default(now()) @db.Timestamp(6)
+  updatedAt DateTime @updatedAt @db.Timestamp(6)
+
+  // Index for cleanup queries to find expired entries
+  @@index([resetTime])
+  @@map("rate_limit")
+}

yarn.lock

@@ -5474,14 +5474,6 @@
     "@eslint/core" "^0.12.0"
     levn "^0.4.1"
 
-"@express-rate-limit/cluster-memory-store@^0.3.1":
-  version "0.3.1"
-  resolved "https://registry.yarnpkg.com/@express-rate-limit/cluster-memory-store/-/cluster-memory-store-0.3.1.tgz#28177e93a3fb1d01bbf8d2fc83704636c197fd5c"
-  integrity sha512-rB1NrctHayZI1foj6DnPV0EO8VnfRakcIaksC+RO1VYl/IbXyDIKYKbeyaqneuq5oUv6fTCv6/GOuUYcYX8KTw==
-  dependencies:
-    "@types/debug" "4.1.12"
-    debug "4.3.4"
-
 "@floating-ui/core@^1.7.3":
   version "1.7.3"
   resolved "https://registry.yarnpkg.com/@floating-ui/core/-/core-1.7.3.tgz#462d722f001e23e46d86fd2bd0d21b7693ccb8b7"
@@ -9445,7 +9437,7 @@
   dependencies:
     "@types/node" "*"
 
-"@types/debug@4.1.12", "@types/debug@^4.1.6":
+"@types/debug@^4.1.6":
   version "4.1.12"
   resolved "https://registry.yarnpkg.com/@types/debug/-/debug-4.1.12.tgz#a155f21690871953410df4b6b6f53187f0500917"
   integrity sha512-vIChWdVG3LG1SMxEvI/AK+FWJthlrqlTu7fbrlywTkkaONwk/UAGaULXRlf8vkzFBLVm0zkMdCquhL5aOjhXPQ==