From 320f92fb9a77103af33afdc063c3059c52c925bb Mon Sep 17 00:00:00 2001 From: Arnas Donauskas Date: Tue, 14 Jul 2026 16:25:10 +0300 Subject: [PATCH] fix: harden inbound webhook + broadcast creation against replays, races, and partial failure MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Addresses four reliability bugs (#367, #368, #369, #370). Each pairs a DB-level guarantee (new migration 037) with the corresponding code change, since the application layer alone can't make these operations atomic. downstream side effect. Adds a unique index on (conversation_id, message_id) — the correctly-scoped idempotency key — and turns the message INSERT into an ON CONFLICT DO NOTHING upsert. An empty result means "replay": acknowledged as a no-op before the unread bump and all fan-out. after(), so a serverless freeze could interrupt them. They are now awaited via Promise.all; each keeps its own .catch so one failure neither blocks the others nor becomes an unhandled rejection. read-modify-write. Replaced with a DB-side atomic increment (bump_conversation_on_inbound RPC), mirroring migration 007. no recipients. Parent + recipients are now inserted in one transaction (create_broadcast_with_recipients RPC); a recipient failure rolls the parent back. Adds regression tests for all four (webhook route + broadcast-core). Co-Authored-By: Claude Opus 4.8 --- src/app/api/whatsapp/webhook/route.test.ts | 276 ++++++++++++++++++ src/app/api/whatsapp/webhook/route.ts | 80 +++-- src/lib/whatsapp/broadcast-core.test.ts | 107 ++++++- src/lib/whatsapp/broadcast-core.ts | 63 ++-- .../037_webhook_broadcast_reliability.sql | 154 ++++++++++ 5 files changed, 620 insertions(+), 60 deletions(-) create mode 100644 src/app/api/whatsapp/webhook/route.test.ts create mode 100644 supabase/migrations/037_webhook_broadcast_reliability.sql diff --git a/src/app/api/whatsapp/webhook/route.test.ts b/src/app/api/whatsapp/webhook/route.test.ts new file mode 100644 index 0000000000..9a3a090c21 --- /dev/null +++ b/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; options: unknown }[], + rpcCalls: [] as { name: string; args: Record }[], + afterCallbacks: [] as (() => Promise | void)[], + automationStarted: 0, + automationCompleted: 0, + }, +})) + +vi.mock('next/server', () => ({ + after: (cb: () => Promise | 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, 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) => { + 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((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) + }) +}) diff --git a/src/app/api/whatsapp/webhook/route.ts b/src/app/api/whatsapp/webhook/route.ts index f642e2db79..6f7160fd86 100644 --- a/src/app/api/whatsapp/webhook/route.ts +++ b/src/app/api/whatsapp/webhook/route.ts @@ -666,37 +666,67 @@ async function processMessage( .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) diff --git a/src/lib/whatsapp/broadcast-core.test.ts b/src/lib/whatsapp/broadcast-core.test.ts index 4b4e449802..777cfa8644 100644 --- a/src/lib/whatsapp/broadcast-core.test.ts +++ b/src/lib/whatsapp/broadcast-core.test.ts @@ -1,7 +1,16 @@ -import { describe, it, expect } from 'vitest'; +import { describe, it, expect, vi } from 'vitest'; import type { SupabaseClient } from '@supabase/supabase-js'; import { createBroadcast, BroadcastError } from './broadcast-core'; +// Contact resolution and token decryption are exercised elsewhere — stub +// them so these tests focus on the persistence boundary. +vi.mock('@/lib/whatsapp/encryption', () => ({ + decrypt: () => 'plain-access-token', +})); +vi.mock('@/lib/api/v1/contacts', () => ({ + findOrCreateContact: vi.fn(async () => ({ id: 'c1' })), +})); + // These assertions all fire in the pure validation prologue, before // any Supabase call — a bare stub is enough. const db = {} as SupabaseClient; @@ -34,3 +43,99 @@ describe('createBroadcast validation', () => { ).rejects.toMatchObject({ status: 400 }); }); }); + +// Build a Supabase-shaped mock that gets createBroadcast past its config + +// template lookups and into persistence. `rpcResult` is what the atomic +// create_broadcast_with_recipients RPC returns. +function makeDb(rpcResult: { data: unknown; error: unknown }) { + const calls = { + rpc: [] as { name: string; args: unknown }[], + // Incremented if the OLD non-atomic path (a direct broadcasts / + // broadcast_recipients insert) is ever reached — it must not be. + usedDirectInsert: 0, + }; + const database = { + from(table: string) { + if (table === 'whatsapp_config') { + return { + select: () => ({ + eq: () => ({ + single: () => + Promise.resolve({ + data: { phone_number_id: 'pn-1', access_token: 'enc' }, + error: null, + }), + }), + }), + }; + } + if (table === 'message_templates') { + const chain: Record = { + select: () => chain, + eq: () => chain, + maybeSingle: () => Promise.resolve({ data: null, error: null }), + }; + return chain; + } + if (table === 'broadcasts' || table === 'broadcast_recipients') { + calls.usedDirectInsert++; + return { + insert: () => ({ + select: () => ({ + single: () => + Promise.resolve({ data: { id: 'orphan' }, error: null }), + }), + }), + }; + } + throw new Error(`unexpected table: ${table}`); + }, + rpc(name: string, args: unknown) { + calls.rpc.push({ name, args }); + return Promise.resolve(rpcResult); + }, + } as unknown as SupabaseClient; + return { db: database, calls }; +} + +describe('createBroadcast atomicity (#370)', () => { + it('creates parent + recipients through the atomic RPC, never a bare parent insert', async () => { + const { db, calls } = makeDb({ + data: [{ broadcast_id: 'b-1', recipient_id: 'r-1', contact_id: 'c1' }], + error: null, + }); + + const plan = await createBroadcast(db, 'acc', 'user', { + templateName: 'promo', + recipients: [{ to: '+14155550123' }], + }); + + expect(calls.rpc).toHaveLength(1); + expect(calls.rpc[0].name).toBe('create_broadcast_with_recipients'); + expect(calls.usedDirectInsert).toBe(0); + expect(plan.broadcastId).toBe('b-1'); + expect(plan.planned).toEqual([ + { recipientRowId: 'r-1', phone: '14155550123', params: [] }, + ]); + }); + + it('throws and leaves no orphaned parent when the atomic create fails', async () => { + const { db, calls } = makeDb({ + data: null, + error: { message: 'recipient insert failed' }, + }); + + await expect( + createBroadcast(db, 'acc', 'user', { + templateName: 'promo', + recipients: [{ to: '+14155550123' }], + }) + ).rejects.toBeInstanceOf(BroadcastError); + + // The RPC was the only persistence attempt; because it runs both + // inserts in a single transaction, its failure rolls the parent back — + // there is no separate parent insert that could survive as an orphan. + expect(calls.rpc).toHaveLength(1); + expect(calls.usedDirectInsert).toBe(0); + }); +}); diff --git a/src/lib/whatsapp/broadcast-core.ts b/src/lib/whatsapp/broadcast-core.ts index 672ce6ada6..d0f08508ba 100644 --- a/src/lib/whatsapp/broadcast-core.ts +++ b/src/lib/whatsapp/broadcast-core.ts @@ -193,49 +193,44 @@ export async function createBroadcast( // recipient change). `rejected` phones have no recipient row, so they // are reported to the caller in the POST response, not in these // persisted counts. - const { data: broadcast, error: bErr } = await db - .from('broadcasts') - .insert({ - account_id: accountId, - user_id: auditUserId, - name: name || `API broadcast (${templateName})`, - template_name: templateName, - template_language: templateLanguage, - status: 'sending', - total_recipients: deduped.length, - }) - .select('id') - .single(); - if (bErr || !broadcast) { - console.error('[broadcast-core] create broadcast error:', bErr); + // Insert the parent broadcast and its recipient rows in ONE transaction + // (migration 037's create_broadcast_with_recipients). Previously these + // were two separate inserts: if the recipient insert failed, the parent + // was already persisted with status 'sending' and no recipients, leaving + // an orphaned campaign that looked like it was sending but had no + // delivery plan (issue #370). The function body is atomic, so a recipient + // failure now rolls the parent back and nothing orphaned survives. + const { data: createdRows, error: createErr } = await db.rpc( + 'create_broadcast_with_recipients', + { + p_account_id: accountId, + p_user_id: auditUserId, + p_name: name || `API broadcast (${templateName})`, + p_template_name: templateName, + p_template_language: templateLanguage, + p_total_recipients: deduped.length, + p_contact_ids: deduped.map((r) => r.contactId), + } + ); + if (createErr || !createdRows || createdRows.length === 0) { + console.error('[broadcast-core] create broadcast error:', createErr); throw new BroadcastError('internal', 'Failed to create broadcast', 500); } - const { data: recipientRows, error: rErr } = await db - .from('broadcast_recipients') - .insert( - deduped.map((r) => ({ - broadcast_id: broadcast.id, - contact_id: r.contactId, - status: 'pending' as const, - })) - ) - .select('id, contact_id'); - if (rErr || !recipientRows) { - console.error('[broadcast-core] create recipients error:', rErr); - throw new BroadcastError('internal', 'Failed to create broadcast', 500); - } + const broadcastId = createdRows[0].broadcast_id as string; // Pair each inserted recipient row back to its phone/params by // contact_id — unambiguous now that duplicates are collapsed. const byContact = new Map(deduped.map((r) => [r.contactId, r])); - const planned: PlannedRecipient[] = recipientRows.map((row) => { - const r = byContact.get(row.contact_id as string)!; - return { recipientRowId: row.id as string, phone: r.phone, params: r.params }; - }); + const planned: PlannedRecipient[] = createdRows.map( + (row: { recipient_id: string; contact_id: string }) => { + const r = byContact.get(row.contact_id)!; + return { recipientRowId: row.recipient_id, phone: r.phone, params: r.params }; + } + ); return { - broadcastId: broadcast.id, + broadcastId, templateName, templateLanguage, phoneNumberId: config.phone_number_id, diff --git a/supabase/migrations/037_webhook_broadcast_reliability.sql b/supabase/migrations/037_webhook_broadcast_reliability.sql new file mode 100644 index 0000000000..89a6192303 --- /dev/null +++ b/supabase/migrations/037_webhook_broadcast_reliability.sql @@ -0,0 +1,154 @@ +-- ============================================================ +-- 037_webhook_broadcast_reliability +-- +-- Three independent reliability fixes that all need a DB-level +-- guarantee the application layer can't provide on its own: +-- +-- #367 Inbound-webhook idempotency. Meta retries webhook +-- deliveries; an unconditional message INSERT persisted the +-- same inbound message twice and re-ran every downstream +-- side effect. A unique index on (conversation_id, +-- message_id) turns a replay into an ON CONFLICT no-op. +-- +-- #369 Concurrent inbound messages lost unread-count increments. +-- The webhook did a read-modify-write of unread_count, so +-- two concurrent deliveries for one conversation both read N +-- and both wrote N+1. Moved to a DB-side atomic increment +-- (mirrors migration 007's automation-counter fix). +-- +-- #370 Broadcast creation persisted the parent row before the +-- recipients, leaving an orphaned `sending` broadcast with +-- no recipients when the recipient insert failed. A single +-- function runs both inserts in one transaction, so a +-- recipient failure rolls the parent back. +-- +-- Idempotent — safe to re-run. +-- ============================================================ + +-- ============================================================ +-- #367 — inbound webhook idempotency +-- +-- The Meta message id is unique per receiving number, and a given +-- (account, contact) always resolves to the same conversation +-- (guaranteed by migration 036), so (conversation_id, message_id) +-- is the correct idempotency key — `message_id` alone is NOT +-- globally unique across phone numbers (see migration 009). +-- +-- A plain (non-partial) unique index is used deliberately so +-- PostgREST's `ON CONFLICT` arbiter inference works from the column +-- list alone. NULL `message_id`s (outbound rows mid-send, before the +-- Meta wamid lands) are treated as distinct by a standard unique +-- index, so they never collide with each other. +-- ============================================================ + +-- Collapse pre-existing duplicates (keep the earliest row per key — +-- it's the one whose downstream side effects already ran) so the +-- unique index can be created. Only rows with a non-NULL message_id +-- can collide. reply_to_message_id is ON DELETE SET NULL (migration +-- 009) and reactions cascade, so removing a strict duplicate is safe. +WITH ranked AS ( + SELECT id, + row_number() OVER ( + PARTITION BY conversation_id, message_id + ORDER BY created_at ASC, id ASC + ) AS rn + FROM messages + WHERE message_id IS NOT NULL +) +DELETE FROM messages m +USING ranked r +WHERE m.id = r.id + AND r.rn > 1; + +CREATE UNIQUE INDEX IF NOT EXISTS idx_messages_conversation_message_id + ON messages (conversation_id, message_id); + +-- ============================================================ +-- #369 — atomic unread-count increment on inbound +-- +-- Replaces the webhook's read-modify-write. The increment happens +-- entirely inside the UPDATE so concurrent inbound deliveries for +-- the same conversation can't lose each other's bump. Also refreshes +-- the last-message summary in the same statement (matching the old +-- code's semantics: last_message_at = now, not the Meta timestamp). +-- ============================================================ +CREATE OR REPLACE FUNCTION public.bump_conversation_on_inbound( + p_conversation_id UUID, + p_last_message_text TEXT +) +RETURNS VOID +LANGUAGE sql +SECURITY DEFINER +SET search_path = public +AS $$ + UPDATE conversations + SET unread_count = COALESCE(unread_count, 0) + 1, + last_message_text = p_last_message_text, + last_message_at = NOW(), + updated_at = NOW() + WHERE id = p_conversation_id; +$$; + +-- Only the service role (webhook) calls this. Lock everyone else out +-- so an authenticated user can't bump another account's unread count. +REVOKE ALL ON FUNCTION public.bump_conversation_on_inbound(UUID, TEXT) FROM PUBLIC; +REVOKE ALL ON FUNCTION public.bump_conversation_on_inbound(UUID, TEXT) FROM anon; +REVOKE ALL ON FUNCTION public.bump_conversation_on_inbound(UUID, TEXT) FROM authenticated; +GRANT EXECUTE ON FUNCTION public.bump_conversation_on_inbound(UUID, TEXT) TO service_role; + +-- ============================================================ +-- #370 — atomic broadcast creation +-- +-- Inserts the parent `broadcasts` row and all `broadcast_recipients` +-- rows in a single transaction (a function body is atomic), then +-- returns the created ids so the caller can build its send plan. If +-- the recipient insert fails, the parent insert rolls back and no +-- orphaned `sending` broadcast survives. +-- +-- Per-status count columns are intentionally NOT seeded — they're +-- owned by the aggregate trigger (migrations 003/005), same as the +-- previous application-side insert. +-- ============================================================ +CREATE OR REPLACE FUNCTION public.create_broadcast_with_recipients( + p_account_id UUID, + p_user_id UUID, + p_name TEXT, + p_template_name TEXT, + p_template_language TEXT, + p_total_recipients INTEGER, + p_contact_ids UUID[] +) +RETURNS TABLE(broadcast_id UUID, recipient_id UUID, contact_id UUID) +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = public +AS $$ +DECLARE + v_broadcast_id UUID; +BEGIN + INSERT INTO broadcasts ( + account_id, user_id, name, template_name, + template_language, status, total_recipients + ) + VALUES ( + p_account_id, p_user_id, p_name, p_template_name, + p_template_language, 'sending', p_total_recipients + ) + RETURNING id INTO v_broadcast_id; + + RETURN QUERY + WITH ins AS ( + INSERT INTO broadcast_recipients (broadcast_id, contact_id, status) + SELECT v_broadcast_id, cid, 'pending' + FROM unnest(p_contact_ids) AS cid + RETURNING id, contact_id + ) + SELECT v_broadcast_id, ins.id, ins.contact_id + FROM ins; +END; +$$; + +REVOKE ALL ON FUNCTION public.create_broadcast_with_recipients(UUID, UUID, TEXT, TEXT, TEXT, INTEGER, UUID[]) FROM PUBLIC; +REVOKE ALL ON FUNCTION public.create_broadcast_with_recipients(UUID, UUID, TEXT, TEXT, TEXT, INTEGER, UUID[]) FROM anon; +REVOKE ALL ON FUNCTION public.create_broadcast_with_recipients(UUID, UUID, TEXT, TEXT, TEXT, INTEGER, UUID[]) FROM authenticated; +GRANT EXECUTE ON FUNCTION public.create_broadcast_with_recipients(UUID, UUID, TEXT, TEXT, TEXT, INTEGER, UUID[]) TO service_role;