feat(messaging): improve reliability of Telegram and WhatsApp message delivery (#493)

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)
 - **[WEBHOOK_SECURITY.md](WEBHOOK_SECURITY.md)** - Inbound fiat webhook HMAC validation, freshness windows, replay receipts, and failure responses
 - **[PRISMA_QUERY_PERFORMANCE.md](PRISMA_QUERY_PERFORMANCE.md)** - Hot webhook query plans, supporting indexes, and the query-index regression check
 - **[NET_WORTH.md](NET_WORTH.md)** - Read-only external Stellar links, valuation/staleness semantics, and private scope

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 |

docs/NOTIFICATIONS.md

@@ -49,3 +49,16 @@ To protect deliverability and prevent spam, email addresses must pass a double o
     2. The certificate is fetched and cached per URL under a 5-minute TTL (failed fetches are never cached, so SNS retries get a genuine attempt).
     3. The canonical string is rebuilt with the exact field set for the message `Type` (Notification vs SubscriptionConfirmation differ), and `crypto.verify` checks `Signature` with `SignatureVersion` (1 → SHA-1, 2 → SHA-256; absent defaults to 1).
   - A delivery that fails verification — tampered body, spoofed certificate host, or certificate fetch failure — is rejected as **unauthenticated (401)**, logged, counted in `mail_webhook_rejections_total{reason}`, and **never** processed. A body that is not an SNS envelope at all has no verifiable provenance and is likewise rejected. The controller maps a verified-but-unprocessable payload to 400 and unexpected failures to 500.
+
+---
+
+## 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).

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/<message-id>/retry \
+     -H "Authorization: Bearer $ADMIN_API_TOKEN"
+   ```
+4. **Cancel Poison Messages**:
+   ```bash
+   curl -X DELETE http://localhost:3001/api/v1/admin/messages/<message-id> \
+     -H "Authorization: Bearer $ADMIN_API_TOKEN"
+   ```
+

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");

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";

prisma/schema.prisma

@@ -325,14 +325,6 @@ model User {
   isActive          Boolean          @default(true)
   createdAt         DateTime         @default(now())
   updatedAt         DateTime         @updatedAt
-  phone               String?          @unique
-  riskTolerance       Int              @default(5)
-  rebalanceStrategy   String? // 'MAX_YIELD' | 'TARGET_ALLOCATION' | null (defaults to MAX_YIELD)
-  strategyConfig      Json? // e.g. { "targetAllocations": { "Blend": 50, "Stellar DEX": 30, "Luma": 20 } }
-  taxAwareRebalancing Boolean          @default(false) // #550 - opt-in wash-sale-aware rebalancing
-  isActive            Boolean          @default(true)
-  createdAt           DateTime         @default(now())
-  updatedAt           DateTime         @updatedAt
   // #317 — cost-basis consumption method for the tax report. Changing this
   // is forward-only: methodEffectiveAt stamps when a change took effect, so
   // disposals recorded before it are never retroactively recomputed (see
@@ -2444,6 +2436,8 @@ model ProtocolLiquiditySnapshot {
   @@index([protocolName, assetSymbol])
   @@index([fetchedAt])
   @@map("protocol_liquidity_snapshots")
+}
+
 // ─────────────────────────────────────────────────────────────────────────────
 // Treasury accounts (#jobs/treasurySweep.ts, #outbox/executors.ts)
 // ─────────────────────────────────────────────────────────────────────────────
@@ -2729,3 +2723,53 @@ model WebAuthnChallenge {
   @@map("webauthn_challenges")
 }
 
+// ── 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")
+}

src/accounting/exporters/qbo.ts

@@ -32,16 +32,19 @@ export interface QboTransaction {
  * Convert transactions to QuickBooks Online-compatible QBO format.
  * This is a simplified OFX-style format that QBO can import.
  */
-export function toQbo(transactions: Array<{
-  type: TransactionType
-  amount: string
-  assetSymbol: string
-  fee?: string | null
-  createdAt: Date
-  protocolName?: string | null
-}>): string {
+export function toQbo(
+  transactions: Array<{
+    type: TransactionType
+    amount: string
+    assetSymbol: string
+    fee?: string | null
+    createdAt: Date
+    protocolName?: string | null
+  }>
+): string {
   const qboTransactions: QboTransaction[] = transactions.map((tx) => {
-    const category = TRANSACTION_TYPE_TO_QBO_CATEGORY[tx.type] || 'Uncategorized'
+    const category =
+      TRANSACTION_TYPE_TO_QBO_CATEGORY[tx.type] || 'Uncategorized'
     const amount = Number(tx.amount)
     const fee = tx.fee ? Number(tx.fee) : 0