From afadcbeb826ca6bcc7dca89c4adc8a44261b1ed5 Mon Sep 17 00:00:00 2001 From: Khilta Date: Mon, 10 Aug 2026 23:14:04 +0530 Subject: [PATCH 01/18] fix: consolidated inbox reliability, AI resilience, and Meta compliance Six changes per the spec in #23: 1. Direction fix: accept "outgoing" (Zernio's actual enum value) alongside legacy "outbound" in both the messages route mapper and the webhook guard. Without this, every message in every thread renders on the customer's side, including our own replies. 2. Local incoming DM storage + Zernio merge: store inbound DMs locally in the webhook using platformMessageId (not Zernio's internal msg.id) so the dedup key matches when the messages route merges local + API responses. Messages route now fetches both sources and deduplicates on platform_message_id. 3. Comment stored as inbound message: the triggering comment is now inserted into the messages table in comment-processor.ts so the inbox shows what the contact commented, not just the DM flow. 4. AI retry with backoff + user-facing fallback: transient errors (rate limits, timeouts, 5xx) are retried with exponential backoff. Permanent errors skip retries. On exhaustion, a configurable fallback message is sent to the contact instead of ghosting them with [AI response failed]. New fields: maxRetries, fallbackMessage. 5. {{contact_name}} personalization + 24h-window logging: engine seeds {{contact_name}} from the incoming message sender name. Webhook logs 24h messaging-window failures as message_failed_24h_window instead of collapsing them into generic message_failed. 6. STOP/START confirmation DMs: handleGlobalKeywords now sends a confirmation DM when a contact opts in or out. Refactored to use a single object parameter instead of 8 positional arguments. No-ops outside the 24h window and when there is no late_conversation_id. Build: tsc --noEmit passes (0 errors). npm run build passes. Branch based on current main (f7b056f, post-#28 merge). Closes #23 --- app/api/v1/messages/route.ts | 49 +++++- app/api/webhooks/late/route.ts | 161 ++++++++++++++++-- .../flow-builder/panels/AiResponsePanel.tsx | 43 +++++ lib/comment-processor.ts | 10 ++ lib/flow-engine/engine.ts | 9 + lib/flow-engine/nodes/ai-response.ts | 99 +++++++++-- lib/flow-engine/types.ts | 11 ++ 7 files changed, 347 insertions(+), 35 deletions(-) diff --git a/app/api/v1/messages/route.ts b/app/api/v1/messages/route.ts index 2f98e27..6636557 100644 --- a/app/api/v1/messages/route.ts +++ b/app/api/v1/messages/route.ts @@ -60,11 +60,13 @@ export async function GET(request: NextRequest) { (res.data as { data?: unknown[] })?.data ?? []; - // Map Zernio messages to the shape the inbox UI expects - const messages = zernioMessages.map((m: any) => ({ + // Map Zernio messages to the shape the inbox UI expects. + // Accept both "outgoing" (Zernio's actual enum value) and "outbound" + // (legacy spelling) so our own replies are correctly shown on our side. + const zernioMapped = zernioMessages.map((m: any) => ({ id: m.id, conversation_id: conversationId, - direction: m.direction === "outbound" ? "outbound" : "inbound", + direction: m.direction === "outgoing" || m.direction === "outbound" ? "outbound" : "inbound", text: m.text ?? m.message ?? null, attachments: m.attachments?.length ? m.attachments : null, quick_reply_payload: null, @@ -78,7 +80,46 @@ export async function GET(request: NextRequest) { created_at: m.sentAt ?? m.createdAt ?? new Date().toISOString(), })); - return NextResponse.json(messages); + // Merge with locally stored messages. Zernio is the source of truth but + // its API can lag or miss messages (especially outbound sends). The local + // table has messages the webhook stored (inbound) and flow engine stored + // (outbound). Dedup on platform_message_id so messages that appear in both + // don't render twice. + const { data: localMessages } = await supabase + .from("messages") + .select("*") + .eq("conversation_id", conversationId) + .order("created_at", { ascending: true }); + + const seenIds = new Set(); + const merged: Array> = []; + + for (const m of zernioMapped) { + const key = (m.platform_message_id || m.id) as string | null; + if (key && !seenIds.has(key)) { + seenIds.add(key); + merged.push(m as unknown as Record); + } + } + + if (localMessages) { + for (const m of localMessages) { + const key = m.platform_message_id || m.id; + if (key && !seenIds.has(key)) { + seenIds.add(key); + merged.push(m as unknown as Record); + } + } + } + + // Sort merged messages by created_at for consistent chronological display + merged.sort((a, b) => { + const aTime = new Date(a.created_at as string).getTime(); + const bTime = new Date(b.created_at as string).getTime(); + return aTime - bTime; + }); + + return NextResponse.json(merged); } catch (error) { console.error("Failed to fetch messages from Zernio API:", error); return NextResponse.json( diff --git a/app/api/webhooks/late/route.ts b/app/api/webhooks/late/route.ts index df73226..070f90b 100644 --- a/app/api/webhooks/late/route.ts +++ b/app/api/webhooks/late/route.ts @@ -6,6 +6,7 @@ import { matchTrigger } from "@/lib/flow-engine/trigger-matcher"; import { resolveWebhookSecret, verifyWebhookSignature } from "@/lib/zernio-webhook"; import { upsertContactForSender } from "@/lib/inbox-sync"; import { processComment } from "@/lib/comment-processor"; +import { createZernioClient } from "@/lib/zernio-client"; import type { Database } from "@/lib/types/database"; import { messagePreview } from "@/lib/message-preview"; @@ -137,8 +138,10 @@ async function handleWebhook(request: NextRequest) { const { message: msg, account } = payload; - // Ignore outbound messages (sent by the bot itself) to prevent loops - if (msg.direction === "outbound") { + // Ignore outbound messages (sent by the bot itself) to prevent loops. + // Accept both "outgoing" (Zernio's actual enum value) and the legacy + // "outbound" spelling for resilience. + if (msg.direction === "outgoing" || msg.direction === "outbound") { return NextResponse.json({ ok: true, skipped: true }); } @@ -264,7 +267,27 @@ async function processMessageEvent( .then(() => {}); } - // Messages are stored by Zernio (source of truth) — no local insert needed. + // Store the incoming DM locally so the inbox has a reliable history even + // when Zernio's API returns stale or incomplete data. Use platformMessageId + // (not msg.id which is Zernio's internal id) so it matches what the messages + // route returns and dedup works when both sources are merged. + const { data: existingMsg } = await supabase + .from("messages") + .select("id") + .eq("conversation_id", conversation.id) + .eq("platform_message_id", msg.platformMessageId || msg.id) + .maybeSingle(); + + if (!existingMsg) { + await supabase.from("messages").insert({ + conversation_id: conversation.id, + direction: "inbound", + text: msg.text, + attachments: msg.attachments?.length ? msg.attachments : null, + platform_message_id: msg.platformMessageId || msg.id, + status: "sent", + }); + } // ── Flow engine ─────────────────────────────────────────────────────────── @@ -281,12 +304,22 @@ async function processMessageEvent( }, }; - const handled = await handleGlobalKeywords( + // Look up workspace API key for keyword confirmation DMs + const { data: kwWorkspace } = await supabase + .from("workspaces") + .select("late_api_key_encrypted") + .eq("id", channel.workspace_id) + .single(); + + const handled = await handleGlobalKeywords({ supabase, - channel.workspace_id, + workspaceId: channel.workspace_id, + workspaceApiKey: kwWorkspace?.late_api_key_encrypted ?? null, + lateConversationId: conv.id, + lateAccountId: account.id, contactId, - msg.text || undefined - ); + text: msg.text || undefined, + }); if (!handled) { const trigger = await matchTrigger(supabase, { @@ -310,7 +343,30 @@ async function processMessageEvent( lateAccountId: account.id, }); } catch (err) { - console.error("Flow execution error:", err); + const errMsg = err instanceof Error ? err.message : String(err); + + // Distinguish 24h-window failures from other errors. Meta only + // allows businesses to message within 24h of the contact's last + // message; sending after the window fails with a specific error. + // Logging it distinctly surfaces this operational gap instead of + // collapsing it into a generic message_failed event. + const is24hWindowError = /24.*hour|window|outside.*allowed/i.test(errMsg); + + console.error("Flow execution error:", errMsg); + + await supabase.from("analytics_events").insert({ + workspace_id: channel.workspace_id, + contact_id: contactId, + flow_id: trigger.flow_id, + event_type: is24hWindowError + ? "message_failed_24h_window" + : "message_failed", + metadata: { + error: errMsg, + triggerId: trigger.id, + ...(is24hWindowError ? { reason: "24h_messaging_window" } : {}), + }, + }); } } } @@ -384,12 +440,39 @@ async function handleCommentWebhook( // ── Global keywords ───────────────────────────────────────────────────────── -async function handleGlobalKeywords( - supabase: Awaited>, - workspaceId: string, - contactId: string, - text: string | undefined -): Promise { +/** + * Context object for keyword confirmation DMs. Passing a single object avoids + * the "too many positional parameters" problem where a caller silently passes + * arguments in the wrong order. + */ +interface KeywordConfirmationParams { + supabase: Awaited>; + workspaceId: string; + workspaceApiKey: string | null; + lateConversationId: string | null; + lateAccountId: string | null; + contactId: string; + text: string | undefined; +} + +/** + * Checks the incoming text against the workspace's global keywords. When a + * STOP/START keyword matches, updates the contact's subscription and sends a + * confirmation DM so the contact knows their opt-in/opt-out was received. + * + * The confirmation DM is only sent when we have a late_conversation_id (i.e. + * a DM context). Comment-triggered flows do not have one at keyword-check + * time, so the confirmation is a no-op there — the early return below guards it. + */ +async function handleGlobalKeywords({ + supabase, + workspaceId, + workspaceApiKey, + lateConversationId, + lateAccountId, + contactId, + text, +}: KeywordConfirmationParams): Promise { if (!text) return false; const { data: workspace } = await supabase @@ -415,6 +498,17 @@ async function handleGlobalKeywords( .from("contacts") .update({ is_subscribed: false }) .eq("id", contactId); + + // Send confirmation DM. No-ops when there is no DM conversation yet + // (e.g. keyword sent on a comment-triggered flow) or when outside the + // 24h messaging window. + await sendKeywordConfirmation({ + workspaceApiKey, + lateConversationId, + lateAccountId, + message: "You've been unsubscribed from our messages. Text START to resubscribe anytime.", + }); + return true; } if (kw.action === "subscribe") { @@ -422,6 +516,14 @@ async function handleGlobalKeywords( .from("contacts") .update({ is_subscribed: true }) .eq("id", contactId); + + await sendKeywordConfirmation({ + workspaceApiKey, + lateConversationId, + lateAccountId, + message: "You're subscribed! Reply STOP anytime to opt out.", + }); + return true; } return false; @@ -430,3 +532,34 @@ async function handleGlobalKeywords( return false; } + +/** + * Sends a single confirmation DM. Silently no-ops when the prerequisites for + * sending (API key, conversation ID, account ID) are missing — this happens + * for comment-triggered flows where no DM conversation exists yet. + */ +async function sendKeywordConfirmation({ + workspaceApiKey, + lateConversationId, + lateAccountId, + message, +}: { + workspaceApiKey: string | null; + lateConversationId: string | null; + lateAccountId: string | null; + message: string; +}): Promise { + if (!workspaceApiKey || !lateConversationId || !lateAccountId) return; + + try { + const zernio = createZernioClient(workspaceApiKey); + await zernio.messages.sendInboxMessage({ + path: { conversationId: lateConversationId }, + body: { accountId: lateAccountId, message }, + }); + } catch (err) { + // Most likely cause is the 24h messaging window having elapsed; log and + // move on rather than surfacing this as a flow failure. + console.error("Failed to send keyword confirmation DM:", err); + } +} diff --git a/components/flow-builder/panels/AiResponsePanel.tsx b/components/flow-builder/panels/AiResponsePanel.tsx index 0386842..e6e9446 100644 --- a/components/flow-builder/panels/AiResponsePanel.tsx +++ b/components/flow-builder/panels/AiResponsePanel.tsx @@ -14,6 +14,8 @@ interface AiResponsePanelData { maxTokens?: number; contextMessages?: number; sendDirectly?: boolean; + maxRetries?: number; + fallbackMessage?: string; [key: string]: unknown; } @@ -161,6 +163,47 @@ export function AiResponsePanel({ data: rawData, onChange }: AiResponsePanelProp yourself with a Send Message node.

+ + {/* Resilience */} +
+

Resilience

+ + {/* Max Retries */} +
+ + onChange({ ...data, maxRetries: parseInt(e.target.value) || 0 })} + min={0} + max={10} + className="w-full rounded-lg border border-border bg-card px-3 py-2 text-sm text-foreground focus:border-blue-500 focus:outline-none focus:ring-1 focus:ring-blue-500" + /> +

+ Retry attempts with exponential backoff on transient errors (rate + limits, timeouts). Set to 0 to disable. +

+
+ + {/* Fallback Message */} +
+ +