diff --git a/backend/src/app.ts b/backend/src/app.ts index 56c3035e..7ec53a22 100644 --- a/backend/src/app.ts +++ b/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"; diff --git a/backend/src/middleware/rate-limiter.middleware.ts b/backend/src/middleware/rate-limiter.middleware.ts index 61079a03..e35081d2 100644 --- a/backend/src/middleware/rate-limiter.middleware.ts +++ b/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) { + 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, diff --git a/backend/src/middleware/stream-rate-limiter.middleware.ts b/backend/src/middleware/stream-rate-limiter.middleware.ts index 19fe84a2..4b551a30 100644 --- a/backend/src/middleware/stream-rate-limiter.middleware.ts +++ b/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.', diff --git a/backend/src/workers/soroban-event-worker.ts b/backend/src/workers/soroban-event-worker.ts index 495e2539..179e58af 100644 --- a/backend/src/workers/soroban-event-worker.ts +++ b/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 { 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 { 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,