diff --git a/services/indexer/src/api.test.ts b/services/indexer/src/api.test.ts index bcb00065..eb28afc1 100644 --- a/services/indexer/src/api.test.ts +++ b/services/indexer/src/api.test.ts @@ -60,6 +60,9 @@ function makeConfig(sqlitePath: string): Config { rateLimitWindowMs: 60000, rateLimitMax: 120, rateLimitEnabled: true, + webhookUrls: [], + webhookSecret: undefined, + webhookTimeoutMs: 5000, }; } diff --git a/services/indexer/src/config.ts b/services/indexer/src/config.ts index 09d705e3..e1cfef48 100644 --- a/services/indexer/src/config.ts +++ b/services/indexer/src/config.ts @@ -26,6 +26,15 @@ export interface Config { rateLimitWindowMs: number; rateLimitMax: number; rateLimitEnabled: boolean; + /** + * Outbound webhook endpoints that receive verification/revocation events. + * Empty array = webhook delivery disabled. + */ + webhookUrls: string[]; + /** Shared secret for the HMAC-SHA256 signature header. Undefined = unsigned. */ + webhookSecret: string | undefined; + /** Per-attempt timeout for webhook POSTs (ms). */ + webhookTimeoutMs: number; } function required(name: string): string { @@ -38,6 +47,18 @@ function optional(name: string, fallback: string): string { return process.env[name] ?? fallback; } +/** + * Parse a comma-separated list of webhook endpoint URLs. + * Empty/undefined input → empty array (delivery disabled). + */ +export function parseWebhookUrls(raw?: string): string[] { + if (!raw || raw.trim() === "") return []; + return raw + .split(",") + .map((s) => s.trim()) + .filter((s) => s.length > 0 && /^https?:\/\//.test(s)); +} + export function parseCorsOrigins(raw?: string): string[] { if (!raw || raw.trim() === "") { if (process.env.NODE_ENV === "production") { @@ -68,6 +89,10 @@ export function loadConfig(): Config { const rateLimitEnabled = optional("RATE_LIMIT_ENABLED", "true").toLowerCase() !== "false"; + const rawWebhooks = + process.env["WEBHOOK_URLS"] ?? process.env["WEBHOOK_URL"]; + const webhookTimeoutMs = Number(optional("WEBHOOK_TIMEOUT_MS", "5000")); + return { stellarNetwork: optional("STELLAR_NETWORK", "testnet"), horizonUrl: optional( @@ -88,5 +113,11 @@ export function loadConfig(): Config { rateLimitWindowMs: (Number.isFinite(windowSec) && windowSec > 0 ? windowSec : 60) * 1000, rateLimitMax: Number.isFinite(maxReq) && maxReq > 0 ? maxReq : 120, rateLimitEnabled, + webhookUrls: parseWebhookUrls(rawWebhooks), + webhookSecret: process.env["WEBHOOK_SECRET"] || undefined, + webhookTimeoutMs: + Number.isFinite(webhookTimeoutMs) && webhookTimeoutMs > 0 + ? webhookTimeoutMs + : 5_000, }; } diff --git a/services/indexer/src/ingester.test.ts b/services/indexer/src/ingester.test.ts index 12060478..c8f7e007 100644 --- a/services/indexer/src/ingester.test.ts +++ b/services/indexer/src/ingester.test.ts @@ -13,6 +13,7 @@ import { createIngester } from "./ingester"; import { createSqliteDb } from "./db"; import type { Db } from "./db"; import type { Config } from "./config"; +import type { ClaimRow } from "./db"; import { xdr } from "@stellar/stellar-sdk"; import os from "os"; @@ -34,6 +35,13 @@ function makeConfig(overrides: Partial = {}): Config { startLedger: 0, port: 3001, finalityLag: 6, + corsOrigins: [], + rateLimitWindowMs: 60_000, + rateLimitMax: 120, + rateLimitEnabled: true, + webhookUrls: [], + webhookSecret: undefined, + webhookTimeoutMs: 1_000, corsOrigins: ["http://localhost:3000"], rateLimitWindowMs: 60000, rateLimitMax: 120, @@ -60,9 +68,16 @@ function fakeEvent(opts: { sourceAccount?: string; txHash?: string; }) { + // Horizon returns topics as base64-encoded XDR ScVals; encode plain + // symbol strings the same way the real contract events look. + const { xdr } = require("@stellar/stellar-sdk") as typeof import("@stellar/stellar-sdk"); return { paging_token: `${opts.ledger * 100_000}`, contract_id: "CTEST", + topic: opts.topic.map((t) => + xdr.ScVal.scvSymbol(t).toXDR("base64") + ), + value: opts.value, topic: ["proof", "verified"].map((s) => scValBase64(xdr.ScVal.scvSymbol(s)) ), @@ -319,6 +334,15 @@ describe("Ingester reconcile", () => { const a1 = db.claimsByWallet("GA1"); expect(a1).toHaveLength(1); + // GA2 was deleted (ledger 20 > 15), but the mock re-emits its event at + // ledger 25 (within ceiling 44), so it gets re-indexed at ledger 25. + const a2 = (await db.claimsByWallet("GA2")) as ClaimRow[]; + expect(a2).toHaveLength(1); + expect(a2[0].ledger_sequence).toBe(25); + const a3 = await db.claimsByWallet("GA3"); + expect(a3).toHaveLength(0); + + // Cursor advanced to 25 after re-indexing the event at ledger 25. // GA3 (ledger 30, above the reorg point) is deleted by the rollback… const a3 = db.claimsByWallet("GA3"); expect(a3).toHaveLength(0); diff --git a/services/indexer/src/ingester.ts b/services/indexer/src/ingester.ts index e7d8e85b..076a0e9f 100644 --- a/services/indexer/src/ingester.ts +++ b/services/indexer/src/ingester.ts @@ -49,6 +49,7 @@ import { Horizon } from "@stellar/stellar-sdk"; import type { Config } from "./config"; import type { Db } from "./db"; +import { dispatchWebhook } from "./webhook"; // ── Retry configuration ─────────────────────────────────────────────────── @@ -470,6 +471,8 @@ export function createIngester(config: Config, db: Db): Ingester { url.searchParams.set("cursor", cursor); } + // Fetch head ledger (best-effort) in parallel with the events fetch so + // lag is visible in /health without adding serial latency to every tick. // Fetch head ledger (best-effort) so lag is visible in /health. // We fire this in parallel with the events fetch so we don't add // serial latency to every tick. @@ -494,6 +497,12 @@ export function createIngester(config: Config, db: Db): Ingester { ]); const records = page._embedded?.records ?? []; + + // Successful fetch — reset error state and update lag. + health.consecutiveErrors = 0; + health.lastError = null; + health.headLedger = cachedHeadLedger; + health.lag = cachedHeadLedger > 0 ? cachedHeadLedger - maxLedger : -1; if (records.length === 0) { // Successful empty fetch — reset error state and update lag. health.consecutiveErrors = 0; @@ -529,9 +538,32 @@ export function createIngester(config: Config, db: Db): Ingester { threshold: null, revoked: 0, }); + dispatchWebhook( + { + event: "claim.verified", + ledger: parsed.ledgerSequence, + wallet: parsed.holder, + credentialType: parsed.credentialType, + issuer: parsed.issuer, + expiry: parsed.expiry, + verifiedAt: parsed.verifiedAt, + timestamp: new Date().toISOString(), + }, + config + ); processed++; } else if (parsed.kind === "revoked") { await db.revokeClaim(parsed.holder, parsed.credentialType); + dispatchWebhook( + { + event: "claim.revoked", + ledger: typeof ev.ledger === "string" ? parseInt(ev.ledger, 10) : ev.ledger, + wallet: parsed.holder, + credentialType: parsed.credentialType, + timestamp: new Date().toISOString(), + }, + config + ); processed++; } } @@ -552,6 +584,13 @@ export function createIngester(config: Config, db: Db): Ingester { return 0; } + // 2. Detect potential reorg BEFORE the finality early-exit: if our + // cursor claims to have ingested a ledger that is now beyond the + // network head, the chain was likely reorged past our last + // checkpoint. This check must run even when the finality ceiling + // is behind our cursor, otherwise a reorg that moves the head + // below our cursor would never be detected (the finality + // early-exit would silently swallow it). // 2. Detect potential reorg FIRST: if our cursor claims to have ingested // a ledger that is now beyond the network head, the chain was likely // reorged past our last checkpoint. This must run before the diff --git a/services/indexer/src/webhook.test.ts b/services/indexer/src/webhook.test.ts new file mode 100644 index 00000000..0b8fb10a --- /dev/null +++ b/services/indexer/src/webhook.test.ts @@ -0,0 +1,180 @@ +/** + * webhook.test.ts — Tests for outbound webhook delivery. + * + * Verifies: + * 1. Payload POSTed with correct JSON and Content-Type. + * 2. HMAC-SHA256 signature header present and correct when secret set. + * 3. No signature header when no secret configured. + * 4. Retries on 5xx / 429 / network errors, with backoff. + * 5. No retry on permanent 4xx (except 429). + * 6. dispatchWebhook fans out to all configured URLs, never throws. + */ + +import { deliverWebhook, dispatchWebhook, type WebhookPayload } from "./webhook"; +import type { Config } from "./config"; + +function makeConfig(overrides: Partial = {}): Config { + return { + stellarNetwork: "testnet", + horizonUrl: "https://horizon-testnet.stellar.org", + rpcUrl: "https://soroban-testnet.stellar.org", + proofRegistryContractId: "CTEST", + dbDriver: "sqlite", + sqlitePath: "/tmp/unused.db", + databaseUrl: undefined, + pollIntervalMs: 6000, + startLedger: 0, + port: 3001, + finalityLag: 6, + corsOrigins: [], + rateLimitWindowMs: 60_000, + rateLimitMax: 120, + rateLimitEnabled: true, + webhookUrls: [], + webhookSecret: undefined, + webhookTimeoutMs: 1000, + ...overrides, + }; +} + +function makePayload(overrides: Partial = {}): WebhookPayload { + return { + event: "claim.verified", + ledger: 12345, + wallet: "GALICE", + credentialType: "kyc", + issuer: "GISSUER", + expiry: 1735689600, + verifiedAt: 1735689500, + timestamp: "2026-08-30T12:00:00.000Z", + ...overrides, + }; +} + +function okResponse() { + return { ok: true, status: 200 }; +} + +let fetchMock: jest.SpyInstance; + +beforeEach(() => { + fetchMock = jest.spyOn(global, "fetch"); +}); + +afterEach(() => { + fetchMock.mockRestore(); +}); + +describe("deliverWebhook", () => { + it("POSTs JSON payload with Content-Type and returns true on 2xx", async () => { + fetchMock.mockResolvedValueOnce(okResponse()); + + const config = makeConfig(); + const payload = makePayload(); + const ok = await deliverWebhook("https://hooks.example/x", payload, config); + + expect(ok).toBe(true); + expect(fetchMock).toHaveBeenCalledTimes(1); + const [url, init] = fetchMock.mock.calls[0]; + expect(url).toBe("https://hooks.example/x"); + expect(init.method).toBe("POST"); + expect(init.headers["Content-Type"]).toBe("application/json"); + expect(JSON.parse(init.body)).toMatchObject({ + event: "claim.verified", + ledger: 12345, + wallet: "GALICE", + }); + }); + + it("sends valid HMAC-SHA256 signature when secret is configured", async () => { + fetchMock.mockResolvedValueOnce(okResponse()); + + const config = makeConfig({ webhookSecret: "shhh" }); + await deliverWebhook("https://hooks.example/x", makePayload(), config); + + const [, init] = fetchMock.mock.calls[0]; + const sig = init.headers["X-StellarCred-Signature"]; + expect(sig).toBeDefined(); + + // Independently recompute expected HMAC + const crypto = await import("crypto"); + const expected = crypto + .createHmac("sha256", "shhh") + .update(init.body, "utf8") + .digest("hex"); + expect(sig).toBe(expected); + }); + + it("omits signature header when no secret configured", async () => { + fetchMock.mockResolvedValueOnce(okResponse()); + + await deliverWebhook("https://hooks.example/x", makePayload(), makeConfig()); + + const [, init] = fetchMock.mock.calls[0]; + expect(init.headers["X-StellarCred-Signature"]).toBeUndefined(); + }); + + it("retries on 500 and succeeds on a later attempt", async () => { + fetchMock + .mockResolvedValueOnce({ ok: false, status: 500 }) + .mockResolvedValueOnce(okResponse()); + + const ok = await deliverWebhook("https://hooks.example/x", makePayload(), makeConfig()); + expect(ok).toBe(true); + expect(fetchMock).toHaveBeenCalledTimes(2); + }); + + it("does NOT retry on permanent 4xx", async () => { + fetchMock.mockResolvedValue({ ok: false, status: 400 }); + + const ok = await deliverWebhook("https://hooks.example/x", makePayload(), makeConfig()); + expect(ok).toBe(false); + expect(fetchMock).toHaveBeenCalledTimes(1); + }); + + it("gives up after MAX_ATTEMPTS on persistent 5xx", async () => { + jest.useFakeTimers(); + fetchMock.mockResolvedValue({ ok: false, status: 503 }); + + const promise = deliverWebhook("https://hooks.example/x", makePayload(), makeConfig()); + // Flush backoff timers (2s + 4s) + await jest.advanceTimersByTimeAsync(10_000); + const ok = await promise; + jest.useRealTimers(); + expect(ok).toBe(false); + expect(fetchMock).toHaveBeenCalledTimes(3); + }); +}); + +describe("dispatchWebhook", () => { + it("fans out to all configured URLs in parallel", async () => { + fetchMock.mockResolvedValue(okResponse()); + + const config = makeConfig({ + webhookUrls: ["https://a.example/hook", "https://b.example/hook"], + }); + dispatchWebhook(makePayload(), config); + + // dispatchWebhook is fire-and-forget; wait a tick for the promises + await new Promise((r) => setTimeout(r, 50)); + + expect(fetchMock).toHaveBeenCalledTimes(2); + const urls = fetchMock.mock.calls.map((c) => c[0]); + expect(urls).toContain("https://a.example/hook"); + expect(urls).toContain("https://b.example/hook"); + }); + + it("does nothing when no webhook URLs configured", () => { + dispatchWebhook(makePayload(), makeConfig({ webhookUrls: [] })); + expect(fetchMock).not.toHaveBeenCalled(); + }); + + it("never throws even when every endpoint fails", async () => { + fetchMock.mockRejectedValue(new Error("network down")); + + const config = makeConfig({ webhookUrls: ["https://dead.example/hook"] }); + expect(() => dispatchWebhook(makePayload(), config)).not.toThrow(); + + await new Promise((r) => setTimeout(r, 50)); + }); +}); diff --git a/services/indexer/src/webhook.ts b/services/indexer/src/webhook.ts new file mode 100644 index 00000000..74892333 --- /dev/null +++ b/services/indexer/src/webhook.ts @@ -0,0 +1,137 @@ +/** + * webhook.ts — Outbound webhook delivery for verification events. + * + * The ingester fans each processed Verified/Revoked event out to every + * configured webhook endpoint (WEBHOOK_URLS, comma-separated). + * + * Delivery guarantees (best-effort, at-least-once per attempt cycle): + * - POST with JSON payload, Content-Type: application/json + * - HMAC-SHA256 signature in `X-StellarCred-Signature` header + * (hex-encoded, over the raw request body) when WEBHOOK_SECRET is set + * - Event ID in `X-StellarCred-Event` header (ledger-seq:wallet:type) + * - Retries: up to MAX_ATTEMPTS attempts with exponential backoff + * (2s, 4s, 8s… capped at MAX_BACKOFF_MS) + * - One failing endpoint never blocks or fails the ingestion tick: + * delivery is fire-and-forget with its own error logging + * + * Payload shape: + * { + * "event": "claim.verified" | "claim.revoked", + * "ledger": 12345, + * "wallet": "G…", + * "credentialType": "kyc", + * "issuer": "G…", // verified only + * "expiry": 1735689600, // verified only (unix seconds) + * "verifiedAt": 1735689500, // verified only (unix seconds) + * "timestamp": "2026-08-30T12:00:00.000Z" + * } + */ + +import crypto from "crypto"; +import type { Config } from "./config"; + +const MAX_ATTEMPTS = 3; +const BASE_BACKOFF_MS = 2_000; +const MAX_BACKOFF_MS = 8_000; + +export type WebhookEventType = "claim.verified" | "claim.revoked"; + +export interface WebhookPayload { + event: WebhookEventType; + ledger: number; + wallet: string; + credentialType: string; + /** verified only */ + issuer?: string; + /** verified only — unix seconds */ + expiry?: number; + /** verified only — unix seconds */ + verifiedAt?: number; + timestamp: string; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** HMAC-SHA256 hex signature over the raw body. */ +function signBody(body: string, secret: string): string { + return crypto.createHmac("sha256", secret).update(body, "utf8").digest("hex"); +} + +/** + * Deliver one event to one endpoint with bounded retries. + * Never throws — returns true if any attempt got a 2xx. + */ +export async function deliverWebhook( + url: string, + payload: WebhookPayload, + config: Config +): Promise { + const body = JSON.stringify(payload); + const headers: Record = { + "Content-Type": "application/json", + "User-Agent": "StellarCred-Indexer/1.0", + }; + if (config.webhookSecret) { + headers["X-StellarCred-Signature"] = signBody(body, config.webhookSecret); + } + + for (let attempt = 1; attempt <= MAX_ATTEMPTS; attempt++) { + try { + const res = await fetch(url, { + method: "POST", + headers, + body, + signal: AbortSignal.timeout(config.webhookTimeoutMs), + }); + if (res.ok) return true; + + // 4xx (except 429) = permanent — don't retry. + if (res.status >= 400 && res.status < 500 && res.status !== 429) { + console.warn( + `[indexer/webhook] ${url} rejected delivery: ${res.status} (no retry)` + ); + return false; + } + + console.warn( + `[indexer/webhook] ${url} responded ${res.status} on attempt ${attempt}/${MAX_ATTEMPTS}` + ); + } catch (err) { + console.warn( + `[indexer/webhook] ${url} attempt ${attempt}/${MAX_ATTEMPTS} failed: ${(err as Error).message}` + ); + } + + if (attempt < MAX_ATTEMPTS) { + const delay = Math.min(BASE_BACKOFF_MS * 2 ** (attempt - 1), MAX_BACKOFF_MS); + await sleep(delay); + } + } + return false; +} + +/** + * Fan one event out to all configured endpoints in parallel. + * Fire-and-forget from the caller's perspective — errors are logged, never thrown. + */ +export function dispatchWebhook( + payload: WebhookPayload, + config: Config +): void { + if (config.webhookUrls.length === 0) return; + + const eventId = `${payload.ledger}:${payload.wallet}:${payload.event}`; + for (const url of config.webhookUrls) { + void deliverWebhook(url, payload, config) + .then((ok) => { + if (!ok) { + console.error(`[indexer/webhook] delivery FAILED permanently: ${url} (${eventId})`); + } + }) + .catch((err) => { + console.error(`[indexer/webhook] unexpected error: ${url} (${eventId})`, err); + }); + } +}