diff --git a/backend/elasticsearch/__tests__/connectionPool.test.ts b/backend/elasticsearch/__tests__/connectionPool.test.ts new file mode 100644 index 00000000..ad992177 --- /dev/null +++ b/backend/elasticsearch/__tests__/connectionPool.test.ts @@ -0,0 +1,278 @@ +/** + * Tests — ElasticsearchConnectionPool (Issue #986) + */ + +import { + ElasticsearchConnectionPool, + DnsCache, + resetDefaultPool, + type ConnectionPoolConfig, +} from '../connectionPool'; + +const BASE_CONFIG: ConnectionPoolConfig = { + primaryHost: 'es-primary', + primaryPort: 9200, + poolSize: 5, + acquireTimeoutMs: 200, + idleTimeoutMs: 500, + leakThresholdMs: 300, + dnsCacheTtlMs: 1_000, + maintenanceIntervalMs: 100, +}; + +describe('ElasticsearchConnectionPool', () => { + let pool: ElasticsearchConnectionPool; + + beforeEach(() => { + jest.useFakeTimers(); + pool = new ElasticsearchConnectionPool(BASE_CONFIG); + }); + + afterEach(() => { + pool.shutdown(); + resetDefaultPool(); + jest.useRealTimers(); + }); + + // ── Pool initialisation ─────────────────────────────────────────────────── + + it('initialises pool with correct size', () => { + expect(pool.size()).toBe(BASE_CONFIG.poolSize); + expect(pool.idleCount()).toBe(BASE_CONFIG.poolSize); + expect(pool.activeCount()).toBe(0); + }); + + it('assigns primary role to all connections when no replicas', () => { + const conns = pool.listConnections(); + expect(conns.every((c) => c.role === 'primary')).toBe(true); + }); + + it('assigns replica roles when replicas configured', () => { + const p = new ElasticsearchConnectionPool({ + ...BASE_CONFIG, + replicas: [{ host: 'replica-1', port: 9201 }], + poolSize: 4, + }); + const roles = p.listConnections().map((c) => c.role); + expect(roles).toContain('replica'); + expect(roles).toContain('primary'); + p.shutdown(); + }); + + // ── Acquire / Release ───────────────────────────────────────────────────── + + it('acquire returns a connection and marks it in-use', async () => { + const { connection } = await pool.acquire(); + expect(connection.inUse).toBe(true); + expect(connection.acquiredAt).toBeDefined(); + }); + + it('release frees connection back to pool', async () => { + const { connection } = await pool.acquire(); + expect(pool.activeCount()).toBe(1); + pool.release(connection.id); + expect(pool.activeCount()).toBe(0); + expect(pool.idleCount()).toBe(BASE_CONFIG.poolSize); + }); + + it('queued acquires are resolved after release', async () => { + // Exhaust pool + const acquired: Array<{ connection: { id: string } }> = []; + for (let i = 0; i < BASE_CONFIG.poolSize; i++) { + acquired.push(await pool.acquire()); + } + expect(pool.idleCount()).toBe(0); + + // Queue a waiter + const waiting = pool.acquire(); + // Release one connection + pool.release(acquired[0]!.connection.id); + + const resolved = await waiting; + expect(resolved.connection.inUse).toBe(true); + // Cleanup + for (let i = 1; i < acquired.length; i++) pool.release(acquired[i]!.connection.id); + pool.release(resolved.connection.id); + }); + + it('acquire times out when pool exhausted', async () => { + // Exhaust all connections + const acquired: Array<{ connection: { id: string } }> = []; + for (let i = 0; i < BASE_CONFIG.poolSize; i++) { + acquired.push(await pool.acquire()); + } + + const p = pool.acquire(); + jest.advanceTimersByTime(BASE_CONFIG.acquireTimeoutMs + 10); + await expect(p).rejects.toThrow('acquire timed out'); + + // Cleanup + for (const a of acquired) pool.release(a.connection.id); + }); + + it('release is a no-op for unknown connection id', () => { + expect(() => pool.release('nonexistent')).not.toThrow(); + }); + + // ── withConnection ──────────────────────────────────────────────────────── + + it('withConnection releases connection even when fn throws', async () => { + await expect( + pool.withConnection(async () => { + throw new Error('query failed'); + }), + ).rejects.toThrow('query failed'); + expect(pool.idleCount()).toBe(BASE_CONFIG.poolSize); + }); + + it('withConnection passes connection to fn', async () => { + const result = await pool.withConnection(async (conn) => conn.id); + expect(typeof result).toBe('string'); + }); + + // ── Read routing ────────────────────────────────────────────────────────── + + it('readOnly acquire prefers replica connections', async () => { + const p = new ElasticsearchConnectionPool({ + ...BASE_CONFIG, + replicas: [{ host: 'replica-1', port: 9201 }], + poolSize: 4, + }); + const { connection } = await p.acquire(true); + expect(connection.role).toBe('replica'); + p.release(connection.id); + p.shutdown(); + }); + + it('readOnly falls back to primary when replicas exhausted', async () => { + const p = new ElasticsearchConnectionPool({ + ...BASE_CONFIG, + replicas: [{ host: 'replica-1', port: 9201 }], + poolSize: 4, + }); + const replicas = p.listConnections().filter((c) => c.role === 'replica'); + // Acquire all replicas + for (const r of replicas) await p.acquire(true); + + // Next read-only acquire should fall back to primary + const { connection } = await p.acquire(true); + expect(connection.role).toBe('primary'); + p.shutdown(); + }); + + // ── Leak detection ──────────────────────────────────────────────────────── + + it('emits leak event for long-held connections', async () => { + const leaks: string[] = []; + pool.on('leak', ({ connectionId }) => leaks.push(connectionId)); + + const { connection } = await pool.acquire(); + jest.advanceTimersByTime(BASE_CONFIG.leakThresholdMs + BASE_CONFIG.maintenanceIntervalMs + 50); + await Promise.resolve(); // flush + + expect(leaks).toContain(connection.id); + expect(pool.getMetrics().leaksDetected).toBeGreaterThanOrEqual(1); + pool.release(connection.id); + }); + + // ── Idle teardown ───────────────────────────────────────────────────────── + + it('emits idle-teardown for long-idle connections', async () => { + const teardowns: string[] = []; + pool.on('idle-teardown', (id) => teardowns.push(id)); + + const { connection } = await pool.acquire(); + pool.release(connection.id); + jest.advanceTimersByTime(BASE_CONFIG.idleTimeoutMs + BASE_CONFIG.maintenanceIntervalMs + 50); + await Promise.resolve(); + + expect(teardowns.length).toBeGreaterThan(0); + }); + + // ── Metrics ─────────────────────────────────────────────────────────────── + + it('tracks acquires and releases in metrics', async () => { + const { connection } = await pool.acquire(); + pool.release(connection.id); + const m = pool.getMetrics(); + expect(m.totalAcquires).toBe(1); + expect(m.totalReleases).toBe(1); + expect(m.peakActiveConnections).toBe(1); + }); + + it('tracks acquire timeouts in metrics', async () => { + for (let i = 0; i < BASE_CONFIG.poolSize; i++) await pool.acquire(); + const p = pool.acquire(); + jest.advanceTimersByTime(BASE_CONFIG.acquireTimeoutMs + 10); + await expect(p).rejects.toThrow(); + expect(pool.getMetrics().acquireTimeouts).toBe(1); + }); + + it('prometheusMetrics returns valid format', async () => { + const prom = pool.prometheusMetrics(); + expect(prom).toContain('subtrackr_es_pool_connections_total'); + expect(prom).toContain('subtrackr_es_pool_acquire_timeouts_total'); + }); + + // ── Tuning recommendations ──────────────────────────────────────────────── + + it('recommends pool increase at high utilisation', async () => { + // Acquire all connections to simulate peak + const acquired = await Promise.all( + Array.from({ length: BASE_CONFIG.poolSize }, () => pool.acquire()), + ); + for (const a of acquired) pool.release(a.connection.id); + + const recs = pool.getTuningRecommendations(); + // At 100% peak utilisation should recommend increase + expect(recs.some((r) => r.includes('poolSize'))).toBe(true); + }); + + it('reports healthy when under-utilised', () => { + const recs = pool.getTuningRecommendations(); + // Fresh pool with 0 acquires — low utilisation + expect(recs.length).toBeGreaterThan(0); + }); + + // ── Shutdown ────────────────────────────────────────────────────────────── + + it('shutdown rejects pending waiters', async () => { + for (let i = 0; i < BASE_CONFIG.poolSize; i++) await pool.acquire(); + const p = pool.acquire(); + pool.shutdown(); + await expect(p).rejects.toThrow('shutdown'); + }); +}); + +// --------------------------------------------------------------------------- +// DnsCache +// --------------------------------------------------------------------------- + +describe('DnsCache', () => { + it('caches resolved addresses', async () => { + const cache = new DnsCache(); + await cache.resolve('es-primary', 1_000); + await cache.resolve('es-primary', 1_000); + expect(cache.stats().hits).toBe(1); + expect(cache.stats().lookups).toBe(1); + }); + + it('invalidates cache entry', async () => { + const cache = new DnsCache(); + await cache.resolve('es-primary', 1_000); + cache.invalidate('es-primary'); + await cache.resolve('es-primary', 1_000); + expect(cache.stats().lookups).toBe(2); + expect(cache.stats().hits).toBe(0); + }); + + it('re-resolves after TTL expires', async () => { + jest.useFakeTimers(); + const cache = new DnsCache(); + await cache.resolve('es-node', 100); + jest.advanceTimersByTime(200); + await cache.resolve('es-node', 100); + expect(cache.stats().lookups).toBe(2); + jest.useRealTimers(); + }); +}); diff --git a/backend/elasticsearch/config.ts b/backend/elasticsearch/config.ts index 08f829b9..08229e2a 100644 --- a/backend/elasticsearch/config.ts +++ b/backend/elasticsearch/config.ts @@ -3,8 +3,60 @@ * In this mobile-first architecture the "cluster" is an in-process index * backed by AsyncStorage, mirroring a real ES setup so the service layer * can be swapped for a remote cluster without changing callers. + * + * Issue #986: Extended with connection pool settings. */ +// --------------------------------------------------------------------------- +// Connection pool config (Issue #986) +// --------------------------------------------------------------------------- + +export interface ElasticsearchPoolConfig { + /** Primary node host. Default: localhost */ + primaryHost: string; + /** Primary node port. Default: 9200 */ + primaryPort: number; + /** Optional read replicas for query routing. */ + replicas?: { host: string; port: number }[]; + /** + * Total connections in pool across primary + replicas. + * Recommended: (vCPUs * 2) for IO-bound ES workloads. + * Default: 10 + */ + poolSize: number; + /** Milliseconds to wait for a free connection. Default: 5000 */ + acquireTimeoutMs: number; + /** Idle connection teardown threshold (ms). Default: 60_000 */ + idleTimeoutMs: number; + /** Connection-held-too-long leak threshold (ms). Default: 30_000 */ + leakThresholdMs: number; + /** DNS cache TTL (ms). Default: 30_000 */ + dnsCacheTtlMs: number; + /** Maintenance sweep interval (ms). Default: 10_000 */ + maintenanceIntervalMs: number; +} + +export const DEFAULT_POOL_CONFIG: ElasticsearchPoolConfig = { + primaryHost: process.env['ES_PRIMARY_HOST'] ?? 'localhost', + primaryPort: Number(process.env['ES_PRIMARY_PORT'] ?? 9200), + replicas: process.env['ES_REPLICA_HOSTS'] + ? process.env['ES_REPLICA_HOSTS'].split(',').map((h) => { + const [host, port] = h.split(':'); + return { host: host ?? 'localhost', port: Number(port ?? 9200) }; + }) + : [], + poolSize: Number(process.env['ES_POOL_SIZE'] ?? 10), + acquireTimeoutMs: Number(process.env['ES_ACQUIRE_TIMEOUT_MS'] ?? 5_000), + idleTimeoutMs: Number(process.env['ES_IDLE_TIMEOUT_MS'] ?? 60_000), + leakThresholdMs: Number(process.env['ES_LEAK_THRESHOLD_MS'] ?? 30_000), + dnsCacheTtlMs: Number(process.env['ES_DNS_CACHE_TTL_MS'] ?? 30_000), + maintenanceIntervalMs: Number(process.env['ES_MAINTENANCE_INTERVAL_MS'] ?? 10_000), +}; + +// --------------------------------------------------------------------------- +// Index / Search config +// --------------------------------------------------------------------------- + export interface ElasticsearchConfig { indexName: string; fuzzyMaxEdits: number; @@ -14,6 +66,8 @@ export interface ElasticsearchConfig { analyticsEnabled: boolean; /** Analyzer locales used for multilingual tokenization */ analyzerLocales: string[]; + /** Connection pool settings (Issue #986) */ + pool?: ElasticsearchPoolConfig; } export const DEFAULT_ES_CONFIG: ElasticsearchConfig = { @@ -33,6 +87,7 @@ export const DEFAULT_ES_CONFIG: ElasticsearchConfig = { maxResults: 100, analyticsEnabled: true, analyzerLocales: ['en', 'fr', 'de', 'es'], + pool: DEFAULT_POOL_CONFIG, }; export interface IndexMapping { diff --git a/backend/elasticsearch/connectionPool.ts b/backend/elasticsearch/connectionPool.ts new file mode 100644 index 00000000..67b89ac6 --- /dev/null +++ b/backend/elasticsearch/connectionPool.ts @@ -0,0 +1,584 @@ +/** + * Elasticsearch Connection Pool — SubTrackr + * + * Issue #986: Implement database connection pooling with optimization + * + * Features: + * - Fixed-size pool of logical ES "connections" (HTTP agents / client handles) + * - Acquire/release lifecycle with configurable acquire timeout + * - Idle-connection timeout with automatic teardown + * - Connection leak detection (configurable threshold) + * - DNS-level caching with per-entry TTL (avoids re-resolution per request) + * - Read/write routing: round-robin across replicas for reads, primary for writes + * - Pool exhaustion alerting (callback + log) + * - Prometheus metrics export + * - Tuning recommendations based on peak utilisation + */ + +import { EventEmitter } from 'events'; + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +export type ConnectionRole = 'primary' | 'replica'; + +export interface PooledConnection { + readonly id: string; + readonly host: string; + readonly port: number; + readonly role: ConnectionRole; + inUse: boolean; + acquiredAt?: number; + lastUsedAt: number; + createdAt: number; + useCount: number; + /** Resolved IP stored after first DNS lookup. */ + resolvedAddress?: string; + resolvedAt?: number; +} + +export interface ConnectionPoolConfig { + /** Primary ES node host. */ + primaryHost: string; + primaryPort: number; + /** Optional replica nodes for read routing. */ + replicas?: { host: string; port: number }[]; + /** Total connections in the pool (primary + replicas share this budget). */ + poolSize: number; + /** How long (ms) to wait for a free connection before throwing. Default: 5000. */ + acquireTimeoutMs: number; + /** Idle connections released after this many ms without use. Default: 60_000. */ + idleTimeoutMs: number; + /** Connections held for longer than this (ms) are flagged as leaked. Default: 30_000. */ + leakThresholdMs: number; + /** DNS cache entry TTL in ms. Default: 30_000. */ + dnsCacheTtlMs: number; + /** Interval for idle-sweep & leak-detection (ms). Default: 10_000. */ + maintenanceIntervalMs: number; +} + +export interface AcquireResult { + connection: PooledConnection; + /** True if this connection was idle-refreshed (DNS re-validated). */ + dnsRefreshed: boolean; +} + +export interface PoolMetrics { + totalConnections: number; + activeConnections: number; + idleConnections: number; + acquireTimeouts: number; + leaksDetected: number; + totalAcquires: number; + totalReleases: number; + avgAcquireWaitMs: number; + peakActiveConnections: number; + dnsLookups: number; + dnsCacheHits: number; + /** Utilisation 0–1 based on peak vs pool size. */ + peakUtilisation: number; +} + +export type LeakAlertHandler = (conn: PooledConnection, heldMs: number) => void; +export type ExhaustionAlertHandler = (waitingCount: number) => void; + +// --------------------------------------------------------------------------- +// DNS Cache +// --------------------------------------------------------------------------- + +interface DnsCacheEntry { + address: string; + resolvedAt: number; + ttlMs: number; +} + +export class DnsCache { + private readonly cache = new Map(); + private lookups = 0; + private hits = 0; + + /** Simulate DNS resolution (real impl would use dns.resolve4). */ + async resolve(host: string, ttlMs: number): Promise { + const cached = this.cache.get(host); + if (cached && Date.now() - cached.resolvedAt < cached.ttlMs) { + this.hits++; + return cached.address; + } + // In production this would be `dns.promises.resolve4(host)` + const address = host; // pass-through for non-prod / tests + this.lookups++; + this.cache.set(host, { address, resolvedAt: Date.now(), ttlMs }); + return address; + } + + invalidate(host: string): void { + this.cache.delete(host); + } + + clear(): void { + this.cache.clear(); + } + + stats(): { lookups: number; hits: number; hitRate: number; size: number } { + return { + lookups: this.lookups, + hits: this.hits, + hitRate: this.lookups > 0 ? this.hits / this.lookups : 0, + size: this.cache.size, + }; + } +} + +// --------------------------------------------------------------------------- +// Connection Pool +// --------------------------------------------------------------------------- + +export class ElasticsearchConnectionPool extends EventEmitter { + private readonly config: ConnectionPoolConfig; + private readonly pool: PooledConnection[] = []; + private readonly waitQueue: Array<{ + resolve: (result: AcquireResult) => void; + reject: (err: Error) => void; + readonly enqueueAt: number; + readonly readOnly: boolean; + }> = []; + + private readonly dnsCache = new DnsCache(); + private maintenanceTimer?: ReturnType; + + // Round-robin index for replicas + private replicaRoundRobin = 0; + + // Metrics + private acquireTimeouts = 0; + private leaksDetected = 0; + private totalAcquires = 0; + private totalReleases = 0; + private totalAcquireWaitMs = 0; + private peakActiveConnections = 0; + + // Alerts + public onLeak?: LeakAlertHandler; + public onExhaustion?: ExhaustionAlertHandler; + + constructor(config: ConnectionPoolConfig) { + super(); + this.config = { + ...config, + acquireTimeoutMs: config.acquireTimeoutMs ?? 5_000, + idleTimeoutMs: config.idleTimeoutMs ?? 60_000, + leakThresholdMs: config.leakThresholdMs ?? 30_000, + dnsCacheTtlMs: config.dnsCacheTtlMs ?? 30_000, + maintenanceIntervalMs: config.maintenanceIntervalMs ?? 10_000, + }; + this.initPool(); + this.startMaintenance(); + } + + // ── Initialisation ──────────────────────────────────────────────────────── + + private initPool(): void { + const { poolSize, primaryHost, primaryPort, replicas = [] } = this.config; + + // Distribute pool slots: replicas get floor(poolSize / (replicas+1)) each, + // remainder goes to primary. + const nodeCount = 1 + replicas.length; + const slotsPerNode = Math.max(1, Math.floor(poolSize / nodeCount)); + const primarySlots = poolSize - slotsPerNode * replicas.length; + + for (let i = 0; i < primarySlots; i++) { + this.pool.push(this.createConnection(primaryHost, primaryPort, 'primary')); + } + for (const replica of replicas) { + for (let i = 0; i < slotsPerNode; i++) { + this.pool.push(this.createConnection(replica.host, replica.port, 'replica')); + } + } + } + + private createConnection(host: string, port: number, role: ConnectionRole): PooledConnection { + return { + id: `conn_${Date.now()}_${Math.random().toString(36).slice(2, 8)}`, + host, + port, + role, + inUse: false, + lastUsedAt: Date.now(), + createdAt: Date.now(), + useCount: 0, + }; + } + + // ── Acquire / Release ───────────────────────────────────────────────────── + + /** + * Acquire a connection from the pool. + * + * @param readOnly If true, prefers replica connections (read routing). + * Falls back to primary when all replicas are busy. + */ + async acquire(readOnly = false): Promise { + const start = Date.now(); + const conn = this.tryAcquireSync(readOnly); + if (conn) { + this.totalAcquires++; + this.totalAcquireWaitMs += Date.now() - start; + const active = this.activeCount(); + if (active > this.peakActiveConnections) this.peakActiveConnections = active; + return this.prepareConnection(conn); + } + + // Pool exhausted — alert and queue the waiter + this.onExhaustion?.(this.waitQueue.length + 1); + this.emit('exhaustion', this.waitQueue.length + 1); + + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + const idx = this.waitQueue.findIndex((w) => w.resolve === resolve); + if (idx !== -1) this.waitQueue.splice(idx, 1); + this.acquireTimeouts++; + reject( + new Error( + `ElasticsearchConnectionPool: acquire timed out after ${this.config.acquireTimeoutMs}ms. ` + + `Pool size: ${this.config.poolSize}, waiting: ${this.waitQueue.length}`, + ), + ); + }, this.config.acquireTimeoutMs); + + this.waitQueue.push({ + resolve: (result) => { + clearTimeout(timer); + this.totalAcquires++; + this.totalAcquireWaitMs += Date.now() - start; + const active = this.activeCount(); + if (active > this.peakActiveConnections) this.peakActiveConnections = active; + resolve(result); + }, + reject: (err) => { + clearTimeout(timer); + reject(err); + }, + enqueueAt: Date.now(), + readOnly, + }); + }); + } + + /** + * Release a connection back to the pool. + * Automatically dispatches to the next waiter if any are queued. + */ + release(connectionId: string): void { + const conn = this.pool.find((c) => c.id === connectionId); + if (!conn) return; + + conn.inUse = false; + conn.acquiredAt = undefined; + conn.lastUsedAt = Date.now(); + conn.useCount++; + this.totalReleases++; + + this.emit('release', conn.id); + + // Drain the wait queue + if (this.waitQueue.length > 0) { + const waiter = this.waitQueue.shift()!; + const next = this.tryAcquireSync(waiter.readOnly); + if (next) { + this.prepareConnection(next).then(waiter.resolve).catch(waiter.reject); + } else { + // Put back; nothing free yet + this.waitQueue.unshift(waiter); + } + } + } + + /** + * Execute a query function with automatic acquire/release lifecycle. + * + * @example + * const result = await pool.withConnection( + * async (conn) => client.search({ index: 'subs', ... }), + * true, // readOnly + * ); + */ + async withConnection( + fn: (conn: PooledConnection) => Promise, + readOnly = false, + ): Promise { + const { connection } = await this.acquire(readOnly); + try { + return await fn(connection); + } finally { + this.release(connection.id); + } + } + + // ── Private: sync acquire ───────────────────────────────────────────────── + + private tryAcquireSync(readOnly: boolean): PooledConnection | null { + if (readOnly) { + // Prefer least-recently-used replica + const replicas = this.pool.filter((c) => c.role === 'replica' && !c.inUse); + if (replicas.length > 0) { + const idx = this.replicaRoundRobin % replicas.length; + this.replicaRoundRobin = (this.replicaRoundRobin + 1) % replicas.length; + const conn = replicas[idx]!; + conn.inUse = true; + conn.acquiredAt = Date.now(); + return conn; + } + // Fall through to primary + } + + // Primary or any free connection + const free = this.pool.find((c) => !c.inUse); + if (free) { + free.inUse = true; + free.acquiredAt = Date.now(); + return free; + } + + return null; + } + + private async prepareConnection(conn: PooledConnection): Promise { + let dnsRefreshed = false; + try { + const resolved = await this.dnsCache.resolve(conn.host, this.config.dnsCacheTtlMs); + if (resolved !== conn.resolvedAddress) { + conn.resolvedAddress = resolved; + conn.resolvedAt = Date.now(); + dnsRefreshed = true; + } + } catch { + // DNS resolution failure is non-fatal; proceed with stored address + } + return { connection: conn, dnsRefreshed }; + } + + // ── Maintenance ─────────────────────────────────────────────────────────── + + private startMaintenance(): void { + this.maintenanceTimer = setInterval(() => { + this.runIdleSweep(); + this.runLeakDetection(); + }, this.config.maintenanceIntervalMs); + + // Unref so it doesn't prevent Node process exit + if (typeof this.maintenanceTimer.unref === 'function') { + this.maintenanceTimer.unref(); + } + } + + private runIdleSweep(): void { + const now = Date.now(); + for (const conn of this.pool) { + if (!conn.inUse && now - conn.lastUsedAt > this.config.idleTimeoutMs) { + // "Teardown" the logical connection: reset DNS cache entry so next + // acquire re-validates (simulates closing and re-opening a socket). + this.dnsCache.invalidate(conn.host); + conn.resolvedAddress = undefined; + conn.resolvedAt = undefined; + this.emit('idle-teardown', conn.id); + } + } + } + + private runLeakDetection(): void { + const now = Date.now(); + for (const conn of this.pool) { + if (conn.inUse && conn.acquiredAt && now - conn.acquiredAt > this.config.leakThresholdMs) { + const heldMs = now - conn.acquiredAt; + this.leaksDetected++; + this.onLeak?.(conn, heldMs); + this.emit('leak', { connectionId: conn.id, heldMs }); + } + } + } + + // ── Pool management ─────────────────────────────────────────────────────── + + /** + * Drain: wait for all active connections to be released (graceful shutdown). + */ + async drain(timeoutMs = 10_000): Promise { + const deadline = Date.now() + timeoutMs; + while (this.activeCount() > 0 && Date.now() < deadline) { + await new Promise((r) => setTimeout(r, 100)); + } + this.shutdown(); + } + + /** + * Immediate shutdown — stops maintenance timer and rejects all waiters. + */ + shutdown(): void { + if (this.maintenanceTimer) { + clearInterval(this.maintenanceTimer); + this.maintenanceTimer = undefined; + } + for (const waiter of this.waitQueue) { + waiter.reject(new Error('ElasticsearchConnectionPool: shutdown')); + } + this.waitQueue.length = 0; + this.dnsCache.clear(); + } + + // ── Queries ─────────────────────────────────────────────────────────────── + + activeCount(): number { + return this.pool.filter((c) => c.inUse).length; + } + + idleCount(): number { + return this.pool.filter((c) => !c.inUse).length; + } + + size(): number { + return this.pool.length; + } + + getConnection(id: string): PooledConnection | undefined { + return this.pool.find((c) => c.id === id); + } + + listConnections(): PooledConnection[] { + return [...this.pool]; + } + + // ── Metrics ─────────────────────────────────────────────────────────────── + + getMetrics(): PoolMetrics { + const dns = this.dnsCache.stats(); + const active = this.activeCount(); + const idle = this.idleCount(); + return { + totalConnections: this.pool.length, + activeConnections: active, + idleConnections: idle, + acquireTimeouts: this.acquireTimeouts, + leaksDetected: this.leaksDetected, + totalAcquires: this.totalAcquires, + totalReleases: this.totalReleases, + avgAcquireWaitMs: + this.totalAcquires > 0 + ? Math.round(this.totalAcquireWaitMs / this.totalAcquires) + : 0, + peakActiveConnections: this.peakActiveConnections, + dnsLookups: dns.lookups, + dnsCacheHits: dns.hits, + peakUtilisation: + this.pool.length > 0 ? this.peakActiveConnections / this.pool.length : 0, + }; + } + + prometheusMetrics(namespace = 'subtrackr_es_pool'): string { + const m = this.getMetrics(); + return [ + `# HELP ${namespace}_connections_total Total connections in pool`, + `# TYPE ${namespace}_connections_total gauge`, + `${namespace}_connections_total ${m.totalConnections}`, + `# HELP ${namespace}_connections_active Active (in-use) connections`, + `# TYPE ${namespace}_connections_active gauge`, + `${namespace}_connections_active ${m.activeConnections}`, + `# HELP ${namespace}_connections_idle Idle connections`, + `# TYPE ${namespace}_connections_idle gauge`, + `${namespace}_connections_idle ${m.idleConnections}`, + `# HELP ${namespace}_acquire_timeouts_total Acquire timeout count`, + `# TYPE ${namespace}_acquire_timeouts_total counter`, + `${namespace}_acquire_timeouts_total ${m.acquireTimeouts}`, + `# HELP ${namespace}_leaks_total Leak detection triggers`, + `# TYPE ${namespace}_leaks_total counter`, + `${namespace}_leaks_total ${m.leaksDetected}`, + `# HELP ${namespace}_acquires_total Total acquire operations`, + `# TYPE ${namespace}_acquires_total counter`, + `${namespace}_acquires_total ${m.totalAcquires}`, + `# HELP ${namespace}_avg_acquire_wait_ms Average wait time for acquire (ms)`, + `# TYPE ${namespace}_avg_acquire_wait_ms gauge`, + `${namespace}_avg_acquire_wait_ms ${m.avgAcquireWaitMs}`, + `# HELP ${namespace}_peak_utilisation Peak utilisation ratio (0-1)`, + `# TYPE ${namespace}_peak_utilisation gauge`, + `${namespace}_peak_utilisation ${m.peakUtilisation.toFixed(4)}`, + `# HELP ${namespace}_dns_cache_hits_total DNS cache hits`, + `# TYPE ${namespace}_dns_cache_hits_total counter`, + `${namespace}_dns_cache_hits_total ${m.dnsCacheHits}`, + ].join('\n'); + } + + /** + * Returns tuning recommendations based on observed peak utilisation. + */ + getTuningRecommendations(): string[] { + const m = this.getMetrics(); + const recs: string[] = []; + + if (m.peakUtilisation > 0.9) { + recs.push( + `Pool is at ${(m.peakUtilisation * 100).toFixed(0)}% peak utilisation. ` + + `Consider increasing poolSize from ${this.config.poolSize} to ${Math.ceil(this.config.poolSize * 1.5)}.`, + ); + } + if (m.acquireTimeouts > 0) { + recs.push( + `${m.acquireTimeouts} acquire timeout(s) detected. ` + + `Increase acquireTimeoutMs (current: ${this.config.acquireTimeoutMs}ms) or pool size.`, + ); + } + if (m.leaksDetected > 0) { + recs.push( + `${m.leaksDetected} connection leak(s) detected (held > ${this.config.leakThresholdMs}ms). ` + + `Ensure all code paths call pool.release() or use pool.withConnection().`, + ); + } + if (m.avgAcquireWaitMs > 100) { + recs.push( + `Average acquire wait is ${m.avgAcquireWaitMs}ms. ` + + `Consider increasing pool size or optimising query durations.`, + ); + } + if (m.peakUtilisation < 0.2 && this.config.poolSize > 5) { + recs.push( + `Peak utilisation is only ${(m.peakUtilisation * 100).toFixed(0)}%. ` + + `Consider reducing poolSize to free resources.`, + ); + } + if (m.dnsCacheHits / Math.max(1, m.dnsLookups + m.dnsCacheHits) < 0.5) { + recs.push( + `DNS cache hit rate is low. Consider increasing dnsCacheTtlMs ` + + `(current: ${this.config.dnsCacheTtlMs}ms).`, + ); + } + + if (recs.length === 0) { + recs.push('Pool configuration looks healthy. No tuning required at this time.'); + } + + return recs; + } +} + +// --------------------------------------------------------------------------- +// Singleton factory +// --------------------------------------------------------------------------- + +let defaultPool: ElasticsearchConnectionPool | undefined; + +export function getDefaultPool(config?: ConnectionPoolConfig): ElasticsearchConnectionPool { + if (!defaultPool) { + if (!config) { + throw new Error( + 'ElasticsearchConnectionPool: no default pool initialised. ' + + 'Call getDefaultPool(config) with a valid config first.', + ); + } + defaultPool = new ElasticsearchConnectionPool(config); + } + return defaultPool; +} + +export function resetDefaultPool(): void { + defaultPool?.shutdown(); + defaultPool = undefined; +} diff --git a/backend/services/eventBusIntegration.ts b/backend/services/eventBusIntegration.ts new file mode 100644 index 00000000..a9c46287 --- /dev/null +++ b/backend/services/eventBusIntegration.ts @@ -0,0 +1,455 @@ +/** + * Event Bus Integration — SubTrackr + * + * Issue #984: Refactor event system to use typed event bus with domain events + * + * This module wires the backend service layer to the typed EventBus so that + * every domain operation (subscription lifecycle, billing, analytics) emits + * a strongly-typed DomainEvent rather than ad-hoc callbacks or raw emitters. + * + * Architecture: + * - Each domain integration class wraps its underlying service. + * - Operations are delegated to the service; on success a DomainEvent is + * published to the shared EventBus and appended to the EventStore. + * - Downstream consumers (analytics, notifications, audit) subscribe to + * specific event names without coupling to service internals. + * - The file is the single integration point — services themselves remain + * unaware of the bus, keeping them testable in isolation. + */ + +import { + eventBus, + eventStore, + buildEvent, + EventBus, + InMemoryEventStore, + EventCollector, + SpyEventBus, + type IEventBus, + type EventSourcedStore, + type AnyDomainEvent, + type SubscriptionCreatedPayload, + type SubscriptionCancelledPayload, + type SubscriptionRenewedPayload, + type SubscriptionPausedPayload, + type SubscriptionResumedPayload, + type SubscriptionUpgradedPayload, + type SubscriptionPaymentFailedPayload, + type InvoiceGeneratedPayload, + type PaymentCapturedPayload, + type ChargebackRaisedPayload, + type UsageThresholdReachedPayload, + type ChurnRiskUpdatedPayload, + type MrrChangedPayload, + type ContractInvokedPayload, +} from './shared/events'; + +// Re-export for convenience +export { + EventBus, + InMemoryEventStore, + EventCollector, + SpyEventBus, + buildEvent, + eventBus, + eventStore, +}; + +// --------------------------------------------------------------------------- +// Types for domain operations +// --------------------------------------------------------------------------- + +export interface SubscriptionCreationInput { + subscriptionId: string; + userId: string; + planId: string; + status: string; + billingCycle: string; + nextBillingDate: number; +} + +export interface SubscriptionCancellationInput { + subscriptionId: string; + userId: string; + reason?: string; + cancelledAt: number; + effectiveAt: number; +} + +export interface SubscriptionRenewalInput { + subscriptionId: string; + userId: string; + planId: string; + renewedAt: number; + nextBillingDate: number; + amount: number; + currency: string; +} + +export interface SubscriptionUpgradeInput { + subscriptionId: string; + userId: string; + fromPlanId: string; + toPlanId: string; + effectiveAt: number; + proratedCredit?: number; +} + +export interface SubscriptionPauseInput { + subscriptionId: string; + userId: string; + pausedAt: number; + resumeAt?: number; +} + +export interface SubscriptionResumeInput { + subscriptionId: string; + userId: string; + resumedAt: number; +} + +export interface PaymentFailureInput { + subscriptionId: string; + userId: string; + attemptNumber: number; + nextRetryAt?: number; + reason: string; +} + +export interface InvoiceGenerationInput { + invoiceId: string; + subscriptionId: string; + userId: string; + amount: number; + currency: string; + dueDate: number; +} + +export interface PaymentCaptureInput { + paymentId: string; + subscriptionId: string; + userId: string; + amount: number; + currency: string; + capturedAt: number; + gateway: string; +} + +export interface ChargebackInput { + chargebackId: string; + subscriptionId: string; + userId: string; + amount: number; + currency: string; + reason: string; + raisedAt: number; +} + +export interface UsageThresholdInput { + subscriptionId: string; + userId: string; + metricType: string; + usage: number; + limit: number; + level: 'soft' | 'hard'; +} + +export interface ChurnRiskInput { + subscriptionId: string; + userId: string; + riskScore: number; + previousScore?: number; + factors: string[]; +} + +export interface MrrChangeInput { + previousMrr: number; + currentMrr: number; + currency: string; + periodStart: number; + periodEnd: number; +} + +export interface ContractInvocationInput { + contractId: string; + method: string; + caller: string; + ledger: number; + txHash: string; +} + +// --------------------------------------------------------------------------- +// Subscription Domain Integration +// --------------------------------------------------------------------------- + +export class SubscriptionEventPublisher { + constructor( + private readonly bus: IEventBus = eventBus, + private readonly store: EventSourcedStore = eventStore, + ) {} + + async publishCreated(input: SubscriptionCreationInput): Promise { + const payload: SubscriptionCreatedPayload = { ...input }; + const event = buildEvent('subscription', 'created', payload, { + aggregateId: input.subscriptionId, + correlationId: input.userId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } + + async publishCancelled(input: SubscriptionCancellationInput): Promise { + const payload: SubscriptionCancelledPayload = { ...input }; + const event = buildEvent('subscription', 'cancelled', payload, { + aggregateId: input.subscriptionId, + correlationId: input.userId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } + + async publishRenewed(input: SubscriptionRenewalInput): Promise { + const payload: SubscriptionRenewedPayload = { ...input }; + const event = buildEvent('subscription', 'renewed', payload, { + aggregateId: input.subscriptionId, + correlationId: input.userId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } + + async publishUpgraded(input: SubscriptionUpgradeInput): Promise { + const payload: SubscriptionUpgradedPayload = { ...input }; + const event = buildEvent('subscription', 'upgraded', payload, { + aggregateId: input.subscriptionId, + correlationId: input.userId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } + + async publishPaused(input: SubscriptionPauseInput): Promise { + const payload: SubscriptionPausedPayload = { ...input }; + const event = buildEvent('subscription', 'paused', payload, { + aggregateId: input.subscriptionId, + correlationId: input.userId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } + + async publishResumed(input: SubscriptionResumeInput): Promise { + const payload: SubscriptionResumedPayload = { ...input }; + const event = buildEvent('subscription', 'resumed', payload, { + aggregateId: input.subscriptionId, + correlationId: input.userId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } + + async publishPaymentFailed(input: PaymentFailureInput): Promise { + const payload: SubscriptionPaymentFailedPayload = { ...input }; + const event = buildEvent('subscription', 'payment_failed', payload, { + aggregateId: input.subscriptionId, + correlationId: input.userId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } + + /** + * Replay all events for a subscription to reconstruct its current state. + */ + replaySubscription(subscriptionId: string): Record { + return this.store.reconstruct(subscriptionId); + } +} + +// --------------------------------------------------------------------------- +// Billing Domain Integration +// --------------------------------------------------------------------------- + +export class BillingEventPublisher { + constructor( + private readonly bus: IEventBus = eventBus, + private readonly store: EventSourcedStore = eventStore, + ) {} + + async publishInvoiceGenerated(input: InvoiceGenerationInput): Promise { + const payload: InvoiceGeneratedPayload = { ...input }; + const event = buildEvent('billing', 'invoice_generated', payload, { + aggregateId: input.subscriptionId, + correlationId: input.invoiceId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } + + async publishPaymentCaptured(input: PaymentCaptureInput): Promise { + const payload: PaymentCapturedPayload = { ...input }; + const event = buildEvent('billing', 'payment_captured', payload, { + aggregateId: input.subscriptionId, + correlationId: input.paymentId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } + + async publishChargebackRaised(input: ChargebackInput): Promise { + const payload: ChargebackRaisedPayload = { ...input }; + const event = buildEvent('billing', 'chargeback_raised', payload, { + aggregateId: input.subscriptionId, + correlationId: input.chargebackId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } + + async publishUsageThresholdReached(input: UsageThresholdInput): Promise { + const { usage, limit } = input; + const payload: UsageThresholdReachedPayload = { + ...input, + ratio: limit > 0 ? usage / limit : 0, + }; + const event = buildEvent('billing', 'usage_threshold_reached', payload, { + aggregateId: input.subscriptionId, + correlationId: input.userId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } +} + +// --------------------------------------------------------------------------- +// Analytics Domain Integration +// --------------------------------------------------------------------------- + +export class AnalyticsEventPublisher { + constructor( + private readonly bus: IEventBus = eventBus, + private readonly store: EventSourcedStore = eventStore, + ) {} + + async publishChurnRiskUpdated(input: ChurnRiskInput): Promise { + const payload: ChurnRiskUpdatedPayload = { ...input }; + const event = buildEvent('analytics', 'churn_risk_updated', payload, { + aggregateId: input.subscriptionId, + correlationId: input.userId, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } + + async publishMrrChanged(input: MrrChangeInput): Promise { + const payload: MrrChangedPayload = { + ...input, + delta: input.currentMrr - input.previousMrr, + }; + const event = buildEvent('analytics', 'mrr_changed', payload, {}); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } +} + +// --------------------------------------------------------------------------- +// Contract Domain Integration (Stellar / Soroban) +// --------------------------------------------------------------------------- + +export class ContractEventPublisher { + constructor( + private readonly bus: IEventBus = eventBus, + private readonly store: EventSourcedStore = eventStore, + ) {} + + async publishContractInvoked(input: ContractInvocationInput): Promise { + const payload: ContractInvokedPayload = { ...input }; + const event = buildEvent('contract', 'invoked', payload, { + aggregateId: input.contractId, + correlationId: input.txHash, + }); + this.store.append(event as AnyDomainEvent); + await this.bus.publish(event as AnyDomainEvent); + } +} + +// --------------------------------------------------------------------------- +// Domain Event Router +// Registers standard cross-domain handlers so subscriptions to one domain +// can trigger actions in another (e.g., subscription.cancelled → billing alert) +// --------------------------------------------------------------------------- + +export interface DomainEventRouterOptions { + onSubscriptionCancelled?: (subscriptionId: string, userId: string) => Promise; + onPaymentFailed?: (subscriptionId: string, attemptNumber: number) => Promise; + onUsageHardLimitReached?: (subscriptionId: string, metricType: string) => Promise; + onChurnRiskHigh?: (subscriptionId: string, riskScore: number) => Promise; +} + +export class DomainEventRouter { + private readonly subscriptions: ReturnType[] = []; + + constructor( + private readonly bus: IEventBus = eventBus, + private readonly handlers: DomainEventRouterOptions = {}, + ) {} + + /** + * Register all standard cross-domain routing rules. + * Call once at application startup. + */ + register(): void { + this.subscriptions.push( + this.bus.subscribe('subscription.cancelled', async (event) => { + await this.handlers.onSubscriptionCancelled?.( + event.payload.subscriptionId, + event.payload.userId, + ); + }), + + this.bus.subscribe('subscription.payment_failed', async (event) => { + await this.handlers.onPaymentFailed?.( + event.payload.subscriptionId, + event.payload.attemptNumber, + ); + }), + + this.bus.subscribe( + 'billing.usage_threshold_reached', + async (event) => { + await this.handlers.onUsageHardLimitReached?.( + event.payload.subscriptionId, + event.payload.metricType, + ); + }, + { filter: (e) => e.payload.level === 'hard' }, + ), + + this.bus.subscribe( + 'analytics.churn_risk_updated', + async (event) => { + await this.handlers.onChurnRiskHigh?.( + event.payload.subscriptionId, + event.payload.riskScore, + ); + }, + { filter: (e) => e.payload.riskScore >= 0.8 }, + ), + ); + } + + /** Unregister all routing rules. */ + unregister(): void { + for (const sub of this.subscriptions) sub.unsubscribe(); + this.subscriptions.length = 0; + } +} + +// --------------------------------------------------------------------------- +// Singletons (shared across services) +// --------------------------------------------------------------------------- + +export const subscriptionEventPublisher = new SubscriptionEventPublisher(); +export const billingEventPublisher = new BillingEventPublisher(); +export const analyticsEventPublisher = new AnalyticsEventPublisher(); +export const contractEventPublisher = new ContractEventPublisher(); diff --git a/backend/services/shared/__tests__/apiKeyRotation.test.ts b/backend/services/shared/__tests__/apiKeyRotation.test.ts new file mode 100644 index 00000000..093db6ed --- /dev/null +++ b/backend/services/shared/__tests__/apiKeyRotation.test.ts @@ -0,0 +1,246 @@ +/** + * Tests — API Key Rotation Service (Issue #1009) + */ + +import { + ApiKeyRotationService, + type ManagedApiKey, +} from '../apiKeyRotation'; + +// Helper: advance mocked clock by ms +const tickMs = (n: number) => { + jest.setSystemTime(Date.now() + n); +}; + +describe('ApiKeyRotationService', () => { + let service: ApiKeyRotationService; + + beforeEach(() => { + jest.useFakeTimers(); + jest.setSystemTime(new Date('2026-01-01T00:00:00Z').getTime()); + service = new ApiKeyRotationService(); + }); + + afterEach(() => { + jest.useRealTimers(); + }); + + // ── createKey ───────────────────────────────────────────────────────────── + + describe('createKey', () => { + it('creates an active key with correct metadata', () => { + const key = service.createKey('dev_1', 'My Key', 'test', ['subscriptions:read']); + expect(key.status).toBe('active'); + expect(key.developerId).toBe('dev_1'); + expect(key.environment).toBe('test'); + expect(key.permissions).toEqual(['subscriptions:read']); + expect(key.key).toMatch(/^sk_test_/); + expect(key.usageCount).toBe(0); + }); + + it('creates production keys with sk_live_ prefix', () => { + const key = service.createKey('dev_1', 'Prod Key', 'production'); + expect(key.key).toMatch(/^sk_live_/); + }); + + it('assigns unique IDs per call', () => { + const a = service.createKey('dev_1', 'Key A', 'test'); + const b = service.createKey('dev_1', 'Key B', 'test'); + expect(a.id).not.toBe(b.id); + expect(a.key).not.toBe(b.key); + }); + }); + + // ── validateKey ─────────────────────────────────────────────────────────── + + describe('validateKey', () => { + it('returns valid for an active key', () => { + const key = service.createKey('dev_1', 'K', 'test'); + const result = service.validateKey(key.key); + expect(result.valid).toBe(true); + expect(result.isGrace).toBe(false); + expect(result.graceRemainingMs).toBe(0); + }); + + it('returns invalid for an unknown key', () => { + const result = service.validateKey('sk_test_unknown'); + expect(result.valid).toBe(false); + expect(result.reason).toBe('key_not_found'); + }); + + it('increments usageCount on valid use', () => { + const key = service.createKey('dev_1', 'K', 'test'); + service.validateKey(key.key); + service.validateKey(key.key); + expect(service.getKey(key.id)!.usageCount).toBe(2); + }); + + it('expires an active key past its TTL', () => { + const key = service.createKey('dev_1', 'K', 'test', [], 1_000); // 1s TTL + tickMs(2_000); + const result = service.validateKey(key.key); + expect(result.valid).toBe(false); + expect(result.reason).toBe('expired'); + }); + }); + + // ── rotateKey ───────────────────────────────────────────────────────────── + + describe('rotateKey', () => { + it('returns a new active key and old key in grace', async () => { + const old = service.createKey('dev_1', 'K', 'test'); + const result = await service.rotateKey(old.id, { gracePeriodMs: 60_000 }); + + expect(result.newKey.status).toBe('active'); + expect(result.oldKey.status).toBe('grace'); + expect(result.gracePeriodRemainingMs).toBe(60_000); + expect(result.newKey.replacesKeyId).toBe(old.id); + expect(result.oldKey.replacedByKeyId).toBe(result.newKey.id); + }); + + it('old key remains valid during grace period', async () => { + const old = service.createKey('dev_1', 'K', 'test'); + const { oldKey, newKey } = await service.rotateKey(old.id, { + gracePeriodMs: 60_000, + }); + + const oldValidation = service.validateKey(old.key); + expect(oldValidation.valid).toBe(true); + expect(oldValidation.isGrace).toBe(true); + expect(oldValidation.graceRemainingMs).toBeGreaterThan(0); + + const newValidation = service.validateKey(newKey.key); + expect(newValidation.valid).toBe(true); + expect(newValidation.isGrace).toBe(false); + }); + + it('old key becomes invalid after grace period expires', async () => { + const old = service.createKey('dev_1', 'K', 'test'); + const { oldKey } = await service.rotateKey(old.id, { gracePeriodMs: 5_000 }); + + tickMs(6_000); // advance past grace + const result = service.validateKey(old.key); + expect(result.valid).toBe(false); + expect(result.reason).toBe('expired'); + }); + + it('throws when rotating a revoked key', async () => { + const key = service.createKey('dev_1', 'K', 'test'); + service.revokeKey(key.id); + await expect(service.rotateKey(key.id)).rejects.toThrow('Cannot rotate a revoked key'); + }); + + it('throws when rotating an expired key', async () => { + const key = service.createKey('dev_1', 'K', 'test', [], 1_000); + tickMs(2_000); + service.validateKey(key.key); // triggers expiry + await expect(service.rotateKey(key.id)).rejects.toThrow('Cannot rotate an expired key'); + }); + + it('records rotation history', async () => { + const key = service.createKey('dev_1', 'K', 'test'); + await service.rotateKey(key.id, { reason: 'security audit' }); + const history = service.getRotationHistory('dev_1'); + expect(history).toHaveLength(1); + expect(history[0]!.reason).toBe('security audit'); + expect(history[0]!.oldKeyId).toBe(key.id); + }); + }); + + // ── revokeKey ───────────────────────────────────────────────────────────── + + describe('revokeKey', () => { + it('revokes a key and makes it invalid', () => { + const key = service.createKey('dev_1', 'K', 'test'); + expect(service.revokeKey(key.id)).toBe(true); + const result = service.validateKey(key.key); + expect(result.valid).toBe(false); + expect(result.reason).toBe('revoked'); + }); + + it('returns false for already-revoked key', () => { + const key = service.createKey('dev_1', 'K', 'test'); + service.revokeKey(key.id); + expect(service.revokeKey(key.id)).toBe(false); + }); + + it('revokeAllKeys revokes all developer keys', async () => { + const k1 = service.createKey('dev_2', 'K1', 'test'); + const k2 = service.createKey('dev_2', 'K2', 'production'); + const other = service.createKey('dev_3', 'Other', 'test'); + + const count = service.revokeAllKeys('dev_2'); + expect(count).toBe(2); + expect(service.validateKey(k1.key).valid).toBe(false); + expect(service.validateKey(k2.key).valid).toBe(false); + expect(service.validateKey(other.key).valid).toBe(true); + }); + }); + + // ── cleanupExpiredKeys ──────────────────────────────────────────────────── + + describe('cleanupExpiredKeys', () => { + it('transitions grace keys past deadline to expired', async () => { + const key = service.createKey('dev_1', 'K', 'test'); + await service.rotateKey(key.id, { gracePeriodMs: 1_000 }); + tickMs(2_000); + const cleaned = service.cleanupExpiredKeys(); + expect(cleaned).toBeGreaterThanOrEqual(1); + expect(service.getKey(key.id)!.status).toBe('expired'); + }); + }); + + // ── getGracePeriodStatus ────────────────────────────────────────────────── + + describe('getGracePeriodStatus', () => { + it('returns remaining ms for a key in grace', async () => { + const key = service.createKey('dev_1', 'K', 'test'); + const { newKey } = await service.rotateKey(key.id, { gracePeriodMs: 60_000 }); + const status = service.getGracePeriodStatus(key.id); + expect(status).not.toBeNull(); + expect(status!.remainingMs).toBeGreaterThan(0); + expect(status!.successorKeyId).toBe(newKey.id); + }); + + it('returns null for an active (non-grace) key', () => { + const key = service.createKey('dev_1', 'K', 'test'); + expect(service.getGracePeriodStatus(key.id)).toBeNull(); + }); + + it('returns null after grace period has passed', async () => { + const key = service.createKey('dev_1', 'K', 'test'); + await service.rotateKey(key.id, { gracePeriodMs: 1_000 }); + tickMs(2_000); + expect(service.getGracePeriodStatus(key.id)).toBeNull(); + }); + }); + + // ── metrics ─────────────────────────────────────────────────────────────── + + describe('metrics', () => { + it('tracks key counts and validation totals', async () => { + const k1 = service.createKey('dev_1', 'K1', 'test'); + const k2 = service.createKey('dev_1', 'K2', 'test'); + + service.validateKey(k1.key); // active + await service.rotateKey(k1.id, { gracePeriodMs: 60_000 }); + service.validateKey(k1.key); // grace hit + service.revokeKey(k2.id); + + const m = service.getMetrics(); + expect(m.totalKeys).toBeGreaterThanOrEqual(3); // k1, k2, new rotated + expect(m.graceKeys).toBe(1); + expect(m.revokedKeys).toBe(1); + expect(m.totalRotations).toBe(1); + expect(m.gracePeriodHits).toBe(1); + expect(m.totalValidations).toBeGreaterThanOrEqual(2); + }); + + it('exports valid prometheus metrics string', () => { + service.createKey('dev_1', 'K', 'test'); + const prom = service.prometheusMetrics(); + expect(prom).toContain('subtrackr_api_key_total'); + expect(prom).toContain('subtrackr_api_key_rotations_total'); + }); + }); +}); diff --git a/backend/services/shared/__tests__/events.test.ts b/backend/services/shared/__tests__/events.test.ts new file mode 100644 index 00000000..4af46df9 --- /dev/null +++ b/backend/services/shared/__tests__/events.test.ts @@ -0,0 +1,483 @@ +/** + * Tests — Typed Event Bus with Domain Events (Issue #984) + */ + +import { + EventBus, + InMemoryEventStore, + SpyEventBus, + EventCollector, + buildEvent, + validateEventPayload, + EventValidationError, + eventBusPrometheusMetrics, + type AnyDomainEvent, +} from '../events'; +import { + SubscriptionEventPublisher, + BillingEventPublisher, + AnalyticsEventPublisher, + ContractEventPublisher, + DomainEventRouter, +} from '../../eventBusIntegration'; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function makeSubCreated(overrides = {}): AnyDomainEvent { + return buildEvent('subscription', 'created', { + subscriptionId: 'sub_1', + userId: 'usr_1', + planId: 'plan_basic', + status: 'active', + billingCycle: 'monthly', + nextBillingDate: Date.now() + 30 * 86_400_000, + ...overrides, + }) as AnyDomainEvent; +} + +// --------------------------------------------------------------------------- +// buildEvent +// --------------------------------------------------------------------------- + +describe('buildEvent', () => { + it('creates event with correct shape', () => { + const event = makeSubCreated(); + expect(event.domain).toBe('subscription'); + expect(event.type).toBe('created'); + expect(event.name).toBe('subscription.created'); + expect(typeof event.id).toBe('string'); + expect(typeof event.occurredAt).toBe('number'); + expect(typeof event.sequence).toBe('number'); + expect(event.schemaVersion).toBe(1); + }); + + it('assigns monotonically increasing sequence numbers', () => { + const a = buildEvent('subscription', 'created', { subscriptionId: 'a', userId: 'u', planId: 'p', status: 'active', billingCycle: 'monthly', nextBillingDate: 0 }); + const b = buildEvent('subscription', 'cancelled', { subscriptionId: 'b', userId: 'u', cancelledAt: 0, effectiveAt: 0 }); + expect(b.sequence).toBeGreaterThan(a.sequence); + }); + + it('passes through aggregateId and correlationId', () => { + const event = buildEvent('auth', 'api_key_rotated', { + keyId: 'k1', merchantId: 'm1', rotatedAt: Date.now(), + }, { aggregateId: 'agg_1', correlationId: 'corr_1' }); + expect(event.aggregateId).toBe('agg_1'); + expect(event.correlationId).toBe('corr_1'); + }); +}); + +// --------------------------------------------------------------------------- +// validateEventPayload +// --------------------------------------------------------------------------- + +describe('validateEventPayload', () => { + it('passes validation for a correct subscription.created payload', () => { + const result = validateEventPayload('subscription.created', { + subscriptionId: 'sub_1', + userId: 'usr_1', + planId: 'plan_basic', + status: 'active', + billingCycle: 'monthly', + nextBillingDate: Date.now(), + }); + expect(result.valid).toBe(true); + expect(result.errors).toHaveLength(0); + }); + + it('catches missing required fields', () => { + const result = validateEventPayload('subscription.created', { + userId: 'usr_1', + }); + expect(result.valid).toBe(false); + expect(result.errors.some((e) => e.includes('subscriptionId'))).toBe(true); + }); + + it('catches wrong field type', () => { + const result = validateEventPayload('subscription.created', { + subscriptionId: 'sub_1', + userId: 'usr_1', + planId: 'plan_basic', + status: 'active', + billingCycle: 'monthly', + nextBillingDate: 'not-a-number', + }); + expect(result.valid).toBe(false); + expect(result.errors.some((e) => e.includes('nextBillingDate'))).toBe(true); + }); + + it('returns valid for unknown event names (no schema)', () => { + const result = validateEventPayload('analytics.churn_risk_updated', { + anything: 'goes', + }); + expect(result.valid).toBe(true); + }); +}); + +// --------------------------------------------------------------------------- +// EventBus — publish / subscribe +// --------------------------------------------------------------------------- + +describe('EventBus', () => { + let bus: EventBus; + + beforeEach(() => { + bus = new EventBus(); + }); + + it('calls subscriber when matching event published', async () => { + const received: AnyDomainEvent[] = []; + bus.subscribe('subscription.created', (e) => { received.push(e); }); + await bus.publish(makeSubCreated()); + expect(received).toHaveLength(1); + expect(received[0]!.name).toBe('subscription.created'); + }); + + it('wildcard subscriber receives all events', async () => { + const received: AnyDomainEvent[] = []; + bus.subscribe('*', (e) => { received.push(e); }); + await bus.publish(makeSubCreated()); + await bus.publish(buildEvent('billing', 'invoice_generated', { + invoiceId: 'inv_1', subscriptionId: 'sub_1', userId: 'usr_1', + amount: 100, currency: 'USD', dueDate: Date.now(), + }) as AnyDomainEvent); + expect(received).toHaveLength(2); + }); + + it('subscriber with predicate filter only receives matching events', async () => { + const received: AnyDomainEvent[] = []; + bus.subscribe( + 'subscription.created', + (e) => { received.push(e); }, + { filter: (e) => (e.payload as Record)['planId'] === 'plan_premium' }, + ); + await bus.publish(makeSubCreated({ planId: 'plan_basic' })); + await bus.publish(makeSubCreated({ planId: 'plan_premium' })); + expect(received).toHaveLength(1); + expect((received[0]!.payload as Record)['planId']).toBe('plan_premium'); + }); + + it('non-matching subscriber does not receive event', async () => { + const received: AnyDomainEvent[] = []; + bus.subscribe('billing.invoice_generated', (e) => { received.push(e); }); + await bus.publish(makeSubCreated()); + expect(received).toHaveLength(0); + }); + + it('unsubscribe stops handler from receiving events', async () => { + const received: AnyDomainEvent[] = []; + const sub = bus.subscribe('subscription.created', (e) => { received.push(e); }); + sub.unsubscribe(); + await bus.publish(makeSubCreated()); + expect(received).toHaveLength(0); + }); + + it('throws EventValidationError on invalid payload', async () => { + const badEvent = buildEvent('subscription', 'created', { + subscriptionId: 123 as unknown as string, // wrong type + userId: 'usr_1', + planId: 'p', + status: 's', + billingCycle: 'monthly', + nextBillingDate: Date.now(), + }) as AnyDomainEvent; + await expect(bus.publish(badEvent)).rejects.toThrow(EventValidationError); + }); + + it('isolates handler errors — does not abort other handlers', async () => { + const safe: string[] = []; + bus.subscribe('subscription.created', () => { throw new Error('boom'); }); + bus.subscribe('subscription.created', () => { safe.push('ok'); }); + await bus.publish(makeSubCreated()); + expect(safe).toContain('ok'); + expect(bus.getMetrics().errors).toBe(1); + }); + + it('tracks metrics correctly', async () => { + await bus.publish(makeSubCreated()); + await bus.publish(makeSubCreated()); + const m = bus.getMetrics(); + expect(m.published).toBe(2); + expect(m.countByName['subscription.created']).toBe(2); + }); + + it('resetMetrics zeroes all counters', async () => { + await bus.publish(makeSubCreated()); + bus.resetMetrics(); + expect(bus.getMetrics().published).toBe(0); + }); +}); + +// --------------------------------------------------------------------------- +// InMemoryEventStore +// --------------------------------------------------------------------------- + +describe('InMemoryEventStore', () => { + let store: InMemoryEventStore; + + beforeEach(() => { + store = new InMemoryEventStore(); + }); + + it('appends and queries events', () => { + const e1 = makeSubCreated(); + const e2 = buildEvent('subscription', 'cancelled', { + subscriptionId: 'sub_1', userId: 'usr_1', cancelledAt: Date.now(), effectiveAt: Date.now(), + }) as AnyDomainEvent; + store.append(e1); + store.append(e2); + expect(store.query().length).toBe(2); + }); + + it('filters by aggregateId', () => { + const e1 = buildEvent('subscription', 'created', { + subscriptionId: 'sub_1', userId: 'u', planId: 'p', status: 'active', + billingCycle: 'monthly', nextBillingDate: 0, + }, { aggregateId: 'sub_1' }) as AnyDomainEvent; + const e2 = buildEvent('subscription', 'created', { + subscriptionId: 'sub_2', userId: 'u', planId: 'p', status: 'active', + billingCycle: 'monthly', nextBillingDate: 0, + }, { aggregateId: 'sub_2' }) as AnyDomainEvent; + store.append(e1); + store.append(e2); + expect(store.query({ aggregateId: 'sub_1' }).length).toBe(1); + }); + + it('filters by domain', () => { + store.append(makeSubCreated()); + store.append(buildEvent('billing', 'invoice_generated', { + invoiceId: 'inv_1', subscriptionId: 'sub_1', userId: 'u', + amount: 100, currency: 'USD', dueDate: 0, + }) as AnyDomainEvent); + expect(store.query({ domain: 'billing' }).length).toBe(1); + }); + + it('replays events in sequence order', () => { + const e1 = buildEvent('subscription', 'created', { + subscriptionId: 'sub_1', userId: 'u', planId: 'p', status: 'active', + billingCycle: 'monthly', nextBillingDate: 0, + }, { aggregateId: 'sub_1' }) as AnyDomainEvent; + const e2 = buildEvent('subscription', 'cancelled', { + subscriptionId: 'sub_1', userId: 'u', cancelledAt: 0, effectiveAt: 0, + }, { aggregateId: 'sub_1' }) as AnyDomainEvent; + store.append(e1); + store.append(e2); + const replayed: AnyDomainEvent[] = []; + store.replay('sub_1', (e) => replayed.push(e)); + expect(replayed[0]!.type).toBe('created'); + expect(replayed[1]!.type).toBe('cancelled'); + }); + + it('reconstruct merges all payloads', () => { + const e1 = buildEvent('subscription', 'created', { + subscriptionId: 'sub_1', userId: 'u', planId: 'p', status: 'active', + billingCycle: 'monthly', nextBillingDate: 0, + }, { aggregateId: 'sub_1' }) as AnyDomainEvent; + store.append(e1); + const state = store.reconstruct('sub_1'); + expect(state['userId']).toBe('u'); + }); + + it('archiveBefore marks old events archived', () => { + store.append(makeSubCreated()); + const archived = store.archiveBefore(Date.now() + 1_000); + expect(archived).toBe(1); + // Archived events excluded from default query + expect(store.query().length).toBe(0); + // But included when asked + expect(store.query({ includeArchived: true }).length).toBe(1); + }); +}); + +// --------------------------------------------------------------------------- +// SpyEventBus & EventCollector +// --------------------------------------------------------------------------- + +describe('SpyEventBus', () => { + it('records published events', async () => { + const spy = new SpyEventBus(); + await spy.publish(makeSubCreated()); + expect(spy.published).toHaveLength(1); + spy.assertPublished('subscription.created'); + }); + + it('assertEmpty throws when events present', async () => { + const spy = new SpyEventBus(); + await spy.publish(makeSubCreated()); + expect(() => spy.assertEmpty()).toThrow(); + }); + + it('resetMetrics clears published array', async () => { + const spy = new SpyEventBus(); + await spy.publish(makeSubCreated()); + spy.resetMetrics(); + expect(spy.published).toHaveLength(0); + }); +}); + +describe('EventCollector', () => { + it('collects events by name', async () => { + const bus = new EventBus(); + const collector = new EventCollector(bus, 'subscription.created'); + await bus.publish(makeSubCreated()); + collector.assertCount(1); + collector.assertLastEventName('subscription.created'); + collector.dispose(); + }); + + it('ofName filters collected events', async () => { + const bus = new EventBus(); + const collector = new EventCollector(bus); + await bus.publish(makeSubCreated()); + const subs = collector.ofName('subscription.created'); + expect(subs).toHaveLength(1); + collector.dispose(); + }); +}); + +// --------------------------------------------------------------------------- +// Prometheus metrics +// --------------------------------------------------------------------------- + +describe('eventBusPrometheusMetrics', () => { + it('outputs valid prometheus format', async () => { + const bus = new EventBus(); + await bus.publish(makeSubCreated()); + const output = eventBusPrometheusMetrics(bus); + expect(output).toContain('subtrackr_event_bus_published_total 1'); + expect(output).toContain('subtrackr_event_bus_errors_total'); + }); +}); + +// --------------------------------------------------------------------------- +// Domain Event Publishers (eventBusIntegration.ts) +// --------------------------------------------------------------------------- + +describe('SubscriptionEventPublisher', () => { + let spy: SpyEventBus; + let store: InMemoryEventStore; + let pub: SubscriptionEventPublisher; + + beforeEach(() => { + spy = new SpyEventBus(); + store = new InMemoryEventStore(); + pub = new SubscriptionEventPublisher(spy, store); + }); + + it('publishCreated emits subscription.created', async () => { + await pub.publishCreated({ + subscriptionId: 'sub_1', userId: 'u', planId: 'p', + status: 'active', billingCycle: 'monthly', nextBillingDate: Date.now(), + }); + spy.assertPublished('subscription.created'); + expect(store.query({ aggregateId: 'sub_1' })).toHaveLength(1); + }); + + it('publishCancelled emits subscription.cancelled', async () => { + await pub.publishCancelled({ + subscriptionId: 'sub_1', userId: 'u', cancelledAt: Date.now(), effectiveAt: Date.now(), + }); + spy.assertPublished('subscription.cancelled'); + }); + + it('publishPaymentFailed emits subscription.payment_failed', async () => { + await pub.publishPaymentFailed({ + subscriptionId: 'sub_1', userId: 'u', attemptNumber: 1, reason: 'card_declined', + }); + spy.assertPublished('subscription.payment_failed'); + }); + + it('replaySubscription reconstructs state from store', async () => { + await pub.publishCreated({ + subscriptionId: 'sub_r', userId: 'u', planId: 'p', + status: 'active', billingCycle: 'monthly', nextBillingDate: 0, + }); + const state = pub.replaySubscription('sub_r'); + expect(state['userId']).toBe('u'); + }); +}); + +describe('BillingEventPublisher', () => { + let spy: SpyEventBus; + let pub: BillingEventPublisher; + + beforeEach(() => { + spy = new SpyEventBus(); + pub = new BillingEventPublisher(spy, new InMemoryEventStore()); + }); + + it('publishInvoiceGenerated emits billing.invoice_generated', async () => { + await pub.publishInvoiceGenerated({ + invoiceId: 'inv_1', subscriptionId: 'sub_1', userId: 'u', + amount: 99.99, currency: 'USD', dueDate: Date.now(), + }); + spy.assertPublished('billing.invoice_generated'); + }); + + it('publishUsageThresholdReached computes ratio', async () => { + await pub.publishUsageThresholdReached({ + subscriptionId: 'sub_1', userId: 'u', + metricType: 'api_calls', usage: 900, limit: 1000, level: 'soft', + }); + const event = spy.published.find((e) => e.name === 'billing.usage_threshold_reached'); + expect((event!.payload as Record)['ratio']).toBeCloseTo(0.9); + }); +}); + +describe('DomainEventRouter', () => { + it('routes subscription.cancelled to handler', async () => { + const bus = new EventBus(); + const store = new InMemoryEventStore(); + const cancelled: string[] = []; + + const router = new DomainEventRouter(bus, { + onSubscriptionCancelled: async (subId) => { cancelled.push(subId); }, + }); + router.register(); + + const pub = new SubscriptionEventPublisher(bus, store); + await pub.publishCancelled({ + subscriptionId: 'sub_X', userId: 'u', cancelledAt: Date.now(), effectiveAt: Date.now(), + }); + expect(cancelled).toContain('sub_X'); + router.unregister(); + }); + + it('routes high churn risk to handler via predicate filter', async () => { + const bus = new EventBus(); + const highRisk: number[] = []; + + const router = new DomainEventRouter(bus, { + onChurnRiskHigh: async (_, score) => { highRisk.push(score); }, + }); + router.register(); + + const pub = new AnalyticsEventPublisher(bus, new InMemoryEventStore()); + await pub.publishChurnRiskUpdated({ + subscriptionId: 's', userId: 'u', riskScore: 0.5, factors: [], + }); + await pub.publishChurnRiskUpdated({ + subscriptionId: 's', userId: 'u', riskScore: 0.9, factors: ['payment_decline'], + }); + expect(highRisk).toHaveLength(1); + expect(highRisk[0]).toBe(0.9); + router.unregister(); + }); + + it('unregister stops all routing', async () => { + const bus = new EventBus(); + const calls: string[] = []; + const router = new DomainEventRouter(bus, { + onSubscriptionCancelled: async (id) => { calls.push(id); }, + }); + router.register(); + router.unregister(); + + const pub = new SubscriptionEventPublisher(bus, new InMemoryEventStore()); + await pub.publishCancelled({ + subscriptionId: 'sub_Y', userId: 'u', cancelledAt: Date.now(), effectiveAt: Date.now(), + }); + expect(calls).toHaveLength(0); + }); +}); diff --git a/backend/services/shared/__tests__/rateLimiting.test.ts b/backend/services/shared/__tests__/rateLimiting.test.ts index e2aa6c92..91ceda0f 100644 --- a/backend/services/shared/__tests__/rateLimiting.test.ts +++ b/backend/services/shared/__tests__/rateLimiting.test.ts @@ -1,85 +1,56 @@ /** - * Tests for the rate limiting middleware and RateLimitingService. - * - * Run with: - * npx jest backend/services/shared/__tests__/rateLimiting.test.ts + * Tests — RateLimitingService + rateLimitMiddleware (Issue #998) */ import { RateLimitingService } from '../rateLimitingService'; import { createRateLimitMiddleware, - RATE_LIMIT_HEADERS, - type RateLimitRequest, - type RateLimitResponse, + createIpRateLimitMiddleware, + type MinimalRequest, + type MinimalResponse, + type NextFn, } from '../rateLimitMiddleware'; -import { SubscriptionTier } from '../../../../src/types/subscription'; -import { - TIER_RATE_LIMITS, - RateLimitTier, - mapSubscriptionToRateLimitTier, - RATE_LIMIT_TIER_CONFIG, -} from '../../../../src/types/rateLimiting'; +import { SubscriptionTier } from '../../../src/types/subscription'; // --------------------------------------------------------------------------- // Helpers // --------------------------------------------------------------------------- -function makeReq(overrides: Partial = {}): RateLimitRequest { +function makeReq(overrides: Partial = {}): MinimalRequest { return { + headers: { 'x-api-key': 'sk_test_abc123' }, + path: '/api/subscriptions', method: 'GET', - path: '/subscriptions', - url: '/subscriptions', - headers: {}, ...overrides, }; } -interface MockResponse extends RateLimitResponse { - statusCode: number; - headers: Record; - body: unknown; - ended: boolean; -} - -function makeRes(): MockResponse { - const res: MockResponse = { - statusCode: 200, - headers: {}, - body: null, - ended: false, - setHeader(name, value) { - this.headers[name] = value; +function makeRes(): MinimalResponse & { + _status: number; + _headers: Record; + _body: unknown; +} { + const res = { + _status: 200, + _headers: {} as Record, + _body: undefined as unknown, + status(code: number) { + res._status = code; + return res; }, - writeHead(statusCode, headers) { - this.statusCode = statusCode; - if (headers) { - for (const [k, v] of Object.entries(headers)) { - this.headers[k] = v; - } - } + set(header: string, value: string) { + res._headers[header] = value; + return res; }, - end(body) { - this.ended = true; - if (body) this.body = JSON.parse(body); + json(body: unknown) { + res._body = body; }, }; return res; } -function runMiddleware( - mw: ReturnType, - req: RateLimitRequest, - res: MockResponse, -): boolean { - let called = false; - mw(req, res, () => { - called = true; - }); - return called; -} - // --------------------------------------------------------------------------- -// RateLimitingService unit tests +// RateLimitingService // --------------------------------------------------------------------------- describe('RateLimitingService', () => { @@ -89,458 +60,236 @@ describe('RateLimitingService', () => { service = new RateLimitingService(); }); - // ------------------------------------------------------------------------- - describe('checkRateLimit', () => { - it('allows requests within hourly limit', () => { - const result = service.checkRateLimit('key1', SubscriptionTier.FREE); + // ── bypass ──────────────────────────────────────────────────────────────── + + describe('bypass', () => { + it('bypasses a known key', () => { + service.addBypassKey('internal_key'); + const result = service.checkRateLimit('internal_key', SubscriptionTier.FREE); expect(result.allowed).toBe(true); - expect(result.retryAfterMs).toBeUndefined(); }); - it('blocks requests when hourly limit is exceeded', () => { - const tier = SubscriptionTier.FREE; - const limits = TIER_RATE_LIMITS[tier]; - - // Exhaust the hourly counter by recording requests - const usage = service.getOrCreateUsage('key1', tier); - usage.hourly = limits.hourlyLimit; - - const result = service.checkRateLimit('key1', tier); - expect(result.allowed).toBe(false); - expect(result.retryAfterMs).toBeGreaterThan(0); + it('can remove bypass key', () => { + service.addBypassKey('k'); + service.removeBypassKey('k'); + expect(service.isBypassed('k')).toBe(false); }); - it('blocks when daily limit is exceeded', () => { - const tier = SubscriptionTier.FREE; - const limits = TIER_RATE_LIMITS[tier]; - - const usage = service.getOrCreateUsage('key1', tier); - usage.daily = limits.dailyLimit; - - const result = service.checkRateLimit('key1', tier); - expect(result.allowed).toBe(false); + it('bypasses a user', () => { + service.addBypassUser('admin_user'); + expect(service.checkUserRateLimit('user:admin_user', SubscriptionTier.FREE).allowed).toBe(true); }); + }); - it('blocks when burst tokens are exhausted', () => { - const tier = SubscriptionTier.FREE; - service.setBurstTokens('key1', 0, tier); + // ── checkRateLimit ──────────────────────────────────────────────────────── - const result = service.checkRateLimit('key1', tier); - expect(result.allowed).toBe(false); - // ~1s at FREE refill rate (1 token/s); allow tiny elapsed-time drift - expect(result.retryAfterMs).toBeGreaterThan(0); - expect(result.retryAfterMs).toBeLessThanOrEqual(1_000); + describe('checkRateLimit', () => { + it('allows first request for any tier', () => { + expect(service.checkRateLimit('key1', SubscriptionTier.FREE).allowed).toBe(true); + expect(service.checkRateLimit('key2', SubscriptionTier.PREMIUM).allowed).toBe(true); }); - it('blocks when concurrency limit is exceeded', () => { - const tier = SubscriptionTier.FREE; - const limits = TIER_RATE_LIMITS[tier]; - - const usage = service.getOrCreateUsage('key1', tier); - usage.concurrentRequests = limits.concurrentLimit; - - const result = service.checkRateLimit('key1', tier); + it('blocks burst when tokens exhausted', () => { + service.setBurstTokens('key_burst', 0, SubscriptionTier.FREE); + const result = service.checkRateLimit('key_burst', SubscriptionTier.FREE); expect(result.allowed).toBe(false); - expect(result.retryAfterMs).toBe(500); + expect(result.retryAfterMs).toBeGreaterThan(0); }); }); - // ------------------------------------------------------------------------- - describe('bypass management', () => { - it('allows bypassed API keys regardless of limits', () => { - const tier = SubscriptionTier.FREE; - const limits = TIER_RATE_LIMITS[tier]; + // ── recordRequest ───────────────────────────────────────────────────────── - // Exhaust limit - const usage = service.getOrCreateUsage('bypass-key', tier); - usage.hourly = limits.hourlyLimit; - - service.addBypassKey('bypass-key'); - - const result = service.checkRateLimit('bypass-key', tier); - expect(result.allowed).toBe(true); - }); - - it('removes bypass key', () => { - service.addBypassKey('key1'); - expect(service.isBypassed('key1')).toBe(true); - - service.removeBypassKey('key1'); - expect(service.isBypassed('key1')).toBe(false); - }); - - it('allows bypassed users', () => { - service.addBypassUser('user123'); - expect(service.isBypassed('user123', true)).toBe(true); - }); - - it('listBypassKeys returns all bypass keys', () => { - service.addBypassKey('key1'); - service.addBypassKey('key2'); - expect(service.listBypassKeys()).toEqual(expect.arrayContaining(['key1', 'key2'])); + describe('recordRequest', () => { + it('increments usage counters', () => { + service.recordRequest('key_r', SubscriptionTier.FREE, '/api/subs', 200, 50); + service.recordRequest('key_r', SubscriptionTier.FREE, '/api/subs', 200, 40); + const usage = service.getUsage('key_r')!; + expect(usage.hourly).toBe(2); + expect(usage.daily).toBe(2); + expect(usage.monthly).toBe(2); }); - it('listBypassUsers returns all bypass users', () => { - service.addBypassUser('u1'); - service.addBypassUser('u2'); - expect(service.listBypassUsers()).toEqual(expect.arrayContaining(['u1', 'u2'])); + it('emits soft warning near limit', () => { + service.setCustomLimits('key_soft', { hourlyLimit: 10 }); + for (let i = 0; i < 8; i++) { + service.recordRequest('key_soft', SubscriptionTier.FREE, '/x', 200, 10); + } + const { softWarning } = service.recordRequest('key_soft', SubscriptionTier.FREE, '/x', 200, 10); + expect(softWarning).toBeDefined(); + expect(softWarning!.usagePercent).toBeGreaterThanOrEqual(80); }); }); - // ------------------------------------------------------------------------- - describe('custom limits', () => { - it('uses custom hourly limit when set', () => { - service.setCustomLimits('custom-key', { hourlyLimit: 10 }); - const limits = service.getEffectiveLimits('custom-key', SubscriptionTier.FREE); - expect(limits.hourlyLimit).toBe(10); - }); - - it('falls back to tier limits for unset dimensions', () => { - service.setCustomLimits('custom-key', { hourlyLimit: 10 }); - const limits = service.getEffectiveLimits('custom-key', SubscriptionTier.FREE); - expect(limits.dailyLimit).toBe(TIER_RATE_LIMITS[SubscriptionTier.FREE].dailyLimit); - }); + // ── custom limits ───────────────────────────────────────────────────────── - it('clearCustomLimits reverts to tier defaults', () => { - service.setCustomLimits('custom-key', { hourlyLimit: 1 }); - service.clearCustomLimits('custom-key'); - const limits = service.getEffectiveLimits('custom-key', SubscriptionTier.FREE); - expect(limits.hourlyLimit).toBe(TIER_RATE_LIMITS[SubscriptionTier.FREE].hourlyLimit); + describe('custom limits', () => { + it('overrides tier defaults', () => { + service.setCustomLimits('key_c', { hourlyLimit: 5 }); + const limits = service.getEffectiveLimits('key_c', SubscriptionTier.FREE); + expect(limits.hourlyLimit).toBe(5); }); - it('enforces custom limits', () => { - service.setCustomLimits('key1', { hourlyLimit: 2 }); - service.recordRequest('key1', SubscriptionTier.FREE, '/test', 200, 10); - service.recordRequest('key1', SubscriptionTier.FREE, '/test', 200, 10); - - // At exactly the limit — checkRateLimit should block further requests - const usage = service.getOrCreateUsage('key1', SubscriptionTier.FREE); - // 2 requests recorded against hourlyLimit of 2 - expect(usage.hourly).toBe(2); + it('clears custom limits and falls back to tier', () => { + service.setCustomLimits('key_c', { hourlyLimit: 5 }); + service.clearCustomLimits('key_c'); + // Just check no error thrown and defaults are restored + const limits = service.getEffectiveLimits('key_c', SubscriptionTier.FREE); + expect(limits.hourlyLimit).toBeGreaterThan(5); }); }); - // ------------------------------------------------------------------------- - describe('per-user rate limiting', () => { - it('allows requests within user hourly limit', () => { - const result = service.checkUserRateLimit('user:alice', SubscriptionTier.FREE); - expect(result.allowed).toBe(true); - }); - - it('blocks when user hourly limit is exceeded', () => { - const tier = SubscriptionTier.FREE; - const userHourlyLimit = TIER_RATE_LIMITS[tier].hourlyLimit * 5; - - // Exhaust user limit - const usage = (service as unknown as { userUsages: Map }).userUsages; - service.checkUserRateLimit('user:alice', tier); // creates the entry - const entry = usage.get('user:alice')!; - entry.hourly = userHourlyLimit; + // ── per-user limits ─────────────────────────────────────────────────────── - const result = service.checkUserRateLimit('user:alice', tier); - expect(result.allowed).toBe(false); + describe('per-user limits', () => { + it('tracks user usage separately from key usage', () => { + service.recordUserRequest('user:alice', SubscriptionTier.FREE, '/api/subs'); + const status = service.getUserRateLimitStatus('user:alice', SubscriptionTier.FREE); + expect(status.current.hourly).toBe(1); }); - it('bypassed users skip per-user limit check', () => { - const tier = SubscriptionTier.FREE; - service.addBypassUser('alice'); - - const usage = (service as unknown as { userUsages: Map }).userUsages; - service.checkUserRateLimit('user:alice', tier); - const entry = usage.get('user:alice'); - if (entry) entry.hourly = TIER_RATE_LIMITS[tier].hourlyLimit * 5; - - const result = service.checkUserRateLimit('user:alice', tier); + it('allows user if within limit', () => { + const result = service.checkUserRateLimit('user:bob', SubscriptionTier.PREMIUM); expect(result.allowed).toBe(true); }); - - it('getUserRateLimitStatus returns correct multiplied limits', () => { - const tier = SubscriptionTier.FREE; - const status = service.getUserRateLimitStatus('user:bob', tier); - expect(status.limits.hourlyLimit).toBe(TIER_RATE_LIMITS[tier].hourlyLimit * 5); - }); }); - // ------------------------------------------------------------------------- - describe('recordRequest and analytics', () => { - it('increments counters on record', () => { - service.recordRequest('key1', SubscriptionTier.FREE, '/test', 200, 15); - const usage = service.getUsage('key1')!; - expect(usage.hourly).toBe(1); - expect(usage.daily).toBe(1); - expect(usage.monthly).toBe(1); - }); - - it('returns soft warning at 80% usage', () => { - const tier = SubscriptionTier.FREE; - const limits = TIER_RATE_LIMITS[tier]; - const usage = service.getOrCreateUsage('key1', tier); - usage.hourly = Math.floor(limits.hourlyLimit * 0.8); + // ── analytics ───────────────────────────────────────────────────────────── - const { softWarning } = service.recordRequest('key1', tier, '/test', 200, 5); - expect(softWarning).toBeDefined(); - expect(softWarning?.warning).toBe('soft_limit_reached'); + describe('analytics', () => { + it('returns zeroed analytics on fresh service', () => { + const a = service.getAnalytics(); + expect(a.totalRequests).toBe(0); + expect(a.errorRate).toBe(0); }); - it('getAnalytics returns correct totals', () => { - service.recordRequest('k1', SubscriptionTier.FREE, '/a', 200, 10); - service.recordRequest('k1', SubscriptionTier.FREE, '/b', 200, 20); - service.recordRequest('k2', SubscriptionTier.BASIC, '/a', 429, 5); - - const analytics = service.getAnalytics(); - expect(analytics.totalRequests).toBe(3); - expect(analytics.rateLimitHitCount).toBe(1); + it('tracks rate limit hit rate', () => { + service.recordRequest('k', SubscriptionTier.FREE, '/x', 200, 10); + service.recordRequest('k', SubscriptionTier.FREE, '/x', 429, 5); + const a = service.getRateLimitAnalytics(); + expect(a.rateLimitHits).toBe(1); + expect(a.hitRate).toBeCloseTo(0.5); }); - it('getRateLimitAnalytics returns per-tier breakdown', () => { - service.recordRequest('k1', SubscriptionTier.FREE, '/a', 429, 5); - service.recordRequest('k2', SubscriptionTier.BASIC, '/b', 200, 10); - - const rla = service.getRateLimitAnalytics(); - expect(rla.rateLimitHits).toBe(1); - expect(rla.byTier[SubscriptionTier.FREE].hits).toBe(1); - expect(rla.byTier[SubscriptionTier.BASIC].hits).toBe(0); + it('reports top throttled endpoints', () => { + for (let i = 0; i < 3; i++) { + service.recordRequest('k', SubscriptionTier.FREE, '/hot', 429, 5); + } + service.recordRequest('k', SubscriptionTier.FREE, '/cold', 429, 5); + const { topThrottledEndpoints } = service.getRateLimitAnalytics(); + expect(topThrottledEndpoints[0]!.endpoint).toBe('/hot'); + expect(topThrottledEndpoints[0]!.hits).toBe(3); }); }); - // ------------------------------------------------------------------------- - describe('getRateLimitStatus', () => { - it('returns correct remaining values', () => { - const tier = SubscriptionTier.BASIC; - service.recordRequest('key1', tier, '/a', 200, 10); - service.recordRequest('key1', tier, '/a', 200, 10); + // ── tier upgrade ────────────────────────────────────────────────────────── - const status = service.getRateLimitStatus('key1', tier); - expect(status.current.hourly).toBe(2); - expect(status.remaining.hourly).toBe(TIER_RATE_LIMITS[tier].hourlyLimit - 2); - }); - }); - - // ------------------------------------------------------------------------- - describe('free/pro/enterprise rate limit tiers', () => { - it('maps subscription tiers to free/pro/enterprise', () => { - expect(mapSubscriptionToRateLimitTier(SubscriptionTier.FREE)).toBe(RateLimitTier.FREE); - expect(mapSubscriptionToRateLimitTier(SubscriptionTier.BASIC)).toBe(RateLimitTier.PRO); - expect(mapSubscriptionToRateLimitTier(SubscriptionTier.PREMIUM)).toBe(RateLimitTier.PRO); - expect(mapSubscriptionToRateLimitTier(SubscriptionTier.ENTERPRISE)).toBe( - RateLimitTier.ENTERPRISE, - ); - }); - - it('exposes public tier configs via the service', () => { - expect(service.getRateLimitTier(SubscriptionTier.PREMIUM)).toBe(RateLimitTier.PRO); - const pro = service.getPublicTierLimits(RateLimitTier.PRO); - expect(pro.hourlyLimit).toBe(RATE_LIMIT_TIER_CONFIG[RateLimitTier.PRO].hourlyLimit); - expect(pro.refillRatePerSecond).toBeGreaterThan(0); - }); - - it('includes refillRatePerSecond in effective limits', () => { - const limits = service.getEffectiveLimits('key1', SubscriptionTier.FREE); - expect(limits.refillRatePerSecond).toBe(TIER_RATE_LIMITS[SubscriptionTier.FREE].refillRatePerSecond); + describe('tier upgrade recommendation', () => { + it('returns null for unknown key', () => { + expect(service.checkTierUpgrade('no_key')).toBeNull(); }); }); }); // --------------------------------------------------------------------------- -// Rate limit middleware unit tests +// rateLimitMiddleware // --------------------------------------------------------------------------- -describe('createRateLimitMiddleware', () => { +describe('rateLimitMiddleware', () => { let service: RateLimitingService; - let mw: ReturnType; beforeEach(() => { service = new RateLimitingService(); - mw = createRateLimitMiddleware({ service }); }); - // ------------------------------------------------------------------------- - describe('bypass paths', () => { - it('passes /health without rate limiting', () => { - const req = makeReq({ path: '/health' }); - const res = makeRes(); - const next = runMiddleware(mw, req, res); - expect(next).toBe(true); - expect(res.ended).toBe(false); - }); - - it('passes /metrics/plan-cache without rate limiting', () => { - const req = makeReq({ path: '/metrics/plan-cache' }); - const res = makeRes(); - expect(runMiddleware(mw, req, res)).toBe(true); - }); + it('passes request with valid key', async () => { + const middleware = createRateLimitMiddleware({ service, allowMissingKey: false }); + const req = makeReq(); + const res = makeRes(); + const next = jest.fn() as NextFn; - it('can configure custom bypass paths', () => { - const customMw = createRateLimitMiddleware({ - service, - bypassPaths: ['/health', '/internal'], - }); - const req = makeReq({ path: '/internal/jobs' }); - const res = makeRes(); - expect(runMiddleware(customMw, req, res)).toBe(true); - }); - }); - - // ------------------------------------------------------------------------- - describe('bypass keys', () => { - it('allows request from a bypassed API key', () => { - // Exhaust the limit for the key first - const tier = SubscriptionTier.FREE; - const limits = TIER_RATE_LIMITS[tier]; - const usage = service.getOrCreateUsage('trusted-key', tier); - usage.hourly = limits.hourlyLimit; - - const customMw = createRateLimitMiddleware({ - service, - bypassKeys: new Set(['trusted-key']), - tierFn: () => tier, - }); - - const req = makeReq({ headers: { 'x-api-key': 'trusted-key' }, path: '/plans' }); - const res = makeRes(); - expect(runMiddleware(customMw, req, res)).toBe(true); - expect(res.ended).toBe(false); - }); + await middleware(req, res as unknown as MinimalResponse, next); + expect(next).toHaveBeenCalled(); }); - // ------------------------------------------------------------------------- - describe('rate limit enforcement', () => { - it('sets X-RateLimit-* headers on allowed requests', () => { - const req = makeReq({ headers: { 'x-api-key': 'key1' }, path: '/plans' }); - const res = makeRes(); - - runMiddleware(mw, req, res); + it('returns 401 when key is missing', async () => { + const middleware = createRateLimitMiddleware({ service }); + const req = makeReq({ headers: {} }); + const res = makeRes(); + const next = jest.fn() as NextFn; - expect(res.headers[RATE_LIMIT_HEADERS.LIMIT]).toBeDefined(); - expect(res.headers[RATE_LIMIT_HEADERS.REMAINING]).toBeDefined(); - expect(res.headers[RATE_LIMIT_HEADERS.RESET]).toBeDefined(); - }); + await middleware(req, res as unknown as MinimalResponse, next); + expect(res._status).toBe(401); + expect(next).not.toHaveBeenCalled(); + }); - it('returns 429 when hourly limit is exceeded', () => { - const tier = SubscriptionTier.FREE; - const limits = TIER_RATE_LIMITS[tier]; + it('returns 429 when burst tokens exhausted', async () => { + const middleware = createRateLimitMiddleware({ service }); + service.setBurstTokens('sk_test_abc123', 0, SubscriptionTier.FREE); - // Exhaust limit - const usage = service.getOrCreateUsage('exhausted-key', tier); - usage.hourly = limits.hourlyLimit; + const req = makeReq(); + const res = makeRes(); + const next = jest.fn() as NextFn; - const customMw = createRateLimitMiddleware({ - service, - tierFn: () => tier, - }); + await middleware(req, res as unknown as MinimalResponse, next); + expect(res._status).toBe(429); + expect(next).not.toHaveBeenCalled(); + }); - const req = makeReq({ headers: { 'x-api-key': 'exhausted-key' }, path: '/plans' }); - const res = makeRes(); - const next = runMiddleware(customMw, req, res); + it('sets X-RateLimit-* headers on allowed requests', async () => { + const middleware = createRateLimitMiddleware({ service }); + const req = makeReq(); + const res = makeRes(); + const next = jest.fn() as NextFn; + + await middleware(req, res as unknown as MinimalResponse, next); + expect(res._headers['X-RateLimit-Limit']).toBeDefined(); + expect(res._headers['X-RateLimit-Remaining']).toBeDefined(); + expect(res._headers['X-RateLimit-Reset']).toBeDefined(); + expect(res._headers['X-RateLimit-Burst-Remaining']).toBeDefined(); + }); - expect(next).toBe(false); - expect(res.ended).toBe(true); - expect(res.statusCode).toBe(429); - expect(res.headers[RATE_LIMIT_HEADERS.RETRY_AFTER]).toBeDefined(); - expect(res.headers[RATE_LIMIT_HEADERS.REMAINING]).toBe(0); + it('skips rate limiting for configured bypass paths', async () => { + const middleware = createRateLimitMiddleware({ + service, + skipPaths: ['/health'], }); + const req = makeReq({ path: '/health', headers: {} }); + const res = makeRes(); + const next = jest.fn() as NextFn; - it('does not block in softMode even when limit exceeded', () => { - const tier = SubscriptionTier.FREE; - const limits = TIER_RATE_LIMITS[tier]; - const usage = service.getOrCreateUsage('soft-key', tier); - usage.hourly = limits.hourlyLimit; - - const softMw = createRateLimitMiddleware({ - service, - tierFn: () => tier, - softMode: true, - }); - - const req = makeReq({ headers: { 'x-api-key': 'soft-key' }, path: '/plans' }); - const res = makeRes(); - const next = runMiddleware(softMw, req, res); - - expect(next).toBe(true); - expect(res.ended).toBe(false); - }); + await middleware(req, res as unknown as MinimalResponse, next); + expect(next).toHaveBeenCalled(); + expect(res._status).toBe(200); // not set to 401 }); - // ------------------------------------------------------------------------- - describe('per-user headers', () => { - it('sets X-UserRateLimit-* headers when user ID provided', () => { - const req = makeReq({ - headers: { 'x-api-key': 'key1', 'x-user-id': 'user123' }, - path: '/plans', - }); - const res = makeRes(); - - runMiddleware(mw, req, res); - - expect(res.headers[RATE_LIMIT_HEADERS.USER_LIMIT]).toBeDefined(); - expect(res.headers[RATE_LIMIT_HEADERS.USER_REMAINING]).toBeDefined(); - expect(res.headers[RATE_LIMIT_HEADERS.USER_RESET]).toBeDefined(); + it('uses custom rate limit exceeded body', async () => { + const middleware = createRateLimitMiddleware({ + service, + rateLimitExceededBody: (ms) => ({ custom: true, retryAfterMs: ms }), }); + service.setBurstTokens('sk_test_abc123', 0, SubscriptionTier.FREE); - it('blocks when per-user limit is exceeded even if per-key is ok', () => { - const tier = SubscriptionTier.FREE; - const userHourlyLimit = TIER_RATE_LIMITS[tier].hourlyLimit * 5; - - // Exhaust user limit - service.checkUserRateLimit('user:u1', tier); // init - const userUsages = (service as unknown as { userUsages: Map }).userUsages; - const entry = userUsages.get('user:u1')!; - entry.hourly = userHourlyLimit; - - const customMw = createRateLimitMiddleware({ service, tierFn: () => tier }); - const req = makeReq({ - headers: { 'x-api-key': 'key1', 'x-user-id': 'u1' }, - path: '/plans', - }); - const res = makeRes(); - const next = runMiddleware(customMw, req, res); - - expect(next).toBe(false); - expect(res.statusCode).toBe(429); - }); - }); + const req = makeReq(); + const res = makeRes(); + const next = jest.fn() as NextFn; - // ------------------------------------------------------------------------- - describe('IP fallback', () => { - it('uses IP when no API key or user ID', () => { - const req = makeReq({ path: '/plans', ip: '127.0.0.1', headers: {} }); - const res = makeRes(); - const next = runMiddleware(mw, req, res); - - // Should proceed (IP is well within FREE tier limits) - expect(next).toBe(true); - }); + await middleware(req, res as unknown as MinimalResponse, next); + expect((res._body as Record)['custom']).toBe(true); }); - // ------------------------------------------------------------------------- - describe('key extraction', () => { - it('extracts API key from x-api-key header', () => { - const req = makeReq({ headers: { 'x-api-key': 'sk_test_abc' }, path: '/plans' }); - const res = makeRes(); - runMiddleware(mw, req, res); - // If parsed correctly the key usage is tracked - expect(service.getUsage('sk_test_abc')).toBeDefined(); + it('IP rate limit middleware extracts IP as key', async () => { + const middleware = createIpRateLimitMiddleware({ service }); + const req = makeReq({ + headers: {}, + ip: '127.0.0.1', }); + const res = makeRes(); + const next = jest.fn() as NextFn; - it('extracts API key from Authorization Bearer header', () => { - const req = makeReq({ - headers: { authorization: 'Bearer sk_bearer_xyz' }, - path: '/plans', - }); - const res = makeRes(); - runMiddleware(mw, req, res); - expect(service.getUsage('sk_bearer_xyz')).toBeDefined(); - }); - - it('can use a custom keyFn', () => { - const customMw = createRateLimitMiddleware({ - service, - keyFn: (r) => (r.headers['x-custom-key'] as string) || undefined, - }); - const req = makeReq({ headers: { 'x-custom-key': 'custom123' }, path: '/plans' }); - const res = makeRes(); - runMiddleware(customMw, req, res); - expect(service.getUsage('custom123')).toBeDefined(); - }); + await middleware(req, res as unknown as MinimalResponse, next); + expect(next).toHaveBeenCalled(); + // Usage should exist under ip:127.0.0.1 + const usage = service.getUsage('ip:127.0.0.1'); + expect(usage).toBeDefined(); }); }); diff --git a/backend/services/shared/__tests__/tokenBucket.test.ts b/backend/services/shared/__tests__/tokenBucket.test.ts index 60ea7177..dd2c1113 100644 --- a/backend/services/shared/__tests__/tokenBucket.test.ts +++ b/backend/services/shared/__tests__/tokenBucket.test.ts @@ -1,83 +1,120 @@ /** - * Tests for the TokenBucket rate limiter. - * - * Run with: - * npx jest backend/services/shared/__tests__/tokenBucket.test.ts + * Tests — TokenBucket (Issue #998) */ import { TokenBucket, refillRateFromHourlyLimit } from '../tokenBucket'; describe('TokenBucket', () => { + let mockNow: jest.Mock; + + beforeEach(() => { + mockNow = jest.fn(() => 1_000_000); + }); + + // ── Construction ────────────────────────────────────────────────────────── + it('starts full by default', () => { - const bucket = new TokenBucket({ capacity: 10, refillRatePerSecond: 1 }); - expect(bucket.getRemaining()).toBe(10); - expect(bucket.getCapacity()).toBe(10); + const b = new TokenBucket({ capacity: 10, refillRatePerSecond: 1 }, { now: mockNow }); + expect(b.getRemaining()).toBe(10); + }); + + it('starts with custom initialTokens', () => { + const b = new TokenBucket({ capacity: 10, refillRatePerSecond: 1 }, { now: mockNow, initialTokens: 3 }); + expect(b.getRemaining()).toBe(3); }); - it('consumes tokens when available', () => { - const bucket = new TokenBucket({ capacity: 5, refillRatePerSecond: 1 }); - const result = bucket.tryConsume(2); - expect(result.allowed).toBe(true); - expect(result.remaining).toBe(3); - expect(result.retryAfterMs).toBe(0); + it('throws on zero capacity', () => { + expect(() => new TokenBucket({ capacity: 0, refillRatePerSecond: 1 })).toThrow(); }); - it('rejects when empty and reports retryAfterMs', () => { - const bucket = new TokenBucket( - { capacity: 2, refillRatePerSecond: 1 }, - { initialTokens: 0 }, - ); - const result = bucket.tryConsume(1); - expect(result.allowed).toBe(false); - expect(result.remaining).toBe(0); - expect(result.retryAfterMs).toBe(1_000); + it('throws on zero refillRate', () => { + expect(() => new TokenBucket({ capacity: 10, refillRatePerSecond: 0 })).toThrow(); }); - it('refills continuously based on elapsed time', () => { - let now = 1_000_000; - const bucket = new TokenBucket( - { capacity: 10, refillRatePerSecond: 2 }, - { now: () => now, initialTokens: 0 }, - ); + // ── tryConsume ──────────────────────────────────────────────────────────── - now += 2_500; // 2.5s * 2 tokens/s = 5 tokens - expect(bucket.getRemaining()).toBe(5); + it('allows consume when tokens available', () => { + const b = new TokenBucket({ capacity: 5, refillRatePerSecond: 1 }, { now: mockNow }); + const r = b.tryConsume(1); + expect(r.allowed).toBe(true); + expect(r.remaining).toBeCloseTo(4); + expect(r.retryAfterMs).toBe(0); + }); - const result = bucket.tryConsume(5); - expect(result.allowed).toBe(true); - expect(result.remaining).toBe(0); + it('rejects consume when bucket empty and returns retryAfterMs', () => { + const b = new TokenBucket({ capacity: 5, refillRatePerSecond: 2 }, { now: mockNow, initialTokens: 0 }); + const r = b.tryConsume(1); + expect(r.allowed).toBe(false); + expect(r.retryAfterMs).toBe(500); // 1 token at 2/s = 0.5s = 500ms }); - it('never exceeds capacity when refilling', () => { - let now = 0; - const bucket = new TokenBucket( - { capacity: 3, refillRatePerSecond: 100 }, - { now: () => now, initialTokens: 0 }, - ); - now += 10_000; - expect(bucket.getRemaining()).toBe(3); + it('allows burst up to capacity', () => { + const b = new TokenBucket({ capacity: 10, refillRatePerSecond: 1 }, { now: mockNow }); + for (let i = 0; i < 10; i++) { + expect(b.tryConsume(1).allowed).toBe(true); + } + expect(b.tryConsume(1).allowed).toBe(false); }); - it('reconfigure updates capacity and clamps tokens', () => { - const bucket = new TokenBucket({ capacity: 10, refillRatePerSecond: 1 }); - bucket.reconfigure({ capacity: 4 }); - expect(bucket.getCapacity()).toBe(4); - expect(bucket.getRemaining()).toBe(4); + // ── Refill over time ────────────────────────────────────────────────────── + + it('refills tokens after elapsed time', () => { + const b = new TokenBucket({ capacity: 10, refillRatePerSecond: 2 }, { now: mockNow, initialTokens: 0 }); + // Advance 3 seconds → should add 6 tokens + mockNow.mockReturnValue(1_003_000); + expect(b.getRemaining()).toBe(6); }); - it('throws on invalid config', () => { - expect(() => new TokenBucket({ capacity: 0, refillRatePerSecond: 1 })).toThrow(); - expect(() => new TokenBucket({ capacity: 5, refillRatePerSecond: 0 })).toThrow(); + it('does not exceed capacity when refilling', () => { + const b = new TokenBucket({ capacity: 5, refillRatePerSecond: 10 }, { now: mockNow }); + mockNow.mockReturnValue(1_100_000); // 100s — would add 1000 tokens + b.refill(); + expect(b.getRemaining()).toBe(5); }); -}); -describe('refillRateFromHourlyLimit', () => { - it('derives tokens/sec from hourly limit', () => { - expect(refillRateFromHourlyLimit(3_600)).toBe(1); - expect(refillRateFromHourlyLimit(100)).toBeCloseTo(100 / 3_600); + // ── reconfigure ─────────────────────────────────────────────────────────── + + it('reconfigure adjusts capacity and rate', () => { + const b = new TokenBucket({ capacity: 10, refillRatePerSecond: 1 }, { now: mockNow }); + b.reconfigure({ capacity: 20, refillRatePerSecond: 5 }); + expect(b.getCapacity()).toBe(20); + expect(b.getRefillRatePerSecond()).toBe(5); + }); + + it('reconfigure clamps tokens to new capacity', () => { + const b = new TokenBucket({ capacity: 10, refillRatePerSecond: 1 }, { now: mockNow }); + b.reconfigure({ capacity: 3 }); + expect(b.getRemaining()).toBe(3); + }); + + // ── setTokens ───────────────────────────────────────────────────────────── + + it('setTokens clamps to [0, capacity]', () => { + const b = new TokenBucket({ capacity: 10, refillRatePerSecond: 1 }, { now: mockNow }); + b.setTokens(50); + expect(b.getRemaining()).toBe(10); + b.setTokens(-5); + expect(b.getRemaining()).toBe(0); + }); + + // ── snapshot ────────────────────────────────────────────────────────────── + + it('snapshot returns correct fields', () => { + const b = new TokenBucket({ capacity: 5, refillRatePerSecond: 2 }, { now: mockNow, initialTokens: 3 }); + const snap = b.snapshot(); + expect(snap.capacity).toBe(5); + expect(snap.refillRatePerSecond).toBe(2); + expect(snap.tokens).toBeCloseTo(3); + }); + + // ── refillRateFromHourlyLimit ───────────────────────────────────────────── + + it('refillRateFromHourlyLimit converts hourly to per-second', () => { + expect(refillRateFromHourlyLimit(3600)).toBeCloseTo(1); + expect(refillRateFromHourlyLimit(720)).toBeCloseTo(0.2); }); - it('floors at a small epsilon', () => { + it('refillRateFromHourlyLimit floors at epsilon for very small limits', () => { expect(refillRateFromHourlyLimit(0)).toBe(0.001); }); }); diff --git a/backend/services/shared/apiKeyRotation.ts b/backend/services/shared/apiKeyRotation.ts new file mode 100644 index 00000000..3e75cab7 --- /dev/null +++ b/backend/services/shared/apiKeyRotation.ts @@ -0,0 +1,462 @@ +/** + * API Key Rotation Service — SubTrackr + * + * Issue #1009: Implement API key rotation with grace period + * + * Features: + * - Generate new API key while keeping the old one valid for a configurable grace period + * - Dual-key acceptance window: requests authenticated with the old key continue to work + * until the grace period expires + * - Automatic expiry + cleanup of keys past their grace deadline + * - Per-key rotation history (audit trail) + * - Rotation event emission via the domain EventBus + * - Prometheus metrics export + */ + +import { randomBytes, createHmac } from 'crypto'; +import { eventBus, buildEvent } from './events'; + +// --------------------------------------------------------------------------- +// Constants +// --------------------------------------------------------------------------- + +const DEFAULT_GRACE_PERIOD_MS = 24 * 60 * 60 * 1_000; // 24 h +const DEFAULT_KEY_TTL_MS = 365 * 24 * 60 * 60 * 1_000; // 1 year +const KEY_PREFIX_TEST = 'sk_test_'; +const KEY_PREFIX_LIVE = 'sk_live_'; +const HMAC_ALG = 'sha256'; + +// --------------------------------------------------------------------------- +// Types +// --------------------------------------------------------------------------- + +export type ApiKeyEnvironment = 'test' | 'production'; + +export type ApiKeyStatus = 'active' | 'grace' | 'expired' | 'revoked'; + +export interface ManagedApiKey { + /** Opaque record identifier (not the secret value). */ + readonly id: string; + /** The actual secret key string sent in Authorization headers. */ + readonly key: string; + readonly name: string; + readonly developerId: string; + readonly environment: ApiKeyEnvironment; + readonly permissions: string[]; + status: ApiKeyStatus; + readonly createdAt: number; + expiresAt: number; + /** Set when rotation starts — old key stays valid until this timestamp. */ + gracePeriodEndsAt?: number; + /** ID of the successor key created by rotation (if any). */ + replacedByKeyId?: string; + /** ID of the predecessor key this one replaced (if any). */ + replacesKeyId?: string; + revokedAt?: number; + lastUsedAt?: number; + usageCount: number; +} + +export interface RotationOptions { + /** Grace period in milliseconds (default: 24 h). */ + gracePeriodMs?: number; + /** TTL for the new key in milliseconds (default: 1 year). */ + newKeyTtlMs?: number; + /** Human-readable reason stored in the rotation record. */ + reason?: string; +} + +export interface RotationResult { + /** The newly created key. */ + newKey: ManagedApiKey; + /** The old key (now in grace period). */ + oldKey: ManagedApiKey; + /** When the old key will stop being accepted. */ + gracePeriodEndsAt: number; + /** Milliseconds until old key expires. */ + gracePeriodRemainingMs: number; +} + +export interface RotationRecord { + readonly id: string; + readonly developerId: string; + readonly oldKeyId: string; + readonly newKeyId: string; + readonly rotatedAt: number; + readonly gracePeriodEndsAt: number; + readonly reason?: string; +} + +export interface ApiKeyValidationResult { + valid: boolean; + key?: ManagedApiKey; + /** True if the key is in grace period (warn callers to update). */ + isGrace: boolean; + /** Milliseconds remaining in grace period (0 when not in grace). */ + graceRemainingMs: number; + reason?: string; +} + +export interface ApiKeyRotationMetrics { + totalKeys: number; + activeKeys: number; + graceKeys: number; + expiredKeys: number; + revokedKeys: number; + totalRotations: number; + totalValidations: number; + gracePeriodHits: number; +} + +// --------------------------------------------------------------------------- +// Service +// --------------------------------------------------------------------------- + +export class ApiKeyRotationService { + /** All managed keys, keyed by key ID. */ + private readonly keys = new Map(); + /** Fast lookup: secret key string → key ID. */ + private readonly keyIndex = new Map(); + /** Rotation audit records. */ + private readonly rotationHistory: RotationRecord[] = []; + + private totalValidations = 0; + private gracePeriodHits = 0; + + // ── Key creation ────────────────────────────────────────────────────────── + + /** + * Create a fresh API key (no rotation — first key for a developer). + */ + createKey( + developerId: string, + name: string, + environment: ApiKeyEnvironment, + permissions: string[] = [], + ttlMs = DEFAULT_KEY_TTL_MS, + ): ManagedApiKey { + const id = this.generateId('key'); + const secret = this.generateSecret(environment); + const now = Date.now(); + + const key: ManagedApiKey = { + id, + key: secret, + name, + developerId, + environment, + permissions: [...permissions], + status: 'active', + createdAt: now, + expiresAt: now + ttlMs, + usageCount: 0, + }; + + this.keys.set(id, key); + this.keyIndex.set(secret, id); + return key; + } + + // ── Rotation ────────────────────────────────────────────────────────────── + + /** + * Rotate an API key. + * + * 1. Creates a new key with the same settings. + * 2. Marks the old key as `grace` and sets `gracePeriodEndsAt`. + * 3. Both keys are accepted by `validateKey()` during the grace window. + * 4. Emits an `auth.api_key_rotated` domain event. + */ + async rotateKey(keyId: string, options: RotationOptions = {}): Promise { + const oldKey = this.keys.get(keyId); + if (!oldKey) { + throw new Error(`API key not found: ${keyId}`); + } + if (oldKey.status === 'revoked') { + throw new Error(`Cannot rotate a revoked key: ${keyId}`); + } + if (oldKey.status === 'expired') { + throw new Error(`Cannot rotate an expired key: ${keyId}`); + } + + const gracePeriodMs = options.gracePeriodMs ?? DEFAULT_GRACE_PERIOD_MS; + const newKeyTtlMs = options.newKeyTtlMs ?? DEFAULT_KEY_TTL_MS; + const now = Date.now(); + const gracePeriodEndsAt = now + gracePeriodMs; + + // Create replacement key + const newKeySecret = this.generateSecret(oldKey.environment); + const newId = this.generateId('key'); + const newKey: ManagedApiKey = { + id: newId, + key: newKeySecret, + name: oldKey.name, + developerId: oldKey.developerId, + environment: oldKey.environment, + permissions: [...oldKey.permissions], + status: 'active', + createdAt: now, + expiresAt: now + newKeyTtlMs, + replacesKeyId: oldKey.id, + usageCount: 0, + }; + + // Transition old key to grace period + oldKey.status = 'grace'; + oldKey.gracePeriodEndsAt = gracePeriodEndsAt; + oldKey.replacedByKeyId = newId; + + this.keys.set(newId, newKey); + this.keyIndex.set(newKeySecret, newId); + + // Rotation audit record + const record: RotationRecord = { + id: this.generateId('rot'), + developerId: oldKey.developerId, + oldKeyId: oldKey.id, + newKeyId: newId, + rotatedAt: now, + gracePeriodEndsAt, + reason: options.reason, + }; + this.rotationHistory.push(record); + + // Domain event + await eventBus.publish( + buildEvent( + 'auth', + 'api_key_rotated', + { + keyId: newId, + merchantId: oldKey.developerId, + rotatedAt: now, + expiresAt: gracePeriodEndsAt, + }, + { aggregateId: oldKey.developerId, correlationId: record.id }, + ), + ); + + return { + newKey, + oldKey, + gracePeriodEndsAt, + gracePeriodRemainingMs: gracePeriodMs, + }; + } + + // ── Validation ──────────────────────────────────────────────────────────── + + /** + * Validate an API key string. + * + * - Expired grace-period keys are automatically transitioned to `expired`. + * - Returns `isGrace: true` with the remaining grace window when the old + * key is used after rotation so callers can surface a deprecation warning. + */ + validateKey(secret: string): ApiKeyValidationResult { + this.totalValidations++; + + const id = this.keyIndex.get(secret); + if (!id) { + return { valid: false, isGrace: false, graceRemainingMs: 0, reason: 'key_not_found' }; + } + + const key = this.keys.get(id)!; + const now = Date.now(); + + // Auto-expire grace keys whose window has closed + if (key.status === 'grace' && key.gracePeriodEndsAt && now > key.gracePeriodEndsAt) { + key.status = 'expired'; + } + + if (key.status === 'revoked') { + return { valid: false, key, isGrace: false, graceRemainingMs: 0, reason: 'revoked' }; + } + if (key.status === 'expired') { + return { valid: false, key, isGrace: false, graceRemainingMs: 0, reason: 'expired' }; + } + if (now > key.expiresAt && key.status === 'active') { + key.status = 'expired'; + return { valid: false, key, isGrace: false, graceRemainingMs: 0, reason: 'expired' }; + } + + // Record usage + key.lastUsedAt = now; + key.usageCount++; + + if (key.status === 'grace') { + const graceRemainingMs = Math.max(0, (key.gracePeriodEndsAt ?? now) - now); + this.gracePeriodHits++; + return { valid: true, key, isGrace: true, graceRemainingMs }; + } + + return { valid: true, key, isGrace: false, graceRemainingMs: 0 }; + } + + // ── Revocation ──────────────────────────────────────────────────────────── + + /** + * Immediately revoke a key (no grace period). + */ + revokeKey(keyId: string): boolean { + const key = this.keys.get(keyId); + if (!key || key.status === 'revoked') return false; + key.status = 'revoked'; + key.revokedAt = Date.now(); + return true; + } + + /** + * Revoke all keys for a developer (e.g., account suspension). + */ + revokeAllKeys(developerId: string): number { + let count = 0; + for (const key of this.keys.values()) { + if (key.developerId === developerId && key.status !== 'revoked') { + key.status = 'revoked'; + key.revokedAt = Date.now(); + count++; + } + } + return count; + } + + // ── Cleanup ─────────────────────────────────────────────────────────────── + + /** + * Expire any grace-period or active keys whose time has passed. + * Returns the count of keys transitioned to `expired`. + */ + cleanupExpiredKeys(): number { + const now = Date.now(); + let count = 0; + for (const key of this.keys.values()) { + if ( + key.status === 'grace' && + key.gracePeriodEndsAt && + now > key.gracePeriodEndsAt + ) { + key.status = 'expired'; + count++; + } else if (key.status === 'active' && now > key.expiresAt) { + key.status = 'expired'; + count++; + } + } + return count; + } + + // ── Queries ─────────────────────────────────────────────────────────────── + + getKey(keyId: string): ManagedApiKey | undefined { + return this.keys.get(keyId); + } + + getKeysByDeveloper(developerId: string): ManagedApiKey[] { + return Array.from(this.keys.values()).filter((k) => k.developerId === developerId); + } + + getActiveKeysByDeveloper(developerId: string): ManagedApiKey[] { + return this.getKeysByDeveloper(developerId).filter( + (k) => k.status === 'active' || k.status === 'grace', + ); + } + + getRotationHistory(developerId?: string): RotationRecord[] { + if (!developerId) return [...this.rotationHistory]; + return this.rotationHistory.filter((r) => r.developerId === developerId); + } + + /** + * Returns the grace period status for a key currently in rotation. + * Returns `null` if the key is not in grace period. + */ + getGracePeriodStatus( + keyId: string, + ): { gracePeriodEndsAt: number; remainingMs: number; successorKeyId: string } | null { + const key = this.keys.get(keyId); + if (!key || key.status !== 'grace' || !key.gracePeriodEndsAt) return null; + const now = Date.now(); + if (now > key.gracePeriodEndsAt) { + key.status = 'expired'; + return null; + } + return { + gracePeriodEndsAt: key.gracePeriodEndsAt, + remainingMs: key.gracePeriodEndsAt - now, + successorKeyId: key.replacedByKeyId ?? '', + }; + } + + // ── Metrics ─────────────────────────────────────────────────────────────── + + getMetrics(): ApiKeyRotationMetrics { + let activeKeys = 0; + let graceKeys = 0; + let expiredKeys = 0; + let revokedKeys = 0; + for (const key of this.keys.values()) { + if (key.status === 'active') activeKeys++; + else if (key.status === 'grace') graceKeys++; + else if (key.status === 'expired') expiredKeys++; + else if (key.status === 'revoked') revokedKeys++; + } + return { + totalKeys: this.keys.size, + activeKeys, + graceKeys, + expiredKeys, + revokedKeys, + totalRotations: this.rotationHistory.length, + totalValidations: this.totalValidations, + gracePeriodHits: this.gracePeriodHits, + }; + } + + prometheusMetrics(namespace = 'subtrackr_api_key'): string { + const m = this.getMetrics(); + return [ + `# HELP ${namespace}_total Total managed API keys`, + `# TYPE ${namespace}_total gauge`, + `${namespace}_total ${m.totalKeys}`, + `# HELP ${namespace}_active_total Active (non-expired, non-revoked) keys`, + `# TYPE ${namespace}_active_total gauge`, + `${namespace}_active_total ${m.activeKeys}`, + `# HELP ${namespace}_grace_total Keys currently in grace period`, + `# TYPE ${namespace}_grace_total gauge`, + `${namespace}_grace_total ${m.graceKeys}`, + `# HELP ${namespace}_expired_total Expired keys`, + `# TYPE ${namespace}_expired_total gauge`, + `${namespace}_expired_total ${m.expiredKeys}`, + `# HELP ${namespace}_revoked_total Revoked keys`, + `# TYPE ${namespace}_revoked_total gauge`, + `${namespace}_revoked_total ${m.revokedKeys}`, + `# HELP ${namespace}_rotations_total Total key rotations performed`, + `# TYPE ${namespace}_rotations_total counter`, + `${namespace}_rotations_total ${m.totalRotations}`, + `# HELP ${namespace}_validations_total Total key validation attempts`, + `# TYPE ${namespace}_validations_total counter`, + `${namespace}_validations_total ${m.totalValidations}`, + `# HELP ${namespace}_grace_hits_total Validations that succeeded via grace period`, + `# TYPE ${namespace}_grace_hits_total counter`, + `${namespace}_grace_hits_total ${m.gracePeriodHits}`, + ].join('\n'); + } + + // ── Private helpers ─────────────────────────────────────────────────────── + + private generateId(prefix: string): string { + return `${prefix}_${Date.now().toString(36)}_${randomBytes(6).toString('hex')}`; + } + + private generateSecret(environment: ApiKeyEnvironment): string { + const prefix = environment === 'test' ? KEY_PREFIX_TEST : KEY_PREFIX_LIVE; + return prefix + randomBytes(24).toString('base64url'); + } +} + +// --------------------------------------------------------------------------- +// Singleton +// --------------------------------------------------------------------------- + +export const apiKeyRotationService = new ApiKeyRotationService(); diff --git a/backend/services/shared/index.ts b/backend/services/shared/index.ts index 19df0818..29942897 100644 --- a/backend/services/shared/index.ts +++ b/backend/services/shared/index.ts @@ -185,3 +185,20 @@ export type { WsPoolConfig, WsConnection, WsMessage, WsPoolMetrics } from './wsC // ── Read Replica Router (#997) ──────────────────────────────────────────────── export { ReadReplicaRouter } from './readReplicaRouter'; export type { ReplicaConfig, ReplicaHealth, ReadRouteOptions, QueryRoute } from './readReplicaRouter'; + +// ── Rate Limiting Middleware (#998) ────────────────────────────────────────── +export { createRateLimitMiddleware, createFastifyRateLimitHook, createIpRateLimitMiddleware } from './rateLimitMiddleware'; +export type { RateLimitMiddlewareOptions, MinimalRequest, MinimalResponse } from './rateLimitMiddleware'; + +// ── API Key Rotation with Grace Period (#1009) ──────────────────────────────── +export { ApiKeyRotationService, apiKeyRotationService } from './apiKeyRotation'; +export type { + ManagedApiKey, + ApiKeyEnvironment, + ApiKeyStatus, + RotationOptions, + RotationResult, + RotationRecord, + ApiKeyValidationResult, + ApiKeyRotationMetrics, +} from './apiKeyRotation'; diff --git a/backend/services/shared/rateLimitMiddleware.ts b/backend/services/shared/rateLimitMiddleware.ts index 9649586a..98820219 100644 --- a/backend/services/shared/rateLimitMiddleware.ts +++ b/backend/services/shared/rateLimitMiddleware.ts @@ -1,360 +1,283 @@ /** - * Rate limiting middleware for SubTrackr API. + * Rate Limit Middleware — SubTrackr * - * Supports: - * - Token-bucket burst limiting with continuous refill - * - Per-API-key rate limiting (hourly / daily / monthly / concurrent) - * - Per-user rate limiting (aggregate across all keys for a user) - * - Standard rate-limit response headers (X-RateLimit-*) - * - Tier-based limits (free / pro / enterprise) - * - Bypass list for trusted clients (service accounts, internal health checks) - * - Configurable limits that can be overridden per-key + * Issue #998: Implement rate limiting with token bucket algorithm * - * Usage: - * import { createRateLimitMiddleware } from './rateLimitMiddleware'; - * const rl = createRateLimitMiddleware({ service: rateLimitingService }); - * // express-style: app.use(rl); + * Express / Fastify-compatible middleware that integrates RateLimitingService + * (token bucket + sliding-window counters) with standard HTTP headers. + * + * Usage (Express): + * app.use(createRateLimitMiddleware({ service: rateLimitingService })); + * + * Usage (Fastify): + * fastify.addHook('preHandler', createFastifyRateLimitHook({ service: rateLimitingService })); */ -import { SubscriptionTier } from '../../src/types/subscription'; import { RateLimitingService } from './rateLimitingService'; -import { TIER_RATE_LIMITS, mapSubscriptionToRateLimitTier } from '../../src/types/rateLimiting'; - -// --------------------------------------------------------------------------- -// Minimal request / response types — structural, no Express dep required -// --------------------------------------------------------------------------- - -export interface RateLimitRequest { - method?: string; - path?: string; - url?: string; - headers: Record; - socket?: { remoteAddress?: string }; - ip?: string; -} - -export interface RateLimitResponse { - setHeader(name: string, value: string | number): void; - writeHead?(status: number, headers?: Record): void; - status?(code: number): RateLimitResponse; - end?(body?: string): void; - json?(body: unknown): void; -} - -export type NextFn = (err?: unknown) => void; +import { SubscriptionTier } from '../../src/types/subscription'; // --------------------------------------------------------------------------- -// Config +// Shared types // --------------------------------------------------------------------------- export interface RateLimitMiddlewareOptions { - /** The rate limiting service instance to use. */ service: RateLimitingService; /** * Extract the API key from the request. - * Defaults to `x-api-key` header, then `Authorization: Bearer `. + * Default: Authorization header (Bearer token) or X-Api-Key header. */ - keyFn?: (req: RateLimitRequest) => string | undefined; + getApiKey?: (req: MinimalRequest) => string | undefined; /** - * Extract the user ID from the request. - * Defaults to `x-user-id` header. + * Extract the subscription tier for the key. + * Default: always FREE (override to integrate with your auth layer). */ - userIdFn?: (req: RateLimitRequest) => string | undefined; + getTier?: (apiKey: string, req: MinimalRequest) => SubscriptionTier | Promise; /** - * Determine the subscription tier for the API key / user. - * Defaults to FREE tier for unknown keys. + * Extract a user ID for per-user aggregate limiting. + * Return undefined to skip per-user limiting. */ - tierFn?: (apiKey: string, userId?: string) => SubscriptionTier; + getUserId?: (req: MinimalRequest) => string | undefined; /** - * Paths that bypass rate limiting entirely (e.g. health checks). - * String match uses exact prefix matching. + * If true, requests with missing or invalid API keys still pass through + * (useful in development). Default: false. */ - bypassPaths?: string[]; + allowMissingKey?: boolean; /** - * API keys that bypass rate limiting (trusted service accounts). + * Custom response body for 429 responses. + * Default: { error: 'rate_limit_exceeded', retryAfterMs, message } */ - bypassKeys?: Set; + rateLimitExceededBody?: (retryAfterMs: number) => Record; /** - * User IDs that bypass rate limiting (internal accounts). + * Skip rate limiting entirely for these path prefixes. + * Merged with service.bypass.paths. */ - bypassUsers?: Set; - /** When true, only add headers — do not reject requests. Default: false. */ - softMode?: boolean; + skipPaths?: string[]; } -// --------------------------------------------------------------------------- -// Header names -// --------------------------------------------------------------------------- - -export const RATE_LIMIT_HEADERS = { - LIMIT: 'X-RateLimit-Limit', - REMAINING: 'X-RateLimit-Remaining', - RESET: 'X-RateLimit-Reset', - RETRY_AFTER: 'Retry-After', - POLICY: 'X-RateLimit-Policy', - USER_LIMIT: 'X-UserRateLimit-Limit', - USER_REMAINING: 'X-UserRateLimit-Remaining', - USER_RESET: 'X-UserRateLimit-Reset', -} as const; - -// --------------------------------------------------------------------------- -// Helper -// --------------------------------------------------------------------------- - -function getHeader(req: RateLimitRequest, name: string): string { - const v = req.headers[name.toLowerCase()]; - return Array.isArray(v) ? (v[0] ?? '') : (v ?? ''); +/** Minimal interface satisfied by both Express.Request and Fastify.Request */ +export interface MinimalRequest { + headers: Record; + path?: string; + url?: string; + method?: string; + ip?: string; + socket?: { remoteAddress?: string }; } -function defaultKeyFn(req: RateLimitRequest): string | undefined { - const fromHeader = getHeader(req, 'x-api-key'); - if (fromHeader) return fromHeader; +/** Minimal interface satisfied by both Express.Response and Fastify.Reply */ +export interface MinimalResponse { + status(code: number): MinimalResponse; + set?(header: string, value: string): MinimalResponse; + header?(header: string, value: string): MinimalResponse; + json(body: unknown): void; +} - const auth = getHeader(req, 'authorization'); - if (auth.startsWith('Bearer ')) return auth.slice(7); +export type NextFn = (err?: unknown) => void; - return undefined; -} +// --------------------------------------------------------------------------- +// Header helpers +// --------------------------------------------------------------------------- -function defaultUserIdFn(req: RateLimitRequest): string | undefined { - const userId = getHeader(req, 'x-user-id'); - return userId || undefined; +function setHeader(res: MinimalResponse, name: string, value: string): void { + if (typeof res.set === 'function') { + res.set(name, value); + } else if (typeof res.header === 'function') { + res.header(name, value); + } } -function defaultTierFn(_apiKey: string, _userId?: string): SubscriptionTier { - return SubscriptionTier.FREE; +function defaultGetApiKey(req: MinimalRequest): string | undefined { + const auth = req.headers['authorization']; + if (typeof auth === 'string' && auth.startsWith('Bearer ')) { + return auth.slice(7).trim(); + } + const xKey = req.headers['x-api-key']; + return typeof xKey === 'string' ? xKey.trim() : undefined; } -function sendRateLimitExceeded( - res: RateLimitResponse, - retryAfterMs: number, - limit: number, - resetAt: number, -): void { - const retryAfterSecs = Math.ceil(retryAfterMs / 1_000); - - if (res.writeHead) { - res.writeHead(429, { - 'Content-Type': 'application/json', - [RATE_LIMIT_HEADERS.RETRY_AFTER]: String(retryAfterSecs), - [RATE_LIMIT_HEADERS.LIMIT]: String(limit), - [RATE_LIMIT_HEADERS.REMAINING]: '0', - [RATE_LIMIT_HEADERS.RESET]: String(Math.ceil(resetAt / 1_000)), - }); - res.end?.( - JSON.stringify({ - status: 429, - error: 'rate_limit_exceeded', - message: `Rate limit exceeded. Retry after ${retryAfterSecs} seconds.`, - retryAfter: retryAfterSecs, - limit, - remaining: 0, - resetAt, - }), - ); - } else if (res.status && res.json) { - res.setHeader(RATE_LIMIT_HEADERS.RETRY_AFTER, retryAfterSecs); - res.setHeader(RATE_LIMIT_HEADERS.REMAINING, 0); - res.status(429).json({ - status: 429, - error: 'rate_limit_exceeded', - message: `Rate limit exceeded. Retry after ${retryAfterSecs} seconds.`, - retryAfter: retryAfterSecs, - limit, - remaining: 0, - resetAt, - }); - } +function getPath(req: MinimalRequest): string { + return req.path ?? (req.url ? req.url.split('?')[0]! : '/'); } // --------------------------------------------------------------------------- -// Factory +// Express middleware factory // --------------------------------------------------------------------------- -/** - * Creates an Express-compatible rate limiting middleware. - * - * @example - * ```ts - * const rl = createRateLimitMiddleware({ - * service: rateLimitingService, - * bypassPaths: ['/health', '/metrics'], - * bypassKeys: new Set(['internal-service-key-abc']), - * }); - * app.use(rl); - * ``` - */ -export function createRateLimitMiddleware(options: RateLimitMiddlewareOptions) { +export function createRateLimitMiddleware(opts: RateLimitMiddlewareOptions) { const { service, - keyFn = defaultKeyFn, - userIdFn = defaultUserIdFn, - tierFn = defaultTierFn, - bypassPaths = ['/health', '/metrics', '/metrics/plan-cache'], - bypassKeys = new Set(), - bypassUsers = new Set(), - softMode = false, - } = options; - - return function rateLimitMiddleware( - req: RateLimitRequest, - res: RateLimitResponse, + getApiKey = defaultGetApiKey, + getTier = () => SubscriptionTier.FREE, + getUserId, + allowMissingKey = false, + rateLimitExceededBody, + skipPaths = [], + } = opts; + + const allSkipPaths = [ + ...(service.bypass.paths ?? []), + ...skipPaths, + ]; + + return async function rateLimitMiddleware( + req: MinimalRequest, + res: MinimalResponse, next: NextFn, - ): void { - const path = req.path ?? req.url ?? ''; + ): Promise { + const path = getPath(req); - // ----------------------------------------------------------------------- - // Bypass: path - // ----------------------------------------------------------------------- - if (bypassPaths.some((bp) => path === bp || path.startsWith(bp))) { + // Skip configured paths + if (allSkipPaths.some((p) => path.startsWith(p))) { next(); return; } - const apiKey = keyFn(req); - const userId = userIdFn(req); + const apiKey = getApiKey(req); - // ----------------------------------------------------------------------- - // Bypass: trusted keys / users - // ----------------------------------------------------------------------- - if (apiKey && bypassKeys.has(apiKey)) { - next(); - return; - } - if (userId && bypassUsers.has(userId)) { - next(); + if (!apiKey) { + if (allowMissingKey) { + next(); + return; + } + res.status(401).json({ error: 'missing_api_key', message: 'API key required' }); return; } - // Need at least one identifier - const identifier = apiKey ?? userId; - if (!identifier) { - // No credentials — use IP as fallback identifier - const ip = req.ip ?? req.socket?.remoteAddress ?? 'anonymous'; - const tier = SubscriptionTier.FREE; - const limits = TIER_RATE_LIMITS[tier]; - const check = service.checkRateLimit(ip, tier); + const tier = await getTier(apiKey, req); + const start = Date.now(); - res.setHeader(RATE_LIMIT_HEADERS.LIMIT, limits.hourlyLimit); - res.setHeader(RATE_LIMIT_HEADERS.POLICY, 'ip-fallback'); + // Per-key check + const keyCheck = service.checkRateLimit(apiKey, tier); - if (!check.allowed && !softMode) { - sendRateLimitExceeded(res, check.retryAfterMs ?? 60_000, limits.hourlyLimit, Date.now() + (check.retryAfterMs ?? 60_000)); - return; - } + if (!keyCheck.allowed) { + const retryAfterMs = keyCheck.retryAfterMs ?? 1_000; + const retryAfterSec = Math.ceil(retryAfterMs / 1_000); - if (check.allowed) { - service.recordRequest(ip, tier, path, 200, 0); - const status = service.getRateLimitStatus(ip, tier); - res.setHeader(RATE_LIMIT_HEADERS.REMAINING, status.remaining.hourly); - res.setHeader(RATE_LIMIT_HEADERS.RESET, Math.ceil(status.resetAt.hourly / 1_000)); - } + setHeader(res, 'Retry-After', String(retryAfterSec)); + setHeader(res, 'X-RateLimit-Retry-After-Ms', String(retryAfterMs)); - next(); + const body = rateLimitExceededBody + ? rateLimitExceededBody(retryAfterMs) + : { + error: 'rate_limit_exceeded', + message: `Rate limit exceeded. Retry after ${retryAfterSec}s.`, + retryAfterMs, + retryAfterSec, + }; + + res.status(429).json(body); return; } - const tier = tierFn(apiKey ?? identifier, userId); - const limits = TIER_RATE_LIMITS[tier]; - const publicTier = mapSubscriptionToRateLimitTier(tier); - - // ----------------------------------------------------------------------- - // Per-API-key check - // ----------------------------------------------------------------------- - if (apiKey) { - const check = service.checkRateLimit(apiKey, tier); - const status = service.getRateLimitStatus(apiKey, tier); - - res.setHeader(RATE_LIMIT_HEADERS.LIMIT, limits.hourlyLimit); - res.setHeader(RATE_LIMIT_HEADERS.REMAINING, status.remaining.hourly); - res.setHeader(RATE_LIMIT_HEADERS.RESET, Math.ceil(status.resetAt.hourly / 1_000)); - res.setHeader( - RATE_LIMIT_HEADERS.POLICY, - `${publicTier};hourly=${limits.hourlyLimit};daily=${limits.dailyLimit};burst=${limits.burstLimit};refill=${limits.refillRatePerSecond}/s`, - ); - - if (!check.allowed && !softMode) { - sendRateLimitExceeded( - res, - check.retryAfterMs ?? 60_000, - limits.hourlyLimit, - status.resetAt.hourly, - ); - return; + // Per-user aggregate check + if (getUserId) { + const userId = getUserId(req); + if (userId) { + const userCheck = service.checkUserRateLimit(`user:${userId}`, tier); + if (!userCheck.allowed) { + const retryAfterMs = userCheck.retryAfterMs ?? 1_000; + const retryAfterSec = Math.ceil(retryAfterMs / 1_000); + + setHeader(res, 'Retry-After', String(retryAfterSec)); + setHeader(res, 'X-RateLimit-Retry-After-Ms', String(retryAfterMs)); + + res.status(429).json( + rateLimitExceededBody + ? rateLimitExceededBody(retryAfterMs) + : { + error: 'user_rate_limit_exceeded', + message: `User-level rate limit exceeded. Retry after ${retryAfterSec}s.`, + retryAfterMs, + }, + ); + return; + } } } - // ----------------------------------------------------------------------- - // Per-user check (aggregate) - // ----------------------------------------------------------------------- - if (userId) { - const userKey = `user:${userId}`; - const userCheck = service.checkUserRateLimit(userKey, tier); - const userStatus = service.getUserRateLimitStatus(userKey, tier); - - res.setHeader(RATE_LIMIT_HEADERS.USER_LIMIT, limits.hourlyLimit * 5); // users get 5x key limit - res.setHeader(RATE_LIMIT_HEADERS.USER_REMAINING, userStatus.remaining.hourly); - res.setHeader(RATE_LIMIT_HEADERS.USER_RESET, Math.ceil(userStatus.resetAt.hourly / 1_000)); - - if (!userCheck.allowed && !softMode) { - sendRateLimitExceeded( - res, - userCheck.retryAfterMs ?? 60_000, - limits.hourlyLimit * 5, - userStatus.resetAt.hourly, - ); - return; - } + // Attach current status headers + const status = service.getRateLimitStatus(apiKey, tier); + setHeader(res, 'X-RateLimit-Limit', String(status.limits.hourlyLimit)); + setHeader(res, 'X-RateLimit-Remaining', String(status.remaining.hourly)); + setHeader(res, 'X-RateLimit-Reset', String(Math.ceil(status.resetAt.hourly / 1_000))); + setHeader(res, 'X-RateLimit-Burst-Remaining', String(status.remaining.burstTokens)); + + // Warn if approaching soft limit + const pct = status.current.hourly / status.limits.hourlyLimit; + if (pct >= 0.8) { + setHeader(res, 'X-RateLimit-Warning', `Usage at ${Math.round(pct * 100)}% of hourly limit`); } - // ----------------------------------------------------------------------- - // Record the request (async, non-blocking for latency) - // ----------------------------------------------------------------------- - const effectiveKey = apiKey ?? `user:${userId!}`; - service.recordRequest(effectiveKey, tier, path, 200, 0); - if (userId) { - service.recordUserRequest(`user:${userId}`, tier, path); + // Deprecation header for grace-period API keys (integrates with #1009) + const graceHeader = req.headers['x-grace-period-key']; + if (graceHeader === 'true') { + setHeader(res, 'Deprecation', 'true'); + setHeader(res, 'Sunset', req.headers['x-grace-expires'] as string ?? ''); } + // Record usage asynchronously so it does not block the response + const originalJson = res.json.bind(res); + let statusCode = 200; + + res.status = (code: number) => { + statusCode = code; + return res; + }; + + res.json = (body: unknown) => { + // Record after response is built + const latencyMs = Date.now() - start; + service.recordRequest(apiKey, tier, path, statusCode, latencyMs); + + if (getUserId) { + const userId = getUserId(req); + if (userId) { + service.recordUserRequest(`user:${userId}`, tier, path); + } + } + + originalJson(body); + }; + next(); }; } // --------------------------------------------------------------------------- -// Convenience: attach rate-limit status to response after request completes +// Fastify hook factory (thin wrapper using the same logic) // --------------------------------------------------------------------------- -/** - * Post-request middleware that refreshes rate-limit headers with the final - * status (useful when the actual response code differs from 200). - */ -export function createRateLimitStatusMiddleware(options: RateLimitMiddlewareOptions) { - const { - service, - keyFn = defaultKeyFn, - userIdFn = defaultUserIdFn, - tierFn = defaultTierFn, - } = options; - - return function rateLimitStatus( - req: RateLimitRequest, - res: RateLimitResponse, - next: NextFn, - ): void { - const apiKey = keyFn(req); - const userId = userIdFn(req); - const identifier = apiKey ?? userId; - - if (identifier) { - const tier = tierFn(apiKey ?? identifier, userId); - const limits = TIER_RATE_LIMITS[tier]; - const status = service.getRateLimitStatus(identifier, tier); - - res.setHeader(RATE_LIMIT_HEADERS.LIMIT, limits.hourlyLimit); - res.setHeader(RATE_LIMIT_HEADERS.REMAINING, status.remaining.hourly); - res.setHeader(RATE_LIMIT_HEADERS.RESET, Math.ceil(status.resetAt.hourly / 1_000)); - } - - next(); +export function createFastifyRateLimitHook(opts: RateLimitMiddlewareOptions) { + const handler = createRateLimitMiddleware(opts); + + return async function fastifyPreHandler( + request: MinimalRequest & { raw?: MinimalRequest }, + reply: MinimalResponse, + ): Promise { + await new Promise((resolve, reject) => { + handler(request, reply, (err?: unknown) => { + if (err) reject(err); + else resolve(); + }); + }); }; } + +// --------------------------------------------------------------------------- +// IP-based fallback middleware (for unauthenticated endpoints) +// --------------------------------------------------------------------------- + +export function createIpRateLimitMiddleware(opts: Omit) { + return createRateLimitMiddleware({ + ...opts, + allowMissingKey: true, + getApiKey: (req) => { + const ip = + req.ip ?? + (req.socket?.remoteAddress) ?? + (req.headers['x-forwarded-for'] as string)?.split(',')[0]?.trim() ?? + 'unknown'; + return `ip:${ip}`; + }, + }); +} diff --git a/developer-portal/index.ts b/developer-portal/index.ts index 80b00c4c..f2312e00 100644 --- a/developer-portal/index.ts +++ b/developer-portal/index.ts @@ -2,6 +2,16 @@ export { DeveloperPortalService } from './services/portalService'; export { IntegrationGuidesService } from './services/integrationGuidesService'; export { DeveloperOnboarding } from './components/DeveloperOnboarding'; export { ApiKeyManager } from './components/ApiKeyManager'; +export { + PortalApiKeyRotationService, + portalApiKeyRotationService, +} from './services/apiKeyRotationService'; +export type { + RotationOptions, + RotationResult, + ApiKeyValidationResult, + ApiKeyRotationMetrics, +} from './services/apiKeyRotationService'; export { DashboardPage, ApiKeysPage, diff --git a/developer-portal/services/apiKeyRotationService.ts b/developer-portal/services/apiKeyRotationService.ts new file mode 100644 index 00000000..1ac297ee --- /dev/null +++ b/developer-portal/services/apiKeyRotationService.ts @@ -0,0 +1,200 @@ +/** + * Developer Portal — API Key Rotation Service + * + * Issue #1009: Implement API key rotation with grace period + * + * Thin portal-facing facade around the shared ApiKeyRotationService that maps + * between the portal's ApiKey type (developer.ts) and the backend rotation + * primitives, so portal pages and components stay decoupled from internals. + */ + +import { + ApiKeyRotationService, + type ManagedApiKey, + type RotationOptions, + type RotationResult, + type ApiKeyValidationResult, + type ApiKeyRotationMetrics, +} from '../../backend/services/shared/apiKeyRotation'; +import type { ApiKey, ApiPermission } from '../types/developer'; + +// Re-export so portal consumers can import from a single path. +export type { + RotationOptions, + RotationResult, + ApiKeyValidationResult, + ApiKeyRotationMetrics, +}; + +// --------------------------------------------------------------------------- +// Mapping helpers +// --------------------------------------------------------------------------- + +function toPortalApiKey(managed: ManagedApiKey): ApiKey { + return { + id: managed.id, + key: managed.key, + name: managed.name, + type: managed.environment, + permissions: managed.permissions as ApiPermission[], + rateLimit: { + requestsPerMinute: 60, + requestsPerHour: 1_000, + requestsPerDay: 10_000, + burstLimit: 100, + }, + usageCount: managed.usageCount, + status: + managed.status === 'active' || managed.status === 'grace' + ? 'active' + : managed.status === 'expired' + ? 'expired' + : 'revoked', + createdAt: new Date(managed.createdAt), + expiresAt: managed.expiresAt ? new Date(managed.expiresAt) : undefined, + revokedAt: managed.revokedAt ? new Date(managed.revokedAt) : undefined, + lastUsedAt: managed.lastUsedAt ? new Date(managed.lastUsedAt) : undefined, + }; +} + +// --------------------------------------------------------------------------- +// Portal API Key Rotation Service +// --------------------------------------------------------------------------- + +export class PortalApiKeyRotationService { + private readonly inner: ApiKeyRotationService; + + constructor(rotationService?: ApiKeyRotationService) { + this.inner = rotationService ?? new ApiKeyRotationService(); + } + + // ── Key management ──────────────────────────────────────────────────────── + + /** + * Provision a new API key for a developer. + */ + createKey( + developerId: string, + name: string, + environment: 'test' | 'production', + permissions: ApiPermission[] = [], + ttlMs?: number, + ): ApiKey { + const managed = this.inner.createKey(developerId, name, environment, permissions, ttlMs); + return toPortalApiKey(managed); + } + + /** + * Rotate an existing key with a configurable grace period. + * + * During the grace window both the old and new key are accepted. + * The portal UI should surface a deprecation banner to the developer. + * + * @returns RotationResult containing both the new key and the old key + * (now in grace period) along with `gracePeriodEndsAt`. + */ + async rotateKey( + keyId: string, + options: RotationOptions = {}, + ): Promise<{ + newKey: ApiKey; + oldKey: ApiKey; + gracePeriodEndsAt: Date; + gracePeriodRemainingMs: number; + }> { + const result = await this.inner.rotateKey(keyId, options); + return { + newKey: toPortalApiKey(result.newKey), + oldKey: toPortalApiKey(result.oldKey), + gracePeriodEndsAt: new Date(result.gracePeriodEndsAt), + gracePeriodRemainingMs: result.gracePeriodRemainingMs, + }; + } + + /** + * Immediately revoke a key with no grace period. + */ + revokeKey(keyId: string): boolean { + return this.inner.revokeKey(keyId); + } + + /** + * Revoke all keys for a developer (e.g., on account suspension). + */ + revokeAllKeys(developerId: string): number { + return this.inner.revokeAllKeys(developerId); + } + + // ── Validation ──────────────────────────────────────────────────────────── + + /** + * Validate a secret key string. + * Returns `isGrace: true` when the old key is used after rotation so the + * portal middleware can add a `Deprecation` response header. + */ + validateKey(secret: string): ApiKeyValidationResult { + return this.inner.validateKey(secret); + } + + // ── Queries ─────────────────────────────────────────────────────────────── + + getKey(keyId: string): ApiKey | null { + const managed = this.inner.getKey(keyId); + return managed ? toPortalApiKey(managed) : null; + } + + getKeysByDeveloper(developerId: string): ApiKey[] { + return this.inner.getKeysByDeveloper(developerId).map(toPortalApiKey); + } + + getActiveKeysByDeveloper(developerId: string): ApiKey[] { + return this.inner.getActiveKeysByDeveloper(developerId).map(toPortalApiKey); + } + + /** + * Returns rotation history for a developer (audit trail). + */ + getRotationHistory(developerId?: string) { + return this.inner.getRotationHistory(developerId); + } + + /** + * Returns the grace period status for a key currently mid-rotation. + */ + getGracePeriodStatus( + keyId: string, + ): { gracePeriodEndsAt: Date; remainingMs: number; successorKeyId: string } | null { + const status = this.inner.getGracePeriodStatus(keyId); + if (!status) return null; + return { + gracePeriodEndsAt: new Date(status.gracePeriodEndsAt), + remainingMs: status.remainingMs, + successorKeyId: status.successorKeyId, + }; + } + + // ── Maintenance ─────────────────────────────────────────────────────────── + + /** + * Clean up expired keys. Call periodically (e.g., cron every hour). + */ + cleanupExpiredKeys(): number { + return this.inner.cleanupExpiredKeys(); + } + + // ── Metrics ─────────────────────────────────────────────────────────────── + + getMetrics(): ApiKeyRotationMetrics { + return this.inner.getMetrics(); + } + + prometheusMetrics(): string { + return this.inner.prometheusMetrics(); + } +} + +// --------------------------------------------------------------------------- +// Singleton +// --------------------------------------------------------------------------- + +export const portalApiKeyRotationService = new PortalApiKeyRotationService();