'use client'; import { useState } from 'react'; import { createClient } from '@/lib/supabase/client'; import { Contact, MessageTemplate } from '@/types'; export type CustomFieldOperator = 'is' | 'is_not' | 'contains'; export interface CustomFieldFilter { fieldId: string; operator: CustomFieldOperator; value: string; } export interface AudienceConfig { type: 'all' | 'tags' | 'custom_field' | 'csv'; tagIds?: string[]; customField?: CustomFieldFilter; csvContacts?: { phone: string; name?: string }[]; /** Contacts carrying any of these tags are subtracted from the result. */ excludeTagIds?: string[]; } /** * Variable mapping — each template placeholder (by key, usually "1", * "2", …) is resolved at send time. `field` maps to a built-in contact * field (name/phone/email/company); `custom_field` maps to a * contact_custom_values.value row keyed by the custom_fields.id stored * in `value`. */ export type VariableMapping = | { type: 'static'; value: string } | { type: 'field'; value: string } | { type: 'custom_field'; value: string }; interface BroadcastPayload { name: string; template: MessageTemplate; audience: AudienceConfig; variables: Record; } interface UseBroadcastSendingReturn { createAndSendBroadcast: (payload: BroadcastPayload) => Promise; isProcessing: boolean; progress: number; } /** * Meta rate-limit buffer. 10 per batch + 1 s pause matches the spec * and keeps us comfortably under 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. */ 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; } /** contactId → (customFieldId → value). */ type CustomValueIndex = Map>; /** * Per-contact resolution of custom-field placeholders. Static and * built-in-field mappings resolve synchronously; custom fields read * from a pre-built index to avoid N+1 queries during the send loop. */ export function resolveVariables( variables: Record, contact: Contact, customValues?: Map, ): string[] { // Keys are typically "1","2",... — 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; if (v.type === 'field') { const fieldMap: Record = { name: contact.name, phone: contact.phone, email: contact.email, company: contact.company, }; return fieldMap[v.value] ?? ''; } // custom_field return customValues?.get(v.value) ?? ''; }); } /** * Bulk-fetch contact_custom_values for a set of contacts. Returns an * index keyed by contact_id → field_id → value. */ async function fetchCustomValueIndex( supabase: ReturnType, contactIds: string[], ): Promise { const index: CustomValueIndex = new Map(); if (contactIds.length === 0) return index; // Supabase PostgREST caps the .in(...) IN-clause roughly at 1000 // values. Page through to stay safe. const PAGE = 500; for (let i = 0; i < contactIds.length; i += PAGE) { const slice = contactIds.slice(i, i + PAGE); const { data } = await supabase .from('contact_custom_values') .select('contact_id, custom_field_id, value') .in('contact_id', slice); for (const row of data ?? []) { const bucket = index.get(row.contact_id) ?? new Map(); bucket.set(row.custom_field_id, row.value ?? ''); index.set(row.contact_id, bucket); } } return index; } export function useBroadcastSending(): UseBroadcastSendingReturn { const [isProcessing, setIsProcessing] = useState(false); const [progress, setProgress] = useState(0); async function resolveAudience(audience: AudienceConfig): Promise { const supabase = createClient(); let contacts: Contact[] = []; if (audience.type === 'all') { const { data, error } = await supabase.from('contacts').select('*'); if (error) throw new Error(`Failed to fetch contacts: ${error.message}`); contacts = data ?? []; } else if ( audience.type === 'tags' && audience.tagIds && audience.tagIds.length > 0 ) { const { data: contactTags, error: tagError } = await supabase .from('contact_tags') .select('contact_id') .in('tag_id', audience.tagIds); if (tagError) throw new Error(`Failed to fetch contact tags: ${tagError.message}`); if (contactTags && contactTags.length > 0) { const uniqueContactIds = [ ...new Set(contactTags.map((ct) => ct.contact_id)), ]; const { data, error } = await supabase .from('contacts') .select('*') .in('id', uniqueContactIds); if (error) throw new Error(`Failed to fetch contacts: ${error.message}`); contacts = data ?? []; } } else if (audience.type === 'custom_field' && audience.customField) { contacts = await resolveCustomFieldAudience(supabase, audience.customField); } else if (audience.type === 'csv' && audience.csvContacts) { // CSV path — contacts are synthesized client-side. Note the // synthesized ids (`csv-N`) are NOT real DB ids; the send flow // still works because the broadcast_recipients insert uses // contact_id. For CSV we skip the recipient row path entirely // at the API boundary (out of scope of this refactor — keeping // prior behavior). return audience.csvContacts.map((c, i) => ({ id: `csv-${i}`, user_id: '', phone: c.phone, name: c.name, created_at: new Date().toISOString(), updated_at: new Date().toISOString(), })); } // Apply exclude tags (works across all contact-derived audience // types). CSV contacts are synthetic so exclusion doesn't apply. if (audience.excludeTagIds && audience.excludeTagIds.length > 0) { const { data: excludeRows } = await supabase .from('contact_tags') .select('contact_id') .in('tag_id', audience.excludeTagIds); const excludedIds = new Set((excludeRows ?? []).map((r) => r.contact_id)); contacts = contacts.filter((c) => !excludedIds.has(c.id)); } return contacts; } async function resolveCustomFieldAudience( supabase: ReturnType, filter: CustomFieldFilter, ): Promise { const { fieldId, operator, value } = filter; // Build the WHERE clause for the operator. PostgREST supports // eq/neq/ilike via the query builder — use ilike with wildcards // for "contains" so the match is case-insensitive. let query = supabase .from('contact_custom_values') .select('contact_id') .eq('custom_field_id', fieldId); if (operator === 'is') query = query.eq('value', value); else if (operator === 'is_not') query = query.neq('value', value); else if (operator === 'contains') query = query.ilike('value', `%${value}%`); const { data: matches, error: matchErr } = await query; if (matchErr) throw new Error(`Custom-field filter failed: ${matchErr.message}`); const contactIds = [...new Set((matches ?? []).map((m) => m.contact_id))]; if (contactIds.length === 0) return []; const { data, error } = await supabase .from('contacts') .select('*') .in('id', contactIds); if (error) throw new Error(`Failed to fetch contacts: ${error.message}`); return data ?? []; } async function createAndSendBroadcast(payload: BroadcastPayload): Promise { setIsProcessing(true); setProgress(0); const supabase = createClient(); try { // ── Step 1: Resolve audience contacts ───────────────────────── setProgress(5); const contacts = await resolveAudience(payload.audience); if (contacts.length === 0) { throw new Error('No contacts found for this audience.'); } // ── Step 2: Create broadcast row ────────────────────────────── setProgress(10); const { data: broadcast, error: broadcastError } = await supabase .from('broadcasts') .insert({ name: payload.name, template_name: payload.template.name, template_language: payload.template.language ?? 'en_US', template_variables: payload.variables, audience_filter: { type: payload.audience.type, tagIds: payload.audience.tagIds, customField: payload.audience.customField, excludeTagIds: payload.audience.excludeTagIds, }, status: 'sending', total_recipients: contacts.length, sent_count: 0, delivered_count: 0, read_count: 0, replied_count: 0, failed_count: 0, }) .select() .single(); if (broadcastError || !broadcast) { throw new Error(`Failed to create broadcast: ${broadcastError?.message}`); } // ── Step 3: Insert recipient rows ───────────────────────────── setProgress(20); 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 += INSERT_BATCH_SIZE) { const batch = recipientRows.slice(i, i + INSERT_BATCH_SIZE); const { error: recipientError } = await supabase .from('broadcast_recipients') .insert(batch); if (recipientError) { console.error('Failed to insert recipient batch:', recipientError); } } // ── Step 4: Fetch recipients (joined contact) + preload custom values setProgress(30); const { data: recipients, error: recipientsFetchError } = await supabase .from('broadcast_recipients') .select('*, contact:contacts(*)') .eq('broadcast_id', broadcast.id); if (recipientsFetchError || !recipients) { throw new Error('Failed to fetch broadcast recipients'); } // One bulk fetch of custom values for every contact in this // broadcast, avoiding N+1 during the send loop. const contactIds = recipients .map((r) => r.contact?.id) .filter((id): id is string => Boolean(id)); const customValueIndex = await fetchCustomValueIndex( supabase, contactIds, ); let failedCount = 0; const totalRecipients = recipients.length; for (let i = 0; i < recipients.length; i += SEND_BATCH_SIZE) { const batch = recipients.slice(i, i + SEND_BATCH_SIZE); const apiRecipients = batch .filter((r) => r.contact?.phone) .map((r) => ({ phone: r.contact!.phone as string, params: r.contact ? resolveVariables( payload.variables, r.contact, customValueIndex.get(r.contact.id), ) : [], })); if (apiRecipients.length === 0) continue; try { const res = await fetch('/api/whatsapp/broadcast', { method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ recipients: apiRecipients, template_name: payload.template.name, template_language: payload.template.language ?? 'en_US', }), }); const data = await res.json(); if (!res.ok) { throw new Error(data.error || 'Broadcast API request failed'); } const resultsByPhone = new Map(); for (const r of (data.results ?? []) as BroadcastApiResult[]) { resultsByPhone.set(r.phone, r); } for (const recipient of batch) { const phone = recipient.contact?.phone; const result = phone ? resultsByPhone.get(phone) : undefined; if (!result) { failedCount++; await supabase .from('broadcast_recipients') .update({ status: 'failed', error_message: 'No phone number on contact', }) .eq('id', recipient.id); continue; } if (result.status === 'sent') { 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) { for (const recipient of batch) { failedCount++; await supabase .from('broadcast_recipients') .update({ status: 'failed', error_message: err instanceof Error ? err.message : 'Unknown error', }) .eq('id', recipient.id); } } const progressPct = 30 + Math.round(((i + batch.length) / totalRecipients) * 60); setProgress(progressPct); if (i + SEND_BATCH_SIZE < recipients.length) { await sleep(SEND_BATCH_DELAY_MS); } } // ── Step 5: Finalize status ─────────────────────────────────── // Aggregate counts are maintained by the DB trigger (migration // 003); we only flip the final status here. setProgress(95); const finalStatus = failedCount === totalRecipients ? 'failed' : 'sent'; await supabase .from('broadcasts') .update({ status: finalStatus }) .eq('id', broadcast.id); setProgress(100); return broadcast.id; } finally { setIsProcessing(false); } } return { createAndSendBroadcast, isProcessing, progress }; }