LaLeV/wacrm
1
0
Fork 0
mirror of https://github.com/Luchexx/wacrm.git synced 2026-11-02 09:19:51 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

fix: stop inbound chats fragmenting into duplicate conversations (#363)

Browse Source 
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
after() runs). There was no UNIQUE index and no unique-violation
backstop on conversations, unlike contacts (migration 022). Worse, once
two conversations 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.

- findOrCreateConversation now resolves oldest-first with .limit(1)
  instead of .single(), so pre-existing duplicates converge, and catches
  23505 on insert to re-resolve the winning row.
- Same hardening applied to the public-API resolver (resolve-conversation).
- Migration 036 merges existing duplicate conversations into the oldest
  thread (re-pointing messages, reactions, deals, flow_runs,
  notifications, ai_usage_log — no data loss) then adds a
  UNIQUE (account_id, contact_id) index as the authoritative guarantee.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Arnas Donauskas 2026-07-10 12:27:46 +03:00
parent 137760e8f8
commit faf0f740a4
No known key found for this signature in database
5 changed files with 283 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
CHANGELOG.md
View File

@ -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

44
src/app/api/whatsapp/webhook/route.ts
View File

@ -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
}

53
src/lib/whatsapp/resolve-conversation.test.ts
View File

@ -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,
});
});
});

61
src/lib/whatsapp/resolve-conversation.ts
View File

@ -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;
}

126
supabase/migrations/036_conversation_contact_dedup.sql Normal file
View File

@ -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);