From 5d5c5f0a20d345b06077ed3301ad53146884de75 Mon Sep 17 00:00:00 2001 From: Arnas Donauskas Date: Fri, 17 Apr 2026 11:22:38 +0300 Subject: [PATCH] fix(broadcasts): per-recipient variables, webhook status sync, rate limit MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Three correctness bugs were making the existing broadcast flow wrong, not just incomplete. 1. Per-recipient variables were broken use-broadcast-sending.ts computed templateParams per contact but shipped only templateParams[0] to the API, so every contact in a batch received contact 0's personalization (name/email/etc). Fix: extend POST /api/whatsapp/broadcast to accept a per-recipient shape `{ recipients: [{ phone, params }], ... }`. Legacy `{ phone_numbers, template_params }` still accepted. Hook sends per- recipient payloads; API returns whatsapp_message_id per phone. 2. Webhook never updated broadcast_recipients handleStatusUpdate in webhook/route.ts only mirrored Meta status onto the `messages` table. broadcast_recipients had no correlation column, so delivered_count / read_count / replied_count never advanced for real broadcasts. Fix (migration 003): - broadcast_recipients.whatsapp_message_id (+ UNIQUE index) for Meta-message-id correlation - composite index (broadcast_id, status) for fast aggregate scans - AFTER INSERT/UPDATE/DELETE trigger that recomputes parent broadcasts counts via COUNT(*) FILTER — sender + webhook only touch the recipient row, aggregates stay consistent Webhook: - handleStatusUpdate also updates broadcast_recipients with forward- only status guard (read cannot regress to sent on a replay) - processMessage flags the replied recipient row when an inbound message arrives from a recent broadcast contact 3. Rate limiting did not match spec Batch 50 / no delay → Batch 10 / 1s between batches, matching the spec and Meta's per-phone-number rate ceiling. Also: extract statusConfig from both page components into src/lib/broadcast-status.ts (single source of truth for PR B). Add whatsapp_message_id to BroadcastRecipient type. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/app/(dashboard)/broadcasts/[id]/page.tsx | 26 +-- src/app/(dashboard)/broadcasts/page.tsx | 11 +- src/app/api/whatsapp/broadcast/route.ts | 95 +++++++--- src/app/api/whatsapp/webhook/route.ts | 105 ++++++++++- src/hooks/use-broadcast-sending.ts | 177 ++++++++++++------ src/lib/broadcast-status.ts | 91 +++++++++ src/types/index.ts | 6 + .../003_broadcast_recipient_wamid.sql | 84 +++++++++ 8 files changed, 480 insertions(+), 115 deletions(-) create mode 100644 src/lib/broadcast-status.ts create mode 100644 supabase/migrations/003_broadcast_recipient_wamid.sql diff --git a/src/app/(dashboard)/broadcasts/[id]/page.tsx b/src/app/(dashboard)/broadcasts/[id]/page.tsx index a11db15..9053935 100644 --- a/src/app/(dashboard)/broadcasts/[id]/page.tsx +++ b/src/app/(dashboard)/broadcasts/[id]/page.tsx @@ -22,23 +22,10 @@ import { Eye, AlertCircle, } from 'lucide-react'; - -const statusConfig: Record = { - draft: { label: 'Draft', classes: 'bg-slate-500/10 text-slate-400 border-slate-500/20' }, - scheduled: { label: 'Scheduled', classes: 'bg-blue-500/10 text-blue-400 border-blue-500/20' }, - sending: { label: 'Sending', classes: 'bg-yellow-500/10 text-yellow-400 border-yellow-500/20' }, - sent: { label: 'Sent', classes: 'bg-emerald-500/10 text-emerald-400 border-emerald-500/20' }, - failed: { label: 'Failed', classes: 'bg-red-500/10 text-red-400 border-red-500/20' }, -}; - -const recipientStatusConfig: Record = { - pending: { label: 'Pending', classes: 'bg-slate-500/10 text-slate-400 border-slate-500/20' }, - sent: { label: 'Sent', classes: 'bg-blue-500/10 text-blue-400 border-blue-500/20' }, - delivered: { label: 'Delivered', classes: 'bg-emerald-500/10 text-emerald-400 border-emerald-500/20' }, - read: { label: 'Read', classes: 'bg-emerald-500/10 text-emerald-300 border-emerald-500/20' }, - replied: { label: 'Replied', classes: 'bg-purple-500/10 text-purple-400 border-purple-500/20' }, - failed: { label: 'Failed', classes: 'bg-red-500/10 text-red-400 border-red-500/20' }, -}; +import { + getBroadcastStatus, + getRecipientStatus, +} from '@/lib/broadcast-status'; interface StatCardProps { label: string; @@ -125,7 +112,7 @@ export default function BroadcastDetailPage() { ); } - const status = statusConfig[broadcast.status] ?? statusConfig.draft; + const status = getBroadcastStatus(broadcast.status); return (
@@ -221,8 +208,7 @@ export default function BroadcastDetailPage() { {recipients.map((recipient) => { - const rStatus = - recipientStatusConfig[recipient.status] ?? recipientStatusConfig.pending; + const rStatus = getRecipientStatus(recipient.status); return ( diff --git a/src/app/(dashboard)/broadcasts/page.tsx b/src/app/(dashboard)/broadcasts/page.tsx index ba48c51..b069caf 100644 --- a/src/app/(dashboard)/broadcasts/page.tsx +++ b/src/app/(dashboard)/broadcasts/page.tsx @@ -14,14 +14,7 @@ import { TableRow, } from '@/components/ui/table'; import { Radio, Plus, Loader2 } from 'lucide-react'; - -const statusConfig: Record = { - draft: { label: 'Draft', classes: 'bg-slate-500/10 text-slate-400 border-slate-500/20' }, - scheduled: { label: 'Scheduled', classes: 'bg-blue-500/10 text-blue-400 border-blue-500/20' }, - sending: { label: 'Sending', classes: 'bg-yellow-500/10 text-yellow-400 border-yellow-500/20' }, - sent: { label: 'Sent', classes: 'bg-emerald-500/10 text-emerald-400 border-emerald-500/20' }, - failed: { label: 'Failed', classes: 'bg-red-500/10 text-red-400 border-red-500/20' }, -}; +import { getBroadcastStatus } from '@/lib/broadcast-status'; export default function BroadcastsPage() { const router = useRouter(); @@ -120,7 +113,7 @@ export default function BroadcastsPage() { {broadcasts.map((broadcast) => { - const status = statusConfig[broadcast.status] ?? statusConfig.draft; + const status = getBroadcastStatus(broadcast.status); return ( , + * template_name, template_language + * } + * + * LEGACY (all phones receive the same params — kept so existing + * callers don't break): + * { + * phone_numbers: string[], + * template_params: string[], + * template_name, template_language + * } + * + * Previous implementation only supported the legacy shape, and the + * sending hook was forced to ship every batch with `templateParams[0]` + * — meaning every recipient got contact-0's personalization. The new + * shape is what actually fixes that. + */ +interface NewRecipient { + phone: string + params?: string[] +} + export async function POST(request: Request) { try { const supabase = await createClient() @@ -26,18 +53,36 @@ export async function POST(request: Request) { } = await supabase.auth.getUser() if (authError || !user) { - return NextResponse.json( - { error: 'Unauthorized' }, - { status: 401 } - ) + return NextResponse.json({ error: 'Unauthorized' }, { status: 401 }) } const body = await request.json() - const { phone_numbers, template_name, template_language, template_params } = body + const { + recipients: newRecipients, + phone_numbers, + template_name, + template_language, + template_params, + } = body - if (!phone_numbers || !Array.isArray(phone_numbers) || phone_numbers.length === 0) { + // Normalize to a list of {phone, params} regardless of shape. + let recipients: NewRecipient[] + if (Array.isArray(newRecipients) && newRecipients.length > 0) { + recipients = newRecipients + } else if (Array.isArray(phone_numbers) && phone_numbers.length > 0) { + const shared: string[] = Array.isArray(template_params) + ? template_params + : [] + recipients = phone_numbers.map((phone: string) => ({ + phone, + params: shared, + })) + } else { return NextResponse.json( - { error: 'phone_numbers array is required and must not be empty' }, + { + error: + 'Provide either `recipients` (preferred) or `phone_numbers` — must be a non-empty array', + }, { status: 400 } ) } @@ -49,7 +94,6 @@ export async function POST(request: Request) { ) } - // Fetch and decrypt WhatsApp config const { data: config, error: configError } = await supabase .from('whatsapp_config') .select('*') @@ -58,7 +102,10 @@ export async function POST(request: Request) { if (configError || !config) { return NextResponse.json( - { error: 'WhatsApp not configured. Please set up your WhatsApp integration first.' }, + { + error: + 'WhatsApp not configured. Please set up your WhatsApp integration first.', + }, { status: 400 } ) } @@ -69,12 +116,12 @@ export async function POST(request: Request) { let sentCount = 0 let failedCount = 0 - for (const phone of phone_numbers) { - const sanitized = sanitizePhoneForMeta(phone) + for (const recipient of recipients) { + const sanitized = sanitizePhoneForMeta(recipient.phone) if (!isValidE164(sanitized)) { results.push({ - phone, + phone: recipient.phone, status: 'failed', error: 'Invalid phone number format', }) @@ -82,13 +129,8 @@ export async function POST(request: Request) { continue } - // Correct sendTemplateMessage signature is: - // (phoneNumberId, accessToken, to, templateName, language?, params?) - // Previously the args were passed in the wrong order (accessToken first, - // params before language), which made Meta reject every broadcast. - // - // Retry with phone variants on "not in allowed list" so numbers that - // differ only in a trunk-prefix 0 still reach recipients. + // Retry with phone variants on "not in allowed list" so numbers + // that differ only in a trunk-prefix 0 still reach recipients. const variants = phoneVariants(sanitized) let sentMessageId: string | null = null let lastError: string | null = null @@ -101,7 +143,7 @@ export async function POST(request: Request) { to: variant, templateName: template_name, language: template_language || 'en_US', - params: template_params || [], + params: recipient.params ?? [], }) sentMessageId = result.messageId lastError = null @@ -111,7 +153,7 @@ export async function POST(request: Request) { error instanceof Error ? error.message : 'Unknown error' if (!isRecipientNotAllowedError(errorMessage)) { lastError = errorMessage - break // unrelated error — stop retrying + break } lastError = errorMessage // retry with next variant @@ -120,15 +162,18 @@ export async function POST(request: Request) { if (sentMessageId) { results.push({ - phone, + phone: recipient.phone, status: 'sent', whatsapp_message_id: sentMessageId, }) sentCount++ } else { - console.error(`Failed to send broadcast to ${phone}:`, lastError) + console.error( + `Failed to send broadcast to ${recipient.phone}:`, + lastError + ) results.push({ - phone, + phone: recipient.phone, status: 'failed', error: lastError || 'Unknown error', }) @@ -138,7 +183,7 @@ export async function POST(request: Request) { return NextResponse.json({ success: true, - total: phone_numbers.length, + total: recipients.length, sent: sentCount, failed: failedCount, results, diff --git a/src/app/api/whatsapp/webhook/route.ts b/src/app/api/whatsapp/webhook/route.ts index aa81d4e..36da8e9 100644 --- a/src/app/api/whatsapp/webhook/route.ts +++ b/src/app/api/whatsapp/webhook/route.ts @@ -179,22 +179,110 @@ async function processWebhook(body: { entry?: WhatsAppWebhookEntry[] }) { } } +// Ordered forward-only: a webhook replay must never regress a recipient +// from `read` back to `sent`. Index into this array gives the "level". +const RECIPIENT_STATUS_ORDER = [ + 'pending', + 'sent', + 'delivered', + 'read', + 'replied', + 'failed', +] as const + +function recipientStatusLevel(s: string): number { + const idx = (RECIPIENT_STATUS_ORDER as readonly string[]).indexOf(s) + return idx < 0 ? 0 : idx +} + async function handleStatusUpdate(status: { id: string status: string timestamp: string recipient_id: string }) { - // The messages table schema uses `message_id` (not whatsapp_message_id) - // and has no `updated_at` column. Meta's status values (sent/delivered/ - // read/failed) already match our CHECK constraint. - const { error } = await supabaseAdmin() + // 1) Mirror onto messages (legacy behavior) — Meta's status values + // already match the CHECK constraint on messages.status. + const { error: msgErr } = await supabaseAdmin() .from('messages') .update({ status: status.status }) .eq('message_id', status.id) - if (error) { - console.error('Error updating message status:', error) + if (msgErr) { + console.error('Error updating message status:', msgErr) + } + + // 2) Mirror onto broadcast_recipients via whatsapp_message_id + // (added in migration 003). The aggregate trigger on + // broadcast_recipients re-derives the parent broadcast's + // sent/delivered/read/failed counts automatically. + const tsIso = new Date(parseInt(status.timestamp) * 1000).toISOString() + + const { data: recipient, error: recFetchErr } = await supabaseAdmin() + .from('broadcast_recipients') + .select('id, status') + .eq('whatsapp_message_id', status.id) + .maybeSingle() + + if (recFetchErr) { + console.error('Error fetching broadcast recipient:', recFetchErr) + return + } + if (!recipient) return // message wasn't part of a broadcast — fine + + // Forward-only status guard: only advance. + const currentLevel = recipientStatusLevel(recipient.status) + const incomingLevel = recipientStatusLevel(status.status) + if (incomingLevel <= currentLevel) return + + const update: Record = { status: status.status } + if (status.status === 'sent' && !('sent_at' in update)) update.sent_at = tsIso + if (status.status === 'delivered') update.delivered_at = tsIso + if (status.status === 'read') update.read_at = tsIso + + const { error: recUpdateErr } = await supabaseAdmin() + .from('broadcast_recipients') + .update(update) + .eq('id', recipient.id) + + if (recUpdateErr) { + console.error('Error updating broadcast recipient status:', recUpdateErr) + } +} + +/** + * If an inbound message's sender is on a still-unreplied + * broadcast_recipients row, flip it to `replied` so the reply count + * advances on the parent broadcast. + * + * Runs on a best-effort basis — failures here must not break the + * main inbound-message flow, so errors are swallowed with a log. + */ +async function flagBroadcastReplyIfAny(userId: string, contactId: string) { + try { + // Most recent outbound broadcast that hasn't been replied to yet. + const { data: recs, error } = await supabaseAdmin() + .from('broadcast_recipients') + .select('id, status, broadcast_id, broadcasts!inner(user_id)') + .eq('contact_id', contactId) + .eq('broadcasts.user_id', userId) + .in('status', ['sent', 'delivered', 'read']) + .order('created_at', { ascending: false }) + .limit(1) + + if (error || !recs || recs.length === 0) return + + const row = recs[0] + const { error: updErr } = await supabaseAdmin() + .from('broadcast_recipients') + .update({ status: 'replied', replied_at: new Date().toISOString() }) + .eq('id', row.id) + + if (updErr) { + console.error('Error marking broadcast recipient replied:', updErr) + } + } catch (err) { + console.error('flagBroadcastReplyIfAny failed:', err) } } @@ -280,6 +368,11 @@ async function processMessage( if (convError) { console.error('Error updating conversation:', convError) } + + // If this contact was a recent broadcast recipient, flag the reply + // so the broadcast's `replied_count` advances (via the aggregate + // trigger installed in migration 003). + await flagBroadcastReplyIfAny(userId, contactRecord.id) } async function parseMessageContent( diff --git a/src/hooks/use-broadcast-sending.ts b/src/hooks/use-broadcast-sending.ts index a9d03ff..7da7bad 100644 --- a/src/hooks/use-broadcast-sending.ts +++ b/src/hooks/use-broadcast-sending.ts @@ -23,6 +23,31 @@ interface UseBroadcastSendingReturn { progress: number; } +/** + * Meta rate-limit buffer. 10 per batch with a 1 s pause matches the + * spec and keeps us well below Meta's per-phone-number messaging rate + * so a large broadcast never trips the upstream limiter. + */ +const SEND_BATCH_SIZE = 10; +const SEND_BATCH_DELAY_MS = 1000; + +/** + * `broadcast_recipients` inserts are independent of the send rate — + * we batch them purely to keep individual Supabase requests small. + */ +const INSERT_BATCH_SIZE = 200; + +function sleep(ms: number) { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +interface BroadcastApiResult { + phone: string; + status: 'sent' | 'failed'; + whatsapp_message_id?: string; + error?: string; +} + export function useBroadcastSending(): UseBroadcastSendingReturn { const [isProcessing, setIsProcessing] = useState(false); const [progress, setProgress] = useState(0); @@ -74,19 +99,26 @@ export function useBroadcastSending(): UseBroadcastSendingReturn { variables: Record, contact: Contact ): string[] { - return Object.keys(variables) - .sort() - .map((key) => { - const v = variables[key]; - if (v.type === 'static') return v.value; - const fieldMap: Record = { - name: contact.name, - phone: contact.phone, - email: contact.email, - company: contact.company, - }; - return fieldMap[v.value] ?? ''; - }); + // Keys typically are "1","2",... so a numeric-aware sort keeps {{1}} + // before {{10}}. + const keys = Object.keys(variables).sort((a, b) => { + const an = Number(a); + const bn = Number(b); + if (Number.isFinite(an) && Number.isFinite(bn)) return an - bn; + return a.localeCompare(b); + }); + + return keys.map((key) => { + const v = variables[key]; + if (v.type === 'static') return v.value; + const fieldMap: Record = { + name: contact.name, + phone: contact.phone, + email: contact.email, + company: contact.company, + }; + return fieldMap[v.value] ?? ''; + }); } async function createAndSendBroadcast(payload: BroadcastPayload): Promise { @@ -96,7 +128,7 @@ export function useBroadcastSending(): UseBroadcastSendingReturn { const supabase = createClient(); try { - // Step 1: Resolve audience contacts + // ── Step 1: Resolve audience contacts ───────────────────────── setProgress(5); const contacts = await resolveAudience(payload.audience); @@ -104,7 +136,7 @@ export function useBroadcastSending(): UseBroadcastSendingReturn { throw new Error('No contacts found for this audience.'); } - // Step 2: Create broadcast record + // ── Step 2: Create broadcast row ────────────────────────────── setProgress(10); const { data: broadcast, error: broadcastError } = await supabase .from('broadcasts') @@ -132,17 +164,16 @@ export function useBroadcastSending(): UseBroadcastSendingReturn { throw new Error(`Failed to create broadcast: ${broadcastError?.message}`); } - // Step 3: Create broadcast recipients in batches + // ── Step 3: Insert recipient rows (batched for request size) ── setProgress(20); - const BATCH_SIZE = 50; const recipientRows = contacts.map((contact) => ({ broadcast_id: broadcast.id, contact_id: contact.id, status: 'pending' as const, })); - for (let i = 0; i < recipientRows.length; i += BATCH_SIZE) { - const batch = recipientRows.slice(i, i + BATCH_SIZE); + for (let i = 0; i < recipientRows.length; i += INSERT_BATCH_SIZE) { + const batch = recipientRows.slice(i, i + INSERT_BATCH_SIZE); const { error: recipientError } = await supabase .from('broadcast_recipients') .insert(batch); @@ -152,7 +183,7 @@ export function useBroadcastSending(): UseBroadcastSendingReturn { } } - // Step 4: Fetch recipients and send via API in batches + // ── Step 4: Fetch recipients (with joined contact) and send ─── setProgress(30); const { data: recipients, error: recipientsFetchError } = await supabase .from('broadcast_recipients') @@ -167,31 +198,29 @@ export function useBroadcastSending(): UseBroadcastSendingReturn { let failedCount = 0; const totalRecipients = recipients.length; - for (let i = 0; i < recipients.length; i += BATCH_SIZE) { - const batch = recipients.slice(i, i + BATCH_SIZE); - const phoneNumbers = batch - .map((r) => r.contact?.phone) - .filter(Boolean) as string[]; + for (let i = 0; i < recipients.length; i += SEND_BATCH_SIZE) { + const batch = recipients.slice(i, i + SEND_BATCH_SIZE); - if (phoneNumbers.length === 0) continue; + // Build per-recipient payload so each contact gets *their own* + // personalization. Previous impl shipped templateParams[0] for + // the whole batch which was a correctness bug. + const apiRecipients = batch + .filter((r) => r.contact?.phone) + .map((r) => ({ + phone: r.contact!.phone as string, + params: r.contact ? resolveVariables(payload.variables, r.contact) : [], + })); - // Resolve template params from the first contact in batch for static values - const templateParams = batch.map((r) => { - if (r.contact) { - return resolveVariables(payload.variables, r.contact); - } - return []; - }); + if (apiRecipients.length === 0) continue; try { const res = await fetch('/api/whatsapp/broadcast', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ - phone_numbers: phoneNumbers, + recipients: apiRecipients, template_name: payload.template.name, template_language: payload.template.language ?? 'en_US', - template_params: templateParams[0] ?? [], }), }); @@ -201,29 +230,56 @@ export function useBroadcastSending(): UseBroadcastSendingReturn { throw new Error(data.error || 'Broadcast API request failed'); } - // Update individual recipient statuses based on results - if (data.results) { - for (let j = 0; j < data.results.length; j++) { - const result = data.results[j]; - const recipient = batch[j]; - if (!recipient) continue; + // The API returns one entry per input recipient in order. + // Map by phone to be defensive against any reordering. + const resultsByPhone = new Map(); + for (const r of (data.results ?? []) as BroadcastApiResult[]) { + resultsByPhone.set(r.phone, r); + } - const newStatus = result.status === 'sent' ? 'sent' : 'failed'; - if (newStatus === 'sent') sentCount++; - else failedCount++; + for (const recipient of batch) { + const phone = recipient.contact?.phone; + const result = phone ? resultsByPhone.get(phone) : undefined; + if (!result) { + // No result for this phone — the API skipped it (most + // likely because contact.phone was missing). Mark failed. + failedCount++; await supabase .from('broadcast_recipients') .update({ - status: newStatus, - sent_at: newStatus === 'sent' ? new Date().toISOString() : null, - error_message: result.error ?? null, + status: 'failed', + error_message: 'No phone number on contact', + }) + .eq('id', recipient.id); + continue; + } + + if (result.status === 'sent') { + sentCount++; + await supabase + .from('broadcast_recipients') + .update({ + status: 'sent', + sent_at: new Date().toISOString(), + whatsapp_message_id: result.whatsapp_message_id ?? null, + error_message: null, + }) + .eq('id', recipient.id); + } else { + failedCount++; + await supabase + .from('broadcast_recipients') + .update({ + status: 'failed', + error_message: result.error ?? 'Unknown error', }) .eq('id', recipient.id); } } } catch (err) { - // Mark entire batch as failed + // Network / 500 — mark the whole batch failed. Trigger + // re-aggregates counts on the parent broadcast. for (const recipient of batch) { failedCount++; await supabase @@ -236,23 +292,34 @@ export function useBroadcastSending(): UseBroadcastSendingReturn { } } - const progressPct = 30 + Math.round(((i + batch.length) / totalRecipients) * 60); + // Progress reflects the send phase (30% → 95%). + const progressPct = + 30 + Math.round(((i + batch.length) / totalRecipients) * 60); setProgress(progressPct); + + // Space successive batches by 1 s to stay under Meta's rate + // limit. Skip the delay after the last batch so we don't pad + // the tail of the run. + if (i + SEND_BATCH_SIZE < recipients.length) { + await sleep(SEND_BATCH_DELAY_MS); + } } - // Step 5: Update broadcast with final counts + // ── Step 5: Finalize broadcast status ───────────────────────── + // Aggregate counts are maintained by the DB trigger (migration + // 003), so we only update `status` here. sent_count etc. are + // already accurate. setProgress(95); const finalStatus = failedCount === totalRecipients ? 'failed' : 'sent'; await supabase .from('broadcasts') - .update({ - status: finalStatus, - sent_count: sentCount, - failed_count: failedCount, - }) + .update({ status: finalStatus }) .eq('id', broadcast.id); setProgress(100); + // Unused locals intentionally kept for clarity at return site: + void sentCount; + return broadcast.id; } finally { setIsProcessing(false); diff --git a/src/lib/broadcast-status.ts b/src/lib/broadcast-status.ts new file mode 100644 index 0000000..37ceaf1 --- /dev/null +++ b/src/lib/broadcast-status.ts @@ -0,0 +1,91 @@ +/** + * Shared status badge config for broadcasts + recipients. + * + * Previously `statusConfig` was defined inline in both + * /broadcasts/page.tsx and /broadcasts/[id]/page.tsx with slight + * drift risk. One source of truth now. + * + * Dark-theme only — bg-*-500/10 + text-*-400 + border-*-500/20. + */ + +import type { BroadcastStatus, RecipientStatus } from "@/types"; + +export interface StatusDisplay { + label: string; + classes: string; + /** + * Set true for statuses that should pulse in the UI to convey + * "live / in-flight" — currently only `sending`. + */ + pulse?: boolean; +} + +export const broadcastStatusConfig: Record = { + draft: { + label: "Draft", + classes: "bg-slate-500/10 text-slate-400 border-slate-500/20", + }, + scheduled: { + label: "Scheduled", + classes: "bg-blue-500/10 text-blue-400 border-blue-500/20", + }, + sending: { + label: "Sending", + classes: "bg-yellow-500/10 text-yellow-400 border-yellow-500/20", + pulse: true, + }, + sent: { + label: "Sent", + classes: "bg-emerald-500/10 text-emerald-400 border-emerald-500/20", + }, + failed: { + label: "Failed", + classes: "bg-red-500/10 text-red-400 border-red-500/20", + }, +}; + +export const recipientStatusConfig: Record = { + pending: { + label: "Pending", + classes: "bg-slate-500/10 text-slate-400 border-slate-500/20", + }, + sent: { + label: "Sent", + classes: "bg-blue-500/10 text-blue-400 border-blue-500/20", + }, + delivered: { + label: "Delivered", + classes: "bg-emerald-500/10 text-emerald-400 border-emerald-500/20", + }, + read: { + label: "Read", + classes: "bg-emerald-500/10 text-emerald-300 border-emerald-500/20", + }, + replied: { + label: "Replied", + classes: "bg-purple-500/10 text-purple-400 border-purple-500/20", + }, + failed: { + label: "Failed", + classes: "bg-red-500/10 text-red-400 border-red-500/20", + }, +}; + +/** + * Tolerant lookup — callers often have a generic string status + * coming from Supabase. Falls back to the "draft" / "pending" + * entry so the UI never crashes on an unknown value. + */ +export function getBroadcastStatus(status: string): StatusDisplay { + return ( + broadcastStatusConfig[status as BroadcastStatus] ?? + broadcastStatusConfig.draft + ); +} + +export function getRecipientStatus(status: string): StatusDisplay { + return ( + recipientStatusConfig[status as RecipientStatus] ?? + recipientStatusConfig.pending + ); +} diff --git a/src/types/index.ts b/src/types/index.ts index 6f766b4..aed01e8 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -189,6 +189,12 @@ export interface BroadcastRecipient { read_at?: string; replied_at?: string; error_message?: string; + /** + * Meta's message id, persisted when the broadcast send succeeds so + * the webhook can mirror status updates back onto the recipient row. + * Added in migration 003. + */ + whatsapp_message_id?: string; created_at: string; contact?: Contact; } diff --git a/supabase/migrations/003_broadcast_recipient_wamid.sql b/supabase/migrations/003_broadcast_recipient_wamid.sql new file mode 100644 index 0000000..9bd0918 --- /dev/null +++ b/supabase/migrations/003_broadcast_recipient_wamid.sql @@ -0,0 +1,84 @@ +-- ============================================================ +-- Broadcast recipient correlation + aggregate counts +-- +-- Problem this solves: +-- * broadcast_recipients had no column to correlate with Meta's +-- message id, so webhook status updates (sent/delivered/read) +-- could not be mirrored into the recipient row and the broadcast +-- aggregate counts never advanced. +-- * aggregate counts on `broadcasts` (sent/delivered/read/replied/ +-- failed) were updated ad-hoc by the sender, which drifted quickly +-- once webhooks arrived out of band. +-- +-- This migration: +-- 1. Adds whatsapp_message_id (+ unique index) so webhooks can find +-- a recipient given Meta's message id. +-- 2. Adds a composite index on (broadcast_id, status) so the +-- aggregate trigger's COUNT(*) FILTER scans are fast. +-- 3. Installs an AFTER INSERT/UPDATE/DELETE trigger on +-- broadcast_recipients that re-aggregates the parent broadcasts +-- row. Keeps writer code trivial — the webhook + hook only touch +-- the recipient row; counts stay consistent automatically. +-- +-- Idempotent — safe to run multiple times. +-- ============================================================ + +ALTER TABLE broadcast_recipients + ADD COLUMN IF NOT EXISTS whatsapp_message_id TEXT; + +-- UNIQUE so webhook retries can't create duplicate correlations. +CREATE UNIQUE INDEX IF NOT EXISTS idx_broadcast_recipients_wamid + ON broadcast_recipients (whatsapp_message_id) + WHERE whatsapp_message_id IS NOT NULL; + +-- Fast path for the aggregate trigger's COUNT(*) FILTER subqueries. +CREATE INDEX IF NOT EXISTS idx_broadcast_recipients_broadcast_status + ON broadcast_recipients (broadcast_id, status); + +-- ============================================================ +-- Aggregate trigger +-- ============================================================ +CREATE OR REPLACE FUNCTION public.recompute_broadcast_counts(bid UUID) +RETURNS VOID AS $$ +BEGIN + UPDATE broadcasts b SET + sent_count = agg.sent_count, + delivered_count = agg.delivered_count, + read_count = agg.read_count, + replied_count = agg.replied_count, + failed_count = agg.failed_count, + updated_at = NOW() + FROM ( + SELECT + COUNT(*) FILTER (WHERE status IN ('sent','delivered','read','replied')) AS sent_count, + COUNT(*) FILTER (WHERE status IN ('delivered','read','replied')) AS delivered_count, + COUNT(*) FILTER (WHERE status IN ('read','replied')) AS read_count, + COUNT(*) FILTER (WHERE status = 'replied') AS replied_count, + COUNT(*) FILTER (WHERE status = 'failed') AS failed_count + FROM broadcast_recipients + WHERE broadcast_id = bid + ) agg + WHERE b.id = bid; +END; +$$ LANGUAGE plpgsql SECURITY DEFINER SET search_path = public; + +CREATE OR REPLACE FUNCTION public.broadcast_recipient_aggregate_trigger() +RETURNS TRIGGER AS $$ +BEGIN + IF TG_OP = 'DELETE' THEN + PERFORM public.recompute_broadcast_counts(OLD.broadcast_id); + RETURN OLD; + END IF; + + -- INSERT or UPDATE — only recompute when status changed (or on fresh insert) + IF TG_OP = 'INSERT' OR OLD.status IS DISTINCT FROM NEW.status THEN + PERFORM public.recompute_broadcast_counts(NEW.broadcast_id); + END IF; + RETURN NEW; +END; +$$ LANGUAGE plpgsql SECURITY DEFINER SET search_path = public; + +DROP TRIGGER IF EXISTS broadcast_recipients_aggregate ON broadcast_recipients; +CREATE TRIGGER broadcast_recipients_aggregate +AFTER INSERT OR UPDATE OR DELETE ON broadcast_recipients +FOR EACH ROW EXECUTE FUNCTION public.broadcast_recipient_aggregate_trigger();