diff --git a/CHANGELOG.md b/CHANGELOG.md index e7e9a91c9..15cb025cb 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/src/app/api/whatsapp/webhook/route.ts b/src/app/api/whatsapp/webhook/route.ts index d1c147235..7687637e7 100644 --- a/src/app/api/whatsapp/webhook/route.ts +++ b/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 } diff --git a/src/lib/whatsapp/resolve-conversation.test.ts b/src/lib/whatsapp/resolve-conversation.test.ts index e483d8052..86b94d9b1 100644 --- a/src/lib/whatsapp/resolve-conversation.test.ts +++ b/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 = { 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, + }); + }); }); diff --git a/src/lib/whatsapp/resolve-conversation.ts b/src/lib/whatsapp/resolve-conversation.ts index 5789ea71b..d4b6f8188 100644 --- a/src/lib/whatsapp/resolve-conversation.ts +++ b/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 { + 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 (findErr) { + console.error('[resolve-conversation] conversation lookup error:', findErr); + throw new SendMessageError('db_error', 'Failed to resolve conversation', 500); + } - if (conv?.id) { - return { conversationId: conv.id, contactId, contactCreated }; + 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; } diff --git a/supabase/migrations/036_conversation_contact_dedup.sql b/supabase/migrations/036_conversation_contact_dedup.sql new file mode 100644 index 000000000..a59451e2d --- /dev/null +++ b/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);