diff --git a/docs/mvp-express.md b/docs/mvp-express.md index 05bca5f1..83ef3588 100644 --- a/docs/mvp-express.md +++ b/docs/mvp-express.md @@ -181,6 +181,8 @@ const response = await fetch('https://anchor.example.com/anchor/info', { Your callback receives provider-agnostic payload as `event.payload`. +The `x-webhook-provider` header takes precedence over `provider` in the JSON body. The body value is used when the header is missing or blank, and `generic` is used when neither source has a value. Duplicate webhook responses include the stored event `status` (`pending`, `processed`, or `failed`); first-delivery responses keep their existing shape. + ## 6) Background jobs lifecycle - `await anchor.init()` diff --git a/src/core/factory.ts b/src/core/factory.ts index e8d83298..12e86c39 100644 --- a/src/core/factory.ts +++ b/src/core/factory.ts @@ -10,6 +10,7 @@ import type { DatabaseAdapter, InteractiveTransactionRecord, QueueAdapter, + QueueDrainStatus, Watcher, WebhookProcessor, } from '@/runtime/interfaces.ts'; @@ -38,6 +39,7 @@ export class AnchorInstance { private initPromise: Promise | null = null; private shutdownPromise: Promise | null = null; private backgroundJobsPromise: Promise | null = null; + private backgroundJobsStopPromise: Promise | null = null; constructor(config: Partial) { this.config = new AnchorConfig(config); @@ -159,14 +161,30 @@ export class AnchorInstance { * Stop watcher services and queue workers. */ public async stopBackgroundJobs(): Promise { + if (this.backgroundJobsStopPromise) return this.backgroundJobsStopPromise; + if (this.backgroundJobsPromise) await this.backgroundJobsPromise; + if (this.backgroundJobsStopPromise) return this.backgroundJobsStopPromise; if (!this.initialized || !this.backgroundJobsRunning) return; - for (const watcher of this.watchers) { - await watcher.stop(); - } + this.backgroundJobsStopPromise = (async () => { + try { + for (const watcher of this.watchers) { + await watcher.stop(); + } - await this.requireQueue().stop(); - this.backgroundJobsRunning = false; + await this.requireQueue().stop(); + this.backgroundJobsRunning = false; + } finally { + this.backgroundJobsStopPromise = null; + } + })(); + + return this.backgroundJobsStopPromise; + } + + /** Return pending and active queue counts without exposing job payloads. */ + public getQueueStatus(): QueueDrainStatus { + return this.queue?.status ?? { pending: 0, active: 0 }; } /** diff --git a/src/index.ts b/src/index.ts index fba635bf..9367ec9d 100644 --- a/src/index.ts +++ b/src/index.ts @@ -25,6 +25,7 @@ export type { IdempotencyRecord, InteractiveTransactionRecord, QueueJob, + QueueDrainStatus, QueueAdapter, Watcher, WebhookProcessor, diff --git a/src/runtime/http/express-router-impl.ts b/src/runtime/http/express-router-impl.ts index 09f3553d..1bd8c84a 100644 --- a/src/runtime/http/express-router-impl.ts +++ b/src/runtime/http/express-router-impl.ts @@ -1307,13 +1307,15 @@ async function handleWebhook( signature, }); - sendJson(res, 200, { + const response: Record = { received: true, duplicate: result.duplicate, event_id: result.eventId, received_at: new Date().toISOString(), provider: result.provider, - }); + }; + if (result.duplicate && result.status) response.status = result.status; + sendJson(res, 200, response); } catch { sendJson(res, 400, { error: 'webhook_error', diff --git a/src/runtime/interfaces.ts b/src/runtime/interfaces.ts index e9eddf45..acbbe23b 100644 --- a/src/runtime/interfaces.ts +++ b/src/runtime/interfaces.ts @@ -159,7 +159,13 @@ export type QueueJob = | { type: 'process_watcher_task'; payload: { watcherTaskId: string } } | { type: 'cleanup_records'; payload: { retentionDays: number } }; +export interface QueueDrainStatus { + readonly pending: number; + readonly active: number; +} + export interface QueueAdapter { + readonly status?: QueueDrainStatus; enqueue(job: QueueJob): Promise; start(worker: (job: QueueJob) => Promise): Promise; stop(): Promise; @@ -178,5 +184,10 @@ export interface WebhookProcessor { payload: Record; rawBody: string | Buffer | Uint8Array; signature?: string; - }): Promise<{ duplicate: boolean; eventId: string; provider: string }>; + }): Promise<{ + duplicate: boolean; + eventId: string; + provider: string; + status?: WebhookEventRecord['status']; + }>; } diff --git a/src/runtime/queue/in-memory-queue.ts b/src/runtime/queue/in-memory-queue.ts index 77c3da96..e4001acf 100644 --- a/src/runtime/queue/in-memory-queue.ts +++ b/src/runtime/queue/in-memory-queue.ts @@ -1,5 +1,5 @@ import { ConfigError } from '@/core/errors.ts'; -import type { QueueAdapter, QueueJob } from '@/runtime/interfaces.ts'; +import type { QueueAdapter, QueueDrainStatus, QueueJob } from '@/runtime/interfaces.ts'; interface InMemoryQueueOptions { concurrency: number; @@ -22,6 +22,10 @@ export class InMemoryQueueAdapter implements QueueAdapter { this.concurrency = options.concurrency; } + public get status(): QueueDrainStatus { + return { pending: this.jobs.length, active: this.activeWorkers }; + } + public async enqueue(job: QueueJob): Promise { this.jobs.push(job); this.kick(); diff --git a/src/runtime/webhooks/default-webhook-processor.ts b/src/runtime/webhooks/default-webhook-processor.ts index c0d76c1c..ef84b647 100644 --- a/src/runtime/webhooks/default-webhook-processor.ts +++ b/src/runtime/webhooks/default-webhook-processor.ts @@ -37,7 +37,12 @@ export class DefaultWebhookProcessor implements WebhookProcessor { payload: Record; rawBody: string | Buffer | Uint8Array; signature?: string; - }): Promise<{ duplicate: boolean; eventId: string; provider: string }> { + }): Promise<{ + duplicate: boolean; + eventId: string; + provider: string; + status?: 'pending' | 'processed' | 'failed'; + }> { this.verifySignatureIfEnabled(input); const insertion = await this.database.insertOrGetWebhookEvent({ @@ -52,6 +57,7 @@ export class DefaultWebhookProcessor implements WebhookProcessor { duplicate: true, eventId: insertion.record.eventId, provider: insertion.record.provider, + status: insertion.record.status, }; } diff --git a/tests/core/factory-background-startup.test.ts b/tests/core/factory-background-startup.test.ts index 956871d1..45499b96 100644 --- a/tests/core/factory-background-startup.test.ts +++ b/tests/core/factory-background-startup.test.ts @@ -1,12 +1,25 @@ import { Keypair } from '@stellar/stellar-sdk'; import type { QueueAdapter, QueueJob } from '@/runtime/interfaces.ts'; +import { InMemoryQueueAdapter } from '@/runtime/queue/in-memory-queue.ts'; import { TransactionWatcher } from '@/runtime/watchers/transaction-watcher.ts'; import { describe, expect, it } from 'vitest'; import { createAnchor } from '@/index.ts'; +function deferred() { + let resolve!: () => void; + const promise = new Promise((res) => { + resolve = res; + }); + + return { promise, resolve }; +} + class MockQueueAdapter { public started = false; public startCalls = 0; + public stopCalls = 0; + public stopGate: Promise | null = null; + public stopEntered = deferred(); public worker: | ((job: { type: string; payload: Record }) => Promise) | null = null; @@ -20,22 +33,91 @@ class MockQueueAdapter { } public async stop(): Promise { + this.stopCalls += 1; + this.stopEntered.resolve(); + if (this.stopGate) await this.stopGate; this.started = false; } } class MockWatcher { public startCalls = 0; + public stopCalls = 0; + public stopGate: Promise | null = null; + public stopError: Error | null = null; + public stopEntered = deferred(); public async start(): Promise { this.startCalls += 1; } public async stop(): Promise { - // no-op + this.stopCalls += 1; + this.stopEntered.resolve(); + if (this.stopGate) await this.stopGate; + if (this.stopError) { + const error = this.stopError; + this.stopError = null; + throw error; + } } } describe('AnchorInstance concurrent background startup', () => { + it('exposes queue counts without exposing queued job payloads', async () => { + const anchor = createAnchor({ + network: { network: 'testnet' }, + server: { interactiveDomain: 'https://anchor.example.com' }, + security: { + sep10SigningKey: Keypair.random().secret(), + interactiveJwtSecret: 'jwt-test-secret', + distributionAccountSecret: 'distribution-test-secret', + }, + assets: { + assets: [ + { + code: 'USDC', + issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5', + }, + ], + }, + framework: { + database: { provider: 'sqlite', url: 'file::memory:' }, + watchers: { enabled: false }, + }, + }); + await anchor.init(); + + const queue = new InMemoryQueueAdapter({ concurrency: 1 }); + (anchor as unknown as { queue: InMemoryQueueAdapter }).queue = queue; + expect(anchor.getQueueStatus()).toEqual({ pending: 0, active: 0 }); + + await queue.enqueue({ + type: 'expire_transaction', + payload: { transactionId: 'private-transaction-id' }, + }); + expect(anchor.getQueueStatus()).toEqual({ pending: 1, active: 0 }); + + let releaseWorker: (() => void) | undefined; + const workerGate = new Promise((resolve) => { + releaseWorker = resolve; + }); + let signalStarted: (() => void) | undefined; + const workerStarted = new Promise((resolve) => { + signalStarted = resolve; + }); + await queue.start(async () => { + signalStarted?.(); + await workerGate; + }); + await workerStarted; + expect(anchor.getQueueStatus()).toEqual({ pending: 0, active: 1 }); + + releaseWorker?.(); + await queue.stop(); + expect(anchor.getQueueStatus()).toEqual({ pending: 0, active: 0 }); + await anchor.shutdown(); + }); + it.each([ ['uses the operational retention value when both settings are configured', 21, 45, 21], ['uses the watcher retention value when the operational setting is absent', undefined, 45, 45], @@ -126,7 +208,83 @@ describe('AnchorInstance concurrent background startup', () => { expect(queue.startCalls).toBe(1); expect(watcher.startCalls).toBe(1); + const watcherStopGate = deferred(); + const queueStopGate = deferred(); + watcher.stopGate = watcherStopGate.promise; + queue.stopGate = queueStopGate.promise; + + let stopComplete = false; + const stops = Promise.all([anchor.stopBackgroundJobs(), anchor.stopBackgroundJobs()]).then( + () => { + stopComplete = true; + }, + ); + + await watcher.stopEntered.promise; + expect(watcher.stopCalls).toBe(1); + expect(queue.stopCalls).toBe(0); + + watcherStopGate.resolve(); + await queue.stopEntered.promise; + expect(queue.stopCalls).toBe(1); + expect(stopComplete).toBe(false); + + queueStopGate.resolve(); + await stops; + expect(stopComplete).toBe(true); + await anchor.shutdown(); + }); + + it('shares a stop failure with concurrent callers without stopping the service twice', async () => { + const anchor = createAnchor({ + network: { network: 'testnet' }, + server: { interactiveDomain: 'https://anchor.example.com' }, + security: { + sep10SigningKey: Keypair.random().secret(), + interactiveJwtSecret: 'jwt-test-secret', + distributionAccountSecret: 'distribution-test-secret', + }, + assets: { + assets: [ + { + code: 'USDC', + issuer: 'GBBD47IF6LWK7P7MDEVSCWR7DPUWV3NY3DTQEVFL4NAT4AQH3ZLLFLA5', + }, + ], + }, + framework: { + database: { + provider: 'sqlite', + url: 'file::memory:', + }, + }, + }); + + const queue = new MockQueueAdapter(); + const watcher = new MockWatcher(); + const stopError = new Error('watcher stop failed'); + watcher.stopError = stopError; + + await anchor.init(); + (anchor as unknown as { queue: unknown }).queue = queue; + (anchor as unknown as { watchers: unknown[] }).watchers = [watcher as never]; + await anchor.startBackgroundJobs(); + + const results = await Promise.allSettled([ + anchor.stopBackgroundJobs(), + anchor.stopBackgroundJobs(), + ]); + + expect(watcher.stopCalls).toBe(1); + expect(queue.stopCalls).toBe(0); + expect(results.map((result) => result.status)).toEqual(['rejected', 'rejected']); + expect( + results.map((result) => (result.status === 'rejected' ? result.reason : undefined)), + ).toEqual([stopError, stopError]); + await anchor.stopBackgroundJobs(); + expect(watcher.stopCalls).toBe(2); + expect(queue.stopCalls).toBe(1); await anchor.shutdown(); }); }); diff --git a/tests/exports.test.ts b/tests/exports.test.ts index 5dd864fc..6bab4baf 100644 --- a/tests/exports.test.ts +++ b/tests/exports.test.ts @@ -16,6 +16,7 @@ describe('package root exports', () => { const job: QueueJob = { type: 'cleanup_records', payload: { retentionDays: 90 } }; const adapter: QueueAdapter = { + status: { pending: 0, active: 0 }, async enqueue(_job) {}, async start(_worker) {}, async stop() {}, diff --git a/tests/mvp-express.integration.test.ts b/tests/mvp-express.integration.test.ts index 8c301548..fbc0143e 100644 --- a/tests/mvp-express.integration.test.ts +++ b/tests/mvp-express.integration.test.ts @@ -1898,6 +1898,7 @@ describe('MVP Express-mounted integration', () => { expect(firstResponse.body.duplicate).toBe(false); expect(firstResponse.body.event_id).toBe('evt_1'); expect(firstResponse.body.provider).toBe('generic'); + expect(firstResponse.body.status).toBeUndefined(); expect(webhookCallbackCount).toBe(1); const duplicateResponse = await invoke({ diff --git a/tests/webhook-fallback.test.ts b/tests/webhook-fallback.test.ts index 5c765d89..680f8d1d 100644 --- a/tests/webhook-fallback.test.ts +++ b/tests/webhook-fallback.test.ts @@ -3,6 +3,7 @@ import { createAnchor, type AnchorInstance } from '@/index.ts'; import { Keypair } from '@stellar/stellar-sdk'; import { createHmac } from 'node:crypto'; import { unlinkSync } from 'node:fs'; +import type { DatabaseAdapter } from '@/runtime/interfaces.ts'; import type { IncomingMessage, ServerResponse } from 'node:http'; import { Readable } from 'node:stream'; import { afterAll, beforeAll, describe, expect, it } from 'vitest'; @@ -200,6 +201,76 @@ describe('Webhook Provider Fallback', () => { expect(lastProvider).toBe('header-provider'); }); + it('matching header and body provider values remain unchanged', async () => { + const payload = { id: 'evt_matching', type: 'test', provider: 'same-provider' }; + const signature = createHmac('sha256', 'webhook-test-secret') + .update(JSON.stringify(payload)) + .digest('hex'); + + const response = await invoke({ + method: 'POST', + path: '/webhooks/events', + headers: { + 'content-type': 'application/json', + 'x-webhook-provider': 'same-provider', + 'x-anchor-signature': signature, + }, + body: payload, + }); + + expect(response.status).toBe(200); + expect(lastProvider).toBe('same-provider'); + }); + + it('duplicate responses include persisted status without exposing failure details', async () => { + const database = (anchor as unknown as { database: DatabaseAdapter }).database; + const errorMessage = 'private downstream error'; + + for (const status of ['pending', 'processed', 'failed'] as const) { + const eventId = `evt_duplicate_${status}`; + const seeded = await database.insertOrGetWebhookEvent({ + id: `${eventId}_internal`, + eventId, + provider: 'stored-provider', + payload: { type: 'test' }, + }); + if (status !== 'pending') { + await database.updateWebhookEventStatus({ + id: seeded.record.id, + status, + ...(status === 'failed' ? { errorMessage } : {}), + }); + } + + const payload = { id: eventId, type: 'test' }; + const signature = createHmac('sha256', 'webhook-test-secret') + .update(JSON.stringify(payload)) + .digest('hex'); + const response = await invoke({ + method: 'POST', + path: '/webhooks/events', + headers: { + 'content-type': 'application/json', + 'x-webhook-provider': 'stored-provider', + 'x-anchor-signature': signature, + }, + body: payload, + }); + + expect(response.status).toBe(200); + if (status === 'failed') { + // Failed events are intentionally retried; the retry is a new + // processing attempt rather than a duplicate response. + expect(response.body.duplicate).toBe(false); + expect(response.body).not.toHaveProperty('status'); + } else { + expect(response.body.duplicate).toBe(true); + expect(response.body.status).toBe(status); + } + expect(JSON.stringify(response.body)).not.toContain(errorMessage); + } + }); + it('Neither present: should fall back to generic', async () => { const payload = { id: 'evt_none', type: 'test' }; const signature = createHmac('sha256', 'webhook-test-secret')