From ef0a9ef2126115ef63a8dd3630ce45506c1ce200 Mon Sep 17 00:00:00 2001 From: MerlinTheWhiz Date: Tue, 28 Jul 2026 22:20:37 +0100 Subject: [PATCH] feat: wrap all external dependencies with timeout/retry/circuit-breaker MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Guard every outbound integration — MongoDB, Stellar (Horizon + RPC), Pinata, Nodemailer, Redis, and webhook dispatcher — with consistent timeout, retry, and circuit-breaker primitives. Translate dependency errors to proper HTTP 503 responses, emit circuit-breaker metrics, propagate traceparent/correlation IDs, and add integration tests. Also fixes pre-existing module corruption in entitlement.js. --- src/app/api/materials/upload/route.js | 14 +- src/app/api/ready/route.js | 10 +- src/app/api/upload/route.js | 56 +----- src/lib/api/hardening.js | 12 ++ src/lib/cache/redis.js | 76 +++++-- src/lib/email.js | 56 +++++- src/lib/entitlement.js | 138 ++----------- src/lib/indexer/stellarIndexer.js | 28 +-- src/lib/mongodb.js | 76 +++++-- src/lib/pinata.js | 66 +++++- src/lib/resilience/circuitBreaker.js | 162 +++++++++++++++ src/lib/resilience/classification.js | 95 +++++++++ src/lib/resilience/index.js | 41 ++++ src/lib/resilience/retry.js | 100 ++++++++++ src/lib/resilience/timeout.js | 61 ++++++ src/lib/stellar/horizonClient.js | 35 ++++ src/lib/stellar/rpcClient.js | 151 ++++++++++++++ src/lib/telemetry/context.js | 1 + src/lib/webhooks/dispatcher.js | 43 ++++ test/fixtures/resilience.js | 159 +++++++++++++++ .../resilience/circuitBreaker.test.js | 188 ++++++++++++++++++ test/integration/resilience/email.test.js | 34 ++++ test/integration/resilience/hardening.test.js | 62 ++++++ test/integration/resilience/pinata.test.js | 49 +++++ test/integration/resilience/redis.test.js | 54 +++++ test/integration/resilience/retry.test.js | 142 +++++++++++++ test/integration/resilience/timeout.test.js | 133 +++++++++++++ test/integration/resilience/webhook.test.js | 27 +++ test/mocks/next-server.js | 12 ++ vitest.config.mjs | 2 + 30 files changed, 1839 insertions(+), 244 deletions(-) create mode 100644 src/lib/resilience/circuitBreaker.js create mode 100644 src/lib/resilience/classification.js create mode 100644 src/lib/resilience/index.js create mode 100644 src/lib/resilience/retry.js create mode 100644 src/lib/resilience/timeout.js create mode 100644 src/lib/stellar/rpcClient.js create mode 100644 test/fixtures/resilience.js create mode 100644 test/integration/resilience/circuitBreaker.test.js create mode 100644 test/integration/resilience/email.test.js create mode 100644 test/integration/resilience/hardening.test.js create mode 100644 test/integration/resilience/pinata.test.js create mode 100644 test/integration/resilience/redis.test.js create mode 100644 test/integration/resilience/retry.test.js create mode 100644 test/integration/resilience/timeout.test.js create mode 100644 test/integration/resilience/webhook.test.js create mode 100644 test/mocks/next-server.js diff --git a/src/app/api/materials/upload/route.js b/src/app/api/materials/upload/route.js index cf77d76..c59725b 100644 --- a/src/app/api/materials/upload/route.js +++ b/src/app/api/materials/upload/route.js @@ -10,7 +10,7 @@ import { validateUploadPayload, validateUploadFileMetadata, } from '@/lib/api/validation' -import { pinata } from '@/lib/pinata' +import { pinata, callPinata } from '@/lib/pinata' import { validateUploadedFile } from '@/lib/ipfs/uploadValidator' const MAX_FILE_SIZE_BYTES = 10 * 1024 * 1024 // 10 MB @@ -119,13 +119,13 @@ export async function POST(request) { // All checks passed — dispatch to Pinata const results = {} - const uploadedFile = await pinata.upload.public.file(file) - const fileUrl = await pinata.gateways.public.convert(uploadedFile.cid) + const uploadedFile = await callPinata('upload.file', () => pinata.upload.public.file(file)) + const fileUrl = await callPinata('gateway.file', () => pinata.gateways.public.convert(uploadedFile.cid)) results.fileUrl = fileUrl if (image) { - const fileThumb = await pinata.upload.public.file(image) - const imgUrl = await pinata.gateways.public.convert(fileThumb.cid) + const fileThumb = await callPinata('upload.thumbnail', () => pinata.upload.public.file(image)) + const imgUrl = await callPinata('gateway.thumbnail', () => pinata.gateways.public.convert(fileThumb.cid)) results.imgUrl = imgUrl } @@ -168,8 +168,8 @@ export async function POST(request) { timestamp: new Date().toISOString(), } - const uploadedJson = await pinata.upload.public.json(metadataJSON) - const jsonUrl = await pinata.gateways.public.convert(uploadedJson.cid) + const uploadedJson = await callPinata('upload.metadata', () => pinata.upload.public.json(metadataJSON)) + const jsonUrl = await callPinata('gateway.metadata', () => pinata.gateways.public.convert(uploadedJson.cid)) results.metadataUrl = jsonUrl auditLog({ event: 'upload_complete', route: 'materials/upload', method: 'POST', status: 200 }) diff --git a/src/app/api/ready/route.js b/src/app/api/ready/route.js index f043581..3cac805 100644 --- a/src/app/api/ready/route.js +++ b/src/app/api/ready/route.js @@ -2,12 +2,13 @@ export const dynamic = "force-dynamic"; import { NextResponse } from "next/server"; import { getDb } from "@/lib/mongodb"; -import { pinata } from "@/lib/pinata"; +import { pinata, callPinata } from "@/lib/pinata"; import { verifyEmailConnection } from "@/lib/email"; import { withApiHardening } from "@/lib/api/hardening"; -import { Server, rpc } from "@stellar/stellar-sdk"; +import { Server } from "@stellar/stellar-sdk"; import { HORIZON_URL, STELLAR_RPC_URL } from "@/lib/config/chain"; import { setGauge } from "@/lib/telemetry/metrics"; +import { getRpcHealth } from "@/lib/stellar/rpcClient"; /** * Readiness probe (#20): "can this instance actually serve traffic right now?" @@ -58,7 +59,7 @@ export async function GET(request) { status.pinata = "offline: PINATA_JWT not configured"; recordFailure("pinata"); } else { - await pinata.testAuthentication(); + await callPinata('testAuthentication', () => pinata.testAuthentication()); status.pinata = "online"; } } catch (err) { @@ -80,8 +81,7 @@ export async function GET(request) { const horizonServer = new Server(HORIZON_URL); await horizonServer.root(); - const rpcServer = new rpc.Server(STELLAR_RPC_URL); - const health = await rpcServer.getHealth(); + const health = await getRpcHealth(); if (health && health.status === "healthy") { status.stellar = "online"; diff --git a/src/app/api/upload/route.js b/src/app/api/upload/route.js index dd60576..08b0864 100644 --- a/src/app/api/upload/route.js +++ b/src/app/api/upload/route.js @@ -9,12 +9,11 @@ import { validateUploadPayload, } from '@/lib/api/validation' import { - retryWithBackoff, validateGatewayUrl, validatePinataResponse, } from '@/lib/api/storage' import { getDb } from '@/lib/mongodb' -import { pinata } from '@/lib/pinata' +import { pinata, callPinata } from '@/lib/pinata' import { storeManifest } from '@/lib/provenance/registry' import { hashFileBytes } from '@/lib/provenance/manifest' import { validateUploadedFile } from '@/lib/ipfs/uploadValidator' @@ -215,23 +214,10 @@ export async function POST(request) { // 4️⃣ Upload the main file try { - const uploadedFile = await retryWithBackoff( - () => pinata.upload.public.file(file), - 3, - 1000, - (err, attempt) => { - console.warn(`[Storage] Document upload attempt ${attempt} failed: ${err.message}`); - } - ) + const uploadedFile = await callPinata('upload.file', () => pinata.upload.public.file(file)) validatePinataResponse(uploadedFile, 'document') - const fileUrl = await retryWithBackoff( - () => pinata.gateways.public.convert(uploadedFile.cid), - 3, - 1000, - (err, attempt) => { - console.warn(`[Storage] Document gateway conversion attempt ${attempt} failed: ${err.message}`); - } + const fileUrl = await callPinata('gateway.convert', () => pinata.gateways.public.convert(uploadedFile.cid)) ) validateGatewayUrl(fileUrl, 'document') results.fileUrl = fileUrl @@ -253,24 +239,10 @@ export async function POST(request) { // 5️⃣ Upload thumbnail (if provided) if (image) { try { - const fileThumb = await retryWithBackoff( - () => pinata.upload.public.file(image), - 3, - 1000, - (err, attempt) => { - console.warn(`[Storage] Thumbnail upload attempt ${attempt} failed: ${err.message}`); - } - ) + const fileThumb = await callPinata('upload.thumbnail', () => pinata.upload.public.file(image)) validatePinataResponse(fileThumb, 'thumbnail') - const imgUrl = await retryWithBackoff( - () => pinata.gateways.public.convert(fileThumb.cid), - 3, - 1000, - (err, attempt) => { - console.warn(`[Storage] Thumbnail gateway conversion attempt ${attempt} failed: ${err.message}`); - } - ) + const imgUrl = await callPinata('gateway.thumbnail', () => pinata.gateways.public.convert(fileThumb.cid)) validateGatewayUrl(imgUrl, 'thumbnail') results.imgUrl = imgUrl } catch (err) { @@ -344,24 +316,10 @@ export async function POST(request) { // 7️⃣ Upload metadata JSON to Pinata try { - const uploadedJson = await retryWithBackoff( - () => pinata.upload.public.json(metadataJSON), - 3, - 1000, - (err, attempt) => { - console.warn(`[Storage] Metadata upload attempt ${attempt} failed: ${err.message}`); - } - ) + const uploadedJson = await callPinata('upload.metadata', () => pinata.upload.public.json(metadataJSON)) validatePinataResponse(uploadedJson, 'metadata') - const jsonUrl = await retryWithBackoff( - () => pinata.gateways.public.convert(uploadedJson.cid), - 3, - 1000, - (err, attempt) => { - console.warn(`[Storage] Metadata gateway conversion attempt ${attempt} failed: ${err.message}`); - } - ) + const jsonUrl = await callPinata('gateway.metadata', () => pinata.gateways.public.convert(uploadedJson.cid)) validateGatewayUrl(jsonUrl, 'metadata') results.metadataUrl = jsonUrl } catch (err) { diff --git a/src/lib/api/hardening.js b/src/lib/api/hardening.js index 1726f94..a89e6f3 100644 --- a/src/lib/api/hardening.js +++ b/src/lib/api/hardening.js @@ -2,6 +2,7 @@ import { NextResponse } from "next/server"; import { auditLog } from "./audit"; import { checkRateLimit } from "./rateLimit"; import { ValidationError } from "./validation"; +import { DependencyError } from "@/lib/resilience/index.js"; import { captureException } from "@/lib/sentry"; import { runWithContext, currentTraceparent, currentCorrelationId } from "@/lib/telemetry/context"; import { withSpan } from "@/lib/telemetry/tracing"; @@ -141,6 +142,17 @@ export async function withApiHardening(request, options, handler) { return finalize(res); } + if (error instanceof DependencyError) { + auditLog({ event: "dependency_failed", route, method, status: error.statusCode || 503, dependency: error.dependency, action: error.action, reason: error.message }); + incrementCounter("http_requests_total", { route, method, outcome: "dependency_error" }); + incrementCounter("rpc_errors_total", { operation: `${error.dependency}.${error.action}` }); + const statusCode = error.statusCode || 503; + return finalize(NextResponse.json( + { error: error.userMessage || "Service temporarily unavailable", code: "dependency_error", dependency: error.dependency }, + { status: statusCode, headers: { "x-correlation-id": currentCorrelationId() } } + )); + } + incrementCounter("http_requests_total", { route, method, outcome: "error" }); captureException(error, { route, method, correlationId: currentCorrelationId() }); return finalize(NextResponse.json({ error: "Internal Server Error", code: "internal_error" }, { status: 500 })); diff --git a/src/lib/cache/redis.js b/src/lib/cache/redis.js index fa7a6e4..4416ff5 100644 --- a/src/lib/cache/redis.js +++ b/src/lib/cache/redis.js @@ -1,36 +1,80 @@ import { createClient } from 'redis'; +import { createCircuitBreaker, CircuitState, DependencyError } from "@/lib/resilience/index.js"; +import { withTimeout } from "@/lib/resilience/timeout.js"; +import { withRetry } from "@/lib/resilience/retry.js"; +import { setGauge, incrementCounter } from "@/lib/telemetry/metrics"; + +const REDIS_TIMEOUT_MS = Number(process.env.REDIS_TIMEOUT_MS || 5000); +const REDIS_CONNECT_TIMEOUT_MS = Number(process.env.REDIS_CONNECT_TIMEOUT_MS || 10000); + +const redisCircuitBreaker = createCircuitBreaker("redis", { + failureThreshold: Number(process.env.REDIS_CB_FAILURE_THRESHOLD || 3), + successThreshold: 2, + resetTimeoutMs: Number(process.env.REDIS_CB_RESET_TIMEOUT_MS || 30000), + onStateChange(name, from, to) { + setGauge("circuit_breaker_state", { dependency: name, state: to }, 1); + if (from && from !== to) { + setGauge("circuit_breaker_state", { dependency: name, state: from }, 0); + } + if (to === CircuitState.OPEN) { + incrementCounter("circuit_breaker_open_total", { dependency: name }); + } + const level = to === CircuitState.OPEN ? "warn" : "info"; + console[level](`[circuit-breaker] redis: ${from} -> ${to}`); + }, +}); + +export function getRedisCircuitBreakerState() { + return redisCircuitBreaker.getState(); +} let client = null; export async function getRedisClient() { if (!process.env.REDIS_URL) return null; + if (redisCircuitBreaker.getState() === CircuitState.OPEN) { + throw new DependencyError({ + dependency: "redis", + action: "connect", + retryable: false, + statusCode: 503, + userMessage: "Cache service is temporarily unavailable.", + }); + } if (!client) { - client = createClient({ url: process.env.REDIS_URL }); + client = createClient({ url: process.env.REDIS_URL, socket: { reconnectStrategy: false } }); client.on('error', (err) => console.error('Redis error', err.message)); - await client.connect(); + await withTimeout(client.connect(), REDIS_CONNECT_TIMEOUT_MS, 'redis.connect'); } return client; } export async function cacheGet(key) { - const redis = await getRedisClient(); - if (!redis) return null; - try { - const val = await redis.get(key); - return val ? JSON.parse(val) : null; - } catch { return null; } + return redisCircuitBreaker.call(async () => { + const redis = await getRedisClient(); + if (!redis) return null; + return withRetry( + async () => { + const val = await withTimeout(redis.get(key), REDIS_TIMEOUT_MS, 'redis.get'); + return val ? JSON.parse(val) : null; + }, + { idempotent: true, maxAttempts: 2, baseDelayMs: 200, maxDelayMs: 1000 } + ); + }); } export async function cacheSet(key, value, ttlSeconds = 600) { - const redis = await getRedisClient(); - if (!redis) return; - try { - await redis.set(key, JSON.stringify(value), { EX: ttlSeconds }); - } catch { /* no-op */ } + return redisCircuitBreaker.call(async () => { + const redis = await getRedisClient(); + if (!redis) return; + await withTimeout(redis.set(key, JSON.stringify(value), { EX: ttlSeconds }), REDIS_TIMEOUT_MS, 'redis.set'); + }); } export async function cacheDel(key) { - const redis = await getRedisClient(); - if (!redis) return; - try { await redis.del(key); } catch { /* no-op */ } + return redisCircuitBreaker.call(async () => { + const redis = await getRedisClient(); + if (!redis) return; + await withTimeout(redis.del(key), REDIS_TIMEOUT_MS, 'redis.del'); + }); } diff --git a/src/lib/email.js b/src/lib/email.js index bef0d0c..865965c 100644 --- a/src/lib/email.js +++ b/src/lib/email.js @@ -1,4 +1,56 @@ import nodemailer from "nodemailer"; +import { createCircuitBreaker, CircuitState, DependencyError } from "@/lib/resilience/index.js"; +import { withTimeout } from "@/lib/resilience/timeout.js"; +import { withRetry } from "@/lib/resilience/retry.js"; +import { setGauge, incrementCounter } from "@/lib/telemetry/metrics"; +import { currentTraceparent } from "@/lib/telemetry/context"; + +const EMAIL_TIMEOUT_MS = Number(process.env.EMAIL_TIMEOUT_MS || 15000); + +const emailCircuitBreaker = createCircuitBreaker("email", { + failureThreshold: Number(process.env.EMAIL_CB_FAILURE_THRESHOLD || 3), + successThreshold: 2, + resetTimeoutMs: Number(process.env.EMAIL_CB_RESET_TIMEOUT_MS || 30000), + onStateChange(name, from, to) { + setGauge("circuit_breaker_state", { dependency: name, state: to }, 1); + if (from && from !== to) { + setGauge("circuit_breaker_state", { dependency: name, state: from }, 0); + } + if (to === CircuitState.OPEN) { + incrementCounter("circuit_breaker_open_total", { dependency: name }); + } + const level = to === CircuitState.OPEN ? "warn" : "info"; + console[level](`[circuit-breaker] email: ${from} -> ${to}`); + }, +}); + +export function getEmailCircuitBreakerState() { + return emailCircuitBreaker.getState(); +} + +async function withEmailResilience(action, fn, opts = {}) { + if (emailCircuitBreaker.getState() === CircuitState.OPEN) { + incrementCounter("rpc_errors_total", { operation: `email.${action}` }); + throw new DependencyError({ + dependency: "email", + action, + retryable: false, + statusCode: 503, + userMessage: "Email service is temporarily unavailable.", + }); + } + return emailCircuitBreaker.call(async () => { + return withRetry( + async () => withTimeout(fn(), EMAIL_TIMEOUT_MS, `email.${action}`), + { + idempotent: opts.idempotent !== false, + maxAttempts: Number(process.env.EMAIL_RETRIES || 2), + baseDelayMs: 500, + maxDelayMs: 5000, + } + ); + }); +} function createTransporter() { // Prefer explicit SMTP settings; fallback to Gmail using EMAIL_USER/PASS @@ -112,10 +164,10 @@ export async function sendWelcomeEmail(to, name) { `; - await transporter.sendMail({ from, to, subject, text, html }); + await withEmailResilience('send.welcome', () => transporter.sendMail({ from, to, subject, text, html }), { idempotent: false }); } export async function verifyEmailConnection() { const transporter = createTransporter(); - await transporter.verify(); + await withEmailResilience('verify', () => transporter.verify()); } diff --git a/src/lib/entitlement.js b/src/lib/entitlement.js index e634380..c801aa1 100644 --- a/src/lib/entitlement.js +++ b/src/lib/entitlement.js @@ -2,6 +2,7 @@ import { getDb } from '@/lib/mongodb'; import { PURCHASE_MANAGER_CONTRACT_ID, STELLAR_RPC_URL, NETWORK_PASSPHRASE } from '@/lib/config/chain'; import { isCompletedPurchaseStatus, normalizeBuyerAddress } from '@/lib/purchases/access'; import { Contract, Address, nativeToScVal, scValToNative, xdr, TransactionBuilder, Account } from '@stellar/stellar-sdk'; +import { simulateTransaction } from '@/lib/stellar/rpcClient'; import logger from '@/lib/logger'; const CACHE_TTL_MS = 5 * 60 * 1000; // 5 minutes positive TTL @@ -52,111 +53,34 @@ export function formatMaterialIdBytes(materialId) { return buf; } -export function buildHasEntitlementXdr(materialId, buyerAddress) { - const contract = new Contract(PURCHASE_MANAGER_CONTRACT_ID); - const matIdScVal = nativeToScVal(formatMaterialIdBytes(materialId), { type: 'bytesN', size: 32 }); - const buyerScVal = new Address(buyerAddress).toScVal(); - - const op = contract.call('has_entitlement', matIdScVal, buyerScVal); - - const tx = new TransactionBuilder(new Account("GAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAWHF", "1"), { - fee: "100", - networkPassphrase: NETWORK_PASSPHRASE, - }) - .addOperation(op) - .setTimeout(30) - .build(); +export async function checkChainEntitlement(materialId, buyerAddress) { + if (!PURCHASE_MANAGER_CONTRACT_ID || !STELLAR_RPC_URL) return null; - return tx.toXDR(); -} + const xdrBlob = buildHasEntitlementXdr(materialId, buyerAddress); + if (!xdrBlob) return null; -export function decodeBoolean(xdrBase64) { + let result; try { - const scval = xdr.ScVal.fromXDR(xdrBase64, 'base64'); - return scValToNative(scval) === true; + result = await simulateTransaction(xdrBlob); } catch (err) { - logger?.warn({ err: err.message }, 'Failed to decode boolean from SCVal'); - return false; + logger?.error({ err: err.message }, 'RPC Error in checkChainEntitlement'); + return null; } -} - -export async function checkChainEntitlement(materialId, buyerAddress) { - if (!PURCHASE_MANAGER_CONTRACT_ID || !STELLAR_RPC_URL) return null; -import { client } from './backend/db'; // Your app's MongoDB client instance -import { CacheEngine } from './cache/engine'; -import { getDb } from './mongodb.js'; -import { PURCHASE_MANAGER_CONTRACT_ID, STELLAR_RPC_URL, NETWORK_PASSPHRASE } from './config/chain.js'; -import { isCompletedPurchaseStatus, normalizeBuyerAddress } from './purchases/access.js'; -import { - Keypair, - TransactionBuilder, - BASE_FEE, - Contract, - xdr, - Account, - Address, - nativeToScVal, -} from '@stellar/stellar-sdk'; - -export async function updateEntitlement(tenant, network, userId, authScope, updates) { - const db = client.db(); - const session = client.startSession(); - - try { - let result; - await session.withTransaction(async () => { - // 1. Update source of truth - result = await db.collection('entitlements').findOneAndUpdate( - { tenant, network, userId, authScope }, - { - $set: { ...updates, updatedAt: new Date() }, - $inc: { version: 1 } - }, - { returnDocument: 'after', session } - ); - const xdrBlob = buildHasEntitlementXdr(materialId, buyerAddress); - if (!xdrBlob) return null; - - const body = { - jsonrpc: '2.0', - id: 1, - method: 'simulateTransaction', - params: { - transaction: xdrBlob, - }, - }; - const res = await fetch(STELLAR_RPC_URL, { - method: 'POST', - headers: { 'content-type': 'application/json' }, - body: JSON.stringify(body), - signal: AbortSignal.timeout(8_000), - }); - - const payload = await res.json(); - if (payload.error) { - logger?.error({ err: payload.error }, 'RPC Error in checkChainEntitlement'); - return null; - } - - if (payload.result?.restorePreamble) { - logger?.warn({ materialId, buyerAddress }, 'Archived state detected in checkChainEntitlement (restorePreamble)'); - return null; - } - - const retval = payload.result?.results?.[0]?.xdr; - if (!retval) { - logger?.warn({ result: payload.result }, 'Malformed result in checkChainEntitlement'); - return null; - } + if (result?.restorePreamble) { + logger?.warn({ materialId, buyerAddress }, 'Archived state detected in checkChainEntitlement (restorePreamble)'); + return null; + } - const hasAccess = decodeBoolean(retval); - logger?.info({ materialId, buyerAddress, hasAccess }, 'Chain read entitlement'); - return hasAccess; - } catch (err) { - logger?.error({ err: err.message }, 'Timeout or network error in checkChainEntitlement'); + const retval = result?.results?.[0]?.xdr; + if (!retval) { + logger?.warn({ result }, 'Malformed result in checkChainEntitlement'); return null; } + + const hasAccess = decodeBoolean(retval); + logger?.info({ materialId, buyerAddress, hasAccess }, 'Chain read entitlement'); + return hasAccess; } function buildHasEntitlementXdr(materialId, buyerAddress) { @@ -342,26 +266,8 @@ export async function verifyEntitlementLogic(materialId, buyerAddress, { db, che if (onChain === true) { await setCache(db, materialId, normalised, true, 'chain'); return { hasAccess: true, source: 'chain' }; - const targetCacheKey = CacheEngine.buildKey('entitlements', { - tenant, network, authScope, id: userId - }); - - // 2. Queue the invalidation to the transactional outbox - await db.collection('cache_outbox').insertOne({ - eventId: `evt_${Date.now()}_${Math.random().toString(36).substring(2, 11)}`, - cacheKey: targetCacheKey, - targetRegistry: 'entitlements', - status: 'PENDING', - createdAt: new Date(), - attempts: 0 - }, { session }); - }); - - return result.value; - } finally { - await session.endSession(); } - + if (onChain === false) { await setCache(db, materialId, normalised, false, 'chain'); return { hasAccess: false, source: 'chain-miss' }; @@ -445,5 +351,5 @@ export function requireEntitlement(handler, getMaterialId) { }; } -export { buildHasEntitlementXdr, checkChainEntitlement }; +export { buildHasEntitlementXdr }; diff --git a/src/lib/indexer/stellarIndexer.js b/src/lib/indexer/stellarIndexer.js index 4b44e3a..da4eec2 100644 --- a/src/lib/indexer/stellarIndexer.js +++ b/src/lib/indexer/stellarIndexer.js @@ -7,6 +7,7 @@ import { auditLog } from "../api/audit.js"; import { runWithContext } from "../telemetry/context.js"; import { withSpan } from "../telemetry/tracing.js"; import { decodeContractEvent } from "./eventDecoder.js"; +import { getRpcEvents } from "../stellar/rpcClient.js"; function duplicateKey(error) { return error?.code === 11000; @@ -364,7 +365,6 @@ export async function runIndexerBatch({ db, eventSource, source = "stellar", lim export function createJsonRpcEventSource({ rpcUrl, contractId, - fetchImpl = fetch, networkPassphrase, manifestOverrides, }) { @@ -376,26 +376,10 @@ export function createJsonRpcEventSource({ return { async getEvents({ cursor, limit, startLedger }) { - const response = await fetchImpl(rpcUrl, { - method: "POST", - headers: { "content-type": "application/json" }, - body: JSON.stringify({ - jsonrpc: "2.0", - id: 1, - method: "getEvents", - params: { - ...(cursor ? {} : startLedger ? { startLedger } : {}), - filters: contractIds.length > 0 ? [{ contractIds }] : [], - pagination: { cursor, limit }, - }, - }), - }); - const payload = await response.json(); - if (payload.error) { - throw new Error(payload.error.message || "Stellar RPC getEvents failed"); - } + const filters = contractIds.length > 0 ? [{ contractIds }] : []; + const result = await getRpcEvents({ cursor, limit, startLedger, filters }, { rpcUrl }); - const rawEvents = payload.result?.events || []; + const rawEvents = result?.events || []; // Decode/validate each raw RPC event against the versioned schema // (#7) before it reaches projection code, which expects normalized @@ -420,8 +404,8 @@ export function createJsonRpcEventSource({ return { events, - nextCursor: payload.result?.cursor || null, - lastLedger: payload.result?.latestLedger || null, + nextCursor: result?.cursor || null, + lastLedger: result?.latestLedger || null, }; }, }; diff --git a/src/lib/mongodb.js b/src/lib/mongodb.js index 4ad16b7..48f78d1 100644 --- a/src/lib/mongodb.js +++ b/src/lib/mongodb.js @@ -2,9 +2,31 @@ import { cpus } from "node:os"; import { MongoClient } from "mongodb"; import { ensureChallengeIndexes } from "./auth/challenge.js"; import { updatePressureSignal } from "./capacity/shed.js"; +import { createCircuitBreaker, CircuitState, DependencyError } from "@/lib/resilience/index.js"; +import { withTimeout } from "@/lib/resilience/timeout.js"; +import { setGauge, incrementCounter } from "@/lib/telemetry/metrics.js"; const globalForMongo = globalThis; +const mongoCircuitBreaker = createCircuitBreaker("mongodb", { + failureThreshold: Number(process.env.MONGODB_CB_FAILURE_THRESHOLD || 3), + successThreshold: 2, + resetTimeoutMs: Number(process.env.MONGODB_CB_RESET_TIMEOUT_MS || 30000), + onStateChange(name, from, to) { + setGauge("circuit_breaker_state", { dependency: name, state: to }, 1); + if (from && from !== to) { + setGauge("circuit_breaker_state", { dependency: name, state: from }, 0); + } + if (to === CircuitState.OPEN) { + incrementCounter("circuit_breaker_open_total", { dependency: name }); + } + const level = to === CircuitState.OPEN ? "warn" : "info"; + console[level]( + `[circuit-breaker] mongodb: ${from} -> ${to}`, + ); + }, +}); + function parsePositiveInteger(value, fallback, variableName) { const parsed = Number.parseInt(value ?? String(fallback), 10); @@ -66,7 +88,20 @@ function getMongoConfiguration() { }; } +export function getMongoCircuitBreakerState() { + return mongoCircuitBreaker.getState(); +} + export function getMongoClientPromise() { + if (mongoCircuitBreaker.getState() === CircuitState.OPEN) { + throw new DependencyError({ + dependency: "mongodb", + action: "connect", + retryable: true, + userMessage: "The database is temporarily unavailable. Please try again later.", + }); + } + if (!globalForMongo._mongoClientPromise) { const { uri, clientOptions } = getMongoConfiguration(); const client = new MongoClient(uri, clientOptions); @@ -85,29 +120,32 @@ export function getMongoClientPromise() { // Event monitoring is not available in all MongoDB driver environments. } - globalForMongo._mongoClientPromise = client.connect().catch((error) => { - globalForMongo._mongoClient = null; - globalForMongo._mongoClientPromise = null; - updatePressureSignal("mongoPoolExhausted", true); - - console.error("[mongodb] Connection failed", { - name: error?.name, - code: error?.code, - codeName: error?.codeName, - message: error?.message, - }); - - throw error; - }); + globalForMongo._mongoClientPromise = client.connect().then( + (resolvedClient) => { + mongoCircuitBreaker.recordSuccess(); + return resolvedClient; + }, + (error) => { + globalForMongo._mongoClient = null; + globalForMongo._mongoClientPromise = null; + updatePressureSignal("mongoPoolExhausted", true); + mongoCircuitBreaker.recordFailure(); + + console.error("[mongodb] Connection failed", { + name: error?.name, + code: error?.code, + codeName: error?.codeName, + message: error?.message, + }); + + throw error; + }, + ); } return globalForMongo._mongoClientPromise; } -export function getClientPromise() { - return getMongoClientPromise(); -} - export async function getMongoClient() { return getMongoClientPromise(); } @@ -142,7 +180,7 @@ export async function ensureMongoIndexes() { export async function pingDatabase() { const db = await getDb(); - await db.command({ ping: 1 }); + await withTimeout(db.command({ ping: 1 }), 5000, "mongodb.ping"); return true; } diff --git a/src/lib/pinata.js b/src/lib/pinata.js index 607bbd8..2e08fd0 100644 --- a/src/lib/pinata.js +++ b/src/lib/pinata.js @@ -1,8 +1,58 @@ -// utils/config.ts -// server only -import { PinataSDK } from "pinata"; - -export const pinata = new PinataSDK({ - pinataJwt: process.env.PINATA_JWT, - pinataGateway: process.env.NEXT_PUBLIC_GATEWAY_URL, -}); +import { PinataSDK } from "pinata"; +import { createCircuitBreaker, CircuitState, DependencyError } from "@/lib/resilience/index.js"; +import { withTimeout } from "@/lib/resilience/timeout.js"; +import { withRetry } from "@/lib/resilience/retry.js"; +import { setGauge, incrementCounter } from "@/lib/telemetry/metrics"; + +const PINATA_TIMEOUT_MS = Number(process.env.PINATA_TIMEOUT_MS || 30000); + +const pinataCircuitBreaker = createCircuitBreaker("pinata", { + failureThreshold: Number(process.env.PINATA_CB_FAILURE_THRESHOLD || 3), + successThreshold: 2, + resetTimeoutMs: Number(process.env.PINATA_CB_RESET_TIMEOUT_MS || 30000), + onStateChange(name, from, to) { + setGauge("circuit_breaker_state", { dependency: name, state: to }, 1); + if (from && from !== to) { + setGauge("circuit_breaker_state", { dependency: name, state: from }, 0); + } + if (to === CircuitState.OPEN) { + incrementCounter("circuit_breaker_open_total", { dependency: name }); + } + const level = to === CircuitState.OPEN ? "warn" : "info"; + console[level](`[circuit-breaker] pinata: ${from} -> ${to}`); + }, +}); + +export function getPinataCircuitBreakerState() { + return pinataCircuitBreaker.getState(); +} + +export async function callPinata(action, fn, opts = {}) { + const idempotent = opts.idempotent !== false; + if (pinataCircuitBreaker.getState() === CircuitState.OPEN) { + incrementCounter("rpc_errors_total", { operation: `pinata.${action}` }); + throw new DependencyError({ + dependency: "pinata", + action, + retryable: false, + statusCode: 503, + userMessage: "Storage service is temporarily unavailable.", + }); + } + return pinataCircuitBreaker.call(async () => { + return withRetry( + async () => withTimeout(fn(), PINATA_TIMEOUT_MS, `pinata.${action}`), + { + idempotent, + maxAttempts: Number(process.env.PINATA_RETRIES || 2), + baseDelayMs: 500, + maxDelayMs: 5000, + } + ); + }); +} + +export const pinata = new PinataSDK({ + pinataJwt: process.env.PINATA_JWT, + pinataGateway: process.env.NEXT_PUBLIC_GATEWAY_URL, +}); diff --git a/src/lib/resilience/circuitBreaker.js b/src/lib/resilience/circuitBreaker.js new file mode 100644 index 0000000..785b845 --- /dev/null +++ b/src/lib/resilience/circuitBreaker.js @@ -0,0 +1,162 @@ +/** + * Circuit breaker states. + */ +export const CircuitState = Object.freeze({ + CLOSED: "closed", + OPEN: "open", + HALF_OPEN: "half_open", +}); + +/** + * Create a circuit breaker for a named dependency. + * + * State machine: + * CLOSED → (failure threshold reached) → OPEN + * OPEN → (reset timeout elapsed) → HALF_OPEN + * HALF_OPEN → (probe succeeds) → CLOSED + * HALF_OPEN → (probe fails) → OPEN + * + * State transitions emit a callback (`onStateChange`) so callers can + * update metrics, logs, or readiness checks. + * + * @param {string} name — dependency name (used in metrics/errors) + * @param {object} [opts] + * @param {number} [opts.failureThreshold=5] — consecutive failures before opening + * @param {number} [opts.successThreshold=2] — consecutive successes in half-open to close + * @param {number} [opts.resetTimeoutMs=30000] — time before transitioning open→half-open + * @param {(name: string, from: string, to: string) => void} [opts.onStateChange] + * @returns {{ + * call: (fn: () => Promise, fallback?: () => Promise) => Promise, + * recordSuccess: () => void, + * recordFailure: () => void, + * getState: () => string, + * forceState: (state: string) => void, + * getName: () => string, + * }} + */ +export function createCircuitBreaker(name, opts = {}) { + const { + failureThreshold = 5, + successThreshold = 2, + resetTimeoutMs = 30000, + onStateChange, + } = opts; + + let state = CircuitState.CLOSED; + let failureCount = 0; + let successCount = 0; + let nextAttemptAt = 0; + + function transition(to) { + const from = state; + state = to; + if (onStateChange) { + onStateChange(name, from, to); + } + } + + function resetCounters() { + failureCount = 0; + successCount = 0; + } + + function maybeTransitionToHalfOpen() { + if (state === CircuitState.OPEN && Date.now() >= nextAttemptAt) { + transition(CircuitState.HALF_OPEN); + } + } + + function recordFailure() { + maybeTransitionToHalfOpen(); + + failureCount++; + + if (state === CircuitState.HALF_OPEN) { + // A single failure in half-open trips back to open + transition(CircuitState.OPEN); + nextAttemptAt = Date.now() + resetTimeoutMs; + return; + } + + if (state === CircuitState.CLOSED && failureCount >= failureThreshold) { + transition(CircuitState.OPEN); + nextAttemptAt = Date.now() + resetTimeoutMs; + } + } + + function recordSuccess() { + maybeTransitionToHalfOpen(); + + if (state === CircuitState.HALF_OPEN) { + successCount++; + if (successCount >= successThreshold) { + resetCounters(); + transition(CircuitState.CLOSED); + } + return; + } + + // Reset failure count on success in closed state + if (state === CircuitState.CLOSED) { + failureCount = 0; + } + } + + /** + * Call a function through the circuit breaker. + * + * If the circuit is OPEN, the call is rejected immediately (fast-fail) + * unless a `fallback` is provided, in which case the fallback is called. + * + * @template T + * @param {() => Promise} fn + * @param {() => Promise} [fallback] + * @returns {Promise} + */ + async function call(fn, fallback) { + if (state === CircuitState.OPEN) { + if (Date.now() >= nextAttemptAt) { + transition(CircuitState.HALF_OPEN); + } else { + if (fallback) return fallback(); + const err = new Error(`Circuit breaker open for ${name}`); + err.name = "CircuitBreakerError"; + err.dependency = name; + throw err; + } + } + + try { + const result = await fn(); + recordSuccess(); + return result; + } catch (error) { + recordFailure(); + throw error; + } + } + + function getState() { + return state; + } + + function forceState(s, opts = {}) { + state = s; + failureCount = opts.failureCount ?? 0; + successCount = opts.successCount ?? 0; + nextAttemptAt = opts.nextAttemptAt ?? 0; + } + + function getName() { + return name; + } + + return { + call, + recordSuccess, + recordFailure, + getState, + forceState, + getName, + }; +} diff --git a/src/lib/resilience/classification.js b/src/lib/resilience/classification.js new file mode 100644 index 0000000..208c432 --- /dev/null +++ b/src/lib/resilience/classification.js @@ -0,0 +1,95 @@ +/** + * Idempotency classification for external-boundary operations. + * + * Categories: + * - read — safe to retry, no side effects + * - idempotent_write — safe to retry (same result regardless of count) + * - non_idempotent_write — NOT safe to retry (duplicate causes side effects) + */ + +/** + * @typedef {'read' | 'idempotent_write' | 'non_idempotent_write'} OperationCategory + */ + +/** + * @type {Record} + */ +const CLASSIFICATIONS = { + // ── MongoDB ────────────────────────────────────────────────────────── + "mongodb.find": { category: "read", description: "Database query (no side effects)" }, + "mongodb.aggregate": { category: "read", description: "Database aggregation pipeline" }, + "mongodb.findOne": { category: "read", description: "Single document lookup" }, + "mongodb.insertOne": { category: "non_idempotent_write", description: "Document insert (duplicate creates duplicate docs)" }, + "mongodb.updateOne": { category: "idempotent_write", description: "Document update (same result if repeated)" }, + "mongodb.findOneAndUpdate": { category: "idempotent_write", description: "Atomic find-and-update" }, + "mongodb.replaceOne": { category: "idempotent_write", description: "Document replacement" }, + "mongodb.deleteOne": { category: "idempotent_write", description: "Document deletion (idempotent)" }, + "mongodb.bulkWrite": { category: "idempotent_write", description: "Bulk write (caller manages idempotency via ordered ops)" }, + "mongodb.insertOne.no-idempotency-key": { category: "non_idempotent_write", description: "Insert without idempotency key" }, + "mongodb.command": { category: "idempotent_write", description: "Database command (e.g. ping)" }, + + // ── Stellar Horizon ────────────────────────────────────────────────── + "stellar.loadAccount": { category: "read", description: "Horizon account lookup" }, + "stellar.feeStats": { category: "read", description: "Horizon fee statistics" }, + "stellar.submitTransaction": { category: "non_idempotent_write", description: "Transaction submission (same tx = duplicate on-chain)" }, + "stellar.root": { category: "read", description: "Horizon root endpoint (health)" }, + + // ── Stellar RPC / Soroban ─────────────────────────────────────────── + "stellar.rpc.getEvents": { category: "read", description: "RPC event query" }, + "stellar.rpc.getHealth": { category: "read", description: "RPC health check" }, + "stellar.rpc.getLedgerEntries": { category: "read", description: "RPC ledger entry query" }, + "stellar.rpc.simulateTransaction": { category: "read", description: "RPC transaction simulation (read-only)" }, + "stellar.rpc.sendTransaction": { category: "non_idempotent_write", description: "RPC transaction submission" }, + + // ── Pinata / IPFS ─────────────────────────────────────────────────── + "pinata.testAuthentication": { category: "read", description: "Pinata auth test" }, + "pinata.upload": { category: "idempotent_write", description: "Pinata file upload (CID-based dedup)" }, + "pinata.unpin": { category: "idempotent_write", description: "Pinata unpin (idempotent)" }, + "pinata.list": { category: "read", description: "Pinata file listing" }, + + // ── Nodemailer / Email ────────────────────────────────────────────── + "email.send": { category: "non_idempotent_write", description: "Send email (duplicate sends duplicate emails)" }, + "email.verify": { category: "read", description: "SMTP connection verification" }, + + // ── Redis ──────────────────────────────────────────────────────────── + "redis.get": { category: "read", description: "Cache read" }, + "redis.set": { category: "idempotent_write", description: "Cache write (same key+value)" }, + "redis.del": { category: "idempotent_write", description: "Cache delete" }, + "redis.eval": { category: "idempotent_write", description: "Lua script execution (assumed idempotent)" }, + + // ── Outbound Webhooks ─────────────────────────────────────────────── + "webhook.deliver": { category: "non_idempotent_write", description: "POST to external URL (duplicate = double notification)" }, + + // ── Vercel Blob (reserved for future) ─────────────────────────────── + "blob.put": { category: "idempotent_write", description: "Blob upload (path-based, idempotent)" }, + "blob.delete": { category: "idempotent_write", description: "Blob delete" }, + "blob.list": { category: "read", description: "Blob listing" }, +}; + +/** + * Classify an operation by key. + * + * @param {string} key — dot-separated e.g. "mongodb.findOne" + * @returns {{ category: OperationCategory, description: string }} + */ +export function classifyOperation(key) { + return CLASSIFICATIONS[key] || { category: "non_idempotent_write", description: `Unknown operation: ${key}. Defaulting to non-idempotent.` }; +} + +/** + * Check whether an operation key is safe to retry. + * + * @param {string} key + * @returns {boolean} + */ +export function isIdempotent(key) { + const op = classifyOperation(key); + return op.category === "read" || op.category === "idempotent_write"; +} + +/** + * Return the full classification table (for inspection / testing). + */ +export function getClassificationTable() { + return { ...CLASSIFICATIONS }; +} diff --git a/src/lib/resilience/index.js b/src/lib/resilience/index.js new file mode 100644 index 0000000..b758d9d --- /dev/null +++ b/src/lib/resilience/index.js @@ -0,0 +1,41 @@ +export { withTimeout } from "./timeout.js"; +export { withRetry, defaultIsTransientError } from "./retry.js"; +export { createCircuitBreaker, CircuitState } from "./circuitBreaker.js"; +export { classifyOperation, isIdempotent, getClassificationTable } from "./classification.js"; + +/** + * Application-level error for external dependency failures. + * + * Carries enough structure for API error handlers to produce appropriate + * HTTP responses with user-facing recovery hints and correlation IDs. + */ +export class DependencyError extends Error { + /** + * @param {object} params + * @param {string} params.dependency — e.g. "mongodb", "stellar-horizon" + * @param {string} params.action — e.g. "loadAccount", "insertOne" + * @param {boolean} [params.retryable] — true if caller can retry safely + * @param {number} [params.statusCode] — suggested HTTP status (503, 502, etc.) + * @param {string} [params.userMessage] — user-facing message for error response + * @param {Error} [params.cause] — original error + */ + constructor({ dependency, action, retryable = true, statusCode = 503, userMessage, cause }) { + const msg = `[${dependency}] ${action} failed`; + super(msg); + this.name = "DependencyError"; + this.dependency = dependency; + this.action = action; + this.retryable = retryable; + this.statusCode = statusCode; + this.userMessage = userMessage || `${dependency} is temporarily unavailable. Please try again later.`; + this.cause = cause; + } +} + +/** + * Generalized transient-error checker. + * + * Covers HTTP status codes, Node system error codes, and common message + * keywords. Delegates to the retry module's default predicate. + */ +export { defaultIsTransientError as isTransientError }; diff --git a/src/lib/resilience/retry.js b/src/lib/resilience/retry.js new file mode 100644 index 0000000..f75bb9a --- /dev/null +++ b/src/lib/resilience/retry.js @@ -0,0 +1,100 @@ +/** + * Bounded exponential backoff with full jitter. + * + * delay = random(0, min(maxDelay, baseDelay * 2^attempt)) + */ +function backoffDelay(attempt, baseDelayMs, maxDelayMs) { + const capped = Math.min(maxDelayMs, baseDelayMs * Math.pow(2, attempt)); + return Math.random() * capped; +} + +/** + * Default transient error predicate: checks status codes, error codes, + * and message keywords commonly associated with transient failures. + * + * @param {Error} error + * @returns {boolean} + */ +export function defaultIsTransientError(error) { + if (!error) return false; + + const status = error?.response?.status ?? error?.status ?? error?.statusCode; + if (status === 429 || status === 503 || status === 502 || status === 504) return true; + + const code = error?.code || ""; + if (code === "ECONNRESET" || code === "ECONNREFUSED" || code === "ETIMEDOUT" || code === "EAI_AGAIN") return true; + + const name = error?.name || ""; + if (name === "TimeoutError" || name === "AbortError") return true; + + const message = String(error?.message || "").toLowerCase(); + return ( + message.includes("timeout") || + message.includes("timed out") || + message.includes("econnreset") || + message.includes("econnrefused") || + message.includes("etimedout") || + message.includes("eai_again") || + message.includes("network") || + message.includes("socket") || + message.includes("connection") + ); +} + +/** + * Retry a function on transient failures with bounded exponential backoff + * and full jitter. + * + * **Idempotency guard**: By default (`idempotent: false`), retries are + * skipped entirely — the function is called exactly once and the error is + * propagated immediately. Pass `{ idempotent: true }` to enable retry logic. + * This prevents accidental duplication of non-idempotent side effects. + * + * @param {() => Promise} fn — the operation to retry (must be a factory so it can be called again) + * @param {object} [opts] + * @param {number} [opts.maxAttempts=3] + * @param {number} [opts.baseDelayMs=200] + * @param {number} [opts.maxDelayMs=10000] + * @param {boolean} [opts.idempotent=false] — true to enable retry; false = single attempt + * @param {(error: Error) => boolean} [opts.isTransientError] — defaults to defaultIsTransientError + * @param {(attempt: number, error: Error, delayMs: number) => void} [opts.onRetry] — called before each retry delay + * @returns {Promise} + * @template T + */ +export async function withRetry(fn, opts = {}) { + const { + maxAttempts = 3, + baseDelayMs = 200, + maxDelayMs = 10000, + idempotent = false, + isTransientError = defaultIsTransientError, + onRetry, + } = opts; + + const attempts = Math.max(1, maxAttempts); + + let lastError; + + for (let attempt = 0; attempt < attempts; attempt++) { + try { + return await fn(); + } catch (error) { + lastError = error; + + const isLastAttempt = attempt >= attempts - 1; + const canRetry = idempotent && attempt < attempts - 1 && isTransientError(error); + + if (isLastAttempt || !canRetry) { + break; + } + + const delay = backoffDelay(attempt, baseDelayMs, maxDelayMs); + if (onRetry) { + onRetry(attempt + 1, error, Math.round(delay)); + } + await new Promise((resolve) => setTimeout(resolve, delay)); + } + } + + throw lastError; +} diff --git a/src/lib/resilience/timeout.js b/src/lib/resilience/timeout.js new file mode 100644 index 0000000..7f72ba1 --- /dev/null +++ b/src/lib/resilience/timeout.js @@ -0,0 +1,61 @@ +import { getContext } from "@/lib/telemetry/context"; + +/** + * Race a promise against a deadline, with optional external abort signal. + * + * If both a timeout and an external signal are provided, whichever fires + * first wins — the other timer/listener is cleaned up to avoid leaks. + * + * @param {Promise} promise + * @param {number} ms + * @param {string} label — used in the timeout error message for diagnostics + * @param {{ signal?: AbortSignal }} [options] + * @returns {Promise} + * @template T + */ +export function withTimeout(promise, ms, label, { signal } = {}) { + if (ms <= 0) { + return promise; + } + + let timer; + let abortListener; + + const timeout = new Promise((_, reject) => { + timer = setTimeout(() => { + const ctx = getContext(); + const correlationId = ctx?.correlationId || null; + const err = new Error(`Request timed out after ${ms}ms for ${label}`); + err.name = "TimeoutError"; + err.correlationId = correlationId; + reject(err); + }, ms); + }); + + const externalAbort = signal + ? new Promise((_, reject) => { + abortListener = () => { + const err = new Error(`Request aborted for ${label}: external signal`); + err.name = "AbortError"; + reject(err); + }; + if (signal.aborted) { + abortListener(); + } else { + signal.addEventListener("abort", abortListener, { once: true }); + } + }) + : null; + + const race = externalAbort ? Promise.race([promise, timeout, externalAbort]) : Promise.race([promise, timeout]); + + const cleanup = () => { + clearTimeout(timer); + if (signal && abortListener) { + signal.removeEventListener("abort", abortListener); + } + }; + + race.then(cleanup, cleanup); + return race; +} diff --git a/src/lib/stellar/horizonClient.js b/src/lib/stellar/horizonClient.js index 6d1d625..d0dd786 100644 --- a/src/lib/stellar/horizonClient.js +++ b/src/lib/stellar/horizonClient.js @@ -1,5 +1,24 @@ import { Horizon } from "@stellar/stellar-sdk"; import { HORIZON_URL, isMainnet } from "@/lib/config/chain"; +import { createCircuitBreaker, CircuitState, DependencyError } from "@/lib/resilience/index.js"; +import { setGauge, incrementCounter } from "@/lib/telemetry/metrics.js"; + +const horizonCircuitBreaker = createCircuitBreaker("stellar-horizon", { + failureThreshold: Number(process.env.STELLAR_HORIZON_CB_FAILURE_THRESHOLD || 3), + successThreshold: 2, + resetTimeoutMs: Number(process.env.STELLAR_HORIZON_CB_RESET_TIMEOUT_MS || 30000), + onStateChange(name, from, to) { + setGauge("circuit_breaker_state", { dependency: name, state: to }, 1); + if (from && from !== to) { + setGauge("circuit_breaker_state", { dependency: name, state: from }, 0); + } + if (to === CircuitState.OPEN) { + incrementCounter("circuit_breaker_open_total", { dependency: name }); + } + const level = to === CircuitState.OPEN ? "warn" : "info"; + console[level](`[circuit-breaker] stellar-horizon: ${from} -> ${to}`); + }, +}); const horizonLogger = { warn(data, message) { @@ -88,7 +107,20 @@ async function withTimeout(promise, ms, label) { * @param {{ timeoutMs?: number, retries?: number }} [opts] * @returns {Promise} */ +export function getHorizonCircuitBreakerState() { + return horizonCircuitBreaker.getState(); +} + export async function withFailover(fn, { timeoutMs = DEFAULT_TIMEOUT_MS, retries = DEFAULT_RETRIES } = {}) { + if (horizonCircuitBreaker.getState() === CircuitState.OPEN) { + throw new DependencyError({ + dependency: "stellar-horizon", + action: "withFailover", + retryable: true, + userMessage: "The Stellar network is temporarily unavailable. Please try again later.", + }); + } + const errors = []; for (let attempt = 0; attempt <= retries; attempt++) { @@ -107,6 +139,7 @@ export async function withFailover(fn, { timeoutMs = DEFAULT_TIMEOUT_MS, retries "Horizon failover succeeded", ); } + horizonCircuitBreaker.recordSuccess(); return result; } catch (err) { errors.push({ url, message: err.message }); @@ -142,6 +175,8 @@ export async function withFailover(fn, { timeoutMs = DEFAULT_TIMEOUT_MS, retries .map(({ url, message }) => `${url}: ${message}`) .join(" | "); + horizonCircuitBreaker.recordFailure(); + throw new Error( `All Horizon endpoints failed after ${retries + 1} attempts. Errors: ${summary}`, ); diff --git a/src/lib/stellar/rpcClient.js b/src/lib/stellar/rpcClient.js new file mode 100644 index 0000000..3ca54d4 --- /dev/null +++ b/src/lib/stellar/rpcClient.js @@ -0,0 +1,151 @@ +import { STELLAR_RPC_URL } from "@/lib/config/chain"; +import { createCircuitBreaker, CircuitState, DependencyError } from "@/lib/resilience/index.js"; +import { withTimeout } from "@/lib/resilience/timeout.js"; +import { withRetry } from "@/lib/resilience/retry.js"; +import { setGauge, incrementCounter } from "@/lib/telemetry/metrics.js"; +import { currentTraceparent } from "@/lib/telemetry/context"; + +const DEFAULT_TIMEOUT_MS = Number(process.env.STELLAR_RPC_TIMEOUT_MS || 10000); +const DEFAULT_RETRIES = Number(process.env.STELLAR_RPC_RETRIES || 2); + +const rpcCircuitBreaker = createCircuitBreaker("stellar-rpc", { + failureThreshold: Number(process.env.STELLAR_RPC_CB_FAILURE_THRESHOLD || 3), + successThreshold: 2, + resetTimeoutMs: Number(process.env.STELLAR_RPC_CB_RESET_TIMEOUT_MS || 30000), + onStateChange(name, from, to) { + setGauge("circuit_breaker_state", { dependency: name, state: to }, 1); + if (from && from !== to) { + setGauge("circuit_breaker_state", { dependency: name, state: from }, 0); + } + if (to === CircuitState.OPEN) { + incrementCounter("circuit_breaker_open_total", { dependency: name }); + } + const level = to === CircuitState.OPEN ? "warn" : "info"; + console[level](`[circuit-breaker] stellar-rpc: ${from} -> ${to}`); + }, +}); + +export function getRpcCircuitBreakerState() { + return rpcCircuitBreaker.getState(); +} + +function isRpcTransientError(error) { + if (!error) return false; + const status = error?.response?.status ?? error?.status; + if (status === 429 || status === 503 || status === 502 || status === 504) return true; + const code = error?.code || ""; + if (code === "ECONNRESET" || code === "ECONNREFUSED" || code === "ETIMEDOUT") return true; + const message = String(error?.message || "").toLowerCase(); + return message.includes("timeout") || message.includes("timed out") || message.includes("econnreset") || message.includes("network") || message.includes("socket"); +} + +/** + * Execute a JSON-RPC call to the Stellar RPC endpoint. + * + * @param {object} rpcRequest — { method, params } + * @param {object} [opts] + * @param {string} [opts.rpcUrl] + * @param {number} [opts.timeoutMs] + * @returns {Promise} — JSON-RPC result + */ +export async function rpcCall(rpcRequest, opts = {}) { + const rpcUrl = opts.rpcUrl || STELLAR_RPC_URL; + const timeoutMs = opts.timeoutMs || DEFAULT_TIMEOUT_MS; + + if (!rpcUrl) { + throw new DependencyError({ + dependency: "stellar-rpc", + action: rpcRequest.method, + retryable: false, + userMessage: "Stellar RPC is not configured.", + }); + } + + const traceparent = currentTraceparent(); + + const exec = async () => { + const body = JSON.stringify({ + jsonrpc: "2.0", + id: Date.now(), + method: rpcRequest.method, + params: rpcRequest.params || {}, + }); + + const headers = { + "Content-Type": "application/json", + }; + if (traceparent) { + headers["traceparent"] = traceparent; + } + + const timerLabel = `stellar-rpc.${rpcRequest.method}`; + const res = await withTimeout(fetch(rpcUrl, { method: "POST", headers, body }), timeoutMs, timerLabel); + + const text = await res.text(); + let payload; + try { + payload = JSON.parse(text); + } catch { + throw Object.assign(new Error(`Stellar RPC returned non-JSON response: ${res.status}`), { status: res.status }); + } + + if (payload.error) { + const err = new Error(payload.error.message || `Stellar RPC error: ${rpcRequest.method}`); + err.code = payload.error.code; + err.status = payload.error.code; + throw err; + } + + return payload.result; + }; + + try { + return await rpcCircuitBreaker.call(async () => { + return withRetry(exec, { + idempotent: true, + maxAttempts: DEFAULT_RETRIES, + baseDelayMs: 500, + maxDelayMs: 5000, + isTransientError: isRpcTransientError, + }); + }); + } catch (error) { + incrementCounter("rpc_errors_total", { operation: rpcRequest.method }); + throw error; + } +} + +/** + * Simulate a Soroban contract transaction. + */ +export async function simulateTransaction(xdr, opts = {}) { + return rpcCall({ method: "simulateTransaction", params: { transaction: xdr } }, opts); +} + +/** + * Fetch events from the Stellar RPC. + * + * @param {object} [query] + * @param {string} [query.cursor] - Pagination cursor + * @param {number} [query.limit] - Page size + * @param {string} [query.startLedger] - Start ledger sequence + * @param {Array<{contractIds?: string[], type?: string}>} [query.filters] - Event filters + */ +export async function getRpcEvents({ cursor, limit, startLedger, filters } = {}, opts = {}) { + const params = {}; + if (startLedger) params.startLedger = startLedger; + if (filters && filters.length > 0) params.filters = filters; + if (cursor || limit) { + params.pagination = {}; + if (cursor) params.pagination.cursor = cursor; + if (limit) params.pagination.limit = limit; + } + return rpcCall({ method: "getEvents", params }, opts); +} + +/** + * Get RPC health status. + */ +export async function getRpcHealth(opts = {}) { + return rpcCall({ method: "getHealth" }, opts); +} diff --git a/src/lib/telemetry/context.js b/src/lib/telemetry/context.js index 470a2a5..b683658 100644 --- a/src/lib/telemetry/context.js +++ b/src/lib/telemetry/context.js @@ -24,6 +24,7 @@ import { AsyncLocalStorage } from "node:async_hooks"; let _storage = null; let _degraded = false; +let _AsyncLocalStorage; /** No-op storage for browser bundles, where async_hooks does not exist. */ class NoopAsyncLocalStorage { diff --git a/src/lib/webhooks/dispatcher.js b/src/lib/webhooks/dispatcher.js index 8f34798..5be3f8b 100644 --- a/src/lib/webhooks/dispatcher.js +++ b/src/lib/webhooks/dispatcher.js @@ -2,6 +2,30 @@ import https from 'node:https'; import dns from 'node:dns/promises'; import { URL } from 'node:url'; import net from 'node:net'; +import { createCircuitBreaker, CircuitState, DependencyError } from "@/lib/resilience/index.js"; +import { setGauge, incrementCounter } from "@/lib/telemetry/metrics"; +import { currentTraceparent } from "@/lib/telemetry/context"; + +const webhookCircuitBreaker = createCircuitBreaker("webhook", { + failureThreshold: Number(process.env.WEBHOOK_CB_FAILURE_THRESHOLD || 5), + successThreshold: 2, + resetTimeoutMs: Number(process.env.WEBHOOK_CB_RESET_TIMEOUT_MS || 30000), + onStateChange(name, from, to) { + setGauge("circuit_breaker_state", { dependency: name, state: to }, 1); + if (from && from !== to) { + setGauge("circuit_breaker_state", { dependency: name, state: from }, 0); + } + if (to === CircuitState.OPEN) { + incrementCounter("circuit_breaker_open_total", { dependency: name }); + } + const level = to === CircuitState.OPEN ? "warn" : "info"; + console[level](`[circuit-breaker] webhook: ${from} -> ${to}`); + }, +}); + +export function getWebhookCircuitBreakerState() { + return webhookCircuitBreaker.getState(); +} export function isPrivateIP(ip) { if (net.isIPv4(ip)) { @@ -50,6 +74,20 @@ export function isPrivateIP(ip) { } export async function dispatchWebhook(url, payloadStr, signatureHeader) { + if (webhookCircuitBreaker.getState() === CircuitState.OPEN) { + incrementCounter("rpc_errors_total", { operation: "webhook.dispatch" }); + throw new DependencyError({ + dependency: "webhook", + action: "dispatch", + retryable: false, + statusCode: 503, + userMessage: "Webhook delivery service is temporarily unavailable.", + }); + } + return webhookCircuitBreaker.call(async () => _dispatchWebhook(url, payloadStr, signatureHeader)); +} + +async function _dispatchWebhook(url, payloadStr, signatureHeader) { let currentUrl = url; let redirects = 0; const maxRedirects = 3; @@ -108,6 +146,11 @@ export async function dispatchWebhook(url, payloadStr, signatureHeader) { requestOptions.headers['Eduvault-Signature'] = signatureHeader; } + const traceparent = currentTraceparent(); + if (traceparent) { + requestOptions.headers['traceparent'] = traceparent; + } + const response = await new Promise((resolve, reject) => { const req = https.request(requestOptions, (res) => { if (res.statusCode >= 300 && res.statusCode < 400 && res.headers.location) { diff --git a/test/fixtures/resilience.js b/test/fixtures/resilience.js new file mode 100644 index 0000000..4c23580 --- /dev/null +++ b/test/fixtures/resilience.js @@ -0,0 +1,159 @@ +import { vi } from "vitest"; +import { CircuitState } from "@/lib/resilience/circuitBreaker"; + +/** + * Track all circuit breakers created in tests so they can be reset. + * @type {Set<{ forceState: (s: string) => void, getName: () => string }>} + */ +const allCircuitBreakers = new Set(); + +/** + * Register a circuit breaker for auto-cleanup. + * Called by test code when creating breakers. + */ +export function trackCircuitBreaker(cb) { + allCircuitBreakers.add(cb); + return cb; +} + +/** + * Reset all tracked circuit breakers to closed state. + */ +export function resetCircuitBreakers() { + for (const cb of allCircuitBreakers) { + cb.forceState(CircuitState.CLOSED, {}); + } + allCircuitBreakers.clear(); +} + +/** + * Assert that a circuit breaker is in the expected state. + */ +export function expectCircuitState(cb, expectedState) { + const actual = cb.getState(); + if (actual !== expectedState) { + throw new Error( + `Expected circuit breaker "${cb.getName()}" to be in state "${expectedState}", got "${actual}"`, + ); + } +} + +/** + * Create a mock adapter function that simulates various failure modes. + * + * @param {object} opts + * @param {'latency' | 'timeout' | 'connection_reset' | 'http_429' | 'http_5xx'} opts.mode + * @param {number} [opts.delayMs] — how long to wait before responding (latency mode) + * @param {number} [opts.failAfter] — number of calls before switching from success to failure + * @param {number} [opts.statusCode] — for http_5xx mode, the specific status code + * @param {boolean} [opts.thenRecover] — after failing, start succeeding again? + * @returns {{ fn: (...args) => Promise, reset: () => void }} + */ +export function createFaultyAdapter(opts = {}) { + const { + mode, + delayMs = 0, + failAfter = 0, + statusCode = 500, + thenRecover = false, + } = opts; + + let callCount = 0; + let failureCount = 0; + const maxFailures = failAfter; + + async function fn(...args) { + callCount++; + + // Latency simulation + if (delayMs > 0) { + await new Promise((r) => setTimeout(r, delayMs)); + } + + // Determine if this call should fail + const shouldFail = maxFailures > 0 && failureCount < maxFailures; + const isRecovered = thenRecover && failureCount >= maxFailures; + + if (shouldFail) { + failureCount++; + + switch (mode) { + case "timeout": { + const err = new Error("simulated timeout"); + err.name = "TimeoutError"; + throw err; + } + case "connection_reset": { + const err = new Error("read ECONNRESET"); + err.code = "ECONNRESET"; + throw err; + } + case "http_429": { + const err = new Error("Too Many Requests"); + err.status = 429; + err.response = { status: 429 }; + throw err; + } + case "http_5xx": { + const err = new Error(`HTTP ${statusCode}`); + err.status = statusCode; + err.response = { status: statusCode }; + throw err; + } + case "latency": + // Latency mode fails with a timeout after the delay + const err = new Error("simulated timeout after latency"); + err.name = "TimeoutError"; + throw err; + default: + throw new Error(`Unknown fault mode: ${mode}`); + } + } + + // If recovered from failures, return success + if (isRecovered) { + return { status: "ok", recovered: true }; + } + + // Success path (when failAfter === 0) + return { status: "ok" }; + } + + function reset() { + callCount = 0; + failureCount = 0; + } + + return { fn, reset }; +} + +/** + * Create a mock that simulates a flaky network connection. + * Alternates between success and failure to test retry logic. + * + * @param {number} successRate — probability of success (0.0 to 1.0) + * @param {number} [delayMs] — simulated latency + * @returns {() => Promise} + */ +export function createFlakyAdapter(successRate = 0.5, delayMs = 0) { + return async (...args) => { + if (delayMs > 0) { + await new Promise((r) => setTimeout(r, delayMs)); + } + + if (Math.random() < successRate) { + return { status: "ok" }; + } + + const err = new Error("Flaky network error"); + err.code = "ECONNRESET"; + throw err; + }; +} + +/** + * Clean slate before each test: clear circuit breaker tracking. + */ +export function resetResilienceFixtures() { + allCircuitBreakers.clear(); +} diff --git a/test/integration/resilience/circuitBreaker.test.js b/test/integration/resilience/circuitBreaker.test.js new file mode 100644 index 0000000..2ed9c14 --- /dev/null +++ b/test/integration/resilience/circuitBreaker.test.js @@ -0,0 +1,188 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { createCircuitBreaker, CircuitState } from "@/lib/resilience/circuitBreaker"; + +describe("createCircuitBreaker", () => { + let breaker; + + beforeEach(() => { + vi.useFakeTimers(); + breaker = createCircuitBreaker("test-dep", { + failureThreshold: 3, + successThreshold: 2, + resetTimeoutMs: 10000, + }); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + describe("initial state", () => { + it("starts closed", () => { + expect(breaker.getState()).toBe(CircuitState.CLOSED); + }); + + it("has the correct name", () => { + expect(breaker.getName()).toBe("test-dep"); + }); + }); + + describe("closed state", () => { + it("calls the function and returns result", async () => { + const fn = vi.fn().mockResolvedValue("ok"); + await expect(breaker.call(fn)).resolves.toBe("ok"); + }); + + it("records failures and transitions to open at threshold", () => { + expect(breaker.getState()).toBe(CircuitState.CLOSED); + + // 3 failures should trip the breaker + for (let i = 0; i < 3; i++) { + breaker.recordFailure(); + } + + expect(breaker.getState()).toBe(CircuitState.OPEN); + }); + + it("resets failure count on success", () => { + breaker.recordFailure(); + breaker.recordFailure(); + breaker.recordSuccess(); + breaker.recordFailure(); + + // Should still be closed — success reset the count + expect(breaker.getState()).toBe(CircuitState.CLOSED); + }); + + it("transitions to open only after reaching threshold", () => { + breaker.recordFailure(); + breaker.recordFailure(); + expect(breaker.getState()).toBe(CircuitState.CLOSED); + + breaker.recordFailure(); + expect(breaker.getState()).toBe(CircuitState.OPEN); + }); + + it("propagates errors from the wrapped function", async () => { + const fn = vi.fn().mockRejectedValue(new Error("boom")); + await expect(breaker.call(fn)).rejects.toThrow("boom"); + }); + + it("counts failures from call() rejections", async () => { + const fn = vi.fn().mockRejectedValue(new Error("boom")); + for (let i = 0; i < 3; i++) { + await expect(breaker.call(fn)).rejects.toThrow("boom"); + } + expect(breaker.getState()).toBe(CircuitState.OPEN); + }); + }); + + describe("open state", () => { + beforeEach(() => { + // Trip the breaker + for (let i = 0; i < 3; i++) { + breaker.recordFailure(); + } + }); + + it("rejects calls immediately with CircuitBreakerError", async () => { + const fn = vi.fn().mockResolvedValue("should not reach"); + await expect(breaker.call(fn)).rejects.toThrow("Circuit breaker open for test-dep"); + expect(fn).not.toHaveBeenCalled(); + }); + + it("calls fallback when provided instead of rejecting", async () => { + const fn = vi.fn().mockResolvedValue("should not reach"); + const fallback = vi.fn().mockResolvedValue("fallback value"); + + await expect(breaker.call(fn, fallback)).resolves.toBe("fallback value"); + expect(fn).not.toHaveBeenCalled(); + expect(fallback).toHaveBeenCalledTimes(1); + }); + + it("auto-transitions to half_open after reset timeout", () => { + expect(breaker.getState()).toBe(CircuitState.OPEN); + + // Advance past reset timeout + vi.advanceTimersByTime(10000); + + // create a dummy call to trigger transition check + const fn = vi.fn().mockResolvedValue("probe"); + breaker.call(fn); // don't await, just trigger the state check + + expect(breaker.getState()).toBe(CircuitState.HALF_OPEN); + }); + }); + + describe("half_open state", () => { + beforeEach(() => { + for (let i = 0; i < 3; i++) { + breaker.recordFailure(); + } + vi.advanceTimersByTime(10000); + }); + + it("transitions to closed after enough probe successes", () => { + // Trigger state transition by calling + breaker.recordSuccess(); + expect(breaker.getState()).toBe(CircuitState.HALF_OPEN); + + breaker.recordSuccess(); + expect(breaker.getState()).toBe(CircuitState.CLOSED); + }); + + it("transitions back to open on probe failure", () => { + breaker.recordSuccess(); + expect(breaker.getState()).toBe(CircuitState.HALF_OPEN); + + breaker.recordFailure(); + expect(breaker.getState()).toBe(CircuitState.OPEN); + }); + + it("requires successThreshold successes to close", async () => { + breaker.recordSuccess(); + expect(breaker.getState()).toBe(CircuitState.HALF_OPEN); + + // One more success should close + breaker.recordSuccess(); + expect(breaker.getState()).toBe(CircuitState.CLOSED); + }); + }); + + describe("forceState", () => { + it("forces the breaker into a given state", () => { + breaker.forceState(CircuitState.OPEN, {}); + expect(breaker.getState()).toBe(CircuitState.OPEN); + + breaker.forceState(CircuitState.CLOSED, {}); + expect(breaker.getState()).toBe(CircuitState.CLOSED); + }); + }); + + describe("onStateChange callback", () => { + it("calls onStateChange on each transition", () => { + const onStateChange = vi.fn(); + const cb = createCircuitBreaker("test-cb", { + failureThreshold: 2, + onStateChange, + }); + + cb.recordFailure(); + cb.recordFailure(); + expect(onStateChange).toHaveBeenCalledWith("test-cb", CircuitState.CLOSED, CircuitState.OPEN); + }); + + it("provides from and to state names", () => { + const onStateChange = vi.fn(); + const cb = createCircuitBreaker("named-cb", { + failureThreshold: 2, + onStateChange, + }); + + cb.recordFailure(); + cb.recordFailure(); + + expect(onStateChange).toHaveBeenCalledWith("named-cb", "closed", "open"); + }); + }); +}); diff --git a/test/integration/resilience/email.test.js b/test/integration/resilience/email.test.js new file mode 100644 index 0000000..3d7ac05 --- /dev/null +++ b/test/integration/resilience/email.test.js @@ -0,0 +1,34 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { CircuitState } from "@/lib/resilience/circuitBreaker"; + +vi.mock("@/lib/telemetry/metrics", () => ({ setGauge: vi.fn(), incrementCounter: vi.fn() })); +vi.mock("@/lib/telemetry/context", () => ({ currentTraceparent: vi.fn() })); +vi.mock("nodemailer", () => ({ + default: { createTransport: vi.fn(() => ({ sendMail: vi.fn().mockResolvedValue("ok"), verify: vi.fn().mockResolvedValue("ok") })) }, +})); + +describe("Email resilience", () => { + beforeEach(() => { + vi.resetModules(); + }); + + it("exports getEmailCircuitBreakerState", async () => { + const { getEmailCircuitBreakerState } = await import("@/lib/email"); + const state = getEmailCircuitBreakerState(); + expect([CircuitState.CLOSED, CircuitState.OPEN, CircuitState.HALF_OPEN]).toContain(state); + }); + + it("sendWelcomeEmail succeeds", async () => { + const { sendWelcomeEmail } = await import("@/lib/email"); + process.env.EMAIL_USER = "test@test.com"; + process.env.EMAIL_PASS = "pass"; + await expect(sendWelcomeEmail("to@test.com", "Test")).resolves.toBeUndefined(); + }); + + it("verifyEmailConnection succeeds", async () => { + const { verifyEmailConnection } = await import("@/lib/email"); + process.env.EMAIL_USER = "test@test.com"; + process.env.EMAIL_PASS = "pass"; + await expect(verifyEmailConnection()).resolves.toBeUndefined(); + }); +}); diff --git a/test/integration/resilience/hardening.test.js b/test/integration/resilience/hardening.test.js new file mode 100644 index 0000000..97505b8 --- /dev/null +++ b/test/integration/resilience/hardening.test.js @@ -0,0 +1,62 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { DependencyError } from "@/lib/resilience/index"; + +vi.mock("next/server", () => ({ + NextResponse: { + json: vi.fn((body, init) => ({ body, status: init?.status || 200, headers: new Map(), headersSet: {} })), + }, +})); + +vi.mock("@/lib/api/audit", () => ({ auditLog: vi.fn() })); +vi.mock("@/lib/api/rateLimit", () => ({ checkRateLimit: vi.fn(() => ({ allowed: true, limit: 100, remaining: 99, retryAfter: 0 })) })); +vi.mock("@/lib/sentry", () => ({ captureException: vi.fn() })); +vi.mock("@/lib/telemetry/context", () => ({ runWithContext: vi.fn((ctx, fn) => fn()), currentTraceparent: vi.fn(), currentCorrelationId: vi.fn(() => "test-cid") })); +vi.mock("@/lib/telemetry/tracing", () => ({ withSpan: vi.fn((n, o, fn) => fn({ setAttribute: vi.fn() })) })); +vi.mock("@/lib/telemetry/metrics", () => ({ incrementCounter: vi.fn(), recordHistogram: vi.fn() })); +vi.mock("@/lib/capacity/concurrency", () => ({ acquireSlot: vi.fn(() => ({ acquired: true, release: vi.fn(), overload: false })) })); +vi.mock("@/lib/capacity/shed", () => ({ preRequestShed: vi.fn(() => ({ shed: false })) })); +vi.mock("@/lib/capacity/budgets", () => ({ resolveRouteBudget: vi.fn(() => ({ maxPayloadBytes: 0 })) })); +vi.mock("@/lib/capacity/backpressure", () => ({ createDisconnectSignal: vi.fn(() => ({ signal: null, cleanup: vi.fn() })) })); +vi.mock("@/lib/api/contract", () => ({ enforceApiResponse: vi.fn((r) => r), negotiateApiVersion: vi.fn() })); +vi.mock("@/lib/security/clientAddress", () => ({ resolveTrustedClientIp: vi.fn(() => "127.0.0.1") })); + +function makeRequest() { + const store = {}; + const h = { + get: (k) => store[k] || null, + set: (k, v) => { store[k] = v; }, + }; + return { method: "GET", headers: h, nextUrl: { pathname: "/api/test" }, url: "http://localhost/api/test" }; +} + +describe("Hardening layer DependencyError handling", () => { + beforeEach(() => { + vi.clearAllMocks(); + }); + + it("maps DependencyError to 503 with userMessage", async () => { + const { withApiHardening } = await import("@/lib/api/hardening"); + const { NextResponse } = await import("next/server"); + const handler = vi.fn().mockRejectedValue( + new DependencyError({ dependency: "test-dep", action: "test-action", userMessage: "Test is down" }) + ); + + await withApiHardening(makeRequest(), { route: "test-route" }, handler); + expect(NextResponse.json).toHaveBeenCalledWith( + expect.objectContaining({ error: "Test is down", dependency: "test-dep" }), + expect.objectContaining({ status: 503 }) + ); + }); + + it("generic error maps to 500", async () => { + const { withApiHardening } = await import("@/lib/api/hardening"); + const { NextResponse } = await import("next/server"); + const handler = vi.fn().mockRejectedValue(new Error("something broke")); + + await withApiHardening(makeRequest(), { route: "test-route" }, handler); + expect(NextResponse.json).toHaveBeenCalledWith( + expect.objectContaining({ error: "Internal Server Error" }), + expect.objectContaining({ status: 500 }) + ); + }); +}); diff --git a/test/integration/resilience/pinata.test.js b/test/integration/resilience/pinata.test.js new file mode 100644 index 0000000..b3ba85f --- /dev/null +++ b/test/integration/resilience/pinata.test.js @@ -0,0 +1,49 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { CircuitState } from "@/lib/resilience/circuitBreaker"; + +vi.mock("@/lib/telemetry/metrics", () => ({ setGauge: vi.fn(), incrementCounter: vi.fn() })); +vi.mock("@/lib/telemetry/context", () => ({ currentTraceparent: vi.fn() })); +vi.mock("pinata", () => ({ + PinataSDK: vi.fn().mockImplementation(function() { + return { + testAuthentication: vi.fn(), + upload: { public: { file: vi.fn(), json: vi.fn() } }, + gateways: { public: { convert: vi.fn() } }, + unpin: vi.fn(), + }; + }), +})); + +describe("Pinata resilience", () => { + beforeEach(() => { + vi.resetModules(); + }); + + it("exports getPinataCircuitBreakerState", async () => { + const { getPinataCircuitBreakerState } = await import("@/lib/pinata"); + const state = getPinataCircuitBreakerState(); + expect([CircuitState.CLOSED, CircuitState.OPEN, CircuitState.HALF_OPEN]).toContain(state); + }); + + it("callPinata wraps with CB and returns result on success", async () => { + const { callPinata } = await import("@/lib/pinata"); + const fn = vi.fn().mockResolvedValue("ok"); + await expect(callPinata("test", fn)).resolves.toBe("ok"); + expect(fn).toHaveBeenCalledOnce(); + }); + + it("callPinata throws DependencyError when CB is open", async () => { + const mod = await import("@/lib/pinata"); + for (let i = 0; i < 3; i++) { + try { await mod.callPinata("test", () => Promise.reject(new Error("fail"))); } catch {} + } + const err = await mod.callPinata("test", () => Promise.resolve("ok")).catch(e => e); + expect(err.name).toBe("DependencyError"); + expect(err.userMessage).toBe("Storage service is temporarily unavailable."); + }); + + it("exports original pinata instance", async () => { + const { pinata } = await import("@/lib/pinata"); + expect(pinata).toBeDefined(); + }); +}); diff --git a/test/integration/resilience/redis.test.js b/test/integration/resilience/redis.test.js new file mode 100644 index 0000000..bb51474 --- /dev/null +++ b/test/integration/resilience/redis.test.js @@ -0,0 +1,54 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { CircuitState } from "@/lib/resilience/circuitBreaker"; + +const mockRedis = { + get: vi.fn(), + set: vi.fn(), + del: vi.fn(), + on: vi.fn(), + connect: vi.fn().mockResolvedValue(), +}; + +vi.mock("@/lib/telemetry/metrics", () => ({ setGauge: vi.fn(), incrementCounter: vi.fn() })); +vi.mock("@/lib/telemetry/context", () => ({ currentTraceparent: vi.fn() })); +vi.mock("redis", () => ({ createClient: vi.fn(() => mockRedis) })); + +describe("Redis resilience", () => { + beforeEach(() => { + vi.resetModules(); + vi.clearAllMocks(); + process.env.REDIS_URL = "redis://localhost:6379"; + }); + + it("exports getRedisCircuitBreakerState", async () => { + const { getRedisCircuitBreakerState } = await import("@/lib/cache/redis"); + const state = getRedisCircuitBreakerState(); + expect([CircuitState.CLOSED, CircuitState.OPEN, CircuitState.HALF_OPEN]).toContain(state); + }); + + it("cacheGet returns null when key missing", async () => { + mockRedis.get.mockResolvedValue(null); + const { cacheGet } = await import("@/lib/cache/redis"); + await expect(cacheGet("missing")).resolves.toBeNull(); + }); + + it("cacheGet returns parsed value when key exists", async () => { + mockRedis.get.mockResolvedValue(JSON.stringify({ foo: "bar" })); + const { cacheGet } = await import("@/lib/cache/redis"); + await expect(cacheGet("exists")).resolves.toEqual({ foo: "bar" }); + }); + + it("cacheSet writes to redis", async () => { + mockRedis.set.mockResolvedValue("OK"); + const { cacheSet } = await import("@/lib/cache/redis"); + await expect(cacheSet("key", { data: 1 }, 300)).resolves.toBeUndefined(); + expect(mockRedis.set).toHaveBeenCalled(); + }); + + it("cacheDel removes key", async () => { + mockRedis.del.mockResolvedValue(1); + const { cacheDel } = await import("@/lib/cache/redis"); + await expect(cacheDel("key")).resolves.toBeUndefined(); + expect(mockRedis.del).toHaveBeenCalledWith("key"); + }); +}); diff --git a/test/integration/resilience/retry.test.js b/test/integration/resilience/retry.test.js new file mode 100644 index 0000000..7d89ff8 --- /dev/null +++ b/test/integration/resilience/retry.test.js @@ -0,0 +1,142 @@ +import { describe, it, expect, vi } from "vitest"; +import { withRetry, defaultIsTransientError } from "@/lib/resilience/retry"; + +// Small base delay so tests don't wait long with real timers +const FAST = { baseDelayMs: 1, maxDelayMs: 10 }; + +describe("withRetry", () => { + it("succeeds on first attempt", async () => { + const fn = vi.fn().mockResolvedValue("ok"); + await expect(withRetry(fn, { idempotent: true })).resolves.toBe("ok"); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it("retries on transient failure and eventually succeeds", async () => { + const fn = vi + .fn() + .mockRejectedValueOnce(Object.assign(new Error("timeout"), { name: "TimeoutError" })) + .mockRejectedValueOnce(Object.assign(new Error("ECONNRESET"), { code: "ECONNRESET" })) + .mockResolvedValue("recovered"); + + await expect(withRetry(fn, { idempotent: true, ...FAST })).resolves.toBe("recovered"); + expect(fn).toHaveBeenCalledTimes(3); + }, 15000); + + it("stops retrying after maxAttempts and throws last error", async () => { + const error = Object.assign(new Error("persistent"), { code: "ECONNRESET" }); + const fn = vi.fn().mockRejectedValue(error); + + await expect(withRetry(fn, { idempotent: true, maxAttempts: 3, ...FAST })).rejects.toThrow("persistent"); + expect(fn).toHaveBeenCalledTimes(3); + }, 15000); + + it("does not retry on non-transient errors", async () => { + const error = new Error("bad request"); + error.status = 400; + const fn = vi.fn().mockRejectedValue(error); + + await expect(withRetry(fn, { idempotent: true, maxAttempts: 3 })).rejects.toThrow("bad request"); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it("does not retry when idempotent is false (default)", async () => { + const error = Object.assign(new Error("transient"), { code: "ECONNRESET" }); + const fn = vi.fn().mockRejectedValue(error); + + await expect(withRetry(fn, { maxAttempts: 3 })).rejects.toThrow("transient"); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it("retries when idempotent is true, even for transient errors", async () => { + const fn = vi + .fn() + .mockRejectedValueOnce(Object.assign(new Error("transient"), { code: "ECONNRESET" })) + .mockResolvedValue("ok"); + + await expect(withRetry(fn, { idempotent: true, ...FAST })).resolves.toBe("ok"); + expect(fn).toHaveBeenCalledTimes(2); + }, 15000); + + it("calls onRetry callback with attempt info before each delay", async () => { + const onRetry = vi.fn(); + const fn = vi + .fn() + .mockRejectedValueOnce(Object.assign(new Error("fail1"), { code: "ECONNRESET" })) + .mockRejectedValueOnce(Object.assign(new Error("fail2"), { code: "ECONNRESET" })) + .mockResolvedValue("ok"); + + await withRetry(fn, { + idempotent: true, + maxAttempts: 3, + ...FAST, + onRetry, + }); + + expect(onRetry).toHaveBeenCalledTimes(2); + expect(onRetry).toHaveBeenNthCalledWith(1, 1, expect.objectContaining({ message: "fail1" }), expect.any(Number)); + expect(onRetry).toHaveBeenNthCalledWith(2, 2, expect.objectContaining({ message: "fail2" }), expect.any(Number)); + }, 15000); + + it("uses custom isTransientError predicate", async () => { + const isTransient = vi.fn().mockReturnValue(false); + const fn = vi.fn().mockRejectedValue(new Error("custom transient")); + + await expect(withRetry(fn, { idempotent: true, isTransientError: isTransient })).rejects.toThrow("custom transient"); + expect(isTransient).toHaveBeenCalled(); + expect(fn).toHaveBeenCalledTimes(1); + }); + + it("respects maxAttempts = 1 (no retry)", async () => { + const fn = vi.fn().mockRejectedValue(Object.assign(new Error("fail"), { code: "ECONNRESET" })); + + await expect(withRetry(fn, { idempotent: true, maxAttempts: 1 })).rejects.toThrow("fail"); + expect(fn).toHaveBeenCalledTimes(1); + }); +}); + +describe("defaultIsTransientError", () => { + it("returns true for 429 status", () => { + const err = new Error("rate limited"); + err.status = 429; + expect(defaultIsTransientError(err)).toBe(true); + }); + + it("returns true for 503 status", () => { + const err = new Error("unavailable"); + err.status = 503; + expect(defaultIsTransientError(err)).toBe(true); + }); + + it("returns true for ECONNRESET code", () => { + const err = new Error("connection reset"); + err.code = "ECONNRESET"; + expect(defaultIsTransientError(err)).toBe(true); + }); + + it("returns true for TimeoutError name", () => { + const err = new Error("timed out"); + err.name = "TimeoutError"; + expect(defaultIsTransientError(err)).toBe(true); + }); + + it("returns false for 400 status", () => { + const err = new Error("bad request"); + err.status = 400; + expect(defaultIsTransientError(err)).toBe(false); + }); + + it("returns false for null/undefined", () => { + expect(defaultIsTransientError(null)).toBe(false); + expect(defaultIsTransientError(undefined)).toBe(false); + }); + + it("detects timeout in error message", () => { + const err = new Error("request timed out"); + expect(defaultIsTransientError(err)).toBe(true); + }); + + it("detects network in error message", () => { + const err = new Error("network error"); + expect(defaultIsTransientError(err)).toBe(true); + }); +}); diff --git a/test/integration/resilience/timeout.test.js b/test/integration/resilience/timeout.test.js new file mode 100644 index 0000000..a05effd --- /dev/null +++ b/test/integration/resilience/timeout.test.js @@ -0,0 +1,133 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; +import { withTimeout } from "@/lib/resilience/timeout"; + +describe("withTimeout", () => { + beforeEach(() => { + vi.useFakeTimers(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("resolves successfully when promise completes before deadline", async () => { + const promise = Promise.resolve("ok"); + await expect(withTimeout(promise, 1000, "test")).resolves.toBe("ok"); + }); + + it("rejects with TimeoutError when promise exceeds deadline", async () => { + const slow = new Promise((resolve) => setTimeout(resolve, 2000)); + const result = withTimeout(slow, 100, "test"); + + vi.advanceTimersByTime(100); + + await expect(result).rejects.toThrow("timed out after 100ms"); + await expect(result).rejects.toHaveProperty("name", "TimeoutError"); + }); + + it("returns resolved value when deadline matches exactly", async () => { + const promise = new Promise((resolve) => setTimeout(() => resolve("exact"), 50)); + const result = withTimeout(promise, 100, "test"); + + vi.advanceTimersByTime(50); + + await expect(result).resolves.toBe("exact"); + }); + + it("returns the promise result immediately when ms is 0 or negative", async () => { + await expect(withTimeout(Promise.resolve("fast"), 0, "test")).resolves.toBe("fast"); + await expect(withTimeout(Promise.resolve("fast"), -1, "test")).resolves.toBe("fast"); + }); + + it("rejects immediately when external signal is already aborted", async () => { + const controller = new AbortController(); + controller.abort(); + + const promise = new Promise((resolve) => setTimeout(resolve, 1000)); + const result = withTimeout(promise, 5000, "test", { signal: controller.signal }); + + await expect(result).rejects.toThrow("external signal"); + await expect(result).rejects.toHaveProperty("name", "AbortError"); + }); + + it("rejects on external signal before timeout", async () => { + const controller = new AbortController(); + const promise = new Promise((resolve) => setTimeout(resolve, 2000)); + const result = withTimeout(promise, 5000, "test", { signal: controller.signal }); + + // Fire abort before timeout + controller.abort(); + + await expect(result).rejects.toThrow("external signal"); + }); + + it("rejects on timeout before external signal", async () => { + const controller = new AbortController(); + const promise = new Promise((resolve) => setTimeout(resolve, 2000)); + const result = withTimeout(promise, 100, "test", { signal: controller.signal }); + + vi.advanceTimersByTime(100); + + await expect(result).rejects.toThrow("timed out after 100ms"); + // Signal didn't fire — verify controller not aborted + expect(controller.signal.aborted).toBe(false); + }); + + it("removes abort listener on completion to avoid leaks", async () => { + const controller = new AbortController(); + const spy = vi.fn(); + + const promise = Promise.resolve("ok"); + await withTimeout(promise, 1000, "test", { signal: controller.signal }); + + // Create a new listener to verify the old one was removed + controller.signal.addEventListener("abort", spy); + controller.abort(); + expect(spy).toHaveBeenCalledTimes(1); // only the new listener fired + }); + + it("removes abort listener on timeout to avoid leaks", async () => { + const controller = new AbortController(); + const spy = vi.fn(); + + const promise = new Promise((resolve) => setTimeout(resolve, 2000)); + const result = withTimeout(promise, 100, "test", { signal: controller.signal }); + + vi.advanceTimersByTime(100); + await expect(result).rejects.toThrow("timed out"); + + controller.signal.addEventListener("abort", spy); + controller.abort(); + expect(spy).toHaveBeenCalledTimes(1); + }); + + it("propagates the promise rejection when it fails before deadline", async () => { + const failing = Promise.reject(new Error("business logic error")); + await expect(withTimeout(failing, 1000, "test")).rejects.toThrow("business logic error"); + }); + + it("includes the label in the timeout error message", async () => { + const slow = new Promise((resolve) => setTimeout(resolve, 2000)); + const result = withTimeout(slow, 50, "my-label"); + + vi.advanceTimersByTime(50); + + await expect(result).rejects.toThrow("my-label"); + }); + + it("handles multiple concurrent timeouts independently", async () => { + const fast = new Promise((resolve) => setTimeout(() => resolve("fast"), 50)); + const slow = new Promise((resolve) => setTimeout(() => resolve("slow"), 200)); + + const resultFast = withTimeout(fast, 100, "fast"); + const resultSlow = withTimeout(slow, 100, "slow"); + + vi.advanceTimersByTime(50); + // fast should resolve + await expect(resultFast).resolves.toBe("fast"); + + vi.advanceTimersByTime(50); + // slow should time out + await expect(resultSlow).rejects.toThrow("timed out after 100ms for slow"); + }); +}); diff --git a/test/integration/resilience/webhook.test.js b/test/integration/resilience/webhook.test.js new file mode 100644 index 0000000..e466a9e --- /dev/null +++ b/test/integration/resilience/webhook.test.js @@ -0,0 +1,27 @@ +import { describe, it, expect, vi, beforeEach } from "vitest"; +import { CircuitState } from "@/lib/resilience/circuitBreaker"; + +vi.mock("@/lib/telemetry/metrics", () => ({ setGauge: vi.fn(), incrementCounter: vi.fn() })); +vi.mock("@/lib/telemetry/context", () => ({ currentTraceparent: vi.fn(() => "00-abc-xyz-01") })); + +describe("Webhook resilience", () => { + beforeEach(() => { + vi.resetModules(); + }); + + it("exports getWebhookCircuitBreakerState", async () => { + const { getWebhookCircuitBreakerState } = await import("@/lib/webhooks/dispatcher"); + const state = getWebhookCircuitBreakerState(); + expect([CircuitState.CLOSED, CircuitState.OPEN, CircuitState.HALF_OPEN]).toContain(state); + }); + + it("dispatchWebhook validates HTTPS", async () => { + const { dispatchWebhook } = await import("@/lib/webhooks/dispatcher"); + await expect(dispatchWebhook("http://example.com", "{}")).rejects.toThrow("Only HTTPS is allowed"); + }); + + it("dispatchWebhook validates port", async () => { + const { dispatchWebhook } = await import("@/lib/webhooks/dispatcher"); + await expect(dispatchWebhook("https://example.com:8080", "{}")).rejects.toThrow("Unsafe port"); + }); +}); diff --git a/test/mocks/next-server.js b/test/mocks/next-server.js new file mode 100644 index 0000000..2b77789 --- /dev/null +++ b/test/mocks/next-server.js @@ -0,0 +1,12 @@ +export class NextResponse { + static json(body, init = {}) { + return { + body, + status: init.status || 200, + headers: new Map(), + headers: { + set(k, v) { this._headers = this._headers || {}; this._headers[k] = v; }, + }, + }; + } +} diff --git a/vitest.config.mjs b/vitest.config.mjs index d5376e3..3b27b34 100644 --- a/vitest.config.mjs +++ b/vitest.config.mjs @@ -4,6 +4,7 @@ import { defineConfig } from "vitest/config"; const srcPath = fileURLToPath(new URL("./src", import.meta.url)); const sentryMockPath = fileURLToPath(new URL("./test/mocks/sentry-nextjs.js", import.meta.url)); const otelMockPath = fileURLToPath(new URL("./test/mocks/opentelemetry-api.js", import.meta.url)); +const nextServerMockPath = fileURLToPath(new URL("./test/mocks/next-server.js", import.meta.url)); export default defineConfig({ resolve: { @@ -11,6 +12,7 @@ export default defineConfig({ "@": srcPath, "@sentry/nextjs": sentryMockPath, "@opentelemetry/api": otelMockPath, + "next/server": nextServerMockPath, }, }, test: {