Merge pull request #364 from ArnasDon/fix/duplicate-conversations

fix: stop inbox creating duplicate chats for one contact (#363)

CHANGELOG.md

@@ -9,6 +9,28 @@ Versions follow [Keep a Changelog](https://keepachangelog.com/en/1.1.0/).
 Pre-1.0, `MINOR` bumps cover new modules; `PATCH` bumps cover bug fixes
 and polish.
 
+## [0.8.1] — 2026-07-10
+
+Fixes inbound chats fragmenting into multiple threads for the same
+number.
+
+> **Migration required:** apply `supabase/migrations/036_conversation_contact_dedup.sql`
+> (merges any existing duplicate conversations into the oldest thread —
+> no messages are lost — then adds a `UNIQUE (account_id, contact_id)`
+> index so one contact can only ever have one conversation).
+
+### Fixed
+
+- **Duplicate chats for a single contact.** An inbound message could
+  create a second conversation for a contact under a race (Meta retries a
+  delivery, or a batch fans out to concurrent runs). Once two existed,
+  the `.single()` lookup errored on every later message and the webhook
+  created yet another conversation each time, snowballing into a wall of
+  duplicate chats. The find-or-create now resolves to the oldest existing
+  thread and a DB unique index makes the one-conversation-per-contact
+  rule authoritative. The same hardening was applied to the public-API
+  conversation resolver. (Issue #363)
+
 ## [0.8.0] — 2026-07-08
 
 Polishes the AI auto-reply bot: it's now **visible and controllable from

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

@@ -1046,16 +1046,34 @@ async function findOrCreateConversation(
   configOwnerUserId: string,
   contactId: string,
 ) {
-  // Look for existing conversation in this account
-  const { data: existing, error: findError } = await supabaseAdmin()
+  // Look for an existing conversation in this account, oldest-first.
+  //
+  // We deliberately do NOT use `.single()` here. `.single()` errors on
+  // *both* 0 rows and ≥2 rows, and the old code treated any error as
+  // "none found" and inserted a new row. So once two conversations
+  // existed for a contact (from a race — Meta retries a delivery, or a
+  // batch fans out to concurrent runs), every subsequent inbound
+  // message errored on the lookup and created yet another conversation,
+  // snowballing into a wall of duplicate chats (issue #363).
+  //
+  // Ordering oldest-first and taking one row makes the lookup resolve to
+  // the same canonical survivor the dedup migration (036) keeps, so any
+  // pre-existing duplicates converge instead of compounding.
+  const { data: existingRows, error: findError } = await supabaseAdmin()
     .from('conversations')
     .select('*')
     .eq('account_id', accountId)
     .eq('contact_id', contactId)
-    .single()
+    .order('created_at', { ascending: true })
+    .limit(1)
 
-  if (!findError && existing) {
-    return { conversation: existing, created: false }
+  if (findError) {
+    console.error('Error finding conversation:', findError)
+    return null
+  }
+
+  if (existingRows && existingRows.length > 0) {
+    return { conversation: existingRows[0], created: false }
   }
 
   // Create new conversation. Same tenancy + audit split as
@@ -1071,6 +1089,22 @@ async function findOrCreateConversation(
     .single()
 
   if (createError) {
+    // Lost a race: a concurrent inbound delivery created the
+    // conversation between our lookup and insert, and the unique index
+    // (migration 036) rejected the duplicate. Re-resolve the winning
+    // row instead of dropping the message — mirrors findOrCreateContact.
+    if (isUniqueViolation(createError)) {
+      const { data: raced } = await supabaseAdmin()
+        .from('conversations')
+        .select('*')
+        .eq('account_id', accountId)
+        .eq('contact_id', contactId)
+        .order('created_at', { ascending: true })
+        .limit(1)
+      if (raced && raced.length > 0) {
+        return { conversation: raced[0], created: false }
+      }
+    }
     console.error('Error creating conversation:', createError)
     return null
   }

src/lib/whatsapp/resolve-conversation.test.ts

@@ -19,14 +19,21 @@ interface Script {
   contactCandidatesByCall?: ContactRow[][];
   insertedContactId?: string; // contacts insert -> single
   insertContactError?: { code?: string } | null;
-  existingConversation?: { id: string } | null; // conversations select.maybeSingle
+  /** Conversation lookup result (oldest-first `.order().limit(1)`).
+   *  A single row or null; wrapped into a one-element array internally. */
+  existingConversation?: { id: string } | null; // conversations select.limit(1)
+  /** Per-call conversation lookup results — overrides existingConversation.
+   *  Lets a test simulate "miss, then hit" for the unique-race path. */
+  existingConversationByCall?: (({ id: string } | null))[];
   insertedConversationId?: string; // conversations insert -> single
+  insertConversationError?: { code?: string } | null;
 }
 
 function makeDb(script: Script): SupabaseClient {
   let table = '';
   let mode: 'select' | 'insert' | 'update' = 'select';
   let likeCalls = 0;
+  let convLookupCalls = 0;
 
   const builder: Record<string, unknown> = {
     select: () => builder,
@@ -39,6 +46,18 @@ function makeDb(script: Script): SupabaseClient {
       return builder;
     },
     eq: () => builder,
+    order: () => builder,
+    limit: () => {
+      // Only the conversation lookup terminates on `.limit(1)`.
+      if (table === 'conversations' && mode === 'select') {
+        const row = script.existingConversationByCall
+          ? (script.existingConversationByCall[convLookupCalls] ?? null)
+          : (script.existingConversation ?? null);
+        convLookupCalls++;
+        return Promise.resolve({ data: row ? [row] : [], error: null });
+      }
+      return Promise.resolve({ data: [], error: null });
+    },
     like: () => {
       const data = script.contactCandidatesByCall
         ? (script.contactCandidatesByCall[likeCalls] ?? [])
@@ -49,11 +68,6 @@ function makeDb(script: Script): SupabaseClient {
     maybeSingle: () => {
       if (table === 'whatsapp_config')
         return Promise.resolve({ data: script.config ?? null, error: null });
-      if (table === 'conversations' && mode === 'select')
-        return Promise.resolve({
-          data: script.existingConversation ?? null,
-          error: null,
-        });
       return Promise.resolve({ data: null, error: null });
     },
     single: () => {
@@ -68,11 +82,17 @@ function makeDb(script: Script): SupabaseClient {
           error: null,
         });
       }
-      if (table === 'conversations' && mode === 'insert')
+      if (table === 'conversations' && mode === 'insert') {
+        if (script.insertConversationError)
+          return Promise.resolve({
+            data: null,
+            error: script.insertConversationError,
+          });
         return Promise.resolve({
           data: { id: script.insertedConversationId },
           error: null,
         });
+      }
       return Promise.resolve({ data: null, error: null });
     },
     // Thenable: `await db.from().update().eq()` lands here.
@@ -168,4 +188,23 @@ describe('resolveConversationByPhone', () => {
     expect(res.contactCreated).toBe(false);
     expect(res.conversationId).toBe('cv-raced');
   });
+
+  it('re-resolves the conversation when the insert loses a unique race', async () => {
+    // Existing contact, conversation lookup misses first (→ attempt an
+    // insert), the insert hits a 23505 from a concurrent create, and the
+    // post-race re-lookup returns the winning conversation — no duplicate
+    // conversation is created (issue #363).
+    const db = makeDb({
+      config: { user_id: 'owner-1' },
+      contactCandidates: [{ id: 'c1', phone: '14155550123' }],
+      existingConversationByCall: [null, { id: 'cv-raced' }],
+      insertConversationError: { code: '23505' },
+    });
+    const res = await resolveConversationByPhone(db, 'acct', '+14155550123');
+    expect(res).toEqual({
+      conversationId: 'cv-raced',
+      contactId: 'c1',
+      contactCreated: false,
+    });
+  });
 });

src/lib/whatsapp/resolve-conversation.ts

@@ -138,16 +138,47 @@ export async function resolveConversationByPhone(
 
   // ---- conversation -------------------------------------------
   // One conversation per (account, contact) — same convention as the
-  // webhook.
-  const { data: conv } = await db
+  // webhook. Order oldest-first and take one row rather than
+  // `.maybeSingle()`, which errors on ≥2 rows: if duplicates predate the
+  // unique index (migration 036), we resolve to the canonical survivor
+  // instead of falling through and creating yet another (issue #363).
+  const conversationId = await findOrCreateConversationRow(
+    db,
+    accountId,
+    contactId,
+    ownerUserId
+  );
+
+  return { conversationId, contactId, contactCreated };
+}
+
+/**
+ * Find (oldest-first) or create the single conversation for
+ * `(accountId, contactId)`. Handles the unique-index race the same way
+ * the inbound webhook does: on a 23505 from a concurrent create,
+ * re-resolve the winning row rather than failing the send.
+ */
+async function findOrCreateConversationRow(
+  db: SupabaseClient,
+  accountId: string,
+  contactId: string,
+  ownerUserId: string
+): Promise<string> {
+  const { data: existing, error: findErr } = await db
     .from('conversations')
     .select('id')
     .eq('account_id', accountId)
     .eq('contact_id', contactId)
-    .maybeSingle();
+    .order('created_at', { ascending: true })
+    .limit(1);
 
-  if (conv?.id) {
-    return { conversationId: conv.id, contactId, contactCreated };
+  if (findErr) {
+    console.error('[resolve-conversation] conversation lookup error:', findErr);
+    throw new SendMessageError('db_error', 'Failed to resolve conversation', 500);
+  }
+
+  if (existing && existing.length > 0) {
+    return existing[0].id;
   }
 
   const { data: newConv, error: convErr } = await db
@@ -161,13 +192,21 @@ export async function resolveConversationByPhone(
     .single();
 
   if (convErr || !newConv) {
+    if (isUniqueViolation(convErr)) {
+      const { data: raced } = await db
+        .from('conversations')
+        .select('id')
+        .eq('account_id', accountId)
+        .eq('contact_id', contactId)
+        .order('created_at', { ascending: true })
+        .limit(1);
+      if (raced && raced.length > 0) {
+        return raced[0].id;
+      }
+    }
     console.error('[resolve-conversation] conversation create error:', convErr);
-    throw new SendMessageError(
-      'db_error',
-      'Failed to create conversation',
-      500
-    );
+    throw new SendMessageError('db_error', 'Failed to create conversation', 500);
   }
 
-  return { conversationId: newConv.id, contactId, contactCreated };
+  return newConv.id;
 }

supabase/migrations/036_conversation_contact_dedup.sql

@@ -0,0 +1,126 @@
+-- ============================================================
+-- 036_conversation_contact_dedup
+--
+-- Prevent the same contact from fragmenting into multiple
+-- conversations within one account (issue #363).
+--
+-- The inbound webhook and the public-API resolver both follow a
+-- "one conversation per (account, contact)" convention, but that
+-- convention was only ever enforced in application code with a
+-- `.single()` / `.maybeSingle()` lookup and no DB constraint. Two
+-- problems compounded:
+--
+--   1. A race (Meta retries a delivery, or a batch delivers two
+--      messages that fan out to concurrent `after()` runs) let two
+--      inserts both miss the lookup and create two conversations —
+--      unlike contacts (migration 022) there was no unique index and
+--      no unique-violation backstop.
+--   2. Once ≥2 conversations existed for a contact, the `.single()`
+--      lookup errored on *every* subsequent inbound message, so the
+--      code fell through and created yet another conversation each
+--      time — the duplication snowballed, which is what the reporter
+--      saw (a wall of duplicate chats for one number).
+--
+-- This migration mirrors 022_contact_phone_dedup:
+--   1. merges existing duplicate conversations into the oldest row,
+--      re-pointing every conversation-scoped child first so nothing
+--      is lost;
+--   2. adds a UNIQUE index on (account_id, contact_id) — the
+--      authoritative guarantee that covers every write path.
+--
+-- Idempotent. **No data loss** — duplicate conversations are merged,
+-- not dropped: child rows (messages, message_reactions, deals,
+-- flow_runs, notifications, ai_usage_log) are re-pointed to the
+-- surviving (oldest) conversation before the losers are deleted.
+-- ============================================================
+
+-- 1) One-time (re-runnable) merge of existing duplicates.
+--    SECURITY DEFINER so it can re-point rows across tables
+--    regardless of the caller's RLS; it only ever collapses
+--    conversations that share the same (account_id, contact_id).
+CREATE OR REPLACE FUNCTION public.merge_duplicate_conversations()
+RETURNS INTEGER
+LANGUAGE plpgsql
+SECURITY DEFINER
+SET search_path = public
+AS $$
+DECLARE
+  v_group    RECORD;
+  v_survivor UUID;
+  v_losers   UUID[];
+  v_all      UUID[];
+  v_merged   INTEGER := 0;
+BEGIN
+  FOR v_group IN
+    SELECT account_id,
+           contact_id,
+           array_agg(id ORDER BY created_at ASC, id ASC) AS ids,
+           COALESCE(SUM(unread_count), 0)                AS total_unread
+    FROM conversations
+    GROUP BY account_id, contact_id
+    HAVING count(*) > 1
+  LOOP
+    v_all      := v_group.ids;
+    v_survivor := v_all[1];
+    v_losers   := v_all[2:array_length(v_all, 1)];
+
+    -- Re-point every conversation-scoped child from the losers onto
+    -- the survivor. None of these carry a conversation-scoped unique
+    -- constraint (message_id is intentionally non-unique — see
+    -- migration 009), so a plain UPDATE is safe. Doing this BEFORE the
+    -- delete is what saves the ON DELETE CASCADE children (messages,
+    -- message_reactions, notifications) from being removed with the
+    -- loser conversations.
+    UPDATE messages          SET conversation_id = v_survivor WHERE conversation_id = ANY(v_losers);
+    UPDATE message_reactions SET conversation_id = v_survivor WHERE conversation_id = ANY(v_losers);
+    UPDATE deals             SET conversation_id = v_survivor WHERE conversation_id = ANY(v_losers);
+    UPDATE flow_runs         SET conversation_id = v_survivor WHERE conversation_id = ANY(v_losers);
+    UPDATE notifications     SET conversation_id = v_survivor WHERE conversation_id = ANY(v_losers);
+    UPDATE ai_usage_log      SET conversation_id = v_survivor WHERE conversation_id = ANY(v_losers);
+
+    -- Roll the merged unread counts onto the survivor and re-derive
+    -- its last-message summary from the now-complete message set, so
+    -- the surviving thread reflects the full history.
+    UPDATE conversations c
+    SET unread_count      = v_group.total_unread,
+        last_message_text = lm.content_text,
+        last_message_at   = lm.created_at,
+        updated_at        = NOW()
+    FROM (
+      SELECT content_text, created_at
+      FROM messages
+      WHERE conversation_id = v_survivor
+      ORDER BY created_at DESC
+      LIMIT 1
+    ) lm
+    WHERE c.id = v_survivor;
+
+    -- Survivor may have no messages at all (edge case). Still fold in
+    -- the merged unread count in that case.
+    UPDATE conversations
+    SET unread_count = v_group.total_unread,
+        updated_at   = NOW()
+    WHERE id = v_survivor
+      AND NOT EXISTS (SELECT 1 FROM messages WHERE conversation_id = v_survivor);
+
+    DELETE FROM conversations WHERE id = ANY(v_losers);
+
+    v_merged := v_merged + COALESCE(array_length(v_losers, 1), 0);
+  END LOOP;
+
+  RETURN v_merged;
+END;
+$$;
+
+ALTER FUNCTION public.merge_duplicate_conversations() OWNER TO postgres;
+REVOKE ALL ON FUNCTION public.merge_duplicate_conversations() FROM PUBLIC;
+
+-- Collapse whatever duplicates exist right now.
+SELECT public.merge_duplicate_conversations();
+
+-- 2) Authoritative guarantee: one conversation per (account, contact).
+--    Every write path (inbound webhook, public-API resolver) now has a
+--    DB-level backstop, and its unique-violation handling can re-resolve
+--    the winning row instead of compounding duplicates.
+CREATE UNIQUE INDEX IF NOT EXISTS idx_conversations_account_contact
+  ON conversations (account_id, contact_id);