From ed9372560cbb57fc2b6b82671befc9c4bae26fca Mon Sep 17 00:00:00 2001 From: woahwhattheheck Date: Sun, 4 Oct 2026 08:57:19 -0400 Subject: [PATCH 1/3] Preserve seller webhook UI and server integration [skip ci] --- backend/env.example.txt | 9 + backend/env.mvp.example | 9 + backend/src/observability/log-events.ts | 1 + backend/src/routes/invoice.routes.ts | 2 + backend/src/server-dual.ts | 8 + backend/src/server-mvp.ts | 15 +- backend/src/server.ts | 11 +- .../tests/fixtures/webhook-postgres-worker.ts | 112 +++++++ .../webhook-postgres.integration.test.ts | 316 ++++++++++++++++++ frontend/app/dashboard/page.tsx | 2 + frontend/components/SellerWebhooksPanel.tsx | 315 +++++++++++++++++ frontend/lib/webhook-api.ts | 183 ++++++++++ frontend/tests/seller-webhooks.test.js | 238 +++++++++++++ frontend/tests/support/stubs/freighter-api.js | 3 +- shared/webhooks.ts | 52 +++ 15 files changed, 1271 insertions(+), 5 deletions(-) create mode 100644 backend/tests/fixtures/webhook-postgres-worker.ts create mode 100644 backend/tests/webhook-postgres.integration.test.ts create mode 100644 frontend/components/SellerWebhooksPanel.tsx create mode 100644 frontend/lib/webhook-api.ts create mode 100644 frontend/tests/seller-webhooks.test.js create mode 100644 shared/webhooks.ts diff --git a/backend/env.example.txt b/backend/env.example.txt index 0367ffb4..1ce2e60f 100644 --- a/backend/env.example.txt +++ b/backend/env.example.txt @@ -104,3 +104,12 @@ VERIFY_CONCURRENCY_RETRY_AFTER_SECONDS=5 # Require cryptographic signatures for cancel (production always requires). REQUIRE_SIGNATURES=false + +# Signed seller webhooks. Run db:migrate before enabling PostgreSQL delivery. +# MVP endpoints/outbox are volatile; use PostgreSQL for durable delivery. +# Generate a stable server-only encryption key with: openssl rand -hex 32 +# Share the key across replicas; never expose it with NEXT_PUBLIC. +WEBHOOKS_ENABLED=false +WEBHOOK_ENCRYPTION_KEY= +WEBHOOK_MAX_ATTEMPTS=8 +WEBHOOK_DISABLE_AFTER_FAILURES=10 diff --git a/backend/env.mvp.example b/backend/env.mvp.example index 69594946..57388839 100644 --- a/backend/env.mvp.example +++ b/backend/env.mvp.example @@ -100,3 +100,12 @@ REQUIRE_SIGNATURES=false # Redis connection for verify-cache (optional). Rate limiters use in-memory. # Falls back to in-memory cache if not configured. REDIS_URL= + +# Signed seller webhooks. Run db:migrate before enabling PostgreSQL delivery. +# MVP endpoints/outbox are volatile; use PostgreSQL for durable delivery. +# Generate a stable server-only encryption key with: openssl rand -hex 32 +# Share the key across replicas; never expose it with NEXT_PUBLIC. +WEBHOOKS_ENABLED=false +WEBHOOK_ENCRYPTION_KEY= +WEBHOOK_MAX_ATTEMPTS=8 +WEBHOOK_DISABLE_AFTER_FAILURES=10 diff --git a/backend/src/observability/log-events.ts b/backend/src/observability/log-events.ts index 44e54de0..adc3c7de 100644 --- a/backend/src/observability/log-events.ts +++ b/backend/src/observability/log-events.ts @@ -50,6 +50,7 @@ const FAILURE_OPERATIONS = [ 'stellar.streamPayment', 'stellar.payments', 'stellar.submit', 'cache.connect', 'cache.get', 'cache.set', 'cache.clear', 'cache.verify', 'proof.invariants', 'proof.json', 'proof.pdf', 'proof.handoff', + 'webhook.worker', 'webhook.manage', ] as const; type FailureOperation = typeof FAILURE_OPERATIONS[number]; diff --git a/backend/src/routes/invoice.routes.ts b/backend/src/routes/invoice.routes.ts index 8b062058..4ab3c163 100644 --- a/backend/src/routes/invoice.routes.ts +++ b/backend/src/routes/invoice.routes.ts @@ -13,6 +13,7 @@ import { } from '../middleware/rate-limit'; import { createInvoiceCeilingMiddleware } from '../middleware/invoice-ceiling'; import { createVerifyCacheMiddleware, verificationCache } from '../middleware/verify-cache'; +import { createWebhookRouter } from './webhook.routes'; export interface InvoiceRouterOptions extends InvoiceHandlerOptions { enableRateLimiting?: boolean; @@ -48,6 +49,7 @@ export function createInvoiceRouter(options: InvoiceRouterOptions): Router { const handlers = createInvoiceHandlers(options); const router = Router(); router.use(requestCorrelationMiddleware); + if (options.storage.webhooks) router.use(createWebhookRouter(options.storage.webhooks)); const enableRateLimiting = options.enableRateLimiting ?? diff --git a/backend/src/server-dual.ts b/backend/src/server-dual.ts index 4021decd..6873cc18 100644 --- a/backend/src/server-dual.ts +++ b/backend/src/server-dual.ts @@ -38,6 +38,7 @@ import { configuredFrontendOrigins, configuredStorageMode, corsOptions } from '. import { healthHandler, readinessHandler } from './health'; import { bodyLimitErrorHandler } from './middleware/body-limit'; import { getEdgeControlConfig } from './middleware/edge-config'; +import { startConfiguredWebhookWorker, type WebhookWorker } from './services/webhook-worker'; import type { InvoiceStorage } from './storage/invoice-storage'; dotenv.config(); @@ -74,6 +75,7 @@ paymentMonitorService.configure({ const app: Application = express(); const PORT = process.env.PORT || 3001; const maxBodyBytes = getEdgeControlConfig().maxBodyBytes; +let webhookWorker: WebhookWorker | undefined; app.use(cors(corsOptions())); app.use(requestCorrelationMiddleware); @@ -165,6 +167,8 @@ async function initialize(): Promise { * configured one. Importing this module never starts a server on its own. */ export function startServer(port: number | string = PORT) { + const worker = startConfiguredWebhookWorker(storage.webhooks); + webhookWorker = worker; const server = app.listen(port, async () => { await initialize(); const modeLabel = storageMode === 'postgres' ? 'PostgreSQL' : 'In-Memory (MVP)'; @@ -179,6 +183,8 @@ export function startServer(port: number | string = PORT) { console.log(`๐ŸŒ Frontends: ${configuredFrontendOrigins().join(', ') || 'not configured'}`); console.log('โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”\n'); }); + server.once('close', () => { void worker?.stop(); }); + server.once('error', () => { void worker?.stop(); }); return server; } @@ -188,6 +194,7 @@ if (/server-dual(\.[cm]?[jt]s)?$/.test(entryPoint)) { } process.on('SIGTERM', async () => { + await webhookWorker?.stop(); paymentMonitorService.stop(); if (storageMode === 'postgres') { await pool.end(); @@ -196,6 +203,7 @@ process.on('SIGTERM', async () => { }); process.on('SIGINT', async () => { + await webhookWorker?.stop(); paymentMonitorService.stop(); if (storageMode === 'postgres') { await pool.end(); diff --git a/backend/src/server-mvp.ts b/backend/src/server-mvp.ts index ea5555d0..c43bbaf0 100644 --- a/backend/src/server-mvp.ts +++ b/backend/src/server-mvp.ts @@ -21,6 +21,7 @@ import { configuredFrontendOrigins, corsOptions } from './config/runtime'; import { healthHandler, readinessHandler } from './health'; import { bodyLimitErrorHandler } from './middleware/body-limit'; import { getEdgeControlConfig } from './middleware/edge-config'; +import { startConfiguredWebhookWorker, type WebhookWorker } from './services/webhook-worker'; dotenv.config(); @@ -37,6 +38,7 @@ paymentMonitorService.configure({ const app: Application = express(); const PORT = process.env.PORT || 3001; const maxBodyBytes = getEdgeControlConfig().maxBodyBytes; +let webhookWorker: WebhookWorker | undefined; // Middleware app.use(cors(corsOptions())); @@ -119,6 +121,8 @@ app.use((req: Request, res: Response) => { * configured one, and so importing this module never starts a server. */ export function startServer(port: number | string = PORT) { + const worker = startConfiguredWebhookWorker(memoryInvoiceStorage.webhooks); + webhookWorker = worker; if (SELLER_PUBLIC_KEY) { try { paymentMonitorService.start(); @@ -127,7 +131,7 @@ export function startServer(port: number | string = PORT) { } } - return app.listen(port, () => { + const server = app.listen(port, () => { console.log('\n๐Ÿš€ Quittance Backend (MVP Mode)'); console.log('โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”'); console.log(`โœ… Server running on port ${port}`); @@ -138,6 +142,9 @@ export function startServer(port: number | string = PORT) { console.log(`๐ŸŒ Frontends: ${configuredFrontendOrigins().join(', ') || 'not configured'}`); console.log('โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”\n'); }); + server.once('close', () => { void worker?.stop(); }); + server.once('error', () => { void worker?.stop(); }); + return server; } const entryPoint = process.argv[1] ?? ''; @@ -145,12 +152,14 @@ if (/server-mvp(\.[cm]?[jt]s)?$/.test(entryPoint)) { startServer(); } -process.on('SIGTERM', () => { +process.on('SIGTERM', async () => { + await webhookWorker?.stop(); paymentMonitorService.stop(); process.exit(0); }); -process.on('SIGINT', () => { +process.on('SIGINT', async () => { + await webhookWorker?.stop(); paymentMonitorService.stop(); process.exit(0); }); diff --git a/backend/src/server.ts b/backend/src/server.ts index 78dfbd54..07fefcaa 100644 --- a/backend/src/server.ts +++ b/backend/src/server.ts @@ -13,12 +13,14 @@ import postgresInvoiceStorage from './storage/postgres-invoice-storage'; import { healthHandler, readinessHandler } from './health'; import { bodyLimitErrorHandler } from './middleware/body-limit'; import { getEdgeControlConfig } from './middleware/edge-config'; +import { startConfiguredWebhookWorker, type WebhookWorker } from './services/webhook-worker'; dotenv.config(); const app: Application = express(); const PORT = process.env.PORT || 3001; const maxBodyBytes = getEdgeControlConfig().maxBodyBytes; +let webhookWorker: WebhookWorker | undefined; app.use(cors(corsOptions())); app.use(requestCorrelationMiddleware); @@ -86,7 +88,9 @@ async function initialize() { } export function startServer(port: number | string = PORT) { - return app.listen(port, async () => { + const worker = startConfiguredWebhookWorker(postgresInvoiceStorage.webhooks); + webhookWorker = worker; + const server = app.listen(port, async () => { await initialize(); console.log('\n๐Ÿš€ Quittance Backend (Postgres Mode)'); console.log('โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”'); @@ -98,6 +102,9 @@ export function startServer(port: number | string = PORT) { console.log(`๐ŸŒ Frontends: ${configuredFrontendOrigins().join(', ') || 'not configured'}`); console.log('โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”โ”\n'); }); + server.once('close', () => { void worker?.stop(); }); + server.once('error', () => { void worker?.stop(); }); + return server; } const entryPoint = process.argv[1] ?? ''; @@ -106,6 +113,7 @@ if (/server(\.[cm]?[jt]s)?$/.test(entryPoint)) { } process.on('SIGTERM', async () => { + await webhookWorker?.stop(); console.log('Shutting down...'); paymentMonitorService.stop(); await pool.end(); @@ -113,6 +121,7 @@ process.on('SIGTERM', async () => { }); process.on('SIGINT', async () => { + await webhookWorker?.stop(); console.log('Shutting down...'); paymentMonitorService.stop(); await pool.end(); diff --git a/backend/tests/fixtures/webhook-postgres-worker.ts b/backend/tests/fixtures/webhook-postgres-worker.ts new file mode 100644 index 00000000..fe902fed --- /dev/null +++ b/backend/tests/fixtures/webhook-postgres-worker.ts @@ -0,0 +1,112 @@ +import { Pool } from 'pg'; +import { PostgresWebhookStorage } from '../../src/storage/postgres-webhook-storage'; + +// One real database transaction per child. The parent controls only when a +// claimed delivery may finish; no webhook request or payload crosses IPC. +let pool: Pool | undefined; +let closing: Promise | undefined; +let running = false; +let finished = false; +let releaseHold: (() => void) | undefined; + +function send(message: Record): Promise { + return new Promise((resolve, reject) => { + if (!process.send || !process.connected) { + reject(Object.assign(new Error('Worker requires IPC'), { code: 'IPC_UNAVAILABLE' })); + return; + } + process.send(message, error => error ? reject(error) : resolve()); + }); +} + +function closePool(): Promise { + return closing ??= pool ? pool.end() : Promise.resolve(); +} + +async function fail(error: unknown): Promise { + if (finished) return; + finished = true; + process.exitCode = 1; + const code = (error as { code?: unknown } | null)?.code; + // Database messages can contain connection details or row values. Report + // only a stable explanation and the diagnostic code, never the raw error. + await send({ + type: 'error', + message: 'PostgreSQL webhook worker failed', + ...(typeof code === 'string' && /^[A-Z0-9_]{1,40}$/.test(code) ? { code } : {}), + }).catch(() => {}); + await closePool().catch(() => {}); + if (process.connected) process.disconnect(); +} + +async function main(): Promise { + const schema = process.env.Q584_PG_SCHEMA; + const connectionString = process.env.DATABASE_URL; + if (!schema || !/^q584_webhooks_[a-f0-9]+$/.test(schema) || !connectionString) { + throw Object.assign(new Error('Invalid worker configuration'), { code: 'INVALID_WORKER_CONFIG' }); + } + pool = new Pool({ + connectionString, + max: 1, + options: `-c search_path=${schema}`, + application_name: `q584-worker-${process.pid}`, + }); + pool.on('error', error => { void fail(error); }); + const storage = new PostgresWebhookStorage(pool); + + async function run(command: Record): Promise { + const mode = command.mode; + const rawNow = command.now; + if (typeof rawNow !== 'string' || (mode !== 'hold' && mode !== 'finish')) { + throw Object.assign(new Error('Invalid run command'), { code: 'INVALID_WORKER_COMMAND' }); + } + const now = new Date(rawNow); + if (!Number.isFinite(now.getTime())) { + throw Object.assign(new Error('Invalid run time'), { code: 'INVALID_WORKER_COMMAND' }); + } + const timestamp = now.toISOString(); + const processed = await storage.processNext(now, async (delivery, _endpoint) => { + // Register release before announcing the claim: the parent may reply + // immediately, before the send callback has completed in this child. + const held = mode === 'hold' + ? new Promise(resolve => { releaseHold = resolve; }) + : undefined; + await send({ + type: 'claimed', pid: process.pid, id: delivery.id, + eventId: delivery.eventId, attempt: delivery.attempt, + }); + if (held) await held; + releaseHold = undefined; + return { + attempt: delivery.attempt + 1, + status: 'delivered', + nextAttemptAt: timestamp, + lastResponseCode: 200, + completedAt: timestamp, + failureCount: 0, + disableEndpoint: false, + }; + }); + // processNext returns only after COMMIT. Close the connection before the + // receipt so a parent can inspect durable state without a lingering pool. + await closePool(); + await send({ type: 'done', pid: process.pid, processed }); + finished = true; + if (process.connected) process.disconnect(); + } + + process.on('message', message => { + if (finished || !message || typeof message !== 'object') return; + const command = message as Record; + if (command.type === 'release') { + releaseHold?.(); + return; + } + if (command.type !== 'run' || running) return; + running = true; + void run(command).catch(fail); + }); + await send({ type: 'ready', pid: process.pid }); +} + +void main().catch(fail); diff --git a/backend/tests/webhook-postgres.integration.test.ts b/backend/tests/webhook-postgres.integration.test.ts new file mode 100644 index 00000000..afefb61e --- /dev/null +++ b/backend/tests/webhook-postgres.integration.test.ts @@ -0,0 +1,316 @@ +import assert from 'node:assert/strict'; +import { after, before, beforeEach, describe, it, type TestContext } from 'node:test'; +import { randomUUID } from 'node:crypto'; +import { readFileSync } from 'node:fs'; +import path from 'node:path'; +import { fork, type ChildProcess } from 'node:child_process'; +import { setTimeout as delay } from 'node:timers/promises'; +import { Pool } from 'pg'; +import { InvoiceService } from '../src/services/invoice.service'; +import { PostgresWebhookStorage } from '../src/storage/postgres-webhook-storage'; +import { sealWebhookSecret } from '../src/services/webhook-crypto'; +import { WEBHOOK_EVENT_TYPES, type WebhookEventType } from '../../shared/webhooks'; +import type { WebhookEndpoint } from '../src/storage/webhook-storage'; + +const DATABASE_URL = process.env.DATABASE_URL; +const SELLER = 'GAYF33NNNMI2Z6VNRFXQ64D4E4SF77PM46NW3ZUZEEU5X7FCHAZCMHKU'; +const PAYER = 'GCT7D6S5VTFGEURS6ZYIO33YZRPQMA3LNWB4GEOHDFDXZGWTA4EPIM5E'; +const TX_HASH = 'a'.repeat(64); +const WORKER = path.join(__dirname, 'fixtures/webhook-postgres-worker.ts'); + +interface WorkerMessage { + type: string; + pid: number; + id?: string; + eventId?: string; + attempt?: number; + processed?: boolean; + message?: string; +} + +/** IPC barriers establish the overlap; no timing guess decides whether workers raced. */ +function spawnWorker(t: TestContext, schema: string) { + const child: ChildProcess = fork(WORKER, [], { + execArgv: ['--import', 'tsx'], + env: { ...process.env, Q584_PG_SCHEMA: schema }, + stdio: ['ignore', 'pipe', 'pipe', 'ipc'], + }); + const queued: WorkerMessage[] = []; + const waiting: Array<{ type: string; resolve: (message: WorkerMessage) => void; reject: (error: Error) => void }> = []; + let stderr = ''; + let exit: { code: number | null; signal: NodeJS.Signals | null } | undefined; + let failure: Error | undefined; + let resolveExit: (value: { code: number | null; signal: NodeJS.Signals | null }) => void; + const exited = new Promise<{ code: number | null; signal: NodeJS.Signals | null }>((resolve) => { resolveExit = resolve; }); + child.stderr!.on('data', (chunk) => { stderr = (stderr + chunk.toString()).slice(-4000); }); + child.on('message', (value) => { + const message = value as WorkerMessage; + if (message.type === 'error') { + failure = new Error(message.message || 'Webhook worker failed'); + for (const waiter of waiting.splice(0)) waiter.reject(failure); + return; + } + const index = waiting.findIndex((item) => item.type === message.type); + if (index === -1) queued.push(message); + else waiting.splice(index, 1)[0].resolve(message); + }); + child.on('error', (error) => { + failure = error; + for (const waiter of waiting.splice(0)) waiter.reject(error); + resolveExit({ code: null, signal: null }); + }); + child.on('exit', (code, signal) => { + exit = { code, signal }; + for (const waiter of waiting.splice(0)) waiter.reject(new Error(`Worker exited before ${waiter.type}: ${code ?? signal}; ${stderr}`)); + resolveExit(exit); + }); + t.after(async () => { + if (!exit && child.pid) child.kill('SIGKILL'); + await exited; + }); + return { + child, exited, + send(message: Record) { child.send(message); }, + wait(type: string): Promise { + const index = queued.findIndex((item) => item.type === type); + if (index !== -1) return Promise.resolve(queued.splice(index, 1)[0]); + if (failure) return Promise.reject(failure); + if (exit) return Promise.reject(new Error(`Worker already exited before ${type}: ${stderr}`)); + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + const index = waiting.indexOf(waiter); + if (index !== -1) waiting.splice(index, 1); + reject(new Error(`Timed out waiting for worker ${type}: ${stderr}`)); + }, 10_000); + const waiter = { + type, + resolve(message: WorkerMessage) { clearTimeout(timer); resolve(message); }, + reject(error: Error) { clearTimeout(timer); reject(error); }, + }; + waiting.push(waiter); + }); + }, + }; +} + +describe('PostgreSQL webhook outbox acceptance', { + skip: DATABASE_URL ? false : 'DATABASE_URL is not set', + concurrency: false, +}, () => { + const schema = `q584_webhooks_${randomUUID().replaceAll('-', '')}`; + let admin: Pool; + let database: Pool; + let invoices: InvoiceService; + let webhooks: PostgresWebhookStorage; + + before(async () => { + admin = new Pool({ connectionString: DATABASE_URL }); + await admin.query(`CREATE SCHEMA "${schema}"`); + database = new Pool({ connectionString: DATABASE_URL, options: `-c search_path=${schema},public` }); + const sql = readFileSync(path.join(__dirname, '../../db/schema.sql'), 'utf8'); + await database.query(sql); + await database.query(sql); + invoices = new InvoiceService(database); + webhooks = new PostgresWebhookStorage(database); + }); + + beforeEach(async () => { + await database.query('DROP TRIGGER IF EXISTS test_reject_outbox ON webhook_deliveries'); + await database.query('TRUNCATE webhook_deliveries, webhook_endpoints, webhook_proofs, payment_events, transactions, invoices CASCADE'); + }); + + after(async () => { + await database?.end(); + if (admin) { + await admin.query(`DROP SCHEMA IF EXISTS "${schema}" CASCADE`); + await admin.end(); + } + }); + + async function endpoint(events: readonly WebhookEventType[] = WEBHOOK_EVENT_TYPES) { + const value: WebhookEndpoint = { + id: randomUUID(), sellerPublicKey: SELLER, url: 'https://merchant.example/webhook', + events: [...events], enabled: true, failureCount: 0, createdAt: new Date().toISOString(), + ...sealWebhookSecret('whsec_integration_test_only', Buffer.alloc(32, 7)), + }; + await webhooks.register(value); + return value; + } + const createInvoice = (idempotencyKey?: string) => invoices.createInvoice({ + sellerPublicKey: SELLER, amount: 10, assetCode: 'XLM', idempotencyKey, + }); + async function deliveries(invoiceId: string) { + const result = await database.query( + `SELECT * FROM webhook_deliveries WHERE payload->'invoice'->>'id'=$1 ORDER BY created_at,id`, [invoiceId]); + return result.rows; + } + async function failOutboxInserts() { + await database.query(`CREATE OR REPLACE FUNCTION test_reject_outbox() RETURNS TRIGGER LANGUAGE plpgsql AS $$ + BEGIN RAISE EXCEPTION 'Q584_TEST_OUTBOX_FAILURE'; RETURN NEW; END; $$; + CREATE TRIGGER test_reject_outbox BEFORE INSERT ON webhook_deliveries + FOR EACH ROW EXECUTE FUNCTION test_reject_outbox();`); + } + + it('enqueues each creation and paid/cancelled/expired transition exactly once', async () => { + await endpoint(); + const paid = await createInvoice('same-create'); + const cancelled = await createInvoice(); + const expired = await createInvoice(); + assert.equal((await createInvoice('same-create')).id, paid.id); + await invoices.markAsPaid(paid.id, TX_HASH, PAYER, undefined, { settledAt: new Date() }); + await invoices.cancelInvoice(cancelled.id, SELLER); + await database.query(`UPDATE invoices SET expires_at=now()-interval '1 second' WHERE id=$1`, [expired.id]); + assert.equal(await invoices.markExpiredInvoices(), 1); + assert.equal(await invoices.markExpiredInvoices(), 0); + await assert.rejects(invoices.markAsPaid(paid.id, TX_HASH, PAYER, undefined, { settledAt: new Date() })); + await assert.rejects(invoices.cancelInvoice(cancelled.id, SELLER)); + await database.query('UPDATE invoices SET status=status'); + + for (const [invoice, kind, audit] of [ + [paid, 'invoice.paid', 'PAYMENT_CONFIRMED'], + [cancelled, 'invoice.cancelled', 'INVOICE_CANCELLED'], + [expired, 'invoice.expired', 'INVOICE_EXPIRED'], + ] as const) { + const rows = await deliveries(invoice.id); + assert.deepEqual(rows.map((row) => row.event_type).sort(), ['invoice.created', kind].sort()); + assert.equal(new Set(rows.map((row) => row.event_id)).size, 2); + for (const row of rows) { + assert.equal(row.payload.id, row.event_id); + assert.equal(row.payload.invoice.id, invoice.id); + assert.equal(row.payload.invoice.amount, '10.0000000'); + } + const events = await invoices.getPaymentEvents(invoice.id); + assert.equal(events.filter((row) => row.eventType === audit).length, 1); + } + }); + + it('rolls back invoice creation when its outbox insert fails', async () => { + await endpoint(); + await failOutboxInserts(); + await assert.rejects(createInvoice('failed-create'), /Q584_TEST_OUTBOX_FAILURE/); + for (const table of ['invoices', 'payment_events', 'webhook_deliveries']) { + assert.equal(Number((await database.query(`SELECT count(*) FROM ${table}`)).rows[0].count), 0); + } + await database.query('DROP TRIGGER test_reject_outbox ON webhook_deliveries'); + const invoice = await createInvoice('failed-create'); + assert.equal((await deliveries(invoice.id)).length, 1); + }); + + for (const transition of ['paid', 'cancelled', 'expired'] as const) { + it(`rolls back ${transition} invoice and audit state when its outbox insert fails`, async () => { + await endpoint(); + const invoice = await createInvoice(); + if (transition === 'expired') { + await database.query(`UPDATE invoices SET expires_at=now()-interval '1 second' WHERE id=$1`, [invoice.id]); + } + const mutate = () => transition === 'paid' + ? invoices.markAsPaid(invoice.id, TX_HASH, PAYER, undefined, { settledAt: new Date() }) + : transition === 'cancelled' + ? invoices.cancelInvoice(invoice.id, SELLER) + : invoices.markExpiredInvoices(); + await failOutboxInserts(); + await assert.rejects(mutate(), /Q584_TEST_OUTBOX_FAILURE/); + // Direct read intentionally avoids InvoiceService's expiry-on-read sweep. + const row = (await database.query('SELECT * FROM invoices WHERE id=$1', [invoice.id])).rows[0]; + assert.equal(row.status, 'PENDING'); + assert.equal(row.payment_tx_hash, null); + assert.equal(row.paid_at, null); + assert.equal(row.cancelled_at, null); + assert.equal(row.settled_at, null); + assert.equal((await invoices.getPaymentEvents(invoice.id)).length, 0); + assert.deepEqual((await deliveries(invoice.id)).map((row) => row.event_type), ['invoice.created']); + + await database.query('DROP TRIGGER test_reject_outbox ON webhook_deliveries'); + await mutate(); + assert.equal((await deliveries(invoice.id)).filter((row) => row.event_type === `invoice.${transition}`).length, 1); + assert.equal((await invoices.getPaymentEvents(invoice.id)).length, 1); + }); + } + + it('commits a payment rejection audit and outbox together, with the same event id', async () => { + await endpoint(); + const invoice = await createInvoice(); + const rejection = { code: 'ASSET_MISMATCH', txHash: TX_HASH, customerEmail: 'private@example.invalid', memo: 'private-memo' }; + await failOutboxInserts(); + await assert.rejects(invoices.logPaymentEvent(invoice.id, 'PAYMENT_REJECTED', rejection), /Q584_TEST_OUTBOX_FAILURE/); + assert.equal((await invoices.getPaymentEvents(invoice.id)).length, 0); + assert.equal((await deliveries(invoice.id)).filter((row) => row.event_type === 'payment.rejected').length, 0); + await database.query('DROP TRIGGER test_reject_outbox ON webhook_deliveries'); + await invoices.logPaymentEvent(invoice.id, 'PAYMENT_REJECTED', rejection); + const audit = (await invoices.getPaymentEvents(invoice.id))[0]; + const outbox = (await deliveries(invoice.id)).filter((row) => row.event_type === 'payment.rejected'); + assert.equal(outbox.length, 1); + assert.equal(outbox[0].event_id, audit.id); + assert.deepEqual(outbox[0].payload.payment, { code: rejection.code, txHash: TX_HASH }); + assert.doesNotMatch(JSON.stringify(outbox[0].payload), /private@example|private-memo/); + assert.equal((await invoices.getInvoiceById(invoice.id))?.status, 'PENDING'); + }); + + it('lets only one of two separate worker processes own a due delivery', { timeout: 30_000 }, async (t) => { + await endpoint(['invoice.created']); + const invoice = await createInvoice(); + const [row] = await deliveries(invoice.id); + const first = spawnWorker(t, schema); + await first.wait('ready'); + first.send({ type: 'run', mode: 'hold', now: new Date().toISOString() }); + const claim = await first.wait('claimed'); + assert.equal(claim.id, row.id); + + const second = spawnWorker(t, schema); + await second.wait('ready'); + assert.notEqual(first.child.pid, second.child.pid); + second.send({ type: 'run', mode: 'finish', now: new Date().toISOString() }); + const refused = await second.wait('done'); + assert.equal(refused.processed, false, 'second worker must finish without claiming the held row'); + assert.equal((await second.exited).code, 0); + assert.equal((await deliveries(invoice.id))[0].attempt, 0, 'first worker has not committed its delivery'); + + first.send({ type: 'release' }); + assert.equal((await first.wait('done')).processed, true); + assert.equal((await first.exited).code, 0); + const [delivered] = await deliveries(invoice.id); + assert.equal(delivered.id, row.id); + assert.equal(delivered.event_id, row.event_id); + assert.equal(delivered.status, 'delivered'); + assert.equal(delivered.attempt, 1); + }); + + it('retains the pending event after a worker crash and delivers it on restart', { timeout: 30_000 }, async (t) => { + await endpoint(['invoice.created']); + const invoice = await createInvoice(); + const [row] = await deliveries(invoice.id); + const first = spawnWorker(t, schema); + await first.wait('ready'); + first.send({ type: 'run', mode: 'hold', now: new Date().toISOString() }); + assert.equal((await first.wait('claimed')).id, row.id); + first.child.kill('SIGKILL'); + assert.equal((await first.exited).signal, 'SIGKILL'); + // Wait for PostgreSQL to observe that exact terminated client's disconnect. + // This is cleanup synchronization, not a claimed latency benchmark. + const deadline = Date.now() + 5000; + for (;;) { + const active = await admin.query('SELECT count(*) FROM pg_stat_activity WHERE application_name=$1', [`q584-worker-${first.child.pid}`]); + if (Number(active.rows[0].count) === 0) break; + assert.ok(Date.now() < deadline, 'database must release the crashed worker connection'); + await delay(20); + } + const [pending] = await deliveries(invoice.id); + assert.equal(pending.id, row.id); + assert.equal(pending.event_id, row.event_id); + assert.equal(pending.status, 'pending'); + assert.equal(pending.attempt, 0); + + const restarted = spawnWorker(t, schema); + await restarted.wait('ready'); + restarted.send({ type: 'run', mode: 'finish', now: new Date().toISOString() }); + const claim = await restarted.wait('claimed'); + assert.equal(claim.eventId, row.event_id); + assert.equal(claim.attempt, 0); + assert.equal((await restarted.wait('done')).processed, true); + assert.equal((await restarted.exited).code, 0); + const [delivered] = await deliveries(invoice.id); + assert.equal(delivered.status, 'delivered'); + assert.equal(delivered.attempt, 1); + assert.equal((await deliveries(invoice.id)).length, 1); + }); +}); diff --git a/frontend/app/dashboard/page.tsx b/frontend/app/dashboard/page.tsx index 16567bb2..842f45fb 100644 --- a/frontend/app/dashboard/page.tsx +++ b/frontend/app/dashboard/page.tsx @@ -5,6 +5,7 @@ import { invoiceApi } from '@/lib/api'; import InvoiceCard from '@/components/InvoiceCard'; import WalletConnect from '@/components/WalletConnect'; import UserProfile from '@/components/UserProfile'; +import SellerWebhooksPanel from '@/components/SellerWebhooksPanel'; import FreighterInstallPrompt from '@/components/FreighterInstallPrompt'; import AssetLogo from '@/components/AssetLogo'; import { useWalletStore } from '@/lib/store'; @@ -510,6 +511,7 @@ export default function DashboardPage() { )} + )} diff --git a/frontend/components/SellerWebhooksPanel.tsx b/frontend/components/SellerWebhooksPanel.tsx new file mode 100644 index 00000000..9c7db293 --- /dev/null +++ b/frontend/components/SellerWebhooksPanel.tsx @@ -0,0 +1,315 @@ +'use client'; + +import { useEffect, useRef, useState, type FormEvent } from 'react'; +import { WEBHOOK_EVENT_TYPES, type WebhookEventType } from '@shared/webhooks'; +import { useWalletStore } from '@/lib/store'; +import { EXPECTED_WALLET_NETWORK } from '@/lib/stellar'; +import { walletSessionGate } from '@/lib/wallet-session'; +import { + webhookApi, + webhookWalletContext, + type WebhookEndpoint, + type WebhookList, + type WebhookRequestContext, + type WebhookSecretResult, +} from '@/lib/webhook-api'; + +const EVENT_LABELS: Record = { + 'invoice.created': 'Invoice created', + 'invoice.paid': 'Invoice paid', + 'invoice.cancelled': 'Invoice cancelled', + 'invoice.expired': 'Invoice expired', + 'payment.rejected': 'Payment rejected', +}; + +const STATUS_LABELS = { + pending: 'Pending', + delivered: 'Delivered', + dead: 'Failed permanently', + cancelled: 'Cancelled', +}; + +function dateLabel(value?: string | null): string { + if (!value) return 'โ€”'; + const date = new Date(value); + return Number.isNaN(date.getTime()) ? 'โ€”' : date.toLocaleString(); +} + +/** The keyed subtree clears secrets and responses on any wallet/network change. */ +export default function SellerWebhooksPanel() { + const wallet = useWalletStore(); + const gate = walletSessionGate(wallet, EXPECTED_WALLET_NETWORK); + if (!gate.ready || !wallet.publicKey) return null; + const context = webhookWalletContext(wallet); + return ; +} + +function SellerWebhooksContent({ sellerPublicKey, walletContext }: Omit) { + const [data, setData] = useState(null); + const [hasLoaded, setHasLoaded] = useState(false); + const [url, setUrl] = useState(''); + const [events, setEvents] = useState([...WEBHOOK_EVENT_TYPES]); + const [busy, setBusy] = useState(null); + const [error, setError] = useState(null); + const [notice, setNotice] = useState(null); + const [secret, setSecret] = useState(null); + const [copied, setCopied] = useState(false); + const [removeId, setRemoveId] = useState(null); + const active = useRef(true); + const inFlight = useRef(false); + const abort = useRef(null); + + useEffect(() => { + active.current = true; + return () => { + active.current = false; + abort.current?.abort(); + }; + }, []); + + async function run(label: string, action: (context: WebhookRequestContext) => Promise<() => void>) { + if (inFlight.current) return; + inFlight.current = true; + const controller = new AbortController(); + abort.current = controller; + setBusy(label); + setError(null); + setNotice(null); + try { + const commit = await action({ sellerPublicKey, walletContext, signal: controller.signal }); + if (active.current && !controller.signal.aborted && + webhookWalletContext(useWalletStore.getState()) === walletContext) commit(); + } catch (caught) { + if (active.current && !controller.signal.aborted) { + setError(caught instanceof Error ? caught.message : 'The webhook action could not be completed.'); + } + } finally { + inFlight.current = false; + if (active.current) setBusy(null); + } + } + + function upsertEndpoint(endpoint: WebhookEndpoint) { + setData((previous) => ({ + endpoints: [endpoint, ...(previous?.endpoints ?? []).filter((item) => item.id !== endpoint.id)], + deliveries: previous?.deliveries ?? [], + })); + } + + function revealSecret(result: WebhookSecretResult) { + upsertEndpoint(result.endpoint); + setSecret(result); + setCopied(false); + } + + function load() { + void run('Loading endpoints', async (context) => { + const result = await webhookApi.list(context); + return () => { + setData(result); + setHasLoaded(true); + setRemoveId(null); + setNotice('Endpoints and recent deliveries are up to date.'); + }; + }); + } + + function register(event: FormEvent) { + event.preventDefault(); + if (secret) return; + void run('Adding endpoint', async (context) => { + const result = await webhookApi.register(context, url, events); + return () => { + revealSecret(result); + setUrl(''); + setNotice('Endpoint added. Save its signing secret below.'); + }; + }); + } + + function rotate(endpointId: string) { + if (secret) return; + void run('Rotating signing secret', async (context) => { + const result = await webhookApi.rotate(context, endpointId); + return () => { + revealSecret(result); + setNotice('Signing secret rotated. Update the secret in your receiver.'); + }; + }); + } + + function remove(endpointId: string) { + void run('Removing endpoint', async (context) => { + await webhookApi.remove(context, endpointId); + return () => { + setData((previous) => previous ? { + endpoints: previous.endpoints.filter((endpoint) => endpoint.id !== endpointId), + // A refresh obtains the final delivery status from the server. + deliveries: previous.deliveries.filter((delivery) => delivery.endpointId !== endpointId), + } : null); + if (secret?.endpoint.id === endpointId) setSecret(null); + setRemoveId(null); + setNotice('Endpoint removed. Refresh to load the latest delivery history.'); + }; + }); + } + + function sendTest(endpointId: string) { + void run('Queuing test delivery', async (context) => { + const result = await webhookApi.test(context, endpointId); + return () => setNotice(`Test event queued: ${result.eventId}. Refresh shortly to check its delivery.`); + }); + } + + async function copySecret() { + if (!secret) return; + try { + await navigator.clipboard.writeText(secret.secret); + if (active.current && webhookWalletContext(useWalletStore.getState()) === walletContext) setCopied(true); + } catch { + if (active.current) setError('Could not copy automatically. Select and copy the signing secret below.'); + } + } + + return ( +
+
+
+

Webhooks

+

Receive signed invoice updates in your own system.

+

Approve a Freighter wallet signature for each action.

+
+ +
+ +

+ {busy ? `${busy}. Check Freighter if a signature is requested.` : notice} +

+ {error &&

{error}

} + + {secret && ( +
+

Copy and save this signing secret now

+

It is shown once. Store it securely on the server that receives your webhooks.

+

{secret.endpoint.url}

+ +