diff --git a/backend/.env.example b/backend/.env.example index 324e86aa..69a43f84 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -151,8 +151,22 @@ PAYMENT_RECONCILE_BACKOFF_MAX_MINUTES=60 PAYMENT_RECONCILE_MAX_BATCH=500 # A payment unresolved this long escalates to MANUAL_REVIEW — but only on a # pass where the provider was actually reachable; a provider outage never -# by itself causes escalation (see ReconciliationService). +# by itself causes escalation (see ReconciliationService). The two hard caps +# below are independent safety bounds: they stop further provider polling and +# hand the payment to a human as soon as either cap is reached, even before +# the age threshold. If a provider-error attempt hits the total-attempt cap +# first, the next pass escalates it without making another provider call. +# The provider-error-streak cap deliberately overrides the outage safeguard +# only after a long, bounded per-payment streak; one bad run still does not +# mass-flag payments. PAYMENT_MANUAL_REVIEW_AFTER_HOURS=24 +# Total scheduled confirmation polls for one payment before it is escalated +# to MANUAL_REVIEW and excluded from future reconciliation selection. +PAYMENT_CONFIRMATION_MAX_ATTEMPTS=20 +# Consecutive provider-unreachable polls for one payment before it is +# escalated to MANUAL_REVIEW, even when the provider-error attempt is the one +# that trips the cap. +PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK=10 # When the manual-review queue depth exceeds this the reconciliation cron # logs a WARN, emails SUPPORT_EMAIL, and the depth stays visible on the # Prometheus /metrics endpoint (managehub_manual_review_queue_depth) for @@ -204,6 +218,17 @@ CREDITS_SETTLEMENT_MAX_PAYOUT_ATTEMPTS=5 # Upper bound on how many CONFIRMED payments one pass of the credit sweep # (top-ups and payment revenue splits) will apply. CREDITS_PAYMENT_SWEEP_MAX_BATCH=200 +# Usage-spike anomaly detection. After a new usage event is charged, the +# service compares the tenant's current rolling window total with the +# average of comparable prior windows. It alerts only when there are at +# least MIN_SAMPLES prior events, the window is at least MULTIPLIER times +# that baseline, and the window total is at least MIN_AMOUNT minor units. +# The absolute floor prevents tiny tenants from alerting on rounding; SMTP +# delivery is best-effort and never rolls back the charge. +CREDITS_USAGE_SPIKE_WINDOW_MINUTES=60 +CREDITS_USAGE_SPIKE_MULTIPLIER=5.0 +CREDITS_USAGE_SPIKE_MIN_AMOUNT=10000 +CREDITS_USAGE_SPIKE_MIN_SAMPLES=3 # Data retention (issue BE-144) # Set to false to stop the daily retention job entirely. diff --git a/backend/src/common/metrics.service.ts b/backend/src/common/metrics.service.ts index 877442f6..fe9ddbe5 100644 --- a/backend/src/common/metrics.service.ts +++ b/backend/src/common/metrics.service.ts @@ -42,6 +42,11 @@ export class MetricsService { this.incrementCounter('managehub_settlement_payout_failures_total', {}); } + /** Counts a detected tenant credit-usage anomaly by metered resource. */ + recordCreditUsageSpike(resource: string): void { + this.incrementCounter('managehub_credit_usage_spikes_total', { resource }); + } + renderPrometheus(): string { const lines: string[] = []; this.appendCounters(lines); diff --git a/backend/src/credits/metered-usage.service.spec.ts b/backend/src/credits/metered-usage.service.spec.ts index e3ddcec5..778559f8 100644 --- a/backend/src/credits/metered-usage.service.spec.ts +++ b/backend/src/credits/metered-usage.service.spec.ts @@ -4,27 +4,37 @@ import { InsufficientCreditException, LedgerService } from './ledger.service'; import { MeteredUsageService } from './metered-usage.service'; import { LedgerAccountKind } from './enums/ledger-account-kind.enum'; import { MeteredResource } from './enums/metered-resource.enum'; +import { createTransport } from 'nodemailer'; import { createLedgerHarness, fakeConfigService, } from './testing/in-memory-ledger'; -function build() { +jest.mock('nodemailer', () => ({ + createTransport: jest.fn(), +})); + +function build(configValues: Record = {}) { const harness = createLedgerHarness(); + const config = fakeConfigService({ + CREDITS_DEFAULT_CURRENCY: 'USD', + CREDITS_DEFAULT_OVERDRAFT_LIMIT: 0, + ...configValues, + }); const ledger = new LedgerService( harness.accounts as any, harness.transactions as any, harness.entries as any, ); - const credits = new CreditsService( - ledger, - fakeConfigService({ - CREDITS_DEFAULT_CURRENCY: 'USD', - CREDITS_DEFAULT_OVERDRAFT_LIMIT: 0, - }), + const credits = new CreditsService(ledger, config); + const metrics = { recordCreditUsageSpike: jest.fn() }; + const usage = new MeteredUsageService( + harness.usageEvents as any, + credits, + config as any, + metrics as any, ); - const usage = new MeteredUsageService(harness.usageEvents as any, credits); - return { harness, ledger, credits, usage }; + return { harness, ledger, credits, usage, config, metrics }; } async function fund(credits: CreditsService, amount: number) { @@ -45,6 +55,28 @@ const meterReading = { }; describe('MeteredUsageService', () => { + beforeEach(() => { + (createTransport as jest.Mock).mockReset(); + }); + + function seedPriorUsage(harness: any, amounts: number[]) { + const now = Date.now(); + amounts.forEach((amount, index) => { + harness.usageEvents.rows.push({ + id: `prior-${index}`, + userId: 'user-1', + resource: MeteredResource.RESOURCE_MINUTES, + units: 1, + unitPrice: amount, + amount, + currency: 'USD', + usageReference: `prior-${index}`, + ledgerTransactionId: `prior-transaction-${index}`, + createdAt: new Date(now - (70 + index * 10) * 60_000), + }); + }); + } + it('prices the reading and charges it against the credit balance', async () => { const { usage, credits, harness } = build(); await fund(credits, 1000); @@ -163,4 +195,86 @@ describe('MeteredUsageService', () => { expect((await credits.getBalance('user-1')).balance).toBe(1000 - 250); expect(await usage.listForUser('user-1')).toHaveLength(3); }); + + it('does not alert when the current window is below the spike threshold', async () => { + const { usage, credits, harness, metrics } = build({ + CREDITS_USAGE_SPIKE_WINDOW_MINUTES: 60, + CREDITS_USAGE_SPIKE_MULTIPLIER: 5, + CREDITS_USAGE_SPIKE_MIN_AMOUNT: 100, + CREDITS_USAGE_SPIKE_MIN_SAMPLES: 3, + }); + seedPriorUsage(harness, [100, 100, 100]); + await fund(credits, 5000); + + await usage.recordUsage({ + ...meterReading, + units: 80, + usageReference: 'below-spike', + }); + + expect(metrics.recordCreditUsageSpike).not.toHaveBeenCalled(); + }); + + it('does not alert without the minimum number of prior samples', async () => { + const { usage, credits, harness, metrics } = build({ + CREDITS_USAGE_SPIKE_MIN_AMOUNT: 100, + CREDITS_USAGE_SPIKE_MIN_SAMPLES: 3, + }); + seedPriorUsage(harness, [10, 10]); + await fund(credits, 5000); + + await usage.recordUsage({ + ...meterReading, + units: 200, + usageReference: 'insufficient-history', + }); + + expect(metrics.recordCreditUsageSpike).not.toHaveBeenCalled(); + }); + + it('increments the usage-spike counter when a sufficiently large spike is detected', async () => { + const { usage, credits, harness, metrics } = build({ + CREDITS_USAGE_SPIKE_MIN_AMOUNT: 100, + CREDITS_USAGE_SPIKE_MIN_SAMPLES: 3, + }); + seedPriorUsage(harness, [10, 10, 10]); + await fund(credits, 5000); + + await usage.recordUsage({ + ...meterReading, + units: 200, + usageReference: 'large-spike', + }); + + expect(metrics.recordCreditUsageSpike).toHaveBeenCalledWith( + MeteredResource.RESOURCE_MINUTES, + ); + }); + + it('does not let a failing SMTP alert break usage recording', async () => { + const sendMail = jest.fn().mockRejectedValue(new Error('smtp unavailable')); + (createTransport as jest.Mock).mockReturnValue({ sendMail }); + const { usage, credits, harness, metrics } = build({ + CREDITS_USAGE_SPIKE_MIN_AMOUNT: 100, + CREDITS_USAGE_SPIKE_MIN_SAMPLES: 3, + SMTP_HOST: 'smtp.example.test', + SMTP_USER: 'smtp-user', + SMTP_PASS: 'smtp-pass', + SMTP_FROM_EMAIL: 'alerts@example.test', + SUPPORT_EMAIL: 'support@example.test', + }); + seedPriorUsage(harness, [10, 10, 10]); + await fund(credits, 5000); + + await expect( + usage.recordUsage({ + ...meterReading, + units: 200, + usageReference: 'smtp-failure-spike', + }), + ).resolves.toMatchObject({ charged: true }); + expect(metrics.recordCreditUsageSpike).toHaveBeenCalled(); + expect(sendMail).toHaveBeenCalledTimes(1); + expect((await credits.getBalance('user-1')).balance).toBe(4000); + }); }); diff --git a/backend/src/credits/metered-usage.service.ts b/backend/src/credits/metered-usage.service.ts index 05e45d70..22c8fa14 100644 --- a/backend/src/credits/metered-usage.service.ts +++ b/backend/src/credits/metered-usage.service.ts @@ -1,9 +1,14 @@ -import { BadRequestException, Injectable } from '@nestjs/common'; +import { BadRequestException, Injectable, Logger } from '@nestjs/common'; +import { ConfigService } from '@nestjs/config'; import { InjectRepository } from '@nestjs/typeorm'; -import { Repository } from 'typeorm'; +import { MoreThanOrEqual, Repository } from 'typeorm'; import { MeteredUsageEvent } from './entities/metered-usage-event.entity'; import { MeteredResource } from './enums/metered-resource.enum'; import { CreditsService } from './credits.service'; +import { MetricsService } from '../common/metrics.service'; +import { withRequestId } from '../common/request-context'; +import { createTransport } from 'nodemailer'; +import Handlebars from 'handlebars'; const POSTGRES_UNIQUE_VIOLATION = '23505'; @@ -46,10 +51,14 @@ export interface RecordUsageResult { */ @Injectable() export class MeteredUsageService { + private readonly logger = new Logger(MeteredUsageService.name); + constructor( @InjectRepository(MeteredUsageEvent) private readonly usageRepository: Repository, private readonly credits: CreditsService, + private readonly config: ConfigService, + private readonly metrics: MetricsService, ) {} async recordUsage(input: RecordUsageInput): Promise { @@ -95,8 +104,9 @@ export class MeteredUsageService { actorId: input.actorId ?? null, }); + let event: MeteredUsageEvent; try { - const event = await this.usageRepository.save( + event = await this.usageRepository.save( this.usageRepository.create({ userId: input.userId, resource: input.resource, @@ -108,7 +118,6 @@ export class MeteredUsageService { ledgerTransactionId: charge.transaction.id, }), ); - return { event, charged: charge.posted }; } catch (error) { // Two deliveries of the same meter reading raced. The ledger already // deduped the charge, so the loser just reports the winner's event. @@ -122,6 +131,233 @@ export class MeteredUsageService { } throw error; } + + // Detection is deliberately after both durable writes. A detector, + // metrics, or email failure is observational only and must never undo + // the charge or make the caller retry a successfully recorded event. + if (charge.posted) { + try { + await this.detectUsageSpike(event); + } catch (error) { + this.logger.warn( + withRequestId( + `Unable to evaluate credit usage spike for ${input.userId}: ${ + error instanceof Error ? error.message : String(error) + }`, + ), + ); + } + } + + return { event, charged: charge.posted }; + } + + /** + * Detects a tenant-level usage spike after the new event is durable. + * + * The comparison is deliberately tenant-wide rather than resource-wide: + * an integration bug can move several metered resources at once, and a + * shared tenant total catches that pattern. The current window is the + * configured number of minutes (default 60) ending at the new event and + * includes that event. The baseline is the average total of the immediately + * preceding equal-length windows that contain at least one prior event; + * requiring at + * least CREDITS_USAGE_SPIKE_MIN_SAMPLES prior events (default 3) keeps a new + * tenant from producing a meaningless percentage. The absolute minimum + * amount (default 10,000 minor units) suppresses alerts for tiny tenants + * where normal rounding or one small event would otherwise look like a + * large ratio. A spike therefore needs + * all three signals: enough history, a meaningful absolute total, and a + * total at least CREDITS_USAGE_SPIKE_MULTIPLIER (default 5.0) times the + * baseline. + * + * This method is observational. Its caller catches every failure after + * the charge, and the SMTP implementation below is independently + * best-effort, so neither an alert problem nor a provider outage can roll + * back usage accounting. + */ + private async detectUsageSpike(event: MeteredUsageEvent): Promise { + const windowMinutes = this.getUsageSpikeWindowMinutes(); + const multiplier = this.getUsageSpikeMultiplier(); + const minimumAmount = this.getUsageSpikeMinimumAmount(); + const minimumSamples = this.getUsageSpikeMinimumSamples(); + const windowMs = windowMinutes * 60_000; + const baselineWindowCount = Math.max(3, minimumSamples); + const eventTime = + event.createdAt instanceof Date + ? new Date(event.createdAt.getTime()) + : new Date(); + const currentWindowStart = eventTime.getTime() - windowMs; + const historyStart = currentWindowStart - windowMs * baselineWindowCount; + const history = await this.usageRepository.find({ + where: { + userId: event.userId, + createdAt: MoreThanOrEqual(new Date(historyStart)), + }, + order: { createdAt: 'ASC' }, + }); + + const relevantHistory = (history ?? []).filter((candidate) => { + const timestamp = candidate.createdAt.getTime(); + return timestamp >= historyStart && timestamp <= eventTime.getTime(); + }); + const priorEvents = relevantHistory.filter( + (candidate) => candidate.createdAt.getTime() < currentWindowStart, + ); + if (priorEvents.length < minimumSamples) { + return; + } + + const priorWindowTotals = new Map(); + for (const priorEvent of priorEvents) { + const windowIndex = Math.floor( + (currentWindowStart - priorEvent.createdAt.getTime()) / windowMs, + ); + if (windowIndex >= 0 && windowIndex < baselineWindowCount) { + priorWindowTotals.set( + windowIndex, + (priorWindowTotals.get(windowIndex) ?? 0) + priorEvent.amount, + ); + } + } + const observedPriorWindowTotals = [...priorWindowTotals.values()]; + if (observedPriorWindowTotals.length === 0) { + return; + } + const baselineAverage = + observedPriorWindowTotals.reduce((sum, total) => sum + total, 0) / + observedPriorWindowTotals.length; + + const currentEvents = relevantHistory.filter( + (candidate) => candidate.createdAt.getTime() >= currentWindowStart, + ); + if (!currentEvents.some((candidate) => candidate.id === event.id)) { + currentEvents.push(event); + } + const windowTotal = currentEvents.reduce( + (sum, candidate) => sum + candidate.amount, + 0, + ); + if ( + windowTotal < minimumAmount || + windowTotal < baselineAverage * multiplier + ) { + return; + } + + const eventAmount = event.amount; + this.logger.warn( + withRequestId( + `ALERT: credit usage spike tenantId=${event.userId} ` + + `resource=${event.resource} eventAmount=${eventAmount} ` + + `windowTotal=${windowTotal} baselineAverage=${baselineAverage.toFixed(2)} ` + + `multiplier=${multiplier} threshold=${minimumAmount}`, + ), + ); + this.metrics.recordCreditUsageSpike(event.resource); + await this.sendUsageSpikeAlert({ + tenantId: event.userId, + resource: event.resource, + eventAmount, + windowTotal, + baselineAverage, + multiplier, + minimumAmount, + }); + } + + private async sendUsageSpikeAlert(input: { + tenantId: string; + resource: string; + eventAmount: number; + windowTotal: number; + baselineAverage: number; + multiplier: number; + minimumAmount: number; + }): Promise { + try { + const host = this.config.get('SMTP_HOST'); + const user = this.config.get('SMTP_USER'); + const pass = this.config.get('SMTP_PASS'); + const from = this.config.get('SMTP_FROM_EMAIL'); + const supportEmail = this.config.get('SUPPORT_EMAIL'); + if (!host || !user || !pass || !from || !supportEmail) { + return; + } + + const transport = createTransport({ + host, + port: this.config.get('SMTP_PORT', 587), + secure: this.config.get('SMTP_SECURE', 'false') === 'true', + auth: { user, pass }, + }); + const template = Handlebars.compile( + '{{company}} credit usage spike alert: tenant {{tenantId}} recorded {{windowTotal}} minor units for {{resource}} in the current {{windowMinutes}} minute window (event {{eventAmount}}, baseline {{baselineAverage}}, multiplier {{multiplier}}, minimum {{minimumAmount}}). Please investigate.', + ); + await transport.sendMail({ + from, + to: supportEmail, + subject: `${this.config.get('COMPANY_NAME', 'ManageHub')} credit usage spike alert`, + text: template({ + company: this.config.get('COMPANY_NAME', 'ManageHub'), + tenantId: input.tenantId, + resource: input.resource, + windowMinutes: this.getUsageSpikeWindowMinutes(), + eventAmount: input.eventAmount, + windowTotal: input.windowTotal, + baselineAverage: input.baselineAverage.toFixed(2), + multiplier: input.multiplier, + minimumAmount: input.minimumAmount, + }), + }); + } catch (error) { + this.logger.warn( + withRequestId( + `Unable to send credit usage spike alert email: ${ + error instanceof Error ? error.message : String(error) + }`, + ), + ); + } + } + + private getUsageSpikeWindowMinutes(): number { + return this.readPositiveNumberConfig( + 'CREDITS_USAGE_SPIKE_WINDOW_MINUTES', + 60, + ); + } + + private getUsageSpikeMultiplier(): number { + return this.readPositiveNumberConfig( + 'CREDITS_USAGE_SPIKE_MULTIPLIER', + 5.0, + ); + } + + private getUsageSpikeMinimumAmount(): number { + return this.readPositiveNumberConfig( + 'CREDITS_USAGE_SPIKE_MIN_AMOUNT', + 10_000, + ); + } + + private getUsageSpikeMinimumSamples(): number { + return this.readPositiveIntegerConfig( + 'CREDITS_USAGE_SPIKE_MIN_SAMPLES', + 3, + ); + } + + private readPositiveNumberConfig(key: string, fallback: number): number { + const configured = Number(this.config.get(key, fallback)); + return Number.isFinite(configured) && configured > 0 + ? configured + : fallback; + } + + private readPositiveIntegerConfig(key: string, fallback: number): number { + return Math.floor(this.readPositiveNumberConfig(key, fallback)); } async listForUser(userId: string, limit = 100): Promise { diff --git a/backend/src/database/migrations/1790003000000-CreateReconciliationRunsTable.ts b/backend/src/database/migrations/1790003000000-CreateReconciliationRunsTable.ts new file mode 100644 index 00000000..3200b1ea --- /dev/null +++ b/backend/src/database/migrations/1790003000000-CreateReconciliationRunsTable.ts @@ -0,0 +1,41 @@ +import { MigrationInterface, QueryRunner } from 'typeorm'; + +export class CreateReconciliationRunsTable1790003000000 implements MigrationInterface { + name = 'CreateReconciliationRunsTable1790003000000'; + + public async up(queryRunner: QueryRunner): Promise { + await queryRunner.query(` + CREATE TABLE "reconciliation_runs" ( + "id" uuid NOT NULL DEFAULT gen_random_uuid(), + "started_at" timestamptz NOT NULL DEFAULT now(), + "finished_at" timestamptz, + "duration_ms" integer, + "candidates" integer NOT NULL DEFAULT 0, + "resolved" integer NOT NULL DEFAULT 0, + "pending" integer NOT NULL DEFAULT 0, + "provider_errors" integer NOT NULL DEFAULT 0, + "escalated_to_manual_review" integer NOT NULL DEFAULT 0, + "expired_swept" integer NOT NULL DEFAULT 0, + "outcome" text, + "error" text, + "details" jsonb, + CONSTRAINT "pk_reconciliation_runs" PRIMARY KEY ("id") + ) + `); + + await queryRunner.query(` + CREATE INDEX "idx_reconciliation_runs_started_at" + ON "reconciliation_runs" ("started_at") + `); + await queryRunner.query(` + CREATE INDEX "idx_reconciliation_runs_outcome" + ON "reconciliation_runs" ("outcome") + `); + } + + public async down(queryRunner: QueryRunner): Promise { + await queryRunner.query(`DROP INDEX "idx_reconciliation_runs_outcome"`); + await queryRunner.query(`DROP INDEX "idx_reconciliation_runs_started_at"`); + await queryRunner.query(`DROP TABLE "reconciliation_runs"`); + } +} diff --git a/backend/src/payments/dto/reconciliation-run-response.dto.spec.ts b/backend/src/payments/dto/reconciliation-run-response.dto.spec.ts new file mode 100644 index 00000000..c690a6b2 --- /dev/null +++ b/backend/src/payments/dto/reconciliation-run-response.dto.spec.ts @@ -0,0 +1,36 @@ +import { ReconciliationRunResponseDto } from './reconciliation-run-response.dto'; + +describe('ReconciliationRunResponseDto', () => { + it('maps a persisted reconciliation run without exposing entity-only state', () => { + const run = { + id: 'run-1', + startedAt: new Date('2025-01-01T00:00:00.000Z'), + finishedAt: new Date('2025-01-01T00:00:01.000Z'), + durationMs: 1000, + candidates: 3, + resolved: 1, + pending: 1, + providerErrors: 1, + escalatedToManualReview: 0, + expiredSwept: 0, + outcome: 'succeeded' as const, + error: null, + details: { attemptCapEscalations: 0 }, + }; + + const dto = ReconciliationRunResponseDto.fromEntity(run as any); + + expect(dto).toMatchObject({ + id: 'run-1', + candidates: 3, + resolved: 1, + pending: 1, + providerErrors: 1, + outcome: 'succeeded', + error: null, + details: { attemptCapEscalations: 0 }, + }); + expect(dto.durationMs).toBe(1000); + expect(dto.finishedAt).toBe(run.finishedAt); + }); +}); diff --git a/backend/src/payments/dto/reconciliation-run-response.dto.ts b/backend/src/payments/dto/reconciliation-run-response.dto.ts new file mode 100644 index 00000000..96252396 --- /dev/null +++ b/backend/src/payments/dto/reconciliation-run-response.dto.ts @@ -0,0 +1,47 @@ +import { ApiProperty } from '@nestjs/swagger'; +import { + ReconciliationRun, + ReconciliationRunOutcome, +} from '../entities/reconciliation-run.entity'; + +export class ReconciliationRunResponseDto { + @ApiProperty() id: string; + @ApiProperty() startedAt: Date; + @ApiProperty({ nullable: true }) finishedAt: Date | null; + @ApiProperty({ nullable: true }) durationMs: number | null; + @ApiProperty() candidates: number; + @ApiProperty() resolved: number; + @ApiProperty() pending: number; + @ApiProperty() providerErrors: number; + @ApiProperty() escalatedToManualReview: number; + @ApiProperty() expiredSwept: number; + @ApiProperty({ + enum: ['succeeded', 'failed'], + nullable: true, + description: 'Null only while a run is still in progress.', + }) + outcome: ReconciliationRunOutcome | null; + @ApiProperty({ nullable: true }) error: string | null; + @ApiProperty({ nullable: true, type: Object }) details: Record< + string, + unknown + > | null; + + static fromEntity(run: ReconciliationRun): ReconciliationRunResponseDto { + const dto = new ReconciliationRunResponseDto(); + dto.id = run.id; + dto.startedAt = run.startedAt; + dto.finishedAt = run.finishedAt; + dto.durationMs = run.durationMs; + dto.candidates = run.candidates; + dto.resolved = run.resolved; + dto.pending = run.pending; + dto.providerErrors = run.providerErrors; + dto.escalatedToManualReview = run.escalatedToManualReview; + dto.expiredSwept = run.expiredSwept; + dto.outcome = run.outcome; + dto.error = run.error; + dto.details = run.details; + return dto; + } +} diff --git a/backend/src/payments/entities/reconciliation-run.entity.ts b/backend/src/payments/entities/reconciliation-run.entity.ts new file mode 100644 index 00000000..4abad7d2 --- /dev/null +++ b/backend/src/payments/entities/reconciliation-run.entity.ts @@ -0,0 +1,64 @@ +import { + Column, + CreateDateColumn, + Entity, + Index, + PrimaryGeneratedColumn, +} from 'typeorm'; + +export type ReconciliationRunOutcome = 'succeeded' | 'failed'; + +/** + * Immutable-ish audit record for one scheduled reconciliation pass. The + * counters are copied from ReconciliationSummary when the pass completes; + * a row remains queryable with nullable completion fields if the process + * fails before it can finish the pass, and `error` captures that failure. + */ +@Entity('reconciliation_runs') +@Index('idx_reconciliation_runs_started_at', ['startedAt']) +@Index('idx_reconciliation_runs_outcome', ['outcome']) +export class ReconciliationRun { + @PrimaryGeneratedColumn('uuid') + id: string; + + @CreateDateColumn({ type: 'timestamptz', name: 'started_at' }) + startedAt: Date; + + @Column({ type: 'timestamptz', name: 'finished_at', nullable: true }) + finishedAt: Date | null; + + @Column({ type: 'int', name: 'duration_ms', nullable: true }) + durationMs: number | null; + + @Column({ type: 'int', name: 'candidates', default: 0 }) + candidates: number; + + @Column({ type: 'int', name: 'resolved', default: 0 }) + resolved: number; + + @Column({ type: 'int', name: 'pending', default: 0 }) + pending: number; + + @Column({ type: 'int', name: 'provider_errors', default: 0 }) + providerErrors: number; + + @Column({ + type: 'int', + name: 'escalated_to_manual_review', + default: 0, + }) + escalatedToManualReview: number; + + @Column({ type: 'int', name: 'expired_swept', default: 0 }) + expiredSwept: number; + + @Column({ type: 'text', name: 'outcome', nullable: true }) + outcome: ReconciliationRunOutcome | null; + + @Column({ type: 'text', name: 'error', nullable: true }) + error: string | null; + + /** Additional run context such as cap configuration and escalation counts. */ + @Column({ type: 'jsonb', name: 'details', nullable: true }) + details: Record | null; +} diff --git a/backend/src/payments/payment-confirmation.service.ts b/backend/src/payments/payment-confirmation.service.ts index eeec181a..903ef8b4 100644 --- a/backend/src/payments/payment-confirmation.service.ts +++ b/backend/src/payments/payment-confirmation.service.ts @@ -55,7 +55,10 @@ export class PaymentConfirmationService { /** * Applies a confirmation outcome for the Payment matching * `providerReference`. Returns the (possibly unchanged) Payment, or null - * if no Payment matches the reference at all. + * if no Payment matches the reference at all. Scheduled confirmation + * checks are bounded by PAYMENT_CONFIRMATION_MAX_ATTEMPTS and + * PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK; reaching either cap + * moves the payment to MANUAL_REVIEW before another poll is scheduled. */ async apply( providerReference: string, @@ -153,7 +156,11 @@ export class PaymentConfirmationService { * * Hard rule: `verified: false` (timeout, provider error, or a `pending` * response) NEVER implies a status change. Only an authoritative - * confirmed/failed response from the provider moves the Payment. + * confirmed/failed response from the provider moves the Payment. This + * method performs one bounded provider check per invocation; the total + * scheduled confirmation checks a payment may receive are capped by + * PAYMENT_CONFIRMATION_MAX_ATTEMPTS and + * PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK in ReconciliationService. */ async verifyOnReturn( payment: Payment, diff --git a/backend/src/payments/payments-admin.controller.ts b/backend/src/payments/payments-admin.controller.ts index 2d0a2c54..9e67e2ed 100644 --- a/backend/src/payments/payments-admin.controller.ts +++ b/backend/src/payments/payments-admin.controller.ts @@ -5,11 +5,13 @@ import { Param, ParseUUIDPipe, Post, + Query, UseGuards, } from '@nestjs/common'; import { ApiBearerAuth, ApiOperation, + ApiQuery, ApiResponse, ApiTags, } from '@nestjs/swagger'; @@ -24,6 +26,7 @@ import { RefundsService } from './refunds.service'; import { AdminActionLogService } from '../admin-audit/admin-action-log.service'; import { AdminActionType } from '../admin-audit/admin-action-type.enum'; import { PaymentResponseDto } from './dto/payment-response.dto'; +import { ReconciliationRunResponseDto } from './dto/reconciliation-run-response.dto'; import { ResolvePaymentManuallyDto } from './dto/resolve-payment-manually.dto'; import { VoidPaymentDto } from './dto/void-payment.dto'; import { CreateRefundDto } from './dto/create-refund.dto'; @@ -69,6 +72,26 @@ export class PaymentsAdminController { return this.reconciliationService.getMetrics(); } + @Get('reconciliation-runs') + @ApiOperation({ + summary: 'List persisted reconciliation runs, newest first', + description: + 'Returns the most recent reconciliation run records. The optional limit is clamped to 1–500 and defaults to 50.', + }) + @ApiQuery({ + name: 'limit', + required: false, + type: Number, + description: 'Number of runs to return (1–500, default 50)', + }) + @ApiResponse({ status: 200, type: [ReconciliationRunResponseDto] }) + async listReconciliationRuns( + @Query('limit') limit?: string, + ): Promise { + const runs = await this.reconciliationService.listRecentRuns(limit ?? 50); + return runs.map((run) => ReconciliationRunResponseDto.fromEntity(run)); + } + @Post(':id/force-reconcile') @ApiOperation({ summary: diff --git a/backend/src/payments/payments.module.ts b/backend/src/payments/payments.module.ts index 64e56205..761bcb62 100644 --- a/backend/src/payments/payments.module.ts +++ b/backend/src/payments/payments.module.ts @@ -6,6 +6,7 @@ import { BullModule } from '@nestjs/bull'; import { Payment } from './entities/payment.entity'; import { ConfirmationEvent } from './entities/confirmation-event.entity'; import { Refund } from './entities/refund.entity'; +import { ReconciliationRun } from './entities/reconciliation-run.entity'; import { PaymentWebhookDeadLetter } from './entities/payment-webhook-dead-letter.entity'; import { PaymentsService } from './payments.service'; import { PaymentConfirmationService } from './payment-confirmation.service'; @@ -46,6 +47,7 @@ import { Payment, ConfirmationEvent, Refund, + ReconciliationRun, PaymentWebhookDeadLetter, ]), WalletsModule, diff --git a/backend/src/payments/reconciliation.service.spec.ts b/backend/src/payments/reconciliation.service.spec.ts index aed0ba4d..743820e2 100644 --- a/backend/src/payments/reconciliation.service.spec.ts +++ b/backend/src/payments/reconciliation.service.spec.ts @@ -112,12 +112,58 @@ function makePaymentRepository(seed: Payment[]) { }; } +function makeRunRepository(seed: any[] = []) { + const rows = [...seed]; + let sequence = rows.length; + + return { + create: jest.fn((entity: any) => ({ id: `run-${++sequence}`, ...entity })), + save: jest.fn(async (entity: any) => { + const row = { ...entity }; + if (!row.id) { + row.id = `run-${++sequence}`; + } + const index = rows.findIndex((candidate) => candidate.id === row.id); + if (index >= 0) { + rows[index] = { ...rows[index], ...row }; + } else { + rows.push(row); + } + return { ...row }; + }), + update: jest.fn(async (id: string, partial: any) => { + const index = rows.findIndex((candidate) => candidate.id === id); + if (index >= 0) { + rows[index] = { ...rows[index], ...partial }; + } + return { affected: index >= 0 ? 1 : 0 }; + }), + find: jest.fn(async (options?: { order?: any; take?: number }) => { + let result = [...rows]; + if (options?.order?.startedAt === 'DESC') { + result.sort( + (left, right) => + new Date(right.startedAt).getTime() - + new Date(left.startedAt).getTime(), + ); + } + if (options?.take) { + result = result.slice(0, options.take); + } + return result.map((row) => ({ ...row })); + }), + _rows: rows, + }; +} + function makeConfigService(overrides: Record = {}) { const values: Record = { PAYMENT_RECONCILE_DUE_AFTER_MINUTES: 5, PAYMENT_RECONCILE_BACKOFF_BASE_MINUTES: 5, PAYMENT_RECONCILE_BACKOFF_MAX_MINUTES: 60, PAYMENT_MANUAL_REVIEW_AFTER_HOURS: 24, + PAYMENT_CONFIRMATION_MAX_ATTEMPTS: 20, + PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK: 10, PAYMENT_VERIFY_TIMEOUT_MS: 3000, PAYMENT_RECONCILE_MAX_BATCH: 500, PAYMENT_MANUAL_REVIEW_ALERT_THRESHOLD: 20, @@ -152,18 +198,21 @@ describe('ReconciliationService', () => { function build( seed: Payment[], configOverrides: Record = {}, + runSeed: any[] = [], ) { const paymentRepository = makePaymentRepository(seed); + const runRepository = makeRunRepository(runSeed); const config = makeConfigService(configOverrides); - const service = new ReconciliationService( - paymentRepository as any, - confirmationService as any, - railRegistry as any, - gateway as any, - config as any, - metrics as any, - ); - return { service, paymentRepository }; + const service = new ReconciliationService( + paymentRepository as any, + runRepository as any, + confirmationService as any, + railRegistry as any, + gateway as any, + config as any, + metrics as any, + ); + return { service, paymentRepository, runRepository }; } const minutesAgo = (n: number, from: Date = new Date()) => @@ -238,6 +287,111 @@ describe('ReconciliationService', () => { }); }); + describe('reconcileDueBatch — confirmation polling caps', () => { + it('continues polling a payment that is below the attempt cap', async () => { + const now = new Date(); + const payment = makePayment({ + createdAt: minutesAgo(10, now), + reconciliationAttempts: 1, + }); + railAdapter.verifyByReference.mockResolvedValue({ outcome: 'pending' }); + + const { service } = build([payment], { + PAYMENT_CONFIRMATION_MAX_ATTEMPTS: 3, + }); + const summary = await service.reconcileDueBatch(now); + + expect(summary.candidates).toBe(1); + expect(summary.pending).toBe(1); + expect(railAdapter.verifyByReference).toHaveBeenCalledWith( + payment.providerReference, + ); + }); + + it('escalates at the attempt cap and never polls the capped payment again', async () => { + const now = new Date(); + const payment = makePayment({ + createdAt: minutesAgo(10, now), + reconciliationAttempts: 2, + }); + railAdapter.verifyByReference.mockResolvedValue({ outcome: 'pending' }); + + const { service, paymentRepository } = build([payment], { + PAYMENT_CONFIRMATION_MAX_ATTEMPTS: 3, + }); + const first = await service.reconcileDueBatch(now); + + expect(first.escalatedToManualReview).toBe(1); + expect(first.attemptCapEscalations).toBe(1); + const updated = await paymentRepository.findOne({ + where: { id: payment.id }, + }); + expect(updated!.status).toBe(PaymentStatus.MANUAL_REVIEW); + expect(updated!.manualReviewReason).toMatch(/attempt cap 3/); + expect(updated!.manualReviewReason).toMatch( + /3 total reconciliation attempts/, + ); + + const second = await service.reconcileDueBatch( + new Date(now.getTime() + 60_000), + ); + expect(second.candidates).toBe(0); + expect(railAdapter.verifyByReference).toHaveBeenCalledTimes(1); + }); + + it('escalates a pre-capped AWAITING_CONFIRMATION row without another poll', async () => { + const now = new Date(); + const payment = makePayment({ + createdAt: minutesAgo(10, now), + reconciliationAttempts: 3, + }); + const { service, paymentRepository } = build([payment], { + PAYMENT_CONFIRMATION_MAX_ATTEMPTS: 3, + }); + + const summary = await service.reconcileDueBatch(now); + + expect(summary.candidates).toBe(0); + expect(summary.escalatedToManualReview).toBe(1); + expect(railAdapter.verifyByReference).not.toHaveBeenCalled(); + const updated = await paymentRepository.findOne({ + where: { id: payment.id }, + }); + expect(updated!.status).toBe(PaymentStatus.MANUAL_REVIEW); + }); + + it('escalates at the provider-error-streak cap even on the erroring attempt', async () => { + const now = new Date(); + const payment = makePayment({ + createdAt: hoursAgo(48, now), + reconciliationAttempts: 2, + providerErrorStreak: 2, + }); + railAdapter.verifyByReference.mockRejectedValue( + new Error('provider unreachable'), + ); + + const { service, paymentRepository } = build([payment], { + PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK: 3, + }); + const summary = await service.reconcileDueBatch(now); + + expect(summary.providerErrors).toBe(1); + expect(summary.escalatedToManualReview).toBe(1); + expect(summary.providerErrorStreakCapEscalations).toBe(1); + const updated = await paymentRepository.findOne({ + where: { id: payment.id }, + }); + expect(updated!.status).toBe(PaymentStatus.MANUAL_REVIEW); + expect(updated!.manualReviewReason).toMatch( + /provider-error streak cap 3/, + ); + expect(updated!.manualReviewReason).toMatch( + /3 consecutive provider errors/, + ); + }); + }); + describe('reconcileDueBatch — provider outage does not cause false-positive escalation', () => { it('does not escalate an old payment to MANUAL_REVIEW when this attempt is itself a provider outage', async () => { const now = new Date(); @@ -405,6 +559,43 @@ describe('ReconciliationService', () => { expect(railAdapter.verifyByReference).toHaveBeenCalledTimes(500); }); + it('persists a started run and completes it with the pass summary', async () => { + railAdapter.verifyByReference.mockResolvedValue({ outcome: 'pending' }); + const payment = makePayment({ createdAt: minutesAgo(10) }); + const { service, runRepository } = build([payment]); + + await service.handleCron(); + + expect(runRepository.save).toHaveBeenCalledTimes(1); + expect(runRepository.update).toHaveBeenCalledWith( + 'run-1', + expect.objectContaining({ + outcome: 'succeeded', + candidates: 1, + pending: 1, + finishedAt: expect.any(Date), + durationMs: expect.any(Number), + }), + ); + }); + + it('marks a failed pass without masking the original error and resets the guard', async () => { + const { service, paymentRepository, runRepository } = build([]); + paymentRepository.find.mockRejectedValueOnce(new Error('database unavailable')); + + await expect(service.handleCron()).rejects.toThrow('database unavailable'); + expect(runRepository.update).toHaveBeenCalledWith( + 'run-1', + expect.objectContaining({ + outcome: 'failed', + error: 'database unavailable', + finishedAt: expect.any(Date), + }), + ); + + await expect(service.handleCron()).resolves.toBeUndefined(); + }); + it('skips a cron tick that fires while the previous pass is still running', async () => { const now = new Date(); const payment = makePayment({ createdAt: minutesAgo(10, now) }); @@ -560,6 +751,31 @@ describe('ReconciliationService', () => { expect(result.map((p) => p.id)).toEqual(['p1']); }); + it('lists recent reconciliation runs newest first with a bounded limit', async () => { + const { service, runRepository } = build( + [], + {}, + [ + { + id: 'run-old', + startedAt: new Date('2025-01-01T00:00:00.000Z'), + }, + { + id: 'run-new', + startedAt: new Date('2025-01-02T00:00:00.000Z'), + }, + ], + ); + + const result = await service.listRecentRuns(1); + + expect(result.map((run) => run.id)).toEqual(['run-new']); + expect(runRepository.find).toHaveBeenCalledWith({ + order: { startedAt: 'DESC' }, + take: 1, + }); + }); + it('getMetrics reports the manual-review queue depth and alerting flag', async () => { const payments = Array.from({ length: 25 }, (_, i) => makePayment({ id: `p${i}`, status: PaymentStatus.MANUAL_REVIEW }), @@ -570,6 +786,8 @@ describe('ReconciliationService', () => { expect(metrics.manualReviewQueueDepth).toBe(25); expect(metrics.alertThreshold).toBe(20); expect(metrics.alerting).toBe(true); + expect(metrics.confirmationMaxAttempts).toBe(20); + expect(metrics.confirmationMaxProviderErrorStreak).toBe(10); }); }); }); diff --git a/backend/src/payments/reconciliation.service.ts b/backend/src/payments/reconciliation.service.ts index 26be2615..56a6dc23 100644 --- a/backend/src/payments/reconciliation.service.ts +++ b/backend/src/payments/reconciliation.service.ts @@ -10,6 +10,7 @@ import { InjectRepository } from '@nestjs/typeorm'; import { Cron, CronExpression } from '@nestjs/schedule'; import { LessThan, Repository } from 'typeorm'; import { Payment } from './entities/payment.entity'; +import { ReconciliationRun } from './entities/reconciliation-run.entity'; import { PaymentStatus } from './enums/payment-status.enum'; import { PaymentFailureReason } from './enums/payment-failure-reason.enum'; import { ConfirmationSource } from './enums/confirmation-source.enum'; @@ -30,15 +31,28 @@ export interface ReconciliationSummary { providerErrors: number; escalatedToManualReview: number; expiredSwept: number; + /** Payments escalated specifically because the total-attempt cap tripped. */ + attemptCapEscalations: number; + /** Payments escalated specifically because the provider-error streak cap tripped. */ + providerErrorStreakCapEscalations: number; } export interface ReconciliationMetrics { manualReviewQueueDepth: number; alertThreshold: number; alerting: boolean; + confirmationMaxAttempts: number; + confirmationMaxProviderErrorStreak: number; + attemptCapEscalations: number; + providerErrorStreakCapEscalations: number; } type ReconcileOutcome = 'resolved' | 'pending' | 'provider_error' | 'escalated'; +type ReconciliationCap = 'attempt' | 'provider_error_streak'; +type ReconcileResult = { + outcome: ReconcileOutcome; + caps?: ReconciliationCap[]; +}; /** * Scheduled reconciliation (issue #1572): treats provider truth as @@ -65,9 +79,16 @@ export class ReconciliationService { */ private isRunning = false; + private lastRunCapState = { + attemptCapEscalations: 0, + providerErrorStreakCapEscalations: 0, + }; + constructor( @InjectRepository(Payment) private readonly paymentRepository: Repository, + @InjectRepository(ReconciliationRun) + private readonly runRepository: Repository, private readonly confirmationService: PaymentConfirmationService, private readonly railRegistry: PaymentRailRegistry, private readonly gateway: PaymentsGateway, @@ -87,11 +108,20 @@ export class ReconciliationService { } this.isRunning = true; + const startedAt = new Date(); + let run: ReconciliationRun | null = null; try { - const started = Date.now(); + run = await this.startRunRecord(startedAt); const summary = await this.reconcileDueBatch(); - this.metrics.recordReconciliationPass(Date.now() - started); - this.logger.log(withRequestId(`Reconciliation pass: ${JSON.stringify(summary)}`)); + this.lastRunCapState = { + attemptCapEscalations: summary.attemptCapEscalations, + providerErrorStreakCapEscalations: + summary.providerErrorStreakCapEscalations, + }; + this.metrics.recordReconciliationPass(Date.now() - startedAt.getTime()); + this.logger.log( + withRequestId(`Reconciliation pass: ${JSON.stringify(summary)}`), + ); const metrics = await this.getMetrics(); this.metrics.setManualReviewDepth(metrics.manualReviewQueueDepth); if (metrics.alerting) { @@ -101,8 +131,15 @@ export class ReconciliationService { `exceeds threshold ${metrics.alertThreshold}`, ), ); - await this.sendManualReviewAlert(metrics.manualReviewQueueDepth, metrics.alertThreshold); + await this.sendManualReviewAlert( + metrics.manualReviewQueueDepth, + metrics.alertThreshold, + ); } + await this.completeRunRecord(run, summary, startedAt); + } catch (error) { + await this.failRunRecord(run, error, startedAt); + throw error; } finally { this.isRunning = false; } @@ -122,6 +159,10 @@ export class ReconciliationService { 'PAYMENT_RECONCILE_MAX_BATCH', 500, ); + // MANUAL_REVIEW is deliberately excluded by the status predicate, and + // isDueForPoll repeats that exclusion as a defensive guard. A payment + // that has crossed a cap is therefore never selected on a later pass, + // even if a stale read briefly shows its previous AWAITING state. const awaiting = await this.paymentRepository.find({ where: { status: PaymentStatus.AWAITING_CONFIRMATION }, take: maxBatch, @@ -134,15 +175,35 @@ export class ReconciliationService { providerErrors: 0, escalatedToManualReview: 0, expiredSwept, + attemptCapEscalations: 0, + providerErrorStreakCapEscalations: 0, }; for (const payment of awaiting) { + // A row that was already at a cap when this process started (for + // example, data written by the pre-cap deployment) is escalated + // without making one more provider call. This preserves the hard cap + // instead of merely skipping it forever in AWAITING_CONFIRMATION. + const reachedCaps = this.getReachedCaps(payment); + if (reachedCaps.length > 0) { + if (await this.escalateCappedPayment(payment, reachedCaps)) { + summary.escalatedToManualReview++; + for (const cap of reachedCaps) { + if (cap === 'attempt') { + summary.attemptCapEscalations++; + } else { + summary.providerErrorStreakCapEscalations++; + } + } + } + continue; + } if (!this.isDueForPoll(payment, now)) { continue; } summary.candidates++; - const outcome = await this.reconcileOne(payment, now); - switch (outcome) { + const result = await this.reconcileOne(payment, now); + switch (result.outcome) { case 'resolved': summary.resolved++; break; @@ -156,8 +217,20 @@ export class ReconciliationService { summary.escalatedToManualReview++; break; } + for (const cap of result.caps ?? []) { + if (cap === 'attempt') { + summary.attemptCapEscalations++; + } else { + summary.providerErrorStreakCapEscalations++; + } + } } + this.lastRunCapState = { + attemptCapEscalations: summary.attemptCapEscalations, + providerErrorStreakCapEscalations: + summary.providerErrorStreakCapEscalations, + }; return summary; } @@ -169,7 +242,12 @@ export class ReconciliationService { `Payment in status ${payment.status} is not awaiting confirmation`, ); } - await this.reconcileOne(payment, new Date()); + const reachedCaps = this.getReachedCaps(payment); + if (reachedCaps.length > 0) { + await this.escalateCappedPayment(payment, reachedCaps); + } else { + await this.reconcileOne(payment, new Date()); + } return this.getPaymentOrThrow(paymentId); } @@ -217,6 +295,20 @@ export class ReconciliationService { }); } + /** Returns persisted reconciliation history, newest first, with a bounded take. */ + async listRecentRuns( + limit: number | string = 50, + ): Promise { + const numericLimit = Number(limit); + const boundedLimit = Number.isFinite(numericLimit) + ? Math.min(Math.max(Math.floor(numericLimit), 1), 500) + : 50; + return this.runRepository.find({ + order: { startedAt: 'DESC' }, + take: boundedLimit, + }); + } + async getMetrics(): Promise { const manualReviewQueueDepth = await this.paymentRepository.count({ where: { status: PaymentStatus.MANUAL_REVIEW }, @@ -229,9 +321,139 @@ export class ReconciliationService { manualReviewQueueDepth, alertThreshold, alerting: manualReviewQueueDepth > alertThreshold, + confirmationMaxAttempts: this.getConfirmationMaxAttempts(), + confirmationMaxProviderErrorStreak: + this.getConfirmationMaxProviderErrorStreak(), + attemptCapEscalations: this.lastRunCapState.attemptCapEscalations, + providerErrorStreakCapEscalations: + this.lastRunCapState.providerErrorStreakCapEscalations, }; } + private async startRunRecord( + startedAt: Date, + ): Promise { + try { + const run = this.runRepository.create({ + startedAt, + finishedAt: null, + durationMs: null, + candidates: 0, + resolved: 0, + pending: 0, + providerErrors: 0, + escalatedToManualReview: 0, + expiredSwept: 0, + outcome: null, + error: null, + details: null, + }); + return await this.runRepository.save(run); + } catch (error) { + this.logRunPersistenceFailure('create', error); + return null; + } + } + + private async completeRunRecord( + run: ReconciliationRun | null, + summary: ReconciliationSummary, + startedAt: Date, + ): Promise { + if (!run) { + return; + } + const finishedAt = new Date(); + try { + await this.runRepository.update(run.id, { + candidates: summary.candidates, + resolved: summary.resolved, + pending: summary.pending, + providerErrors: summary.providerErrors, + escalatedToManualReview: summary.escalatedToManualReview, + expiredSwept: summary.expiredSwept, + finishedAt, + durationMs: finishedAt.getTime() - startedAt.getTime(), + outcome: 'succeeded', + error: null, + details: { + attemptCap: this.getConfirmationMaxAttempts(), + providerErrorStreakCap: + this.getConfirmationMaxProviderErrorStreak(), + attemptCapEscalations: summary.attemptCapEscalations, + providerErrorStreakCapEscalations: + summary.providerErrorStreakCapEscalations, + }, + }); + } catch (error) { + this.logRunPersistenceFailure('complete', error); + } + } + + private async failRunRecord( + run: ReconciliationRun | null, + error: unknown, + startedAt: Date, + ): Promise { + if (!run) { + return; + } + const finishedAt = new Date(); + try { + await this.runRepository.update(run.id, { + finishedAt, + durationMs: finishedAt.getTime() - startedAt.getTime(), + outcome: 'failed', + error: + error instanceof Error + ? error.message + : String(error ?? 'Unknown error'), + details: { + phase: 'reconciliation', + attemptCap: this.getConfirmationMaxAttempts(), + providerErrorStreakCap: + this.getConfirmationMaxProviderErrorStreak(), + }, + }); + } catch (persistenceError) { + this.logRunPersistenceFailure('failure', persistenceError); + } + } + + private logRunPersistenceFailure( + operation: 'create' | 'complete' | 'failure', + error: unknown, + ): void { + this.logger.warn( + withRequestId( + `Unable to ${operation} reconciliation run record: ${ + error instanceof Error ? error.message : String(error) + }`, + ), + ); + } + + private getConfirmationMaxAttempts(): number { + return this.readPositiveIntegerConfig( + 'PAYMENT_CONFIRMATION_MAX_ATTEMPTS', + 20, + ); + } + + private getConfirmationMaxProviderErrorStreak(): number { + return this.readPositiveIntegerConfig( + 'PAYMENT_CONFIRMATION_MAX_PROVIDER_ERROR_STREAK', + 10, + ); + } + + private readPositiveIntegerConfig(key: string, fallback: number): number { + const configured = Number(this.config.get(key, fallback)); + return Number.isFinite(configured) && configured > 0 + ? Math.floor(configured) + : fallback; + } + private async sendManualReviewAlert( depth: number, threshold: number, @@ -311,7 +533,7 @@ export class ReconciliationService { private async reconcileOne( payment: Payment, now: Date, - ): Promise { + ): Promise { if (!payment.providerReference) { this.logger.warn( `Payment ${payment.id} is AWAITING_CONFIRMATION with no providerReference`, @@ -371,7 +593,63 @@ export class ReconciliationService { : {}), }); - return 'resolved'; + return { outcome: 'resolved' }; + } + + private getReachedCaps(payment: Payment): ReconciliationCap[] { + const caps: ReconciliationCap[] = []; + if ( + payment.reconciliationAttempts >= this.getConfirmationMaxAttempts() + ) { + caps.push('attempt'); + } + if ( + payment.providerErrorStreak >= + this.getConfirmationMaxProviderErrorStreak() + ) { + caps.push('provider_error_streak'); + } + return caps; + } + + private async escalateCappedPayment( + payment: Payment, + caps: ReconciliationCap[], + ): Promise { + const fresh = await this.getPaymentOrThrow(payment.id); + if (fresh.status !== PaymentStatus.AWAITING_CONFIRMATION) { + return false; + } + assertValidTransition(fresh.status, PaymentStatus.MANUAL_REVIEW); + const maxAttempts = this.getConfirmationMaxAttempts(); + const maxProviderErrorStreak = + this.getConfirmationMaxProviderErrorStreak(); + const capDescriptions: string[] = []; + if (caps.includes('attempt')) { + capDescriptions.push( + `attempt cap ${maxAttempts} reached (${fresh.reconciliationAttempts} total reconciliation attempts)`, + ); + } + if (caps.includes('provider_error_streak')) { + capDescriptions.push( + `provider-error streak cap ${maxProviderErrorStreak} reached (${fresh.providerErrorStreak} consecutive provider errors)`, + ); + } + const description = capDescriptions.join('; '); + fresh.status = PaymentStatus.MANUAL_REVIEW; + fresh.manualReviewReason = + `Confirmation polling stopped: ${description}; ` + + `total reconciliation attempts: ${fresh.reconciliationAttempts}; ` + + `provider error streak: ${fresh.providerErrorStreak}.`; + this.logger.warn( + withRequestId( + `ALERT: payment ${fresh.id} reached a confirmation polling cap ` + + `(${description})`, + ), + ); + await this.paymentRepository.save(fresh); + this.gateway.emitPaymentUpdate(fresh.id, fresh.status); + return true; } /** @@ -383,17 +661,26 @@ export class ReconciliationService { * provider error — a provider-side outage must never mass-flag every * in-flight payment after a single bad run. Only an attempt that * successfully reached the provider (and got "still pending") can push - * an old-enough payment into MANUAL_REVIEW. + * an old-enough payment into MANUAL_REVIEW or trip the total-attempt cap. + * The provider-error streak cap is the deliberate exception: a long, + * bounded streak on one payment is evidence that this payment needs a + * human, whereas one failed run is evidence only that a provider may be + * experiencing a broad outage. If a provider-error attempt happens to + * reach the total-attempt cap first, the next scheduler pass escalates + * the already-capped row without making another provider call. */ private async recordAttempt( payment: Payment, now: Date, opts: { providerError: boolean }, - ): Promise { + ): Promise { const attempts = payment.reconciliationAttempts + 1; const providerErrorStreak = opts.providerError ? payment.providerErrorStreak + 1 : 0; + const maxAttempts = this.getConfirmationMaxAttempts(); + const maxProviderErrorStreak = + this.getConfirmationMaxProviderErrorStreak(); await this.paymentRepository.update(payment.id, { reconciliationAttempts: attempts, @@ -401,25 +688,68 @@ export class ReconciliationService { lastReconciledAt: now, }); + const caps: ReconciliationCap[] = []; + if (attempts >= maxAttempts) { + caps.push('attempt'); + } + if (providerErrorStreak >= maxProviderErrorStreak) { + caps.push('provider_error_streak'); + } + const manualReviewAfterHours = this.config.get( 'PAYMENT_MANUAL_REVIEW_AFTER_HOURS', 24, ); const ageMs = now.getTime() - payment.createdAt.getTime(); + const attemptCapReached = caps.includes('attempt'); + const providerErrorStreakCapReached = + caps.includes('provider_error_streak'); const shouldEscalate = - !opts.providerError && ageMs >= manualReviewAfterHours * 3_600_000; + providerErrorStreakCapReached || + (!opts.providerError && + (attemptCapReached || + ageMs >= manualReviewAfterHours * 3_600_000)); if (!shouldEscalate) { - return opts.providerError ? 'provider_error' : 'pending'; + return { + outcome: opts.providerError ? 'provider_error' : 'pending', + caps: [], + }; } const fresh = await this.getPaymentOrThrow(payment.id); assertValidTransition(fresh.status, PaymentStatus.MANUAL_REVIEW); fresh.status = PaymentStatus.MANUAL_REVIEW; - fresh.manualReviewReason = `Unresolved after ${attempts} reconciliation attempts (${(ageMs / 3_600_000).toFixed(1)}h old)`; + if (caps.length > 0) { + const capDescriptions: string[] = []; + if (caps.includes('attempt')) { + capDescriptions.push( + `attempt cap ${maxAttempts} reached (${attempts} total reconciliation attempts)`, + ); + } + if (caps.includes('provider_error_streak')) { + capDescriptions.push( + `provider-error streak cap ${maxProviderErrorStreak} reached (${providerErrorStreak} consecutive provider errors)`, + ); + } + fresh.manualReviewReason = + `Confirmation polling stopped: ${capDescriptions.join('; ')}; ` + + `total reconciliation attempts: ${attempts}; ` + + `provider error streak: ${providerErrorStreak}.`; + this.logger.warn( + withRequestId( + `ALERT: payment ${payment.id} reached a confirmation polling cap ` + + `(${capDescriptions.join('; ')})`, + ), + ); + } else { + fresh.manualReviewReason = + `Unresolved after ${attempts} reconciliation attempts ` + + `(${(ageMs / 3_600_000).toFixed(1)}h old)`; + } await this.paymentRepository.save(fresh); this.gateway.emitPaymentUpdate(fresh.id, fresh.status); - return 'escalated'; + return { outcome: 'escalated', caps }; } /** @@ -428,6 +758,16 @@ export class ReconciliationService { * often instead of every single cron tick. */ private isDueForPoll(payment: Payment, now: Date): boolean { + if (payment.status !== PaymentStatus.AWAITING_CONFIRMATION) { + return false; + } + if ( + payment.reconciliationAttempts >= this.getConfirmationMaxAttempts() || + payment.providerErrorStreak >= + this.getConfirmationMaxProviderErrorStreak() + ) { + return false; + } const dueAfterMinutes = this.config.get( 'PAYMENT_RECONCILE_DUE_AFTER_MINUTES', 5, diff --git a/backend/src/wallets/wallets.service.spec.ts b/backend/src/wallets/wallets.service.spec.ts index 5d0b77b8..6e257fab 100644 --- a/backend/src/wallets/wallets.service.spec.ts +++ b/backend/src/wallets/wallets.service.spec.ts @@ -327,12 +327,55 @@ describe('WalletsService', () => { }); describe('getWalletStatus', () => { + function makeStatusManager( + account: WalletAccount | null, + balance: number, + events: string[] = [], + ) { + const accountQueryBuilder: any = { + setLock: jest.fn(() => { + events.push('account-lock'); + return accountQueryBuilder; + }), + where: jest.fn(() => { + events.push('account-where'); + return accountQueryBuilder; + }), + getOne: jest.fn(async () => { + events.push('account-read'); + return account; + }), + }; + const ledgerQueryBuilder: any = { + select: jest.fn(() => ledgerQueryBuilder), + where: jest.fn(() => ledgerQueryBuilder), + getRawOne: jest.fn(async () => { + events.push('ledger-aggregate'); + return { balance: String(balance) }; + }), + }; + const accountRepository = { + createQueryBuilder: jest.fn(() => accountQueryBuilder), + }; + const ledgerRepository = { + createQueryBuilder: jest.fn(() => ledgerQueryBuilder), + }; + const manager = { + getRepository: jest.fn((entity: unknown) => + entity === WalletAccount ? accountRepository : ledgerRepository, + ), + }; + return { manager, accountQueryBuilder, ledgerQueryBuilder }; + } + it('reports an unprovisioned wallet as such, with a zero balance', async () => { - walletAccountRepository.findOne.mockResolvedValueOnce(null); + const { manager } = makeStatusManager(null, 0); + transaction.mockImplementationOnce((cb: any) => cb(manager)); const status = await service.getWalletStatus('user-1'); expect(status).toEqual({ account: null, balance: 0, currency: 'XLM' }); + expect(transaction).toHaveBeenCalledTimes(1); }); it('reports a soft-deleted wallet as absent without reading its balance', async () => { @@ -348,17 +391,92 @@ describe('WalletsService', () => { it('sums the ledger to compute the balance for a provisioned wallet', async () => { const account = makeAccount(); - walletAccountRepository.findOne.mockResolvedValueOnce(account); - const queryBuilder = { - select: jest.fn().mockReturnThis(), - where: jest.fn().mockReturnThis(), - getRawOne: jest.fn().mockResolvedValue({ balance: '5000' }), - }; - ledgerRepository.createQueryBuilder.mockReturnValue(queryBuilder); + const { manager, accountQueryBuilder, ledgerQueryBuilder } = + makeStatusManager(account, 5000); + transaction.mockImplementationOnce((cb: any) => cb(manager)); const status = await service.getWalletStatus('user-1'); expect(status).toEqual({ account, balance: 5000, currency: 'XLM' }); + expect(accountQueryBuilder.setLock).toHaveBeenCalledWith( + 'pessimistic_write', + ); + expect(ledgerQueryBuilder.getRawOne).toHaveBeenCalledTimes(1); + }); + + it('serializes concurrent balance reads around an interleaved writer', async () => { + const account = makeAccount(); + const events: string[] = []; + let balance = 100; + let releaseFirstAggregate!: () => void; + let firstAggregateStarted!: () => void; + const firstAggregateStartedPromise = new Promise((resolve) => { + firstAggregateStarted = resolve; + }); + const firstAggregateRelease = new Promise((resolve) => { + releaseFirstAggregate = resolve; + }); + const { manager, ledgerQueryBuilder } = makeStatusManager( + account, + balance, + events, + ); + ledgerQueryBuilder.getRawOne.mockImplementation(async () => { + const snapshot = balance; + events.push(`ledger-aggregate:${snapshot}`); + const aggregateCount = events.filter((event) => + event.startsWith('ledger-aggregate:'), + ).length; + if (aggregateCount === 1) { + firstAggregateStarted(); + await firstAggregateRelease; + } + return { balance: String(snapshot) }; + }); + + let queue = Promise.resolve(); + const enqueue = (work: () => Promise) => { + const run = queue.then(work); + queue = run.then( + () => undefined, + () => undefined, + ); + return run; + }; + transaction.mockImplementation((cb: any) => enqueue(() => cb(manager))); + const interleaveWrite = () => + enqueue(async () => { + events.push('writer-lock'); + balance = 200; + events.push('writer-commit'); + }); + + const first = service.getWalletStatus('user-1'); + await firstAggregateStartedPromise; + const write = interleaveWrite(); + const second = service.getWalletStatus('user-1'); + releaseFirstAggregate(); + const [firstStatus, secondStatus] = await Promise.all([ + first, + write, + second, + ]); + + expect(firstStatus.balance).toBe(100); + expect(secondStatus.balance).toBe(200); + expect(events).toEqual([ + 'account-lock', + 'account-where', + 'account-read', + 'ledger-aggregate:100', + 'writer-lock', + 'writer-commit', + 'account-lock', + 'account-where', + 'account-read', + 'ledger-aggregate:200', + ]); + expect(transaction).toHaveBeenCalledTimes(2); }); }); diff --git a/backend/src/wallets/wallets.service.ts b/backend/src/wallets/wallets.service.ts index 9ed8d842..95e3f40b 100644 --- a/backend/src/wallets/wallets.service.ts +++ b/backend/src/wallets/wallets.service.ts @@ -99,15 +99,44 @@ export class WalletsService { } } + /** + * PostgreSQL uses READ COMMITTED by default (unless a connection overrides + * it), so the account lookup and a bare ledger aggregate would otherwise + * be two independent statement snapshots. This transaction takes the same + * `pessimistic_write` row lock as fundCustodialWallet before reading the + * aggregate, and runs the aggregate through that transaction's EntityManager. + * `pessimistic_write` is chosen over a share/read lock for consistency over + * maximum read concurrency: it is the same lock fundCustodialWallet takes, + * so the two operations cannot pass one another. Against writers in this + * service, the lock serializes the read either + * before the writer (it sees the pre-write committed ledger) or after it + * (it waits for the writer's commit and sees the post-write ledger), never + * a partially applied application. + * + * This guarantee does not cover writers outside WalletsService that insert + * wallet_ledger_entries without first taking the wallet_accounts row lock + * (direct SQL, another service, or a future debit path). The current + * repository has no settlement/credit wallet-debit path; fundCustodialWallet + * is the only ledger writer and uses this lock, but any new writer must + * participate in the same locking protocol. + */ async getWalletStatus(userId: string): Promise { - const account = await this.walletAccountRepository.findOne({ - where: { userId, deletedAt: IsNull() }, - }); - if (!account || account.deletedAt) { - return { account: null, balance: 0, currency: LEDGER_ASSET }; - } - const balance = await this.getBalance(account.id); - return { account, balance, currency: LEDGER_ASSET }; + return this.walletAccountRepository.manager.transaction( + async (manager) => { + const account = await manager + .getRepository(WalletAccount) + .createQueryBuilder('wallet_account') + .setLock('pessimistic_write') + .where('wallet_account.user_id = :userId', { userId }) + .andWhere('wallet_account.deleted_at IS NULL') + .getOne(); + if (!account) { + return { account: null, balance: 0, currency: LEDGER_ASSET }; + } + const balance = await this.getBalance(manager, account.id); + return { account, balance, currency: LEDGER_ASSET }; + }, + ); } /** @@ -361,14 +390,18 @@ export class WalletsService { } /** - * Sums the append-only ledger for an account that has already passed the - * active-wallet lookup. There is intentionally no deleted-entry predicate: - * financial history remains auditable after an account-level soft delete, - * while the account lookup prevents a deleted wallet from exposing it as a - * current balance. + * Runs the ledger aggregate through the caller's locked transaction + * manager. There is intentionally no deleted-entry predicate: financial + * history remains auditable after an account-level soft delete, while the + * caller's active-wallet lookup prevents a deleted wallet from exposing it + * as a current balance. */ - private async getBalance(walletAccountId: string): Promise { - const result = await this.ledgerRepository + private async getBalance( + manager: EntityManager, + walletAccountId: string, + ): Promise { + const result = await manager + .getRepository(WalletLedgerEntry) .createQueryBuilder('entry') .select( `COALESCE(SUM(CASE WHEN entry.type = 'CREDIT' THEN entry.amount ELSE -entry.amount END), 0)`,