diff --git a/docs/DOCUMENTATION_INDEX.md b/docs/DOCUMENTATION_INDEX.md index 15ecd736..7e3edbf8 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) - **[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 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 adf6d6d6..66b5d569 100644 --- a/docs/NOTIFICATIONS.md +++ b/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). 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 845b6c32..0ba6f28d 100644 --- a/prisma/schema.prisma +++ b/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") +} diff --git a/src/accounting/exporters/qbo.ts b/src/accounting/exporters/qbo.ts index 05714f83..ef53b492 100644 --- a/src/accounting/exporters/qbo.ts +++ b/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 diff --git a/src/accounting/exporters/xeroCsv.ts b/src/accounting/exporters/xeroCsv.ts index c340324c..6b9a6db3 100644 --- a/src/accounting/exporters/xeroCsv.ts +++ b/src/accounting/exporters/xeroCsv.ts @@ -31,16 +31,19 @@ export interface XeroTransaction { * Convert transactions to Xero-compatible bank statement CSV format. * Xero expects specific column headers for bank statement imports. */ -export function toXeroCsv(transactions: Array<{ - type: TransactionType - amount: string - assetSymbol: string - fee?: string | null - createdAt: Date - protocolName?: string | null -}>): string { +export function toXeroCsv( + transactions: Array<{ + type: TransactionType + amount: string + assetSymbol: string + fee?: string | null + createdAt: Date + protocolName?: string | null + }> +): string { const xeroTransactions: XeroTransaction[] = transactions.map((tx) => { - const category = TRANSACTION_TYPE_TO_XERO_CATEGORY[tx.type] || 'Uncategorized' + const category = + TRANSACTION_TYPE_TO_XERO_CATEGORY[tx.type] || 'Uncategorized' const amount = Number(tx.amount) const fee = tx.fee ? Number(tx.fee) : 0 const netAmount = amount - fee @@ -60,7 +63,14 @@ export function toXeroCsv(transactions: Array<{ } }) - const headers = ['Date', 'Amount', 'Payee', 'Description', 'Reference', 'Category'] + const headers = [ + 'Date', + 'Amount', + 'Payee', + 'Description', + 'Reference', + 'Category', + ] const rows = xeroTransactions.map((tx) => [ tx.date, tx.amount, diff --git a/src/agent/loop.ts b/src/agent/loop.ts index c39b06ec..5e6704aa 100644 --- a/src/agent/loop.ts +++ b/src/agent/loop.ts @@ -350,8 +350,8 @@ async function rebalanceCheckJob(): Promise { const hasStrategyContext = Boolean(lead.config.strategyName) || Boolean(lead.follow) || - (protocolPositions[0].user as any).liquidityFloor !== null && - (protocolPositions[0].user as any).liquidityFloor !== undefined + ((protocolPositions[0].user as any).liquidityFloor !== null && + (protocolPositions[0].user as any).liquidityFloor !== undefined) const userStrategyPreferences = hasStrategyContext ? protocolPositions.map((p: PositionWithUser) => { @@ -376,9 +376,7 @@ async function rebalanceCheckJob(): Promise { (p.user as any).liquidityFloor === undefined ? {} : { - liquidityFloorUsd: Number( - (p.user as any).liquidityFloor - ), + liquidityFloorUsd: Number((p.user as any).liquidityFloor), }), followedStrategyId: follow?.followedStrategyId ?? undefined, } diff --git a/src/agent/router.ts b/src/agent/router.ts index b2d9264e..c89ef6e0 100644 --- a/src/agent/router.ts +++ b/src/agent/router.ts @@ -739,7 +739,10 @@ export async function executeRebalanceIfNeeded( .map(([protocol]) => protocol) ) - if (positionsToMove.length === 0 || liquidityRestoreTargets.size === 0) { + if ( + positionsToMove.length === 0 || + liquidityRestoreTargets.size === 0 + ) { await recordDecision({ outcome: 'BLOCKED', blockedReason: 'liquidity_floor_shortfall', @@ -767,17 +770,14 @@ export async function executeRebalanceIfNeeded( (sum, position) => sum + BigInt(position.amount), BigInt(0) ) - totalAmount = (restoreAmount < positionAmount - ? restoreAmount - : positionAmount + totalAmount = ( + restoreAmount < positionAmount ? restoreAmount : positionAmount ).toString() allProtocols = allProtocols.filter((protocol) => liquidityRestoreTargets!.has(protocol.name) ) } else { - const instantLiquidIds = new Set( - floorStatus.instantLiquidPositionIds - ) + const instantLiquidIds = new Set(floorStatus.instantLiquidPositionIds) const lockedIds = new Set(floorStatus.lockedPositionIds) positionsToMove = userPositions.filter( (position) => @@ -793,9 +793,8 @@ export async function executeRebalanceIfNeeded( ) const yieldLimit = yieldAmount < portfolioAmount ? yieldAmount : portfolioAmount - totalAmount = (yieldLimit < movableAmount - ? yieldLimit - : movableAmount + totalAmount = ( + yieldLimit < movableAmount ? yieldLimit : movableAmount ).toString() if (totalAmount === '0' || positionsToMove.length === 0) { await recordDecision({ @@ -838,10 +837,10 @@ export async function executeRebalanceIfNeeded( const strategy: RebalanceStrategy = liquidityRestoreTargets ? new MaxYieldStrategy() : activeGoal - ? new GoalTrackingStrategy() - : preferredStrategy === 'TARGET_ALLOCATION' - ? new TargetAllocationStrategy() - : new MaxYieldStrategy() + ? new GoalTrackingStrategy() + : preferredStrategy === 'TARGET_ALLOCATION' + ? new TargetAllocationStrategy() + : new MaxYieldStrategy() // Risk ceiling is opt-in per user (or per goal). Only when a ceiling is // set do we load the current ProtocolRiskScore rows and pass them to the @@ -890,7 +889,8 @@ export async function executeRebalanceIfNeeded( // #550 - Wash-sale-aware rebalancing check (informational + opt-in soft deprioritization) let washSaleRisk = false - const taxAwareRebalancing = userStrategyPreferences[0]?.taxAwareRebalancing ?? false + const taxAwareRebalancing = + userStrategyPreferences[0]?.taxAwareRebalancing ?? false if (decision.shouldRebalance) { const likelyRebuy = isSameAssetRebuyLikely( userStrategyPreferences[0]?.strategyName ?? null, @@ -921,13 +921,17 @@ export async function executeRebalanceIfNeeded( }) if (taxAwareRebalancing) { - const higherThreshold = effectiveThresholds.minimumImprovement * 2 - const targetProtocolData = allProtocols.find(p => p.name === decision.targetProtocol) + const higherThreshold = + effectiveThresholds.minimumImprovement * 2 + const targetProtocolData = allProtocols.find( + (p) => p.name === decision.targetProtocol + ) if (targetProtocolData) { const netImprovement = targetProtocolData.apy - currentApy if (netImprovement < higherThreshold) { decision.shouldRebalance = false - decision.reasoning = 'Wash-sale risk: improvement below elevated threshold' + decision.reasoning = + 'Wash-sale risk: improvement below elevated threshold' } } } diff --git a/src/approvals/executors.ts b/src/approvals/executors.ts index a98793ec..ee3485ab 100644 --- a/src/approvals/executors.ts +++ b/src/approvals/executors.ts @@ -38,6 +38,21 @@ export type ApprovalPayload = selectedLotIds?: string[] acknowledgeGoalImpact?: boolean } + | { + type: 'loan_origination' + userId: string + positionId: string + principal: number + borrowedAsset: string + actingAsUserId?: string | null + } + | { + type: 'loan_repayment' + userId: string + loanId: string + amount: number + actingAsUserId?: string | null + } /** * Re-run an approved payload through the exact same deposit/withdraw path a @@ -50,10 +65,33 @@ export type ApprovalPayload = export async function runApprovedPayload( requestId: string, payload: ApprovalPayload -): Promise { +): Promise { const { executeDeposit, executeWithdraw } = await import('../controllers/transaction-controller') + if (payload.type === 'loan_origination') { + const { originateLoan } = await import('../lending/service') + return originateLoan({ + userId: payload.userId, + positionId: payload.positionId, + principal: payload.principal, + borrowedAsset: payload.borrowedAsset, + actingAsUserId: payload.actingAsUserId, + skipApprovalGuard: true, + }) + } + + if (payload.type === 'loan_repayment') { + const { repayLoan } = await import('../lending/service') + return repayLoan({ + userId: payload.userId, + loanId: payload.loanId, + amount: payload.amount, + actingAsUserId: payload.actingAsUserId, + skipApprovalGuard: true, + }) + } + const memo = payload.memo ? `${payload.memo} (approval:${requestId})` : `approval:${requestId}` diff --git a/src/config/env.ts b/src/config/env.ts index 136f08d8..d10c3fc6 100644 --- a/src/config/env.ts +++ b/src/config/env.ts @@ -96,7 +96,10 @@ function validateAllRequiredEnvVars(): void { // ── 1. Missing vars ────────────────────────────────────────────────────── for (const key of requiredVars) { - if (!process.env[key] && !(process.env.SECRET_BACKEND === 'aws-ssm' && ssmManagedSecrets.has(key))) { + if ( + !process.env[key] && + !(process.env.SECRET_BACKEND === 'aws-ssm' && ssmManagedSecrets.has(key)) + ) { errors.push(`Missing required environment variable: ${key}`) } } @@ -519,6 +522,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', /** @@ -845,7 +865,8 @@ export const config = { ), /** Minimum hold duration (ms) before a position is eligible for coverage. */ minHoldDurationMs: parseInt( - process.env.PROTECTION_FUND_MIN_HOLD_DURATION_MS || String(7 * 24 * 60 * 60 * 1000) + process.env.PROTECTION_FUND_MIN_HOLD_DURATION_MS || + String(7 * 24 * 60 * 60 * 1000) ), /** Default asset symbol for the fund. */ defaultAssetSymbol: process.env.PROTECTION_FUND_DEFAULT_ASSET || 'USDC', diff --git a/src/config/secrets.ts b/src/config/secrets.ts index 5703b7fe..37fd0272 100644 --- a/src/config/secrets.ts +++ b/src/config/secrets.ts @@ -42,14 +42,16 @@ export const OPTIONAL_SECRET_KEYS = [ export type SecretKey = (typeof SECRET_KEYS)[number] export type OptionalSecretKey = (typeof OPTIONAL_SECRET_KEYS)[number] -const secretValidators: Partial boolean>> = { - JWT_SEED: (value) => value.length >= 32, - WALLET_ENCRYPTION_KEY: (value) => /^[0-9a-f]{64}$/i.test(value), - STELLAR_AGENT_SECRET_KEY: (value) => value.startsWith('S') && value.length === 56, - ANTHROPIC_API_KEY: (value) => value.startsWith('sk-ant-'), - TWILIO_AUTH_TOKEN: (value) => value.length >= 32, - DATABASE_URL: (value) => /^postgres(?:ql)?:\/\//.test(value), -} +const secretValidators: Partial boolean>> = + { + JWT_SEED: (value) => value.length >= 32, + WALLET_ENCRYPTION_KEY: (value) => /^[0-9a-f]{64}$/i.test(value), + STELLAR_AGENT_SECRET_KEY: (value) => + value.startsWith('S') && value.length === 56, + ANTHROPIC_API_KEY: (value) => value.startsWith('sk-ant-'), + TWILIO_AUTH_TOKEN: (value) => value.length >= 32, + DATABASE_URL: (value) => /^postgres(?:ql)?:\/\//.test(value), + } const optionalSecretValidators: Partial< Record boolean> @@ -67,7 +69,8 @@ export function validateSecretCredentials(now = new Date()): string[] { errors.push(`${key} is missing`) continue } - if (!secretValidators[key]?.(value)) errors.push(`${key} has an invalid format`) + if (!secretValidators[key]?.(value)) + errors.push(`${key} has an invalid format`) const expiresAt = process.env[`SECRET_EXPIRY_${key}`] if (expiresAt) { @@ -80,7 +83,11 @@ export function validateSecretCredentials(now = new Date()): string[] { for (const key of OPTIONAL_SECRET_KEYS) { const value = process.env[key] - if (value && optionalSecretValidators[key] && !optionalSecretValidators[key]!(value)) { + if ( + value && + optionalSecretValidators[key] && + !optionalSecretValidators[key]!(value) + ) { errors.push(`${key} has an invalid format`) } const expiresAt = process.env[`SECRET_EXPIRY_${key}`] diff --git a/src/controllers/accounting-controller.ts b/src/controllers/accounting-controller.ts index b5aa5c4e..7d04976f 100644 --- a/src/controllers/accounting-controller.ts +++ b/src/controllers/accounting-controller.ts @@ -12,7 +12,13 @@ const accountingExportSchema = z.object({ export async function exportAccountingData( userId: string, query: unknown -): Promise<{ success: boolean; data?: string; contentType?: string; filename?: string; error?: string }> { +): Promise<{ + success: boolean + data?: string + contentType?: string + filename?: string + error?: string +}> { try { const parsed = accountingExportSchema.parse(query) @@ -47,7 +53,9 @@ export async function exportAccountingData( })) if (transactions.length === 0) { - logger.info(`No transactions found for user ${userId} in the specified date range`) + logger.info( + `No transactions found for user ${userId} in the specified date range` + ) } let data: string diff --git a/src/controllers/admin-impersonation-controller.ts b/src/controllers/admin-impersonation-controller.ts index b7d5030f..1ba6414e 100644 --- a/src/controllers/admin-impersonation-controller.ts +++ b/src/controllers/admin-impersonation-controller.ts @@ -24,7 +24,9 @@ export async function startImpersonation( const token = generateImpersonationToken(adminId, adminName, targetUserId) - logger.info(`[AdminImpersonation] Admin ${adminName} (${adminId}) started impersonating user ${targetUserId}`) + logger.info( + `[AdminImpersonation] Admin ${adminName} (${adminId}) started impersonating user ${targetUserId}` + ) await publishUserEvent( targetUserId, diff --git a/src/controllers/auth-controller.ts b/src/controllers/auth-controller.ts index f1936f7c..4b9ffc0e 100644 --- a/src/controllers/auth-controller.ts +++ b/src/controllers/auth-controller.ts @@ -23,7 +23,11 @@ import { rotateRefreshToken, type RefreshFailureReason, } from '../services/refresh-token.service' -import { getActiveTotpCredential, issueTotpChallenge, completeTotpChallenge } from '../services/totp.service' +import { + getActiveTotpCredential, + issueTotpChallenge, + completeTotpChallenge, +} from '../services/totp.service' // ── Helpers ──────────────────────────────────────────────────────────────── @@ -251,7 +255,10 @@ export async function verifyTotp(req: Request, res: Response): Promise { code?: unknown } - if (typeof totpChallengeToken !== 'string' || totpChallengeToken.length === 0) { + if ( + typeof totpChallengeToken !== 'string' || + totpChallengeToken.length === 0 + ) { res.status(400).json({ error: 'totpChallengeToken is required' }) return } diff --git a/src/controllers/push-controller.ts b/src/controllers/push-controller.ts index 5c4c9ee2..a9100884 100644 --- a/src/controllers/push-controller.ts +++ b/src/controllers/push-controller.ts @@ -120,7 +120,9 @@ export async function sendPushNotification( }) if (activeTokens.length === 0) { - logger.info(`No active push tokens for user ${userId}, skipping push notification`) + logger.info( + `No active push tokens for user ${userId}, skipping push notification` + ) return } @@ -138,7 +140,10 @@ export async function sendPushNotification( error: result.error, }) - if (result.error?.includes('InvalidRegistration') || result.error?.includes('Unregistered')) { + if ( + result.error?.includes('InvalidRegistration') || + result.error?.includes('Unregistered') + ) { await db.pushDeviceToken.update({ where: { id: deviceToken.id }, data: { isActive: false }, diff --git a/src/events/types.ts b/src/events/types.ts index 55df7088..121168e7 100644 --- a/src/events/types.ts +++ b/src/events/types.ts @@ -96,6 +96,8 @@ export const EVENT_TYPE_TOPIC: Record = { 'approval.executed': 'transactions', 'approval.expired': 'transactions', 'approval.cancelled': 'transactions', + 'loan.liquidation_executed': 'transactions', + 'loan.charged_off': 'transactions', 'agent.rebalanced': 'agent', 'agent.decision_recorded': 'agent', 'agent.circuit_breaker_tripped': 'agent', diff --git a/src/index.ts b/src/index.ts index 50515c6b..480d58b9 100644 --- a/src/index.ts +++ b/src/index.ts @@ -65,6 +65,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 { scheduleLoanAccrual } from './jobs/loanAccrual' import { scheduleLoanLiquidationMonitor } from './jobs/loanLiquidationMonitor' import { scheduleLinkedExternalWalletSync } from './jobs/linkedExternalWalletSync' @@ -155,6 +156,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 let outboundNotificationsHandle: NodeJS.Timeout | null = null let linkedExternalWalletSyncHandle: NodeJS.Timeout | null = null let loanAccrualHandle: NodeJS.Timeout | null = null @@ -381,7 +383,10 @@ const apiRoutes: ApiRoute[] = [ { path: 'strategies', handlers: [strategiesRouter] }, { path: 'sub-accounts', handlers: [subAccountsRouter] }, { path: 'support/tickets', handlers: [supportTicketsRouter] }, - { path: 'admin/support/tickets', handlers: [adminRateLimiter, adminSupportTicketsRouter] }, + { + path: 'admin/support/tickets', + handlers: [adminRateLimiter, adminSupportTicketsRouter], + }, { path: 'assistant', handlers: [assistantRouter] }, { path: 'approvals', handlers: [approvalsRouter] }, { path: 'approval-policies', handlers: [approvalPoliciesRouter] }, @@ -390,7 +395,10 @@ const apiRoutes: ApiRoute[] = [ { path: 'stream', handlers: [streamRouter] }, { path: 'notifications', handlers: [notificationsRouter] }, { path: 'liquidity-floor', handlers: [liquidityFloorRouter] }, - { path: 'admin/notifications/dlq', handlers: [adminRateLimiter, notificationDlqRouter] }, + { + path: 'admin/notifications/dlq', + handlers: [adminRateLimiter, notificationDlqRouter], + }, { path: 'admin', handlers: [adminRateLimiter, adminRouter] }, ] @@ -526,6 +534,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') + } + if (linkedExternalWalletSyncHandle) { clearInterval(linkedExternalWalletSyncHandle) linkedExternalWalletSyncHandle = null @@ -608,7 +622,9 @@ async function initServices(): Promise { const { validateSecretCredentials } = await import('./config/secrets') const secretErrors = validateSecretCredentials() if (secretErrors.length) { - throw new Error(`[Startup] Secret validation failed: ${secretErrors.join('; ')}`) + throw new Error( + `[Startup] Secret validation failed: ${secretErrors.join('; ')}` + ) } if (config.nodeEnv === 'production') { @@ -762,6 +778,7 @@ async function main(): Promise { portfolioRiskJobHandle = schedulePortfolioRiskJob() approvalExpiryHandle = scheduleApprovalExpiry() reserveReconciliationHandle = scheduleReserveReconciliation() + messageDeliverySweepHandle = scheduleMessageDeliverySweep() outboundNotificationsHandle = scheduleOutboundNotifications() linkedExternalWalletSyncHandle = scheduleLinkedExternalWalletSync() // #532 — accrual first, then the monitor: the monitor values loans, and it diff --git a/src/jobs/alertRules.ts b/src/jobs/alertRules.ts index b57149f4..90006488 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 { sendPushNotification } from '../controllers/push-controller' import { @@ -235,7 +236,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, + }) } } @@ -243,9 +250,12 @@ async function deliverAlert( const title = `Alert: ${rule.metric}` const body = `${rule.metric} is ${observedValue} (threshold: ${threshold})` await sendPushNotification(rule.userId, title, body, data).catch((err) => { - logger.warn(`[AlertRules] Failed to send push notification for rule ${rule.id}`, { - error: err instanceof Error ? err.message : String(err), - }) + logger.warn( + `[AlertRules] Failed to send push notification for rule ${rule.id}`, + { + error: err instanceof Error ? err.message : String(err), + } + ) }) } } 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 2f309cd2..bca80773 100644 --- a/src/middleware/adminAuth.ts +++ b/src/middleware/adminAuth.ts @@ -49,6 +49,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', // #528 — treasury sweep policies, emergency sweep, signer rotation. 'treasury:read', 'treasury:write', diff --git a/src/middleware/consentPolicy.ts b/src/middleware/consentPolicy.ts index eb02f521..d21922c0 100644 --- a/src/middleware/consentPolicy.ts +++ b/src/middleware/consentPolicy.ts @@ -27,9 +27,12 @@ export function requireConsentPolicy( return (req: ConsentRequest, res: Response, next: NextFunction): void => { const userId = (req as any).user?.id || (req as any).userId || 'anonymous' const userConsents = req.userConsents || (req as any).user?.consents || {} - - const bypassReasonHeader = req.headers['x-consent-bypass-reason'] as AuditableBypassReason | undefined - const bypassJustificationHeader = req.headers['x-consent-bypass-justification'] as string | undefined + + const bypassReasonHeader = req.headers['x-consent-bypass-reason'] as + AuditableBypassReason | undefined + const bypassJustificationHeader = req.headers[ + 'x-consent-bypass-justification' + ] as string | undefined const validation = validateWorkflowConsent({ userId, diff --git a/src/middleware/impersonation.ts b/src/middleware/impersonation.ts index a55dcbb9..31c6ca71 100644 --- a/src/middleware/impersonation.ts +++ b/src/middleware/impersonation.ts @@ -5,7 +5,8 @@ import db from '../db' import { logger } from '../utils/logger' const IMPERSONATION_TTL_MINUTES = 30 -const IMPERSONATION_SECRET = process.env.IMPERSONATION_SECRET || 'impersonation-secret' +const IMPERSONATION_SECRET = + process.env.IMPERSONATION_SECRET || 'impersonation-secret' export interface ImpersonationContext { adminId: string @@ -26,7 +27,9 @@ export function generateImpersonationToken( ): string { const sessionId = crypto.randomUUID() const createdAt = new Date() - const expiresAt = new Date(createdAt.getTime() + IMPERSONATION_TTL_MINUTES * 60 * 1000) + const expiresAt = new Date( + createdAt.getTime() + IMPERSONATION_TTL_MINUTES * 60 * 1000 + ) const payload: ImpersonationContext = { adminId, @@ -49,7 +52,10 @@ export function verifyImpersonationToken( token: string ): ImpersonationContext | null { try { - const decoded = jwt.verify(token, IMPERSONATION_SECRET) as ImpersonationContext + const decoded = jwt.verify( + token, + IMPERSONATION_SECRET + ) as ImpersonationContext if (new Date() > new Date(decoded.expiresAt)) { return null diff --git a/src/notifications/service.ts b/src/notifications/service.ts index 32bc9707..0ea05dbe 100644 --- a/src/notifications/service.ts +++ b/src/notifications/service.ts @@ -5,7 +5,10 @@ import db from '../db' import { logger } from '../utils/logger' -import { generateCorrelationId, runWithCorrelationIdAsync } from '../utils/correlation' +import { + generateCorrelationId, + runWithCorrelationIdAsync, +} from '../utils/correlation' import { NotificationChannel, NotificationCategory, @@ -263,10 +266,7 @@ function simpleHash(str: string): string { /** * Verify unsubscribe token */ -export function verifyUnsubscribeToken( - token: string, - userId: string -): boolean { +export function verifyUnsubscribeToken(token: string, userId: string): boolean { try { const [hash, timestamp] = token.split(':') @@ -321,13 +321,9 @@ export async function processUnsubscribe( // Disable preferences for (const category of categories) { - await setPreference( - request.userId, - request.channel, - category, - false, - { updatedBy: 'user' } - ) + await setPreference(request.userId, request.channel, category, false, { + updatedBy: 'user', + }) // Log unsubscribe action await logPreferenceChange({ diff --git a/src/privacy/consentPolicy.ts b/src/privacy/consentPolicy.ts index 32c661cf..28f11ddc 100644 --- a/src/privacy/consentPolicy.ts +++ b/src/privacy/consentPolicy.ts @@ -54,7 +54,10 @@ export interface ConsentValidationResult { } /** Mapping of high-risk workflows to their mandatory consent scope */ -export const WORKFLOW_CONSENT_REQUIREMENTS: Record = { +export const WORKFLOW_CONSENT_REQUIREMENTS: Record< + HighRiskWorkflow, + ConsentScope +> = { AUTOMATED_TRADING: 'consent:automated_trading', PROFILE_ANALYTICS: 'consent:data_processing', THIRD_PARTY_EXPORT: 'consent:third_party_sharing', @@ -70,7 +73,14 @@ export const WORKFLOW_CONSENT_REQUIREMENTS: Record= cappedLoss ? 'FULL' : payoutAmount > 0 ? 'PRORATED' : 'ZERO' + const coverageType = + payoutAmount >= cappedLoss + ? 'FULL' + : payoutAmount > 0 + ? 'PRORATED' + : 'ZERO' const claim = await db.coverageClaim.create({ data: { @@ -229,17 +256,21 @@ export async function computeClaimsForEvent(eventId: string) { if (remainingFund <= 0) break } - logger.info(`[ProtectionFund] Computed ${claims.length} claims for event ${eventId}`) + logger.info( + `[ProtectionFund] Computed ${claims.length} claims for event ${eventId}` + ) return claims } -function computePositionLoss(position: any, lossStart: Date, lossEnd: Date): number { - const relevantTxs = position.transactions.filter( - (tx: any) => { - const txDate = new Date(tx.createdAt) - return txDate >= lossStart && txDate <= lossEnd - } - ) +function computePositionLoss( + position: any, + lossStart: Date, + lossEnd: Date +): number { + const relevantTxs = position.transactions.filter((tx: any) => { + const txDate = new Date(tx.createdAt) + return txDate >= lossStart && txDate <= lossEnd + }) let totalDeposited = 0 let totalWithdrawn = 0 @@ -266,7 +297,8 @@ export async function executePayout(claimId: string) { include: { event: true }, }) if (!claim) throw new Error('Claim not found') - if (claim.status !== 'PENDING') throw new Error(`Claim is not pending (status: ${claim.status})`) + if (claim.status !== 'PENDING') + throw new Error(`Claim is not pending (status: ${claim.status})`) if (claim.payoutAmount.lte(0)) throw new Error('Claim has zero payout') const outboxOp = await enqueueOutboxOp(db, { @@ -301,7 +333,13 @@ export async function executePayout(claimId: string) { payloadHash: 'sha256:0', blockType: 'TXN_BATCH', createdAt: new Date(), - payloads: [{ type: 'PROTECTION_FUND_PAYOUT', claimId, amount: Number(claim.payoutAmount) }], + payloads: [ + { + type: 'PROTECTION_FUND_PAYOUT', + claimId, + amount: Number(claim.payoutAmount), + }, + ], }) return outboxOp @@ -348,7 +386,10 @@ export async function getMyCoverage(userId: string) { coverageTerms: terms, estimatedCoverage: exposure .filter((e) => e.eligibleForCoverage) - .reduce((sum, e) => sum + Math.min(e.currentValue, terms.perUserCoverageCap), 0), + .reduce( + (sum, e) => sum + Math.min(e.currentValue, terms.perUserCoverageCap), + 0 + ), } } diff --git a/src/push/pushProvider.ts b/src/push/pushProvider.ts index 4a8be2ae..7ea30b5b 100644 --- a/src/push/pushProvider.ts +++ b/src/push/pushProvider.ts @@ -49,9 +49,12 @@ export class FcmPushProvider implements PushProvider { try { const messageId = `push_fcm_${Date.now()}` - logger.info(`[FcmPushProvider] Sent push via FCM to ${message.deviceToken}`, { - messageId, - }) + logger.info( + `[FcmPushProvider] Sent push via FCM to ${message.deviceToken}`, + { + messageId, + } + ) return { messageId, provider: this.name, success: true } } catch (err) { @@ -83,9 +86,12 @@ export class ApnsPushProvider implements PushProvider { try { const messageId = `push_apns_${Date.now()}` - logger.info(`[ApnsPushProvider] Sent push via APNs to ${message.deviceToken}`, { - messageId, - }) + logger.info( + `[ApnsPushProvider] Sent push via APNs to ${message.deviceToken}`, + { + messageId, + } + ) return { messageId, provider: this.name, success: true } } catch (err) { @@ -126,16 +132,15 @@ export class PushRegistry { getProvider(platform: string): PushProvider { const provider = this.providers.get(platform.toLowerCase()) if (!provider) { - logger.warn(`[PushRegistry] No provider for platform "${platform}", falling back to mock`) + logger.warn( + `[PushRegistry] No provider for platform "${platform}", falling back to mock` + ) return this.providers.get('mock')! } return provider } - async send( - platform: string, - message: PushMessage - ): Promise { + async send(platform: string, message: PushMessage): Promise { const provider = this.getProvider(platform) return provider.send(message) } diff --git a/src/referral/service.ts b/src/referral/service.ts index 5a9b7da6..fd74f61f 100644 --- a/src/referral/service.ts +++ b/src/referral/service.ts @@ -1089,7 +1089,14 @@ export async function referralLeaderboard( page: number = 1, limit: number = 20, includeDisplayName: boolean = false -): Promise<{ ownerId: string; code: string; conversions: number; displayName?: string | null }[]> { +): Promise< + { + ownerId: string + code: string + conversions: number + displayName?: string | null + }[] +> { const codes = await db.referralCode.findMany({ include: { owner: includeDisplayName @@ -1116,7 +1123,8 @@ export async function referralLeaderboard( ownerId: c.ownerUserId, code: c.code, conversions: c._count.conversions, - ...(includeDisplayName ? { displayName: (c as any).owner?.displayName ?? null } : {}), + ...(includeDisplayName + ? { displayName: (c as any).owner?.displayName ?? null } + : {}), })) } - diff --git a/src/routes/admin.ts b/src/routes/admin.ts index e2fa2753..db3c8c8b 100644 --- a/src/routes/admin.ts +++ b/src/routes/admin.ts @@ -31,6 +31,7 @@ import { manualTripBreaker, manualResetBreaker, } from '../agent/breakerService' +import { messageDeliveryService } from '../messaging' const router = Router() const prisma = db @@ -181,8 +182,12 @@ router.get( }) } catch (error) { const message = error instanceof Error ? error.message : 'Unknown error' - logger.error('[Admin] Failed to generate job dashboard:', { error: message }) - res.status(500).json({ error: 'Failed to generate job dashboard', message }) + logger.error('[Admin] Failed to generate job dashboard:', { + error: message, + }) + res + .status(500) + .json({ error: 'Failed to generate job dashboard', message }) } } ) @@ -2324,6 +2329,226 @@ 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 // ── Privacy Erasure Routes (#489) ─────────────────────────────────────────── diff --git a/src/routes/alerts.ts b/src/routes/alerts.ts index b5e023db..88be76fd 100644 --- a/src/routes/alerts.ts +++ b/src/routes/alerts.ts @@ -140,7 +140,7 @@ router.get( sortOrder: 'asc' | 'desc' } - const where: any = { userId } + const where: any = { userId, deletedAt: null } if (query.isActive !== undefined) where.isActive = query.isActive === 'true' if (query.metric) where.metric = query.metric if (query.protocolName) where.protocolName = query.protocolName @@ -155,11 +155,6 @@ router.get( take: limit, }), ]) - const rules = await (db as any).alertRule.findMany({ - where: { userId, deletedAt: null }, - select: alertSelect, - orderBy: { createdAt: 'desc' }, - }) return res .status(200) diff --git a/src/routes/analytics.ts b/src/routes/analytics.ts index 557eb6c5..ae5b93a6 100644 --- a/src/routes/analytics.ts +++ b/src/routes/analytics.ts @@ -340,15 +340,18 @@ router.get( // PDF export (#538) — same data as the CSV export, rendered as a // paginated plain-text report rather than a formatted table. if (format.data.format === 'pdf') { - const pdf = toSimplePdf(`Portfolio Yield Report — ${period.data.period}`, [ - `User: ${userId}`, - `Total yield: ${totalYield}`, - `Period yield: ${periodYield}`, - `Average APY: ${averageApy}`, - '', - 'Date Yield Amount APY', - ...points.map((p) => `${p.date} ${p.yieldAmount} ${p.apy}`), - ]) + const pdf = toSimplePdf( + `Portfolio Yield Report — ${period.data.period}`, + [ + `User: ${userId}`, + `Total yield: ${totalYield}`, + `Period yield: ${periodYield}`, + `Average APY: ${averageApy}`, + '', + 'Date Yield Amount APY', + ...points.map((p) => `${p.date} ${p.yieldAmount} ${p.apy}`), + ] + ) res.setHeader('Content-Type', 'application/pdf') res.setHeader( 'Content-Disposition', diff --git a/src/routes/notification-dlq.ts b/src/routes/notification-dlq.ts index d326671e..bd9676dc 100644 --- a/src/routes/notification-dlq.ts +++ b/src/routes/notification-dlq.ts @@ -1,8 +1,5 @@ import { Router } from 'express' -import { - requireAdminAuth, - requireAdminScope, -} from '../middleware/adminAuth' +import { requireAdminAuth, requireAdminScope } from '../middleware/adminAuth' import { listDeadOutboundNotifications, retryDeadOutboundNotification, @@ -16,15 +13,15 @@ router.get( requireAdminAuth, requireAdminScope('dlq:read'), async (_req, res) => { - try { - const items = await listDeadOutboundNotifications() - res.status(200).json({ items, count: items.length }) - } catch (error) { - logger.error('[Admin] Failed to inspect outbound notification DLQ', { - error: error instanceof Error ? error.message : String(error), - }) - res.status(500).json({ error: 'Failed to inspect notification queue' }) - } + try { + const items = await listDeadOutboundNotifications() + res.status(200).json({ items, count: items.length }) + } catch (error) { + logger.error('[Admin] Failed to inspect outbound notification DLQ', { + error: error instanceof Error ? error.message : String(error), + }) + res.status(500).json({ error: 'Failed to inspect notification queue' }) + } } ) @@ -50,4 +47,4 @@ router.post( } ) -export default router \ No newline at end of file +export default router diff --git a/src/routes/portfolio.ts b/src/routes/portfolio.ts index ba0611ca..6e407ca7 100644 --- a/src/routes/portfolio.ts +++ b/src/routes/portfolio.ts @@ -255,8 +255,14 @@ router.get( const displayCurrency = user.displayCurrency || 'USD' const { convertUsdToCurrency } = await import('../utils/fxConvert') - const displayTotalBalance = convertUsdToCurrency(totalBalance, displayCurrency) - const displayTotalEarnings = convertUsdToCurrency(totalEarnings, displayCurrency) + const displayTotalBalance = convertUsdToCurrency( + totalBalance, + displayCurrency + ) + const displayTotalEarnings = convertUsdToCurrency( + totalEarnings, + displayCurrency + ) return res.status(200).json({ userId: user.id, diff --git a/src/routes/protection-fund.ts b/src/routes/protection-fund.ts index 4b474006..f8239d77 100644 --- a/src/routes/protection-fund.ts +++ b/src/routes/protection-fund.ts @@ -33,7 +33,9 @@ router.get('/status', async (_req: Request, res: Response) => { const status = await getProtectionFundStatus() return res.json(status) } catch (err) { - logger.error('[ProtectionFund] Status query failed', { error: err instanceof Error ? err.message : String(err) }) + logger.error('[ProtectionFund] Status query failed', { + error: err instanceof Error ? err.message : String(err), + }) return sendError(res, 500, 'Failed to load protection fund status') } }) @@ -44,7 +46,9 @@ router.get('/my-coverage', requireAuth, async (req: Request, res: Response) => { const coverage = await getMyCoverage(req.auth!.userId) return res.json(coverage) } catch (err) { - logger.error('[ProtectionFund] My-coverage query failed', { error: err instanceof Error ? err.message : String(err) }) + logger.error('[ProtectionFund] My-coverage query failed', { + error: err instanceof Error ? err.message : String(err), + }) return sendError(res, 500, 'Failed to load coverage') } }) @@ -68,10 +72,15 @@ router.post( message: 'Coverage event declared and pending review', }) } catch (err) { - if (err instanceof Error && err.message.includes('not a covered event type')) { + if ( + err instanceof Error && + err.message.includes('not a covered event type') + ) { return sendError(res, 400, err.message) } - logger.error('[ProtectionFund] Declare event failed', { error: err instanceof Error ? err.message : String(err) }) + logger.error('[ProtectionFund] Declare event failed', { + error: err instanceof Error ? err.message : String(err), + }) return sendError(res, 500, 'Failed to declare coverage event') } } @@ -89,11 +98,17 @@ router.post( try { const { eventId } = req.params as unknown as { eventId: string } const { approved } = req.body as { approved: boolean } - const event = await reviewCoverageEvent(eventId, approved, req.auth!.userId) + const event = await reviewCoverageEvent( + eventId, + approved, + req.auth!.userId + ) return res.json({ eventId: event.id, status: event.status, - message: approved ? 'Event approved, claims computed' : 'Event rejected', + message: approved + ? 'Event approved, claims computed' + : 'Event rejected', }) } catch (err) { if (err instanceof Error && err.message === 'Coverage event not found') { @@ -102,7 +117,9 @@ router.post( if (err instanceof Error && err.message.includes('not pending review')) { return sendError(res, 409, err.message) } - logger.error('[ProtectionFund] Review event failed', { error: err instanceof Error ? err.message : String(err) }) + logger.error('[ProtectionFund] Review event failed', { + error: err instanceof Error ? err.message : String(err), + }) return sendError(res, 500, 'Failed to review event') } } @@ -135,7 +152,9 @@ router.post( assetSymbol: contribution.assetSymbol, }) } catch (err) { - logger.error('[ProtectionFund] Record contribution failed', { error: err instanceof Error ? err.message : String(err) }) + logger.error('[ProtectionFund] Record contribution failed', { + error: err instanceof Error ? err.message : String(err), + }) return sendError(res, 500, 'Failed to record contribution') } } @@ -160,7 +179,9 @@ router.post( if (err instanceof Error && err.message === 'Claim not found') { return sendNotFound(res, 'Claim not found') } - logger.error('[ProtectionFund] Payout failed', { error: err instanceof Error ? err.message : String(err) }) + logger.error('[ProtectionFund] Payout failed', { + error: err instanceof Error ? err.message : String(err), + }) return sendError(res, 500, 'Failed to execute payout') } } diff --git a/src/routes/recurring-withdrawals.ts b/src/routes/recurring-withdrawals.ts index 2b9f6c87..14b13a70 100644 --- a/src/routes/recurring-withdrawals.ts +++ b/src/routes/recurring-withdrawals.ts @@ -303,12 +303,10 @@ router.patch( (amountMode === 'PERCENT_OF_BALANCE' && Number(updates.percentage ?? existing.percentage) <= 0) ) { - res - .status(400) - .json({ - error: - 'The selected amount mode requires a positive amount or percentage', - }) + res.status(400).json({ + error: + 'The selected amount mode requires a positive amount or percentage', + }) return } if (existing.lastRunStatus === 'executing') { diff --git a/src/routes/settings.ts b/src/routes/settings.ts index fe21c587..2f7b8279 100644 --- a/src/routes/settings.ts +++ b/src/routes/settings.ts @@ -43,7 +43,9 @@ router.put( displayCurrency: updated.displayCurrency, }) } catch (err) { - logger.error('[Settings] Update failed', { error: err instanceof Error ? err.message : String(err) }) + logger.error('[Settings] Update failed', { + error: err instanceof Error ? err.message : String(err), + }) return sendError(res, 500, 'Failed to update settings') } } diff --git a/src/routes/sub-accounts.ts b/src/routes/sub-accounts.ts index 90431212..ea87938f 100644 --- a/src/routes/sub-accounts.ts +++ b/src/routes/sub-accounts.ts @@ -101,11 +101,9 @@ router.post( const { operations, atomic } = bulkSubAccountsSchema.parse(req.body) const parentUserId = req.auth!.userId if (operations.length > MAX_BULK_BATCH_SIZE) { - res - .status(400) - .json({ - error: `Batch size exceeds maximum limit of ${MAX_BULK_BATCH_SIZE} operations`, - }) + res.status(400).json({ + error: `Batch size exceeds maximum limit of ${MAX_BULK_BATCH_SIZE} operations`, + }) return } const errorText = (err: unknown) => @@ -140,24 +138,22 @@ router.post( audit(parentUserId, result.action, result.data) res.json({ atomic: true, results }) } catch (err) { - res - .status(400) - .json({ - atomic: true, - rolledBack: true, - failedIndex, - error: 'Atomic batch failed; transaction rolled back', - message: errorText(err), - results: operations.map((_, index) => ({ - index, - success: false, - ...(index === failedIndex - ? { error: errorText(err) } - : index < failedIndex - ? { rolledBack: true } - : { skipped: true }), - })), - }) + res.status(400).json({ + atomic: true, + rolledBack: true, + failedIndex, + error: 'Atomic batch failed; transaction rolled back', + message: errorText(err), + results: operations.map((_, index) => ({ + index, + success: false, + ...(index === failedIndex + ? { error: errorText(err) } + : index < failedIndex + ? { rolledBack: true } + : { skipped: true }), + })), + }) } return } diff --git a/src/routes/telegram.ts b/src/routes/telegram.ts index 52ed8081..83e53cae 100644 --- a/src/routes/telegram.ts +++ b/src/routes/telegram.ts @@ -1,7 +1,7 @@ import express, { Request, Response } from 'express' import { handleTelegramMessage } from '../telegram/handler' +import { messageDeliveryService } from '../messaging' import { logger } from '../utils/logger' -import { enqueueOutboundNotification } from '../services/outboundNotifications' const router = express.Router() @@ -31,13 +31,20 @@ 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 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') } - - await enqueueOutboundNotification('telegram', payload) return res.status(200).send('OK') } catch (error) { logger.error('[Telegram webhook] error handling message:', error) diff --git a/src/services/outboundNotifications.ts b/src/services/outboundNotifications.ts index e9d6fe32..4c06230a 100644 --- a/src/services/outboundNotifications.ts +++ b/src/services/outboundNotifications.ts @@ -6,14 +6,20 @@ import { } from '../utils/metrics' export type OutboundNotificationChannel = - | 'email' - | 'sms' - | 'whatsapp' - | 'telegram' + 'email' | 'sms' | 'whatsapp' | 'telegram' -const MAX_ATTEMPTS = Math.max(1, Number(process.env.NOTIFICATION_MAX_ATTEMPTS) || 8) -const BASE_DELAY_MS = Math.max(100, Number(process.env.NOTIFICATION_BASE_DELAY_MS) || 1000) -const MAX_DELAY_MS = Math.max(BASE_DELAY_MS, Number(process.env.NOTIFICATION_MAX_DELAY_MS) || 300000) +const MAX_ATTEMPTS = Math.max( + 1, + Number(process.env.NOTIFICATION_MAX_ATTEMPTS) || 8 +) +const BASE_DELAY_MS = Math.max( + 100, + Number(process.env.NOTIFICATION_BASE_DELAY_MS) || 1000 +) +const MAX_DELAY_MS = Math.max( + BASE_DELAY_MS, + Number(process.env.NOTIFICATION_MAX_DELAY_MS) || 300000 +) const BATCH_SIZE = 50 type NotificationRecord = { @@ -67,12 +73,17 @@ async function deliver(notification: NotificationRecord): Promise { signal: AbortSignal.timeout(10_000), } ) - if (!response.ok) throw new Error(`Telegram API returned HTTP ${response.status}`) - const result = (await response.json()) as { result?: { message_id?: number } } + if (!response.ok) + throw new Error(`Telegram API returned HTTP ${response.status}`) + const result = (await response.json()) as { + result?: { message_id?: number } + } return String(result.result?.message_id ?? 'accepted') } } - throw new Error(`Unsupported outbound notification channel: ${notification.channel}`) + throw new Error( + `Unsupported outbound notification channel: ${notification.channel}` + ) } function retryDelayMs(attempt: number): number { @@ -118,7 +129,10 @@ export async function dispatchOutboundNotifications(): Promise { where: { id: notification.id }, data: { status: 'DELIVERED', providerId, lastError: null }, }) - outboundNotificationAttempts.inc({ channel: notification.channel, status: 'delivered' }) + outboundNotificationAttempts.inc({ + channel: notification.channel, + status: 'delivered', + }) } catch (error) { const message = error instanceof Error ? error.message : String(error) const dead = attempt >= MAX_ATTEMPTS @@ -164,7 +178,9 @@ export async function listDeadOutboundNotifications(limit = 100) { }) } -export async function retryDeadOutboundNotification(id: string): Promise { +export async function retryDeadOutboundNotification( + id: string +): Promise { const result = await (db as any).outboundNotification.updateMany({ where: { id, status: 'DEAD' }, data: { @@ -187,4 +203,4 @@ export function scheduleOutboundNotifications(): NodeJS.Timeout { }, 10_000) handle.unref() return handle -} \ No newline at end of file +} diff --git a/src/services/totp.service.ts b/src/services/totp.service.ts index af830806..cb968bec 100644 --- a/src/services/totp.service.ts +++ b/src/services/totp.service.ts @@ -23,7 +23,13 @@ export interface TotpChallenge { export interface TotpChallengeResult { ok: true - user: { id: string; walletAddress: string; displayName: string | null; email: string | null; network: Network } + user: { + id: string + walletAddress: string + displayName: string | null + email: string | null + network: Network + } stellarPubKey: string network: Network referralCode?: string | null @@ -148,7 +154,10 @@ export async function completeTotpChallenge( select: { secretEncrypted: true, verifiedAt: true }, }) - if (!credential?.verifiedAt || !verifyTotpCode(credential.secretEncrypted, code)) { + if ( + !credential?.verifiedAt || + !verifyTotpCode(credential.secretEncrypted, code) + ) { return { ok: false, reason: 'invalid_code' } } @@ -156,7 +165,13 @@ export async function completeTotpChallenge( const user = await db.user.findUniqueOrThrow({ where: { id: challenge.userId }, - select: { id: true, walletAddress: true, displayName: true, email: true, network: true }, + select: { + id: true, + walletAddress: true, + displayName: true, + email: true, + network: true, + }, }) logger.info('[TOTP] Challenge completed', { userId: challenge.userId }) diff --git a/src/stellar/driftCheck.ts b/src/stellar/driftCheck.ts index 2cf5f24a..6342353f 100644 --- a/src/stellar/driftCheck.ts +++ b/src/stellar/driftCheck.ts @@ -47,11 +47,22 @@ const EXPECTED_DEFAULT_RPCS: Record = { export function checkStellarIntegrationDrift( customConfig?: StellarDriftConfig ): DriftCheckReport { - const envNetwork = (customConfig?.network || process.env.STELLAR_NETWORK || 'testnet').toLowerCase() - const rpcUrl = customConfig?.rpcUrl || process.env.STELLAR_RPC_URL || process.env.STELLAR_RPC_URLS || '' - const vaultContractId = customConfig?.vaultContractId || process.env.VAULT_CONTRACT_ID || '' - const usdcTokenAddress = customConfig?.usdcTokenAddress || process.env.USDC_TOKEN_ADDRESS || '' - const agentSecretKey = customConfig?.agentSecretKey || process.env.STELLAR_AGENT_SECRET_KEY || '' + const envNetwork = ( + customConfig?.network || + process.env.STELLAR_NETWORK || + 'testnet' + ).toLowerCase() + const rpcUrl = + customConfig?.rpcUrl || + process.env.STELLAR_RPC_URL || + process.env.STELLAR_RPC_URLS || + '' + const vaultContractId = + customConfig?.vaultContractId || process.env.VAULT_CONTRACT_ID || '' + const usdcTokenAddress = + customConfig?.usdcTokenAddress || process.env.USDC_TOKEN_ADDRESS || '' + const agentSecretKey = + customConfig?.agentSecretKey || process.env.STELLAR_AGENT_SECRET_KEY || '' const drifts: DriftIssue[] = [] let checksPassed = 0 @@ -112,7 +123,10 @@ export function checkStellarIntegrationDrift( }) } else { // Check for network mismatch in RPC URL - if (envNetwork === 'mainnet' && (rpcUrl.includes('testnet') || rpcUrl.includes('futurenet'))) { + if ( + envNetwork === 'mainnet' && + (rpcUrl.includes('testnet') || rpcUrl.includes('futurenet')) + ) { drifts.push({ component: 'network', field: 'STELLAR_RPC_URL', @@ -170,7 +184,10 @@ export function checkStellarIntegrationDrift( severity: 'error', description: 'USDC_TOKEN_ADDRESS is missing from deployment metadata', }) - } else if (!isValidContractId(usdcTokenAddress) && !StrKey.isValidEd25519PublicKey(usdcTokenAddress)) { + } else if ( + !isValidContractId(usdcTokenAddress) && + !StrKey.isValidEd25519PublicKey(usdcTokenAddress) + ) { drifts.push({ component: 'contract', field: 'USDC_TOKEN_ADDRESS', @@ -192,7 +209,8 @@ export function checkStellarIntegrationDrift( expected: 'Valid Stellar Secret Key (56 chars starting with S)', actual: 'missing', severity: 'error', - description: 'STELLAR_AGENT_SECRET_KEY is missing from environment metadata', + description: + 'STELLAR_AGENT_SECRET_KEY is missing from environment metadata', }) } else if (!StrKey.isValidEd25519SecretSeed(agentSecretKey)) { drifts.push({ @@ -201,7 +219,8 @@ export function checkStellarIntegrationDrift( expected: 'Valid Stellar Secret Key (56 chars starting with S)', actual: 'invalid_secret_key_format', severity: 'error', - description: 'STELLAR_AGENT_SECRET_KEY fails Ed25519 secret seed validation', + description: + 'STELLAR_AGENT_SECRET_KEY fails Ed25519 secret seed validation', }) } else { checksPassed++ @@ -220,5 +239,7 @@ export function checkStellarIntegrationDrift( } function isValidContractId(id: string): boolean { - return Boolean(id && id.length === 56 && id.startsWith('C') && StrKey.isValidContract(id)) + return Boolean( + id && id.length === 56 && id.startsWith('C') && StrKey.isValidContract(id) + ) } diff --git a/src/strategy/service.ts b/src/strategy/service.ts index 14e5d463..1bce8501 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, @@ -720,8 +721,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/job-metrics.ts b/src/utils/job-metrics.ts index 4c9f60ac..1dd6f8a7 100644 --- a/src/utils/job-metrics.ts +++ b/src/utils/job-metrics.ts @@ -138,7 +138,8 @@ export async function getJobDashboardSnapshot(): Promise<{ } else if (metric.name === 'job_completion_status') { for (const val of metric.values) { const job = (val.labels as any).job_name - if (job) getOrCreate(job).lastStatus = val.value === 1 ? 'success' : 'failed' + if (job) + getOrCreate(job).lastStatus = val.value === 1 ? 'success' : 'failed' } } } @@ -148,4 +149,3 @@ export async function getJobDashboardSnapshot(): Promise<{ jobs: jobMap, } } - diff --git a/src/utils/metrics.ts b/src/utils/metrics.ts index 33a9145a..33e2d6aa 100644 --- a/src/utils/metrics.ts +++ b/src/utils/metrics.ts @@ -1027,6 +1027,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) +} + // ── Queue Health Metrics (#520) ───────────────────────────────────────────── export const queueLagSeconds = new client.Gauge({ diff --git a/src/utils/pdf.ts b/src/utils/pdf.ts index bad466fa..e6ff9973 100644 --- a/src/utils/pdf.ts +++ b/src/utils/pdf.ts @@ -20,7 +20,8 @@ function escapePdfText(text: string): string { function chunk(items: T[], size: number): T[][] { const out: T[][] = [] - for (let i = 0; i < items.length; i += size) out.push(items.slice(i, i + size)) + for (let i = 0; i < items.length; i += size) + out.push(items.slice(i, i + size)) return out } @@ -51,7 +52,9 @@ export function toSimplePdf(title: string, lines: string[]): Buffer { let y = PAGE_HEIGHT - MARGIN streamLines.push(`BT /F1 ${FONT_SIZE} Tf ${MARGIN} ${y} Td`) if (i === 0) { - streamLines.push(`(${escapePdfText(title)}) Tj 0 -${LINE_HEIGHT * 1.5} Td`) + streamLines.push( + `(${escapePdfText(title)}) Tj 0 -${LINE_HEIGHT * 1.5} Td` + ) } for (const line of pageLines) { streamLines.push(`(${escapePdfText(line)}) Tj 0 -${LINE_HEIGHT} Td`) 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 efacdadc..f7c069b8 100644 --- a/src/utils/twilio-client.ts +++ b/src/utils/twilio-client.ts @@ -39,11 +39,19 @@ export interface SendMessageParams { export async function sendWhatsAppMessage( params: SendMessageParams ): Promise { - return enqueueOutboundNotification('whatsapp', params as unknown as Record) + return enqueueOutboundNotification( + 'whatsapp', + params as unknown as Record + ) } -export async function sendSmsMessage(params: SendMessageParams): Promise { - return enqueueOutboundNotification('sms', params as unknown as Record) +export async function sendSmsMessage( + params: SendMessageParams +): Promise { + return enqueueOutboundNotification( + 'sms', + params as unknown as Record + ) } export async function sendTwilioMessageNow( @@ -75,3 +83,37 @@ 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 ac412215..5b64ec82 100644 --- a/src/validators/webhook-validators.ts +++ b/src/validators/webhook-validators.ts @@ -53,6 +53,9 @@ const WEBHOOK_EVENTS = [ 'approval.executed', 'approval.expired', 'approval.cancelled', + // Lending events (#532) + 'loan.liquidation_executed', + 'loan.charged_off', ] as const export const createWebhookSchema = z.object({ diff --git a/tests/integration/agent/rebalanceCost.integration.test.ts b/tests/integration/agent/rebalanceCost.integration.test.ts index 760a3e9f..41cca32d 100644 --- a/tests/integration/agent/rebalanceCost.integration.test.ts +++ b/tests/integration/agent/rebalanceCost.integration.test.ts @@ -45,7 +45,8 @@ const mockOutboxOpFindUnique = jest.fn().mockResolvedValue(null) const mockPersistDecision = jest.fn().mockResolvedValue('decision-1') jest.mock('../../../src/agent/rebalanceDecision', () => ({ - persistRebalanceDecision: (...args: unknown[]) => mockPersistDecision(...args), + persistRebalanceDecision: (...args: unknown[]) => + mockPersistDecision(...args), })) jest.mock('../../../src/db', () => { diff --git a/tests/integration/support-tickets.integration.test.ts b/tests/integration/support-tickets.integration.test.ts index 83150dee..d5840561 100644 --- a/tests/integration/support-tickets.integration.test.ts +++ b/tests/integration/support-tickets.integration.test.ts @@ -154,13 +154,11 @@ describe('Support Ticket Integration Tests', () => { it('enforces internal visibility in both user-facing database read queries', async () => { const app = buildApp() - const created = await request(app) - .post('/api/v1/support/tickets') - .send({ - subject: 'Private notes', - category: 'OTHER', - body: 'Please help with this issue', - }) + const created = await request(app).post('/api/v1/support/tickets').send({ + subject: 'Private notes', + category: 'OTHER', + body: 'Please help with this issue', + }) await request(app).get('/api/v1/support/tickets') expect(db.supportTicket.findMany).toHaveBeenLastCalledWith( expect.objectContaining({ @@ -180,13 +178,11 @@ describe('Support Ticket Integration Tests', () => { it('rejects reads and replies for another owner', async () => { const app = buildApp() - const created = await request(app) - .post('/api/v1/support/tickets') - .send({ - subject: 'Ownership test', - category: 'OTHER', - body: 'Please help with this issue', - }) + const created = await request(app).post('/api/v1/support/tickets').send({ + subject: 'Ownership test', + category: 'OTHER', + body: 'Please help with this issue', + }) tickets.get(created.body.ticket.id).userId = 'another-user' expect( ( @@ -206,14 +202,12 @@ describe('Support Ticket Integration Tests', () => { it('computes urgent SLA breaches and excludes closed tickets', async () => { const app = buildApp() - const created = await request(app) - .post('/api/v1/support/tickets') - .send({ - subject: 'Urgent test', - category: 'OTHER', - priority: 'URGENT', - body: 'Please help with this issue', - }) + const created = await request(app).post('/api/v1/support/tickets').send({ + subject: 'Urgent test', + category: 'OTHER', + priority: 'URGENT', + body: 'Please help with this issue', + }) tickets.get(created.body.ticket.id).createdAt = new Date( Date.now() - 3 * 3600000 ) diff --git a/tests/integration/webauthn.integration.test.ts b/tests/integration/webauthn.integration.test.ts index 0e5fc74f..48cd3278 100644 --- a/tests/integration/webauthn.integration.test.ts +++ b/tests/integration/webauthn.integration.test.ts @@ -269,12 +269,10 @@ describe('passkey security and session integration', () => { mockChallenges.get(options.body.challengeId).sessionId = 'another-session' expect( ( - await request(app) - .post('/api/v1/webauthn/register-verify') - .send({ - challengeId: options.body.challengeId, - response: registration, - }) + await request(app).post('/api/v1/webauthn/register-verify').send({ + challengeId: options.body.challengeId, + response: registration, + }) ).status ).toBe(400) expect(verifyRegistrationResponse).not.toHaveBeenCalled() 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/mail/mailFailover.test.ts b/tests/unit/mail/mailFailover.test.ts index 0831e7b6..96dcd6c7 100644 --- a/tests/unit/mail/mailFailover.test.ts +++ b/tests/unit/mail/mailFailover.test.ts @@ -145,13 +145,11 @@ describe('MailRegistry failover (#496)', () => { const reject = new SesSignatureVerificationError('invalid_signature') ;(primary.parseWebhook as jest.Mock).mockRejectedValue(reject) const fallback = makeProvider('fallback') - fallback.parseWebhook = jest - .fn() - .mockResolvedValue({ - type: 'delivery', - messageId: 'm-2', - recipient: 'u@e.co', - }) + fallback.parseWebhook = jest.fn().mockResolvedValue({ + type: 'delivery', + messageId: 'm-2', + recipient: 'u@e.co', + }) const registry = new MailRegistry(primary, fallback) // A forged/tampered SES payload must never be rescued into processing by a 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/privacy/consentPolicy.test.ts b/tests/unit/privacy/consentPolicy.test.ts index 040b3c5b..1ff0060c 100644 --- a/tests/unit/privacy/consentPolicy.test.ts +++ b/tests/unit/privacy/consentPolicy.test.ts @@ -12,12 +12,24 @@ describe('User Consent and Data-Processing Policy Engine (#510)', () => { const userId = 'user_test_123' it('maps all high-risk workflows to their required consent scopes', () => { - expect(WORKFLOW_CONSENT_REQUIREMENTS.AUTOMATED_TRADING).toBe('consent:automated_trading') - expect(WORKFLOW_CONSENT_REQUIREMENTS.PROFILE_ANALYTICS).toBe('consent:data_processing') - expect(WORKFLOW_CONSENT_REQUIREMENTS.THIRD_PARTY_EXPORT).toBe('consent:third_party_sharing') - expect(WORKFLOW_CONSENT_REQUIREMENTS.NOTIFICATIONS_MARKETING).toBe('consent:marketing') - expect(WORKFLOW_CONSENT_REQUIREMENTS.AI_PORTFOLIO_ADVISORY).toBe('consent:ai_advisory') - expect(WORKFLOW_CONSENT_REQUIREMENTS.TELEMETRY_COLLECTION).toBe('consent:telemetry') + expect(WORKFLOW_CONSENT_REQUIREMENTS.AUTOMATED_TRADING).toBe( + 'consent:automated_trading' + ) + expect(WORKFLOW_CONSENT_REQUIREMENTS.PROFILE_ANALYTICS).toBe( + 'consent:data_processing' + ) + expect(WORKFLOW_CONSENT_REQUIREMENTS.THIRD_PARTY_EXPORT).toBe( + 'consent:third_party_sharing' + ) + expect(WORKFLOW_CONSENT_REQUIREMENTS.NOTIFICATIONS_MARKETING).toBe( + 'consent:marketing' + ) + expect(WORKFLOW_CONSENT_REQUIREMENTS.AI_PORTFOLIO_ADVISORY).toBe( + 'consent:ai_advisory' + ) + expect(WORKFLOW_CONSENT_REQUIREMENTS.TELEMETRY_COLLECTION).toBe( + 'consent:telemetry' + ) }) it('approves workflow when explicit consent is present', () => { @@ -59,7 +71,9 @@ describe('User Consent and Data-Processing Policy Engine (#510)', () => { }) expect(result.allowed).toBe(false) - expect(result.reason).toContain("Required user consent 'consent:automated_trading' is missing or revoked") + expect(result.reason).toContain( + "Required user consent 'consent:automated_trading' is missing or revoked" + ) }) it('allows workflow with valid auditable bypass and justification', () => { @@ -68,7 +82,8 @@ describe('User Consent and Data-Processing Policy Engine (#510)', () => { workflow: 'AUTOMATED_TRADING', requiredScope: 'consent:automated_trading', bypassReason: 'EMERGENCY_CIRCUIT_BREAKER', - bypassJustification: 'Market volatility circuit breaker triggered by system risk engine', + bypassJustification: + 'Market volatility circuit breaker triggered by system risk engine', }) expect(result.allowed).toBe(true) @@ -86,6 +101,8 @@ describe('User Consent and Data-Processing Policy Engine (#510)', () => { }) expect(result.allowed).toBe(false) - expect(result.reason).toContain('Consent bypass rejected: justification must be at least 10 characters') + expect(result.reason).toContain( + 'Consent bypass rejected: justification must be at least 10 characters' + ) }) }) diff --git a/tests/unit/protectionFund/service.test.ts b/tests/unit/protectionFund/service.test.ts index 2f74411f..8e76efc2 100644 --- a/tests/unit/protectionFund/service.test.ts +++ b/tests/unit/protectionFund/service.test.ts @@ -86,7 +86,11 @@ describe('src/protectionFund/service', () => { }) it('returns the balance for a known asset', async () => { - mockFindUnique.mockResolvedValue({ assetSymbol: 'USDC', amount: '50000', status: 'ACTIVE' }) + mockFindUnique.mockResolvedValue({ + assetSymbol: 'USDC', + amount: '50000', + status: 'ACTIVE', + }) const balance = await getFundBalance('USDC') expect(balance.amount).toBe(50000) expect(balance.status).toBe('ACTIVE') @@ -145,7 +149,10 @@ describe('src/protectionFund/service', () => { describe('reviewCoverageEvent', () => { it('approves and computes claims', async () => { - mockFindUnique.mockResolvedValue({ id: 'event-1', status: 'PENDING_REVIEW' }) + mockFindUnique.mockResolvedValue({ + id: 'event-1', + status: 'PENDING_REVIEW', + }) mockUpdate.mockResolvedValue({ id: 'event-1', status: 'APPROVED' }) mockFindMany.mockResolvedValue([]) @@ -157,16 +164,28 @@ describe('src/protectionFund/service', () => { it('rejects an event that is not pending', async () => { mockFindUnique.mockResolvedValue({ id: 'event-1', status: 'APPROVED' }) - await expect(reviewCoverageEvent('event-1', true, 'admin-2')).rejects.toThrow('not pending review') + await expect( + reviewCoverageEvent('event-1', true, 'admin-2') + ).rejects.toThrow('not pending review') }) }) describe('getMyCoverage', () => { it('returns coverage info for a user', async () => { mockFindMany.mockResolvedValue([ - { id: 'pos-1', protocolName: 'Blend', assetSymbol: 'USDC', currentValue: '5000', createdAt: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000) }, + { + id: 'pos-1', + protocolName: 'Blend', + assetSymbol: 'USDC', + currentValue: '5000', + createdAt: new Date(Date.now() - 30 * 24 * 60 * 60 * 1000), + }, ]) - mockFindUnique.mockResolvedValue({ assetSymbol: 'USDC', amount: '50000', status: 'ACTIVE' }) + mockFindUnique.mockResolvedValue({ + assetSymbol: 'USDC', + amount: '50000', + status: 'ACTIVE', + }) const coverage = await getMyCoverage('user-1') @@ -178,10 +197,33 @@ describe('src/protectionFund/service', () => { describe('getProtectionFundStatus', () => { it('returns the public status', async () => { - mockFindUnique.mockResolvedValue({ assetSymbol: 'USDC', amount: '50000', status: 'ACTIVE' }) + mockFindUnique.mockResolvedValue({ + assetSymbol: 'USDC', + amount: '50000', + status: 'ACTIVE', + }) mockFindMany - .mockResolvedValueOnce([{ id: 'c1', source: 'REVENUE_SKIM', assetSymbol: 'USDC', amount: '1000', createdAt: new Date() }]) - .mockResolvedValueOnce([{ id: 'e1', protocolName: 'Blend', cause: 'EXPLOIT', status: 'APPROVED', totalPlatformExposure: '100000', lossWindowStart: '2026-09-01', lossWindowEnd: '2026-09-02', createdAt: new Date() }]) + .mockResolvedValueOnce([ + { + id: 'c1', + source: 'REVENUE_SKIM', + assetSymbol: 'USDC', + amount: '1000', + createdAt: new Date(), + }, + ]) + .mockResolvedValueOnce([ + { + id: 'e1', + protocolName: 'Blend', + cause: 'EXPLOIT', + status: 'APPROVED', + totalPlatformExposure: '100000', + lossWindowStart: '2026-09-01', + lossWindowEnd: '2026-09-02', + createdAt: new Date(), + }, + ]) const status = await getProtectionFundStatus() 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') + }) +}) diff --git a/tests/unit/routes/jobDashboard.test.ts b/tests/unit/routes/jobDashboard.test.ts index e8cfc805..67d18e5e 100644 --- a/tests/unit/routes/jobDashboard.test.ts +++ b/tests/unit/routes/jobDashboard.test.ts @@ -4,14 +4,18 @@ import fs from 'fs' import path from 'path' -import { getJobDashboardSnapshot, recordJobCompletion, recordJobRetry } from '../../../src/utils/job-metrics' +import { + getJobDashboardSnapshot, + recordJobCompletion, + recordJobRetry, +} from '../../../src/utils/job-metrics' const ROOT = path.resolve(__dirname, '../../..') describe('Operational Job Dashboard & Telemetry (#509)', () => { it('records job retries and completion status in telemetry', async () => { const jobName = 'test_retention_sweep' - + recordJobRetry(jobName) recordJobCompletion(jobName, 'success') @@ -36,7 +40,10 @@ describe('Operational Job Dashboard & Telemetry (#509)', () => { }) it('indexes JOB_TRIAGE_RUNBOOK.md in DOCUMENTATION_INDEX.md', () => { - const indexContent = fs.readFileSync(path.join(ROOT, 'docs', 'DOCUMENTATION_INDEX.md'), 'utf8') + const indexContent = fs.readFileSync( + path.join(ROOT, 'docs', 'DOCUMENTATION_INDEX.md'), + 'utf8' + ) expect(indexContent).toContain('JOB_TRIAGE_RUNBOOK.md') }) }) diff --git a/tests/unit/scripts/disasterRecoveryRehearsal.test.ts b/tests/unit/scripts/disasterRecoveryRehearsal.test.ts index 4bf364df..b642af87 100644 --- a/tests/unit/scripts/disasterRecoveryRehearsal.test.ts +++ b/tests/unit/scripts/disasterRecoveryRehearsal.test.ts @@ -42,7 +42,10 @@ describe('Disaster Recovery Rehearsal & Backup Validation (#511)', () => { }) it('is indexed in DOCUMENTATION_INDEX.md', () => { - const docIndex = fs.readFileSync(path.join(ROOT, 'docs', 'DOCUMENTATION_INDEX.md'), 'utf8') + const docIndex = fs.readFileSync( + path.join(ROOT, 'docs', 'DOCUMENTATION_INDEX.md'), + 'utf8' + ) expect(docIndex).toContain('DISASTER_RECOVERY_REHEARSAL.md') }) }) diff --git a/tests/unit/stellar/driftCheck.test.ts b/tests/unit/stellar/driftCheck.test.ts index 50d8026c..2dfc8f85 100644 --- a/tests/unit/stellar/driftCheck.test.ts +++ b/tests/unit/stellar/driftCheck.test.ts @@ -8,7 +8,9 @@ import { checkStellarIntegrationDrift } from '../../../src/stellar/driftCheck' describe('Stellar Contract & Schema Drift Detection Engine (#508)', () => { const validContractId = StrKey.encodeContract(Buffer.alloc(32, 1)) const validUsdcTokenAddress = StrKey.encodeContract(Buffer.alloc(32, 2)) - const validAgentSecretKey = StrKey.encodeEd25519SecretSeed(Buffer.alloc(32, 3)) + const validAgentSecretKey = StrKey.encodeEd25519SecretSeed( + Buffer.alloc(32, 3) + ) it('passes validation when network, contract IDs, and RPC URLs are aligned', () => { const report = checkStellarIntegrationDrift({ diff --git a/tests/unit/utils/fxConvert.test.ts b/tests/unit/utils/fxConvert.test.ts index 3df9f0c2..400bee62 100644 --- a/tests/unit/utils/fxConvert.test.ts +++ b/tests/unit/utils/fxConvert.test.ts @@ -75,7 +75,10 @@ describe('src/utils/fxConvert', () => { }) it('preserves all original fields', () => { - const result = withDisplayCurrency({ amountUsd: 100, name: 'test', id: '1' }, 'GBP') + const result = withDisplayCurrency( + { amountUsd: 100, name: 'test', id: '1' }, + 'GBP' + ) expect(result.name).toBe('test') expect(result.id).toBe('1') expect(result.amountUsd).toBe(100)