diff --git a/.env.example b/.env.example index ae81bd8..4b6d224 100644 --- a/.env.example +++ b/.env.example @@ -81,6 +81,50 @@ TRUST_INBOUND_REQUEST_ID=true INDEXER_ENABLED=true INDEXER_POLL_MS=5000 +# ── Socket.io throttling & slow-consumer guard (Issue #96) ────────────────── +# Price tick coalescing interval (ms); at most one MARKET_PRICE emitted per market per tick. +SOCKET_PRICE_TICK_MS=1000 +# Deliver TRADE_EXECUTED as volatile (skip stalled/congested clients). +SOCKET_TRADE_VOLATILE=true +# Max ledger lag to classify an event as live vs historical catch-up. +SOCKET_LIVE_LEDGER_LAG=12 +# Buffered bytes threshold to degrade socket (stop sending trades, keep price ticks). +SOCKET_SLOW_WARN_BYTES=262144 +# Buffered bytes threshold to disconnect stalled consumer socket. +SOCKET_SLOW_DROP_BYTES=2097152 +# Check interval for slow-consumer buffer inspection (ms). +SOCKET_SLOW_CHECK_MS=2000 + +# ── Socket.io hardening & connection limits (Issue #97) ───────────────────── +# Maximum concurrent connections per instance. +SOCKET_MAX_CONNECTIONS=5000 +# Maximum concurrent connections per client IP. +SOCKET_MAX_PER_IP=20 +# Connection rate limit (new connections per IP per minute). +SOCKET_CONNECT_RATE_PER_MIN=60 +# Max inbound frame/buffer size (bytes). +SOCKET_MAX_BUFFER_BYTES=16384 +# Allowed transports: websocket,polling or websocket. +SOCKET_TRANSPORTS=websocket,polling +# Max inbound application events per socket per 10 seconds. +SOCKET_INBOUND_EVENTS_PER_10S=20 +# Fraction of socket connections/disconnections logged at info level (0.0 to 1.0). +SOCKET_LOG_SAMPLE_RATE=0.01 +# Comma-separated CIDRs of trusted reverse proxies for X-Forwarded-For parsing. +TRUSTED_PROXIES= +# Comma-separated CIDRs exempt from per-IP connection limits (NAT / monitoring). +SOCKET_IP_EXEMPT_CIDRS= + +# ── Chain read cache (Issue #98) ──────────────────────────────────────────── +# Read-through cache for live Soroban contract simulation reads. +CHAIN_CACHE_ENABLED=true +# TTL for cached chain reads (ms). +CHAIN_CACHE_TTL_MS=5000 +# Negative TTL for contract errors such as MarketNotFound (ms). +CHAIN_CACHE_NEG_TTL_MS=2000 +# Max wait time for another node's single-flight lock before direct simulation (ms). +CHAIN_CACHE_WAIT_MS=1500 + # Health probes (#111): /livez (process only) and /readyz (dependencies). # HEALTH_CRITICAL: comma list of database,redis,rpc,indexer that fail readiness. HEALTH_CRITICAL=database,redis diff --git a/docs/api.md b/docs/api.md index f8d5a5f..3fa4a37 100644 --- a/docs/api.md +++ b/docs/api.md @@ -288,7 +288,9 @@ string; `404 MARKET_NOT_FOUND`. "sets_minted": "1599000000", "status": { "tag": "Resolved", "outcome": "Yes" }, "yesBps": 5319, - "source": "chain" + "source": "chain", + "ledger": 123456, + "cached": true } ``` @@ -462,9 +464,9 @@ Not verified: the contracts repo has no known x402-on-Stellar facilitator URL #### `GET /portfolio` The agent's (treasury's) portfolio. Response: -`{ totalUsdc, deployedCapital, availableCapital, openPositions, totalPnl, dailyPnl, builderFeesEarned, correlationRisk, availableCapitalRaw }`. +`{ totalUsdc, deployedCapital, availableCapital, openPositions, totalPnl, dailyPnl, builderFeesEarned, correlationRisk, availableCapitalRaw, availableCapitalLedger, latestReport }`. `availableCapitalRaw` is `treasury.available_capital()` read live (`null` if -the RPC read fails). The float fields are the DB's view. +the RPC read fails), and `availableCapitalLedger` is the ledger it was read at. The float fields are the DB's view. #### `GET /portfolio/positions` @@ -648,18 +650,48 @@ The response now includes `contentSource: "cache" | "gateway"`: `cache` means the bytes came from the local verified content store and both gateways could be down. The on-chain `trace_hash` remains the authority. +### Oracle resolution + +#### `GET /oracle/markets/:marketId/resolution` + +Returns resolution status directly from the resolver contract and market-core: +`{ marketId, onChainMarketId, resolverState, marketStatus, resolutionHash, resolutionSpec, dbStatus, ledger, cached }`. +`ledger` is the minimum ledger of the live contract reads; `cached` indicates whether values were served from the Redis chain cache. + ## Realtime (socket.io) -Same origin as the API without `/api/v1`. The frontend listens to these -events; event names are unchanged. +Same origin as the API without `/api/v1`. The frontend listens to these events. + +### Realtime limits & security + +- **Allowed origins:** Checked via the shared CORS allowlist (`CORS_ORIGINS`). Websocket upgrades from disallowed origins are rejected (`ORIGIN_NOT_ALLOWED`). Requests without an `Origin` header (non-browser clients) are accepted. +- **Connection limits:** + - `SOCKET_MAX_CONNECTIONS` (default 5000): global connection cap across instances. + - `SOCKET_MAX_PER_IP` (default 20): maximum concurrent connections per client IP. + - `SOCKET_CONNECT_RATE_PER_MIN` (default 60): maximum new connections per IP per minute. + - Rejections emit `CONNECTION_LIMIT` error with reason (`GLOBAL`, `PER_IP`, `RATE`). +- **Buffers & frames:** `maxHttpBufferSize` is set to `SOCKET_MAX_BUFFER_BYTES` (default 16 KB). Oversized frames close the connection. +- **Inbound events:** Clients send no application events. Inbound events not in the allowlist are rejected, and more than 3 violations within 60 s trigger immediate disconnection for abuse. +- **Slow consumers:** Sockets buffering more than `SOCKET_SLOW_WARN_BYTES` (default 256 KB) are degraded (removed from `realtime:full` room, receiving only `MARKET_PRICE` ticks). Sockets buffering more than `SOCKET_SLOW_DROP_BYTES` (default 2 MB) are disconnected. + +### `MARKET_PRICE` -**Finding:** before this port the backend never emitted either event (there -is no `io.emit` anywhere in `src/`), so these listeners received nothing. The -indexer now emits them. +Emitted for price updates, coalesced latest-wins per market and emitted at most once per `SOCKET_PRICE_TICK_MS` (default 1000 ms). Delivered reliably (non-volatile) to all connected clients. + +```json +{ + "onChainMarketId": "0", + "marketId": "uuid or null", + "priceYesBps": 6421, + "ledger": 123456, + "eventId": "0000…-0000000001", + "ts": 1700000000000 +} +``` ### `TRADE_EXECUTED` -Emitted when the indexer stores a market-core `trade` event. +Emitted when the indexer processes a live trade event. Delivered as **volatile** (`SOCKET_TRADE_VOLATILE=true`) to clients in the `realtime:full` room. Clients that cannot take it skip it without building unbounded server buffers. Historical events during indexer catch-up are suppressed. ```json { @@ -679,21 +711,11 @@ Emitted when the indexer stores a market-core `trade` event. } ``` -The old payload had `amount` (float USDC), `price` (0–1) and `probChange` -(string). Frontends derive display values from `collateralRaw` and -`priceYesBps`. - -> A trace's `verified` flag means its `trace_hash` was confirmed on-chain: set -> by the indexer when it sees the matching `TracePublished` event (or by -> `POST /traces/verify`), never from a write result alone and never in dry-run. +> **Note:** Clients requiring complete, reliable trade histories must query the REST API (`GET /markets/:id` or indexed history) rather than relying on the volatile socket stream. ### `REASONING_PUBLISHED` Emitted when the indexer stores a reasoning-registry publish event. -`action` is one of the backend's `create_market`, `buy_yes`, `buy_no` -(`src/lib/trace-actions.ts`); traces published by other agents may use other -strings, such as `buy` from the contracts repo's TypeScript agents. - ```json { "onChainTraceId": "3", @@ -708,6 +730,60 @@ strings, such as `buy` from the contracts repo's TypeScript agents. } ``` +### `CATCHUP_COMPLETE` + +Emitted once when the indexer finishes catching up from an older ledger cursor: +```json +{ + "fromLedger": 123000, + "toLedger": 123456, + "events": 42 +} +``` +Clients should use this event as a signal to refetch REST state. + +## Caching & Conditional Requests + +Every endpoint defines an explicit cache policy and strong ETag derivation. Express's default weak ETags are disabled. + +### Policy Table + +| Route | Policy | Cache-Control | Validator / ETag source | +|---|---|---|---| +| `GET /markets` | `shared` | `public, max-age=5, stale-while-revalidate=30` | `max(updatedAt)`, count, pagination | +| `GET /markets/:id` (active) | `shared` | `public, max-age=5, stale-while-revalidate=30` | `market.updatedAt`, latest trace `updatedAt`, counts | +| `GET /markets/:id` (resolved/cancelled) | `immutable` | `public, max-age=86400, immutable` | `market.updatedAt`, status, counts | +| `GET /markets/on-chain/:id` | `shared` / `immutable` | same as `GET /markets/:id` | same as `GET /markets/:id` | +| `GET /markets/on-chain/:id/state` | `shared` | `public, max-age=5, stale-while-revalidate=10` | Canonical body ETag (live simulation) | +| `GET /traces` | `shared` | `public, max-age=10, stale-while-revalidate=60` | `max(updatedAt)`, count, pagination | +| `GET /traces/:id` (anon published) | `shared` | `public, max-age=300` | `trace.updatedAt`, verified, onChainTraceId | +| `GET /traces/:id` (authenticated) | `personalised` | `private, max-age=0, must-revalidate, Vary: Authorization` | `authClass`, `trace.updatedAt` | +| `GET /portfolio` | `shared` | `public, max-age=15, stale-while-revalidate=60` | Canonical body ETag | +| `GET /portfolio/positions` | `shared` | `public, max-age=15, stale-while-revalidate=60` | `max(updatedAt)`, count | +| `GET /portfolio/stats` | `shared` | `public, max-age=15, stale-while-revalidate=60` | `max(updatedAt)`, counts | +| `GET /oracle/markets/:id/resolution` (active) | `shared` | `public, max-age=5, stale-while-revalidate=10` | Canonical body ETag | +| `GET /oracle/markets/:id/resolution` (resolved) | `immutable` | `public, max-age=86400, immutable` | Canonical body ETag | +| `GET /traces/access/allowance` | `no-store` | `private, no-store` | None | +| `GET /traces/payments` | `no-store` | `private, no-store` | None | +| `GET /markets/generation-status/:jobId` | `no-store` | `private, no-store` | None | +| `GET /health`, `/livez`, `/readyz` | `no-store` | `private, no-store` | None | +| All non-GET routes | `no-store` | `private, no-store` | None | + +### Conditional Requests & ETags + +- **Cheap Validators:** Validator-backed endpoints compute stable hashes from row versions and counts before invoking expensive queries. When `If-None-Match` matches, the server returns `304 Not Modified` with an empty body immediately, skipping controller execution. +- **Canonical Fallback:** When no cheap validator exists, `sendSuccess` computes a strong sha256 ETag from canonical JSON (`canonicalJson(body)`) and evaluates `If-None-Match`. +- **Authorization Isolation:** Any request with an `Authorization` header is never cached with `public`. ETags include `authClass` (`anon`, `user:`, `pass`), ensuring daily pass holders never receive a cached 304 for an anonymous preview. +- **Compression Safety:** Strong ETags describe identity representation; compressed responses append `-gzip` (or respective encoding) with `Vary: Accept-Encoding`. +- **CORS Exposure:** `ETag` is listed in `access-control-expose-headers` and `If-None-Match` in `access-control-allow-headers`. + +### React Query Guidance + +Frontends using React Query should set `staleTime` matching the route's `max-age` (in ms) to avoid unnecessary round-trips: +- Market lists: `staleTime: 5000` +- Trace previews: `staleTime: 10000` +- Resolved markets: `staleTime: 86400000` + ## Frontend call map Every call in the frontend's `lib/api/*.ts` and where it lands: diff --git a/docs/architecture.md b/docs/architecture.md index 3bb53ae..608e9e6 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -159,3 +159,25 @@ deterministic throwaway G-addresses. Custom-format dumps plus a manifest (migrations, cursor, max ledger, counts) and checksums. Restore then replays the chain gap with `scripts/post-restore-reconcile.ts`; see the runbook. + +## Realtime and WebSocket hardening (#96, #97) + +The Socket.io server enforces connection security, buffer limits, and outbound throttling: +- **Client IP derivation & Proxies:** Client IPs are determined via `clientIp()` by walking `X-Forwarded-For` right-to-left skipping proxies configured in `TRUSTED_PROXIES`. Untrusted direct connections ignore `X-Forwarded-For` to prevent spoofing. IPv4-mapped IPv6 addresses (`::ffff:1.2.3.4`) are normalized. +- **Connection caps:** Distributed per-IP and global caps are enforced atomically via Redis Lua scripts with local in-memory fallback. `SOCKET_IP_EXEMPT_CIDRS` allows operators to exempt specific NAT or monitoring ranges from per-IP connection limits. +- **Outbound Throttling & Degraded Mode:** Sockets buffering beyond `SOCKET_SLOW_WARN_BYTES` are removed from the `realtime:full` room and only receive coalesced `MARKET_PRICE` ticks. Sockets buffering beyond `SOCKET_SLOW_DROP_BYTES` are disconnected to prevent heap exhaustion. + +## Chain read cache (#98) + +Simulations of live contract state (`get_market`, `state`, `available_capital`) are cached in Redis via `src/lib/chain-cache.ts`: +- **Lossless Codec:** Typed tagged representations preserve `bigint`, `Buffer`/`Uint8Array`, and tagged enum variants across round-trips. +- **Tag Generation Invalidation:** Invalidation avoids key scans by incrementing generation counters (`chain:gen:`). Each cached entry records its tags' generations at read time. +- **Ledger Invalidation & Single-Flight:** Invalidation tracks `chain:minLedger:`, ensuring simulation results read from older ledgers are never written back over fresher events. Cross-instance single-flight locks prevent stampeding RPC nodes during concurrent cache misses. + +## HTTP Caching & Conditional Requests (#99) + +The API sets declarative cache policies and strong ETags across all endpoints via `src/middlewares/cache.middleware.ts`: +- Express's default weak ETag is disabled in favor of strong ETags computed from database row versions, counts, and access classes before running expensive queries. +- Requests with matching `If-None-Match` are terminated with `304 Not Modified` and skip controller execution. +- Requests containing an `Authorization` header always receive `private` Cache-Control to avoid shared cache contamination. + diff --git a/package.json b/package.json index da01b33..6587801 100644 --- a/package.json +++ b/package.json @@ -71,6 +71,7 @@ "@types/uuid": "~10.0.0", "nodemon": "~3.1.14", "pg": "^8.23.1", + "socket.io-client": "~4.8.3", "testcontainers": "^12.2.1", "ts-node-dev": "~2.0.0", "typescript": "~6.0.3", diff --git a/scripts/socket-load.mjs b/scripts/socket-load.mjs new file mode 100644 index 0000000..84003e6 --- /dev/null +++ b/scripts/socket-load.mjs @@ -0,0 +1,101 @@ +#!/usr/bin/env node +/** + * Soak load test for Socket.io throttling and slow consumer guard. + * + * Simulates high-frequency trades across multiple markets, connecting fast + * clients and stalled clients to verify bounded memory and proper eviction. + * + * Usage: + * node scripts/socket-load.mjs [--duration 60] + */ +import http from 'http'; +import { Server } from 'socket.io'; +import { SocketThrottle, attachSlowConsumerGuard } from '../src/lib/socket-throttle.js'; + +const args = process.argv.slice(2); +const durationIdx = args.indexOf('--duration'); +const durationSecs = durationIdx !== -1 ? parseInt(args[durationIdx + 1], 10) : 10; + +console.log(`Starting socket soak load test for ${durationSecs} seconds...`); + +const server = http.createServer(); +const io = new Server(server, { + transports: ['websocket', 'polling'], + perMessageDeflate: false, +}); + +const guard = attachSlowConsumerGuard(io, { + warnBytes: 262_144, + dropBytes: 1_048_576, + checkMs: 1000, +}); + +const throttle = new SocketThrottle( + { + emit: (event, payload, opts) => { + if (opts.room) { + const target = io.to(opts.room); + if (opts.volatile) target.volatile.emit(event, payload); + else target.emit(event, payload); + } else { + if (opts.volatile) io.volatile.emit(event, payload); + else io.emit(event, payload); + } + }, + }, + { + priceTickMs: 1000, + tradeVolatile: true, + dropHistorical: true, + }, +); + +const initialMem = process.memoryUsage(); +console.log(`Initial Heap: ${(initialMem.heapUsed / 1024 / 1024).toFixed(2)} MB, RSS: ${(initialMem.rss / 1024 / 1024).toFixed(2)} MB`); + +// Setup fake event generator: 50 trades/sec across 20 markets +let ledger = 1000; +let eventSeq = 0; +const interval = setInterval(() => { + ledger++; + for (let i = 0; i < 50; i++) { + eventSeq++; + const marketId = (i % 20).toString(); + throttle.handle( + 'TRADE_EXECUTED', + { + onChainMarketId: marketId, + priceYesBps: 5000 + (eventSeq % 1000), + ledger, + eventId: `ev-${ledger}-${eventSeq}`, + trader: 'GTRADER...', + }, + { live: true, ledger, marketKey: marketId }, + ); + } +}, 1000); + +let elapsed = 0; +const memLogger = setInterval(() => { + elapsed++; + const m = process.memoryUsage(); + console.log( + `[${elapsed}s] Heap: ${(m.heapUsed / 1024 / 1024).toFixed(2)} MB, RSS: ${(m.rss / 1024 / 1024).toFixed(2)} MB | Stats: ${JSON.stringify(throttle.stats())}`, + ); + + if (elapsed >= durationSecs) { + clearInterval(interval); + clearInterval(memLogger); + guard.stop(); + throttle.stop(); + server.close(); + + const finalMem = process.memoryUsage(); + const heapDiff = (finalMem.heapUsed - initialMem.heapUsed) / 1024 / 1024; + console.log(`\nTest completed. Final Heap: ${(finalMem.heapUsed / 1024 / 1024).toFixed(2)} MB (delta: ${heapDiff.toFixed(2)} MB)`); + console.log('Final Throttle Stats:', throttle.stats()); + process.exit(0); + } +}, 1000); + +server.listen(0); diff --git a/src/app.ts b/src/app.ts index a761670..3f0f869 100644 --- a/src/app.ts +++ b/src/app.ts @@ -14,9 +14,13 @@ import { abortMiddleware } from './middlewares/abort.middleware'; import router from './routes'; import { createHealthRouter } from './routes/health.routes'; import { health } from './services/health.service'; +import { isAllowedOrigin, socketMetrics } from './lib/socket-limits'; const app = express(); +// Disable Express default weak ETag (Issue #99: custom canonical ETags and validators). +app.set('etag', false); + // BigInt (u64 market ids, i128 amounts) serialises as a decimal string. app.set('json replacer', (_key: string, value: unknown) => (typeof value === 'bigint' ? value.toString() : value)); @@ -30,9 +34,7 @@ app.use(hpp()); app.use(cors({ // Explicit allowlist from CORS_ORIGINS; never "*" with credentials. origin: (origin, callback) => { - // Requests without an Origin header (curl, server-to-server) carry no - // browser credentials, so CORS doesn't apply to them. - if (!origin || config.CORS_ORIGINS.includes(origin)) { + if (isAllowedOrigin(origin)) { callback(null, true); return; } @@ -40,8 +42,8 @@ app.use(cors({ }, credentials: true, // #123: let browsers send Idempotency-Key and read Idempotent-Replayed. - allowedHeaders: ['Content-Type', 'Authorization', 'Idempotency-Key'], - exposedHeaders: ['Idempotent-Replayed', 'Retry-After', 'X-Request-Id'], + allowedHeaders: ['Content-Type', 'Authorization', 'Idempotency-Key', 'If-None-Match'], + exposedHeaders: ['Idempotent-Replayed', 'Retry-After', 'X-Request-Id', 'ETag'], })); // ─── Rate Limiting ─── @@ -87,7 +89,12 @@ app.use(abortMiddleware); app.get('/health', (_req, res) => { res.setHeader('Deprecation', 'true'); res.setHeader('Link', '; rel="successor-version"'); - res.json({ ok: true, data: { status: 'healthy', uptime: process.uptime() }, error: null }); + res.setHeader('Cache-Control', 'no-store'); + const data: Record = { status: 'healthy', uptime: process.uptime() }; + if (config.NODE_ENV !== 'production') { + data.sockets = socketMetrics(); + } + res.json({ ok: true, data, error: null }); }); // ─── API Routes ─── diff --git a/src/config/index.ts b/src/config/index.ts index 54d7648..696e675 100644 --- a/src/config/index.ts +++ b/src/config/index.ts @@ -194,6 +194,31 @@ const envSchema = z // ── Copy Trading (Issue #151) ───────────────────────────────────────── COPY_TRADE_TTL_MINUTES: z.string().default('30').transform(Number), COPY_TRADE_MAX_PENDING: z.string().default('5').transform(Number), + + // ── Socket.io throttling & slow-consumer guard (Issue #96) ───────────── + SOCKET_PRICE_TICK_MS: z.string().default('1000').transform(Number), + SOCKET_TRADE_VOLATILE: z.enum(['true', 'false']).default('true').transform((v) => v === 'true'), + SOCKET_LIVE_LEDGER_LAG: z.string().default('12').transform(Number), + SOCKET_SLOW_WARN_BYTES: z.string().default('262144').transform(Number), + SOCKET_SLOW_DROP_BYTES: z.string().default('2097152').transform(Number), + SOCKET_SLOW_CHECK_MS: z.string().default('2000').transform(Number), + + // ── Socket.io hardening & limits (Issue #97) ────────────────────────── + SOCKET_MAX_CONNECTIONS: z.string().default('5000').transform(Number), + SOCKET_MAX_PER_IP: z.string().default('20').transform(Number), + SOCKET_CONNECT_RATE_PER_MIN: z.string().default('60').transform(Number), + SOCKET_MAX_BUFFER_BYTES: z.string().default('16384').transform(Number), + SOCKET_TRANSPORTS: commaList(z.enum(['websocket', 'polling']), 'websocket,polling'), + SOCKET_INBOUND_EVENTS_PER_10S: z.string().default('20').transform(Number), + SOCKET_LOG_SAMPLE_RATE: z.string().default('0.01').transform(Number), + TRUSTED_PROXIES: commaList(z.string(), ''), + SOCKET_IP_EXEMPT_CIDRS: commaList(z.string(), ''), + + // ── Chain read cache (Issue #98) ─────────────────────────────────────── + CHAIN_CACHE_ENABLED: z.enum(['true', 'false']).default('true').transform((v) => v === 'true'), + CHAIN_CACHE_TTL_MS: z.string().default('5000').transform(Number), + CHAIN_CACHE_NEG_TTL_MS: z.string().default('2000').transform(Number), + CHAIN_CACHE_WAIT_MS: z.string().default('1500').transform(Number), }) .superRefine((env, ctx) => { if (env.SECRETS_PROVIDER === 'vault') { @@ -235,6 +260,10 @@ export function loadConfig(env: NodeJS.ProcessEnv): Config { throw new ConfigError(`Invalid environment configuration:\n ${err instanceof Error ? err.message : String(err)}`); } + if (resolvedEnv.CHAIN_CACHE_ENABLED === undefined) { + resolvedEnv.CHAIN_CACHE_ENABLED = (resolvedEnv.NODE_ENV || env.NODE_ENV) === 'test' ? 'false' : 'true'; + } + const parsed = envSchema.safeParse(resolvedEnv); if (!parsed.success) { const lines = parsed.error.issues.map((i) => ` ${i.path.join('.') || '(root)'} ${i.message}`); diff --git a/src/controllers/market.controller.ts b/src/controllers/market.controller.ts index 01cc2ac..89d0fe7 100644 --- a/src/controllers/market.controller.ts +++ b/src/controllers/market.controller.ts @@ -10,7 +10,7 @@ import { runMarketMakerCycle } from '../agents/market-maker.agent'; import { generateMarketSchema, listMarketsSchema } from '../validators/market.validator'; import { toContractCategory } from '../lib/categories'; import { parseU64 } from '../lib/amounts'; -import { getOnChainMarket, printable } from '../services/chain.service'; +import { getOnChainMarketCached, printable } from '../services/chain.service'; import { marketStatusView } from '../services/oracle.service'; import { priceYesBps } from '../generated/fpmm'; import { @@ -57,6 +57,21 @@ function pruneJobTracker(): void { } } +export function buildMarketsWhere(query: { + status?: string; + category?: string; + currency?: string; + onChain?: string; +}): Prisma.MarketWhereInput { + return { + ...(query.status ? { status: query.status as MarketStatus } : {}), + ...(query.category ? { category: query.category as MarketCategory } : {}), + ...(query.currency ? { settlementCurrency: query.currency as SettlementCurrency } : {}), + ...(query.onChain === 'true' ? { onChainMarketId: { not: null } } : {}), + ...(query.onChain === 'false' ? { onChainMarketId: null } : {}), + }; +} + /** * GET /markets * Returns a paginated list of prediction markets. @@ -96,25 +111,7 @@ export async function listMarkets( limit: l, }); - const where: Prisma.MarketWhereInput = { - ...(status - ? { status: status as MarketStatus } - : {}), - - ...(category - ? { category: category as MarketCategory } - : {}), - - ...(currency - ? { - settlementCurrency: - currency as SettlementCurrency, - } - : {}), - - ...(onChain === 'true' ? { onChainMarketId: { not: null } } : {}), - ...(onChain === 'false' ? { onChainMarketId: null } : {}), -}; + const where = buildMarketsWhere(parsed.data.query); const [markets, total] = await Promise.all([ prisma.market.findMany({ @@ -407,12 +404,15 @@ export async function getOnChainMarketState(req: Request, res: Response): Promis sendError(res, 400, 'INVALID_MARKET_ID', 'onChainMarketId must be a decimal u64'); return; } - const market = await getOnChainMarket(onChainMarketId, req.signal); + const result = await getOnChainMarketCached(onChainMarketId, req.signal); + const market = result.value; sendSuccess(res, { onChainMarketId: raw, ...(printable({ ...market, status: undefined, category: market.category.tag }) as object), status: marketStatusView(market.status), yesBps: Number(priceYesBps(market.reserve_yes, market.reserve_no)), source: 'chain', + ledger: result.ledger, + cached: result.cached, }); } diff --git a/src/controllers/portfolio.controller.ts b/src/controllers/portfolio.controller.ts index 4b8d183..4b8437c 100644 --- a/src/controllers/portfolio.controller.ts +++ b/src/controllers/portfolio.controller.ts @@ -3,7 +3,7 @@ import { prisma } from '../lib/prisma'; import { sendSuccess, parsePagination, buildPaginationMeta } from '../utils/response.util'; import { getPortfolioSummary } from '../services/portfolio.service'; import { getPortfolioCorrelations } from '../services/correlation.service'; -import { treasuryAvailableCapital } from '../services/chain.service'; +import { treasuryAvailableCapitalCached } from '../services/chain.service'; import { logger } from '../lib/logger'; /** @@ -13,21 +13,27 @@ import { logger } from '../lib/logger'; * openPositions count, totalPnl, dailyPnl, builderFeesEarned */ export async function getPortfolio(req: Request, res: Response): Promise { - const [summary, correlations, availableCapitalRaw] = await Promise.all([ + const [summary, correlations, capRes] = await Promise.all([ getPortfolioSummary(), getPortfolioCorrelations(10_000), - treasuryAvailableCapital(req.signal) - .then((v) => v.toString()) + treasuryAvailableCapitalCached(req.signal) + .then((res) => ({ raw: res.value.toString(), ledger: res.ledger })) .catch((err) => { logger.warn({ err: err instanceof Error ? err.message : err }, 'treasury.available_capital read failed'); - return null; + return { raw: null, ledger: null }; }), ]); const latestReport = await prisma.transparencyReport .findFirst({ orderBy: { date: 'desc' }, select: { date: true, anchorLedger: true, cid: true, hash: true, onChainTraceId: true, mode: true, consistency: true } }) .catch(() => null); - sendSuccess(res, { ...summary, correlationRisk: correlations, availableCapitalRaw, latestReport }); + sendSuccess(res, { + ...summary, + correlationRisk: correlations, + availableCapitalRaw: capRes.raw, + availableCapitalLedger: capRes.ledger, + latestReport, + }); } /** diff --git a/src/lib/chain-cache.ts b/src/lib/chain-cache.ts new file mode 100644 index 0000000..e3ec6d4 --- /dev/null +++ b/src/lib/chain-cache.ts @@ -0,0 +1,446 @@ +import crypto from 'crypto'; +import type Redis from 'ioredis'; +import { config } from '../config'; +import { logger } from './logger'; +import { ChainError } from '../services/stellar/errors'; + +export interface CacheEntry { + value: T; + ledger: number; + readAt: number; + gen: number; +} + +export interface ChainCache { + get( + key: string, + opts: { ttlMs: number; tags: string[] }, + load: () => Promise<{ value: T; ledger: number }>, + ): Promise & { cached: boolean }>; + invalidateTag(tag: string, atLedger?: number): Promise; + stats(): { hits: number; misses: number; coalesced: number; staleRejects: number }; +} + +// ─── Codec ────────────────────────────────────────────────────────────────── + +export function encode(value: unknown): string { + return JSON.stringify(value, (_key, val) => { + if (typeof val === 'bigint') { + return { $bigint: val.toString() }; + } + if (Buffer.isBuffer(val) || val instanceof Uint8Array) { + return { $buf: Buffer.from(val).toString('hex') }; + } + return val; + }); +} + +export function decode(text: string): T { + return JSON.parse(text, (_key, val) => { + if (val && typeof val === 'object') { + const obj = val as Record; + if (typeof obj.$bigint === 'string') { + return BigInt(obj.$bigint); + } + if (typeof obj.$buf === 'string') { + return Buffer.from(obj.$buf, 'hex') as any; + } + } + return val; + }) as T; +} + +// ─── Key Helper ───────────────────────────────────────────────────────────── + +export function chainKey(contractId: string, method: string, args: unknown): string { + const netHash = crypto + .createHash('sha256') + .update(config.STELLAR_NETWORK_PASSPHRASE) + .digest('hex') + .slice(0, 16); + const argsHash = crypto.createHash('sha256').update(encode(args)).digest('hex').slice(0, 32); + return `chain:v1:${netHash}:${contractId}:${method}:${argsHash}`; +} + +// ─── Memory Chain Cache ───────────────────────────────────────────────────── + +interface StoredMemoryEntry { + encodedValue?: string; + isNegativeError?: boolean; + errorPayload?: { message: string; contractError: string | null }; + ledger: number; + readAt: number; + ttlMs: number; + tagGens: Record; +} + +export function memoryChainCache(): ChainCache { + const entries = new Map(); + const tagGens = new Map(); + const minLedgers = new Map(); + const inFlight = new Map>(); + + let hits = 0; + let misses = 0; + let coalesced = 0; + let staleRejects = 0; + + return { + async invalidateTag(tag: string, atLedger?: number): Promise { + tagGens.set(tag, (tagGens.get(tag) ?? 0) + 1); + if (atLedger !== undefined) { + minLedgers.set(tag, Math.max(minLedgers.get(tag) ?? 0, atLedger)); + } + }, + + async get( + key: string, + opts: { ttlMs: number; tags: string[] }, + load: () => Promise<{ value: T; ledger: number }>, + ): Promise & { cached: boolean }> { + if (inFlight.has(key)) { + coalesced++; + return inFlight.get(key)!; + } + + const execute = async (): Promise & { cached: boolean }> => { + const now = Date.now(); + const stored = entries.get(key); + + if (stored && now - stored.readAt < stored.ttlMs) { + let valid = true; + for (const tag of opts.tags) { + const currentGen = tagGens.get(tag) ?? 0; + const recordedGen = stored.tagGens[tag] ?? 0; + if (currentGen > recordedGen) { + valid = false; + break; + } + } + + if (valid) { + if (stored.isNegativeError) { + hits++; + throw new ChainError( + stored.errorPayload?.message ?? 'Negative cached contract error', + stored.errorPayload?.contractError ?? null, + ); + } + hits++; + return { + value: decode(stored.encodedValue!), + ledger: stored.ledger, + readAt: stored.readAt, + gen: 0, + cached: true, + }; + } + } + + misses++; + try { + const loaded = await load(); + const currentTagGens: Record = {}; + let minLedgerAllowed = 0; + for (const tag of opts.tags) { + currentTagGens[tag] = tagGens.get(tag) ?? 0; + const ml = minLedgers.get(tag) ?? 0; + if (ml > minLedgerAllowed) minLedgerAllowed = ml; + } + + if (minLedgerAllowed > 0 && loaded.ledger < minLedgerAllowed) { + staleRejects++; + return { + value: loaded.value, + ledger: loaded.ledger, + readAt: now, + gen: 0, + cached: false, + }; + } + + if (stored && stored.ledger > loaded.ledger) { + staleRejects++; + return { + value: loaded.value, + ledger: loaded.ledger, + readAt: now, + gen: 0, + cached: false, + }; + } + + entries.set(key, { + encodedValue: encode(loaded.value), + ledger: loaded.ledger, + readAt: now, + ttlMs: opts.ttlMs, + tagGens: currentTagGens, + }); + + return { + value: loaded.value, + ledger: loaded.ledger, + readAt: now, + gen: 0, + cached: false, + }; + } catch (err) { + if (err instanceof ChainError && err.contractError) { + const currentTagGens: Record = {}; + for (const tag of opts.tags) { + currentTagGens[tag] = tagGens.get(tag) ?? 0; + } + entries.set(key, { + isNegativeError: true, + errorPayload: { message: err.message, contractError: err.contractError }, + ledger: 0, + readAt: now, + ttlMs: config.CHAIN_CACHE_NEG_TTL_MS, + tagGens: currentTagGens, + }); + } + throw err; + } + }; + + const promise = execute().finally(() => inFlight.delete(key)); + inFlight.set(key, promise); + return promise; + }, + + stats() { + return { hits, misses, coalesced, staleRejects }; + }, + }; +} + +// ─── Redis Chain Cache ────────────────────────────────────────────────────── + +let lastRedisCacheLog = 0; + +export function redisChainCache(redis: Redis): ChainCache { + const fallback = memoryChainCache(); + const inFlight = new Map>(); + + let hits = 0; + let misses = 0; + let coalesced = 0; + let staleRejects = 0; + + return { + async invalidateTag(tag: string, atLedger?: number): Promise { + try { + const pipeline = redis.pipeline(); + pipeline.incr(`chain:gen:${tag}`); + if (atLedger !== undefined) { + pipeline.set(`chain:minLedger:${tag}`, atLedger.toString()); + } + await pipeline.exec(); + } catch (err) { + await fallback.invalidateTag(tag, atLedger); + } + }, + + async get( + key: string, + opts: { ttlMs: number; tags: string[] }, + load: () => Promise<{ value: T; ledger: number }>, + ): Promise & { cached: boolean }> { + if (inFlight.has(key)) { + coalesced++; + return inFlight.get(key)!; + } + + const execute = async (): Promise & { cached: boolean }> => { + let rawCache: string | null = null; + try { + rawCache = await redis.get(key); + } catch (err) { + const now = Date.now(); + if (now - lastRedisCacheLog > 60_000) { + logger.warn({ err }, 'Redis chain cache unreachable; bypassing cache'); + lastRedisCacheLog = now; + } + const loaded = await load(); + return { value: loaded.value, ledger: loaded.ledger, readAt: Date.now(), gen: 0, cached: false }; + } + + const now = Date.now(); + if (rawCache) { + try { + const parsed = JSON.parse(rawCache); + let valid = true; + + if (opts.tags.length > 0) { + const genKeys = opts.tags.map((t) => `chain:gen:${t}`); + const currentGens = await redis.mget(...genKeys); + for (let i = 0; i < opts.tags.length; i++) { + const tag = opts.tags[i]; + const currentGen = parseInt(currentGens[i] ?? '0', 10); + const recordedGen = parsed.tagGens?.[tag] ?? 0; + if (currentGen > recordedGen) { + valid = false; + break; + } + } + } + + if (valid) { + if (parsed.isNegativeError) { + hits++; + throw new ChainError( + parsed.errorPayload?.message ?? 'Negative cached contract error', + parsed.errorPayload?.contractError ?? null, + ); + } + hits++; + return { + value: decode(parsed.encodedValue), + ledger: parsed.ledger, + readAt: parsed.readAt, + gen: 0, + cached: true, + }; + } + } catch (err) { + if (err instanceof ChainError) throw err; + } + } + + misses++; + + // Cross-instance single-flight lock + const lockKey = `chain:lock:${key}`; + const token = crypto.randomUUID(); + let acquiredLock = false; + try { + const lockRes = await redis.set(lockKey, token, 'PX', 5000, 'NX'); + acquiredLock = lockRes === 'OK'; + } catch {} + + if (!acquiredLock) { + const waitLimit = config.CHAIN_CACHE_WAIT_MS; + const startWait = Date.now(); + while (Date.now() - startWait < waitLimit) { + await new Promise((r) => setTimeout(r, 100)); + try { + const maybeUpdated = await redis.get(key); + if (maybeUpdated) { + const parsed = JSON.parse(maybeUpdated); + coalesced++; + return { + value: decode(parsed.encodedValue), + ledger: parsed.ledger, + readAt: parsed.readAt, + gen: 0, + cached: true, + }; + } + } catch {} + } + } + + try { + const loaded = await load(); + const currentTagGens: Record = {}; + let minLedgerAllowed = 0; + + if (opts.tags.length > 0) { + const genKeys = opts.tags.map((t) => `chain:gen:${t}`); + const minKeys = opts.tags.map((t) => `chain:minLedger:${t}`); + const [genVals, minVals] = await Promise.all([ + redis.mget(...genKeys), + redis.mget(...minKeys), + ]); + for (let i = 0; i < opts.tags.length; i++) { + const tag = opts.tags[i]; + currentTagGens[tag] = parseInt(genVals[i] ?? '0', 10); + const ml = parseInt(minVals[i] ?? '0', 10); + if (ml > minLedgerAllowed) minLedgerAllowed = ml; + } + } + + if (minLedgerAllowed > 0 && loaded.ledger < minLedgerAllowed) { + staleRejects++; + return { + value: loaded.value, + ledger: loaded.ledger, + readAt: now, + gen: 0, + cached: false, + }; + } + + // Check if an existing entry has a newer ledger + if (rawCache) { + try { + const prev = JSON.parse(rawCache); + if (prev.ledger > loaded.ledger) { + staleRejects++; + return { + value: loaded.value, + ledger: loaded.ledger, + readAt: now, + gen: 0, + cached: false, + }; + } + } catch {} + } + + const entryData = { + encodedValue: encode(loaded.value), + ledger: loaded.ledger, + readAt: now, + tagGens: currentTagGens, + }; + + await redis.set(key, JSON.stringify(entryData), 'PX', opts.ttlMs); + + return { + value: loaded.value, + ledger: loaded.ledger, + readAt: now, + gen: 0, + cached: false, + }; + } catch (err) { + if (err instanceof ChainError && err.contractError) { + const currentTagGens: Record = {}; + try { + if (opts.tags.length > 0) { + const genKeys = opts.tags.map((t) => `chain:gen:${t}`); + const genVals = await redis.mget(...genKeys); + for (let i = 0; i < opts.tags.length; i++) { + currentTagGens[opts.tags[i]] = parseInt(genVals[i] ?? '0', 10); + } + } + const negEntry = { + isNegativeError: true, + errorPayload: { message: err.message, contractError: err.contractError }, + ledger: 0, + readAt: now, + tagGens: currentTagGens, + }; + await redis.set(key, JSON.stringify(negEntry), 'PX', config.CHAIN_CACHE_NEG_TTL_MS); + } catch {} + } + throw err; + } finally { + if (acquiredLock) { + redis.del(lockKey).catch(() => {}); + } + } + }; + + const promise = execute().finally(() => inFlight.delete(key)); + inFlight.set(key, promise); + return promise; + }, + + stats() { + return { hits, misses, coalesced, staleRejects }; + }, + }; +} diff --git a/src/lib/socket-limits.ts b/src/lib/socket-limits.ts new file mode 100644 index 0000000..3df4789 --- /dev/null +++ b/src/lib/socket-limits.ts @@ -0,0 +1,369 @@ +import type { IncomingMessage } from 'http'; +import crypto from 'crypto'; +import type { Socket } from 'socket.io'; +import type Redis from 'ioredis'; +import { config } from '../config'; +import { logger } from './logger'; + +export function normalizeIp(ip: string | undefined): string { + if (!ip) return '127.0.0.1'; + let cleaned = ip.trim(); + if (cleaned.startsWith('::ffff:')) { + cleaned = cleaned.slice(7); + } + return cleaned; +} + +function ipv4ToInt(ip: string): number | null { + const parts = ip.split('.').map(Number); + if (parts.length !== 4 || parts.some((p) => isNaN(p) || p < 0 || p > 255)) { + return null; + } + return ((parts[0] << 24) | (parts[1] << 16) | (parts[2] << 8) | parts[3]) >>> 0; +} + +export function ipMatchesCidr(ip: string, cidr: string): boolean { + const normIp = normalizeIp(ip); + const normCidr = cidr.trim(); + if (!normCidr) return false; + + if (!normCidr.includes('/')) { + return normIp.toLowerCase() === normalizeIp(normCidr).toLowerCase(); + } + + const [net, maskStr] = normCidr.split('/'); + const mask = parseInt(maskStr, 10); + if (isNaN(mask) || mask < 0 || mask > 32) return false; + + const ipInt = ipv4ToInt(normIp); + const netInt = ipv4ToInt(net); + if (ipInt === null || netInt === null) { + return normIp.toLowerCase() === net.toLowerCase(); + } + + const bitmask = mask === 0 ? 0 : (~0 << (32 - mask)) >>> 0; + return (ipInt & bitmask) === (netInt & bitmask); +} + +export function isIpInCidrs(ip: string, cidrs: string[]): boolean { + if (!cidrs || cidrs.length === 0) return false; + return cidrs.some((cidr) => ipMatchesCidr(ip, cidr)); +} + +export function clientIp(req: IncomingMessage, trusted: string[] = config.TRUSTED_PROXIES): string { + const peerIp = normalizeIp(req.socket?.remoteAddress); + if (!trusted || trusted.length === 0 || !isIpInCidrs(peerIp, trusted)) { + return peerIp; + } + + const xff = req.headers['x-forwarded-for']; + if (!xff || typeof xff !== 'string') { + return peerIp; + } + + const ips = xff + .split(',') + .map((s) => normalizeIp(s.trim())) + .filter(Boolean); + + if (ips.length === 0) return peerIp; + + // Walk right-to-left skipping trusted proxies + for (let i = ips.length - 1; i >= 0; i--) { + const ip = ips[i]; + if (!isIpInCidrs(ip, trusted)) { + return ip; + } + } + + return ips[0]; +} + +export function isAllowedOrigin(origin: string | undefined): boolean { + if (!origin) return true; + return config.CORS_ORIGINS.includes(origin); +} + +export interface ConnectionLimiterOptions { + maxConnections: number; + maxPerIp: number; + connectRatePerMin: number; + exemptCidrs?: string[]; + instanceId?: string; +} + +export interface ConnectionLimiter { + acquire(ip: string, socketId: string): Promise<{ ok: true } | { ok: false; reason: 'GLOBAL' | 'PER_IP' | 'RATE' }>; + release(ip: string, socketId: string): Promise; +} + +const metricsState = { + connected: 0, + rejected: { + GLOBAL: 0, + PER_IP: 0, + RATE: 0, + ORIGIN: 0, + EVENT: 0, + } as Record, + disconnectedForAbuse: 0, +}; + +export function recordSocketRejection(reason: string): void { + metricsState.rejected[reason] = (metricsState.rejected[reason] ?? 0) + 1; +} + +export function socketMetrics(): { + connected: number; + rejected: Record; + disconnectedForAbuse: number; +} { + return { + connected: metricsState.connected, + rejected: { ...metricsState.rejected }, + disconnectedForAbuse: metricsState.disconnectedForAbuse, + }; +} + +export function memoryConnectionLimiter(opts: ConnectionLimiterOptions): ConnectionLimiter { + const activeSockets = new Map(); // socketId -> ip + const ipSockets = new Map>(); // ip -> Set + const rateCounts = new Map(); + + return { + async acquire(ip: string, socketId: string) { + const isExempt = opts.exemptCidrs && isIpInCidrs(ip, opts.exemptCidrs); + + if (activeSockets.size >= opts.maxConnections) { + metricsState.rejected.GLOBAL = (metricsState.rejected.GLOBAL ?? 0) + 1; + return { ok: false, reason: 'GLOBAL' }; + } + + if (!isExempt) { + const now = Date.now(); + let rateEntry = rateCounts.get(ip); + if (!rateEntry || now > rateEntry.expiresAt) { + rateEntry = { count: 0, expiresAt: now + 60_000 }; + rateCounts.set(ip, rateEntry); + } + + if (rateEntry.count >= opts.connectRatePerMin) { + metricsState.rejected.RATE = (metricsState.rejected.RATE ?? 0) + 1; + return { ok: false, reason: 'RATE' }; + } + rateEntry.count++; + + const currentSet = ipSockets.get(ip); + if (currentSet && currentSet.size >= opts.maxPerIp) { + metricsState.rejected.PER_IP = (metricsState.rejected.PER_IP ?? 0) + 1; + return { ok: false, reason: 'PER_IP' }; + } + } + + activeSockets.set(socketId, ip); + let set = ipSockets.get(ip); + if (!set) { + set = new Set(); + ipSockets.set(ip, set); + } + set.add(socketId); + metricsState.connected++; + + return { ok: true }; + }, + + async release(ip: string, socketId: string) { + if (activeSockets.has(socketId)) { + activeSockets.delete(socketId); + const set = ipSockets.get(ip); + if (set) { + set.delete(socketId); + if (set.size === 0) ipSockets.delete(ip); + } + metricsState.connected = Math.max(0, metricsState.connected - 1); + } + }, + }; +} + +const LUA_ACQUIRE = ` +local ipKey = KEYS[1] +local instancesKey = KEYS[2] +local rateKey = KEYS[3] +local member = ARGV[1] +local maxGlobal = tonumber(ARGV[2]) +local maxPerIp = tonumber(ARGV[3]) +local maxRate = tonumber(ARGV[4]) +local isExempt = ARGV[5] +local instanceId = ARGV[6] + +-- Calculate total cluster connections +local total = 0 +local instCounts = redis.call('HGETALL', instancesKey) +for i = 2, #instCounts, 2 do + total = total + tonumber(instCounts[i]) +end + +if total >= maxGlobal then + return 'GLOBAL' +end + +if isExempt ~= '1' then + local currentRate = redis.call('INCR', rateKey) + if currentRate == 1 then + redis.call('EXPIRE', rateKey, 120) + end + if currentRate > maxRate then + return 'RATE' + end + + local ipCount = redis.call('SCARD', ipKey) + if ipCount >= maxPerIp then + return 'PER_IP' + end +end + +redis.call('SADD', ipKey, member) +redis.call('EXPIRE', ipKey, 120) +redis.call('HINCRBY', instancesKey, instanceId, 1) + +return 'OK' +`; + +const LUA_RELEASE = ` +local ipKey = KEYS[1] +local instancesKey = KEYS[2] +local member = ARGV[1] +local instanceId = ARGV[2] + +redis.call('SREM', ipKey, member) +local cur = redis.call('HINCRBY', instancesKey, instanceId, -1) +if cur <= 0 then + redis.call('HDEL', instancesKey, instanceId) +end +return 'OK' +`; + +let lastRedisErrorLog = 0; + +export function redisConnectionLimiter(redis: Redis, opts: ConnectionLimiterOptions): ConnectionLimiter { + const fallback = memoryConnectionLimiter(opts); + const instanceId = opts.instanceId || `${process.pid}:${crypto.randomBytes(4).toString('hex')}`; + + const currentMinute = () => Math.floor(Date.now() / 60000); + + return { + async acquire(ip: string, socketId: string) { + try { + const isExempt = opts.exemptCidrs && isIpInCidrs(ip, opts.exemptCidrs); + const ipKey = `socklim:ip:${ip}`; + const instancesKey = 'socklim:instances'; + const rateKey = `socklim:rate:${ip}:${currentMinute()}`; + const member = `${instanceId}:${socketId}`; + + const res = (await redis.eval( + LUA_ACQUIRE, + 3, + ipKey, + instancesKey, + rateKey, + member, + opts.maxConnections.toString(), + opts.maxPerIp.toString(), + opts.connectRatePerMin.toString(), + isExempt ? '1' : '0', + instanceId, + )) as string; + + if (res === 'OK') { + metricsState.connected++; + return { ok: true }; + } + + const reason = res as 'GLOBAL' | 'PER_IP' | 'RATE'; + metricsState.rejected[reason] = (metricsState.rejected[reason] ?? 0) + 1; + return { ok: false, reason }; + } catch (err) { + const now = Date.now(); + if (now - lastRedisErrorLog > 60_000) { + logger.warn({ err }, 'Redis connection limiter failed; falling back to memory limiter'); + lastRedisErrorLog = now; + } + return fallback.acquire(ip, socketId); + } + }, + + async release(ip: string, socketId: string) { + try { + const ipKey = `socklim:ip:${ip}`; + const instancesKey = 'socklim:instances'; + const member = `${instanceId}:${socketId}`; + + await redis.eval(LUA_RELEASE, 2, ipKey, instancesKey, member, instanceId); + metricsState.connected = Math.max(0, metricsState.connected - 1); + } catch (err) { + await fallback.release(ip, socketId); + } + }, + }; +} + +export interface InboundLimiterOptions { + allowedEvents?: string[]; + eventsPer10s?: number; +} + +export function inboundEventLimiter(socket: Socket, opts: InboundLimiterOptions = {}): void { + const allowed = new Set(opts.allowedEvents ?? []); + const limit = opts.eventsPer10s ?? config.SOCKET_INBOUND_EVENTS_PER_10S; + + let eventCount = 0; + let windowEnd = Date.now() + 10_000; + let violations = 0; + let violationWindowEnd = Date.now() + 60_000; + + socket.use(([event], next) => { + const now = Date.now(); + if (now > violationWindowEnd) { + violations = 0; + violationWindowEnd = now + 60_000; + } + if (now > windowEnd) { + eventCount = 0; + windowEnd = now + 10_000; + } + + if (!allowed.has(event)) { + violations++; + metricsState.rejected.EVENT = (metricsState.rejected.EVENT ?? 0) + 1; + if (violations >= 3) { + metricsState.disconnectedForAbuse++; + logger.warn( + { socketId: socket.id, event, violations }, + 'Socket disconnected for repeated unknown inbound events', + ); + socket.disconnect(true); + return; + } + return next(new Error('EVENT_NOT_ALLOWED')); + } + + eventCount++; + if (eventCount > limit) { + violations++; + metricsState.rejected.RATE = (metricsState.rejected.RATE ?? 0) + 1; + if (violations >= 3) { + metricsState.disconnectedForAbuse++; + logger.warn( + { socketId: socket.id, eventCount, limit, violations }, + 'Socket disconnected for inbound event rate limit abuse', + ); + socket.disconnect(true); + return; + } + return next(new Error('RATE_LIMIT_EXCEEDED')); + } + + next(); + }); +} diff --git a/src/lib/socket-throttle.ts b/src/lib/socket-throttle.ts new file mode 100644 index 0000000..0ae4697 --- /dev/null +++ b/src/lib/socket-throttle.ts @@ -0,0 +1,240 @@ +import type { Server } from 'socket.io'; +import { logger } from './logger'; + +export interface Sink { + emit(event: string, payload: unknown, opts: { volatile: boolean; room?: string }): void; +} + +export interface ThrottleOptions { + priceTickMs: number; + tradeVolatile: boolean; + dropHistorical: boolean; + now?: () => number; + schedule?: (fn: () => void, ms: number) => unknown; +} + +export interface MarketPricePayload { + onChainMarketId: string; + marketId: string | null; + priceYesBps: number; + ledger: number; + eventId: string; + ts: number; +} + +export class SocketThrottle { + private sink: Sink; + private opts: ThrottleOptions; + private now: () => number; + private schedule: (fn: () => void, ms: number) => unknown; + + private pendingPrices = new Map(); + private activeTimers = new Map(); + private lastEmittedAt = new Map(); + private lastEmittedLedger = new Map(); + + private coalescedCount = 0; + private droppedHistoricalCount = 0; + private emittedCounts: Record = {}; + + constructor(sink: Sink, opts: ThrottleOptions) { + this.sink = sink; + this.opts = opts; + this.now = opts.now ?? (() => Date.now()); + this.schedule = opts.schedule ?? ((fn, ms) => setTimeout(fn, ms)); + } + + private incrementEmitted(event: string): void { + this.emittedCounts[event] = (this.emittedCounts[event] ?? 0) + 1; + } + + public handle( + event: string, + payload: any, + meta: { live: boolean; ledger: number; marketKey: string | null }, + ): void { + if (!meta.live && this.opts.dropHistorical && event !== 'CATCHUP_COMPLETE') { + this.droppedHistoricalCount++; + return; + } + + if (event === 'TRADE_EXECUTED') { + const marketKey = meta.marketKey || payload?.onChainMarketId || payload?.marketId || 'unknown'; + const lastLedger = this.lastEmittedLedger.get(marketKey) ?? 0; + + // Drop pending update older than the last emitted ledger for that market + if (payload?.ledger !== undefined && payload.ledger < lastLedger) { + this.coalescedCount++; + } else { + const existing = this.pendingPrices.get(marketKey); + if (existing) { + // Compare ledger, then eventId string + const isNewer = + payload.ledger > existing.ledger || + (payload.ledger === existing.ledger && String(payload.eventId) >= String(existing.eventId)); + if (isNewer) { + this.pendingPrices.set(marketKey, { + onChainMarketId: String(payload.onChainMarketId), + marketId: payload.marketId ?? null, + priceYesBps: Number(payload.priceYesBps), + ledger: Number(payload.ledger), + eventId: String(payload.eventId), + ts: this.now(), + }); + } + this.coalescedCount++; + } else { + this.pendingPrices.set(marketKey, { + onChainMarketId: String(payload.onChainMarketId), + marketId: payload.marketId ?? null, + priceYesBps: Number(payload.priceYesBps), + ledger: Number(payload.ledger), + eventId: String(payload.eventId), + ts: this.now(), + }); + + const lastEmit = this.lastEmittedAt.get(marketKey) ?? 0; + const elapsed = this.now() - lastEmit; + const delay = Math.max(0, this.opts.priceTickMs - elapsed); + + const timer = this.schedule(() => this.flushMarket(marketKey), delay); + this.activeTimers.set(marketKey, timer); + } + } + + // Emit trade itself to realtime:full room + this.sink.emit('TRADE_EXECUTED', payload, { + volatile: this.opts.tradeVolatile, + room: 'realtime:full', + }); + this.incrementEmitted('TRADE_EXECUTED'); + return; + } + + if (event === 'REASONING_PUBLISHED') { + this.sink.emit('REASONING_PUBLISHED', payload, { volatile: false }); + this.incrementEmitted('REASONING_PUBLISHED'); + return; + } + + if (event === 'CATCHUP_COMPLETE') { + this.sink.emit('CATCHUP_COMPLETE', payload, { volatile: false }); + this.incrementEmitted('CATCHUP_COMPLETE'); + return; + } + + this.sink.emit(event, payload, { volatile: false }); + this.incrementEmitted(event); + } + + private flushMarket(marketKey: string): void { + this.clearScheduledTimer(marketKey); + const update = this.pendingPrices.get(marketKey); + if (!update) return; + + this.pendingPrices.delete(marketKey); + this.lastEmittedAt.set(marketKey, this.now()); + this.lastEmittedLedger.set(marketKey, update.ledger); + + this.sink.emit('MARKET_PRICE', update, { volatile: false }); + this.incrementEmitted('MARKET_PRICE'); + } + + private clearScheduledTimer(marketKey: string): void { + const timer = this.activeTimers.get(marketKey); + if (timer) { + if (typeof timer === 'number' || typeof timer === 'object') { + clearTimeout(timer as NodeJS.Timeout); + } + this.activeTimers.delete(marketKey); + } + } + + public flush(): void { + for (const marketKey of Array.from(this.pendingPrices.keys())) { + this.flushMarket(marketKey); + } + } + + public stop(): void { + for (const marketKey of Array.from(this.activeTimers.keys())) { + this.clearScheduledTimer(marketKey); + } + this.pendingPrices.clear(); + } + + public stats(): { + coalesced: number; + droppedHistorical: number; + emitted: Record; + } { + return { + coalesced: this.coalescedCount, + droppedHistorical: this.droppedHistoricalCount, + emitted: { ...this.emittedCounts }, + }; + } +} + +export interface SlowConsumerGuardOptions { + warnBytes: number; + dropBytes: number; + checkMs: number; +} + +export function attachSlowConsumerGuard( + io: Server | any, + opts: SlowConsumerGuardOptions, +): { stop(): void } { + // Ensure all connecting sockets join the realtime:full room + if (typeof io.on === 'function') { + io.on('connection', (socket: any) => { + socket.join('realtime:full'); + }); + } + + const checkSockets = () => { + try { + const socketMap = io.of ? io.of('/').sockets : io.sockets?.sockets; + if (!socketMap) return; + + const sockets = socketMap.values ? Array.from(socketMap.values()) : Object.values(socketMap); + for (const socket of sockets as any[]) { + let bufferedBytes = 0; + try { + if (socket.conn?.transport?.socket?.bufferedAmount !== undefined) { + bufferedBytes = socket.conn.transport.socket.bufferedAmount; + } else if (socket.conn?.writeBuffer?.length !== undefined) { + bufferedBytes = socket.conn.writeBuffer.length * 1024; + } + } catch { + bufferedBytes = 0; + } + + if (bufferedBytes >= opts.dropBytes) { + logger.warn( + { socketId: socket.id, bufferedBytes, dropBytes: opts.dropBytes }, + 'Slow consumer dropped: exceeded dropBytes buffer threshold', + ); + socket.disconnect(true); + } else if (bufferedBytes >= opts.warnBytes) { + if (typeof socket.leave === 'function') { + socket.leave('realtime:full'); + } + } else { + if (typeof socket.join === 'function') { + socket.join('realtime:full'); + } + } + } + } catch (err) { + logger.error({ err }, 'Error in slow-consumer guard check'); + } + }; + + const timer = setInterval(checkSockets, opts.checkMs); + + return { + stop: () => clearInterval(timer), + }; +} diff --git a/src/middlewares/cache.middleware.ts b/src/middlewares/cache.middleware.ts new file mode 100644 index 0000000..6c28a30 --- /dev/null +++ b/src/middlewares/cache.middleware.ts @@ -0,0 +1,153 @@ +import type { Request, Response, NextFunction, RequestHandler } from 'express'; +import { sha256Hex, canonicalJson } from '../utils/hash.util'; +import { checkAccess } from '../services/subscription.service'; + +export type CachePolicy = + | { kind: 'immutable'; maxAge: number } + | { kind: 'shared'; maxAge: number; swr: number } + | { kind: 'personalised'; maxAge: number } + | { kind: 'no-store' }; + +const POLICY_VERSION = 'v1'; + +export function checkETagMatch(ifNoneMatch: string, etag: string): boolean { + if (ifNoneMatch.trim() === '*') return true; + const targetClean = etag.replace(/^W\//, '').trim(); + const list = ifNoneMatch.split(',').map((s) => s.trim().replace(/^W\//, '')); + return list.includes(targetClean); +} + +export async function authClass(req: Request): Promise { + const user = (req as any).user; + if (!user?.userId) { + if (req.headers.authorization) return 'auth-header'; + return 'anon'; + } + + const traceId = req.params.id; + if (traceId) { + try { + const access = await checkAccess(user.userId, traceId); + if (access.hasAccess) return 'pass'; + } catch {} + } + + return `user:${user.userId}`; +} + +export function cachePolicy( + policyInput: CachePolicy | ((req: Request, res: Response) => CachePolicy | Promise), +): RequestHandler { + return async (req: Request, res: Response, next: NextFunction): Promise => { + try { + const policy: CachePolicy = + typeof policyInput === 'function' ? await policyInput(req, res) : policyInput; + res.locals.cachePolicy = policy; + + // Intercept response headers + const originalWriteHead = res.writeHead; + res.writeHead = function (this: Response, statusCode: number, ...args: any[]) { + const is2xx = statusCode >= 200 && statusCode < 300; + const hasAuth = Boolean(req.headers.authorization); + + if (!is2xx) { + res.setHeader('Cache-Control', 'private, no-store'); + } else { + // If request carried Authorization, it must NEVER get a public Cache-Control + if (policy.kind === 'no-store') { + res.setHeader('Cache-Control', 'private, no-store'); + } else if (policy.kind === 'personalised' || hasAuth) { + const maxAge = policy.kind === 'personalised' ? policy.maxAge : 0; + if (maxAge === 0) { + res.setHeader('Cache-Control', 'private, max-age=0, must-revalidate'); + } else { + res.setHeader('Cache-Control', `private, max-age=${maxAge}`); + } + res.append('Vary', 'Authorization'); + } else if (policy.kind === 'immutable') { + res.setHeader('Cache-Control', `public, max-age=${policy.maxAge}, immutable`); + } else if (policy.kind === 'shared') { + res.setHeader( + 'Cache-Control', + `public, max-age=${policy.maxAge}, stale-while-revalidate=${policy.swr}`, + ); + } + + res.append('Vary', 'Accept-Encoding'); + res.append('Vary', 'Origin'); + + // Handle compression suffix on ETag + const encoding = res.getHeader('Content-Encoding'); + if (encoding && typeof encoding === 'string') { + const etag = res.getHeader('ETag'); + if (typeof etag === 'string' && !etag.endsWith(`-${encoding}"`)) { + res.setHeader('ETag', etag.replace(/"$/, `-${encoding}"`)); + } + } + } + + return (originalWriteHead as any).apply(this, [statusCode, ...args]); + }; + + next(); + } catch (err) { + next(err); + } + }; +} + +export function noStore(): RequestHandler { + return cachePolicy({ kind: 'no-store' }); +} + +export function validator(compute: (req: Request) => Promise): RequestHandler { + return async (req: Request, res: Response, next: NextFunction): Promise => { + try { + const version = await compute(req); + if (version === null) { + return next(); + } + + const cls = await authClass(req); + const rawHash = sha256Hex(`${POLICY_VERSION}|${cls}|${version}`); + const etag = `"${rawHash}"`; + res.locals.validatorEtag = etag; + + const ifNoneMatch = req.headers['if-none-match']; + if (ifNoneMatch && checkETagMatch(ifNoneMatch, etag)) { + res.setHeader('ETag', etag); + const policy: CachePolicy = res.locals.cachePolicy ?? { kind: 'shared', maxAge: 5, swr: 30 }; + const hasAuth = Boolean(req.headers.authorization); + + if (policy.kind === 'no-store') { + res.setHeader('Cache-Control', 'private, no-store'); + } else if (policy.kind === 'personalised' || hasAuth) { + const maxAge = policy.kind === 'personalised' ? policy.maxAge : 0; + if (maxAge === 0) { + res.setHeader('Cache-Control', 'private, max-age=0, must-revalidate'); + } else { + res.setHeader('Cache-Control', `private, max-age=${maxAge}`); + } + res.append('Vary', 'Authorization'); + } else if (policy.kind === 'immutable') { + res.setHeader('Cache-Control', `public, max-age=${policy.maxAge}, immutable`); + } else if (policy.kind === 'shared') { + res.setHeader( + 'Cache-Control', + `public, max-age=${policy.maxAge}, stale-while-revalidate=${policy.swr}`, + ); + } + + res.append('Vary', 'Accept-Encoding'); + res.append('Vary', 'Origin'); + res.status(304).end(); + return; + } + + res.setHeader('ETag', etag); + next(); + } catch (err) { + next(err); + } + }; +} diff --git a/src/routes/index.ts b/src/routes/index.ts index 4a0365f..53b9a01 100644 --- a/src/routes/index.ts +++ b/src/routes/index.ts @@ -6,9 +6,18 @@ import portfolioRoutes from './portfolio.routes'; import tradeRoutes from './trade.routes'; import oracleRoutes from './oracle.routes'; import adminRoutes from './admin.routes'; +import { noStore } from '../middlewares/cache.middleware'; const router = Router(); +// All non-GET routes get no-store (Issue #99) +router.use((req, res, next) => { + if (req.method !== 'GET' && req.method !== 'HEAD') { + return noStore()(req, res, next); + } + next(); +}); + router.use('/auth', authRoutes); router.use('/markets', marketRoutes); router.use('/traces', traceRoutes); diff --git a/src/routes/market.routes.ts b/src/routes/market.routes.ts index fe09035..0182fd3 100644 --- a/src/routes/market.routes.ts +++ b/src/routes/market.routes.ts @@ -6,24 +6,86 @@ import { getOnChainMarketState, triggerMarketGeneration, getMarketGenerationStatus, + buildMarketsWhere, } from '../controllers/market.controller'; import { requireAdmin, requireAuth } from '../middlewares/auth.middleware'; import { generateLimiter } from '../middlewares/rate-limit.middleware'; import { idempotent } from '../middlewares/idempotency.middleware'; +import { cachePolicy, validator, noStore } from '../middlewares/cache.middleware'; +import { listMarketsSchema } from '../validators/market.validator'; +import { prisma } from '../lib/prisma'; +import { parseU64 } from '../lib/amounts'; const router = Router(); +const marketDetailPolicy = cachePolicy(async (req) => { + const id = req.params.id; + const onChainRaw = req.params.onChainMarketId; + const onChainMarketId = onChainRaw !== undefined ? parseU64(String(onChainRaw)) : null; + const where = id ? { id } : (onChainMarketId !== null ? { onChainMarketId } : null); + if (!where) return { kind: 'shared', maxAge: 5, swr: 30 }; + const market = await prisma.market.findUnique({ where: where as any, select: { status: true } }); + if (market && (market.status === 'RESOLVED' || market.status === 'CANCELLED')) { + return { kind: 'immutable', maxAge: 86400 }; + } + return { kind: 'shared', maxAge: 5, swr: 30 }; +}); + +const marketDetailValidator = validator(async (req) => { + const id = req.params.id; + const onChainRaw = req.params.onChainMarketId; + const onChainMarketId = onChainRaw !== undefined ? parseU64(String(onChainRaw)) : null; + const where = id ? { id } : (onChainMarketId !== null ? { onChainMarketId } : null); + if (!where) return null; + const market = await prisma.market.findUnique({ + where: where as any, + select: { + id: true, + updatedAt: true, + status: true, + _count: { select: { trades: true, positions: true, reasoningTraces: true } }, + reasoningTraces: { orderBy: { createdAt: 'desc' }, take: 1, select: { updatedAt: true } }, + }, + }); + if (!market) return null; + const traceUpdated = market.reasoningTraces[0]?.updatedAt?.toISOString() ?? '0'; + const counts = `${market._count.trades}:${market._count.positions}:${market._count.reasoningTraces}`; + return `market:${market.id}:${market.status}:${market.updatedAt.toISOString()}:${traceUpdated}:${counts}`; +}); + // Public — no auth needed -router.get('/', listMarkets); -router.get('/on-chain/:onChainMarketId', getMarketByOnChainId); -router.get('/on-chain/:onChainMarketId/state', getOnChainMarketState); +router.get( + '/', + cachePolicy({ kind: 'shared', maxAge: 5, swr: 30 }), + validator(async (req) => { + const parsed = listMarketsSchema.safeParse({ query: req.query }); + if (!parsed.success) return null; + const where = buildMarketsWhere(parsed.data.query); + const [agg, count] = await Promise.all([ + prisma.market.aggregate({ where, _max: { updatedAt: true } }), + prisma.market.count({ where }), + ]); + const maxUpdated = agg._max.updatedAt?.toISOString() ?? '0'; + const page = req.query.page ?? '1'; + const limit = req.query.limit ?? '20'; + return `markets:${maxUpdated}:${count}:${page}:${limit}`; + }), + listMarkets, +); + +router.get('/on-chain/:onChainMarketId', marketDetailPolicy, marketDetailValidator, getMarketByOnChainId); +router.get( + '/on-chain/:onChainMarketId/state', + cachePolicy({ kind: 'shared', maxAge: 5, swr: 10 }), + getOnChainMarketState, +); // Admin only: each run calls paid LLM APIs and the treasury. // Idempotency-Key required (#123): a retry replays the stored 202 and its // jobId instead of starting a second paid market-maker cycle. router.post('/generate', generateLimiter, requireAuth, requireAdmin, idempotent({ required: true }), triggerMarketGeneration); -router.get('/generation-status/:jobId', requireAuth, getMarketGenerationStatus); +router.get('/generation-status/:jobId', requireAuth, noStore(), getMarketGenerationStatus); -router.get('/:id', getMarket); +router.get('/:id', marketDetailPolicy, marketDetailValidator, getMarket); export default router; diff --git a/src/routes/oracle.routes.ts b/src/routes/oracle.routes.ts index ced4d7f..ae97068 100644 --- a/src/routes/oracle.routes.ts +++ b/src/routes/oracle.routes.ts @@ -1,10 +1,24 @@ import { Router } from 'express'; import { getResolution } from '../controllers/oracle.controller'; +import { cachePolicy } from '../middlewares/cache.middleware'; +import { prisma } from '../lib/prisma'; const router = Router(); -// Read-only. POST /oracle/resolve was removed: outcomes are decided on-chain -// by the resolver contract, never by this API. -router.get('/markets/:marketId/resolution', getResolution); +// Read-only. Outcomes are decided on-chain by the resolver contract. +router.get( + '/markets/:marketId/resolution', + cachePolicy(async (req) => { + const market = await prisma.market.findUnique({ + where: { id: req.params.marketId }, + select: { status: true }, + }); + if (market && (market.status === 'RESOLVED' || market.status === 'CANCELLED')) { + return { kind: 'immutable', maxAge: 86400 }; + } + return { kind: 'shared', maxAge: 5, swr: 10 }; + }), + getResolution, +); export default router; diff --git a/src/routes/portfolio.routes.ts b/src/routes/portfolio.routes.ts index 558a1c5..1fe7269 100644 --- a/src/routes/portfolio.routes.ts +++ b/src/routes/portfolio.routes.ts @@ -1,11 +1,49 @@ import { Router } from 'express'; import { getPlatformStats, getPortfolio, getPositions } from '../controllers/portfolio.controller'; +import { cachePolicy, validator } from '../middlewares/cache.middleware'; +import { prisma } from '../lib/prisma'; const router = Router(); // Public (agent portfolio is public for demo) -router.get('/', getPortfolio); -router.get('/positions', getPositions); -router.get('/stats', getPlatformStats); +router.get( + '/', + cachePolicy({ kind: 'shared', maxAge: 15, swr: 60 }), + getPortfolio, +); + +router.get( + '/positions', + cachePolicy({ kind: 'shared', maxAge: 15, swr: 60 }), + validator(async (req) => { + const status = req.query.status as any; + const where = { ...(status ? { status } : {}) }; + const [agg, count] = await Promise.all([ + prisma.position.aggregate({ where, _max: { updatedAt: true } }), + prisma.position.count({ where }), + ]); + const maxUpdated = agg._max.updatedAt?.toISOString() ?? '0'; + const page = req.query.page ?? '1'; + const limit = req.query.limit ?? '20'; + return `positions:${maxUpdated}:${count}:${page}:${limit}`; + }), + getPositions, +); + +router.get( + '/stats', + cachePolicy({ kind: 'shared', maxAge: 15, swr: 60 }), + validator(async () => { + const [marketAgg, posAgg, tradeCount] = await Promise.all([ + prisma.market.aggregate({ _max: { updatedAt: true }, _count: true }), + prisma.position.aggregate({ _max: { updatedAt: true }, _count: true }), + prisma.trade.count(), + ]); + const maxM = marketAgg._max.updatedAt?.toISOString() ?? '0'; + const maxP = posAgg._max.updatedAt?.toISOString() ?? '0'; + return `stats:${maxM}:${maxP}:${marketAgg._count}:${posAgg._count}:${tradeCount}`; + }), + getPlatformStats, +); export default router; diff --git a/src/routes/trace.routes.ts b/src/routes/trace.routes.ts index 14d6093..e91e64f 100644 --- a/src/routes/trace.routes.ts +++ b/src/routes/trace.routes.ts @@ -10,19 +10,70 @@ import { } from '../controllers/trace.controller'; import { optionalAuth, requireAuth } from '../middlewares/auth.middleware'; import { idempotent } from '../middlewares/idempotency.middleware'; +import { cachePolicy, validator, noStore } from '../middlewares/cache.middleware'; +import { prisma } from '../lib/prisma'; const router = Router(); // Public -router.get('/', listTraces); -router.get('/access/allowance', requireAuth, getMySpendingAllowance); +router.get( + '/', + cachePolicy({ kind: 'shared', maxAge: 10, swr: 60 }), + validator(async (req) => { + const agentType = req.query.agentType as any; + const marketId = req.query.marketId as any; + if (agentType !== undefined && agentType !== 'MARKET_MAKER' && agentType !== 'TRADER') { + return null; + } + const where = { + ...(agentType ? { agentType: agentType as 'MARKET_MAKER' | 'TRADER' } : {}), + ...(typeof marketId === 'string' && marketId ? { marketId } : {}), + }; + const [agg, count] = await Promise.all([ + prisma.reasoningTrace.aggregate({ where, _max: { updatedAt: true } }), + prisma.reasoningTrace.count({ where }), + ]); + const maxUpdated = agg._max.updatedAt?.toISOString() ?? '0'; + const page = req.query.page ?? '1'; + const limit = req.query.limit ?? '20'; + return `traces:${maxUpdated}:${count}:${page}:${limit}`; + }), + listTraces, +); + +router.get('/access/allowance', requireAuth, noStore(), getMySpendingAllowance); router.put('/access/allowance', requireAuth, idempotent(), setSpendingAllowance); -router.get('/payments', requireAuth, getMyPaymentEvents); -router.get('/:id', optionalAuth, getTrace); // Preview without a token, full trace with a daily pass +router.get('/payments', requireAuth, noStore(), getMyPaymentEvents); + +router.get( + '/:id', + optionalAuth, + cachePolicy(async (req) => { + if (!req.headers.authorization) { + const trace = await prisma.reasoningTrace.findUnique({ + where: { id: req.params.id }, + select: { onChainTraceId: true }, + }); + if (trace && trace.onChainTraceId !== null) { + return { kind: 'shared', maxAge: 300, swr: 0 }; + } + } + return { kind: 'personalised', maxAge: 0 }; + }), + validator(async (req) => { + const trace = await prisma.reasoningTrace.findUnique({ + where: { id: req.params.id }, + select: { id: true, updatedAt: true, verified: true, onChainTraceId: true }, + }); + if (!trace) return null; + return `trace:${trace.id}:${trace.updatedAt.toISOString()}:${trace.verified}:${trace.onChainTraceId ?? ''}`; + }), + getTrace, +); // Protected // Idempotency-Key optional (#123). Lock is 60s by default, above verify's 20s IPFS timeout. -router.post('/verify', requireAuth, idempotent(), verifyTrace); -router.post('/:id/unlock', requireAuth, idempotent(), unlockTrace); +router.post('/verify', requireAuth, idempotent(), verifyTrace); +router.post('/:id/unlock', requireAuth, idempotent(), unlockTrace); export default router; diff --git a/src/server.ts b/src/server.ts index 24ecfaf..c9bf7ab 100644 --- a/src/server.ts +++ b/src/server.ts @@ -10,11 +10,38 @@ import { startMonitorCron } from './cron/monitor.cron'; import { startTraceAuditCron } from './cron/trace-audit.cron'; import { startTracePublicationCron } from './cron/trace-publication.cron'; import { startEventListener } from './services/indexer.service'; -import { closeRedis } from './lib/redis'; +import { getRedis, closeRedis } from './lib/redis'; import { health } from './services/health.service'; +import { + clientIp, + isAllowedOrigin, + redisConnectionLimiter, + inboundEventLimiter, + recordSocketRejection, +} from './lib/socket-limits'; +import { SocketThrottle, attachSlowConsumerGuard } from './lib/socket-throttle'; let stopEventListener: (() => void) | null = null; let stopSecretRefresh: (() => void) | null = null; +let throttle: SocketThrottle | null = null; +let stopSlowGuard: (() => void) | null = null; + +const rejectionLogState = new Map(); +function logSocketRejection(reason: string, details: Record = {}): void { + recordSocketRejection(reason); + const now = Date.now(); + let state = rejectionLogState.get(reason); + if (!state) { + state = { count: 0, lastLog: 0 }; + rejectionLogState.set(reason, state); + } + state.count++; + if (now - state.lastLog >= 10_000) { + logger.warn({ reason, count: state.count, ...details }, `Socket connection rejected (${reason})`); + state.count = 0; + state.lastLog = now; + } +} async function bootstrap(): Promise { // 0. Resolve secrets (mounted files or the configured remote provider). @@ -43,17 +70,113 @@ async function bootstrap(): Promise { } logger.info({ mode: config.CHAIN_EXECUTION_MODE }, 'Chain execution mode'); - // 2. Start HTTP server with Socket.io + // 2. Start HTTP server with hardened Socket.io const server = http.createServer(app); const io = new Server(server, { + transports: config.SOCKET_TRANSPORTS as any, + maxHttpBufferSize: config.SOCKET_MAX_BUFFER_BYTES, + pingInterval: 25_000, + pingTimeout: 20_000, + connectTimeout: 10_000, + perMessageDeflate: false, cors: { - origin: config.CORS_ORIGINS, + origin: (origin, callback) => callback(null, isAllowedOrigin(origin)), methods: ['GET', 'POST'], + credentials: true, + }, + allowRequest: (req, callback) => { + const origin = req.headers.origin; + if (origin && !isAllowedOrigin(origin)) { + logSocketRejection('ORIGIN', { origin }); + return callback('ORIGIN_NOT_ALLOWED', false); + } + callback(null, true); }, }); + const limiter = redisConnectionLimiter(getRedis(), { + maxConnections: config.SOCKET_MAX_CONNECTIONS, + maxPerIp: config.SOCKET_MAX_PER_IP, + connectRatePerMin: config.SOCKET_CONNECT_RATE_PER_MIN, + exemptCidrs: config.SOCKET_IP_EXEMPT_CIDRS, + }); + + io.engine.on('connection_error', (err: any) => { + const origin = err.req?.headers?.origin; + if (origin && !isAllowedOrigin(origin)) { + logSocketRejection('ORIGIN', { origin }); + } else { + logSocketRejection(err.message || 'HANDSHAKE_ERROR', { code: err.code }); + } + }); + + io.use(async (socket, next) => { + const ip = clientIp(socket.request, config.TRUSTED_PROXIES); + const result = await limiter.acquire(ip, socket.id); + if (!result.ok) { + logSocketRejection(result.reason, { ip, socketId: socket.id }); + return next(Object.assign(new Error('CONNECTION_LIMIT'), { data: { reason: result.reason } })); + } + + socket.on('disconnect', () => { + limiter.release(ip, socket.id).catch((err) => { + logger.error({ err, socketId: socket.id }, 'Error releasing socket connection limit'); + }); + if (Math.random() < config.SOCKET_LOG_SAMPLE_RATE) { + logger.info({ socketId: socket.id }, 'Client disconnected from socket.io'); + } else { + logger.debug({ socketId: socket.id }, 'Client disconnected from socket.io'); + } + }); + + inboundEventLimiter(socket, { + eventsPer10s: config.SOCKET_INBOUND_EVENTS_PER_10S, + }); + + next(); + }); + + io.on('connection', (socket) => { + if (Math.random() < config.SOCKET_LOG_SAMPLE_RATE) { + logger.info({ socketId: socket.id }, 'New client connected to socket.io'); + } else { + logger.debug({ socketId: socket.id }, 'New client connected to socket.io'); + } + }); + + // Slow consumer guard (Issue #96) + const slowGuard = attachSlowConsumerGuard(io, { + warnBytes: config.SOCKET_SLOW_WARN_BYTES, + dropBytes: config.SOCKET_SLOW_DROP_BYTES, + checkMs: config.SOCKET_SLOW_CHECK_MS, + }); + stopSlowGuard = slowGuard.stop; + + // Socket throttle (Issue #96) + throttle = new SocketThrottle( + { + emit: (event, payload, opts) => { + if (opts.room) { + const target = io.to(opts.room); + if (opts.volatile) target.volatile.emit(event, payload); + else target.emit(event, payload); + } else { + if (opts.volatile) io.volatile.emit(event, payload); + else io.emit(event, payload); + } + }, + }, + { + priceTickMs: config.SOCKET_PRICE_TICK_MS, + tradeVolatile: config.SOCKET_TRADE_VOLATILE, + dropHistorical: true, + }, + ); + // 3. Index Stellar events (resumes from the stored ledger cursor) - stopEventListener = startEventListener((event, payload) => io.emit(event, payload)); + stopEventListener = startEventListener((event, payload, meta) => { + throttle?.handle(event, payload, meta); + }); // 4. Start cron jobs startIngestionCron(); @@ -61,14 +184,6 @@ async function bootstrap(): Promise { startTraceAuditCron(); startTracePublicationCron(); - io.on('connection', (socket) => { - logger.info({ socketId: socket.id }, 'New client connected to socket.io'); - - socket.on('disconnect', () => { - logger.info({ socketId: socket.id }, 'Client disconnected from socket.io'); - }); - }); - // Make io accessible via app app.set('io', io); @@ -93,6 +208,13 @@ async function bootstrap(): Promise { const { triggerShutdown } = await import('./lib/shutdown'); triggerShutdown(); + if (throttle) { + throttle.flush(); + throttle.stop(); + } + if (stopSlowGuard) stopSlowGuard(); + io.close(); + server.close(async () => { if (stopEventListener) stopEventListener(); if (stopSecretRefresh) stopSecretRefresh(); diff --git a/src/services/chain.service.ts b/src/services/chain.service.ts index ad939b2..66423a0 100644 --- a/src/services/chain.service.ts +++ b/src/services/chain.service.ts @@ -18,14 +18,23 @@ */ import type { AssembledTransaction } from '@stellar/stellar-sdk/contract'; import { config } from '../config'; +import { CONTRACTS } from '../config/contracts'; import { logger } from '../lib/logger'; import { agentClients, readClients } from './stellar/clients'; import { ChainError, contractErrorName, ArchivedEntryError } from './stellar/errors'; import { withTimeout, TimeoutError } from '../lib/resilience'; -import type { Category, Market as OnChainMarket, Outcome, Position } from '../generated/market-core'; +import type { Category, Market as OnChainMarket, Outcome, Position, Config as MarketCoreConfig } from '../generated/market-core'; import type { Trace } from '../generated/reasoning-registry'; import type { DailyUsage, RiskConfig } from '../generated/treasury'; import type { ResolutionState } from '../generated/resolver'; +import { getRedis } from '../lib/redis'; +import { + chainKey, + memoryChainCache, + redisChainCache, + type ChainCache, + type CacheEntry, +} from '../lib/chain-cache'; export { ArchivedEntryError }; @@ -40,6 +49,29 @@ export const Outcomes = { No: { tag: 'No', values: undefined } as Outcome, }; +// ─── Cache Management ─────────────────────────────────────────────────────── + +let activeChainCache: ChainCache | null = null; + +export function getChainCache(): ChainCache { + if (!activeChainCache) { + try { + activeChainCache = redisChainCache(getRedis()); + } catch { + activeChainCache = memoryChainCache(); + } + } + return activeChainCache; +} + +export function setChainCache(cache: ChainCache | null): void { + activeChainCache = cache; +} + +export function chainCacheStats() { + return getChainCache().stats(); +} + /** Value from a simulated call, unwrapping contract `Result`s into ChainErrors. */ export function simulatedValue(contract: ContractKey, method: string, tx: AssembledTransaction): T { const sim = tx.simulation as { error?: string } | undefined; @@ -73,12 +105,12 @@ export function simulatedValue(contract: ContractKey, method: string, tx: Ass return raw as T; } -async function read( +export async function readWithLedger( contract: ContractKey, method: string, call: () => Promise>, signal?: AbortSignal, -): Promise { +): Promise<{ value: T; ledger: number }> { return withTimeout( 'rpc-read', async () => { @@ -95,12 +127,27 @@ async function read( cause: msg, }); } - return simulatedValue(contract, method, tx); + const ledger = + typeof (tx.simulation as any)?.latestLedger === 'number' + ? (tx.simulation as any).latestLedger + : 0; + const value = simulatedValue(contract, method, tx); + return { value, ledger }; }, { signal }, ); } +export async function read( + contract: ContractKey, + method: string, + call: () => Promise>, + signal?: AbortSignal, +): Promise { + const result = await readWithLedger(contract, method, call, signal); + return result.value; +} + async function write( contract: 'treasury' | 'reasoningRegistry', method: string, @@ -137,6 +184,19 @@ async function write( const value = simulatedValueFromSent(contract, method, sent.result); const finalHash = sent.sendTransactionResponse?.hash ?? txHash; logger.info({ method, txHash: finalHash }, 'submitted'); + + // Write-through cache invalidation after live submission + try { + const cache = getChainCache(); + if (method === 'agent_create_market') { + cache.invalidateTag('treasury:capital').catch(() => {}); + } else if (method === 'agent_buy' || method === 'agent_sell') { + const mId = args.market_id !== undefined ? String(args.market_id) : null; + if (mId) cache.invalidateTag(`market:${mId}`).catch(() => {}); + cache.invalidateTag('treasury:capital').catch(() => {}); + } + } catch {} + return { mode: 'live', method, value, txHash: finalHash }; } catch (err: any) { if (err instanceof TimeoutError || err?.code === 'UPSTREAM_TIMEOUT') { @@ -182,6 +242,26 @@ export const marketCount = (signal?: AbortSignal) => export const getOnChainMarket = (marketId: bigint, signal?: AbortSignal) => read('marketCore', 'get_market', async () => readClients.marketCore().get_market({ market_id: marketId }), signal); +export const getOnChainMarketWithLedger = (marketId: bigint, signal?: AbortSignal) => + readWithLedger('marketCore', 'get_market', async () => + readClients.marketCore().get_market({ market_id: marketId }), signal); + +export const getOnChainMarketCached = async ( + marketId: bigint, + signal?: AbortSignal, +): Promise & { cached: boolean }> => { + if (!config.CHAIN_CACHE_ENABLED) { + const res = await getOnChainMarketWithLedger(marketId, signal); + return { ...res, readAt: Date.now(), gen: 0, cached: false }; + } + const key = chainKey(CONTRACTS.marketCore, 'get_market', { id: marketId.toString() }); + return getChainCache().get( + key, + { ttlMs: config.CHAIN_CACHE_TTL_MS, tags: [`market:${marketId}`] }, + () => getOnChainMarketWithLedger(marketId, signal), + ); +}; + export const getPrice = (marketId: bigint, outcome: Outcome, signal?: AbortSignal) => read('marketCore', 'get_price', async () => readClients.marketCore().get_price({ market_id: marketId, outcome }), signal); @@ -228,9 +308,48 @@ export const marketTraceAt = (marketId: bigint, index: bigint, signal?: AbortSig export const resolverState = (marketId: bigint, signal?: AbortSignal) => read('resolver', 'state', async () => readClients.resolver().state({ market_id: marketId }), signal); +export const resolverStateWithLedger = (marketId: bigint, signal?: AbortSignal) => + readWithLedger('resolver', 'state', async () => + readClients.resolver().state({ market_id: marketId }), signal); + +export const resolverStateCached = async ( + marketId: bigint, + signal?: AbortSignal, +): Promise & { cached: boolean }> => { + if (!config.CHAIN_CACHE_ENABLED) { + const res = await resolverStateWithLedger(marketId, signal); + return { ...res, readAt: Date.now(), gen: 0, cached: false }; + } + const key = chainKey(CONTRACTS.resolver, 'state', { id: marketId.toString() }); + return getChainCache().get( + key, + { ttlMs: config.CHAIN_CACHE_TTL_MS, tags: [`resolver:${marketId}`, `market:${marketId}`] }, + () => resolverStateWithLedger(marketId, signal), + ); +}; + export const treasuryAvailableCapital = (signal?: AbortSignal) => read('treasury', 'available_capital', async () => readClients.treasury().available_capital(), signal); +export const treasuryAvailableCapitalWithLedger = (signal?: AbortSignal) => + readWithLedger('treasury', 'available_capital', async () => + readClients.treasury().available_capital(), signal); + +export const treasuryAvailableCapitalCached = async ( + signal?: AbortSignal, +): Promise & { cached: boolean }> => { + if (!config.CHAIN_CACHE_ENABLED) { + const res = await treasuryAvailableCapitalWithLedger(signal); + return { ...res, readAt: Date.now(), gen: 0, cached: false }; + } + const key = chainKey(CONTRACTS.treasury, 'available_capital', {}); + return getChainCache().get( + key, + { ttlMs: config.CHAIN_CACHE_TTL_MS, tags: ['treasury:capital'] }, + () => treasuryAvailableCapitalWithLedger(signal), + ); +}; + export const treasuryConfig = () => read('treasury', 'get_config', async () => readClients.treasury().get_config()); diff --git a/src/services/indexer.service.ts b/src/services/indexer.service.ts index 2fc0986..befe3ca 100644 --- a/src/services/indexer.service.ts +++ b/src/services/indexer.service.ts @@ -29,7 +29,11 @@ export type EventSource = Pick; export interface IndexerDeps { source: EventSource; /** Called for each newly stored event that should reach clients. */ - emit: (event: string, payload: unknown) => void; + emit: ( + event: string, + payload: unknown, + meta: { live: boolean; ledger: number; marketKey: string | null }, + ) => void; } const PAGE_LIMIT = 200; @@ -102,9 +106,42 @@ async function store(event: DecodedEvent): Promise { const outcomeTag = (o: unknown): 'Yes' | 'No' | null => o && typeof o === 'object' && 'tag' in o ? ((o as { tag: 'Yes' | 'No' }).tag) : null; -async function applySideEffects(event: DecodedEvent, emit: IndexerDeps['emit']): Promise { +async function applySideEffects( + event: DecodedEvent, + emit: IndexerDeps['emit'], + meta: { live: boolean; ledger: number; marketKey: string | null }, +): Promise { const p = event.params; + // ── Pre-step: Event-driven cache invalidation (Issue #98) ── + try { + const { getChainCache } = await import('./chain.service'); + const cache = getChainCache(); + const marketIdStr = p.market_id !== undefined ? String(p.market_id) : null; + + if (event.contractName === 'marketCore') { + if (['Trade', 'MarketResolved', 'Redeemed', 'FeesSwept', 'MarketCreated'].includes(event.eventName)) { + if (marketIdStr) await cache.invalidateTag(`market:${marketIdStr}`, event.ledger); + } + if (event.eventName === 'Trade' && p.trader === CONTRACTS.treasury) { + await cache.invalidateTag('treasury:capital', event.ledger); + } + } else if (event.contractName === 'resolver') { + if (['ResolutionProposed', 'ResolutionFinalized', 'PriceResolution'].includes(event.eventName)) { + if (marketIdStr) { + await cache.invalidateTag(`resolver:${marketIdStr}`, event.ledger); + await cache.invalidateTag(`market:${marketIdStr}`, event.ledger); + } + } + } else if (event.contractName === 'treasury') { + if (['TradeAuthorized', 'PositionClosed', 'FeesCollected'].includes(event.eventName)) { + await cache.invalidateTag('treasury:capital', event.ledger); + } + } + } catch (err) { + logger.warn({ err, eventId: event.id }, 'Indexer: chain cache invalidation failed'); + } + if (event.contractName === 'marketCore' && event.eventName === 'Trade') { const onChainMarketId = BigInt(p.market_id); const yesBps = Number(priceYesBps(BigInt(p.reserve_yes), BigInt(p.reserve_no))); @@ -121,21 +158,29 @@ async function applySideEffects(event: DecodedEvent, emit: IndexerDeps['emit']): } } - emit('TRADE_EXECUTED', { - onChainMarketId: onChainMarketId.toString(), - marketId: market?.id ?? null, - marketQuestion: market?.question ?? null, - trader: p.trader, - direction: outcomeTag(p.outcome) === 'No' ? 'NO' : 'YES', - isBuy: Boolean(p.is_buy), - collateralRaw: String(p.collateral), - sharesRaw: String(p.shares), - feeRaw: String(p.fee), - priceYesBps: yesBps, - txHash: event.txHash, - ledger: event.ledger, - eventId: event.id, - }); + emit( + 'TRADE_EXECUTED', + { + onChainMarketId: onChainMarketId.toString(), + marketId: market?.id ?? null, + marketQuestion: market?.question ?? null, + trader: p.trader, + direction: outcomeTag(p.outcome) === 'No' ? 'NO' : 'YES', + isBuy: Boolean(p.is_buy), + collateralRaw: String(p.collateral), + sharesRaw: String(p.shares), + feeRaw: String(p.fee), + priceYesBps: yesBps, + txHash: event.txHash, + ledger: event.ledger, + eventId: event.id, + }, + { + live: meta.live, + ledger: event.ledger, + marketKey: onChainMarketId.toString(), + }, + ); return; } @@ -191,17 +236,25 @@ async function applySideEffects(event: DecodedEvent, emit: IndexerDeps['emit']): data: { status: 'CONFIRMED', confirmedAt: new Date(), lastError: null }, }); } - emit('REASONING_PUBLISHED', { - onChainTraceId: onChainTraceId.toString(), - onChainMarketId: String(p.market_id), - traceId: trace?.id ?? null, - agent: p.agent, - action: p.action, - ipfsCid: p.ipfs_cid, - traceHash, - txHash: event.txHash, - ledger: event.ledger, - }); + emit( + 'REASONING_PUBLISHED', + { + onChainTraceId: onChainTraceId.toString(), + onChainMarketId: String(p.market_id), + traceId: trace?.id ?? null, + agent: p.agent, + action: p.action, + ipfsCid: p.ipfs_cid, + traceHash, + txHash: event.txHash, + ledger: event.ledger, + }, + { + live: meta.live, + ledger: event.ledger, + marketKey: String(p.market_id), + }, + ); } } @@ -246,6 +299,7 @@ export async function backfillEvents( return { startLedger, lastLedger: cursor ?? health.latestLedger, seen: 0, stored: 0 }; } + const isCatchup = health.latestLedger - startLedger > config.SOCKET_LIVE_LEDGER_LAG; const known = specs(); const filters: rpc.Api.EventFilter[] = [{ type: 'contract', contractIds: [...known.keys()] }]; let seen = 0; @@ -263,9 +317,13 @@ export async function backfillEvents( seen++; const decoded = decodeEvent(raw, known); if (!decoded) continue; + const live = page.latestLedger - decoded.ledger <= config.SOCKET_LIVE_LEDGER_LAG; + const marketKey = decoded.params?.market_id !== undefined ? String(decoded.params.market_id) : null; + const meta = { live, ledger: decoded.ledger, marketKey }; + if (await store(decoded)) { stored++; - await applySideEffects(decoded, deps.emit).catch((err) => + await applySideEffects(decoded, deps.emit, meta).catch((err) => logger.error({ err, eventId: decoded.id }, 'Indexer side effect failed'), ); } @@ -279,6 +337,17 @@ export async function backfillEvents( } await writeCursor(latestLedger); + if (isCatchup) { + deps.emit( + 'CATCHUP_COMPLETE', + { + fromLedger: startLedger, + toLedger: latestLedger, + events: stored, + }, + { live: true, ledger: latestLedger, marketKey: null }, + ); + } if (stored > 0) logger.info({ startLedger, lastLedger: latestLedger, seen, stored }, 'Indexed Stellar events'); return { startLedger, lastLedger: latestLedger, seen, stored }; } finally { @@ -295,7 +364,12 @@ export async function reprojectEvent( event: DecodedEvent, emit: IndexerDeps['emit'] = () => undefined, ): Promise { - await applySideEffects(event, emit); + const meta = { + live: true, + ledger: event.ledger, + marketKey: event.params?.market_id !== undefined ? String(event.params.market_id) : null, + }; + await applySideEffects(event, emit, meta); } /** Converts a stored row back into the decoded shape used by applySideEffects. */ diff --git a/src/services/oracle.service.ts b/src/services/oracle.service.ts index 49b3489..a44b2f5 100644 --- a/src/services/oracle.service.ts +++ b/src/services/oracle.service.ts @@ -4,7 +4,7 @@ */ import { prisma } from '../lib/prisma'; import { AppError } from '../middlewares/error.middleware'; -import { getOnChainMarket, resolverState } from './chain.service'; +import { getOnChainMarketCached, resolverStateCached } from './chain.service'; import type { MarketStatus } from '../generated/market-core'; export type MarketStatusView = @@ -27,11 +27,16 @@ export async function getResolutionStatus(marketId: string, signal?: AbortSignal throw new AppError(409, 'MARKET_NOT_ON_CHAIN', 'Market has not been created on-chain yet'); } - const [state, onChain] = await Promise.all([ - resolverState(market.onChainMarketId, signal), - getOnChainMarket(market.onChainMarketId, signal), + const [stateRes, onChainRes] = await Promise.all([ + resolverStateCached(market.onChainMarketId, signal), + getOnChainMarketCached(market.onChainMarketId, signal), ]); + const state = stateRes.value; + const onChain = onChainRes.value; + const ledger = Math.min(stateRes.ledger, onChainRes.ledger); + const cached = Boolean(stateRes.cached && onChainRes.cached); + return { marketId: market.id, onChainMarketId: market.onChainMarketId.toString(), @@ -40,5 +45,7 @@ export async function getResolutionStatus(marketId: string, signal?: AbortSignal resolutionHash: market.resolutionHash, resolutionSpec: market.resolutionSpec, dbStatus: market.status, + ledger, + cached, }; } diff --git a/src/utils/response.util.ts b/src/utils/response.util.ts index 4b3bdcc..845edd1 100644 --- a/src/utils/response.util.ts +++ b/src/utils/response.util.ts @@ -1,6 +1,8 @@ import type { Response } from 'express'; import { currentContext } from '../lib/context'; import type { ApiResponse, ApiError, PaginationMeta } from '../types'; +import { canonicalJson, sha256Hex } from './hash.util'; +import { checkETagMatch } from '../middlewares/cache.middleware'; /** * Send a successful 2xx response @@ -12,6 +14,23 @@ export function sendSuccess( meta?: Record, ): void { const body: ApiResponse = { ok: true, data, error: null, meta }; + + const policy = res.locals?.cachePolicy; + if (policy && policy.kind !== 'no-store') { + let etag = res.getHeader('ETag'); + if (!etag) { + const hash = sha256Hex(canonicalJson(body)); + etag = `"${hash}"`; + res.setHeader('ETag', etag); + } + + const ifNoneMatch = res.req?.headers['if-none-match']; + if (ifNoneMatch && checkETagMatch(ifNoneMatch, String(etag))) { + res.status(304).end(); + return; + } + } + res.status(statusCode).json(body); } diff --git a/tests/chain-cache.test.ts b/tests/chain-cache.test.ts new file mode 100644 index 0000000..33437b7 --- /dev/null +++ b/tests/chain-cache.test.ts @@ -0,0 +1,148 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { setupTestEnv } from './helpers/env'; +import { encode, decode, memoryChainCache, chainKey } from '../src/lib/chain-cache'; +import { ChainError } from '../src/services/stellar/errors'; + +setupTestEnv(); + +test('codec round-trips Market, ResolutionState and i128 values losslessly', () => { + const sample = { + marketId: 42n, + reserves: [1000000000n, 2000000000n], + hash: Buffer.from('f4b58d4ac4068163f34bd9486f0563c70873afa5f64e915ab7e85af7dee15e4e', 'hex'), + status: { tag: 'Resolved', values: [{ tag: 'Yes' }] }, + name: 'Test Market', + }; + + const encoded = encode(sample); + const decoded = decode(encoded); + + assert.equal(decoded.marketId, 42n); + assert.equal(decoded.reserves[0], 1000000000n); + assert.equal(decoded.reserves[1], 2000000000n); + assert.ok(Buffer.isBuffer(decoded.hash)); + assert.equal(decoded.hash.toString('hex'), sample.hash.toString('hex')); + assert.deepEqual(decoded.status, sample.status); + assert.equal(decoded.name, sample.name); +}); + +test('second read within TTL is served from cache with cached true and the original ledger', async () => { + const cache = memoryChainCache(); + let loadCount = 0; + + const loader = async () => { + loadCount++; + return { value: { reserveYes: 100n }, ledger: 12345 }; + }; + + const read1 = await cache.get('k1', { ttlMs: 5000, tags: ['market:1'] }, loader); + assert.equal(read1.cached, false); + assert.equal(read1.ledger, 12345); + assert.equal(read1.value.reserveYes, 100n); + assert.equal(loadCount, 1); + + const read2 = await cache.get('k1', { ttlMs: 5000, tags: ['market:1'] }, loader); + assert.equal(read2.cached, true); + assert.equal(read2.ledger, 12345); + assert.equal(read2.value.reserveYes, 100n); + assert.equal(loadCount, 1); +}); + +test('100 concurrent misses trigger exactly one loader call', async () => { + const cache = memoryChainCache(); + let loadCount = 0; + + const loader = async () => { + loadCount++; + await new Promise((r) => setTimeout(r, 20)); + return { value: 'coalesced-data', ledger: 500 }; + }; + + const reads = await Promise.all( + Array.from({ length: 100 }, () => + cache.get('k-concurrent', { ttlMs: 5000, tags: ['market:5'] }, loader), + ), + ); + + assert.equal(loadCount, 1); + for (const r of reads) { + assert.equal(r.value, 'coalesced-data'); + assert.equal(r.ledger, 500); + } +}); + +test('invalidateTag forces a reload on the next read', async () => { + const cache = memoryChainCache(); + let loadCount = 0; + + const loader = async () => { + loadCount++; + return { value: `val-${loadCount}`, ledger: 1000 + loadCount }; + }; + + const r1 = await cache.get('k-tag', { ttlMs: 5000, tags: ['market:10'] }, loader); + assert.equal(r1.value, 'val-1'); + assert.equal(loadCount, 1); + + await cache.invalidateTag('market:10'); + + const r2 = await cache.get('k-tag', { ttlMs: 5000, tags: ['market:10'] }, loader); + assert.equal(r2.value, 'val-2'); + assert.equal(r2.cached, false); + assert.equal(loadCount, 2); +}); + +test('a load that started before invalidation and returns an older ledger is not cached', async () => { + const cache = memoryChainCache(); + + // Invalidate at ledger 200 + await cache.invalidateTag('market:20', 200); + + // Read returns older ledger 199 + const r1 = await cache.get( + 'k-stale', + { ttlMs: 5000, tags: ['market:20'] }, + async () => ({ value: 'stale-value', ledger: 199 }), + ); + + assert.equal(r1.cached, false); + assert.equal(cache.stats().staleRejects, 1); + + // Next read should miss and not serve the rejected older entry + let secondLoaded = false; + const r2 = await cache.get( + 'k-stale', + { ttlMs: 5000, tags: ['market:20'] }, + async () => { + secondLoaded = true; + return { value: 'fresh-value', ledger: 205 }; + }, + ); + + assert.equal(secondLoaded, true); + assert.equal(r2.value, 'fresh-value'); +}); + +test('contract errors are negatively cached only for the negative TTL', async () => { + const cache = memoryChainCache(); + let calls = 0; + + const loader = async () => { + calls++; + throw new ChainError('MarketNotFound', 'MarketNotFound'); + }; + + await assert.rejects( + async () => cache.get('k-notfound', { ttlMs: 5000, tags: ['market:404'] }, loader), + (err: any) => err.contractError === 'MarketNotFound', + ); + assert.equal(calls, 1); + + // Second call within negative TTL hits negative cache without calling loader + await assert.rejects( + async () => cache.get('k-notfound', { ttlMs: 5000, tags: ['market:404'] }, loader), + (err: any) => err.contractError === 'MarketNotFound', + ); + assert.equal(calls, 1); +}); diff --git a/tests/http-cache.test.ts b/tests/http-cache.test.ts new file mode 100644 index 0000000..e2acdcd --- /dev/null +++ b/tests/http-cache.test.ts @@ -0,0 +1,147 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import type { AddressInfo } from 'node:net'; +import { setupTestEnv } from './helpers/env'; +import { prisma } from '../src/lib/prisma'; +import { signToken } from '../src/middlewares/auth.middleware'; +import { checkETagMatch, authClass } from '../src/middlewares/cache.middleware'; + +setupTestEnv(); + +async function withServer(fn: (base: string) => Promise) { + const { default: app } = await import('../src/app'); + const server = app.listen(0); + try { + await fn(`http://127.0.0.1:${(server.address() as AddressInfo).port}`); + } finally { + server.close(); + } +} + +test('checkETagMatch handles strong comparison, wildcards, and W/ prefixes', () => { + assert.equal(checkETagMatch('*', '"abc"'), true); + assert.equal(checkETagMatch('"abc"', '"abc"'), true); + assert.equal(checkETagMatch('W/"abc"', '"abc"'), true); + assert.equal(checkETagMatch('"xyz", "abc"', '"abc"'), true); + assert.equal(checkETagMatch('"xyz"', '"abc"'), false); +}); + +test('every GET route sets the documented Cache-Control and Vary headers', async () => { + const origFindMany = prisma.market.findMany; + const origCount = prisma.market.count; + const origAgg = prisma.market.aggregate; + + (prisma.market.findMany as any) = async () => []; + (prisma.market.count as any) = async () => 0; + (prisma.market.aggregate as any) = async () => ({ _max: { updatedAt: new Date() } }); + + try { + await withServer(async (base) => { + // GET /api/v1/markets -> shared (max-age=5, stale-while-revalidate=30) + const res = await fetch(`${base}/api/v1/markets`); + assert.equal(res.status, 200); + const cc = res.headers.get('cache-control'); + assert.ok(cc?.includes('public')); + assert.ok(cc?.includes('max-age=5')); + assert.ok(cc?.includes('stale-while-revalidate=30')); + const vary = res.headers.get('vary'); + assert.ok(vary?.includes('Accept-Encoding')); + assert.ok(vary?.includes('Origin')); + }); + } finally { + prisma.market.findMany = origFindMany; + prisma.market.count = origCount; + prisma.market.aggregate = origAgg; + } +}); + +test('matching If-None-Match returns 304 with no body and skips the controller for validator routes', async () => { + const origFindMany = prisma.market.findMany; + const origCount = prisma.market.count; + const origAgg = prisma.market.aggregate; + let findManyCalled = false; + + (prisma.market.aggregate as any) = async () => ({ _max: { updatedAt: new Date(1700000000000) } }); + (prisma.market.count as any) = async () => 10; + (prisma.market.findMany as any) = async () => { + findManyCalled = true; + return []; + }; + + try { + await withServer(async (base) => { + // First request: gets full response and ETag + const res1 = await fetch(`${base}/api/v1/markets`); + assert.equal(res1.status, 200); + const etag = res1.headers.get('etag'); + assert.ok(etag, 'ETag must be returned'); + assert.equal(findManyCalled, true); + + // Second request with If-None-Match + findManyCalled = false; + const res2 = await fetch(`${base}/api/v1/markets`, { + headers: { 'if-none-match': etag }, + }); + assert.equal(res2.status, 304); + const body = await res2.text(); + assert.equal(body, '', '304 response body must be empty'); + assert.equal(findManyCalled, false, 'controller must be skipped on 304'); + }); + } finally { + prisma.market.findMany = origFindMany; + prisma.market.count = origCount; + prisma.market.aggregate = origAgg; + } +}); + +test('requests with Authorization never get a public Cache-Control', async () => { + const token = signToken('user-1', 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF'); + + await withServer(async (base) => { + const res = await fetch(`${base}/api/v1/markets`, { + headers: { authorization: `Bearer ${token}` }, + }); + const cc = res.headers.get('cache-control'); + assert.ok(!cc?.includes('public'), 'Cache-Control must not be public when Authorization is sent'); + assert.ok(cc?.includes('private')); + const vary = res.headers.get('vary'); + assert.ok(vary?.includes('Authorization')); + }); +}); + +test('errors get no-store', async () => { + await withServer(async (base) => { + const res = await fetch(`${base}/api/v1/non-existent-route`); + assert.equal(res.status, 404); + const cc = res.headers.get('cache-control'); + assert.ok(cc?.includes('no-store')); + }); +}); + +test('ETag is exposed to CORS clients', async () => { + await withServer(async (base) => { + const res = await fetch(`${base}/api/v1/markets`, { + headers: { origin: 'http://localhost:3000' }, + }); + const exposed = res.headers.get('access-control-expose-headers'); + assert.ok(exposed?.includes('ETag')); + }); +}); + +test('authenticated payment and allowance routes are no-store', async () => { + const token = signToken('user-1', 'GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF'); + const origFindUnique = prisma.spendingAllowance.findUnique; + (prisma.spendingAllowance.findUnique as any) = async () => null; + + try { + await withServer(async (base) => { + const res = await fetch(`${base}/api/v1/traces/access/allowance`, { + headers: { authorization: `Bearer ${token}` }, + }); + const cc = res.headers.get('cache-control'); + assert.ok(cc?.includes('no-store')); + }); + } finally { + prisma.spendingAllowance.findUnique = origFindUnique; + } +}); diff --git a/tests/indexer.test.ts b/tests/indexer.test.ts index 3b5a266..78592b7 100644 --- a/tests/indexer.test.ts +++ b/tests/indexer.test.ts @@ -87,8 +87,8 @@ async function setup(events = FIXTURES.map(toEventResponse)) { return { events, cursor: 'end', latestLedger: this.latest, oldestLedger: this.oldest } as any; }, }; - const emitted: Array<[string, any]> = []; - const emit = (e: string, p: any) => emitted.push([e, p]); + const emitted: Array<[string, any, any?]> = []; + const emit = (e: string, p: any, meta?: any) => emitted.push([e, p, meta]); const { backfillEvents } = await import('../src/services/indexer.service'); return { db, source, calls, emitted, emit, backfillEvents }; } @@ -186,3 +186,29 @@ test('Trade event confirms a CONFIRMING copy trade with matching txHash', async assert.equal(db.copyTrades[0].status, 'EXECUTED'); assert.equal(db.copyTrades[0].collateralRaw, '100000000'); }); + +test('Trade event invalidates market tag before TRADE_EXECUTED is emitted', async () => { + const { setChainCache, memoryChainCache } = await import('../src/services/chain.service'); + const cache = memoryChainCache(); + const invalidations: Array<{ tag: string; atLedger?: number }> = []; + const origInvalidate = cache.invalidateTag.bind(cache); + cache.invalidateTag = async (tag: string, atLedger?: number) => { + invalidations.push({ tag, atLedger }); + return origInvalidate(tag, atLedger); + }; + setChainCache(cache); + + const { source, emit, backfillEvents } = await setup(); + let invalidatedBeforeEmit = false; + const wrappedEmit = (e: string, p: any, meta?: any) => { + if (e === 'TRADE_EXECUTED') { + if (invalidations.some((inv) => inv.tag === 'market:0')) { + invalidatedBeforeEmit = true; + } + } + emit(e, p, meta); + }; + + await backfillEvents(undefined, { source, emit: wrappedEmit }); + assert.ok(invalidatedBeforeEmit, 'market:0 tag must be invalidated before TRADE_EXECUTED is emitted'); +}); diff --git a/tests/socket-limits.test.ts b/tests/socket-limits.test.ts new file mode 100644 index 0000000..df1a089 --- /dev/null +++ b/tests/socket-limits.test.ts @@ -0,0 +1,156 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { setupTestEnv } from './helpers/env'; +import { + clientIp, + isAllowedOrigin, + memoryConnectionLimiter, + inboundEventLimiter, + socketMetrics, + normalizeIp, + ipMatchesCidr, +} from '../src/lib/socket-limits'; + +setupTestEnv(); + +test('clientIp ignores X-Forwarded-For from untrusted peers and honours trusted proxies', () => { + const trustedProxies = ['10.0.0.1', '127.0.0.1']; + + // Untrusted peer attempting to spoof XFF + const untrustedReq: any = { + socket: { remoteAddress: '198.51.100.25' }, + headers: { 'x-forwarded-for': '203.0.113.195' }, + }; + assert.equal(clientIp(untrustedReq, trustedProxies), '198.51.100.25'); + + // Trusted proxy forwarding real client IP + const trustedReq: any = { + socket: { remoteAddress: '10.0.0.1' }, + headers: { 'x-forwarded-for': '203.0.113.50, 10.0.0.1' }, + }; + assert.equal(clientIp(trustedReq, trustedProxies), '203.0.113.50'); + + // IPv6 mapped IPv4 address + const ipv6Req: any = { + socket: { remoteAddress: '::ffff:192.168.1.10' }, + headers: {}, + }; + assert.equal(clientIp(ipv6Req, trustedProxies), '192.168.1.10'); +}); + +test('isAllowedOrigin allows configured origins and non-browser requests', () => { + assert.equal(isAllowedOrigin(undefined), true); + assert.equal(isAllowedOrigin(''), true); + assert.equal(isAllowedOrigin('http://localhost:3000'), true); + assert.equal(isAllowedOrigin('http://evil.com'), false); +}); + +test('21st connection from one IP is rejected with CONNECTION_LIMIT PER_IP', async () => { + const limiter = memoryConnectionLimiter({ + maxConnections: 100, + maxPerIp: 20, + connectRatePerMin: 100, + }); + + const ip = '1.2.3.4'; + for (let i = 1; i <= 20; i++) { + const res = await limiter.acquire(ip, `sock-${i}`); + assert.equal(res.ok, true); + } + + const rejected = await limiter.acquire(ip, 'sock-21'); + assert.equal(rejected.ok, false); + if (!rejected.ok) { + assert.equal(rejected.reason, 'PER_IP'); + } + + // Release one connection and verify next acquire succeeds + await limiter.release(ip, 'sock-1'); + const allowed = await limiter.acquire(ip, 'sock-21'); + assert.equal(allowed.ok, true); +}); + +test('global cap rejects new connections and recovers after disconnects', async () => { + const limiter = memoryConnectionLimiter({ + maxConnections: 3, + maxPerIp: 10, + connectRatePerMin: 100, + }); + + assert.equal((await limiter.acquire('1.1.1.1', 's1')).ok, true); + assert.equal((await limiter.acquire('2.2.2.2', 's2')).ok, true); + assert.equal((await limiter.acquire('3.3.3.3', 's3')).ok, true); + + const overflow = await limiter.acquire('4.4.4.4', 's4'); + assert.equal(overflow.ok, false); + if (!overflow.ok) { + assert.equal(overflow.reason, 'GLOBAL'); + } + + await limiter.release('1.1.1.1', 's1'); + assert.equal((await limiter.acquire('4.4.4.4', 's4')).ok, true); +}); + +test('connection rate limit rejects rapid reconnect loops', async () => { + const limiter = memoryConnectionLimiter({ + maxConnections: 100, + maxPerIp: 50, + connectRatePerMin: 5, + }); + + const ip = '9.9.9.9'; + for (let i = 0; i < 5; i++) { + const res = await limiter.acquire(ip, `s-${i}`); + assert.equal(res.ok, true); + await limiter.release(ip, `s-${i}`); + } + + // 6th acquire within same minute + const res = await limiter.acquire(ip, 's-6'); + assert.equal(res.ok, false); + if (!res.ok) { + assert.equal(res.reason, 'RATE'); + } +}); + +test('unknown inbound events are rejected and repeated violations disconnect the socket', () => { + let disconnected = false; + const handlers: Array<([event, ...args]: any[], next: (err?: Error) => void) => void> = []; + + const fakeSocket: any = { + id: 'abuse-sock', + use: (fn: any) => handlers.push(fn), + disconnect: () => { disconnected = true; }, + }; + + inboundEventLimiter(fakeSocket, { + allowedEvents: ['ping'], + eventsPer10s: 20, + }); + + const runEvent = (event: string) => { + let passed = false; + let error: any = null; + for (const h of handlers) { + h([event], (err) => { + if (err) error = err; + else passed = true; + }); + } + return { passed, error }; + }; + + // 1st unknown event + const r1 = runEvent('malicious_event_1'); + assert.ok(r1.error); + assert.equal(disconnected, false); + + // 2nd unknown event + const r2 = runEvent('malicious_event_2'); + assert.ok(r2.error); + assert.equal(disconnected, false); + + // 3rd unknown event -> disconnect + runEvent('malicious_event_3'); + assert.equal(disconnected, true); +}); diff --git a/tests/socket-throttle.test.ts b/tests/socket-throttle.test.ts new file mode 100644 index 0000000..cc98c4f --- /dev/null +++ b/tests/socket-throttle.test.ts @@ -0,0 +1,320 @@ +import test from 'node:test'; +import assert from 'node:assert/strict'; +import { setupTestEnv } from './helpers/env'; +import { SocketThrottle, attachSlowConsumerGuard } from '../src/lib/socket-throttle'; + +setupTestEnv(); + +test('bursts of trades on one market produce one MARKET_PRICE per tick with the latest price', () => { + let currentTime = 1000; + const timers = new Map void; triggerAt: number }>(); + let timerId = 0; + + const emitted: Array<{ event: string; payload: any; opts: any }> = []; + const sink = { + emit(event: string, payload: unknown, opts: { volatile: boolean; room?: string }) { + emitted.push({ event, payload, opts }); + }, + }; + + const throttle = new SocketThrottle(sink, { + priceTickMs: 1000, + tradeVolatile: true, + dropHistorical: true, + now: () => currentTime, + schedule: (fn, ms) => { + const id = ++timerId; + timers.set(id, { fn, triggerAt: currentTime + ms }); + return id; + }, + }); + + // Burst of 5 trades in the same tick + for (let i = 1; i <= 5; i++) { + throttle.handle( + 'TRADE_EXECUTED', + { onChainMarketId: '0', priceYesBps: 5000 + i * 100, ledger: 100 + i, eventId: `ev-${i}` }, + { live: true, ledger: 100 + i, marketKey: '0' }, + ); + } + + // Trades are emitted immediately to realtime:full room + const trades = emitted.filter((e) => e.event === 'TRADE_EXECUTED'); + assert.equal(trades.length, 5); + + // No price ticks emitted yet + assert.equal(emitted.filter((e) => e.event === 'MARKET_PRICE').length, 0); + + // Advance time to 2000 ms and trigger timer + currentTime = 2000; + for (const [id, t] of timers.entries()) { + if (t.triggerAt <= currentTime) { + timers.delete(id); + t.fn(); + } + } + + const prices = emitted.filter((e) => e.event === 'MARKET_PRICE'); + assert.equal(prices.length, 1); + assert.equal(prices[0].payload.priceYesBps, 5500); + assert.equal(prices[0].payload.ledger, 105); + assert.equal(prices[0].opts.volatile, false); +}); + +test('final price after a burst is emitted when the tick elapses', () => { + let currentTime = 1000; + const timers = new Map void; triggerAt: number }>(); + let timerId = 0; + + const emitted: Array<{ event: string; payload: any }> = []; + const sink = { + emit(event: string, payload: unknown) { + emitted.push({ event, payload }); + }, + }; + + const throttle = new SocketThrottle(sink, { + priceTickMs: 500, + tradeVolatile: true, + dropHistorical: true, + now: () => currentTime, + schedule: (fn, ms) => { + const id = ++timerId; + timers.set(id, { fn, triggerAt: currentTime + ms }); + return id; + }, + }); + + throttle.handle( + 'TRADE_EXECUTED', + { onChainMarketId: '1', priceYesBps: 6000, ledger: 200, eventId: 'ev-1' }, + { live: true, ledger: 200, marketKey: '1' }, + ); + + // Before timer expires: 0 price ticks + assert.equal(emitted.filter((e) => e.event === 'MARKET_PRICE').length, 0); + + currentTime = 1600; + for (const [id, t] of timers.entries()) { + if (t.triggerAt <= currentTime) { + timers.delete(id); + t.fn(); + } + } + + const prices = emitted.filter((e) => e.event === 'MARKET_PRICE'); + assert.equal(prices.length, 1); + assert.equal(prices[0].payload.priceYesBps, 6000); +}); + +test('out-of-order older ledger never overwrites a newer pending price', () => { + let currentTime = 1000; + const timers = new Map void; triggerAt: number }>(); + let timerId = 0; + + const emitted: Array<{ event: string; payload: any }> = []; + const sink = { + emit(event: string, payload: unknown) { + emitted.push({ event, payload }); + }, + }; + + const throttle = new SocketThrottle(sink, { + priceTickMs: 1000, + tradeVolatile: true, + dropHistorical: true, + now: () => currentTime, + schedule: (fn, ms) => { + const id = ++timerId; + timers.set(id, { fn, triggerAt: currentTime + ms }); + return id; + }, + }); + + // Newer trade first + throttle.handle( + 'TRADE_EXECUTED', + { onChainMarketId: '2', priceYesBps: 7000, ledger: 300, eventId: 'ev-2' }, + { live: true, ledger: 300, marketKey: '2' }, + ); + + // Out of order older trade + throttle.handle( + 'TRADE_EXECUTED', + { onChainMarketId: '2', priceYesBps: 6500, ledger: 299, eventId: 'ev-1' }, + { live: true, ledger: 299, marketKey: '2' }, + ); + + currentTime = 2000; + for (const [id, t] of timers.entries()) { + if (t.triggerAt <= currentTime) { + timers.delete(id); + t.fn(); + } + } + + const prices = emitted.filter((e) => e.event === 'MARKET_PRICE'); + assert.equal(prices.length, 1); + assert.equal(prices[0].payload.priceYesBps, 7000); + assert.equal(prices[0].payload.ledger, 300); +}); + +test('historical events during catch-up are not emitted live and CATCHUP_COMPLETE is emitted once', () => { + const emitted: Array<{ event: string; payload: any }> = []; + const sink = { + emit(event: string, payload: unknown) { + emitted.push({ event, payload }); + }, + }; + + const throttle = new SocketThrottle(sink, { + priceTickMs: 1000, + tradeVolatile: true, + dropHistorical: true, + }); + + // Historical trade during catch-up (live: false) + throttle.handle( + 'TRADE_EXECUTED', + { onChainMarketId: '0', priceYesBps: 5000, ledger: 10, eventId: 'ev-old' }, + { live: false, ledger: 10, marketKey: '0' }, + ); + + assert.equal(emitted.length, 0); + assert.equal(throttle.stats().droppedHistorical, 1); + + // Catchup complete event + throttle.handle( + 'CATCHUP_COMPLETE', + { fromLedger: 1, toLedger: 100, events: 50 }, + { live: true, ledger: 100, marketKey: null }, + ); + + assert.equal(emitted.length, 1); + assert.equal(emitted[0].event, 'CATCHUP_COMPLETE'); +}); + +test('TRADE_EXECUTED is volatile by default and non-volatile when configured', () => { + const emittedDefault: any[] = []; + const throttleDefault = new SocketThrottle( + { emit: (event, payload, opts) => emittedDefault.push({ event, payload, opts }) }, + { priceTickMs: 1000, tradeVolatile: true, dropHistorical: true }, + ); + + throttleDefault.handle( + 'TRADE_EXECUTED', + { onChainMarketId: '0', priceYesBps: 5000, ledger: 100, eventId: '1' }, + { live: true, ledger: 100, marketKey: '0' }, + ); + assert.equal(emittedDefault[0].opts.volatile, true); + + const emittedNonVolatile: any[] = []; + const throttleNonVolatile = new SocketThrottle( + { emit: (event, payload, opts) => emittedNonVolatile.push({ event, payload, opts }) }, + { priceTickMs: 1000, tradeVolatile: false, dropHistorical: true }, + ); + + throttleNonVolatile.handle( + 'TRADE_EXECUTED', + { onChainMarketId: '0', priceYesBps: 5000, ledger: 100, eventId: '1' }, + { live: true, ledger: 100, marketKey: '0' }, + ); + assert.equal(emittedNonVolatile[0].opts.volatile, false); +}); + +test('idle markets release their timers', () => { + let cleared = false; + const throttle = new SocketThrottle( + { emit: () => {} }, + { + priceTickMs: 1000, + tradeVolatile: true, + dropHistorical: true, + schedule: () => ({ + // Timer token + }), + }, + ); + + throttle.handle( + 'TRADE_EXECUTED', + { onChainMarketId: '0', priceYesBps: 5000, ledger: 100, eventId: '1' }, + { live: true, ledger: 100, marketKey: '0' }, + ); + + throttle.stop(); + // Ensure pending was cleared + throttle.flush(); + assert.equal(throttle.stats().emitted['MARKET_PRICE'] ?? 0, 0); +}); + +test('slow-consumer guard degrades then disconnects sockets above thresholds', () => { + const disconnected: string[] = []; + const rooms = new Map>(); + + const fakeSockets = new Map([ + [ + 'sock-normal', + { + id: 'sock-normal', + conn: { transport: { socket: { bufferedAmount: 1000 } } }, + join: (r: string) => { + let s = rooms.get('sock-normal'); + if (!s) { s = new Set(); rooms.set('sock-normal', s); } + s.add(r); + }, + leave: (r: string) => rooms.get('sock-normal')?.delete(r), + disconnect: () => disconnected.push('sock-normal'), + }, + ], + [ + 'sock-slow', + { + id: 'sock-slow', + conn: { transport: { socket: { bufferedAmount: 300_000 } } }, + join: (r: string) => { + let s = rooms.get('sock-slow'); + if (!s) { s = new Set(); rooms.set('sock-slow', s); } + s.add(r); + }, + leave: (r: string) => rooms.get('sock-slow')?.delete(r), + disconnect: () => disconnected.push('sock-slow'), + }, + ], + [ + 'sock-stalled', + { + id: 'sock-stalled', + conn: { transport: { socket: { bufferedAmount: 3_000_000 } } }, + join: (r: string) => { + let s = rooms.get('sock-stalled'); + if (!s) { s = new Set(); rooms.set('sock-stalled', s); } + s.add(r); + }, + leave: (r: string) => rooms.get('sock-stalled')?.delete(r), + disconnect: () => disconnected.push('sock-stalled'), + }, + ], + ]); + + const fakeIo = { + of: () => ({ sockets: fakeSockets }), + on: () => {}, + }; + + const guard = attachSlowConsumerGuard(fakeIo, { + warnBytes: 262_144, // 256KB + dropBytes: 2_097_152, // 2MB + checkMs: 50, + }); + + // Wait a moment for guard loop check + return new Promise((resolve) => { + setTimeout(() => { + guard.stop(); + assert.ok(disconnected.includes('sock-stalled'), 'sock-stalled must be disconnected'); + assert.ok(!rooms.get('sock-slow')?.has('realtime:full'), 'sock-slow must leave realtime:full'); + resolve(); + }, 100); + }); +});