From 6e5e0ff82a7ce521dacb57ad52555d02c344057f Mon Sep 17 00:00:00 2001 From: Namiiikaze Date: Sat, 25 Jul 2026 05:29:08 -0700 Subject: [PATCH] Implemented dead-letter queue processing for failed notifications with persistent DLQ storage, retry/requeue support, and health-monitor visibility for DLQ growth --- listener/package.json | 3 +- listener/src/database/schema.sql | 21 ++++ listener/src/index.ts | 17 ++- .../src/services/dead-letter-queue.test.ts | 83 +++++++++++++ listener/src/services/notification-api.ts | 14 +++ .../services/notification-health-monitor.ts | 24 +++- .../scheduled-notification-repository.ts | 115 ++++++++++++++++++ listener/src/types/scheduled-notification.ts | 13 ++ 8 files changed, 282 insertions(+), 8 deletions(-) create mode 100644 listener/src/services/dead-letter-queue.test.ts diff --git a/listener/package.json b/listener/package.json index 1de3a88..ab17054 100644 --- a/listener/package.json +++ b/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" }, diff --git a/listener/src/database/schema.sql b/listener/src/database/schema.sql index 57612ff..a965749 100644 --- a/listener/src/database/schema.sql +++ b/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, diff --git a/listener/src/index.ts b/listener/src/index.ts index 9fe2b34..3ef63c5 100644 --- a/listener/src/index.ts +++ b/listener/src/index.ts @@ -43,6 +43,7 @@ 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 { @@ -50,13 +51,17 @@ async function main() { 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,7 +181,9 @@ async function main() { healthMonitor, }); - healthMonitor.start(); + if (healthMonitor) { + healthMonitor.start(); + } const subscriber = new EventSubscriber(config); await subscriber.start(); @@ -184,7 +191,9 @@ async function main() { const shutdown = async () => { logger.info('Shutting down services...'); - healthMonitor.stop(); + if (healthMonitor) { + healthMonitor.stop(); + } if (cleanupService) { await cleanupService.stop(); diff --git a/listener/src/services/dead-letter-queue.test.ts b/listener/src/services/dead-letter-queue.test.ts new file mode 100644 index 0000000..1462015 --- /dev/null +++ b/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); + }); +}); diff --git a/listener/src/services/notification-api.ts b/listener/src/services/notification-api.ts index 6e8fe6f..b2d00eb 100644 --- a/listener/src/services/notification-api.ts +++ b/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 { + return await this.repository.retryDeadLetterNotification(id, requestId); + } } diff --git a/listener/src/services/notification-health-monitor.ts b/listener/src/services/notification-health-monitor.ts index 47a1159..a7eb280 100644 --- a/listener/src/services/notification-health-monitor.ts +++ b/listener/src/services/notification-health-monitor.ts @@ -2,6 +2,7 @@ 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'; @@ -9,6 +10,7 @@ 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 | 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 { diff --git a/listener/src/services/scheduled-notification-repository.ts b/listener/src/services/scheduled-notification-repository.ts index d727108..765a9a8 100644 --- a/listener/src/services/scheduled-notification-repository.ts +++ b/listener/src/services/scheduled-notification-repository.ts @@ -7,6 +7,7 @@ import { CreateScheduledNotificationInput, NotificationStatus, NotificationExecutionLog, + DeadLetterQueueEntry, } from '../types/scheduled-notification'; import { hashPayload } from '../utils/payload-integrity'; @@ -278,6 +279,10 @@ export class ScheduledNotificationRepository { id, ]); + if (isFailed) { + await this.moveToDeadLetterQueue(id, error, errorDetails, nextRetryCount); + } + logger.info('Notification marked for retry or failed', { id, newStatus, @@ -287,6 +292,113 @@ export class ScheduledNotificationRepository { }); } + async moveToDeadLetterQueue( + id: number, + error: Error, + errorDetails: string, + retryCount: number, + requestId?: string + ): Promise { + const notification = await this.getById(id); + if (!notification) { + return false; + } + + const insertSql = ` + INSERT INTO dead_letter_queue ( + scheduled_notification_id, notification_type, target_recipient, payload, failure_reason, error_details, retry_count + ) VALUES (?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(scheduled_notification_id) DO UPDATE SET + failure_reason = excluded.failure_reason, + error_details = excluded.error_details, + retry_count = excluded.retry_count, + last_retried_at = NULL + `; + + await this.db.run(insertSql, [ + id, + notification.notificationType, + notification.targetRecipient, + typeof notification.payload === 'string' ? notification.payload : JSON.stringify(notification.payload), + error.message, + errorDetails, + retryCount, + ]); + + logger.warn('Notification moved to dead letter queue', { + requestId, + id, + notificationType: notification.notificationType, + failureReason: error.message, + }); + + return true; + } + + async getDeadLetterQueue(): Promise { + const sql = ` + SELECT id, scheduled_notification_id, notification_type, target_recipient, payload, failure_reason, error_details, created_at, last_retried_at, retry_count + FROM dead_letter_queue + ORDER BY created_at DESC + `; + + const rows = await this.db.all<{ + id: number; + scheduled_notification_id: number; + notification_type: string; + target_recipient: string; + payload: string; + failure_reason: string; + error_details: string | null; + created_at: string; + last_retried_at: string | null; + retry_count: number; + }>(sql); + + return rows.map((row) => ({ + id: row.id, + originalNotificationId: row.scheduled_notification_id, + notificationType: row.notification_type as any, + targetRecipient: row.target_recipient, + payload: row.payload, + failureReason: row.failure_reason, + errorDetails: row.error_details, + createdAt: new Date(row.created_at), + lastRetriedAt: row.last_retried_at ? new Date(row.last_retried_at) : null, + retryCount: row.retry_count, + })); + } + + async retryDeadLetterNotification(id: number, requestId?: string): Promise { + const entry = await this.db.get<{ scheduled_notification_id: number }>( + 'SELECT scheduled_notification_id FROM dead_letter_queue WHERE id = ?', + [id] + ); + + if (!entry) { + return false; + } + + await this.db.transaction(async () => { + await this.db.run( + ` + UPDATE scheduled_notifications + SET status = ?, next_retry_at = NULL, updated_at = ?, retry_count = 0, last_error = NULL, error_details = NULL + WHERE id = ? + `, + [NotificationStatus.PENDING, new Date().toISOString(), entry.scheduled_notification_id] + ); + + await this.db.run( + `UPDATE dead_letter_queue SET last_retried_at = ?, retry_count = retry_count + 1 WHERE id = ?`, + [new Date().toISOString(), id] + ); + }); + + logger.info('Dead-letter notification requeued', { requestId, id, notificationId: entry.scheduled_notification_id }); + return true; + } + /** * Fetch PENDING notifications whose next_retry_at is due (or null, meaning immediately schedulable). * Used by RetryScheduler to pick up failed notifications for re-attempt. @@ -462,6 +574,7 @@ export class ScheduledNotificationRepository { completed: number; failed: number; overdue: number; + deadLetterQueue: number; }> { const now = new Date().toISOString(); @@ -485,6 +598,7 @@ export class ScheduledNotificationRepository { const counts = await this.db.all<{ adjusted_status: string; count: number }>(countBySql, [now]); const overdueResult = await this.db.get<{ count: number }>(overdueSql, [now, now]); + const dlqResult = await this.db.get<{ count: number }>('SELECT COUNT(*) as count FROM dead_letter_queue'); const stats = { pending: 0, @@ -492,6 +606,7 @@ export class ScheduledNotificationRepository { completed: 0, failed: 0, overdue: overdueResult?.count ?? 0, + deadLetterQueue: dlqResult?.count ?? 0, }; counts.forEach((row) => { diff --git a/listener/src/types/scheduled-notification.ts b/listener/src/types/scheduled-notification.ts index 371eb95..cdaa92d 100644 --- a/listener/src/types/scheduled-notification.ts +++ b/listener/src/types/scheduled-notification.ts @@ -100,3 +100,16 @@ export interface SchedulerConfig { timingBufferMs: number; retryDelayMs?: number; } + +export interface DeadLetterQueueEntry { + id?: number; + originalNotificationId: number; + notificationType: NotificationType; + targetRecipient: string; + payload: string; + failureReason: string; + errorDetails?: string | null; + createdAt?: Date; + lastRetriedAt?: Date | null; + retryCount: number; +}