diff --git a/backend/src/eventOutbox.ts b/backend/src/eventOutbox.ts index d0c46c24d..9947e11f7 100644 --- a/backend/src/eventOutbox.ts +++ b/backend/src/eventOutbox.ts @@ -168,6 +168,10 @@ class EventOutboxService { * Returns the created outbox record. */ async writeEvent(input: OutboxWriteInput): Promise { + if (!prisma.eventOutbox) { + throw new Error('prisma.eventOutbox is not initialized. Ensure Prisma schema includes eventOutbox model.'); + } + assertEventOutboxModelAvailable(); const now = new Date(); const record = await getEventOutboxDelegate().create({ diff --git a/backend/src/idempotency.ts b/backend/src/idempotency.ts index cefbc293f..cf111e0d6 100644 --- a/backend/src/idempotency.ts +++ b/backend/src/idempotency.ts @@ -27,6 +27,8 @@ import { prisma } from './prisma'; import { redisClientManager } from './rateLimiter'; import { logger } from './middleware/structuredLogging'; import type { Request, Response, NextFunction } from 'express'; +import NodeCache from 'node-cache'; +import { redisClientManager } from './rateLimiter'; /** * The replay cache used by money-moving operations (deposits, withdrawals, @@ -394,6 +396,306 @@ declare global { } } +// ─── Error Classes ────────────────────────────────────────────────────── + +export class IdempotencyConflictError extends Error { + constructor(message = 'Idempotency key already used for a different request body') { + super(message); + this.name = 'IdempotencyConflictError'; + } +} + +// ─── Internal Types ───────────────────────────────────────────────────── + +export interface IdempotentOperationResult { + statusCode: number; + body: T; +} + +interface StoredResponse extends IdempotentOperationResult { + fingerprint: string; + metadata: IdempotencyKeyMetadata; +} + +interface PendingOperation { + fingerprint: string; + promise: Promise>; + metadata: IdempotencyKeyMetadata; +} + +export interface IdempotencyKeyInfo { + key: string; + metadata: IdempotencyKeyMetadata; +} + +export interface IdempotencyMetrics { + hits: number; + conflicts: number; + evictions: number; + activeKeys: number; + pendingKeys: number; +} + +// ─── Redis key prefix ─────────────────────────────────────────────────── + +const REDIS_PREFIX = 'idempotency:'; + +// ─── Fingerprint helper ───────────────────────────────────────────────── + +export function getIdempotencyHashThreshold(): number { + return parseInt(process.env.IDEMPOTENCY_HASH_THRESHOLD_BYTES || '4096', 10); +} + +export function buildIdempotencyFingerprint(payload: unknown): string { + const stable = stableStringify(payload); + const byteLength = Buffer.byteLength(stable, 'utf-8'); + if (byteLength > getIdempotencyHashThreshold()) { + return `hashv1:${crypto.createHash('sha256').update(stable).digest('hex')}`; + } + return stable; +} + +function stableStringify(value: unknown): string { + if (value === null) return 'null'; + if (value instanceof Date) return JSON.stringify(value.toISOString()); + if (typeof value !== 'object') return JSON.stringify(value); + + if (Array.isArray(value)) { + return `[${value.map((item) => stableStringify(item)).join(',')}]`; + } + + const record = value as Record; + const keys = Object.keys(record).sort(); + const serialized = keys.map((key) => `${JSON.stringify(key)}:${stableStringify(record[key])}`); + return `{${serialized.join(',')}}`; +} + +// ─── IdempotencyStore ─────────────────────────────────────────────────── + +export class IdempotencyStore { + private readonly completedResponses: NodeCache; + private readonly pendingResponses = new Map>(); + + private _hits = 0; + private _conflicts = 0; + private _evictions = 0; + + constructor(private readonly ttlMs = 24 * 60 * 60 * 1000) { + const ttlSeconds = Math.max(1, Math.ceil(this.ttlMs / 1000)); + this.completedResponses = new NodeCache({ stdTTL: ttlSeconds, checkperiod: ttlSeconds }); + this.completedResponses.on('expired', (_key: string, _value: unknown) => { + this._evictions++; + }); + } + + private redisKey(key: string): string { + return `${REDIS_PREFIX}${key}`; + } + + private get redis() { + const client = redisClientManager.getClient(); + return redisClientManager.isReady() && client ? client : null; + } + + private async redisGet(key: string): Promise | null> { + const r = this.redis; + if (!r) return null; + try { + const raw = await r.get(this.redisKey(key)); + return raw ? (JSON.parse(raw) as StoredResponse) : null; + } catch { + return null; + } + } + + private async redisSet(key: string, value: StoredResponse): Promise { + const r = this.redis; + if (!r) return; + try { + const ttlSeconds = Math.max(1, Math.ceil(this.ttlMs / 1000)); + await r.set(this.redisKey(key), JSON.stringify(value), 'EX', ttlSeconds); + } catch (err) { + console.log(JSON.stringify({ level: 'warn', event: 'idempotency_redis_write_fail', key, reason: (err as Error).message })); + } + } + + private async redisDel(key: string): Promise { + const r = this.redis; + if (!r) return false; + try { + return (await r.del(this.redisKey(key))) > 0; + } catch { + return false; + } + } + + async execute( + key: string, + fingerprint: string, + operation: () => Promise> + ): Promise<{ result: IdempotentOperationResult; replayed: boolean }> { + const now = new Date().toISOString(); + + let completed = await this.redisGet(key); + if (!completed) { + completed = this.completedResponses.get>(key) ?? null; + } + + if (completed) { + if (completed.fingerprint !== fingerprint) { + this._conflicts++; + throw new IdempotencyConflictError(); + } + this._hits++; + completed.metadata.lastAccessedAt = now; + completed.metadata.replayCount++; + await this.redisSet(key, completed); + this.completedResponses.set(key, completed); + return { result: { statusCode: completed.statusCode, body: completed.body }, replayed: true }; + } + + const pendingOperation = this.pendingResponses.get(key) as PendingOperation | undefined; + if (pendingOperation) { + if (pendingOperation.fingerprint !== fingerprint) { + this._conflicts++; + throw new IdempotencyConflictError(); + } + this._hits++; + pendingOperation.metadata.lastAccessedAt = now; + pendingOperation.metadata.replayCount++; + const replayed = await pendingOperation.promise; + return { result: { statusCode: replayed.statusCode, body: replayed.body }, replayed: true }; + } + + const metadata: IdempotencyKeyMetadata = { createdAt: now, lastAccessedAt: now, replayCount: 0, status: 'pending' }; + + const operationPromise = (async () => { + const result = await operation(); + const stored: StoredResponse = { + ...result, + fingerprint, + metadata: { ...metadata, status: 'completed', lastAccessedAt: new Date().toISOString() }, + }; + await this.redisSet(key, stored); + this.completedResponses.set(key, stored, this.ttlMs / 1000); + return stored; + })(); + + this.pendingResponses.set(key, { fingerprint, promise: operationPromise, metadata }); + + try { + const stored = await operationPromise; + return { result: { statusCode: stored.statusCode, body: stored.body }, replayed: false }; + } finally { + this.pendingResponses.delete(key); + } + } + + inspectKeys(prefix?: string): IdempotencyKeyInfo[] { + const results: IdempotencyKeyInfo[] = []; + for (const key of this.completedResponses.keys()) { + if (prefix && !key.startsWith(prefix)) continue; + const entry = this.completedResponses.get>(key); + if (entry) results.push({ key, metadata: { ...entry.metadata } }); + } + for (const [key, pending] of this.pendingResponses.entries()) { + if (prefix && !key.startsWith(prefix)) continue; + if (!results.some((r) => r.key === key)) { + results.push({ key, metadata: { ...pending.metadata } }); + } + } + return results; + } + + async deleteKey(key: string): Promise { + const deletedLocal = this.completedResponses.del(key) > 0; + const deletedPending = this.pendingResponses.delete(key); + const deletedRedis = await this.redisDel(key); + if (deletedLocal || deletedPending || deletedRedis) { + this._evictions++; + return true; + } + return false; + } + + clear(): void { + const count = this.completedResponses.keys().length + this.pendingResponses.size; + this._evictions += count; + this.completedResponses.flushAll(); + this.pendingResponses.clear(); + } + + getMetrics(): IdempotencyMetrics { + return { + hits: this._hits, + conflicts: this._conflicts, + evictions: this._evictions, + activeKeys: this.completedResponses.keys().length, + pendingKeys: this.pendingResponses.size, + }; + } + + async pruneStaleKeys( + retentionMs: number, + dryRun = false, + ): Promise<{ pruned: number; localPruned: number; redisPruned: number }> { + const cutoff = Date.now() - retentionMs; + let localPruned = 0; + let redisPruned = 0; + + for (const key of this.completedResponses.keys()) { + const entry = this.completedResponses.get>(key); + if (!entry) continue; + const createdAt = Date.parse(entry.metadata.createdAt); + if (Number.isNaN(createdAt) || createdAt >= cutoff) continue; + if (!dryRun) { + this.completedResponses.del(key); + this._evictions++; + } + localPruned++; + } + + const r = this.redis; + if (r) { + let cursor = '0'; + do { + const [nextCursor, keys] = await r.scan(cursor, 'MATCH', `${REDIS_PREFIX}*`, 'COUNT', 100); + cursor = nextCursor; + for (const redisKey of keys) { + try { + const raw = await r.get(redisKey); + if (!raw) continue; + const entry = JSON.parse(raw) as StoredResponse; + const createdAt = Date.parse(entry.metadata?.createdAt ?? ''); + const ttl = await r.ttl(redisKey); + const isStale = (!Number.isNaN(createdAt) && createdAt < cutoff) || ttl === 0; + if (!isStale) continue; + if (!dryRun) { + await r.del(redisKey); + this._evictions++; + } + redisPruned++; + } catch { + if (!dryRun) { + await r.del(redisKey); + this._evictions++; + } + redisPruned++; + } + } + } while (cursor !== '0'); + } + + return { pruned: localPruned + redisPruned, localPruned, redisPruned }; + } +} + +// ─── Singleton ────────────────────────────────────────────────────────── + +export const idempotencyStore = new IdempotencyStore( + parseInt(process.env.IDEMPOTENCY_KEY_TTL_MS || '86400000', 10) +); + // ─── Cleanup ──────────────────────────────────────────────────────────────── /** diff --git a/backend/src/middleware/cache.ts b/backend/src/middleware/cache.ts index 45e383c2e..c080b4ca1 100644 --- a/backend/src/middleware/cache.ts +++ b/backend/src/middleware/cache.ts @@ -338,6 +338,7 @@ export function triggerCacheInvalidation( for (const hook of invalidationHooks) { try { const hookPatterns = hook(eventType, metadata); + // Validate and sanitize hook return value if (hookPatterns instanceof Promise) { void hookPatterns.catch((err) => { console.error( @@ -359,12 +360,16 @@ export function triggerCacheInvalidation( console.error( JSON.stringify({ level: 'error', + event: 'invalidation_hook_error', + error: `InvalidationHook must return string[], got ${typeof hookPatterns}`, + eventType, event: 'invalidation_hook_invalid_return', error: 'Invalidation hook must return an array of cache key patterns', }), ); continue; } + patterns.push(...hookPatterns); patterns.push(...hookPatterns.filter((pattern): pattern is string => typeof pattern === 'string')); } catch (err) { console.error(