Feat/durable webhook queue

apps/api/src/db/client.ts

@@ -49,11 +49,28 @@ const BOOTSTRAP_SQL = [
    )`,
   `CREATE TABLE IF NOT EXISTS webhook_deliveries (
      id TEXT PRIMARY KEY, webhook_id TEXT NOT NULL, link_id TEXT NOT NULL,
-     event TEXT NOT NULL, status_code INTEGER, ok INTEGER NOT NULL,
+     event TEXT NOT NULL, attempt INTEGER NOT NULL DEFAULT 1,
+     queue_entry_id TEXT,
+     status_code INTEGER, ok INTEGER NOT NULL,
      error TEXT, created_at INTEGER NOT NULL
    )`,
-  `CREATE INDEX IF NOT EXISTS webhook_deliveries_webhook_id_created_at_idx
-     ON webhook_deliveries (webhook_id, created_at DESC)`,
+  `CREATE TABLE IF NOT EXISTS webhook_queue (
+     id TEXT PRIMARY KEY,
+     webhook_id TEXT NOT NULL,
+     link_id TEXT NOT NULL,
+     event TEXT NOT NULL,
+     payload TEXT NOT NULL,
+     attempts INTEGER NOT NULL DEFAULT 0,
+     next_attempt_at INTEGER NOT NULL,
+     status TEXT NOT NULL DEFAULT 'pending',
+     last_status_code INTEGER,
+     last_error TEXT,
+     created_at INTEGER NOT NULL,
+     updated_at INTEGER NOT NULL
+   )`,
+  // Index to make the worker's "claim due rows" query fast.
+  `CREATE INDEX IF NOT EXISTS idx_webhook_queue_due
+     ON webhook_queue (status, next_attempt_at)`,
   `CREATE TABLE IF NOT EXISTS offramp_quotes (
      quote_id TEXT PRIMARY KEY, link_id TEXT NOT NULL,
      sell_asset_code TEXT NOT NULL, sell_asset_issuer TEXT, sell_amount TEXT NOT NULL,

apps/api/src/db/schema.ts

@@ -98,12 +98,44 @@ export const webhookDeliveries = sqliteTable("webhook_deliveries", {
   webhookId: text("webhook_id").notNull(),
   linkId: text("link_id").notNull(),
   event: text("event").notNull(),
+  /** Which attempt number this row records (1-based). */
+  attempt: integer("attempt").notNull().default(1),
+  /** ID of the queue entry this delivery belongs to. Null for legacy rows. */
+  queueEntryId: text("queue_entry_id"),
   statusCode: integer("status_code"),
   ok: integer("ok", { mode: "boolean" }).notNull(),
   error: text("error"),
   createdAt: integer("created_at").notNull(),
 });
 
+/**
+ * Durable delivery queue.  Each row = one pending (or dead-lettered) delivery.
+ * The worker claims rows by atomically flipping status from "pending" → "claimed",
+ * then delivers, then either resolves → "delivered" or reschedules / dead-letters.
+ *
+ * status lifecycle:
+ *   pending → claimed → delivered
+ *                    ↘ pending  (transient failure, next_attempt_at bumped)
+ *                    ↘ dead     (exhausted max attempts)
+ *   dead   → pending  (manual replay via POST /webhooks/deliveries/:id/replay)
+ */
+export const webhookQueue = sqliteTable("webhook_queue", {
+  id: text("id").primaryKey(),
+  webhookId: text("webhook_id").notNull(),
+  linkId: text("link_id").notNull(),
+  event: text("event").notNull(),
+  /** JSON-serialised event payload — the exact body that will be signed & sent. */
+  payload: text("payload").notNull(),
+  attempts: integer("attempts").notNull().default(0),
+  nextAttemptAt: integer("next_attempt_at").notNull(),
+  /** pending | claimed | delivered | dead */
+  status: text("status").notNull().default("pending"),
+  lastStatusCode: integer("last_status_code"),
+  lastError: text("last_error"),
+  createdAt: integer("created_at").notNull(),
+  updatedAt: integer("updated_at").notNull(),
+});
+
 export const offrampQuotes = sqliteTable("offramp_quotes", {
   quoteId: text("quote_id").primaryKey(),
   linkId: text("link_id").notNull(),

apps/api/src/repos/index.ts

@@ -1,6 +1,4 @@
-import { and, desc, eq, inArray, isNotNull, isNull, lt } from "drizzle-orm";
-import type { ApiKeyScope } from "../services/api-keys";
-import { decodeScopesFromDb, encodeScopesForDb } from "../services/api-keys";
+import { eq, and, inArray, lt, lte } from "drizzle-orm";
 import type {
   CreateLinkInput,
   KycFieldSpec,
@@ -18,6 +16,7 @@ import type {
   StoredOffRampQuote,
   Webhook,
   WebhookDelivery,
+  WebhookQueueEntry,
   WebhookRepository,
   WatcherStateRepository,
   OffRampTelemetryRepository,
@@ -33,6 +32,7 @@ import {
   sellers,
   webhooks,
   webhookDeliveries,
+  webhookQueue,
   watcherCursors,
   processedTx,
   offrampQuotes,
@@ -402,85 +402,124 @@ export class DrizzleWebhookRepository implements WebhookRepository {
     return rows.map(rowToWebhook);
   }
 
-  async getById(id: string, sellerId: string, opts?: { includeDeleted?: boolean }): Promise<Webhook | null> {
-    const conditions = [eq(webhooks.id, id), eq(webhooks.sellerId, sellerId)];
-    if (!opts?.includeDeleted) conditions.push(isNull(webhooks.deletedAt));
-    const rows = await this.db
-      .select()
-      .from(webhooks)
-      .where(and(...conditions))
-      .limit(1);
-    return rows[0] ? rowToWebhook(rows[0]) : null;
+  async findWebhookById(id: string): Promise<Webhook | null> {
+    const rows = await this.db.select().from(webhooks).where(eq(webhooks.id, id)).limit(1);
+    return rows[0] ?? null;
   }
 
-  async rotateSecret(id: string, sellerId: string, newSecret: string, overlapMs: number): Promise<Webhook | null> {
-    const existing = await this.getById(id, sellerId);
-    if (!existing) return null;
-
-    const updated = {
-      secretEncrypted: encryptSecret(newSecret),
-      secretLast4: last4(newSecret),
-      previousSecretEncrypted: existing.secretEncrypted,
-      previousSecretLast4: existing.secretLast4,
-      previousSecretExpiresAt: Date.now() + overlapMs,
-    };
-    await this.db
-      .update(webhooks)
-      .set(updated)
-      .where(and(eq(webhooks.id, id), eq(webhooks.sellerId, sellerId)));
-
-    return { ...existing, ...updated };
-  }
-
-  async softDelete(id: string, sellerId: string): Promise<boolean> {
-    const result = await this.db
-      .update(webhooks)
-      .set({ deletedAt: Date.now() })
-      .where(and(eq(webhooks.id, id), eq(webhooks.sellerId, sellerId), isNull(webhooks.deletedAt)));
-    return (result.rowsAffected ?? 0) > 0;
-  }
-
-  async recordDelivery(d: Omit<WebhookDelivery, "id" | "createdAt">): Promise<void> {
+  async recordDelivery(d: WebhookDelivery): Promise<void> {
     await this.db.insert(webhookDeliveries).values({
       id: newId("whd"),
       webhookId: d.webhookId,
       linkId: d.linkId,
       event: d.event,
+      attempt: d.attempt,
+      queueEntryId: d.queueEntryId,
       statusCode: d.statusCode,
       ok: d.ok,
       error: d.error,
       createdAt: Date.now(),
     });
   }
 
-  async listDeliveries(
-    webhookId: string,
-    sellerId: string,
-    opts: { limit: number; cursor?: string | null },
-  ): Promise<{ deliveries: WebhookDelivery[]; nextCursor: string | null }> {
-    // Ownership check — a merchant may only read deliveries for their own
-    // webhook. Deleted webhooks are included on purpose: history must stay
-    // visible after an endpoint is removed.
-    const owned = await this.getById(webhookId, sellerId, { includeDeleted: true });
-    if (!owned) return { deliveries: [], nextCursor: null };
+  // ---------------------------------------------------------------------------
+  // Queue operations
+  // ---------------------------------------------------------------------------
 
-    const cursorCreatedAt = opts.cursor ? decodeDeliveryCursor(opts.cursor) : null;
-    const conditions = [eq(webhookDeliveries.webhookId, webhookId)];
-    if (cursorCreatedAt !== null) conditions.push(lt(webhookDeliveries.createdAt, cursorCreatedAt));
+  async enqueue(
+    entry: Omit<WebhookQueueEntry, "attempts" | "status" | "lastStatusCode" | "lastError" | "updatedAt">,
+  ): Promise<WebhookQueueEntry> {
+    const now = Date.now();
+    const row = {
+      id: entry.id,
+      webhookId: entry.webhookId,
+      linkId: entry.linkId,
+      event: entry.event,
+      payload: entry.payload,
+      attempts: 0,
+      nextAttemptAt: entry.nextAttemptAt,
+      status: "pending" as const,
+      lastStatusCode: null,
+      lastError: null,
+      createdAt: entry.createdAt,
+      updatedAt: now,
+    };
+    await this.db.insert(webhookQueue).values(row);
+    return row;
+  }
 
-    // Fetch one extra row to know whether there's a next page.
-    const rows = await this.db
+  /**
+   * Claim up to `limit` pending rows whose next_attempt_at <= now.
+   *
+   * SQLite is single-writer, so the read-then-update within a single synchronous
+   * call is safe against concurrent processes sharing the same file. For a
+   * multi-process / Turso setup the `status = 'claimed'` write acts as an
+   * optimistic lock: if two workers race, only one's UPDATE will match the row
+   * (the other will find status ≠ 'pending' on the next SELECT and skip it).
+   */
+  async claimDue(now: number, limit: number): Promise<WebhookQueueEntry[]> {
+    // 1. Find candidates.
+    const candidates = await this.db
       .select()
-      .from(webhookDeliveries)
-      .where(and(...conditions))
-      .orderBy(desc(webhookDeliveries.createdAt))
-      .limit(opts.limit + 1);
+      .from(webhookQueue)
+      .where(
+        and(
+          eq(webhookQueue.status, "pending"),
+          lte(webhookQueue.nextAttemptAt, now),
+        ),
+      )
+      .limit(limit);
 
-    const page = rows.slice(0, opts.limit);
-    const last = page[page.length - 1];
-    const nextCursor = rows.length > opts.limit && last ? encodeDeliveryCursor(last.createdAt) : null;
+    if (candidates.length === 0) return [];
+
+    const ids = candidates.map((r) => r.id);
+
+    // 2. Atomically transition pending → claimed and take the affected rows
+    //    straight off the UPDATE via RETURNING.
+    //
+    //    RETURNING is what makes this safe across instances: it yields exactly
+    //    the rows *this* statement transitioned. Re-SELECTing status='claimed'
+    //    afterwards would also match rows a concurrent worker had just claimed
+    //    between our UPDATE and our SELECT, and both workers would deliver the
+    //    same webhook.
+    const claimed = await this.db
+      .update(webhookQueue)
+      .set({ status: "claimed", updatedAt: Date.now() })
+      .where(
+        and(
+          inArray(webhookQueue.id, ids),
+          eq(webhookQueue.status, "pending"),
+        ),
+      )
+      .returning();
 
-    return { deliveries: page, nextCursor };
+    return claimed.map(rowToQueueEntry);
+  }
+
+  async updateQueueEntry(
+    id: string,
+    patch: Pick<WebhookQueueEntry, "status" | "attempts" | "nextAttemptAt" | "lastStatusCode" | "lastError">,
+  ): Promise<void> {
+    await this.db
+      .update(webhookQueue)
+      .set({
+        status: patch.status,
+        attempts: patch.attempts,
+        nextAttemptAt: patch.nextAttemptAt,
+        lastStatusCode: patch.lastStatusCode,
+        lastError: patch.lastError,
+        updatedAt: Date.now(),
+      })
+      .where(eq(webhookQueue.id, id));
+  }
+
+  async findQueueEntry(id: string): Promise<WebhookQueueEntry | null> {
+    const rows = await this.db
+      .select()
+      .from(webhookQueue)
+      .where(eq(webhookQueue.id, id))
+      .limit(1);
+    return rows[0] ? rowToQueueEntry(rows[0]) : null;
   }
 
   async listDeliveriesByLinkId(linkId: string): Promise<WebhookDelivery[]> {
@@ -494,6 +533,8 @@ export class DrizzleWebhookRepository implements WebhookRepository {
       webhookId: r.webhookId,
       linkId: r.linkId,
       event: r.event,
+      attempt: r.attempt,
+      queueEntryId: r.queueEntryId,
       statusCode: r.statusCode,
       ok: r.ok,
       error: r.error,
@@ -502,6 +543,23 @@ export class DrizzleWebhookRepository implements WebhookRepository {
   }
 }
 
+type QueueRow = typeof webhookQueue.$inferSelect;
+
+function rowToQueueEntry(row: QueueRow): WebhookQueueEntry {
+  return {
+    id: row.id,
+    webhookId: row.webhookId,
+    linkId: row.linkId,
+    event: row.event,
+    payload: row.payload,
+    attempts: row.attempts,
+    nextAttemptAt: row.nextAttemptAt,
+    status: row.status as WebhookQueueEntry["status"],
+    lastStatusCode: row.lastStatusCode,
+    lastError: row.lastError,
+    createdAt: row.createdAt,
+    updatedAt: row.updatedAt,
+  };
 function encodeDeliveryCursor(createdAt: number): string {
   return Buffer.from(String(createdAt), "utf8").toString("base64url");
 }

apps/api/src/routes/webhooks.ts

@@ -111,5 +111,55 @@ export function webhookRoutes(c: Container): Hono<{ Variables: AuthVariables }>
     return ctx.json({ deliveries, nextCursor });
   });
 
+  /**
+   * POST /webhooks/deliveries/:id/replay
+   *
+   * Re-enqueues a dead-lettered (or any) queue entry for immediate redelivery.
+   * Returns 202 Accepted with the updated entry summary. The actual delivery
+   * happens on the next WebhookWorker tick (within seconds).
+   *
+   * Idempotent if called on an entry that is already pending or delivered:
+   *   - pending   → next_attempt_at reset to now (no-op on next tick if already 0)
+   *   - delivered → re-queued as pending (manual replay of a successful delivery)
+   *   - dead      → re-queued as pending (the primary use-case)
+   *   - claimed   → 409 (delivery is in-flight; wait for it to settle first)
+   */
+  app.post("/deliveries/:id/replay", async (ctx) => {
+    const id = ctx.req.param("id");
+    const entry = await c.webhooks.findQueueEntry(id);
+
+    if (!entry) {
+      return ctx.json({ error: "not_found", message: `No delivery queue entry with id "${id}"` }, 404);
+    }
+
+    if (entry.status === "claimed") {
+      return ctx.json(
+        { error: "in_flight", message: "Delivery is currently in-flight; wait for it to settle before replaying." },
+        409,
+      );
+    }
+
+    await c.webhooks.updateQueueEntry(id, {
+      status: "pending",
+      attempts: entry.attempts, // preserve history count; worker increments on next attempt
+      nextAttemptAt: Date.now(),
+      lastStatusCode: entry.lastStatusCode,
+      lastError: entry.lastError,
+    });
+
+    return ctx.json(
+      {
+        id: entry.id,
+        webhookId: entry.webhookId,
+        linkId: entry.linkId,
+        event: entry.event,
+        previousAttempts: entry.attempts,
+        status: "pending",
+        message: "Queued for immediate redelivery.",
+      },
+      202,
+    );
+  });
+
   return app;
 }