diff --git a/docs/architecture/cloud-integrations.md b/docs/architecture/cloud-integrations.md index 63b86b570..43cbcb95f 100644 --- a/docs/architecture/cloud-integrations.md +++ b/docs/architecture/cloud-integrations.md @@ -368,6 +368,20 @@ never re-parsed as tokens. All five automated rules ship disabled (`enabled: fal enforce the blocklist, a per-recipient hourly cap (`RATE_LIMIT_MAX_PER_HOUR`), a 10-minute duplicate-body filter, simulated typing delay, and a content filter that rejects non-HTTPS URLs and URL shorteners. +**ACK timeout, late ACKs, and retry.** Outbound messages remain in `typing` (refreshing `typing_at` when +`socket.sendMessage` completes) until a Baileys `messages.update` server ACK (`status = 2/3/4`) arrives. +`sweepAckTimeouts()` runs every 60 seconds (`ACK_SWEEP_INTERVAL_MS`) while connected, on socket open, +during scheduled automation sweeps, and on startup (`initFromDb()`) to mark non-in-flight outbound rows +still in `queued`, `seen`, or `typing` past `ACK_TIMEOUT_MS` (5 minutes, or any unfinished pre-restart or +unsubmitted rows on startup) as `failed` with `error = 'ack_timeout'`. If a late ACK arrives after a row +was marked `failed`, `advanceStatus()` promotes the row to `sent`, `delivered`, or `read` by +`external_message_id` (including prior retry-attempt IDs), clears `error` and `failed_at`, backfills +earlier timestamps via `COALESCE`, and ignores out-of-order lower-rank updates. +`POST /api/whatsapp/messages/:messageId/retry` (`retryMessage()`) coalesces concurrent retry requests and +atomically re-queues a `failed` row in place without creating duplicate rows, enforcing Offline mode +(`offline_mode`) and all safety controls while excluding the retried row from self-blocking in rate, +weekly-cap, and duplicate-body checks. + **Offline behaviour.** No work is started when the module is disabled, when Offline mode is on, when it is in a terminal shutdown state, or when the request signal is aborted. Work already in flight is cancelled when Offline mode turns on or during shutdown rather than left to fail late. diff --git a/docs/reference/api.md b/docs/reference/api.md index ede674169..7366a2ffd 100644 --- a/docs/reference/api.md +++ b/docs/reference/api.md @@ -766,6 +766,7 @@ Router: `main/routes/whatsapp.ts`. Full path: `/api/whatsapp`. | `POST` | `/automation/broadcast` | `ROLE_ACCESS.ownerManager` | body: `template_id`, `min_visits`, `inactive_days`, `min_points`, `limit` | `{ ok: true, matched, attempted, sent, skipped, errors }` for up to `limit` (1-50, default 25) matching active customers; `503` when disconnected. | | `POST` | `/automation/run-sweep` | `ROLE_ACCESS.ownerManager` | none | `{ ok: true, feedbackSent, winbackSent }` from an immediate scheduled-rule sweep. | | `GET` | `/messages` | `whatsapp.use` | query: `?limit`, `?offset`, `?direction`, `?status`, `?phone`, `?bill_id` | Message history with `trigger_rule` attribution; entries with `kind: "z_report"` are omitted unless the caller also has `reports.view`. This permission filter is applied before pagination. | +| `POST` | `/messages/:messageId/retry` | `whatsapp.use` | path: `messageId` | `{ ok: true, messageId }` after retrying a `failed` outbound message in place (`403` unless the caller also has `reports.view` when `kind: "z_report"`); `503` when disconnected, in cooldown, or in Offline mode; `404` when not found; `400` when invalid, not `failed`, or blocked by safety checks. | | `GET` | `/inbox` | `ROLE_ACCESS.ownerManager` | query: `?limit`, `?offset` | - | | `POST` | `/inbox/:messageId/reply` | `ROLE_ACCESS.ownerManager` | path: `messageId`; body: `body` | Replies to an inbound message. | | `GET` | `/blocklist` | `ROLE_ACCESS.ownerManager` | none | - | diff --git a/frontend/src/app/(dashboard)/whatsapp/page.tsx b/frontend/src/app/(dashboard)/whatsapp/page.tsx index 2f3a430ad..9d3ac8432 100644 --- a/frontend/src/app/(dashboard)/whatsapp/page.tsx +++ b/frontend/src/app/(dashboard)/whatsapp/page.tsx @@ -65,6 +65,7 @@ interface WhatsAppStatus { lastError: string | null; lastErrorReason: string | null; cooldownUntil: string | null; + offlineMode?: boolean; } type WhatsAppApiErrorKey = keyof AppConfig['Messages']['whatsapp']['apiError']; @@ -264,6 +265,9 @@ export default function WhatsAppPage() { const tAutomation = useTranslations('whatsapp.automation'); const tLastError = useTranslations('whatsapp.lastError'); const tApiError = useTranslations('whatsapp.apiError'); + const tSend = useTranslations('whatsapp.send'); + const tStatus = useTranslations('whatsapp.status'); + const tSettings = useTranslations('settings'); const { confirm, ConfirmDialog } = useConfirm(); const setWhatsappEnabled = usePosSettingsStore((s) => s.setWhatsappEnabled); const { currentTenant } = useAuthStore(); @@ -288,6 +292,7 @@ export default function WhatsAppPage() { const [pairingPhone, setPairingPhone] = useState(dialCode); const [sentMessages, setSentMessages] = useState([]); + const [retryingIds, setRetryingIds] = useState>(() => new Set()); const [inbox, setInbox] = useState([]); const [blocklist, setBlocklist] = useState([]); const [blockPhone, setBlockPhone] = useState(''); @@ -409,9 +414,10 @@ export default function WhatsAppPage() { useEffect(() => { const id = setInterval(() => { void refreshStatus(); + void refreshSent(); }, 5000); return () => clearInterval(id); - }, [refreshStatus]); + }, [refreshStatus, refreshSent]); useEffect(() => { if (!isAdmin) return; @@ -489,6 +495,46 @@ export default function WhatsAppPage() { } }; + const retrySentMessage = async (messageId: number) => { + if (retryingIds.has(messageId)) return; + if (status?.offlineMode) { + toast.error(tSettings('offlineModeTitle')); + return; + } + setRetryingIds((prev) => { + const next = new Set(prev); + next.add(messageId); + return next; + }); + try { + await api.post(`/whatsapp/messages/${messageId}/retry`); + toast.success(tStatus('queued')); + await refreshSent(); + } catch (err) { + const reason = (err as { response?: { data?: { reason?: string } } })?.response?.data?.reason; + if (reason === 'offline_mode') { + toast.error(tSettings('offlineModeTitle')); + } else if (reason === 'not_connected' || reason === 'cooldown') { + toast.error(tSend('error.notConnected')); + } else if (reason === 'blocked') { + toast.error(tSend('error.blocked')); + } else if (reason === 'not_on_whatsapp') { + toast.error(tSend('error.notOnWhatsapp')); + } else if (reason === 'rate_limited' || reason === 'weekly_cap') { + toast.error(tSend('error.rateLimited')); + } else { + toastApiError(err, tSend('failed'), tApiError); + } + await refreshSent(); + } finally { + setRetryingIds((prev) => { + const next = new Set(prev); + next.delete(messageId); + return next; + }); + } + }; + const sendBroadcastNow = async () => { setBroadcasting(true); try { @@ -1048,9 +1094,30 @@ export default function WhatsAppPage() { -
+
{m.error &&
{m.error}
} + {m.status === 'failed' && ( + + )}
diff --git a/main/routes/whatsapp.ts b/main/routes/whatsapp.ts index 62220f40f..3db7e231d 100644 --- a/main/routes/whatsapp.ts +++ b/main/routes/whatsapp.ts @@ -2,6 +2,7 @@ import { Router, Request, Response } from 'express'; import { hasPermission, requirePermission } from '../services/authorization'; import { asyncHandler } from '../middleware/async-handler'; import { getDatabase, getSettingValue, upsertSettings } from '../db'; +import { isOfflineMode } from '../services/offline-mode'; import { getHttpRequestSignal, trackHttpRequestWork } from '../shutdown'; import * as whatsapp from '../services/whatsapp'; import * as QRCode from 'qrcode'; @@ -155,7 +156,12 @@ router.post( }), ); if (!result.ok) { - const status = result.reason === 'not_connected' || result.reason === 'cooldown' ? 503 : 400; + const status = + result.reason === 'not_connected' || + result.reason === 'cooldown' || + result.reason === 'offline_mode' + ? 503 + : 400; res.status(status).json({ error: result.error, reason: result.reason }); return; } @@ -163,6 +169,64 @@ router.post( }), ); +router.post( + '/messages/:messageId/retry', + requirePermission('whatsapp.use'), + asyncHandler(async (req, res) => { + if (isOfflineMode()) { + res.status(503).json({ + error: 'Offline mode is on. Turn it off in Settings to use WhatsApp.', + reason: 'offline_mode', + }); + return; + } + const rawId = Number(req.params.messageId); + if (!Number.isInteger(rawId) || rawId <= 0) { + res.status(400).json({ error: 'messageId must be a positive integer', reason: 'not_found' }); + return; + } + const userId = String((req as Request & { user?: { userId?: string } }).user?.userId || ''); + const db = getDatabase(); + const row = db + .prepare('SELECT kind FROM whatsapp_messages WHERE id = ? AND direction = ?') + .get(rawId, 'outbound') as { kind: string } | undefined; + if (!row) { + res.status(404).json({ error: 'Message not found.', reason: 'not_found' }); + return; + } + if (row.kind === 'z_report' && !hasPermission(userId, 'reports.view')) { + res.status(403).json({ + error: 'Insufficient permissions', + required: 'reports.view', + code: 'PERMISSION_DENIED', + }); + return; + } + const result = await trackHttpRequestWork( + req, + whatsapp.retryMessage(rawId, { + userId: userId || null, + signal: getHttpRequestSignal(req), + }), + ); + if (!result.ok) { + const status = + result.reason === 'not_found' + ? 404 + : result.reason === 'not_connected' || + result.reason === 'cooldown' || + result.reason === 'offline_mode' + ? 503 + : 400; + res + .status(status) + .json({ error: result.error, reason: result.reason, messageId: result.messageId }); + return; + } + res.json({ ok: true, messageId: result.messageId }); + }), +); + router.get('/automation', requirePermission('whatsapp.use'), (_req: Request, res: Response) => { res.json({ config: whatsapp.getAutomationConfig(), @@ -289,7 +353,12 @@ router.post( }), ); if (!result.ok) { - const status = result.reason === 'not_connected' || result.reason === 'cooldown' ? 503 : 400; + const status = + result.reason === 'not_connected' || + result.reason === 'cooldown' || + result.reason === 'offline_mode' + ? 503 + : 400; res.status(status).json({ error: result.error, reason: result.reason }); return; } diff --git a/main/services/whatsapp.ts b/main/services/whatsapp.ts index 9764ba6e2..99ee057d2 100644 --- a/main/services/whatsapp.ts +++ b/main/services/whatsapp.ts @@ -193,6 +193,7 @@ export interface WhatsAppStatus { cooldownUntil: string | null; /** Stable reason code for frontend i18n translation of connection errors. */ lastErrorReason?: string | null; + offlineMode?: boolean; qr?: string; pairingCode?: string; } @@ -205,9 +206,20 @@ interface QueuedSend { kind: 'bill_receipt' | 'manual_reply' | 'auto_followup' | 'z_report'; userId: string | null; signal?: AbortSignal; + retryMessageId?: number | null; } const AUTOMATION_SWEEP_INTERVAL_MS = 5 * 60 * 1000; +export const ACK_TIMEOUT_MS = 5 * 60 * 1000; +export const ACK_SWEEP_INTERVAL_MS = 60 * 1000; +const STATUS_RANK: Record = { + queued: 0, + seen: 1, + typing: 2, + sent: 3, + delivered: 4, + read: 5, +}; const state: { enabled: boolean; @@ -223,8 +235,9 @@ const state: { cooldownTimer: NodeJS.Timeout | null; reconnectTimer: NodeJS.Timeout | null; automationTimer: NodeJS.Timeout | null; + ackSweepTimer: NodeJS.Timeout | null; lastSendByPhone: Map; - recentBodies: Map; + recentBodies: Map; sentMessageCache: Map; lidToPhoneMap: Map; } = { @@ -241,6 +254,7 @@ const state: { cooldownTimer: null, reconnectTimer: null, automationTimer: null, + ackSweepTimer: null, lastSendByPhone: new Map(), recentBodies: new Map(), sentMessageCache: new Map(), @@ -435,6 +449,7 @@ export function getStatus(): WhatsAppStatus { lastError: state.lastError, lastErrorReason: state.lastErrorReason, cooldownUntil: state.cooldownUntil ?? null, + offlineMode: isOfflineMode(), qr: state.lastQr ?? undefined, pairingCode: state.lastPairingCode ?? undefined, }; @@ -564,6 +579,7 @@ export function isCurrentlyQuietHours( export function isOverWeeklyAutomatedCap( phoneE164: string, maxPerCustomerPerWeek = getAutomationConfig().safety.maxPerCustomerPerWeek, + excludeMessageId?: number | null, ): boolean { if (!Number.isFinite(maxPerCustomerPerWeek) || maxPerCustomerPerWeek <= 0) return false; const db = getDatabase(); @@ -578,8 +594,11 @@ export function isOverWeeklyAutomatedCap( AND direction = 'outbound' AND (kind = 'auto_followup' OR created_by_user_id LIKE 'rule:%') AND queued_at >= ? + AND (? IS NULL OR id != ?) `) - .get(phoneE164, sevenDaysAgo) as { c: number }; + .get(phoneE164, sevenDaysAgo, excludeMessageId ?? null, excludeMessageId ?? null) as { + c: number; + }; return row.c >= maxPerCustomerPerWeek; } @@ -610,22 +629,40 @@ function hasShortenerOrNonHttps(body: string): string | null { return null; } -function isDuplicateBody(phoneE164: string, body: string): boolean { +function hasDuplicateBody( + phoneE164: string, + body: string, + excludeMessageId?: number | null, +): boolean { const cutoff = Date.now() - BODY_REPEAT_WINDOW_MS; const recent = state.recentBodies.get(phoneE164) ?? []; const fresh = recent.filter((r) => r.at >= cutoff); - for (const r of fresh) { - if (r.body === body) { - state.recentBodies.set(phoneE164, fresh); - return true; - } + if (fresh.length !== recent.length) { + if (fresh.length > 0) state.recentBodies.set(phoneE164, fresh); + else state.recentBodies.delete(phoneE164); } - fresh.push({ body, at: Date.now() }); + return fresh.some( + (r) => r.body === body && (excludeMessageId == null || r.messageId !== excludeMessageId), + ); +} + +function recordRecentBody(phoneE164: string, body: string, messageId: number): void { + const cutoff = Date.now() - BODY_REPEAT_WINDOW_MS; + const recent = state.recentBodies.get(phoneE164) ?? []; + const fresh = recent.filter((r) => r.at >= cutoff && r.messageId !== messageId); + fresh.push({ body, at: Date.now(), messageId }); // Bound the per-phone history to avoid unbounded growth in long-running installs. if (fresh.length > RECENT_BODIES_PER_PHONE_MAX) fresh.splice(0, fresh.length - RECENT_BODIES_PER_PHONE_MAX); state.recentBodies.set(phoneE164, fresh); - return false; +} + +function removeRecentBodyForMessage(phoneE164: string, messageId: number): void { + const recent = state.recentBodies.get(phoneE164); + if (!recent) return; + const filtered = recent.filter((r) => r.messageId !== messageId); + if (filtered.length > 0) state.recentBodies.set(phoneE164, filtered); + else state.recentBodies.delete(phoneE164); } function isBlocked(phoneE164: string): boolean { @@ -639,8 +676,9 @@ function isOverRateLimit( phoneE164: string, maxPerHour = RATE_LIMIT_MAX_PER_HOUR, skipMinGap = false, + excludeMessageId?: number | null, ): { limited: boolean; retryAfterMs?: number } { - if (!skipMinGap && process.env.FLO_WHATSAPP_FAST_DELAY !== '1') { + if (!skipMinGap && excludeMessageId == null && process.env.FLO_WHATSAPP_FAST_DELAY !== '1') { const last = state.lastSendByPhone.get(phoneE164); if (last) { const gap = Date.now() - last; @@ -660,8 +698,11 @@ function isOverRateLimit( .prepare(` SELECT COUNT(*) AS c FROM whatsapp_messages WHERE phone_e164 = ? AND direction = 'outbound' AND queued_at >= ? + AND (? IS NULL OR id != ?) `) - .get(phoneE164, oneHourAgo) as { c: number }; + .get(phoneE164, oneHourAgo, excludeMessageId ?? null, excludeMessageId ?? null) as { + c: number; + }; if (row.c >= maxPerHour) { return { limited: true }; } @@ -786,6 +827,14 @@ function updateMessageRow( /** Advances message status and backfills earlier timestamps using COALESCE. */ function advanceStatus(id: number, latest: 'sent' | 'delivered' | 'read'): void { + const db = getDatabase(); + const current = db.prepare('SELECT status FROM whatsapp_messages WHERE id = ?').get(id) as + | { status: string } + | undefined; + if (!current) return; + const currentRank = STATUS_RANK[current.status] ?? -1; + const targetRank = STATUS_RANK[latest] ?? -1; + const nextStatus = currentRank > targetRank ? current.status : latest; const ts = now(); const stamps = new Set([latest]); if (latest === 'read') { @@ -796,16 +845,122 @@ function advanceStatus(id: number, latest: 'sent' | 'delivered' | 'read'): void } const exprs = Array.from(stamps, (s) => `${s}_at = COALESCE(${s}_at, ?)`).join(', '); const placeholders = Array(stamps.size).fill(ts); - getDatabase() - .prepare(`UPDATE whatsapp_messages SET status = ?, ${exprs} WHERE id = ?`) - .run(latest, ...placeholders, id); + db.prepare( + `UPDATE whatsapp_messages SET status = ?, error = NULL, failed_at = NULL, ${exprs} WHERE id = ?`, + ).run(nextStatus, ...placeholders, id); +} + +const EXTERNAL_ID_MAP_MAX = 1000; +const externalIdToMessageId = new Map(); +const activeSendMessageIds = new Set(); +const inProcessSubmittedIds = new Set(); + +function rememberExternalMessageId(externalId: string | null | undefined, messageId: number): void { + if (!externalId) return; + externalIdToMessageId.delete(externalId); + externalIdToMessageId.set(externalId, messageId); + if (externalIdToMessageId.size > EXTERNAL_ID_MAP_MAX) { + const oldest = externalIdToMessageId.keys().next().value; + if (oldest !== undefined) externalIdToMessageId.delete(oldest); + } +} + +/** + * Sweeps outbound rows still in 'queued', 'seen', or 'typing' past ACK_TIMEOUT_MS + * (or unfinished outbound rows from before startup / interrupted before socket submission + * when includeAllUnfinished is true) and marks them 'failed' with error = 'ack_timeout'. + */ +export function sweepAckTimeouts( + nowMs: number = Date.now(), + opts?: { includeAllUnfinished?: boolean }, +): number { + if (whatsappTerminalCleanup || whatsappShutdownRequested) return 0; + try { + const db = getDatabase(); + const failedAt = new Date(nowMs).toISOString().replace('T', ' ').replace(/\..*$/, ''); + const cutoff = new Date(nowMs - ACK_TIMEOUT_MS) + .toISOString() + .replace('T', ' ') + .replace(/\..*$/, ''); + const candidates = ( + opts?.includeAllUnfinished + ? db + .prepare(` + SELECT id, external_message_id, + CASE WHEN datetime(COALESCE(typing_at, seen_at, queued_at)) <= datetime(?) THEN 1 ELSE 0 END AS expired + FROM whatsapp_messages + WHERE direction = 'outbound' + AND status IN ('queued', 'seen', 'typing') + `) + .all(cutoff) + : db + .prepare(` + SELECT id, external_message_id, 1 AS expired + FROM whatsapp_messages + WHERE direction = 'outbound' + AND status IN ('queued', 'seen', 'typing') + AND datetime(COALESCE(typing_at, seen_at, queued_at)) <= datetime(?) + `) + .all(cutoff) + ) as { id: number; external_message_id: string | null; expired: number }[]; + const staleRows = candidates.filter((row) => { + if (activeSendMessageIds.has(row.id)) return false; + if (row.expired === 1) return true; + if (!opts?.includeAllUnfinished) return false; + return !row.external_message_id || !inProcessSubmittedIds.has(row.id); + }); + if (staleRows.length === 0) return 0; + + const updateUnconditionalStmt = db.prepare(` + UPDATE whatsapp_messages + SET status = 'failed', error = 'ack_timeout', failed_at = ? + WHERE id = ? AND direction = 'outbound' AND status IN ('queued', 'seen', 'typing') + `); + const updateCutoffStmt = db.prepare(` + UPDATE whatsapp_messages + SET status = 'failed', error = 'ack_timeout', failed_at = ? + WHERE id = ? AND direction = 'outbound' AND status IN ('queued', 'seen', 'typing') + AND datetime(COALESCE(typing_at, seen_at, queued_at)) <= datetime(?) + `); + let count = 0; + const sweepTx = db.transaction(() => { + for (const row of staleRows) { + if (activeSendMessageIds.has(row.id)) continue; + const bypassCutoff = + opts?.includeAllUnfinished && + (!row.external_message_id || !inProcessSubmittedIds.has(row.id)); + const result = bypassCutoff + ? updateUnconditionalStmt.run(failedAt, row.id) + : updateCutoffStmt.run(failedAt, row.id, cutoff); + if (result.changes > 0) { + count += result.changes; + } + } + }); + sweepTx(); + return count; + } catch (err) { + logWhatsApp('warn', 'ack_sweep_failed', { error: sanitizeLogText(err) }); + return 0; + } } function findMessageByExternalId(externalId: string): { id: number; phone_e164: string } | null { - const row = getDatabase() + const db = getDatabase(); + const row = db .prepare('SELECT id, phone_e164 FROM whatsapp_messages WHERE external_message_id = ?') .get(externalId) as { id: number; phone_e164: string } | undefined; - return row ?? null; + if (row) return row; + const mappedId = externalIdToMessageId.get(externalId); + if (mappedId !== undefined) { + const mappedRow = db + .prepare( + "SELECT id, phone_e164 FROM whatsapp_messages WHERE id = ? AND direction = 'outbound'", + ) + .get(mappedId) as { id: number; phone_e164: string } | undefined; + if (mappedRow) return mappedRow; + } + return null; } export function processInboundOptOut(phoneE164: string, body: string): boolean { @@ -900,7 +1055,9 @@ function attachSocketHandlers(socket: BaileysSocket): void { state.lastPairingCode = null; state.lastError = null; state.lastErrorReason = null; + sweepAckTimeouts(); ensureAutomationTimer(); + ensureAckSweepTimer(); const user = (socket as any).user; if (user?.id) { const phone = '+' + userFromJid(user.id); @@ -1230,6 +1387,25 @@ function startSocket(requestSignal?: AbortSignal): Promise { return sharedPromise; } +function clearAckSweepTimer(): void { + if (state.ackSweepTimer) { + clearInterval(state.ackSweepTimer); + state.ackSweepTimer = null; + } +} + +function ensureAckSweepTimer(): void { + clearAckSweepTimer(); + if (!state.enabled || isWhatsAppTerminal() || process.env.FLO_WHATSAPP_SOCKET_DISABLED === '1') { + return; + } + state.ackSweepTimer = setInterval(() => { + if (!state.enabled || isWhatsAppTerminal() || state.state !== 'connected') return; + sweepAckTimeouts(); + }, ACK_SWEEP_INTERVAL_MS); + state.ackSweepTimer.unref?.(); +} + function ensureAutomationTimer(): void { if (state.automationTimer) { clearInterval(state.automationTimer); @@ -1260,6 +1436,7 @@ export async function enable(userId: string): Promise<{ ok: boolean; error?: str state.lastError = null; state.lastErrorReason = null; ensureAutomationTimer(); + ensureAckSweepTimer(); // Restore session from existing credentials if present to avoid unneeded re-pairing. const credsPath = path.join(getAuthDir(), 'creds.json'); if (fs.existsSync(credsPath)) { @@ -1292,6 +1469,11 @@ export function disable(): void { clearInterval(state.automationTimer); state.automationTimer = null; } + clearAckSweepTimer(); + inFlightRetries.clear(); + activeSendMessageIds.clear(); + inProcessSubmittedIds.clear(); + externalIdToMessageId.clear(); sweepRecipientCooldownUntil.clear(); inFlightNthVisitKeys.clear(); state.state = 'disconnected'; @@ -1324,6 +1506,7 @@ export async function connectWithQr( state.lastQr = null; state.lastPairingCode = null; ensureAutomationTimer(); + ensureAckSweepTimer(); try { await abortable(() => startSocket(signal), signal); } catch (err) { @@ -1357,6 +1540,7 @@ export async function connectWithPairingCode( : whatsappAbortController.signal; if (signal.aborted) return { ok: false, error: 'WhatsApp request cancelled.' }; ensureAutomationTimer(); + ensureAckSweepTimer(); try { if (!state.socket) { await abortable(() => startSocket(signal), signal); @@ -1383,6 +1567,7 @@ export function suspendForOfflineMode(): void { clearTimeout(state.reconnectTimer); state.reconnectTimer = null; } + clearAckSweepTimer(); if (state.socket) { try { state.socket.end(undefined); @@ -1421,6 +1606,11 @@ export function disconnect(): void { clearInterval(state.automationTimer); state.automationTimer = null; } + clearAckSweepTimer(); + inFlightRetries.clear(); + activeSendMessageIds.clear(); + inProcessSubmittedIds.clear(); + externalIdToMessageId.clear(); sweepRecipientCooldownUntil.clear(); inFlightNthVisitKeys.clear(); state.state = 'disconnected'; @@ -1439,6 +1629,7 @@ export interface SendResult { error?: string; reason?: | 'feature_off' + | 'offline_mode' | 'no_phone' | 'blocked' | 'quiet_hours' @@ -1448,22 +1639,131 @@ export interface SendResult { | 'not_connected' | 'not_on_whatsapp' | 'content_blocked' + | 'not_found' + | 'not_failed' | 'send_failed'; } const sendLocks = new Map>(); +const inFlightRetries = new Map>(); // Serializes outgoing sends per recipient to prevent race conditions during rate-limit checks. export function sendMessage(req: QueuedSend): Promise { return trackWhatsAppWork(sendMessageWithLock(req)); } +export function retryMessage( + messageId: number, + opts?: { userId?: string | null; signal?: AbortSignal }, +): Promise { + if (!Number.isInteger(messageId) || messageId <= 0) { + return Promise.resolve({ ok: false, error: 'Invalid message ID.', reason: 'not_found' }); + } + if (isOfflineMode()) { + return Promise.resolve({ + ok: false, + messageId, + error: 'Offline mode is on.', + reason: 'offline_mode', + }); + } + if (!state.enabled || isWhatsAppTerminal()) { + return Promise.resolve({ + ok: false, + messageId, + error: 'WhatsApp is not enabled.', + reason: 'feature_off', + }); + } + const existingInFlight = inFlightRetries.get(messageId); + if (existingInFlight) { + return existingInFlight; + } + if (activeSendMessageIds.has(messageId)) { + return Promise.resolve({ + ok: false, + messageId, + error: 'Only failed messages can be retried.', + reason: 'not_failed', + }); + } + const db = getDatabase(); + const row = db + .prepare(` + SELECT id, bill_id, customer_id, phone_e164, direction, kind, status, body, external_message_id, created_by_user_id + FROM whatsapp_messages + WHERE id = ? AND direction = 'outbound' + `) + .get(messageId) as + | { + id: number; + bill_id: number | null; + customer_id: string | number | null; + phone_e164: string; + direction: 'outbound'; + kind: QueuedSend['kind']; + status: string; + body: string; + external_message_id: string | null; + created_by_user_id: string | null; + } + | undefined; + if (!row) { + return Promise.resolve({ + ok: false, + error: 'Message not found.', + reason: 'not_found', + }); + } + if (row.status !== 'failed') { + return Promise.resolve({ + ok: false, + messageId: row.id, + error: 'Only failed messages can be retried.', + reason: 'not_failed', + }); + } + if (row.external_message_id) { + rememberExternalMessageId(row.external_message_id, row.id); + } + const preservedUserId = + typeof row.created_by_user_id === 'string' && row.created_by_user_id.startsWith('rule:') + ? row.created_by_user_id + : (opts?.userId ?? row.created_by_user_id ?? null); + const retryPromise = trackWhatsAppWork( + sendMessageWithLock({ + phoneE164: row.phone_e164, + body: row.body, + billId: row.bill_id, + customerId: row.customer_id, + kind: row.kind, + userId: preservedUserId, + signal: opts?.signal, + retryMessageId: row.id, + }), + ).finally(() => { + if (inFlightRetries.get(messageId) === retryPromise) { + inFlightRetries.delete(messageId); + } + }); + inFlightRetries.set(messageId, retryPromise); + return retryPromise; +} + async function sendMessageWithLock(req: QueuedSend): Promise { const startedAt = Date.now(); logWhatsApp('info', 'send_requested', { phone: maskPhone(req.phoneE164) }); + if (isOfflineMode()) { + return { + ok: false, + ...(req.retryMessageId != null ? { messageId: req.retryMessageId } : {}), + error: 'Offline mode is on.', + reason: 'offline_mode', + }; + } const signal = req.signal - ? AbortSignal.any([whatsappAbortController.signal, req.signal]) - : whatsappAbortController.signal; + ? AbortSignal.any([whatsappAbortController.signal, getOfflineModeAbortSignal(), req.signal]) + : AbortSignal.any([whatsappAbortController.signal, getOfflineModeAbortSignal()]); const previous = sendLocks.get(req.phoneE164) ?? Promise.resolve(); let release!: () => void; const current = new Promise((resolve) => { @@ -1472,6 +1772,14 @@ async function sendMessageWithLock(req: QueuedSend): Promise { sendLocks.set(req.phoneE164, current); try { await abortable(() => previous, signal); + if (isOfflineMode() && !isWhatsAppTerminal()) { + return { + ok: false, + ...(req.retryMessageId != null ? { messageId: req.retryMessageId } : {}), + error: 'Offline mode is on.', + reason: 'offline_mode', + }; + } if (isWhatsAppTerminal()) { const result = { ok: false, @@ -1495,6 +1803,14 @@ async function sendMessageWithLock(req: QueuedSend): Promise { }); return result; } catch (error) { + if (isOfflineMode() && !isWhatsAppTerminal()) { + return { + ok: false, + ...(req.retryMessageId != null ? { messageId: req.retryMessageId } : {}), + error: 'Offline mode is on.', + reason: 'offline_mode', + }; + } if (isWhatsAppTerminal() || signal.aborted) { const result = { ok: false, @@ -1524,11 +1840,53 @@ async function sendMessageWithLock(req: QueuedSend): Promise { } async function sendMessageInternal(req: QueuedSend, signal: AbortSignal): Promise { + const retryIdPart = req.retryMessageId != null ? { messageId: req.retryMessageId } : {}; + if (isOfflineMode()) { + return { ok: false, ...retryIdPart, error: 'Offline mode is on.', reason: 'offline_mode' }; + } if (!state.enabled || isWhatsAppTerminal()) - return { ok: false, error: 'WhatsApp is not enabled.', reason: 'feature_off' }; - if (!req.phoneE164) return { ok: false, error: 'Phone number required.', reason: 'no_phone' }; + return { ok: false, ...retryIdPart, error: 'WhatsApp is not enabled.', reason: 'feature_off' }; + if (!req.phoneE164) + return { ok: false, ...retryIdPart, error: 'Phone number required.', reason: 'no_phone' }; + const db = getDatabase(); + if (req.retryMessageId != null) { + if (activeSendMessageIds.has(req.retryMessageId)) { + return { + ok: false, + messageId: req.retryMessageId, + error: 'Only failed messages can be retried.', + reason: 'not_failed', + }; + } + const retryTarget = db + .prepare( + 'SELECT status, external_message_id FROM whatsapp_messages WHERE id = ? AND direction = ?', + ) + .get(req.retryMessageId, 'outbound') as + | { status: string; external_message_id: string | null } + | undefined; + if (!retryTarget) { + return { ok: false, error: 'Message not found.', reason: 'not_found' }; + } + if (retryTarget.status !== 'failed') { + return { + ok: false, + messageId: req.retryMessageId, + error: 'Only failed messages can be retried.', + reason: 'not_failed', + }; + } + if (retryTarget.external_message_id) { + rememberExternalMessageId(retryTarget.external_message_id, req.retryMessageId); + } + } if (isBlocked(req.phoneE164)) { - return { ok: false, error: 'This number asked to stop receiving messages.', reason: 'blocked' }; + return { + ok: false, + ...retryIdPart, + error: 'This number asked to stop receiving messages.', + reason: 'blocked', + }; } const automationConfig = getAutomationConfig(); const automated = isAutomatedSend(req); @@ -1539,55 +1897,92 @@ async function sendMessageInternal(req: QueuedSend, signal: AbortSignal): Promis ) { return { ok: false, + ...retryIdPart, error: 'Automated messages are paused during quiet hours.', reason: 'quiet_hours', }; } - if (isOverWeeklyAutomatedCap(req.phoneE164, automationConfig.safety.maxPerCustomerPerWeek)) { + if ( + isOverWeeklyAutomatedCap( + req.phoneE164, + automationConfig.safety.maxPerCustomerPerWeek, + req.retryMessageId, + ) + ) { return { ok: false, + ...retryIdPart, error: 'Weekly automated message limit reached for this customer.', reason: 'weekly_cap', }; } } if (state.state !== 'connected' || !state.socket) { - return { ok: false, error: 'Flo is not connected to WhatsApp.', reason: 'not_connected' }; + return { + ok: false, + ...retryIdPart, + error: 'Flo is not connected to WhatsApp.', + reason: 'not_connected', + }; } const socket = state.socket; const jid = await resolveJid(req.phoneE164, socket, signal); + if (isOfflineMode() && !isWhatsAppTerminal()) { + return { ok: false, ...retryIdPart, error: 'Offline mode is on.', reason: 'offline_mode' }; + } if (isWhatsAppTerminal()) - return { ok: false, error: 'WhatsApp is shutting down.', reason: 'send_failed' }; + return { + ok: false, + ...retryIdPart, + error: 'WhatsApp is shutting down.', + reason: 'send_failed', + }; if (isBlocked(req.phoneE164)) { - return { ok: false, error: 'This number asked to stop receiving messages.', reason: 'blocked' }; + return { + ok: false, + ...retryIdPart, + error: 'This number asked to stop receiving messages.', + reason: 'blocked', + }; } if (!jid) { return { ok: false, + ...retryIdPart, error: 'This phone is not registered on WhatsApp.', reason: 'not_on_whatsapp', }; } if (isInCooldown()) { - return { ok: false, error: 'Send is temporarily paused.', reason: 'cooldown' }; + return { ok: false, ...retryIdPart, error: 'Send is temporarily paused.', reason: 'cooldown' }; } - const rate = isOverRateLimit(req.phoneE164, RATE_LIMIT_MAX_PER_HOUR, automated); + const rate = isOverRateLimit( + req.phoneE164, + RATE_LIMIT_MAX_PER_HOUR, + automated, + req.retryMessageId, + ); if (rate.limited) { - return { ok: false, error: 'Rate limit reached for this number.', reason: 'rate_limited' }; + return { + ok: false, + ...retryIdPart, + error: 'Rate limit reached for this number.', + reason: 'rate_limited', + }; } const contentErr = hasShortenerOrNonHttps(req.body); if (contentErr) { - return { ok: false, error: contentErr, reason: 'content_blocked' }; + return { ok: false, ...retryIdPart, error: contentErr, reason: 'content_blocked' }; } - if (isDuplicateBody(req.phoneE164, req.body)) { + if (hasDuplicateBody(req.phoneE164, req.body, req.retryMessageId)) { return { ok: false, + ...retryIdPart, error: 'Identical message sent to this number recently.', reason: 'content_blocked', }; } - const db = getDatabase(); const bill = req.billId ? (db .prepare(` @@ -1599,7 +1994,7 @@ async function sendMessageInternal(req: QueuedSend, signal: AbortSignal): Promis .get(req.billId) as any) : null; if (bill && bill.payment_status !== 'paid') { - return { ok: false, error: 'Bill is not paid.', reason: 'send_failed' }; + return { ok: false, ...retryIdPart, error: 'Bill is not paid.', reason: 'send_failed' }; } let resolvedKind = req.kind; let resolvedCustomerId = req.customerId; @@ -1623,42 +2018,110 @@ async function sendMessageInternal(req: QueuedSend, signal: AbortSignal): Promis ) { return { ok: false, + ...retryIdPart, error: 'Visit milestone message already sent for this order.', reason: 'content_blocked', }; } let messageId: number; - try { - messageId = recordMessageRow({ - phone_e164: req.phoneE164, - direction: 'outbound', - kind: resolvedKind, - status: 'queued', - body: req.body, - bill_id: req.billId, - customer_id: resolvedCustomerId, - created_by_user_id: req.userId, - }); - } catch (err: any) { - return { ok: false, error: err.message ?? 'Failed to record message.', reason: 'send_failed' }; + if (req.retryMessageId != null) { + const priorRow = db + .prepare('SELECT external_message_id FROM whatsapp_messages WHERE id = ? AND direction = ?') + .get(req.retryMessageId, 'outbound') as { external_message_id: string | null } | undefined; + if (priorRow?.external_message_id) { + rememberExternalMessageId(priorRow.external_message_id, req.retryMessageId); + } + const queuedAt = now(); + const updated = db + .prepare(` + UPDATE whatsapp_messages + SET status = 'queued', + queued_at = ?, + seen_at = NULL, + typing_at = NULL, + sent_at = NULL, + delivered_at = NULL, + read_at = NULL, + failed_at = NULL, + error = NULL, + kind = ?, + customer_id = COALESCE(?, customer_id), + created_by_user_id = COALESCE(?, created_by_user_id) + WHERE id = ? AND direction = 'outbound' AND status = 'failed' + `) + .run( + queuedAt, + resolvedKind, + resolvedCustomerId !== undefined && resolvedCustomerId !== null + ? String(resolvedCustomerId) + : null, + req.userId ?? null, + req.retryMessageId, + ); + if (updated.changes === 0) { + return { + ok: false, + messageId: req.retryMessageId, + error: 'Only failed messages can be retried.', + reason: 'not_failed', + }; + } + messageId = req.retryMessageId; + } else { + try { + messageId = recordMessageRow({ + phone_e164: req.phoneE164, + direction: 'outbound', + kind: resolvedKind, + status: 'queued', + body: req.body, + bill_id: req.billId, + customer_id: resolvedCustomerId, + created_by_user_id: req.userId, + }); + } catch (err: any) { + return { + ok: false, + error: err.message ?? 'Failed to record message.', + reason: 'send_failed', + }; + } } + activeSendMessageIds.add(messageId); + recordRecentBody(req.phoneE164, req.body, messageId); + let submittedToSocket = false; + + const isAlreadyDeliveredOrSent = (): boolean => { + if (req.retryMessageId == null) return false; + const cur = db.prepare('SELECT status FROM whatsapp_messages WHERE id = ?').get(messageId) as + | { status: string } + | undefined; + return !!cur && (STATUS_RANK[cur.status] ?? -1) >= STATUS_RANK.sent; + }; + const shutdownFailure = (): SendResult => { - if (!whatsappTerminalCleanup) { + if (!submittedToSocket) { + removeRecentBodyForMessage(req.phoneE164, messageId); + } + const offline = isOfflineMode() && !isWhatsAppTerminal(); + const errorMsg = offline ? 'Offline mode is on.' : 'WhatsApp is shutting down.'; + const reason = offline ? ('offline_mode' as const) : ('send_failed' as const); + if (!whatsappTerminalCleanup && !isAlreadyDeliveredOrSent()) { try { updateMessageRow(messageId, { status: 'failed', - error: 'WhatsApp is shutting down.', + error: errorMsg, timestamp_field: 'failed_at', }); } catch {} } - return { ok: false, messageId, error: 'WhatsApp is shutting down.', reason: 'send_failed' }; + return { ok: false, messageId, error: errorMsg, reason }; }; try { - if (resolvedKind === 'manual_reply') { + if (resolvedKind === 'manual_reply' && !isAlreadyDeliveredOrSent()) { try { updateMessageRow(messageId, { status: 'seen', timestamp_field: 'seen_at' }); } catch { @@ -1669,11 +2132,17 @@ async function sendMessageInternal(req: QueuedSend, signal: AbortSignal): Promis if (isWhatsAppTerminal() || signal.aborted) return shutdownFailure(); await abortable(() => socket.sendPresenceUpdate('composing', jid), signal).catch(() => {}); if (isWhatsAppTerminal() || signal.aborted) return shutdownFailure(); - updateMessageRow(messageId, { status: 'typing', timestamp_field: 'typing_at' }); + if (!isAlreadyDeliveredOrSent()) { + updateMessageRow(messageId, { status: 'typing', timestamp_field: 'typing_at' }); + } await abortableDelay(randomDelayMs(req.body), signal); await abortable(() => socket.sendPresenceUpdate('paused', jid), signal).catch(() => {}); if (isWhatsAppTerminal() || signal.aborted) return shutdownFailure(); + if (isAlreadyDeliveredOrSent()) { + return { ok: true, messageId }; + } if (isBlocked(req.phoneE164)) { + removeRecentBodyForMessage(req.phoneE164, messageId); updateMessageRow(messageId, { status: 'failed', error: 'This number asked to stop receiving messages.', @@ -1686,11 +2155,19 @@ async function sendMessageInternal(req: QueuedSend, signal: AbortSignal): Promis reason: 'blocked', }; } + submittedToSocket = true; const sent = await abortable(() => socket.sendMessage(jid, { text: req.body }), signal); if (isWhatsAppTerminal() || signal.aborted) return shutdownFailure(); + const sentExternalId = sent?.key?.id ?? null; + if (sentExternalId) { + rememberExternalMessageId(sentExternalId, messageId); + } + inProcessSubmittedIds.add(messageId); // Do not mark 'sent' immediately; wait for server ACK (status=2) in messages.update. + // Refresh typing_at on submission completion so ACK_TIMEOUT_MS starts from here. updateMessageRow(messageId, { - external_message_id: sent?.key?.id ?? null, + ...(sentExternalId !== null ? { external_message_id: sentExternalId } : {}), + timestamp_field: 'typing_at', }); // Cache sent message payload for session restart re-encryption requests. if (sent?.key?.id && sent?.message) { @@ -1708,11 +2185,13 @@ async function sendMessageInternal(req: QueuedSend, signal: AbortSignal): Promis return { ok: true, messageId }; } catch (err: any) { if (isWhatsAppTerminal() || signal.aborted) return shutdownFailure(); - updateMessageRow(messageId, { - status: 'failed', - error: err?.message ?? 'Send failed', - timestamp_field: 'failed_at', - }); + if (!isAlreadyDeliveredOrSent()) { + updateMessageRow(messageId, { + status: 'failed', + error: err?.message ?? 'Send failed', + timestamp_field: 'failed_at', + }); + } const statusCode = (err as any)?.output?.statusCode; if (typeof statusCode === 'number' && RATE_LIMITED_STATUS_CODES.has(statusCode)) { triggerCooldown( @@ -1722,6 +2201,8 @@ async function sendMessageInternal(req: QueuedSend, signal: AbortSignal): Promis ); } return { ok: false, messageId, error: err?.message ?? 'Send failed', reason: 'send_failed' }; + } finally { + activeSendMessageIds.delete(messageId); } } @@ -2393,6 +2874,7 @@ async function runScheduledAutomationSweepImpl( if (!state.enabled || isWhatsAppTerminal() || state.state !== 'connected' || !state.socket) { return { feedbackSent, winbackSent }; } + sweepAckTimeouts(nowMs); for (const [key, until] of sweepRecipientCooldownUntil) { if (until <= nowMs) sweepRecipientCooldownUntil.delete(key); } @@ -2781,12 +3263,20 @@ export function initFromDb(): void { .prepare("SELECT value FROM settings WHERE key = 'whatsapp_enabled'") .get() as { value: string | null } | undefined; state.enabled = v?.value === 'true'; + if (state.enabled && !whatsappShutdownRequested && !whatsappTerminalCleanup) { + state.shuttingDown = false; + if (whatsappAbortController.signal.aborted) whatsappAbortController = new AbortController(); + } logWhatsApp('info', 'startup_init', { enabled: state.enabled, hasCredentials: fs.existsSync(credsPath), connectedPhone: maskPhone(state.connectedPhone), }); + sweepAckTimeouts(Date.now(), { + includeAllUnfinished: !state.socket && sendLocks.size === 0 && activeSendMessageIds.size === 0, + }); ensureAutomationTimer(); + ensureAckSweepTimer(); if (state.enabled) { void startSocket().catch(() => {}); } @@ -2824,6 +3314,9 @@ export function requestShutdown(): void { clearInterval(state.automationTimer); state.automationTimer = null; } + clearAckSweepTimer(); + inFlightRetries.clear(); + activeSendMessageIds.clear(); const socket = state.socket; shutdownSocket = socket; if (socket) { diff --git a/tests/whatsapp-service.test.ts b/tests/whatsapp-service.test.ts index 77c81e406..26ec07ea8 100644 --- a/tests/whatsapp-service.test.ts +++ b/tests/whatsapp-service.test.ts @@ -1245,6 +1245,487 @@ async function main(): Promise { } } + // 5i. Issue #1344: ACK timeout sweep, startup sweep, late ACK promotion from failed, and idempotent retry + { + whatsapp.disconnect(); + await whatsapp.enable('ack-timeout-test'); + await whatsapp.connectWithQr(); + eventHandlers.get('connection.update')?.({ connection: 'open' }); + await new Promise((resolve) => setImmediate(resolve)); + + assert( + whatsapp.ACK_TIMEOUT_MS === 5 * 60 * 1000, + `ACK_TIMEOUT_MS is 5 minutes (got ${whatsapp.ACK_TIMEOUT_MS})`, + ); + + let extSeq = 0; + const prevSendMessage = fakeSocket.sendMessage; + fakeSocket.sendMessage = async () => ({ + key: { id: `ack-timeout-ext-${++extSeq}` }, + message: { conversation: 'test' }, + }); + + try { + // 1. Send a message that receives no server ACK -> stays 'typing' until ACK_TIMEOUT_MS + const sentNoAck = await whatsapp.sendMessage({ + phoneE164: '+12133734253', + body: 'Receipt #1344 awaiting ACK', + billId: null, + customerId: null, + kind: 'bill_receipt', + userId: 'user-1344', + }); + assert( + sentNoAck.ok === true && typeof sentNoAck.messageId === 'number', + `sendMessage succeeds before ACK (got ${JSON.stringify(sentNoAck)})`, + ); + const msgId1 = sentNoAck.messageId!; + const beforeSweep = db + .prepare( + 'SELECT status, error, external_message_id, typing_at, failed_at FROM whatsapp_messages WHERE id = ?', + ) + .get(msgId1) as { + status: string; + error: string | null; + external_message_id: string | null; + typing_at: string | null; + failed_at: string | null; + }; + assert( + beforeSweep.status === 'typing' && + beforeSweep.external_message_id === 'ack-timeout-ext-1' && + beforeSweep.error === null, + `message starts in typing waiting for ACK (got ${JSON.stringify(beforeSweep)})`, + ); + + // Sweep before timeout does not mark fresh typing row failed + const earlySwept = whatsapp.sweepAckTimeouts(Date.now() + 60 * 1000); + const afterEarlySweep = db + .prepare('SELECT status, error FROM whatsapp_messages WHERE id = ?') + .get(msgId1) as { status: string; error: string | null }; + assert( + earlySwept === 0 && afterEarlySweep.status === 'typing', + `fresh typing message is not swept before ACK_TIMEOUT_MS (got swept=${earlySwept}, status=${afterEarlySweep.status})`, + ); + + // Sweep after ACK_TIMEOUT_MS marks the row failed with error='ack_timeout' + const sweptCount = whatsapp.sweepAckTimeouts(Date.now() + whatsapp.ACK_TIMEOUT_MS + 5_000); + const afterTimeoutSweep = db + .prepare( + 'SELECT status, error, failed_at, external_message_id FROM whatsapp_messages WHERE id = ?', + ) + .get(msgId1) as { + status: string; + error: string | null; + failed_at: string | null; + external_message_id: string | null; + }; + assert( + sweptCount >= 1 && + afterTimeoutSweep.status === 'failed' && + afterTimeoutSweep.error === 'ack_timeout' && + afterTimeoutSweep.failed_at !== null && + afterTimeoutSweep.external_message_id === 'ack-timeout-ext-1', + `stale typing message becomes failed (ack_timeout) after timeout while preserving external_message_id (got ${JSON.stringify(afterTimeoutSweep)})`, + ); + + // 2. Late ACK for a message already marked failed (ack_timeout) promotes it to sent/delivered and clears error/failed_at + eventHandlers.get('messages.update')?.([ + { key: { id: 'ack-timeout-ext-1' }, update: { status: 2 } }, + ]); + await new Promise((resolve) => setImmediate(resolve)); + const afterLateSent = db + .prepare( + 'SELECT status, error, failed_at, sent_at, delivered_at FROM whatsapp_messages WHERE id = ?', + ) + .get(msgId1) as { + status: string; + error: string | null; + failed_at: string | null; + sent_at: string | null; + delivered_at: string | null; + }; + assert( + afterLateSent.status === 'sent' && + afterLateSent.error === null && + afterLateSent.failed_at === null && + afterLateSent.sent_at !== null, + `late SERVER_ACK (status=2) promotes failed (ack_timeout) message to sent and clears error/failed_at (got ${JSON.stringify(afterLateSent)})`, + ); + + // Mark failed again to test direct promotion from failed -> delivered (status=3) and out-of-order status=2 non-regression + db.prepare( + "UPDATE whatsapp_messages SET status = 'failed', error = 'ack_timeout', failed_at = datetime('now'), sent_at = NULL, delivered_at = NULL WHERE id = ?", + ).run(msgId1); + eventHandlers.get('messages.update')?.([ + { key: { id: 'ack-timeout-ext-1' }, update: { status: 3 } }, + ]); + await new Promise((resolve) => setImmediate(resolve)); + const afterLateDelivered = db + .prepare( + 'SELECT status, error, failed_at, sent_at, delivered_at FROM whatsapp_messages WHERE id = ?', + ) + .get(msgId1) as { + status: string; + error: string | null; + failed_at: string | null; + sent_at: string | null; + delivered_at: string | null; + }; + assert( + afterLateDelivered.status === 'delivered' && + afterLateDelivered.error === null && + afterLateDelivered.failed_at === null && + afterLateDelivered.sent_at !== null && + afterLateDelivered.delivered_at !== null, + `late delivered ACK (status=3) promotes failed message to delivered and backfills sent_at (got ${JSON.stringify(afterLateDelivered)})`, + ); + + eventHandlers.get('messages.update')?.([ + { key: { id: 'ack-timeout-ext-1' }, update: { status: 2 } }, + ]); + await new Promise((resolve) => setImmediate(resolve)); + const afterOutOfOrderSent = db + .prepare('SELECT status FROM whatsapp_messages WHERE id = ?') + .get(msgId1) as { status: string }; + assert( + afterOutOfOrderSent.status === 'delivered', + `out-of-order status=2 does not regress delivered back to sent (got ${afterOutOfOrderSent.status})`, + ); + + // 3. Startup sweep: rows stuck in queued/typing from before a restart are swept on initFromDb() + whatsapp.disconnect(); + const staleQueuedInsert = db + .prepare(` + INSERT INTO whatsapp_messages (phone_e164, direction, kind, status, body, queued_at) + VALUES ('+12133734253', 'outbound', 'bill_receipt', 'queued', 'Stuck queued before restart', datetime('now', '-10 minutes')) + `) + .run(); + const staleTypingInsert = db + .prepare(` + INSERT INTO whatsapp_messages (phone_e164, direction, kind, status, body, queued_at, typing_at) + VALUES ('+12133734253', 'outbound', 'manual_reply', 'typing', 'Stuck typing before restart', datetime('now', '-1 minutes'), datetime('now', '-1 minutes')) + `) + .run(); + const staleQueuedId = Number(staleQueuedInsert.lastInsertRowid); + const staleTypingId = Number(staleTypingInsert.lastInsertRowid); + + whatsapp.initFromDb(); + const sweptQueued = db + .prepare('SELECT status, error, failed_at FROM whatsapp_messages WHERE id = ?') + .get(staleQueuedId) as { status: string; error: string | null; failed_at: string | null }; + const sweptTyping = db + .prepare('SELECT status, error, failed_at FROM whatsapp_messages WHERE id = ?') + .get(staleTypingId) as { status: string; error: string | null; failed_at: string | null }; + assert( + sweptQueued.status === 'failed' && + sweptQueued.error === 'ack_timeout' && + sweptQueued.failed_at !== null, + `startup initFromDb sweeps pre-restart queued row to failed/ack_timeout (got ${JSON.stringify(sweptQueued)})`, + ); + assert( + sweptTyping.status === 'failed' && + sweptTyping.error === 'ack_timeout' && + sweptTyping.failed_at !== null, + `startup initFromDb sweeps pre-restart typing row to failed/ack_timeout (got ${JSON.stringify(sweptTyping)})`, + ); + + // 4. Retry: blocked in Offline mode, honours #1315 safety controls, and sends once in-place without duplicates + await whatsapp.enable('retry-test'); + await whatsapp.connectWithQr(); + eventHandlers.get('connection.update')?.({ connection: 'open' }); + await new Promise((resolve) => setImmediate(resolve)); + + // 4a. Retry is blocked in Offline mode + db.prepare( + "INSERT OR REPLACE INTO settings (key, value) VALUES ('offline_mode', 'true')", + ).run(); + const offlineRetry = await whatsapp.retryMessage(staleQueuedId); + const offlineSend = await whatsapp.sendMessage({ + phoneE164: '+12133734253', + body: 'Should not send in offline mode', + billId: null, + customerId: null, + kind: 'manual_reply', + userId: null, + }); + db.prepare( + "INSERT OR REPLACE INTO settings (key, value) VALUES ('offline_mode', 'false')", + ).run(); + assert( + offlineRetry.ok === false && offlineRetry.reason === 'offline_mode', + `retryMessage is blocked in Offline mode with reason=offline_mode (got ${JSON.stringify(offlineRetry)})`, + ); + assert( + offlineSend.ok === false && offlineSend.reason === 'offline_mode', + `sendMessage is blocked in Offline mode with reason=offline_mode (got ${JSON.stringify(offlineSend)})`, + ); + const rowStillFailedAfterOffline = db + .prepare('SELECT status, error FROM whatsapp_messages WHERE id = ?') + .get(staleQueuedId) as { status: string; error: string | null }; + assert( + rowStillFailedAfterOffline.status === 'failed' && + rowStillFailedAfterOffline.error === 'ack_timeout', + `blocked offline retry leaves failed row untouched (got ${JSON.stringify(rowStillFailedAfterOffline)})`, + ); + + // 4b. Retry honours #1315 safety controls (quiet hours for automated messages) + const autoFailedInsert = db + .prepare(` + INSERT INTO whatsapp_messages (phone_e164, direction, kind, status, error, body, created_by_user_id, queued_at, failed_at) + VALUES ('+12133734253', 'outbound', 'auto_followup', 'failed', 'ack_timeout', 'Automated follow-up retry', 'rule:winback', datetime('now', '-10 minutes'), datetime('now', '-5 minutes')) + `) + .run(); + const autoFailedId = Number(autoFailedInsert.lastInsertRowid); + const prevConfig = whatsapp.getAutomationConfig(); + whatsapp.saveAutomationConfig({ + ...prevConfig, + safety: { + ...prevConfig.safety, + quietHoursEnabled: true, + quietHoursStart: '00:00', + quietHoursEnd: '23:59', + }, + }); + const quietHoursRetry = await whatsapp.retryMessage(autoFailedId); + whatsapp.saveAutomationConfig(prevConfig); + assert( + quietHoursRetry.ok === false && quietHoursRetry.reason === 'quiet_hours', + `retryMessage honours #1315 quiet hours for automated messages (got ${JSON.stringify(quietHoursRetry)})`, + ); + + // 4c. Concurrent retry calls send once in place (no duplicate rows, single socket.sendMessage call) + const countBeforeRetry = ( + db.prepare('SELECT COUNT(*) AS c FROM whatsapp_messages').get() as { c: number } + ).c; + const extSeqBeforeRetry = extSeq; + const [retryRes1, retryRes2] = await Promise.all([ + whatsapp.retryMessage(staleQueuedId, { userId: 'staff-retry' }), + whatsapp.retryMessage(staleQueuedId, { userId: 'staff-retry' }), + ]); + const countAfterRetry = ( + db.prepare('SELECT COUNT(*) AS c FROM whatsapp_messages').get() as { c: number } + ).c; + const retriedRow = db + .prepare( + 'SELECT id, status, error, failed_at, external_message_id FROM whatsapp_messages WHERE id = ?', + ) + .get(staleQueuedId) as { + id: number; + status: string; + error: string | null; + failed_at: string | null; + external_message_id: string | null; + }; + assert( + retryRes1.ok === true && + retryRes2.ok === true && + retryRes1.messageId === staleQueuedId && + retryRes2.messageId === staleQueuedId, + `concurrent retryMessage calls resolve to the same existing messageId (got ${JSON.stringify([retryRes1, retryRes2])})`, + ); + assert( + countAfterRetry === countBeforeRetry && extSeq - extSeqBeforeRetry === 1, + `retryMessage updates row in-place without inserting duplicates and calls socket.sendMessage once (rows delta=${countAfterRetry - countBeforeRetry}, send calls=${extSeq - extSeqBeforeRetry})`, + ); + assert( + retriedRow.status === 'typing' && + retriedRow.error === null && + retriedRow.failed_at === null && + retriedRow.external_message_id === `ack-timeout-ext-${extSeq}`, + `retried row transitions to typing with cleared error/failed_at and new external_message_id (got ${JSON.stringify(retriedRow)})`, + ); + + // 4d. Sequential retry after row is already in typing/sent is rejected with not_failed + const secondSequentialRetry = await whatsapp.retryMessage(staleQueuedId); + assert( + secondSequentialRetry.ok === false && secondSequentialRetry.reason === 'not_failed', + `retrying a message that is no longer failed returns reason=not_failed (got ${JSON.stringify(secondSequentialRetry)})`, + ); + + // 4e. Prior attempt external_message_id still matches late ACK after retry overwrites external_message_id + const priorAttemptSend = await whatsapp.sendMessage({ + phoneE164: '+14155550191', + body: 'Prior attempt ACK test', + billId: null, + customerId: null, + kind: 'manual_reply', + userId: null, + }); + assert( + priorAttemptSend.ok === true && typeof priorAttemptSend.messageId === 'number', + 'prior attempt send succeeds', + ); + const priorMsgId = priorAttemptSend.messageId!; + const firstAttemptExtId = ( + db + .prepare('SELECT external_message_id FROM whatsapp_messages WHERE id = ?') + .get(priorMsgId) as { + external_message_id: string; + } + ).external_message_id; + whatsapp.sweepAckTimeouts(Date.now() + whatsapp.ACK_TIMEOUT_MS + 5_000); + // Non-retry duplicate send immediately after ACK timeout is still blocked by recentBodies + const nonRetryDupAfterTimeout = await whatsapp.sendMessage({ + phoneE164: '+14155550191', + body: 'Prior attempt ACK test', + billId: null, + customerId: null, + kind: 'manual_reply', + userId: null, + }); + assert( + nonRetryDupAfterTimeout.ok === false && + nonRetryDupAfterTimeout.reason === 'content_blocked', + `non-retry duplicate send after ACK timeout remains blocked by recentBodies (got ${JSON.stringify(nonRetryDupAfterTimeout)})`, + ); + // Self-retry is allowed even though body is in recentBodies for the same messageId + const secondAttemptRetry = await whatsapp.retryMessage(priorMsgId); + assert( + secondAttemptRetry.ok === true, + `self-retry succeeds (got ${JSON.stringify(secondAttemptRetry)})`, + ); + const secondAttemptExtId = ( + db + .prepare('SELECT external_message_id FROM whatsapp_messages WHERE id = ?') + .get(priorMsgId) as { + external_message_id: string; + } + ).external_message_id; + assert( + firstAttemptExtId !== secondAttemptExtId, + `retry produces a new external_message_id (${firstAttemptExtId} vs ${secondAttemptExtId})`, + ); + // Emit late ACK (status=3 delivered) referencing the FIRST attempt's external_message_id + const updateListener = eventHandlers.get('messages.update'); + updateListener?.([{ key: { id: firstAttemptExtId }, update: { status: 3 } }]); + const promotedFromFirstAttempt = db + .prepare( + 'SELECT status, error, failed_at, sent_at, delivered_at FROM whatsapp_messages WHERE id = ?', + ) + .get(priorMsgId) as { + status: string; + error: string | null; + failed_at: string | null; + sent_at: string | null; + delivered_at: string | null; + }; + assert( + promotedFromFirstAttempt.status === 'delivered' && + promotedFromFirstAttempt.error === null && + promotedFromFirstAttempt.sent_at !== null && + promotedFromFirstAttempt.delivered_at !== null, + `late ACK for prior attempt external_message_id promotes retried row to delivered (got ${JSON.stringify(promotedFromFirstAttempt)})`, + ); + + // 4f. Retry duplicate-body check blocks when a DIFFERENT message with the same body was recently sent + const otherFailedInsert = db + .prepare(` + INSERT INTO whatsapp_messages (phone_e164, direction, kind, status, error, body, queued_at, failed_at) + VALUES ('+14155550191', 'outbound', 'manual_reply', 'failed', 'ack_timeout', 'Prior attempt ACK test', datetime('now', '-20 minutes'), datetime('now', '-15 minutes')) + `) + .run(); + const otherFailedId = Number(otherFailedInsert.lastInsertRowid); + const dupBlockedRetry = await whatsapp.retryMessage(otherFailedId); + assert( + dupBlockedRetry.ok === false && dupBlockedRetry.reason === 'content_blocked', + `retry is blocked with content_blocked when a different message with the same body was recently sent (got ${JSON.stringify(dupBlockedRetry)})`, + ); + + // 4g. Rejected retry (unpaid bill) does not poison recentBodies for a subsequent valid send + const unpaidOrderInsert = db + .prepare( + "INSERT INTO orders (order_number, type, status, subtotal, tax_amount, discount_amount, total, created_at, updated_at) VALUES ('ORD-RETRY-UNPAID', 'dine_in', 'pending', 100, 0, 0, 100, datetime('now'), datetime('now'))", + ) + .run(); + const unpaidBillInsert = db + .prepare( + "INSERT INTO bills (order_id, bill_number, subtotal, tax_amount, discount_amount, total, paid_amount, balance, payment_status, created_at, updated_at) VALUES (?, 'BILL-RETRY-UNPAID', 100, 0, 0, 100, 0, 100, 'unpaid', datetime('now'), datetime('now'))", + ) + .run(unpaidOrderInsert.lastInsertRowid); + const unpaidBillId = Number(unpaidBillInsert.lastInsertRowid); + const unpaidFailedInsert = db + .prepare(` + INSERT INTO whatsapp_messages (bill_id, phone_e164, direction, kind, status, error, body, queued_at, failed_at) + VALUES (?, '+14155550192', 'outbound', 'bill_receipt', 'failed', 'ack_timeout', 'Unpaid bill receipt body', datetime('now', '-10 minutes'), datetime('now', '-5 minutes')) + `) + .run(unpaidBillId); + const unpaidFailedMsgId = Number(unpaidFailedInsert.lastInsertRowid); + const rejectedUnpaidRetry = await whatsapp.retryMessage(unpaidFailedMsgId); + assert( + rejectedUnpaidRetry.ok === false && rejectedUnpaidRetry.reason === 'send_failed', + `retry for unpaid bill is rejected with send_failed (got ${JSON.stringify(rejectedUnpaidRetry)})`, + ); + const validSendAfterRejectedRetry = await whatsapp.sendMessage({ + phoneE164: '+14155550192', + body: 'Unpaid bill receipt body', + billId: null, + customerId: null, + kind: 'manual_reply', + userId: null, + }); + assert( + validSendAfterRejectedRetry.ok === true, + `rejected retry does not poison recentBodies for subsequent valid send (got ${JSON.stringify(validSendAfterRejectedRetry)})`, + ); + + // 4h. Active in-flight send is excluded from sweepAckTimeouts while socket.sendMessage is pending + let releaseSlowSend!: () => void; + const slowSendGate = new Promise((resolve) => { + releaseSlowSend = resolve; + }); + let slowSendEntered!: () => void; + const slowSendEnteredPromise = new Promise((resolve) => { + slowSendEntered = resolve; + }); + fakeSocket.sendMessage = async () => { + slowSendEntered(); + await slowSendGate; + extSeq += 1; + return { key: { id: `ack-timeout-ext-${extSeq}` }, message: { conversation: 'slow' } }; + }; + const inFlightSendPromise = whatsapp.sendMessage({ + phoneE164: '+14155550193', + body: 'Slow in-flight send test', + billId: null, + customerId: null, + kind: 'manual_reply', + userId: null, + }); + await slowSendEnteredPromise; + const inFlightRowBeforeSweep = db + .prepare( + "SELECT id, status FROM whatsapp_messages WHERE phone_e164 = '+14155550193' ORDER BY id DESC LIMIT 1", + ) + .get() as { id: number; status: string }; + // Sweep far in the future while socket.sendMessage is still in flight + whatsapp.sweepAckTimeouts(Date.now() + whatsapp.ACK_TIMEOUT_MS + 60_000); + const inFlightRowDuringSend = db + .prepare('SELECT status, error FROM whatsapp_messages WHERE id = ?') + .get(inFlightRowBeforeSweep.id) as { status: string; error: string | null }; + assert( + inFlightRowDuringSend.status === 'typing' && inFlightRowDuringSend.error === null, + `active in-flight send is excluded from sweepAckTimeouts (got ${JSON.stringify(inFlightRowDuringSend)})`, + ); + releaseSlowSend(); + const finishedSlowSend = await inFlightSendPromise; + assert(finishedSlowSend.ok === true, 'slow in-flight send resolves ok=true'); + + // 4i. Runtime initFromDb() after suspendForOfflineMode() does not sweep a recently submitted typing row + whatsapp.suspendForOfflineMode(); + whatsapp.initFromDb(); + const afterRuntimeInit = db + .prepare('SELECT status, error FROM whatsapp_messages WHERE id = ?') + .get(inFlightRowBeforeSweep.id) as { status: string; error: string | null }; + assert( + afterRuntimeInit.status === 'typing' && afterRuntimeInit.error === null, + `runtime initFromDb after suspendForOfflineMode preserves recently submitted typing row (got ${JSON.stringify(afterRuntimeInit)})`, + ); + } finally { + fakeSocket.sendMessage = prevSendMessage; + } + } + whatsapp.disconnect(); await whatsapp.enable('broadcast-disconnected-check'); const disconnectedBroadcast = await whatsapp.runBroadcast({