From dfe4d1ad97f1886f1aad144b64232d0e1979802e Mon Sep 17 00:00:00 2001 From: jwinterm Date: Wed, 22 Jul 2026 12:04:19 -0400 Subject: [PATCH] feat: export confirmed payments as financial events --- src/index.js | 4 + src/migrations/041_financial_event_outbox.sql | 58 +++++ src/services/financialEventExporter.js | 221 ++++++++++++++++++ test/cloneMigrationRunner.test.js | 10 +- test/financialEventExporter.test.js | 76 ++++++ 5 files changed, 364 insertions(+), 5 deletions(-) create mode 100644 src/migrations/041_financial_event_outbox.sql create mode 100644 src/services/financialEventExporter.js create mode 100644 test/financialEventExporter.test.js diff --git a/src/index.js b/src/index.js index 421c1ce..fa892f9 100644 --- a/src/index.js +++ b/src/index.js @@ -29,6 +29,7 @@ const PayoutRetryService = require('./payments/payoutRetryService'); const AlertService = require('./services/alertService'); const HealthService = require('./services/healthService'); const LatePaymentReconciler = require('./services/latePaymentReconciler'); +const FinancialEventExporter = require('./services/financialEventExporter'); const GameModeManager = require('./game/gameModeManager'); const PaymentConfigManager = require('./config/paymentConfig'); const EnvironmentValidator = require('./config/environmentValidator'); @@ -76,6 +77,7 @@ broadcastManager.setDebugManager(debugManager); // Initialize payment system components (debugManager is now available) const databaseManager = new DatabaseManager(); const db = databaseManager; // Alias for convenience in API endpoints +const financialEventExporter = new FinancialEventExporter({ db: databaseManager }); const walletRPCService = new WalletRPCService(debugManager, { minConfirmations: paymentConfigManager.getConfig().payouts.processing.confirmations }); @@ -1138,6 +1140,7 @@ async function startServer() { await socketHandlers.initializeMatchMode(); financialRecoveryReady = true; payoutDispatchReady = financialRuntime.walletReady && isPayoutProcessingEnabled(); + financialEventExporter.start(); // Socket.io connection handler - ONLY after payment system is ready io.on('connection', function(socket) { @@ -1301,6 +1304,7 @@ async function gracefulShutdown(signal, exitCode = 0) { if (paymentExpiryInterval) clearInterval(paymentExpiryInterval); if (walletHealthInterval) clearInterval(walletHealthInterval); if (databaseHealthInterval) clearInterval(databaseHealthInterval); + financialEventExporter.stop(); // Terminal solo games are held in a serialized settlement-pending state when their atomic // completion transaction fails. Drain those retries before disabling GameModeManager or diff --git a/src/migrations/041_financial_event_outbox.sql b/src/migrations/041_financial_event_outbox.sql new file mode 100644 index 0000000..2adf96a --- /dev/null +++ b/src/migrations/041_financial_event_outbox.sql @@ -0,0 +1,58 @@ +-- Durable, backend-neutral accounting event export. Payment confirmation never waits for the +-- accounting receiver: this trigger only appends an outbox identity inside the payment transaction. + +CREATE TABLE IF NOT EXISTS financial_event_outbox ( + id BIGSERIAL PRIMARY KEY, + event_kind VARCHAR(40) NOT NULL, + aggregate_id INTEGER NOT NULL REFERENCES payments(id) ON DELETE RESTRICT, + status VARCHAR(20) NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'in_flight', 'delivered')), + attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0), + next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + lease_until TIMESTAMPTZ, + delivered_at TIMESTAMPTZ, + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT financial_event_outbox_once UNIQUE (event_kind, aggregate_id), + CONSTRAINT financial_event_outbox_state CHECK ( + (status = 'pending' AND lease_until IS NULL AND delivered_at IS NULL) + OR (status = 'in_flight' AND lease_until IS NOT NULL AND delivered_at IS NULL) + OR (status = 'delivered' AND lease_until IS NULL AND delivered_at IS NOT NULL) + ) +); + +CREATE INDEX IF NOT EXISTS idx_financial_event_outbox_due + ON financial_event_outbox(next_attempt_at, created_at) + WHERE delivered_at IS NULL; + +CREATE OR REPLACE FUNCTION enqueue_confirmed_payment_financial_event() +RETURNS TRIGGER AS $$ +BEGIN + IF NEW.status = 'confirmed' + AND COALESCE(NEW.received_amount, 0) > 0 + AND ( + TG_OP = 'INSERT' + OR OLD.status IS DISTINCT FROM 'confirmed' + OR COALESCE(OLD.received_amount, 0) <= 0 + ) THEN + INSERT INTO financial_event_outbox (event_kind, aggregate_id) + VALUES ('payment.confirmed', NEW.id) + ON CONFLICT (event_kind, aggregate_id) DO NOTHING; + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql; + +DROP TRIGGER IF EXISTS trg_enqueue_confirmed_payment_financial_event ON payments; +CREATE TRIGGER trg_enqueue_confirmed_payment_financial_event +AFTER INSERT OR UPDATE OF status, received_amount ON payments +FOR EACH ROW EXECUTE FUNCTION enqueue_confirmed_payment_financial_event(); + +-- Backfill only receipt-valued rows. Legacy confirmed rows with received_amount=0 lack adequate +-- economic evidence and must not be manufactured into revenue events. +INSERT INTO financial_event_outbox (event_kind, aggregate_id) +SELECT 'payment.confirmed', id +FROM payments +WHERE status = 'confirmed' AND COALESCE(received_amount, 0) > 0 +ON CONFLICT (event_kind, aggregate_id) DO NOTHING; diff --git a/src/services/financialEventExporter.js b/src/services/financialEventExporter.js new file mode 100644 index 0000000..decabfe --- /dev/null +++ b/src/services/financialEventExporter.js @@ -0,0 +1,221 @@ +const { getProfile } = require('../chain/chainProfile'); +const money = require('../money/atomic'); + +const PLACEHOLDER = /(change[_-]?me|replace[_-]?with|example|password|secret)/i; + +function boundedInteger(value, fallback, min, max) { + const parsed = Number.parseInt(String(value ?? ''), 10); + return Number.isInteger(parsed) ? Math.max(min, Math.min(max, parsed)) : fallback; +} + +function isoTimestamp(value) { + const date = value instanceof Date ? value : new Date(value); + if (Number.isNaN(date.getTime())) throw new Error('Confirmed payment has no valid timestamp'); + return date.toISOString(); +} + +function receiptProvenance(receipts, paymentId) { + const rows = Array.isArray(receipts) ? receipts : []; + const normalized = rows.map((receipt) => ({ + kind: receipt.evidence_type === 'chain_output' ? 'monero-output' : 'provider-invoice', + ref: String(receipt.evidence_id || '').slice(0, 500) + })).filter((item) => item.ref); + return normalized.length > 0 + ? normalized + : [{ kind: 'wowngeon-payment', ref: String(paymentId) }]; +} + +function buildFinancialEvent(payment, env = process.env) { + if (!payment || payment.id == null) throw new Error('Payment identity is required'); + const profile = getProfile(env.CRYPTO_TYPE || 'WOW'); + const atomic = money.toBig(payment.received_amount || payment.expected_amount || 0); + if (atomic <= 0n) throw new Error('Confirmed payment has no positive received amount'); + const networkName = profile.id === 'WOW' + ? 'wownero:mainnet' + : `monero:${String(env.MONERO_NETWORK || 'mainnet').toLowerCase()}`; + const accountRef = String(env.FINANCIAL_EVENT_ACCOUNT_REF || 'wowngeon:receipts').trim(); + if (!accountRef || accountRef.length > 300) throw new Error('Invalid financial event account reference'); + + return { + schema: 'financial-event/v1', + id: `payment-${payment.id}`, + occurred_at: isoTimestamp(payment.confirmed_at), + producer: 'wowngeon', + product: 'wowngeon', + activity: 'customer-payment', + account_refs: [accountRef], + provenance: receiptProvenance(payment.receipts, payment.id), + amounts: {}, + legs: [{ + asset: profile.symbol, + quantity: money.format(atomic, profile.decimals), + direction: 'in', + network: networkName, + account_ref: accountRef, + usd_value: '0', + price_source: '', + price_confidence: '' + }], + memo: 'Confirmed customer payment; USD valuation is required before accounting confirmation.', + attributes: { + valuation_required: true, + atomic_amount: atomic.toString(), + atomic_decimals: profile.decimals, + payment_type: String(payment.payment_type || ''), + product_id: String(payment.product_id || ''), + provider_id: String(payment.provider_id || ''), + receipt_count: Array.isArray(payment.receipts) ? payment.receipts.length : 0 + } + }; +} + +class FinancialEventExporter { + constructor({ db, env = process.env, fetchFn = global.fetch, logger = console } = {}) { + if (!db) throw new Error('FinancialEventExporter requires db'); + this.db = db; + this.env = env; + this.fetchFn = fetchFn; + this.logger = logger; + this.endpoint = String(env.FINANCIAL_EVENT_SINK_URL || '').trim(); + this.token = String(env.FINANCIAL_EVENT_SINK_TOKEN || '').trim(); + this.enabled = Boolean(this.endpoint || this.token); + this.intervalMs = boundedInteger(env.FINANCIAL_EVENT_POLL_MS, 60000, 5000, 3600000); + this.batchSize = boundedInteger(env.FINANCIAL_EVENT_BATCH_SIZE, 20, 1, 100); + this.running = false; + this.timer = null; + + if (Boolean(this.endpoint) !== Boolean(this.token)) { + throw new Error('Financial event sink URL and token must be configured together'); + } + if (!this.enabled) return; + if (typeof this.fetchFn !== 'function') throw new Error('Financial event HTTP transport is unavailable'); + let parsed; + try { parsed = new URL(this.endpoint); } catch (_) { throw new Error('Financial event sink URL is invalid'); } + if (!['http:', 'https:'].includes(parsed.protocol) + || (String(env.NODE_ENV).toLowerCase() === 'production' && parsed.protocol !== 'https:')) { + throw new Error('Financial event sink must use HTTPS in production'); + } + if (this.token.length < 32 || PLACEHOLDER.test(this.token)) { + throw new Error('Financial event sink token must be a strong non-placeholder secret'); + } + const profile = getProfile(env.CRYPTO_TYPE || 'WOW'); + const network = String(env.MONERO_NETWORK || 'mainnet').toLowerCase(); + if ((profile.id === 'XMR' && network !== 'mainnet') || profile.id === 'WOW' && network !== 'mainnet') { + throw new Error('Testnet and stagenet payments must not be exported as financial events'); + } + } + + async _claim() { + const result = await this.db.query(` + UPDATE financial_event_outbox + SET status = 'in_flight', attempts = attempts + 1, + lease_until = NOW() + INTERVAL '5 minutes', updated_at = NOW() + WHERE id = ( + SELECT id FROM financial_event_outbox + WHERE delivered_at IS NULL + AND next_attempt_at <= NOW() + AND (status = 'pending' OR (status = 'in_flight' AND lease_until < NOW())) + ORDER BY created_at, id + FOR UPDATE SKIP LOCKED + LIMIT 1 + ) + RETURNING id, aggregate_id, attempts + `); + return result.rows[0] || null; + } + + async _loadPayment(paymentId) { + const result = await this.db.query(` + SELECT p.id, p.payment_type, p.expected_amount, p.received_amount, + p.confirmed_at, p.provider_id, p.product_id, + COALESCE(json_agg(json_build_object( + 'evidence_type', r.evidence_type, + 'evidence_id', r.evidence_id + ) ORDER BY r.id) FILTER (WHERE r.id IS NOT NULL), '[]'::json) AS receipts + FROM payments p + LEFT JOIN payment_receipts r ON r.payment_id = p.id AND r.confirmed = TRUE + WHERE p.id = $1 AND p.status = 'confirmed' + GROUP BY p.id + `, [paymentId]); + if (!result.rows[0]) throw new Error('Outbox payment is missing or no longer confirmed'); + return result.rows[0]; + } + + async _deliver(claim) { + const payment = await this._loadPayment(claim.aggregate_id); + const payload = buildFinancialEvent(payment, this.env); + const response = await this.fetchFn(this.endpoint, { + method: 'POST', + headers: { + authorization: `Bearer ${this.token}`, + 'content-type': 'application/json' + }, + body: JSON.stringify(payload), + redirect: 'manual', + signal: AbortSignal.timeout(10000) + }); + if (!response || !response.ok) { + throw new Error(`Financial event sink returned HTTP ${response?.status || 0}`); + } + await this.db.query(` + UPDATE financial_event_outbox + SET status = 'delivered', delivered_at = NOW(), lease_until = NULL, + last_error = NULL, updated_at = NOW() + WHERE id = $1 + `, [claim.id]); + } + + async _retry(claim, error) { + const seconds = Math.min(3600, 30 * (2 ** Math.min(Number(claim.attempts || 1) - 1, 7))); + const message = String(error?.message || 'Financial event delivery failed').slice(0, 500); + await this.db.query(` + UPDATE financial_event_outbox + SET status = 'pending', lease_until = NULL, + next_attempt_at = NOW() + ($2::integer * INTERVAL '1 second'), + last_error = $3, updated_at = NOW() + WHERE id = $1 + `, [claim.id, seconds, message]); + this.logger.warn?.(`[FinancialEvent] Delivery ${claim.id} deferred: ${message}`); + } + + async runOnce() { + if (!this.enabled || this.running) return { skipped: true, delivered: 0, deferred: 0 }; + this.running = true; + let delivered = 0; + let deferred = 0; + try { + for (let index = 0; index < this.batchSize; index += 1) { + const claim = await this._claim(); + if (!claim) break; + try { + await this._deliver(claim); + delivered += 1; + } catch (error) { + await this._retry(claim, error); + deferred += 1; + } + } + return { skipped: false, delivered, deferred }; + } finally { + this.running = false; + } + } + + start() { + if (!this.enabled || this.timer) return false; + this.runOnce().catch((error) => this.logger.warn?.(`[FinancialEvent] Export tick failed: ${error.message}`)); + this.timer = setInterval(() => { + this.runOnce().catch((error) => this.logger.warn?.(`[FinancialEvent] Export tick failed: ${error.message}`)); + }, this.intervalMs); + this.timer.unref?.(); + return true; + } + + stop() { + if (this.timer) clearInterval(this.timer); + this.timer = null; + } +} + +module.exports = FinancialEventExporter; +module.exports.buildFinancialEvent = buildFinancialEvent; diff --git a/test/cloneMigrationRunner.test.js b/test/cloneMigrationRunner.test.js index 650097b..cce0d53 100644 --- a/test/cloneMigrationRunner.test.js +++ b/test/cloneMigrationRunner.test.js @@ -161,11 +161,11 @@ describe('disposable clone migration execution', () => { logSpy.mockRestore(); }); - test('discovers the contiguous repository manifest through migration 40 dynamically', () => { + test('discovers the contiguous repository manifest through migration 41 dynamically', () => { const manifest = discoverMigrationManifest(); expect(manifest.latestOrdinal).toBeGreaterThanOrEqual(MINIMUM_RELEASE_MIGRATION); - expect(manifest.latestFilename).toBe('040_paid_match_entropy_precommit.sql'); - expect(manifest.files).toHaveLength(40); + expect(manifest.latestFilename).toBe('041_financial_event_outbox.sql'); + expect(manifest.files).toHaveLength(41); }); test('runs the normal ordered migration implementation and proves the exact ledger', async () => { @@ -185,9 +185,9 @@ describe('disposable clone migration execution', () => { }); expect(pool.transactions).toEqual(result.files); - expect(result.latestFilename).toBe('040_paid_match_entropy_precommit.sql'); + expect(result.latestFilename).toBe('041_financial_event_outbox.sql'); expect(output).toContain('clone_migration_status=ok'); - expect(output).toContain('latest_migration=040_paid_match_entropy_precommit.sql'); + expect(output).toContain('latest_migration=041_financial_event_outbox.sql'); expect(output.join('\n')).not.toContain(secret); expect(pool.config).toEqual(expect.objectContaining({ database: env.DB_NAME, diff --git a/test/financialEventExporter.test.js b/test/financialEventExporter.test.js new file mode 100644 index 0000000..833cf96 --- /dev/null +++ b/test/financialEventExporter.test.js @@ -0,0 +1,76 @@ +const FinancialEventExporter = require('../src/services/financialEventExporter'); +const { buildFinancialEvent } = FinancialEventExporter; + +const ENV = { + NODE_ENV: 'production', + CRYPTO_TYPE: 'WOW', + MONERO_NETWORK: 'mainnet', + FINANCIAL_EVENT_SINK_URL: 'https://accounting.example.test/api/events/wowngeon', + FINANCIAL_EVENT_SINK_TOKEN: 'a'.repeat(64), + FINANCIAL_EVENT_ACCOUNT_REF: 'wowngeon:receipts' +}; + +const PAYMENT = { + id: 22, + payment_type: 'credits_package', + expected_amount: '900000000000', + received_amount: '900000000000', + confirmed_at: '2026-07-22T14:30:00Z', + provider_id: 'native-monero', + product_id: 'small', + receipts: [{ evidence_type: 'chain_output', evidence_id: `${'a'.repeat(64)}:${'b'.repeat(64)}` }] +}; + +describe('financial event exporter', () => { + test('builds a backend-neutral, unpriced WOW receipt without customer data', () => { + const event = buildFinancialEvent(PAYMENT, ENV); + + expect(event).toEqual(expect.objectContaining({ + schema: 'financial-event/v1', + id: 'payment-22', + producer: 'wowngeon', + product: 'wowngeon', + activity: 'customer-payment', + account_refs: ['wowngeon:receipts'], + amounts: {}, + attributes: expect.objectContaining({ + valuation_required: true, + atomic_amount: '900000000000', + atomic_decimals: 11 + }) + })); + expect(event.legs[0]).toEqual(expect.objectContaining({ + asset: 'WOW', quantity: '9', direction: 'in', usd_value: '0' + })); + expect(JSON.stringify(event)).not.toMatch(/user|email|address_index/i); + }); + + test('claims and delivers one durable outbox row', async () => { + const db = { + query: jest.fn(async (sql) => { + if (/RETURNING id, aggregate_id, attempts/i.test(sql)) { + return db.claimed ? { rows: [] } : (db.claimed = true, { rows: [{ id: 5, aggregate_id: 22, attempts: 1 }] }); + } + if (/FROM payments p/i.test(sql)) return { rows: [PAYMENT] }; + return { rows: [] }; + }) + }; + const fetchFn = jest.fn(async () => ({ ok: true, status: 201 })); + const exporter = new FinancialEventExporter({ db, env: ENV, fetchFn, logger: { warn: jest.fn() } }); + + await expect(exporter.runOnce()).resolves.toEqual({ skipped: false, delivered: 1, deferred: 0 }); + expect(fetchFn).toHaveBeenCalledWith(ENV.FINANCIAL_EVENT_SINK_URL, expect.objectContaining({ + method: 'POST', + headers: expect.objectContaining({ authorization: `Bearer ${ENV.FINANCIAL_EVENT_SINK_TOKEN}` }) + })); + expect(db.query.mock.calls.some(([sql]) => /SET status = 'delivered'/i.test(sql))).toBe(true); + }); + + test('refuses to export stagenet value into production accounting', () => { + expect(() => new FinancialEventExporter({ + db: { query: jest.fn() }, + env: { ...ENV, CRYPTO_TYPE: 'XMR', MONERO_NETWORK: 'stagenet' }, + fetchFn: jest.fn() + })).toThrow(/must not be exported/); + }); +});