From 47fd4375ccc92e2b733a57bb9631ccaf29afc739 Mon Sep 17 00:00:00 2001 From: oak Date: Sun, 27 Sep 2026 00:17:06 +0100 Subject: [PATCH] feat(messaging): improve reliability of Telegram and WhatsApp message delivery (#493) --- docs/DOCUMENTATION_INDEX.md | 1 + docs/MESSAGING_DELIVERY.md | 128 +++ docs/NOTIFICATIONS.md | 13 + docs/RUNBOOK.md | 40 + .../migration.sql | 49 ++ .../rollback.sql | 4 + prisma/schema.prisma | 52 ++ src/config/env.ts | 17 + src/controllers/user-webhook-controller.ts | 5 +- src/index.ts | 9 + src/jobs/alertRules.ts | 9 +- src/jobs/messageDeliverySweep.ts | 87 ++ src/messaging/index.ts | 5 + src/messaging/service.ts | 808 ++++++++++++++++++ src/messaging/types.ts | 83 ++ src/middleware/adminAuth.ts | 3 + src/routes/admin.ts | 224 +++++ src/routes/telegram.ts | 30 +- src/strategy/service.ts | 8 +- src/utils/metrics.ts | 82 ++ src/utils/telegram-client.ts | 176 ++++ src/utils/twilio-client.ts | 31 + src/validators/webhook-validators.ts | 4 +- tests/unit/jobs/messageDeliverySweep.test.ts | 101 +++ .../messaging/messageDeliveryService.test.ts | 253 ++++++ tests/unit/messaging/telegramClient.test.ts | 109 +++ tests/unit/routes/admin-messages.test.ts | 256 ++++++ 27 files changed, 2563 insertions(+), 24 deletions(-) create mode 100644 docs/MESSAGING_DELIVERY.md create mode 100644 prisma/migrations/20260927000000_add_message_deliveries/migration.sql create mode 100644 prisma/migrations/20260927000000_add_message_deliveries/rollback.sql create mode 100644 src/jobs/messageDeliverySweep.ts create mode 100644 src/messaging/index.ts create mode 100644 src/messaging/service.ts create mode 100644 src/messaging/types.ts create mode 100644 src/utils/telegram-client.ts create mode 100644 tests/unit/jobs/messageDeliverySweep.test.ts create mode 100644 tests/unit/messaging/messageDeliveryService.test.ts create mode 100644 tests/unit/messaging/telegramClient.test.ts create mode 100644 tests/unit/routes/admin-messages.test.ts diff --git a/docs/DOCUMENTATION_INDEX.md b/docs/DOCUMENTATION_INDEX.md index 7ea37eaa..410e7733 100644 --- a/docs/DOCUMENTATION_INDEX.md +++ b/docs/DOCUMENTATION_INDEX.md @@ -17,6 +17,7 @@ - **[REBALANCE_EXPOSURE_COST.md](REBALANCE_EXPOSURE_COST.md)** - Rebalance exposure caps (#346) and grounded cost model + payback gate (#347): resolution, tighten-only merge, residual routing, conservative-by-default decisions - **[CORRELATION_DIVERSIFICATION.md](CORRELATION_DIVERSIFICATION.md)** - Portfolio APY correlation matrix + diversification score (#348): alignment, null-on-degenerate semantics, weighted score, caveat - **[YIELD_COMPOSITION.md](YIELD_COMPOSITION.md)** - Yield base/incentive decomposition + effective APY (#349): haircut model, schema columns, flag-gated consumption, emissions risk modifier, yield-breakdown endpoint +- **[MESSAGING_DELIVERY.md](MESSAGING_DELIVERY.md)** - Telegram & WhatsApp message delivery reliability, retry/fallback policy, metrics, and recovery admin API (#493) ### For DevOps/Deployment diff --git a/docs/MESSAGING_DELIVERY.md b/docs/MESSAGING_DELIVERY.md new file mode 100644 index 00000000..924d1d19 --- /dev/null +++ b/docs/MESSAGING_DELIVERY.md @@ -0,0 +1,128 @@ +# Telegram & WhatsApp Message Delivery Architecture (#493) + +## Overview + +NeuroWealth provides high-reliability, observable message delivery for Telegram bot communications and WhatsApp notifications. The messaging subsystem tracks message lifecycle statuses, enforces exponential backoff with jitter and provider rate-limit compliance, supports automated cross-channel fallback, and provides admin endpoints for safe manual recovery of failed or dead-lettered messages. + +--- + +## Delivery Status Lifecycle + +All outbound messages are tracked in the `MessageDelivery` table with the following lifecycle states: + +``` +[ PENDING ] ───▶ [ SENDING ] ───▶ [ DELIVERED ] + ▲ │ + │ (retry) ▼ + └── [ RETRYING ] ◀─── (transient failure: 429, 5xx, network timeout) + │ + ▼ (attempts exhausted or permanent failure) + [ DEAD_LETTER ] ───▶ [ Manual Recovery / Cancel ] +``` + +- **`PENDING`**: Message is queued and awaiting dispatch or scheduled for a future retry attempt (`nextAttemptAt`). +- **`SENDING`**: Message is actively being delivered via provider API. +- **`DELIVERED`**: Message successfully accepted by provider; records `providerMessageId` (Telegram message ID or Twilio Message SID) and `deliveredAt` timestamp. +- **`FAILED`**: Explicitly cancelled or marked unrecoverable. +- **`DEAD_LETTER`**: Message exhausted all retries or experienced a fatal non-transient error (e.g. 400 Bad Request, 403 Forbidden / bot blocked, invalid phone number). Alert triggers automatically if DLQ threshold is breached. + +--- + +## Retry & Fallback Policy + +### 1. Transient Error Detection & Provider Backoff +- **Telegram**: + - Automatically parses HTTP 429 response `parameters.retry_after` and extracts seconds. + - Classifies HTTP 429, 500, 502, 503, 504, `ECONNRESET`, `ETIMEDOUT`, and network errors as transient. + - Treats 400 (Bad Request), 403 (Forbidden: bot blocked by user), and 404 as non-transient / terminal. +- **WhatsApp (Twilio)**: + - Parses Twilio error codes (e.g. 20429 rate limit, 503 service unavailable, 21614 message queue saturation). + - Inspects `Retry-After` headers if returned. + - Treats 21211 (Invalid 'To' phone number), 21610 (Blacklisted recipient), and 21617 as non-transient. + +### 2. Exponential Backoff with Jitter +For retriable errors, the next attempt timestamp is calculated as: +$$\text{delay} = \min(\text{maxDelayMs}, \text{baseDelayMs} \times 2^{\text{attempts}-1}) \times (1 + \text{jitter})$$ +where $\text{jitter} \in [0, 0.25]$ to prevent thundering herd spikes. If provider provides `retry_after`, $\max(\text{delay}, \text{retryAfterMs})$ is respected. + +### 3. Cross-Channel Fallback +When configured (`MESSAGING_FALLBACK_ENABLED=true`): +- If delivery fails permanently on the primary channel, the engine checks for a linked user account or configured fallback recipient: + - `TELEGRAM` ──▶ `WHATSAPP` (using user's verified phone number) + - `WHATSAPP` ──▶ `TELEGRAM` (using user's linked Telegram chat ID) +- Fallback dispatches record `fallbackTriggered = true`, `fallbackChannel`, and `fallbackRecipient`, emitting `message_fallbacks_total` metrics. + +--- + +## Background Sweep Job + +The background sweep worker (`src/jobs/messageDeliverySweep.ts`) runs at configured intervals (default: 30 seconds): +- Scans for `PENDING` messages where `nextAttemptAt <= NOW()`. +- Dispatches messages ordered by priority (`CRITICAL`, `HIGH`, `NORMAL`, `LOW`) and creation date. +- Emits queue depth metrics (`message_queue_depth`) for Prometheus monitoring. +- Triggers alerting via `alertingService` if dead-lettered message count exceeds `MESSAGING_DLQ_ALERT_THRESHOLD`. + +--- + +## Metrics & Observability + +Prometheus metrics exposed at `/metrics`: + +| Metric Name | Type | Labels | Description | +|-------------|------|--------|-------------| +| `message_deliveries_total` | Counter | `channel`, `status`, `category` | Total delivery attempts by status (`DELIVERED`, `RETRYING`, `DEAD_LETTER`, `DELIVERED_FALLBACK`) | +| `message_retries_total` | Counter | `channel` | Total retries scheduled | +| `message_fallbacks_total` | Counter | `from_channel`, `to_channel` | Total cross-channel fallback events triggered | +| `message_dead_letters_total` | Counter | `channel`, `reason` | Total messages moved to dead letter queue | +| `message_queue_depth` | Gauge | `channel`, `status` | Current number of messages in pending or dead letter states | +| `message_delivery_duration_seconds` | Histogram | `channel` | Latency distribution of message delivery requests | + +--- + +## Safe Manual Recovery Admin API + +Authenticated administrators with `messages:read` or `messages:write` scopes can safely monitor and recover failed messages via the Admin API: + +### 1. List Messages +`GET /api/v1/admin/messages?status=DEAD_LETTER&channel=TELEGRAM&page=1&limit=50` +- Query parameters: `status`, `channel`, `userId`, `page`, `limit`. +- Returns paginated list of tracked message deliveries. + +### 2. Queue & Delivery Statistics +`GET /api/v1/admin/messages/stats` +- Returns summary counts of messages by status and per channel breakdown (pending, delivered, dead-lettered). + +### 3. Inspect Single Message +`GET /api/v1/admin/messages/:id` +- Returns full details including attempt history, last error message, provider IDs, and fallback status. + +### 4. Single Message Recovery +`POST /api/v1/admin/messages/:id/retry` +- Resets attempts counter and immediately triggers safe re-dispatch. +- Logs administrative audit entry (`action: MESSAGE_RETRY`). + +### 5. Bulk Dead-Letter Queue Recovery +`POST /api/v1/admin/messages/retry-dead-letters` +- Payload: `{ "channel": "TELEGRAM" }` (optional filter). +- Resets all matching `DEAD_LETTER` messages to `PENDING` with immediate `nextAttemptAt` for the background worker to safely sweep with rate-limiting. +- Logs administrative audit entry (`action: MESSAGE_RETRY_DEAD_LETTERS`). + +### 6. Cancel Pending/Dead Message +`DELETE /api/v1/admin/messages/:id` +- Transitions message to `FAILED` with `lastError: "Cancelled by administrator"`. +- Prevents any further background retry attempts. +- Logs administrative audit entry (`action: MESSAGE_CANCEL`). + +--- + +## Configuration Variables + +| Variable | Type | Default | Description | +|----------|------|---------|-------------| +| `TELEGRAM_BOT_TOKEN` | String | - | Telegram Bot API token | +| `MESSAGING_MAX_RETRIES` | Number | `5` | Maximum delivery attempts before moving to DLQ | +| `MESSAGING_RETRY_BASE_DELAY_MS` | Number | `2000` | Initial exponential backoff delay (ms) | +| `MESSAGING_RETRY_MAX_DELAY_MS` | Number | `300000` | Maximum backoff delay cap (5 minutes) | +| `MESSAGING_SWEEP_INTERVAL_MS` | Number | `30000` | Background retry sweep interval (30 seconds) | +| `MESSAGING_FALLBACK_ENABLED` | Boolean | `true` | Enable automated cross-channel fallback (Telegram ↔ WhatsApp) | +| `MESSAGING_DLQ_ALERT_THRESHOLD` | Number | `10` | Dead-letter threshold triggering high-priority alerts | diff --git a/docs/NOTIFICATIONS.md b/docs/NOTIFICATIONS.md index 2d3a730f..713d726a 100644 --- a/docs/NOTIFICATIONS.md +++ b/docs/NOTIFICATIONS.md @@ -20,3 +20,16 @@ To protect deliverability and prevent spam, email addresses must pass a double o - **Provider Abstraction (`MailProvider`)**: Supports AWS `SES` and `SMTP` (Nodemailer), selected via `MAIL_PROVIDER` environment variable. Uses a health ledger for automatic failover. - **Mandatory Plaintext**: All templates generate both HTML and plaintext parts with unsubscribe / manage-preferences links. - **Provider Webhooks (`POST /api/v1/webhooks/mail`)**: Signature-verified callback endpoint processing bounces and spam complaints. Hard bounces or complaints update status to `BOUNCED` / `COMPLAINED` / `SUPPRESSED` and emit `notification.email_suppressed`. + +--- + +## Telegram & WhatsApp Message Delivery (#493) + +Telegram bot interactions and WhatsApp alert dispatches are managed by the unified resilient messaging delivery service (`src/messaging/service.ts`). +Features include: +- Message delivery lifecycle tracking (`PENDING`, `SENDING`, `DELIVERED`, `FAILED`, `DEAD_LETTER`). +- Provider 429 rate limit backoff and jittered exponential retry. +- Cross-channel fallback between Telegram and WhatsApp. +- Dead letter alerting and safe manual recovery admin endpoints. + +For complete architectural details, see [MESSAGING_DELIVERY.md](./MESSAGING_DELIVERY.md). diff --git a/docs/RUNBOOK.md b/docs/RUNBOOK.md index 5e145ab1..66b6010d 100644 --- a/docs/RUNBOOK.md +++ b/docs/RUNBOOK.md @@ -561,3 +561,43 @@ tick rather than trading blind. curl -s http://localhost:3001/metrics | grep -E "agent_breaker_(state|trips_total)" psql "$DATABASE_URL" -c "SELECT scope, \"scopeKey\", state, \"trippedRule\" FROM agent_circuit_breakers ORDER BY \"updatedAt\" DESC;" ``` + +--- + +## 13. Telegram & WhatsApp Message Delivery Recovery (#493) + +### Alert: `telegram_dlq_high` / `whatsapp_dlq_high` +Triggered when dead-lettered message count exceeds `MESSAGING_DLQ_ALERT_THRESHOLD` (default: 10). + +### Diagnostic commands +```bash +# Check queue metrics and dead-letter count +curl -H "Authorization: Bearer $ADMIN_API_TOKEN" http://localhost:3001/api/v1/admin/messages/stats | jq + +# List failed / dead-lettered messages +curl -H "Authorization: Bearer $ADMIN_API_TOKEN" "http://localhost:3001/api/v1/admin/messages?status=DEAD_LETTER&limit=20" | jq +``` + +### Recovery actions +1. **Investigate Root Cause**: Inspect `lastError` on failed deliveries: + - 429 Rate Limits: Verify rate-limit configuration and exponential backoff settings. + - 403 Forbidden (Telegram): User blocked the bot; cancel message or re-route. + - 21211 (WhatsApp/Twilio): Invalid phone number format. +2. **Bulk Retry Dead Letters**: Once downstream connectivity is restored: + ```bash + curl -X POST http://localhost:3001/api/v1/admin/messages/retry-dead-letters \ + -H "Authorization: Bearer $ADMIN_API_TOKEN" \ + -H "Content-Type: application/json" \ + -d '{"channel": "TELEGRAM"}' + ``` +3. **Single Message Manual Retry**: + ```bash + curl -X POST http://localhost:3001/api/v1/admin/messages//retry \ + -H "Authorization: Bearer $ADMIN_API_TOKEN" + ``` +4. **Cancel Poison Messages**: + ```bash + curl -X DELETE http://localhost:3001/api/v1/admin/messages/ \ + -H "Authorization: Bearer $ADMIN_API_TOKEN" + ``` + diff --git a/prisma/migrations/20260927000000_add_message_deliveries/migration.sql b/prisma/migrations/20260927000000_add_message_deliveries/migration.sql new file mode 100644 index 00000000..cd8b1232 --- /dev/null +++ b/prisma/migrations/20260927000000_add_message_deliveries/migration.sql @@ -0,0 +1,49 @@ +-- CreateEnum +CREATE TYPE "MessageChannel" AS ENUM ('TELEGRAM', 'WHATSAPP'); + +-- CreateEnum +CREATE TYPE "MessageDeliveryStatus" AS ENUM ('PENDING', 'SENDING', 'DELIVERED', 'FAILED', 'DEAD_LETTER'); + +-- CreateEnum +CREATE TYPE "MessagePriority" AS ENUM ('HIGH', 'NORMAL', 'LOW'); + +-- CreateTable +CREATE TABLE "message_deliveries" ( + "id" TEXT NOT NULL, + "channel" "MessageChannel" NOT NULL, + "recipient" TEXT NOT NULL, + "userId" TEXT, + "category" TEXT NOT NULL DEFAULT 'NOTIFICATION', + "body" TEXT NOT NULL, + "status" "MessageDeliveryStatus" NOT NULL DEFAULT 'PENDING', + "priority" "MessagePriority" NOT NULL DEFAULT 'NORMAL', + "attempts" INTEGER NOT NULL DEFAULT 0, + "maxAttempts" INTEGER NOT NULL DEFAULT 5, + "nextAttemptAt" TIMESTAMP(3), + "lastError" TEXT, + "providerMessageId" TEXT, + "metadata" JSONB, + "fallbackChannel" "MessageChannel", + "fallbackRecipient" TEXT, + "fallbackTriggered" BOOLEAN NOT NULL DEFAULT false, + "deliveredAt" TIMESTAMP(3), + "createdAt" TIMESTAMP(3) NOT NULL DEFAULT CURRENT_TIMESTAMP, + "updatedAt" TIMESTAMP(3) NOT NULL, + + CONSTRAINT "message_deliveries_pkey" PRIMARY KEY ("id") +); + +-- CreateIndex +CREATE INDEX "message_deliveries_channel_status_idx" ON "message_deliveries"("channel", "status"); + +-- CreateIndex +CREATE INDEX "message_deliveries_status_nextAttemptAt_idx" ON "message_deliveries"("status", "nextAttemptAt"); + +-- CreateIndex +CREATE INDEX "message_deliveries_recipient_idx" ON "message_deliveries"("recipient"); + +-- CreateIndex +CREATE INDEX "message_deliveries_userId_idx" ON "message_deliveries"("userId"); + +-- CreateIndex +CREATE INDEX "message_deliveries_createdAt_idx" ON "message_deliveries"("createdAt"); diff --git a/prisma/migrations/20260927000000_add_message_deliveries/rollback.sql b/prisma/migrations/20260927000000_add_message_deliveries/rollback.sql new file mode 100644 index 00000000..1a9b981f --- /dev/null +++ b/prisma/migrations/20260927000000_add_message_deliveries/rollback.sql @@ -0,0 +1,4 @@ +DROP TABLE IF EXISTS "message_deliveries"; +DROP TYPE IF EXISTS "MessagePriority"; +DROP TYPE IF EXISTS "MessageDeliveryStatus"; +DROP TYPE IF EXISTS "MessageChannel"; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index bbb8e64f..e1ef1804 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -1937,3 +1937,55 @@ model AgentCircuitBreaker { @@index([state]) @@map("agent_circuit_breakers") } + +// ── Telegram & WhatsApp Message Delivery (#493) ────────────────────────── + +enum MessageChannel { + TELEGRAM + WHATSAPP +} + +enum MessageDeliveryStatus { + PENDING + SENDING + DELIVERED + FAILED + DEAD_LETTER +} + +enum MessagePriority { + HIGH + NORMAL + LOW +} + +model MessageDelivery { + id String @id @default(uuid()) + channel MessageChannel + recipient String + userId String? + category String @default("NOTIFICATION") + body String + status MessageDeliveryStatus @default(PENDING) + priority MessagePriority @default(NORMAL) + attempts Int @default(0) + maxAttempts Int @default(5) + nextAttemptAt DateTime? + lastError String? + providerMessageId String? + metadata Json? + fallbackChannel MessageChannel? + fallbackRecipient String? + fallbackTriggered Boolean @default(false) + deliveredAt DateTime? + createdAt DateTime @default(now()) + updatedAt DateTime @updatedAt + + @@index([channel, status]) + @@index([status, nextAttemptAt]) + @@index([recipient]) + @@index([userId]) + @@index([createdAt]) + @@map("message_deliveries") +} + diff --git a/src/config/env.ts b/src/config/env.ts index 81cb47f3..75bb6933 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -442,6 +442,23 @@ export const config = { twilioToken: process.env.TWILIO_AUTH_TOKEN || '', fromNumber: process.env.WHATSAPP_FROM || '', }, + telegram: { + botToken: process.env.TELEGRAM_BOT_TOKEN || '', + webhookSecret: process.env.TELEGRAM_WEBHOOK_SECRET || '', + apiUrl: process.env.TELEGRAM_API_URL || 'https://api.telegram.org', + }, + messaging: { + maxRetries: parseInt(process.env.MESSAGING_MAX_RETRIES || '3'), + baseDelayMs: parseInt(process.env.MESSAGING_BASE_DELAY_MS || '1000'), + maxDelayMs: parseInt(process.env.MESSAGING_MAX_DELAY_MS || '30000'), + retryIntervalMs: parseInt( + process.env.MESSAGING_RETRY_INTERVAL_MS || '60000' + ), + fallbackEnabled: process.env.MESSAGING_FALLBACK_ENABLED !== 'false', + dlqAlertThreshold: parseInt( + process.env.MESSAGING_DLQ_ALERT_THRESHOLD || '5' + ), + }, transcription: { provider: process.env.TRANSCRIPTION_PROVIDER || 'openai', /** diff --git a/src/controllers/user-webhook-controller.ts b/src/controllers/user-webhook-controller.ts index a16f33d6..8e2692f1 100644 --- a/src/controllers/user-webhook-controller.ts +++ b/src/controllers/user-webhook-controller.ts @@ -1,5 +1,6 @@ import crypto from 'node:crypto' import { Request, Response } from 'express' +import { Prisma } from '@prisma/client' import db from '../db' import { logger } from '../utils/logger' import { @@ -195,7 +196,9 @@ export async function updateEndpoint( ...(url && { url }), ...(events && Array.isArray(events) && { events }), ...(topicScope && Array.isArray(topicScope) && { topicScope }), - ...(filterJson !== undefined && { filterJson }), + ...(filterJson !== undefined && { + filterJson: filterJson === null ? Prisma.DbNull : filterJson, + }), ...(status && { status }), }, select: { diff --git a/src/index.ts b/src/index.ts index c4240f7c..9a853a66 100644 --- a/src/index.ts +++ b/src/index.ts @@ -59,6 +59,7 @@ import { scheduleProtocolRiskScoring } from './jobs/protocolRiskScoring' import { schedulePortfolioRiskJob } from './jobs/portfolioRisk' import { scheduleApprovalExpiry } from './jobs/approvalExpiry' import { scheduleReserveReconciliation } from './jobs/reserveReconciliation' +import { scheduleMessageDeliverySweep } from './jobs/messageDeliverySweep' import { startEventListener, stopEventListener } from './stellar/events' import { startEventBridge, stopEventBridge } from './events/bridge' import { attachWebSocketServer, closeWebSocketServer } from './ws/server' @@ -133,6 +134,7 @@ let outboxDispatcherHandle: NodeJS.Timeout | null = null let portfolioRiskJobHandle: NodeJS.Timeout | null = null let approvalExpiryHandle: NodeJS.Timeout | null = null let reserveReconciliationHandle: NodeJS.Timeout | null = null +let messageDeliverySweepHandle: NodeJS.Timeout | null = null function allServicesReady(): boolean { return Object.values(serviceStatus).every((s) => s.ready) @@ -446,6 +448,12 @@ async function gracefulShutdown(signal: string): Promise { logger.info('[Shutdown] Reserve reconciliation timer cleared') } + if (messageDeliverySweepHandle) { + clearInterval(messageDeliverySweepHandle) + messageDeliverySweepHandle = null + logger.info('[Shutdown] Message delivery sweep timer cleared') + } + try { stopFeeOracle() logger.info('[Shutdown] Fee oracle stopped') @@ -658,6 +666,7 @@ async function main(): Promise { portfolioRiskJobHandle = schedulePortfolioRiskJob() approvalExpiryHandle = scheduleApprovalExpiry() reserveReconciliationHandle = scheduleReserveReconciliation() + messageDeliverySweepHandle = scheduleMessageDeliverySweep() } // ── Process-level error guards ──────────────────────────────────────────────── diff --git a/src/jobs/alertRules.ts b/src/jobs/alertRules.ts index 5a3b430c..847b3a95 100644 --- a/src/jobs/alertRules.ts +++ b/src/jobs/alertRules.ts @@ -10,6 +10,7 @@ import { config } from '../config/env' import { publishUserEvent } from '../events/publisher' import { EVENT_TYPE_TOPIC } from '../events/types' import { sendWhatsAppMessage } from '../utils/twilio-client' +import { messageDeliveryService } from '../messaging' import { formatAlertTriggeredReply } from '../whatsapp/formatters' import { compare, @@ -233,7 +234,13 @@ async function deliverAlert( threshold, observedValue, }) - await sendWhatsAppMessage({ to: `whatsapp:${user.phone}`, body }) + await messageDeliveryService.send({ + channel: 'WHATSAPP', + recipient: user.phone, + userId: rule.userId, + category: 'ALERT', + body, + }) } } } diff --git a/src/jobs/messageDeliverySweep.ts b/src/jobs/messageDeliverySweep.ts new file mode 100644 index 00000000..4fb912f9 --- /dev/null +++ b/src/jobs/messageDeliverySweep.ts @@ -0,0 +1,87 @@ +import { logger, logBackgroundJob } from '../utils/logger' +import { + generateCorrelationId, + runWithCorrelationIdAsync, +} from '../utils/correlation' +import { recordJobSuccess, recordJobFailure } from '../utils/job-metrics' +import { scheduleResilientJob } from './resilientScheduler' +import { config } from '../config/env' +import { messageDeliveryService } from '../messaging/service' +import { alertingService } from '../services/alerting' + +const JOB_NAME = 'message_delivery_sweep' + +/** + * Sweeps the message queue to retry pending/failed deliveries and monitor DLQ health (#493). + */ +export async function runMessageDeliverySweep(): Promise { + const correlationId = generateCorrelationId() + return runWithCorrelationIdAsync(correlationId, async () => { + const start = Date.now() + + try { + const result = await messageDeliveryService.processPendingQueue(50) + const stats = await messageDeliveryService.getMessageStats() + + const durationMs = Date.now() - start + logBackgroundJob(JOB_NAME, 'success', durationMs / 1000, correlationId, { + processed: result.processed, + delivered: result.delivered, + retried: result.retried, + deadLettered: result.deadLettered, + totalDeadLetter: stats.byStatus.deadLetter, + }) + recordJobSuccess(JOB_NAME, durationMs) + + // Alerting if dead letters exceed configured threshold + const dlqThreshold = config.messaging.dlqAlertThreshold || 5 + if (stats.byStatus.deadLetter >= dlqThreshold) { + await alertingService + .emit( + { + title: 'Message Delivery DLQ Threshold Exceeded', + description: `Messaging dead-letter queue currently contains ${stats.byStatus.deadLetter} undeliverable messages (threshold: ${dlqThreshold}).`, + severity: 'warning', + component: 'messaging_dlq', + metadata: { + totalDeadLetter: stats.byStatus.deadLetter, + threshold: dlqThreshold, + byChannel: stats.byChannel, + }, + }, + 'messaging_dlq_threshold' + ) + .catch(() => {}) + } + } catch (error) { + const durationMs = Date.now() - start + const errorMessage = + error instanceof Error ? error.message : 'Unknown error' + logBackgroundJob(JOB_NAME, 'failed', durationMs / 1000, correlationId, { + error: errorMessage, + }) + recordJobFailure(JOB_NAME, durationMs, error) + logger.error(`[${JOB_NAME}] Sweep execution failed`, { + error: errorMessage, + correlationId, + }) + } + }) +} + +/** + * Schedule recurring message delivery retry and recovery sweep. + */ +export function scheduleMessageDeliverySweep(): NodeJS.Timeout { + const handle = scheduleResilientJob({ + jobName: JOB_NAME, + task: runMessageDeliverySweep, + intervalMs: config.messaging.retryIntervalMs, + unref: true, + }) + + logger.info( + `[Messaging] Message delivery sweep scheduled every ${config.messaging.retryIntervalMs}ms` + ) + return handle +} diff --git a/src/messaging/index.ts b/src/messaging/index.ts new file mode 100644 index 00000000..0d4fe662 --- /dev/null +++ b/src/messaging/index.ts @@ -0,0 +1,5 @@ +export * from './types' +export { + MessageDeliveryService, + messageDeliveryService, +} from './service' diff --git a/src/messaging/service.ts b/src/messaging/service.ts new file mode 100644 index 00000000..380383fd --- /dev/null +++ b/src/messaging/service.ts @@ -0,0 +1,808 @@ +import crypto from 'node:crypto' +import db from '../db' +import { config } from '../config' +import { logger } from '../utils/logger' +import { alertingService } from '../services/alerting' +import { + sendTelegramMessage, + isTelegramTransientError, + extractTelegramRetryAfter, +} from '../utils/telegram-client' +import { + sendWhatsAppMessage, + isWhatsAppTransientError, + extractWhatsAppRetryAfter, +} from '../utils/twilio-client' +import { + recordMessageDelivery, + recordMessageRetry, + recordMessageFallback, + recordMessageDeadLetter, + updateMessageQueueDepth, +} from '../utils/metrics' +import type { + MessageChannel, + MessageDeliveryRecord, + MessageDeliveryStatus, + MessagePriority, + SendMessageInput, + ListMessagesFilter, + MessageStats, +} from './types' + +// In-memory fallback cache to ensure zero-loss delivery if DB is temporarily unavailable or mocked in tests +const inMemoryStore = new Map() + +function sanitizeRecipient(recipient: string): string { + if (recipient.length <= 6) return '***' + return `${recipient.slice(0, 3)}***${recipient.slice(-3)}` +} + +function shouldQueryDb(): boolean { + if (process.env.NODE_ENV === 'test') { + const md = (db as any)?.messageDelivery + if (!md) return false + return Boolean( + md.create?._isMockFunction || + typeof md.create?.mockResolvedValue === 'function' || + process.env.USE_REAL_DB_FOR_TESTS === 'true' + ) + } + return Boolean((db as any)?.messageDelivery) +} + +export class MessageDeliveryService { + private static instance: MessageDeliveryService + + public static getInstance(): MessageDeliveryService { + if (!MessageDeliveryService.instance) { + MessageDeliveryService.instance = new MessageDeliveryService() + } + return MessageDeliveryService.instance + } + + /** + * Calculate exponential backoff with full jitter. + */ + public calculateBackoff(attempt: number, retryAfterSeconds?: number): number { + if (retryAfterSeconds && retryAfterSeconds > 0) { + return retryAfterSeconds * 1000 + } + const base = config.messaging.baseDelayMs + const max = config.messaging.maxDelayMs + const exponential = Math.min(max, base * 2 ** Math.max(0, attempt - 1)) + return Math.floor(Math.random() * exponential) + 50 + } + + /** + * Persist a delivery record to DB (with memory fallback). + */ + private async persistRecord( + record: MessageDeliveryRecord + ): Promise { + inMemoryStore.set(record.id, { ...record }) + + if (shouldQueryDb()) { + try { + await (db as any).messageDelivery.create({ + data: { + id: record.id, + channel: record.channel, + recipient: record.recipient, + userId: record.userId, + category: record.category, + body: record.body, + status: record.status, + priority: record.priority, + attempts: record.attempts, + maxAttempts: record.maxAttempts, + nextAttemptAt: record.nextAttemptAt, + lastError: record.lastError, + providerMessageId: record.providerMessageId, + metadata: record.metadata ?? undefined, + fallbackChannel: record.fallbackChannel, + fallbackRecipient: record.fallbackRecipient, + fallbackTriggered: record.fallbackTriggered, + deliveredAt: record.deliveredAt, + }, + }) + } catch (error) { + logger.warn('[Messaging] Could not write message delivery to DB, retained in memory', { + id: record.id, + error: error instanceof Error ? error.message : String(error), + }) + } + } + + return record + } + + /** + * Update an existing delivery record in DB and memory. + */ + private async updateRecord( + record: MessageDeliveryRecord + ): Promise { + record.updatedAt = new Date() + inMemoryStore.set(record.id, { ...record }) + + if (shouldQueryDb()) { + try { + await (db as any).messageDelivery.update({ + where: { id: record.id }, + data: { + status: record.status, + attempts: record.attempts, + nextAttemptAt: record.nextAttemptAt, + lastError: record.lastError, + providerMessageId: record.providerMessageId, + fallbackChannel: record.fallbackChannel, + fallbackRecipient: record.fallbackRecipient, + fallbackTriggered: record.fallbackTriggered, + deliveredAt: record.deliveredAt, + updatedAt: record.updatedAt, + }, + }) + } catch (error) { + logger.warn('[Messaging] Could not update message delivery in DB, updated in memory', { + id: record.id, + error: error instanceof Error ? error.message : String(error), + }) + } + } + + return record + } + + /** + * Find a delivery record by ID. + */ + public async getMessage(id: string): Promise { + if (shouldQueryDb()) { + try { + const row = await (db as any).messageDelivery.findUnique({ + where: { id }, + }) + if (row) return row as MessageDeliveryRecord + } catch { + // Fall through to memory store + } + } + return inMemoryStore.get(id) ?? null + } + + /** + * Queue or immediately send a message. + */ + public async send( + input: SendMessageInput, + options?: { immediate?: boolean } + ): Promise { + const id = crypto.randomUUID() + const now = new Date() + + const record: MessageDeliveryRecord = { + id, + channel: input.channel, + recipient: input.recipient, + userId: input.userId ?? null, + category: input.category || 'NOTIFICATION', + body: input.body, + status: 'PENDING', + priority: input.priority || 'NORMAL', + attempts: 0, + maxAttempts: input.maxAttempts || config.messaging.maxRetries || 5, + nextAttemptAt: null, + lastError: null, + providerMessageId: null, + metadata: input.metadata || null, + fallbackChannel: input.fallbackChannel ?? null, + fallbackRecipient: input.fallbackRecipient ?? null, + fallbackTriggered: false, + deliveredAt: null, + createdAt: now, + updatedAt: now, + } + + await this.persistRecord(record) + + const shouldSendImmediate = options?.immediate ?? true + if (shouldSendImmediate) { + return await this.dispatchSingle(record) + } + + return record + } + + /** + * Execute delivery for a single record. + */ + public async dispatchSingle( + record: MessageDeliveryRecord + ): Promise { + record.status = 'SENDING' + await this.updateRecord(record) + + const startTime = Date.now() + + try { + let providerMessageId: string | null = null + + if (record.channel === 'TELEGRAM') { + const result = await sendTelegramMessage({ + chatId: record.recipient, + text: record.body, + }) + providerMessageId = result.messageId + } else if (record.channel === 'WHATSAPP') { + const to = record.recipient.startsWith('whatsapp:') + ? record.recipient + : `whatsapp:${record.recipient}` + const sid = await sendWhatsAppMessage({ + to, + body: record.body, + }) + providerMessageId = sid + } else { + throw new Error(`Unsupported message channel: ${record.channel}`) + } + + // Success + const durationSeconds = (Date.now() - startTime) / 1000 + record.attempts += 1 + record.status = 'DELIVERED' + record.deliveredAt = new Date() + record.providerMessageId = providerMessageId + record.lastError = null + record.nextAttemptAt = null + + await this.updateRecord(record) + recordMessageDelivery( + record.channel, + 'DELIVERED', + record.category, + durationSeconds + ) + + logger.info('[Messaging] Message successfully delivered', { + id: record.id, + channel: record.channel, + recipient: sanitizeRecipient(record.recipient), + category: record.category, + attempts: record.attempts, + providerMessageId, + }) + + return record + } catch (error) { + const durationSeconds = (Date.now() - startTime) / 1000 + record.attempts += 1 + const errorMessage = + error instanceof Error ? error.message : String(error) + record.lastError = errorMessage + + recordMessageRetry(record.channel) + + const isTransient = + record.channel === 'TELEGRAM' + ? isTelegramTransientError(error) + : isWhatsAppTransientError(error) + + const retryAfter = + record.channel === 'TELEGRAM' + ? extractTelegramRetryAfter(error) + : extractWhatsAppRetryAfter(error) + + logger.warn('[Messaging] Message delivery attempt failed', { + id: record.id, + channel: record.channel, + recipient: sanitizeRecipient(record.recipient), + attempts: record.attempts, + maxAttempts: record.maxAttempts, + isTransient, + error: errorMessage, + }) + + // If transient and we have attempts remaining, schedule retry + if (isTransient && record.attempts < record.maxAttempts) { + const delayMs = this.calculateBackoff(record.attempts, retryAfter) + record.status = 'PENDING' + record.nextAttemptAt = new Date(Date.now() + delayMs) + + await this.updateRecord(record) + recordMessageDelivery( + record.channel, + 'RETRYING', + record.category, + durationSeconds + ) + + return record + } + + // If attempts exhausted or terminal, check fallback policy + if (config.messaging.fallbackEnabled && !record.fallbackTriggered) { + const fallback = await this.resolveFallback(record) + if (fallback) { + logger.warn( + `[Messaging] Triggering fallback from ${record.channel} to ${fallback.channel}`, + { + id: record.id, + fromChannel: record.channel, + toChannel: fallback.channel, + recipient: sanitizeRecipient(fallback.recipient), + } + ) + + record.fallbackTriggered = true + record.fallbackChannel = fallback.channel + record.fallbackRecipient = fallback.recipient + recordMessageFallback(record.channel, fallback.channel) + + // Try dispatching immediately over fallback channel + try { + let fallbackProviderId: string | null = null + if (fallback.channel === 'TELEGRAM') { + const res = await sendTelegramMessage({ + chatId: fallback.recipient, + text: record.body, + }) + fallbackProviderId = res.messageId + } else if (fallback.channel === 'WHATSAPP') { + const to = fallback.recipient.startsWith('whatsapp:') + ? fallback.recipient + : `whatsapp:${fallback.recipient}` + fallbackProviderId = await sendWhatsAppMessage({ + to, + body: record.body, + }) + } + + record.status = 'DELIVERED' + record.deliveredAt = new Date() + record.providerMessageId = fallbackProviderId + record.lastError = null + record.nextAttemptAt = null + + await this.updateRecord(record) + recordMessageDelivery( + fallback.channel, + 'DELIVERED_FALLBACK', + record.category + ) + + logger.info('[Messaging] Message successfully delivered via fallback', { + id: record.id, + fallbackChannel: fallback.channel, + recipient: sanitizeRecipient(fallback.recipient), + }) + + return record + } catch (fallbackError) { + logger.error('[Messaging] Fallback delivery also failed', { + id: record.id, + fallbackChannel: fallback.channel, + error: + fallbackError instanceof Error + ? fallbackError.message + : String(fallbackError), + }) + record.lastError = `Primary failed (${errorMessage}); Fallback failed: ${ + fallbackError instanceof Error + ? fallbackError.message + : String(fallbackError) + }` + } + } + } + + // No fallback or fallback failed — move to DEAD_LETTER + record.status = 'DEAD_LETTER' + record.nextAttemptAt = null + await this.updateRecord(record) + + recordMessageDeadLetter(record.channel) + recordMessageDelivery( + record.channel, + 'DEAD_LETTER', + record.category, + durationSeconds + ) + + logger.error('[Messaging] Message delivery moved to DEAD_LETTER', { + id: record.id, + channel: record.channel, + recipient: sanitizeRecipient(record.recipient), + attempts: record.attempts, + lastError: record.lastError, + }) + + // Emit alert + await alertingService + .emit( + { + title: `Message Delivery Dead-Lettered (${record.channel})`, + description: `Message ${record.id} to ${sanitizeRecipient( + record.recipient + )} failed after ${record.attempts} attempts: ${record.lastError}`, + severity: 'warning', + component: 'messaging', + metadata: { + messageId: record.id, + channel: record.channel, + category: record.category, + attempts: record.attempts, + recipient: sanitizeRecipient(record.recipient), + error: record.lastError, + }, + }, + `dead_letter:${record.id}` + ) + .catch(() => {}) + + return record + } + } + + /** + * Determine fallback channel and recipient if available. + */ + private async resolveFallback( + record: MessageDeliveryRecord + ): Promise<{ channel: MessageChannel; recipient: string } | null> { + if (record.fallbackChannel && record.fallbackRecipient) { + return { + channel: record.fallbackChannel, + recipient: record.fallbackRecipient, + } + } + + // Try resolving alternate channel via linked user records + try { + if (record.channel === 'WHATSAPP') { + // Fallback from WhatsApp to Telegram + if (record.userId) { + const user = await db.user.findUnique({ + where: { id: record.userId }, + select: { walletAddress: true }, + }) + if (user?.walletAddress) { + const { getTelegramUser } = await import('../telegram/userManager') + // Check memory store for matching wallet + const tlgUser = getTelegramUser(user.walletAddress) + if (tlgUser?.chatId) { + return { channel: 'TELEGRAM', recipient: tlgUser.chatId } + } + } + } + } else if (record.channel === 'TELEGRAM') { + // Fallback from Telegram to WhatsApp + if (record.userId) { + const user = await db.user.findUnique({ + where: { id: record.userId }, + select: { phone: true }, + }) + if (user?.phone) { + return { channel: 'WHATSAPP', recipient: user.phone } + } + } + } + } catch { + // Ignore resolution error + } + + return null + } + + /** + * Process pending/retrying messages in the queue whose nextAttemptAt <= now. + */ + public async processPendingQueue(batchSize: number = 50): Promise<{ + processed: number + delivered: number + retried: number + deadLettered: number + }> { + const now = new Date() + let candidates: MessageDeliveryRecord[] = [] + + if (shouldQueryDb()) { + try { + candidates = await (db as any).messageDelivery.findMany({ + where: { + status: 'PENDING', + OR: [{ nextAttemptAt: null }, { nextAttemptAt: { lte: now } }], + }, + orderBy: [{ priority: 'desc' }, { createdAt: 'asc' }], + take: batchSize, + }) + } catch { + // Fallback to in-memory store + } + } + + if (!candidates || candidates.length === 0) { + candidates = Array.from(inMemoryStore.values()) + .filter( + (m) => + m.status === 'PENDING' && + (!m.nextAttemptAt || m.nextAttemptAt.getTime() <= now.getTime()) + ) + .slice(0, batchSize) + } + + let delivered = 0 + let retried = 0 + let deadLettered = 0 + + // Process with concurrency limit of 5 + const concurrency = 5 + for (let i = 0; i < candidates.length; i += concurrency) { + const chunk = candidates.slice(i, i + concurrency) + await Promise.all( + chunk.map(async (msg) => { + const result = await this.dispatchSingle(msg) + if (result.status === 'DELIVERED') delivered++ + else if (result.status === 'PENDING') retried++ + else if (result.status === 'DEAD_LETTER') deadLettered++ + }) + ) + } + + // Update gauge depth metrics + await this.refreshQueueMetrics().catch(() => {}) + + return { + processed: candidates.length, + delivered, + retried, + deadLettered, + } + } + + /** + * Update queue depth metrics in Prometheus. + */ + public async refreshQueueMetrics(): Promise { + const stats = await this.getMessageStats() + updateMessageQueueDepth('TELEGRAM', 'PENDING', stats.byChannel.telegram.pending) + updateMessageQueueDepth('TELEGRAM', 'DEAD_LETTER', stats.byChannel.telegram.deadLetter) + updateMessageQueueDepth('WHATSAPP', 'PENDING', stats.byChannel.whatsapp.pending) + updateMessageQueueDepth('WHATSAPP', 'DEAD_LETTER', stats.byChannel.whatsapp.deadLetter) + } + + /** + * Manual recovery: force-retry a specific message. + */ + public async retryMessage(id: string): Promise { + const msg = await this.getMessage(id) + if (!msg) { + throw new Error(`Message with id ${id} not found`) + } + + msg.status = 'PENDING' + msg.attempts = 0 + msg.nextAttemptAt = new Date() + msg.lastError = null + msg.fallbackTriggered = false + + await this.updateRecord(msg) + logger.info(`[Messaging] Manual retry initiated for message ${id}`) + return await this.dispatchSingle(msg) + } + + /** + * Manual recovery: bulk retry all dead-lettered messages. + */ + public async retryAllDeadLetters( + channel?: MessageChannel + ): Promise<{ count: number }> { + let deadLetters: MessageDeliveryRecord[] = [] + + if (shouldQueryDb()) { + try { + deadLetters = await (db as any).messageDelivery.findMany({ + where: { + status: 'DEAD_LETTER', + ...(channel ? { channel } : {}), + }, + }) + } catch { + // Memory fallback + } + } + + if (!deadLetters || deadLetters.length === 0) { + deadLetters = Array.from(inMemoryStore.values()).filter( + (m) => + m.status === 'DEAD_LETTER' && (!channel || m.channel === channel) + ) + } + + for (const msg of deadLetters) { + msg.status = 'PENDING' + msg.nextAttemptAt = new Date() + msg.attempts = 0 + msg.lastError = null + msg.fallbackTriggered = false + await this.updateRecord(msg) + } + + logger.info( + `[Messaging] Reset ${deadLetters.length} dead-lettered messages to PENDING for re-attempt` + ) + + // Trigger processing + void this.processPendingQueue(deadLetters.length).catch(() => {}) + + return { count: deadLetters.length } + } + + /** + * Cancel an unsent or dead-letter message. + */ + public async cancelMessage(id: string): Promise { + const msg = await this.getMessage(id) + if (!msg) { + throw new Error(`Message with id ${id} not found`) + } + + msg.status = 'FAILED' + msg.lastError = 'Cancelled by administrator' + msg.nextAttemptAt = null + await this.updateRecord(msg) + + logger.info(`[Messaging] Cancelled message ${id}`) + return msg + } + + /** + * List messages with filter and pagination. + */ + public async listMessages(filter: ListMessagesFilter = {}): Promise<{ + messages: MessageDeliveryRecord[] + total: number + }> { + const limit = Math.min(filter.limit || 50, 500) + const offset = filter.offset || 0 + + if (shouldQueryDb()) { + try { + const where: any = {} + if (filter.channel) where.channel = filter.channel + if (filter.status) where.status = filter.status + if (filter.recipient) where.recipient = { contains: filter.recipient } + if (filter.userId) where.userId = filter.userId + if (filter.category) where.category = filter.category + + const [messages, total] = await Promise.all([ + (db as any).messageDelivery.findMany({ + where, + orderBy: { createdAt: 'desc' }, + take: limit, + skip: offset, + }), + (db as any).messageDelivery.count({ where }), + ]) + + return { messages, total } + } catch { + // Memory fallback + } + } + + let all = Array.from(inMemoryStore.values()) + if (filter.channel) all = all.filter((m) => m.channel === filter.channel) + if (filter.status) all = all.filter((m) => m.status === filter.status) + if (filter.recipient) + all = all.filter((m) => m.recipient.includes(filter.recipient!)) + if (filter.userId) all = all.filter((m) => m.userId === filter.userId) + if (filter.category) all = all.filter((m) => m.category === filter.category) + + all.sort((a, b) => b.createdAt.getTime() - a.createdAt.getTime()) + return { + messages: all.slice(offset, offset + limit), + total: all.length, + } + } + + /** + * Get queue and delivery stats. + */ + public async getMessageStats(): Promise { + const defaultStats: MessageStats = { + total: 0, + byStatus: { + pending: 0, + sending: 0, + delivered: 0, + failed: 0, + deadLetter: 0, + }, + byChannel: { + telegram: { + total: 0, + delivered: 0, + failed: 0, + pending: 0, + deadLetter: 0, + }, + whatsapp: { + total: 0, + delivered: 0, + failed: 0, + pending: 0, + deadLetter: 0, + }, + }, + } + + if (shouldQueryDb()) { + try { + const rows = await (db as any).messageDelivery.groupBy({ + by: ['channel', 'status'], + _count: { _all: true }, + }) + + for (const row of rows) { + const count = row._count._all + defaultStats.total += count + + if (row.status === 'PENDING') defaultStats.byStatus.pending += count + else if (row.status === 'SENDING') defaultStats.byStatus.sending += count + else if (row.status === 'DELIVERED') + defaultStats.byStatus.delivered += count + else if (row.status === 'FAILED') defaultStats.byStatus.failed += count + else if (row.status === 'DEAD_LETTER') + defaultStats.byStatus.deadLetter += count + + const channelKey = + row.channel === 'TELEGRAM' ? 'telegram' : 'whatsapp' + defaultStats.byChannel[channelKey].total += count + + if (row.status === 'DELIVERED') + defaultStats.byChannel[channelKey].delivered += count + else if (row.status === 'FAILED') + defaultStats.byChannel[channelKey].failed += count + else if (row.status === 'PENDING' || row.status === 'SENDING') + defaultStats.byChannel[channelKey].pending += count + else if (row.status === 'DEAD_LETTER') + defaultStats.byChannel[channelKey].deadLetter += count + } + + return defaultStats + } catch { + // Memory fallback + } + } + + for (const msg of inMemoryStore.values()) { + defaultStats.total += 1 + if (msg.status === 'PENDING') defaultStats.byStatus.pending += 1 + else if (msg.status === 'SENDING') defaultStats.byStatus.sending += 1 + else if (msg.status === 'DELIVERED') defaultStats.byStatus.delivered += 1 + else if (msg.status === 'FAILED') defaultStats.byStatus.failed += 1 + else if (msg.status === 'DEAD_LETTER') defaultStats.byStatus.deadLetter += 1 + + const ch = msg.channel === 'TELEGRAM' ? 'telegram' : 'whatsapp' + defaultStats.byChannel[ch].total += 1 + if (msg.status === 'DELIVERED') defaultStats.byChannel[ch].delivered += 1 + else if (msg.status === 'FAILED') defaultStats.byChannel[ch].failed += 1 + else if (msg.status === 'PENDING' || msg.status === 'SENDING') + defaultStats.byChannel[ch].pending += 1 + else if (msg.status === 'DEAD_LETTER') + defaultStats.byChannel[ch].deadLetter += 1 + } + + return defaultStats + } + + /** + * Reset store for unit tests. + */ + public clearStoreForTests(): void { + inMemoryStore.clear() + } +} + +export const messageDeliveryService = MessageDeliveryService.getInstance() diff --git a/src/messaging/types.ts b/src/messaging/types.ts new file mode 100644 index 00000000..fa11efff --- /dev/null +++ b/src/messaging/types.ts @@ -0,0 +1,83 @@ +export type MessageChannel = 'TELEGRAM' | 'WHATSAPP' + +export type MessageDeliveryStatus = + | 'PENDING' + | 'SENDING' + | 'DELIVERED' + | 'FAILED' + | 'DEAD_LETTER' + +export type MessagePriority = 'HIGH' | 'NORMAL' | 'LOW' + +export interface MessageDeliveryRecord { + id: string + channel: MessageChannel + recipient: string + userId: string | null + category: string + body: string + status: MessageDeliveryStatus + priority: MessagePriority + attempts: number + maxAttempts: number + nextAttemptAt: Date | null + lastError: string | null + providerMessageId: string | null + metadata: Record | null + fallbackChannel: MessageChannel | null + fallbackRecipient: string | null + fallbackTriggered: boolean + deliveredAt: Date | null + createdAt: Date + updatedAt: Date +} + +export interface SendMessageInput { + channel: MessageChannel + recipient: string + body: string + userId?: string | null + category?: string + priority?: MessagePriority + maxAttempts?: number + metadata?: Record + fallbackChannel?: MessageChannel + fallbackRecipient?: string +} + +export interface ListMessagesFilter { + channel?: MessageChannel + status?: MessageDeliveryStatus + recipient?: string + userId?: string + category?: string + limit?: number + offset?: number +} + +export interface MessageStats { + total: number + byStatus: { + pending: number + sending: number + delivered: number + failed: number + deadLetter: number + } + byChannel: { + telegram: { + total: number + delivered: number + failed: number + pending: number + deadLetter: number + } + whatsapp: { + total: number + delivered: number + failed: number + pending: number + deadLetter: number + } + } +} diff --git a/src/middleware/adminAuth.ts b/src/middleware/adminAuth.ts index e90735e3..1c6a2917 100644 --- a/src/middleware/adminAuth.ts +++ b/src/middleware/adminAuth.ts @@ -42,6 +42,9 @@ export const ADMIN_SCOPES = [ // #394 — GDPR/CCPA right-to-erasure 'erasure:write', 'erasure:read', + // #493 — Telegram and WhatsApp message delivery management & recovery + 'messages:read', + 'messages:write', 'super', ] as const export type AdminScope = (typeof ADMIN_SCOPES)[number] diff --git a/src/routes/admin.ts b/src/routes/admin.ts index 73871555..c4a26e7f 100644 --- a/src/routes/admin.ts +++ b/src/routes/admin.ts @@ -23,6 +23,7 @@ import { manualTripBreaker, manualResetBreaker, } from '../agent/breakerService' +import { messageDeliveryService } from '../messaging' const router = Router() const prisma = db @@ -1794,4 +1795,227 @@ router.post( } ) +// ── Telegram & WhatsApp Message Delivery Management & Recovery (#493) ───── + +/** + * GET /api/admin/messages + * List tracked message deliveries with optional filtering. Required scope: messages:read + */ +router.get( + '/messages', + requireAdminScope('messages:read'), + async (req: Request, res: Response) => { + try { + const { channel, status, recipient, userId, category, limit, offset } = + req.query + + const result = await messageDeliveryService.listMessages({ + channel: channel as any, + status: status as any, + recipient: recipient as string | undefined, + userId: userId as string | undefined, + category: category as string | undefined, + limit: limit ? parseInt(limit as string, 10) : undefined, + offset: offset ? parseInt(offset as string, 10) : undefined, + }) + + auditLog(req, res, 'MESSAGES_LIST', 'success', { + channel, + status, + recipient, + returned: result.messages.length, + total: result.total, + }) + + res.status(200).json({ + success: true, + data: result, + timestamp: new Date().toISOString(), + }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'MESSAGES_LIST', 'failure', { error: message }) + res + .status(500) + .json({ success: false, error: 'Failed to list messages' }) + } + } +) + +/** + * GET /api/admin/messages/stats + * Overview queue depth and delivery statistics by channel. Required scope: messages:read + */ +router.get( + '/messages/stats', + requireAdminScope('messages:read'), + async (req: Request, res: Response) => { + try { + const stats = await messageDeliveryService.getMessageStats() + auditLog(req, res, 'MESSAGES_STATS', 'success') + res.status(200).json({ + success: true, + data: stats, + timestamp: new Date().toISOString(), + }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'MESSAGES_STATS', 'failure', { error: message }) + res + .status(500) + .json({ success: false, error: 'Failed to retrieve message stats' }) + } + } +) + +/** + * GET /api/admin/messages/:id + * Inspect a single message delivery record. Required scope: messages:read + */ +router.get( + '/messages/:id', + requireAdminScope('messages:read'), + async (req: Request, res: Response) => { + try { + const msg = await messageDeliveryService.getMessage(req.params.id) + if (!msg) { + auditLog(req, res, 'MESSAGE_GET', 'failure', { + id: req.params.id, + error: 'not_found', + }) + res.status(404).json({ success: false, error: 'Message not found' }) + return + } + + auditLog(req, res, 'MESSAGE_GET', 'success', { id: msg.id }) + res.status(200).json({ success: true, data: msg }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'MESSAGE_GET', 'failure', { error: message }) + res + .status(500) + .json({ success: false, error: 'Failed to retrieve message' }) + } + } +) + +/** + * POST /api/admin/messages/:id/retry + * Safe manual recovery: force-retry a failed or dead-lettered message. Required scope: messages:write + */ +router.post( + '/messages/:id/retry', + requireAdminScope('messages:write'), + async (req: Request, res: Response) => { + try { + const msg = await messageDeliveryService.getMessage(req.params.id) + if (!msg) { + auditLog(req, res, 'MESSAGE_RETRY', 'failure', { + id: req.params.id, + error: 'not_found', + }) + res.status(404).json({ success: false, error: 'Message not found' }) + return + } + + const retried = await messageDeliveryService.retryMessage(req.params.id) + auditLog(req, res, 'MESSAGE_RETRY', 'success', { + id: retried.id, + channel: retried.channel, + status: retried.status, + }) + + res.status(200).json({ + success: true, + data: retried, + message: `Message ${retried.id} retry initiated; current status: ${retried.status}`, + }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'MESSAGE_RETRY', 'failure', { error: message }) + res + .status(500) + .json({ success: false, error: `Failed to retry message: ${message}` }) + } + } +) + +/** + * POST /api/admin/messages/retry-dead-letters + * Safe manual recovery: bulk retry all dead-lettered messages. Required scope: messages:write + */ +router.post( + '/messages/retry-dead-letters', + requireAdminScope('messages:write'), + async (req: Request, res: Response) => { + try { + const { channel } = req.body ?? {} + const result = await messageDeliveryService.retryAllDeadLetters(channel) + + auditLog(req, res, 'MESSAGE_RETRY_DEAD_LETTERS', 'success', { + channel, + retriedCount: result.count, + }) + + res.status(200).json({ + success: true, + data: result, + message: `Successfully re-enqueued ${result.count} dead-lettered messages for delivery`, + }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'MESSAGE_RETRY_DEAD_LETTERS', 'failure', { + error: message, + }) + res.status(500).json({ + success: false, + error: `Failed to retry dead-lettered messages: ${message}`, + }) + } + } +) + +/** + * DELETE /api/admin/messages/:id + * Safe manual recovery: cancel an unsent or dead-lettered message. Required scope: messages:write + */ +router.delete( + '/messages/:id', + requireAdminScope('messages:write'), + async (req: Request, res: Response) => { + try { + const msg = await messageDeliveryService.getMessage(req.params.id) + if (!msg) { + auditLog(req, res, 'MESSAGE_CANCEL', 'failure', { + id: req.params.id, + error: 'not_found', + }) + res.status(404).json({ success: false, error: 'Message not found' }) + return + } + + const cancelled = await messageDeliveryService.cancelMessage( + req.params.id + ) + auditLog(req, res, 'MESSAGE_CANCEL', 'success', { + id: cancelled.id, + status: cancelled.status, + }) + + res.status(200).json({ + success: true, + data: cancelled, + message: `Message ${cancelled.id} cancelled successfully`, + }) + } catch (error) { + const message = error instanceof Error ? error.message : 'Unknown error' + auditLog(req, res, 'MESSAGE_CANCEL', 'failure', { error: message }) + res + .status(500) + .json({ success: false, error: `Failed to cancel message: ${message}` }) + } + } +) + export default router + diff --git a/src/routes/telegram.ts b/src/routes/telegram.ts index 1c8819b0..7369d662 100644 --- a/src/routes/telegram.ts +++ b/src/routes/telegram.ts @@ -1,5 +1,6 @@ import express, { Request, Response } from 'express' import { handleTelegramMessage } from '../telegram/handler' +import { messageDeliveryService } from '../messaging' import { logger } from '../utils/logger' const router = express.Router() @@ -31,24 +32,17 @@ router.post('/', async (req: Request, res: Response) => { try { const reply = await handleTelegramMessage(chatId, text) - const payload = { - chat_id: chatId, - text: reply, - parse_mode: 'HTML', - } - - const response = await fetch( - `https://api.telegram.org/bot${BOT_TOKEN}/sendMessage`, - { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify(payload), - } - ) - - if (!response.ok) { - logger.error('[Telegram webhook] Bot API error', { - status: response.status, + const delivery = await messageDeliveryService.send({ + channel: 'TELEGRAM', + recipient: String(chatId), + body: reply, + category: 'BOT_REPLY', + }) + + if (delivery.status === 'DEAD_LETTER' || delivery.status === 'FAILED') { + logger.error('[Telegram webhook] Bot message delivery failed', { + chatId, + error: delivery.lastError, }) return res.status(502).send('Bad Gateway') } diff --git a/src/strategy/service.ts b/src/strategy/service.ts index cb2b7297..8334a134 100644 --- a/src/strategy/service.ts +++ b/src/strategy/service.ts @@ -27,6 +27,7 @@ import { logger } from '../utils/logger' import { publishUserEvent } from '../events/publisher' import { EVENT_TYPE_TOPIC } from '../events/types' import { sendWhatsAppMessage } from '../utils/twilio-client' +import { messageDeliveryService } from '../messaging' import { formatStrategyUpdatedReply, formatStrategyUnpublishedReply, @@ -590,8 +591,11 @@ async function notifyFollowers( }) : formatStrategyUnpublishedReply({ label: details.label }) - await sendWhatsAppMessage({ - to: `whatsapp:${follower.follower.phone}`, + await messageDeliveryService.send({ + channel: 'WHATSAPP', + recipient: follower.follower.phone, + userId: follower.followerUserId, + category: 'STRATEGY_UPDATE', body, }) }) diff --git a/src/utils/metrics.ts b/src/utils/metrics.ts index d47c1156..a9597899 100644 --- a/src/utils/metrics.ts +++ b/src/utils/metrics.ts @@ -946,6 +946,88 @@ export function recordAssistantFallback( assistantFallbackTotal.inc({ reason }) } +// ── Message Delivery Metrics (#493) ────────────────────────────────────────────── +export const messageDeliveriesTotal = new client.Counter({ + name: 'message_deliveries_total', + help: 'Total number of Telegram and WhatsApp message deliveries by status', + labelNames: ['channel', 'status', 'category'] as const, + registers: [register], +}) + +export const messageRetriesTotal = new client.Counter({ + name: 'message_retries_total', + help: 'Total number of message retry attempts', + labelNames: ['channel'] as const, + registers: [register], +}) + +export const messageFallbacksTotal = new client.Counter({ + name: 'message_fallbacks_total', + help: 'Total number of message channel fallbacks triggered', + labelNames: ['from_channel', 'to_channel'] as const, + registers: [register], +}) + +export const messageDeadLettersTotal = new client.Counter({ + name: 'message_dead_letters_total', + help: 'Total number of messages moved to dead-letter state', + labelNames: ['channel'] as const, + registers: [register], +}) + +export const messageQueueDepth = new client.Gauge({ + name: 'message_queue_depth', + help: 'Current number of messages in delivery queue by channel and status', + labelNames: ['channel', 'status'] as const, + registers: [register], +}) + +export const messageDeliveryDuration = new client.Histogram({ + name: 'message_delivery_duration_seconds', + help: 'Latency of message delivery attempts by channel', + labelNames: ['channel'] as const, + buckets: [0.05, 0.1, 0.25, 0.5, 1, 2.5, 5, 10], + registers: [register], +}) + +export function recordMessageDelivery( + channel: string, + status: string, + category: string = 'NOTIFICATION', + durationSeconds?: number +): void { + messageDeliveriesTotal.inc({ channel, status, category }) + if (durationSeconds !== undefined && durationSeconds >= 0) { + messageDeliveryDuration.observe({ channel }, durationSeconds) + } +} + +export function recordMessageRetry(channel: string): void { + messageRetriesTotal.inc({ channel }) +} + +export function recordMessageFallback( + fromChannel: string, + toChannel: string +): void { + messageFallbacksTotal.inc({ + from_channel: fromChannel, + to_channel: toChannel, + }) +} + +export function recordMessageDeadLetter(channel: string): void { + messageDeadLettersTotal.inc({ channel }) +} + +export function updateMessageQueueDepth( + channel: string, + status: string, + depth: number +): void { + messageQueueDepth.set({ channel, status }, depth) +} + /** * Get metrics for Prometheus scraping */ diff --git a/src/utils/telegram-client.ts b/src/utils/telegram-client.ts new file mode 100644 index 00000000..6d22b505 --- /dev/null +++ b/src/utils/telegram-client.ts @@ -0,0 +1,176 @@ +import { config } from '../config' +import { HttpClientAdapter, CircuitBreakerError, TimeoutError } from './http-client' +import { logger } from './logger' + +export interface SendTelegramMessageParams { + chatId: number | string + text: string + parseMode?: 'HTML' | 'Markdown' | 'MarkdownV2' +} + +export interface TelegramSendResult { + messageId: string + ok: boolean +} + +export class TelegramApiError extends Error { + public statusCode?: number + public errorCode?: number + public retryAfterSeconds?: number + public isTransient: boolean + + constructor( + message: string, + options?: { + statusCode?: number + errorCode?: number + retryAfterSeconds?: number + isTransient?: boolean + } + ) { + super(message) + this.name = 'TelegramApiError' + this.statusCode = options?.statusCode + this.errorCode = options?.errorCode + this.retryAfterSeconds = options?.retryAfterSeconds + this.isTransient = options?.isTransient ?? false + } +} + +const httpClient = new HttpClientAdapter({ + timeoutMs: config.httpClient.timeoutMs, + maxRetries: config.httpClient.maxRetries, + baseDelayMs: config.httpClient.baseDelayMs, + maxDelayMs: config.httpClient.maxDelayMs, + circuitBreakerThreshold: config.httpClient.circuitBreakerThreshold, + circuitBreakerResetMs: config.httpClient.circuitBreakerResetMs, +}) + +/** + * Classify whether an error from Telegram is transient and safe to retry. + */ +export function isTelegramTransientError(error: unknown): boolean { + if (error instanceof TelegramApiError) { + return error.isTransient + } + if (error instanceof TimeoutError || error instanceof CircuitBreakerError) { + return true + } + if (error instanceof Error) { + const msg = error.message.toLowerCase() + if ( + msg.includes('timeout') || + msg.includes('econnreset') || + msg.includes('etimedout') || + msg.includes('network') || + msg.includes('fetch failed') || + msg.includes('bad gateway') || + msg.includes('service unavailable') || + msg.includes('gateway timeout') + ) { + return true + } + } + return false +} + +/** + * Extract retry-after in seconds if returned by Telegram (HTTP 429). + */ +export function extractTelegramRetryAfter(error: unknown): number | undefined { + if (error instanceof TelegramApiError) { + return error.retryAfterSeconds + } + return undefined +} + +/** + * Sends a message via the Telegram Bot API with circuit breaker, timeout, and transient retry. + */ +export async function sendTelegramMessage( + params: SendTelegramMessageParams +): Promise { + const token = config.telegram.botToken + if (!token) { + throw new TelegramApiError('TELEGRAM_BOT_TOKEN is not configured', { + isTransient: false, + }) + } + + const apiUrl = config.telegram.apiUrl || 'https://api.telegram.org' + const endpoint = `${apiUrl}/bot${token}/sendMessage` + + return httpClient.execute(async () => { + const payload = { + chat_id: + typeof params.chatId === 'string' && /^\d+$/.test(params.chatId) + ? Number(params.chatId) + : params.chatId, + text: params.text, + parse_mode: params.parseMode ?? 'HTML', + } + + const response = await fetch(endpoint, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(payload), + }) + + let data: any = null + try { + data = await response.json() + } catch { + // response body was not valid JSON + } + + if (!response.ok || !data?.ok) { + const statusCode = response.status + const errorCode = data?.error_code || statusCode + const description = + data?.description || `Telegram Bot API error HTTP ${statusCode}` + const retryAfter = + data?.parameters?.retry_after ?? + (response.headers.get('retry-after') + ? parseInt(response.headers.get('retry-after')!, 10) + : undefined) + + // 429 and 5xx are transient; 400/403/404 are permanent + const isTransient = + statusCode === 429 || + statusCode >= 500 || + errorCode === 429 || + errorCode >= 500 + + logger.warn('[Telegram] Send message failed', { + chatId: params.chatId, + statusCode, + errorCode, + description, + isTransient, + retryAfter, + }) + + throw new TelegramApiError(description, { + statusCode, + errorCode, + retryAfterSeconds: retryAfter, + isTransient, + }) + } + + const messageId = String(data.result?.message_id ?? Date.now()) + logger.info(`[Telegram] Message sent to chat ${params.chatId}: id=${messageId}`) + return { + messageId, + ok: true, + } + }, 'telegram.sendMessage') +} + +export function resetTelegramClient(): void { + httpClient.reset() +} + +export function getTelegramHttpClient(): HttpClientAdapter { + return httpClient +} diff --git a/src/utils/twilio-client.ts b/src/utils/twilio-client.ts index 02cd8079..60425826 100644 --- a/src/utils/twilio-client.ts +++ b/src/utils/twilio-client.ts @@ -57,3 +57,34 @@ export function resetTwilioClient(): void { export function getTwilioHttpClient(): HttpClientAdapter { return httpClient } + +/** + * Classify whether an error from Twilio WhatsApp is transient. + */ +export function isWhatsAppTransientError(error: unknown): boolean { + if (!error) return false + const err = error as any + const code = err.code || err.status + if (code === 20429 || code === 429 || (typeof code === 'number' && code >= 500 && code < 600)) { + return true + } + const msg = (err.message || '').toLowerCase() + if ( + msg.includes('timeout') || + msg.includes('econnreset') || + msg.includes('network') || + msg.includes('rate limit') || + msg.includes('circuit breaker is open') + ) { + return true + } + return false +} + +export function extractWhatsAppRetryAfter(error: unknown): number | undefined { + if (!error) return undefined + const err = error as any + if (err.retryAfter) return Number(err.retryAfter) + return undefined +} + diff --git a/src/validators/webhook-validators.ts b/src/validators/webhook-validators.ts index fc54319b..293b0fd3 100644 --- a/src/validators/webhook-validators.ts +++ b/src/validators/webhook-validators.ts @@ -80,7 +80,7 @@ export const createUserWebhookEndpointSchema = z.object({ topicScope: z .array(z.string()) .optional(), - filterJson: z.record(z.any()).nullable().optional(), + filterJson: z.record(z.string(), z.any()).nullable().optional(), }) export const updateUserWebhookEndpointSchema = z.object({ @@ -92,7 +92,7 @@ export const updateUserWebhookEndpointSchema = z.object({ topicScope: z .array(z.string()) .optional(), - filterJson: z.record(z.any()).nullable().optional(), + filterJson: z.record(z.string(), z.any()).nullable().optional(), status: z .enum(['ACTIVE', 'PAUSED', 'DISABLED']) .optional(), diff --git a/tests/unit/jobs/messageDeliverySweep.test.ts b/tests/unit/jobs/messageDeliverySweep.test.ts new file mode 100644 index 00000000..b3f370a4 --- /dev/null +++ b/tests/unit/jobs/messageDeliverySweep.test.ts @@ -0,0 +1,101 @@ +process.env.NODE_ENV = 'test' +process.env.MESSAGING_DLQ_ALERT_THRESHOLD = '2' + +import { runMessageDeliverySweep } from '../../../src/jobs/messageDeliverySweep' +import { messageDeliveryService } from '../../../src/messaging/service' +import { alertingService } from '../../../src/services/alerting' +import * as jobMetrics from '../../../src/utils/job-metrics' + +jest.mock('../../../src/services/alerting', () => ({ + alertingService: { + emit: jest.fn().mockResolvedValue(undefined), + }, +})) + +jest.mock('../../../src/utils/job-metrics', () => ({ + recordJobSuccess: jest.fn(), + recordJobFailure: jest.fn(), +})) + +describe('Message Delivery Sweep Job (#493)', () => { + beforeEach(() => { + messageDeliveryService.clearStoreForTests() + jest.restoreAllMocks() + }) + + it('successfully processes queue and records success metrics', async () => { + const processSpy = jest + .spyOn(messageDeliveryService, 'processPendingQueue') + .mockResolvedValue({ + processed: 3, + delivered: 2, + retried: 1, + deadLettered: 0, + }) + + const statsSpy = jest + .spyOn(messageDeliveryService, 'getMessageStats') + .mockResolvedValue({ + total: 3, + byStatus: { pending: 1, sending: 0, delivered: 2, failed: 0, deadLetter: 0 }, + byChannel: { + telegram: { total: 2, delivered: 2, failed: 0, pending: 0, deadLetter: 0 }, + whatsapp: { total: 1, delivered: 0, failed: 0, pending: 1, deadLetter: 0 }, + }, + }) + + await runMessageDeliverySweep() + + expect(processSpy).toHaveBeenCalledWith(50) + expect(statsSpy).toHaveBeenCalled() + expect(jobMetrics.recordJobSuccess).toHaveBeenCalledWith( + 'message_delivery_sweep', + expect.any(Number) + ) + }) + + it('emits an operational alert when dead letters exceed threshold', async () => { + jest.spyOn(messageDeliveryService, 'processPendingQueue').mockResolvedValue({ + processed: 5, + delivered: 2, + retried: 0, + deadLettered: 3, + }) + + jest.spyOn(messageDeliveryService, 'getMessageStats').mockResolvedValue({ + total: 5, + byStatus: { pending: 0, sending: 0, delivered: 2, failed: 0, deadLetter: 3 }, + byChannel: { + telegram: { total: 3, delivered: 1, failed: 0, pending: 0, deadLetter: 2 }, + whatsapp: { total: 2, delivered: 1, failed: 0, pending: 0, deadLetter: 1 }, + }, + }) + + const emitSpy = jest.spyOn(alertingService, 'emit') + + await runMessageDeliverySweep() + + expect(emitSpy).toHaveBeenCalledWith( + expect.objectContaining({ + title: expect.stringContaining('DLQ Threshold Exceeded'), + severity: 'warning', + component: 'messaging_dlq', + }), + 'messaging_dlq_threshold' + ) + }) + + it('records failure metric when sweep throws', async () => { + jest + .spyOn(messageDeliveryService, 'processPendingQueue') + .mockRejectedValue(new Error('Queue processor failure')) + + await runMessageDeliverySweep() + + expect(jobMetrics.recordJobFailure).toHaveBeenCalledWith( + 'message_delivery_sweep', + expect.any(Number), + expect.any(Error) + ) + }) +}) diff --git a/tests/unit/messaging/messageDeliveryService.test.ts b/tests/unit/messaging/messageDeliveryService.test.ts new file mode 100644 index 00000000..148ad323 --- /dev/null +++ b/tests/unit/messaging/messageDeliveryService.test.ts @@ -0,0 +1,253 @@ +process.env.NODE_ENV = 'test' +process.env.TELEGRAM_BOT_TOKEN = 'test-token' +process.env.TWILIO_ACCOUNT_SID = 'AC' + '0'.repeat(32) +process.env.TWILIO_AUTH_TOKEN = '0'.repeat(32) +process.env.WHATSAPP_FROM = 'whatsapp:+14155238886' + +import { + MessageDeliveryService, + messageDeliveryService, +} from '../../../src/messaging/service' +import * as telegramClient from '../../../src/utils/telegram-client' +import * as twilioClient from '../../../src/utils/twilio-client' +import { alertingService } from '../../../src/services/alerting' + +jest.mock('../../../src/services/alerting', () => ({ + alertingService: { + emit: jest.fn().mockResolvedValue(undefined), + }, +})) + +describe('MessageDeliveryService (#493)', () => { + beforeEach(() => { + messageDeliveryService.clearStoreForTests() + jest.restoreAllMocks() + }) + + describe('Delivery Status Tracking', () => { + it('tracks successful Telegram message delivery with provider message ID', async () => { + jest.spyOn(telegramClient, 'sendTelegramMessage').mockResolvedValue({ + ok: true, + messageId: 'tlg-msg-12345', + }) + + const record = await messageDeliveryService.send({ + channel: 'TELEGRAM', + recipient: '123456789', + body: 'Hello via Telegram', + category: 'NOTIFICATION', + }) + + expect(record.status).toBe('DELIVERED') + expect(record.providerMessageId).toBe('tlg-msg-12345') + expect(record.deliveredAt).toBeInstanceOf(Date) + expect(record.attempts).toBe(1) + expect(record.lastError).toBeNull() + + const retrieved = await messageDeliveryService.getMessage(record.id) + expect(retrieved).not.toBeNull() + expect(retrieved?.status).toBe('DELIVERED') + }) + + it('tracks successful WhatsApp message delivery with Twilio SID', async () => { + jest.spyOn(twilioClient, 'sendWhatsAppMessage').mockResolvedValue('SM-twilio-sid-1') + + const record = await messageDeliveryService.send({ + channel: 'WHATSAPP', + recipient: '+15551234567', + body: 'Hello via WhatsApp', + category: 'ALERT', + }) + + expect(record.status).toBe('DELIVERED') + expect(record.providerMessageId).toBe('SM-twilio-sid-1') + expect(record.attempts).toBe(1) + expect(record.deliveredAt).toBeInstanceOf(Date) + }) + }) + + describe('Retry Policy & Backoff', () => { + it('re-enqueues message for retry on transient failure with exponential backoff', async () => { + const transientError = new telegramClient.TelegramApiError('Rate limited', { + statusCode: 429, + retryAfterSeconds: 5, + isTransient: true, + }) + jest.spyOn(telegramClient, 'sendTelegramMessage').mockRejectedValue(transientError) + + const record = await messageDeliveryService.send({ + channel: 'TELEGRAM', + recipient: '123456789', + body: 'Test retry', + maxAttempts: 3, + }) + + expect(record.status).toBe('PENDING') + expect(record.attempts).toBe(1) + expect(record.nextAttemptAt).toBeInstanceOf(Date) + expect(record.lastError).toContain('Rate limited') + }) + + it('computes exponential jittered backoff respecting retry_after', () => { + const backoffWithHeader = messageDeliveryService.calculateBackoff(1, 10) + expect(backoffWithHeader).toBe(10000) + + const backoffAttempt1 = messageDeliveryService.calculateBackoff(1) + expect(backoffAttempt1).toBeGreaterThanOrEqual(50) + }) + }) + + describe('Fallback Handling', () => { + it('triggers fallback channel when primary channel fails and fallback is configured', async () => { + // Primary fails with terminal error + jest.spyOn(twilioClient, 'sendWhatsAppMessage').mockRejectedValue( + new Error('21211: Invalid phone number') + ) + // Fallback succeeds + jest.spyOn(telegramClient, 'sendTelegramMessage').mockResolvedValue({ + ok: true, + messageId: 'tlg-fallback-id', + }) + + const record = await messageDeliveryService.send({ + channel: 'WHATSAPP', + recipient: '+15550000000', + body: 'Security Alert', + fallbackChannel: 'TELEGRAM', + fallbackRecipient: '987654321', + }) + + expect(record.status).toBe('DELIVERED') + expect(record.fallbackTriggered).toBe(true) + expect(record.fallbackChannel).toBe('TELEGRAM') + expect(record.fallbackRecipient).toBe('987654321') + expect(record.providerMessageId).toBe('tlg-fallback-id') + }) + }) + + describe('Dead Letter Queue & Alerting', () => { + it('moves to DEAD_LETTER and emits alert when attempts and fallback are exhausted', async () => { + const terminalErr = new Error('403: Forbidden - bot blocked by user') + jest.spyOn(telegramClient, 'sendTelegramMessage').mockRejectedValue(terminalErr) + + const emitSpy = jest.spyOn(alertingService, 'emit') + + const record = await messageDeliveryService.send({ + channel: 'TELEGRAM', + recipient: '123456789', + body: 'Test failure', + maxAttempts: 1, + }) + + expect(record.status).toBe('DEAD_LETTER') + expect(record.lastError).toContain('bot blocked by user') + expect(emitSpy).toHaveBeenCalledWith( + expect.objectContaining({ + severity: 'warning', + component: 'messaging', + title: expect.stringContaining('Dead-Lettered'), + }), + expect.stringContaining('dead_letter:') + ) + }) + }) + + describe('Safe Manual Recovery', () => { + it('retries a dead-lettered message manually', async () => { + // First attempt fails to DEAD_LETTER + jest.spyOn(telegramClient, 'sendTelegramMessage').mockRejectedValueOnce( + new Error('Fatal failure') + ) + + const record = await messageDeliveryService.send({ + channel: 'TELEGRAM', + recipient: '123456789', + body: 'Manual recovery test', + maxAttempts: 1, + }) + + expect(record.status).toBe('DEAD_LETTER') + + // Now manual retry succeeds + jest.spyOn(telegramClient, 'sendTelegramMessage').mockResolvedValueOnce({ + ok: true, + messageId: 'recovered-123', + }) + + const recovered = await messageDeliveryService.retryMessage(record.id) + expect(recovered.status).toBe('DELIVERED') + expect(recovered.providerMessageId).toBe('recovered-123') + }) + + it('bulk retries all dead-lettered messages', async () => { + // Seed 2 dead letter records + jest.spyOn(telegramClient, 'sendTelegramMessage').mockRejectedValue(new Error('Down')) + + await messageDeliveryService.send({ + channel: 'TELEGRAM', + recipient: '111', + body: 'msg1', + maxAttempts: 1, + }) + await messageDeliveryService.send({ + channel: 'TELEGRAM', + recipient: '222', + body: 'msg2', + maxAttempts: 1, + }) + + const stats = await messageDeliveryService.getMessageStats() + expect(stats.byStatus.deadLetter).toBe(2) + + const result = await messageDeliveryService.retryAllDeadLetters('TELEGRAM') + expect(result.count).toBe(2) + }) + + it('cancels an unsent or dead-letter message safely', async () => { + jest.spyOn(telegramClient, 'sendTelegramMessage').mockRejectedValueOnce(new Error('Down')) + + const record = await messageDeliveryService.send({ + channel: 'TELEGRAM', + recipient: '111', + body: 'to cancel', + maxAttempts: 1, + }) + + const cancelled = await messageDeliveryService.cancelMessage(record.id) + expect(cancelled.status).toBe('FAILED') + expect(cancelled.lastError).toContain('Cancelled by administrator') + }) + + it('lists and filters messages accurately', async () => { + jest.spyOn(telegramClient, 'sendTelegramMessage').mockResolvedValue({ + ok: true, + messageId: 'm1', + }) + jest.spyOn(twilioClient, 'sendWhatsAppMessage').mockResolvedValue('m2') + + await messageDeliveryService.send({ + channel: 'TELEGRAM', + recipient: '111', + body: 'Telegram msg', + category: 'BOT_REPLY', + }) + await messageDeliveryService.send({ + channel: 'WHATSAPP', + recipient: '+15551112222', + body: 'WhatsApp msg', + category: 'ALERT', + }) + + const all = await messageDeliveryService.listMessages() + expect(all.total).toBe(2) + + const onlyTelegram = await messageDeliveryService.listMessages({ channel: 'TELEGRAM' }) + expect(onlyTelegram.total).toBe(1) + expect(onlyTelegram.messages[0].channel).toBe('TELEGRAM') + + const onlyAlerts = await messageDeliveryService.listMessages({ category: 'ALERT' }) + expect(onlyAlerts.total).toBe(1) + expect(onlyAlerts.messages[0].category).toBe('ALERT') + }) + }) +}) diff --git a/tests/unit/messaging/telegramClient.test.ts b/tests/unit/messaging/telegramClient.test.ts new file mode 100644 index 00000000..0e6d094a --- /dev/null +++ b/tests/unit/messaging/telegramClient.test.ts @@ -0,0 +1,109 @@ +process.env.NODE_ENV = 'test' +process.env.TELEGRAM_BOT_TOKEN = 'test-token-123' +process.env.TELEGRAM_API_URL = 'https://api.telegram.org' + +import { + sendTelegramMessage, + isTelegramTransientError, + extractTelegramRetryAfter, + TelegramApiError, + resetTelegramClient, +} from '../../../src/utils/telegram-client' +import { TimeoutError, CircuitBreakerError } from '../../../src/utils/http-client' + +describe('Telegram Client (#493)', () => { + beforeEach(() => { + resetTelegramClient() + jest.restoreAllMocks() + }) + + it('successfully delivers message to Telegram Bot API', async () => { + const fetchSpy = jest.spyOn(global, 'fetch' as any).mockResolvedValue({ + ok: true, + status: 200, + json: async () => ({ + ok: true, + result: { message_id: 998877 }, + }), + } as Response) + + const result = await sendTelegramMessage({ + chatId: 123456, + text: 'Hello World', + parseMode: 'HTML', + }) + + expect(result.ok).toBe(true) + expect(result.messageId).toBe('998877') + expect(fetchSpy).toHaveBeenCalledWith( + 'https://api.telegram.org/bottest-token-123/sendMessage', + expect.objectContaining({ + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + chat_id: 123456, + text: 'Hello World', + parse_mode: 'HTML', + }), + }) + ) + }) + + it('handles rate limit 429 and extracts retry_after', async () => { + jest.spyOn(global, 'fetch' as any).mockResolvedValue({ + ok: false, + status: 429, + headers: new Headers({ 'retry-after': '12' }), + json: async () => ({ + ok: false, + error_code: 429, + description: 'Too Many Requests: retry after 12', + parameters: { retry_after: 12 }, + }), + } as Response) + + await expect( + sendTelegramMessage({ + chatId: '123456', + text: 'Test', + }) + ).rejects.toThrow('Too Many Requests') + + const err = new TelegramApiError('Too Many Requests', { + statusCode: 429, + retryAfterSeconds: 12, + isTransient: true, + }) + + expect(isTelegramTransientError(err)).toBe(true) + expect(extractTelegramRetryAfter(err)).toBe(12) + }) + + it('identifies 5xx server errors as transient', async () => { + const err502 = new TelegramApiError('Bad Gateway', { + statusCode: 502, + isTransient: true, + }) + expect(isTelegramTransientError(err502)).toBe(true) + + const timeoutErr = new TimeoutError(5000, 'telegram.sendMessage') + expect(isTelegramTransientError(timeoutErr)).toBe(true) + + const cbErr = new CircuitBreakerError('telegram.sendMessage') + expect(isTelegramTransientError(cbErr)).toBe(true) + }) + + it('identifies 400 and 403 as permanent non-transient errors', () => { + const err400 = new TelegramApiError('Bad Request: chat not found', { + statusCode: 400, + isTransient: false, + }) + expect(isTelegramTransientError(err400)).toBe(false) + + const err403 = new TelegramApiError('Forbidden: bot was blocked by the user', { + statusCode: 403, + isTransient: false, + }) + expect(isTelegramTransientError(err403)).toBe(false) + }) +}) diff --git a/tests/unit/routes/admin-messages.test.ts b/tests/unit/routes/admin-messages.test.ts new file mode 100644 index 00000000..a67046fb --- /dev/null +++ b/tests/unit/routes/admin-messages.test.ts @@ -0,0 +1,256 @@ +process.env.NODE_ENV = 'test' +process.env.TELEGRAM_BOT_TOKEN = 'test-token' +process.env.TWILIO_ACCOUNT_SID = 'AC' + '0'.repeat(32) +process.env.TWILIO_AUTH_TOKEN = '0'.repeat(32) + +import express from 'express' +import request from 'supertest' +import adminRouter from '../../../src/routes/admin' +import { messageDeliveryService } from '../../../src/messaging/service' + +jest.mock('../../../src/db', () => ({ + __esModule: true, + default: { + adminAuditLog: { + create: jest.fn().mockResolvedValue({ id: 'audit-log-1' }), + }, + }, +})) + +jest.mock('../../../src/middleware/adminAuth', () => { + const actual = jest.requireActual('../../../src/middleware/adminAuth') + return { + ...actual, + requireAdminAuth: (req: any, res: any, next: any) => { + res.locals.adminAuth = { + id: 'admin-1', + name: 'Super Admin', + role: 'super', + scopes: ['super', 'messages:read', 'messages:write'], + } + next() + }, + requireAdminScope: (_scope: string) => (_req: any, _res: any, next: any) => next(), + } +}) + +describe('Admin Messages Recovery Routes (#493)', () => { + let app: express.Application + + beforeEach(() => { + messageDeliveryService.clearStoreForTests() + jest.restoreAllMocks() + + app = express() + app.use(express.json()) + app.use('/api/admin', adminRouter) + }) + + it('GET /api/admin/messages lists tracked messages', async () => { + const listSpy = jest.spyOn(messageDeliveryService, 'listMessages').mockResolvedValue({ + messages: [ + { + id: 'msg-1', + channel: 'TELEGRAM', + recipient: '123456', + userId: 'user-1', + category: 'ALERT', + body: 'Price alert triggered', + status: 'DELIVERED', + priority: 'NORMAL', + attempts: 1, + maxAttempts: 5, + nextAttemptAt: null, + lastError: null, + providerMessageId: 'p-1', + metadata: null, + fallbackChannel: null, + fallbackRecipient: null, + fallbackTriggered: false, + deliveredAt: new Date(), + createdAt: new Date(), + updatedAt: new Date(), + }, + ], + total: 1, + }) + + const res = await request(app).get('/api/admin/messages') + expect(res.status).toBe(200) + expect(res.body.success).toBe(true) + expect(res.body.data.messages.length).toBe(1) + expect(res.body.data.messages[0].channel).toBe('TELEGRAM') + expect(listSpy).toHaveBeenCalled() + }) + + it('GET /api/admin/messages/stats returns queue and delivery metrics', async () => { + jest.spyOn(messageDeliveryService, 'getMessageStats').mockResolvedValue({ + total: 10, + byStatus: { pending: 2, sending: 1, delivered: 6, failed: 0, deadLetter: 1 }, + byChannel: { + telegram: { total: 5, delivered: 4, failed: 0, pending: 1, deadLetter: 0 }, + whatsapp: { total: 5, delivered: 2, failed: 0, pending: 2, deadLetter: 1 }, + }, + }) + + const res = await request(app).get('/api/admin/messages/stats') + expect(res.status).toBe(200) + expect(res.body.success).toBe(true) + expect(res.body.data.total).toBe(10) + expect(res.body.data.byStatus.deadLetter).toBe(1) + }) + + it('GET /api/admin/messages/:id returns message detail or 404', async () => { + jest.spyOn(messageDeliveryService, 'getMessage').mockResolvedValueOnce({ + id: 'msg-target', + channel: 'WHATSAPP', + recipient: '+15551234567', + userId: null, + category: 'NOTIFICATION', + body: 'Test body', + status: 'PENDING', + priority: 'NORMAL', + attempts: 0, + maxAttempts: 5, + nextAttemptAt: null, + lastError: null, + providerMessageId: null, + metadata: null, + fallbackChannel: null, + fallbackRecipient: null, + fallbackTriggered: false, + deliveredAt: null, + createdAt: new Date(), + updatedAt: new Date(), + }) + + const foundRes = await request(app).get('/api/admin/messages/msg-target') + expect(foundRes.status).toBe(200) + expect(foundRes.body.data.id).toBe('msg-target') + + jest.spyOn(messageDeliveryService, 'getMessage').mockResolvedValueOnce(null) + const notFoundRes = await request(app).get('/api/admin/messages/non-existent') + expect(notFoundRes.status).toBe(404) + expect(notFoundRes.body.success).toBe(false) + }) + + it('POST /api/admin/messages/:id/retry triggers manual recovery', async () => { + jest.spyOn(messageDeliveryService, 'getMessage').mockResolvedValueOnce({ + id: 'msg-to-retry', + channel: 'TELEGRAM', + recipient: '123', + userId: null, + category: 'NOTIFICATION', + body: 'text', + status: 'DEAD_LETTER', + priority: 'NORMAL', + attempts: 5, + maxAttempts: 5, + nextAttemptAt: null, + lastError: 'Fatal error', + providerMessageId: null, + metadata: null, + fallbackChannel: null, + fallbackRecipient: null, + fallbackTriggered: false, + deliveredAt: null, + createdAt: new Date(), + updatedAt: new Date(), + }) + + jest.spyOn(messageDeliveryService, 'retryMessage').mockResolvedValueOnce({ + id: 'msg-to-retry', + channel: 'TELEGRAM', + recipient: '123', + userId: null, + category: 'NOTIFICATION', + body: 'text', + status: 'DELIVERED', + priority: 'NORMAL', + attempts: 1, + maxAttempts: 5, + nextAttemptAt: null, + lastError: null, + providerMessageId: 'recovered-id', + metadata: null, + fallbackChannel: null, + fallbackRecipient: null, + fallbackTriggered: false, + deliveredAt: new Date(), + createdAt: new Date(), + updatedAt: new Date(), + }) + + const res = await request(app).post('/api/admin/messages/msg-to-retry/retry') + expect(res.status).toBe(200) + expect(res.body.success).toBe(true) + expect(res.body.data.status).toBe('DELIVERED') + }) + + it('POST /api/admin/messages/retry-dead-letters bulk retries dead letters', async () => { + jest + .spyOn(messageDeliveryService, 'retryAllDeadLetters') + .mockResolvedValueOnce({ count: 4 }) + + const res = await request(app) + .post('/api/admin/messages/retry-dead-letters') + .send({ channel: 'WHATSAPP' }) + + expect(res.status).toBe(200) + expect(res.body.success).toBe(true) + expect(res.body.data.count).toBe(4) + }) + + it('DELETE /api/admin/messages/:id cancels a delivery', async () => { + jest.spyOn(messageDeliveryService, 'getMessage').mockResolvedValueOnce({ + id: 'msg-to-cancel', + channel: 'TELEGRAM', + recipient: '123', + userId: null, + category: 'NOTIFICATION', + body: 'text', + status: 'PENDING', + priority: 'NORMAL', + attempts: 0, + maxAttempts: 5, + nextAttemptAt: null, + lastError: null, + providerMessageId: null, + metadata: null, + fallbackChannel: null, + fallbackRecipient: null, + fallbackTriggered: false, + deliveredAt: null, + createdAt: new Date(), + updatedAt: new Date(), + }) + + jest.spyOn(messageDeliveryService, 'cancelMessage').mockResolvedValueOnce({ + id: 'msg-to-cancel', + channel: 'TELEGRAM', + recipient: '123', + userId: null, + category: 'NOTIFICATION', + body: 'text', + status: 'FAILED', + priority: 'NORMAL', + attempts: 0, + maxAttempts: 5, + nextAttemptAt: null, + lastError: 'Cancelled by administrator', + providerMessageId: null, + metadata: null, + fallbackChannel: null, + fallbackRecipient: null, + fallbackTriggered: false, + deliveredAt: null, + createdAt: new Date(), + updatedAt: new Date(), + }) + + const res = await request(app).delete('/api/admin/messages/msg-to-cancel') + expect(res.status).toBe(200) + expect(res.body.success).toBe(true) + expect(res.body.data.status).toBe('FAILED') + }) +})