diff --git a/health.ts b/health.ts new file mode 100644 index 0000000..6d72999 --- /dev/null +++ b/health.ts @@ -0,0 +1,464 @@ +import { readFileSync } from 'node:fs'; +import { dirname, join } from 'node:path'; +import { fileURLToPath } from 'node:url'; +import type { AggregatedHealthResponse, DependencyHealth, HealthResponse, HealthStatus } from './schemas.js'; + +export const HEALTH_CHECK_TIMEOUT_MS = 3_000; +export const UPSTREAM_HEALTH_TIMEOUT_MS = 5_000; + +export function readServiceVersion(importMetaUrl: string): string { + const serviceDir = dirname(fileURLToPath(importMetaUrl)); + const pkgPath = join(serviceDir, '../package.json'); + try { + const pkg = JSON.parse(readFileSync(pkgPath, 'utf8')) as { version?: string }; + return pkg.version ?? '0.0.0'; + } catch { + return '0.0.0'; + } +} + +export function computeOverallStatus( + dependencies: DependencyHealth[], + options?: { criticalNames?: string[] }, +): HealthStatus { + if (dependencies.length === 0) { + return 'healthy'; + } + + const critical = new Set(options?.criticalNames ?? dependencies.map((dep) => dep.name)); + const criticalDeps = dependencies.filter((dep) => critical.has(dep.name)); + const hasCriticalFailure = criticalDeps.some((dep) => dep.status === 'disconnected'); + const hasAnyFailure = dependencies.some((dep) => dep.status === 'disconnected'); + + if (hasCriticalFailure) return 'unhealthy'; + if (hasAnyFailure) return 'degraded'; + return 'healthy'; +} + +export async function withLatency( + fn: () => Promise, + timeoutMs: number, +): Promise<{ ok: true; latencyMs: number; value: T } | { ok: false; latencyMs: number; error: unknown }> { + const start = Date.now(); + let timeoutId: ReturnType | undefined; + + try { + const value = await Promise.race([ + fn(), + new Promise((_, reject) => { + timeoutId = setTimeout(() => reject(new Error('Health check timed out')), timeoutMs); + }), + ]); + return { ok: true, latencyMs: Date.now() - start, value }; + } catch (error) { + return { ok: false, latencyMs: Date.now() - start, error }; + } finally { + if (timeoutId) clearTimeout(timeoutId); + } +} + +export async function checkPostgresql( + queryFn: () => Promise, + timeoutMs = HEALTH_CHECK_TIMEOUT_MS, +): Promise { + const result = await withLatency(queryFn, timeoutMs); + if (result.ok) { + return { name: 'postgresql', status: 'connected', latencyMs: result.latencyMs }; + } + return { name: 'postgresql', status: 'disconnected', latencyMs: result.latencyMs }; +} + +export async function checkRedisPing( + pingFn: () => Promise, + timeoutMs = HEALTH_CHECK_TIMEOUT_MS, +): Promise { + const result = await withLatency(pingFn, timeoutMs); + if (result.ok && result.value === 'PONG') { + return { name: 'redis', status: 'connected', latencyMs: result.latencyMs }; + } + return { name: 'redis', status: 'disconnected', latencyMs: result.latencyMs }; +} + +export const BULLMQ_WAITING_DEGRADED_THRESHOLD = 1000; + +export async function checkBullMQ( + getJobCounts: () => Promise>, + queueName: string, + timeoutMs = HEALTH_CHECK_TIMEOUT_MS, + getIsPaused?: () => Promise, +): Promise { + const result = await withLatency( + async () => ({ + counts: await getJobCounts(), + isPaused: getIsPaused ? await getIsPaused() : false, + }), + timeoutMs, + ); + + if (!result.ok) { + return { name: queueName, status: 'disconnected', latencyMs: result.latencyMs }; + } + + const { counts, isPaused } = result.value; + const details = { + queueName, + isPaused, + waiting: counts.waiting ?? 0, + active: counts.active ?? 0, + failed: counts.failed ?? 0, + delayed: counts.delayed ?? 0, + }; + + // A paused queue processes no jobs, so treat it as down regardless of job counts. + if (isPaused) { + return { name: queueName, status: 'disconnected', latencyMs: result.latencyMs, details }; + } + + return { name: queueName, status: 'connected', latencyMs: result.latencyMs, details }; +} + +export async function checkHttpEndpoint( + url: string, + options: { + name: string; + timeoutMs?: number; + fetchImpl?: typeof fetch; + method?: 'GET' | 'HEAD'; + }, +): Promise { + const { name, timeoutMs = HEALTH_CHECK_TIMEOUT_MS, fetchImpl = fetch, method = 'GET' } = options; + const start = Date.now(); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + + try { + const response = await fetchImpl(url, { method, signal: controller.signal }); + const latencyMs = Date.now() - start; + if (!response.ok) { + return { name, status: 'disconnected', latencyMs, details: { httpStatus: response.status } }; + } + return { name, status: 'connected', latencyMs }; + } catch (error) { + return { + name, + status: 'disconnected', + latencyMs: Date.now() - start, + details: { error: error instanceof Error ? error.message : String(error) }, + }; + } finally { + clearTimeout(timer); + } +} + +export interface UpstreamHealthResult extends DependencyHealth { + body?: HealthResponse; +} + +export async function checkUpstreamServiceHealth( + baseUrl: string, + serviceName: string, + options: { + timeoutMs?: number; + fetchImpl?: typeof fetch; + } = {}, +): Promise { + const { timeoutMs = UPSTREAM_HEALTH_TIMEOUT_MS, fetchImpl = fetch } = options; + const url = `${baseUrl.replace(/\/+$/, '')}/api/health`; + const start = Date.now(); + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), timeoutMs); + + try { + const response = await fetchImpl(url, { signal: controller.signal }); + const latencyMs = Date.now() - start; + + if (!response.ok) { + return { + name: serviceName, + status: 'disconnected', + latencyMs, + details: { httpStatus: response.status }, + }; + } + + const body = (await response.json()) as HealthResponse; + return { + name: serviceName, + status: body.status === 'unhealthy' ? 'disconnected' : 'connected', + latencyMs, + details: { + status: body.status, + version: body.version, + service: body.service, + }, + body, + }; + } catch (error) { + return { + name: serviceName, + status: 'disconnected', + latencyMs: Date.now() - start, + details: { error: error instanceof Error ? error.message : String(error) }, + }; + } finally { + clearTimeout(timer); + } +} + +export function buildHealthResponse(params: { + service: string; + version: string; + startTime: number; + dependencies: DependencyHealth[]; + upstream?: DependencyHealth[]; + criticalDependencyNames?: string[]; +}): HealthResponse { + const combined = [...params.dependencies, ...(params.upstream ?? [])]; + const status = computeOverallStatus(combined, { + criticalNames: params.criticalDependencyNames ?? params.dependencies.map((dep) => dep.name), + }); + + return { + status, + service: params.service, + version: params.version, + uptime: Math.floor((Date.now() - params.startTime) / 1000), + lastDependencyCheck: new Date().toISOString(), + dependencies: params.dependencies, + ...(params.upstream?.length ? { upstream: params.upstream } : {}), + }; +} + +export interface ServiceHealthTarget { + name: string; + baseUrl: string; +} + +export interface AggregateHealthOptions { + gatewayHealth: HealthResponse; + targets: ServiceHealthTarget[]; + timeoutMs?: number; + fetchImpl?: typeof fetch; +} + +export async function aggregateAllHealth(options: AggregateHealthOptions): Promise { + const { gatewayHealth, targets, timeoutMs = UPSTREAM_HEALTH_TIMEOUT_MS, fetchImpl = fetch } = options; + const checkedAt = new Date().toISOString(); + + const downstreamResults = await Promise.allSettled( + targets.map(async (target) => { + const result = await checkUpstreamServiceHealth(target.baseUrl, target.name, { + timeoutMs, + fetchImpl, + }); + return { target, result }; + }), + ); + + const services: AggregatedHealthResponse['services'] = { + 'api-gateway': gatewayHealth, + }; + + const serviceStatuses: HealthStatus[] = [gatewayHealth.status]; + + for (let i = 0; i < downstreamResults.length; i++) { + const settled = downstreamResults[i]; + const target = targets[i]; + + if (settled.status === 'fulfilled') { + const { result } = settled.value; + if (result.body) { + services[target.name] = result.body; + serviceStatuses.push(result.body.status); + } else { + services[target.name] = { + status: 'unhealthy', + error: result.details?.error ? String(result.details.error) : 'Service unreachable', + }; + serviceStatuses.push('unhealthy'); + } + } else { + services[target.name] = { + status: 'unhealthy', + error: settled.reason instanceof Error ? settled.reason.message : 'Health check failed', + }; + serviceStatuses.push('unhealthy'); + } + } + + let status: HealthStatus = 'healthy'; + if (serviceStatuses.some((s) => s === 'unhealthy')) { + status = 'unhealthy'; + } else if (serviceStatuses.some((s) => s === 'degraded')) { + status = 'degraded'; + } + + return { + status, + service: 'api-gateway', + version: gatewayHealth.version, + uptime: gatewayHealth.uptime, + lastDependencyCheck: checkedAt, + dependencies: gatewayHealth.dependencies, + upstream: gatewayHealth.upstream, + services, + }; +} + +async function checkStellarRpc( + getLatestLedger: () => Promise<{ sequence: number }>, +): Promise { + const result = await withLatency(getLatestLedger, HEALTH_CHECK_TIMEOUT_MS); + if (result.ok) { + return { + name: 'stellar-rpc', + status: 'connected', + latencyMs: result.latencyMs, + details: { reachable: true, latestLedgerSequence: result.value.sequence }, + }; + } + return { + name: 'stellar-rpc', + status: 'disconnected', + latencyMs: result.latencyMs, + details: { reachable: false }, + }; +} + +export async function buildFxEngineHealthResponse(options: { + pingRedis: () => Promise; + ratesApiUrl: string; + startTime: number; + service: string; + version: string; + fetchImpl?: typeof fetch; +}): Promise { + const { pingRedis, ratesApiUrl, startTime, service, version, fetchImpl = fetch } = options; + + const [redisDep, ratesApi] = await Promise.all([ + checkRedisPing(pingRedis), + checkHttpEndpoint(ratesApiUrl, { + name: 'rates-api', + fetchImpl, + method: 'GET', + }), + ]); + + return buildHealthResponse({ + service, + version, + startTime, + dependencies: [redisDep], + upstream: [ratesApi], + criticalDependencyNames: ['redis'], + }); +} + +export async function buildSettlementEngineHealthResponse(options: { + queryDatabase: () => Promise; + pingRedis: () => Promise; + getQueueJobCounts: () => Promise>; + getQueueIsPaused?: () => Promise; + startTime: number; + service: string; + version: string; +}): Promise { + const { queryDatabase, pingRedis, getQueueJobCounts, getQueueIsPaused, startTime, service, version } = options; + + const [postgresql, redisDep, bullmq] = await Promise.all([ + checkPostgresql(queryDatabase), + checkRedisPing(pingRedis), + checkBullMQ(getQueueJobCounts, 'bullmq-settlement', HEALTH_CHECK_TIMEOUT_MS, getQueueIsPaused), + ]); + + const health = buildHealthResponse({ + service, + version, + startTime, + dependencies: [postgresql, redisDep, bullmq], + criticalDependencyNames: ['postgresql', 'redis', 'bullmq-settlement'], + }); + + const waiting = bullmq.details?.waiting; + if (typeof waiting === 'number' && waiting > BULLMQ_WAITING_DEGRADED_THRESHOLD && health.status === 'healthy') { + health.status = 'degraded'; + } + + return health; +} + +export async function buildIndexerHealthResponse(options: { + queryDatabase: () => Promise; + pingRedis: () => Promise; + getQueueJobCounts: () => Promise>; + getQueueIsPaused?: () => Promise; + getLatestLedger: () => Promise<{ sequence: number }>; + latestLedgerCursor?: number; + latestLedgerSequence?: number; + lagWarnThreshold: number; + startTime: number; + service: string; + version: string; +}): Promise { + const { + queryDatabase, + pingRedis, + getQueueJobCounts, + getQueueIsPaused, + getLatestLedger, + latestLedgerCursor, + latestLedgerSequence, + lagWarnThreshold, + startTime, + service, + version, + } = options; + + const [postgresql, redisDep, bullmq, stellarRpc] = await Promise.all([ + checkPostgresql(queryDatabase), + checkRedisPing(pingRedis), + checkBullMQ(getQueueJobCounts, 'bullmq-webhooks', HEALTH_CHECK_TIMEOUT_MS, getQueueIsPaused), + checkStellarRpc(getLatestLedger), + ]); + + const lag = + latestLedgerSequence !== undefined && latestLedgerCursor !== undefined + ? latestLedgerSequence - latestLedgerCursor + : undefined; + + const health = buildHealthResponse({ + service, + version, + startTime, + dependencies: [postgresql, redisDep, bullmq, stellarRpc], + criticalDependencyNames: ['postgresql', 'redis', 'bullmq-webhooks', 'stellar-rpc'], + }); + + if (lag !== undefined) { + health.dependencies = health.dependencies.map((dep) => + dep.name === 'stellar-rpc' + ? { + ...dep, + details: { + ...(dep.details ?? {}), + latestLedgerCursor, + latestLedgerSequence, + lag, + lagWarnThreshold, + }, + } + : dep, + ); + + if (lag > lagWarnThreshold && health.status === 'healthy') { + health.status = 'degraded'; + } + } + + const waiting = bullmq.details?.waiting; + if (typeof waiting === 'number' && waiting > BULLMQ_WAITING_DEGRADED_THRESHOLD && health.status === 'healthy') { + health.status = 'degraded'; + } + + return health; +} diff --git a/index.ts b/index.ts index 38cebf2..52b76d0 100644 --- a/index.ts +++ b/index.ts @@ -1,2772 +1,1358 @@ /** - * API Gateway — BettaPay Backend + * Settlement Engine — BettaPay Backend * - * Unified REST entry point for the BettaPay platform. - * Handles merchant registration, payment sessions, and settlement requests. + * Handles settlement processing with fee deduction and audit trail. * * Endpoints: - * GET /api/health — liveness and dependency probe - * GET /api/health/all — aggregated health across all services - * POST /api/merchants — register merchant (protected) - * GET /api/merchants/:id — fetch merchant (protected) - * DELETE /api/merchants/:id — soft-delete merchant (protected) - * POST /api/merchants/:id/restore — restore soft-deleted merchant (protected) - * PATCH /api/merchants/:id/settings — update merchant fee rules / settings (protected) - * POST /api/payments — initiate payment session (protected) - * GET /api/payments/:id — fetch payment session - * PATCH /api/payments/:id/status — transition payment status (protected) - * POST /api/settlements — trigger settlement (protected) - * GET /api/deployments — Soroban contract addresses (testnet) - * GET /api/rates — proxy to FX engine (timeout-aware) - * GET /api/currencies — proxy to FX engine (timeout-aware) - * GET /api/quote — proxy to FX engine (timeout-aware) + * GET /api/health — dependency and upstream health probe + * GET /api/settlements — list settlements (paginated) + * POST /api/settlements — create and process a settlement + * + * Precision strategy + * ────────────────── + * All monetary arithmetic uses BigNumber.js (ROUND_DOWN, no floating-point). + * Fee basis points are applied as: + * feeAmount = floor(grossAmount × feeBps / 10 000, asset decimals) + * netAmount = grossAmount − feeAmount + * + * All three amounts (grossAmount, feeAmount, netAmount) are stored as + * decimal strings so the database never loses sub-cent precision for + * assets like USDC (6 dp) or XLM (7 dp). */ -import Fastify, { - type FastifyBaseLogger, - type FastifyRequest, - type FastifyReply, -} from "fastify"; -import cors from "@fastify/cors"; -import fastifyJwt from "@fastify/jwt"; -import rateLimit from "@fastify/rate-limit"; -import crypto from "crypto"; -import zlib from "zlib"; -import { Transform } from "stream"; -import { z } from "zod"; +import Fastify from 'fastify'; +import { z } from 'zod'; +import cors from '@fastify/cors'; +import helmet from '@fastify/helmet'; +import rateLimit from '@fastify/rate-limit'; +import * as promClient from 'prom-client'; +import * as crypto from 'crypto'; +import { Queue, Worker } from 'bullmq'; +import { PrismaClient } from '@prisma/client'; +import pg from 'pg'; +import { PrismaPg } from '@prisma/adapter-pg'; +import BigNumber from 'bignumber.js'; +import { createWebhookQueue, createWebhookWorker } from '@bettapay/webhook-delivery'; +import { computeSettlementAmounts, SettlementAmountError } from './settlement-amounts.js'; +import type { DiscountTier } from './settlement-amounts.js'; +import { acquireSemaphore, releaseSemaphore, getActiveCount } from './redis-semaphore.js'; +import { closeWorkerWithTimeout, trackActiveJob } from './worker-shutdown.js'; import { validateEnvOrExit, - type Env, + CreateSettlementBody, + BulkSettlementBody, + registerErrorHandler, + registerRequestId, + createErrorResponse, + ErrorCodes, + FeeRule, + SettlementListQuery, getPrismaLogLevels, setupPrismaQueryLogging, buildPrismaConnectionUrl, connectWithRetry, - registerRequestId, createLoggerOptions, registerTracing, + buildSettlementEngineHealthResponse, + readServiceVersion, createRedisClient, waitForRedis, startRedisMemoryMonitor, startMetricsServer, - logFeatureFlags, + runStartupChecks, startPrismaPoolMetricsCollector, - encryptField, - decryptField, - encryptSensitiveFields, - decryptSensitiveFields, -} from "@bettapay/validation"; -import * as promClient from "prom-client"; -import { createFxClient } from "./clients/fx-client.js"; -import { - createIndexerClient, - type IndexerClient, -} from "./clients/indexer-client.js"; -import { UpstreamReadTimeoutError } from "./upstream-fetch.js"; -import { - createSettlementClient, - SettlementEngineUnavailableError, -} from "./clients/settlement-client.js"; -import { - CreateMerchantBody, - CreatePaymentBody, - CreateSettlementBody, - CreateSupportedAssetBody, - UpdateSupportedAssetBody, - UpdatePaymentStatusBody, - UpdateSettlementStatusBody, - UpdateMerchantSettingsBody, - UpdateMerchantNameBody, - WalletChallengeQuery, - WalletVerifyBody, - SettlementListQuery, - PaginationQuery, - BulkCancelPaymentsBody, - UpdateMerchantKycBody, - PAYMENT_STATUS_TRANSITIONS, - SETTLEMENT_STATUS_TRANSITIONS, - isValidTransition, - createErrorResponse, - ErrorCodes, - registerErrorHandler, - registerServiceAuth, - createAuditLogger, - timingSafeStrEqual, } from "@bettapay/validation"; -import type { Merchant } from "@prisma/client"; -import type { ApiResponse, PaginatedResponse } from "@bettapay/shared-types"; -import { buildPaginationMeta } from "@bettapay/shared-types"; -import { PrismaClient } from "@prisma/client"; -import pg from "pg"; -import helmet from "@fastify/helmet"; -import { PrismaPg } from "@prisma/adapter-pg"; -import { fetchUpstream, UpstreamTimeoutError } from "./upstream-fetch.js"; -import { Keypair } from "@stellar/stellar-sdk"; -import { OAuth2Client } from "google-auth-library"; -import { registerGatewayHealthRoutes } from "./health.js"; -import { - startAbandonedPaymentsCron, - stopAbandonedPaymentsCron, -} from "./abandoned-payments-cron.js"; -import { - createWebhookQueue, - type WebhookJobData, -} from "@bettapay/webhook-delivery"; -import { Queue } from "bullmq"; -import { readServiceVersion } from "@bettapay/validation"; - -declare module "fastify" { - export interface FastifyInstance { - authenticate: ( - request: FastifyRequest, - reply: FastifyReply, - ) => Promise; - } -} +import type { PaginatedResponse, ApiResponse } from '@bettapay/shared-types'; +import { buildPaginationMeta } from '@bettapay/shared-types'; -const IDEMPOTENCY_KEY_MAX_LEN = 255; -const IDEMPOTENCY_TTL_MS = 24 * 60 * 60 * 1000; // 24 hours -// Caps decompressed request bodies. Fastify's own bodyLimit only sees the -// compressed (on-the-wire) byte count, so a small gzip payload can otherwise -// decompress to many times its transmitted size before bodyLimit ever applies. -const MAX_DECOMPRESSED_BODY_BYTES = 1_048_576; -class DecompressedSizeLimitError extends Error { - statusCode = 413; - code = "DECOMPRESSED_BODY_TOO_LARGE"; - constructor() { - super("Decompressed request body exceeds the maximum allowed size"); - } -} +const env = validateEnvOrExit(process.env); +const PORT = Number(process.env.PORT ?? '3001'); +const startTime = Date.now(); +const SERVICE_VERSION = readServiceVersion(import.meta.url); -class InvalidGzipStreamError extends Error { - statusCode = 400; - code = "INVALID_GZIP_STREAM"; - constructor() { - super("Request body is not a valid gzip stream"); - } -} +const pool = new pg.Pool({ + connectionString: buildPrismaConnectionUrl(env.DATABASE_URL, env.DATABASE_POOL_SIZE, env.DATABASE_POOL_TIMEOUT), + max: env.DATABASE_POOL_SIZE, + connectionTimeoutMillis: env.DATABASE_POOL_TIMEOUT * 1000, +}); +const adapter = new PrismaPg(pool); +const prisma = new PrismaClient({ adapter, log: getPrismaLogLevels() }); + +type SettlementJobData = { + id: string; + merchantId: string; + grossAmount: string; + asset: string; + traceId?: string; +}; -// Wraps a gunzip stream with a byte counter so an oversized decompressed -// payload is rejected (413) before it is buffered into memory, and any -// decompression failure surfaces as a 400 rather than a hung connection. -function createLimitedGunzipStream(maxBytes: number): { - input: zlib.Gunzip; - output: Transform; -} { - const gunzip = zlib.createGunzip(); - let received = 0; - - const limiter = new Transform({ - transform(chunk: Buffer, _encoding, callback) { - received += chunk.length; - if (received > maxBytes) { - callback(new DecompressedSizeLimitError()); - return; - } - callback(null, chunk); - }, - }); +type SettlementRecord = NonNullable>>; - gunzip.on("error", () => { - limiter.destroy(new InvalidGzipStreamError()); - }); +const fastify = Fastify({ + logger: createLoggerOptions({ level: env.LOG_LEVEL }), + // Explicitly set body limit to 1MB (Fastify's default) + bodyLimit: 1_048_576, +}); - gunzip.pipe(limiter); +registerRequestId(fastify); +setupPrismaQueryLogging(prisma, fastify.log); +startPrismaPoolMetricsCollector(pool, promClient.register, 10000, fastify.log, promClient); - return { input: gunzip, output: limiter }; -} +// #386 — exponential backoff retry strategy +const redis = createRedisClient(env.REDIS_URL, fastify.log); -function readIdempotencyKey(request: FastifyRequest): string | null { - const raw = request.headers["idempotency-key"]; - if (!raw) return null; - const key = Array.isArray(raw) ? raw[0] : raw; - return (key as string).trim() || null; -} +fastify.addHook('onClose', async () => { + await redis.quit(); +}); -const isProduction = process.env.NODE_ENV === "production"; +fastify.register(cors, { + origin: env.ALLOWED_ORIGINS +}); -const env = validateEnvOrExit(process.env); -const PORT = Number(process.env.PORT ?? "3000"); -const startTime = Date.now(); -const SERVICE_VERSION = readServiceVersion(import.meta.url); +fastify.register(helmet, { contentSecurityPolicy: false }); -// --- Request lifecycle timeouts --------------------------------------------- +fastify.register(rateLimit, { + global: true, + max: 1000, + timeWindow: 60 * 1000, + errorResponseBuilder: (_request, context) => ({ + error: { + code: 'RATE_LIMIT_EXCEEDED', + message: `Too many requests — rate limit is ${context.max} requests per ${context.after}`, + }, + }), +}); -// REQUEST_TIMEOUT_MS bounds how long a single request may run. If a handler -// (e.g. a slow DB query or a hung upstream service) exceeds it, the per-request -// hook below replies 408 Request Timeout so the client connection is released -// instead of being held open and exhausting the connection pool. -// -// CONNECTION_TIMEOUT_MS is the socket-level backstop (set 1s higher). It closes -// any connection the request timeout did not already finish. -// -// IMPORTANT: keep both values BELOW any upstream load balancer / reverse proxy -// idle timeout (commonly 60s) so this gateway returns a clean 408 rather than -// the load balancer cutting the connection first. -const REQUEST_TIMEOUT_MS = 30_000; -const CONNECTION_TIMEOUT_MS = 31_000; - -// --- App Factory & Configuration Options ------------------------------------ -export interface AppOptions { - prisma?: PrismaClient; - indexerClient?: ReturnType; - settlementClient?: ReturnType; - fxClient?: ReturnType; - redis?: ReturnType; - logger?: any; - fetchImpl?: typeof fetch; -} +registerErrorHandler(fastify); +// Distributed tracing: log + propagate x-request-id / x-trace-id (#118). +registerTracing(fastify); -let defaultPrisma: PrismaClient | null = null; -let sharedPgPool: pg.Pool | null = null; -export function getDefaultPrisma(): PrismaClient { - if (!defaultPrisma) { - sharedPgPool = new pg.Pool({ - connectionString: buildPrismaConnectionUrl( - env.DATABASE_URL, - env.DATABASE_POOL_SIZE, - env.DATABASE_POOL_TIMEOUT, - ), - max: env.DATABASE_POOL_SIZE, - connectionTimeoutMillis: env.DATABASE_POOL_TIMEOUT * 1000, - }); - const adapter = new PrismaPg(sharedPgPool); - defaultPrisma = new PrismaClient({ adapter, log: getPrismaLogLevels() }); - startPrismaPoolMetricsCollector( - sharedPgPool, - promClient.register, - 10000, - undefined, - promClient, - ); - } - return defaultPrisma; -} +// #386 — BullMQ connection also uses exponential backoff +const redisConnection = new URL(env.REDIS_URL); +const connectionParams = { + host: redisConnection.hostname, + port: parseInt(redisConnection.port || '6379', 10), + maxRetriesPerRequest: env.REDIS_MAX_RETRIES, + enableReadyCheck: false, + retryStrategy: (attempt: number) => { + const delay = Math.min(Math.pow(2, attempt) * 100, 5_000); + fastify.log.warn({ attempt, delayMs: delay }, 'BullMQ Redis connection retry'); + return delay; + }, +}; -// Set by buildApp() when it creates the app's Redis client — shutdown()/start() -// (defined after buildApp, at module scope) need it but don't have their own -// handle on the instance buildApp created internally. -let sharedRedis: ReturnType | null = null; +// ── Settlement processing queue ──────────────────────────────────────────────── -const redis = new Redis(env.REDIS_URL, { - maxRetriesPerRequest: env.REDIS_MAX_RETRIES, - lazyConnect: true, +const settlementQueue = new Queue('settlements', { + connection: connectionParams, + defaultJobOptions: { + attempts: 3, + backoff: { type: 'exponential', delay: 2000 }, + removeOnComplete: { count: 1000 }, + removeOnFail: { count: 5000 }, + }, +}); +const settlementDLQ = new Queue('settlements-dlq', { connection: connectionParams }); + +// ── Webhook delivery queue & worker (shared @bettapay/webhook-delivery) ─────── +// +// Webhook delivery is now decoupled from the settlement worker: after updating +// the settlement status the worker enqueues a WebhookJobData onto +// 'settlement-webhooks' and returns immediately. The shared webhookWorker +// handles retries with BullMQ's built-in exponential back-off — no in-process +// sleep loop required. +// +// Migration note: the previous sendWebhookWithRetries had no persistence, so +// there are no in-flight webhook jobs to migrate. The queue name +// 'settlement-webhooks' is fresh. +const webhookQueue = createWebhookQueue('settlement-webhooks', connectionParams); +const webhookWorker = createWebhookWorker('settlement-webhooks', connectionParams, { + logger: { + info: (obj, msg) => fastify.log.info(obj, msg), + warn: (obj, msg) => fastify.log.warn(obj, msg), + error: (obj, msg) => fastify.log.error(obj, msg), + }, }); +const getActiveWebhookJob = trackActiveJob(webhookWorker); -redis.on('error', (err) => { - fastify.log.warn({ err: err.message }, 'Redis connection error'); +// ── Metrics ───────────────────────────────────────────────────────────────── +const feeFallbackCounter = new promClient.Counter({ + name: 'settlement_fee_fallback_total', + help: 'Total number of times fee resolution fell back to the default rate due to malformed settings', + labelNames: ['merchant_id'], }); -// --- Response logging hooks ------------------------------------------------- -const SENSITIVE_FIELDS = new Set([ - "token", - "secret", - "secretHash", - "password", - "privateKey", - "secretKey", -]); -const CONTROL_CHARS_EXCEPT_NEWLINES_AND_TABS = - /[\u0000-\u0008\u000B\u000C\u000E-\u001F\u007F]/g; - -function sanitizeString(value: string): string { - return value - .trim() - .replace(CONTROL_CHARS_EXCEPT_NEWLINES_AND_TABS, "") - .normalize("NFC"); -} +const settlementDelayCounter = new promClient.Counter({ + name: 'settlement_semaphore_delay_total', + help: 'Total number of settlements delayed due to per-merchant concurrency limit', + labelNames: ['merchant_id'], +}); -function sanitizeInput(value: unknown, seen = new WeakSet()): unknown { - if (typeof value === "string") { - return sanitizeString(value); - } +// Served on its own port (see startMetricsServer below), not on the +// application port — keeps the scrape endpoint unauthenticated without +// exposing it alongside application traffic. +const metricsServer = startMetricsServer({ + appPort: PORT, + contentType: promClient.register.contentType, + getMetrics: () => promClient.register.metrics(), + log: fastify.log, +}); - if (Array.isArray(value)) { - return value.map((item) => sanitizeInput(item, seen)); - } +// ── Database & Redis Setup ─────────────────────────────────────────────────────── + +// ── Monthly volume helper (Redis-cached, 5-min TTL) ───────────────────────── +// +// Used by volume-based fee discounts (#323). Queries the sum of grossAmount +// for the current calendar month for a given merchant, caching the result in +// Redis for MONTHLY_VOLUME_CACHE_TTL_SECONDS to avoid a DB round-trip on +// every settlement request. +// +// Cache key: `monthlyVol:{merchantId}:{YYYY-MM}` +// On Redis miss or error: falls back to a live DB query; on DB error: returns 0. +const MONTHLY_VOLUME_CACHE_TTL_SECONDS = 300; // 5 minutes - if (value && typeof value === "object") { - if (seen.has(value)) return value; - seen.add(value); +async function getMonthlyVolume(merchantId: string): Promise { + const now = new Date(); + const yearMonth = `${now.getFullYear()}-${String(now.getMonth() + 1).padStart(2, '0')}`; + const cacheKey = `monthlyVol:${merchantId}:${yearMonth}`; - const record = value as Record; - for (const [key, nestedValue] of Object.entries(record)) { - record[key] = sanitizeInput(nestedValue, seen); + try { + const cached = await redis.get(cacheKey); + if (cached !== null) { + const parsed = parseFloat(cached); + return isFinite(parsed) ? parsed : 0; } + } catch { + // Redis unavailable — fall through to DB query } - return value; + try { + const monthStart = new Date(now.getFullYear(), now.getMonth(), 1); + const result = await prisma.$queryRaw<[{ sum: string | null }]>` + SELECT COALESCE(SUM(CAST("grossAmount" AS DECIMAL)), 0)::text AS sum + FROM "Settlement" + WHERE "merchantId" = ${merchantId} + AND "initiatedAt" >= ${monthStart} + AND "status" IN ('completed', 'pending', 'processing') + `; + const volume = parseFloat(result[0]?.sum ?? '0'); + const safeVolume = isFinite(volume) ? volume : 0; + + // Populate cache (best-effort; ignore Redis errors) + await redis.set(cacheKey, String(safeVolume), 'EX', MONTHLY_VOLUME_CACHE_TTL_SECONDS).catch(() => {}); + + return safeVolume; + } catch { + return 0; + } } -export const QUERY_PARAM_CONTROL_CHARS_REGEX = - /[\u0000-\u0008\u000A-\u001F\u007F]/g; +const worker = new Worker('settlements', async job => { + const settlementId = job.data.id; + const merchantId = job.data.merchantId; + const traceId = job.data.traceId; + + const log = traceId + ? fastify.log.child({ traceId }) + : fastify.log; + + if (job.attemptsMade > 0) { + log.warn({ + jobId: job.id, + attempt: job.attemptsMade + 1, + maxAttempts: 3, + settlementId, + }, 'Retrying settlement job'); + } -export function sanitizeParamString(value: string): string { - return value.replace(QUERY_PARAM_CONTROL_CHARS_REGEX, ""); -} + // ── Per-merchant concurrency semaphore ────────────────────────────────────── + const maxRetries = 3; + const requeueDelayMs = 5000; + let acquired = false; + + log.info({ + jobId: job.id, + merchantId, + amount: job.data.grossAmount, + asset: job.data.asset, + jobName: job.name, + }, 'Processing settlement job'); + + for (let attempt = 0; attempt <= maxRetries; attempt++) { + acquired = await acquireSemaphore(redis, merchantId); + if (acquired) break; + + if (attempt < maxRetries) { + log.info({ + merchantId, + settlementId, + attempt: attempt + 1, + maxRetries, + }, 'Settlement delayed: merchant at concurrency limit, re-queuing'); + + settlementDelayCounter.inc({ merchant_id: merchantId }); + + await settlementQueue.add('process-settlement', job.data, { + delay: requeueDelayMs, + attempts: job.opts.attempts, + backoff: job.opts.backoff, + }); + return; + } -export function sanitizeParamsValue( - value: unknown, - seen = new WeakSet(), -): unknown { - if (typeof value === "string") { - return sanitizeParamString(value); + log.error({ + merchantId, + settlementId, + }, 'Settlement failed: merchant concurrency limit exceeded after max retries'); + throw new Error(`Merchant ${merchantId} at concurrency limit after ${maxRetries} retries`); } - if (Array.isArray(value)) { - return value.map((item) => sanitizeParamsValue(item, seen)); - } + try { + // In a real app this interacts with Soroban; here we mark completed. + const updatedSettlement = await prisma.settlement.update({ + where: { id: settlementId }, + data: { status: 'completed', completedAt: new Date() }, + }); + + log.info({ settlementId }, 'Settlement completed in database'); - if (value && typeof value === "object") { - if (seen.has(value)) return value; - seen.add(value); + if (updatedSettlement.webhookUrl) { + await webhookQueue.add('deliver', { + url: updatedSettlement.webhookUrl, + event: { event: 'settlement.completed', data: updatedSettlement as unknown as Record }, + }); + } + } catch (error) { + log.error({ error, settlementId }, 'Settlement processing failed'); + + const updatedSettlement = await prisma.settlement.update({ + where: { id: settlementId }, + data: { status: 'failed', completedAt: new Date() }, + }).catch(() => null); + + if (updatedSettlement?.webhookUrl) { + // Best-effort enqueue — don't let a queue error mask the original failure. + await webhookQueue.add('deliver', { + url: updatedSettlement.webhookUrl, + event: { event: 'settlement.failed', data: updatedSettlement as unknown as Record }, + }).catch((err: unknown) => { + log.error({ err, settlementId }, 'Failed to enqueue failure webhook'); + }); + } - const record = value as Record; - for (const [key, nestedValue] of Object.entries(record)) { - record[key] = sanitizeParamsValue(nestedValue, seen); + throw error; + } finally { + if (acquired) { + await releaseSemaphore(redis, merchantId).catch(() => {}); } } +}, { + connection: connectionParams, + concurrency: 5, +}); - return value; -} +const getActiveSettlementJob = trackActiveJob(worker); -function redactValue(value: any): any { - if (value === null || value === undefined) return value; - if (Array.isArray(value)) return value.map(redactValue); - if (typeof value === "object") return redactObject(value); - return value; -} +worker.on('failed', async (job, err) => { + if (job) { + fastify.log.error({ + jobId: job.id, + settlementId: job.data.id, + attempt: job.attemptsMade, + error: err.message, + }, 'Job failed after all retries, moving to DLQ'); -function redactObject(obj: Record) { - const out: Record = {}; - for (const k of Object.keys(obj)) { - try { - if (SENSITIVE_FIELDS.has(k)) { - out[k] = "[REDACTED]"; - } else { - out[k] = redactValue(obj[k]); - } - } catch (e) { - out[k] = "[REDACTION_ERROR]"; - } + await settlementDLQ.add(job.name, job.data, { + jobId: job.id, + attempts: 1, + }); } - return out; -} +}); -function hashSecret(secret: string): string { - return crypto.createHash("sha256").update(secret).digest("hex"); -} +settlementQueue.on('error', (err) => { + fastify.log.error({ err: err.message }, 'BullMQ queue connection error'); +}); +settlementDLQ.on('error', (err) => { + fastify.log.error({ err: err.message }, 'BullMQ DLQ connection error'); +}); +worker.on('error', (err) => { + fastify.log.error({ err: err.message }, 'BullMQ worker connection error'); +}); +webhookQueue.on('error', (err) => { + fastify.log.error({ err: err.message }, 'BullMQ webhook queue connection error'); +}); +webhookWorker.on('error', (err) => { + fastify.log.error({ err: err.message }, 'BullMQ webhook worker connection error'); +}); -export function buildApp(opts: AppOptions = {}) { - const fastify = Fastify({ - logger: - opts.logger !== undefined - ? opts.logger - : createLoggerOptions({ level: env.LOG_LEVEL }), - requestTimeout: REQUEST_TIMEOUT_MS, - connectionTimeout: CONNECTION_TIMEOUT_MS, - bodyLimit: 1_048_576, +fastify.get('/api/health', async (_request, reply) => { + const health = await buildSettlementEngineHealthResponse({ + queryDatabase: () => prisma.$queryRaw`SELECT 1`, + pingRedis: () => redis.ping(), + getQueueJobCounts: () => settlementQueue.getJobCounts(), + getQueueIsPaused: () => settlementQueue.isPaused(), + startTime, + service: 'settlement-engine', + version: SERVICE_VERSION, }); + const statusCode = health.status === 'unhealthy' ? 503 : 200; + return reply.code(statusCode).send(health); +}); - registerRequestId(fastify); - registerErrorHandler(fastify); - registerTracing(fastify); - registerServiceAuth(fastify, env.INTER_SERVICE_SECRET); - - // Centralized query and path parameter sanitization preHandler hook: - // Recursively strips unsafe ASCII control characters (0x00-0x1F except \t, 0x7F) - fastify.addHook("preHandler", async (request: FastifyRequest) => { - if (request.query && typeof request.query === "object") { - sanitizeParamsValue(request.query); - } - if (request.params && typeof request.params === "object") { - sanitizeParamsValue(request.params); - } +fastify.get('/api/settlements', async (request, reply): Promise> => { + const { page, limit, status, from, to, includeDeleted } = SettlementListQuery.parse(request.query ?? {}); + const where: any = {}; + if (status) where.status = status; + if (from || to) { + where.initiatedAt = {}; + if (from) where.initiatedAt.gte = new Date(from); + if (to) where.initiatedAt.lte = new Date(to); + } + // Exclude superseded settlements by default (#322) + if (!includeDeleted) { + where.supersededById = null; + } + const records = await prisma.settlement.findMany({ + where, + take: limit, + skip: (page - 1) * limit, + orderBy: { initiatedAt: 'desc' }, }); + const total = await prisma.settlement.count({ where }); + return { + data: records, + pagination: buildPaginationMeta(page, limit, total) + }; +}); - // Transparent field-level decryption before sending API responses: - fastify.addHook("preSerialization", async (_request, _reply, payload) => { - return decryptSensitiveFields(payload); - }); +// ============================================================================ +// SETTLEMENT RETRY (#322) +// ============================================================================ + +fastify.post<{ Params: { id: string } }>( + '/api/settlements/:id/retry', + async (request, reply) => { + const { id } = request.params; - // Guards against decompression bombs: Fastify's own bodyLimit only checks - // the compressed (on-the-wire) size, so a small gzip payload could otherwise - // decompress to well beyond the intended cap before anything notices. - fastify.addHook("preParsing", async (request, _reply, payload) => { - const contentEncoding = request.headers["content-encoding"]; - if (!contentEncoding || contentEncoding === "identity") { - return payload; + // Fetch the original settlement + const original = await prisma.settlement.findUnique({ + where: { id }, + }); + + if (!original) { + return reply.code(404).send(createErrorResponse(ErrorCodes.NOT_FOUND, 'Settlement not found')); } - if (contentEncoding !== "gzip") { - return payload; + + // Only failed settlements can be retried + if (original.status !== 'failed') { + return reply.code(422).send(createErrorResponse( + ErrorCodes.VALIDATION_ERROR, + 'Only failed settlements can be retried', + { currentStatus: original.status } + )); } - // The decompressed size no longer matches the original (compressed) - // Content-Length, so drop it — otherwise Fastify's own body reader - // rejects the request with FST_ERR_CTP_INVALID_CONTENT_LENGTH. - delete request.headers["content-length"]; + // Count the retry chain to enforce max 3 retries + const retryChain = await prisma.settlement.findMany({ + where: { + OR: [ + { supersededById: id }, + { id: original.supersededById ?? '' }, + ], + }, + }); + + // Find the root of the chain + let current = original; + let chainLength = 0; + const visited = new Set(); - const { input, output } = createLimitedGunzipStream( - MAX_DECOMPRESSED_BODY_BYTES, - ); - payload.pipe(input); - return output; - }); + while (current.supersededById && !visited.has(current.id)) { + visited.add(current.id); + chainLength++; + const parent = await prisma.settlement.findUnique({ + where: { id: current.supersededById }, + }); + if (!parent) break; + current = parent; + } - const prisma = opts.prisma ?? getDefaultPrisma(); - const indexerClient = - opts.indexerClient ?? - createIndexerClient({ - baseUrl: env.INDEXER_URL, - serviceToken: env.INTER_SERVICE_SECRET, - logger: fastify.log, - timeoutMs: env.READ_TIMEOUT_MS, - }); - const settlementClient = - opts.settlementClient ?? - createSettlementClient({ - baseUrl: env.SETTLEMENT_ENGINE_URL, - serviceToken: env.INTER_SERVICE_SECRET, - logger: fastify.log, - timeoutMs: env.WRITE_TIMEOUT_MS, - }); - const fxClient = - opts.fxClient ?? - createFxClient({ - baseUrl: env.FX_ENGINE_URL, - serviceToken: env.INTER_SERVICE_SECRET, - logger: fastify.log, - timeoutMs: env.READ_TIMEOUT_MS, + // Count forward retries from original + const forwardRetries = await prisma.settlement.count({ + where: { supersededById: id }, }); - const logAuditEvent = createAuditLogger( - prisma as unknown as Parameters[0], - fastify.log, - ); - - // Setup plugins - fastify.register(helmet, { - contentSecurityPolicy: false, - crossOriginEmbedderPolicy: { policy: "require-corp" }, - crossOriginOpenerPolicy: { policy: "same-origin" }, - crossOriginResourcePolicy: { policy: "same-origin" }, - referrerPolicy: { policy: "strict-origin-when-cross-origin" }, - hsts: { maxAge: 31536000 }, - }); - fastify.addHook("onSend", async (_request, reply, _payload) => { - if (!reply.getHeader("permissions-policy")) { - reply.header( - "Permissions-Policy", - "geolocation=(), microphone=(), camera=()", - ); + const totalRetries = chainLength + forwardRetries; + + if (totalRetries >= 3) { + return reply.code(422).send(createErrorResponse( + ErrorCodes.VALIDATION_ERROR, + 'Maximum retry limit (3) exceeded', + { retryCount: totalRetries } + )); } - }); - fastify.register(cors, { - origin: env.ALLOWED_ORIGINS, - credentials: true, - }); + // Clone the settlement + const newSettlementId = 'set_' + crypto.randomUUID().replace(/-/g, ''); + const traceId = crypto.randomUUID(); - fastify.register(fastifyJwt, { - secret: env.JWT_SECRET, - sign: { - expiresIn: env.JWT_EXPIRES_IN, - }, - }); + const newSettlement = await prisma.settlement.create({ + data: { + id: newSettlementId, + merchantId: original.merchantId, + totalAmount: original.totalAmount, + grossAmount: original.grossAmount, + feeAmount: original.feeAmount, + netAmount: original.netAmount, + feeBps: original.feeBps, + asset: original.asset, + status: 'pending', + webhookUrl: original.webhookUrl, + feeSnapshot: (original.feeSnapshot ?? undefined) as any, + }, + }); - // Rate limiting: global default and route overrides - fastify.register(rateLimit, { - max: 1000, - timeWindow: "1 minute", - addHeaders: { - "x-ratelimit-limit": true, - "x-ratelimit-remaining": true, - "x-ratelimit-reset": true, - "retry-after": true, - }, - }); + // Mark original as superseded + await prisma.settlement.update({ + where: { id }, + data: { supersededById: newSettlementId }, + }); - // Exposes standard X-RateLimit-* response headers on every rate-limited - // route. The installed @fastify/rate-limit version tracks hit counts in a - // store that's private to its own onRequest hook, with no read-only "peek" - // API, so we mirror it with our own counter built from the same - // `createRateLimit` helper and the route's own (global or overridden) limit - // config. One `checkRateLimit` instance is cached per route so its counter - // persists (and accumulates) across requests exactly like the real one — - // both increment exactly once per request against the same max/window, so - // they always agree on the numbers. Routes opted out via - // `config: { rateLimit: false }` (e.g. health checks) are skipped. - const rateLimitCheckers = new WeakMap< - object, - ReturnType - >(); - - fastify.addHook( - "onSend", - async (request: FastifyRequest, reply: FastifyReply, payload) => { - const routeConfig = request.routeOptions?.config as - | { rateLimit?: false | Record } - | undefined; - if (!routeConfig || routeConfig.rateLimit === false) { - return payload; - } + // Queue the new settlement for processing + await settlementQueue.add('process-settlement', { + id: newSettlementId, + merchantId: newSettlement.merchantId, + grossAmount: newSettlement.grossAmount, + asset: newSettlement.asset, + traceId, + }); - let checkRateLimit = rateLimitCheckers.get(routeConfig); - if (!checkRateLimit) { - checkRateLimit = fastify.createRateLimit( - typeof routeConfig.rateLimit === "object" - ? routeConfig.rateLimit - : {}, - ); - rateLimitCheckers.set(routeConfig, checkRateLimit); - } + fastify.log.info({ originalId: id, newId: newSettlementId, retryCount: totalRetries + 1 }, 'Settlement retried'); - const result = (await checkRateLimit(request)) as { - max?: number; - remaining?: number; - ttlInSeconds?: number; - }; - - if (typeof result.max === "number") { - reply.header("X-RateLimit-Limit", result.max); - reply.header("X-RateLimit-Remaining", result.remaining ?? 0); - reply.header( - "X-RateLimit-Reset", - Math.ceil(Date.now() / 1000) + (result.ttlInSeconds ?? 0), - ); - } + return reply.code(201).send({ data: newSettlement }); + } +); - return payload; - }, - ); - - // --- Same-origin enforcement -------------------------------------------------- - // Reject cross-origin mutations that lack an explicit CORS preflight. - // Server-to-server calls (no Origin header, authenticated via x-service-token) - // are exempt. GET/HEAD are also exempt since they cannot cause state changes. - const ALLOWED_ORIGINS_SET = new Set( - env.ALLOWED_ORIGINS.map((o) => o.toLowerCase()), - ); - - fastify.addHook( - "onRequest", - async (request: FastifyRequest, reply: FastifyReply) => { - const method = request.method; - if (method === "GET" || method === "HEAD" || method === "OPTIONS") return; - - const origin = request.headers.origin; - if (!origin) return; - - const normalised = origin.trim().replace(/\/+$/, "").toLowerCase(); - const isAllowed = [...ALLOWED_ORIGINS_SET].some((allowed) => - timingSafeStrEqual(normalised, allowed), - ); +interface ReconcileQuery { + merchantId?: string; + from?: string; + to?: string; +} - if (!isAllowed) { - request.log.warn( - { origin, method, url: request.url }, - "Rejected cross-origin mutation", - ); - return reply - .code(403) - .send( - createErrorResponse( - ErrorCodes.INVALID_ORIGIN, - "Request origin is not allowed", - ), - ); +/** + * Local Consistency Check for Settlements + * + * This endpoint performs internal validation of settlement records to ensure + * data integrity. It verifies: + * - Mathematical consistency: grossAmount - feeAmount = netAmount + * - Fee calculation accuracy: feeAmount matches feeBps applied to grossAmount + * - Merchant reference validity: all settlements reference existing merchants + * + * This is a LOCAL consistency check - it does not make external HTTP calls. + * All validation is performed against the settlement engine's own database. + */ +fastify.get<{ Querystring: ReconcileQuery }>('/api/settlements/reconcile', async (request, reply) => { + try { + const { merchantId, from, to } = request.query; + + const where: Record = {}; + if (merchantId) { + where.merchantId = merchantId; + } + if (from || to) { + where.initiatedAt = {}; + if (from) { + (where.initiatedAt as Record).gte = new Date(from); } - }, - ); - - // Request body logging for mutation endpoints - async function logRequestBody(request: FastifyRequest, reply: FastifyReply) { - if (request.body && typeof request.body === "object") { - const cloned = JSON.parse(JSON.stringify(request.body)); - for (const key of SENSITIVE_FIELDS) { - if (key in cloned) { - cloned[key] = "[REDACTED]"; - } + if (to) { + (where.initiatedAt as Record).lte = new Date(to); } - const logLevel = isProduction ? "debug" : "info"; - request.log[logLevel]( - { requestId: request.id, body: cloned }, - "incoming request body", - ); } - } - - // Authentication hook - fastify.decorate( - "authenticate", - async function (request: FastifyRequest, reply: FastifyReply) { - try { - await request.jwtVerify(); - } catch (err) { - request.log.error(err); - return reply - .code(401) - .send(createErrorResponse(ErrorCodes.UNAUTHORIZED, "Unauthorized")); - } - const jti = (request.user as any)?.jti; - const merchantId = (request.user as any)?.merchantId; - if (!jti || !merchantId) { - return; - } + // Query settlements from local database + const settlements = await prisma.settlement.findMany({ + where, + orderBy: { initiatedAt: 'desc' }, + }); -// Authentication hook -fastify.decorate('authenticate', async function (request: FastifyRequest, reply: FastifyReply) { - try { - await request.jwtVerify(); - const payload = request.user as MerchantJwtPayload; - if (payload.jti && await isJtiRevoked(payload.jti)) { - return reply.code(401).send(createErrorResponse(ErrorCodes.UNAUTHORIZED, 'Unauthorized')); - } - } catch (err) { - request.log.error(err); - return reply.code(401).send(createErrorResponse(ErrorCodes.UNAUTHORIZED, 'Unauthorized')); - } -}); - try { - const ok = await updateSessionLastUsed(jti, merchantId); - if (!ok) { - request.log.warn( - { jti, merchantId }, - "[Auth] JWT session missing or revoked", - ); - return reply - .code(401) - .send(createErrorResponse(ErrorCodes.UNAUTHORIZED, "Unauthorized")); - } - } catch (err: any) { - request.log.error( - { err, jti, merchantId }, - "[Auth] Session validation failed", - ); - return reply - .code(503) - .send( - createErrorResponse( - ErrorCodes.INTERNAL_ERROR, - "Authentication service unavailable", - ), - ); - } - }, - ); + // 2. Fetch api-gateway records via HTTP call + const gatewayUrl = process.env.API_GATEWAY_URL || 'http://localhost:3000'; + const url = new URL(`${gatewayUrl}/api/settlements`); + if (merchantId) url.searchParams.append('merchantId', merchantId); + if (from) url.searchParams.append('from', from); + if (to) url.searchParams.append('to', to); - // Per-merchant concurrent request limiting via Redis. - // Uses INCR with a TTL so that abandoned connections (e.g. dropped before - // onResponse fires) are automatically cleaned up after 30 seconds. - const MERCHANT_CONCURRENCY_TTL_SEC = 30; - const merchantMaxConcurrency = env.MERCHANT_MAX_CONCURRENCY; + const token = env.INTER_SERVICE_SECRET; - fastify.addHook( - "preHandler", - async (request: FastifyRequest, reply: FastifyReply) => { - const merchantId = (request.user as any)?.merchantId; - if (!merchantId) return; + let gatewayRecords: any[] = []; + try { + const response = await fetch(url.toString(), { + headers: { + 'x-service-token': token, + 'Content-Type': 'application/json', + }, + }); - const key = `concurrency:${merchantId}`; - try { - const count = await redis.incr(key); - if (count === 1) { - await redis.expire(key, MERCHANT_CONCURRENCY_TTL_SEC); - } - if (count > merchantMaxConcurrency) { - await redis.decr(key); - return reply - .code(429) - .header("Retry-After", "1") - .send( - createErrorResponse( - ErrorCodes.CONCURRENCY_EXCEEDED, - "Too many concurrent requests", - ), - ); - } - } catch (err) { - request.log.error( - { err, merchantId }, - "Concurrency limiter Redis error — allowing request through", - ); + if (!response.ok) { + throw new Error(`API Gateway returned status ${response.status}`); } - }, - ); - fastify.addHook( - "onResponse", - async (request: FastifyRequest, _reply: FastifyReply) => { - const merchantId = (request.user as any)?.merchantId; - if (!merchantId) return; + const data = await response.json() as { data: any[] }; + gatewayRecords = data.data; + } catch (error) { + fastify.log.error({ error }, 'Failed to fetch settlements from API Gateway'); + return reply.code(502).send({ + error: { code: 'UPSTREAM_ERROR', message: 'Failed to fetch settlement records from api-gateway', details: error instanceof Error ? error.message : String(error) } + }); + } - const key = `concurrency:${merchantId}`; - try { - await redis.decr(key); - } catch (err) { - request.log.error( - { err, merchantId }, - "Concurrency limiter Redis DECR error", - ); - } - }, - ); + // 3. Diff the two sets by settlement ID and compare records + const localMap = new Map(); + for (const r of settlements) { + localMap.set(r.id, r); + } - fastify.addHook("preHandler", async (request) => { - if (request.body !== undefined) { - request.body = sanitizeInput(request.body); + const gatewayMap = new Map(); + for (const r of gatewayRecords) { + gatewayMap.set(r.id, r); } - }); - // Zod validation runs inside route handlers after this global preHandler, so - // schemas receive trimmed, control-character-free, NFC-normalized strings. - - // Routes - registerGatewayHealthRoutes({ - fastify, - prisma, - env: { - FX_ENGINE_URL: env.FX_ENGINE_URL, - SETTLEMENT_ENGINE_URL: env.SETTLEMENT_ENGINE_URL, - INDEXER_URL: env.INDEXER_URL, - }, - startTime, - serviceVersion: SERVICE_VERSION, - fetchImpl: opts.fetchImpl, - }); + const matchedIds = new Set(); + const missing: any[] = []; // In gateway, but missing in local + const extra: any[] = []; // In local, but missing in gateway + const mismatched: any[] = []; // In both, but fields differ - // --- Wallet Auth Challenge Store ---------------------------------------------- - // #386 — exponential backoff retry strategy - const redis = createRedisClient(env.REDIS_URL, fastify.log); - sharedRedis = redis; - - const GOOGLE_AUTH_GRACE_PERIOD_MS = 30_000; - const GOOGLE_AUTH_LOCKOUT_KEY_PREFIX = "auth_fail:google:"; - const SESSION_KEY_PREFIX = "session:"; - const SESSION_INDEX_PREFIX = "sessions:"; - const SESSION_LIMIT_PER_MERCHANT = 10; - - async function getSessionMetadata(jti: string) { - const sessionRaw = await redis.get(`${SESSION_KEY_PREFIX}${jti}`); - if (!sessionRaw) return null; - return JSON.parse(sessionRaw) as { - merchantId: string; - deviceInfo: string; - createdAt: string; - lastUsedAt: string; - }; - } + let localGrossTotal = new BigNumber(0); + let localFeeTotal = new BigNumber(0); + let localNetTotal = new BigNumber(0); - async function updateSessionLastUsed(jti: string, merchantId: string) { - const session = await getSessionMetadata(jti); - if (!session || session.merchantId !== merchantId) return false; - session.lastUsedAt = new Date().toISOString(); - await Promise.all([ - redis.set(`${SESSION_KEY_PREFIX}${jti}`, JSON.stringify(session)), - redis.zadd( - `${SESSION_INDEX_PREFIX}${merchantId}`, - Date.parse(session.lastUsedAt), - jti, - ), - ]); - return true; - } + let gatewayGrossTotal = new BigNumber(0); + let gatewayFeeTotal = new BigNumber(0); + let gatewayNetTotal = new BigNumber(0); - async function createAuthSession(merchantId: string, deviceInfo: string) { - const jti = crypto.randomBytes(16).toString("hex"); - const session = { - merchantId, - deviceInfo, - createdAt: new Date().toISOString(), - lastUsedAt: new Date().toISOString(), + const parseBN = (val: unknown): BigNumber => { + const bn = new BigNumber(val as string ?? 0); + return bn.isFinite() ? bn : new BigNumber(0); }; - const indexKey = `${SESSION_INDEX_PREFIX}${merchantId}`; - await redis.set(`${SESSION_KEY_PREFIX}${jti}`, JSON.stringify(session)); - await redis.zadd(indexKey, Date.parse(session.lastUsedAt), jti); + const inconsistencies: Array<{ + settlementId: string; + type: 'amount_mismatch' | 'fee_calculation' | 'missing_merchant'; + details: Record; + }> = []; + + let totalGross = new BigNumber(0); + let totalFee = new BigNumber(0); + let totalNet = new BigNumber(0); + let validCount = 0; + + const statusCounts: Record = { + pending: 0, + processing: 0, + completed: 0, + failed: 0, + }; - const totalSessions = await redis.zcard(indexKey); - if (totalSessions > SESSION_LIMIT_PER_MERCHANT) { - const toRemove = await redis.zrange( - indexKey, - 0, - totalSessions - SESSION_LIMIT_PER_MERCHANT - 1, - ); - if (toRemove.length > 0) { - await Promise.all( - toRemove.map((oldJti) => redis.del(`${SESSION_KEY_PREFIX}${oldJti}`)), - ); - await redis.zrem(indexKey, ...toRemove); + const merchants = await prisma.merchant.findMany({ select: { id: true } }); + const existingMerchantIds = new Set(merchants.map(m => m.id)); + + for (const settlement of settlements) { + const gross = parseBN(settlement.grossAmount); + const fee = parseBN(settlement.feeAmount); + const net = parseBN(settlement.netAmount); + + totalGross = totalGross.plus(gross); + totalFee = totalFee.plus(fee); + totalNet = totalNet.plus(net); + + statusCounts[settlement.status] = (statusCounts[settlement.status] || 0) + 1; + + // Check 1: Verify grossAmount - feeAmount = netAmount + const expectedNet = gross.minus(fee); + if (!expectedNet.isEqualTo(net)) { + inconsistencies.push({ + settlementId: settlement.id, + type: 'amount_mismatch', + details: { + grossAmount: settlement.grossAmount, + feeAmount: settlement.feeAmount, + netAmount: settlement.netAmount, + expectedNet: expectedNet.toString(), + }, + }); + continue; } - } - return jti; - } - - async function listAuthSessions(merchantId: string) { - const indexKey = `${SESSION_INDEX_PREFIX}${merchantId}`; - const jtis = await redis.zrange(indexKey, 0, -1); - const sessions = await Promise.all( - jtis.map(async (jti) => { - const metadata = await getSessionMetadata(jti); - return metadata ? { jti, ...metadata } : null; - }), - ); - return sessions.filter( - ( - session, - ): session is { - jti: string; - merchantId: string; - deviceInfo: string; - createdAt: string; - lastUsedAt: string; - } => Boolean(session), - ); - } - - async function revokeAuthSession(merchantId: string, jti: string) { - const metadata = await getSessionMetadata(jti); - if (!metadata || metadata.merchantId !== merchantId) return false; - await Promise.all([ - redis.del(`${SESSION_KEY_PREFIX}${jti}`), - redis.zrem(`${SESSION_INDEX_PREFIX}${merchantId}`, jti), - ]); - return true; - } + // Check 2: Verify fee calculation matches feeBps + // feeAmount = floor(grossAmount × feeBps / 10000) + const expectedFee = gross.times(settlement.feeBps).dividedBy(10000).integerValue(BigNumber.ROUND_DOWN); + // Allow for minor precision differences (within 1 unit) + if (expectedFee.minus(fee).abs().isGreaterThan(1)) { + inconsistencies.push({ + settlementId: settlement.id, + type: 'fee_calculation', + details: { + grossAmount: settlement.grossAmount, + feeBps: settlement.feeBps, + actualFee: settlement.feeAmount, + expectedFee: expectedFee.toString(), + }, + }); + continue; + } - async function getGoogleAuthLockoutCount(email: string) { - return parseInt( - (await redis.get(`${GOOGLE_AUTH_LOCKOUT_KEY_PREFIX}${email}`)) || "0", - 10, - ); - } + // Check 3: Verify merchant exists + if (!existingMerchantIds.has(settlement.merchantId)) { + inconsistencies.push({ + settlementId: settlement.id, + type: 'missing_merchant', + details: { + merchantId: settlement.merchantId, + }, + }); + continue; + } - async function incrementGoogleAuthLockout(email: string) { - const lockoutKey = `${GOOGLE_AUTH_LOCKOUT_KEY_PREFIX}${email}`; - const count = await redis.incr(lockoutKey); - await redis.expire(lockoutKey, env.AUTH_LOCKOUT_MINUTES * 60); - return count; - } + validCount++; + } - async function resetGoogleAuthLockout(email: string) { - await redis - .del(`${GOOGLE_AUTH_LOCKOUT_KEY_PREFIX}${email}`) - .catch(() => {}); + return { + summary: { + total: settlements.length, + valid: validCount, + inconsistent: inconsistencies.length, + }, + statusBreakdown: statusCounts, + totals: { + gross: totalGross.toString(), + fee: totalFee.toString(), + net: totalNet.toString(), + }, + inconsistencies, + reconciliationType: 'local_consistency_check', + }; + } catch (error) { + fastify.log.error({ error }, 'Reconciliation error'); + return reply.code(400).send({ error: 'Failed to perform reconciliation' }); } +}); - fastify.get<{ Querystring: WalletChallengeQuery }>( - "/api/auth/wallet/challenge", - { - config: { rateLimit: { max: 10, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - const { address } = WalletChallengeQuery.parse(request.query); - const nonce = crypto.randomBytes(32).toString("hex"); - const challenge = `BettaPay:${address}:${nonce}`; - const expiresAt = Date.now() + 2 * 60 * 1000; // 2 minutes - try { - await redis.set( - `wallet_challenge:${address}`, - JSON.stringify({ challenge, expiresAt }), - "PX", - 120000, - ); - } catch (err) { - request.log.error({ err }, "Failed to set wallet challenge in Redis"); - return reply - .code(503) - .send({ error: "Authentication service unavailable" }); - } - return reply.send({ challenge, expiresAt }); +fastify.post<{ Body: z.infer }>( + '/api/settlements', + { + config: { + rateLimit: { + max: 60, + timeWindow: 60 * 1000, + }, }, - ); + }, + async (request, reply) => { + const d = CreateSettlementBody.parse(request.body); - fastify.post<{ Body: WalletVerifyBody }>( - "/api/auth/wallet/verify", - { - config: { rateLimit: { max: 10, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - const d = WalletVerifyBody.parse(request.body); - const ip = request.ip; - const lockoutKey = `wallet_lockout:${d.address}`; - - const failedAttempts = parseInt((await redis.get(lockoutKey)) || "0", 10); - const maxAttempts = parseInt( - process.env.AUTH_MAX_FAILED_ATTEMPTS || "5", - 10, - ); - if (failedAttempts >= maxAttempts) { - request.log.warn( - { address: d.address, ip }, - "[Auth] Wallet verify locked out due to too many failed attempts", - ); - return reply - .code(429) - .send({ error: "Too many failed attempts. Try again later." }); - } + if (!d.amount || !d.asset) { + return reply.code(400).send(createErrorResponse(ErrorCodes.VALIDATION_ERROR, 'amount and asset are required')); + } - let storedRaw; - try { - storedRaw = await redis.get(`wallet_challenge:${d.address}`); - } catch (err) { - request.log.error({ err }, "Failed to get wallet challenge from Redis"); - return reply - .code(503) - .send({ error: "Authentication service unavailable" }); - } + // Validate that the amount is positive without floating-point conversion + const grossBN = new BigNumber(d.amount); + if (!grossBN.isFinite() || grossBN.isLessThanOrEqualTo(0)) { + return reply.code(400).send(createErrorResponse(ErrorCodes.VALIDATION_ERROR, 'amount must be > 0')); + } - if (!storedRaw) { - return reply - .code(400) - .send({ error: "Challenge expired or not found" }); - } + const merchant = await prisma.merchant.findUnique({ where: { id: d.merchantId } }); + const parsedFeeRule = FeeRule.passthrough().safeParse(merchant?.settings); + let feeBps = env.FEES_DEFAULT_BPS; + let maxFeeBps: number | undefined; + let maxFeeThreshold: string | undefined; + + if (parsedFeeRule.success) { + feeBps = parsedFeeRule.data.feeBps; + const settings = parsedFeeRule.data as Record; + maxFeeBps = settings.maxFeeBps as number | undefined; + maxFeeThreshold = settings.maxFeeThreshold as string | undefined; + } else { + feeFallbackCounter.inc({ merchant_id: d.merchantId }); + fastify.log.warn({ + merchantId: d.merchantId, + rawSettings: merchant?.settings, + issues: parsedFeeRule.error?.issues + }, '[Settlement] FeeRule parsing failed, falling back to FEES_DEFAULT_BPS'); + } + const webhookUrl = parsedFeeRule.success ? (parsedFeeRule.data as Record).webhookUrl as string ?? null : null; - const stored = JSON.parse(storedRaw); + // Fetch monthly volume for volume-based fee discount (#323). + // Redis-cached with a 5-min TTL; falls back to DB query on cache miss. + const monthlyVolume = await getMonthlyVolume(d.merchantId); + const discountTiers: DiscountTier[] = env.FEE_DISCOUNT_TIERS ?? []; - if (Date.now() > stored.expiresAt) { - await redis.del(`wallet_challenge:${d.address}`).catch(() => {}); - return reply.code(400).send({ error: "Challenge expired" }); - } - if (stored.challenge !== d.challenge) { - return reply.code(400).send({ error: "Invalid challenge" }); + let computeResult; + try { + computeResult = computeSettlementAmounts( + d.amount, + feeBps, + monthlyVolume, + discountTiers, + ); + } catch (error) { + if (error instanceof SettlementAmountError) { + return reply.code(422).send(createErrorResponse(ErrorCodes.VALIDATION_ERROR, error.message)); } + throw error; + } + const { grossAmount, feeAmount, netAmount, feeSnapshot } = computeResult; + + if (feeSnapshot.discountApplied > 0) { + fastify.log.info({ + merchantId: d.merchantId, + monthlyVolume, + baseBps: feeBps, + effectiveBps: feeSnapshot.feeBpsApplied, + discountBps: feeSnapshot.discountApplied, + }, '[Settlement] Volume-based fee discount applied'); + } - await redis.del(`wallet_challenge:${d.address}`).catch(() => {}); // Single use + const rawIdempotencyKey = request.headers['idempotency-key']; + const idempotencyKey = Array.isArray(rawIdempotencyKey) ? rawIdempotencyKey[0] : rawIdempotencyKey; + const settlementId = 'set_' + crypto.randomUUID().replace(/-/g, ''); + + if (idempotencyKey) { + let claimed: string | null = null; try { - const keypair = Keypair.fromPublicKey(d.address); - const isValid = keypair.verify( - Buffer.from(d.challenge, "utf-8"), - Buffer.from(d.signature, "base64"), - ); - if (!isValid) { - const lockoutMinutes = parseInt( - process.env.AUTH_LOCKOUT_MINUTES || "15", - 10, - ); - await redis.incr(lockoutKey); - await redis.expire(lockoutKey, lockoutMinutes * 60); - request.log.warn( - { address: d.address, ip }, - "[Auth] Invalid signature during wallet verify", - ); - return reply.code(401).send({ error: "Invalid signature" }); + claimed = await redis.set(`idempotency:${idempotencyKey}`, settlementId, 'EX', 86400, 'NX'); + } catch { + // Redis unavailable — fall through to DB @unique constraint + } + + if (claimed === null) { + // Another request atomically claimed this idempotency key first + const existingId = await redis.get(`idempotency:${idempotencyKey}`).catch(() => null); + if (existingId) { + const existingSettlement = await prisma.settlement.findUnique({ + where: { id: existingId }, + }); + if (existingSettlement) { + return reply.code(200).send({ data: existingSettlement }); + } } - } catch (err) { - const lockoutMinutes = parseInt( - process.env.AUTH_LOCKOUT_MINUTES || "15", - 10, - ); - await redis.incr(lockoutKey); - await redis.expire(lockoutKey, lockoutMinutes * 60); - request.log.warn( - { address: d.address, ip }, - "[Auth] Signature verification failed", - ); - return reply.code(401).send({ error: "Signature verification failed" }); } + } + + const settlement = await prisma.settlement.create({ + data: { + id: settlementId, + merchantId: d.merchantId, + totalAmount: grossAmount, + grossAmount, + feeAmount, + netAmount, + feeBps, + asset: d.asset, + status: 'pending', + webhookUrl, + feeSnapshot: feeSnapshot as any, + idempotencyKey: idempotencyKey ?? undefined, + idempotencyKeyExpiresAt: idempotencyKey ? new Date(Date.now() + 86400_000) : undefined, + }, + }); - await redis.del(lockoutKey).catch(() => {}); // reset on success + const traceId = (request as unknown as { traceId?: string }).traceId; - const merchant = await prisma.merchant.upsert({ - where: { ownerId: d.address }, - update: {}, - create: { - id: crypto.randomUUID(), - name: "My Business", - ownerId: d.address, - settings: {}, - }, - }); + const jobData: SettlementJobData = { + id: settlement.id, + merchantId: settlement.merchantId, + grossAmount: settlement.grossAmount, + asset: settlement.asset, + traceId, + }; - const jwtToken = fastify.jwt.sign({ merchantId: merchant.id, ownerId: merchant.ownerId }); - return reply.send({ token: jwtToken }); - } catch (err: any) { - request.log.error({ err }, '[Auth] Google OAuth failed'); - return reply.code(401).send(createErrorResponse(ErrorCodes.UNAUTHORIZED, 'Google token verification failed')); - } + await settlementQueue.add('process-settlement', jobData); + + return reply.code(201).send({ data: settlement }); }); -fastify.post('/api/auth/refresh', { - preHandler: [enforceAuthIpReputation] -}, async (request, reply) => { - try { - await request.jwtVerify(); - } catch (err) { - request.log.error(err); - await recordAuthIpFailure(request); - return reply.code(401).send(createErrorResponse(ErrorCodes.UNAUTHORIZED, 'Unauthorized')); - } - - const payload = request.user as MerchantJwtPayload; - if (!payload.merchantId || !payload.ownerId || !payload.jti || !payload.exp) { - await recordAuthIpFailure(request); - return reply.code(401).send(createErrorResponse(ErrorCodes.UNAUTHORIZED, 'Unauthorized')); - } - - if (await isJtiRevoked(payload.jti)) { - await recordAuthIpFailure(request); - return reply.code(401).send(createErrorResponse(ErrorCodes.UNAUTHORIZED, 'Unauthorized')); - } - - const remainingLifetime = payload.exp - Math.floor(Date.now() / 1000); - if (remainingLifetime <= 0) { - await recordAuthIpFailure(request); - return reply.code(401).send(createErrorResponse(ErrorCodes.UNAUTHORIZED, 'Unauthorized')); - } - - const refreshCount = await incrementRefreshRate(payload.merchantId); - if (refreshCount > REFRESH_RATE_LIMIT_MAX) { - return reply - .header('Retry-After', String(REFRESH_RATE_LIMIT_SECONDS)) - .code(429) - .send(createErrorResponse(ErrorCodes.RATE_LIMITED, 'Too many token refresh requests')); - } - - await revokeJti(payload.jti, remainingLifetime); - await recordAuthIpSuccess(request); - - return reply.send({ token: signMerchantJwt(payload.merchantId, payload.ownerId) }); -}); - -fastify.post<{ Body: WalletVerifyRouteBody }>('/api/auth/wallet/verify', { - preHandler: [enforceAuthIpReputation], - config: { rateLimit: { max: 30, timeWindow: '1 minute' } } -}, async (request, reply) => { - let d; - try { - d = WalletVerifyBody.parse(request.body); - } catch (err) { - await recordAuthIpFailure(request); - throw err; - } - - if (await isNonceUsed(d.nonce)) { - await recordAuthIpFailure(request); - return reply - .code(409) - .send(createErrorResponse(ErrorCodes.INVALID_REQUEST, 'Nonce has already been used')); - } - - if (!verifyWalletSignature(d.address, walletChallenge(d), d.signature)) { - await recordAuthIpFailure(request); - return reply.code(401).send(createErrorResponse(ErrorCodes.UNAUTHORIZED, 'Invalid wallet signature')); - } - - if (!await markNonceUsed(d.nonce)) { - await recordAuthIpFailure(request); - return reply - .code(409) - .send(createErrorResponse(ErrorCodes.INVALID_REQUEST, 'Nonce has already been used')); - } - - await recordAuthIpSuccess(request); - - const merchant = await prisma.merchant.findFirst({ - where: { - deletedAt: null, - OR: [{ id: d.address }, { ownerId: d.address }], +fastify.post<{ Body: z.infer }>( + '/api/settlements/bulk', + { + config: { + rateLimit: { + max: 30, + timeWindow: 60 * 1000, + }, }, - }); - - const response: Record = { success: true, address: d.address }; - if (merchant) { - response.token = signMerchantJwt(merchant.id, merchant.ownerId); - } - - return reply.send(response); -}); + }, + async (request, reply) => { + const d = BulkSettlementBody.parse(request.body); -fastify.get('/api/admin/auth/ip-score', { - preValidation: [fastify.authenticate] -}, async (request, reply) => { - const payload = request.user as MerchantJwtPayload; - if (payload.merchantId !== env.ADMIN_ADDRESS) { - return reply.code(403).send(createErrorResponse(ErrorCodes.FORBIDDEN, 'Forbidden')); - } - - const { ip } = AuthIpScoreQuery.parse(request.query ?? {}); - return { ip, score: await getAuthIpScore(ip) }; -}); - -// Merchants -fastify.post<{ Body: z.infer }>('/api/merchants', { - preValidation: [fastify.authenticate], - preHandler: [logRequestBody], - config: { rateLimit: { max: 30, timeWindow: '1 minute' } } -}, async (request, reply) => { - const d = CreateMerchantBody.parse(request.body); - const secret = d.secret || crypto.randomBytes(24).toString('hex'); - const secretHash = encryptField(hashSecret(secret)); - const merchant = await prisma.$transaction(async (tx) => { - const created = await tx.merchant.create({ - data: { - id: d.id, - name: d.name, - ownerId: d.ownerId, - settings: d.settings as any ?? {}, - secretHash, - } - }); - await logAuditEvent('merchant.created', 'merchant', created.id, { before: null, after: created }, request, tx as unknown as Parameters[5]); - return created; - }); - if (!d.secret) { - fastify.log.warn({ merchantId: merchant.id }, 'Auto-generated merchant secret returned in response. This will only be shown once.'); + if (d.settlements.length > 100) { + return reply.code(400).send(createErrorResponse(ErrorCodes.VALIDATION_ERROR, 'Batch size exceeds maximum limit of 100 settlements')); } - const { secretHash: _hash, ...safeMerchant } = merchant; - return reply.code(201).send({ data: { merchant: safeMerchant, secret } }); -}); - -fastify.get<{ Params: { id: string } }>('/api/merchants/:id', { - preValidation: [fastify.authenticate] -}, async (request, reply): Promise> => { - const { id } = request.params; - const merchant = await prisma.merchant.findFirst({ - where: { id, deletedAt: null }, - }); - if (!merchant) { - reply.code(404); - return { error: createErrorResponse(ErrorCodes.NOT_FOUND, 'Merchant not found') }; - } - return { data: merchant }; -}); - -fastify.delete<{ Params: { id: string } }>('/api/merchants/:id', { - preValidation: [fastify.authenticate], - config: { rateLimit: { max: 30, timeWindow: '1 minute' } } -}, async (request, reply) => { - const { id } = request.params; - const merchant = await prisma.merchant.findFirst({ - where: { id, deletedAt: null }, - }); - if (!merchant) return reply.code(404).send(createErrorResponse(ErrorCodes.NOT_FOUND, 'Merchant not found')); - const deviceInfo = `${request.ip || "unknown"} ${request.headers["user-agent"] ?? "unknown"}`; - const jti = await createAuthSession(merchant.id, deviceInfo); - const token = fastify.jwt.sign( - { merchantId: merchant.id, ownerId: merchant.ownerId }, - { jwtid: jti }, - ); - return reply.send({ token }); - }, - ); - - const walletChallenges = new Map< - string, - { challenge: string; expiresAt: number } - >(); - - interface WalletChallengeRouteBody { - address?: unknown; - } - - const WalletChallengeBody = z.object({ - address: z.string().min(1, "address is required"), - }); - fastify.post<{ Body: WalletChallengeRouteBody }>( - "/api/auth/challenge", - async (request, reply) => { - const d = WalletChallengeBody.parse(request.body); - const challenge = crypto.randomBytes(32).toString("hex"); - const expiresAt = Date.now() + 5 * 60 * 1000; // 5 mins - walletChallenges.set(d.address, { challenge, expiresAt }); - return reply.send({ - challenge, - expiresAt: new Date(expiresAt).toISOString(), - }); - }, - ); - - interface WalletVerifyRouteBody { - address?: unknown; - signature?: unknown; - } - - const LegacyWalletVerifyBody = z.object({ - address: z.string().min(1, "address is required"), - signature: z.string().min(1, "signature is required"), - }); + const merchant = await prisma.merchant.findUnique({ where: { id: d.merchantId } }); + if (!merchant) { + return reply.code(404).send(createErrorResponse(ErrorCodes.NOT_FOUND, 'Merchant not found')); + } - fastify.post<{ Body: WalletVerifyRouteBody }>( - "/api/auth/verify", - async (request, reply) => { - const d = LegacyWalletVerifyBody.parse(request.body); - const challengeInfo = walletChallenges.get(d.address); - - if (!challengeInfo) { - return reply - .code(400) - .send( - createErrorResponse( - ErrorCodes.INVALID_REQUEST, - "Challenge not found or expired", - ), - ); + const settings = merchant.settings as { + webhookUrl?: string; + minSettlementAmount?: string; + maxSettlementAmount?: string; + dailySettlementLimit?: string; + } | null | undefined; + + const parsedFeeRule = FeeRule.passthrough().safeParse(merchant?.settings); + const feeBps = parsedFeeRule.success ? parsedFeeRule.data.feeBps : env.FEES_DEFAULT_BPS; + const settings_data = parsedFeeRule.success ? (parsedFeeRule.data as Record) : {}; + const maxFeeBps = settings_data.maxFeeBps as number | undefined; + const maxFeeThreshold = settings_data.maxFeeThreshold as string | undefined; + const webhookUrl = settings_data.webhookUrl as string ?? null; + + // Fetch monthly volume for volume-based fee discount (#323). + const monthlyVolume = await getMonthlyVolume(d.merchantId); + const discountTiers: DiscountTier[] = env.FEE_DISCOUNT_TIERS ?? []; + + // Fetch current daily total + const todayStart = new Date(); + todayStart.setHours(0, 0, 0, 0); + + const aggregateResult = await prisma.$queryRaw<[{ sum: string | null }]>` + SELECT COALESCE(SUM(CAST("totalAmount" AS DECIMAL)), 0)::text as sum + FROM "Settlement" + WHERE "merchantId" = ${d.merchantId} + AND "initiatedAt" >= ${todayStart} + `; + + const currentDailyTotal = aggregateResult?.[0]?.sum ? parseFloat(aggregateResult[0].sum) : 0; + + let runningBatchTotal = 0; + const validItems: Array<{ amount: string; asset: string; id: string; grossAmount: string; feeAmount: string; netAmount: string }> = []; + const errors: Array<{ index: number; reason: string }> = []; + + for (let i = 0; i < d.settlements.length; i++) { + const item = d.settlements[i]; + const amount = parseFloat(item.amount); + + if (isNaN(amount) || amount <= 0) { + errors.push({ index: i, reason: 'amount must be greater than zero' }); + continue; } - if (Date.now() > challengeInfo.expiresAt) { - walletChallenges.delete(d.address); - return reply - .code(400) - .send( - createErrorResponse( - ErrorCodes.INVALID_REQUEST, - "Challenge expired", - ), - ); - } - - try { - const keypair = Keypair.fromPublicKey(d.address); - const isValid = keypair.verify( - Buffer.from(challengeInfo.challenge), - Buffer.from(d.signature, "hex"), - ); - if (!isValid) { - return reply - .code(401) - .send( - createErrorResponse(ErrorCodes.UNAUTHORIZED, "Invalid signature"), - ); + // Check min/max amount limits + if (settings?.minSettlementAmount) { + const minAmount = parseFloat(settings.minSettlementAmount); + if (amount < minAmount) { + errors.push({ + index: i, + reason: `Settlement amount ${item.amount} is below minimum ${settings.minSettlementAmount}` + }); + continue; } - } catch (err) { - return reply - .code(401) - .send( - createErrorResponse(ErrorCodes.UNAUTHORIZED, "Invalid signature"), - ); } - walletChallenges.delete(d.address); - - let merchant; - try { - merchant = await prisma.merchant.upsert({ - where: { id: d.address }, - update: {}, - create: { - id: d.address, - name: `Merchant ${d.address.substring(0, 6)}`, - ownerId: `owner-${d.address.substring(0, 6)}`, - settings: {}, - }, - }); - } catch (err: any) { - if (err.code === "P2002") { - merchant = await prisma.merchant.findUnique({ - where: { id: d.address }, + if (settings?.maxSettlementAmount) { + const maxAmount = parseFloat(settings.maxSettlementAmount); + if (amount > maxAmount) { + errors.push({ + index: i, + reason: `Settlement amount ${item.amount} exceeds maximum ${settings.maxSettlementAmount}` }); - } else { - throw err; + continue; } } - if (!merchant) { - return reply - .code(500) - .send( - createErrorResponse( - ErrorCodes.INTERNAL_ERROR, - "Failed to upsert merchant", - ), - ); + // Check daily settlement limits + if (settings?.dailySettlementLimit) { + const dailyLimit = parseFloat(settings.dailySettlementLimit); + if (currentDailyTotal + runningBatchTotal + amount > dailyLimit) { + errors.push({ + index: i, + reason: `Daily settlement limit exceeded. Current: ${currentDailyTotal + runningBatchTotal}, Requested: ${amount}, Limit: ${settings.dailySettlementLimit}` + }); + continue; + } } - const token = fastify.jwt.sign({ - merchantId: merchant.id, - ownerId: merchant.ownerId, - }); - return reply.send({ token }); - }, - ); - - interface GoogleAuthRouteBody { - token?: unknown; - } - - const GoogleAuthBody = z.object({ - token: z.string().min(1, "token is required"), - }); - - fastify.post<{ Body: GoogleAuthRouteBody }>( - "/api/auth/google", - async (request, reply) => { - const d = GoogleAuthBody.parse(request.body); - + let itemResult; try { - const client = new OAuth2Client(); - const ticket = await client.verifyIdToken({ - idToken: d.token, - audience: process.env.GOOGLE_CLIENT_ID, - }); - const payload = ticket.getPayload(); - if (!payload) { - return reply - .code(401) - .send( - createErrorResponse( - ErrorCodes.UNAUTHORIZED, - "Google token verification failed: invalid token payload", - ), - ); - } - const email = payload.email; - if (!email) { - return reply - .code(400) - .send( - createErrorResponse( - ErrorCodes.INVALID_REQUEST, - "Email missing in Google token payload", - ), - ); - } - - const lockoutCount = await getGoogleAuthLockoutCount(email); - if (lockoutCount >= env.AUTH_MAX_FAILED_ATTEMPTS) { - request.log.warn( - { email, lockoutCount }, - "[Auth] Google OAuth locked out due to too many failed attempts", - ); - return reply - .code(429) - .send( - createErrorResponse( - ErrorCodes.UNAUTHORIZED, - "Too many failed attempts. Try again later.", - ), - ); - } - - if (env.ALLOWED_EMAIL_DOMAINS.length > 0) { - const domain = email.split("@")[1]?.toLowerCase(); - if (!domain || !env.ALLOWED_EMAIL_DOMAINS.includes(domain)) { - await incrementGoogleAuthLockout(email); - request.log.info( - { email, domain }, - "[Auth] Google OAuth rejected: email domain not allowed", - ); - return reply - .code(403) - .send( - createErrorResponse( - ErrorCodes.INVALID_ORIGIN, - "Email domain not allowed", - { domain }, - ), - ); - } - } - - const tokenExpired = - typeof payload.exp === "number" && Date.now() / 1000 > payload.exp; - const tokenAgeMs = tokenExpired ? Date.now() - payload.exp * 1000 : 0; - if (tokenExpired && tokenAgeMs > GOOGLE_AUTH_GRACE_PERIOD_MS) { - await incrementGoogleAuthLockout(email); - request.log.warn( - { email, expiredMs: tokenAgeMs }, - "[Auth] Google OAuth rejected: token expired", - ); - return reply - .code(401) - .send( - createErrorResponse( - ErrorCodes.UNAUTHORIZED, - "Google token expired", - ), - ); - } - - if (tokenExpired) { - request.log.warn( - { email, expiredMs: tokenAgeMs }, - "[Auth] Google OAuth accepted with expired token within grace period", - ); + itemResult = computeSettlementAmounts(item.amount, feeBps, monthlyVolume, discountTiers); + } catch (error) { + if (error instanceof SettlementAmountError) { + errors.push({ index: i, reason: error.message }); + continue; } + throw error; + } + const { grossAmount, feeAmount, netAmount } = itemResult; + const settlementId = 'set_' + crypto.randomUUID().replace(/-/g, ''); + + validItems.push({ + id: settlementId, + amount: item.amount, + asset: item.asset, + grossAmount, + feeAmount, + netAmount + }); + runningBatchTotal += amount; + } - request.log.info({ email }, "[Auth] Google OAuth accepted"); - - await resetGoogleAuthLockout(email); + const batchId = 'batch_' + crypto.randomUUID().replace(/-/g, ''); - let merchant = await prisma.merchant.findFirst({ - where: { ownerId: email, deletedAt: null }, - }); - if (!merchant) { - const merchantId = `google_${crypto.randomBytes(8).toString("hex")}`; - merchant = await prisma.merchant.create({ + if (validItems.length > 0) { + await prisma.$transaction(async (tx) => { + for (const item of validItems) { + await tx.settlement.create({ data: { - id: merchantId, - name: email.split("@")[0] + " Merchant", - ownerId: email, - settings: {}, + id: item.id, + merchantId: d.merchantId, + totalAmount: item.grossAmount, + grossAmount: item.grossAmount, + feeAmount: item.feeAmount, + netAmount: item.netAmount, + feeBps, + asset: item.asset, + status: 'pending', + webhookUrl, + batchId, }, }); } - - const deviceInfo = `${request.ip || "unknown"} ${request.headers["user-agent"] ?? "unknown"}`; - const jti = await createAuthSession(merchant.id, deviceInfo); - const jwtToken = fastify.jwt.sign( - { merchantId: merchant.id, ownerId: merchant.ownerId }, - { jwtid: jti }, - ); - return reply.send({ token: jwtToken }); - } catch (err: any) { - request.log.error({ err }, "[Auth] Google OAuth failed"); - return reply - .code(401) - .send( - createErrorResponse( - ErrorCodes.UNAUTHORIZED, - "Google token verification failed", - ), - ); - } - }, - ); - - fastify.get( - "/api/auth/sessions", - { - preValidation: [fastify.authenticate], - }, - async (request, reply) => { - const merchantId = (request.user as any)?.merchantId; - if (!merchantId) { - return reply - .code(401) - .send(createErrorResponse(ErrorCodes.UNAUTHORIZED, "Unauthorized")); - } - - const sessions = await listAuthSessions(merchantId); - return reply.send({ data: { sessions } }); - }, - ); - - fastify.delete<{ Params: { jti: string } }>( - "/api/auth/sessions/:jti", - { - preValidation: [fastify.authenticate], - }, - async (request, reply) => { - const merchantId = (request.user as any)?.merchantId; - const { jti } = request.params; - - if (!merchantId) { - return reply - .code(401) - .send(createErrorResponse(ErrorCodes.UNAUTHORIZED, "Unauthorized")); - } - - const revoked = await revokeAuthSession(merchantId, jti); - if (!revoked) { - return reply - .code(404) - .send(createErrorResponse(ErrorCodes.NOT_FOUND, "Session not found")); - } - - return reply.send({ status: "revoked" }); - }, - ); - - // Merchants - fastify.post<{ Body: z.infer }>( - "/api/merchants", - { - preValidation: [fastify.authenticate], - preHandler: [logRequestBody], - config: { rateLimit: { max: 30, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - const d = CreateMerchantBody.parse(request.body); - const secret = d.secret || crypto.randomBytes(24).toString("hex"); - const secretHash = encryptField(hashSecret(secret)); - const merchant = await prisma.$transaction(async (tx) => { - const created = await tx.merchant.create({ - data: { - id: d.id, - name: d.name, - ownerId: d.ownerId, - settings: (d.settings as any) ?? {}, - secretHash, - }, - }); - await logAuditEvent( - "merchant.created", - "merchant", - created.id, - { before: null, after: created }, - request, - tx as unknown as Parameters[5], - ); - return created; }); - if (!d.secret) { - fastify.log.warn( - { merchantId: merchant.id }, - "Auto-generated merchant secret returned in response. This will only be shown once.", - ); - } - const { secretHash: _hash, ...safeMerchant } = merchant; - return reply.code(201).send({ data: { merchant: safeMerchant, secret } }); - }, - ); - fastify.get<{ Params: { id: string } }>( - "/api/merchants/:id", - { - preValidation: [fastify.authenticate], - }, - async (request, reply): Promise> => { - const { id } = request.params; - const merchant = await prisma.merchant.findFirst({ - where: { id, deletedAt: null }, - }); - if (!merchant) { - reply.code(404); - return { - error: createErrorResponse( - ErrorCodes.NOT_FOUND, - "Merchant not found", - ), + // Enqueue job for each successfully created settlement record + for (const item of validItems) { + const jobData: SettlementJobData = { + id: item.id, + merchantId: d.merchantId, + grossAmount: item.grossAmount, + asset: item.asset, }; - } - return { data: merchant }; - }, - ); - - fastify.delete<{ Params: { id: string } }>( - "/api/merchants/:id", - { - preValidation: [fastify.authenticate], - config: { rateLimit: { max: 30, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - const { id } = request.params; - const merchant = await prisma.merchant.findFirst({ - where: { id, deletedAt: null }, - }); - if (!merchant) - return reply - .code(404) - .send( - createErrorResponse(ErrorCodes.NOT_FOUND, "Merchant not found"), - ); - - await prisma.$transaction(async (tx) => { - const updated = await tx.merchant.update({ - where: { id }, - data: { deletedAt: new Date() }, - }); - await logAuditEvent( - "merchant.deleted", - "merchant", - updated.id, - { before: merchant, after: updated }, - request, - tx as unknown as Parameters[5], - ); - }); - - return reply.code(200).send({ success: true }); - }, - ); - - fastify.post<{ Params: { id: string } }>( - "/api/merchants/:id/restore", - { - preValidation: [fastify.authenticate], - config: { rateLimit: { max: 30, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - const { id } = request.params; - const merchant = await prisma.merchant.findUnique({ where: { id } }); - if (!merchant) - return reply - .code(404) - .send( - createErrorResponse(ErrorCodes.NOT_FOUND, "Merchant not found"), - ); - if (!merchant.deletedAt) { - return reply - .code(400) - .send( - createErrorResponse( - ErrorCodes.INVALID_REQUEST, - "Merchant is not soft-deleted", - ), - ); - } - - const restored = await prisma.merchant.update({ - where: { id }, - data: { deletedAt: null }, - }); - - return reply.code(200).send({ success: true, merchant: restored }); - }, - ); - - // Update per-merchant settings (fee rules, tier). Merges into existing settings so - // a partial update does not wipe unrelated keys. The settlement engine reads - // settings.feeBps from here when computing fees. - fastify.patch<{ - Params: { id: string }; - Body: z.infer; - }>( - "/api/merchants/:id/settings", - { - preValidation: [fastify.authenticate], - preHandler: [logRequestBody], - config: { rateLimit: { max: 30, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - const d = UpdateMerchantSettingsBody.parse(request.body); - - // Reject attempts to set kycStatus via the merchant settings endpoint - if ("kycStatus" in (request.body as Record)) { - return reply - .code(403) - .send( - createErrorResponse( - ErrorCodes.UNAUTHORIZED, - "kycStatus cannot be updated via this endpoint", - ), - ); - } - - const { id } = request.params; - const merchant = await prisma.merchant.findFirst({ - where: { id, deletedAt: null }, - }); - if (!merchant) - return reply - .code(404) - .send( - createErrorResponse(ErrorCodes.NOT_FOUND, "Merchant not found"), - ); - - const currentSettings = (merchant.settings ?? {}) as Record< - string, - unknown - >; - const nextSettings = { ...currentSettings, ...d }; - - const updated = await prisma.$transaction(async (tx) => { - const merchantUpdate = await tx.merchant.update({ - where: { id }, - data: { settings: nextSettings as object }, - }); - await logAuditEvent( - "merchant.updated", - "merchant", - merchantUpdate.id, - { before: merchant, after: merchantUpdate }, - request, - tx as unknown as Parameters[5], - ); - return merchantUpdate; - }); - - return reply.code(200).send({ data: { merchant: updated } }); - }, - ); - - // Admin-only: update merchant KYC status - fastify.patch<{ - Params: { id: string }; - Body: z.infer; - }>( - "/api/admin/merchants/:id/kyc", - { - preValidation: [fastify.serviceAuth], - preHandler: [logRequestBody], - config: { rateLimit: { max: 30, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - const d = UpdateMerchantKycBody.parse(request.body); - const { id } = request.params; - - const merchant = await prisma.merchant.findFirst({ - where: { id, deletedAt: null }, - }); - if (!merchant) - return reply - .code(404) - .send( - createErrorResponse(ErrorCodes.NOT_FOUND, "Merchant not found"), - ); - - const updated = await prisma.$transaction(async (tx) => { - const merchantUpdate = await tx.merchant.update({ - where: { id }, - data: { kycStatus: d.kycStatus }, + await settlementQueue.add('process-settlement', jobData).catch((err) => { + request.log.error({ err, settlementId: item.id }, 'Failed to enqueue bulk settlement job'); }); - await logAuditEvent( - "merchant.kyc.updated", - "merchant", - merchantUpdate.id, - { - before: { kycStatus: merchant.kycStatus }, - after: { kycStatus: merchantUpdate.kycStatus }, - }, - request, - tx as unknown as Parameters[5], - ); - return merchantUpdate; - }); - - return reply.code(200).send({ data: { merchant: updated } }); - }, - ); - - // Payments - fastify.post<{ Body: z.infer }>( - "/api/payments", - { - preValidation: [fastify.authenticate], - preHandler: [logRequestBody], - config: { rateLimit: { max: 300, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - // ── 1. Parse and validate request body ────────────────────────────────────── - const d = CreatePaymentBody.parse(request.body); - - // ── 2. Read and validate optional Idempotency-Key header ──────────────────── - const idempotencyKey = readIdempotencyKey(request); - - if ( - idempotencyKey !== null && - idempotencyKey.length > IDEMPOTENCY_KEY_MAX_LEN - ) { - return reply - .code(400) - .send( - createErrorResponse( - ErrorCodes.VALIDATION_ERROR, - "Idempotency-Key must not exceed 255 characters", - ), - ); - } - - // ── 3. Idempotency check: look for a non-expired record with the same key ─── - if (idempotencyKey !== null) { - const now = new Date(); - const existing = await prisma.payment.findFirst({ - where: { - idempotencyKey, - idempotencyKeyExpiresAt: { gt: now }, - }, - }); - - if (existing) { - request.log.info( - { idempotencyKey, paymentId: existing.id }, - "Idempotency hit — returning cached payment", - ); - return reply.code(200).send({ data: existing }); - } - } - - // ── 4. Create the payment (with idempotency fields when a key was supplied) ── - const idempotencyKeyExpiresAt = idempotencyKey - ? new Date(Date.now() + IDEMPOTENCY_TTL_MS) - : null; - - let fxQuote: Awaited> = null; - if (d.convertTo) { - try { - fxQuote = await fxClient.getQuote( - { from: d.asset, to: d.convertTo, amount: d.amount }, - request.headers, - ); - } catch (err) { - if (err instanceof UpstreamReadTimeoutError) { - request.log.warn( - { service: err.service, endpoint: err.endpoint }, - "fx-service read timeout — no cached quote available, returning 503", - ); - return reply - .code(503) - .header("Retry-After", "5") - .send( - createErrorResponse( - ErrorCodes.GATEWAY_TIMEOUT, - "FX service temporarily unavailable, please retry", - ), - ); - } - throw err; - } - } - - const payment = await prisma.$transaction(async (tx) => { - const created = await tx.payment.create({ - data: { - id: "pay_" + crypto.randomUUID().replace(/-/g, ""), - merchantId: d.merchantId, - payerId: d.payerId, - amount: d.amount, - asset: d.asset, - reference: d.reference, - status: "initiated", - idempotencyKey: idempotencyKey ?? undefined, - idempotencyKeyExpiresAt: idempotencyKeyExpiresAt ?? undefined, - }, - }); - await logAuditEvent( - "payment.created", - "payment", - created.id, - { before: null, after: created }, - request, - tx as unknown as Parameters[5], - ); - return created; - }); - - request.log.info( - { idempotencyKey, paymentId: payment.id }, - idempotencyKey - ? "Idempotency miss — payment created" - : "Payment created (no idempotency key)", - ); - - if (d.convertTo) { - return reply.code(201).send({ data: { ...payment, fxQuote } }); } - - return reply.code(201).send({ data: payment }); - }, - ); - - fastify.get<{ - Params: { id: string }; - Querystring: { includeEvents?: string }; - }>("/api/payments/:id", async (request, reply) => { - const { id } = request.params; - const payment = await prisma.payment.findUnique({ where: { id } }); - if (!payment) - return reply - .code(404) - .send(createErrorResponse(ErrorCodes.NOT_FOUND, "Payment not found")); - - // Optional on-chain event enrichment (?includeEvents=true). The indexer is an - // enrichment source only: if it is unavailable, `events` is null and the - // payment is still returned so the endpoint never fails on indexer issues. - if (request.query.includeEvents === "true") { - // Forward tracing headers so the indexer call is part of the same trace (#118). - const events = await indexerClient.getPaymentEvents( - payment.merchantId, - request.headers, - ); - return { data: { ...payment, events } }; } - return { data: payment }; - }); - - // Enforce valid status transitions. The DB enum and Prisma allow any status, so - // this route is the single place that guards the payment state machine. - fastify.patch<{ - Params: { id: string }; - Body: z.infer; - }>( - "/api/payments/:id/status", - { - preValidation: [fastify.authenticate], - preHandler: [logRequestBody], - config: { rateLimit: { max: 300, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - const d = UpdatePaymentStatusBody.parse(request.body); - - const { id } = request.params; - const payment = await prisma.payment.findUnique({ where: { id } }); - if (!payment) - return reply - .code(404) - .send(createErrorResponse(ErrorCodes.NOT_FOUND, "Payment not found")); - - const allowed = PAYMENT_STATUS_TRANSITIONS[payment.status] ?? []; - if ( - !isValidTransition(PAYMENT_STATUS_TRANSITIONS, payment.status, d.status) - ) { - return reply.code(422).send( - createErrorResponse( - ErrorCodes.VALIDATION_ERROR, - "Invalid status transition", - { - from: payment.status, - to: d.status, - allowedTransitions: allowed, - }, - ), - ); - } - - const updated = await prisma.$transaction(async (tx) => { - const paymentUpdate = await tx.payment.update({ - where: { id }, - data: { status: d.status }, - }); - await logAuditEvent( - "payment.status.changed", - "payment", - paymentUpdate.id, - { before: payment, after: paymentUpdate }, - request, - tx as unknown as Parameters[5], - ); - return paymentUpdate; - }); - return reply.send({ data: updated }); - }, - ); - - // Bulk-cancel initiated payments belonging to the authenticated merchant. - fastify.post<{ Body: z.infer }>( - "/api/payments/bulk-cancel", - { - preValidation: [fastify.authenticate], - preHandler: [logRequestBody], - config: { rateLimit: { max: 30, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - const d = BulkCancelPaymentsBody.parse(request.body); - const merchantId = (request.user as any).merchantId as string; - - // Deduplicate IDs - const uniqueIds = [...new Set(d.paymentIds)]; - - const payments = await prisma.payment.findMany({ - where: { id: { in: uniqueIds } }, - }); - - const paymentMap = new Map(payments.map((p) => [p.id, p])); - - const cancelledIds: string[] = []; - const skippedIds: string[] = []; - const errors: { id: string; reason: string }[] = []; - - for (const id of uniqueIds) { - const payment = paymentMap.get(id); - if (!payment) { - skippedIds.push(id); - continue; - } - if (payment.merchantId !== merchantId) { - skippedIds.push(id); - continue; - } - if (payment.status !== "initiated") { - skippedIds.push(id); - continue; - } - cancelledIds.push(id); - } - - if (cancelledIds.length > 0) { - await prisma.$transaction(async (tx) => { - for (const id of cancelledIds) { - const before = paymentMap.get(id)!; - const updated = await tx.payment.update({ - where: { id }, - data: { status: "cancelled" }, - }); - await logAuditEvent( - "payment.status.changed", - "payment", - id, - { before, after: updated }, - request, - tx as unknown as Parameters[5], - ); - } - }); - } - - return reply.code(200).send({ - cancelled: cancelledIds.length, - skipped: skippedIds.length, - errors: errors.length, - cancelledIds, - skippedIds, - }); - }, - ); - - fastify.patch<{ - Params: { id: string }; - Body: z.infer; - }>( - "/api/settlements/:id/status", - { - preValidation: [fastify.authenticate], - preHandler: [logRequestBody], - config: { rateLimit: { max: 30, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - let d; - try { - d = UpdateSettlementStatusBody.parse(request.body); - } catch (error) { - return reply - .code(400) - .send( - createErrorResponse( - ErrorCodes.VALIDATION_ERROR, - "Invalid request body", - error, - ), - ); - } - - const { id } = request.params; - const settlement = await prisma.settlement.findUnique({ where: { id } }); - if (!settlement) - return reply - .code(404) - .send( - createErrorResponse(ErrorCodes.NOT_FOUND, "Settlement not found"), - ); - - const allowed = SETTLEMENT_STATUS_TRANSITIONS[settlement.status] ?? []; - if ( - !isValidTransition( - SETTLEMENT_STATUS_TRANSITIONS, - settlement.status, - d.status, - ) - ) { - return reply.code(422).send( - createErrorResponse( - ErrorCodes.VALIDATION_ERROR, - "Invalid status transition", - { - from: settlement.status, - to: d.status, - allowedTransitions: allowed, - }, - ), - ); - } - - const updated = await prisma.$transaction(async (tx) => { - const settlementUpdate = await tx.settlement.update({ - where: { id }, - data: { - status: d.status, - ...(d.status === "completed" || d.status === "failed" - ? { completedAt: new Date() } - : {}), - }, - }); - await logAuditEvent( - "settlement.status.changed", - "settlement", - settlementUpdate.id, - { before: settlement, after: settlementUpdate }, - request, - tx as unknown as Parameters[5], - ); - return settlementUpdate; - }); - return reply.send({ data: updated }); - }, - ); - - // Settlements - // - // Authorization: service-to-service callers (x-service-token) may pass - // merchantId to filter across merchants. Merchant-authenticated callers - // (JWT) are always scoped to their own merchantId — the query parameter is - // ignored for them so one merchant can never read another's settlements. - fastify.get<{ - Querystring: z.infer & { merchantId?: string }; - }>( - "/api/settlements", - { - preValidation: async (request: FastifyRequest, reply: FastifyReply) => { - if (request.headers["x-service-token"]) { - await fastify.serviceAuth(request, reply); - return; - } - await fastify.authenticate(request, reply); + return reply.code(201).send({ + data: { + batchId, + total: d.settlements.length, + created: validItems.length, + errors, }, - config: { rateLimit: { max: 100, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - const query = SettlementListQuery.parse(request.query); - const { status, from, to, limit, page } = query; - const requestedMerchantId = (request.query as { merchantId?: string }) - .merchantId; - - const isServiceAuth = Boolean(request.headers["x-service-token"]); - const scopedMerchantId = isServiceAuth - ? requestedMerchantId - : (request.user as { merchantId?: string } | undefined)?.merchantId; - - const { startDate, endDate, includeDeleted } = query as any; - const where: any = {}; - if (scopedMerchantId) { - where.merchantId = scopedMerchantId; - } - if (status) { - where.status = status; - } - const effectiveFrom = startDate ?? from; - const effectiveTo = endDate ?? to; - if (effectiveFrom || effectiveTo) { - where.initiatedAt = {}; - if (effectiveFrom) { - where.initiatedAt.gte = new Date(effectiveFrom); - } - if (effectiveTo) { - where.initiatedAt.lte = new Date(effectiveTo); - } - } - - // When includeDeleted is false, exclude settlements belonging to soft-deleted merchants. - // Service-auth callers (preValidation: [fastify.serviceAuth]) always see everything. - if (!includeDeleted) { - where.merchant = { deletedAt: null }; - } + }); + } +); - const [records, total] = await Promise.all([ - prisma.settlement.findMany({ - where, - orderBy: { initiatedAt: "desc" }, - take: limit, - skip: (page - 1) * limit, - }), - prisma.settlement.count({ where }), - ]); - - return { - data: records, - pagination: buildPaginationMeta(page, limit, total), - }; - }, - ); - - fastify.post<{ Body: z.infer }>( - "/api/settlements", - { - preValidation: [fastify.authenticate], - preHandler: [logRequestBody], - config: { rateLimit: { max: 30, timeWindow: "1 minute" } }, +fastify.get<{ Params: { batchId: string } }>( + '/api/settlements/batch/:batchId/status', + { + config: { + rateLimit: { + max: 60, + timeWindow: 60 * 1000, + }, }, - async (request, reply) => { - const d = CreateSettlementBody.parse(request.body); - const merchant = await prisma.merchant.findUnique({ - where: { id: d.merchantId }, - }); - - if (!merchant) { - return reply - .code(404) - .send( - createErrorResponse(ErrorCodes.NOT_FOUND, "Merchant not found"), - ); - } - - const settings = merchant.settings as - | { - webhookUrl?: string; - minSettlementAmount?: string; - maxSettlementAmount?: string; - dailySettlementLimit?: string; - } - | null - | undefined; - - // Normalize to items array (backward compatibility: single amount/asset becomes single-item batch) - const items = - d.items || - (d.amount && d.asset ? [{ amount: d.amount, asset: d.asset }] : []); - - // #319 — Validate each asset against SupportedAsset table - for (const item of items) { - const supportedAsset = await prisma.supportedAsset.findUnique({ - where: { code: item.asset }, - }); + }, + async (request, reply) => { + const { batchId } = request.params; - if (!supportedAsset || !supportedAsset.isActive) { - return reply - .code(422) - .send( - createErrorResponse( - ErrorCodes.VALIDATION_ERROR, - `Asset ${item.asset} is not supported`, - { asset: item.asset }, - ), - ); - } - } - - // Validate each settlement item against merchant limits - for (const item of items) { - const amount = parseFloat(item.amount); - - // Check minimum settlement amount - if (settings?.minSettlementAmount) { - const minAmount = parseFloat(settings.minSettlementAmount); - if (amount < minAmount) { - return reply.code(422).send( - createErrorResponse( - ErrorCodes.VALIDATION_ERROR, - `Settlement amount ${item.amount} is below minimum ${settings.minSettlementAmount}`, - { - amount: item.amount, - minSettlementAmount: settings.minSettlementAmount, - }, - ), - ); - } - } - - // Check maximum settlement amount - if (settings?.maxSettlementAmount) { - const maxAmount = parseFloat(settings.maxSettlementAmount); - if (amount > maxAmount) { - return reply.code(422).send( - createErrorResponse( - ErrorCodes.VALIDATION_ERROR, - `Settlement amount ${item.amount} exceeds maximum ${settings.maxSettlementAmount}`, - { - amount: item.amount, - maxSettlementAmount: settings.maxSettlementAmount, - }, - ), - ); - } - } - } - - // Check daily settlement limit (aggregate all assets) - if (settings?.dailySettlementLimit) { - const todayStart = new Date(); - todayStart.setHours(0, 0, 0, 0); - const startTimeMs = Date.now(); - - const aggregateResult = await prisma.$queryRaw< - [{ sum: string | null }] - >` - SELECT COALESCE(SUM(CAST("totalAmount" AS DECIMAL)), 0)::text as sum - FROM "Settlement" - WHERE "merchantId" = ${d.merchantId} - AND "initiatedAt" >= ${todayStart} - `; - - const currentDailyTotal = aggregateResult?.[0]?.sum - ? parseFloat(aggregateResult[0].sum) - : 0; - const queryDurationMs = Date.now() - startTimeMs; - request.log.debug( - { queryDurationMs, merchantId: d.merchantId }, - "Daily settlement aggregate query", - ); - - const requestTotal = items.reduce( - (sum: number, item: any) => sum + parseFloat(item.amount), - 0, - ); - const newDailyTotal = currentDailyTotal + requestTotal; - const dailyLimit = parseFloat(settings.dailySettlementLimit); - - if (newDailyTotal > dailyLimit) { - return reply.code(422).send( - createErrorResponse( - ErrorCodes.VALIDATION_ERROR, - `Daily settlement limit exceeded. Current: ${currentDailyTotal}, Requested: ${requestTotal}, Limit: ${settings.dailySettlementLimit}`, - { - currentDailyTotal: currentDailyTotal.toString(), - requestedAmount: requestTotal.toString(), - dailySettlementLimit: settings.dailySettlementLimit, - }, - ), - ); - } - } + if (!batchId || !batchId.startsWith('batch_')) { + return reply.code(400).send(createErrorResponse(ErrorCodes.VALIDATION_ERROR, 'Invalid batchId format')); + } - try { - const settlementResponse = await settlementClient.createSettlement( - d, - request.headers, - ); - return reply - .code(settlementResponse.status) - .type(settlementResponse.contentType) - .send(settlementResponse.body); - } catch (err) { - if (err instanceof SettlementEngineUnavailableError) { - request.log.warn( - { err }, - "settlement-engine unavailable during settlement creation", - ); - return reply - .code(504) - .send( - createErrorResponse( - ErrorCodes.GATEWAY_TIMEOUT, - "Settlement engine unavailable", - ), - ); - } - throw err; - } - }, - ); + const settlements = await prisma.settlement.findMany({ + where: { batchId }, + }); - fastify.get( - "/api/admin/audit-log", - { - preValidation: [fastify.serviceAuth], - config: { rateLimit: { max: 100, timeWindow: "1 minute" } }, - }, - async (request, reply) => { - const { page, limit } = PaginationQuery.parse(request.query ?? {}); - const query = request.query as Record; - const where: Record = {}; + if (settlements.length === 0) { + return reply.code(404).send(createErrorResponse(ErrorCodes.NOT_FOUND, `Batch ${batchId} not found`)); + } - if (query.entityType) { - where.entityType = query.entityType; - } - if (query.action) { - where.action = query.action; - } - if (query.startDate || query.endDate) { - where.createdAt = {}; - if (query.startDate) { - (where.createdAt as Record).gte = new Date( - query.startDate, - ); - } - if (query.endDate) { - (where.createdAt as Record).lte = new Date( - query.endDate, - ); - } - } + const total = settlements.length; + let pending = 0; + let processing = 0; + let completed = 0; + let failed = 0; + + for (const s of settlements) { + if (s.status === 'pending') pending++; + else if (s.status === 'processing') processing++; + else if (s.status === 'completed') completed++; + else if (s.status === 'failed') failed++; + } - const [rows, total] = await Promise.all([ - prisma.auditLog.findMany({ - where, - orderBy: { createdAt: "desc" }, - take: limit, - skip: (page - 1) * limit, - }), - prisma.auditLog.count({ where }), - ]); - - return reply.send({ - data: rows, - pagination: buildPaginationMeta(page, limit, total), - }); - }, - ); + let overallStatus = 'processing'; + if (completed === total) overallStatus = 'completed'; + else if (failed === total) overallStatus = 'failed'; + else if (pending === total) overallStatus = 'pending'; - fastify.get("/api/deployments", async (request, reply) => { return { data: { - network: env.STELLAR_NETWORK_PASSPHRASE, - contracts: [ - { - name: "Settlement contract", - contractId: env.SETTLEMENT_CONTRACT_ID, - explorerUrl: `https://lab.stellar.org/r/testnet/contract/${env.SETTLEMENT_CONTRACT_ID}`, - }, - { - name: "Governance contract", - contractId: env.GOVERNANCE_CONTRACT_ID, - explorerUrl: `https://lab.stellar.org/r/testnet/contract/${env.GOVERNANCE_CONTRACT_ID}`, - }, - ], - updatedAt: new Date().toISOString(), + batchId, + total, + pending, + processing, + completed, + failed, + status: overallStatus, }, }; - }); - - async function proxyFxUpstream( - request: FastifyRequest, - reply: FastifyReply, - path: string, - ) { - const targetUrl = new URL(path, env.FX_ENGINE_URL).toString(); - - try { - const response = await fetchUpstream(request, targetUrl, {}, request.log); - const body = await response.text(); - const contentType = - response.headers.get("content-type") ?? "application/json"; - return reply.code(response.status).type(contentType).send(body); - } catch (err) { - if (err instanceof UpstreamTimeoutError) { - return reply - .code(504) - .send( - createErrorResponse(ErrorCodes.GATEWAY_TIMEOUT, "Gateway Timeout"), - ); - } - throw err; - } } +); - fastify.get("/api/rates", async (request, reply) => - proxyFxUpstream(request, reply, "/api/rates"), - ); - fastify.get("/api/currencies", async (request, reply) => - proxyFxUpstream(request, reply, "/api/currencies"), - ); +// ============================================================================ +// SETTLEMENT BATCHING JOB (#320) +// ============================================================================ + +// BullMQ repeatable job that runs every BATCH_INTERVAL_SECONDS to batch +// pending settlements by asset. Only creates batches for assets with +// >= BATCH_MIN_COUNT settlements. + +const batchQueue = new Queue('settlement-batching', { + connection: redis, + defaultJobOptions: { + attempts: 3, + backoff: { type: 'exponential', delay: 5000 }, + removeOnComplete: 100, + removeOnFail: 100, + }, +}); - // ============================================================================ - // SUPPORTED ASSETS (#319) - // ============================================================================ +const batchWorker = new Worker( + 'settlement-batching', + async (job) => { + const traceId = job.data.traceId || crypto.randomUUID(); + fastify.log.info({ traceId }, 'Starting settlement batching job'); - // GET /api/assets — list all supported assets - fastify.get("/api/assets", async (request, reply) => { try { - const assets = await prisma.supportedAsset.findMany({ - where: { isActive: true }, - select: { - code: true, - contractId: true, - decimals: true, - name: true, - isActive: true, - }, + // Fetch all pending settlements + const pendingSettlements = await prisma.settlement.findMany({ + where: { status: 'pending' }, }); - return { data: assets }; - } catch (error) { - request.log.error({ error }, "Failed to fetch supported assets"); - return reply - .code(500) - .send( - createErrorResponse( - ErrorCodes.INTERNAL_ERROR, - "Internal server error", - ), - ); - } - }); + if (pendingSettlements.length === 0) { + fastify.log.info({ traceId }, 'No pending settlements to batch'); + return { batched: 0 }; + } - // POST /api/admin/assets — admin endpoint to add new asset - fastify.post( - "/api/admin/assets", - { - preValidation: [fastify.serviceAuth], - schema: { - body: z.object({ - code: z.string().min(1), - contractId: z.string().min(1), - decimals: z.number().int().min(0), - name: z.string().min(1), - isActive: z.boolean().default(true), - }), - }, - }, - async (request, reply) => { - const body = request.body as z.infer; + // Group by asset + const grouped = pendingSettlements.reduce((acc, s) => { + if (!acc[s.asset]) acc[s.asset] = []; + acc[s.asset].push(s); + return acc; + }, {} as Record); + + let batchedCount = 0; + + // Create batches for assets with >= BATCH_MIN_COUNT + for (const [asset, settlements] of Object.entries(grouped)) { + if (settlements.length >= env.BATCH_MIN_COUNT) { + const totalGross = settlements.reduce( + (sum, s) => sum.plus(s.grossAmount), + new BigNumber(0) + ).toString(); + const totalFees = settlements.reduce( + (sum, s) => sum.plus(s.feeAmount), + new BigNumber(0) + ).toString(); + const totalNet = settlements.reduce( + (sum, s) => sum.plus(s.netAmount), + new BigNumber(0) + ).toString(); + + const batch = await prisma.settlementBatch.create({ + data: { + asset, + totalCount: settlements.length, + totalGross, + totalFees, + totalNet, + }, + }); - try { - const asset = await prisma.supportedAsset.create({ - data: body, - }); + // Update settlements with batchId and mark completed + await prisma.settlement.updateMany({ + where: { id: { in: settlements.map((s) => s.id) } }, + data: { batchId: batch.id, status: 'completed' }, + }); - await logAuditEvent( - "asset.created", - "SupportedAsset", - asset.code, - { before: null, after: asset }, - request, - ); - - return reply.code(201).send({ data: asset }); - } catch (error: any) { - if (error.code === "P2002") { - return reply - .code(409) - .send( - createErrorResponse( - ErrorCodes.INVALID_REQUEST, - "Asset code already exists", - ), - ); - } - request.log.error({ error }, "Failed to create supported asset"); - return reply - .code(500) - .send( - createErrorResponse( - ErrorCodes.INTERNAL_ERROR, - "Internal server error", - ), + fastify.log.info( + { traceId, batchId: batch.id, asset, count: settlements.length }, + 'Created settlement batch' ); - } - }, - ); - - // PATCH /api/admin/assets/:code — admin endpoint to update asset - fastify.patch( - "/api/admin/assets/:code", - { - preValidation: [fastify.serviceAuth], - schema: { - params: z.object({ code: z.string().min(1) }), - body: z.object({ - contractId: z.string().min(1).optional(), - decimals: z.number().int().min(0).optional(), - name: z.string().min(1).optional(), - isActive: z.boolean().optional(), - }), - }, - }, - async (request, reply) => { - const { code } = request.params as { code: string }; - const body = request.body as z.infer; - - try { - const asset = await prisma.supportedAsset.update({ - where: { code }, - data: body, - }); - await logAuditEvent( - "asset.updated", - "SupportedAsset", - asset.code, - { before: null, after: asset }, - request, - ); - - return { data: asset }; - } catch (error: any) { - if (error.code === "P2025") { - return reply - .code(404) - .send(createErrorResponse(ErrorCodes.NOT_FOUND, "Asset not found")); - } - request.log.error({ error }, "Failed to update supported asset"); - return reply - .code(500) - .send( - createErrorResponse( - ErrorCodes.INTERNAL_ERROR, - "Internal server error", - ), + batchedCount += settlements.length; + } else { + fastify.log.info( + { traceId, asset, count: settlements.length }, + 'Skipping batch (below min count)' ); - } - }, - ); - - // DELETE /api/admin/assets/:code — admin endpoint to delete asset - fastify.delete( - "/api/admin/assets/:code", - { - preValidation: [fastify.serviceAuth], - schema: { - params: z.object({ code: z.string().min(1) }), - }, - }, - async (request, reply) => { - const { code } = request.params as { code: string }; - - try { - await prisma.supportedAsset.delete({ - where: { code }, - }); - - await logAuditEvent( - "asset.deleted", - "SupportedAsset", - code, - { before: null, after: null }, - request, - ); - - return reply.code(204).send(); - } catch (error: any) { - if (error.code === "P2025") { - return reply - .code(404) - .send(createErrorResponse(ErrorCodes.NOT_FOUND, "Asset not found")); } - request.log.error({ error }, "Failed to delete supported asset"); - return reply - .code(500) - .send( - createErrorResponse( - ErrorCodes.INTERNAL_ERROR, - "Internal server error", - ), - ); } - }, - ); - - fastify.get("/api/quote", async (request, reply) => { - const query = new URLSearchParams( - request.query as Record, - ).toString(); - const path = query ? `/api/quote?${query}` : "/api/quote"; - return proxyFxUpstream(request, reply, path); - }); - return fastify; -} + fastify.log.info({ traceId, batchedCount }, 'Settlement batching job completed'); + return { batched: batchedCount }; + } catch (error) { + fastify.log.error({ traceId, error }, 'Settlement batching job failed'); + throw error; + } + }, + { connection: redis, concurrency: 1 } +); -// ─── Warmup ───────────────────────────────────────────────────────────────── +// Schedule the batching job to run every BATCH_INTERVAL_SECONDS +await batchQueue.add( + 'batch-pending-settlements', + { traceId: crypto.randomUUID() }, + { + repeat: { + every: env.BATCH_INTERVAL_SECONDS * 1000, + }, + } +); -interface DownstreamService { - name: string; - healthUrl: string; -} +batchWorker.on('completed', (job) => { + fastify.log.info({ jobId: job.id }, 'Batching job completed'); +}); -function getDownstreamServices(env: Env): DownstreamService[] { - return [ - { name: "fx-engine", healthUrl: `${env.FX_ENGINE_URL}/api/health` }, - { name: "indexer", healthUrl: `${env.INDEXER_URL}/api/health` }, - ]; -} +batchWorker.on('failed', (job, err) => { + fastify.log.error({ jobId: job?.id, error: err }, 'Batching job failed'); +}); -/** - * Make best-effort warmup requests to downstream services so their caches, - * connection pools, and health state are ready before the gateway accepts - * traffic. Each call carries a unique x-trace-id so operators can correlate - * startup events across services. - * - * Errors are logged but never thrown — a downstream that is still warming up - * should not prevent the gateway from starting. - */ -async function warmupDownstreamServices( - env: Env, - logger: FastifyBaseLogger, -): Promise { - const services = getDownstreamServices(env); - await Promise.allSettled( - services.map(async (svc) => { - const traceId = crypto.randomUUID(); - const startTime = Date.now(); - try { - const response = await fetch(svc.healthUrl, { - headers: { "x-trace-id": traceId }, - signal: AbortSignal.timeout(5_000), - }); - const durationMs = Date.now() - startTime; - logger.info( - { - traceId, - targetService: svc.name, - statusCode: response.status, - durationMs, - }, - "Warmup completed", - ); - } catch (err) { - const durationMs = Date.now() - startTime; - logger.warn( - { traceId, targetService: svc.name, durationMs, err }, - "Warmup failed — downstream may not be ready", - ); - } - }), - ); -} +// ============================================================================ +// GRACEFUL SHUTDOWN +// ============================================================================ -// Graceful shutdown -let mainApp: ReturnType | null = null; -let metricsServer: ReturnType | null = null; -let shuttingDown = false; +let isShuttingDown = false; -async function shutdown(signal: string) { - if (shuttingDown) return; - shuttingDown = true; +async function gracefulShutdown(signal: string): Promise { + // Prevent multiple shutdown attempts + if (isShuttingDown) { + fastify.log.warn({ signal }, 'Shutdown already in progress, ignoring duplicate signal'); + return; + } + + isShuttingDown = true; + fastify.log.info({ signal }, 'Received shutdown signal, starting graceful shutdown'); - const app = mainApp!; - app.log.info(`Received ${signal}, shutting down gracefully...`); + // Set a timeout to force exit if shutdown hangs + const forceExitTimeout = setTimeout(() => { + fastify.log.error('Graceful shutdown timed out after 30 seconds, forcing exit'); + process.exit(1); + }, 30000); try { - await app.close(); - if (metricsServer) { - await new Promise((resolve) => - metricsServer!.close(() => resolve()), - ); - } - await getDefaultPrisma().$disconnect(); - stopAbandonedPaymentsCron(); + // 1. Close Fastify server (stops accepting new connections) + fastify.log.info('Closing Fastify server...'); + await fastify.close(); + fastify.log.info('Fastify server closed'); + + // 1b. Close the metrics server + await new Promise((resolve) => metricsServer.close(() => resolve())); + + // 2. Close BullMQ workers (drain and close gracefully, force-stop after 10s) + fastify.log.info('Closing BullMQ workers...'); + await closeWorkerWithTimeout(worker, 'settlements', fastify.log, getActiveSettlementJob); + await closeWorkerWithTimeout(batchWorker, 'batching', fastify.log, () => undefined); + fastify.log.info('BullMQ workers closed'); + + // 3. Close BullMQ queues + fastify.log.info('Closing BullMQ queues...'); + await settlementQueue.close(); + await settlementDLQ.close(); + await batchQueue.close(); + await closeWorkerWithTimeout(webhookWorker, 'settlement-webhooks', fastify.log, getActiveWebhookJob); + await webhookQueue.close(); + fastify.log.info('BullMQ queues closed'); + + // 4. Close Redis connection + fastify.log.info('Closing Redis connection...'); + await redis.quit(); + fastify.log.info('Redis connection closed'); + + // 5. Disconnect Prisma + fastify.log.info('Disconnecting Prisma...'); + await prisma.$disconnect(); + fastify.log.info('Prisma disconnected'); + + // Clear the force exit timeout + clearTimeout(forceExitTimeout); + + fastify.log.info({ signal }, 'Graceful shutdown completed successfully'); process.exit(0); - } catch (err) { - app.log.error(err, "Error during shutdown"); + } catch (error) { + fastify.log.error({ error, signal }, 'Error during graceful shutdown'); + clearTimeout(forceExitTimeout); process.exit(1); } } -process.on("SIGTERM", () => shutdown("SIGTERM")); -process.on("SIGINT", () => shutdown("SIGINT")); +// Register shutdown handlers for SIGTERM and SIGINT +process.on('SIGTERM', () => { + void gracefulShutdown('SIGTERM'); +}); + +process.on('SIGINT', () => { + void gracefulShutdown('SIGINT'); +}); + +// ============================================================================ +// STARTUP +// ============================================================================ const start = async () => { try { - const app = mainApp!; - const prisma = getDefaultPrisma(); - const redis = sharedRedis!; - - // #391 — wait for dependencies before accepting traffic - await connectWithRetry(prisma, app.log); - await waitForRedis(redis, app.log); - - // #314 — warmup downstream services with unique trace IDs - await warmupDownstreamServices(env, app.log); + await runStartupChecks({ + service: 'settlement-engine', + version: SERVICE_VERSION, + logger: fastify.log, + checks: [ + { + name: 'prisma', + fn: () => connectWithRetry(prisma, fastify.log), + critical: true, + }, + { + name: 'redis', + fn: () => waitForRedis(redis, fastify.log), + critical: true, + }, + { + name: 'bullmq', + fn: async () => { + const counts = await settlementQueue.getJobCounts(); + fastify.log.info({ counts }, 'BullMQ queue reachable'); + }, + critical: false, + }, + ], + }); // #387 — Redis memory monitoring - startRedisMemoryMonitor(redis, app.log); + startRedisMemoryMonitor(redis, fastify.log); - if (process.env.NODE_ENV !== "test") { - const webhookQueue = createWebhookQueue("gateway-expired-webhooks", { - url: env.REDIS_URL, - }); - startAbandonedPaymentsCron( - prisma, - app.log, - (env as any).PAYMENT_ABANDONMENT_HOURS ?? 24, - webhookQueue, - ); - } - await app.listen({ port: PORT, host: "0.0.0.0" }); + await fastify.listen({ port: PORT, host: '0.0.0.0' }); + fastify.log.info({ port: PORT }, 'Settlement Engine started successfully'); } catch (err) { - if (mainApp) mainApp.log.error(err); - else console.error(err); + fastify.log.error(err); process.exit(1); } }; -const isDirectRun = Boolean( - process.argv[1] && - (process.argv[1].endsWith("index.ts") || - process.argv[1].endsWith("index.js")), -); -if (isDirectRun) { - mainApp = buildApp(); - - // Served on its own port (see startMetricsServer), not the application - // port — keeps the scrape endpoint unauthenticated without exposing it - // alongside application traffic. Started only for the real server process, - // not when buildApp() is called directly by tests. - promClient.collectDefaultMetrics(); - metricsServer = startMetricsServer({ - appPort: PORT, - contentType: promClient.register.contentType, - getMetrics: () => promClient.register.metrics(), - log: mainApp.log, - }); +export { fastify, prisma, settlementQueue }; + +const isDirectRun = + !process.argv[1] || + process.argv[1].endsWith('index.ts') || + process.argv[1].endsWith('index.js') || + process.argv[1].endsWith('dist/index.js'); - logFeatureFlags(mainApp.log); +if (isDirectRun && process.env.NODE_ENV !== 'test') { start(); -} +} \ No newline at end of file diff --git a/prisma.test.ts b/prisma.test.ts new file mode 100644 index 0000000..1055bc2 --- /dev/null +++ b/prisma.test.ts @@ -0,0 +1,228 @@ +import test from 'node:test'; +import assert from 'node:assert'; +import { + buildPrismaConnectionUrl, + connectWithRetry, + getPrismaLogLevels, + shouldEnablePrismaQueryLogging, +} from './prisma.js'; + +test('getPrismaLogLevels includes query in development', () => { + const originalNodeEnv = process.env.NODE_ENV; + const originalLogLevel = process.env.LOG_LEVEL; + + process.env.NODE_ENV = 'development'; + delete process.env.LOG_LEVEL; + assert.ok(getPrismaLogLevels().includes('query')); + + process.env.NODE_ENV = originalNodeEnv; + process.env.LOG_LEVEL = originalLogLevel; +}); + +test('getPrismaLogLevels excludes query in production', () => { + const originalNodeEnv = process.env.NODE_ENV; + const originalLogLevel = process.env.LOG_LEVEL; + + process.env.NODE_ENV = 'production'; + process.env.LOG_LEVEL = 'debug'; + assert.ok(!shouldEnablePrismaQueryLogging()); + assert.ok(!getPrismaLogLevels().includes('query')); + + process.env.NODE_ENV = originalNodeEnv; + process.env.LOG_LEVEL = originalLogLevel; +}); + +test('getPrismaLogLevels returns error and warn only in production', () => { + const originalNodeEnv = process.env.NODE_ENV; + const originalOverride = process.env.PRISMA_LOG_LEVELS; + + process.env.NODE_ENV = 'production'; + delete process.env.PRISMA_LOG_LEVELS; + assert.deepStrictEqual(getPrismaLogLevels(), ['error', 'warn']); + + process.env.NODE_ENV = originalNodeEnv; + process.env.PRISMA_LOG_LEVELS = originalOverride; +}); + +test('getPrismaLogLevels returns query, info, warn, error in development', () => { + const originalNodeEnv = process.env.NODE_ENV; + const originalOverride = process.env.PRISMA_LOG_LEVELS; + + process.env.NODE_ENV = 'development'; + delete process.env.PRISMA_LOG_LEVELS; + assert.deepStrictEqual(getPrismaLogLevels(), ['query', 'info', 'warn', 'error']); + + process.env.NODE_ENV = originalNodeEnv; + process.env.PRISMA_LOG_LEVELS = originalOverride; +}); + +test('getPrismaLogLevels returns error only in test', () => { + const originalNodeEnv = process.env.NODE_ENV; + const originalOverride = process.env.PRISMA_LOG_LEVELS; + + process.env.NODE_ENV = 'test'; + delete process.env.PRISMA_LOG_LEVELS; + assert.deepStrictEqual(getPrismaLogLevels(), ['error']); + + process.env.NODE_ENV = originalNodeEnv; + process.env.PRISMA_LOG_LEVELS = originalOverride; +}); + +test('getPrismaLogLevels honors PRISMA_LOG_LEVELS override regardless of NODE_ENV', () => { + const originalNodeEnv = process.env.NODE_ENV; + const originalOverride = process.env.PRISMA_LOG_LEVELS; + + process.env.NODE_ENV = 'production'; + process.env.PRISMA_LOG_LEVELS = 'query, warn'; + assert.deepStrictEqual(getPrismaLogLevels(), ['query', 'warn']); + + process.env.NODE_ENV = originalNodeEnv; + process.env.PRISMA_LOG_LEVELS = originalOverride; +}); + +test('getPrismaLogLevels falls back to the NODE_ENV default when the override has no valid levels', () => { + const originalNodeEnv = process.env.NODE_ENV; + const originalOverride = process.env.PRISMA_LOG_LEVELS; + + process.env.NODE_ENV = 'test'; + process.env.PRISMA_LOG_LEVELS = 'not-a-level, also-invalid'; + assert.deepStrictEqual(getPrismaLogLevels(), ['error']); + + process.env.NODE_ENV = originalNodeEnv; + process.env.PRISMA_LOG_LEVELS = originalOverride; +}); + +test('connectWithRetry succeeds after transient failures', async () => { + let attempts = 0; + const prisma = { + async $connect() { + attempts += 1; + if (attempts < 3) { + throw new Error('connection refused'); + } + }, + }; + + const warnings: object[] = []; + await connectWithRetry(prisma, { + debug: () => undefined, + warn: (obj) => warnings.push(obj), + }, { baseDelayMs: 1, maxRetries: 5 }); + + assert.strictEqual(attempts, 3); + assert.strictEqual(warnings.length, 2); +}); + +test('connectWithRetry throws after exhausting retries', async () => { + const prisma = { + async $connect() { + throw new Error('database unavailable'); + }, + }; + + await assert.rejects( + () => + connectWithRetry(prisma, { + debug: () => undefined, + warn: () => undefined, + }, { baseDelayMs: 1, maxRetries: 3 }), + ); +}); + +test('buildPrismaConnectionUrl appends params with ? when URL has no query string', () => { + const url = 'postgresql://user:pass@localhost:5432/bettapay'; + const result = buildPrismaConnectionUrl(url, 15, 10); + assert.strictEqual(result, 'postgresql://user:pass@localhost:5432/bettapay?connection_limit=15&pool_timeout=10'); +}); + +test('buildPrismaConnectionUrl appends params with & when URL already has a query string', () => { + const url = 'postgresql://user:pass@localhost:5432/bettapay?sslmode=require'; + const result = buildPrismaConnectionUrl(url, 20, 5); + assert.strictEqual(result, 'postgresql://user:pass@localhost:5432/bettapay?sslmode=require&connection_limit=20&pool_timeout=5'); +}); + +test('buildPrismaConnectionUrl uses defaults when no poolSize or timeout given', () => { + const url = 'postgresql://user:pass@localhost:5432/bettapay'; + const result = buildPrismaConnectionUrl(url); + assert.strictEqual(result, 'postgresql://user:pass@localhost:5432/bettapay?connection_limit=10&pool_timeout=10'); +}); + +import { + resetRotation, + hasRotated, + getActiveConnectionUrl, + connectWithRetryWithRotation, +} from './prisma.js'; + +test('connectWithRetryWithRotation switches to rotate URL on 28P01', async () => { + resetRotation(); + let attempts = 0; + const prisma = { + async $connect() { + attempts += 1; + if (attempts === 1) { + const err = new Error('authentication failed'); + (err as any).code = '28P01'; + throw err; + } + }, + }; + + await connectWithRetryWithRotation(prisma, { + debug: () => undefined, + warn: () => undefined, + }, { rotationUrl: 'postgres://rotated/db', baseDelayMs: 1, maxRetries: 5 }); + + assert.strictEqual(hasRotated(), true); + assert.strictEqual(getActiveConnectionUrl('postgres://primary/db'), 'postgres://rotated/db'); + resetRotation(); +}); + +test('connectWithRetryWithRotation does not switch on primary URL success', async () => { + resetRotation(); + const prisma = { + async $connect() {}, + }; + + await connectWithRetryWithRotation(prisma, { + debug: () => undefined, + warn: () => undefined, + }, { rotationUrl: 'postgres://rotated/db', baseDelayMs: 1, maxRetries: 3 }); + + assert.strictEqual(hasRotated(), false); + assert.strictEqual(getActiveConnectionUrl('postgres://primary/db'), 'postgres://primary/db'); + resetRotation(); +}); + +test('rotation logged at warn level', async () => { + resetRotation(); + let attempts = 0; + const prisma = { + async $connect() { + attempts += 1; + if (attempts === 1) { + const err = new Error('authentication failed'); + (err as any).code = '28P01'; + throw err; + } + }, + }; + + const warnMessages: string[] = []; + const logger = { + debug: () => undefined, + warn: (_obj: object, msg?: string) => { + if (msg) warnMessages.push(msg); + }, + }; + + await connectWithRetryWithRotation(prisma, logger, { rotationUrl: 'postgres://rotated/db', baseDelayMs: 1, maxRetries: 5 }); + + assert.ok(warnMessages.some(m => m.includes('credential rotation') || m.includes('authentication error'))); + resetRotation(); +}); + +test('getActiveConnectionUrl returns primary URL when no rotation', () => { + resetRotation(); + assert.strictEqual(getActiveConnectionUrl('postgres://primary/db'), 'postgres://primary/db'); +}); diff --git a/prisma.ts b/prisma.ts new file mode 100644 index 0000000..9d0cfd2 --- /dev/null +++ b/prisma.ts @@ -0,0 +1,202 @@ +export type PrismaLogLevel = 'query' | 'info' | 'warn' | 'error'; + +const ALL_PRISMA_LOG_LEVELS: readonly PrismaLogLevel[] = ['query', 'info', 'warn', 'error']; + +let _rotateUrl: string | undefined; +let _hasRotated = false; + +export function setRotationUrl(url: string | undefined): void { + _rotateUrl = url; + _hasRotated = false; +} + +export function resetRotation(): void { + _rotateUrl = undefined; + _hasRotated = false; +} + +export function hasRotated(): boolean { + return _hasRotated; +} + +export function getActiveConnectionUrl(primaryUrl: string): string { + return _hasRotated && _rotateUrl !== undefined ? _rotateUrl : primaryUrl; +} + +function isPrismaLogLevel(value: string): value is PrismaLogLevel { + return (ALL_PRISMA_LOG_LEVELS as readonly string[]).includes(value); +} + +// PRISMA_LOG_LEVELS is a comma-separated override, e.g. "error,warn". Falls +// back to the NODE_ENV default when unset, empty, or containing no valid level. +function parsePrismaLogLevelsOverride(raw: string | undefined): PrismaLogLevel[] | undefined { + if (!raw) return undefined; + const levels = raw + .split(',') + .map((level) => level.trim().toLowerCase()) + .filter(isPrismaLogLevel); + return levels.length > 0 ? levels : undefined; +} + +function defaultPrismaLogLevelsForEnv(nodeEnv: string | undefined): PrismaLogLevel[] { + if (nodeEnv === 'production') return ['error', 'warn']; + if (nodeEnv === 'test') return ['error']; + return ['query', 'info', 'warn', 'error']; +} + +export interface PrismaQueryEvent { + query: string; + duration: number; +} + +export interface PrismaConnectable { + $connect: () => Promise; +} + +export interface PrismaQueryable { + $on: (event: 'query', callback: (event: PrismaQueryEvent) => void) => void; +} + +export interface PrismaLogger { + debug: (obj: object, msg?: string) => void; + warn: (obj: object, msg?: string) => void; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +export function shouldEnablePrismaQueryLogging(): boolean { + if (process.env.NODE_ENV === 'production') return false; + return process.env.LOG_LEVEL === 'debug' || process.env.NODE_ENV === 'development'; +} + +export function getPrismaLogLevels(): PrismaLogLevel[] { + const override = parsePrismaLogLevelsOverride(process.env.PRISMA_LOG_LEVELS); + const levels = override ?? defaultPrismaLogLevelsForEnv(process.env.NODE_ENV); + const source = override ? 'PRISMA_LOG_LEVELS override' : `NODE_ENV=${process.env.NODE_ENV ?? 'development'} default`; + console.log(`[Prisma] log levels: ${levels.join(', ')} (${source})`); + return levels; +} + +export function setupPrismaQueryLogging(prisma: PrismaQueryable, logger: PrismaLogger): void { + if (!shouldEnablePrismaQueryLogging()) return; + + prisma.$on('query', (event) => { + logger.debug({ query: event.query, duration: event.duration }, 'Prisma query'); + }); +} + +export interface ConnectWithRetryOptions { + maxRetries?: number; + baseDelayMs?: number; + maxDelayMs?: number; +} + +export async function connectWithRetry( + prisma: PrismaConnectable, + logger: PrismaLogger, + options: ConnectWithRetryOptions = {} +): Promise { + const maxRetries = options.maxRetries ?? 10; + const baseDelayMs = options.baseDelayMs ?? 1000; + const maxDelayMs = options.maxDelayMs ?? 30000; + let lastError: unknown; + + for (let attempt = 0; attempt < maxRetries; attempt++) { + try { + await prisma.$connect(); + return; + } catch (err) { + lastError = err; + if (attempt < maxRetries - 1) { + const delay = Math.min(baseDelayMs * Math.pow(2, attempt), maxDelayMs); + logger.warn( + { attempt: attempt + 1, maxRetries, delayMs: delay, err }, + 'Database connection failed, retrying' + ); + await sleep(delay); + } + } + } + + const message = + lastError instanceof Error ? lastError.message : String(lastError ?? 'unknown error'); + throw new Error(`Failed to connect to database after ${maxRetries} attempts: ${message}`); +} + +/** + * Build a Prisma-compatible connection URL with pool and timeout parameters. + * + * Appends `connection_limit` and `pool_timeout` as query parameters to the + * raw DATABASE_URL. These tell Prisma's internal query engine how many + * concurrent connections to allow and how long to wait before timing out a + * pooled connection request. Without explicit values Prisma defaults to an + * unbounded pool — a recipe for connection exhaustion under load. + * + * The function safely detects whether the URL already carries a query string + * (i.e. contains "?") and uses "&" instead of "?" to avoid clobbering any + * pre-existing parameters such as sslmode, schema, or application_name. + * + * @param rawUrl - The base DATABASE_URL (e.g. postgresql://user:pass@host:5432/db). + * @param poolSize - Max connections in the Prisma pool (default: 10). + * @param timeout - Max seconds to wait for a connection from the pool (default: 10). + */ +export function buildPrismaConnectionUrl( + rawUrl: string, + poolSize: number = 10, + timeout: number = 10, +): string { + const sep = rawUrl.includes('?') ? '&' : '?'; + return `${rawUrl}${sep}connection_limit=${poolSize}&pool_timeout=${timeout}`; +} + +export interface ConnectWithRotationOptions extends ConnectWithRetryOptions { + rotationUrl?: string; + logger?: PrismaLogger; +} + +export async function connectWithRetryWithRotation( + prisma: PrismaConnectable, + logger: PrismaLogger, + options: ConnectWithRotationOptions = {} +): Promise { + const maxRetries = options.maxRetries ?? 10; + const baseDelayMs = options.baseDelayMs ?? 1000; + const maxDelayMs = options.maxDelayMs ?? 30000; + const rotationUrl = options.rotationUrl; + let lastError: unknown; + + for (let attempt = 0; attempt < maxRetries; attempt++) { + try { + await prisma.$connect(); + return; + } catch (err) { + lastError = err; + const isAuthError = (err as { code?: string })?.code === '28P01'; + + if (isAuthError && rotationUrl && !_hasRotated) { + _hasRotated = true; + _rotateUrl = rotationUrl; + const rotationLogger = options.logger ?? logger; + rotationLogger.warn( + { attempt: attempt + 1 }, + 'Database credential rotation: switching to rotation URL due to authentication error (28P01)' + ); + } + + if (attempt < maxRetries - 1) { + const delay = Math.min(baseDelayMs * Math.pow(2, attempt), maxDelayMs); + logger.warn( + { attempt: attempt + 1, maxRetries, delayMs: delay, err }, + 'Database connection failed, retrying' + ); + await sleep(delay); + } + } + } + + const message = + lastError instanceof Error ? lastError.message : String(lastError ?? 'unknown error'); + throw new Error(`Failed to connect to database after ${maxRetries} attempts: ${message}`); +}