From 8ef481ad513c850e71a6e723e16d5581a7158943 Mon Sep 17 00:00:00 2001 From: AI Date: Sun, 30 Aug 2026 16:04:12 +0100 Subject: [PATCH] feat: Implement Circuit Breaker pattern for external services --- src/services/kycFraud.service.ts | 78 ++++++---- src/services/notification.service.ts | 44 ++++-- src/services/payment/stellarProvider.ts | 76 ++++++---- src/services/payment/stripeProvider.ts | 75 ++++++---- src/utils/circuitBreaker.test.ts | 46 ++++++ src/utils/circuitBreaker.ts | 188 ++++++++++++++++++++++++ src/utils/horizonClient.ts | 55 ++++--- 7 files changed, 435 insertions(+), 127 deletions(-) create mode 100644 src/utils/circuitBreaker.test.ts create mode 100644 src/utils/circuitBreaker.ts diff --git a/src/services/kycFraud.service.ts b/src/services/kycFraud.service.ts index e333493..46176d6 100644 --- a/src/services/kycFraud.service.ts +++ b/src/services/kycFraud.service.ts @@ -481,6 +481,8 @@ function applyPlattScaling(rawScore: number, params: { A: number; B: number }): // ─── Third-Party Fraud Service ──────────────────────────────────────────────── +import { CircuitBreaker, CircuitBreakerRegistry, DefaultValueFallback } from '../utils/circuitBreaker'; + export async function getThirdPartyFraudScore( input: FraudInput, ): Promise<{ score: number; signals: FraudSignal[] } | null> { @@ -493,41 +495,55 @@ export async function getThirdPartyFraudScore( return null; } - const controller = new AbortController(); - const timer = setTimeout(() => controller.abort(), config.kycFraud.thirdPartyTimeoutMs); + let cb = CircuitBreakerRegistry.get('kyc_fraud'); + if (!cb) { + cb = new CircuitBreaker('kyc_fraud', { + failureRateThreshold: 0.5, + minimumRequests: 5, + latencyThresholdMs: 5000, + openTimeoutMs: 60000, + halfOpenMaxRequests: 3 + }, new DefaultValueFallback<{ score: number; signals: FraudSignal[] } | null>(null)); + CircuitBreakerRegistry.set('kyc_fraud', cb); + } - const response = await fetch(config.kycFraud.thirdPartyApiUrl, { - method: 'POST', - headers: { - 'Content-Type': 'application/json', - Authorization: `Bearer ${config.kycFraud.thirdPartyApiKey}`, - }, - body: JSON.stringify({ - userId: input.userId, - ipAddress: input.ipAddress, - userAgent: input.userAgent, - deviceFingerprint: input.deviceFingerprint, - documentType: input.documentType, - }), - signal: controller.signal, - }); + return cb.execute(async () => { + const controller = new AbortController(); + const timer = setTimeout(() => controller.abort(), config.kycFraud.thirdPartyTimeoutMs); - clearTimeout(timer); + const response = await fetch(config.kycFraud.thirdPartyApiUrl, { + method: 'POST', + headers: { + 'Content-Type': 'application/json', + Authorization: `Bearer ${config.kycFraud.thirdPartyApiKey}`, + }, + body: JSON.stringify({ + userId: input.userId, + ipAddress: input.ipAddress, + userAgent: input.userAgent, + deviceFingerprint: input.deviceFingerprint, + documentType: input.documentType, + }), + signal: controller.signal, + }); - if (!response.ok) { - logger.warn(`Third-party fraud service returned ${response.status}`); - return null; - } + clearTimeout(timer); - const result = await response.json() as { score?: number; signals?: FraudSignal[] }; - return { - score: result.score ?? 0, - signals: (result.signals ?? []).map((s: any) => ({ - signal: s.signal ?? 'thirdPartyFlag', - severity: s.severity ?? 'medium', - detail: s.detail ?? 'Third-party fraud signal', - })), - }; + if (!response.ok) { + logger.warn(`Third-party fraud service returned ${response.status}`); + throw new Error(`Third-party fraud service returned ${response.status}`); + } + + const result = await response.json() as { score?: number; signals?: FraudSignal[] }; + return { + score: result.score ?? 0, + signals: (result.signals ?? []).map((s: any) => ({ + signal: s.signal ?? 'thirdPartyFlag', + severity: s.severity ?? 'medium', + detail: s.detail ?? 'Third-party fraud signal', + })), + }; + }); } catch (err: any) { logger.warn('Third-party fraud service unavailable, skipping', { error: err.message }); return null; diff --git a/src/services/notification.service.ts b/src/services/notification.service.ts index 16a6023..69bf3c9 100644 --- a/src/services/notification.service.ts +++ b/src/services/notification.service.ts @@ -9,7 +9,7 @@ import { EmailPreferenceService } from './email-preference.service'; import { sanitizeObject, sanitizeString } from '../utils/sanitization'; import { classifyDeliveryError } from '../utils/deliveryErrors'; import { AdminNotificationPreferenceService } from './adminNotificationPreference.service'; - +import { CircuitBreaker, CircuitBreakerRegistry, FailFastFallback } from '../utils/circuitBreaker'; const EMAIL_DELIVERY_MAX_ATTEMPTS = 3; const EMAIL_DELIVERY_BASE_DELAY_MS = 200; @@ -180,21 +180,35 @@ export class NotificationService { attachments?: Array<{ filename: string; content: Buffer; contentType?: string }>; } = {} ): Promise { - try { - await this.transporter.sendMail({ - from: options.from || config.email.from, - to, - subject, - html, - text: options.text, - attachments: options.attachments, - }); - - logger.info(`Email sent to ${to}`); - } catch (error) { - logger.error('Error sending email:', error); - throw error; + let cb = CircuitBreakerRegistry.get('email'); + if (!cb) { + cb = new CircuitBreaker('email', { + failureRateThreshold: 0.5, + minimumRequests: 5, + latencyThresholdMs: 10000, + openTimeoutMs: 60000, + halfOpenMaxRequests: 3 + }, new FailFastFallback()); + CircuitBreakerRegistry.set('email', cb); } + + return cb.execute(async () => { + try { + await this.transporter.sendMail({ + from: options.from || config.email.from, + to, + subject, + html, + text: options.text, + attachments: options.attachments, + }); + + logger.info(`Email sent to ${to}`); + } catch (error) { + logger.error('Error sending email:', error); + throw error; + } + }); } /** diff --git a/src/services/payment/stellarProvider.ts b/src/services/payment/stellarProvider.ts index 52a41fd..27eb0c7 100644 --- a/src/services/payment/stellarProvider.ts +++ b/src/services/payment/stellarProvider.ts @@ -20,6 +20,8 @@ import logger from '../../config/logger'; * ever called) — see pledge.worker.ts. Do not rely on this provider alone * for idempotency. */ +import { CircuitBreaker, CircuitBreakerRegistry, FailFastFallback } from '../../utils/circuitBreaker'; + export class StellarProvider implements PaymentProvider { readonly name = 'stellar' as const; private server: Server; @@ -30,43 +32,57 @@ export class StellarProvider implements PaymentProvider { this.server = new Server(networkUrl); this.escrowKeypair = Keypair.fromSecret(escrowSecretKey); this.networkPassphrase = networkPassphrase; + + if (!CircuitBreakerRegistry.has('stellar')) { + CircuitBreakerRegistry.set('stellar', new CircuitBreaker('stellar', { + failureRateThreshold: 0.5, + minimumRequests: 5, + latencyThresholdMs: 5000, + openTimeoutMs: 60000, + halfOpenMaxRequests: 3 + }, new FailFastFallback())); + } } async charge(options: ChargeOptions): Promise { - try { - const account = await this.server.getAccount(this.escrowKeypair.publicKey()); + const cb = CircuitBreakerRegistry.get('stellar')!; - const tx = new TransactionBuilder(account, { - fee: BASE_FEE, - networkPassphrase: this.networkPassphrase, - }) - .addOperation( - Operation.payment({ - destination: this.escrowKeypair.publicKey(), - asset: Asset.native(), - amount: options.amount.toFixed(7), - }), - ) - .addMemo(Memo.text(options.idempotencyKey.slice(0, 28))) - .setTimeout(30) - .build(); + return cb.execute(async () => { + try { + const account = await this.server.getAccount(this.escrowKeypair.publicKey()); - tx.sign(this.escrowKeypair); + const tx = new TransactionBuilder(account, { + fee: BASE_FEE, + networkPassphrase: this.networkPassphrase, + }) + .addOperation( + Operation.payment({ + destination: this.escrowKeypair.publicKey(), + asset: Asset.native(), + amount: options.amount.toFixed(7), + }), + ) + .addMemo(Memo.text(options.idempotencyKey.slice(0, 28))) + .setTimeout(30) + .build(); - const result: any = await this.server.sendTransaction(tx); + tx.sign(this.escrowKeypair); - if (result.status === 'ERROR' || result.status === 'FAILED') { - throw new PaymentError( - `Stellar transaction failed (status: ${result.status}): ${JSON.stringify(result.errorResult ?? '')}`, - true, - ); - } + const result: any = await this.server.sendTransaction(tx); - return { providerReference: result.hash, provider: 'stellar', raw: result }; - } catch (error: any) { - if (error instanceof PaymentError) throw error; - logger.error('Stellar charge failed', { pledgeId: options.pledgeId, error: error.message }); - throw new PaymentError(error.message ?? 'Stellar charge failed', true); - } + if (result.status === 'ERROR' || result.status === 'FAILED') { + throw new PaymentError( + `Stellar transaction failed (status: ${result.status}): ${JSON.stringify(result.errorResult ?? '')}`, + true, + ); + } + + return { providerReference: result.hash, provider: 'stellar', raw: result }; + } catch (error: any) { + if (error instanceof PaymentError) throw error; + logger.error('Stellar charge failed', { pledgeId: options.pledgeId, error: error.message }); + throw new PaymentError(error.message ?? 'Stellar charge failed', true); + } + }); } } diff --git a/src/services/payment/stripeProvider.ts b/src/services/payment/stripeProvider.ts index 3bdfa3b..f18dafe 100644 --- a/src/services/payment/stripeProvider.ts +++ b/src/services/payment/stripeProvider.ts @@ -15,12 +15,23 @@ import logger from '../../config/logger'; * and are expected to be wired in separately; `stripeCustomerId` / * `stripePaymentMethodId` on ChargeOptions are the seam for that future work. */ +import { CircuitBreaker, CircuitBreakerRegistry, FailFastFallback } from '../../utils/circuitBreaker'; + export class StripeProvider implements PaymentProvider { readonly name = 'stripe' as const; private stripe: Stripe; constructor(secretKey: string) { this.stripe = new Stripe(secretKey, { apiVersion: '2024-06-20' }); + if (!CircuitBreakerRegistry.has('stripe')) { + CircuitBreakerRegistry.set('stripe', new CircuitBreaker('stripe', { + failureRateThreshold: 0.5, + minimumRequests: 5, + latencyThresholdMs: 5000, + openTimeoutMs: 60000, + halfOpenMaxRequests: 3 + }, new FailFastFallback())); + } } async charge(options: ChargeOptions): Promise { @@ -31,39 +42,41 @@ export class StripeProvider implements PaymentProvider { ); } - try { - const intent = await this.stripe.paymentIntents.create( - { - amount: Math.round(options.amount * 100), - currency: options.currency.toLowerCase(), - customer: options.stripeCustomerId, - payment_method: options.stripePaymentMethodId, - off_session: true, - confirm: true, - metadata: { - pledgeId: options.pledgeId, - donorId: options.donorId, - campaignId: options.campaignId, - }, - }, - { idempotencyKey: options.idempotencyKey }, - ); + const cb = CircuitBreakerRegistry.get('stripe')!; - if (intent.status !== 'succeeded') { - throw new PaymentError( - `Stripe payment intent ${intent.id} did not succeed (status: ${intent.status})`, - true, + return cb.execute(async () => { + try { + const intent = await this.stripe.paymentIntents.create( + { + amount: Math.round(options.amount * 100), + currency: options.currency.toLowerCase(), + customer: options.stripeCustomerId, + payment_method: options.stripePaymentMethodId, + off_session: true, + confirm: true, + metadata: { + pledgeId: options.pledgeId, + donorId: options.donorId, + campaignId: options.campaignId, + }, + }, + { idempotencyKey: options.idempotencyKey }, ); - } - return { providerReference: intent.id, provider: 'stripe', raw: intent }; - } catch (error: any) { - if (error instanceof PaymentError) throw error; - logger.error('Stripe charge failed', { pledgeId: options.pledgeId, error: error.message }); - // Stripe card errors are declines and not worth infinite retry, but the - // worker's existing MAX_RETRIES + dead-letter path already bounds - // retries, so we mark everything retryable here and let that path work. - throw new PaymentError(error.message ?? 'Stripe charge failed', true); - } + if (intent.status !== 'succeeded') { + throw new PaymentError( + `Stripe payment intent ${intent.id} did not succeed (status: ${intent.status})`, + true, + ); + } + + return { providerReference: intent.id, provider: 'stripe', raw: intent }; + } catch (error: any) { + if (error instanceof PaymentError) throw error; + logger.error('Stripe charge failed', { pledgeId: options.pledgeId, error: error.message }); + throw new PaymentError(error.message ?? 'Stripe charge failed', true); + } + }); } } + diff --git a/src/utils/circuitBreaker.test.ts b/src/utils/circuitBreaker.test.ts new file mode 100644 index 0000000..985f854 --- /dev/null +++ b/src/utils/circuitBreaker.test.ts @@ -0,0 +1,46 @@ +import { CircuitBreaker, CircuitBreakerState, CircuitBreakerRegistry, FailFastFallback } from './circuitBreaker'; + +describe('CircuitBreaker', () => { + beforeEach(() => { + CircuitBreakerRegistry.clear(); + }); + + it('should transition to OPEN when failure threshold is exceeded', async () => { + const cb = new CircuitBreaker('test', { + failureRateThreshold: 0.5, + minimumRequests: 2, + latencyThresholdMs: 5000, + openTimeoutMs: 1000, + halfOpenMaxRequests: 1 + }, new FailFastFallback()); + + await expect(cb.execute(async () => { throw new Error('fail'); })).rejects.toThrow('fail'); + await expect(cb.execute(async () => { throw new Error('fail'); })).rejects.toThrow('fail'); + + expect(cb.getState()).toBe(CircuitBreakerState.OPEN); + }); + + it('should transition to HALF_OPEN after timeout', async () => { + jest.useFakeTimers(); + const cb = new CircuitBreaker('test2', { + failureRateThreshold: 0.5, + minimumRequests: 2, + latencyThresholdMs: 5000, + openTimeoutMs: 1000, + halfOpenMaxRequests: 1 + }, new FailFastFallback()); + + await expect(cb.execute(async () => { throw new Error('fail'); })).rejects.toThrow('fail'); + await expect(cb.execute(async () => { throw new Error('fail'); })).rejects.toThrow('fail'); + + expect(cb.getState()).toBe(CircuitBreakerState.OPEN); + + jest.advanceTimersByTime(1100); + + // Should allow one request as HALF_OPEN + await expect(cb.execute(async () => 'success')).resolves.toBe('success'); + expect(cb.getState()).toBe(CircuitBreakerState.CLOSED); + + jest.useRealTimers(); + }); +}); diff --git a/src/utils/circuitBreaker.ts b/src/utils/circuitBreaker.ts new file mode 100644 index 0000000..a23cf45 --- /dev/null +++ b/src/utils/circuitBreaker.ts @@ -0,0 +1,188 @@ +import logger from '../config/logger'; + +export enum CircuitBreakerState { + CLOSED = 'CLOSED', + OPEN = 'OPEN', + HALF_OPEN = 'HALF_OPEN', +} + +export interface CircuitBreakerConfig { + failureRateThreshold: number; + minimumRequests: number; + latencyThresholdMs: number; + openTimeoutMs: number; + halfOpenMaxRequests: number; + windowSize?: number; +} + +export class CircuitBreakerOpenError extends Error { + constructor(message: string) { + super(message); + this.name = 'CircuitBreakerOpenError'; + } +} + +export interface FallbackStrategy { + fallback(error: Error): Promise | T; +} + +export class FailFastFallback implements FallbackStrategy { + fallback(error: Error): never { + throw error; + } +} + +export class DefaultValueFallback implements FallbackStrategy { + constructor(private readonly defaultValue: T) {} + fallback(error: Error): T { + return this.defaultValue; + } +} + +export class CircularBuffer { + private outcomes: boolean[]; + private latencies: number[]; + private head: number = 0; + private count: number = 0; + + constructor(private readonly capacity: number) { + this.outcomes = new Array(capacity).fill(true); + this.latencies = new Array(capacity).fill(0); + } + + add(success: boolean, latency: number) { + this.outcomes[this.head] = success; + this.latencies[this.head] = latency; + this.head = (this.head + 1) % this.capacity; + if (this.count < this.capacity) { + this.count++; + } + } + + getMetrics() { + if (this.count === 0) return { failureRate: 0, p95Latency: 0, total: 0 }; + let failures = 0; + const validLatencies = new Float64Array(this.count); + for (let i = 0; i < this.count; i++) { + if (!this.outcomes[i]) failures++; + validLatencies[i] = this.latencies[i]; + } + validLatencies.sort(); + const p95Index = Math.floor(validLatencies.length * 0.95); + const p95Latency = validLatencies[p95Index] || 0; + + return { + failureRate: failures / this.count, + p95Latency, + total: this.count + }; + } + + reset() { + this.head = 0; + this.count = 0; + } +} + +export class CircuitBreaker { + private state: CircuitBreakerState = CircuitBreakerState.CLOSED; + private nextAttemptAt: number = 0; + private halfOpenSuccesses: number = 0; + private halfOpenRequests: number = 0; + private buffer: CircularBuffer; + + constructor( + public readonly name: string, + private readonly config: CircuitBreakerConfig, + private readonly fallbackStrategy: FallbackStrategy + ) { + this.buffer = new CircularBuffer(config.windowSize || 100); + } + + getState(): CircuitBreakerState { + return this.state; + } + + async execute(action: () => Promise): Promise { + if (this.state === CircuitBreakerState.OPEN) { + if (Date.now() >= this.nextAttemptAt) { + this.transitionTo(CircuitBreakerState.HALF_OPEN); + } else { + return this.fallbackStrategy.fallback( + new CircuitBreakerOpenError(`Circuit ${this.name} is OPEN.`) + ); + } + } + + if (this.state === CircuitBreakerState.HALF_OPEN) { + if (this.halfOpenRequests >= this.config.halfOpenMaxRequests) { + return this.fallbackStrategy.fallback( + new CircuitBreakerOpenError(`Circuit ${this.name} is HALF_OPEN and testing.`) + ); + } + this.halfOpenRequests++; + } + + const startTime = performance.now(); + try { + const result = await action(); + const latency = performance.now() - startTime; + this.onSuccess(latency); + return result; + } catch (error) { + const latency = performance.now() - startTime; + this.onFailure(latency); + return this.fallbackStrategy.fallback(error as Error); + } + } + + private onSuccess(latency: number) { + if (this.state === CircuitBreakerState.HALF_OPEN) { + this.halfOpenSuccesses++; + if (this.halfOpenSuccesses >= this.config.halfOpenMaxRequests) { + this.transitionTo(CircuitBreakerState.CLOSED); + } + return; + } + + this.buffer.add(true, latency); + this.checkThresholds(); + } + + private onFailure(latency: number) { + if (this.state === CircuitBreakerState.HALF_OPEN) { + this.transitionTo(CircuitBreakerState.OPEN); + return; + } + + this.buffer.add(false, latency); + this.checkThresholds(); + } + + private checkThresholds() { + const metrics = this.buffer.getMetrics(); + if (metrics.total >= this.config.minimumRequests) { + if (metrics.failureRate >= this.config.failureRateThreshold || metrics.p95Latency >= this.config.latencyThresholdMs) { + this.transitionTo(CircuitBreakerState.OPEN); + } + } + } + + private transitionTo(newState: CircuitBreakerState) { + const oldState = this.state; + this.state = newState; + + logger.info(`CircuitBreaker [${this.name}] transitioned from ${oldState} to ${newState}`); + + if (newState === CircuitBreakerState.OPEN) { + this.nextAttemptAt = Date.now() + this.config.openTimeoutMs; + } else if (newState === CircuitBreakerState.HALF_OPEN) { + this.halfOpenRequests = 0; + this.halfOpenSuccesses = 0; + } else if (newState === CircuitBreakerState.CLOSED) { + this.buffer.reset(); + } + } +} + +export const CircuitBreakerRegistry = new Map>(); diff --git a/src/utils/horizonClient.ts b/src/utils/horizonClient.ts index bd2af56..3b30937 100644 --- a/src/utils/horizonClient.ts +++ b/src/utils/horizonClient.ts @@ -13,6 +13,7 @@ */ import logger from '../config/logger'; +import { CircuitBreaker, CircuitBreakerRegistry, FailFastFallback } from './circuitBreaker'; // ── Types ───────────────────────────────────────────────────────────────────── @@ -115,27 +116,41 @@ export class HorizonClient { private async getJson(url: string): Promise { logger.debug(`HorizonClient GET ${url}`); - let response: Response; - try { - response = await fetch(url, { - headers: { Accept: 'application/json' }, - }); - } catch (networkError) { - throw new HorizonError(url, 0, null, `Network error fetching ${url}: ${String(networkError)}`); + let cb = CircuitBreakerRegistry.get('horizon'); + if (!cb) { + cb = new CircuitBreaker('horizon', { + failureRateThreshold: 0.5, + minimumRequests: 10, + latencyThresholdMs: 5000, + openTimeoutMs: 60000, + halfOpenMaxRequests: 5 + }, new FailFastFallback()); + CircuitBreakerRegistry.set('horizon', cb); } - if (!response.ok) { - const retryAfterHeader = response.headers.get('Retry-After'); - const retryAfterMs = retryAfterHeader ? parseFloat(retryAfterHeader) * 1_000 : null; - - throw new HorizonError( - url, - response.status, - retryAfterMs, - `Horizon returned HTTP ${response.status} for ${url}`, - ); - } - - return response.json() as Promise; + return cb.execute(async () => { + let response: Response; + try { + response = await fetch(url, { + headers: { Accept: 'application/json' }, + }); + } catch (networkError) { + throw new HorizonError(url, 0, null, `Network error fetching ${url}: ${String(networkError)}`); + } + + if (!response.ok) { + const retryAfterHeader = response.headers.get('Retry-After'); + const retryAfterMs = retryAfterHeader ? parseFloat(retryAfterHeader) * 1_000 : null; + + throw new HorizonError( + url, + response.status, + retryAfterMs, + `Horizon returned HTTP ${response.status} for ${url}`, + ); + } + + return response.json() as Promise; + }); } }