feat(indexer): add verification webhook delivery and fix reorg detect…

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,
   };
 }
 

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,
   };
 }

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> = {}): 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);

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

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> = {}): 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> = {}): 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));
+  });
+});