Extract shared createRateLimiter factory

backend/src/app.ts

@@ -14,7 +14,6 @@ import { sandboxMiddleware } from "./middleware/sandbox.middleware.js";
 import { globalRateLimiter } from "./middleware/rate-limiter.middleware.js";
 import { requestIdMiddleware } from "./middleware/requestId.js";
 import v1Routes from "./routes/v1/index.js";
-
 import healthRoutes from "./routes/health.routes.js";
 import "./lib/stream-id.js";
 

backend/src/middleware/rate-limiter.middleware.ts

@@ -1,10 +1,22 @@
-import { rateLimit } from 'express-rate-limit';
+import { rateLimit, type Options } from 'express-rate-limit';
 
-export const globalRateLimiter = rateLimit({
+/**
+ * Shared factory to create an express-rate-limit instance with common configuration.
+ * 
+ * @param options Configuration options for express-rate-limit
+ * @returns Express rate limit middleware
+ */
+export function createRateLimiter(options: Partial<Options>) {
+  return rateLimit({
+    standardHeaders: true, // Return rate limit info in the `RateLimit-*` headers
+    legacyHeaders: false, // Disable the `X-RateLimit-*` headers
+    ...options,
+  });
+}
+
+export const globalRateLimiter = createRateLimiter({
   windowMs: 1 * 60 * 1000, // 1 minute
   max: 100, // Limit each IP to 100 requests per `window` (here, per minute)
-  standardHeaders: true, // Return rate limit info in the `RateLimit-*` headers
-  legacyHeaders: false, // Disable the `X-RateLimit-*` headers
   message: {
     message: 'Too many requests, please try again later.',
     status: 429,

backend/src/middleware/stream-rate-limiter.middleware.ts

@@ -1,4 +1,4 @@
-import { rateLimit } from 'express-rate-limit';
+import { createRateLimiter } from './rate-limiter.middleware.js';
 import { type Request, type Response, type NextFunction } from 'express';
 import type { AuthenticatedRequest } from '../types/auth.types.js';
 import logger from '../logger.js';
@@ -22,11 +22,9 @@ export function createStreamRateLimiter(
   // Read from environment variable, default to 10 if not set
   const max = options?.max ?? (process.env.STREAM_CREATE_RATE_LIMIT ? parseInt(process.env.STREAM_CREATE_RATE_LIMIT, 10) : 10);
 
-  return rateLimit({
+  return createRateLimiter({
     windowMs,
     max,
-    standardHeaders: true, // Return rate limit info in the `RateLimit-*` headers
-    legacyHeaders: false, // Disable the `X-RateLimit-*` headers
     message: {
       error: 'Too many stream creation requests - rate limit exceeded',
       message: 'You have exceeded the rate limit for stream creation. Please try again later.',

backend/src/workers/soroban-event-worker.ts

@@ -8,8 +8,6 @@ import { Prisma } from "../generated/prisma/index.js";
 import "../lib/stream-id.js";
 import { rpcPool } from "../lib/rpc-pool.js";
 
-// ─── Config ──────────────────────────────────────────────────────────────────
-
 // ─── XDR Decoding Helpers ────────────────────────────────────────────────────
 
 /** Decode an ScVal symbol to a string. */
@@ -116,7 +114,8 @@ export class SorobanEventWorker {
 
   constructor() {
     this.contractId = process.env.STREAM_CONTRACT_ID ?? "";
-    const rpcUrl = process.env.SOROBAN_RPC_URL ?? "https://soroban-testnet.stellar.org";
+    const rpcUrl =
+      process.env.SOROBAN_RPC_URL ?? "https://soroban-testnet.stellar.org";
     this.server = new rpc.Server(rpcUrl, { allowHttp: true });
     this.pollIntervalMs = parseInt(
       process.env.INDEXER_POLL_INTERVAL_MS ?? "5000",
@@ -205,16 +204,22 @@ export class SorobanEventWorker {
    */
   async triggerPoll(customRequestId?: string): Promise<string> {
     if (!this.isRunning) {
-      return customRequestId || requestContext?.getStore?.()?.requestId || randomUUID();
+      return (
+        customRequestId ||
+        requestContext?.getStore?.()?.requestId ||
+        randomUUID()
+      );
     }
 
     const requestId =
-      customRequestId || requestContext?.getStore?.()?.requestId || randomUUID();
+      customRequestId ||
+      requestContext?.getStore?.()?.requestId ||
+      randomUUID();
 
     try {
       await this.runExclusive(() => {
         const runBatch = () => this.fetchAndProcessEvents();
-        return requestContext && typeof requestContext.run === 'function'
+        return requestContext && typeof requestContext.run === "function"
           ? requestContext.run({ requestId }, runBatch)
           : runBatch();
       });
@@ -292,7 +297,7 @@ export class SorobanEventWorker {
           this.fetchAndProcessEvents().catch((err) => {
             logger.error("[SorobanWorker] Unhandled error during poll:", err);
           });
-        return requestContext && typeof requestContext.run === 'function'
+        return requestContext && typeof requestContext.run === "function"
           ? requestContext.run({ requestId }, execute)
           : execute();
       });
@@ -307,7 +312,11 @@ export class SorobanEventWorker {
    */
   private async fetchAndProcessEvents(): Promise<void> {
     const currentCtx = requestContext?.getStore?.();
-    if (!currentCtx?.requestId && requestContext && typeof requestContext.run === 'function') {
+    if (
+      !currentCtx?.requestId &&
+      requestContext &&
+      typeof requestContext.run === "function"
+    ) {
       const requestId = randomUUID();
       return requestContext.run({ requestId }, () =>
         this.fetchAndProcessEvents(),
@@ -336,7 +345,9 @@ export class SorobanEventWorker {
       ? { ...baseFilter, cursor: state.lastCursor }
       : { ...baseFilter, startLedger: state.lastLedger || this.startLedger };
 
-    const response = await (this.server ? this.server.getEvents(params) : rpcPool.execute("getEvents", (server) => server.getEvents(params)));
+    const response = await (this.server
+      ? this.server.getEvents(params)
+      : rpcPool.execute("getEvents", (server) => server.getEvents(params)));
 
     if (response.events.length === 0) return;
 
@@ -384,7 +395,7 @@ export class SorobanEventWorker {
     // Use the response's final cursor if provided and no error occurred, otherwise the last valid event's ID
     const finalCursor = hasError
       ? lastCursor
-      : ((response as any).latestCursor || lastCursor);
+      : (response as any).latestCursor || lastCursor;
 
     await prisma.indexerState.upsert({
       where: { id: INDEXER_STATE_ID },
@@ -717,11 +728,18 @@ export class SorobanEventWorker {
       // Check for a duplicate BEFORE mutating any Stream fields so that a
       // replayed event never re-applies the top-up.
       const existingEvent = await tx.streamEvent.findUnique({
-        where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'TOPPED_UP' } },
+        where: {
+          transactionHash_eventType: {
+            transactionHash: event.txHash,
+            eventType: "TOPPED_UP",
+          },
+        },
         select: { id: true },
       });
       if (existingEvent) {
-        logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=TOPPED_UP`);
+        logger.warn(
+          `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=TOPPED_UP`,
+        );
         return;
       }
 
@@ -739,7 +757,7 @@ export class SorobanEventWorker {
         ratePerSecondBigInt === 0n
           ? null
           : BigInt(stream.startTime) +
-            (BigInt(newDepositedAmount) / ratePerSecondBigInt) +
+            BigInt(newDepositedAmount) / ratePerSecondBigInt +
             BigInt(stream.totalPausedDuration);
 
       await tx.stream.update({
@@ -752,10 +770,15 @@ export class SorobanEventWorker {
       });
 
       await tx.streamEvent.upsert({
-        where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'TOPPED_UP' } },
+        where: {
+          transactionHash_eventType: {
+            transactionHash: event.txHash,
+            eventType: "TOPPED_UP",
+          },
+        },
         create: {
           streamId,
-          eventType: 'TOPPED_UP',
+          eventType: "TOPPED_UP",
           amount,
           transactionHash: event.txHash,
           ledgerSequence: event.ledger,
@@ -798,11 +821,18 @@ export class SorobanEventWorker {
       // Check for a duplicate BEFORE mutating any Stream fields so that a
       // replayed event never double-increments withdrawnAmount.
       const existingEvent = await tx.streamEvent.findUnique({
-        where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
+        where: {
+          transactionHash_eventType: {
+            transactionHash: event.txHash,
+            eventType: "WITHDRAWN",
+          },
+        },
         select: { id: true },
       });
       if (existingEvent) {
-        logger.warn(`[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=WITHDRAWN`);
+        logger.warn(
+          `[SorobanWorker] Duplicate StreamEvent skipped: txHash=${event.txHash} type=WITHDRAWN`,
+        );
         return;
       }
 
@@ -824,10 +854,15 @@ export class SorobanEventWorker {
       });
 
       await tx.streamEvent.upsert({
-        where: { transactionHash_eventType: { transactionHash: event.txHash, eventType: 'WITHDRAWN' } },
+        where: {
+          transactionHash_eventType: {
+            transactionHash: event.txHash,
+            eventType: "WITHDRAWN",
+          },
+        },
         create: {
           streamId,
-          eventType: 'WITHDRAWN',
+          eventType: "WITHDRAWN",
           amount,
           transactionHash: event.txHash,
           ledgerSequence: event.ledger,