fix: land the stranded inbound-webhook + broadcast reliability work on main

src/app/api/whatsapp/webhook/route.test.ts

@@ -0,0 +1,276 @@
+import { describe, it, expect, vi, beforeEach } from 'vitest'
+
+// Shared, hoisted state the module mocks close over. Reset per test.
+const h = vi.hoisted(() => ({
+  runAutomationsForTrigger: vi.fn(),
+  dispatchInboundToFlows: vi.fn(),
+  dispatchInboundToAiReply: vi.fn(),
+  dispatchWebhookEvent: vi.fn(),
+  state: {
+    // Result the message upsert's .select() resolves to. A genuine insert
+    // returns the row; a replayed delivery conflicts and returns [].
+    messageUpsertResult: [{ id: 'msg-1' }] as { id: string }[],
+    priorCustomerMsgCount: 0,
+    conversation: { id: 'conv-1', unread_count: 0, account_id: 'acc-1' },
+    upsertCalls: [] as { row: Record<string, unknown>; options: unknown }[],
+    rpcCalls: [] as { name: string; args: Record<string, unknown> }[],
+    afterCallbacks: [] as (() => Promise<void> | void)[],
+    automationStarted: 0,
+    automationCompleted: 0,
+  },
+}))
+
+vi.mock('next/server', () => ({
+  after: (cb: () => Promise<void> | void) => {
+    h.state.afterCallbacks.push(cb)
+  },
+  NextResponse: {
+    json: (body: unknown, init?: { status?: number }) => ({ body, init }),
+  },
+}))
+
+vi.mock('@supabase/supabase-js', () => ({
+  createClient: () => ({
+    from(table: string) {
+      switch (table) {
+        case 'whatsapp_config':
+          return {
+            select: () => ({
+              eq: () =>
+                Promise.resolve({
+                  data: [
+                    {
+                      account_id: 'acc-1',
+                      user_id: 'user-1',
+                      access_token: 'enc',
+                    },
+                  ],
+                  error: null,
+                }),
+            }),
+          }
+        case 'conversations':
+          // findOrCreateConversation: select().eq().eq().order().limit()
+          return {
+            select: () => ({
+              eq: () => ({
+                eq: () => ({
+                  order: () => ({
+                    limit: () =>
+                      Promise.resolve({
+                        data: [h.state.conversation],
+                        error: null,
+                      }),
+                  }),
+                }),
+              }),
+            }),
+          }
+        case 'broadcast_recipients':
+          // flagBroadcastReplyIfAny: select().eq().eq().in().order().limit()
+          return {
+            select: () => ({
+              eq: () => ({
+                eq: () => ({
+                  in: () => ({
+                    order: () => ({
+                      limit: () =>
+                        Promise.resolve({ data: [], error: null }),
+                    }),
+                  }),
+                }),
+              }),
+            }),
+          }
+        case 'messages':
+          return {
+            // priorCustomerMsgCount: select('id',{count,head}).eq().eq()
+            select: () => ({
+              eq: () => ({
+                eq: () =>
+                  Promise.resolve({
+                    count: h.state.priorCustomerMsgCount,
+                    error: null,
+                  }),
+              }),
+            }),
+            // Idempotent insert: upsert(...).select('id')
+            upsert: (row: Record<string, unknown>, options: unknown) => {
+              h.state.upsertCalls.push({ row, options })
+              return {
+                select: () =>
+                  Promise.resolve({
+                    data: h.state.messageUpsertResult,
+                    error: null,
+                  }),
+              }
+            },
+          }
+        default:
+          throw new Error(`unexpected table: ${table}`)
+      }
+    },
+    rpc: (name: string, args: Record<string, unknown>) => {
+      h.state.rpcCalls.push({ name, args })
+      return Promise.resolve({ data: null, error: null })
+    },
+  }),
+}))
+
+vi.mock('@/lib/whatsapp/encryption', () => ({
+  decrypt: () => 'plain-token',
+  encrypt: (v: string) => v,
+  isLegacyFormat: () => false,
+}))
+vi.mock('@/lib/whatsapp/meta-api', () => ({
+  getMediaUrl: vi.fn(),
+  downloadMedia: vi.fn(),
+}))
+vi.mock('@/lib/contacts/dedupe', () => ({
+  findExistingContact: vi.fn(async () => ({
+    id: 'contact-1',
+    name: 'Ada',
+    phone: '15551230000',
+  })),
+  isUniqueViolation: () => false,
+}))
+vi.mock('@/lib/whatsapp/webhook-signature', () => ({
+  verifyMetaWebhookSignature: () => true,
+}))
+vi.mock('@/lib/whatsapp/template-webhook', () => ({
+  isTemplateWebhookField: () => false,
+  handleTemplateWebhookChange: vi.fn(),
+}))
+vi.mock('@/lib/automations/engine', () => ({
+  runAutomationsForTrigger: h.runAutomationsForTrigger,
+}))
+vi.mock('@/lib/flows/engine', () => ({
+  dispatchInboundToFlows: h.dispatchInboundToFlows,
+}))
+vi.mock('@/lib/ai/auto-reply', () => ({
+  dispatchInboundToAiReply: h.dispatchInboundToAiReply,
+}))
+vi.mock('@/lib/webhooks/deliver', () => ({
+  dispatchWebhookEvent: h.dispatchWebhookEvent,
+}))
+
+import { POST } from './route'
+
+function inboundRequest() {
+  const body = {
+    entry: [
+      {
+        changes: [
+          {
+            field: 'messages',
+            value: {
+              metadata: { phone_number_id: 'pn-1' },
+              contacts: [{ wa_id: '15551230000', profile: { name: 'Ada' } }],
+              messages: [
+                {
+                  id: 'wamid.TEST1',
+                  from: '15551230000',
+                  timestamp: '1700000000',
+                  type: 'text',
+                  text: { body: 'hello' },
+                },
+              ],
+            },
+          },
+        ],
+      },
+    ],
+  }
+  return {
+    text: async () => JSON.stringify(body),
+    headers: { get: () => 'sha256=stub' },
+  } as unknown as Request
+}
+
+async function runWebhook() {
+  const res = await POST(inboundRequest())
+  // Drain the after() callback exactly as the runtime would.
+  for (const cb of h.state.afterCallbacks) await cb()
+  return res
+}
+
+beforeEach(() => {
+  vi.clearAllMocks()
+  h.state.messageUpsertResult = [{ id: 'msg-1' }]
+  h.state.priorCustomerMsgCount = 0
+  h.state.conversation = { id: 'conv-1', unread_count: 0, account_id: 'acc-1' }
+  h.state.upsertCalls = []
+  h.state.rpcCalls = []
+  h.state.afterCallbacks = []
+  h.state.automationStarted = 0
+  h.state.automationCompleted = 0
+  h.dispatchInboundToFlows.mockResolvedValue({ consumed: false })
+  h.dispatchInboundToAiReply.mockResolvedValue(undefined)
+  h.dispatchWebhookEvent.mockResolvedValue(undefined)
+  h.runAutomationsForTrigger.mockImplementation(() => {
+    h.state.automationStarted++
+    return new Promise<void>((resolve) => {
+      setTimeout(() => {
+        h.state.automationCompleted++
+        resolve()
+      }, 0)
+    })
+  })
+})
+
+describe('inbound webhook: idempotent insert (#367)', () => {
+  it('a genuine first delivery persists once and fans out downstream', async () => {
+    await runWebhook()
+
+    // Inserted via upsert with the (conversation_id, message_id) conflict
+    // target — not a bare insert.
+    expect(h.state.upsertCalls).toHaveLength(1)
+    expect(h.state.upsertCalls[0].options).toMatchObject({
+      onConflict: 'conversation_id,message_id',
+      ignoreDuplicates: true,
+    })
+    // Downstream side effects ran exactly once.
+    expect(h.state.rpcCalls).toHaveLength(1)
+    expect(h.dispatchInboundToFlows).toHaveBeenCalledTimes(1)
+    expect(h.dispatchWebhookEvent).toHaveBeenCalledTimes(1)
+  })
+
+  it('a replayed delivery is a no-op: no unread bump, no fan-out', async () => {
+    // Upsert hits the unique index and returns no row.
+    h.state.messageUpsertResult = []
+
+    await runWebhook()
+
+    expect(h.state.upsertCalls).toHaveLength(1)
+    // None of the downstream side effects fire on a replay.
+    expect(h.state.rpcCalls).toHaveLength(0)
+    expect(h.dispatchInboundToFlows).not.toHaveBeenCalled()
+    expect(h.runAutomationsForTrigger).not.toHaveBeenCalled()
+    expect(h.dispatchInboundToAiReply).not.toHaveBeenCalled()
+    expect(h.dispatchWebhookEvent).not.toHaveBeenCalled()
+  })
+})
+
+describe('inbound webhook: atomic unread bump (#369)', () => {
+  it('increments unread through the DB-side RPC, not a read-modify-write', async () => {
+    await runWebhook()
+
+    expect(h.state.rpcCalls).toHaveLength(1)
+    expect(h.state.rpcCalls[0]).toMatchObject({
+      name: 'bump_conversation_on_inbound',
+      args: { p_conversation_id: 'conv-1' },
+    })
+  })
+})
+
+describe('inbound webhook: after() awaits automations (#368)', () => {
+  it('every triggered automation settles before the after() callback resolves', async () => {
+    await runWebhook()
+
+    // first_inbound_message + new_message_received + keyword_match.
+    expect(h.state.automationStarted).toBe(3)
+    // If the dispatches were fire-and-forget, completed would still be 0
+    // here — the callback would have resolved before the timers fired.
+    expect(h.state.automationCompleted).toBe(3)
+  })
+})

src/app/api/whatsapp/webhook/route.ts

@@ -1,7 +1,7 @@
 import { NextResponse, after } from 'next/server'
 import { createClient } from '@supabase/supabase-js'
 import { decrypt, encrypt, isLegacyFormat } from '@/lib/whatsapp/encryption'
 import { getMediaUrl, downloadMedia } from '@/lib/whatsapp/meta-api'
 import { normalizePhone } from '@/lib/whatsapp/phone-utils'
 import { findExistingContact, isUniqueViolation } from '@/lib/contacts/dedupe'
 import { verifyMetaWebhookSignature } from '@/lib/whatsapp/webhook-signature'
@@ -666,37 +666,67 @@
     .eq('sender_type', 'customer')
   const isFirstInboundMessage = (priorCustomerMsgCount ?? 0) === 0
 
-  const { error: msgError } = await supabaseAdmin().from('messages').insert({
-    conversation_id: conversation.id,
-    sender_type: 'customer',
-    content_type: contentType,
-    content_text: contentText,
-    media_url: mediaUrl,
-    message_id: message.id,
-    status: 'delivered',
-    created_at: new Date(parseInt(message.timestamp) * 1000).toISOString(),
-    reply_to_message_id: replyToInternalId,
-    // Only populated for content_type='interactive'. Migration 010 added
-    // the column; null for every other content_type so existing inserts
-    // behave identically.
-    interactive_reply_id: interactiveReplyId,
-  })
+  // Idempotent insert. Meta retries webhook deliveries (a slow ack, a
+  // transient 5xx), and each retry replays the exact same message.id. The
+  // unique index on (conversation_id, message_id) added in migration 037
+  // makes a replay conflict; `ignoreDuplicates` turns that into an ON
+  // CONFLICT DO NOTHING, and the `.select()` then returns the inserted row
+  // ONLY on a genuine first insert — an empty result means this delivery
+  // was a replay. This is the single idempotency boundary that must sit
+  // BEFORE the unread bump and all downstream fan-out below (issue #367).
+  const { data: insertedRows, error: msgError } = await supabaseAdmin()
+    .from('messages')
+    .upsert(
+      {
+        conversation_id: conversation.id,
+        sender_type: 'customer',
+        content_type: contentType,
+        content_text: contentText,
+        media_url: mediaUrl,
+        message_id: message.id,
+        status: 'delivered',
+        created_at: new Date(parseInt(message.timestamp) * 1000).toISOString(),
+        reply_to_message_id: replyToInternalId,
+        // Only populated for content_type='interactive'. Migration 010 added
+        // the column; null for every other content_type so existing inserts
+        // behave identically.
+        interactive_reply_id: interactiveReplyId,
+      },
+      { onConflict: 'conversation_id,message_id', ignoreDuplicates: true }
+    )
+    .select('id')
 
   if (msgError) {
     console.error('Error inserting message:', msgError)
     return
   }
 
-  // Update conversation
-  const { error: convError } = await supabaseAdmin()
-    .from('conversations')
-    .update({
-      last_message_text: contentText || `[${message.type}]`,
-      last_message_at: new Date().toISOString(),
-      unread_count: (conversation.unread_count || 0) + 1,
-      updated_at: new Date().toISOString(),
-    })
-    .eq('id', conversation.id)
+  // Replayed delivery: the message already exists, so acknowledge it as a
+  // no-op. Returning here is what keeps a retry from double-bumping unread,
+  // re-advancing flows, re-firing automations, re-invoking AI handling, and
+  // re-dispatching public webhooks (issue #367).
+  if (!insertedRows || insertedRows.length === 0) {
+    console.info(
+      '[webhook] duplicate inbound message ignored (idempotent replay):',
+      message.id
+    )
+    return
+  }
+
+  // Update conversation. The unread bump is done DB-side (migration 037's
+  // bump_conversation_on_inbound) rather than as a read-modify-write of the
+  // snapshot loaded above: two inbound messages for the same conversation
+  // can process concurrently, and computing `snapshot + 1` in the app let
+  // both reads see the same value and write the same increment, losing one
+  // (issue #369). The RPC increments in a single UPDATE and refreshes the
+  // last-message summary in the same statement.
+  const { error: convError } = await supabaseAdmin().rpc(
+    'bump_conversation_on_inbound',
+    {
+      p_conversation_id: conversation.id,
+      p_last_message_text: contentText || `[${message.type}]`,
+    }
+  )
 
   if (convError) {
     console.error('Error updating conversation:', convError)