feat(dlq): add dead-letter isolation and operator remediation system (#706)

docs/DEAD_LETTER_HANDLING.md

@@ -0,0 +1,32 @@
+# 📮 Dead-Letter Isolation & Operator Remediation
+
+This document details the Dead Letter Queue (DLQ) isolation and remediation architecture for NotifyChain (Issue #706).
+
+---
+
+## 1. Overview & Non-blocking Guarantee
+
+When a notification fails delivery and exhausts its maximum retry quota, it is immediately moved out of the active dispatch loop into the DLQ. This prevents poison-pill payloads or persistently failing endpoints from causing head-of-line blocking for other notifications.
+
+---
+
+## 2. DLQ Record Schema
+
+| Field | Type | Description |
+|---|---|---|
+| `dlqId` | `string` | Unique identifier (`dlq-uuid`) |
+| `originalNotificationId` | `string` | Foreign key linking to the source notification |
+| `provider` | `string` | Target transport (e.g. `Discord`, `Webhook`) |
+| `targetRecipientSanitized` | `string` | Redacted recipient URL/address |
+| `failureReason` | `string` | Sanitized error message |
+| `errorStack` | `string` | Redacted error stack trace |
+| `retryAttempts` | `number` | Total number of attempts executed before isolation |
+| `status` | `string` | `isolated` \| `requeued` \| `discarded` |
+
+---
+
+## 3. Operator Remediation
+
+* **Inspection**: Query dead-lettered entries via `listDeadLetters()`.
+* **Requeueing**: Re-inject isolated payloads back into the dispatch loop via `requeue(dlqId)`.
+* **Discarding**: Mark permanently invalid webhooks as discarded via `discard(dlqId)`.

listener/src/services/dead-letter-manager.test.ts

@@ -0,0 +1,68 @@
+import { DeadLetterManager } from './dead-letter-manager';
+
+describe('Dead-Letter Isolation & Operator Diagnostics (Issue #706)', () => {
+  let manager: DeadLetterManager;
+
+  beforeEach(() => {
+    manager = new DeadLetterManager();
+  });
+
+  test('isolates failed notifications with redacted secret information', () => {
+    const entry = manager.isolateNotification({
+      originalNotificationId: 'notif-12345',
+      provider: 'Discord',
+      targetRecipient: 'https://discord.com/api/webhooks/999/SecretTokenABC',
+      error: new Error('Discord rate limited (429) on secret-endpoint'),
+      retryAttempts: 5,
+      payload: { eventId: 'evt-001', amount: '100' },
+    });
+
+    expect(entry.dlqId).toMatch(/^dlq-/);
+    expect(entry.originalNotificationId).toBe('notif-12345');
+    expect(entry.status).toBe('isolated');
+    expect(entry.targetRecipientSanitized).not.toContain('SecretTokenABC');
+    expect(entry.retryAttempts).toBe(5);
+  });
+
+  test('allows operator inspection and listing with status filters', () => {
+    manager.isolateNotification({
+      originalNotificationId: 'n1',
+      provider: 'Discord',
+      targetRecipient: 'hook1',
+      error: 'timeout',
+      retryAttempts: 3,
+      payload: {},
+    });
+
+    manager.isolateNotification({
+      originalNotificationId: 'n2',
+      provider: 'Webhook',
+      targetRecipient: 'hook2',
+      error: '500 Internal Server Error',
+      retryAttempts: 5,
+      payload: {},
+    });
+
+    const list = manager.listDeadLetters({ provider: 'Discord' });
+    expect(list.total).toBe(1);
+    expect(list.entries[0].provider).toBe('Discord');
+  });
+
+  test('supports requeuing and metrics tracking', () => {
+    const entry = manager.isolateNotification({
+      originalNotificationId: 'n-requeue',
+      provider: 'Webhook',
+      targetRecipient: 'hook',
+      error: 'temp network failure',
+      retryAttempts: 3,
+      payload: {},
+    });
+
+    expect(manager.getMetrics().activeDepth).toBe(1);
+
+    const requeued = manager.requeue(entry.dlqId);
+    expect(requeued.status).toBe('requeued');
+    expect(manager.getMetrics().activeDepth).toBe(0);
+    expect(manager.getMetrics().requeuedCount).toBe(1);
+  });
+});

listener/src/services/dead-letter-manager.ts

@@ -0,0 +1,129 @@
+/**
+ * Dead-Letter Isolation & Remediation System (Issue #706)
+ *
+ * Isolates exhausted or unrecoverable notifications into a dedicated DLQ store,
+ * preventing pipeline head-of-line blocking while maintaining full operator traceability.
+ */
+
+import { randomUUID } from 'crypto';
+import { redactString } from '../utils/redact';
+
+export interface DeadLetterEntry {
+  dlqId: string;
+  originalNotificationId: string;
+  provider: string;
+  targetRecipientSanitized: string;
+  failureReason: string;
+  errorStack?: string;
+  retryAttempts: number;
+  payload: Record<string, unknown>;
+  isolatedAt: string;
+  status: 'isolated' | 'requeued' | 'discarded';
+}
+
+export class DeadLetterManager {
+  private readonly store = new Map<string, DeadLetterEntry>();
+
+  /**
+   * Isolates a failed notification into the Dead Letter Queue.
+   */
+  public isolateNotification(params: {
+    originalNotificationId: string;
+    provider: string;
+    targetRecipient: string;
+    error: Error | string;
+    retryAttempts: number;
+    payload: Record<string, unknown>;
+  }): DeadLetterEntry {
+    const dlqId = `dlq-${randomUUID()}`;
+    const rawReason = params.error instanceof Error ? params.error.message : String(params.error);
+    const rawStack = params.error instanceof Error ? params.error.stack : undefined;
+
+    const entry: DeadLetterEntry = {
+      dlqId,
+      originalNotificationId: params.originalNotificationId,
+      provider: params.provider,
+      targetRecipientSanitized: redactString(params.targetRecipient),
+      failureReason: redactString(rawReason),
+      errorStack: rawStack ? redactString(rawStack) : undefined,
+      retryAttempts: params.retryAttempts,
+      payload: params.payload,
+      isolatedAt: new Date().toISOString(),
+      status: 'isolated',
+    };
+
+    this.store.set(dlqId, entry);
+    return entry;
+  }
+
+  /**
+   * Retrieves a single DLQ entry for operator diagnostic inspection.
+   */
+  public getDeadLetter(dlqId: string): DeadLetterEntry | undefined {
+    return this.store.get(dlqId);
+  }
+
+  /**
+   * Lists isolated dead-letter entries with pagination and provider filtering.
+   */
+  public listDeadLetters(options: {
+    provider?: string;
+    status?: 'isolated' | 'requeued' | 'discarded';
+    limit?: number;
+    offset?: number;
+  } = {}): { entries: DeadLetterEntry[]; total: number } {
+    let all = Array.from(this.store.values());
+
+    if (options.provider) {
+      all = all.filter((e) => e.provider === options.provider);
+    }
+    if (options.status) {
+      all = all.filter((e) => e.status === options.status);
+    }
+
+    const total = all.length;
+    const offset = options.offset || 0;
+    const limit = options.limit || 50;
+
+    return {
+      entries: all.slice(offset, offset + limit),
+      total,
+    };
+  }
+
+  /**
+   * Marks a dead-letter entry as requeued for pipeline re-processing.
+   */
+  public requeue(dlqId: string): DeadLetterEntry {
+    const entry = this.store.get(dlqId);
+    if (!entry) {
+      throw new Error(`DLQ entry '${dlqId}' not found.`);
+    }
+    entry.status = 'requeued';
+    return entry;
+  }
+
+  /**
+   * Discards / acknowledges a dead-letter entry without re-delivery.
+   */
+  public discard(dlqId: string): DeadLetterEntry {
+    const entry = this.store.get(dlqId);
+    if (!entry) {
+      throw new Error(`DLQ entry '${dlqId}' not found.`);
+    }
+    entry.status = 'discarded';
+    return entry;
+  }
+
+  /**
+   * Returns DLQ metrics for monitoring dashboards.
+   */
+  public getMetrics(): { totalIsolated: number; activeDepth: number; requeuedCount: number } {
+    const all = Array.from(this.store.values());
+    return {
+      totalIsolated: all.length,
+      activeDepth: all.filter((e) => e.status === 'isolated').length,
+      requeuedCount: all.filter((e) => e.status === 'requeued').length,
+    };
+  }
+}