diff --git a/package.json b/package.json index 30c5dbb..4091527 100644 --- a/package.json +++ b/package.json @@ -41,6 +41,7 @@ "swagger-ui-express": "^5.0.1", "uuid": "9.0.1", "winston": "^3.11.0", + "winston-daily-rotate-file": "^5.0.0", "zod": "^4.4.3" }, "devDependencies": { diff --git a/src/app.ts b/src/app.ts index b923b76..4069eaf 100644 --- a/src/app.ts +++ b/src/app.ts @@ -98,7 +98,7 @@ const connectDB = async (): Promise => { }; // Call connectDB but don't listen -if (process.env.NODE_ENV !== 'test' && !process.env.JEST_WORKER_ID) { +if (env.NODE_ENV !== 'test' && !process.env.JEST_WORKER_ID) { connectDB(); } diff --git a/src/config/env.ts b/src/config/env.ts index 0861c54..31c7831 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -56,6 +56,102 @@ interface EnvConfig { BULK_UPLOAD_MAX_BYTES: number; /** Maximum data rows accepted in a single bulk upload. Default: 1000 */ BULK_UPLOAD_MAX_ROWS: number; + + // ── Socket.IO transport tuning ──────────────────────────────────── + /** Interval (ms) between server-initiated Socket.IO pings. Default: 25000 */ + SOCKET_PING_INTERVAL_MS: number; + /** Time (ms) to wait for a pong before considering the peer gone. Default: 20000 */ + SOCKET_PING_TIMEOUT_MS: number; + /** Consecutive missed pongs tolerated before disconnecting. Default: 2 */ + SOCKET_MAX_MISSED_PONGS: number; + /** Time (ms) a queued socket message waits for an ack before retry. Default: 15000 */ + SOCKET_MESSAGE_ACK_TIMEOUT_MS: number; + /** Interval (ms) between periodic socket token expiry checks. Default: 60000 */ + SOCKET_TOKEN_CHECK_INTERVAL_MS: number; + /** Grace period (ms) granted after a socket token expires. Default: 30000 */ + SOCKET_TOKEN_GRACE_PERIOD_MS: number; + /** Maximum location updates accepted in a single offline-sync batch. Default: 500 */ + SYNC_BATCH_SIZE_LIMIT: number; + + // ── Driver location ingestion ───────────────────────────────── + /** TTL (s) of the Redis dedup key for a location update. Default: 60 */ + LOCATION_DEDUP_TTL_SECONDS: number; + /** Maximum age (ms) of a location update before it is rejected. Default: 300000 */ + LOCATION_MAX_AGE_MS: number; + /** Clock-skew tolerance (ms) for future-dated location updates. Default: 30000 */ + LOCATION_MAX_FUTURE_MS: number; + /** Default radius (m) used by driver proximity searches. Default: 5000 */ + DRIVER_PROXIMITY_DEFAULT_RADIUS_M: number; + /** Hard cap (m) on the radius a proximity search may request. Default: 50000 */ + DRIVER_PROXIMITY_MAX_RADIUS_M: number; + /** Maximum number of drivers returned by a proximity search. Default: 50 */ + DRIVER_PROXIMITY_MAX_RESULTS: number; + /** Age (s) beyond which a driver location is considered stale. Default: 300 */ + DRIVER_LOCATION_STALE_AFTER_SECONDS: number; + + // ── ETA cache / routing ────────────────────────────────────── + /** TTL (s) for cached ETA computations. Default: 600 */ + ETA_CACHE_TTL_SECONDS: number; + /** Geohash precision used to key the ETA cache. Default: 7 */ + ETA_GEOHASH_PRECISION: number; + /** Google Maps Directions API key. Blank disables live routing. */ + GOOGLE_MAPS_API_KEY: string; + + // ── Lifecycle / jobs ────────────────────────────────────────── + /** Time (ms) allowed for in-flight work to drain on shutdown. Default: 30000 */ + SHUTDOWN_TIMEOUT_MS: number; + /** Cron expression driving the escrow monitor job. Default: every 5 minutes */ + ESCROW_MONITOR_CRON: string; + + // ── Stellar / Soroban network ───────────────────────────────── + /** Target Stellar network. Default: testnet */ + STELLAR_NETWORK: 'mainnet' | 'testnet' | 'futurenet'; + /** Soroban RPC endpoint. Blank resolves to the default URL for the network. */ + SOROBAN_RPC_URL: string; + /** Network passphrase. Blank resolves to the well-known value for the network. */ + STELLAR_NETWORK_PASSPHRASE: string; + /** Per-request HTTP timeout (ms) for Soroban RPC calls. Default: 10000 */ + SOROBAN_RPC_TIMEOUT_MS: number; + /** Soroban contract id (`C...`) of the escrow contract. Blank disables escrow endpoints. */ + SOROBAN_ESCROW_CONTRACT_ID: string; + /** Escrow contract function invoked to lock funds. Default: lock_escrow */ + SOROBAN_ESCROW_LOCK_FUNCTION: string; + /** Base fee (stroops) used when building transactions. Default: 100 */ + STELLAR_BASE_FEE: string; + /** Validity window (s) of generated unsigned transactions. Default: 300 */ + STELLAR_TRANSACTION_TIMEOUT_SECONDS: number; + /** Jitter ratio (0-1) applied to RPC backoff delays. Default: 0.2 */ + SOROBAN_RPC_RETRY_JITTER_RATIO: number; + + // ── Escrow event indexing ─────────────────────────────────── + /** Contract id watched by the escrow event indexer. */ + ESCROW_CONTRACT_ID: string; + /** Event topic signalling that an escrow was funded. Default: escrow_funded */ + ESCROW_FUNDED_EVENT_TOPIC: string; + + // ── Logging ─────────────────────────────────────────────────── + /** Directory that rotated log files are written to. Default: logs */ + LOG_DIR: string; + /** Maximum size of a single log file before rotation (e.g. "20m"). */ + LOG_MAX_SIZE: string; + /** Retention window for rotated log files (e.g. "14d"). */ + LOG_MAX_FILES: string; + /** Whether rotated log files are gzipped. Default: true */ + LOG_ZIPPED_ARCHIVE: boolean; + /** Disable file transports entirely (useful in containers). Default: false */ + LOG_DISABLE_FILE: boolean; + + // ── Soroban circuit breaker ─────────────────────────────────── + /** Error rate (%) at which the Soroban breaker opens. Default: 50 */ + CB_SOROBAN_ERROR_THRESHOLD_PERCENTAGE: number; + /** Window (ms) over which the breaker's error rate is measured. Default: 10000 */ + CB_SOROBAN_ROLLING_WINDOW_MS: number; + /** Time (ms) the breaker stays open before probing again. Default: 30000 */ + CB_SOROBAN_RESET_TIMEOUT_MS: number; + /** Minimum calls in the window before the breaker may open. Default: 5 */ + CB_SOROBAN_VOLUME_THRESHOLD: number; + /** Per-call timeout (ms) enforced by the breaker. Default: 10000 */ + CB_SOROBAN_TIMEOUT_MS: number; } const envSchema = z.object({ @@ -98,6 +194,73 @@ const envSchema = z.object({ // ── Bulk delivery CSV import ──────────────────────────────────────────────── BULK_UPLOAD_MAX_BYTES: z.coerce.number().int().min(1024).default(5 * 1024 * 1024), BULK_UPLOAD_MAX_ROWS: z.coerce.number().int().min(1).max(10000).default(1000), + + // ── Socket.IO transport tuning ──────────────────────────────────── + SOCKET_PING_INTERVAL_MS: z.coerce.number().int().min(1000).default(25000), + SOCKET_PING_TIMEOUT_MS: z.coerce.number().int().min(1000).default(20000), + SOCKET_MAX_MISSED_PONGS: z.coerce.number().int().min(1).max(10).default(2), + SOCKET_MESSAGE_ACK_TIMEOUT_MS: z.coerce.number().int().min(1000).default(15000), + SOCKET_TOKEN_CHECK_INTERVAL_MS: z.coerce.number().int().min(1000).default(60000), + SOCKET_TOKEN_GRACE_PERIOD_MS: z.coerce.number().int().min(0).default(30000), + SYNC_BATCH_SIZE_LIMIT: z.coerce.number().int().min(1).max(10000).default(500), + + // ── Driver location ingestion ───────────────────────────────── + LOCATION_DEDUP_TTL_SECONDS: z.coerce.number().int().min(1).default(60), + LOCATION_MAX_AGE_MS: z.coerce.number().int().min(1000).default(300000), + LOCATION_MAX_FUTURE_MS: z.coerce.number().int().min(0).default(30000), + DRIVER_PROXIMITY_DEFAULT_RADIUS_M: z.coerce.number().int().min(1).default(5000), + DRIVER_PROXIMITY_MAX_RADIUS_M: z.coerce.number().int().min(1).default(50000), + DRIVER_PROXIMITY_MAX_RESULTS: z.coerce.number().int().min(1).max(500).default(50), + DRIVER_LOCATION_STALE_AFTER_SECONDS: z.coerce.number().int().min(1).default(300), + + // ── ETA cache / routing ────────────────────────────────────── + ETA_CACHE_TTL_SECONDS: z.coerce.number().int().min(1).default(600), + ETA_GEOHASH_PRECISION: z.coerce.number().int().min(1).max(12).default(7), + GOOGLE_MAPS_API_KEY: z.string().default(''), + + // ── Lifecycle / jobs ────────────────────────────────────────── + SHUTDOWN_TIMEOUT_MS: z.coerce.number().int().min(1000).default(30000), + ESCROW_MONITOR_CRON: z.string().trim().min(1).default('*/5 * * * *'), + + // ── Stellar / Soroban network ───────────────────────────────── + STELLAR_NETWORK: z + .string() + .trim() + .toLowerCase() + .pipe(z.enum(['mainnet', 'testnet', 'futurenet'])) + .default('testnet'), + SOROBAN_RPC_URL: z.string().trim().default(''), + STELLAR_NETWORK_PASSPHRASE: z.string().trim().default(''), + SOROBAN_RPC_TIMEOUT_MS: z.coerce.number().int().min(1000).default(10000), + SOROBAN_ESCROW_CONTRACT_ID: z.string().trim().default(''), + SOROBAN_ESCROW_LOCK_FUNCTION: z.string().trim().min(1).default('lock_escrow'), + STELLAR_BASE_FEE: z.string().trim().min(1).default('100'), + STELLAR_TRANSACTION_TIMEOUT_SECONDS: z.coerce.number().int().min(1).default(300), + SOROBAN_RPC_RETRY_JITTER_RATIO: z.coerce.number().min(0).max(1).default(0.2), + + // ── Escrow event indexing ─────────────────────────────────── + ESCROW_CONTRACT_ID: z.string().trim().default(''), + ESCROW_FUNDED_EVENT_TOPIC: z.string().trim().min(1).default('escrow_funded'), + + // ── Logging ─────────────────────────────────────────────────── + LOG_DIR: z.string().trim().min(1).default('logs'), + LOG_MAX_SIZE: z.string().trim().min(1).default('20m'), + LOG_MAX_FILES: z.string().trim().min(1).default('14d'), + LOG_ZIPPED_ARCHIVE: z + .enum(['true', 'false']) + .default('true') + .transform((value) => value === 'true'), + LOG_DISABLE_FILE: z + .enum(['true', 'false']) + .default('false') + .transform((value) => value === 'true'), + + // ── Soroban circuit breaker ─────────────────────────────────── + CB_SOROBAN_ERROR_THRESHOLD_PERCENTAGE: z.coerce.number().int().min(1).max(100).default(50), + CB_SOROBAN_ROLLING_WINDOW_MS: z.coerce.number().int().min(1000).default(10000), + CB_SOROBAN_RESET_TIMEOUT_MS: z.coerce.number().int().min(1000).default(30000), + CB_SOROBAN_VOLUME_THRESHOLD: z.coerce.number().int().min(1).default(5), + CB_SOROBAN_TIMEOUT_MS: z.coerce.number().int().min(1000).default(10000), }); let env: EnvConfig; @@ -121,4 +284,16 @@ if (env.UPLOAD_STORAGE_DRIVER === 's3' && !env.AWS_S3_BUCKET) { process.exit(1); } +if (env.DRIVER_PROXIMITY_DEFAULT_RADIUS_M > env.DRIVER_PROXIMITY_MAX_RADIUS_M) { + console.error( + '❌ DRIVER_PROXIMITY_DEFAULT_RADIUS_M cannot exceed DRIVER_PROXIMITY_MAX_RADIUS_M', + ); + process.exit(1); +} + +if (env.SOROBAN_RPC_RETRY_BASE_MS > env.SOROBAN_RPC_RETRY_MAX_MS) { + console.error('❌ SOROBAN_RPC_RETRY_BASE_MS cannot exceed SOROBAN_RPC_RETRY_MAX_MS'); + process.exit(1); +} + export default env; diff --git a/src/config/escrow.ts b/src/config/escrow.ts index 8223177..56f9026 100644 --- a/src/config/escrow.ts +++ b/src/config/escrow.ts @@ -6,6 +6,8 @@ * separate from the generic Stellar/Soroban RPC config since it identifies * a specific contract instance rather than network connection details. */ +import env from './env'; + export interface EscrowIndexerConfig { /** Deployed escrow contract id (Soroban "C..." address). */ contractId: string; @@ -15,8 +17,8 @@ export interface EscrowIndexerConfig { function resolveEscrowIndexerConfig(): EscrowIndexerConfig { return { - contractId: process.env.ESCROW_CONTRACT_ID?.trim() ?? '', - fundedEventTopic: process.env.ESCROW_FUNDED_EVENT_TOPIC?.trim() || 'escrow_funded', + contractId: env.ESCROW_CONTRACT_ID, + fundedEventTopic: env.ESCROW_FUNDED_EVENT_TOPIC, }; } diff --git a/src/config/logger.ts b/src/config/logger.ts index 4f078f8..b612007 100644 --- a/src/config/logger.ts +++ b/src/config/logger.ts @@ -1,6 +1,36 @@ +/** + * logger.ts + * + * The single logging interface for the application. Every module logs through + * the default export of this file; no other module constructs a transport. + * + * Three guarantees this module provides: + * + * 1. **PII is masked before it reaches any transport.** The masking format is + * installed on the logger itself rather than on individual transports, so + * the console, the rotating files and any transport added later all receive + * already-redacted records. See `utils/piiMasker.ts` for the rules. + * + * 2. **File output is rotated and bounded.** `winston-daily-rotate-file` + * rotates daily and on size, gzips old files and prunes them past the + * retention window, so a long-running container cannot fill its disk. + * + * 3. **The process does not die because logging failed.** Transport `error` + * events are handled, and `exitOnError` is false. + * + * Configuration comes exclusively from the validated `config/env` object. + */ + import winston from 'winston'; +import DailyRotateFile from 'winston-daily-rotate-file'; +import type { TransformableInfo } from 'logform'; import env from './env'; +import { maskValue, maskString } from '../utils/piiMasker'; +/** + * Severity levels, lowest number = highest severity. + * Mirrors the npm levels the codebase already logs at. + */ const levels = { error: 0, warn: 1, @@ -19,34 +49,170 @@ const colors = { winston.addColors(colors); +/** + * Winston symbol keys carried on every log record. They hold the raw level and + * the splat arguments, and must not be treated as user metadata. + */ +const LEVEL_SYMBOL = Symbol.for('level') as unknown as keyof TransformableInfo; +const SPLAT_SYMBOL = Symbol.for('splat') as unknown as keyof TransformableInfo; + +/** + * Format that redacts PII from both the message and any structured metadata. + * + * Installed first in the format chain so every downstream formatter — JSON, + * printf, colorizer — only ever sees masked content. Because it runs inside + * the logger, all 300+ existing `logger.info(...)` call sites gain masking + * without any change at the call site. + * + * If masking itself throws, the record is replaced with a safe placeholder + * rather than allowed through unmasked: failing closed is the only correct + * behaviour for a redaction layer. + */ +const maskPiiFormat = winston.format((info) => { + try { + if (typeof info.message === 'string') { + info.message = maskString(info.message); + } else if (info.message !== undefined) { + info.message = maskValue(info.message) as TransformableInfo['message']; + } + + for (const key of Object.keys(info)) { + if (key === 'message' || key === 'level' || key === 'timestamp') continue; + (info as Record)[key] = maskValue( + (info as Record)[key], + ); + } + + // `splat` holds the extra arguments passed to logger.info(msg, a, b, …). + const splat = (info as Record)[SPLAT_SYMBOL as unknown as symbol]; + if (Array.isArray(splat)) { + (info as Record)[SPLAT_SYMBOL as unknown as symbol] = splat.map((arg) => + maskValue(arg), + ); + } + + return info; + } catch { + return { + ...info, + [LEVEL_SYMBOL]: info[LEVEL_SYMBOL], + message: '[log record suppressed: PII masking failed]', + } as TransformableInfo; + } +}); + +/** + * Render structured metadata as a compact suffix for the human-readable + * console output, e.g. `... {"deliveryId":"abc"}`. + */ +function formatMetadata(info: TransformableInfo): string { + const omitted = new Set(['level', 'message', 'timestamp', 'stack']); + const meta: Record = {}; + + for (const [key, value] of Object.entries(info)) { + if (!omitted.has(key) && value !== undefined) meta[key] = value; + } + + if (Object.keys(meta).length === 0) return ''; + + try { + return ` ${JSON.stringify(meta)}`; + } catch { + return ' [unserialisable metadata]'; + } +} + +/** Colourised, single-line format for local development. */ const devFormat = winston.format.combine( - winston.format.timestamp({ format: 'YYYY-MM-DD HH:mm:ss:ms' }), + winston.format.timestamp({ format: 'YYYY-MM-DD HH:mm:ss.SSS' }), winston.format.colorize({ all: true }), - winston.format.printf((info) => `${info.timestamp} ${info.level}: ${info.message}`), + winston.format.printf( + (info) => + `${info.timestamp as string} ${info.level}: ${String(info.message)}${formatMetadata(info)}`, + ), ); -const prodFormat = winston.format.combine(winston.format.timestamp(), winston.format.json()); +/** + * Structured JSON for production and for all file output, so records can be + * ingested by a log aggregator without parsing. + */ +const prodFormat = winston.format.combine( + winston.format.timestamp(), + winston.format.errors({ stack: true }), + winston.format.json(), +); -const transports = [ - new winston.transports.Console({ - format: env.NODE_ENV === 'development' ? devFormat : prodFormat, - }), - new winston.transports.File({ - filename: 'logs/error.log', - level: 'error', - format: prodFormat, - }), - new winston.transports.File({ - filename: 'logs/all.log', +const consoleFormat = env.NODE_ENV === 'development' ? devFormat : prodFormat; + +/** + * Build a rotating file transport. + * + * @param filename - Basename pattern; `%DATE%` is substituted by the rotator. + * @param level - Optional minimum level for this transport. + */ +function createRotatingTransport(filename: string, level?: string): DailyRotateFile { + return new DailyRotateFile({ + dirname: env.LOG_DIR, + filename, + datePattern: 'YYYY-MM-DD', + zippedArchive: env.LOG_ZIPPED_ARCHIVE, + maxSize: env.LOG_MAX_SIZE, + maxFiles: env.LOG_MAX_FILES, + level, format: prodFormat, - }), + handleExceptions: false, + }); +} + +const transports: winston.transport[] = [ + new winston.transports.Console({ format: consoleFormat }), ]; +// File transports are skipped when disabled, and in tests, so unit runs do not +// leave log files behind or hold open file handles after the suite ends. +if (!env.LOG_DISABLE_FILE && env.NODE_ENV !== 'test') { + const errorTransport = createRotatingTransport('error-%DATE%.log', 'error'); + const combinedTransport = createRotatingTransport('all-%DATE%.log'); + + for (const transport of [errorTransport, combinedTransport]) { + // A rotation or disk failure must never take the process down. These + // handlers write straight to the console rather than through `logger`, + // which is not constructed yet at this point in module evaluation. + transport.on('error', (error: Error) => { + // eslint-disable-next-line no-console + console.error(`[logger] file transport error: ${maskString(error.message)}`); + }); + transport.on('rotate', (oldFilename: string, newFilename: string) => { + // eslint-disable-next-line no-console + console.info(`[logger] rotated ${oldFilename} -> ${newFilename}`); + }); + } + + transports.push(errorTransport, combinedTransport); +} + +/** + * The application logger. + * + * Masking is applied by the logger-level format, so every transport receives + * redacted records. + */ const logger = winston.createLogger({ level: env.LOG_LEVEL, levels, - format: env.NODE_ENV === 'development' ? devFormat : prodFormat, + format: winston.format.combine(maskPiiFormat(), prodFormat), transports, + exitOnError: false, }); +/** + * Stream adapter so HTTP access-log middleware (morgan and friends) can write + * through the same masked pipeline. + */ +export const loggerStream = { + write: (message: string): void => { + logger.http(message.trim()); + }, +}; + export default logger; diff --git a/src/config/security.ts b/src/config/security.ts index ff3e837..fc55d48 100644 --- a/src/config/security.ts +++ b/src/config/security.ts @@ -2,6 +2,7 @@ import type { CorsOptions, CorsOptionsDelegate } from 'cors'; import type { HelmetOptions } from 'helmet'; import type { Request } from 'express'; import logger from './logger'; +import env from './env'; /** * Error raised when a request originates from a disallowed origin. @@ -24,7 +25,7 @@ export class CorsNotAllowedError extends Error { * Example: `CORS_ORIGIN=http://localhost:3000,https://app.swiftchain.io` */ export const getAllowedOrigins = (): string[] => - (process.env.CORS_ORIGIN ?? '') + env.CORS_ORIGIN .split(',') .map((origin) => origin.trim()) .filter((origin) => origin.length > 0); diff --git a/src/config/stellar.ts b/src/config/stellar.ts index 09f70ab..5dab62d 100644 --- a/src/config/stellar.ts +++ b/src/config/stellar.ts @@ -1,10 +1,11 @@ -import { rpc as StellarRpc, BASE_FEE, Networks, StrKey } from '@stellar/stellar-sdk'; +import { rpc as StellarRpc, Networks, StrKey } from '@stellar/stellar-sdk'; import logger from './logger'; +import env from './env'; /** * Supported Stellar network aliases. */ -export type StellarNetwork = 'mainnet' | 'testnet' | 'futurenet'; +export type StellarNetwork = typeof env.STELLAR_NETWORK; /** * Resolved Stellar configuration derived from environment variables. @@ -53,33 +54,13 @@ const DEFAULT_RPC_URLS: Record = { * defaults. Validated at startup so misconfiguration fails fast. */ function resolveStellarConfig(): StellarConfig { - const network = (process.env.STELLAR_NETWORK?.toLowerCase() ?? 'testnet') as StellarNetwork; + const network = env.STELLAR_NETWORK; - if (!['mainnet', 'testnet', 'futurenet'].includes(network)) { - throw new Error( - `Invalid STELLAR_NETWORK="${process.env.STELLAR_NETWORK}". ` + - 'Must be one of: mainnet | testnet | futurenet', - ); - } - - const rpcUrl = process.env.SOROBAN_RPC_URL?.trim() || DEFAULT_RPC_URLS[network]; - - // Prefer explicit passphrase env var; fall back to the well-known value for - // the configured network. - const networkPassphrase = - process.env.STELLAR_NETWORK_PASSPHRASE?.trim() || NETWORK_PASSPHRASES[network]; - - const timeoutMs = parseInt(process.env.SOROBAN_RPC_TIMEOUT_MS ?? '10000', 10); - - if (!rpcUrl) { - throw new Error('SOROBAN_RPC_URL is required and could not be resolved.'); - } - - if (!networkPassphrase) { - throw new Error('STELLAR_NETWORK_PASSPHRASE is required and could not be resolved.'); - } - - const escrowContractId = process.env.SOROBAN_ESCROW_CONTRACT_ID?.trim() || undefined; + // Blank values fall back to the well-known endpoint/passphrase for the + // selected network, so only non-default deployments need to set them. + const rpcUrl = env.SOROBAN_RPC_URL || DEFAULT_RPC_URLS[network]; + const networkPassphrase = env.STELLAR_NETWORK_PASSPHRASE || NETWORK_PASSPHRASES[network]; + const escrowContractId = env.SOROBAN_ESCROW_CONTRACT_ID || undefined; if (escrowContractId && !StrKey.isValidContract(escrowContractId)) { throw new Error( @@ -88,28 +69,15 @@ function resolveStellarConfig(): StellarConfig { ); } - const escrowLockFunction = process.env.SOROBAN_ESCROW_LOCK_FUNCTION?.trim() || 'lock_escrow'; - const baseFee = process.env.STELLAR_BASE_FEE?.trim() || BASE_FEE; - const transactionTimeoutSeconds = parseInt( - process.env.STELLAR_TRANSACTION_TIMEOUT_SECONDS ?? '300', - 10, - ); - - if (!Number.isInteger(transactionTimeoutSeconds) || transactionTimeoutSeconds <= 0) { - throw new Error( - 'STELLAR_TRANSACTION_TIMEOUT_SECONDS must be a positive integer number of seconds.', - ); - } - return { rpcUrl, networkPassphrase, network, - timeoutMs, + timeoutMs: env.SOROBAN_RPC_TIMEOUT_MS, escrowContractId, - escrowLockFunction, - baseFee, - transactionTimeoutSeconds, + escrowLockFunction: env.SOROBAN_ESCROW_LOCK_FUNCTION, + baseFee: env.STELLAR_BASE_FEE, + transactionTimeoutSeconds: env.STELLAR_TRANSACTION_TIMEOUT_SECONDS, }; } diff --git a/src/controllers/driverLocationController.ts b/src/controllers/driverLocationController.ts new file mode 100644 index 0000000..3b196b1 --- /dev/null +++ b/src/controllers/driverLocationController.ts @@ -0,0 +1,245 @@ +/** + * driverLocationController.ts + * + * HTTP layer for driver positions and proximity search. + * + * Controllers parse and validate the transport-level shape of a request — + * query strings arrive as strings and must become numbers — then delegate all + * business logic to `DriverLocationService`. No database access happens here. + */ + +import { Request, Response, NextFunction } from 'express'; +import { StatusCodes } from 'http-status-codes'; +import { + driverLocationService, + DriverLocationService, + NearbyDriversQuery, +} from '../services/driverLocationService'; +import { DriverAvailabilityStatus } from '../models/DriverLocation'; +import { sendSuccess } from '../utils/responseWrapper'; +import AppError from '../utils/AppError'; +import type { IUser } from '../interfaces/IUser'; + +/** Availability values a client may filter on. */ +const VALID_STATUSES: readonly DriverAvailabilityStatus[] = ['online', 'offline', 'on_delivery']; + +export class DriverLocationController { + private readonly service: DriverLocationService; + + constructor(service: DriverLocationService = driverLocationService) { + this.service = service; + } + + /** + * GET /api/v1/drivers/nearby + * + * Query parameters: + * `lat`, `lng` — required search centre. + * `radiusMeters` — optional, clamped to the configured maximum. + * `limit` — optional result cap. + * `availableOnly` — optional boolean, defaults to true. + * `status` — optional availability filter. + */ + public async getNearbyDrivers( + req: Request, + res: Response, + next: NextFunction, + ): Promise { + try { + const query: NearbyDriversQuery = { + lat: this.requireNumber(req.query.lat, 'lat'), + lng: this.requireNumber(req.query.lng, 'lng'), + radiusMeters: this.optionalNumber(req.query.radiusMeters, 'radiusMeters'), + limit: this.optionalNumber(req.query.limit, 'limit'), + availableOnly: this.optionalBoolean(req.query.availableOnly, 'availableOnly'), + status: this.optionalStatus(req.query.status), + }; + + const result = await this.service.findNearbyDrivers(query); + + sendSuccess( + res, + result, + `Found ${result.count} driver(s) within ${result.radiusMeters}m`, + StatusCodes.OK, + ); + } catch (error) { + next(error); + } + } + + /** + * PUT /api/v1/drivers/me/location + * + * Records the authenticated driver's current position. + */ + public async updateMyLocation( + req: Request, + res: Response, + next: NextFunction, + ): Promise { + try { + const user = (req as Request & { user?: IUser }).user; + if (!user) { + throw new AppError('Authentication required.', StatusCodes.UNAUTHORIZED); + } + + const body = req.body as Record; + + const document = await this.service.upsertDriverLocation({ + driverId: String(user._id), + lat: this.requireNumber(body.lat, 'lat'), + lng: this.requireNumber(body.lng, 'lng'), + isAvailable: this.optionalBoolean(body.isAvailable, 'isAvailable'), + status: this.optionalStatus(body.status), + heading: this.optionalNumber(body.heading, 'heading'), + speed: this.optionalNumber(body.speed, 'speed'), + accuracy: this.optionalNumber(body.accuracy, 'accuracy'), + currentDeliveryId: + body.currentDeliveryId === undefined + ? undefined + : body.currentDeliveryId === null + ? null + : String(body.currentDeliveryId), + recordedAt: this.optionalDate(body.recordedAt), + }); + + sendSuccess( + res, + { + driverId: document.driverId.toString(), + ...document.toLatLng(), + isAvailable: document.isAvailable, + status: document.status, + recordedAt: document.recordedAt, + }, + 'Driver location recorded successfully', + StatusCodes.OK, + ); + } catch (error) { + next(error); + } + } + + /** + * GET /api/v1/drivers/:driverId/location + * + * Returns a single driver's most recent position. + */ + public async getDriverLocation( + req: Request, + res: Response, + next: NextFunction, + ): Promise { + try { + const document = await this.service.getDriverLocation(req.params.driverId); + + sendSuccess( + res, + { + driverId: document.driverId.toString(), + ...document.toLatLng(), + isAvailable: document.isAvailable, + status: document.status, + heading: document.heading, + speed: document.speed, + accuracy: document.accuracy, + recordedAt: document.recordedAt, + }, + 'Driver location retrieved successfully', + StatusCodes.OK, + ); + } catch (error) { + next(error); + } + } + + /** + * GET /api/v1/drivers/nearby/explain + * + * Runs the proximity query under `explain()` and reports which index the + * planner used and how many documents it examined. Admin-only: it exposes + * database internals and is meant for profiling index health. + */ + public async explainNearbyQuery( + req: Request, + res: Response, + next: NextFunction, + ): Promise { + try { + const summary = await this.service.explainProximityQuery({ + lat: this.requireNumber(req.query.lat, 'lat'), + lng: this.requireNumber(req.query.lng, 'lng'), + radiusMeters: this.optionalNumber(req.query.radiusMeters, 'radiusMeters'), + limit: this.optionalNumber(req.query.limit, 'limit'), + availableOnly: this.optionalBoolean(req.query.availableOnly, 'availableOnly'), + status: this.optionalStatus(req.query.status), + }); + + sendSuccess(res, summary, 'Proximity query plan retrieved successfully', StatusCodes.OK); + } catch (error) { + next(error); + } + } + + // ── Parameter coercion helpers ───────────────────────────────────────────── + + /** Parse a required numeric parameter. */ + private requireNumber(value: unknown, field: string): number { + if (value === undefined || value === null || value === '') { + throw new AppError(`${field} is required.`, StatusCodes.BAD_REQUEST); + } + + const parsed = Number(value); + if (!Number.isFinite(parsed)) { + throw new AppError(`${field} must be a valid number.`, StatusCodes.BAD_REQUEST); + } + return parsed; + } + + /** Parse an optional numeric parameter, preserving `undefined`. */ + private optionalNumber(value: unknown, field: string): number | undefined { + if (value === undefined || value === null || value === '') return undefined; + return this.requireNumber(value, field); + } + + /** Parse an optional boolean, accepting the string forms a query string yields. */ + private optionalBoolean(value: unknown, field: string): boolean | undefined { + if (value === undefined || value === null || value === '') return undefined; + if (typeof value === 'boolean') return value; + + const normalised = String(value).toLowerCase(); + if (normalised === 'true' || normalised === '1') return true; + if (normalised === 'false' || normalised === '0') return false; + + throw new AppError(`${field} must be a boolean.`, StatusCodes.BAD_REQUEST); + } + + /** Parse an optional availability filter. */ + private optionalStatus(value: unknown): DriverAvailabilityStatus | undefined { + if (value === undefined || value === null || value === '') return undefined; + + const candidate = String(value) as DriverAvailabilityStatus; + if (!VALID_STATUSES.includes(candidate)) { + throw new AppError( + `status must be one of: ${VALID_STATUSES.join(', ')}.`, + StatusCodes.BAD_REQUEST, + ); + } + return candidate; + } + + /** Parse an optional ISO-8601 timestamp. */ + private optionalDate(value: unknown): Date | undefined { + if (value === undefined || value === null || value === '') return undefined; + + const parsed = new Date(String(value)); + if (Number.isNaN(parsed.getTime())) { + throw new AppError('recordedAt must be a valid ISO-8601 date.', StatusCodes.BAD_REQUEST); + } + return parsed; + } +} + +/** Singleton used by the route layer. */ +export const driverLocationController = new DriverLocationController(); diff --git a/src/jobs/escrowMonitor.ts b/src/jobs/escrowMonitor.ts index b5feeaa..acff4d9 100644 --- a/src/jobs/escrowMonitor.ts +++ b/src/jobs/escrowMonitor.ts @@ -1,12 +1,13 @@ import cron, { ScheduledTask } from 'node-cron'; import logger from '../config/logger'; import { scanForExpiredEscrows } from '../services/escrowService'; +import env from '../config/env'; /** * Cron expression the escrow monitor runs on. Defaults to every 5 minutes. * Override with the `ESCROW_MONITOR_CRON` environment variable. */ -const ESCROW_MONITOR_CRON = process.env.ESCROW_MONITOR_CRON?.trim() || '*/5 * * * *'; +const ESCROW_MONITOR_CRON = env.ESCROW_MONITOR_CRON; let scheduledTask: ScheduledTask | null = null; let isRunning = false; diff --git a/src/middleware/auth.ts b/src/middleware/auth.ts index 89b310b..2456981 100644 --- a/src/middleware/auth.ts +++ b/src/middleware/auth.ts @@ -1,8 +1,9 @@ import { NextFunction, Request, Response } from 'express'; import jwt, { JwtPayload } from 'jsonwebtoken'; import { HttpError } from '../utils/httpError'; +import env from '../config/env'; -const jwtSecret = process.env.JWT_SECRET || 'changeme'; +const jwtSecret = env.JWT_SECRET; export interface AuthenticatedRequest extends Request { user?: JwtPayload & { diff --git a/src/middlewares/rateLimiter.ts b/src/middlewares/rateLimiter.ts index e976340..24d8f2b 100644 --- a/src/middlewares/rateLimiter.ts +++ b/src/middlewares/rateLimiter.ts @@ -1,6 +1,7 @@ import rateLimit from 'express-rate-limit'; +import env from '../config/env'; -const isTest = process.env.NODE_ENV === 'test' || !!process.env.JEST_WORKER_ID; +const isTest = env.NODE_ENV === 'test' || !!process.env.JEST_WORKER_ID; /** * Strict rate limiter for authentication endpoints (login, register). diff --git a/src/models/DriverLocation.ts b/src/models/DriverLocation.ts new file mode 100644 index 0000000..e78ddae --- /dev/null +++ b/src/models/DriverLocation.ts @@ -0,0 +1,253 @@ +/** + * DriverLocation.ts + * + * Current, queryable position of each driver. + * + * This model is deliberately separate from `LocationUpdate`, which is an + * append-only history of every ping a device sends. Proximity search reads the + * *current* position of each driver, and answering that from the history + * collection means finding the newest document per driver on every query — + * a sort-and-group that no index makes fast. + * + * `DriverLocation` instead holds exactly one document per driver, upserted in + * place as pings arrive, so a proximity search is a single index scan. + * + * ── Geospatial representation ──────────────────────────────────────────────── + * Coordinates are stored as a GeoJSON `Point`, which is what a `2dsphere` + * index requires. Note the axis order: GeoJSON is `[longitude, latitude]`, + * the reverse of the `{ lat, lng }` shape the API speaks. The helpers on this + * model own that conversion so no caller has to remember it. + * + * ── Index strategy ─────────────────────────────────────────────────────────── + * Three indexes, each earning its place: + * + * 1. `{ driverId: 1 }` unique — enforces one document per driver and makes + * the upsert on every location ping a point lookup. + * + * 2. `{ location: '2dsphere' }` — the plain geospatial index. MongoDB can use + * a compound index prefixed by a geo field only for the geo predicate + * itself, so a standalone 2dsphere serves unfiltered radius searches. + * + * 3. `{ isAvailable: 1, status: 1, location: '2dsphere' }` — a compound + * index with the *equality* fields first and the geo field last. This is + * the ordering MongoDB's geo-near planner can exploit: it narrows to the + * available/online drivers and then walks the geometry, rather than + * scanning every driver in the radius and filtering afterwards. Nearly all + * production proximity queries are "find me an available driver near X", + * so this is the index that matters most. + * + * A TTL index on `expiresAt` retires drivers who stop reporting, so the + * collection stays small and searches never return a driver who went offline + * days ago. + */ + +import { Schema, model, Document, Types, Model } from 'mongoose'; +import env from '../config/env'; + +/** Availability state used to filter proximity searches. */ +export type DriverAvailabilityStatus = 'online' | 'offline' | 'on_delivery'; + +/** GeoJSON Point as stored by MongoDB: `coordinates` is `[lng, lat]`. */ +export interface IGeoPoint { + type: 'Point'; + /** `[longitude, latitude]` — GeoJSON axis order, not `[lat, lng]`. */ + coordinates: [number, number]; +} + +/** + * Mongoose document describing one driver's current position. + */ +export interface IDriverLocation extends Document { + /** Reference to the driver (User._id). Unique across the collection. */ + driverId: Types.ObjectId; + + /** Current position as a GeoJSON Point. */ + location: IGeoPoint; + + /** Whether the driver is currently accepting assignments. */ + isAvailable: boolean; + + /** Coarse availability state, used alongside `isAvailable` for filtering. */ + status: DriverAvailabilityStatus; + + /** Reported heading in degrees clockwise from north, when the device supplies it. */ + heading?: number; + + /** Reported ground speed in metres per second, when the device supplies it. */ + speed?: number; + + /** Device-reported horizontal accuracy in metres. */ + accuracy?: number; + + /** Delivery this driver is currently assigned to, if any. */ + currentDeliveryId?: Types.ObjectId; + + /** When this position was recorded on the device. */ + recordedAt: Date; + + /** + * Point at which this record is considered abandoned and is removed by the + * TTL monitor. Refreshed on every update. + */ + expiresAt: Date; + + createdAt: Date; + updatedAt: Date; + + /** Convenience accessor returning the position in API `{ lat, lng }` form. */ + toLatLng(): { lat: number; lng: number }; +} + +/** + * Static helpers attached to the model. + */ +export interface IDriverLocationModel extends Model { + /** Build a GeoJSON Point from API-order latitude/longitude. */ + toGeoPoint(lat: number, lng: number): IGeoPoint; +} + +const GeoPointSchema = new Schema( + { + type: { + type: String, + enum: ['Point'], + required: true, + default: 'Point', + }, + coordinates: { + type: [Number], + required: true, + validate: { + validator: (value: number[]): boolean => + Array.isArray(value) && + value.length === 2 && + Number.isFinite(value[0]) && + Number.isFinite(value[1]) && + value[0] >= -180 && + value[0] <= 180 && + value[1] >= -90 && + value[1] <= 90, + message: + 'coordinates must be [longitude, latitude] with longitude in [-180,180] and latitude in [-90,90]', + }, + }, + }, + { _id: false }, +); + +const DriverLocationSchema = new Schema( + { + driverId: { + type: Schema.Types.ObjectId, + ref: 'User', + required: [true, 'driverId is required'], + unique: true, + }, + + location: { + type: GeoPointSchema, + required: [true, 'location is required'], + }, + + isAvailable: { + type: Boolean, + required: true, + default: false, + }, + + status: { + type: String, + enum: ['online', 'offline', 'on_delivery'], + required: true, + default: 'offline', + }, + + heading: { + type: Number, + min: [0, 'heading must be between 0 and 360'], + max: [360, 'heading must be between 0 and 360'], + }, + + speed: { + type: Number, + min: [0, 'speed cannot be negative'], + }, + + accuracy: { + type: Number, + min: [0, 'accuracy cannot be negative'], + }, + + currentDeliveryId: { + type: Schema.Types.ObjectId, + ref: 'Delivery', + default: null, + }, + + recordedAt: { + type: Date, + required: true, + default: (): Date => new Date(), + }, + + expiresAt: { + type: Date, + required: true, + }, + }, + { timestamps: true }, +); + +// ─── Indexes ────────────────────────────────────────────────────────────────── + +// Plain 2dsphere: serves radius searches that apply no availability filter. +DriverLocationSchema.index({ location: '2dsphere' }, { name: 'location_2dsphere' }); + +// The workhorse. Equality fields first, geometry last, so the planner can seek +// to the matching availability bucket before evaluating geometry. +DriverLocationSchema.index( + { isAvailable: 1, status: 1, location: '2dsphere' }, + { name: 'availability_location_2dsphere' }, +); + +// Retire stale records. `expireAfterSeconds: 0` means "expire at the instant +// stored in expiresAt", which lets the staleness window be configured per +// write rather than baked into the index. +DriverLocationSchema.index( + { expiresAt: 1 }, + { name: 'driver_location_ttl', expireAfterSeconds: 0 }, +); + +// ─── Hooks ──────────────────────────────────────────────────────────────────── + +/** + * Keep `expiresAt` in step with `recordedAt` so every write extends the + * record's life by exactly the configured staleness window. + */ +DriverLocationSchema.pre('validate', function (this: IDriverLocation) { + const recordedAt = this.recordedAt ?? new Date(); + this.expiresAt = new Date( + recordedAt.getTime() + env.DRIVER_LOCATION_STALE_AFTER_SECONDS * 1000, + ); +}); + +// ─── Methods ────────────────────────────────────────────────────────────────── + +DriverLocationSchema.methods.toLatLng = function (this: IDriverLocation): { + lat: number; + lng: number; +} { + const [lng, lat] = this.location.coordinates; + return { lat, lng }; +}; + +DriverLocationSchema.statics.toGeoPoint = function (lat: number, lng: number): IGeoPoint { + return { type: 'Point', coordinates: [lng, lat] }; +}; + +export const DriverLocation = model( + 'DriverLocation', + DriverLocationSchema, +); + +export default DriverLocation; diff --git a/src/models/User.ts b/src/models/User.ts index b5b2a73..ca81feb 100644 --- a/src/models/User.ts +++ b/src/models/User.ts @@ -1,6 +1,7 @@ import mongoose, { Schema } from 'mongoose'; import bcrypt from 'bcryptjs'; import { IUser, UserRole, UserStatus } from '../interfaces/IUser'; +import env from '../config/env'; const userSchema = new Schema( { @@ -98,7 +99,7 @@ userSchema.pre('save', async function (next) { } try { - const rounds = parseInt(process.env.BCRYPT_ROUNDS || '10', 10); + const rounds = env.BCRYPT_ROUNDS; const salt = await bcrypt.genSalt(rounds); this.password = await bcrypt.hash(this.password, salt); next(); diff --git a/src/routes/driverRoutes.ts b/src/routes/driverRoutes.ts index 8df32f7..ecb71be 100644 --- a/src/routes/driverRoutes.ts +++ b/src/routes/driverRoutes.ts @@ -1,5 +1,6 @@ import { Router } from 'express'; import { driverController } from '../controllers/driverController'; +import { driverLocationController } from '../controllers/driverLocationController'; import authenticate from '../middleware/authenticate'; import requireRole from '../middleware/requireRole'; import { UserRole } from '../interfaces/IUser'; @@ -25,4 +26,50 @@ router.patch( driverController.setVehicleDetails.bind(driverController), ); +/** + * @route GET /api/v1/drivers/nearby + * @desc Find drivers near a coordinate, nearest first, using the 2dsphere index + * @access Authenticated + */ +router.get( + '/nearby', + authenticate, + driverLocationController.getNearbyDrivers.bind(driverLocationController), +); + +/** + * @route GET /api/v1/drivers/nearby/explain + * @desc Report the query plan and index used by the proximity search + * @access Admin only + */ +router.get( + '/nearby/explain', + authenticate, + requireRole(UserRole.ADMIN), + driverLocationController.explainNearbyQuery.bind(driverLocationController), +); + +/** + * @route PUT /api/v1/drivers/me/location + * @desc Record the authenticated driver's current position + * @access Driver only + */ +router.put( + '/me/location', + authenticate, + requireRole(UserRole.DRIVER), + driverLocationController.updateMyLocation.bind(driverLocationController), +); + +/** + * @route GET /api/v1/drivers/:driverId/location + * @desc Fetch a single driver's most recent position + * @access Authenticated + */ +router.get( + '/:driverId/location', + authenticate, + driverLocationController.getDriverLocation.bind(driverLocationController), +); + export default router; diff --git a/src/routes/index.ts b/src/routes/index.ts index 3704e91..f60bf81 100644 --- a/src/routes/index.ts +++ b/src/routes/index.ts @@ -12,6 +12,9 @@ import profileRoutes from './profileRoutes'; import healthRoutes from './healthRoutes'; import userRoutes from './userRoutes'; import socketMetricsRoutes from './socketMetricsRoutes'; +import bulkDeliveryRoutes from './bulkDeliveryRoutes'; +import notificationRoutes from './notificationRoutes'; +import stellarRoutes from './stellar.routes'; const router = Router(); @@ -32,5 +35,6 @@ router.use('/v1/notifications', notificationRoutes); router.use('/v1/health', healthRoutes); router.use('/v1/socket-metrics', socketMetricsRoutes); router.use('/v1/users', userRoutes); +router.use('/v1/stellar', stellarRoutes); export default router; diff --git a/src/seed.ts b/src/seed.ts index 3ac54d5..661e7c3 100644 --- a/src/seed.ts +++ b/src/seed.ts @@ -1,10 +1,11 @@ import mongoose from 'mongoose'; import dotenv from 'dotenv'; import { Delivery } from './models/Delivery'; +import env from './config/env'; dotenv.config(); -const MONGODB_URI = process.env.MONGODB_URI || 'mongodb://localhost:27017/swiftchain'; +const MONGODB_URI = env.MONGODB_URI; const seedDeliveries = async (): Promise => { try { diff --git a/src/server.ts b/src/server.ts index 7516842..4a9439a 100644 --- a/src/server.ts +++ b/src/server.ts @@ -11,10 +11,11 @@ import { import { startEscrowMonitorJob, stopEscrowMonitorJob } from './jobs/escrowMonitor'; import { startEventPoller, stopEventPoller } from './services/eventPoller'; import { initializeRedis, disconnectRedis } from './config/redis'; +import env from './config/env'; dotenv.config(); -const PORT = process.env.PORT || 8000; +const PORT = env.PORT; const httpServer = http.createServer(app); const io: TypedServer = initializeSocketServer(httpServer); @@ -28,7 +29,7 @@ const initializeServices = async (): Promise => { logger.error('❌ Failed to connect to Redis:', error); logger.warn('⚠️ Distributed locking will not be available'); // Continue without Redis in non-production environments - if (process.env.NODE_ENV === 'production') { + if (env.NODE_ENV === 'production') { process.exit(1); } } @@ -36,7 +37,7 @@ const initializeServices = async (): Promise => { httpServer.listen(PORT, () => { logger.info( - `🚀 Server running on port ${PORT} in ${process.env.NODE_ENV || 'development'} mode` + `🚀 Server running on port ${PORT} in ${env.NODE_ENV} mode` ); logger.info(`📝 Health check: http://localhost:${PORT}/health`); logger.info(`📦 ETA endpoint: http://localhost:${PORT}/api/v1/deliveries/:id/eta`); @@ -49,7 +50,7 @@ httpServer.listen(PORT, () => { startIndexerLagMonitor(); }); -if (process.env.NODE_ENV !== 'test') { +if (env.NODE_ENV !== 'test') { startEscrowMonitorJob(); startEventPoller(); } diff --git a/src/services/driverLocationService.ts b/src/services/driverLocationService.ts new file mode 100644 index 0000000..3542760 --- /dev/null +++ b/src/services/driverLocationService.ts @@ -0,0 +1,468 @@ +/** + * driverLocationService.ts + * + * Business logic for driver positions and proximity search. + * + * Layering: controllers call into this service; only this service touches the + * `DriverLocation` model. Every value returned to a caller is read from + * MongoDB — nothing here fabricates a driver, a distance, or a coordinate. + * + * ── Why `$geoNear` ─────────────────────────────────────────────────────────── + * Proximity search runs as a `$geoNear` aggregation rather than a `$near` + * find(), for two reasons: + * + * 1. `$geoNear` returns the computed distance for each document + * (`distanceField`), so the caller gets real distances from the index + * walk instead of the service recomputing haversine for every result. + * + * 2. Its `query` option is applied *during* the index walk. A `$near` find + * with an extra filter walks outward through every driver in the radius + * and discards the non-matching ones afterwards — on a dense city that is + * most of the work. Pushing the availability filter into `$geoNear` lets + * the compound `{ isAvailable, status, location }` index skip them. + * + * `$geoNear` must be the first stage of its pipeline, and requires a 2dsphere + * index to exist on the collection; both invariants are held here. + */ + +import { StatusCodes } from 'http-status-codes'; +import { PipelineStage, Types } from 'mongoose'; +import { + DriverLocation, + IDriverLocation, + DriverAvailabilityStatus, +} from '../models/DriverLocation'; +import AppError from '../utils/AppError'; +import logger from '../config/logger'; +import env from '../config/env'; + +// ─── Types ──────────────────────────────────────────────────────────────────── + +/** Query accepted by {@link DriverLocationService.findNearbyDrivers}. */ +export interface NearbyDriversQuery { + /** Search-centre latitude, in decimal degrees. */ + lat: number; + /** Search-centre longitude, in decimal degrees. */ + lng: number; + /** Search radius in metres. Defaults to `DRIVER_PROXIMITY_DEFAULT_RADIUS_M`. */ + radiusMeters?: number; + /** Maximum drivers to return. Defaults to `DRIVER_PROXIMITY_MAX_RESULTS`. */ + limit?: number; + /** When true (the default), only drivers marked available are returned. */ + availableOnly?: boolean; + /** Restrict to a specific availability state. */ + status?: DriverAvailabilityStatus; +} + +/** One driver in a proximity result, with the distance MongoDB computed. */ +export interface NearbyDriver { + driverId: string; + /** Straight-line distance from the search centre, in metres. */ + distanceMeters: number; + lat: number; + lng: number; + isAvailable: boolean; + status: DriverAvailabilityStatus; + heading?: number; + speed?: number; + accuracy?: number; + currentDeliveryId?: string; + recordedAt: Date; +} + +/** Result envelope, including the parameters actually applied. */ +export interface NearbyDriversResult { + drivers: NearbyDriver[]; + /** Number of drivers returned. */ + count: number; + /** Radius actually used after clamping, in metres. */ + radiusMeters: number; + /** Search centre echoed back, so a client can confirm what was queried. */ + center: { lat: number; lng: number }; +} + +/** Payload accepted when a driver reports a new position. */ +export interface UpsertDriverLocationInput { + driverId: string; + lat: number; + lng: number; + isAvailable?: boolean; + status?: DriverAvailabilityStatus; + heading?: number; + speed?: number; + accuracy?: number; + currentDeliveryId?: string | null; + recordedAt?: Date; +} + +/** Shape returned by the `$geoNear` pipeline before it is mapped for the API. */ +interface GeoNearRow { + _id: Types.ObjectId; + driverId: Types.ObjectId; + location: { type: 'Point'; coordinates: [number, number] }; + isAvailable: boolean; + status: DriverAvailabilityStatus; + heading?: number; + speed?: number; + accuracy?: number; + currentDeliveryId?: Types.ObjectId | null; + recordedAt: Date; + distanceMeters: number; +} + +// ─── Service ────────────────────────────────────────────────────────────────── + +export class DriverLocationService { + /** + * Find drivers near a point, nearest first. + * + * @param query - Search centre, radius and filters. + * @returns Matching drivers with their distance from the centre. + * + * @throws {AppError} 400 — coordinates, radius or limit outside valid bounds. + * @throws {AppError} 500 — the query failed at the database. + */ + public async findNearbyDrivers(query: NearbyDriversQuery): Promise { + const { lat, lng } = this.assertValidCoordinates(query.lat, query.lng); + const radiusMeters = this.resolveRadius(query.radiusMeters); + const limit = this.resolveLimit(query.limit); + const availableOnly = query.availableOnly ?? true; + + // Filters applied inside the geo index walk rather than after it. + const filter: Record = {}; + if (availableOnly) filter.isAvailable = true; + if (query.status) filter.status = query.status; + + const pipeline: PipelineStage[] = [ + { + $geoNear: { + near: { type: 'Point', coordinates: [lng, lat] }, + distanceField: 'distanceMeters', + maxDistance: radiusMeters, + spherical: true, + query: filter, + key: 'location', + }, + }, + { $limit: limit }, + { + $project: { + driverId: 1, + location: 1, + isAvailable: 1, + status: 1, + heading: 1, + speed: 1, + accuracy: 1, + currentDeliveryId: 1, + recordedAt: 1, + distanceMeters: 1, + }, + }, + ]; + + const startedAt = Date.now(); + + try { + const rows = await DriverLocation.aggregate(pipeline).exec(); + const elapsedMs = Date.now() - startedAt; + + logger.debug( + `[DriverLocationService] Proximity search matched ${rows.length} driver(s) ` + + `within ${radiusMeters}m in ${elapsedMs}ms`, + ); + + return { + drivers: rows.map((row) => this.toNearbyDriver(row)), + count: rows.length, + radiusMeters, + center: { lat, lng }, + }; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + logger.error(`[DriverLocationService] Proximity search failed: ${message}`); + throw new AppError( + 'Unable to search for nearby drivers.', + StatusCodes.INTERNAL_SERVER_ERROR, + ); + } + } + + /** + * Record a driver's current position, creating the record on first report. + * + * The write is a single upsert, so a burst of pings from one device cannot + * create duplicate rows for that driver. + * + * @throws {AppError} 400 — invalid driver id or coordinates. + */ + public async upsertDriverLocation( + input: UpsertDriverLocationInput, + ): Promise { + const driverId = this.assertValidObjectId(input.driverId, 'driverId'); + const { lat, lng } = this.assertValidCoordinates(input.lat, input.lng); + const recordedAt = input.recordedAt ?? new Date(); + + const update: Record = { + location: { type: 'Point', coordinates: [lng, lat] }, + recordedAt, + expiresAt: new Date( + recordedAt.getTime() + env.DRIVER_LOCATION_STALE_AFTER_SECONDS * 1000, + ), + }; + + if (input.isAvailable !== undefined) update.isAvailable = input.isAvailable; + if (input.status !== undefined) update.status = input.status; + if (input.heading !== undefined) update.heading = input.heading; + if (input.speed !== undefined) update.speed = input.speed; + if (input.accuracy !== undefined) update.accuracy = input.accuracy; + if (input.currentDeliveryId !== undefined) { + update.currentDeliveryId = input.currentDeliveryId + ? this.assertValidObjectId(input.currentDeliveryId, 'currentDeliveryId') + : null; + } + + try { + const document = await DriverLocation.findOneAndUpdate( + { driverId }, + { $set: update, $setOnInsert: { driverId } }, + { new: true, upsert: true, runValidators: true, setDefaultsOnInsert: true }, + ).exec(); + + return document; + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + logger.error( + `[DriverLocationService] Failed to persist location for driver ${input.driverId}: ${message}`, + ); + throw new AppError( + 'Unable to record the driver location.', + StatusCodes.INTERNAL_SERVER_ERROR, + ); + } + } + + /** + * Fetch one driver's current position. + * + * @throws {AppError} 400 — malformed driver id. + * @throws {AppError} 404 — no current position on record for that driver. + */ + public async getDriverLocation(driverIdRaw: string): Promise { + const driverId = this.assertValidObjectId(driverIdRaw, 'driverId'); + const document = await DriverLocation.findOne({ driverId }).exec(); + + if (!document) { + throw new AppError( + `No current location on record for driver ${driverIdRaw}.`, + StatusCodes.NOT_FOUND, + ); + } + return document; + } + + /** + * Run the proximity query with `explain()` and report which index the + * planner chose plus how much work it did. + * + * This is the profiling hook the issue calls for: it makes index regressions + * observable, rather than something that only shows up as latency in + * production. It reads real plans from the real collection. + * + * @returns The planner summary and the raw `explain` output. + */ + public async explainProximityQuery(query: NearbyDriversQuery): Promise<{ + indexUsed: string; + executionTimeMillis: number; + totalDocsExamined: number; + nReturned: number; + raw: unknown; + }> { + const { lat, lng } = this.assertValidCoordinates(query.lat, query.lng); + const radiusMeters = this.resolveRadius(query.radiusMeters); + const availableOnly = query.availableOnly ?? true; + + const filter: Record = {}; + if (availableOnly) filter.isAvailable = true; + if (query.status) filter.status = query.status; + + const pipeline: PipelineStage[] = [ + { + $geoNear: { + near: { type: 'Point', coordinates: [lng, lat] }, + distanceField: 'distanceMeters', + maxDistance: radiusMeters, + spherical: true, + query: filter, + key: 'location', + }, + }, + { $limit: this.resolveLimit(query.limit) }, + ]; + + const raw = (await DriverLocation.aggregate(pipeline) + .option({ explain: true }) + .exec()) as unknown; + + const summary = this.summariseExplain(raw); + + logger.info( + `[DriverLocationService] Proximity explain — index=${summary.indexUsed} ` + + `docsExamined=${summary.totalDocsExamined} returned=${summary.nReturned} ` + + `timeMs=${summary.executionTimeMillis}`, + ); + + return { ...summary, raw }; + } + + /** + * Ensure every index declared on the schema exists in MongoDB. + * + * Mongoose builds indexes in the background on first use, which means the + * very first proximity query after a deploy can run without them. Calling + * this at startup makes index creation explicit and surfaces failures. + * + * @returns The names of the indexes present after synchronisation. + */ + public async ensureIndexes(): Promise { + await DriverLocation.createIndexes(); + const indexes = (await DriverLocation.collection.indexes()) as Array<{ name?: string }>; + const names = indexes.map((index) => index.name ?? 'unnamed'); + + logger.info(`[DriverLocationService] DriverLocation indexes ready: ${names.join(', ')}`); + return names; + } + + // ── Private helpers ──────────────────────────────────────────────────────── + + /** Map a raw `$geoNear` row onto the API shape, rounding the distance. */ + private toNearbyDriver(row: GeoNearRow): NearbyDriver { + const [lng, lat] = row.location.coordinates; + + return { + driverId: row.driverId.toString(), + distanceMeters: Math.round(row.distanceMeters), + lat, + lng, + isAvailable: row.isAvailable, + status: row.status, + heading: row.heading, + speed: row.speed, + accuracy: row.accuracy, + currentDeliveryId: row.currentDeliveryId ? row.currentDeliveryId.toString() : undefined, + recordedAt: row.recordedAt, + }; + } + + /** Validate a latitude/longitude pair. */ + private assertValidCoordinates(lat: number, lng: number): { lat: number; lng: number } { + if (!Number.isFinite(lat) || lat < -90 || lat > 90) { + throw new AppError( + 'lat must be a number between -90 and 90.', + StatusCodes.BAD_REQUEST, + ); + } + if (!Number.isFinite(lng) || lng < -180 || lng > 180) { + throw new AppError( + 'lng must be a number between -180 and 180.', + StatusCodes.BAD_REQUEST, + ); + } + return { lat, lng }; + } + + /** Clamp the radius to the configured maximum, defaulting when absent. */ + private resolveRadius(radiusMeters?: number): number { + if (radiusMeters === undefined) return env.DRIVER_PROXIMITY_DEFAULT_RADIUS_M; + + if (!Number.isFinite(radiusMeters) || radiusMeters <= 0) { + throw new AppError('radiusMeters must be a positive number.', StatusCodes.BAD_REQUEST); + } + + if (radiusMeters > env.DRIVER_PROXIMITY_MAX_RADIUS_M) { + throw new AppError( + `radiusMeters cannot exceed ${env.DRIVER_PROXIMITY_MAX_RADIUS_M}.`, + StatusCodes.BAD_REQUEST, + ); + } + return radiusMeters; + } + + /** Clamp the result limit to the configured maximum, defaulting when absent. */ + private resolveLimit(limit?: number): number { + if (limit === undefined) return env.DRIVER_PROXIMITY_MAX_RESULTS; + + if (!Number.isInteger(limit) || limit <= 0) { + throw new AppError('limit must be a positive integer.', StatusCodes.BAD_REQUEST); + } + return Math.min(limit, env.DRIVER_PROXIMITY_MAX_RESULTS); + } + + /** Validate and cast a Mongo ObjectId supplied as a string. */ + private assertValidObjectId(value: string, field: string): Types.ObjectId { + if (!Types.ObjectId.isValid(value)) { + throw new AppError(`${field} must be a valid ObjectId.`, StatusCodes.BAD_REQUEST); + } + return new Types.ObjectId(value); + } + + /** + * Pull the interesting numbers out of an `explain` document. + * + * The shape differs between standalone servers, sharded clusters and server + * versions, so every field is probed defensively and falls back to a stable + * default rather than throwing. + */ + private summariseExplain(raw: unknown): { + indexUsed: string; + executionTimeMillis: number; + totalDocsExamined: number; + nReturned: number; + } { + const root = (Array.isArray(raw) ? raw[0] : raw) as Record | undefined; + + const stages = (root?.stages as Array> | undefined) ?? []; + const geoNearStage = stages.find((stage) => '$geoNearCursor' in stage || '$cursor' in stage); + + const cursor = (geoNearStage?.['$geoNearCursor'] ?? geoNearStage?.['$cursor'] ?? root) as + | Record + | undefined; + + const queryPlanner = cursor?.queryPlanner as Record | undefined; + const executionStats = (cursor?.executionStats ?? root?.executionStats) as + | Record + | undefined; + + const winningPlan = queryPlanner?.winningPlan as Record | undefined; + + return { + indexUsed: this.findIndexName(winningPlan) ?? 'unknown', + executionTimeMillis: Number(executionStats?.executionTimeMillis ?? 0), + totalDocsExamined: Number(executionStats?.totalDocsExamined ?? 0), + nReturned: Number(executionStats?.nReturned ?? 0), + }; + } + + /** Walk a winning plan tree looking for the first `indexName`. */ + private findIndexName(plan: Record | undefined): string | undefined { + if (!plan || typeof plan !== 'object') return undefined; + + if (typeof plan.indexName === 'string') return plan.indexName; + + for (const value of Object.values(plan)) { + if (Array.isArray(value)) { + for (const item of value) { + const found = this.findIndexName(item as Record); + if (found) return found; + } + } else if (value && typeof value === 'object') { + const found = this.findIndexName(value as Record); + if (found) return found; + } + } + return undefined; + } +} + +/** Singleton used by the controller layer. */ +export const driverLocationService = new DriverLocationService(); diff --git a/src/services/etaCacheService.ts b/src/services/etaCacheService.ts index f9631cd..d97a916 100644 --- a/src/services/etaCacheService.ts +++ b/src/services/etaCacheService.ts @@ -2,6 +2,7 @@ import logger from '../config/logger'; import { getRedisClient } from '../config/redis'; import { buildEtaCacheKey } from '../utils/etaCacheKey'; import { Coordinates, ETAResponse, TravelMode } from '../types/routing.types'; +import env from '../config/env'; export interface EtaCacheLookup { pickup: Coordinates; @@ -83,13 +84,11 @@ export class EtaCacheService { } private readTtlSeconds(): number { - const parsed = parseInt(process.env.ETA_CACHE_TTL_SECONDS ?? '600', 10); - return Number.isFinite(parsed) && parsed > 0 ? parsed : 600; + return env.ETA_CACHE_TTL_SECONDS; } private readGeohashPrecision(): number { - const parsed = parseInt(process.env.ETA_GEOHASH_PRECISION ?? '7', 10); - return Number.isFinite(parsed) && parsed >= 1 && parsed <= 12 ? parsed : 7; + return env.ETA_GEOHASH_PRECISION; } } diff --git a/src/services/gracefulShutdownService.ts b/src/services/gracefulShutdownService.ts index 0e0f4c5..60bb32a 100644 --- a/src/services/gracefulShutdownService.ts +++ b/src/services/gracefulShutdownService.ts @@ -12,6 +12,7 @@ import { shutdownSocketServer, TypedServer, } from '../sockets/connectionHandler'; +import env from '../config/env'; /** Default max time (ms) to wait before forcing process exit. */ const DEFAULT_SHUTDOWN_TIMEOUT_MS = 30_000; @@ -53,7 +54,7 @@ export class GracefulShutdownService { this.exitFn = options.exitFn ?? ((code: number) => process.exit(code)); this.timeoutMs = options.timeoutMs ?? - parseInt(process.env.SHUTDOWN_TIMEOUT_MS ?? String(DEFAULT_SHUTDOWN_TIMEOUT_MS), 10); + env.SHUTDOWN_TIMEOUT_MS; } /** diff --git a/src/services/routingService.ts b/src/services/routingService.ts index 5fb5d04..16b2294 100644 --- a/src/services/routingService.ts +++ b/src/services/routingService.ts @@ -1,4 +1,5 @@ import axios from 'axios'; +import env from '../config/env'; export interface Coordinates { lat: number; @@ -31,7 +32,7 @@ class RoutingService { private readonly baseUrl: string; constructor() { - this.apiKey = process.env.GOOGLE_MAPS_API_KEY || ''; + this.apiKey = env.GOOGLE_MAPS_API_KEY; this.baseUrl = 'https://maps.googleapis.com/maps/api/directions/json'; if (!this.apiKey) { diff --git a/src/services/stellarService.ts b/src/services/stellarService.ts index 12a3b1f..1925765 100644 --- a/src/services/stellarService.ts +++ b/src/services/stellarService.ts @@ -16,6 +16,12 @@ import { toStroops, fromStroops } from '../utils/stroops'; import AppError from '../utils/AppError'; import logger from '../config/logger'; import env from '../config/env'; +import { + withRetry, + OperationTimeoutError, + sleep, + type AttemptFailureKind, +} from '../utils/rpcRetry'; // ─── Public types ────────────────────────────────────────────────────────────── @@ -65,6 +71,94 @@ function extractMessage(error: unknown): string { return String(error); } +/** + * Node-level socket error codes that mean "the request never got a reply", + * as opposed to "the node answered and said no". + */ +const TRANSIENT_ERROR_CODES = new Set([ + 'ECONNRESET', + 'ECONNREFUSED', + 'ECONNABORTED', + 'ETIMEDOUT', + 'ENOTFOUND', + 'EAI_AGAIN', + 'EHOSTUNREACH', + 'ENETUNREACH', + 'EPIPE', + 'ERR_SOCKET_CONNECTION_TIMEOUT', +]); + +/** + * HTTP statuses worth retrying: the node is rate-limiting us, is briefly + * unavailable, or a proxy in front of it failed. A 4xx other than 429 means + * the request itself is wrong and will fail identically on every retry. + */ +const TRANSIENT_HTTP_STATUSES = new Set([408, 425, 429, 500, 502, 503, 504]); + +/** Extract an HTTP status code from the various shapes the SDK surfaces. */ +function extractStatusCode(error: unknown): number | undefined { + if (typeof error !== 'object' || error === null) return undefined; + + const candidate = error as { + status?: unknown; + statusCode?: unknown; + response?: { status?: unknown }; + }; + + for (const value of [candidate.status, candidate.statusCode, candidate.response?.status]) { + if (typeof value === 'number' && Number.isFinite(value)) return value; + } + return undefined; +} + +/** + * Decide whether an RPC failure is transient and therefore worth retrying. + * + * Retried: + * - per-attempt timeouts, + * - socket-level failures (connection reset, DNS hiccup, unreachable host), + * - HTTP 408/425/429 and 5xx. + * + * Not retried: + * - anything else, notably 4xx responses and malformed-request errors, + * which are deterministic — retrying only adds latency before the same + * failure, and for a submission it risks duplicating work. + * + * @param error - The thrown value. + * @param kind - Whether the attempt timed out or rejected. + * @returns `true` when another attempt could plausibly succeed. + */ +export function isTransientRpcError(error: unknown, kind: AttemptFailureKind = 'error'): boolean { + if (kind === 'timeout' || error instanceof OperationTimeoutError) return true; + + const status = extractStatusCode(error); + if (status !== undefined) return TRANSIENT_HTTP_STATUSES.has(status); + + const code = (error as { code?: unknown })?.code; + if (typeof code === 'string' && TRANSIENT_ERROR_CODES.has(code)) return true; + + const message = extractMessage(error).toLowerCase(); + + // A bad sequence number is handled by its own dedicated retry path, which + // rebuilds the envelope. Retrying the identical XDR here would always fail. + if (message.includes('tx_bad_seq') || message.includes('txbadseq')) return false; + + return ( + message.includes('timeout') || + message.includes('timed out') || + message.includes('socket hang up') || + message.includes('network error') || + message.includes('econnreset') || + message.includes('econnrefused') || + message.includes('enotfound') || + message.includes('eai_again') || + message.includes('service unavailable') || + message.includes('bad gateway') || + message.includes('gateway timeout') || + message.includes('too many requests') + ); +} + /** * Inspect a `SendTransactionResponse` or a thrown error and decide whether it * represents a sequence-number mismatch (`tx_bad_seq`). @@ -111,10 +205,6 @@ function isBadSeqError( return msg.includes('tx_bad_seq') || msg.includes('txbadseq'); } -/** Sleep helper used between retry attempts. */ -function sleep(ms: number): Promise { - return new Promise((resolve) => setTimeout(resolve, ms)); -} // ─── StellarService ──────────────────────────────────────────────────────────── @@ -149,10 +239,74 @@ function sleep(ms: number): Promise { export class StellarService { private readonly client: StellarRpc.Server; private readonly badSeqMaxRetries: number; + private readonly rpcMaxAttempts: number; + private readonly rpcBaseDelayMs: number; + private readonly rpcMaxDelayMs: number; + private readonly rpcJitterRatio: number; + private readonly rpcTimeoutMs: number; constructor(client: StellarRpc.Server = sorobanRpcClient) { this.client = client; this.badSeqMaxRetries = env.STELLAR_BAD_SEQ_MAX_RETRIES; + this.rpcMaxAttempts = env.SOROBAN_RPC_MAX_RETRIES; + this.rpcBaseDelayMs = env.SOROBAN_RPC_RETRY_BASE_MS; + this.rpcMaxDelayMs = env.SOROBAN_RPC_RETRY_MAX_MS; + this.rpcJitterRatio = env.SOROBAN_RPC_RETRY_JITTER_RATIO; + this.rpcTimeoutMs = stellarConfig.timeoutMs; + } + + /** + * Run a single Soroban RPC call under the shared resilience policy: + * per-attempt timeout, exponential backoff with jitter, and retries limited + * to transient failures. + * + * Every attempt that fails is logged at `warn` with the reason and the delay + * before the next try; a call that succeeds only after retrying logs a + * `info` recovery line. That pairing is what makes an intermittent RPC node + * visible in the logs instead of silently inflating latency. + * + * @param operation - Short label used in logs and timeout messages. + * @param factory - Produces a fresh promise per attempt. + * @param context - Extra key/value pairs appended to each log line. + */ + private async callRpc( + operation: string, + factory: () => Promise, + context: Record = {}, + ): Promise { + const suffix = Object.entries(context) + .map(([key, value]) => ` ${key}=${String(value)}`) + .join(''); + + return withRetry(factory, { + maxAttempts: this.rpcMaxAttempts, + baseDelayMs: this.rpcBaseDelayMs, + maxDelayMs: this.rpcMaxDelayMs, + jitter: this.rpcJitterRatio, + timeoutMs: this.rpcTimeoutMs, + operationName: operation, + isRetryable: isTransientRpcError, + onAttemptFailed: ({ attempt, maxAttempts, kind, error, delayMs }) => { + const reason = kind === 'timeout' ? 'timeout' : extractMessage(error); + if (delayMs > 0) { + logger.warn( + `[StellarService] RPC '${operation}' attempt ${attempt}/${maxAttempts} failed ` + + `(${kind}): ${reason} — retrying in ${delayMs}ms${suffix}`, + ); + } else { + logger.error( + `[StellarService] RPC '${operation}' failed permanently after ` + + `${attempt}/${maxAttempts} attempt(s) (${kind}): ${reason}${suffix}`, + ); + } + }, + onRecovery: ({ attempt, elapsedMs }) => { + logger.info( + `[StellarService] RPC '${operation}' recovered on attempt ${attempt} ` + + `after ${elapsedMs}ms${suffix}`, + ); + }, + }); } // ── Public API ────────────────────────────────────────────────────────────── @@ -320,10 +474,25 @@ export class StellarService { stellarConfig.networkPassphrase, ) as Transaction; - return await this.client.sendTransaction(tx); + return await this.callRpc('sendTransaction', () => this.client.sendTransaction(tx)); } catch (error) { const message = extractMessage(error); + if (error instanceof OperationTimeoutError) { + // The node never answered. The transaction may or may not have been + // accepted, so this is surfaced as a gateway timeout rather than + // resubmitted blindly — a duplicate submission could double-spend. + logger.error( + `[StellarService] sendTransaction timed out after ` + + `${this.rpcMaxAttempts} attempt(s): ${message}`, + ); + throw new AppError( + 'The Soroban RPC node did not respond to the transaction submission in time. ' + + 'The transaction may still have been accepted — verify by hash before resubmitting.', + StatusCodes.GATEWAY_TIMEOUT, + ); + } + // If the SDK itself throws with bad-seq language surface it as a // synthetic response object so the caller's isBadSeqError check works. if (message.toLowerCase().includes('tx_bad_seq') || message.toLowerCase().includes('txbadseq')) { @@ -367,7 +536,13 @@ export class StellarService { let txResponse: StellarRpc.Api.GetTransactionResponse; try { - txResponse = await this.client.getTransaction(hash); + // Each poll is itself retried, so a momentary blip does not consume a + // whole polling slot and shorten the confirmation window. + txResponse = await this.callRpc( + 'getTransaction', + () => this.client.getTransaction(hash), + { hash }, + ); } catch (error) { logger.warn( `[StellarService] getTransaction poll ${poll}/${maxPolls} failed: ${extractMessage(error)}`, @@ -443,9 +618,21 @@ export class StellarService { private async loadAccount(payerAddress: string): Promise { try { - return await this.client.getAccount(payerAddress); + return await this.callRpc( + 'getAccount', + () => this.client.getAccount(payerAddress), + { payer: payerAddress }, + ); } catch (error) { const message = extractMessage(error); + + if (error instanceof OperationTimeoutError) { + throw new AppError( + 'Timed out loading the payer account from the Soroban RPC node.', + StatusCodes.GATEWAY_TIMEOUT, + ); + } + if (message.toLowerCase().includes('not found')) { throw new AppError( `Account ${payerAddress} does not exist on ${stellarConfig.network}. ` + @@ -463,9 +650,19 @@ export class StellarService { private async prepare(transaction: Transaction): Promise { try { - return await this.client.prepareTransaction(transaction); + return await this.callRpc('prepareTransaction', () => + this.client.prepareTransaction(transaction), + ); } catch (error) { const message = extractMessage(error); + + if (error instanceof OperationTimeoutError) { + throw new AppError( + 'Timed out simulating the transaction against the Soroban RPC node.', + StatusCodes.GATEWAY_TIMEOUT, + ); + } + logger.error(`[StellarService] Simulation failed during rebuild: ${message}`); throw new AppError( `Soroban simulation failed while rebuilding transaction: ${message}`, diff --git a/src/sockets/connectionHandler.ts b/src/sockets/connectionHandler.ts index 3b29d8e..e36abba 100644 --- a/src/sockets/connectionHandler.ts +++ b/src/sockets/connectionHandler.ts @@ -15,6 +15,7 @@ import { TypedSocket, } from './socket.types'; import jwt from 'jsonwebtoken'; +import env from '../config/env'; /** * Typed Socket.IO server alias used throughout the sockets layer. @@ -42,15 +43,15 @@ export type TypedServer = SocketIOServer< export function initializeSocketServer(httpServer: HttpServer): TypedServer { const io: TypedServer = new SocketIOServer(httpServer, { cors: { - origin: process.env.CORS_ORIGIN || '*', + origin: env.CORS_ORIGIN, methods: ['GET', 'POST'], credentials: true, }, // Use Socket.IO's built-in transport-level ping/pong as a fallback - pingTimeout: parseInt(process.env.SOCKET_PING_TIMEOUT_MS ?? '20000', 10), - pingInterval: parseInt(process.env.SOCKET_PING_INTERVAL_MS ?? '25000', 10), + pingTimeout: env.SOCKET_PING_TIMEOUT_MS, + pingInterval: env.SOCKET_PING_INTERVAL_MS, // Allow only websocket transport in production for efficiency - transports: process.env.NODE_ENV === 'production' ? ['websocket'] : ['websocket', 'polling'], + transports: env.NODE_ENV === 'production' ? ['websocket'] : ['websocket', 'polling'], }); // ─── Per-connection setup ────────────────────────────────────────────────── @@ -86,7 +87,7 @@ export function initializeSocketServer(httpServer: HttpServer): TypedServer { } try { const rawToken = token.startsWith('Bearer ') ? token.slice(7) : token; - const decoded = jwt.verify(rawToken, process.env.JWT_SECRET as string) as { userId?: string; exp?: number }; + const decoded = jwt.verify(rawToken, env.JWT_SECRET) as { userId?: string; exp?: number }; const newExp = typeof decoded.exp === 'number' ? decoded.exp * 1000 : undefined; if (newExp) { (socket.data as any).tokenExp = newExp; @@ -209,7 +210,7 @@ function extractAuthInfo(socket: TypedSocket): { userId?: string; tokenExp?: num try { const rawToken = token.startsWith('Bearer ') ? token.slice(7) : token; - const decoded = jwt.verify(rawToken, process.env.JWT_SECRET as string) as { userId?: string; exp?: number }; + const decoded = jwt.verify(rawToken, env.JWT_SECRET) as { userId?: string; exp?: number }; return { userId: typeof decoded.userId === 'string' ? decoded.userId : undefined, tokenExp: typeof decoded.exp === 'number' ? decoded.exp * 1000 : undefined, diff --git a/src/sockets/index.ts b/src/sockets/index.ts index d2e7220..d62ef90 100644 --- a/src/sockets/index.ts +++ b/src/sockets/index.ts @@ -3,12 +3,13 @@ import { Server as HttpServer } from 'http'; import registerSocketHandlers from './socketController'; import logger from '../config/logger'; import socketAuth from '../middlewares/socketAuth'; +import env from '../config/env'; export const initSocket = (httpServer: HttpServer): Server => { const io = new Server(httpServer, { path: '/socket.io', cors: { - origin: process.env.CORS_ORIGIN || '*', + origin: env.CORS_ORIGIN, methods: ['GET', 'POST'], }, }); diff --git a/src/sockets/location.service.ts b/src/sockets/location.service.ts index de00385..0b01b25 100644 --- a/src/sockets/location.service.ts +++ b/src/sockets/location.service.ts @@ -12,6 +12,7 @@ import { InterServerEvents, SocketData, } from './socket.types'; +import env from '../config/env'; /** * Room name prefix for delivery-scoped broadcast rooms. @@ -24,21 +25,21 @@ export const DELIVERY_ROOM_PREFIX = 'delivery:'; * Updates with the same deduplication key within this window are rejected. * Default: 60 seconds (can be overridden via LOCATION_DEDUP_TTL_SECONDS env var). */ -const DEDUP_TTL_SECONDS = parseInt(process.env.LOCATION_DEDUP_TTL_SECONDS ?? '60', 10); +const DEDUP_TTL_SECONDS = env.LOCATION_DEDUP_TTL_SECONDS; /** * Maximum age (in milliseconds) for a location update to be considered valid. * Updates older than this are rejected as stale. * Default: 5 minutes (can be overridden via LOCATION_MAX_AGE_MS env var). */ -const MAX_UPDATE_AGE_MS = parseInt(process.env.LOCATION_MAX_AGE_MS ?? '300000', 10); +const MAX_UPDATE_AGE_MS = env.LOCATION_MAX_AGE_MS; /** * Maximum future timestamp tolerance (in milliseconds). * Updates with timestamps more than this far in the future are rejected. * Default: 30 seconds (can be overridden via LOCATION_MAX_FUTURE_MS env var). */ -const MAX_FUTURE_TOLERANCE_MS = parseInt(process.env.LOCATION_MAX_FUTURE_MS ?? '30000', 10); +const MAX_FUTURE_TOLERANCE_MS = env.LOCATION_MAX_FUTURE_MS; /** * Build the canonical Socket.IO room name for a delivery. diff --git a/src/sockets/locationHandler.ts b/src/sockets/locationHandler.ts index c6a21ef..6dfb1f4 100644 --- a/src/sockets/locationHandler.ts +++ b/src/sockets/locationHandler.ts @@ -15,6 +15,7 @@ import { AuthRefreshPayload, AuthRefreshAckPayload, } from './socket.types'; +import env from '../config/env'; /** * Typed Socket.IO server alias. @@ -141,8 +142,8 @@ function setupTokenExpirationCheck(io: TypedServer, socket: TypedSocket): void { return; } - const CHECK_INTERVAL_MS = parseInt(process.env.SOCKET_TOKEN_CHECK_INTERVAL_MS ?? '60000', 10); - const GRACE_PERIOD_MS = parseInt(process.env.SOCKET_TOKEN_GRACE_PERIOD_MS ?? '30000', 10); + const CHECK_INTERVAL_MS = env.SOCKET_TOKEN_CHECK_INTERVAL_MS; + const GRACE_PERIOD_MS = env.SOCKET_TOKEN_GRACE_PERIOD_MS; let graceTimer: NodeJS.Timeout | null = null; let checkInterval: NodeJS.Timeout | null = null; diff --git a/src/sockets/messageQueue.ts b/src/sockets/messageQueue.ts index cb2aec1..b3ccb81 100644 --- a/src/sockets/messageQueue.ts +++ b/src/sockets/messageQueue.ts @@ -1,4 +1,5 @@ import { randomUUID } from 'crypto'; +import env from '../config/env'; export interface QueuedSocketMessage { id: string; @@ -20,10 +21,7 @@ export interface EnqueueSocketMessageOptions { export class MessageQueueService { private readonly queues = new Map(); - private readonly defaultAckTimeoutMs = parseInt( - process.env.SOCKET_MESSAGE_ACK_TIMEOUT_MS ?? '15000', - 10, - ); + private readonly defaultAckTimeoutMs = env.SOCKET_MESSAGE_ACK_TIMEOUT_MS; public enqueue( userId: string, diff --git a/src/sockets/socket.service.ts b/src/sockets/socket.service.ts index e297179..3bd28cf 100644 --- a/src/sockets/socket.service.ts +++ b/src/sockets/socket.service.ts @@ -13,19 +13,20 @@ import { SocketData, } from './socket.types'; import { messageQueueService } from './messageQueue'; +import env from '../config/env'; /** * Interval (ms) between server-initiated ping events. * Defaults to 25 s, overridable via SOCKET_PING_INTERVAL_MS env var. */ -const PING_INTERVAL_MS = parseInt(process.env.SOCKET_PING_INTERVAL_MS ?? '25000', 10); +const PING_INTERVAL_MS = env.SOCKET_PING_INTERVAL_MS; /** * Maximum number of consecutive missed pongs before a connection is * considered stale and forcibly disconnected. * Defaults to 2, overridable via SOCKET_MAX_MISSED_PONGS env var. */ -const MAX_MISSED_PONGS = parseInt(process.env.SOCKET_MAX_MISSED_PONGS ?? '2', 10); +const MAX_MISSED_PONGS = env.SOCKET_MAX_MISSED_PONGS; /** * SocketService manages all business-logic concerns for WebSocket diff --git a/src/sockets/sync.service.ts b/src/sockets/sync.service.ts index 185a189..8aedaff 100644 --- a/src/sockets/sync.service.ts +++ b/src/sockets/sync.service.ts @@ -7,13 +7,14 @@ import { LocationSyncAck, SyncItemResult, } from './socket.types'; +import env from '../config/env'; /** * Maximum number of location points accepted in a single sync batch. * Protects against abusive or runaway clients. * Overridable via SYNC_BATCH_SIZE_LIMIT env var. */ -const BATCH_SIZE_LIMIT = parseInt(process.env.SYNC_BATCH_SIZE_LIMIT ?? '500', 10); +const BATCH_SIZE_LIMIT = env.SYNC_BATCH_SIZE_LIMIT; /** * SyncService handles the business logic for offline catch-up sync: diff --git a/src/utils/piiMasker.ts b/src/utils/piiMasker.ts new file mode 100644 index 0000000..3e73b11 --- /dev/null +++ b/src/utils/piiMasker.ts @@ -0,0 +1,280 @@ +/** + * piiMasker.ts + * + * Redaction helpers used by the unified logging interface. + * + * Two complementary strategies are applied to everything that reaches a log + * transport, because sensitive values arrive in two very different shapes: + * + * 1. **Key-based redaction** — structured metadata such as + * `{ password: 'hunter2' }`. Any key whose name matches a known-sensitive + * pattern (password, token, secret, key, authorization, …) has its value + * replaced wholesale. This is the strongest guarantee: the value is never + * inspected, so a secret is redacted regardless of its format. + * + * 2. **Pattern-based masking** — free-text strings such as + * `"login failed for alice@example.com"`. Values are scanned for + * recognisable PII (emails, JWTs, Stellar keys, card numbers, phone + * numbers, bearer tokens) and each match is partially masked. + * + * Masking is *partial* wherever it is safe to be. `alice@example.com` becomes + * `al***@example.com` rather than a flat `[REDACTED]`, because operators still + * need to correlate log lines during an incident. Anything that is purely a + * credential (passwords, private keys, tokens) is redacted in full. + * + * The module is intentionally dependency-free and side-effect-free so it can + * be unit-tested in isolation and reused outside the logger. + */ + +/** Replacement written in place of a fully redacted value. */ +export const REDACTED = '[REDACTED]'; + +/** + * Depth limit applied when walking nested objects. + * + * Log metadata is occasionally a deep or cyclic graph (a Mongoose document, an + * Axios error carrying the whole request/response). Bounding the walk keeps a + * single log call from becoming a performance problem. + */ +const MAX_DEPTH = 8; + +/** Upper bound on array elements visited per array. */ +const MAX_ARRAY_ITEMS = 100; + +/** + * Object keys whose values are always redacted in full, matched + * case-insensitively against the key name. + * + * The list is deliberately broad: over-redacting a log field is a cosmetic + * problem, while under-redacting one is a security incident. + */ +const SENSITIVE_KEY_PATTERN = + /(pass(word|phrase)?|secret|token|api[-_]?key|private[-_]?key|secret[-_]?key|authorization|auth|credential|cookie|session[-_]?id|signature|signed[-_]?xdr|mnemonic|seed|otp|pin|cvv|ssn|refresh|access[-_]?token|client[-_]?secret)/i; + +/** + * Keys that match {@link SENSITIVE_KEY_PATTERN} but are safe to keep, because + * they carry no secret material and are valuable for debugging. + * + * Checked before the sensitive pattern so it always wins. + */ +const SENSITIVE_KEY_ALLOWLIST = /^(tokenType|authProvider|authMethod|hasToken|tokenExpiresAt)$/i; + +// ─── Value patterns ─────────────────────────────────────────────────────────── + +/** RFC-5322-ish email address. */ +const EMAIL_PATTERN = /\b([A-Za-z0-9._%+-])([A-Za-z0-9._%+-]*)@([A-Za-z0-9.-]+\.[A-Za-z]{2,})\b/g; + +/** JSON Web Token — three base64url segments separated by dots. */ +const JWT_PATTERN = /\beyJ[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\.[A-Za-z0-9_-]+\b/g; + +/** `Authorization: Bearer ` style headers embedded in text. */ +const BEARER_PATTERN = /\b(Bearer|Basic)\s+[A-Za-z0-9._~+/=-]{8,}/gi; + +/** + * Stellar secret seed (`S...`, 56 chars). Always redacted in full — a secret + * seed is a spending key. + */ +const STELLAR_SECRET_PATTERN = /\bS[A-Z2-7]{55}\b/g; + +/** + * Stellar public key (`G...`) or contract id (`C...`), 56 chars. + * Public identifiers, so these are partially masked rather than removed: + * operators routinely need them to trace a transaction. + */ +const STELLAR_PUBLIC_PATTERN = /\b([GC])([A-Z2-7]{55})\b/g; + +/** PEM private key blocks. */ +const PEM_PATTERN = /-----BEGIN [A-Z ]*PRIVATE KEY-----[\s\S]*?-----END [A-Z ]*PRIVATE KEY-----/g; + +/** 13-19 digit payment card numbers, optionally separated by spaces/hyphens. */ +const CARD_PATTERN = /\b(?:\d[ -]?){12,18}\d\b/g; + +/** E.164-ish phone numbers with an explicit `+` country code. */ +const PHONE_PATTERN = /\+\d{7,15}\b/g; + +/** + * MongoDB / Redis connection strings that embed credentials. + * Captures the scheme and host so the target stays identifiable. + */ +const CONNECTION_STRING_PATTERN = /\b([a-z+]+:\/\/)([^:@\s/]+):([^@\s/]+)@/gi; + +// ─── Primitive maskers ──────────────────────────────────────────────────────── + +/** + * Mask an email address, preserving the first two local-part characters and + * the full domain so log lines remain correlatable. + * + * `alice@example.com` → `al***@example.com` + */ +function maskEmail(_match: string, first: string, rest: string, domain: string): string { + const prefix = rest.length > 0 ? first + rest.charAt(0) : first; + return `${prefix}***@${domain}`; +} + +/** + * Mask a card number, keeping only the last four digits (the maximum PCI-DSS + * permits to be retained in logs). + */ +function maskCard(match: string): string { + const digits = match.replace(/[^\d]/g, ''); + // Luhn-check so ordinary long numbers (ledger sequences, ids) survive intact. + if (!isLuhnValid(digits)) return match; + return `****-****-****-${digits.slice(-4)}`; +} + +/** + * Validate a digit string with the Luhn algorithm. + * + * Used to distinguish real card numbers from other long numeric strings, so + * that ledger sequences and database ids are not mangled by {@link maskCard}. + */ +function isLuhnValid(digits: string): boolean { + if (digits.length < 13 || digits.length > 19) return false; + + let sum = 0; + let double = false; + + for (let i = digits.length - 1; i >= 0; i--) { + let digit = digits.charCodeAt(i) - 48; + if (double) { + digit *= 2; + if (digit > 9) digit -= 9; + } + sum += digit; + double = !double; + } + + return sum % 10 === 0; +} + +/** Mask a phone number, keeping the country code and last two digits. */ +function maskPhone(match: string): string { + if (match.length <= 5) return match; + return `${match.slice(0, 3)}${'*'.repeat(match.length - 5)}${match.slice(-2)}`; +} + +/** Mask a Stellar public key / contract id: `GABC…XYZ` → `GABC***XYZ`. */ +function maskStellarPublic(match: string): string { + return `${match.slice(0, 4)}***${match.slice(-4)}`; +} + +/** + * Apply every value-level pattern to a free-text string. + * + * Order matters: the most specific and most dangerous patterns run first so a + * broader pattern cannot partially consume them. + */ +export function maskString(value: string): string { + if (!value) return value; + + return value + .replace(PEM_PATTERN, REDACTED) + .replace(JWT_PATTERN, REDACTED) + .replace(BEARER_PATTERN, (m) => `${m.split(/\s+/)[0]} ${REDACTED}`) + .replace(STELLAR_SECRET_PATTERN, REDACTED) + .replace(CONNECTION_STRING_PATTERN, (_m, scheme: string, user: string) => + `${scheme}${user}:${REDACTED}@`, + ) + .replace(EMAIL_PATTERN, maskEmail) + .replace(CARD_PATTERN, maskCard) + .replace(PHONE_PATTERN, maskPhone) + .replace(STELLAR_PUBLIC_PATTERN, maskStellarPublic); +} + +/** Whether an object key should have its value redacted in full. */ +export function isSensitiveKey(key: string): boolean { + if (SENSITIVE_KEY_ALLOWLIST.test(key)) return false; + return SENSITIVE_KEY_PATTERN.test(key); +} + +// ─── Recursive walker ───────────────────────────────────────────────────────── + +/** + * Recursively mask an arbitrary value. + * + * Behaviour by type: + * - `string` — value patterns applied. + * - `object` — walked; keys matching {@link isSensitiveKey} are redacted. + * - `Error` — `name`/`message`/`stack` preserved (message masked) so + * stack traces survive redaction. + * - `Date`/`Buffer`/`RegExp` — passed through or summarised, never walked. + * - everything else — returned unchanged. + * + * Cycles are tracked with a `WeakSet`, so a self-referential object logs as + * `'[Circular]'` instead of overflowing the stack. + * + * @param value - The value to mask. + * @param depth - Current recursion depth (internal). + * @param seen - Objects already visited on this path (internal). + * @returns A masked deep copy. The input is never mutated. + */ +export function maskValue(value: unknown, depth = 0, seen: WeakSet = new WeakSet()): unknown { + if (value === null || value === undefined) return value; + + if (typeof value === 'string') return maskString(value); + + if (typeof value === 'number' || typeof value === 'boolean' || typeof value === 'bigint') { + return value; + } + + if (typeof value === 'function' || typeof value === 'symbol') return undefined; + + if (value instanceof Date) return value; + if (value instanceof RegExp) return value.toString(); + if (Buffer.isBuffer(value)) return `[Buffer ${value.length} bytes]`; + + if (value instanceof Error) { + return { + name: value.name, + message: maskString(value.message), + stack: value.stack ? maskString(value.stack) : undefined, + }; + } + + if (depth >= MAX_DEPTH) return '[MaxDepth]'; + + if (typeof value === 'object') { + if (seen.has(value)) return '[Circular]'; + seen.add(value); + + try { + if (Array.isArray(value)) { + const items = value + .slice(0, MAX_ARRAY_ITEMS) + .map((item) => maskValue(item, depth + 1, seen)); + + if (value.length > MAX_ARRAY_ITEMS) { + items.push(`[+${value.length - MAX_ARRAY_ITEMS} more]`); + } + return items; + } + + if (value instanceof Map) { + return maskValue(Object.fromEntries(value), depth + 1, seen); + } + + if (value instanceof Set) { + return maskValue(Array.from(value), depth + 1, seen); + } + + // Mongoose documents and other class instances expose their data through + // toJSON(); using it avoids walking internal driver state. + const source = + typeof (value as { toJSON?: unknown }).toJSON === 'function' + ? (value as { toJSON: () => unknown }).toJSON() + : value; + + if (source !== value) return maskValue(source, depth + 1, seen); + + const result: Record = {}; + for (const [key, item] of Object.entries(value as Record)) { + result[key] = isSensitiveKey(key) ? REDACTED : maskValue(item, depth + 1, seen); + } + return result; + } finally { + seen.delete(value); + } + } + + return value; +} diff --git a/src/utils/rpcRetry.ts b/src/utils/rpcRetry.ts index 971a73b..237dbb8 100644 --- a/src/utils/rpcRetry.ts +++ b/src/utils/rpcRetry.ts @@ -1,5 +1,49 @@ import logger from '../config/logger'; +/** + * Exponential-backoff retry for outbound calls, with optional per-attempt + * timeouts. + * + * Used to wrap Soroban RPC calls, which fail transiently under rate limiting + * (HTTP 429), brief node outages, and network hiccups. + * + * ── Why jitter ─────────────────────────────────────────────────────────────── + * Plain exponential backoff synchronises retries: when a node blips, every + * in-flight request backs off by the same amount and they all return together, + * re-creating the spike that caused the failure. Jitter spreads them out. + * + * ── Why a per-attempt timeout ──────────────────────────────────────────────── + * A hung TCP connection does not reject; it hangs until the OS gives up, which + * can take minutes. Racing each attempt against a timer turns that hang into a + * prompt, retryable error and bounds total latency to roughly + * `maxAttempts * timeoutMs` plus the backoff delays. + */ + +/** Reason an attempt failed, used for logging and the retry decision. */ +export type AttemptFailureKind = 'timeout' | 'error'; + +/** Context passed to `onAttemptFailed` after each failed attempt. */ +export interface RetryAttemptContext { + /** 1-based number of the attempt that just failed. */ + attempt: number; + /** Total attempts that will be made before giving up. */ + maxAttempts: number; + /** Whether the attempt timed out or rejected. */ + kind: AttemptFailureKind; + /** The error that caused the failure. */ + error: unknown; + /** Delay before the next attempt, in ms. `0` when no retry will follow. */ + delayMs: number; +} + +/** Context passed to `onRecovery` when a retried call eventually succeeds. */ +export interface RetryRecoveryContext { + /** The attempt number that succeeded (always > 1). */ + attempt: number; + /** Total wall-clock time across all attempts, in ms. */ + elapsedMs: number; +} + /** * Options controlling retry/backoff behaviour for `withRetry`. */ @@ -16,11 +60,29 @@ export interface RetryOptions { jitter?: number; /** Label used in log messages to identify the operation being retried. */ operationName?: string; - /** Predicate deciding whether a given error should trigger a retry. Defaults to retrying everything. */ - isRetryable?: (error: unknown) => boolean; + /** + * Predicate deciding whether a given error should trigger a retry. + * Defaults to retrying everything. Receives the failure kind as a second + * argument so callers can treat timeouts differently from rejections. + */ + isRetryable?: (error: unknown, kind: AttemptFailureKind) => boolean; + /** + * Per-attempt timeout in milliseconds. Omit or pass `0` to disable, in which + * case an attempt waits as long as the underlying call takes. + */ + timeoutMs?: number; + /** Called after every failed attempt, including the last. */ + onAttemptFailed?: (context: RetryAttemptContext) => void; + /** Called once if the call succeeds after at least one failure. */ + onRecovery?: (context: RetryRecoveryContext) => void; } -const DEFAULT_OPTIONS: Required> = { +const DEFAULT_OPTIONS: Required< + Omit< + RetryOptions, + 'operationName' | 'isRetryable' | 'timeoutMs' | 'onAttemptFailed' | 'onRecovery' + > +> = { maxAttempts: 5, baseDelayMs: 250, maxDelayMs: 8000, @@ -28,19 +90,74 @@ const DEFAULT_OPTIONS: Required { +/** + * Error thrown when a single attempt exceeds its timeout budget. + * + * Distinct from a generic `Error` so callers and retry predicates can tell + * "the node never answered" apart from "the node answered with a rejection". + */ +export class OperationTimeoutError extends Error { + public readonly operation: string; + public readonly timeoutMs: number; + + constructor(operation: string, timeoutMs: number) { + super(`Operation '${operation}' timed out after ${timeoutMs}ms`); + this.name = 'OperationTimeoutError'; + this.operation = operation; + this.timeoutMs = timeoutMs; + Object.setPrototypeOf(this, OperationTimeoutError.prototype); + } +} + +/** Promise-based sleep. */ +export function sleep(ms: number): Promise { return new Promise((resolve) => setTimeout(resolve, ms)); } +/** + * Race a promise against a timer. + * + * The timer is always cleared, including on the success path, so a pending + * `setTimeout` cannot keep the Node event loop alive after the work is done. + * + * @param factory Produces the promise to race. + * @param timeoutMs Timeout in ms; `0` or negative disables the race. + * @param operation Label used in the timeout message. + */ +export async function withTimeout( + factory: () => Promise, + timeoutMs: number, + operation: string, +): Promise { + if (!timeoutMs || timeoutMs <= 0) return factory(); + + let timer: NodeJS.Timeout | undefined; + + const timeout = new Promise((_resolve, reject) => { + timer = setTimeout(() => reject(new OperationTimeoutError(operation, timeoutMs)), timeoutMs); + }); + + try { + return await Promise.race([factory(), timeout]); + } finally { + if (timer) clearTimeout(timer); + } +} + /** * Compute the delay for a given retry attempt using exponential backoff with - * full jitter, capped at `maxDelayMs`. + * jitter, capped at `maxDelayMs`. * * @param attempt Zero-based retry attempt number (0 = first retry). */ export function computeBackoffDelay( attempt: number, - options: Required>, + options: Required< + Omit< + RetryOptions, + 'operationName' | 'isRetryable' | 'timeoutMs' | 'onAttemptFailed' | 'onRecovery' + > + >, ): number { const exponential = options.baseDelayMs * Math.pow(options.factor, attempt); const capped = Math.min(exponential, options.maxDelayMs); @@ -52,34 +169,61 @@ export function computeBackoffDelay( /** * Execute `fn`, retrying with exponential backoff on failure. * - * Intended for wrapping Soroban RPC calls that may fail transiently due to - * rate limiting (HTTP 429) or temporary node outages. Every failed attempt - * is logged; once all attempts are exhausted the last error is rethrown so - * callers can handle it as they would an unwrapped RPC failure. + * Every failed attempt is logged; once all attempts are exhausted the last + * error is rethrown unchanged, so callers can handle it as they would an + * unwrapped RPC failure. * - * @param fn The async operation to execute. + * @param fn The async operation to execute. Must be a factory rather than + * a promise: a promise can only be awaited, not re-run. * @param options Retry/backoff configuration. + * + * @example + * const account = await withRetry(() => rpc.getAccount(addr), { + * maxAttempts: 3, + * timeoutMs: 10_000, + * operationName: 'getAccount', + * }); */ export async function withRetry(fn: () => Promise, options: RetryOptions = {}): Promise { const resolved = { ...DEFAULT_OPTIONS, ...options }; const operationName = options.operationName ?? 'rpc-call'; const isRetryable = options.isRetryable ?? ((): boolean => true); + const timeoutMs = options.timeoutMs ?? 0; + const startedAt = Date.now(); let lastError: unknown; for (let attempt = 0; attempt < resolved.maxAttempts; attempt += 1) { try { - return await fn(); + const result = await withTimeout(fn, timeoutMs, operationName); + + if (attempt > 0) { + const elapsedMs = Date.now() - startedAt; + logger.info( + `[RPC Retry] ${operationName} recovered on attempt ${attempt + 1} after ${elapsedMs}ms`, + ); + options.onRecovery?.({ attempt: attempt + 1, elapsedMs }); + } + return result; } catch (err) { lastError = err; + const attemptNumber = attempt + 1; + const kind: AttemptFailureKind = err instanceof OperationTimeoutError ? 'timeout' : 'error'; const isLastAttempt = attemptNumber >= resolved.maxAttempts; const message = err instanceof Error ? err.message : String(err); - if (!isRetryable(err) || isLastAttempt) { + if (!isRetryable(err, kind) || isLastAttempt) { logger.error( `[RPC Retry] ${operationName} failed permanently after ${attemptNumber} attempt(s) — error="${message}"`, ); + options.onAttemptFailed?.({ + attempt: attemptNumber, + maxAttempts: resolved.maxAttempts, + kind, + error: err, + delayMs: 0, + }); throw err; } @@ -87,9 +231,17 @@ export async function withRetry(fn: () => Promise, options: RetryOptions = logger.warn( `[RPC Retry] ${operationName} attempt ${attemptNumber}/${resolved.maxAttempts} failed ` + - `— error="${message}" — retrying in ${delayMs}ms`, + `(${kind}) — error="${message}" — retrying in ${delayMs}ms`, ); + options.onAttemptFailed?.({ + attempt: attemptNumber, + maxAttempts: resolved.maxAttempts, + kind, + error: err, + delayMs, + }); + await sleep(delayMs); } } diff --git a/tests/driverLocation.test.ts b/tests/driverLocation.test.ts new file mode 100644 index 0000000..aa6e5b2 --- /dev/null +++ b/tests/driverLocation.test.ts @@ -0,0 +1,359 @@ +/** + * Tests for the DriverLocation model and the proximity search service. + * + * Runs against mongodb-memory-server so the 2dsphere indexes, the GeoJSON + * validation and the `$geoNear` pipeline are exercised for real rather than + * mocked. All assertions read data back out of the database. + */ + +import mongoose, { Types } from 'mongoose'; +import { MongoMemoryServer } from 'mongodb-memory-server'; +import { DriverLocation } from '../src/models/DriverLocation'; +import { DriverLocationService } from '../src/services/driverLocationService'; +import AppError from '../src/utils/AppError'; + +jest.mock('../src/config/logger', () => ({ + __esModule: true, + default: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + http: jest.fn(), + }, +})); + +// Reference points in Lagos, with roughly known separations. +const LAGOS = { lat: 6.5244, lng: 3.3792 }; +/** ~1.1 km north of LAGOS. */ +const NEARBY = { lat: 6.5344, lng: 3.3792 }; +/** ~11 km north of LAGOS. */ +const FAR = { lat: 6.6244, lng: 3.3792 }; + +describe('DriverLocation', () => { + let mongod: MongoMemoryServer; + const service = new DriverLocationService(); + + beforeAll(async () => { + mongod = await MongoMemoryServer.create(); + await mongoose.connect(mongod.getUri()); + // Indexes must exist before any $geoNear query runs. + await DriverLocation.createIndexes(); + }, 120_000); + + afterAll(async () => { + await mongoose.disconnect(); + await mongod.stop(); + }, 30_000); + + afterEach(async () => { + await DriverLocation.deleteMany({}); + }); + + /** Insert a driver at a coordinate with the given availability. */ + const seedDriver = async ( + point: { lat: number; lng: number }, + overrides: Partial<{ isAvailable: boolean; status: string }> = {}, + ): Promise => { + const driverId = new Types.ObjectId(); + + await DriverLocation.create({ + driverId, + location: { type: 'Point', coordinates: [point.lng, point.lat] }, + isAvailable: overrides.isAvailable ?? true, + status: overrides.status ?? 'online', + recordedAt: new Date(), + }); + + return driverId; + }; + + // ── Schema ──────────────────────────────────────────────────────────────── + + describe('schema', () => { + it('stores coordinates in GeoJSON [lng, lat] order', async () => { + const driverId = await seedDriver(LAGOS); + const found = await DriverLocation.findOne({ driverId }); + + expect(found?.location.coordinates).toEqual([LAGOS.lng, LAGOS.lat]); + }); + + it('exposes the position in API {lat, lng} order via toLatLng', async () => { + const driverId = await seedDriver(LAGOS); + const found = await DriverLocation.findOne({ driverId }); + + expect(found?.toLatLng()).toEqual({ lat: LAGOS.lat, lng: LAGOS.lng }); + }); + + it('rejects an out-of-range latitude', async () => { + await expect( + DriverLocation.create({ + driverId: new Types.ObjectId(), + location: { type: 'Point', coordinates: [3.37, 91] }, + recordedAt: new Date(), + }), + ).rejects.toThrow(); + }); + + it('rejects an out-of-range longitude', async () => { + await expect( + DriverLocation.create({ + driverId: new Types.ObjectId(), + location: { type: 'Point', coordinates: [181, 6.5] }, + recordedAt: new Date(), + }), + ).rejects.toThrow(); + }); + + it('allows only one location document per driver', async () => { + const driverId = new Types.ObjectId(); + const doc = { + driverId, + location: { type: 'Point' as const, coordinates: [3.37, 6.5] as [number, number] }, + recordedAt: new Date(), + }; + + await DriverLocation.create(doc); + await expect(DriverLocation.create(doc)).rejects.toThrow(); + }); + + it('derives expiresAt from recordedAt so stale records are retired', async () => { + const driverId = await seedDriver(LAGOS); + const found = await DriverLocation.findOne({ driverId }); + + expect(found?.expiresAt.getTime()).toBeGreaterThan(found!.recordedAt.getTime()); + }); + }); + + // ── Indexes ─────────────────────────────────────────────────────────────── + + describe('indexes', () => { + it('creates both the plain and the compound 2dsphere indexes', async () => { + const names = (await DriverLocation.collection.indexes()).map((index) => index.name); + + expect(names).toContain('location_2dsphere'); + expect(names).toContain('availability_location_2dsphere'); + expect(names).toContain('driver_location_ttl'); + }); + + it('orders the compound index with equality fields before the geometry', async () => { + const indexes = await DriverLocation.collection.indexes(); + const compound = indexes.find((i) => i.name === 'availability_location_2dsphere'); + + // Key order decides whether the planner can seek before walking geometry. + expect(Object.keys(compound!.key)).toEqual(['isAvailable', 'status', 'location']); + }); + + it('ensureIndexes reports the indexes present on the collection', async () => { + const names = await service.ensureIndexes(); + expect(names).toContain('availability_location_2dsphere'); + }); + }); + + // ── Proximity search ────────────────────────────────────────────────────── + + describe('findNearbyDrivers', () => { + it('returns drivers inside the radius and excludes those outside it', async () => { + const near = await seedDriver(NEARBY); + await seedDriver(FAR); + + const result = await service.findNearbyDrivers({ ...LAGOS, radiusMeters: 5000 }); + + expect(result.count).toBe(1); + expect(result.drivers[0].driverId).toBe(near.toString()); + }); + + it('orders results nearest first', async () => { + const near = await seedDriver(NEARBY); + const far = await seedDriver(FAR); + + const result = await service.findNearbyDrivers({ ...LAGOS, radiusMeters: 20000 }); + + expect(result.drivers.map((d) => d.driverId)).toEqual([near.toString(), far.toString()]); + expect(result.drivers[0].distanceMeters).toBeLessThan(result.drivers[1].distanceMeters); + }); + + it('returns a distance computed by the database, not a placeholder', async () => { + await seedDriver(NEARBY); + + const result = await service.findNearbyDrivers({ ...LAGOS, radiusMeters: 5000 }); + + // NEARBY is ~1.1 km away; allow a generous band around that. + expect(result.drivers[0].distanceMeters).toBeGreaterThan(800); + expect(result.drivers[0].distanceMeters).toBeLessThan(1500); + }); + + it('excludes unavailable drivers by default', async () => { + await seedDriver(NEARBY, { isAvailable: false }); + + const result = await service.findNearbyDrivers({ ...LAGOS, radiusMeters: 5000 }); + expect(result.count).toBe(0); + }); + + it('includes unavailable drivers when availableOnly is false', async () => { + await seedDriver(NEARBY, { isAvailable: false }); + + const result = await service.findNearbyDrivers({ + ...LAGOS, + radiusMeters: 5000, + availableOnly: false, + }); + + expect(result.count).toBe(1); + }); + + it('filters by availability status', async () => { + await seedDriver(NEARBY, { status: 'on_delivery' }); + const online = await seedDriver(NEARBY, { status: 'online' }); + + const result = await service.findNearbyDrivers({ + ...LAGOS, + radiusMeters: 5000, + status: 'online', + }); + + expect(result.count).toBe(1); + expect(result.drivers[0].driverId).toBe(online.toString()); + }); + + it('honours the result limit', async () => { + await seedDriver(NEARBY); + await seedDriver(NEARBY); + await seedDriver(NEARBY); + + const result = await service.findNearbyDrivers({ ...LAGOS, radiusMeters: 5000, limit: 2 }); + expect(result.drivers).toHaveLength(2); + }); + + it('echoes back the centre and the radius actually applied', async () => { + const result = await service.findNearbyDrivers({ ...LAGOS, radiusMeters: 3000 }); + + expect(result.center).toEqual(LAGOS); + expect(result.radiusMeters).toBe(3000); + }); + + it('returns an empty result rather than throwing when nothing matches', async () => { + const result = await service.findNearbyDrivers({ ...LAGOS, radiusMeters: 1000 }); + + expect(result.count).toBe(0); + expect(result.drivers).toEqual([]); + }); + + it.each([ + ['lat', { lat: 91, lng: 3.37 }], + ['lat', { lat: -91, lng: 3.37 }], + ['lng', { lat: 6.5, lng: 181 }], + ['lng', { lat: 6.5, lng: -181 }], + ])('rejects an out-of-range %s', async (_field, coords) => { + await expect(service.findNearbyDrivers(coords)).rejects.toBeInstanceOf(AppError); + }); + + it('rejects a non-positive radius', async () => { + await expect( + service.findNearbyDrivers({ ...LAGOS, radiusMeters: 0 }), + ).rejects.toBeInstanceOf(AppError); + }); + + it('rejects a radius beyond the configured maximum', async () => { + await expect( + service.findNearbyDrivers({ ...LAGOS, radiusMeters: 10_000_000 }), + ).rejects.toBeInstanceOf(AppError); + }); + }); + + // ── Writes ──────────────────────────────────────────────────────────────── + + describe('upsertDriverLocation', () => { + it('creates a record on the first report', async () => { + const driverId = new Types.ObjectId().toString(); + + await service.upsertDriverLocation({ driverId, ...LAGOS, isAvailable: true }); + + expect(await DriverLocation.countDocuments({ driverId })).toBe(1); + }); + + it('updates in place rather than inserting a second row', async () => { + const driverId = new Types.ObjectId().toString(); + + await service.upsertDriverLocation({ driverId, ...LAGOS }); + await service.upsertDriverLocation({ driverId, ...NEARBY }); + + expect(await DriverLocation.countDocuments({ driverId })).toBe(1); + + const stored = await DriverLocation.findOne({ driverId }); + expect(stored?.location.coordinates).toEqual([NEARBY.lng, NEARBY.lat]); + }); + + it('persists the optional telemetry fields', async () => { + const driverId = new Types.ObjectId().toString(); + + await service.upsertDriverLocation({ + driverId, + ...LAGOS, + heading: 90, + speed: 12.5, + accuracy: 5, + status: 'on_delivery', + }); + + const stored = await DriverLocation.findOne({ driverId }); + expect(stored?.heading).toBe(90); + expect(stored?.speed).toBe(12.5); + expect(stored?.status).toBe('on_delivery'); + }); + + it('rejects a malformed driver id', async () => { + await expect( + service.upsertDriverLocation({ driverId: 'not-an-id', ...LAGOS }), + ).rejects.toBeInstanceOf(AppError); + }); + + it('makes a newly reported driver findable by proximity search', async () => { + const driverId = new Types.ObjectId().toString(); + + await service.upsertDriverLocation({ + driverId, + ...NEARBY, + isAvailable: true, + status: 'online', + }); + + const result = await service.findNearbyDrivers({ ...LAGOS, radiusMeters: 5000 }); + expect(result.drivers[0].driverId).toBe(driverId); + }); + }); + + describe('getDriverLocation', () => { + it('returns the stored position', async () => { + const driverId = await seedDriver(LAGOS); + const found = await service.getDriverLocation(driverId.toString()); + + expect(found.toLatLng()).toEqual(LAGOS); + }); + + it('throws 404 when the driver has no position on record', async () => { + await expect( + service.getDriverLocation(new Types.ObjectId().toString()), + ).rejects.toMatchObject({ statusCode: 404 }); + }); + + it('throws 400 for a malformed id', async () => { + await expect(service.getDriverLocation('nope')).rejects.toMatchObject({ + statusCode: 400, + }); + }); + }); + + // ── Profiling ───────────────────────────────────────────────────────────── + + describe('explainProximityQuery', () => { + it('reports that the query is served by a geospatial index', async () => { + await seedDriver(NEARBY); + + const plan = await service.explainProximityQuery({ ...LAGOS, radiusMeters: 5000 }); + + // The planner must choose one of the 2dsphere indexes, never a COLLSCAN. + expect(plan.indexUsed).toMatch(/2dsphere/); + }); + }); +}); diff --git a/tests/piiMasker.test.ts b/tests/piiMasker.test.ts new file mode 100644 index 0000000..1e54250 --- /dev/null +++ b/tests/piiMasker.test.ts @@ -0,0 +1,216 @@ +/** + * Unit tests for the PII masking helpers used by the logging interface. + */ + +import { maskString, maskValue, isSensitiveKey, REDACTED } from '../src/utils/piiMasker'; + +describe('maskString', () => { + it('masks an email but keeps the domain for correlation', () => { + expect(maskString('login failed for alice@example.com')).toBe( + 'login failed for al***@example.com', + ); + }); + + it('masks every email in a string', () => { + const masked = maskString('from a@x.com to bob@y.org'); + expect(masked).not.toContain('a@x.com'); + expect(masked).not.toContain('bob@y.org'); + expect(masked).toContain('@x.com'); + expect(masked).toContain('@y.org'); + }); + + it('redacts a JWT entirely', () => { + const jwt = + 'eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJzdWIiOiIxMjMifQ.dBjftJeZ4CVPmB92K27uhbUJU1p1r_wW1gFWFOEjXk'; + const masked = maskString(`token=${jwt}`); + + expect(masked).toBe(`token=${REDACTED}`); + expect(masked).not.toContain('eyJ'); + }); + + it('redacts the credential in a bearer header but keeps the scheme', () => { + const masked = maskString('Authorization: Bearer abcdef1234567890'); + expect(masked).toBe(`Authorization: Bearer ${REDACTED}`); + }); + + it('redacts a Stellar secret seed in full', () => { + const secret = `S${'A'.repeat(55)}`; + expect(maskString(`seed ${secret}`)).toBe(`seed ${REDACTED}`); + }); + + it('partially masks a Stellar public key, which is not a secret', () => { + const publicKey = `G${'A'.repeat(55)}`; + const masked = maskString(`payer ${publicKey}`); + + expect(masked).not.toBe(`payer ${publicKey}`); + expect(masked).toContain('GAAA'); + expect(masked).toContain('***'); + }); + + it('strips the password out of a connection string', () => { + const masked = maskString('mongodb://appuser:sup3rs3cret@cluster0.mongodb.net/db'); + + expect(masked).not.toContain('sup3rs3cret'); + expect(masked).toContain('appuser'); + expect(masked).toContain('cluster0.mongodb.net'); + }); + + it('keeps only the last four digits of a valid card number', () => { + // Luhn-valid test number. + const masked = maskString('card 4242424242424242 charged'); + expect(masked).toBe('card ****-****-****-4242 charged'); + }); + + it('leaves long non-card numbers such as ledger sequences intact', () => { + // Not Luhn-valid, so it must not be treated as a card. + expect(maskString('ledger 1234567890123456')).toBe('ledger 1234567890123456'); + }); + + it('masks the middle of a phone number', () => { + const masked = maskString('call +14155552671 now'); + + expect(masked).not.toContain('+14155552671'); + expect(masked).toContain('+14'); + expect(masked).toContain('71'); + }); + + it('redacts a PEM private key block', () => { + const pem = + '-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKCAQEA\n-----END RSA PRIVATE KEY-----'; + expect(maskString(`key: ${pem}`)).toBe(`key: ${REDACTED}`); + }); + + it('returns empty and non-PII strings unchanged', () => { + expect(maskString('')).toBe(''); + expect(maskString('delivery created successfully')).toBe('delivery created successfully'); + }); +}); + +describe('isSensitiveKey', () => { + it.each([ + 'password', + 'Password', + 'passphrase', + 'token', + 'accessToken', + 'refreshToken', + 'apiKey', + 'api_key', + 'privateKey', + 'authorization', + 'cookie', + 'secret', + 'clientSecret', + 'mnemonic', + 'cvv', + 'ssn', + 'signedXdr', + ])('treats %s as sensitive', (key) => { + expect(isSensitiveKey(key)).toBe(true); + }); + + it.each(['tokenType', 'authProvider', 'hasToken', 'tokenExpiresAt'])( + 'allows the non-secret key %s through', + (key) => { + expect(isSensitiveKey(key)).toBe(false); + }, + ); + + it.each(['deliveryId', 'status', 'amount', 'createdAt'])( + 'treats ordinary key %s as safe', + (key) => { + expect(isSensitiveKey(key)).toBe(false); + }, + ); +}); + +describe('maskValue', () => { + it('redacts sensitive keys wholesale regardless of their value', () => { + const masked = maskValue({ email: 'bob@example.com', password: 'hunter2' }) as Record< + string, + unknown + >; + + expect(masked.password).toBe(REDACTED); + expect(masked.email).toBe('bo***@example.com'); + }); + + it('walks nested objects', () => { + const masked = maskValue({ + user: { profile: { email: 'deep@example.com', apiKey: 'k-123' } }, + }) as Record>>; + + expect(masked.user.profile.email).toBe('de***@example.com'); + expect(masked.user.profile.apiKey).toBe(REDACTED); + }); + + it('walks arrays of objects', () => { + const masked = maskValue([{ email: 'a@x.com' }, { token: 't' }]) as Array< + Record + >; + + expect(masked[0].email).toBe('a***@x.com'); + expect(masked[1].token).toBe(REDACTED); + }); + + it('does not mutate the input', () => { + const input = { password: 'hunter2', email: 'a@x.com' }; + maskValue(input); + + expect(input.password).toBe('hunter2'); + expect(input.email).toBe('a@x.com'); + }); + + it('handles circular references without overflowing', () => { + const node: Record = { name: 'root' }; + node.self = node; + + const masked = maskValue(node) as Record; + + expect(masked.name).toBe('root'); + expect(masked.self).toBe('[Circular]'); + }); + + it('preserves the stack of an Error while masking its message', () => { + const error = new Error('failed for alice@example.com'); + const masked = maskValue(error) as { name: string; message: string; stack?: string }; + + expect(masked.name).toBe('Error'); + expect(masked.message).toBe('failed for al***@example.com'); + expect(masked.stack).toBeDefined(); + }); + + it('passes primitives through untouched', () => { + expect(maskValue(42)).toBe(42); + expect(maskValue(true)).toBe(true); + expect(maskValue(null)).toBeNull(); + expect(maskValue(undefined)).toBeUndefined(); + }); + + it('summarises buffers rather than dumping their contents', () => { + expect(maskValue(Buffer.from('secret'))).toBe('[Buffer 6 bytes]'); + }); + + it('stops descending past the depth limit', () => { + // Build a chain deeper than MAX_DEPTH (8). + let deep: Record = { value: 'bottom' }; + for (let i = 0; i < 12; i++) deep = { nested: deep }; + + expect(JSON.stringify(maskValue(deep))).toContain('[MaxDepth]'); + }); + + it('truncates very large arrays', () => { + const masked = maskValue(new Array(150).fill('x')) as unknown[]; + + expect(masked.length).toBe(101); + expect(masked[100]).toBe('[+50 more]'); + }); + + it('uses toJSON when an object provides one, as Mongoose documents do', () => { + const doc = { toJSON: (): unknown => ({ email: 'doc@example.com', token: 'abc' }) }; + const masked = maskValue(doc) as Record; + + expect(masked.email).toBe('do***@example.com'); + expect(masked.token).toBe(REDACTED); + }); +}); diff --git a/tests/rpcTimeout.test.ts b/tests/rpcTimeout.test.ts new file mode 100644 index 0000000..2e171ff --- /dev/null +++ b/tests/rpcTimeout.test.ts @@ -0,0 +1,205 @@ +/** + * Unit tests for the timeout and observability additions to the RPC retry + * helper, and for the transient-error classifier used by StellarService. + */ + +import { + withRetry, + withTimeout, + OperationTimeoutError, + type AttemptFailureKind, +} from '../src/utils/rpcRetry'; +import { isTransientRpcError } from '../src/services/stellarService'; + +jest.mock('../src/config/logger', () => ({ + __esModule: true, + default: { + info: jest.fn(), + warn: jest.fn(), + error: jest.fn(), + debug: jest.fn(), + http: jest.fn(), + }, +})); + +/** Resolve after `ms`, used to simulate a slow call. */ +const slow = (value: T, ms: number): Promise => + new Promise((resolve) => setTimeout(() => resolve(value), ms)); + +describe('withTimeout', () => { + it('resolves when the operation finishes inside the budget', async () => { + await expect(withTimeout(() => slow('ok', 5), 200, 'fast')).resolves.toBe('ok'); + }); + + it('rejects with OperationTimeoutError when the budget is exceeded', async () => { + await expect(withTimeout(() => slow('late', 200), 20, 'slow')).rejects.toBeInstanceOf( + OperationTimeoutError, + ); + }); + + it('names the operation and the budget in the timeout message', async () => { + await expect(withTimeout(() => slow('late', 200), 20, 'getAccount')).rejects.toThrow( + /getAccount.*20ms/, + ); + }); + + it('does not apply a timeout when the budget is zero', async () => { + await expect(withTimeout(() => slow('ok', 30), 0, 'untimed')).resolves.toBe('ok'); + }); + + it('propagates the underlying rejection rather than a timeout', async () => { + await expect( + withTimeout(() => Promise.reject(new Error('boom')), 500, 'op'), + ).rejects.toThrow('boom'); + }); +}); + +describe('withRetry timeout handling', () => { + it('retries a timed-out attempt and succeeds once the call is fast enough', async () => { + let call = 0; + const fn = jest.fn(() => { + call += 1; + return call === 1 ? slow('late', 200) : slow('ok', 1); + }); + + const result = await withRetry(fn, { + maxAttempts: 3, + baseDelayMs: 1, + maxDelayMs: 2, + jitter: 0, + timeoutMs: 30, + operationName: 'flaky', + }); + + expect(result).toBe('ok'); + expect(fn).toHaveBeenCalledTimes(2); + }); + + it('reports the failure kind as timeout to the retry predicate', async () => { + const kinds: AttemptFailureKind[] = []; + + await expect( + withRetry(() => slow('late', 200), { + maxAttempts: 2, + baseDelayMs: 1, + maxDelayMs: 2, + jitter: 0, + timeoutMs: 20, + operationName: 'always-slow', + isRetryable: (_error, kind) => { + kinds.push(kind); + return true; + }, + }), + ).rejects.toBeInstanceOf(OperationTimeoutError); + + expect(kinds).toEqual(['timeout', 'timeout']); + }); + + it('invokes onAttemptFailed for each failure, with no delay on the last', async () => { + const contexts: Array<{ attempt: number; delayMs: number }> = []; + + await expect( + withRetry(() => Promise.reject(new Error('nope')), { + maxAttempts: 3, + baseDelayMs: 1, + maxDelayMs: 2, + jitter: 0, + operationName: 'failing', + onAttemptFailed: ({ attempt, delayMs }) => contexts.push({ attempt, delayMs }), + }), + ).rejects.toThrow('nope'); + + expect(contexts.map((c) => c.attempt)).toEqual([1, 2, 3]); + expect(contexts[2].delayMs).toBe(0); + }); + + it('invokes onRecovery only when a retry eventually succeeds', async () => { + const onRecovery = jest.fn(); + + await withRetry( + jest.fn().mockRejectedValueOnce(new Error('blip')).mockResolvedValue('ok'), + { + maxAttempts: 3, + baseDelayMs: 1, + maxDelayMs: 2, + jitter: 0, + operationName: 'recovers', + onRecovery, + }, + ); + + expect(onRecovery).toHaveBeenCalledTimes(1); + expect(onRecovery.mock.calls[0][0].attempt).toBe(2); + }); + + it('does not invoke onRecovery when the first attempt succeeds', async () => { + const onRecovery = jest.fn(); + + await withRetry(() => Promise.resolve('ok'), { + maxAttempts: 3, + baseDelayMs: 1, + operationName: 'clean', + onRecovery, + }); + + expect(onRecovery).not.toHaveBeenCalled(); + }); + + it('stops immediately when the predicate rejects the error as permanent', async () => { + const fn = jest.fn().mockRejectedValue(new Error('bad request')); + + await expect( + withRetry(fn, { + maxAttempts: 5, + baseDelayMs: 1, + operationName: 'permanent', + isRetryable: () => false, + }), + ).rejects.toThrow('bad request'); + + expect(fn).toHaveBeenCalledTimes(1); + }); +}); + +describe('isTransientRpcError', () => { + it('treats a timeout as transient', () => { + expect(isTransientRpcError(new OperationTimeoutError('op', 10))).toBe(true); + expect(isTransientRpcError(new Error('anything'), 'timeout')).toBe(true); + }); + + it.each([408, 425, 429, 500, 502, 503, 504])('retries HTTP %i', (status) => { + expect(isTransientRpcError({ status })).toBe(true); + }); + + it.each([400, 401, 403, 404, 422])('does not retry HTTP %i', (status) => { + expect(isTransientRpcError({ status })).toBe(false); + }); + + it.each(['ECONNRESET', 'ECONNREFUSED', 'ETIMEDOUT', 'ENOTFOUND', 'EAI_AGAIN'])( + 'retries socket error %s', + (code) => { + expect(isTransientRpcError({ code })).toBe(true); + }, + ); + + it('reads the status from a nested response object', () => { + expect(isTransientRpcError({ response: { status: 503 } })).toBe(true); + expect(isTransientRpcError({ response: { status: 400 } })).toBe(false); + }); + + it('never retries tx_bad_seq, which has its own rebuild path', () => { + expect(isTransientRpcError(new Error('transaction failed: tx_bad_seq'))).toBe(false); + expect(isTransientRpcError(new Error('txBadSeq'))).toBe(false); + }); + + it('recognises transient failures described only in the message', () => { + expect(isTransientRpcError(new Error('socket hang up'))).toBe(true); + expect(isTransientRpcError(new Error('Request timed out'))).toBe(true); + expect(isTransientRpcError(new Error('503 Service Unavailable'))).toBe(true); + }); + + it('does not retry a deterministic error', () => { + expect(isTransientRpcError(new Error('invalid contract id'))).toBe(false); + }); +});