Implemented dead-letter queue processing for failed notifications wit…

listener/package.json

@@ -10,10 +10,9 @@
     "lint": "node ./node_modules/typescript/bin/tsc --noEmit",
     "test": "node ./node_modules/jest/bin/jest.js",
     "migrate": "ts-node src/scripts/migrate-db.ts",
-    "migrate:templates": "ts-node src/scripts/migrate-templates.ts"
+    "migrate:templates": "ts-node src/scripts/migrate-templates.ts",
     "typecheck": "node ./node_modules/typescript/bin/tsc --noEmit",
     "lint": "node ./node_modules/typescript/bin/tsc --noEmit",
-    "migrate": "ts-node src/scripts/migrate-db.ts",
     "check-migrations": "ts-node src/scripts/check-migrations.ts",
     "validate:batch": "ts-node src/utils/batch-validator.ts"
   },

listener/src/database/schema.sql

@@ -65,6 +65,27 @@ CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_event_id
 CREATE INDEX IF NOT EXISTS idx_scheduled_notifications_target 
   ON scheduled_notifications(target_recipient, status);
 
+-- Dead-letter queue for permanently failed notifications
+CREATE TABLE IF NOT EXISTS dead_letter_queue (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  scheduled_notification_id INTEGER NOT NULL UNIQUE,
+  notification_type VARCHAR(50) NOT NULL,
+  target_recipient TEXT NOT NULL,
+  payload TEXT NOT NULL,
+  failure_reason TEXT NOT NULL,
+  error_details TEXT,
+  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  last_retried_at DATETIME,
+  retry_count INTEGER NOT NULL DEFAULT 0,
+  FOREIGN KEY (scheduled_notification_id) REFERENCES scheduled_notifications(id) ON DELETE CASCADE
+);
+
+CREATE INDEX IF NOT EXISTS idx_dead_letter_queue_created_at
+  ON dead_letter_queue(created_at);
+
+CREATE INDEX IF NOT EXISTS idx_dead_letter_queue_notification_type
+  ON dead_letter_queue(notification_type);
+
 -- Notification execution history for auditing
 CREATE TABLE IF NOT EXISTS notification_execution_log (
   id INTEGER PRIMARY KEY AUTOINCREMENT,

listener/src/index.ts

@@ -43,20 +43,25 @@ async function main() {
   let retryScheduler: RetryScheduler | null = null;
   let notificationAPI: NotificationAPI | null = null;
   let templateService: TemplateService | null = null;
+  let healthMonitor: NotificationHealthMonitor | null = null;
 
   if (config.scheduler?.enabled) {
     try {
       logger.info('Initializing database for scheduled notifications and templates');
       const db = await initializeDatabase(config.databasePath);
   let templateService: NotificationTemplateService | null = null;
   let cleanupService: CleanupService | null = null;
+  let repository: ScheduledNotificationRepository | null = null;
   let reconciliationEngine: IndexingReconciliationEngine | null = null;
   let archiveService: ArchiveService | null = null;
   let archiveStore: ArchiveStore | null = null;
   let metricsRunner: NotificationMetricsRunner | null = null;
   let metricsStore: NotificationMetricsStore | null = null;
 
-  const healthMonitor = new NotificationHealthMonitor(null, getWorkerManager());
+  repository = new ScheduledNotificationRepository(db);
+  healthMonitor = new NotificationHealthMonitor(null, getWorkerManager(), {
+    repository,
+  });
 
   if (config.analytics?.enabled) {
     initNotificationAnalyticsAggregator(config.analytics);
@@ -119,7 +124,7 @@ async function main() {
     templateService = new NotificationTemplateService(templateRepository);
 
     if (config.scheduler?.enabled) {
-      const repository = new ScheduledNotificationRepository(db);
+      repository = new ScheduledNotificationRepository(db);
       notificationAPI = new NotificationAPI(repository);
 
       // Initialize template service
@@ -176,15 +181,19 @@ async function main() {
     healthMonitor,
   });
 
-  healthMonitor.start();
+  if (healthMonitor) {
+    healthMonitor.start();
+  }
 
   const subscriber = new EventSubscriber(config);
   await subscriber.start();
 
   const shutdown = async () => {
     logger.info('Shutting down services...');
 
-    healthMonitor.stop();
+    if (healthMonitor) {
+      healthMonitor.stop();
+    }
 
     if (cleanupService) {
       await cleanupService.stop();

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

@@ -0,0 +1,83 @@
+import * as fs from 'fs';
+import * as path from 'path';
+import { Database } from '../database/database';
+import { ScheduledNotificationRepository } from './scheduled-notification-repository';
+import { NotificationStatus, NotificationType } from '../types/scheduled-notification';
+
+describe('Dead letter queue processing', () => {
+  const dbPath = './data/test-dead-letter-queue.db';
+  let db: Database;
+  let repository: ScheduledNotificationRepository;
+
+  beforeEach(async () => {
+    if (fs.existsSync(dbPath)) fs.unlinkSync(dbPath);
+    const dbDir = path.dirname(dbPath);
+    if (!fs.existsSync(dbDir)) fs.mkdirSync(dbDir, { recursive: true });
+
+    db = new Database(dbPath);
+    await db.initialize();
+    repository = new ScheduledNotificationRepository(db);
+  });
+
+  afterEach(async () => {
+    await db.close();
+    if (fs.existsSync(dbPath)) fs.unlinkSync(dbPath);
+  });
+
+  it('moves exhausted notifications into the dead letter queue', async () => {
+    const notificationId = await repository.create({
+      payload: { message: 'hello' },
+      notificationType: NotificationType.DISCORD,
+      targetRecipient: 'https://example.test/hook',
+      executeAt: new Date(Date.now() + 60_000),
+      maxRetries: 1,
+    });
+
+    await repository.markAsFailedOrRetry(
+      notificationId,
+      new Error('permanent failure'),
+      0,
+      1,
+      new Date(Date.now() + 5_000)
+    );
+
+    const dlqEntries = await repository.getDeadLetterQueue();
+    expect(dlqEntries).toHaveLength(1);
+    expect(dlqEntries[0].originalNotificationId).toBe(notificationId);
+    expect(dlqEntries[0].failureReason).toBe('permanent failure');
+
+    const row = await repository.getById(notificationId);
+    expect(row?.status).toBe(NotificationStatus.FAILED);
+  });
+
+  it('requeues a dead-lettered notification for retry', async () => {
+    const notificationId = await repository.create({
+      payload: { message: 'retry me' },
+      notificationType: NotificationType.EMAIL,
+      targetRecipient: 'test@example.com',
+      executeAt: new Date(Date.now() + 60_000),
+      maxRetries: 1,
+    });
+
+    await repository.markAsFailedOrRetry(
+      notificationId,
+      new Error('permanent failure'),
+      0,
+      1,
+      new Date(Date.now() + 5_000)
+    );
+
+    const [entry] = await repository.getDeadLetterQueue();
+    const requeued = await repository.retryDeadLetterNotification(entry.id!, 'req-2');
+
+    expect(requeued).toBe(true);
+
+    const row = await repository.getById(notificationId);
+    expect(row?.status).toBe(NotificationStatus.PENDING);
+    expect(row?.retryCount).toBe(0);
+    expect(row?.nextRetryAt).toBeNull();
+
+    const remainingEntries = await repository.getDeadLetterQueue();
+    expect(remainingEntries).toHaveLength(0);
+  });
+});

listener/src/services/notification-api.ts

@@ -133,4 +133,18 @@ export class NotificationAPI {
   async getRetryDistribution() {
     return await this.repository.getRetryDistribution();
   }
+
+  /**
+   * Get all notifications currently in the dead-letter queue.
+   */
+  async getDeadLetterQueue() {
+    return await this.repository.getDeadLetterQueue();
+  }
+
+  /**
+   * Requeue a dead-lettered notification so it can be retried again.
+   */
+  async retryDeadLetterNotification(id: number, requestId?: string): Promise<boolean> {
+    return await this.repository.retryDeadLetterNotification(id, requestId);
+  }
 }

listener/src/services/notification-health-monitor.ts

@@ -2,13 +2,15 @@ import logger from '../utils/logger';
 import { EventProcessingQueue } from './event-processing-queue';
 import { WorkerManager } from './worker-manager';
 import { eventRegistry } from '../store/event-registry';
+import { ScheduledNotificationRepository } from './scheduled-notification-repository';
 
 export type ComponentStatus = 'healthy' | 'degraded' | 'unhealthy';
 
 export interface QueueHealth {
   status: ComponentStatus;
   pendingJobs: number;
   stalledSince: number | null;
+  deadLetterQueueDepth: number;
 }
 
 export interface WorkerHealth {
@@ -41,6 +43,8 @@ export interface NotificationHealthMonitorOptions {
   maxProcessingDelayMs?: number;
   /** Injected clock for tests. */
   now?: () => number;
+  /** Optional repository used to surface DLQ depth in the health report. */
+  repository?: ScheduledNotificationRepository | null;
 }
 
 /**
@@ -58,6 +62,7 @@ export class NotificationHealthMonitor {
 
   private queue: EventProcessingQueue | null;
   private workerManager: WorkerManager | null;
+  private repository: ScheduledNotificationRepository | null;
 
   private timer: ReturnType<typeof setInterval> | null = null;
   private lastReport: HealthReport | null = null;
@@ -74,6 +79,7 @@ export class NotificationHealthMonitor {
   ) {
     this.queue = queue;
     this.workerManager = workerManager;
+    this.repository = options.repository ?? null;
     this.intervalMs = options.intervalMs ?? 30_000;
     this.stallThresholdCycles = options.stallThresholdCycles ?? 3;
     this.maxProcessingDelayMs = options.maxProcessingDelayMs ?? 60_000;
@@ -139,7 +145,7 @@ export class NotificationHealthMonitor {
 
   private checkQueue(): QueueHealth {
     if (!this.queue) {
-      return { status: 'healthy', pendingJobs: 0, stalledSince: null };
+      return { status: 'healthy', pendingJobs: 0, stalledSince: null, deadLetterQueueDepth: this.getDeadLetterQueueDepth() };
     }
 
     const pending = this.queue.pendingCount();
@@ -167,7 +173,21 @@ export class NotificationHealthMonitor {
       status = 'degraded';
     }
 
-    return { status, pendingJobs: pending, stalledSince: this.stalledSince };
+    return { status, pendingJobs: pending, stalledSince: this.stalledSince, deadLetterQueueDepth: this.getDeadLetterQueueDepth() };
+  }
+
+  private getDeadLetterQueueDepth(): number {
+    if (!this.repository) {
+      return 0;
+    }
+
+    try {
+      const stats = this.repository.getStats();
+      return (stats as any).deadLetterQueue ?? 0;
+    } catch (error) {
+      logger.warn('Unable to determine dead letter queue depth', { error });
+      return 0;
+    }
   }
 
   private checkWorkers(): WorkerHealth {