From 9c000c67201254263d626d0a2c0edf67aad81211 Mon Sep 17 00:00:00 2001 From: Chinaza007 Date: Sun, 30 Aug 2026 02:26:01 +0100 Subject: [PATCH] feat: add websocket pooling and streaming optimizations Closes #741. Closes #743. Closes #744. Closes #734. --- ENV_VARS.md | 6 + PERFORMANCE_GUIDE.md | 89 +++++-- backend/.env.example | 6 + backend/src/config.ts | 10 + .../src/config/database-read-replica.test.ts | 70 ++++++ backend/src/config/database.ts | 77 ++++++ backend/src/index.ts | 33 +-- .../middleware/__tests__/compression.test.ts | 50 ++++ .../middleware/__tests__/streaming.test.ts | 106 ++++++++ backend/src/middleware/compression.ts | 237 +++++++----------- backend/src/middleware/streaming.ts | 145 +++++++++++ backend/src/routes/pool-monitor.ts | 6 +- backend/src/routes/streaming-export.ts | 27 +- .../src/services/exports/streaming-export.ts | 3 +- backend/src/websocket/pool.test.ts | 101 ++++++++ backend/src/websocket/pool.ts | 131 ++++++++++ backend/src/websocket/server.ts | 34 +-- infra/main.tf | 49 +++- infra/variables.tf | 18 ++ 19 files changed, 982 insertions(+), 216 deletions(-) create mode 100644 backend/src/config/database-read-replica.test.ts create mode 100644 backend/src/middleware/__tests__/compression.test.ts create mode 100644 backend/src/middleware/__tests__/streaming.test.ts create mode 100644 backend/src/middleware/streaming.ts create mode 100644 backend/src/websocket/pool.test.ts create mode 100644 backend/src/websocket/pool.ts diff --git a/ENV_VARS.md b/ENV_VARS.md index 8348bf0e..9c1c9f9f 100644 --- a/ENV_VARS.md +++ b/ENV_VARS.md @@ -14,6 +14,10 @@ | VAPID_PRIVATE_KEY | VAPID private key for Web Push API | auto-generated | No | | WS_ENABLED | Enable/disable WebSocket support | true | No | | WS_PORT | WebSocket port | 3001 | No | +| DB_READ_REPLICA_URLS | Comma-separated PostgreSQL read replica URLs | - | No | +| DB_REPLICA_MAX_LAG_MS | Maximum replica lag before primary failover | 5000 | No | +| DB_REPLICA_HEALTH_CHECK_INTERVAL_MS | Interval for replica health checks | 30000 | No | +| DB_REPLICA_FAILOVER_COOLDOWN_MS | Cooldown after replica failover | 15000 | No | ## Frontend @@ -37,6 +41,8 @@ AGENTICPAY_ALLOWED_SIGNATURE_ORIGINS=https://agenticpay.com,http://localhost:300 VAPID_PUBLIC_KEY=your-vapid-public-key VAPID_PRIVATE_KEY=your-vapid-private-key WS_ENABLED=true +DB_READ_REPLICA_URLS= +DB_REPLICA_MAX_LAG_MS=5000 ``` - `.env.development` — local development diff --git a/PERFORMANCE_GUIDE.md b/PERFORMANCE_GUIDE.md index 1cdc2f26..7f9109ac 100644 --- a/PERFORMANCE_GUIDE.md +++ b/PERFORMANCE_GUIDE.md @@ -65,13 +65,12 @@ Reduces payload size by 60-80% on average, improving: ### Compression Methods -1. **Brotli** (preferred): 20-30% smaller than gzip - - Quality level: 5 (balanced speed/compression) - - Mode: Text optimization +AgenticPay uses the maintained Express `compression` middleware with streaming +backpressure support, negotiated encodings, and route-level filters. -2. **Gzip** (fallback): Universal support - - Compression level: 6 - - Minimum size threshold: 1KB +- Compression level: 6 +- Minimum size threshold: 1KB by default +- Skips images, audio, video, archives, and already-compressed content ### Implementation @@ -79,8 +78,7 @@ Located in: `backend/src/middleware/compression.ts` ```typescript app.use(compressionMiddleware({ - brotliLevel: 5, - gzipLevel: 6, + level: 6, minSizeBytes: 1024, })); ``` @@ -90,7 +88,7 @@ app.use(compressionMiddleware({ Access compression metrics via: ``` -GET /api/v1/monitoring/pool/compression +GET /api/v1/monitoring/compression ``` Returns: @@ -108,6 +106,35 @@ Returns: --- +## API Response Streaming + +Large exports are streamed with chunked transfer encoding instead of being buffered in memory. + +Endpoints: + +```bash +GET /api/v1/exports/audit/stream?format=csv&limit=100000 +GET /api/v1/exports/audit/stream?format=jsonl&batchSize=1000 +GET /api/v1/exports/payments/stream?format=csv +``` + +Reusable helpers live in `backend/src/middleware/streaming.ts`: + +```typescript +const query = parseStreamingQuery(req.query); +await streamDataset({ + req, + res, + items: takeStreamItems(fetchRows(), query.limit), + format: query.format, +}); +``` + +The streaming helpers set `Transfer-Encoding: chunked`, disable proxy buffering with +`X-Accel-Buffering: no`, honor HTTP backpressure, and track completed, aborted, and failed streams. + +--- + ## Cursor-Based Pagination ### Overview @@ -198,6 +225,40 @@ GET /api/v1/payments -H "If-None-Match: abc123def456" Optimized connection pooling with PgBouncer for efficient resource utilization: +### Read Replicas and Failover + +Read replica routing is configured with: + +```bash +DB_READ_REPLICA_URLS=postgresql://user:pass@replica-a:5432/agenticpay,postgresql://user:pass@replica-b:5432/agenticpay +DB_REPLICA_MAX_LAG_MS=5000 +DB_REPLICA_HEALTH_CHECK_INTERVAL_MS=30000 +DB_REPLICA_FAILOVER_COOLDOWN_MS=15000 +``` + +`backend/src/config/database.ts` exposes `ReadReplicaRouter`, which routes `SELECT` and `WITH` +queries across healthy replicas and falls back to `DATABASE_URL` when no replica is available or +replica lag exceeds the configured threshold. Terraform can provision replicas with +`db_read_replica_count` and wires `DB_READ_REPLICA_URLS` into the backend service. + +### WebSocket Pooling + +WebSocket connections are managed by `backend/src/websocket/pool.ts`. The pool enforces capacity, +tracks active and queued connections, batches outbound messages through `ManagedConnection`, and +supports clean shutdown. Tune batching with: + +```typescript +attachWebSocketServer({ + server, + options: { + maxConnections: 250, + maxQueueSizePerConnection: 500, + flushIntervalMs: 25, + maxBatchSize: 50, + }, +}); +``` + **Benefits:** - Prevents connection exhaustion - Detects and prevents connection leaks @@ -239,7 +300,7 @@ Located in: `backend/src/config/database.ts` Access pool health via: ``` -GET /api/v1/monitoring/pool/health +GET /api/v1/monitoring/health ``` Returns: @@ -261,7 +322,7 @@ Returns: Automatic detection of connection leaks: ``` -GET /api/v1/monitoring/pool/leaks +GET /api/v1/monitoring/leaks ``` - Monitors connection acquisition/release @@ -271,7 +332,7 @@ GET /api/v1/monitoring/pool/leaks ### Metrics Endpoint ``` -GET /api/v1/monitoring/pool/metrics +GET /api/v1/monitoring/metrics ``` Returns comprehensive pool statistics including: @@ -358,7 +419,7 @@ cache.registerWarmer('dashboard:overview', ### Metrics Endpoint ``` -GET /api/v1/monitoring/pool/cache +GET /api/v1/monitoring/cache ``` Returns: @@ -383,7 +444,7 @@ Returns: Comprehensive view of all performance metrics: ``` -GET /api/v1/monitoring/pool/performance +GET /api/v1/monitoring/performance ``` Returns combined metrics: diff --git a/backend/.env.example b/backend/.env.example index 319ffb82..e40e5a52 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -10,6 +10,12 @@ RATE_LIMIT_ENTERPRISE=1000 RATE_LIMIT_WINDOW_MS=900000 COMPRESSION_THRESHOLD=1024 +# Database read replicas (comma-separated PostgreSQL URLs) +DB_READ_REPLICA_URLS= +DB_REPLICA_MAX_LAG_MS=5000 +DB_REPLICA_HEALTH_CHECK_INTERVAL_MS=30000 +DB_REPLICA_FAILOVER_COOLDOWN_MS=15000 + # Security headers HSTS_MAX_AGE_SECONDS=31536000 PERMISSIONS_POLICY=camera=(), microphone=(), geolocation=(), payment=(), usb=(), magnetometer=(), gyroscope=(), interest-cohort=() diff --git a/backend/src/config.ts b/backend/src/config.ts index 27ab83c1..85799390 100644 --- a/backend/src/config.ts +++ b/backend/src/config.ts @@ -26,6 +26,10 @@ const envSchema = z.object({ DB_POOL_ACQUIRE_TIMEOUT_MS: z.string().default('30000'), DB_POOL_MAX_USES: z.string().default('7500'), DB_STATEMENT_TIMEOUT_MS: z.string().default('30000'), + DB_READ_REPLICA_URLS: z.string().default(''), + DB_REPLICA_MAX_LAG_MS: z.string().default('5000'), + DB_REPLICA_HEALTH_CHECK_INTERVAL_MS: z.string().default('30000'), + DB_REPLICA_FAILOVER_COOLDOWN_MS: z.string().default('15000'), HSTS_MAX_AGE_SECONDS: z.string().default('31536000'), PERMISSIONS_POLICY: z .string() @@ -83,6 +87,12 @@ export const config = { maxUses: Number(env.DB_POOL_MAX_USES), statementTimeoutMs: Number(env.DB_STATEMENT_TIMEOUT_MS), }, + replicas: { + urls: env.DB_READ_REPLICA_URLS.split(',').map((url) => url.trim()).filter(Boolean), + maxLagMs: Number(env.DB_REPLICA_MAX_LAG_MS), + healthCheckIntervalMs: Number(env.DB_REPLICA_HEALTH_CHECK_INTERVAL_MS), + failoverCooldownMs: Number(env.DB_REPLICA_FAILOVER_COOLDOWN_MS), + }, }, security: { hsts: { diff --git a/backend/src/config/database-read-replica.test.ts b/backend/src/config/database-read-replica.test.ts new file mode 100644 index 00000000..2a37a479 --- /dev/null +++ b/backend/src/config/database-read-replica.test.ts @@ -0,0 +1,70 @@ +import { describe, expect, it } from 'vitest'; +import { + ReadReplicaRouter, + buildReplicaConfigs, + isReadQuery, +} from './database'; + +describe('read replica routing', () => { + it('detects read queries conservatively', () => { + expect(isReadQuery('select * from payments')).toBe(true); + expect(isReadQuery(' WITH recent AS (select 1) select * from recent')).toBe(true); + expect(isReadQuery('update payments set status = $1')).toBe(false); + }); + + it('routes writes to primary and reads to healthy replicas', () => { + const router = new ReadReplicaRouter( + ['postgres://replica-1/db', 'postgres://replica-2/db'], + 'postgres://primary/db', + 5000, + ); + + expect(router.select('UPDATE payments SET status = $1')).toEqual({ + url: 'postgres://primary/db', + source: 'primary', + reason: 'write_query', + }); + + expect(router.select('SELECT * FROM payments')).toEqual({ + url: 'postgres://replica-1/db', + source: 'replica', + reason: 'healthy_replica', + }); + expect(router.select('SELECT * FROM invoices')).toMatchObject({ + url: 'postgres://replica-2/db', + source: 'replica', + }); + }); + + it('fails over to primary when all replicas are unhealthy or lagging', () => { + const router = new ReadReplicaRouter( + ['postgres://replica-1/db', 'postgres://replica-2/db'], + 'postgres://primary/db', + 100, + ); + + router.updateHealth('postgres://replica-1/db', { healthy: false }); + router.updateHealth('postgres://replica-2/db', { healthy: true, lagMs: 500 }); + + expect(router.select('SELECT * FROM payments')).toEqual({ + url: 'postgres://primary/db', + source: 'primary', + reason: 'replica_unavailable', + }); + }); + + it('builds replica configs from environment URLs', () => { + const previous = process.env.DB_READ_REPLICA_URLS; + process.env.DB_READ_REPLICA_URLS = 'postgres://user:pass@replica-a:5432/app, postgres://user:pass@replica-b/app'; + + try { + expect(buildReplicaConfigs()).toMatchObject([ + { host: 'replica-a', port: 5432, database: 'app', user: 'user', enabled: true }, + { host: 'replica-b', port: 5432, database: 'app', user: 'user', enabled: true }, + ]); + } finally { + if (previous === undefined) delete process.env.DB_READ_REPLICA_URLS; + else process.env.DB_READ_REPLICA_URLS = previous; + } + }); +}); diff --git a/backend/src/config/database.ts b/backend/src/config/database.ts index ee0db488..b1f4e24b 100644 --- a/backend/src/config/database.ts +++ b/backend/src/config/database.ts @@ -874,6 +874,83 @@ export function isReadQuery(sql: string): boolean { return /^\s*(SELECT|WITH\s)/i.test(sql); } +export type ReplicaHealth = "healthy" | "lagging" | "unhealthy"; + +export interface ReadReplicaTarget { + url: string; + health: ReplicaHealth; + lagMs: number; + lastCheckedAt: number; + failureCount: number; +} + +export interface ReplicaSelection { + url: string; + source: "primary" | "replica"; + reason: "write_query" | "no_replicas" | "healthy_replica" | "replica_unavailable"; +} + +export class ReadReplicaRouter { + private replicas: ReadReplicaTarget[]; + private nextReplicaIndex = 0; + + constructor( + replicaUrls = buildReplicaUrls(), + private readonly primaryUrl = process.env.DATABASE_URL ?? "", + private readonly maxLagMs = envInt("DB_REPLICA_MAX_LAG_MS", 5000), + ) { + this.replicas = replicaUrls.map((url) => ({ + url, + health: "healthy", + lagMs: 0, + lastCheckedAt: 0, + failureCount: 0, + })); + } + + select(sql: string): ReplicaSelection { + if (!isReadQuery(sql)) { + return { url: this.primaryUrl, source: "primary", reason: "write_query" }; + } + + const healthyReplicas = this.replicas.filter( + (replica) => replica.health === "healthy" && replica.lagMs <= this.maxLagMs, + ); + + if (this.replicas.length === 0) { + return { url: this.primaryUrl, source: "primary", reason: "no_replicas" }; + } + + if (healthyReplicas.length === 0) { + return { url: this.primaryUrl, source: "primary", reason: "replica_unavailable" }; + } + + const replica = healthyReplicas[this.nextReplicaIndex % healthyReplicas.length]; + this.nextReplicaIndex = (this.nextReplicaIndex + 1) % healthyReplicas.length; + return { url: replica.url, source: "replica", reason: "healthy_replica" }; + } + + updateHealth(url: string, params: { healthy: boolean; lagMs?: number; checkedAt?: number }): void { + const replica = this.replicas.find((candidate) => candidate.url === url); + if (!replica) return; + + replica.lagMs = params.lagMs ?? replica.lagMs; + replica.lastCheckedAt = params.checkedAt ?? Date.now(); + replica.health = !params.healthy + ? "unhealthy" + : replica.lagMs > this.maxLagMs + ? "lagging" + : "healthy"; + replica.failureCount = replica.health === "healthy" ? 0 : replica.failureCount + 1; + } + + snapshot(): ReadReplicaTarget[] { + return this.replicas.map((replica) => ({ ...replica })); + } +} + +export const readReplicaRouter = new ReadReplicaRouter(); + // ── Query Profiler ──────────────────────────────────────────────────────────── export interface QueryProfile { diff --git a/backend/src/index.ts b/backend/src/index.ts index 69974ffe..77ab9a26 100644 --- a/backend/src/index.ts +++ b/backend/src/index.ts @@ -4,7 +4,6 @@ import express, { Request, Response, NextFunction } from 'express'; import cors from 'cors'; import dotenv from 'dotenv'; import rateLimit from 'express-rate-limit'; -import compression from 'compression'; import { verificationRouter } from './routes/verification.js'; import { invoiceRouter } from './routes/invoice.js'; import { stellarRouter } from './routes/stellar.js'; @@ -48,6 +47,12 @@ import commentsRouter from './routes/comments.js'; import collaborationRouter from './routes/collaboration.js'; import { paymentStrategiesRouter } from './routes/payment-strategies.js'; import { registerDefaultPaymentProviders } from './services/payments/bootstrap.js'; +import { compressionMiddleware } from './middleware/compression.js'; +import { streamingExportRouter } from './routes/streaming-export.js'; +import { poolMonitorRouter } from './routes/pool-monitor.js'; +import { legacyRouter } from './routes/legacy.js'; +import { splitsRouter } from './routes/splits.js'; +import { refundsRouter } from './routes/refunds.js'; dotenv.config(); @@ -180,24 +185,7 @@ app.use(contentSecurityPolicy()); app.use(express.json()); -app.use( - compression({ - threshold: config.compression.threshold, - filter: (req, res) => { - if (req.headers['x-no-compression']) { - return false; - } - const contentType = res.getHeader('Content-Type'); - if (typeof contentType === 'string' && contentType.includes('application/json')) { - return true; - } - if (Array.isArray(contentType) && contentType.some((ct) => ct.includes('application/json'))) { - return true; - } - return compression.filter(req, res); - }, - }) -); +app.use(compressionMiddleware({ minSizeBytes: config.compression.threshold })); app.use(requestIdMiddleware); app.use(auditMiddleware()); @@ -241,9 +229,6 @@ app.use('/docs', docsRouter); import { versionMiddleware } from './middleware/versioning.js'; -import { portfolioRouter } from './routes/portfolio.js'; -import { emailRouter } from './routes/email.js'; - // Apply tiered limiter to all API routes app.use('/api/', tieredRateLimit); @@ -297,6 +282,10 @@ apiV1Router.use('/comments', commentsRouter); apiV1Router.use('/collaboration', collaborationRouter); // Multi-chain payment processing via the PaymentProvider strategy pattern — Issue #726 apiV1Router.use('/payment-strategies', paymentStrategiesRouter); +// Large dataset streaming exports +apiV1Router.use('/exports', streamingExportRouter); +// Performance and pool monitoring +apiV1Router.use('/monitoring', poolMonitorRouter); // Explicit URL-based mounting app.use('/api/v1', apiV1Router); diff --git a/backend/src/middleware/__tests__/compression.test.ts b/backend/src/middleware/__tests__/compression.test.ts new file mode 100644 index 00000000..062041f2 --- /dev/null +++ b/backend/src/middleware/__tests__/compression.test.ts @@ -0,0 +1,50 @@ +import { describe, expect, it } from 'vitest'; +import type { Request, Response } from 'express'; +import { + getCompressionMetrics, + recordCompressionMetric, + shouldCompressResponse, +} from '../compression'; + +function req(headers: Record = {}): Request { + return { headers } as Request; +} + +function res(contentType?: string): Response { + const headers: Record = {}; + if (contentType) headers['Content-Type'] = contentType; + + return { + getHeader(name: string) { + return headers[name]; + }, + } as unknown as Response; +} + +describe('compression middleware helpers', () => { + it('skips compression when explicitly disabled', () => { + expect(shouldCompressResponse(req({ 'x-no-compression': '1' }), res('application/json'))).toBe(false); + }); + + it('allows JSON and text responses through the compression filter', () => { + expect(shouldCompressResponse(req({ 'accept-encoding': 'gzip' }), res('application/json'))).toBe(true); + expect(shouldCompressResponse(req({ 'accept-encoding': 'gzip' }), res('text/csv'))).toBe(true); + }); + + it('skips pre-compressed or binary content types', () => { + expect(shouldCompressResponse(req({ 'accept-encoding': 'gzip' }), res('image/png'))).toBe(false); + expect(shouldCompressResponse(req({ 'accept-encoding': 'gzip' }), res('application/zip'))).toBe(false); + }); + + it('records compression metrics', () => { + const before = getCompressionMetrics(); + + recordCompressionMetric(1000, 250, 'gzip', 4); + const after = getCompressionMetrics(); + + expect(after.totalRequests).toBe(before.totalRequests + 1); + expect(after.compressedRequests).toBeGreaterThanOrEqual(before.compressedRequests + 1); + expect(after.gzipRequests).toBeGreaterThanOrEqual(before.gzipRequests + 1); + expect(after.compressionRatio).toBeGreaterThan(0); + }); +}); diff --git a/backend/src/middleware/__tests__/streaming.test.ts b/backend/src/middleware/__tests__/streaming.test.ts new file mode 100644 index 00000000..048f7f9f --- /dev/null +++ b/backend/src/middleware/__tests__/streaming.test.ts @@ -0,0 +1,106 @@ +import { EventEmitter } from 'node:events'; +import { describe, expect, it, beforeEach, vi } from 'vitest'; +import type { Request, Response } from 'express'; +import { + getStreamingMetrics, + parseStreamingQuery, + resetStreamingMetrics, + streamDataset, + takeStreamItems, +} from '../streaming'; + +function makeReq(): Request { + return new EventEmitter() as Request; +} + +function makeRes() { + const chunks: string[] = []; + const headers: Record = {}; + const res = new EventEmitter() as Response & { + chunks: string[]; + writableEnded: boolean; + destroyed: boolean; + }; + + res.chunks = chunks; + res.writableEnded = false; + res.destroyed = false; + res.status = vi.fn(() => res) as any; + res.setHeader = vi.fn((name: string, value: string) => { + headers[name] = value; + return res; + }) as any; + res.write = vi.fn((chunk: string) => { + chunks.push(chunk); + return true; + }) as any; + res.end = vi.fn(() => { + res.writableEnded = true; + return res; + }) as any; + res.json = vi.fn((body: unknown) => { + chunks.push(JSON.stringify(body)); + res.writableEnded = true; + return res; + }) as any; + + return { res, chunks, headers }; +} + +async function* rows() { + yield { id: 1, name: 'one' }; + yield { id: 2, name: 'two' }; +} + +describe('streaming middleware helpers', () => { + beforeEach(() => resetStreamingMetrics()); + + it('parses streaming query defaults and caps batch size', () => { + expect(parseStreamingQuery({})).toEqual({ format: 'json', limit: undefined, batchSize: 500 }); + expect(parseStreamingQuery({ format: 'jsonl', limit: '10', batchSize: '99999' })).toEqual({ + format: 'jsonl', + limit: 10, + batchSize: 5000, + }); + }); + + it('streams a JSON array without buffering every row first', async () => { + const req = makeReq(); + const { res, chunks, headers } = makeRes(); + + await streamDataset({ req, res, items: rows(), format: 'json' }); + + expect(headers['Content-Type']).toBe('application/json'); + expect(chunks.join('')).toBe('{"data":[{"id":1,"name":"one"},{"id":2,"name":"two"}]}'); + expect(getStreamingMetrics()).toMatchObject({ + completedStreams: 1, + rowsStreamed: 2, + activeStreams: 0, + }); + }); + + it('streams newline-delimited JSON and applies transforms', async () => { + const req = makeReq(); + const { res, chunks, headers } = makeRes(); + + await streamDataset({ + req, + res, + items: rows(), + format: 'jsonl', + transform: (row) => ({ id: row.id }), + }); + + expect(headers['Content-Type']).toBe('application/x-ndjson'); + expect(chunks.join('')).toBe('{"id":1}\n{"id":2}\n'); + }); + + it('limits async iterable output', async () => { + const limited: Array<{ id: number; name: string }> = []; + for await (const row of takeStreamItems(rows(), 1)) { + limited.push(row); + } + + expect(limited).toEqual([{ id: 1, name: 'one' }]); + }); +}); diff --git a/backend/src/middleware/compression.ts b/backend/src/middleware/compression.ts index 56cb429b..14a1e06e 100644 --- a/backend/src/middleware/compression.ts +++ b/backend/src/middleware/compression.ts @@ -1,16 +1,25 @@ -import { brotliCompressSync, gzipSync, constants } from 'node:zlib'; +import compression from 'compression'; import type { Request, Response, NextFunction } from 'express'; interface CompressionConfig { - brotliLevel: number; - gzipLevel: number; + level: number; minSizeBytes: number; excludeContentTypes: string[]; } +interface CompressionMetrics { + totalRequests: number; + compressedRequests: number; + totalOriginalSize: number; + totalCompressedSize: number; + compressionRatio: number; + brotliRequests: number; + gzipRequests: number; + averageCompressionTimeMs: number; +} + const DEFAULT_CONFIG: CompressionConfig = { - brotliLevel: 5, - gzipLevel: 6, + level: 6, minSizeBytes: 1024, excludeContentTypes: [ 'image/', @@ -33,174 +42,114 @@ const TEXT_TYPES = [ ]; const configs = new Map(); +const compressionTimes: number[] = []; + +const metrics: CompressionMetrics = { + totalRequests: 0, + compressedRequests: 0, + totalOriginalSize: 0, + totalCompressedSize: 0, + compressionRatio: 0, + brotliRequests: 0, + gzipRequests: 0, + averageCompressionTimeMs: 0, +}; export function configureEndpoint(endpoint: string, config: Partial): void { const existing = configs.get(endpoint) ?? { ...DEFAULT_CONFIG }; - Object.assign(existing, config); - configs.set(endpoint, existing); + configs.set(endpoint, { ...existing, ...config }); } -function getConfig(req: Request): CompressionConfig { - const endpoint = req.route?.path ?? req.path; - return configs.get(endpoint) ?? DEFAULT_CONFIG; +function endpointConfig(req: Request, globalConfig: CompressionConfig): CompressionConfig { + return configs.get(req.route?.path ?? req.path) ?? configs.get(req.path) ?? globalConfig; } -function shouldCompress(req: Request, res: Response, config: CompressionConfig): boolean { +function isCompressibleContentType(contentType: string, config: CompressionConfig): boolean { + const normalized = contentType.toLowerCase(); + if (config.excludeContentTypes.some((excluded) => normalized.startsWith(excluded))) { + return false; + } + return TEXT_TYPES.some((type) => normalized.startsWith(type)); +} + +export function shouldCompressResponse(req: Request, res: Response, config = DEFAULT_CONFIG): boolean { if (req.headers['x-no-compression']) return false; - const contentLength = parseInt(res.getHeader('Content-Length') as string || '0', 10); - if (contentLength > 0 && contentLength < config.minSizeBytes) return false; + const contentType = String(res.getHeader('Content-Type') ?? '').toLowerCase(); + if (contentType && !isCompressibleContentType(contentType, config)) { + return false; + } - const contentType = (res.getHeader('Content-Type') as string || '').toLowerCase(); - if (!contentType) return false; + return compression.filter(req, res); +} - for (const exclude of config.excludeContentTypes) { - if (contentType.startsWith(exclude)) return false; - } +function updateMetrics(startedAt: bigint, res: Response): void { + metrics.totalRequests += 1; - if (TEXT_TYPES.some(t => contentType.startsWith(t))) return true; + const elapsedMs = Number(process.hrtime.bigint() - startedAt) / 1_000_000; + compressionTimes.push(elapsedMs); + if (compressionTimes.length > 1000) compressionTimes.shift(); + metrics.averageCompressionTimeMs = + compressionTimes.reduce((sum, value) => sum + value, 0) / compressionTimes.length; - return false; -} + const encoding = String(res.getHeader('Content-Encoding') ?? '').toLowerCase(); + if (!encoding) return; -function getCompressedResponse(res: Response, body: unknown, config: CompressionConfig): Buffer | null { - const bodyStr = typeof body === 'string' ? body : JSON.stringify(body); - const bodyBuf = Buffer.from(bodyStr, 'utf-8'); - const len = bodyBuf.length; - - const acceptEncoding = (res.req.headers['accept-encoding'] as string) || ''; - - if (acceptEncoding.includes('br')) { - try { - const compressed = brotliCompressSync(bodyBuf, { - params: { - [constants.BROTLI_PARAM_QUALITY]: config.brotliLevel, - [constants.BROTLI_PARAM_MODE]: constants.BROTLI_MODE_TEXT, - }, - }); - if (compressed.length < len) { - res.setHeader('Content-Encoding', 'br'); - res.setHeader('X-Compression', 'brotli'); - return compressed; - } - } catch { - // Fall through to gzip - } + metrics.compressedRequests += 1; + if (encoding === 'br') metrics.brotliRequests += 1; + if (encoding === 'gzip') metrics.gzipRequests += 1; + + const contentLength = Number(res.getHeader('Content-Length') ?? 0); + if (Number.isFinite(contentLength) && contentLength > 0) { + metrics.totalCompressedSize += contentLength; } - if (acceptEncoding.includes('gzip')) { - try { - const compressed = gzipSync(bodyBuf, { level: config.gzipLevel }); - if (compressed.length < len) { - res.setHeader('Content-Encoding', 'gzip'); - res.setHeader('X-Compression', 'gzip'); - return compressed; - } - } catch { - // Fall through to uncompressed - } + if (metrics.totalOriginalSize > 0 && metrics.totalCompressedSize > 0) { + metrics.compressionRatio = + (1 - metrics.totalCompressedSize / metrics.totalOriginalSize) * 100; } +} - if (acceptEncoding.includes('deflate')) { - return null; +export function recordCompressionMetric( + originalSize: number, + compressedSize: number, + encoding: string, + timeMs: number, +): void { + metrics.totalRequests += 1; + if (compressedSize < originalSize) { + metrics.compressedRequests += 1; + metrics.totalOriginalSize += originalSize; + metrics.totalCompressedSize += compressedSize; + if (encoding === 'br') metrics.brotliRequests += 1; + if (encoding === 'gzip') metrics.gzipRequests += 1; } - return null; + compressionTimes.push(timeMs); + if (compressionTimes.length > 1000) compressionTimes.shift(); + metrics.compressionRatio = + metrics.totalOriginalSize > 0 + ? (1 - metrics.totalCompressedSize / metrics.totalOriginalSize) * 100 + : 0; + metrics.averageCompressionTimeMs = + compressionTimes.reduce((sum, value) => sum + value, 0) / compressionTimes.length; } export function compressionMiddleware(config?: Partial) { const globalConfig = { ...DEFAULT_CONFIG, ...config }; + const middleware = compression({ + threshold: globalConfig.minSizeBytes, + level: globalConfig.level, + filter: (req, res) => shouldCompressResponse(req, res, endpointConfig(req, globalConfig)), + }); return (req: Request, res: Response, next: NextFunction): void => { - if (req.method === 'HEAD') { - next(); - return; - } - - const endpointConfig = configs.get(req.path) ?? globalConfig; - - if (!shouldCompress(req, res, endpointConfig)) { - next(); - return; - } - - const originalSend = res.send.bind(res); - const originalJson = res.json.bind(res); - const originalEnd = res.end.bind(res); - - let responseBody: unknown = null; - let contentType: string | undefined; - - res.send = (body: unknown): Response => { - responseBody = body; - contentType = res.getHeader('Content-Type') as string || undefined; - return originalSend(''); // Will be replaced by our compressed version in end - }; - - res.json = (body: unknown): Response => { - responseBody = body; - contentType = 'application/json'; - return originalJson(body); - }; - - res.end = (data?: unknown, encoding?: BufferEncoding | (() => void), cb?: () => void): Response => { - const body = data ?? responseBody; - if (!body) { - return originalEnd(data as Buffer, (encoding as BufferEncoding) || 'utf-8', cb as (() => void) | undefined); - } - - const compressed = getCompressedResponse(res, body, endpointConfig); - if (compressed) { - res.removeHeader('Content-Length'); - return originalEnd(compressed, cb as (() => void) | undefined); - } - - return originalEnd(data as Buffer, (encoding as BufferEncoding) || 'utf-8', cb as (() => void) | undefined); - }; - - next(); + const startedAt = process.hrtime.bigint(); + res.once('finish', () => updateMetrics(startedAt, res)); + middleware(req, res, next); }; } -interface CompressionMetrics { - totalRequests: number; - compressedRequests: number; - totalOriginalSize: number; - totalCompressedSize: number; - compressionRatio: number; - brotliRequests: number; - gzipRequests: number; - averageCompressionTimeMs: number; -} - -const metrics: CompressionMetrics = { - totalRequests: 0, - compressedRequests: 0, - totalOriginalSize: 0, - totalCompressedSize: 0, - compressionRatio: 1, - brotliRequests: 0, - gzipRequests: 0, - averageCompressionTimeMs: 0, -}; - -const compressionTimes: number[] = []; - -export function recordCompressionMetric(originalSize: number, compressedSize: number, encoding: string, timeMs: number): void { - metrics.totalRequests++; - if (compressedSize < originalSize) { - metrics.compressedRequests++; - metrics.totalOriginalSize += originalSize; - metrics.totalCompressedSize += compressedSize; - if (encoding === 'br') metrics.brotliRequests++; - else if (encoding === 'gzip') metrics.gzipRequests++; - } - compressionTimes.push(timeMs); - if (compressionTimes.length > 1000) compressionTimes.shift(); - metrics.compressionRatio = metrics.totalOriginalSize > 0 ? (1 - metrics.totalCompressedSize / metrics.totalOriginalSize) * 100 : 0; - metrics.averageCompressionTimeMs = compressionTimes.reduce((a, b) => a + b, 0) / compressionTimes.length || 0; -} - export function getCompressionMetrics() { return { activeEndpoints: Array.from(configs.keys()), diff --git a/backend/src/middleware/streaming.ts b/backend/src/middleware/streaming.ts new file mode 100644 index 00000000..7743bfec --- /dev/null +++ b/backend/src/middleware/streaming.ts @@ -0,0 +1,145 @@ +import { once } from 'node:events'; +import type { Request, Response } from 'express'; + +export type StreamFormat = 'json' | 'jsonl'; + +export interface StreamingQuery { + format: StreamFormat; + limit?: number; + batchSize: number; +} + +export interface StreamingMetrics { + activeStreams: number; + completedStreams: number; + abortedStreams: number; + failedStreams: number; + rowsStreamed: number; +} + +export interface StreamDatasetOptions { + req: Request; + res: Response; + items: AsyncIterable; + format?: StreamFormat; + filename?: string; + transform?: (item: T) => unknown; +} + +const DEFAULT_BATCH_SIZE = 500; +const MAX_BATCH_SIZE = 5000; +const MAX_LIMIT = 10_000_000; + +const metrics: StreamingMetrics = { + activeStreams: 0, + completedStreams: 0, + abortedStreams: 0, + failedStreams: 0, + rowsStreamed: 0, +}; + +function parsePositiveInt(value: unknown, fallback: number, max: number): number { + const parsed = Number(Array.isArray(value) ? value[0] : value); + if (!Number.isFinite(parsed) || parsed < 1) return fallback; + return Math.min(Math.floor(parsed), max); +} + +export function parseStreamingQuery(query: Request['query']): StreamingQuery { + const rawFormat = Array.isArray(query.format) ? query.format[0] : query.format; + const format: StreamFormat = rawFormat === 'jsonl' ? 'jsonl' : 'json'; + const limitValue = Array.isArray(query.limit) ? query.limit[0] : query.limit; + const limit = limitValue === undefined ? undefined : parsePositiveInt(limitValue, MAX_LIMIT, MAX_LIMIT); + + return { + format, + limit, + batchSize: parsePositiveInt(query.batchSize, DEFAULT_BATCH_SIZE, MAX_BATCH_SIZE), + }; +} + +async function writeChunk(res: Response, chunk: string): Promise { + if (res.write(chunk)) return; + await once(res, 'drain'); +} + +export async function streamDataset(options: StreamDatasetOptions): Promise { + const { req, res, items, filename, transform = (item: T) => item } = options; + const format = options.format ?? 'json'; + let first = true; + let aborted = false; + + const onClose = () => { + if (!res.writableEnded) aborted = true; + }; + + req.once('close', onClose); + metrics.activeStreams += 1; + + try { + res.status(200); + res.setHeader('Content-Type', format === 'jsonl' ? 'application/x-ndjson' : 'application/json'); + res.setHeader('Cache-Control', 'no-store'); + res.setHeader('Transfer-Encoding', 'chunked'); + res.setHeader('X-Accel-Buffering', 'no'); + if (filename) { + res.setHeader('Content-Disposition', `attachment; filename="${filename}"`); + } + + if (format === 'json') { + await writeChunk(res, '{"data":['); + } + + for await (const item of items) { + if (aborted || res.destroyed) break; + + const serialized = JSON.stringify(transform(item)); + if (format === 'jsonl') { + await writeChunk(res, `${serialized}\n`); + } else { + await writeChunk(res, `${first ? '' : ','}${serialized}`); + first = false; + } + metrics.rowsStreamed += 1; + } + + if (format === 'json') { + await writeChunk(res, ']}'); + } + + if (!res.writableEnded) res.end(); + + if (aborted) metrics.abortedStreams += 1; + else metrics.completedStreams += 1; + } catch (error) { + metrics.failedStreams += 1; + if (!res.headersSent) { + res.status(500).json({ error: 'STREAM_FAILED' }); + } else if (!res.writableEnded) { + res.destroy(error instanceof Error ? error : new Error(String(error))); + } + } finally { + req.removeListener('close', onClose); + metrics.activeStreams = Math.max(0, metrics.activeStreams - 1); + } +} + +export async function* takeStreamItems(items: AsyncIterable, limit?: number): AsyncIterable { + let count = 0; + for await (const item of items) { + if (limit !== undefined && count >= limit) return; + count += 1; + yield item; + } +} + +export function getStreamingMetrics(): StreamingMetrics { + return { ...metrics }; +} + +export function resetStreamingMetrics(): void { + metrics.activeStreams = 0; + metrics.completedStreams = 0; + metrics.abortedStreams = 0; + metrics.failedStreams = 0; + metrics.rowsStreamed = 0; +} diff --git a/backend/src/routes/pool-monitor.ts b/backend/src/routes/pool-monitor.ts index 23629875..b85a2fe8 100644 --- a/backend/src/routes/pool-monitor.ts +++ b/backend/src/routes/pool-monitor.ts @@ -12,7 +12,7 @@ import { poolMetrics, getPgBouncerConfig } from '../config/database.js'; export const poolMonitorRouter = Router(); /** - * GET /api/v1/monitoring/pool/health + * GET /api/v1/monitoring/health * Get comprehensive pool health report */ poolMonitorRouter.get('/health', async (req: Request, res: Response) => { @@ -32,7 +32,7 @@ poolMonitorRouter.get('/health', async (req: Request, res: Response) => { }); /** - * GET /api/v1/monitoring/pool/metrics + * GET /api/v1/monitoring/metrics * Get detailed pool metrics */ poolMonitorRouter.get('/metrics', async (req: Request, res: Response) => { @@ -51,7 +51,7 @@ poolMonitorRouter.get('/metrics', async (req: Request, res: Response) => { }); /** - * GET /api/v1/monitoring/pool/config + * GET /api/v1/monitoring/config * Get PgBouncer configuration */ poolMonitorRouter.get('/config', async (req: Request, res: Response) => { diff --git a/backend/src/routes/streaming-export.ts b/backend/src/routes/streaming-export.ts index a2d3d871..5058196e 100644 --- a/backend/src/routes/streaming-export.ts +++ b/backend/src/routes/streaming-export.ts @@ -3,6 +3,7 @@ import { asyncHandler } from '../middleware/errorHandler.js'; import { streamingExportService } from '../services/exports/streaming-export.js'; import { auditService } from '../services/auditService.js'; import { pipeStreamToResponse } from '../middleware/stream-cancel.js'; +import { getStreamingMetrics, parseStreamingQuery } from '../middleware/streaming.js'; import type { ExportFormat } from '../services/exports/streaming-export.js'; import type { AuditEntry, AuditQuery } from '../services/auditService.js'; @@ -13,10 +14,12 @@ export const streamingExportRouter = Router(); * Stream audit log entries using cursor-based pagination. */ streamingExportRouter.get('/audit/stream', asyncHandler(async (req: Request, res: Response) => { - const format = (req.query.format as ExportFormat) || 'csv'; - const limit = req.query.limit ? Number(req.query.limit) : undefined; + const streamQuery = parseStreamingQuery(req.query); + const rawFormat = (req.query.format as ExportFormat | undefined) || 'csv'; + const format = rawFormat === 'jsonl' ? 'jsonl' : 'csv'; + const limit = streamQuery.limit; - if (format !== 'csv' && format !== 'jsonl') { + if (rawFormat !== 'csv' && rawFormat !== 'jsonl') { res.status(400).json({ error: 'Format must be csv or jsonl' }); return; } @@ -50,6 +53,7 @@ streamingExportRouter.get('/audit/stream', asyncHandler(async (req: Request, res format, headers: csvHeaders, rowLimit: limit, + chunkSize: streamQuery.batchSize, fetchPage: async (cursor, chunkLimit) => { const offset = cursor ? Number(cursor) : pageOffset; const result = await auditService.queryEntries({ @@ -103,10 +107,12 @@ streamingExportRouter.get('/audit/stream', asyncHandler(async (req: Request, res * Uses in-memory demo data; in production this would query Prisma. */ streamingExportRouter.get('/payments/stream', asyncHandler(async (req: Request, res: Response) => { - const format = (req.query.format as ExportFormat) || 'csv'; - const limit = req.query.limit ? Number(req.query.limit) : undefined; + const streamQuery = parseStreamingQuery(req.query); + const rawFormat = (req.query.format as ExportFormat | undefined) || 'csv'; + const format = rawFormat === 'jsonl' ? 'jsonl' : 'csv'; + const limit = streamQuery.limit; - if (format !== 'csv' && format !== 'jsonl') { + if (rawFormat !== 'csv' && rawFormat !== 'jsonl') { res.status(400).json({ error: 'Format must be csv or jsonl' }); return; } @@ -128,6 +134,7 @@ streamingExportRouter.get('/payments/stream', asyncHandler(async (req: Request, format, headers: csvHeaders, rowLimit: limit, + chunkSize: streamQuery.batchSize, fetchPage: async (_cursor, _chunkLimit) => { // Placeholder: in production, use Prisma cursor-based pagination: // const payments = await prisma.payment.findMany({ @@ -165,6 +172,14 @@ streamingExportRouter.get('/payments/stream', asyncHandler(async (req: Request, }); })); +/** + * GET /metrics + * Get in-process streaming response metrics. + */ +streamingExportRouter.get('/metrics', asyncHandler(async (_req: Request, res: Response) => { + res.status(200).json({ timestamp: new Date().toISOString(), data: getStreamingMetrics() }); +})); + /** * GET /:exportId/status * Get export job status and progress. diff --git a/backend/src/services/exports/streaming-export.ts b/backend/src/services/exports/streaming-export.ts index 764064b9..9f6a3841 100644 --- a/backend/src/services/exports/streaming-export.ts +++ b/backend/src/services/exports/streaming-export.ts @@ -111,9 +111,10 @@ export class StreamingExportService extends BaseService { headers?: string[]; rowSerializer: (item: T) => string; rowLimit?: number; + chunkSize?: number; }): Readable { const { exportId, fetchPage, format, headers, rowSerializer, rowLimit } = params; - const chunkSize = this.config.chunkSize; + const chunkSize = params.chunkSize ?? this.config.chunkSize; const limit = rowLimit ?? this.config.maxRowLimit; const job = this.activeJobs.get(exportId); const controller = this.abortControllers.get(exportId); diff --git a/backend/src/websocket/pool.test.ts b/backend/src/websocket/pool.test.ts new file mode 100644 index 00000000..d3ae8899 --- /dev/null +++ b/backend/src/websocket/pool.test.ts @@ -0,0 +1,101 @@ +import { EventEmitter } from 'node:events'; +import { describe, expect, it, vi } from 'vitest'; +import { WebSocketConnectionPool } from './pool'; +import type { WebSocketServerMetrics } from './types'; + +function metrics(): WebSocketServerMetrics { + return { + activeConnections: 0, + acceptedConnections: 0, + rejectedConnections: 0, + closedConnections: 0, + enqueuedMessages: 0, + droppedMessages: 0, + sentMessages: 0, + subscribedChannels: {}, + }; +} + +function socket() { + const ws = new EventEmitter() as any; + ws.OPEN = 1; + ws.readyState = 1; + ws.bufferedAmount = 0; + ws.send = vi.fn(); + ws.close = vi.fn(); + ws.terminate = vi.fn(); + return ws; +} + +describe('WebSocketConnectionPool', () => { + it('tracks accepted and closed connections', () => { + const poolMetrics = metrics(); + const pool = new WebSocketConnectionPool(poolMetrics, { + maxConnections: 2, + maxQueueSizePerConnection: 10, + maxBufferedAmountBytes: 1024, + maxBatchSize: 5, + defaultChannels: ['payment.events'], + }); + + const ws = socket(); + pool.addConnection({ ws }); + + expect(pool.size).toBe(1); + expect(poolMetrics.activeConnections).toBe(1); + expect(poolMetrics.acceptedConnections).toBe(1); + expect(poolMetrics.subscribedChannels['payment.events']).toBe(1); + + expect(pool.removeConnection(ws)).toBe(true); + expect(poolMetrics.activeConnections).toBe(0); + expect(poolMetrics.closedConnections).toBe(1); + expect(poolMetrics.subscribedChannels['payment.events']).toBeUndefined(); + }); + + it('rejects connections above capacity', () => { + const poolMetrics = metrics(); + const pool = new WebSocketConnectionPool(poolMetrics, { + maxConnections: 1, + maxQueueSizePerConnection: 10, + maxBufferedAmountBytes: 1024, + maxBatchSize: 5, + defaultChannels: [], + }); + + pool.addConnection({ ws: socket() }); + + expect(() => pool.addConnection({ ws: socket() })).toThrow('WEBSOCKET_POOL_EXHAUSTED'); + expect(poolMetrics.rejectedConnections).toBe(1); + expect(pool.snapshot().saturated).toBe(true); + }); + + it('broadcasts subscribed messages in batches', () => { + const poolMetrics = metrics(); + const pool = new WebSocketConnectionPool(poolMetrics, { + maxConnections: 2, + maxQueueSizePerConnection: 10, + maxBufferedAmountBytes: 1024, + maxBatchSize: 10, + defaultChannels: ['analytics.updates'], + }); + const wsA = socket(); + const wsB = socket(); + pool.addConnection({ ws: wsA }); + pool.addConnection({ ws: wsB }); + + const accepted = pool.broadcast({ + type: 'metric.updated', + channel: 'analytics.updates', + payload: { p95: 40 }, + }); + + expect(accepted).toBe(2); + expect(pool.snapshot().queuedMessages).toBe(2); + + pool.flushAll(); + + expect(wsA.send).toHaveBeenCalledOnce(); + expect(wsB.send).toHaveBeenCalledOnce(); + expect(poolMetrics.sentMessages).toBe(2); + }); +}); diff --git a/backend/src/websocket/pool.ts b/backend/src/websocket/pool.ts new file mode 100644 index 00000000..185aff42 --- /dev/null +++ b/backend/src/websocket/pool.ts @@ -0,0 +1,131 @@ +import type WebSocket from 'ws'; +import { ManagedConnection } from './managedConnection.js'; +import type { + WebSocketChannel, + WebSocketOutboundMessage, + WebSocketServerMetrics, +} from './types.js'; + +export interface WebSocketConnectionPoolOptions { + maxConnections: number; + maxQueueSizePerConnection: number; + maxBufferedAmountBytes: number; + maxBatchSize: number; + defaultChannels: WebSocketChannel[]; +} + +export interface AddConnectionParams { + ws: WebSocket; + authExpiresAtMs?: number; + useBinary?: boolean; +} + +export interface WebSocketConnectionPoolSnapshot { + activeConnections: number; + queuedMessages: number; + averageQueuedMessages: number; + saturated: boolean; +} + +export class WebSocketConnectionPool { + private readonly connections = new Map(); + + constructor( + private readonly metrics: WebSocketServerMetrics, + private readonly options: WebSocketConnectionPoolOptions, + ) {} + + get size(): number { + return this.connections.size; + } + + canAccept(): boolean { + return this.connections.size < this.options.maxConnections; + } + + addConnection(params: AddConnectionParams): ManagedConnection { + if (!this.canAccept()) { + this.metrics.rejectedConnections += 1; + this.metrics.lastOverloadAtMs = Date.now(); + throw new Error('WEBSOCKET_POOL_EXHAUSTED'); + } + + const managed = new ManagedConnection({ + ws: params.ws, + metrics: this.metrics, + maxQueueSize: this.options.maxQueueSizePerConnection, + maxBufferedAmountBytes: this.options.maxBufferedAmountBytes, + maxBatchSize: this.options.maxBatchSize, + defaultChannels: this.options.defaultChannels, + authExpiresAtMs: params.authExpiresAtMs, + useBinary: params.useBinary, + }); + + this.connections.set(params.ws, managed); + this.metrics.activeConnections = this.connections.size; + this.metrics.acceptedConnections += 1; + return managed; + } + + getConnection(ws: WebSocket): ManagedConnection | undefined { + return this.connections.get(ws); + } + + removeConnection(ws: WebSocket): boolean { + const managed = this.connections.get(ws); + if (!managed) return false; + + managed.close(); + this.connections.delete(ws); + this.metrics.activeConnections = this.connections.size; + this.metrics.closedConnections += 1; + return true; + } + + broadcast(message: WebSocketOutboundMessage): number { + let accepted = 0; + for (const managed of this.connections.values()) { + if (managed.enqueue(message).accepted) accepted += 1; + } + return accepted; + } + + flushAll(): void { + for (const managed of this.connections.values()) { + managed.flush(); + } + } + + closeAll(code = 1001, reason = 'Server shutting down'): void { + for (const ws of this.connections.keys()) { + try { + ws.close(code, reason); + } catch { + ws.terminate(); + } + this.removeConnection(ws); + } + } + + values(): IterableIterator { + return this.connections.values(); + } + + sockets(): IterableIterator { + return this.connections.keys(); + } + + snapshot(): WebSocketConnectionPoolSnapshot { + let queuedMessages = 0; + for (const managed of this.connections.values()) { + queuedMessages += managed.getQueuedCount(); + } + + return { + activeConnections: this.connections.size, + queuedMessages, + averageQueuedMessages: this.connections.size === 0 ? 0 : queuedMessages / this.connections.size, + saturated: !this.canAccept(), + }; + } +} diff --git a/backend/src/websocket/server.ts b/backend/src/websocket/server.ts index 3a625062..e6039e20 100644 --- a/backend/src/websocket/server.ts +++ b/backend/src/websocket/server.ts @@ -1,7 +1,6 @@ import type http from 'node:http'; import { WebSocketServer } from 'ws'; import type WebSocket from 'ws'; -import { ManagedConnection } from './managedConnection.js'; import type { WebSocketChannel, WebSocketClientMessage, @@ -10,6 +9,7 @@ import type { WebSocketServerOptions, } from './types.js'; import type { WebSocketScalingAdapter } from './scaling.js'; +import { WebSocketConnectionPool } from './pool.js'; export type AgenticPayWebSocketServer = { wss: WebSocketServer; @@ -72,7 +72,7 @@ export function attachWebSocketServer(params: { const metrics = createMetrics(); const wss = new WebSocketServer({ noServer: true }); - const connections = new Map(); + const pool = new WebSocketConnectionPool(metrics, options); const lastPongAt = new Map(); let unsubscribeScaling: (() => void) | undefined; @@ -81,7 +81,7 @@ export function attachWebSocketServer(params: { const url = new URL(req.url || '', `http://${req.headers.host || 'localhost'}`); if (url.pathname !== options.path) return; - if (metrics.activeConnections >= options.maxConnections) { + if (!pool.canAccept()) { metrics.rejectedConnections += 1; metrics.lastOverloadAtMs = Date.now(); socket.write('HTTP/1.1 503 Service Unavailable\r\nConnection: close\r\n\r\n'); @@ -98,22 +98,14 @@ export function attachWebSocketServer(params: { }); wss.on('connection', (ws: WebSocket, req) => { - metrics.activeConnections += 1; - metrics.acceptedConnections += 1; const url = new URL(req.url || '', `http://${req.headers.host || 'localhost'}`); - const managed = new ManagedConnection({ + const managed = pool.addConnection({ ws, - metrics, - maxQueueSize: options.maxQueueSizePerConnection, - maxBufferedAmountBytes: options.maxBufferedAmountBytes, - maxBatchSize: options.maxBatchSize, - defaultChannels: options.defaultChannels, authExpiresAtMs: parseAuthExpiry(url.searchParams.get('expiresAt'), options.maxAuthAgeMs), useBinary: options.enableBinaryProtocol && url.searchParams.get('proto') === '1', }); - connections.set(ws, managed); lastPongAt.set(ws, Date.now()); ws.on('pong', () => lastPongAt.set(ws, Date.now())); @@ -137,30 +129,25 @@ export function attachWebSocketServer(params: { }); ws.on('close', () => { - managed.close(); - connections.delete(ws); + pool.removeConnection(ws); lastPongAt.delete(ws); - metrics.activeConnections = Math.max(0, metrics.activeConnections - 1); - metrics.closedConnections += 1; }); }); const flushTimer = setInterval(() => { - for (const managed of connections.values()) { - managed.flush(); - } + pool.flushAll(); }, options.flushIntervalMs); const pingTimer = setInterval(() => { const now = Date.now(); - for (const ws of connections.keys()) { + for (const ws of pool.sockets()) { if (ws.readyState !== ws.OPEN) continue; const lastPong = lastPongAt.get(ws) ?? 0; if (now - lastPong > options.pingIntervalMs + options.pongTimeoutMs) { ws.terminate(); continue; } - const managed = connections.get(ws); + const managed = pool.getConnection(ws); if (managed?.isAuthExpired(now)) { managed.enqueue({ type: 'auth.expired', priority: 'high' }); ws.close(4001, 'Auth token expired'); @@ -171,9 +158,7 @@ export function attachWebSocketServer(params: { }, options.pingIntervalMs); const broadcastLocal = (message: WebSocketOutboundMessage) => { - for (const managed of connections.values()) { - managed.enqueue(message); - } + pool.broadcast(message); }; const broadcast = (message: WebSocketOutboundMessage) => { @@ -200,6 +185,7 @@ export function attachWebSocketServer(params: { clearInterval(flushTimer); clearInterval(pingTimer); unsubscribeScaling?.(); + pool.closeAll(); await new Promise((resolve) => wss.close(() => resolve())); }; diff --git a/infra/main.tf b/infra/main.tf index b6f8c383..9926300c 100644 --- a/infra/main.tf +++ b/infra/main.tf @@ -139,6 +139,30 @@ resource "aws_db_instance" "postgres" { } } +resource "aws_db_instance" "postgres_read_replica" { + count = var.db_read_replica_count + + identifier = "agenticpay-${var.environment}-ro-${count.index + 1}" + replicate_source_db = aws_db_instance.postgres.identifier + instance_class = var.db_read_replica_instance_class + + vpc_security_group_ids = [aws_security_group.rds.id] + + auto_minor_version_upgrade = true + deletion_protection = var.environment == "prod" + skip_final_snapshot = var.environment != "prod" + copy_tags_to_snapshot = true + performance_insights_enabled = var.environment == "prod" + performance_insights_kms_key_id = var.environment == "prod" ? aws_kms_key.data_at_rest.arn : null + + enabled_cloudwatch_logs_exports = ["postgresql"] + + tags = { + Name = "agenticpay-${var.environment}-read-replica-${count.index + 1}" + Role = "read-replica" + } +} + # RDS Proxy (AWS-managed PgBouncer in transaction mode) resource "aws_security_group" "rds_proxy" { name = "agenticpay-${var.environment}-rds-proxy-sg" @@ -375,8 +399,10 @@ resource "aws_apprunner_service" "backend" { STELLAR_NETWORK = var.stellar_network PGBOUNCER_ENABLED = "true" DATABASE_URL = "postgresql://${var.db_username}:${var.db_password}@${aws_db_proxy.pgbouncer.endpoint}:5432/agenticpay" - DB_POOL_MAX = var.db_proxy_pool_max - DB_POOL_MIN = var.db_proxy_pool_min + DB_READ_REPLICA_URLS = join(",", [for replica in aws_db_instance.postgres_read_replica : "postgresql://${var.db_username}:${var.db_password}@${replica.address}:5432/agenticpay"]) + DB_REPLICA_MAX_LAG_MS = tostring(var.db_replica_max_lag_ms) + DB_POOL_MAX = tostring(var.db_proxy_pool_max) + DB_POOL_MIN = tostring(var.db_proxy_pool_min) } } image_identifier = "${aws_ecr_repository.backend.repository_url}:latest" @@ -756,6 +782,25 @@ resource "aws_cloudwatch_metric_alarm" "db_connection_usage" { } } +resource "aws_cloudwatch_metric_alarm" "db_read_replica_lag" { + count = var.db_read_replica_count + + alarm_name = "agenticpay-db-replica-lag-${var.environment}-${count.index + 1}" + comparison_operator = "GreaterThanThreshold" + evaluation_periods = "2" + metric_name = "ReplicaLag" + namespace = "AWS/RDS" + period = "60" + statistic = "Average" + threshold = ceil(var.db_replica_max_lag_ms / 1000) + alarm_description = "Alert when PostgreSQL read replica lag exceeds backend routing threshold" + alarm_actions = var.environment == "prod" ? [aws_sns_topic.alerts.arn] : [] + + dimensions = { + DBInstanceIdentifier = aws_db_instance.postgres_read_replica[count.index].identifier + } +} + # SNS Topic for Alerts resource "aws_sns_topic" "alerts" { name = "agenticpay-alerts-${var.environment}" diff --git a/infra/variables.tf b/infra/variables.tf index 529d49c2..d04e1ade 100644 --- a/infra/variables.tf +++ b/infra/variables.tf @@ -71,6 +71,24 @@ variable "db_max_allocated_storage" { default = 100 } +variable "db_read_replica_count" { + description = "Number of PostgreSQL read replicas to provision for read scaling and failover" + type = number + default = 0 +} + +variable "db_read_replica_instance_class" { + description = "RDS instance class for PostgreSQL read replicas" + type = string + default = "db.t4g.medium" +} + +variable "db_replica_max_lag_ms" { + description = "Maximum tolerated read replica lag before backend read routing falls back to primary" + type = number + default = 5000 +} + # ── RDS Proxy (PgBouncer) Variables ──────────────────────────────────────────── variable "db_proxy_borrow_timeout" {