Merge pull request #297 from lorenzo-romano/feat/issue-165-webhook-lag

feat(webhooks): add dead-letter status and lag metrics to the delivery queue (#165)

Author: wagmiiii

apps/web/src/app/api/webhooks/deliver/route.ts

@@ -1,6 +1,6 @@
 import { NextResponse } from 'next/server';
 import { withClient, ensureSchema } from '@/lib/db';
-import { deliverDue } from '@/lib/webhooks';
+import { deliverDue, pendingDue } from '@/lib/webhooks';
 
 export const dynamic = 'force-dynamic';
 export const maxDuration = 30;
@@ -23,7 +23,11 @@ export async function GET(request: Request) {
   try {
     const result = await withClient(async (client) => {
       await ensureSchema(client);
-      return deliverDue(client);
+      const outcome = await deliverDue(client);
+      // Remaining lag after this run: the signal a scheduler uses to scale
+      // consumer frequency to backlog (#165).
+      const lag = await pendingDue(client);
+      return { ...outcome, lag };
     });
     return NextResponse.json({ success: true, ...result });
   } catch (error: unknown) {

apps/web/src/app/api/webhooks/route.ts

@@ -14,7 +14,9 @@ export async function GET() {
       configured: false,
       pending: 0,
       failed: 0,
+      deadLetter: 0,
       delivered: 0,
+      lag: 0,
       recentFailed: [],
     });
   }

apps/web/src/lib/webhooks.test.ts

@@ -10,6 +10,8 @@ import {
   deliverDue,
   enqueueWebhookDelivery,
   payloadFromRow,
+  pendingDue,
+  webhookSummary,
 } from './webhooks';
 
 describe('shouldRetry', () => {
@@ -176,3 +178,78 @@ describe('deliverDue — a sleeping host cannot stall the caller past the budget
     expect(elapsed).toBeLessThan(5_000);
   });
 });
+
+describe('pendingDue — the lag signal a consumer fleet scales on (#165)', () => {
+  it('counts only pending rows whose retry time has passed (or never set)', async () => {
+    const queries: string[] = [];
+    const query = vi.fn(async (sql: string) => {
+      queries.push(sql);
+      return { rows: [{ count: '7' }] };
+    });
+
+    const lag = await pendingDue({ query } as never, { now: new Date('2026-08-01T00:00:00Z') });
+
+    expect(lag).toBe(7);
+    const sql = queries[0];
+    expect(sql).toContain("status = 'pending'");
+    // Null next_retry_at (never attempted) is always due.
+    expect(sql).toContain('next_retry_at IS NULL OR next_retry_at <= $1::timestamptz');
+  });
+});
+
+describe('webhookSummary', () => {
+  it('reports lag and the dead-letter count alongside the existing tallies', async () => {
+    const query = vi.fn(async (sql: string) => {
+      if (/^SELECT status, count/.test(sql)) {
+        return {
+          rows: [
+            { status: 'pending', n: '2' },
+            { status: 'delivered', n: '5' },
+            { status: 'dead_letter', n: '3' },
+          ],
+        };
+      }
+      if (/^SELECT count\(\*\)::text AS count/.test(sql)) {
+        return { rows: [{ count: '2' }] };
+      }
+      return { rows: [] };
+    });
+
+    const summary = await webhookSummary({ query } as never);
+
+    expect(summary.pending).toBe(2);
+    expect(summary.delivered).toBe(5);
+    expect(summary.deadLetter).toBe(3);
+    expect(summary.lag).toBe(2);
+    expect(summary.recentFailed).toEqual([]);
+  });
+
+  it('lists dead-lettered deliveries in recentFailed for operator inspection', async () => {
+    const query = vi.fn(async (sql: string) => {
+      if (/^SELECT status, count/.test(sql)) return { rows: [{ status: 'dead_letter', n: '1' }] };
+      if (/^SELECT count\(\*\)::text AS count/.test(sql)) return { rows: [{ count: '0' }] };
+      if (/^SELECT id, payment_tx_hash/.test(sql)) {
+        return {
+          rows: [
+            {
+              id: '9',
+              payment_tx_hash: 'a'.repeat(64),
+              status: 'dead_letter',
+              attempts: 8,
+              last_status_code: 503,
+              last_error: 'HTTP 503',
+              updated_at: new Date('2026-08-01T00:00:00.000Z'),
+            },
+          ],
+        };
+      }
+      return { rows: [] };
+    });
+
+    const summary = await webhookSummary({ query } as never);
+
+    expect(summary.deadLetter).toBe(1);
+    expect(summary.recentFailed[0].status).toBe('dead_letter');
+    expect(summary.recentFailed[0].attempts).toBe(8);
+  });
+});

apps/web/src/lib/webhooks.ts

@@ -15,7 +15,12 @@ export const DELIVERY_WINDOW_MS = 24 * 60 * 60 * 1000;
 export const ATTEMPT_TIMEOUT_MS = 2_000;
 export const MAX_BACKOFF_MS = 60 * 60 * 1000;
 
-export type DeliveryStatus = 'pending' | 'delivering' | 'delivered' | 'failed';
+export type DeliveryStatus =
+  | 'pending'
+  | 'delivering'
+  | 'delivered'
+  | 'failed'
+  | 'dead_letter';
 
 export interface PaymentPayload {
   tx_hash: string;
@@ -274,7 +279,9 @@ export async function deliverDue(
       transportError,
     });
     if (terminal.status === 'delivered') delivered++;
-    else if (terminal.status === 'failed') failed++;
+    // A dead-lettered row is terminal, not a retry — count it as failed here
+    // so the run's tallies reflect deliveries that gave up (#165).
+    else if (terminal.status === 'failed' || terminal.status === 'dead_letter') failed++;
     else retried++;
   }
 
@@ -308,7 +315,11 @@ async function recordAttempt(
   let status: DeliveryStatus;
   if (ok) status = 'delivered';
   else if (next) status = 'pending';
-  else status = 'failed';
+  // A delivery that exhausts its attempt budget (or its 24h delivery window)
+  // is dead-lettered: it is kept for operator inspection and never retried
+  // again (#165). This is the queue's explicit dead-letter state, distinct
+  // from a transient 'failed' row.
+  else status = 'dead_letter';
 
   await client.query(
     `INSERT INTO webhook_attempts (delivery_id, attempt_number, status_code, error)
@@ -332,10 +343,37 @@ async function recordAttempt(
   return { id: input.id, status, statusCode: input.statusCode, error: input.error };
 }
 
+/**
+ * The queue's lag: deliveries that are due for (re)delivery right now.
+ *
+ * This is the signal a consumer fleet auto-scales on (#165) — when lag stays
+ * high, schedule more frequent or overlapping `/api/webhooks/deliver` runs;
+ * when it is zero, the queue is drained. Rows with a null `next_retry_at`
+ * (never attempted) are always due; the rest are due once their retry time
+ * has passed.
+ */
+export async function pendingDue(
+  client: Client,
+  opts: { now?: Date } = {},
+): Promise<number> {
+  const now = (opts.now ?? new Date()).toISOString();
+  const res = await client.query<{ count: string }>(
+    `SELECT count(*)::text AS count
+     FROM webhook_deliveries
+     WHERE status = 'pending'
+       AND (next_retry_at IS NULL OR next_retry_at <= $1::timestamptz)`,
+    [now],
+  );
+  return Number(res.rows[0]?.count ?? 0);
+}
+
 export async function webhookSummary(client: Client): Promise<{
   pending: number;
   failed: number;
+  deadLetter: number;
   delivered: number;
+  /** Deliveries due right now — the lag a consumer fleet scales on (#165). */
+  lag: number;
   recentFailed: Array<{
     id: number;
     paymentTxHash: string;
@@ -349,7 +387,12 @@ export async function webhookSummary(client: Client): Promise<{
   const counts = await client.query<{ status: string; n: string }>(
     `SELECT status, count(*)::text AS n FROM webhook_deliveries GROUP BY status`,
   );
-  const byStatus: Record<string, number> = { pending: 0, failed: 0, delivered: 0 };
+  const byStatus: Record<string, number> = {
+    pending: 0,
+    failed: 0,
+    delivered: 0,
+    dead_letter: 0,
+  };
   for (const row of counts.rows) byStatus[row.status] = Number(row.n);
 
   const recent = await client.query<{
@@ -363,15 +406,17 @@ export async function webhookSummary(client: Client): Promise<{
   }>(
     `SELECT id, payment_tx_hash, status, attempts, last_status_code, last_error, updated_at
      FROM webhook_deliveries
-     WHERE status = 'failed'
+     WHERE status IN ('failed', 'dead_letter')
      ORDER BY updated_at DESC
      LIMIT 20`,
   );
 
   return {
     pending: (byStatus.pending ?? 0) + (byStatus.delivering ?? 0),
     failed: byStatus.failed ?? 0,
+    deadLetter: byStatus.dead_letter ?? 0,
     delivered: byStatus.delivered ?? 0,
+    lag: await pendingDue(client),
     recentFailed: recent.rows.map((row) => ({
       id: Number(row.id),
       paymentTxHash: row.payment_tx_hash,