diff --git a/.env.example b/.env.example index 8bc5259..dd25a1c 100644 --- a/.env.example +++ b/.env.example @@ -23,6 +23,20 @@ DATABASE_URL="postgresql://parashield:non-production-dev-pg-password-change-in-p # Redis password used by docker-compose.yml # REDIS_PASSWORD=non-production-dev-redis-password +# ─── Redis connection (#710) ─────────────────────────────────────────────────── +# Connection URL for the shared Redis client (rate limiting, idempotency, +# distributed locks, policy-state cache). Falls back to redis://localhost:6379 +# when unset. Include credentials in the URL when Redis requires auth +# (e.g. redis://:password@host:6379) — they are never logged. +# REDIS_URL="redis://localhost:6379" + +# Startup connection retry. The app retries the first Redis connection with +# exponential backoff + jitter before failing startup. All optional; defaults shown. +# REDIS_STARTUP_MAX_ATTEMPTS=5 # total attempts, including the first +# REDIS_STARTUP_INITIAL_DELAY_MS=200 # delay before the 2nd attempt (doubles each retry) +# REDIS_STARTUP_MAX_DELAY_MS=3000 # cap on the backoff delay +# REDIS_STARTUP_ATTEMPT_TIMEOUT_MS=5000 # per-attempt timeout; a hung connect cannot stall startup + # Distributed locking for scheduled background jobs (#675) — optional. # Stable lock keys are used per job (lock:job:auth-cleanup, lock:job:oracle-worker, # lock:job:claims-worker). Lock TTL in milliseconds. Defaults: 300000 (5m) for auth cleanup, diff --git a/README.md b/README.md index 2d1bd5b..2aec10b 100644 --- a/README.md +++ b/README.md @@ -132,6 +132,26 @@ npm run start:prod --- +## API versioning (#713) + +Every response carries an `X-API-Version` header advertising the semantic +version that served it (`v1` → `1.0.0`, `v2` → `2.0.0`), including error +responses, streamed responses (SSE/NDJSON) and responses that never reach a +route handler (CORS preflights, request timeouts). + +- The version is taken from the route's `vN` segment (`/api/v1/...`), falling + back to a request `x-api-version` header, then to `v1`. +- Requests for an unknown version (`/api/v3/...`) are rejected with `400`. +- `v1` responses also include `Deprecation: true` and + `Link: ; rel="successor"`. +- The header is exposed via CORS so browser clients can read it. + +Resolution lives in `src/common/versioning/api-version.ts` and is applied by the +versioning interceptor (success paths), the global exception filter (error +paths) and an early middleware (everything else). + +--- + ## CORS configuration CORS is enabled in `src/main.ts` and controlled entirely via environment variables (see `.env.example`): @@ -370,6 +390,7 @@ All `BigInt` values (such as oracle readings, large policy IDs, and 64-bit times - **Input sanitization**: a global middleware (`InputSanitizationMiddleware`, registered in `src/main.ts`) trims every string in JSON/urlencoded request bodies and escapes `<`/`>` before validation and persistence, so markup cannot survive into stored values or reflected responses. - **Connection pooling**: Prisma pool sizing is applied at runtime to the datasource URL (`connection_limit=10`, `pool_timeout=10`, `connect_timeout=5`). Override each value with `DATABASE_CONNECTION_LIMIT`, `DATABASE_POOL_TIMEOUT`, `DATABASE_CONNECT_TIMEOUT`, or by putting the parameter directly in `DATABASE_URL`. +- **Response header hardening (#711)**: `applySecurityHeaders()` (`src/common/security/response-header-hardening.ts`, called from `src/main.ts`) disables Express's `X-Powered-By`, applies [helmet](https://helmetjs.github.io/) and strips implementation-revealing headers (`Server`, `X-Powered-By`, `X-AspNet-Version`, `X-AspNetMvc-Version`, `X-Runtime`, `Via`) from every response. Stripping is done at the header level, so headers added later by the framework or an upstream proxy are removed too. Extra names can be added via `extraHeadersToStrip`. --- @@ -411,6 +432,8 @@ CONCURRENCY_TEST_DATABASE_URL=postgresql://.../parashield_test npx jest claims.c | Flight delay | [AviationStack](https://aviationstack.com) | Yes (`AVIATIONSTACK_API_KEY`) | | DeFi exploit | Stellar RPC event stream | No | +**Threshold evaluation (#712):** oracle readings are 7-decimal fixed point (e.g. `324000000` = 32.4) and product thresholds are `Decimal(20,7)`. Trigger comparisons are done on exact scaled `BigInt` values (`src/common/utils/decimal.ts`) rather than `parseFloat`, so the last decimal place is never lost and a reading exactly equal to the threshold never triggers (strict `LessThan` / greater-than). + --- ## Project layout @@ -473,6 +496,36 @@ Fee per tx: ~0.00001 XLM. Fund via `stellar keys fund
--network testne --- +## Redis connection & startup retry (#710) + +The app creates a single shared Redis client (`REDIS_URL`) used for rate limiting, +idempotency, distributed locks and the policy-state cache. The client is created +with `lazyConnect`, so construction never blocks; the first connection is performed +during startup by `RedisConnectionService` (`src/redis/redis-connection.service.ts`). + +Startup behaviour: + +- The first connection is retried with exponential backoff, a capped delay and + jitter. A per-attempt timeout guarantees a hung connect cannot stall startup. +- If every attempt fails, startup fails fast with one actionable error naming the + Redis `host:port` (never the username/password or full URL). +- Once connected, runtime reconnection is handed back to ioredis (its + `retryStrategy`/`reconnectOnError` handle transient drops). + +Tuning (all optional — defaults shown): + +| Variable | Default | Description | +| --- | --- | --- | +| `REDIS_STARTUP_MAX_ATTEMPTS` | `5` | Total connection attempts, including the first. | +| `REDIS_STARTUP_INITIAL_DELAY_MS` | `200` | Delay before the second attempt; doubles each retry. | +| `REDIS_STARTUP_MAX_DELAY_MS` | `3000` | Upper bound for the backoff delay. | +| `REDIS_STARTUP_ATTEMPT_TIMEOUT_MS` | `5000` | Per-attempt timeout. | + +Invalid values (negative, non-integer or out of range) fail startup with a clear +validation error. Credentials embedded in `REDIS_URL` are never logged. + +--- + ## v2 roadmap - Full Soroban SDK transaction builder for all write paths (currently stubbed with `// TODO` markers) diff --git a/src/app.module.ts b/src/app.module.ts index 1a7af93..66db48d 100644 --- a/src/app.module.ts +++ b/src/app.module.ts @@ -37,6 +37,24 @@ function validateConfig(config: Record) { logger.warn('REDIS_URL is not set — falling back to redis://localhost:6379'); } + // #710 — Validate Redis startup-retry tuning. All optional; defaults live in + // src/redis/redis-startup.ts and are documented in .env.example / README. + const redisRetryBounds: Array<{ key: string; min: number; max: number }> = [ + { key: 'REDIS_STARTUP_MAX_ATTEMPTS', min: 1, max: 100 }, + { key: 'REDIS_STARTUP_INITIAL_DELAY_MS', min: 0, max: 60000 }, + { key: 'REDIS_STARTUP_MAX_DELAY_MS', min: 0, max: 300000 }, + { key: 'REDIS_STARTUP_ATTEMPT_TIMEOUT_MS', min: 1, max: 120000 }, + ]; + for (const { key, min, max } of redisRetryBounds) { + const raw = config[key] as string | undefined; + if (raw !== undefined && raw !== '') { + const parsed = Number(raw); + if (!Number.isInteger(parsed) || parsed < min || parsed > max) { + errors.push(`${key} must be an integer between ${min} and ${max}`); + } + } + } + // #716 — Validate all secrets for length, format, entropy, and placeholders const nodeEnv = (config['NODE_ENV'] as string) || process.env.NODE_ENV || 'development'; try { diff --git a/src/claims/claims.service.spec.ts b/src/claims/claims.service.spec.ts index bcd552f..1169561 100644 --- a/src/claims/claims.service.spec.ts +++ b/src/claims/claims.service.spec.ts @@ -631,6 +631,95 @@ describe('ClaimsService', () => { expect(mockStellarService.invokeContract).not.toHaveBeenCalled(); }); + // #712 — The threshold comparison must be exact. A reading equal to a + // LessThan threshold is NOT a trigger (strict), and this must hold at full + // 7-decimal resolution rather than via parseFloat's binary rounding. + it('#712 — a reading exactly equal to the LessThan threshold does not trigger', async () => { + mockPrismaService.policy.findUnique.mockResolvedValue(ACTIVE_POLICY); + mockPrismaService.claim.findFirst.mockResolvedValue(null); + mockPrismaService.claim.create.mockResolvedValue({ id: 'claim-eq-lt', status: 'PROCESSING' }); + mockOracleService.getLatestReading.mockResolvedValue({ + key: ACTIVE_POLICY.oracleKey, + value: BigInt(500_000_000), // exactly 50.0000000 — equal to the threshold + confidence: 90, + }); + mockPolicyService.getProductById.mockResolvedValue(MOCK_PRODUCT); // threshold 50.0000000, LessThan + mockPrismaService.claim.update.mockResolvedValue({}); + + const result = await service.autoProcess(POLICY_ID); + + expect(result).toBe('Rejected'); + expect(mockStellarService.invokeContract).not.toHaveBeenCalled(); + expect(mockPrismaService.claim.update).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ status: 'REJECTED', triggerMet: false }), + }), + ); + }); + + it('#712 — a reading exactly equal to the GreaterThan threshold does not trigger', async () => { + const GT_PRODUCT = { ...MOCK_PRODUCT, comparison: 'GreaterThan', threshold: '30.0000000' }; + mockPrismaService.policy.findUnique.mockResolvedValue(ACTIVE_POLICY); + mockPrismaService.claim.findFirst.mockResolvedValue(null); + mockPrismaService.claim.create.mockResolvedValue({ id: 'claim-eq-gt', status: 'PROCESSING' }); + mockOracleService.getLatestReading.mockResolvedValue({ + key: ACTIVE_POLICY.oracleKey, + value: BigInt(300_000_000), // exactly 30.0000000 — equal to the threshold + confidence: 90, + }); + mockPolicyService.getProductById.mockResolvedValue(GT_PRODUCT); + mockPrismaService.claim.update.mockResolvedValue({}); + + const result = await service.autoProcess(POLICY_ID); + + expect(result).toBe('Rejected'); + expect(mockStellarService.invokeContract).not.toHaveBeenCalled(); + }); + + it('#712 — the last decimal place of the threshold is honoured (one unit below triggers)', async () => { + // Reading is exactly 1e-7 below the threshold — must trigger LessThan. + const PRODUCT = { ...MOCK_PRODUCT, threshold: '50.0000001' }; + mockPrismaService.policy.findUnique.mockResolvedValue(ACTIVE_POLICY); + mockPrismaService.claim.findFirst.mockResolvedValue(null); + mockPrismaService.claim.create.mockResolvedValue({ id: 'claim-last-dp', status: 'PROCESSING' }); + mockOracleService.getLatestReading.mockResolvedValue({ + key: ACTIVE_POLICY.oracleKey, + value: BigInt(500_000_000), // 50.0000000 < 50.0000001 + confidence: 90, + }); + mockPolicyService.getProductById.mockResolvedValue(PRODUCT); + mockStellarService.invokeContract.mockResolvedValue('aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa'); + + const result = await service.autoProcess(POLICY_ID); + + expect(result).toBe('Paid'); + expect(mockStellarService.invokeContract).toHaveBeenCalled(); + }); + + // #712 regression: with the old `parseFloat(threshold) * 1e7` path, a large + // threshold loses low-order digits to double rounding, so a reading exactly + // equal to the threshold can drift below it and pay out. The scaled BigInt + // comparison keeps the boundary exact. + it('#712 — exact comparison holds for large thresholds (no float drift)', async () => { + const PRODUCT = { ...MOCK_PRODUCT, threshold: '1000000000000.0000001' }; + mockPrismaService.policy.findUnique.mockResolvedValue(ACTIVE_POLICY); + mockPrismaService.claim.findFirst.mockResolvedValue(null); + mockPrismaService.claim.create.mockResolvedValue({ id: 'claim-large', status: 'PROCESSING' }); + mockOracleService.getLatestReading.mockResolvedValue({ + key: ACTIVE_POLICY.oracleKey, + // Exactly equal to the threshold, as scaled BigInt. + value: BigInt('10000000000000000001'), + confidence: 90, + }); + mockPolicyService.getProductById.mockResolvedValue(PRODUCT); + mockPrismaService.claim.update.mockResolvedValue({}); + + const result = await service.autoProcess(POLICY_ID); + + expect(result).toBe('Rejected'); + expect(mockStellarService.invokeContract).not.toHaveBeenCalled(); + }); + // #259 — When getProductById returns null/undefined (product deactivated after // the policy was sold), the claim must fail loud for manual review rather than // silently substituting a threshold/comparison that may not match what the diff --git a/src/claims/claims.service.ts b/src/claims/claims.service.ts index b13f4f6..8e2560f 100644 --- a/src/claims/claims.service.ts +++ b/src/claims/claims.service.ts @@ -10,6 +10,7 @@ import { Prisma, ClaimStatus, PolicyStatus } from '@prisma/client'; import { WebhooksService } from '../common/events/webhooks.service'; import { StatusEventsService } from '../common/events/status-events.service'; import { ErrorCode } from '../common/errors/error-codes'; +import { compareScaledToThreshold, parseDecimalScaled, parseFixedPointBigInt } from '../common/utils/decimal'; import { financialBatch, financialTransaction, isSerializationFailure } from '../prisma/financial-transaction'; export type ClaimResult = 'Paid' | 'Rejected' | 'Expired' | 'AlreadyClaimed' | 'AlreadyProcessed' | 'PolicyNotActive' | 'PendingFinalPeriod'; @@ -260,12 +261,14 @@ export class ClaimsService { } // #245 — Guard against non-numeric threshold values. Product.threshold is a // free-text string at the service boundary (Decimal serialised via .toString()). - // parseFloat('') or parseFloat('N/A') yields NaN, and BigInt(NaN) throws a - // RangeError — crashing mid-flow after the claim row is already PROCESSING. - // Treat an unparseable threshold the same as a missing product: fail loud for - // manual review and revert the atomic gate so the policy can be retried. - const rawThreshold = parseFloat(product.threshold); - if (!isFinite(rawThreshold)) { + // #712 — Parse it exactly (scaled BigInt) instead of `parseFloat`, which loses + // precision and could put a boundary value on the wrong side of the trigger. + // An unparseable threshold is treated the same as a missing product: fail loud + // for manual review and revert the atomic gate so the policy can be retried. + let threshold: bigint; + try { + threshold = parseDecimalScaled(product.threshold); + } catch { this.logger.error( `Product ${policy.productId} has non-numeric threshold "${product.threshold}" — cannot evaluate trigger for policy ${policyId}, marking claim FAILED for manual review`, ); @@ -291,13 +294,13 @@ this.auditOp('Claim', claim.id, ClaimStatus.PROCESSING, ClaimStatus.FAILED, 'Non await this.policyStateCache?.invalidate(policyId); return 'Rejected'; } - const threshold = BigInt(Math.round(rawThreshold * 1e7)); const comparison = product.comparison; - const readingValue = BigInt(reading.value); - const triggerMet = comparison === 'LessThan' - ? readingValue < threshold - : readingValue > threshold; + // #712 — Compare the exact scaled values. Semantics are unchanged: 'LessThan' + // triggers strictly below the threshold, otherwise strictly above; a reading + // exactly equal to the threshold never triggers. + const readingValue = parseFixedPointBigInt(reading.value); + const triggerMet = compareScaledToThreshold(readingValue, threshold, comparison); this.logger.log( `Trigger eval: key=${reading.key} value=${reading.value} threshold=${threshold} triggerMet=${triggerMet}`, diff --git a/src/claims/policy-state-cache.service.spec.ts b/src/claims/policy-state-cache.service.spec.ts index 97d80e8..2d37a2b 100644 --- a/src/claims/policy-state-cache.service.spec.ts +++ b/src/claims/policy-state-cache.service.spec.ts @@ -44,7 +44,8 @@ describe('PolicyStateCacheService (#714)', () => { beforeEach(async () => { mockPrisma = { policy: { - findUnique: jest.fn().mockResolvedValue(JSON.parse(JSON.stringify(mockPolicy))), + // structuredClone (not JSON round-trip) so BigInt/Date fields survive. + findUnique: jest.fn().mockResolvedValue(structuredClone(mockPolicy)), findMany: jest.fn().mockResolvedValue([{ id: 'policy-uuid-1' }]), }, }; @@ -316,4 +317,44 @@ describe('PolicyStateCacheService (#714)', () => { expect(dbQueryReductionPct).toBe(95); }); }); + + describe('#712 — exact oracle threshold comparison', () => { + it('does not trigger when the reading equals the threshold', async () => { + mockOracle.getLatestReading.mockResolvedValue({ + ...mockReading, + value: BigInt(500_000_000), // exactly 50.0, equal to the '50.0' threshold + }); + + const state = await service.computePolicyState('policy-uuid-1'); + + expect(state?.triggerMet).toBe(false); + expect(state?.latestReadingValue).toBe('500000000'); + }); + + it('triggers when the reading is one unit below a LessThan threshold', async () => { + mockOracle.getLatestReading.mockResolvedValue({ + ...mockReading, + value: BigInt(499_999_999), // 49.9999999 < 50.0 + }); + + const state = await service.computePolicyState('policy-uuid-1'); + + expect(state?.triggerMet).toBe(true); + }); + + it('falls back to the reading timestamp when bucketStart is absent', async () => { + mockOracle.getLatestReading.mockResolvedValue({ + key: 'rainfall:station-101:monthly', + value: '450000000', + confidence: 90, + timestamp: 1_700_000_000, + source: 'open-meteo', + }); + + const state = await service.computePolicyState('policy-uuid-1'); + + expect(state?.latestReadingTimestamp).toBe(1_700_000_000_000); + expect(state?.triggerMet).toBe(true); + }); + }); }); diff --git a/src/claims/policy-state-cache.service.ts b/src/claims/policy-state-cache.service.ts index fe36f71..aa478c0 100644 --- a/src/claims/policy-state-cache.service.ts +++ b/src/claims/policy-state-cache.service.ts @@ -4,6 +4,7 @@ import { PrismaService } from '../prisma/prisma.service'; import { OracleService } from '../oracle/oracle.service'; import { PolicyStatus, ClaimStatus } from '@prisma/client'; import { StatusEventsService, PolicyStatusEvent, PolicyOracleKeyChangeEvent } from '../common/events/status-events.service'; +import { compareScaledToThreshold, parseDecimalScaled, parseFixedPointBigInt } from '../common/utils/decimal'; export const POLICY_DERIVED_STATE_TTL_SECONDS = 60; const IN_MEMORY_MAX_ENTRIES = 2000; @@ -305,8 +306,15 @@ export class PolicyStateCacheService implements OnModuleInit, OnModuleDestroy { const reading = await this.oracleService.getLatestReading(policy.oracleKey); const product = policy.product; - const rawThreshold = parseFloat(product.threshold.toString()); - const thresholdBigInt = isFinite(rawThreshold) ? BigInt(Math.round(rawThreshold * 1e7)) : 0n; + // #712 — Parse the threshold exactly (scaled BigInt) instead of parseFloat. + // Unlike the payout path, a non-numeric threshold here keeps the previous + // graceful behaviour (treated as 0) rather than throwing. + let thresholdBigInt: bigint; + try { + thresholdBigInt = parseDecimalScaled(product.threshold.toString()); + } catch { + thresholdBigInt = 0n; + } let triggerMet = false; let readingValueStr: string | null = null; @@ -314,12 +322,13 @@ export class PolicyStateCacheService implements OnModuleInit, OnModuleDestroy { if (reading) { readingValueStr = reading.value.toString(); - readingTimestamp = reading.bucketStart.getTime(); - const readingVal = BigInt(reading.value); - triggerMet = - product.comparison === 'LessThan' - ? readingVal < thresholdBigInt - : readingVal > thresholdBigInt; + // #712 — Prefer the reading's own bucket start, falling back to its + // timestamp: OracleReading does not always carry a bucketStart. + readingTimestamp = reading.bucketStart + ? reading.bucketStart.getTime() + : reading.timestamp * 1000; + const readingVal = parseFixedPointBigInt(reading.value); + triggerMet = compareScaledToThreshold(readingVal, thresholdBigInt, product.comparison); } const claims = policy.claims ?? []; diff --git a/src/common/filters/http-exception.filter.spec.ts b/src/common/filters/http-exception.filter.spec.ts index e1b5d82..67c0867 100644 --- a/src/common/filters/http-exception.filter.spec.ts +++ b/src/common/filters/http-exception.filter.spec.ts @@ -150,6 +150,27 @@ describe('GlobalExceptionFilter', () => { const body = json.mock.calls[0][0]; expect(body.correlationId).toBe('req-corr-123'); }); + + // #713 — error responses bypass the versioning interceptor, so the filter + // must stamp X-API-Version itself. + it('#713 — stamps X-API-Version on error responses', () => { + const host = mockHost('GET', '/api/v2/policies/missing-uuid'); + const error = new HttpException('Policy not found', HttpStatus.NOT_FOUND); + + filter.catch(error, host); + + expect(res.setHeader).toHaveBeenCalledWith('X-API-Version', '2.0.0'); + }); + + it('#713 — defaults to 1.0.0 for unversioned error paths', () => { + const host = mockHost('GET', '/health'); + const error = new HttpException('Unavailable', HttpStatus.BAD_REQUEST); + + filter.catch(error, host); + + expect(res.setHeader).toHaveBeenCalledWith('X-API-Version', '1.0.0'); + expect(res.setHeader).toHaveBeenCalledWith('Deprecation', 'true'); + }); }); describe('Logging and edge cases', () => { diff --git a/src/common/filters/http-exception.filter.ts b/src/common/filters/http-exception.filter.ts index ed2bbcc..0f1e35e 100644 --- a/src/common/filters/http-exception.filter.ts +++ b/src/common/filters/http-exception.filter.ts @@ -6,6 +6,7 @@ import { Request, Response } from 'express'; import { errorCodeFromBody, errorCodeFromStatus } from '../errors/error-codes'; import { serializeBigInt } from '../interceptors/bigint-serializer.interceptor'; import { buildProblemDetails } from '../errors/problem-details'; +import { applyApiVersionHeaders, resolveRequestedVersion } from '../versioning/api-version'; @Catch() export class GlobalExceptionFilter implements ExceptionFilter { @@ -48,6 +49,14 @@ export class GlobalExceptionFilter implements ExceptionFilter { const errorCode = errorCodeFromBody(rawResponse) ?? errorCodeFromStatus(status); const path = req.originalUrl || req.url; + // #713 — Error responses never pass through the versioning interceptor's + // success path, so stamp the version header here too. + const versionHeader = req.headers?.['x-api-version']; + applyApiVersionHeaders( + res, + resolveRequestedVersion(path, Array.isArray(versionHeader) ? versionHeader[0] : versionHeader), + ); + // #677 — Adopt RFC 7807 problem details with Content-Type application/problem+json. // Includes type, title, status, detail, instance, extension members (errors, correlationId), // and preserves legacy fields (success, statusCode, error, message, errorCode, path, timestamp) diff --git a/src/common/interceptors/versioning.interceptor.spec.ts b/src/common/interceptors/versioning.interceptor.spec.ts new file mode 100644 index 0000000..ecc7ff6 --- /dev/null +++ b/src/common/interceptors/versioning.interceptor.spec.ts @@ -0,0 +1,80 @@ +import { BadRequestException } from '@nestjs/common'; +import { lastValueFrom, of } from 'rxjs'; +import { VersioningInterceptor } from './versioning.interceptor'; +import { API_VERSION_HEADER } from '../versioning/api-version'; + +function makeContext(url: string, headers: Record = {}) { + const setHeader = jest.fn(); + const res = { setHeader, headersSent: false }; + const req = { originalUrl: url, url, headers }; + const ctx = { + switchToHttp: () => ({ + getRequest: () => req, + getResponse: () => res, + }), + } as any; + return { ctx, setHeader }; +} + +describe('VersioningInterceptor (#713)', () => { + const interceptor = new VersioningInterceptor({} as any); + + it('sets X-API-Version before the handler runs (so streamed responses get it)', () => { + const { ctx, setHeader } = makeContext('/api/v1/policies'); + + let headerSetBeforeHandle = false; + const next = { + handle: jest.fn(() => { + headerSetBeforeHandle = setHeader.mock.calls.length > 0; + return of({ success: true }); + }), + }; + + interceptor.intercept(ctx, next as any); + + expect(headerSetBeforeHandle).toBe(true); + expect(setHeader).toHaveBeenCalledWith(API_VERSION_HEADER, '1.0.0'); + }); + + it('marks v1 as deprecated', () => { + const { ctx, setHeader } = makeContext('/api/v1/policies'); + interceptor.intercept(ctx, { handle: () => of({}) } as any); + + expect(setHeader).toHaveBeenCalledWith('Deprecation', 'true'); + expect(setHeader).toHaveBeenCalledWith('Link', '; rel="successor"'); + }); + + it('serves v2 with the right version and no deprecation headers', () => { + const { ctx, setHeader } = makeContext('/api/v2/policies'); + interceptor.intercept(ctx, { handle: () => of({}) } as any); + + expect(setHeader).toHaveBeenCalledWith(API_VERSION_HEADER, '2.0.0'); + expect(setHeader).not.toHaveBeenCalledWith('Deprecation', expect.anything()); + }); + + it('honours an explicit x-api-version header on an unversioned path', () => { + const { ctx, setHeader } = makeContext('/health', { 'x-api-version': 'v2' }); + interceptor.intercept(ctx, { handle: () => of({}) } as any); + + expect(setHeader).toHaveBeenCalledWith(API_VERSION_HEADER, '2.0.0'); + }); + + it('rejects unsupported versions before handling', () => { + const { ctx } = makeContext('/api/v3/policies'); + const next = { handle: jest.fn(() => of({})) }; + + expect(() => interceptor.intercept(ctx, next as any)).toThrow(BadRequestException); + expect(next.handle).not.toHaveBeenCalled(); + }); + + it('adds the version to a success envelope metadata', async () => { + const { ctx } = makeContext('/api/v1/policies'); + const data: any = { success: true }; + + const result = await lastValueFrom( + interceptor.intercept(ctx, { handle: () => of(data) } as any), + ); + + expect(result.metadata.version).toBe('1.0.0'); + }); +}); diff --git a/src/common/interceptors/versioning.interceptor.ts b/src/common/interceptors/versioning.interceptor.ts index 908c219..e66b307 100644 --- a/src/common/interceptors/versioning.interceptor.ts +++ b/src/common/interceptors/versioning.interceptor.ts @@ -2,13 +2,13 @@ import { Injectable, NestInterceptor, ExecutionContext, CallHandler, BadRequestE import { Reflector } from '@nestjs/core'; import { Observable } from 'rxjs'; import { tap } from 'rxjs/operators'; - -const API_VERSIONS: Record = { - v1: '1.0.0', - v2: '2.0.0', -}; - -const SUPPORTED_VERSIONS = Object.keys(API_VERSIONS); +import { + SUPPORTED_API_VERSIONS, + applyApiVersionHeaders, + apiVersionHeaderValue, + isSupportedVersion, + resolveRequestedVersion, +} from '../versioning/api-version'; @Injectable() export class VersioningInterceptor implements NestInterceptor { @@ -18,28 +18,29 @@ export class VersioningInterceptor implements NestInterceptor { const request = context.switchToHttp().getRequest(); const response = context.switchToHttp().getResponse(); - const path = request.path; - const versionMatch = path.match(/^\/api\/(v\d+)/); - const version = versionMatch ? versionMatch[1] : 'v1'; + const path = request.originalUrl || request.url || request.path; + const header = request.headers?.['x-api-version']; + const version = resolveRequestedVersion( + path, + Array.isArray(header) ? header[0] : header, + ); // #691 — Reject unknown versions instead of silently defaulting to v1. - if (!API_VERSIONS[version]) { + if (!isSupportedVersion(version)) { throw new BadRequestException( - `Unsupported API version: ${version}. Supported versions: ${SUPPORTED_VERSIONS.join(', ')}.` + `Unsupported API version: ${version}. Supported versions: ${SUPPORTED_API_VERSIONS.join(', ')}.`, ); } - const apiVersion = API_VERSIONS[version]; + const apiVersion = apiVersionHeaderValue(version)!; + + // #713 — Emit the version headers up-front rather than in `tap`. `tap` only + // runs once the handler emits, which is too late for streamed responses + // (SSE/NDJSON) whose headers are flushed before the body completes. + applyApiVersionHeaders(response, version); return next.handle().pipe( tap((data) => { - response.setHeader('X-API-Version', apiVersion); - - if (version === 'v1') { - response.setHeader('Deprecation', 'true'); - response.setHeader('Link', '; rel="successor"'); - } - if (data && data.success !== undefined) { data.metadata = data.metadata || {}; data.metadata.version = apiVersion; @@ -47,4 +48,4 @@ export class VersioningInterceptor implements NestInterceptor { }), ); } -} \ No newline at end of file +} diff --git a/src/common/security/response-header-hardening.spec.ts b/src/common/security/response-header-hardening.spec.ts new file mode 100644 index 0000000..9c7c253 --- /dev/null +++ b/src/common/security/response-header-hardening.spec.ts @@ -0,0 +1,202 @@ +import express from 'express'; +import type { Express } from 'express'; +import http from 'http'; +import type { AddressInfo } from 'net'; +import { + LEAKY_RESPONSE_HEADERS, + SecurityHeaderApp, + applySecurityHeaders, + createHeaderStrippingMiddleware, +} from './response-header-hardening'; + +/** Minimal fake Express `Response` that records whatever reaches the socket. */ +function fakeRes() { + const headers = new Map(); + const res = { + lastWriteHead: undefined as unknown[] | undefined, + setHeader(name: string, value: unknown) { + headers.set(name.toLowerCase(), value); + return res; + }, + getHeader(name: string) { + return headers.get(name.toLowerCase()); + }, + removeHeader(name: string) { + headers.delete(name.toLowerCase()); + }, + getHeaderNames() { + return Array.from(headers.keys()); + }, + writeHead(...args: unknown[]) { + res.lastWriteHead = args; + return res; + }, + has(name: string) { + return headers.has(name.toLowerCase()); + }, + }; + return res; +} + +function runMiddleware(res: ReturnType) { + const middleware = createHeaderStrippingMiddleware(); + const next = jest.fn(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + middleware({} as any, res as any, next); + return next; +} + +async function withServer( + app: Express, + fn: (port: number) => Promise, +): Promise { + const server = app.listen(0); + await new Promise((resolve) => server.once('listening', () => resolve())); + const { port } = server.address() as AddressInfo; + try { + await fn(port); + } finally { + await new Promise((resolve) => server.close(() => resolve())); + } +} + +function request( + port: number, + path: string, +): Promise<{ status: number; headers: http.IncomingHttpHeaders }> { + return new Promise((resolve, reject) => { + http + .get({ host: '127.0.0.1', port, path }, (res) => { + res.resume(); + res.on('end', () => resolve({ status: res.statusCode ?? 0, headers: res.headers })); + }) + .on('error', reject); + }); +} + +describe('response header hardening (#711)', () => { + describe('createHeaderStrippingMiddleware', () => { + it('strips every leaky header set after the middleware runs', () => { + const res = fakeRes(); + const next = runMiddleware(res); + expect(next).toHaveBeenCalledTimes(1); + + for (const name of LEAKY_RESPONSE_HEADERS) { + res.setHeader(name, 'some-value'); + } + + for (const name of LEAKY_RESPONSE_HEADERS) { + expect(res.has(name)).toBe(false); + } + }); + + it('is case-insensitive', () => { + const res = fakeRes(); + runMiddleware(res); + + res.setHeader('sErVeR', 'nginx'); + res.setHeader('X-POWERED-BY', 'Express'); + + expect(res.has('Server')).toBe(false); + expect(res.has('X-Powered-By')).toBe(false); + }); + + it('leaves other headers untouched', () => { + const res = fakeRes(); + runMiddleware(res); + + res.setHeader('Content-Type', 'application/json'); + res.setHeader('X-Total-Count', '42'); + + expect(res.getHeader('Content-Type')).toBe('application/json'); + expect(res.getHeader('X-Total-Count')).toBe('42'); + }); + + it('removes leaky headers that were already set before the middleware', () => { + const res = fakeRes(); + res.setHeader('Server', 'upstream/1.2.3'); + res.setHeader('X-Runtime', '0.012'); + + runMiddleware(res); + + expect(res.has('Server')).toBe(false); + expect(res.has('X-Runtime')).toBe(false); + }); + + it('filters headers passed directly to writeHead', () => { + const res = fakeRes(); + runMiddleware(res); + + res.writeHead(200, { + Server: 'nginx', + 'X-Powered-By': 'Express', + 'Content-Type': 'text/plain', + }); + + const headersArg = res.lastWriteHead?.[1] as Record; + expect(headersArg).not.toHaveProperty('Server'); + expect(headersArg).not.toHaveProperty('X-Powered-By'); + expect(headersArg).toHaveProperty('Content-Type', 'text/plain'); + }); + + it('supports a custom denylist', () => { + const res = fakeRes(); + const middleware = createHeaderStrippingMiddleware(['X-Internal-Trace']); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + middleware({} as any, res as any, jest.fn()); + + res.setHeader('X-Internal-Trace', 'abc'); + res.setHeader('Server', 'nginx'); + + expect(res.has('X-Internal-Trace')).toBe(false); + // Not in the custom list, so it survives this layer. + expect(res.has('Server')).toBe(true); + }); + }); + + describe('applySecurityHeaders', () => { + it('disables x-powered-by and installs helmet + stripping middleware', () => { + const app: SecurityHeaderApp = { + use: jest.fn().mockReturnThis(), + disable: jest.fn().mockReturnThis(), + }; + + const strip = applySecurityHeaders(app); + + expect(app.disable).toHaveBeenCalledWith('x-powered-by'); + expect(app.use).toHaveBeenCalledTimes(2); + expect(typeof strip).toBe('function'); + }); + + it('removes X-Powered-By and Server end-to-end', async () => { + const app = express(); + applySecurityHeaders(app as unknown as SecurityHeaderApp); + app.get('/ok', (_req, res) => { + res.json({ ok: true }); + }); + app.get('/upstream', (_req, res) => { + // Simulate a reverse proxy / upstream stack adding leaky headers. + res.setHeader('Server', 'nginx/1.25.0'); + res.setHeader('X-Runtime', '0.010'); + res.setHeader('Via', '1.1 vegur'); + res.json({ ok: true }); + }); + + await withServer(app, async (port) => { + const plain = await request(port, '/ok'); + expect(plain.status).toBe(200); + expect(plain.headers['x-powered-by']).toBeUndefined(); + expect(plain.headers['server']).toBeUndefined(); + // helmet's standard headers are still applied. + expect(plain.headers['x-content-type-options']).toBe('nosniff'); + + const upstream = await request(port, '/upstream'); + expect(upstream.status).toBe(200); + expect(upstream.headers['server']).toBeUndefined(); + expect(upstream.headers['x-runtime']).toBeUndefined(); + expect(upstream.headers['via']).toBeUndefined(); + expect(upstream.headers['x-powered-by']).toBeUndefined(); + }); + }); + }); +}); diff --git a/src/common/security/response-header-hardening.ts b/src/common/security/response-header-hardening.ts new file mode 100644 index 0000000..53df0df --- /dev/null +++ b/src/common/security/response-header-hardening.ts @@ -0,0 +1,139 @@ +import type { NextFunction, Request, RequestHandler, Response } from 'express'; +import type { HelmetOptions } from 'helmet'; +import helmet from 'helmet'; + +/** Minimal surface needed from an Express/Nest HTTP application. */ +export interface SecurityHeaderApp { + use(...args: unknown[]): unknown; + /** Present on Express/NestExpressApplication; absent on some adapters. */ + disable?(setting: string): unknown; +} + +/** + * #711 — Response header hardening. + * + * Two classes of header leak implementation details and give attackers an easy + * target list: + * - `X-Powered-By: Express` is set by Express itself. + * - `Server` (and friends such as `X-AspNet-Version`, `X-Runtime`, `Via`) are + * set by reverse proxies / upstream stacks that sit in front of the API. + * + * We strip them from every response. Stripping is done at the header level (by + * wrapping `res.setHeader` / `res.writeHead`) so it also covers headers added + * after this middleware runs, by the framework or any downstream middleware. + */ +export const LEAKY_RESPONSE_HEADERS: readonly string[] = [ + 'Server', + 'X-Powered-By', + 'X-AspNet-Version', + 'X-AspNetMvc-Version', + 'X-Runtime', + 'Via', +]; + +/** + * Express middleware that drops any response header whose (case-insensitive) + * name is in `denylist`. Safe to apply more than once; each layer simply + * filters the set of headers that reach the socket. + */ +export function createHeaderStrippingMiddleware( + denylist: readonly string[] = LEAKY_RESPONSE_HEADERS, +): RequestHandler { + const denied = new Set(denylist.map((name) => name.toLowerCase())); + + return function stripLeakyResponseHeaders(_req: Request, res: Response, next: NextFunction): void { + const originalSetHeader = res.setHeader.bind(res); + const originalWriteHead = res.writeHead.bind(res); + + res.setHeader = function patchedSetHeader( + this: Response, + name: string, + value: number | string | readonly string[], + ): Response { + if (denied.has(String(name).toLowerCase())) { + return this; + } + originalSetHeader(name, value); + return this; + } as unknown as typeof res.setHeader; + + // `writeHead(status, headers)` can be called directly with a header bag, + // bypassing `setHeader`; filter that bag too. + res.writeHead = function patchedWriteHead(this: Response, ...args: unknown[]): Response { + const headersIndex = args.findIndex( + (arg) => arg !== null && typeof arg === 'object' && !Array.isArray(arg), + ); + if (headersIndex !== -1) { + const headers = { ...(args[headersIndex] as Record) }; + for (const key of Object.keys(headers)) { + if (denied.has(key.toLowerCase())) delete headers[key]; + } + args[headersIndex] = headers; + } + return (originalWriteHead as (...innerArgs: unknown[]) => Response).apply(this, args); + } as unknown as typeof res.writeHead; + + // Headers already set (e.g. by an upstream layer before this runs) must go. + const existing = typeof res.getHeaderNames === 'function' ? res.getHeaderNames() : []; + for (const name of existing) { + if (denied.has(name.toLowerCase())) { + res.removeHeader(name); + } + } + + next(); + }; +} + +export interface SecurityHeaderOptions { + /** Extra header names to strip, in addition to {@link LEAKY_RESPONSE_HEADERS}. */ + extraHeadersToStrip?: readonly string[]; + /** Helmet options; merged over the secure defaults. */ + helmetOptions?: HelmetOptions; +} + +/** + * Apply the full response-header hardening stack to an Express/Nest application: + * 1. disable Express's `X-Powered-By` at the framework level (cheapest, so it + * is never even generated); + * 2. install helmet for the standard security headers (with `hidePoweredBy` + * on as a second line of defence); + * 3. strip the remaining leaky headers from every response. + * + * Returns the header-stripping middleware so callers (and tests) can inspect or + * reuse it. + */ +export function applySecurityHeaders( + app: SecurityHeaderApp, + options: SecurityHeaderOptions = {}, +): RequestHandler { + // 1. Express-level (also mirrors helmet's hidePoweredBy behaviour). + if (typeof app.disable === 'function') { + app.disable('x-powered-by'); + } + + // 2. Standard security headers. + app.use( + helmet({ + referrerPolicy: { policy: 'strict-origin-when-cross-origin' }, + hsts: { + maxAge: 31536000, + includeSubDomains: true, + preload: true, + }, + ...options.helmetOptions, + // Hard-code this on regardless of caller options: leaking the stack is + // the thing this module exists to prevent. + hidePoweredBy: true, + }), + ); + + // 3. Strip leftover leaky headers. + const denylist = [ + ...LEAKY_RESPONSE_HEADERS, + ...(options.extraHeadersToStrip ?? []), + ]; + const strip = createHeaderStrippingMiddleware(denylist); + app.use(strip); + return strip; +} diff --git a/src/common/utils/decimal.spec.ts b/src/common/utils/decimal.spec.ts new file mode 100644 index 0000000..a439fd4 --- /dev/null +++ b/src/common/utils/decimal.spec.ts @@ -0,0 +1,153 @@ +import { + DEFAULT_DECIMAL_SCALE, + InvalidDecimalError, + compareScaledToThreshold, + evaluateThreshold, + parseDecimalScaled, + parseFixedPointBigInt, +} from './decimal'; + +describe('decimal utils (#712)', () => { + describe('parseDecimalScaled', () => { + it('uses a 7-decimal default scale', () => { + expect(DEFAULT_DECIMAL_SCALE).toBe(7); + }); + + it('parses exact 7-dp strings without float error', () => { + expect(parseDecimalScaled('50.0000000')).toBe(500_000_000n); + expect(parseDecimalScaled('0.0000001')).toBe(1n); + expect(parseDecimalScaled('0.1')).toBe(1_000_000n); + expect(parseDecimalScaled('-2.5')).toBe(-25_000_000n); + expect(parseDecimalScaled('123456789.1234567')).toBe(1_234_567_891_234_567n); + expect(parseDecimalScaled('1')).toBe(10_000_000n); + expect(parseDecimalScaled('0')).toBe(0n); + }); + + it('accepts numbers, bigints and sign forms', () => { + expect(parseDecimalScaled(50)).toBe(500_000_000n); + expect(parseDecimalScaled(50n)).toBe(500_000_000n); + expect(parseDecimalScaled('+5')).toBe(50_000_000n); + expect(parseDecimalScaled('.5')).toBe(5_000_000n); + expect(parseDecimalScaled('5.')).toBe(50_000_000n); + }); + + it('accepts decimal-like objects (e.g. Prisma.Decimal) via toString', () => { + expect(parseDecimalScaled({ toString: () => '12.5' })).toBe(125_000_000n); + }); + + it('is exact where parseFloat(binary) is not', () => { + // 20 significant digits: the old `parseFloat(x) * 1e7` path cannot hold this. + const exact = parseDecimalScaled('1000000000000.0000001'); + expect(exact).toBe(10_000_000_000_000_000_001n); + }); + + it('recovers the intended value from float noise (0.1 + 0.2)', () => { + // String(0.1 + 0.2) === '0.30000000000000004' + expect(parseDecimalScaled(0.1 + 0.2)).toBe(3_000_000n); + }); + + it('expands scientific notation exactly', () => { + expect(parseDecimalScaled('1e-7')).toBe(1n); + expect(parseDecimalScaled('1.5e2')).toBe(1_500_000_000n); + expect(parseDecimalScaled('1.5e-1')).toBe(1_500_000n); + expect(parseDecimalScaled(1e21)).toBe(10n ** 28n); + }); + + it('rounds digits beyond the scale half away from zero', () => { + // 0.00000005 -> 0.0000001 + expect(parseDecimalScaled('0.00000005')).toBe(1n); + // 1.23456789 -> 1.2345679 + expect(parseDecimalScaled('1.23456789')).toBe(12_345_679n); + // -1.23456785 -> -1.2345679 + expect(parseDecimalScaled('-1.23456785')).toBe(-12_345_679n); + }); + + it('honours a custom scale', () => { + expect(parseDecimalScaled('1.2345', 2)).toBe(123n); + expect(parseDecimalScaled('1.235', 2)).toBe(124n); + expect(parseDecimalScaled('10', 0)).toBe(10n); + }); + + it.each([ + ['', 'empty string'], + [' ', 'whitespace'], + ['N/A', 'non-numeric text'], + ['abc', 'letters'], + ['1.2.3', 'multiple points'], + ['.', 'bare point'], + ['--5', 'double sign'], + [null, 'null'], + [undefined, 'undefined'], + [NaN, 'NaN'], + [Infinity, '-/Infinity'], + [{}, 'object'], + ])('throws InvalidDecimalError for %p (%s)', (value) => { + expect(() => parseDecimalScaled(value)).toThrow(InvalidDecimalError); + }); + }); + + describe('parseFixedPointBigInt', () => { + it('parses integer fixed-point values', () => { + expect(parseFixedPointBigInt('200000000')).toBe(200_000_000n); + expect(parseFixedPointBigInt(200_000_000)).toBe(200_000_000n); + expect(parseFixedPointBigInt(200_000_000n)).toBe(200_000_000n); + expect(parseFixedPointBigInt('-5')).toBe(-5n); + }); + + it('parses a decimal form rather than truncating it', () => { + expect(parseFixedPointBigInt('200.0000000')).toBe(2_000_000_000n); + }); + + it('rejects non-numeric values', () => { + expect(() => parseFixedPointBigInt('nope')).toThrow(InvalidDecimalError); + expect(() => parseFixedPointBigInt(null)).toThrow(InvalidDecimalError); + }); + }); + + describe('compareScaledToThreshold', () => { + const threshold = 500_000_000n; // 50.0000000 + + it('triggers strictly below for LessThan', () => { + expect(compareScaledToThreshold(499_999_999n, threshold, 'LessThan')).toBe(true); + expect(compareScaledToThreshold(500_000_001n, threshold, 'LessThan')).toBe(false); + }); + + it('triggers strictly above for everything else', () => { + expect(compareScaledToThreshold(500_000_001n, threshold, 'GreaterThan')).toBe(true); + expect(compareScaledToThreshold(499_999_999n, threshold, 'GreaterThan')).toBe(false); + // Unknown comparisons keep the historical else-branch (greater-than). + expect(compareScaledToThreshold(500_000_001n, threshold, 'gte')).toBe(true); + }); + + it('never triggers when the reading equals the threshold', () => { + expect(compareScaledToThreshold(threshold, threshold, 'LessThan')).toBe(false); + expect(compareScaledToThreshold(threshold, threshold, 'GreaterThan')).toBe(false); + }); + }); + + describe('evaluateThreshold', () => { + it('matches the documented claim-boundary behaviour', () => { + // LessThan: value below threshold triggers. + expect(evaluateThreshold('200000000', '50.0000000', 'LessThan')).toBe(true); + expect(evaluateThreshold('800000000', '50.0000000', 'LessThan')).toBe(false); + // GreaterThan: value above threshold triggers. + expect(evaluateThreshold('450000000', '30.0000000', 'GreaterThan')).toBe(true); + expect(evaluateThreshold('150000000', '30.0000000', 'GreaterThan')).toBe(false); + }); + + it('treats an exactly-equal reading as not triggered', () => { + expect(evaluateThreshold('500000000', '50.0000000', 'LessThan')).toBe(false); + expect(evaluateThreshold('500000000', '50.0000000', 'GreaterThan')).toBe(false); + }); + + it('is exact for high-precision thresholds (regression)', () => { + // Reading is 1 unit of 1e-7 below the threshold; must trigger. + expect( + evaluateThreshold('500000000', '50.0000001', 'LessThan'), + ).toBe(true); + expect( + evaluateThreshold('500000000', '49.9999999', 'GreaterThan'), + ).toBe(true); + }); + }); +}); diff --git a/src/common/utils/decimal.ts b/src/common/utils/decimal.ts new file mode 100644 index 0000000..837ab92 --- /dev/null +++ b/src/common/utils/decimal.ts @@ -0,0 +1,219 @@ +import { BadRequestException } from '@nestjs/common'; + +/** + * #712 — Exact decimal arithmetic for oracle threshold comparison. + * + * Product thresholds are stored as `Decimal(20,7)` and oracle reading values as + * 7-decimal fixed point. Comparing them via `parseFloat(threshold) * 1e7` loses + * precision (binary floating point cannot represent most decimal fractions), so + * a threshold the operator typed as `0.3000000` can end up on the wrong side of + * an exact boundary and pay out (or not) incorrectly. + * + * Everything here works on `BigInt` scaled by 10^`DEFAULT_DECIMAL_SCALE`, which + * is exact for any value representable at 7 decimal places. + */ + +/** Decimal places used across the money/oracle contracts (Stellar USDC precision). */ +export const DEFAULT_DECIMAL_SCALE = 7; + +/** Raised when a value cannot be interpreted as a decimal number. */ +export class InvalidDecimalError extends BadRequestException { + constructor(message = 'Value is not a valid decimal number') { + super(message); + this.name = 'InvalidDecimalError'; + } +} + +/** `true` if the value has an exponent, e.g. `1e-7` / `1.2E+3`. */ +const EXPONENT_PATTERN = /^[+-]?(?:\d+\.?\d*|\.\d+)[eE][+-]?\d+$/; +/** `true` if the value is a plain (non-exponent) decimal, e.g. `-12.34` / `.5` / `5.`. */ +const PLAIN_DECIMAL_PATTERN = /^([+-]?)(?:(\d+)(?:\.(\d*))?|\.(\d+))$/; + +/** + * Rewrite scientific notation (`1e-7`, `2.5E+3`) into plain decimal notation + * without any rounding, so the caller's regexes never need to know about it. + */ +function expandExponent(input: string): string { + if (!EXPONENT_PATTERN.test(input)) return input; + + const match = /^([+-]?)(\d*)(?:\.(\d*))?[eE]([+-]?\d+)$/.exec(input); + if (!match) return input; + + const negative = match[1] === '-'; + const intPart = match[2] ?? ''; + const fracPart = match[3] ?? ''; + const exponent = parseInt(match[4], 10); + + const digits = intPart + fracPart; + const pointPosition = intPart.length + exponent; + + let plain: string; + if (pointPosition <= 0) { + plain = `0.${'0'.repeat(-pointPosition)}${digits}`; + } else if (pointPosition >= digits.length) { + plain = `${digits}${'0'.repeat(pointPosition - digits.length)}`; + } else { + plain = `${digits.slice(0, pointPosition)}.${digits.slice(pointPosition)}`; + } + + return negative ? `-${plain}` : plain; +} + +/** Normalise any accepted input type to a plain (non-exponent) decimal string. */ +function toPlainDecimalString(value: unknown): string { + if (typeof value === 'bigint') return value.toString(); + + if (typeof value === 'number') { + if (!Number.isFinite(value)) { + throw new InvalidDecimalError(`Value is not a finite number: ${value}`); + } + return expandExponent(String(value)); + } + + if (typeof value === 'string') { + const trimmed = value.trim(); + if (trimmed === '') { + throw new InvalidDecimalError('Value is an empty string'); + } + return expandExponent(trimmed); + } + + // Accept decimal-like objects (e.g. Prisma.Decimal) via their string form, + // but not arbitrary objects whose default toString is "[object Object]". + if (value !== null && typeof value === 'object') { + const stringified = String(value).trim(); + if (stringified !== '' && !stringified.startsWith('[object ')) { + return expandExponent(stringified); + } + } + + throw new InvalidDecimalError( + `Unsupported decimal value type: ${value === null ? 'null' : typeof value}`, + ); +} + +/** + * Parse a decimal value into an exact `BigInt` scaled by 10^`scale` + * (default {@link DEFAULT_DECIMAL_SCALE}). + * + * Accepts a string, finite number or bigint. Sign is preserved; scientific + * notation (including the `1e-7` form that `String(number)` produces for small + * floats) is expanded exactly. Digits beyond `scale` are rounded half-away-from + * -zero, matching how the value would be stored in a 7-dp column. + * + * Throws {@link InvalidDecimalError} for empty strings, `NaN`, `Infinity` and + * anything that is not a number at all. + * + * @example + * parseDecimalScaled('50.0000000') // 500000000n + * parseDecimalScaled('0.0000001') // 1n + * parseDecimalScaled(50) // 500000000n + */ +export function parseDecimalScaled( + value: unknown, + scale: number = DEFAULT_DECIMAL_SCALE, +): bigint { + if (!Number.isInteger(scale) || scale < 0) { + throw new InvalidDecimalError(`Invalid decimal scale: ${scale}`); + } + + const plain = toPlainDecimalString(value); + const match = PLAIN_DECIMAL_PATTERN.exec(plain); + if (!match) { + throw new InvalidDecimalError(`Value is not a valid decimal number: "${plain}"`); + } + + const sign = match[1] === '-' ? -1n : 1n; + const intPart = match[2] ?? '0'; + const fracPart = match[3] ?? match[4] ?? ''; + + const combined = intPart + fracPart; + if (!/\d/.test(combined)) { + throw new InvalidDecimalError(`Value is not a valid decimal number: "${plain}"`); + } + + let mantissa = BigInt(combined); + const excessDigits = fracPart.length - scale; + + if (excessDigits > 0) { + const divisor = 10n ** BigInt(excessDigits); + const quotient = mantissa / divisor; + const remainder = mantissa % divisor; + // Round half away from zero. + mantissa = remainder * 2n >= divisor ? quotient + 1n : quotient; + } else if (excessDigits < 0) { + mantissa = mantissa * 10n ** BigInt(-excessDigits); + } + + return sign * mantissa; +} + +/** + * Parse an already-scaled oracle reading value (an integer count of 10^-scale + * units, e.g. `200000000` for 20.0 at scale 7) into a `BigInt`. + * + * Accepts the exact integer string form produced by `BigInt#toString`, a bigint + * or an integer number. A value carrying a decimal point is parsed as a decimal + * (so a caller passing `200.0000000` still works) rather than silently truncated. + */ +export function parseFixedPointBigInt( + value: unknown, + scale: number = DEFAULT_DECIMAL_SCALE, +): bigint { + if (typeof value === 'bigint') return value; + + if (typeof value === 'number') { + if (!Number.isInteger(value)) { + return parseDecimalScaled(value, scale); + } + return BigInt(value); + } + + if (typeof value === 'string') { + const trimmed = value.trim(); + if (/^[+-]?\d+$/.test(trimmed)) { + return BigInt(trimmed); + } + return parseDecimalScaled(trimmed, scale); + } + + throw new InvalidDecimalError( + `Unsupported reading value type: ${value === null ? 'null' : typeof value}`, + ); +} + +/** + * Compare two values already scaled by 10^`scale`, preserving the trigger + * semantics used across the app: `'LessThan'` triggers when the reading is + * strictly below the threshold, anything else triggers when it is strictly + * above. A reading exactly equal to the threshold never triggers. + */ +export function compareScaledToThreshold( + readingScaled: bigint, + thresholdScaled: bigint, + comparison: unknown, +): boolean { + return comparison === 'LessThan' + ? readingScaled < thresholdScaled + : readingScaled > thresholdScaled; +} + +/** + * Convenience wrapper that parses `threshold`/`reading` exactly and compares + * them. Accepts the raw product threshold (any decimal form) and the raw oracle + * reading value (7-dp fixed point). Throws {@link InvalidDecimalError} if either + * value is not numeric — callers that must degrade gracefully should parse the + * threshold themselves and use {@link compareScaledToThreshold}. + */ +export function evaluateThreshold( + reading: unknown, + threshold: unknown, + comparison: unknown, + scale: number = DEFAULT_DECIMAL_SCALE, +): boolean { + return compareScaledToThreshold( + parseFixedPointBigInt(reading, scale), + parseDecimalScaled(threshold, scale), + comparison, + ); +} diff --git a/src/common/versioning/api-version.spec.ts b/src/common/versioning/api-version.spec.ts new file mode 100644 index 0000000..b101331 --- /dev/null +++ b/src/common/versioning/api-version.spec.ts @@ -0,0 +1,146 @@ +import { + API_VERSION_HEADER, + API_VERSIONS, + DEFAULT_API_VERSION, + SUPPORTED_API_VERSIONS, + apiVersionHeaderValue, + applyApiVersionHeaders, + createApiVersionMiddleware, + isSupportedVersion, + normalizeVersionToken, + resolveRequestedVersion, + versionSegmentFromPath, +} from './api-version'; + +describe('api-version helpers (#713)', () => { + describe('versionSegmentFromPath', () => { + it('extracts the version segment from a path', () => { + expect(versionSegmentFromPath('/api/v1/policies')).toBe('v1'); + expect(versionSegmentFromPath('/api/v2/claims?status=open')).toBe('v2'); + expect(versionSegmentFromPath('/api/v2')).toBe('v2'); + }); + + it('returns the (possibly unsupported) segment so it can be rejected', () => { + expect(versionSegmentFromPath('/api/v3/policies')).toBe('v3'); + }); + + it('returns undefined when the path has no version segment', () => { + expect(versionSegmentFromPath('/health')).toBeUndefined(); + expect(versionSegmentFromPath(undefined)).toBeUndefined(); + }); + }); + + describe('normalizeVersionToken', () => { + it.each([ + ['v1', 'v1'], + ['V2', 'v2'], + ['1', 'v1'], + ['2', 'v2'], + ['1.0', 'v1'], + ['1.0.0', 'v1'], + ['2.0.0', 'v2'], + ])('normalizes %p to %p', (input, expected) => { + expect(normalizeVersionToken(input)).toBe(expected); + }); + + it('rejects unknown values', () => { + expect(normalizeVersionToken('nope')).toBeUndefined(); + expect(normalizeVersionToken('3')).toBeUndefined(); + expect(normalizeVersionToken('')).toBeUndefined(); + expect(normalizeVersionToken(undefined)).toBeUndefined(); + }); + }); + + describe('resolveRequestedVersion', () => { + it('prefers the path segment', () => { + expect(resolveRequestedVersion('/api/v2/policies', 'v1')).toBe('v2'); + }); + + it('falls back to the header, then the default', () => { + expect(resolveRequestedVersion('/health', 'v2')).toBe('v2'); + expect(resolveRequestedVersion('/health')).toBe(DEFAULT_API_VERSION); + expect(resolveRequestedVersion(undefined)).toBe(DEFAULT_API_VERSION); + }); + }); + + describe('version metadata', () => { + it('exposes the supported versions', () => { + expect(SUPPORTED_API_VERSIONS).toEqual(['v1', 'v2']); + expect(API_VERSIONS.v1).toBe('1.0.0'); + }); + + it('maps route segments to advertised versions', () => { + expect(apiVersionHeaderValue('v1')).toBe('1.0.0'); + expect(apiVersionHeaderValue('v2')).toBe('2.0.0'); + expect(apiVersionHeaderValue('v9')).toBeUndefined(); + expect(isSupportedVersion('v1')).toBe(true); + expect(isSupportedVersion('v9')).toBe(false); + }); + }); + + describe('applyApiVersionHeaders', () => { + function fakeRes() { + const headers: Record = {}; + return { + headersSent: false, + setHeader: jest.fn((name: string, value: string) => { + headers[name] = value; + }), + headers, + }; + } + + it('sets X-API-Version and, for v1, the deprecation headers', () => { + const res = fakeRes(); + applyApiVersionHeaders(res, 'v1'); + + expect(res.setHeader).toHaveBeenCalledWith(API_VERSION_HEADER, '1.0.0'); + expect(res.setHeader).toHaveBeenCalledWith('Deprecation', 'true'); + expect(res.setHeader).toHaveBeenCalledWith('Link', '; rel="successor"'); + }); + + it('does not mark v2 as deprecated', () => { + const res = fakeRes(); + applyApiVersionHeaders(res, 'v2'); + + expect(res.setHeader).toHaveBeenCalledWith(API_VERSION_HEADER, '2.0.0'); + expect(res.setHeader).not.toHaveBeenCalledWith('Deprecation', expect.anything()); + }); + + it('is a no-op for unknown versions or already-sent responses', () => { + const unknown = fakeRes(); + applyApiVersionHeaders(unknown, 'v9'); + expect(unknown.setHeader).not.toHaveBeenCalled(); + + const sent = fakeRes(); + sent.headersSent = true; + applyApiVersionHeaders(sent, 'v1'); + expect(sent.setHeader).not.toHaveBeenCalled(); + }); + + it('is a no-op when there is no response', () => { + expect(() => applyApiVersionHeaders(undefined, 'v1')).not.toThrow(); + }); + }); + + describe('createApiVersionMiddleware', () => { + function run(url: string, headers: Record = {}) { + const setHeader = jest.fn(); + const next = jest.fn(); + const middleware = createApiVersionMiddleware(); + middleware({ url, headers } as any, { setHeader, headersSent: false } as any, next); + return { setHeader, next }; + } + + it('stamps the header for a versioned path', () => { + const { setHeader, next } = run('/api/v1/policies'); + expect(setHeader).toHaveBeenCalledWith(API_VERSION_HEADER, '1.0.0'); + expect(next).toHaveBeenCalled(); + }); + + it('honours the x-api-version header (including array form)', () => { + const { setHeader } = run('/health', { 'x-api-version': ['v2'] }); + expect(setHeader).toHaveBeenCalledWith(API_VERSION_HEADER, '2.0.0'); + }); + }); +}); diff --git a/src/common/versioning/api-version.ts b/src/common/versioning/api-version.ts new file mode 100644 index 0000000..bffe5a7 --- /dev/null +++ b/src/common/versioning/api-version.ts @@ -0,0 +1,145 @@ +import type { NextFunction, Request, RequestHandler, Response } from 'express'; + +/** + * #713 — API version discovery. + * + * Every response carries an `X-API-Version` header so clients can tell which + * version of the API served their request without parsing the URL. The value is + * the semantic version advertised for the matched route segment (`v1` → + * `1.0.0`, `v2` → `2.0.0`). + * + * This module is the single source of truth for version resolution, shared by + * the interceptor (success paths), the global exception filter (error paths) + * and an early middleware (responses that never reach either, e.g. CORS + * preflights). + */ + +export const API_VERSION_HEADER = 'X-API-Version'; + +/** Route-segment version → advertised semantic version. */ +export const API_VERSIONS: Record = { + v1: '1.0.0', + v2: '2.0.0', +}; + +/** Version used when the request does not name one. */ +export const DEFAULT_API_VERSION = 'v1'; + +/** Route-segment versions the API currently supports. */ +export const SUPPORTED_API_VERSIONS = Object.keys(API_VERSIONS); + +/** `true` when `version` (e.g. `v1`) is a supported route segment. */ +export function isSupportedVersion(version: string): boolean { + return Object.prototype.hasOwnProperty.call(API_VERSIONS, version); +} + +/** Advertised semantic version for a route segment, or `undefined` if unknown. */ +export function apiVersionHeaderValue(version: string): string | undefined { + return isSupportedVersion(version) ? API_VERSIONS[version] : undefined; +} + +/** Strip the query string and fragment from a URL/path. */ +function pathnameOf(pathOrUrl: string): string { + const withoutHash = pathOrUrl.split('#', 1)[0]; + return withoutHash.split('?', 1)[0]; +} + +/** Pull the `vN` segment out of a path such as `/api/v1/policies`. */ +export function versionSegmentFromPath(pathOrUrl?: string): string | undefined { + if (!pathOrUrl) return undefined; + const match = /^\/api\/(v\d+)(?:[\/?#]|$)/.exec(pathnameOf(pathOrUrl)); + return match?.[1]; +} + +/** + * Normalise a request `x-api-version` value into a route-segment version. + * Accepts `v1`, `1`, `1.0` and `1.0.0` (case-insensitive). + */ +export function normalizeVersionToken(value?: string): string | undefined { + if (!value) return undefined; + const trimmed = value.trim(); + if (!trimmed) return undefined; + + if (isSupportedVersion(trimmed.toLowerCase())) { + return trimmed.toLowerCase(); + } + + if (/^\d+(?:\.\d+)*$/.test(trimmed)) { + const major = trimmed.split('.')[0]; + const candidate = `v${major}`; + if (isSupportedVersion(candidate)) { + // Accept a bare major version ("1"), the advertised version ("1.0.0"), or + // a prefix of it ("1.0"). + const advertised = API_VERSIONS[candidate]; + if (trimmed === major || trimmed === advertised || advertised.startsWith(`${trimmed}.`)) { + return candidate; + } + } + } + + return undefined; +} + +/** + * Resolve the version for a request: the path segment wins (it is what routing + * uses), then an explicit `x-api-version` header, then the default. + */ +export function resolveRequestedVersion( + pathOrUrl?: string, + headerValue?: string, +): string { + return ( + versionSegmentFromPath(pathOrUrl) ?? + normalizeVersionToken(headerValue) ?? + DEFAULT_API_VERSION + ); +} + +/** Minimal response surface needed to set version headers. */ +export interface VersionHeaderTarget { + setHeader?(name: string, value: string): unknown; + headersSent?: boolean; +} + +/** + * Set `X-API-Version` (and, for `v1`, the deprecation headers) on a response if + * it is still writable. No-op when the version is unknown or the headers have + * already been flushed, so it is safe to call from any layer. + */ +export function applyApiVersionHeaders( + response: VersionHeaderTarget | undefined, + version: string, +): void { + if (!response || typeof response.setHeader !== 'function') return; + if (response.headersSent) return; + + const value = apiVersionHeaderValue(version); + if (!value) return; + + response.setHeader(API_VERSION_HEADER, value); + + if (version === DEFAULT_API_VERSION) { + response.setHeader('Deprecation', 'true'); + response.setHeader('Link', '; rel="successor"'); + } +} + +/** + * Early middleware that stamps `X-API-Version` on every response, including + * ones produced before or outside the Nest interceptor/filter pipeline. + */ +export function createApiVersionMiddleware(): RequestHandler { + return function apiVersionHeader( + req: Request, + res: Response, + next: NextFunction, + ): void { + const header = req.headers?.['x-api-version']; + const version = resolveRequestedVersion( + req.originalUrl || req.url, + Array.isArray(header) ? header[0] : header, + ); + applyApiVersionHeaders(res, version); + next(); + }; +} diff --git a/src/config/env.d.ts b/src/config/env.d.ts index 0e249e1..3215a18 100644 --- a/src/config/env.d.ts +++ b/src/config/env.d.ts @@ -19,6 +19,11 @@ export interface EnvironmentVariables { HORIZON_URL?: string; STELLAR_NETWORK?: string; REDIS_URL?: string; + // #710 — Redis startup connection retry tuning (defaults in src/redis/redis-startup.ts) + REDIS_STARTUP_MAX_ATTEMPTS?: string; + REDIS_STARTUP_INITIAL_DELAY_MS?: string; + REDIS_STARTUP_MAX_DELAY_MS?: string; + REDIS_STARTUP_ATTEMPT_TIMEOUT_MS?: string; ADMIN_API_KEY?: string; ORACLE_OPERATOR_API_KEY?: string; AVIATIONSTACK_API_KEY?: string; diff --git a/src/main.ts b/src/main.ts index 3c04c2b..9c34c83 100644 --- a/src/main.ts +++ b/src/main.ts @@ -18,7 +18,8 @@ import { REALTIME_DOCS } from './common/swagger/realtime-docs'; import { ErrorResponseDto } from './common/swagger/error-response.dto'; import { initializeOpenTelemetry } from './common/telemetry/opentelemetry'; import { parseTrustProxy } from './common/network/trust-proxy'; -import helmet from 'helmet'; +import { applySecurityHeaders } from './common/security/response-header-hardening'; +import { createApiVersionMiddleware } from './common/versioning/api-version'; import compression from 'compression'; import { ConfigService } from '@nestjs/config'; import { json, urlencoded } from 'express'; @@ -92,16 +93,16 @@ async function bootstrap() { typeof value === 'bigint' ? value.toString() : value, ); - // Security headers (X-Content-Type-Options, X-Frame-Options, HSTS, etc.) - app.use(helmet({ - referrerPolicy: { policy: 'strict-origin-when-cross-origin' }, - // #705 — enable HSTS to prevent SSL stripping attacks - hsts: { - maxAge: 31536000, // 1 year - includeSubDomains: true, - preload: true, - }, - })); + // #713 — Stamp X-API-Version on every response as early as possible so it is + // present even on responses that bypass the interceptor/exception-filter + // pipeline (CORS preflights, timeouts, static docs). + app.use(createApiVersionMiddleware()); + + // Security headers (X-Content-Type-Options, X-Frame-Options, HSTS, etc.). + // #711 — also disables Express's X-Powered-By and strips leaky headers such + // as Server/X-Runtime/Via from every response. + // #705 — HSTS is enabled to prevent SSL stripping attacks (see the helper). + applySecurityHeaders(app); // Response compression — gzip/deflate for text-based responses (JSON, HTML, // SSE streams). Runs before route handlers so every response is eligible. @@ -230,6 +231,8 @@ async function bootstrap() { methods: corsMethods, allowedHeaders: corsAllowedHeaders, credentials: corsCredentials, + // #713 — let browser clients read the version/deprecation headers. + exposedHeaders: ['X-API-Version', 'Deprecation', 'Link'], }); app.setGlobalPrefix('api/v1'); diff --git a/src/oracle/oracle.service.ts b/src/oracle/oracle.service.ts index 18e6b65..29bc9ae 100644 --- a/src/oracle/oracle.service.ts +++ b/src/oracle/oracle.service.ts @@ -129,6 +129,12 @@ export interface OracleReading { source: string; /** OK when the upstream returned real data, NO_DATA when the value is unknown (#171). */ status?: "OK" | "NO_DATA"; + /** + * #712 — Start of the idempotency bucket this reading was stored under. Set + * when a reading is loaded from the database; optional because upstream + * fetches build readings before they are persisted. + */ + bucketStart?: Date; } /** Length of an oracle idempotency bucket in ms — matches the hourly cron cadence (#172). */ @@ -342,6 +348,8 @@ export class OracleService { confidence: record.confidence, timestamp: Math.floor(record.submittedAt.getTime() / 1000), source: record.source, + // #712 — expose the bucket start so callers can report it as the reading time. + bucketStart: record.bucketStart, })); } @@ -374,6 +382,8 @@ export class OracleService { confidence: record.confidence, timestamp: Math.floor(record.submittedAt.getTime() / 1000), source: record.source, + // #712 — expose the bucket start so callers can report it as the reading time. + bucketStart: record.bucketStart, } : null; diff --git a/src/redis/redis-connection.service.spec.ts b/src/redis/redis-connection.service.spec.ts new file mode 100644 index 0000000..ff8578d --- /dev/null +++ b/src/redis/redis-connection.service.spec.ts @@ -0,0 +1,171 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { ConfigService } from '@nestjs/config'; +import { RedisConnectionService, runtimeRetryDelay } from './redis-connection.service'; +import { registerRedisErrorHandler } from './redis.module'; +import { RedisStartupConnectionError } from './redis-startup'; + +/** Fake ioredis-like client. */ +class FakeRedisClient { + attempts = 0; + disconnects = 0; + quits = 0; + status = 'wait'; + readonly errorHandlers: Array<(err: Error) => void> = []; + options: Record = {}; + + constructor(private readonly failuresBeforeSuccess: number) {} + + connect = jest.fn(async () => { + this.attempts += 1; + if (this.attempts <= this.failuresBeforeSuccess) { + throw new Error('connect ECONNREFUSED 127.0.0.1:6379'); + } + this.status = 'ready'; + }); + + disconnect = jest.fn((_reconnect?: boolean) => { + this.disconnects += 1; + this.status = 'end'; + }); + + quit = jest.fn(async () => { + this.quits += 1; + this.status = 'end'; + }); + + on = jest.fn((event: string, handler: (err: Error) => void) => { + if (event === 'error') this.errorHandlers.push(handler); + return this; + }); +} + +async function buildService( + failuresBeforeSuccess: number, + env: Record = {}, +): Promise<{ service: RedisConnectionService; client: FakeRedisClient }> { + const client = new FakeRedisClient(failuresBeforeSuccess); + const config = { get: jest.fn((key: string) => env[key]) }; + + const moduleRef: TestingModule = await Test.createTestingModule({ + providers: [ + RedisConnectionService, + { provide: 'REDIS_CLIENT', useValue: client }, + { provide: ConfigService, useValue: config }, + ], + }).compile(); + + const service = moduleRef.get(RedisConnectionService); + // No real sleeps in tests. + jest.spyOn(service as any, 'sleep').mockResolvedValue(undefined); + jest.spyOn(service as any, 'random').mockReturnValue(1); + jest.spyOn(service as any, 'now').mockReturnValue(0); + return { service, client }; +} + +describe('RedisConnectionService (#710)', () => { + afterEach(() => jest.clearAllMocks()); + + it('starts when Redis is available on the first attempt', async () => { + const { service, client } = await buildService(0, { REDIS_URL: 'redis://localhost:6379' }); + + await expect(service.onModuleInit()).resolves.toBeUndefined(); + + expect(client.connect).toHaveBeenCalledTimes(1); + expect(client.disconnect).not.toHaveBeenCalled(); + expect(client.options.retryStrategy).toBe(runtimeRetryDelay); + }); + + it('starts when Redis becomes available after N failed attempts', async () => { + const { service, client } = await buildService(2, { + REDIS_URL: 'redis://localhost:6379', + REDIS_STARTUP_MAX_ATTEMPTS: '5', + }); + + await expect(service.onModuleInit()).resolves.toBeUndefined(); + + expect(client.connect).toHaveBeenCalledTimes(3); + expect(client.disconnect).not.toHaveBeenCalled(); + }); + + it('fails after exactly the configured maximum attempts', async () => { + const { service, client } = await buildService(Infinity, { + REDIS_URL: 'redis://localhost:6379', + REDIS_STARTUP_MAX_ATTEMPTS: '3', + }); + + await expect(service.onModuleInit()).rejects.toBeInstanceOf(RedisStartupConnectionError); + expect(client.connect).toHaveBeenCalledTimes(3); + // Failed client is cleaned up so no connection/listener is leaked. + expect(client.disconnect).toHaveBeenCalledTimes(1); + }); + + it('honors the configured retry values', async () => { + const { service, client } = await buildService(Infinity, { + REDIS_URL: 'redis://localhost:6379', + REDIS_STARTUP_MAX_ATTEMPTS: '2', + REDIS_STARTUP_INITIAL_DELAY_MS: '10', + REDIS_STARTUP_MAX_DELAY_MS: '1000', + }); + + await expect(service.onModuleInit()).rejects.toBeInstanceOf(RedisStartupConnectionError); + expect(client.connect).toHaveBeenCalledTimes(2); + const sleep = (service as any).sleep as jest.Mock; + expect(sleep).toHaveBeenCalledTimes(1); + }); + + it('produces an actionable error naming host and port', async () => { + const { service } = await buildService(Infinity, { + REDIS_URL: 'redis://redis.internal:6380', + REDIS_STARTUP_MAX_ATTEMPTS: '2', + }); + + await expect(service.onModuleInit()).rejects.toThrow(/redis\.internal:6380/); + }); + + it('never includes credentials in the failure error', async () => { + const { service } = await buildService(Infinity, { + REDIS_URL: 'redis://admin:supersecret@redis.internal:6380', + REDIS_STARTUP_MAX_ATTEMPTS: '2', + }); + + const error = await service.onModuleInit().then( + () => undefined, + (err: unknown) => err as Error, + ); + expect(error).toBeInstanceOf(RedisStartupConnectionError); + expect(error!.message).not.toContain('supersecret'); + expect(error!.message).not.toContain('admin'); + expect(error!.message).not.toContain('redis://'); + expect(error!.message).toContain('redis.internal:6380'); + }); + + it('closes the client on shutdown exactly once (idempotent)', async () => { + const { service, client } = await buildService(0, { REDIS_URL: 'redis://localhost:6379' }); + await service.onModuleInit(); + + await service.onModuleDestroy(); + await service.onModuleDestroy(); + + expect(client.quit).toHaveBeenCalledTimes(1); + expect(client.disconnect).not.toHaveBeenCalled(); + }); +}); + +describe('registerRedisErrorHandler (#710)', () => { + it('always attaches an error listener and is idempotent', () => { + const listeners: Array<(err: Error) => void> = []; + const client = { + on: jest.fn((event: string, handler: (err: Error) => void) => { + if (event === 'error') listeners.push(handler); + return client; + }), + } as any; + + registerRedisErrorHandler(client); + registerRedisErrorHandler(client); + + expect(listeners).toHaveLength(1); + // Emitting must not throw (i.e. the error is handled). + expect(() => listeners[0](new Error('boom'))).not.toThrow(); + }); +}); diff --git a/src/redis/redis-connection.service.ts b/src/redis/redis-connection.service.ts new file mode 100644 index 0000000..3baefc3 --- /dev/null +++ b/src/redis/redis-connection.service.ts @@ -0,0 +1,121 @@ +import { Inject, Injectable, Logger, OnModuleDestroy, OnModuleInit } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; +import type Redis from 'ioredis'; +import { + ConnectRetryDeps, + ConnectRetryLogger, + RedisTarget, + connectWithRetry, + parseRedisTarget, + readRedisStartupRetryOptions, +} from './redis-startup'; + +/** + * Runtime reconnection strategy handed to ioredis once startup succeeds. + * Kept identical to the pre-#710 behaviour: exponential-ish backoff capped at + * 2s. During startup the strategy is `null` (set in redis.module.ts) so it does + * not race {@link connectWithRetry}. + */ +export function runtimeRetryDelay(times: number): number { + return Math.min(times * 200, 2_000); +} + +/** + * RedisConnectionService (#710) + * + * Owns the startup handshake for the shared `REDIS_CLIENT`: + * - retries the first connection with backoff/jitter/timeout + * - fails startup with one actionable error after the configured attempts + * - hands runtime reconnection back to ioredis once connected + * - closes the client on shutdown (idempotently) + * + * It never creates a client itself — the single instance is provided by + * RedisModule's `REDIS_CLIENT` factory. + */ +@Injectable() +export class RedisConnectionService implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(RedisConnectionService.name); + private closed = false; + + constructor( + @Inject('REDIS_CLIENT') private readonly client: Redis, + private readonly config: ConfigService, + ) {} + + async onModuleInit(): Promise { + const options = readRedisStartupRetryOptions( + (key) => this.config.get(key) ?? undefined, + ); + const target: RedisTarget = parseRedisTarget(this.config.get('REDIS_URL')); + + await connectWithRetry(this.client, options, target, this.retryDeps(), this.retryLogger()); + this.enableRuntimeReconnect(); + + this.logger.log( + `Redis connected at ${target.host}:${target.port} (startup retry: up to ${options.maxAttempts} attempts)`, + ); + } + + async onModuleDestroy(): Promise { + if (this.closed) return; + this.closed = true; + await this.closeClient(); + } + + /** Resume ioredis-owned runtime reconnection now that startup is complete. */ + private enableRuntimeReconnect(): void { + const client = this.client as unknown as { options?: Record }; + if (client?.options) { + client.options.retryStrategy = runtimeRetryDelay; + } + } + + private async closeClient(): Promise { + if (!this.client) return; + const client = this.client as unknown as { + status?: string; + quit?: () => Promise; + disconnect?: (reconnect?: boolean) => void; + }; + try { + if (typeof client.quit === 'function' && client.status === 'ready') { + await client.quit(); + } else if (typeof client.disconnect === 'function') { + client.disconnect(false); + } + } catch (err) { + this.logger.warn( + `Error closing Redis client: ${err instanceof Error ? err.message : String(err)}`, + ); + } + } + + /** Injectable for tests (no real sleeps). */ + protected sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); + } + + /** Injectable for tests. */ + protected now(): number { + return Date.now(); + } + + /** Injectable for tests. */ + protected random(): number { + return Math.random(); + } + + private retryDeps(): ConnectRetryDeps { + return { + sleep: (ms) => this.sleep(ms), + now: () => this.now(), + random: () => this.random(), + }; + } + + private retryLogger(): ConnectRetryLogger { + return { + warn: (message) => this.logger.warn(message), + }; + } +} diff --git a/src/redis/redis-startup.spec.ts b/src/redis/redis-startup.spec.ts new file mode 100644 index 0000000..ec86775 --- /dev/null +++ b/src/redis/redis-startup.spec.ts @@ -0,0 +1,244 @@ +import { + DEFAULT_REDIS_STARTUP_RETRY, + RedisConnectable, + RedisStartupConnectionError, + RedisStartupRetryOptions, + RedisAttemptTimeoutError, + computeBackoffDelay, + connectWithRetry, + parseRedisTarget, + readRedisStartupRetryOptions, +} from './redis-startup'; + +/** Fake ioredis-like client whose connect() fails a fixed number of times. */ +class FakeRedisClient implements RedisConnectable { + attempts = 0; + disconnects = 0; + status = 'wait'; + + constructor(private readonly failuresBeforeSuccess: number) {} + + connect = jest.fn(async () => { + this.attempts += 1; + if (this.attempts <= this.failuresBeforeSuccess) { + throw new Error('connect ECONNREFUSED 127.0.0.1:6379'); + } + this.status = 'ready'; + }); + + disconnect = jest.fn((_reconnect?: boolean) => { + this.disconnects += 1; + this.status = 'end'; + }); +} + +const options = (overrides: Partial = {}): RedisStartupRetryOptions => ({ + ...DEFAULT_REDIS_STARTUP_RETRY, + ...overrides, +}); + +const target = { host: 'redis.internal', port: 6380 }; + +const deps = (random = () => 1) => ({ + sleep: jest.fn().mockResolvedValue(undefined), + now: jest.fn(() => 1000), + random, +}); + +const logger = { warn: jest.fn() }; + +describe('redis-startup helpers (#710)', () => { + beforeEach(() => jest.clearAllMocks()); + + describe('readRedisStartupRetryOptions', () => { + it('returns the documented defaults when nothing is configured', () => { + expect(readRedisStartupRetryOptions(() => undefined)).toEqual(DEFAULT_REDIS_STARTUP_RETRY); + }); + + it('honors configured values', () => { + const env: Record = { + REDIS_STARTUP_MAX_ATTEMPTS: '9', + REDIS_STARTUP_INITIAL_DELAY_MS: '50', + REDIS_STARTUP_MAX_DELAY_MS: '5000', + REDIS_STARTUP_ATTEMPT_TIMEOUT_MS: '750', + }; + expect(readRedisStartupRetryOptions((key) => env[key])).toEqual({ + maxAttempts: 9, + initialDelayMs: 50, + maxDelayMs: 5000, + attemptTimeoutMs: 750, + }); + }); + + it('ignores invalid values and falls back to defaults', () => { + const env: Record = { + REDIS_STARTUP_MAX_ATTEMPTS: '0', + REDIS_STARTUP_ATTEMPT_TIMEOUT_MS: 'abc', + }; + const result = readRedisStartupRetryOptions((key) => env[key]); + expect(result.maxAttempts).toBe(DEFAULT_REDIS_STARTUP_RETRY.maxAttempts); + expect(result.attemptTimeoutMs).toBe(DEFAULT_REDIS_STARTUP_RETRY.attemptTimeoutMs); + }); + }); + + describe('parseRedisTarget', () => { + it('extracts host and port', () => { + expect(parseRedisTarget('redis://localhost:6379')).toEqual({ host: 'localhost', port: 6379 }); + }); + + it('never retains credentials from the URL', () => { + const parsed = parseRedisTarget('redis://admin:supersecret@redis.example.com:6380/1'); + expect(parsed).toEqual({ host: 'redis.example.com', port: 6380 }); + expect(JSON.stringify(parsed)).not.toContain('supersecret'); + expect(JSON.stringify(parsed)).not.toContain('admin'); + }); + + it('falls back safely for unparseable URLs', () => { + expect(parseRedisTarget('not a url')).toEqual({ host: 'localhost', port: 6379 }); + expect(parseRedisTarget(undefined)).toEqual({ host: 'localhost', port: 6379 }); + }); + }); + + describe('computeBackoffDelay', () => { + it('grows exponentially and is capped at maxDelayMs', () => { + const opts = options({ initialDelayMs: 100, maxDelayMs: 400 }); + expect(computeBackoffDelay(1, opts, () => 1)).toBe(100); + expect(computeBackoffDelay(2, opts, () => 1)).toBe(200); + expect(computeBackoffDelay(3, opts, () => 1)).toBe(400); + // 4th attempt would be 800ms but is capped at 400ms. + expect(computeBackoffDelay(4, opts, () => 1)).toBe(400); + }); + + it('applies jitter between 50% and 100% of the capped delay', () => { + const opts = options({ initialDelayMs: 100, maxDelayMs: 1000 }); + expect(computeBackoffDelay(1, opts, () => 0)).toBe(50); + expect(computeBackoffDelay(1, opts, () => 1)).toBe(100); + }); + }); + + describe('RedisStartupConnectionError', () => { + it('names host/port and attempt count, keeps the cause, and leaks no credentials', () => { + const cause = new Error('connect ECONNREFUSED'); + const error = new RedisStartupConnectionError( + { host: 'redis.internal', port: 6380 }, + 5, + 1234, + cause, + ); + expect(error.message).toContain('redis.internal:6380'); + expect(error.message).toContain('5 attempt(s)'); + expect(error.message).toContain('1234ms'); + expect((error as { cause?: unknown }).cause).toBe(cause); + expect(error.message).not.toContain('supersecret'); + expect(error.message).not.toContain('redis://'); + }); + }); + + describe('connectWithRetry', () => { + it('succeeds when Redis becomes available after failed attempts (slow startup)', async () => { + const client = new FakeRedisClient(2); + const d = deps(); + + await connectWithRetry(client, options({ maxAttempts: 5 }), target, d, logger); + + expect(client.connect).toHaveBeenCalledTimes(3); + expect(d.sleep).toHaveBeenCalledTimes(2); + expect(client.disconnect).not.toHaveBeenCalled(); + }); + + it('uses the configured max attempts and fails with one clear error', async () => { + const client = new FakeRedisClient(Infinity); + const d = deps(); + + await expect( + connectWithRetry(client, options({ maxAttempts: 3 }), target, d, logger), + ).rejects.toBeInstanceOf(RedisStartupConnectionError); + + expect(client.connect).toHaveBeenCalledTimes(3); + expect(d.sleep).toHaveBeenCalledTimes(2); + // The failed client is closed so no connection/listener is leaked. + expect(client.disconnect).toHaveBeenCalledTimes(1); + }); + + it('honors the configured backoff delays', async () => { + const client = new FakeRedisClient(Infinity); + const d = deps(() => 1); + + await expect( + connectWithRetry( + client, + options({ maxAttempts: 3, initialDelayMs: 100, maxDelayMs: 1000 }), + target, + d, + logger, + ), + ).rejects.toBeInstanceOf(RedisStartupConnectionError); + + expect(d.sleep.mock.calls.map((call) => call[0])).toEqual([100, 200]); + }); + + it('gives up on a hung attempt once the per-attempt timeout elapses', async () => { + jest.useFakeTimers(); + try { + const client: RedisConnectable = { + connect: jest.fn(() => new Promise(() => undefined)), + disconnect: jest.fn(), + }; + const d = deps(); + const promise = connectWithRetry( + client, + options({ maxAttempts: 1, attemptTimeoutMs: 50 }), + target, + d, + logger, + ); + const assertion = expect(promise).rejects.toBeInstanceOf(RedisStartupConnectionError); + + jest.advanceTimersByTime(60); + await assertion; + + expect(client.disconnect).toHaveBeenCalledWith(false); + } finally { + jest.useRealTimers(); + } + }); + + it('never leaks credentials through attempt logs', async () => { + const client = new FakeRedisClient(Infinity); + const d = deps(); + + await expect( + connectWithRetry(client, options({ maxAttempts: 1 }), target, d, logger), + ).rejects.toBeInstanceOf(RedisStartupConnectionError); + + const logged = logger.warn.mock.calls.map((call) => String(call[0])).join('\n'); + expect(logged).not.toContain('supersecret'); + expect(logged).not.toContain('redis://'); + }); + + it('propagates an attempt timeout as the cause after exhausting attempts', async () => { + jest.useFakeTimers(); + try { + const client: RedisConnectable = { + connect: jest.fn(() => new Promise(() => undefined)), + disconnect: jest.fn(), + }; + const promise = connectWithRetry( + client, + options({ maxAttempts: 1, attemptTimeoutMs: 25 }), + target, + deps(), + logger, + ); + const assertion = promise.catch((error: unknown) => error); + + jest.advanceTimersByTime(30); + const error = (await assertion) as RedisStartupConnectionError; + expect(error).toBeInstanceOf(RedisStartupConnectionError); + expect((error as { cause?: unknown }).cause).toBeInstanceOf(RedisAttemptTimeoutError); + } finally { + jest.useRealTimers(); + } + }); + }); +}); diff --git a/src/redis/redis-startup.ts b/src/redis/redis-startup.ts new file mode 100644 index 0000000..e840b36 --- /dev/null +++ b/src/redis/redis-startup.ts @@ -0,0 +1,254 @@ +/** + * #710 — Redis startup connection retry. + * + * The Redis client is created once (see redis.module.ts) with `lazyConnect`, + * so creating it never blocks. This module owns the *startup* handshake: + * it retries the first connection with exponential backoff, a capped delay, + * jitter and a per-attempt timeout, and fails with one actionable error when + * every attempt is exhausted. + * + * Division of responsibility (issue #710, point 5): + * - Startup connection attempts are owned by `connectWithRetry` below (driven + * by RedisConnectionService). While startup is in progress the client's + * `retryStrategy` returns `null`, so ioredis does not schedule background + * reconnects that would race this loop. + * - Runtime reconnection (after the first successful connection) is owned by + * ioredis via `retryStrategy`/`reconnectOnError`; RedisConnectionService + * swaps the runtime strategy in once startup succeeds. + */ + +export interface RedisStartupRetryOptions { + /** Maximum number of connection attempts (including the first). */ + maxAttempts: number; + /** Delay before the second attempt, in milliseconds. Doubles each attempt. */ + initialDelayMs: number; + /** Upper bound for the backoff delay, in milliseconds. */ + maxDelayMs: number; + /** Hard timeout for a single connection attempt, in milliseconds. */ + attemptTimeoutMs: number; +} + +export const DEFAULT_REDIS_STARTUP_RETRY: RedisStartupRetryOptions = { + maxAttempts: 5, + initialDelayMs: 200, + maxDelayMs: 3_000, + attemptTimeoutMs: 5_000, +}; + +export const REDIS_STARTUP_ENV_KEYS = { + maxAttempts: 'REDIS_STARTUP_MAX_ATTEMPTS', + initialDelayMs: 'REDIS_STARTUP_INITIAL_DELAY_MS', + maxDelayMs: 'REDIS_STARTUP_MAX_DELAY_MS', + attemptTimeoutMs: 'REDIS_STARTUP_ATTEMPT_TIMEOUT_MS', +} as const; + +export interface RedisTarget { + host: string; + port: number; +} + +/** Minimal shape of the ioredis client this module needs. */ +export interface RedisConnectable { + connect(): Promise; + disconnect?(reconnect?: boolean): void; +} + +export interface ConnectRetryDeps { + sleep(ms: number): Promise; + now(): number; + /** Returns a number in [0, 1). Injected so tests are deterministic. */ + random(): number; +} + +export interface ConnectRetryLogger { + warn(message: string): void; +} + +/** Raised when a single connection attempt exceeds its timeout. */ +export class RedisAttemptTimeoutError extends Error { + constructor(timeoutMs: number) { + super(`Redis connection attempt timed out after ${timeoutMs}ms`); + this.name = 'RedisAttemptTimeoutError'; + } +} + +/** + * Final, actionable startup failure. Deliberately contains only the host and + * port — never the username, password or full connection URL. + */ +export class RedisStartupConnectionError extends Error { + constructor( + target: RedisTarget, + attempts: number, + elapsedMs: number, + cause?: unknown, + ) { + super( + `Unable to connect to Redis at ${target.host}:${target.port} after ${attempts} ` + + `attempt(s) over ${elapsedMs}ms. Verify that Redis is running and reachable ` + + `and that REDIS_URL is correct. Startup retries can be tuned with ` + + `REDIS_STARTUP_MAX_ATTEMPTS, REDIS_STARTUP_INITIAL_DELAY_MS, ` + + `REDIS_STARTUP_MAX_DELAY_MS and REDIS_STARTUP_ATTEMPT_TIMEOUT_MS.`, + ); + this.name = 'RedisStartupConnectionError'; + if (cause !== undefined) { + // ES2021 target: assign manually rather than using the Error options bag. + (this as { cause?: unknown }).cause = cause; + } + } +} + +function readNonNegativeInt( + get: (key: string) => string | undefined, + key: string, + fallback: number, + minimum: number, +): number { + const raw = get(key); + if (raw === undefined || raw === null || raw.trim() === '') return fallback; + const parsed = Number(raw); + if (!Number.isInteger(parsed) || parsed < minimum) return fallback; + return parsed; +} + +/** + * Read the startup retry knobs from configuration, falling back to the + * documented defaults. Values are validated up-front in app.module.ts's + * `validateConfig`; the fallback here keeps the client usable even if an + * invalid value slips through. + */ +export function readRedisStartupRetryOptions( + get: (key: string) => string | undefined, +): RedisStartupRetryOptions { + const maxAttempts = readNonNegativeInt(get, REDIS_STARTUP_ENV_KEYS.maxAttempts, DEFAULT_REDIS_STARTUP_RETRY.maxAttempts, 1); + const initialDelayMs = readNonNegativeInt(get, REDIS_STARTUP_ENV_KEYS.initialDelayMs, DEFAULT_REDIS_STARTUP_RETRY.initialDelayMs, 0); + const maxDelayMs = Math.max( + initialDelayMs, + readNonNegativeInt(get, REDIS_STARTUP_ENV_KEYS.maxDelayMs, DEFAULT_REDIS_STARTUP_RETRY.maxDelayMs, 0), + ); + const attemptTimeoutMs = readNonNegativeInt(get, REDIS_STARTUP_ENV_KEYS.attemptTimeoutMs, DEFAULT_REDIS_STARTUP_RETRY.attemptTimeoutMs, 1); + return { maxAttempts, initialDelayMs, maxDelayMs, attemptTimeoutMs }; +} + +/** + * Parse host/port out of a connection URL without retaining any credentials. + * Falls back to a best-effort result so an unusual URL shape can never leak + * the raw value into an error message. + */ +export function parseRedisTarget(redisUrl: string | undefined): RedisTarget { + if (redisUrl) { + try { + const parsed = new URL(redisUrl); + const port = parsed.port + ? Number(parsed.port) + : parsed.protocol === 'rediss:' + ? 6380 + : 6379; + return { host: parsed.hostname || 'localhost', port }; + } catch { + // Fall through to the sanitized default below. + } + } + return { host: 'localhost', port: 6379 }; +} + +/** Exponential backoff with a cap and 50–100% jitter. */ +export function computeBackoffDelay( + attempt: number, + options: RedisStartupRetryOptions, + random: () => number, +): number { + const exponential = options.initialDelayMs * 2 ** Math.max(0, attempt - 1); + const capped = Math.min(exponential, options.maxDelayMs); + const jitter = 0.5 + random() * 0.5; + return Math.max(0, Math.round(capped * jitter)); +} + +function describeError(error: unknown): string { + if (error instanceof Error) return error.message; + return String(error); +} + +function closeQuietly(client: RedisConnectable): void { + try { + client.disconnect?.(false); + } catch { + // Best effort: the client may already be closed. + } +} + +function connectOnce( + client: RedisConnectable, + timeoutMs: number, +): Promise { + return new Promise((resolve, reject) => { + let settled = false; + const timer = setTimeout(() => { + if (settled) return; + settled = true; + // A hung attempt must not hold the socket open across retries. + closeQuietly(client); + reject(new RedisAttemptTimeoutError(timeoutMs)); + }, timeoutMs); + + Promise.resolve() + .then(() => client.connect()) + .then( + () => { + if (settled) return; + settled = true; + clearTimeout(timer); + resolve(); + }, + (error) => { + if (settled) return; + settled = true; + clearTimeout(timer); + reject(error); + }, + ); + }); +} + +/** + * Connect once, retrying with backoff until `options.maxAttempts` is reached. + * + * On success the promise resolves. On final failure the client is closed so no + * connection or listener is leaked, and a RedisStartupConnectionError (with the + * original error attached as `cause`) is thrown. + */ +export async function connectWithRetry( + client: RedisConnectable, + options: RedisStartupRetryOptions, + target: RedisTarget, + deps: ConnectRetryDeps, + logger: ConnectRetryLogger, +): Promise { + const startedAt = deps.now(); + let lastError: unknown; + + for (let attempt = 1; attempt <= options.maxAttempts; attempt++) { + try { + await connectOnce(client, options.attemptTimeoutMs); + return; + } catch (error) { + lastError = error; + const elapsedMs = Math.max(0, deps.now() - startedAt); + const summary = + `Redis connection attempt ${attempt}/${options.maxAttempts} to ` + + `${target.host}:${target.port} failed after ${elapsedMs}ms: ${describeError(error)}`; + + if (attempt < options.maxAttempts) { + const delay = computeBackoffDelay(attempt, options, deps.random); + logger.warn(`${summary}. Retrying in ${delay}ms`); + await deps.sleep(delay); + } else { + logger.warn(summary); + } + } + } + + const elapsedMs = Math.max(0, deps.now() - startedAt); + closeQuietly(client); + throw new RedisStartupConnectionError(target, options.maxAttempts, elapsedMs, lastError); +} diff --git a/src/redis/redis.module.ts b/src/redis/redis.module.ts index 41c8723..47667e4 100644 --- a/src/redis/redis.module.ts +++ b/src/redis/redis.module.ts @@ -1,7 +1,60 @@ -import { Module, Global } from '@nestjs/common'; +import { Module, Global, Logger } from '@nestjs/common'; import { ConfigService } from '@nestjs/config'; import Redis from 'ioredis'; import { DistributedLockService } from './distributed-lock.service'; +import { RedisConnectionService } from './redis-connection.service'; +import { RedisStartupRetryOptions, readRedisStartupRetryOptions } from './redis-startup'; + +const redisLogger = new Logger('RedisClient'); + +/** + * Attach an error listener to the client so a connection error can never + * surface as an unhandled 'error' event (which would crash the process). + * Idempotent: a second call is a no-op. + */ +export function registerRedisErrorHandler(client: Redis): void { + const holder = client as unknown as { __parashieldErrorHandlerAttached?: boolean }; + if (holder.__parashieldErrorHandlerAttached) return; + holder.__parashieldErrorHandlerAttached = true; + // The message is safe to log — it never contains credentials. + client.on('error', (err: Error) => { + redisLogger.warn(`Redis client error: ${err?.message ?? String(err)}`); + }); +} + +/** + * Create the single shared Redis client. + * + * `lazyConnect` keeps construction non-blocking; the actual startup connect + * (with retries) is performed by RedisConnectionService.onModuleInit. + * + * `retryStrategy` returns `null` until startup completes, so ioredis does not + * schedule background reconnects that would race the startup loop. Once + * connected, RedisConnectionService swaps in {@link runtimeRetryDelay} and + * ioredis owns runtime reconnection. + */ +export function createRedisClient( + redisUrl: string, + options: RedisStartupRetryOptions, +): Redis { + const client = new Redis(redisUrl, { + lazyConnect: true, + maxRetriesPerRequest: 3, + retryStrategy: () => null, + // Reconnect with backoff on connection loss once startup has completed. + reconnectOnError(err: Error) { + const target = ['READONLY', 'ECONNRESET', 'ETIMEDOUT']; + return target.some((code) => err.message.includes(code)); + }, + // Keep connections alive to detect stale connections early. + keepAlive: 10000, + // Per-attempt connection timeout: a hung connect cannot stall startup. + connectTimeout: options.attemptTimeoutMs, + }); + + registerRedisErrorHandler(client); + return client; +} @Global() @Module({ @@ -13,29 +66,16 @@ import { DistributedLockService } from './distributed-lock.service'; if (!redisUrl) { throw new Error('REDIS_URL environment variable is required'); } - return new Redis(redisUrl, { - // Retry configuration for transient connection failures (#664). - maxRetriesPerRequest: 3, - retryStrategy(times: number) { - const delay = Math.min(times * 200, 2000); - return delay; - }, - // Reconnect with exponential backoff on connection loss. - reconnectOnError(err: Error) { - const target = ['READONLY', 'ECONNRESET', 'ETIMEDOUT']; - return target.some((code) => err.message.includes(code)); - }, - // Keep connections alive to detect stale connections early. - keepAlive: 10000, - // Connection timeout to fail fast on unreachable hosts. - connectTimeout: 5000, - }); + const options = readRedisStartupRetryOptions( + (key) => configService.get(key) ?? undefined, + ); + return createRedisClient(redisUrl, options); }, inject: [ConfigService], }, + RedisConnectionService, DistributedLockService, ], exports: ['REDIS_CLIENT', DistributedLockService], }) export class RedisModule {} -