PR: Matched Fund Consistency Verification & Repair System

[marshalfleet]

Closes #216

Summary

Implements a production-grade consistency verification and repair system for the
matchedTotal counter on the Multiplier table. matchedTotal is the running
total of all matched funds allocated against a multiplier's cap — it is the
authoritative value used by claimMatchCap to enforce the cap atomically. If
it drifts from the true sum of MatchedFund.matchedAmount rows, the system
either over-allocates matched funds (financial loss) or under-allocates them
(donor trust loss). This PR adds the safety net that detects and corrects such
drift automatically.

---

Problem

matchedTotal is maintained atomically by two code paths:

• Allocation (MatchedFundAllocationService.claimMatchCap): increments
  matchedTotal with a CTE + FOR UPDATE in a single SQL statement.
• Refund (DonationService.refundDonation): decrements matchedTotal with
  the same pattern.

The ADR in matchedFundAllocation.service.ts explains why this is correct under
concurrent allocation. However, the invariant only holds if every write to
MatchedFund goes through these two paths. Several real-world scenarios break
that assumption:

Scenario	Effect on matchedTotal
Admin script or migration inserts/deletes MatchedFund rows directly	Counter not updated
DB connection drops after MatchedFund INSERT commits but before Multiplier UPDATE commits	Counter behind reality
matchedTotal column backfill in migration is incomplete or wrong	Wrong starting value
Refund partially applied (status updated, counter decrement not committed)	Counter ahead of reality
Future code paths that forget to go through the service	Silent drift

Without a verification layer, drift is invisible until it causes a financial
discrepancy in a report or an audit.

---

Solution Overview

Three new components work together:

MatchedFundVerificationService   ←  pure business logic, injectable DB client
         ↑
matchedFundVerification.worker   ←  BullMQ scheduler + on-demand trigger
         ↑
src/index.ts                     ←  startup registration (feature-flagged)

Configuration is fully externalised — all thresholds, cron expressions, and
feature flags are driven by environment variables with safe defaults.

---

Files Changed

src/config/index.ts — new matchedFundVerification config block

Adds a matchedFundVerification section with 10 tunable parameters:

Parameter	Env var	Default	Purpose
enabled	MATCHED_FUND_VERIFICATION_ENABLED	true	Feature flag
fullVerificationCron	MATCHED_FUND_FULL_VERIFICATION_CRON	0 2 * * *	Daily full sweep cron
samplingVerificationCron	MATCHED_FUND_SAMPLING_CRON	10 * * * *	Hourly sampling cron
samplingPercent	MATCHED_FUND_SAMPLING_PERCENT	10	% of rows in each sampling pass
precisionThreshold	MATCHED_FUND_PRECISION_THRESHOLD	0.00000001	Min discrepancy to classify as inconsistent
alertInconsistencyRateThreshold	MATCHED_FUND_ALERT_INCONSISTENCY_RATE	0.05	Fraction of rows inconsistent before systemic alert
alertLargeDiscrepancyThreshold	MATCHED_FUND_ALERT_LARGE_DISCREPANCY	1000	Single-row discrepancy that triggers a large-discrepancy alert
repairMaxRetries	MATCHED_FUND_REPAIR_MAX_RETRIES	3	Max repair attempts per row
repairRetryDelayMs	MATCHED_FUND_REPAIR_RETRY_DELAY_MS	500	Delay between repair retries
repairBatchLimit	MATCHED_FUND_REPAIR_BATCH_LIMIT	100	Max rows repaired per job invocation

All values are plain strings/numbers — no runtime library construction in the
config module.

---

.env.example — new env vars

All 10 env vars documented with defaults and inline explanations, following the
existing style of the file.

---

src/services/matchedFundVerification.service.ts (new)

The core service. All methods accept an injectable db parameter (defaults to
the module-level Prisma singleton) so they can be unit-tested without a real
database connection.

verify(opts, db?) — main entry point

Orchestrates a complete verification + repair pass. Accepts a VerifyOptions
object:

interface VerifyOptions {
  mode: 'full' | 'sampling' | 'triggered';
  multiplierIds?: string[];   // for triggered mode only
  samplingPercent?: number;   // override per-run sampling %
  repair?: boolean;           // default true
}

Flow:

1. Dispatches to queryFull, querySampling, or queryTriggered depending on
   mode.
2. Checks the systemic-inconsistency rate. If inconsistentCount / examined > alertInconsistencyRateThreshold, logs an ERROR-level systemic_inconsistency
   alert and skips the repair phase to avoid silently patching widespread
   corruption.
3. For each inconsistent row, checks whether its discrepancy exceeds
   alertLargeDiscrepancyThreshold and logs a large_discrepancy alert if so.
4. Calls repairOne for each inconsistent row (up to repairBatchLimit). Logs
   a warning if the batch limit is hit and rows are deferred to the next run.
5. Returns a VerificationResult with full observability data:

interface VerificationResult {
  mode: 'full' | 'sampling' | 'triggered';
  startedAt: string;        // ISO-8601
  finishedAt: string;       // ISO-8601
  examined: number;
  inconsistentCount: number;
  repairedCount: number;
  repairFailureCount: number;
  inconsistencies: InconsistentMultiplier[];
  systemicAlert: boolean;
  durationMs: number;
}

queryFull(precisionThreshold, db?) — full verification

Runs a single aggregation query against all Multiplier rows:

SELECT
  m.id,
  m."matchedTotal"::text               AS "storedTotal",
  COALESCE(SUM(mf."matchedAmount"), 0)::text AS "actualSum",
  (COALESCE(SUM(mf."matchedAmount"), 0) - m."matchedTotal")::text AS discrepancy
FROM "Multiplier" m
LEFT JOIN "MatchedFund" mf
       ON mf."multiplierId" = m.id
      AND mf."refundedAt" IS NULL     -- exclude refunded rows
GROUP BY m.id, m."matchedTotal"
HAVING ABS(COALESCE(SUM(mf."matchedAmount"), 0) - m."matchedTotal")
       > $threshold

Key design points:

• The HAVING clause pushes the threshold filter into Postgres — only
  inconsistent rows are returned over the wire; consistent rows (the vast
  majority in a healthy system) produce zero network traffic.
• refundedAt IS NULL excludes refunded MatchedFund rows because
  refundDonation decrements matchedTotal and sets refundedAt
  atomically — excluding refunded rows keeps both sides of the invariant in
  sync.
• The count query runs in parallel with the aggregate query (Promise.all)
  via a cheap SELECT COUNT(*) FROM "Multiplier" (no JOIN) so the total
  examined count is available for the rate calculation without adding
  overhead to the main query.
• No MatchedFund rows are loaded into Node.js memory regardless of dataset
  size.

querySampling(samplingPercent, precisionThreshold, db?) — sampling pass

Same aggregation query but with TABLESAMPLE SYSTEM($pct) on the Multiplier
scan. PostgreSQL's TABLESAMPLE SYSTEM operates on 8 KB pages, so the actual
sample fraction varies slightly from the nominal percentage — this is acceptable
for an early-warning signal. The percent is clamped to [0.000001, 100] before
use.

queryTriggered(multiplierIds, precisionThreshold, db?) — targeted verification

Adds WHERE m.id = ANY($ids::text[]) to the aggregation. Short-circuits
immediately (no DB call) when multiplierIds is empty. The examined count
is set to multiplierIds.length since those are the rows we intended to check.

repairOne(inconsistent, db?) — single-row repair with retries

for attempt in 0..repairMaxRetries:
  BEGIN TRANSACTION
    SELECT "matchedTotal" FROM "Multiplier" WHERE id = $id FOR UPDATE
    SELECT COALESCE(SUM("matchedAmount"), 0) FROM "MatchedFund"
      WHERE "multiplierId" = $id AND "refundedAt" IS NULL
    if |re-sum - locked_total| <= threshold: COMMIT (no-op)
    UPDATE "Multiplier" SET "matchedTotal" = re-sum WHERE id = $id
  COMMIT
  return success
  on error: sleep repairRetryDelayMs, retry
return failure

Concurrency safety:

• The FOR UPDATE on the Multiplier row serialises against concurrent
  claimMatchCap and refundDonation calls on the same row.
• The re-sum runs inside the same transaction as the lock, so it reads a
  snapshot consistent with the locked row — no TOCTOU gap.
• The re-check before the UPDATE makes the repair idempotent: if a
  concurrent transaction (allocation or a previous repair run) already
  corrected the value between the scan and the repair, the re-check detects
  discrepancy ≤ threshold and exits without writing. The locked value, not
  the stale scan value, is used as oldValue in the result.
• Deadlock prevention: the Multiplier row lock is always the first lock
  acquired (same as in claimMatchCap), so there is no lock-order cycle
  between repair and allocation.

On permanent failure (all retries exhausted), logs an ERROR-level
repair_failure alert and returns success: false. The verify() call site
counts these as repairFailureCount in the result.

injectInconsistency(multiplierId, value, db?) — test helper

Directly sets matchedTotal to an arbitrary value without going through the
allocation service. Used in tests to simulate drift. Must never be called
in production code paths.

---

src/workers/matchedFundVerification.worker.ts (new)

BullMQ queue + worker for the verification system.

Queue: matched-fund-verification

Default job options: 3 attempts, exponential backoff starting at 5 s,
removeOnComplete: true, removeOnFail: 100.

Job types

Type	Trigger	Behaviour
FULL_VERIFICATION	Daily cron (fullVerificationCron)	Verifies and repairs all Multiplier rows
SAMPLING_VERIFICATION	Hourly cron (samplingVerificationCron)	Verifies and repairs a random sample
TRIGGERED_VERIFICATION	On-demand via enqueueTriggeredVerification()	Verifies and optionally repairs specific rows

scheduleVerificationJobs() — idempotent schedule registration

Registers both recurring jobs using repeat: { pattern: cron } + a stable
jobId. BullMQ deduplicates by jobId, so calling this on every startup is
safe and correct. Skips registration entirely when the feature flag is off.

enqueueTriggeredVerification(multiplierIds, repair?) — ad-hoc trigger

await enqueueTriggeredVerification(['mult-abc', 'mult-xyz']);
// or verify-only without repair:
await enqueueTriggeredVerification(['mult-abc'], false);

Enqueues a TRIGGERED_VERIFICATION job with a unique jobId derived from the
sorted multiplier IDs + timestamp. Logs a warning and returns immediately
(without throwing) when the feature flag is off.

Worker concurrency

concurrency: 1 — only one verification job runs at a time. This prevents
two full-scan jobs from hammering the database simultaneously if a previous
run was delayed and two fire close together. It also prevents a full scan
and a sampling pass from running in parallel.

Observability

The worker logs structured JSON at INFO level for job start/complete and
ERROR level for job failures, including jobType and the error stack.

---

src/index.ts — worker registration

Adds the verification worker startup alongside the existing workers:

if (config.matchedFundVerification.enabled) {
  import('./workers/matchedFundVerification.worker.js')
    .then(({ scheduleVerificationJobs }) => scheduleVerificationJobs())
    .then(() => logger.info('Matched fund verification worker started'))
    .catch((error) => logger.error('Failed to start matched fund verification worker:', error));
}

Uses the same dynamic-import + feature-flag pattern as the moderation, receipt,
and email workers. The worker only connects to Redis/BullMQ when the module is
imported, so disabling the flag has zero overhead.

---

src/services/matchedFundVerification.service.test.ts (new)

38 unit tests covering all acceptance criteria. No real database connection is
used — all DB interaction is through an injectable stub built with makeDb().

Test stub design

function makeDb(queryRawResults: QueryRawResult[]) {
  // Sequential queue of $queryRaw return values.
  // $transaction passes the stub itself as the tx client so that
  // tx.$queryRaw uses the same queue — all queries in a single flat array.
  return { $queryRaw, $transaction };
}

This design means a test that stages [countResult, aggregateResult, lockResult, sumResult, updateResult] fully exercises the verify + repair
path without any Jest spy overhead on the Prisma model API.

Test suites

Suite	Tests	What is verified
verify() — full mode	6	Zero inconsistencies; single repair; multi-row repair; systemic alert aborts repair; repair: false; repairBatchLimit
verify() — sampling mode	3	Mode tag; custom pct override; sampling repair
verify() — triggered mode	3	ID filtering; empty-array short-circuit; triggered repair
queryFull() — precision threshold	4	No rows within threshold; rows above threshold; Decimal mapping; zero actualSum
querySampling()	3	Inconsistent rows returned; pct clamped low; pct clamped high
queryTriggered()	3	Empty-array no-op; single query issued; consistent result
repairOne()	8	Success path; old/new value logging; no-op on concurrent fix; retry on transient error; all-retries-exhausted failure; repair_failure alert; missing-row failure; Decimal precision
Large-discrepancy alerting	2	Alert fires above threshold; does not fire below threshold
injectInconsistency()	2	Raw UPDATE issued; Decimal value accepted
Concurrency regression	2	Verify completes during concurrent writes; no-op repair doesn't corrupt concurrent allocation
VerificationResult structure	2	ISO timestamps present; mode string correct

Acceptance criteria cross-reference

AC	Test(s)
Verification detects matchedTotal ≠ sum(matchedAmount)	"detects and repairs a single inconsistent row"
Floating-point precision handling (small diffs ignored)	"returns no rows when discrepancy is within the precision threshold"
Large datasets use server-side aggregation	All queryFull / querySampling tests (mock structure forces single query)
Repair updates matchedTotal to actual sum	"updates matchedTotal to the actual sum and returns success=true"
Repair logs old/new values for audit	"logs old and new values when repair succeeds"
Repair handles failures gracefully (retry)	"retries on transient error and succeeds on the second attempt"
Full verification (all multipliers)	verify() — full mode suite
Sampling verification (random subset)	verify() — sampling mode suite
Triggered verification (after deployment)	verify() — triggered mode suite
Allocation continues during verification	"completes successfully when called while the DB is accepting writes"
No-op repair does not corrupt concurrent allocation	"repair no-op path does not corrupt a concurrent allocation"

---

Alerting Strategy

Three alert conditions emit logger.error with a structured payload:

// 1. Systemic inconsistency — too many rows are wrong at once
logger.error('[matchedFundVerification] ALERT: systemic_inconsistency', {
  alert: 'systemic_inconsistency',
  inconsistentCount, examined, inconsistencyRate, threshold,
});

// 2. Large discrepancy on a single row
logger.error('[matchedFundVerification] ALERT: large_discrepancy', {
  alert: 'large_discrepancy',
  multiplierId, storedTotal, actualSum, discrepancy, threshold,
});

// 3. Repair failed after all retries
logger.error('[matchedFundVerification] ALERT: repair_failure', {
  alert: 'repair_failure',
  multiplierId, error,
});

The service does not integrate with any alerting SDK directly. In production,
these ERROR-level structured log lines should be forwarded to the alerting
backend (PagerDuty, Datadog, etc.) by the log shipper — the same pattern used
everywhere else in this codebase.

---

Performance Characteristics

Operation	Cost	Notes
Full verification (read)	1 DB query, O(M + F) server-side	M = multipliers, F = MatchedFund rows; only inconsistent rows returned over wire
Sampling verification (read)	1 DB query, O(pct × M + pct × F)	TABLESAMPLE prunes at page level
Repair (single row)	1 interactive transaction, 3 queries	FOR UPDATE + re-sum + UPDATE; typically < 5 ms
Repair (batch of N)	N sequential transactions	N ≤ repairBatchLimit (default 100)

The verification query does not hold any locks. Concurrent allocations and
refunds are unaffected during the read phase.

---

How to Use

Normal operation (automatic)

Nothing to do — the worker starts automatically when MATCHED_FUND_VERIFICATION_ENABLED=true
(the default). A daily full sweep runs at 02:00 UTC; an hourly sampling pass
runs at :10.

Trigger manually after a data fix

import { enqueueTriggeredVerification } from './workers/matchedFundVerification.worker';

// After fixing multiplier 'abc' in a migration script:
await enqueueTriggeredVerification(['abc']);

// Verify only, no repair:
await enqueueTriggeredVerification(['abc'], false);

Simulate inconsistency in tests / staging

await MatchedFundVerificationService.injectInconsistency('mult-id', '0');
const result = await MatchedFundVerificationService.verify({ mode: 'triggered', multiplierIds: ['mult-id'] });
// result.inconsistentCount === 1, result.repairedCount === 1

Tune via environment variables

# Run full sweep twice daily instead of once
MATCHED_FUND_FULL_VERIFICATION_CRON="0 2,14 * * *"

# Sample 25% of rows on each hourly pass
MATCHED_FUND_SAMPLING_PERCENT=25

# Alert if any single row is off by more than $100
MATCHED_FUND_ALERT_LARGE_DISCREPANCY=100

# Disable the feature entirely without removing the code
MATCHED_FUND_VERIFICATION_ENABLED=false

---

Testing

# Run just the new tests
npm test -- --testPathPattern=matchedFundVerification.service.test

# Run with coverage
npm run test:coverage -- --testPathPattern=matchedFundVerification

All 38 tests pass. No existing tests were modified or broken.

---

Out of Scope (not in this PR)

• Real-time (per-allocation) consistency checks — periodic verification is
  sufficient and lower overhead.
• A REST endpoint to trigger verification — enqueueTriggeredVerification is
  the programmatic API; exposing it over HTTP is a follow-up.
• Distributed transactions across multiple databases — single DB assumed.
• Full audit trail for every matched-fund operation beyond the existing logging.