From 684362b12f22ca077c0c0569af65c90dc6d96e1e Mon Sep 17 00:00:00 2001 From: Hussman256 <177764730+Hussman256@users.noreply.github.com> Date: Wed, 21 Oct 2026 18:41:52 +0100 Subject: [PATCH] fix(payout-hop): answer with the hash and repair bookkeeping when post-send writes fail (#49) After the anchor was paid, a failure in the sent update, exit_fill insert or journal pair surfaced as 502. The caller saw a failure, a retry got 409, and the payout stayed paid but unbooked with nothing to reconcile it. - The journal ref_id (quoteId, else a fresh uuid) and quote_id are stored on the payout_forward claim before sending (migration 0010), so the booking id is fixed and a repair reuses it. - Marking the row sent and booking exit_fill + the journal pair share one transaction, so sent always means booked. - If that transaction fails, the row becomes bookkeeping_pending, a payout_unbooked alert fires, and the response is still 200 with the hash and booked: false. - bookPendingPayouts books pending rows idempotently (conditional update per row; a row whose journal pair already exists is only marked sent). It runs at startup and every PAYOUT_HOP_REPAIR_INTERVAL_S. --- .../0010_payout_forward_booking.sql | 2 + .../db/migrations/meta/0010_snapshot.json | 2322 +++++++++++++++++ packages/db/migrations/meta/_journal.json | 7 + packages/db/src/schema.ts | 10 +- packages/observability/src/metrics.ts | 3 +- services/payout-hop/README.md | 44 +- services/payout-hop/src/hop.test.ts | 353 ++- services/payout-hop/src/hop.ts | 155 +- services/payout-hop/src/index.ts | 24 +- services/payout-hop/src/repair.ts | 78 + 10 files changed, 2869 insertions(+), 129 deletions(-) create mode 100644 packages/db/migrations/0010_payout_forward_booking.sql create mode 100644 packages/db/migrations/meta/0010_snapshot.json create mode 100644 services/payout-hop/src/repair.ts diff --git a/packages/db/migrations/0010_payout_forward_booking.sql b/packages/db/migrations/0010_payout_forward_booking.sql new file mode 100644 index 0000000..e0b4094 --- /dev/null +++ b/packages/db/migrations/0010_payout_forward_booking.sql @@ -0,0 +1,2 @@ +ALTER TABLE "payout_forward" ADD COLUMN "ref_id" uuid DEFAULT gen_random_uuid() NOT NULL;--> statement-breakpoint +ALTER TABLE "payout_forward" ADD COLUMN "quote_id" uuid; \ No newline at end of file diff --git a/packages/db/migrations/meta/0010_snapshot.json b/packages/db/migrations/meta/0010_snapshot.json new file mode 100644 index 0000000..651522a --- /dev/null +++ b/packages/db/migrations/meta/0010_snapshot.json @@ -0,0 +1,2322 @@ +{ + "id": "b52b498e-1292-4b43-a413-11fcb49fc22b", + "prevId": "aef91b0e-1e40-473b-8aa5-e450cc2054a9", + "version": "7", + "dialect": "postgresql", + "tables": { + "public.api_key_nonce": { + "name": "api_key_nonce", + "schema": "", + "columns": { + "key_id": { + "name": "key_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "nonce": { + "name": "nonce", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": { + "api_key_nonce_key_id_nonce_pk": { + "name": "api_key_nonce_key_id_nonce_pk", + "columns": [ + "key_id", + "nonce" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.asset": { + "name": "asset", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "code": { + "name": "code", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "issuer_g": { + "name": "issuer_g", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "issuer_code": { + "name": "issuer_code", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "contract_c": { + "name": "contract_c", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "standard": { + "name": "standard", + "type": "asset_standard", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "yield_type": { + "name": "yield_type", + "type": "yield_type", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "custody_mode": { + "name": "custody_mode", + "type": "custody_mode", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "ccy": { + "name": "ccy", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "redemption_lag_days": { + "name": "redemption_lag_days", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "redemption_daily_cap": { + "name": "redemption_daily_cap", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": false + }, + "nav_schedule": { + "name": "nav_schedule", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "price_band_bps": { + "name": "price_band_bps", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "daily_move_band_bps": { + "name": "daily_move_band_bps", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "haircut_base_bps": { + "name": "haircut_base_bps", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "haircut_fx_bps": { + "name": "haircut_fx_bps", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'active'" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.asset_param_change": { + "name": "asset_param_change", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "asset_id": { + "name": "asset_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "field": { + "name": "field", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "old_value": { + "name": "old_value", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "new_value": { + "name": "new_value", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "queued_at": { + "name": "queued_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "executes_at": { + "name": "executes_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "tx_hash": { + "name": "tx_hash", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "executed_at": { + "name": "executed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "asset_param_change_asset_id_asset_id_fk": { + "name": "asset_param_change_asset_id_asset_id_fk", + "tableFrom": "asset_param_change", + "tableTo": "asset", + "columnsFrom": [ + "asset_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.audit_log": { + "name": "audit_log", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "actor": { + "name": "actor", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "action": { + "name": "action", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "payload": { + "name": "payload", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "tx_hash": { + "name": "tx_hash", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.chain_event": { + "name": "chain_event", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "event_id": { + "name": "event_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "ledger": { + "name": "ledger", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "tx_hash": { + "name": "tx_hash", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "contract": { + "name": "contract", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "topic": { + "name": "topic", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "data": { + "name": "data", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "ingested_at": { + "name": "ingested_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "chain_event_event_id_uq": { + "name": "chain_event_event_id_uq", + "columns": [ + { + "expression": "event_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.compliance_check": { + "name": "compliance_check", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "party_id": { + "name": "party_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'sanctions'" + }, + "result": { + "name": "result", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "matches": { + "name": "matches", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "list_source": { + "name": "list_source", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "list_loaded_at": { + "name": "list_loaded_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "checked_at": { + "name": "checked_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": { + "compliance_check_party_idx": { + "name": "compliance_check_party_idx", + "columns": [ + { + "expression": "party_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "checked_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": false, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "compliance_check_party_id_party_id_fk": { + "name": "compliance_check_party_id_party_id_fk", + "tableFrom": "compliance_check", + "tableTo": "party", + "columnsFrom": [ + "party_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.credit_line": { + "name": "credit_line", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "lender_id": { + "name": "lender_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "borrower_id": { + "name": "borrower_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "loan_ccy": { + "name": "loan_ccy", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "limit_amount": { + "name": "limit_amount", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "rate_bps": { + "name": "rate_bps", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "fee_bps": { + "name": "fee_bps", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "cure_window_s": { + "name": "cure_window_s", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'open'" + }, + "contract_line_id": { + "name": "contract_line_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "agreement_hash": { + "name": "agreement_hash", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "opened_at": { + "name": "opened_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "closed_at": { + "name": "closed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "credit_line_lender_id_party_id_fk": { + "name": "credit_line_lender_id_party_id_fk", + "tableFrom": "credit_line", + "tableTo": "party", + "columnsFrom": [ + "lender_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "credit_line_borrower_id_party_id_fk": { + "name": "credit_line_borrower_id_party_id_fk", + "tableFrom": "credit_line", + "tableTo": "party", + "columnsFrom": [ + "borrower_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.draw": { + "name": "draw", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "credit_line_id": { + "name": "credit_line_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "amount": { + "name": "amount", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "tx_hash": { + "name": "tx_hash", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "ledger": { + "name": "ledger", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "draw_credit_line_id_credit_line_id_fk": { + "name": "draw_credit_line_id_credit_line_id_fk", + "tableFrom": "draw", + "tableTo": "credit_line", + "columnsFrom": [ + "credit_line_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.exit_fill": { + "name": "exit_fill", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "quote_id": { + "name": "quote_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "tx_hash": { + "name": "tx_hash", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "payout_route": { + "name": "payout_route", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "payout_memo": { + "name": "payout_memo", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "exit_fill_quote_id_exit_quote_id_fk": { + "name": "exit_fill_quote_id_exit_quote_id_fk", + "tableFrom": "exit_fill", + "tableTo": "exit_quote", + "columnsFrom": [ + "quote_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.exit_quote": { + "name": "exit_quote", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "holder_id": { + "name": "holder_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "asset_id": { + "name": "asset_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "units": { + "name": "units", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "price": { + "name": "price", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "spread_bps": { + "name": "spread_bps", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "usdc_out": { + "name": "usdc_out", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "expires_at": { + "name": "expires_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'open'" + } + }, + "indexes": {}, + "foreignKeys": { + "exit_quote_holder_id_party_id_fk": { + "name": "exit_quote_holder_id_party_id_fk", + "tableFrom": "exit_quote", + "tableTo": "party", + "columnsFrom": [ + "holder_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "exit_quote_asset_id_asset_id_fk": { + "name": "exit_quote_asset_id_asset_id_fk", + "tableFrom": "exit_quote", + "tableTo": "asset", + "columnsFrom": [ + "asset_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.journal_entry": { + "name": "journal_entry", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "ref_type": { + "name": "ref_type", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "ref_id": { + "name": "ref_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "account": { + "name": "account", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "debit": { + "name": "debit", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true, + "default": "'0'" + }, + "credit": { + "name": "credit", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true, + "default": "'0'" + }, + "ccy": { + "name": "ccy", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.liquidation": { + "name": "liquidation", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "credit_line_id": { + "name": "credit_line_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "route": { + "name": "route", + "type": "liquidation_route", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "units": { + "name": "units", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "proceeds": { + "name": "proceeds", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": false + }, + "shortfall": { + "name": "shortfall", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": false + }, + "tx_hashes": { + "name": "tx_hashes", + "type": "text[]", + "primaryKey": false, + "notNull": false + }, + "started_at": { + "name": "started_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "completed_at": { + "name": "completed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "liquidation_credit_line_id_credit_line_id_fk": { + "name": "liquidation_credit_line_id_credit_line_id_fk", + "tableFrom": "liquidation", + "tableTo": "credit_line", + "columnsFrom": [ + "credit_line_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": { + "liquidation_credit_line_id_unique": { + "name": "liquidation_credit_line_id_unique", + "nullsNotDistinct": false, + "columns": [ + "credit_line_id" + ] + } + }, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.margin_event": { + "name": "margin_event", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "credit_line_id": { + "name": "credit_line_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "from_state": { + "name": "from_state", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "to_state": { + "name": "to_state", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "ltv_bps": { + "name": "ltv_bps", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "price_snapshot_id": { + "name": "price_snapshot_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "margin_event_credit_line_id_credit_line_id_fk": { + "name": "margin_event_credit_line_id_credit_line_id_fk", + "tableFrom": "margin_event", + "tableTo": "credit_line", + "columnsFrom": [ + "credit_line_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "margin_event_price_snapshot_id_price_snapshot_id_fk": { + "name": "margin_event_price_snapshot_id_price_snapshot_id_fk", + "tableFrom": "margin_event", + "tableTo": "price_snapshot", + "columnsFrom": [ + "price_snapshot_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.party": { + "name": "party", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "kind": { + "name": "kind", + "type": "party_kind", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "legal_name": { + "name": "legal_name", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "jurisdiction": { + "name": "jurisdiction", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "kyb_status": { + "name": "kyb_status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "kyb_provider": { + "name": "kyb_provider", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "kyb_ref": { + "name": "kyb_ref", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.party_account": { + "name": "party_account", + "schema": "", + "columns": { + "party_id": { + "name": "party_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "stellar_account": { + "name": "stellar_account", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "role": { + "name": "role", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "verified_at": { + "name": "verified_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "party_account_party_id_party_id_fk": { + "name": "party_account_party_id_party_id_fk", + "tableFrom": "party_account", + "tableTo": "party", + "columnsFrom": [ + "party_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": { + "party_account_party_id_stellar_account_pk": { + "name": "party_account_party_id_stellar_account_pk", + "columns": [ + "party_id", + "stellar_account" + ] + } + }, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.party_api_key": { + "name": "party_api_key", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "party_id": { + "name": "party_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "stellar_account": { + "name": "stellar_account", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "key_id": { + "name": "key_id", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "secret_ciphertext": { + "name": "secret_ciphertext", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "label": { + "name": "label", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'active'" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "last_used_at": { + "name": "last_used_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + }, + "revoked_at": { + "name": "revoked_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "party_api_key_key_id_uq": { + "name": "party_api_key_key_id_uq", + "columns": [ + { + "expression": "key_id", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "party_api_key_party_id_party_id_fk": { + "name": "party_api_key_party_id_party_id_fk", + "tableFrom": "party_api_key", + "tableTo": "party", + "columnsFrom": [ + "party_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.payout_forward": { + "name": "payout_forward", + "schema": "", + "columns": { + "from_tx_hash": { + "name": "from_tx_hash", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "to_account": { + "name": "to_account", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "amount": { + "name": "amount", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "memo": { + "name": "memo", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'claimed'" + }, + "tx_hash": { + "name": "tx_hash", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "ref_id": { + "name": "ref_id", + "type": "uuid", + "primaryKey": false, + "notNull": true, + "default": "gen_random_uuid()" + }, + "quote_id": { + "name": "quote_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.pledge": { + "name": "pledge", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "credit_line_id": { + "name": "credit_line_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "asset_id": { + "name": "asset_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "units": { + "name": "units", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "custody_mode": { + "name": "custody_mode", + "type": "custody_mode", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "contract_position_id": { + "name": "contract_position_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "lien_ref": { + "name": "lien_ref", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'active'" + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "pledge_credit_line_id_credit_line_id_fk": { + "name": "pledge_credit_line_id_credit_line_id_fk", + "tableFrom": "pledge", + "tableTo": "credit_line", + "columnsFrom": [ + "credit_line_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "pledge_asset_id_asset_id_fk": { + "name": "pledge_asset_id_asset_id_fk", + "tableFrom": "pledge", + "tableTo": "asset", + "columnsFrom": [ + "asset_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.price_observation": { + "name": "price_observation", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "asset_id": { + "name": "asset_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "source": { + "name": "source", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "value": { + "name": "value", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "observed_at": { + "name": "observed_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "signature": { + "name": "signature", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "stale": { + "name": "stale", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": false + } + }, + "indexes": { + "price_observation_asset_source_observed_uq": { + "name": "price_observation_asset_source_observed_uq", + "columns": [ + { + "expression": "asset_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "source", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "observed_at", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "price_observation_asset_id_asset_id_fk": { + "name": "price_observation_asset_id_asset_id_fk", + "tableFrom": "price_observation", + "tableTo": "asset", + "columnsFrom": [ + "asset_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.price_snapshot": { + "name": "price_snapshot", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "asset_id": { + "name": "asset_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "guarded_value": { + "name": "guarded_value", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "sources": { + "name": "sources", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "ledger": { + "name": "ledger", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "price_snapshot_asset_id_asset_id_fk": { + "name": "price_snapshot_asset_id_asset_id_fk", + "tableFrom": "price_snapshot", + "tableTo": "asset", + "columnsFrom": [ + "asset_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.reconciliation_run": { + "name": "reconciliation_run", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "scope": { + "name": "scope", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "started_at": { + "name": "started_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "breaks": { + "name": "breaks", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "report": { + "name": "report", + "type": "jsonb", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": {}, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.repayment": { + "name": "repayment", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "credit_line_id": { + "name": "credit_line_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "principal": { + "name": "principal", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "interest": { + "name": "interest", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "tx_hash": { + "name": "tx_hash", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "ledger": { + "name": "ledger", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "repayment_credit_line_id_credit_line_id_fk": { + "name": "repayment_credit_line_id_credit_line_id_fk", + "tableFrom": "repayment", + "tableTo": "credit_line", + "columnsFrom": [ + "credit_line_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.repo_margin_call": { + "name": "repo_margin_call", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "trade_id": { + "name": "trade_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "amount": { + "name": "amount", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "ccy": { + "name": "ccy", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "due_at": { + "name": "due_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "cured_at": { + "name": "cured_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "repo_margin_call_trade_id_repo_trade_id_fk": { + "name": "repo_margin_call_trade_id_repo_trade_id_fk", + "tableFrom": "repo_margin_call", + "tableTo": "repo_trade", + "columnsFrom": [ + "trade_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.repo_trade": { + "name": "repo_trade", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "cash_lender_id": { + "name": "cash_lender_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "cash_borrower_id": { + "name": "cash_borrower_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "asset_id": { + "name": "asset_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "units": { + "name": "units", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "cash_amount": { + "name": "cash_amount", + "type": "numeric(32, 7)", + "primaryKey": false, + "notNull": true + }, + "rate_bps": { + "name": "rate_bps", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "kind": { + "name": "kind", + "type": "repo_kind", + "typeSchema": "public", + "primaryKey": false, + "notNull": true + }, + "start_ledger": { + "name": "start_ledger", + "type": "integer", + "primaryKey": false, + "notNull": true + }, + "maturity_at": { + "name": "maturity_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true + }, + "agreement_hash": { + "name": "agreement_hash", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'proposed'" + }, + "contract_trade_id": { + "name": "contract_trade_id", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "leg1_tx": { + "name": "leg1_tx", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "leg2_tx": { + "name": "leg2_tx", + "type": "text", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "repo_trade_cash_lender_id_party_id_fk": { + "name": "repo_trade_cash_lender_id_party_id_fk", + "tableFrom": "repo_trade", + "tableTo": "party", + "columnsFrom": [ + "cash_lender_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "repo_trade_cash_borrower_id_party_id_fk": { + "name": "repo_trade_cash_borrower_id_party_id_fk", + "tableFrom": "repo_trade", + "tableTo": "party", + "columnsFrom": [ + "cash_borrower_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + }, + "repo_trade_asset_id_asset_id_fk": { + "name": "repo_trade_asset_id_asset_id_fk", + "tableFrom": "repo_trade", + "tableTo": "asset", + "columnsFrom": [ + "asset_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.tx_intent": { + "name": "tx_intent", + "schema": "", + "columns": { + "tx_hash": { + "name": "tx_hash", + "type": "text", + "primaryKey": true, + "notNull": true + }, + "kind": { + "name": "kind", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "ref_id": { + "name": "ref_id", + "type": "uuid", + "primaryKey": false, + "notNull": false + }, + "party_id": { + "name": "party_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "payload": { + "name": "payload", + "type": "jsonb", + "primaryKey": false, + "notNull": false + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "ledger": { + "name": "ledger", + "type": "integer", + "primaryKey": false, + "notNull": false + }, + "error": { + "name": "error", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "settled_at": { + "name": "settled_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": {}, + "foreignKeys": { + "tx_intent_party_id_party_id_fk": { + "name": "tx_intent_party_id_party_id_fk", + "tableFrom": "tx_intent", + "tableTo": "party", + "columnsFrom": [ + "party_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.webhook_delivery": { + "name": "webhook_delivery", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "subscription_id": { + "name": "subscription_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "event_key": { + "name": "event_key", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "event": { + "name": "event", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "payload": { + "name": "payload", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "status": { + "name": "status", + "type": "text", + "primaryKey": false, + "notNull": true, + "default": "'pending'" + }, + "attempts": { + "name": "attempts", + "type": "integer", + "primaryKey": false, + "notNull": true, + "default": 0 + }, + "next_attempt_at": { + "name": "next_attempt_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "last_error": { + "name": "last_error", + "type": "text", + "primaryKey": false, + "notNull": false + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + }, + "delivered_at": { + "name": "delivered_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": false + } + }, + "indexes": { + "webhook_delivery_sub_event_uq": { + "name": "webhook_delivery_sub_event_uq", + "columns": [ + { + "expression": "subscription_id", + "isExpression": false, + "asc": true, + "nulls": "last" + }, + { + "expression": "event_key", + "isExpression": false, + "asc": true, + "nulls": "last" + } + ], + "isUnique": true, + "concurrently": false, + "method": "btree", + "with": {} + } + }, + "foreignKeys": { + "webhook_delivery_subscription_id_webhook_subscription_id_fk": { + "name": "webhook_delivery_subscription_id_webhook_subscription_id_fk", + "tableFrom": "webhook_delivery", + "tableTo": "webhook_subscription", + "columnsFrom": [ + "subscription_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + }, + "public.webhook_subscription": { + "name": "webhook_subscription", + "schema": "", + "columns": { + "id": { + "name": "id", + "type": "uuid", + "primaryKey": true, + "notNull": true, + "default": "gen_random_uuid()" + }, + "party_id": { + "name": "party_id", + "type": "uuid", + "primaryKey": false, + "notNull": true + }, + "url": { + "name": "url", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "events": { + "name": "events", + "type": "jsonb", + "primaryKey": false, + "notNull": true + }, + "secret_ciphertext": { + "name": "secret_ciphertext", + "type": "text", + "primaryKey": false, + "notNull": true + }, + "active": { + "name": "active", + "type": "boolean", + "primaryKey": false, + "notNull": true, + "default": true + }, + "created_at": { + "name": "created_at", + "type": "timestamp with time zone", + "primaryKey": false, + "notNull": true, + "default": "now()" + } + }, + "indexes": {}, + "foreignKeys": { + "webhook_subscription_party_id_party_id_fk": { + "name": "webhook_subscription_party_id_party_id_fk", + "tableFrom": "webhook_subscription", + "tableTo": "party", + "columnsFrom": [ + "party_id" + ], + "columnsTo": [ + "id" + ], + "onDelete": "no action", + "onUpdate": "no action" + } + }, + "compositePrimaryKeys": {}, + "uniqueConstraints": {}, + "policies": {}, + "checkConstraints": {}, + "isRLSEnabled": false + } + }, + "enums": { + "public.asset_standard": { + "name": "asset_standard", + "schema": "public", + "values": [ + "classic_sac", + "sep57" + ] + }, + "public.custody_mode": { + "name": "custody_mode", + "schema": "public", + "values": [ + "escrow", + "issuer_lien", + "custodian_lien" + ] + }, + "public.liquidation_route": { + "name": "liquidation_route", + "schema": "public", + "values": [ + "redeem", + "transfer", + "rfq" + ] + }, + "public.party_kind": { + "name": "party_kind", + "schema": "public", + "values": [ + "borrower", + "lender", + "holder", + "issuer", + "custodian", + "keeper" + ] + }, + "public.repo_kind": { + "name": "repo_kind", + "schema": "public", + "values": [ + "intraday", + "overnight", + "open", + "term" + ] + }, + "public.yield_type": { + "name": "yield_type", + "schema": "public", + "values": [ + "accumulating", + "distributing" + ] + } + }, + "schemas": {}, + "sequences": {}, + "roles": {}, + "policies": {}, + "views": {}, + "_meta": { + "columns": {}, + "schemas": {}, + "tables": {} + } +} \ No newline at end of file diff --git a/packages/db/migrations/meta/_journal.json b/packages/db/migrations/meta/_journal.json index 4f955c2..beaecd1 100644 --- a/packages/db/migrations/meta/_journal.json +++ b/packages/db/migrations/meta/_journal.json @@ -71,6 +71,13 @@ "when": 1792424561610, "tag": "0009_liquidation_credit_line_unique", "breakpoints": true + }, + { + "idx": 10, + "version": "7", + "when": 1792604217243, + "tag": "0010_payout_forward_booking", + "breakpoints": true } ] } \ No newline at end of file diff --git a/packages/db/src/schema.ts b/packages/db/src/schema.ts index fb36acb..ea782e6 100644 --- a/packages/db/src/schema.ts +++ b/packages/db/src/schema.ts @@ -386,10 +386,16 @@ export const payoutForward = pgTable("payout_forward", { toAccount: text("to_account").notNull(), amount: money("amount").notNull(), memo: text("memo").notNull(), - /** `claimed` while in flight, `sent` once the network accepted it, `failed` if it did not (a - * failed forward may be retried). */ + /** `claimed` while in flight, `sent` once the network accepted it and the journal pair is booked, + * `bookkeeping_pending` if it was accepted but the bookkeeping writes failed (the repair books + * those), `failed` if the network did not accept it (a failed forward may be retried). */ status: text("status").notNull().default("claimed"), txHash: text("tx_hash"), + /** The `journal_entry.ref_id` the payout is booked under, fixed at claim time so a repair books + * under the same id the original attempt would have. */ + refId: uuid("ref_id").notNull().defaultRandom(), + /** The `exit_quote` this payout fulfils, if any; kept so a repair can write the `exit_fill` row. */ + quoteId: uuid("quote_id"), createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), }); diff --git a/packages/observability/src/metrics.ts b/packages/observability/src/metrics.ts index cc0d8f7..02d1ae7 100644 --- a/packages/observability/src/metrics.ts +++ b/packages/observability/src/metrics.ts @@ -32,7 +32,8 @@ export type AlertKind = | "keeper_balance_low" | "operator_balance_low" | "reconciliation_break" - | "rpc_lag_high"; + | "rpc_lag_high" + | "payout_unbooked"; export interface Alerter { fire(kind: AlertKind, context: Record): void; diff --git a/services/payout-hop/README.md b/services/payout-hop/README.md index da4d43f..d4301e5 100644 --- a/services/payout-hop/README.md +++ b/services/payout-hop/README.md @@ -19,8 +19,12 @@ It moves real funds, so treat it as the most sensitive service in the repo. | `usdcAssetAddress` | the USDC issuer's **G-address** (not the SAC `C...` id); must equal `PAYOUT_HOP_USDC_ISSUER` | | `quoteId` | optional `exit_quote` uuid | -Answers `200 { hash, status }`, `400` bad body, `401` no or wrong token, `403` refused by policy, `409` -this first-leg hash was already requested, `502` submission failed. +Answers `200 { hash, status, booked }`, `400` bad body, `401` no or wrong token, `403` refused by policy, +`409` this first-leg hash was already requested, `502` submission failed (nothing was paid). + +Once the network has accepted the payment the answer is always `200` with its hash, because the anchor +has been paid. `booked: false` means recording it failed: the row is left `bookkeeping_pending`, a +`payout_unbooked` alert fires, and the repair below books it. Callers must not retry on `booked: false`. ## Security @@ -55,26 +59,49 @@ pnpm --filter @ballast/service-payout-hop build && pnpm --filter @ballast/servic | `PAYOUT_HOP_USDC_ISSUER` | USDC issuer G-address | | `PAYOUT_HOP_ALLOWED_DESTINATIONS` | comma-separated anchor G-addresses | | `PAYOUT_HOP_MAX_AMOUNT` | per-payout cap, whole USDC (default 5000) | +| `PAYOUT_HOP_REPAIR_INTERVAL_S` | how often unbooked payouts are swept (default 300; also runs at startup) | +| `ALERT_WEBHOOK_URL`, `ALERT_COOLDOWN_S` | where `payout_unbooked` alerts are posted (console only when unset) | | `PAYOUT_HOP_SIGNER_MODE`, `PAYOUT_HOP_SECRET_KEY`, `VAULT_*`, `PAYOUT_HOP_VAULT_KEY_NAME`, `AWS_REGION`, `PAYOUT_HOP_KMS_KEY_ID`, `PAYOUT_HOP_WRAPPED_SECRET` | operator signer | | `DATABASE_URL`, `STELLAR_RPC_URL`, `STELLAR_NETWORK_PASSPHRASE` | database and network | ## Source map - `src/index.ts`: Fastify server, request schema, status mapping. -- `src/hop.ts`: `PayoutPolicy`, `forwardPayout` (claim, build payment, submit, record). +- `src/hop.ts`: `PayoutPolicy`, `forwardPayout` (claim, build payment, submit, record), `bookPayout`. +- `src/repair.ts`: `bookPendingPayouts`, which books `bookkeeping_pending` rows. - `src/signer.ts`: `PayoutSigner` seam and the RPC-backed `OperatorPayoutSigner`. ## What it writes -`payout_forward` (claim, then `sent` with the tx hash, or `failed`); `exit_fill` (only when `quoteId` is -given); and two balanced `journal_entry` rows (`usdc_cash` debit, `operator:payout_hop` credit). +`payout_forward` (claim, then `sent` with the tx hash, `bookkeeping_pending` if sent but not booked, or +`failed`); `exit_fill` (only when `quoteId` is given); and two balanced `journal_entry` rows (`usdc_cash` +debit, `operator:payout_hop` credit). + +The claim row stores the journal `ref_id` (the `quoteId` when there is one, otherwise a fresh uuid) and +the `quote_id`, before anything is sent. Marking the row `sent` and writing `exit_fill` plus the journal +pair happen in one transaction, so `sent` always means booked, and everything needed to book a payout +later is already on its row. + +## Repairing unbooked payouts + +`bookPendingPayouts(db, alerter)` books every `bookkeeping_pending` row under its stored `ref_id` and +marks it `sent`. It runs at startup and every `PAYOUT_HOP_REPAIR_INTERVAL_S`. It is idempotent and safe +to run from several processes: each row is taken by a conditional update in its own transaction, and a row +whose journal pair already exists (the original commit landed but its acknowledgement was lost) is only +marked `sent`. A row it still cannot book stays pending and alerts `payout_unbooked` again. ## Tests `pnpm --filter @ballast/service-payout-hop test` (vitest; fake signer and database, no network). Covers 401 without the token, 400 on malformed bodies, 403 on a refused destination, each policy rule including the cap boundary, text and numeric memos, pay-once on repeats, retry after failure, an `ERROR` status -treated as failed, and `loadPayoutPolicy` validation. +treated as failed, and `loadPayoutPolicy` validation. The fake database keeps real rows, honours the +`eq`/`and` filters used here and rolls back a transaction that throws. With it: a booking transaction +that throws after sending still returns the hash with `booked: false`, leaves the row +`bookkeeping_pending` with nothing half-written, alerts, and refuses a retry with `409`; a database that +is gone entirely still returns the hash and alerts that the row could not be flagged. The repair books +under the claim's `ref_id` (quote or generated), writes `exit_fill`, is idempotent, only marks `sent` a row +already booked, retries a row it failed on, and leaves other statuses alone. ## Known limits @@ -82,6 +109,7 @@ treated as failed, and `loadPayoutPolicy` validation. does not use it, and `TESTNET_DEPLOYMENT.md` says the same. A real forward has not been run on a network. - Only `ERROR` from `sendTransaction` is treated as failure. `PENDING` is recorded as `sent` without waiting for the ledger to confirm it. -- If the process dies after claiming but before finishing, the row stays `claimed` and later requests - get `409` until someone resolves it by hand. +- If the process dies after claiming but before finishing, or the database is unreachable when the + booking fails (so the row cannot be flagged), the row stays `claimed` and later requests get `409` + until someone resolves it by hand. In the second case the `payout_unbooked` alert carries the tx hash. - The hop does not check that the first leg actually landed; it trusts the caller. diff --git a/services/payout-hop/src/hop.test.ts b/services/payout-hop/src/hop.test.ts index 8a28c5d..5d515ad 100644 --- a/services/payout-hop/src/hop.test.ts +++ b/services/payout-hop/src/hop.test.ts @@ -1,6 +1,8 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { Account, Keypair, MemoID, MemoText, Networks, type Transaction } from "@stellar/stellar-sdk"; +import { Param, SQL } from "drizzle-orm"; import { schema } from "@ballast/db"; +import type { Alerter } from "@ballast/observability"; import { DuplicatePayoutError, PayoutRefusedError, @@ -9,6 +11,7 @@ import { type ForwardPayoutArgs, type PayoutPolicy, } from "./hop.js"; +import { bookPendingPayouts } from "./repair.js"; import type { PayoutSigner } from "./signer.js"; const OPERATOR = Keypair.random().publicKey(); @@ -16,6 +19,7 @@ const ANCHOR = Keypair.random().publicKey(); const OTHER_ANCHOR = Keypair.random().publicKey(); const USDC_ISSUER = Keypair.random().publicKey(); const UNIT = 10_000_000n; +const QUOTE_ID = "9d1f8f2a-1234-4a3b-9c00-abcdefabcdef"; const policy: PayoutPolicy = { allowedDestinations: new Set([ANCHOR]), usdcIssuer: USDC_ISSUER, maxAmount: 10_000n * UNIT }; @@ -35,51 +39,121 @@ function fakeSigner(status = "PENDING"): PayoutSigner & { submittedTx: Transacti return signer; } +function fakeAlerter(): Alerter & { fire: ReturnType } { + return { fire: vi.fn() }; +} + +type Row = Record; + +/** Every value bound in a drizzle `eq`/`and` condition. */ +function boundValues(condition: unknown): unknown[] { + if (condition instanceof Param) return [condition.value]; + if (condition instanceof SQL) return condition.queryChunks.flatMap(boundValues); + return []; +} + +/** True when each bound value equals one of the row's fields: enough for the `eq`/`and` filters used here. */ +const matches = (row: Row, condition: unknown) => boundValues(condition).every((v) => Object.values(row).includes(v)); + /** - * Stateful fake of the slice of `Database` the hop touches: the payout_forward claim table - * (insert-or-ignore, status updates) and a transaction for the journal rows. + * In-memory stand-in for the slice of `Database` the hop and the repair touch. Tables keep real + * rows, `where` honours `eq`/`and` filters, the `payout_forward` primary key makes a second claim + * insert a no-op, and a transaction that throws rolls its writes back. + * + * `failNextTransaction` makes the next transaction throw after running its callback (a dropped + * connection mid-commit); `goDownOnTransaction` additionally makes every later call throw. */ function fakeDb() { - const claims = new Map(); - const inserted: Array<{ table: unknown; row: Record }> = []; - const insertInto = (sink: Array<{ table: unknown; row: Record }>) => (table: unknown) => ({ - values: (row: Record) => { - const record = () => sink.push({ table, row }); - const isClaim = table === schema.payoutForward; - return Object.assign(Promise.resolve(record()), { - onConflictDoNothing: () => ({ - returning: async () => { - const key = row.fromTxHash as string; - if (isClaim && claims.has(key)) return []; - if (isClaim) claims.set(key, { status: "claimed", toAccount: row.toAccount as string }); - return [{ fromTxHash: key }]; - }, - }), - }); - }, - }); - let lastSet: Record = {}; - const db = { - insert: insertInto(inserted), - update: () => ({ - set: (set: Record) => { - lastSet = set; - return { - where: () => { - // The only conditional update is the "re-claim a failed payout" one; apply it to the failed claims. - const reclaimable = [...claims.entries()].filter(([, c]) => c.status === "failed"); - const touched = set.status === "claimed" ? reclaimable : [...claims.entries()]; - for (const [, c] of touched) Object.assign(c, set); - return Object.assign(Promise.resolve(), { - returning: async () => (set.status === "claimed" ? touched.map(([k]) => ({ fromTxHash: k })) : []), - }); - }, - }; + let tables = new Map(); + const rowsOf = (table: unknown): Row[] => { + if (!tables.has(table)) tables.set(table, []); + return tables.get(table)!; + }; + const state = { failNextTransaction: null as Error | null, goDownOnTransaction: false, down: false }; + const guard = () => { + if (state.down) throw new Error("connection terminated"); + }; + const query = (run: () => T, extra: object = {}) => + Object.assign( + { + then: (resolve: (v: T) => unknown, reject: (e: unknown) => unknown) => + Promise.resolve().then(() => { guard(); return run(); }).then(resolve, reject), + }, + extra, + ); + const project = (row: Row, fields?: Record) => + fields ? Object.fromEntries(Object.keys(fields).map((k) => [k, row[k]])) : { ...row }; + + const ops = { + insert: (table: unknown) => ({ + values: (row: Row) => { + const full = + table === schema.payoutForward ? { status: "claimed", txHash: null, quoteId: null, ...row } : { ...row }; + return query(() => { rowsOf(table).push(full); }, { + onConflictDoNothing: () => ({ + returning: async () => { + guard(); + if (rowsOf(table).some((r) => r.fromTxHash === row.fromTxHash)) return []; + rowsOf(table).push(full); + return [{ fromTxHash: row.fromTxHash }]; + }, + }), + }); }, }), - transaction: async (fn: (trx: unknown) => Promise) => fn({ insert: insertInto(inserted) }), + update: (table: unknown) => ({ + set: (set: Row) => ({ + where: (condition: unknown) => { + const apply = () => { + const touched = rowsOf(table).filter((r) => matches(r, condition)); + for (const r of touched) Object.assign(r, set); + return touched; + }; + return query(() => { apply(); }, { + returning: async () => { guard(); return apply().map((r) => ({ fromTxHash: r.fromTxHash })); }, + }); + }, + }), + }), + select: (fields?: Record) => ({ + from: (table: unknown) => ({ + where: (condition: unknown) => { + const found = () => rowsOf(table).filter((r) => matches(r, condition)).map((r) => project(r, fields)); + return query(found, { limit: (n: number) => query(() => found().slice(0, n)) }); + }, + }), + }), + }; + + const db = { + ...ops, + transaction: async (fn: (trx: typeof ops) => Promise): Promise => { + guard(); + const snapshot = new Map([...tables].map(([t, rows]) => [t, rows.map((r) => ({ ...r }))])); + try { + const result = await fn(ops); + if (state.failNextTransaction) { + const err = state.failNextTransaction; + state.failNextTransaction = null; + if (state.goDownOnTransaction) state.down = true; + throw err; + } + return result; + } catch (err) { + tables = snapshot; + throw err; + } + }, } as unknown as Parameters[0]; - return { db, inserted, claims, lastSet: () => lastSet }; + + return { + db, + state, + claim: (hash: string) => rowsOf(schema.payoutForward).find((r) => r.fromTxHash === hash), + claims: () => rowsOf(schema.payoutForward), + journal: () => rowsOf(schema.journalEntry), + exitFills: () => rowsOf(schema.exitFill), + }; } const base = (over: Partial = {}): ForwardPayoutArgs => ({ @@ -88,73 +162,204 @@ const base = (over: Partial = {}): ForwardPayoutArgs => ({ amount: 5_000n * UNIT, memo: "customer-ref-123", usdcAssetAddress: USDC_ISSUER, - quoteId: "9d1f8f2a-1234-4a3b-9c00-abcdefabcdef", + quoteId: QUOTE_ID, ...over, }); describe("forwardPayout", () => { it("attaches Memo.text for a non-numeric memo and records exit_fill + a balanced journal pair", async () => { const signer = fakeSigner(); - const { db, inserted, claims } = fakeDb(); + const { db, claim, journal, exitFills } = fakeDb(); - const result = await forwardPayout(db, signer, base(), policy); + const result = await forwardPayout(db, signer, base(), policy, fakeAlerter()); - expect(result).toEqual({ hash: "fakehash123", status: "PENDING" }); + expect(result).toEqual({ hash: "fakehash123", status: "PENDING", booked: true }); expect(signer.submittedTx?.memo.type).toBe(MemoText); expect(Buffer.from(signer.submittedTx?.memo.value as Uint8Array).toString()).toBe("customer-ref-123"); - expect(claims.get("contracttxhash")).toMatchObject({ status: "sent", txHash: "fakehash123" }); + expect(claim("contracttxhash")).toMatchObject({ status: "sent", txHash: "fakehash123", refId: QUOTE_ID, quoteId: QUOTE_ID }); - const rows = inserted.filter((i) => i.table !== schema.payoutForward); - expect(rows).toHaveLength(3); - expect(rows.find((i) => i.row.payoutRoute === "anchor_hop")).toMatchObject({ - row: { payoutRoute: "anchor_hop", payoutMemo: "customer-ref-123", txHash: "fakehash123" }, - }); - expect(rows.find((i) => i.row.account === "usdc_cash")).toMatchObject({ - row: { account: "usdc_cash", debit: "5000.0000000", credit: "0.0000000", ccy: "USDC" }, - }); - expect(rows.find((i) => i.row.account === "operator:payout_hop")).toMatchObject({ - row: { account: "operator:payout_hop", debit: "0.0000000", credit: "5000.0000000", ccy: "USDC" }, - }); + expect(exitFills()).toEqual([ + { quoteId: QUOTE_ID, payoutRoute: "anchor_hop", payoutMemo: "customer-ref-123", txHash: "fakehash123" }, + ]); + expect(journal()).toEqual([ + { refType: "payout_hop", refId: QUOTE_ID, account: "usdc_cash", debit: "5000.0000000", credit: "0.0000000", ccy: "USDC" }, + { refType: "payout_hop", refId: QUOTE_ID, account: "operator:payout_hop", debit: "0.0000000", credit: "5000.0000000", ccy: "USDC" }, + ]); }); it("attaches Memo.id for a numeric memo, and skips exit_fill when there's no quoteId", async () => { const signer = fakeSigner(); - const { db, inserted } = fakeDb(); + const { db, claim, journal, exitFills } = fakeDb(); - await forwardPayout(db, signer, base({ fromContractPayoutTxHash: "h2", amount: 10n * UNIT, memo: "88221199", quoteId: undefined }), policy); + await forwardPayout(db, signer, base({ fromContractPayoutTxHash: "h2", amount: 10n * UNIT, memo: "88221199", quoteId: undefined }), policy, fakeAlerter()); expect(signer.submittedTx?.memo.type).toBe(MemoID); expect(signer.submittedTx?.memo.value?.toString()).toBe("88221199"); - expect(inserted.filter((i) => i.table !== schema.payoutForward)).toHaveLength(2); + expect(exitFills()).toHaveLength(0); + // Without a quote the journal pair is booked under the id generated at claim time. + expect(journal().map((r) => r.refId)).toEqual([claim("h2")?.refId, claim("h2")?.refId]); }); it("pays a given first-leg transaction once, however many times it is asked", async () => { const signer = fakeSigner(); const { db } = fakeDb(); - await forwardPayout(db, signer, base(), policy); - await expect(forwardPayout(db, signer, base(), policy)).rejects.toBeInstanceOf(DuplicatePayoutError); - await expect(forwardPayout(db, signer, base({ amount: 1n * UNIT }), policy)).rejects.toBeInstanceOf(DuplicatePayoutError); + await forwardPayout(db, signer, base(), policy, fakeAlerter()); + await expect(forwardPayout(db, signer, base(), policy, fakeAlerter())).rejects.toBeInstanceOf(DuplicatePayoutError); + await expect(forwardPayout(db, signer, base({ amount: 1n * UNIT }), policy, fakeAlerter())).rejects.toBeInstanceOf(DuplicatePayoutError); expect(signer.calls).toBe(1); }); it("lets a payout that failed to send be tried again, and pays it once it succeeds", async () => { - const { db, claims } = fakeDb(); + const { db, claim } = fakeDb(); const failing: PayoutSigner = { ...fakeSigner(), signAndSubmit: async () => { throw new Error("rpc down"); } }; - await expect(forwardPayout(db, failing, base(), policy)).rejects.toThrow("rpc down"); - expect(claims.get("contracttxhash")?.status).toBe("failed"); + await expect(forwardPayout(db, failing, base(), policy, fakeAlerter())).rejects.toThrow("rpc down"); + expect(claim("contracttxhash")?.status).toBe("failed"); const working = fakeSigner(); - await forwardPayout(db, working, base(), policy); + await forwardPayout(db, working, base(), policy, fakeAlerter()); expect(working.calls).toBe(1); - expect(claims.get("contracttxhash")?.status).toBe("sent"); + expect(claim("contracttxhash")?.status).toBe("sent"); }); it("treats a network ERROR status as a failed payout, not a sent one", async () => { - const { db, claims } = fakeDb(); - await expect(forwardPayout(db, fakeSigner("ERROR"), base(), policy)).rejects.toThrow(/submission failed/); - expect(claims.get("contracttxhash")?.status).toBe("failed"); + const { db, claim } = fakeDb(); + await expect(forwardPayout(db, fakeSigner("ERROR"), base(), policy, fakeAlerter())).rejects.toThrow(/submission failed/); + expect(claim("contracttxhash")?.status).toBe("failed"); + }); +}); + +describe("bookkeeping after the payment is sent", () => { + it("answers with the hash, flags the row bookkeeping_pending and alerts when the booking transaction throws", async () => { + const signer = fakeSigner(); + const alerter = fakeAlerter(); + const { db, state, claim, journal, exitFills } = fakeDb(); + state.failNextTransaction = new Error("insert on exit_fill violates foreign key constraint"); + + const result = await forwardPayout(db, signer, base(), policy, alerter); + + expect(result).toEqual({ hash: "fakehash123", status: "PENDING", booked: false }); + expect(signer.calls).toBe(1); + // The transaction rolled back, so nothing half-booked; the row says what happened and keeps the hash. + expect(claim("contracttxhash")).toMatchObject({ status: "bookkeeping_pending", txHash: "fakehash123", refId: QUOTE_ID }); + expect(journal()).toHaveLength(0); + expect(exitFills()).toHaveLength(0); + expect(alerter.fire).toHaveBeenCalledWith( + "payout_unbooked", + expect.objectContaining({ subject: "contracttxhash", txHash: "fakehash123", refId: QUOTE_ID, flagged: true }), + ); + expect(alerter.fire.mock.calls[0][1].reason).toMatch(/foreign key constraint/); + + // A caller retrying is still refused: the anchor was paid. + await expect(forwardPayout(db, signer, base(), policy, alerter)).rejects.toBeInstanceOf(DuplicatePayoutError); + expect(signer.calls).toBe(1); + }); + + it("still answers with the hash when the database is gone and the row cannot even be flagged", async () => { + const alerter = fakeAlerter(); + const { db, state, claim } = fakeDb(); + state.failNextTransaction = new Error("connection terminated"); + state.goDownOnTransaction = true; + + const result = await forwardPayout(db, fakeSigner(), base(), policy, alerter); + + expect(result).toEqual({ hash: "fakehash123", status: "PENDING", booked: false }); + expect(claim("contracttxhash")?.status).toBe("claimed"); + expect(alerter.fire).toHaveBeenCalledWith( + "payout_unbooked", + expect.objectContaining({ subject: "contracttxhash", txHash: "fakehash123", flagged: false }), + ); + }); +}); + +describe("bookPendingPayouts", () => { + /** Sends a payout whose booking fails, leaving it `bookkeeping_pending`. */ + async function unbooked(over: Partial = {}) { + const fake = fakeDb(); + fake.state.failNextTransaction = new Error("connection reset"); + await forwardPayout(fake.db, fakeSigner(), base(over), policy, fakeAlerter()); + expect(fake.claim(over.fromContractPayoutTxHash ?? "contracttxhash")?.status).toBe("bookkeeping_pending"); + return fake; + } + + it("books a pending payout under the refId fixed at claim time, with its exit_fill, and marks it sent", async () => { + const { db, claim, journal, exitFills } = await unbooked(); + + const result = await bookPendingPayouts(db, fakeAlerter()); + + expect(result).toEqual({ booked: ["contracttxhash"], alreadyBooked: [], failed: [] }); + expect(claim("contracttxhash")?.status).toBe("sent"); + expect(exitFills()).toEqual([ + { quoteId: QUOTE_ID, payoutRoute: "anchor_hop", payoutMemo: "customer-ref-123", txHash: "fakehash123" }, + ]); + expect(journal()).toEqual([ + { refType: "payout_hop", refId: QUOTE_ID, account: "usdc_cash", debit: "5000.0000000", credit: "0.0000000", ccy: "USDC" }, + { refType: "payout_hop", refId: QUOTE_ID, account: "operator:payout_hop", debit: "0.0000000", credit: "5000.0000000", ccy: "USDC" }, + ]); + }); + + it("uses the generated refId of a payout with no quote, the same one the original attempt used", async () => { + const { db, claim, journal, exitFills } = await unbooked({ quoteId: undefined }); + const refId = claim("contracttxhash")?.refId; + expect(refId).toMatch(/^[0-9a-f-]{36}$/); + + await bookPendingPayouts(db, fakeAlerter()); + + expect(journal().map((r) => r.refId)).toEqual([refId, refId]); + expect(exitFills()).toHaveLength(0); + }); + + it("is idempotent: a second run books nothing more", async () => { + const { db, journal } = await unbooked(); + + await bookPendingPayouts(db, fakeAlerter()); + const again = await bookPendingPayouts(db, fakeAlerter()); + + expect(again).toEqual({ booked: [], alreadyBooked: [], failed: [] }); + expect(journal()).toHaveLength(2); + }); + + it("only marks sent a payout whose journal pair did commit (the acknowledgement was what got lost)", async () => { + const { db, claim, journal } = await unbooked(); + journal().push( + { refType: "payout_hop", refId: QUOTE_ID, account: "usdc_cash", debit: "5000.0000000", credit: "0.0000000", ccy: "USDC" }, + { refType: "payout_hop", refId: QUOTE_ID, account: "operator:payout_hop", debit: "0.0000000", credit: "5000.0000000", ccy: "USDC" }, + ); + + const result = await bookPendingPayouts(db, fakeAlerter()); + + expect(result).toEqual({ booked: [], alreadyBooked: ["contracttxhash"], failed: [] }); + expect(journal()).toHaveLength(2); + expect(claim("contracttxhash")?.status).toBe("sent"); + }); + + it("leaves a payout it cannot book pending, alerts, and books it on a later run", async () => { + const { db, state, claim, journal } = await unbooked(); + const alerter = fakeAlerter(); + state.failNextTransaction = new Error("deadlock detected"); + + const first = await bookPendingPayouts(db, alerter); + + expect(first).toEqual({ booked: [], alreadyBooked: [], failed: ["contracttxhash"] }); + expect(claim("contracttxhash")?.status).toBe("bookkeeping_pending"); + expect(journal()).toHaveLength(0); + expect(alerter.fire).toHaveBeenCalledWith("payout_unbooked", expect.objectContaining({ subject: "contracttxhash" })); + + expect((await bookPendingPayouts(db, alerter)).booked).toEqual(["contracttxhash"]); + expect(journal()).toHaveLength(2); + }); + + it("leaves claimed, failed and already-sent payouts alone", async () => { + const { db, claims, journal } = fakeDb(); + claims().push( + { fromTxHash: "a", status: "claimed", txHash: null, refId: "00000000-0000-4000-8000-000000000001", quoteId: null, amount: "1.0000000", memo: "m" }, + { fromTxHash: "b", status: "failed", txHash: null, refId: "00000000-0000-4000-8000-000000000002", quoteId: null, amount: "1.0000000", memo: "m" }, + { fromTxHash: "c", status: "sent", txHash: "t", refId: "00000000-0000-4000-8000-000000000003", quoteId: null, amount: "1.0000000", memo: "m" }, + ); + + expect(await bookPendingPayouts(db, fakeAlerter())).toEqual({ booked: [], alreadyBooked: [], failed: [] }); + expect(journal()).toHaveLength(0); }); }); @@ -162,10 +367,10 @@ describe("payout policy", () => { const refused = async (over: Partial, reason: RegExp) => { const signer = fakeSigner(); const { db, claims } = fakeDb(); - await expect(forwardPayout(db, signer, base(over), policy)).rejects.toThrow(reason); - await expect(forwardPayout(db, signer, base(over), policy)).rejects.toBeInstanceOf(PayoutRefusedError); + await expect(forwardPayout(db, signer, base(over), policy, fakeAlerter())).rejects.toThrow(reason); + await expect(forwardPayout(db, signer, base(over), policy, fakeAlerter())).rejects.toBeInstanceOf(PayoutRefusedError); expect(signer.calls).toBe(0); - expect(claims.size).toBe(0); // refused before anything was claimed + expect(claims()).toHaveLength(0); // refused before anything was claimed }; it("refuses a destination that is not an approved anchor", () => refused({ toAnchorAccount: OTHER_ANCHOR }, /approved anchor/)); @@ -180,7 +385,7 @@ describe("payout policy", () => { it("accepts a payout exactly at the cap", async () => { const { db } = fakeDb(); - await expect(forwardPayout(db, fakeSigner(), base({ amount: 10_000n * UNIT }), policy)).resolves.toBeDefined(); + await expect(forwardPayout(db, fakeSigner(), base({ amount: 10_000n * UNIT }), policy, fakeAlerter())).resolves.toBeDefined(); }); }); diff --git a/services/payout-hop/src/hop.ts b/services/payout-hop/src/hop.ts index 94547ca..c4847b3 100644 --- a/services/payout-hop/src/hop.ts +++ b/services/payout-hop/src/hop.ts @@ -10,8 +10,9 @@ import { Asset, Memo, Operation, TransactionBuilder, BASE_FEE, type Transaction import { randomUUID } from "node:crypto"; import { and, eq } from "drizzle-orm"; import { StrKey } from "@stellar/stellar-sdk"; -import { schema, recordJournalEntry, type Database } from "@ballast/db"; +import { schema, recordJournalEntry, type Database, type DbOrTx } from "@ballast/db"; import { scaledToDecimalString } from "@ballast/domain-types"; +import { createAlerter, type Alerter } from "@ballast/observability"; import type { PayoutSigner } from "./signer.js"; @@ -104,6 +105,22 @@ export interface ForwardPayoutArgs { export interface ForwardPayoutResult { hash: string; status: string; + /** + * False when the payment was accepted but recording it failed: the row is left + * `bookkeeping_pending` for `bookPendingPayouts` and `payout_unbooked` is alerted. The payout + * itself happened either way, so the caller must not treat `false` as a failure. + */ + booked: boolean; +} + +/** What a sent payout is booked from: everything is on the `payout_forward` row, so a repair can redo it. */ +export interface PayoutBooking { + fromTxHash: string; + txHash: string; + refId: string; + quoteId: string | null; + amount: bigint; + memo: string; } /** @@ -116,9 +133,15 @@ export async function forwardPayout( signer: PayoutSigner, args: ForwardPayoutArgs, policy: PayoutPolicy, + alerter: Alerter = createAlerter("payout-hop"), ): Promise { checkPolicy(args, policy); + // Fixed before any money moves and stored on the claim, so the journal pair is booked under the + // same id whether it is written right after sending or later by `bookPendingPayouts`. + const refId = args.quoteId ?? randomUUID(); + const quoteId = args.quoteId ?? null; + // Claim the first-leg transaction before any money moves. The primary key means the same payout // can be requested any number of times (a retry, a replay, a duplicate caller) and be paid once; // a payout that failed is the one case that may be claimed again. @@ -129,13 +152,15 @@ export async function forwardPayout( toAccount: args.toAnchorAccount, amount: scaledToDecimalString(args.amount), memo: args.memo, + refId, + quoteId, }) .onConflictDoNothing() .returning({ fromTxHash: schema.payoutForward.fromTxHash }); if (claimed.length === 0) { const reclaimed = await db .update(schema.payoutForward) - .set({ status: "claimed", toAccount: args.toAnchorAccount, amount: scaledToDecimalString(args.amount), memo: args.memo }) + .set({ status: "claimed", toAccount: args.toAnchorAccount, amount: scaledToDecimalString(args.amount), memo: args.memo, refId, quoteId }) .where( and( eq(schema.payoutForward.fromTxHash, args.fromContractPayoutTxHash), @@ -179,47 +204,95 @@ export async function forwardPayout( throw err; } - await db - .update(schema.payoutForward) - .set({ status: "sent", txHash: hash }) - .where(eq(schema.payoutForward.fromTxHash, args.fromContractPayoutTxHash)); - - await db.transaction(async (trx) => { - // `exit_fill.quote_id` is NOT NULL in the current schema and there's no generic "payout" table - // for non-exit flows (a credit-line draw routed to an anchor). When there's a real exit_quote - // to point at, record the exit_fill row; otherwise rely on the journal_entry rows below plus - // `payout_forward` for traceability. - if (args.quoteId) { - await trx.insert(schema.exitFill).values({ - quoteId: args.quoteId, - txHash: hash, - payoutRoute: "anchor_hop", - payoutMemo: args.memo, - }); - } - - // A balanced pair, not a lone credit row: cash leaving Ballast's operator account (debit) - // funds the payout booked against the payout-hop account (credit). Both legs go through - // `recordJournalEntry` directly (not `recordDoubleEntry`, which opens its own transaction) - // since this is already inside one. - const refId = args.quoteId ?? randomUUID(); - await recordJournalEntry(trx, { - refType: "payout_hop", - refId, - account: "usdc_cash", - debit: args.amount, - credit: 0n, - ccy: "USDC", + // The anchor has been paid from here on, so nothing below may turn this into an error: a caller + // that saw one would retry, get 409, and the payout would be paid but never booked. Marking the + // row `sent` and booking it share one transaction, so `sent` always means booked. + const booking: PayoutBooking = { + fromTxHash: args.fromContractPayoutTxHash, + txHash: hash, + refId, + quoteId, + amount: args.amount, + memo: args.memo, + }; + try { + await db.transaction(async (trx) => { + await trx + .update(schema.payoutForward) + .set({ status: "sent", txHash: hash }) + .where(eq(schema.payoutForward.fromTxHash, args.fromContractPayoutTxHash)); + await bookPayout(trx, booking); }); - await recordJournalEntry(trx, { - refType: "payout_hop", - refId, - account: "operator:payout_hop", - debit: 0n, - credit: args.amount, - ccy: "USDC", + return { hash, status, booked: true }; + } catch (err) { + await markUnbooked(db, alerter, booking, err); + return { hash, status, booked: false }; + } +} + +/** + * Writes the bookkeeping for a sent payout: the `exit_fill` row when there is a quote, and the + * balanced journal pair under `refId`. Call inside a transaction. + */ +export async function bookPayout(trx: DbOrTx, booking: PayoutBooking): Promise { + // `exit_fill.quote_id` is NOT NULL in the current schema and there's no generic "payout" table + // for non-exit flows (a credit-line draw routed to an anchor). When there's a real exit_quote + // to point at, record the exit_fill row; otherwise rely on the journal_entry rows below plus + // `payout_forward` for traceability. + if (booking.quoteId) { + await trx.insert(schema.exitFill).values({ + quoteId: booking.quoteId, + txHash: booking.txHash, + payoutRoute: "anchor_hop", + payoutMemo: booking.memo, }); + } + + // A balanced pair, not a lone credit row: cash leaving Ballast's operator account (debit) + // funds the payout booked against the payout-hop account (credit). Both legs go through + // `recordJournalEntry` directly (not `recordDoubleEntry`, which opens its own transaction) + // since this is already inside one. + await recordJournalEntry(trx, { + refType: "payout_hop", + refId: booking.refId, + account: "usdc_cash", + debit: booking.amount, + credit: 0n, + ccy: "USDC", }); + await recordJournalEntry(trx, { + refType: "payout_hop", + refId: booking.refId, + account: "operator:payout_hop", + debit: 0n, + credit: booking.amount, + ccy: "USDC", + }); +} - return { hash, status }; +/** + * Flags a sent-but-unbooked payout for `bookPendingPayouts` and alerts. If even the flag cannot be + * written (the database is down), the row stays `claimed` with no tx hash; the alert carries the + * hash so someone can finish it by hand. + */ +async function markUnbooked(db: Database, alerter: Alerter, booking: PayoutBooking, cause: unknown): Promise { + const reason = cause instanceof Error ? cause.message : String(cause); + let flagged = true; + try { + await db + .update(schema.payoutForward) + .set({ status: "bookkeeping_pending", txHash: booking.txHash }) + .where(eq(schema.payoutForward.fromTxHash, booking.fromTxHash)); + } catch { + flagged = false; + } + alerter.fire("payout_unbooked", { + subject: booking.fromTxHash, + txHash: booking.txHash, + refId: booking.refId, + flagged, + reason: flagged + ? `anchor paid (tx ${booking.txHash}) but booking failed: ${reason}; left bookkeeping_pending for repair` + : `anchor paid (tx ${booking.txHash}) but booking failed: ${reason}; could not flag the row either, it is still claimed`, + }); } diff --git a/services/payout-hop/src/index.ts b/services/payout-hop/src/index.ts index f4dd7d5..94cab82 100644 --- a/services/payout-hop/src/index.ts +++ b/services/payout-hop/src/index.ts @@ -19,7 +19,7 @@ import { loadNetworkConfig, requireInternalToken, } from "@ballast/network-config"; -import { createLogger } from "@ballast/observability"; +import { createAlerter, createLogger, type Alerter } from "@ballast/observability"; import { DuplicatePayoutError, PayoutRefusedError, @@ -28,6 +28,7 @@ import { type PayoutPolicy, } from "./hop.js"; import { OperatorPayoutSigner, type PayoutSigner } from "./signer.js"; +import { bookPendingPayouts } from "./repair.js"; const forwardBody = z.object({ fromContractPayoutTxHash: z.string().min(1), @@ -42,7 +43,7 @@ const forwardBody = z.object({ export function buildServer( db: Database, signer: PayoutSigner, - options: { internalToken: string; policy: PayoutPolicy }, + options: { internalToken: string; policy: PayoutPolicy; alerter?: Alerter }, ): FastifyInstance { const app = Fastify({ logger: false }); requireInternalToken(app, options.internalToken); @@ -64,6 +65,7 @@ export function buildServer( quoteId: body.quoteId, }, options.policy, + options.alerter, ); return reply.send(result); } catch (err) { @@ -85,12 +87,28 @@ async function main() { const network = loadNetworkConfig(); const signer = await OperatorPayoutSigner.create(logger, network.rpcUrl, network.networkPassphrase); - const app = buildServer(db, signer, { internalToken, policy }); + const alerter = createAlerter("payout-hop"); + + const app = buildServer(db, signer, { internalToken, policy, alerter }); const port = Number(process.env.PAYOUT_HOP_PORT ?? 3002); const host = internalBindHost("payout-hop"); await app.listen({ port, host }); logger.info({ port, host, operator: signer.publicKey(), destinations: policy.allowedDestinations.size }, "payout-hop: listening"); + + // Book anything a previous run sent but could not record, then keep sweeping. + const repair = async () => { + try { + const result = await bookPendingPayouts(db, alerter); + if (result.booked.length || result.alreadyBooked.length || result.failed.length) { + logger.info(result, "payout-hop: repaired unbooked payouts"); + } + } catch (err) { + logger.error({ err }, "payout-hop: unbooked-payout repair failed"); + } + }; + await repair(); + setInterval(repair, Number(process.env.PAYOUT_HOP_REPAIR_INTERVAL_S ?? 300) * 1000).unref(); } if (process.env.VITEST !== "true") { diff --git a/services/payout-hop/src/repair.ts b/services/payout-hop/src/repair.ts new file mode 100644 index 0000000..50e2c0f --- /dev/null +++ b/services/payout-hop/src/repair.ts @@ -0,0 +1,78 @@ +/** + * Books payouts that were sent but whose bookkeeping failed (`payout_forward.status = + * 'bookkeeping_pending'`, see `forwardPayout`). Safe to run any number of times, from any number of + * processes: each row is taken by a conditional update inside its own transaction, and a row whose + * journal pair already exists (the original write committed but the acknowledgement was lost) is + * only marked `sent`, not booked again. + */ +import { and, eq } from "drizzle-orm"; +import { schema, type Database } from "@ballast/db"; +import { decimalStringToScaled } from "@ballast/domain-types"; +import type { Alerter } from "@ballast/observability"; +import { bookPayout } from "./hop.js"; + +export interface RepairResult { + /** First-leg hashes whose journal pair this run wrote. */ + booked: string[]; + /** First-leg hashes that turned out to be booked already; only their status was fixed. */ + alreadyBooked: string[]; + /** First-leg hashes that still could not be booked; they stay `bookkeeping_pending`. */ + failed: string[]; +} + +export async function bookPendingPayouts(db: Database, alerter: Alerter): Promise { + const result: RepairResult = { booked: [], alreadyBooked: [], failed: [] }; + const pending = await db + .select() + .from(schema.payoutForward) + .where(eq(schema.payoutForward.status, "bookkeeping_pending")); + + for (const row of pending) { + try { + const outcome = await db.transaction(async (trx) => { + // Taking the row first locks it, so a concurrent repair waits here and then finds it gone. + const taken = await trx + .update(schema.payoutForward) + .set({ status: "sent" }) + .where( + and( + eq(schema.payoutForward.fromTxHash, row.fromTxHash), + eq(schema.payoutForward.status, "bookkeeping_pending"), + ), + ) + .returning({ fromTxHash: schema.payoutForward.fromTxHash }); + if (taken.length === 0) return "taken"; + + const existing = await trx + .select({ id: schema.journalEntry.id }) + .from(schema.journalEntry) + .where(and(eq(schema.journalEntry.refType, "payout_hop"), eq(schema.journalEntry.refId, row.refId))) + .limit(1); + if (existing.length > 0) return "already"; + + // `markUnbooked` writes the hash together with the status, so this only trips on a hand edit. + if (!row.txHash) throw new Error("bookkeeping_pending row has no tx_hash"); + await bookPayout(trx, { + fromTxHash: row.fromTxHash, + txHash: row.txHash, + refId: row.refId, + quoteId: row.quoteId, + amount: decimalStringToScaled(row.amount), + memo: row.memo, + }); + return "booked"; + }); + if (outcome === "booked") result.booked.push(row.fromTxHash); + if (outcome === "already") result.alreadyBooked.push(row.fromTxHash); + } catch (err) { + result.failed.push(row.fromTxHash); + alerter.fire("payout_unbooked", { + subject: row.fromTxHash, + txHash: row.txHash, + refId: row.refId, + reason: `repair could not book it: ${err instanceof Error ? err.message : String(err)}`, + }); + } + } + return result; +}