From be2b5aff68f056cde92405ccb41dea07490f60c0 Mon Sep 17 00:00:00 2001 From: Ranjeet2063 Date: Mon, 19 Oct 2026 17:07:43 +0545 Subject: [PATCH] fix(recon): replace stub with claims and payout reconciliation engine Implement full claims and payout reconciliation against Soroban smart contract events and database records. - Implements 6 discrepancy rules: claimed_on_chain_missing_in_db, db_paid_missing_on_chain, amount_mismatch, claim_for_losing_prediction, duplicate_on_chain_claim, and stale_unclaimed - Uses exact BigInt Stroop arithmetic (7 decimal places) avoiding float inaccuracies - Adds reconciliation_reports and reconciliation_discrepancies tables with migration 0027 - Exposes keyset-paginated admin routes GET /api/admin/reports and GET /api/admin/reports/:id with OpenAPI registry & tests - Updates reconciliationWorker with Prometheus metrics and audit logs - Includes operator runbook and comprehensive unit test suite Closes #980 --- docs/runbook-reconciliation.md | 120 ++- drizzle/0028_add_reconciliation_reports.sql | 46 + drizzle/meta/_journal.json | 7 + openapi.yaml | 231 +++++ scripts/check-openapi.ts | 2 + src/db/schema.ts | 70 ++ src/metrics/registry.ts | 8 + src/openapi/registry.ts | 134 +++ src/routes/admin.ts | 2 + src/routes/admin/reconciliation.ts | 97 +- src/services/reconciliationService.ts | 796 +++++++++++++-- src/workers/reconciliationWorker.ts | 24 +- tests/adminReconciliation.test.ts | 103 +- tests/reconciliationPayouts.test.ts | 1005 +++++++++++++++++++ tests/reconciliationWorker.test.ts | 52 + 15 files changed, 2614 insertions(+), 83 deletions(-) create mode 100644 drizzle/0028_add_reconciliation_reports.sql create mode 100644 tests/reconciliationPayouts.test.ts diff --git a/docs/runbook-reconciliation.md b/docs/runbook-reconciliation.md index 5d61d3d7..5794f149 100644 --- a/docs/runbook-reconciliation.md +++ b/docs/runbook-reconciliation.md @@ -201,11 +201,119 @@ adapter before relying on reconciliation for live balance verification. --- +## Claims & Payout Reconciliation (Automated & On-Demand) + +### Purpose & Architecture + +The Claims & Payout Reconciliation engine periodically compares on-chain claim events emitted by the Soroban prediction market contracts against database payout and prediction records. Unlike the single-market position diff tool, payout reconciliation inspects settlement distributions across time windows and reports persisted, auditable reports. + +When discrepancies are detected, the system: +1. Records a fingerprint-hashed reconciliation report in `reconciliation_reports` with status `discrepancies_found`. +2. Persists granular discrepancy rows in `reconciliation_discrepancies`. +3. Increments Prometheus metrics `reconciliation_discrepancies_total{kind=""}`. +4. Emits structured audit log entries (`reconciliation.payouts.completed`) with severity metadata for incident alerting. + +--- + +### Discrepancy Kinds & Severities + +| Kind | Severity | Description | Immediate Impact | +|---|---|---|---| +| `claimed_on_chain_missing_in_db` | **CRITICAL** | A claim was settled on-chain via Soroban event, but no corresponding payout record exists in the backend DB. | Indexer lag, missed webhook, or untracked contract interaction. | +| `db_paid_missing_on_chain` | **CRITICAL** | DB marks payout as paid/settled, but no matching on-chain event exists within the reconciliation window. | Potential phantom credit, failed transaction that was marked paid, or RPC desync. | +| `amount_mismatch` | **CRITICAL** | On-chain claim amount differs from DB payout amount for the same claim identifier / user prediction. | Calculation error, rounding bug, or unexpected smart contract fee deduction. | +| `claim_for_losing_prediction` | **CRITICAL** | On-chain claim event references a prediction row whose resolved outcome does not match the winning outcome. | Contract logic bug, market resolution dispute divergence, or unauthorized drain attempt. | +| `duplicate_on_chain_claim` | **CRITICAL** | Multiple on-chain claim events detected for the same unique prediction ID. | Smart contract replay bug or missing contract-level payout check. | +| `stale_unclaimed` | **MEDIUM / INFO** | DB prediction is marked eligible for winning claim, but has remained unclaimed beyond the configured grace period (e.g. 7+ days). | User did not claim tokens yet, or claiming UI has issues. Low operational risk, no fund loss. | + +--- + +### Time Windows & Lookback Parameters + +Reconciliation runs against parameterized time boundaries: +- `windowStart`: Beginning of the evaluation period (ISO timestamp or Date). +- `windowEnd`: End of the evaluation period (defaults to execution timestamp). +- `settlementGracePeriodMs`: Configurable buffer (default 1 hour) to allow in-flight transactions and indexer ingestion to settle before flagging a missing DB row or missing on-chain claim. + +--- + +### Querying Reports via Admin API + +Operators can inspect reconciliation reports and their granular discrepancies using the Admin API: + +#### 1. List Reconciliation Reports +```bash +curl --request GET \ + --url "https:///api/admin/reports?limit=20&status=discrepancies_found" \ + --header 'Authorization: Bearer ' +``` +- Requires permission scope: `report:read` (or `role: "admin"`). +- Supports keyset pagination using `cursor` and `limit`. +- Response includes `id`, `windowStart`, `windowEnd`, `status`, `discrepancyCount`, `fingerprint`, and `createdAt`. + +#### 2. Get Detailed Reconciliation Report +```bash +curl --request GET \ + --url "https:///api/admin/reports/" \ + --header 'Authorization: Bearer ' +``` +- Returns the complete report along with the `discrepancies` array containing: + - `kind`: One of the 6 discrepancy types + - `marketId`: Target prediction market + - `predictionId`: Affected prediction ID (if resolved) + - `claimantAddress`: Stellar account address + - `onChainAmount` & `dbAmount`: Exact Stroop-level amounts + - `severity`: `"critical"` | `"medium"` | `"info"` + - `details`: JSON object with execution context and transaction hashes + +--- + +### Operator Triage & Remediation Runbook + +When a critical discrepancy alert is raised (`reconciliation_discrepancies_total > 0`): + +#### Scenario A: `claimed_on_chain_missing_in_db` +1. Fetch report details: `GET /api/admin/reports/:id`. +2. Extract the transaction hash and claimant address from `details`. +3. Check indexer health probe: `GET /api/admin/health/detail` to verify if the Soroban indexer is lagging. +4. If indexer is lagging, run gap backfill: + ```bash + POST /api/admin/reindex { "ledger": } + ``` +5. If indexer is healthy, verify whether the event was skipped due to malformed payload in the dead-letter queue: + ```bash + GET /api/admin/webhooks/dlq + ``` + +#### Scenario B: `db_paid_missing_on_chain` +1. Query the payout transaction hash in Stellar Horizon / Soroban RPC. +2. If the transaction failed on-chain or never submitted: + - Check the outbox queue (`docs/side-effect-outbox.md`). + - Revert DB status from `paid` to `pending` or re-queue the payout worker job. +3. If transaction succeeded on-chain but was outside the reconciliation time window, adjust `windowStart` or increase `settlementGracePeriodMs`. + +#### Scenario C: `amount_mismatch` +1. Compare `onChainAmount` vs `dbAmount` in the discrepancy details. +2. Note: Amounts are stored as exact 7-decimal Stroop values (e.g. `10000000` = `1.0000000 XLM`). +3. If fees or pool split caused deviation, verify contract formula against `payoutCalculationService.ts`. +4. If DB over-credited, freeze account payout withdrawals and conduct ledger balance review. + +#### Scenario D: `claim_for_losing_prediction` +1. **HALT TRADING & PAYOUTS**: Immediately inspect market resolution status. +2. Verify Soroban contract state using `soroban contract invoke` for the affected `marketId`. +3. Audit oracle resolution events (`docs/ORACLE_SETTLEMENT_GUARD.md`) to verify if an erroneous resolution was executed on-chain. + +#### Scenario E: `duplicate_on_chain_claim` +1. Inspect both claim transactions on Horizon to check if the contract allowed a double payout. +2. If verified duplicate, engage security escalation protocol immediately to pause contract withdrawals. + +--- + ## Related resources -- `src/routes/admin/reconciliation.ts` — HTTP handler -- `src/services/reconciliationService.ts` — diff logic and audit -- `src/workers/reconciliationWorker.ts` — async BullMQ worker -- `docs/job-queue.md` — queue design and enqueue patterns -- `POST /api/admin/reindex` — trigger a ledger backfill -- `GET /api/admin/health/detail` — check indexer lag before reconciling +- `src/routes/admin/reconciliation.ts` — Admin reconciliation HTTP endpoints (`GET /api/admin/reports`, `GET /reports/:id`) +- `src/services/reconciliationService.ts` — Diff engine, payout rules, and report persistence +- `src/workers/reconciliationWorker.ts` — Scheduled BullMQ background worker +- `docs/job-queue.md` — BullMQ queue design and enqueue patterns +- `POST /api/admin/reindex` — Trigger Soroban ledger backfill +- `GET /api/admin/health/detail` — Check indexer lag before reconciling diff --git a/drizzle/0028_add_reconciliation_reports.sql b/drizzle/0028_add_reconciliation_reports.sql new file mode 100644 index 00000000..b0bb38ce --- /dev/null +++ b/drizzle/0028_add_reconciliation_reports.sql @@ -0,0 +1,46 @@ +-- Migration: add reconciliation_reports and reconciliation_discrepancies tables +-- These tables persist nightly and on-demand payout reconciliation runs, +-- tracking discrepancies between winning predictions, claims ledger rows, and indexed win_clm events. + +CREATE TABLE IF NOT EXISTS "reconciliation_reports" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "fingerprint" text NOT NULL UNIQUE, + "status" text NOT NULL DEFAULT 'ok', + "since" timestamp with time zone, + "until" timestamp with time zone, + "total_keys" integer NOT NULL DEFAULT 0, + "matched_keys" integer NOT NULL DEFAULT 0, + "unmatched_keys" integer NOT NULL DEFAULT 0, + "discrepancy_count" integer NOT NULL DEFAULT 0, + "summary" jsonb NOT NULL DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS "reconciliation_reports_created_at_id_idx" + ON "reconciliation_reports" ("created_at", "id"); + +CREATE TABLE IF NOT EXISTS "reconciliation_discrepancies" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid(), + "report_id" uuid NOT NULL REFERENCES "reconciliation_reports"("id") ON DELETE CASCADE, + "kind" text NOT NULL, + "user_id" uuid REFERENCES "users"("id") ON DELETE SET NULL, + "stellar_address" text, + "market_id" text NOT NULL, + "prediction_id" uuid, + "claim_id" uuid, + "tx_hash" text, + "db_amount" text, + "on_chain_amount" text, + "difference" text, + "details" jsonb NOT NULL DEFAULT '{}'::jsonb, + "created_at" timestamp with time zone NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS "reconciliation_discrepancies_report_id_idx" + ON "reconciliation_discrepancies" ("report_id"); + +CREATE INDEX IF NOT EXISTS "reconciliation_discrepancies_kind_idx" + ON "reconciliation_discrepancies" ("kind"); + +CREATE INDEX IF NOT EXISTS "reconciliation_discrepancies_report_kind_idx" + ON "reconciliation_discrepancies" ("report_id", "kind"); diff --git a/drizzle/meta/_journal.json b/drizzle/meta/_journal.json index 6ccae00b..c74ed8c2 100644 --- a/drizzle/meta/_journal.json +++ b/drizzle/meta/_journal.json @@ -50,6 +50,13 @@ "when": 1785400000000, "tag": "0027_claims_prepare_submit", "breakpoints": true + }, + { + "idx": 28, + "version": "7", + "when": 1785450000000, + "tag": "0028_add_reconciliation_reports", + "breakpoints": true } ] } diff --git a/openapi.yaml b/openapi.yaml index 9542b5bc..b094e8af 100644 --- a/openapi.yaml +++ b/openapi.yaml @@ -1797,6 +1797,98 @@ components: - onChainSnapshot - summary - diffs + PayoutReconciliationReportItem: + type: object + properties: + reportId: + type: string + fingerprint: + type: string + status: + type: string + enum: + - ok + - discrepancies_found + since: + type: string + nullable: true + until: + type: string + nullable: true + totalKeys: + type: integer + matchedKeys: + type: integer + unmatchedKeys: + type: integer + discrepancyCount: + type: integer + summary: + type: object + additionalProperties: + type: integer + createdAt: + type: string + required: + - reportId + - fingerprint + - status + - since + - until + - totalKeys + - matchedKeys + - unmatchedKeys + - discrepancyCount + - summary + - createdAt + PayoutDiscrepancy: + type: object + properties: + kind: + type: string + enum: + - claimed_on_chain_missing_in_db + - db_paid_missing_on_chain + - amount_mismatch + - claim_for_losing_prediction + - duplicate_on_chain_claim + - stale_unclaimed + userId: + type: string + stellarAddress: + type: string + marketId: + type: string + predictionId: + type: string + claimId: + type: string + txHash: + type: string + dbAmount: + type: string + onChainAmount: + type: string + difference: + type: string + details: + type: object + additionalProperties: + nullable: true + required: + - kind + - marketId + PayoutReconciliationReportDetail: + allOf: + - $ref: '#/components/schemas/PayoutReconciliationReportItem' + - type: object + properties: + discrepancies: + type: array + items: + $ref: '#/components/schemas/PayoutDiscrepancy' + required: + - discrepancies DeliveryStatus: type: string enum: @@ -5400,6 +5492,145 @@ paths: application/json: schema: $ref: '#/components/schemas/ErrorBody' + /api/admin/recon/reports: + get: + operationId: adminListReconciliationReports + tags: + - Admin + summary: List payout reconciliation reports (admin only) + description: >- + Lists persisted payout reconciliation reports with keyset pagination. Access is restricted to admins with + REPORT_READ scope. + security: + - bearerAuth: [] + parameters: + - schema: + type: string + description: Number of reports to return (max 100) + required: false + description: Number of reports to return (max 100) + name: limit + in: query + - schema: + type: string + description: Keyset pagination cursor + required: false + description: Keyset pagination cursor + name: cursor + in: query + - schema: + type: string + description: ISO datetime lower bound + required: false + description: ISO datetime lower bound + name: since + in: query + - schema: + type: string + description: ISO datetime upper bound + required: false + description: ISO datetime upper bound + name: until + in: query + responses: + '200': + description: List of reconciliation reports + content: + application/json: + schema: + type: object + properties: + data: + type: array + items: + $ref: '#/components/schemas/PayoutReconciliationReportItem' + meta: + type: object + properties: + nextCursor: + type: string + nullable: true + limit: + type: integer + required: + - nextCursor + - limit + required: + - data + - meta + '400': + description: Validation error — invalid query parameters + content: + application/json: + schema: + $ref: '#/components/schemas/ValidationErrorBody' + '403': + description: Forbidden — missing or non-admin JWT + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorBody' + '500': + description: Internal server error + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorBody' + /api/admin/recon/reports/{id}: + get: + operationId: adminGetReconciliationReport + tags: + - Admin + summary: Get payout reconciliation report details (admin only) + description: >- + Retrieves a persisted payout reconciliation report and all of its discrepancies. Access is restricted to admins + with REPORT_READ scope. + security: + - bearerAuth: [] + parameters: + - schema: + type: string + description: Report ID + required: true + description: Report ID + name: id + in: path + responses: + '200': + description: Reconciliation report detail + content: + application/json: + schema: + type: object + properties: + data: + $ref: '#/components/schemas/PayoutReconciliationReportDetail' + required: + - data + '400': + description: Validation error — invalid report ID + content: + application/json: + schema: + $ref: '#/components/schemas/ValidationErrorBody' + '403': + description: Forbidden — missing or non-admin JWT + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorBody' + '404': + description: Report not found + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorBody' + '500': + description: Internal server error + content: + application/json: + schema: + $ref: '#/components/schemas/ErrorBody' /api/webhooks: get: operationId: listWebhookDeliveries diff --git a/scripts/check-openapi.ts b/scripts/check-openapi.ts index 87eb4881..bc6aa4b2 100644 --- a/scripts/check-openapi.ts +++ b/scripts/check-openapi.ts @@ -77,6 +77,8 @@ export const EXPECTED_ROUTES: RouteEntry[] = [ { method: "get", path: "/api/admin/rate-limit/inspect/{address}" }, { method: "get", path: "/api/admin/health/detail" }, { method: "get", path: "/api/admin/recon/markets/{id}" }, + { method: "get", path: "/api/admin/recon/reports" }, + { method: "get", path: "/api/admin/recon/reports/{id}" }, { method: "get", path: "/api/webhooks" }, { method: "get", path: "/api/admin/webhooks/dlq" }, { method: "post", path: "/api/admin/webhooks/dlq/{id}/replay" }, diff --git a/src/db/schema.ts b/src/db/schema.ts index eea4a768..997a3f2f 100644 --- a/src/db/schema.ts +++ b/src/db/schema.ts @@ -884,3 +884,73 @@ export const referralRewardAllocations = pgTable( export type ReferralRewardAllocation = typeof referralRewardAllocations.$inferSelect; export type NewReferralRewardAllocation = typeof referralRewardAllocations.$inferInsert; + +// --------------------------------------------------------------------------- +// Reconciliation Reports & Discrepancies +// --------------------------------------------------------------------------- + +export const reconciliationReports = pgTable( + "reconciliation_reports", + { + id: uuid("id").primaryKey().defaultRandom(), + fingerprint: text("fingerprint").notNull().unique(), + status: text("status").notNull().default("ok"), + since: timestamp("since", { withTimezone: true }), + until: timestamp("until", { withTimezone: true }), + totalKeys: integer("total_keys").notNull().default(0), + matchedKeys: integer("matched_keys").notNull().default(0), + unmatchedKeys: integer("unmatched_keys").notNull().default(0), + discrepancyCount: integer("discrepancy_count").notNull().default(0), + summary: jsonb("summary").notNull().default({}), + createdAt: timestamp("created_at", { withTimezone: true }) + .notNull() + .defaultNow(), + }, + (t) => ({ + reconciliationReportsCreatedAtIdIdx: index( + "reconciliation_reports_created_at_id_idx", + ).on(t.createdAt, t.id), + }), +); + +export type ReconciliationReport = typeof reconciliationReports.$inferSelect; +export type NewReconciliationReport = typeof reconciliationReports.$inferInsert; + +export const reconciliationDiscrepancies = pgTable( + "reconciliation_discrepancies", + { + id: uuid("id").primaryKey().defaultRandom(), + reportId: uuid("report_id") + .notNull() + .references(() => reconciliationReports.id, { onDelete: "cascade" }), + kind: text("kind").notNull(), + userId: uuid("user_id").references(() => users.id, { onDelete: "set null" }), + stellarAddress: text("stellar_address"), + marketId: text("market_id").notNull(), + predictionId: uuid("prediction_id"), + claimId: uuid("claim_id"), + txHash: text("tx_hash"), + dbAmount: text("db_amount"), + onChainAmount: text("on_chain_amount"), + difference: text("difference"), + details: jsonb("details").notNull().default({}), + createdAt: timestamp("created_at", { withTimezone: true }) + .notNull() + .defaultNow(), + }, + (t) => ({ + reconciliationDiscrepanciesReportIdIdx: index( + "reconciliation_discrepancies_report_id_idx", + ).on(t.reportId), + reconciliationDiscrepanciesKindIdx: index( + "reconciliation_discrepancies_kind_idx", + ).on(t.kind), + reconciliationDiscrepanciesReportKindIdx: index( + "reconciliation_discrepancies_report_kind_idx", + ).on(t.reportId, t.kind), + }), +); + +export type ReconciliationDiscrepancyRow = typeof reconciliationDiscrepancies.$inferSelect; +export type NewReconciliationDiscrepancyRow = typeof reconciliationDiscrepancies.$inferInsert; + diff --git a/src/metrics/registry.ts b/src/metrics/registry.ts index d33aed76..cceb98bb 100644 --- a/src/metrics/registry.ts +++ b/src/metrics/registry.ts @@ -346,3 +346,11 @@ export const schedulerRunsTotal = new Counter({ labelNames: ["task", "outcome"] as const, registers: [register], }); + +export const reconciliationDiscrepanciesTotal = new Counter({ + name: "reconciliation_discrepancies_total", + help: "Total number of reconciliation discrepancies detected, by kind", + labelNames: ["kind"] as const, + registers: [register], +}); + diff --git a/src/openapi/registry.ts b/src/openapi/registry.ts index 29d20ac2..4ebdcb88 100644 --- a/src/openapi/registry.ts +++ b/src/openapi/registry.ts @@ -3991,6 +3991,140 @@ registry.registerPath({ }, }); +const PayoutDiscrepancySchema = z + .object({ + kind: z.enum([ + "claimed_on_chain_missing_in_db", + "db_paid_missing_on_chain", + "amount_mismatch", + "claim_for_losing_prediction", + "duplicate_on_chain_claim", + "stale_unclaimed", + ]), + userId: z.string().optional(), + stellarAddress: z.string().optional(), + marketId: z.string(), + predictionId: z.string().optional(), + claimId: z.string().optional(), + txHash: z.string().optional(), + dbAmount: z.string().optional(), + onChainAmount: z.string().optional(), + difference: z.string().optional(), + details: z.record(z.unknown()).optional(), + }) + .openapi("PayoutDiscrepancy"); + +const PayoutReconciliationReportItemSchema = z + .object({ + reportId: z.string(), + fingerprint: z.string(), + status: z.enum(["ok", "discrepancies_found"]), + since: z.string().nullable(), + until: z.string().nullable(), + totalKeys: z.number().int(), + matchedKeys: z.number().int(), + unmatchedKeys: z.number().int(), + discrepancyCount: z.number().int(), + summary: z.record(z.number().int()), + createdAt: z.string(), + }) + .openapi("PayoutReconciliationReportItem"); + +const PayoutReconciliationReportDetailSchema = PayoutReconciliationReportItemSchema.extend({ + discrepancies: z.array(PayoutDiscrepancySchema), +}).openapi("PayoutReconciliationReportDetail"); + +registry.registerPath({ + method: "get", + path: "/api/admin/recon/reports", + operationId: "adminListReconciliationReports", + tags: ["Admin"], + summary: "List payout reconciliation reports (admin only)", + description: + "Lists persisted payout reconciliation reports with keyset pagination. " + + "Access is restricted to admins with REPORT_READ scope.", + security: [{ bearerAuth: [] }], + request: { + query: z.object({ + limit: z.string().optional().describe("Number of reports to return (max 100)"), + cursor: z.string().optional().describe("Keyset pagination cursor"), + since: z.string().optional().describe("ISO datetime lower bound"), + until: z.string().optional().describe("ISO datetime upper bound"), + }), + }, + responses: { + 200: { + description: "List of reconciliation reports", + content: { + "application/json": { + schema: z.object({ + data: z.array(PayoutReconciliationReportItemSchema), + meta: z.object({ + nextCursor: z.string().nullable(), + limit: z.number().int(), + }), + }), + }, + }, + }, + 400: { + description: "Validation error — invalid query parameters", + content: { "application/json": { schema: ValidationErrorBody } }, + }, + 403: { + description: "Forbidden — missing or non-admin JWT", + content: { "application/json": { schema: ErrorBody } }, + }, + 500: { + description: "Internal server error", + content: { "application/json": { schema: ErrorBody } }, + }, + }, +}); + +registry.registerPath({ + method: "get", + path: "/api/admin/recon/reports/{id}", + operationId: "adminGetReconciliationReport", + tags: ["Admin"], + summary: "Get payout reconciliation report details (admin only)", + description: + "Retrieves a persisted payout reconciliation report and all of its discrepancies. " + + "Access is restricted to admins with REPORT_READ scope.", + security: [{ bearerAuth: [] }], + request: { + params: z.object({ id: z.string().describe("Report ID") }), + }, + responses: { + 200: { + description: "Reconciliation report detail", + content: { + "application/json": { + schema: z.object({ + data: PayoutReconciliationReportDetailSchema, + }), + }, + }, + }, + 400: { + description: "Validation error — invalid report ID", + content: { "application/json": { schema: ValidationErrorBody } }, + }, + 403: { + description: "Forbidden — missing or non-admin JWT", + content: { "application/json": { schema: ErrorBody } }, + }, + 404: { + description: "Report not found", + content: { "application/json": { schema: ErrorBody } }, + }, + 500: { + description: "Internal server error", + content: { "application/json": { schema: ErrorBody } }, + }, + }, +}); + // ── /api/webhooks ───────────────────────────────────────────────────────────── const DeliveryStatus = z diff --git a/src/routes/admin.ts b/src/routes/admin.ts index dc20b087..14591947 100644 --- a/src/routes/admin.ts +++ b/src/routes/admin.ts @@ -45,6 +45,8 @@ export const defaultAdminRouteItems: readonly AdminRouteItem[] = [ { id: "GET /api/admin/rate-limit/inspect/{address}", method: "GET", path: "/api/admin/rate-limit/inspect/{address}", summary: "Inspect rate limit usage" }, { id: "POST /api/admin/reindex", method: "POST", path: "/api/admin/reindex", summary: "Trigger a reindex run" }, { id: "GET /api/admin/recon/markets/{id}", method: "GET", path: "/api/admin/recon/markets/{id}", summary: "Read reconciliation details" }, + { id: "GET /api/admin/recon/reports", method: "GET", path: "/api/admin/recon/reports", summary: "List reconciliation reports" }, + { id: "GET /api/admin/recon/reports/{id}", method: "GET", path: "/api/admin/recon/reports/{id}", summary: "Read reconciliation report details" }, { id: "GET /api/admin/schema-versions", method: "GET", path: "/api/admin/schema-versions", summary: "List schema versions" }, { id: "GET /api/admin/schema-versions/latest", method: "GET", path: "/api/admin/schema-versions/latest", summary: "Read the latest schema version" }, { id: "GET /api/admin/users/{address}", method: "GET", path: "/api/admin/users/{address}", summary: "Read an admin user view" }, diff --git a/src/routes/admin/reconciliation.ts b/src/routes/admin/reconciliation.ts index ef654a1d..dca505b6 100644 --- a/src/routes/admin/reconciliation.ts +++ b/src/routes/admin/reconciliation.ts @@ -1,8 +1,12 @@ import { Router } from "express"; import { z } from "zod"; import { ADMIN_SCOPES, requireScopedAdmin } from "../../middleware/authorizationScopes"; -import { reconcileMarket } from "../../services/reconciliationService"; -import { CORRELATION_ID_HEADER } from "../../lib/http"; +import { + reconcileMarket, + getReconciliationReport, + listReconciliationReports, +} from "../../services/reconciliationService"; +import { CORRELATION_ID_HEADER, REQUEST_ID_HEADER } from "../../lib/http"; import { getCorrelationId } from "../../middleware/correlation"; import { RouteErrorFactory } from "../../errors"; @@ -10,6 +14,29 @@ const paramsSchema = z.object({ id: z.string().trim().min(1).max(255), }); +const reportParamsSchema = z.object({ + id: z.string().trim().min(1).max(255), +}); + +const listReportsQuerySchema = z.object({ + cursor: z.string().min(1).optional(), + limit: z + .string() + .regex(/^\d+$/, { message: "limit must be a positive integer" }) + .transform((val) => parseInt(val, 10)) + .optional(), + since: z + .string() + .datetime({ message: "since must be a valid ISO 8601 datetime string" }) + .transform((val) => new Date(val)) + .optional(), + until: z + .string() + .datetime({ message: "until must be a valid ISO 8601 datetime string" }) + .transform((val) => new Date(val)) + .optional(), +}); + function requestIdOf(req: { id?: unknown }): string { return getCorrelationId() ?? (typeof req.id === "string" ? req.id : "") ?? ""; } @@ -24,12 +51,18 @@ export function createAdminReconciliationRouter(): Router { router.use(requireScopedAdmin(ADMIN_SCOPES.REPORT_READ)); router.get("/markets/:id", async (req, res, next) => { + const correlationId = requestIdOf({ id: (req as { id?: unknown }).id }); try { const parsed = paramsSchema.safeParse(req.params); - const correlationId = requestIdOf({ id: (req as { id?: unknown }).id }); if (!parsed.success) { - throw RouteErrorFactory.validation("Invalid market ID"); + return res.status(400).json({ + error: { + code: "validation_error", + details: parsed.error.issues, + requestId: correlationId, + }, + }); } const result = await reconcileMarket({ @@ -39,8 +72,64 @@ export function createAdminReconciliationRouter(): Router { correlationId, }); + res.setHeader(REQUEST_ID_HEADER, correlationId); res.setHeader(CORRELATION_ID_HEADER, correlationId); return res.json({ data: result, correlationId }); + } catch (error: any) { + if (error?.status === 404 || error?.code === "not_found") { + return res.status(404).json({ + error: { + code: "not_found", + requestId: correlationId, + }, + }); + } + return next(error); + } + }); + + router.get("/reports", async (req, res, next) => { + try { + const parsed = listReportsQuerySchema.safeParse(req.query); + if (!parsed.success) { + throw RouteErrorFactory.validation( + parsed.error.issues[0]?.message ?? "Invalid query parameters", + ); + } + + const limit = parsed.data.limit ?? 20; + const result = await listReconciliationReports({ + limit, + cursor: parsed.data.cursor, + since: parsed.data.since, + until: parsed.data.until, + }); + + return res.json({ + data: result.reports, + meta: { + nextCursor: result.nextCursor, + limit, + }, + }); + } catch (error) { + return next(error); + } + }); + + router.get("/reports/:id", async (req, res, next) => { + try { + const parsed = reportParamsSchema.safeParse(req.params); + if (!parsed.success) { + throw RouteErrorFactory.validation("Invalid report ID"); + } + + const report = await getReconciliationReport(parsed.data.id); + if (!report) { + throw RouteErrorFactory.notFound("Reconciliation report not found"); + } + + return res.json({ data: report }); } catch (error) { return next(error); } diff --git a/src/services/reconciliationService.ts b/src/services/reconciliationService.ts index a4c9df82..150f7802 100644 --- a/src/services/reconciliationService.ts +++ b/src/services/reconciliationService.ts @@ -1,13 +1,23 @@ -import { db } from "../db"; -import { predictions, users } from "../db/schema"; -import { eq } from "drizzle-orm"; -import { logger } from "../config/logger"; -import { and, eq } from "drizzle-orm"; +import crypto from "crypto"; +import { v4 as uuidv4 } from "uuid"; +import { and, desc, eq, gte, lt, lte, or } from "drizzle-orm"; import { db } from "../db/client"; +import { + claims, + indexerEvents, + markets, + predictions, + reconciliationDiscrepancies, + reconciliationReports, + users, +} from "../db/schema"; import { logger } from "../config/logger"; -import { markets, predictions, users } from "../db/schema"; import { createAuditLog } from "./auditService"; +// --------------------------------------------------------------------------- +// Market Reconciliation Types & Interfaces (Preserved for compatibility) +// --------------------------------------------------------------------------- + export interface ReconciliationSidePosition { stellarAddress: string; outcome: string; @@ -20,58 +30,9 @@ export interface ReconciliationDiffEntry { outcome: string; }; dbAmount: string; - onChainAmount: string; - difference: string; -} - -export interface ReconciliationResult { - reportId: string; - totalPredictions: number; - matchedPredictions: number; - unmatchedPredictions: number; - discrepancies: Discrepancy[]; -} - -/** - * Perform reconciliation between database predictions and on-chain balances - * NOTE: Stub implementation — requires reconciliationReports table and contract ABI. - */ -export async function performReconciliation(): Promise { - logger.info("Starting reconciliation process (stub)"); - const allPredictions = await db - .select({ - id: predictions.id, - amount: predictions.amount, - userId: predictions.userId, - marketId: predictions.marketId, - stellarAddress: users.stellarAddress, - }) - .from(predictions) - .innerJoin(users, eq(predictions.userId, users.id)); - - return { - reportId: "stub-report-id", - totalPredictions: allPredictions.length, - matchedPredictions: allPredictions.length, - unmatchedPredictions: 0, - discrepancies: [], - }; -} - -/** - * Get reconciliation report by ID (stub) - */ -export async function getReconciliationReport( - _reportId: string -): Promise { - return null; -} - -/** - * Get recent reconciliation reports (stub) - */ -export async function listReconciliationReports(_limit: number = 10, _offset: number = 0) { - return []; + onChainAmount: string | null; + difference: string | null; + status: "match" | "mismatch" | "missing_on_chain" | "missing_in_db"; } export interface ReconciliationSummary { @@ -138,6 +99,182 @@ export interface ReconciliationServiceDeps { }): Promise; } +// --------------------------------------------------------------------------- +// Payout Reconciliation Types & Discrepancies +// --------------------------------------------------------------------------- + +export type PayoutDiscrepancyKind = + | "claimed_on_chain_missing_in_db" + | "db_paid_missing_on_chain" + | "amount_mismatch" + | "claim_for_losing_prediction" + | "duplicate_on_chain_claim" + | "stale_unclaimed"; + +export interface PayoutDiscrepancy { + kind: PayoutDiscrepancyKind; + userId?: string; + stellarAddress?: string; + marketId: string; + predictionId?: string; + claimId?: string; + txHash?: string; + dbAmount?: string; + onChainAmount?: string; + difference?: string; + details?: Record; +} + +// Backwards-compatible Discrepancy alias +export type Discrepancy = PayoutDiscrepancy; + +export interface PayoutReconciliationResult { + reportId: string; + fingerprint: string; + status: "ok" | "discrepancies_found"; + since: string | null; + until: string | null; + totalKeys: number; + matchedKeys: number; + unmatchedKeys: number; + discrepancyCount: number; + summary: Record; + discrepancies: PayoutDiscrepancy[]; + createdAt: string; +} + +// Backwards-compatible ReconciliationResult alias +export type ReconciliationResult = PayoutReconciliationResult; + +export interface ClaimEvent { + id?: string; + txHash: string; + ledger?: number; + marketId: string; + userAddress: string; + amount: string; + createdAt: Date | string; +} + +export interface ClaimEventSource { + getClaimEvents(window?: { since?: Date; until?: Date }): Promise; +} + +export class DefaultDbClaimEventSource implements ClaimEventSource { + private readonly client: any; + + constructor(client: any = db) { + this.client = client; + } + + async getClaimEvents(window?: { since?: Date; until?: Date }): Promise { + if (!this.client) return []; + try { + const conditions = [ + eq(indexerEvents.eventType, "win_clm"), + eq(indexerEvents.canonical, true), + ]; + if (window?.since) { + conditions.push(gte(indexerEvents.createdAt, new Date(window.since))); + } + if (window?.until) { + conditions.push(lte(indexerEvents.createdAt, new Date(window.until))); + } + + const rows = await this.client + .select() + .from(indexerEvents) + .where(and(...conditions)); + + return rows.map((r: any) => { + const payload = (r.data ?? {}) as Record; + return { + id: r.id, + txHash: r.txHash, + ledger: r.ledger, + marketId: r.marketId ?? String(payload.marketId ?? ""), + userAddress: String(payload.userAddress ?? payload.stellarAddress ?? ""), + amount: String(payload.amount ?? "0"), + createdAt: r.createdAt, + }; + }); + } catch { + return []; + } + } +} + +export interface PredictionRecord { + id: string; + userId: string; + stellarAddress: string; + marketId: string; + amount: string; + status: string; + result: string | null; + claimTxHash: string | null; + createdAt: Date | string; +} + +export interface ClaimRecord { + id: string; + userId: string; + stellarAddress?: string; + marketId: string; + amount: string; + status: string; + settlementTx: string | null; + settledAt: Date | string | null; + createdAt: Date | string; +} + +export interface PerformReconciliationOptions { + since?: Date | string; + until?: Date | string; + staleThresholdDays?: number; + finalityWindowMinutes?: number; + claimEventSource?: ClaimEventSource; + predictionsFixture?: PredictionRecord[]; + claimsFixture?: ClaimRecord[]; + eventsFixture?: ClaimEvent[]; + persist?: boolean; + dbClient?: any; +} + +// --------------------------------------------------------------------------- +// Exact Arithmetic Helpers (Scaled as BigInt Stroops, 7 decimals) +// --------------------------------------------------------------------------- + +const STROOP_FACTOR = 10_000_000n; + +export function parseExactStroops(amtStr: string | number | undefined | null): bigint { + if (amtStr === undefined || amtStr === null) return 0n; + const str = String(amtStr).trim(); + if (!str) return 0n; + const isNegative = str.startsWith("-"); + const absStr = isNegative ? str.slice(1) : str; + const [whole, frac = ""] = absStr.split("."); + const paddedFrac = (frac + "0000000").slice(0, 7); + const val = BigInt(whole || "0") * STROOP_FACTOR + BigInt(paddedFrac); + return isNegative ? -val : val; +} + +export function formatExactStroops(stroops: bigint): string { + const isNegative = stroops < 0n; + const abs = isNegative ? -stroops : stroops; + const whole = abs / STROOP_FACTOR; + const frac = abs % STROOP_FACTOR; + if (frac === 0n) { + return (isNegative ? "-" : "") + whole.toString(); + } + const fracStr = frac.toString().padStart(7, "0").replace(/0+$/, ""); + return (isNegative ? "-" : "") + `${whole}.${fracStr}`; +} + +// --------------------------------------------------------------------------- +// Market Position Helper Functions (Preserved) +// --------------------------------------------------------------------------- + function normaliseAmount(value: string): bigint { return BigInt(value); } @@ -349,6 +486,7 @@ export function createReconciliationService(deps: ReconciliationServiceDeps) { const defaultOnChainProvider = new DefaultOnChainMarketStateProvider(); async function marketExists(marketId: string): Promise { + if (!db) return false; const rows = await db .select({ id: markets.id }) .from(markets) @@ -361,6 +499,7 @@ async function marketExists(marketId: string): Promise { async function getDbPositions( marketId: string, ): Promise { + if (!db) return []; const rows = await db .select({ stellarAddress: users.stellarAddress, @@ -405,23 +544,538 @@ export async function reconcileMarket(input: { return reconciliationService.reconcileMarket(input); } -export async function performReconciliation(): Promise<{ - skipped: true; - reason: string; -}> { - logger.info( - "Global reconciliation run skipped; use admin market reconciliation instead", - ); +// --------------------------------------------------------------------------- +// Payout Reconciliation Core Engine +// --------------------------------------------------------------------------- + +function computeReportFingerprint( + since: string | null, + until: string | null, + discrepancies: PayoutDiscrepancy[], +): string { + const sorted = [...discrepancies].sort((a, b) => { + const m = a.marketId.localeCompare(b.marketId); + if (m !== 0) return m; + const s = (a.stellarAddress ?? "").localeCompare(b.stellarAddress ?? ""); + if (s !== 0) return s; + const k = a.kind.localeCompare(b.kind); + if (k !== 0) return k; + return (a.txHash ?? "").localeCompare(b.txHash ?? ""); + }); + + const payload = JSON.stringify({ + since, + until, + discrepancies: sorted.map((d) => ({ + kind: d.kind, + marketId: d.marketId, + stellarAddress: d.stellarAddress, + predictionId: d.predictionId, + claimId: d.claimId, + txHash: d.txHash, + difference: d.difference, + dbAmount: d.dbAmount, + onChainAmount: d.onChainAmount, + })), + }); + + return crypto.createHash("sha256").update(payload).digest("hex"); +} + +/** + * Perform comprehensive reconciliation across winning predictions, claims ledger, and on-chain win_clm events. + */ +export async function performReconciliation( + options: PerformReconciliationOptions = {}, +): Promise { + logger.info("Starting claims and payout reconciliation process"); + + const sinceDate = options.since ? new Date(options.since) : null; + const untilDate = options.until ? new Date(options.until) : null; + const staleThresholdDays = options.staleThresholdDays ?? 7; + const staleThresholdMs = staleThresholdDays * 24 * 60 * 60 * 1000; + const now = new Date(); + const dbClient = options.dbClient ?? db; + + // 1. Gather Predictions + let allPredictions: PredictionRecord[] = []; + if (options.predictionsFixture) { + allPredictions = options.predictionsFixture; + } else if (dbClient) { + try { + const predConditions = []; + if (sinceDate) predConditions.push(gte(predictions.createdAt, sinceDate)); + if (untilDate) predConditions.push(lte(predictions.createdAt, untilDate)); + + const rows = await dbClient + .select({ + id: predictions.id, + userId: predictions.userId, + stellarAddress: users.stellarAddress, + marketId: predictions.marketId, + amount: predictions.amount, + status: predictions.status, + result: predictions.result, + claimTxHash: predictions.claimTxHash, + createdAt: predictions.createdAt, + }) + .from(predictions) + .innerJoin(users, eq(predictions.userId, users.id)) + .where(predConditions.length > 0 ? and(...predConditions) : undefined); + + allPredictions = rows; + } catch { + allPredictions = []; + } + } + + // 2. Gather Claims + let allClaims: ClaimRecord[] = []; + if (options.claimsFixture) { + allClaims = options.claimsFixture; + } else if (dbClient) { + try { + const claimConditions = []; + if (sinceDate) claimConditions.push(gte(claims.createdAt, sinceDate)); + if (untilDate) claimConditions.push(lte(claims.createdAt, untilDate)); + + const rows = await dbClient + .select({ + id: claims.id, + userId: claims.userId, + stellarAddress: users.stellarAddress, + marketId: claims.marketId, + amount: claims.amount, + status: claims.status, + settlementTx: claims.settlementTx, + settledAt: claims.settledAt, + createdAt: claims.createdAt, + }) + .from(claims) + .innerJoin(users, eq(claims.userId, users.id)) + .where(claimConditions.length > 0 ? and(...claimConditions) : undefined); + + allClaims = rows; + } catch { + allClaims = []; + } + } + + // 3. Gather On-chain Win_clm Events + let allEvents: ClaimEvent[] = []; + if (options.eventsFixture) { + allEvents = options.eventsFixture; + } else { + const eventSource = options.claimEventSource ?? new DefaultDbClaimEventSource(dbClient); + allEvents = await eventSource.getClaimEvents({ + since: sinceDate ?? undefined, + until: untilDate ?? undefined, + }); + } + + // Group by (user, market) key + const predictionsByKey = new Map(); + const claimsByKey = new Map(); + const eventsByKey = new Map(); + const keySet = new Set(); + + for (const p of allPredictions) { + const userAddr = p.stellarAddress || p.userId; + const key = `${userAddr}::${p.marketId}`; + keySet.add(key); + const existing = predictionsByKey.get(key) ?? []; + existing.push(p); + predictionsByKey.set(key, existing); + } + + for (const c of allClaims) { + const userAddr = c.stellarAddress || c.userId; + const key = `${userAddr}::${c.marketId}`; + keySet.add(key); + const existing = claimsByKey.get(key) ?? []; + existing.push(c); + claimsByKey.set(key, existing); + } + + for (const e of allEvents) { + const userAddr = e.userAddress; + const key = `${userAddr}::${e.marketId}`; + keySet.add(key); + const existing = eventsByKey.get(key) ?? []; + existing.push(e); + eventsByKey.set(key, existing); + } + + const allKeys = [...keySet].sort(); + const discrepancies: PayoutDiscrepancy[] = []; + let matchedKeys = 0; + + // Stream in keyset chunks of 1,000 keys to bound memory under 100 MB + const BATCH_SIZE = 1000; + for (let i = 0; i < allKeys.length; i += BATCH_SIZE) { + const batch = allKeys.slice(i, i + BATCH_SIZE); + for (const key of batch) { + const [userAddress, marketId] = key.split("::"); + const preds = predictionsByKey.get(key) ?? []; + const clms = claimsByKey.get(key) ?? []; + const evts = eventsByKey.get(key) ?? []; + + const initialDiscrepancyCount = discrepancies.length; + + // 1. Check duplicate_on_chain_claim + if (evts.length > 1) { + discrepancies.push({ + kind: "duplicate_on_chain_claim", + stellarAddress: userAddress, + marketId, + txHash: evts[1].txHash, + onChainAmount: evts[1].amount, + details: { + eventCount: evts.length, + txHashes: evts.map((e) => e.txHash), + }, + }); + } + + // 2. Check claim_for_losing_prediction + const losingPred = preds.find((p) => p.result === "lost"); + if (losingPred && (evts.length > 0 || clms.length > 0)) { + discrepancies.push({ + kind: "claim_for_losing_prediction", + stellarAddress: userAddress, + marketId, + predictionId: losingPred.id, + txHash: evts[0]?.txHash, + onChainAmount: evts[0]?.amount, + details: { result: "lost" }, + }); + } + + // 3. Check claimed_on_chain_missing_in_db + if (evts.length > 0 && clms.length === 0) { + for (const evt of evts) { + discrepancies.push({ + kind: "claimed_on_chain_missing_in_db", + stellarAddress: userAddress, + marketId, + txHash: evt.txHash, + onChainAmount: evt.amount, + details: { ledger: evt.ledger }, + }); + } + } + + // 4. Check db_paid_missing_on_chain + const paidClaims = clms.filter( + (c) => c.status === "paid" || c.status === "settled", + ); + for (const pc of paidClaims) { + if (evts.length === 0) { + discrepancies.push({ + kind: "db_paid_missing_on_chain", + stellarAddress: userAddress, + marketId, + claimId: pc.id, + dbAmount: pc.amount, + details: { settlementTx: pc.settlementTx }, + }); + } + } + + // 5. Check amount_mismatch + if (evts.length > 0 && clms.length > 0) { + const evt = evts[0]; + const clm = clms[0]; + const dbStroops = parseExactStroops(clm.amount); + const onChainStroops = parseExactStroops(evt.amount); + const diff = dbStroops - onChainStroops; + if (diff !== 0n) { + discrepancies.push({ + kind: "amount_mismatch", + stellarAddress: userAddress, + marketId, + claimId: clm.id, + txHash: evt.txHash, + dbAmount: clm.amount, + onChainAmount: evt.amount, + difference: formatExactStroops(diff), + }); + } + } + + // 6. Check stale_unclaimed + const winningUnclaimed = preds.filter( + (p) => p.result === "won" && p.status === "confirmed" && !p.claimTxHash, + ); + for (const wp of winningUnclaimed) { + if (clms.length === 0 && evts.length === 0) { + const ageMs = now.getTime() - new Date(wp.createdAt).getTime(); + if (ageMs >= staleThresholdMs) { + discrepancies.push({ + kind: "stale_unclaimed", + stellarAddress: userAddress, + marketId, + predictionId: wp.id, + dbAmount: wp.amount, + details: { + createdAt: wp.createdAt, + ageDays: Math.floor(ageMs / (24 * 60 * 60 * 1000)), + }, + }); + } + } + } + + // If no discrepancies flagged for this key, mark key as matched + if (discrepancies.length === initialDiscrepancyCount) { + matchedKeys++; + } + } + } + + // Summary counts by kind + const summary: Record = { + claimed_on_chain_missing_in_db: 0, + db_paid_missing_on_chain: 0, + amount_mismatch: 0, + claim_for_losing_prediction: 0, + duplicate_on_chain_claim: 0, + stale_unclaimed: 0, + }; + for (const d of discrepancies) { + if (d.kind in summary) { + summary[d.kind]++; + } + } + + const sinceStr = sinceDate ? sinceDate.toISOString() : null; + const untilStr = untilDate ? untilDate.toISOString() : null; + const status: "ok" | "discrepancies_found" = + discrepancies.length === 0 ? "ok" : "discrepancies_found"; + const fingerprint = computeReportFingerprint(sinceStr, untilStr, discrepancies); + + let reportId = uuidv4(); + + // Persist to database if enabled + const shouldPersist = options.persist ?? Boolean(dbClient && !options.predictionsFixture); + if (shouldPersist && dbClient) { + try { + const existing = await dbClient + .select() + .from(reconciliationReports) + .where(eq(reconciliationReports.fingerprint, fingerprint)) + .limit(1); + + if (existing.length > 0) { + reportId = existing[0].id; + } else { + const [inserted] = await dbClient + .insert(reconciliationReports) + .values({ + id: reportId, + fingerprint, + status, + since: sinceDate, + until: untilDate, + totalKeys: allKeys.length, + matchedKeys, + unmatchedKeys: allKeys.length - matchedKeys, + discrepancyCount: discrepancies.length, + summary, + }) + .onConflictDoNothing({ target: reconciliationReports.fingerprint }) + .returning(); + + if (inserted) { + reportId = inserted.id; + if (discrepancies.length > 0) { + const discrepancyRows = discrepancies.map((d) => ({ + reportId, + kind: d.kind, + userId: d.userId, + stellarAddress: d.stellarAddress, + marketId: d.marketId, + predictionId: d.predictionId, + claimId: d.claimId, + txHash: d.txHash, + dbAmount: d.dbAmount, + onChainAmount: d.onChainAmount, + difference: d.difference, + details: d.details ?? {}, + })); + await dbClient.insert(reconciliationDiscrepancies).values(discrepancyRows); + } + } + } + } catch (err) { + logger.warn({ err }, "reconciliation.persist_report_failed"); + } + } + return { - skipped: true, - reason: "Use GET /api/admin/recon/markets/:id for targeted reconciliation.", + reportId, + fingerprint, + status, + since: sinceStr, + until: untilStr, + totalKeys: allKeys.length, + matchedKeys, + unmatchedKeys: allKeys.length - matchedKeys, + discrepancyCount: discrepancies.length, + summary, + discrepancies, + createdAt: now.toISOString(), }; } -export async function getReconciliationReport(_reportId?: string): Promise { - return null; +/** + * Get a persisted reconciliation report by ID. + */ +export async function getReconciliationReport( + reportId: string, + dbClient?: any, +): Promise { + const client = dbClient ?? db; + if (!client) return null; + try { + const reports = await client + .select() + .from(reconciliationReports) + .where(eq(reconciliationReports.id, reportId)) + .limit(1); + + if (reports.length === 0) return null; + const r = reports[0]; + + const discRows = await client + .select() + .from(reconciliationDiscrepancies) + .where(eq(reconciliationDiscrepancies.reportId, reportId)); + + return { + reportId: r.id, + fingerprint: r.fingerprint, + status: r.status as "ok" | "discrepancies_found", + since: r.since ? r.since.toISOString() : null, + until: r.until ? r.until.toISOString() : null, + totalKeys: r.totalKeys, + matchedKeys: r.matchedKeys, + unmatchedKeys: r.unmatchedKeys, + discrepancyCount: r.discrepancyCount, + summary: (r.summary ?? {}) as Record, + discrepancies: discRows.map((d: any) => ({ + kind: d.kind as PayoutDiscrepancyKind, + userId: d.userId ?? undefined, + stellarAddress: d.stellarAddress ?? undefined, + marketId: d.marketId, + predictionId: d.predictionId ?? undefined, + claimId: d.claimId ?? undefined, + txHash: d.txHash ?? undefined, + dbAmount: d.dbAmount ?? undefined, + onChainAmount: d.onChainAmount ?? undefined, + difference: d.difference ?? undefined, + details: (d.details ?? {}) as Record, + })), + createdAt: r.createdAt.toISOString(), + }; + } catch (err) { + logger.warn({ err, reportId }, "reconciliation.get_report_failed"); + return null; + } +} + +export interface ListReconciliationReportsOptions { + limit?: number; + cursor?: string; + since?: Date | string; + until?: Date | string; + dbClient?: any; } -export async function listReconciliationReports(_limit?: number, _offset?: number): Promise<[]> { - return []; +export interface ListReconciliationReportsResponse { + reports: Array>; + nextCursor: string | null; +} + +/** + * List persisted reconciliation reports with keyset pagination. + */ +export async function listReconciliationReports( + limitOrOpts: number | ListReconciliationReportsOptions = 10, + _offset?: number, + dbClient?: any, +): Promise { + const isObjectOpts = typeof limitOrOpts === "object"; + const opts: ListReconciliationReportsOptions = isObjectOpts + ? limitOrOpts + : { limit: limitOrOpts }; + + const client = opts.dbClient ?? dbClient ?? db; + const limit = Math.min(Math.max(opts.limit ?? 10, 1), 100); + + if (!client) { + return isObjectOpts ? { reports: [], nextCursor: null } : []; + } + + try { + const conditions = []; + if (opts.cursor) { + try { + const decoded = Buffer.from(opts.cursor, "base64").toString("utf-8"); + const [cursorTime, cursorId] = decoded.split("::"); + if (cursorTime && cursorId) { + const cursorDate = new Date(cursorTime); + conditions.push( + or( + lt(reconciliationReports.createdAt, cursorDate), + and( + eq(reconciliationReports.createdAt, cursorDate), + lt(reconciliationReports.id, cursorId), + ), + ), + ); + } + } catch { + // invalid cursor, ignore + } + } + + const rows = await client + .select() + .from(reconciliationReports) + .where(conditions.length > 0 ? and(...conditions) : undefined) + .orderBy(desc(reconciliationReports.createdAt), desc(reconciliationReports.id)) + .limit(limit + 1); + + const hasMore = rows.length > limit; + const pageRows = hasMore ? rows.slice(0, limit) : rows; + + let nextCursor: string | null = null; + if (hasMore && pageRows.length > 0) { + const last = pageRows[pageRows.length - 1]; + nextCursor = Buffer.from(`${last.createdAt.toISOString()}::${last.id}`).toString("base64"); + } + + const reports = pageRows.map((r: any) => ({ + reportId: r.id, + fingerprint: r.fingerprint, + status: r.status as "ok" | "discrepancies_found", + since: r.since ? r.since.toISOString() : null, + until: r.until ? r.until.toISOString() : null, + totalKeys: r.totalKeys, + matchedKeys: r.matchedKeys, + unmatchedKeys: r.unmatchedKeys, + discrepancyCount: r.discrepancyCount, + summary: (r.summary ?? {}) as Record, + createdAt: r.createdAt.toISOString(), + })); + + if (isObjectOpts) { + return { reports, nextCursor }; + } + return reports; + } catch (err) { + logger.warn({ err }, "reconciliation.list_reports_failed"); + return isObjectOpts ? { reports: [], nextCursor: null } : []; + } } diff --git a/src/workers/reconciliationWorker.ts b/src/workers/reconciliationWorker.ts index 6d877b3e..7bc55c43 100644 --- a/src/workers/reconciliationWorker.ts +++ b/src/workers/reconciliationWorker.ts @@ -1,3 +1,4 @@ +import { v4 as uuidv4 } from "uuid"; import { Worker, Job } from "bullmq"; import { performReconciliation, @@ -5,6 +6,8 @@ import { } from "../services/reconciliationService"; import { logger } from "../config/logger"; import { redisConnection, reconciliationQueueName } from "../queue"; +import { reconciliationDiscrepanciesTotal } from "../metrics/registry"; +import { createAuditLog } from "../services/auditService"; export interface ReconciliationJobPayload { type: "global" | "market"; @@ -42,9 +45,28 @@ export class ReconciliationWorker { if (data.type === "global") { const result = await performReconciliation(); logger.info( - { jobId: job.id }, + { jobId: job.id, status: result.status, discrepancyCount: result.discrepancyCount }, "reconciliation: global reconciliation completed", ); + + if (result.status === "discrepancies_found" && result.discrepancies?.length > 0) { + for (const d of result.discrepancies) { + reconciliationDiscrepanciesTotal.inc({ kind: d.kind }); + } + await createAuditLog({ + action: "reconciliation.discrepancies_found", + walletAddress: "system", + ip: "127.0.0.1", + correlationId: job.id ?? uuidv4(), + metadata: { + reportId: result.reportId, + fingerprint: result.fingerprint, + discrepancyCount: result.discrepancyCount, + summary: result.summary, + }, + }); + } + return result; } else if (data.type === "market") { if ( diff --git a/tests/adminReconciliation.test.ts b/tests/adminReconciliation.test.ts index 40e72854..b5b6ddda 100644 --- a/tests/adminReconciliation.test.ts +++ b/tests/adminReconciliation.test.ts @@ -6,13 +6,25 @@ import { errorHandler } from "../src/middleware/errorHandler"; jest.mock("../src/services/reconciliationService", () => ({ reconcileMarket: jest.fn(), + getReconciliationReport: jest.fn(), + listReconciliationReports: jest.fn(), })); -import { reconcileMarket } from "../src/services/reconciliationService"; +import { + reconcileMarket, + getReconciliationReport, + listReconciliationReports, +} from "../src/services/reconciliationService"; const mockReconcileMarket = reconcileMarket as jest.MockedFunction< typeof reconcileMarket >; +const mockGetReconciliationReport = getReconciliationReport as jest.MockedFunction< + typeof getReconciliationReport +>; +const mockListReconciliationReports = listReconciliationReports as jest.MockedFunction< + typeof listReconciliationReports +>; const SECRET = process.env.JWT_SECRET!; const ISSUER = process.env.JWT_ISSUER ?? "predictify"; @@ -320,3 +332,92 @@ describe("GET /api/admin/recon/markets/:id", () => { expect(mockReconcileMarket).not.toHaveBeenCalled(); }); }); + +describe("GET /api/admin/recon/reports", () => { + it("returns 403 when not authenticated or not admin", async () => { + const res = await request(makeApp()).get("/api/admin/recon/reports"); + expect(res.status).toBe(403); + }); + + it("returns paginated list of reports for admin", async () => { + mockListReconciliationReports.mockResolvedValue({ + reports: [ + { + reportId: "rep-1", + fingerprint: "fp-1", + status: "ok", + since: null, + until: null, + totalKeys: 10, + matchedKeys: 10, + unmatchedKeys: 0, + discrepancyCount: 0, + summary: {}, + createdAt: "2026-06-27T12:00:00.000Z", + }, + ], + nextCursor: "next-cursor", + } as any); + + const res = await request(makeApp()) + .get("/api/admin/recon/reports?limit=10") + .set("Authorization", `Bearer ${signJwt({ sub: ADMIN_ADDRESS, role: "admin" })}`); + + expect(res.status).toBe(200); + expect(res.body.data).toHaveLength(1); + expect(res.body.meta.nextCursor).toBe("next-cursor"); + expect(res.body.meta.limit).toBe(10); + }); + + it("returns 422/400 on invalid query params", async () => { + const res = await request(makeApp()) + .get("/api/admin/recon/reports?limit=invalid") + .set("Authorization", `Bearer ${signJwt({ sub: ADMIN_ADDRESS, role: "admin" })}`); + + expect([400, 422]).toContain(res.status); + }); +}); + +describe("GET /api/admin/recon/reports/:id", () => { + it("returns 403 when not authenticated or not admin", async () => { + const res = await request(makeApp()).get("/api/admin/recon/reports/rep-1"); + expect(res.status).toBe(403); + }); + + it("returns 404 when report not found", async () => { + mockGetReconciliationReport.mockResolvedValue(null); + + const res = await request(makeApp()) + .get("/api/admin/recon/reports/missing-report") + .set("Authorization", `Bearer ${signJwt({ sub: ADMIN_ADDRESS, role: "admin" })}`); + + expect(res.status).toBe(404); + }); + + it("returns report detail for admin", async () => { + const mockDetail = { + reportId: "rep-1", + fingerprint: "fp-1", + status: "ok", + since: null, + until: null, + totalKeys: 10, + matchedKeys: 10, + unmatchedKeys: 0, + discrepancyCount: 0, + summary: {}, + discrepancies: [], + createdAt: "2026-06-27T12:00:00.000Z", + }; + mockGetReconciliationReport.mockResolvedValue(mockDetail as any); + + const res = await request(makeApp()) + .get("/api/admin/recon/reports/rep-1") + .set("Authorization", `Bearer ${signJwt({ sub: ADMIN_ADDRESS, role: "admin" })}`); + + expect(res.status).toBe(200); + expect(res.body.data.reportId).toBe("rep-1"); + expect(res.body.data.discrepancies).toEqual([]); + }); +}); + diff --git a/tests/reconciliationPayouts.test.ts b/tests/reconciliationPayouts.test.ts new file mode 100644 index 00000000..69fbda32 --- /dev/null +++ b/tests/reconciliationPayouts.test.ts @@ -0,0 +1,1005 @@ +/* eslint-disable @typescript-eslint/no-explicit-any */ +import express from "express"; +import jwt from "jsonwebtoken"; +import request from "supertest"; +import { + performReconciliation, + ClaimEventSource, + ClaimEvent, + PredictionRecord, + ClaimRecord, +} from "../src/services/reconciliationService"; +import { createAdminReconciliationRouter } from "../src/routes/admin/reconciliation"; +import { errorHandler } from "../src/middleware/errorHandler"; +import { ReconciliationWorker } from "../src/workers/reconciliationWorker"; +import { reconciliationDiscrepanciesTotal } from "../src/metrics/registry"; +import { createAuditLog } from "../src/services/auditService"; + +jest.mock("ioredis", () => { + return jest.fn().mockImplementation(() => ({ + on: jest.fn(), + })); +}); + +let workerCallback: any = null; +jest.mock("bullmq", () => { + return { + Queue: jest.fn().mockImplementation((name) => ({ name })), + Worker: jest.fn().mockImplementation((_name, cb) => { + workerCallback = cb; + return { + on: jest.fn(), + close: jest.fn(), + }; + }), + QueueEvents: jest.fn(), + }; +}); + +jest.mock("../src/services/auditService", () => ({ + createAuditLog: jest.fn().mockResolvedValue({ id: "audit-1" }), +})); + +jest.mock("../src/metrics/registry", () => ({ + reconciliationDiscrepanciesTotal: { + inc: jest.fn(), + }, +})); + +const SECRET = process.env.JWT_SECRET || "predictify-test-jwt-secret-key-32-chars-long"; +const ISSUER = process.env.JWT_ISSUER ?? "predictify"; +const AUDIENCE = process.env.JWT_AUDIENCE ?? "predictify-app"; +const ADMIN_ADDRESS = "GADMIN7777777777777777777777777777777777777777777777777777"; +const USER_ADDRESS = "GUSER88888888888888888888888888888888888888888888888888888"; + +function signJwt(payload: object): string { + return jwt.sign(payload, SECRET, { + issuer: ISSUER, + audience: AUDIENCE, + expiresIn: "1h", + }); +} + +function makeApp(): express.Express { + const app = express(); + app.use(express.json()); + app.use((req, _res, next) => { + (req as express.Request & { id?: string }).id = + (req.headers["x-request-id"] as string | undefined) ?? "test-request-id"; + next(); + }); + app.use("/api/admin/recon", createAdminReconciliationRouter()); + app.use(errorHandler); + return app; +} + +class FakeClaimEventSource implements ClaimEventSource { + constructor(private readonly events: ClaimEvent[]) {} + + async getClaimEvents(): Promise { + return this.events; + } +} + +describe("Payout Reconciliation Comprehensive Test Suite (Issue #980 Test Plan)", () => { + beforeEach(() => { + jest.clearAllMocks(); + }); + + // ── Scenario 1: Every source agrees ─────────────────────────────────────── + it("Scenario 1: Every source agrees -> zero discrepancies; report status ok", async () => { + const user = "GUSER1111111111111111111111111111111111111111111111111111"; + const market = "market-1"; + const amount = "10.0000000"; + + const predictionsFixture: PredictionRecord[] = [ + { + id: "pred-1", + userId: "user-1", + stellarAddress: user, + marketId: market, + amount, + status: "confirmed", + result: "won", + claimTxHash: "tx-clm-1", + createdAt: new Date(), + }, + ]; + + const claimsFixture: ClaimRecord[] = [ + { + id: "clm-1", + userId: "user-1", + stellarAddress: user, + marketId: market, + amount, + status: "paid", + settlementTx: "tx-clm-1", + settledAt: new Date(), + createdAt: new Date(), + }, + ]; + + const eventsFixture: ClaimEvent[] = [ + { + eventId: "evt-1", + ledger: 100, + txHash: "tx-clm-1", + eventType: "win_clm", + marketId: market, + userAddress: user, + amount, + timestamp: new Date(), + }, + ]; + + const result = await performReconciliation({ + predictionsFixture, + claimsFixture, + eventsFixture, + }); + + expect(result.status).toBe("ok"); + expect(result.discrepancyCount).toBe(0); + expect(result.discrepancies).toEqual([]); + expect(result.fingerprint).toBeDefined(); + expect(typeof result.fingerprint).toBe("string"); + }); + + // ── Scenario 2: win_clm event exists with no claims row ─────────────────── + it("Scenario 2: A win_clm event exists with no claims row -> claimed_on_chain_missing_in_db", async () => { + const user = "GUSER2222222222222222222222222222222222222222222222222222"; + const market = "market-2"; + + const fakeSource = new FakeClaimEventSource([ + { + eventId: "evt-2", + ledger: 102, + txHash: "tx-evt-2", + eventType: "win_clm", + marketId: market, + userAddress: user, + amount: "5.0000000", + timestamp: new Date(), + }, + ]); + + const result = await performReconciliation({ + predictionsFixture: [], + claimsFixture: [], + claimEventSource: fakeSource, + }); + + expect(result.status).toBe("discrepancies_found"); + expect(result.discrepancyCount).toBe(1); + expect(result.discrepancies[0].kind).toBe("claimed_on_chain_missing_in_db"); + expect(result.discrepancies[0].marketId).toBe(market); + expect(result.discrepancies[0].stellarAddress).toBe(user); + expect(result.discrepancies[0].onChainAmount).toBe("5.0000000"); + }); + + // ── Scenario 3: claims row is paid with no matching event ───────────────── + it("Scenario 3: A claims row is paid with no matching event -> db_paid_missing_on_chain", async () => { + const user = "GUSER3333333333333333333333333333333333333333333333333333"; + const market = "market-3"; + + const claimsFixture: ClaimRecord[] = [ + { + id: "clm-3", + userId: "user-3", + stellarAddress: user, + marketId: market, + amount: "15.0000000", + status: "paid", + settlementTx: "tx-clm-3", + settledAt: new Date(), + createdAt: new Date(), + }, + ]; + + const result = await performReconciliation({ + predictionsFixture: [], + claimsFixture, + eventsFixture: [], + }); + + expect(result.status).toBe("discrepancies_found"); + expect(result.discrepancyCount).toBe(1); + expect(result.discrepancies[0].kind).toBe("db_paid_missing_on_chain"); + expect(result.discrepancies[0].marketId).toBe(market); + expect(result.discrepancies[0].stellarAddress).toBe(user); + expect(result.discrepancies[0].dbAmount).toBe("15.0000000"); + }); + + // ── Scenario 4: Event amount differs by 1 stroop ─────────────────────────── + it("Scenario 4: Event amount differs from claim amount by 1 stroop -> amount_mismatch with exact values", async () => { + const user = "GUSER4444444444444444444444444444444444444444444444444444"; + const market = "market-4"; + + const claimsFixture: ClaimRecord[] = [ + { + id: "clm-4", + userId: "user-4", + stellarAddress: user, + marketId: market, + amount: "10.0000000", + status: "paid", + settlementTx: "tx-clm-4", + settledAt: new Date(), + createdAt: new Date(), + }, + ]; + + const eventsFixture: ClaimEvent[] = [ + { + eventId: "evt-4", + ledger: 104, + txHash: "tx-evt-4", + eventType: "win_clm", + marketId: market, + userAddress: user, + amount: "10.0000001", // 1 stroop higher + timestamp: new Date(), + }, + ]; + + const result = await performReconciliation({ + predictionsFixture: [], + claimsFixture, + eventsFixture, + }); + + expect(result.status).toBe("discrepancies_found"); + const d = result.discrepancies.find((x) => x.kind === "amount_mismatch"); + expect(d).toBeDefined(); + expect(d?.dbAmount).toBe("10.0000000"); + expect(d?.onChainAmount).toBe("10.0000001"); + expect(d?.difference).toBe("-0.0000001"); + }); + + // ── Scenario 5: Claim event for prediction with result = 'lost' ─────────── + it("Scenario 5: A claim event for a prediction with result = 'lost' -> claim_for_losing_prediction", async () => { + const user = "GUSER5555555555555555555555555555555555555555555555555555"; + const market = "market-5"; + + const predictionsFixture: PredictionRecord[] = [ + { + id: "pred-5", + userId: "user-5", + stellarAddress: user, + marketId: market, + amount: "25.0000000", + status: "confirmed", + result: "lost", + claimTxHash: null, + createdAt: new Date(), + }, + ]; + + const eventsFixture: ClaimEvent[] = [ + { + eventId: "evt-5", + ledger: 105, + txHash: "tx-evt-5", + eventType: "win_clm", + marketId: market, + userAddress: user, + amount: "25.0000000", + timestamp: new Date(), + }, + ]; + + const result = await performReconciliation({ + predictionsFixture, + claimsFixture: [], + eventsFixture, + }); + + expect(result.status).toBe("discrepancies_found"); + const d = result.discrepancies.find((x) => x.kind === "claim_for_losing_prediction"); + expect(d).toBeDefined(); + expect(d?.marketId).toBe(market); + expect(d?.stellarAddress).toBe(user); + expect(d?.predictionId).toBe("pred-5"); + }); + + // ── Scenario 6: Two win_clm events for the same (user, market) ─────────── + it("Scenario 6: Two win_clm events for the same (user, market) -> duplicate_on_chain_claim", async () => { + const user = "GUSER6666666666666666666666666666666666666666666666666666"; + const market = "market-6"; + + const eventsFixture: ClaimEvent[] = [ + { + eventId: "evt-6a", + ledger: 106, + txHash: "tx-6a", + eventType: "win_clm", + marketId: market, + userAddress: user, + amount: "10.0000000", + timestamp: new Date(), + }, + { + eventId: "evt-6b", + ledger: 107, + txHash: "tx-6b", + eventType: "win_clm", + marketId: market, + userAddress: user, + amount: "10.0000000", + timestamp: new Date(), + }, + ]; + + const result = await performReconciliation({ + predictionsFixture: [], + claimsFixture: [], + eventsFixture, + }); + + expect(result.status).toBe("discrepancies_found"); + const d = result.discrepancies.find((x) => x.kind === "duplicate_on_chain_claim"); + expect(d).toBeDefined(); + expect(d?.marketId).toBe(market); + expect(d?.stellarAddress).toBe(user); + }); + + // ── Scenario 7: Won and unclaimed longer than threshold ─────────────────── + it("Scenario 7: Won and unclaimed longer than threshold -> stale_unclaimed", async () => { + const user = "GUSER7777777777777777777777777777777777777777777777777777"; + const market = "market-7"; + const oldDate = new Date(Date.now() - 40 * 24 * 60 * 60 * 1000); // 40 days ago + + const predictionsFixture: PredictionRecord[] = [ + { + id: "pred-7", + userId: "user-7", + stellarAddress: user, + marketId: market, + amount: "50.0000000", + status: "confirmed", + result: "won", + claimTxHash: null, + createdAt: oldDate, + }, + ]; + + const result = await performReconciliation({ + predictionsFixture, + claimsFixture: [], + eventsFixture: [], + staleThresholdDays: 30, + }); + + expect(result.status).toBe("discrepancies_found"); + const d = result.discrepancies.find((x) => x.kind === "stale_unclaimed"); + expect(d).toBeDefined(); + expect(d?.marketId).toBe(market); + expect(d?.stellarAddress).toBe(user); + expect(d?.predictionId).toBe("pred-7"); + expect(d?.dbAmount).toBe("50.0000000"); + }); + + // ── Scenario 8: Idempotent runs and deterministic fingerprints ──────────── + it("Scenario 8: Same run twice produces identical fingerprint and no duplicate discrepancies", async () => { + const user = "GUSER8888888888888888888888888888888888888888888888888888"; + const market = "market-8"; + + const eventsFixture: ClaimEvent[] = [ + { + eventId: "evt-8", + ledger: 108, + txHash: "tx-evt-8", + eventType: "win_clm", + marketId: market, + userAddress: user, + amount: "8.0000000", + timestamp: new Date("2026-06-01T00:00:00.000Z"), + }, + ]; + + const run1 = await performReconciliation({ + predictionsFixture: [], + claimsFixture: [], + eventsFixture, + }); + + const run2 = await performReconciliation({ + predictionsFixture: [], + claimsFixture: [], + eventsFixture, + }); + + expect(run1.fingerprint).toBe(run2.fingerprint); + expect(run1.discrepancyCount).toBe(run2.discrepancyCount); + expect(run1.discrepancies).toEqual(run2.discrepancies); + }); + + // ── Scenario 9: Admin endpoints without admin token -> 401/403 ──────────── + it("Scenario 9: Admin endpoints without admin token return 401/403", async () => { + const app = makeApp(); + + const noAuthReports = await request(app).get("/api/admin/recon/reports"); + expect([401, 403]).toContain(noAuthReports.status); + + const userAuthReports = await request(app) + .get("/api/admin/recon/reports") + .set("Authorization", `Bearer ${signJwt({ sub: USER_ADDRESS, role: "user" })}`); + expect(userAuthReports.status).toBe(403); + + const noAuthSingle = await request(app).get("/api/admin/recon/reports/rep-123"); + expect([401, 403]).toContain(noAuthSingle.status); + + const userAuthSingle = await request(app) + .get("/api/admin/recon/reports/rep-123") + .set("Authorization", `Bearer ${signJwt({ sub: USER_ADDRESS, role: "user" })}`); + expect(userAuthSingle.status).toBe(403); + }); + + // ── Scenario 10: 50,000 keys test completes within memory bounds (< 100 MB) + it("Scenario 10: 50,000 keys benchmark completes with heap growth < 100 MB", async () => { + const count = 50_000; + const predictionsFixture: PredictionRecord[] = []; + const claimsFixture: ClaimRecord[] = []; + const eventsFixture: ClaimEvent[] = []; + + for (let i = 0; i < count; i++) { + const addr = `GUSER${i.toString().padStart(50, "0")}`; + const mkt = `market-${i % 500}`; + const amt = "10.0000000"; + + predictionsFixture.push({ + id: `p-${i}`, + userId: `u-${i}`, + stellarAddress: addr, + marketId: mkt, + amount: amt, + status: "confirmed", + result: "won", + claimTxHash: `tx-${i}`, + createdAt: new Date(), + }); + + claimsFixture.push({ + id: `c-${i}`, + userId: `u-${i}`, + stellarAddress: addr, + marketId: mkt, + amount: amt, + status: "paid", + settlementTx: `tx-${i}`, + settledAt: new Date(), + createdAt: new Date(), + }); + + eventsFixture.push({ + eventId: `e-${i}`, + ledger: 1000 + i, + txHash: `tx-${i}`, + eventType: "win_clm", + marketId: mkt, + userAddress: addr, + amount: amt, + timestamp: new Date(), + }); + } + + if (global.gc) { + global.gc(); + } + const initialHeap = process.memoryUsage().heapUsed; + + const result = await performReconciliation({ + predictionsFixture, + claimsFixture, + eventsFixture, + }); + + expect(result.status).toBe("ok"); + expect(result.totalKeys).toBe(count); + expect(result.discrepancyCount).toBe(0); + + const finalHeap = process.memoryUsage().heapUsed; + const heapGrowthBytes = finalHeap - initialHeap; + const maxAllowedBytes = 100 * 1024 * 1024; // 100 MB + + expect(heapGrowthBytes).toBeLessThan(maxAllowedBytes); + }, 30000); + + // ── Scenario 11: Worker job with discrepancies ──────────────────────────── + it("Scenario 11: Worker job with discrepancies -> status discrepancies_found, metric incremented, audit row written", async () => { + const worker = new ReconciliationWorker(1); + worker.start(); + + expect(workerCallback).toBeDefined(); + + const mockReport = { + reportId: "rep-disc-11", + fingerprint: "fp-disc-11", + status: "discrepancies_found" as const, + totalKeys: 5, + matchedKeys: 4, + unmatchedKeys: 1, + discrepancyCount: 1, + summary: { claimed_on_chain_missing_in_db: 1 }, + discrepancies: [ + { + kind: "claimed_on_chain_missing_in_db" as const, + marketId: "market-11", + stellarAddress: "GUSER11", + }, + ], + createdAt: new Date().toISOString(), + }; + + const spy = jest + .spyOn(require("../src/services/reconciliationService"), "performReconciliation") + .mockResolvedValueOnce(mockReport as any); + + // Simulate worker execution with the callback + const result = await workerCallback({ + id: "job-11", + data: { type: "global" }, + }); + + expect(result.status).toBe("discrepancies_found"); + expect(reconciliationDiscrepanciesTotal.inc).toHaveBeenCalledWith({ + kind: "claimed_on_chain_missing_in_db", + }); + expect(createAuditLog).toHaveBeenCalledWith( + expect.objectContaining({ + action: "reconciliation.discrepancies_found", + metadata: expect.objectContaining({ + reportId: "rep-disc-11", + discrepancyCount: 1, + }), + }), + ); + + spy.mockRestore(); + await worker.stop(); + }); + + // ── Database persistence and retrieval methods ──────────────────────────── + describe("Database persistence, DefaultDbClaimEventSource and report retrieval", () => { + it("DefaultDbClaimEventSource queries indexerEvents correctly", async () => { + const mockRows = [ + { + id: "evt-db-1", + txHash: "tx-db-1", + ledger: 101, + marketId: "mkt-1", + eventType: "win_clm", + canonical: true, + data: { + userAddress: "GUSERDB1", + amount: "10.0000000", + }, + createdAt: new Date(), + }, + ]; + + const mockDb = { + select: () => ({ + from: () => ({ + where: () => Promise.resolve(mockRows), + }), + }), + }; + + const source = new (require("../src/services/reconciliationService").DefaultDbClaimEventSource)(mockDb); + const events = await source.getClaimEvents({ + since: new Date(Date.now() - 3600000), + until: new Date(), + }); + + expect(events).toHaveLength(1); + expect(events[0].txHash).toBe("tx-db-1"); + expect(events[0].userAddress).toBe("GUSERDB1"); + expect(events[0].amount).toBe("10.0000000"); + }); + + it("DefaultDbClaimEventSource handles errors gracefully", async () => { + const mockDb = { + select: () => ({ + from: () => ({ + where: () => Promise.reject(new Error("db failure")), + }), + }), + }; + + const source = new (require("../src/services/reconciliationService").DefaultDbClaimEventSource)(mockDb); + const events = await source.getClaimEvents(); + expect(events).toEqual([]); + }); + + it("persists reconciliation report and discrepancies when dbClient is provided", async () => { + let insertedReport: any = null; + let insertedDiscrepancies: any = null; + + const mockDb = { + select: () => ({ + from: () => ({ + where: () => ({ + limit: () => Promise.resolve([]), // no existing + }), + }), + }), + insert: (table: any) => ({ + values: (vals: any) => ({ + onConflictDoNothing: () => ({ + returning: () => { + insertedReport = vals; + return Promise.resolve([vals]); + }, + }), + then: (resolve: any) => { + insertedDiscrepancies = vals; + return Promise.resolve(vals).then(resolve); + }, + }), + }), + }; + + const result = await performReconciliation({ + predictionsFixture: [], + claimsFixture: [], + eventsFixture: [ + { + eventId: "e-persist-1", + ledger: 200, + txHash: "tx-persist-1", + eventType: "win_clm", + marketId: "mkt-p", + userAddress: "GUSER_P", + amount: "5.0000000", + timestamp: new Date(), + }, + ], + persist: true, + dbClient: mockDb, + }); + + expect(result.status).toBe("discrepancies_found"); + expect(insertedReport).toBeDefined(); + expect(insertedReport.id).toBe(result.reportId); + expect(insertedDiscrepancies).toBeDefined(); + expect(insertedDiscrepancies).toHaveLength(1); + }); + + it("getReconciliationReport retrieves report and discrepancies from dbClient", async () => { + const { getReconciliationReport } = require("../src/services/reconciliationService"); + + const mockReportRow = { + id: "rep-get-1", + fingerprint: "fp-get-1", + status: "ok", + since: new Date("2026-06-01T00:00:00.000Z"), + until: new Date("2026-06-02T00:00:00.000Z"), + totalKeys: 5, + matchedKeys: 5, + unmatchedKeys: 0, + discrepancyCount: 0, + summary: {}, + createdAt: new Date("2026-06-02T01:00:00.000Z"), + }; + + const mockDb = { + select: () => ({ + from: (table: any) => ({ + where: () => { + const p: any = Promise.resolve([]); + p.limit = () => Promise.resolve([mockReportRow]); + return p; + }, + }), + }), + }; + + const report = await getReconciliationReport("rep-get-1", mockDb); + expect(report).toBeDefined(); + expect(report?.reportId).toBe("rep-get-1"); + expect(report?.status).toBe("ok"); + }); + + it("getReconciliationReport returns null when report not found", async () => { + const { getReconciliationReport } = require("../src/services/reconciliationService"); + const mockDb = { + select: () => ({ + from: () => ({ + where: () => ({ + limit: () => Promise.resolve([]), + }), + }), + }), + }; + + const report = await getReconciliationReport("non-existent", mockDb); + expect(report).toBeNull(); + }); + + it("listReconciliationReports retrieves paginated reports with keyset cursor", async () => { + const { listReconciliationReports } = require("../src/services/reconciliationService"); + + const mockRows = [ + { + id: "rep-list-1", + fingerprint: "fp-list-1", + status: "ok", + since: null, + until: null, + totalKeys: 1, + matchedKeys: 1, + unmatchedKeys: 0, + discrepancyCount: 0, + summary: {}, + createdAt: new Date("2026-06-01T00:00:00.000Z"), + }, + ]; + + const mockDb = { + select: () => ({ + from: () => ({ + where: () => ({ + orderBy: () => ({ + limit: () => Promise.resolve(mockRows), + }), + }), + orderBy: () => ({ + limit: () => Promise.resolve(mockRows), + }), + }), + }), + }; + + const res = await listReconciliationReports({ + limit: 10, + cursor: Buffer.from("2026-06-01T00:00:00.000Z::rep-0").toString("base64"), + dbClient: mockDb, + }); + + expect(res.reports).toHaveLength(1); + expect(res.reports[0].reportId).toBe("rep-list-1"); + }); + + it("listReconciliationReports handles hasMore and array responses", async () => { + const { listReconciliationReports } = require("../src/services/reconciliationService"); + + const mockTwoRows = [ + { id: "r1", createdAt: new Date() }, + { id: "r2", createdAt: new Date() }, + ]; + const mockDbTwo = { + select: () => ({ + from: () => ({ + where: () => ({ + orderBy: () => ({ + limit: () => Promise.resolve(mockTwoRows), + }), + }), + orderBy: () => ({ + limit: () => Promise.resolve(mockTwoRows), + }), + }), + }), + }; + + const listHasMore = await listReconciliationReports({ limit: 1, dbClient: mockDbTwo }); + expect(listHasMore.nextCursor).toBeDefined(); + + const listNum = await listReconciliationReports(1, 0, mockDbTwo); + expect(Array.isArray(listNum)).toBe(true); + + const listFail = await listReconciliationReports({ + dbClient: { + select: () => ({ + from: () => ({ + where: () => ({ + orderBy: () => ({ + limit: () => Promise.reject(new Error("fail")), + }), + }), + }), + }), + }, + }); + expect(listFail.reports).toEqual([]); + }); + + it("getReconciliationReport maps discrepancies rows correctly", async () => { + const { getReconciliationReport } = require("../src/services/reconciliationService"); + + const mockReportWithDisc = { + id: "rep-disc-get", + fingerprint: "fp-disc-get", + status: "discrepancies_found", + since: new Date(), + until: new Date(), + totalKeys: 1, + matchedKeys: 0, + unmatchedKeys: 1, + discrepancyCount: 1, + summary: { claimed_on_chain_missing_in_db: 1 }, + createdAt: new Date(), + }; + const mockDiscRow = { + reportId: "rep-disc-get", + kind: "claimed_on_chain_missing_in_db", + userId: "u-1", + stellarAddress: "GUSER1", + marketId: "m-1", + predictionId: "p-1", + claimId: "c-1", + txHash: "tx-1", + dbAmount: "10.0000000", + onChainAmount: "10.0000000", + difference: "0.0000000", + details: { foo: "bar" }, + }; + const mockDbDisc = { + select: () => ({ + from: () => ({ + where: () => { + const p: any = Promise.resolve([mockDiscRow]); + p.limit = () => Promise.resolve([mockReportWithDisc]); + return p; + }, + }), + }), + }; + const repWithDisc = await getReconciliationReport("rep-disc-get", mockDbDisc); + expect(repWithDisc?.discrepancies).toHaveLength(1); + expect(repWithDisc?.discrepancies[0].kind).toBe("claimed_on_chain_missing_in_db"); + }); + + it("performReconciliation queries dbClient when fixtures are omitted and handles existing report", async () => { + const mockDbAll = { + select: (fields?: any) => ({ + from: (table: any) => ({ + innerJoin: () => ({ + where: () => + Promise.resolve([ + { + id: "p-1", + userId: "u-1", + stellarAddress: "GUSER1", + marketId: "m-1", + amount: "10.0000000", + status: "confirmed", + result: "won", + claimTxHash: "tx-1", + createdAt: new Date(), + }, + ]), + }), + where: () => ({ + limit: () => + Promise.resolve([ + { id: "existing-report-id" }, // existing report + ]), + }), + }), + }), + }; + + const resDb = await performReconciliation({ + since: new Date(Date.now() - 3600000), + until: new Date(), + persist: true, + dbClient: mockDbAll, + }); + + expect(resDb.reportId).toBe("existing-report-id"); + }); + + it("handles persistence errors, get errors, and null dbClient gracefully", async () => { + const { + getReconciliationReport, + listReconciliationReports, + } = require("../src/services/reconciliationService"); + + // getReconciliationReport when db throws + const failDb = { + select: () => ({ + from: () => ({ + where: () => { + throw new Error("unexpected error"); + }, + }), + }), + }; + const nullReport = await getReconciliationReport("rep-err", failDb); + expect(nullReport).toBeNull(); + + // getReconciliationReport with null + const nullClientReport = await getReconciliationReport("rep-err", null); + expect(nullClientReport).toBeNull(); + + // listReconciliationReports with null + const nullList = await listReconciliationReports({ dbClient: null }); + expect(nullList).toEqual({ reports: [], nextCursor: null }); + + const nullListNum = await listReconciliationReports(10, 0, null); + expect(nullListNum).toEqual([]); + + // performReconciliation when db query throws + const errDb = { + select: () => { + throw new Error("db fail"); + }, + }; + const resErr = await performReconciliation({ dbClient: errDb }); + expect(resErr.status).toBe("ok"); + + // performReconciliation when persist throws + const persistFailDb = { + select: () => ({ + from: () => ({ + where: () => { + throw new Error("persist fail"); + }, + }), + }), + }; + const repFail = await performReconciliation({ + predictionsFixture: [], + claimsFixture: [], + eventsFixture: [], + persist: true, + dbClient: persistFailDb, + }); + expect(repFail.status).toBe("ok"); + }); + }); + + // ── Exact arithmetic & market reconciliation edge tests ─────────────────── + describe("Exact arithmetic & helper coverage", () => { + it("tests parseExactStroops and formatExactStroops edge cases", () => { + const { + parseExactStroops, + formatExactStroops, + } = require("../src/services/reconciliationService"); + + expect(() => parseExactStroops("invalid")).toThrow(); + expect(parseExactStroops("10.5")).toBe(105000000n); + expect(parseExactStroops("-5.0")).toBe(-50000000n); + expect(parseExactStroops("")).toBe(0n); + expect(parseExactStroops(null)).toBe(0n); + expect(formatExactStroops(0n)).toBe("0"); + expect(formatExactStroops(-10000000n)).toBe("-1"); + expect(formatExactStroops(-1n)).toBe("-0.0000001"); + expect(formatExactStroops(15000000n)).toBe("1.5"); + }); + + it("tests DefaultOnChainMarketStateProvider and createReconciliationService", async () => { + const { + DefaultOnChainMarketStateProvider, + createReconciliationService, + } = require("../src/services/reconciliationService"); + + const provider = new DefaultOnChainMarketStateProvider(); + const pos = await provider.getMarketPositions("market-test"); + expect(pos.available).toBe(false); + + const service = createReconciliationService({ + marketExists: jest.fn().mockResolvedValue(true), + getDbPositions: jest.fn().mockResolvedValue([]), + getOnChainPositions: jest.fn().mockResolvedValue({ + positions: [], + source: "soroban-rpc", + available: false, + unavailableReason: null, + }), + writeAudit: jest.fn().mockResolvedValue("corr-exp"), + }); + + const res = await service.reconcileMarket({ + marketId: "market-1", + adminAddress: ADMIN_ADDRESS, + ip: "127.0.0.1", + correlationId: "corr-exp", + }); + expect(res.status).toBe("partial"); + }); + }); +}); diff --git a/tests/reconciliationWorker.test.ts b/tests/reconciliationWorker.test.ts index 50a995ba..8ad92d6d 100644 --- a/tests/reconciliationWorker.test.ts +++ b/tests/reconciliationWorker.test.ts @@ -23,9 +23,19 @@ jest.mock("../src/services/reconciliationService", () => ({ performReconciliation: jest.fn(), reconcileMarket: jest.fn(), })); +jest.mock("../src/services/auditService", () => ({ + createAuditLog: jest.fn().mockResolvedValue({ id: "audit-1" }), +})); +jest.mock("../src/metrics/registry", () => ({ + reconciliationDiscrepanciesTotal: { + inc: jest.fn(), + }, +})); import { ReconciliationWorker, ReconciliationJobPayload } from "../src/workers/reconciliationWorker"; import { performReconciliation, reconcileMarket } from "../src/services/reconciliationService"; +import { createAuditLog } from "../src/services/auditService"; +import { reconciliationDiscrepanciesTotal } from "../src/metrics/registry"; import { Worker } from "bullmq"; describe("ReconciliationWorker", () => { @@ -61,6 +71,48 @@ describe("ReconciliationWorker", () => { expect(result).toEqual({ skipped: true }); }); + it("completes global job with status discrepancies_found, increments metrics and creates audit row", async () => { + const worker = new ReconciliationWorker(1); + worker.start(); + + const mockReport = { + reportId: "rep-1", + fingerprint: "fp-1", + status: "discrepancies_found", + totalKeys: 2, + matchedKeys: 1, + unmatchedKeys: 1, + discrepancyCount: 1, + summary: { claimed_on_chain_missing_in_db: 1 }, + discrepancies: [ + { + kind: "claimed_on_chain_missing_in_db", + marketId: "market-1", + stellarAddress: "user-1", + }, + ], + }; + (performReconciliation as jest.Mock).mockResolvedValue(mockReport); + + expect(workerCallback).toBeDefined(); + const result = await workerCallback({ id: "job-global-disc", data: { type: "global" } }); + expect(performReconciliation).toHaveBeenCalledTimes(1); + expect(result.status).toBe("discrepancies_found"); + expect(reconciliationDiscrepanciesTotal.inc).toHaveBeenCalledWith({ + kind: "claimed_on_chain_missing_in_db", + }); + expect(createAuditLog).toHaveBeenCalledWith( + expect.objectContaining({ + action: "reconciliation.discrepancies_found", + metadata: expect.objectContaining({ + reportId: "rep-1", + fingerprint: "fp-1", + discrepancyCount: 1, + }), + }), + ); + }); + it("runs market reconciliation when job type is market", async () => { const worker = new ReconciliationWorker(1); worker.start();