mirror of
https://github.com/Luchexx/wacrm.git
synced 2026-11-02 09:19:51 +00:00
fix(broadcasts): per-recipient variables, webhook status sync, rate limit
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) <noreply@anthropic.com>
This commit is contained in:
parent
f746299df3
commit
5d5c5f0a20
@ -22,23 +22,10 @@ import {
|
|||||||
Eye,
|
Eye,
|
||||||
AlertCircle,
|
AlertCircle,
|
||||||
} from 'lucide-react';
|
} from 'lucide-react';
|
||||||
|
import {
|
||||||
const statusConfig: Record<string, { label: string; classes: string }> = {
|
getBroadcastStatus,
|
||||||
draft: { label: 'Draft', classes: 'bg-slate-500/10 text-slate-400 border-slate-500/20' },
|
getRecipientStatus,
|
||||||
scheduled: { label: 'Scheduled', classes: 'bg-blue-500/10 text-blue-400 border-blue-500/20' },
|
} from '@/lib/broadcast-status';
|
||||||
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<string, { label: string; classes: string }> = {
|
|
||||||
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' },
|
|
||||||
};
|
|
||||||
|
|
||||||
interface StatCardProps {
|
interface StatCardProps {
|
||||||
label: string;
|
label: string;
|
||||||
@ -125,7 +112,7 @@ export default function BroadcastDetailPage() {
|
|||||||
);
|
);
|
||||||
}
|
}
|
||||||
|
|
||||||
const status = statusConfig[broadcast.status] ?? statusConfig.draft;
|
const status = getBroadcastStatus(broadcast.status);
|
||||||
|
|
||||||
return (
|
return (
|
||||||
<div className="space-y-6">
|
<div className="space-y-6">
|
||||||
@ -221,8 +208,7 @@ export default function BroadcastDetailPage() {
|
|||||||
</TableHeader>
|
</TableHeader>
|
||||||
<TableBody>
|
<TableBody>
|
||||||
{recipients.map((recipient) => {
|
{recipients.map((recipient) => {
|
||||||
const rStatus =
|
const rStatus = getRecipientStatus(recipient.status);
|
||||||
recipientStatusConfig[recipient.status] ?? recipientStatusConfig.pending;
|
|
||||||
|
|
||||||
return (
|
return (
|
||||||
<TableRow key={recipient.id} className="border-slate-800">
|
<TableRow key={recipient.id} className="border-slate-800">
|
||||||
|
|||||||
@ -14,14 +14,7 @@ import {
|
|||||||
TableRow,
|
TableRow,
|
||||||
} from '@/components/ui/table';
|
} from '@/components/ui/table';
|
||||||
import { Radio, Plus, Loader2 } from 'lucide-react';
|
import { Radio, Plus, Loader2 } from 'lucide-react';
|
||||||
|
import { getBroadcastStatus } from '@/lib/broadcast-status';
|
||||||
const statusConfig: Record<string, { label: string; classes: string }> = {
|
|
||||||
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' },
|
|
||||||
};
|
|
||||||
|
|
||||||
export default function BroadcastsPage() {
|
export default function BroadcastsPage() {
|
||||||
const router = useRouter();
|
const router = useRouter();
|
||||||
@ -120,7 +113,7 @@ export default function BroadcastsPage() {
|
|||||||
</TableHeader>
|
</TableHeader>
|
||||||
<TableBody>
|
<TableBody>
|
||||||
{broadcasts.map((broadcast) => {
|
{broadcasts.map((broadcast) => {
|
||||||
const status = statusConfig[broadcast.status] ?? statusConfig.draft;
|
const status = getBroadcastStatus(broadcast.status);
|
||||||
return (
|
return (
|
||||||
<TableRow
|
<TableRow
|
||||||
key={broadcast.id}
|
key={broadcast.id}
|
||||||
|
|||||||
@ -16,6 +16,33 @@ interface BroadcastResult {
|
|||||||
error?: string
|
error?: string
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Two input shapes are accepted:
|
||||||
|
*
|
||||||
|
* NEW (preferred — supports per-recipient variable substitution):
|
||||||
|
* {
|
||||||
|
* recipients: Array<{ phone: string; params: string[] }>,
|
||||||
|
* 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) {
|
export async function POST(request: Request) {
|
||||||
try {
|
try {
|
||||||
const supabase = await createClient()
|
const supabase = await createClient()
|
||||||
@ -26,18 +53,36 @@ export async function POST(request: Request) {
|
|||||||
} = await supabase.auth.getUser()
|
} = await supabase.auth.getUser()
|
||||||
|
|
||||||
if (authError || !user) {
|
if (authError || !user) {
|
||||||
return NextResponse.json(
|
return NextResponse.json({ error: 'Unauthorized' }, { status: 401 })
|
||||||
{ error: 'Unauthorized' },
|
|
||||||
{ status: 401 }
|
|
||||||
)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
const body = await request.json()
|
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(
|
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 }
|
{ status: 400 }
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@ -49,7 +94,6 @@ export async function POST(request: Request) {
|
|||||||
)
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Fetch and decrypt WhatsApp config
|
|
||||||
const { data: config, error: configError } = await supabase
|
const { data: config, error: configError } = await supabase
|
||||||
.from('whatsapp_config')
|
.from('whatsapp_config')
|
||||||
.select('*')
|
.select('*')
|
||||||
@ -58,7 +102,10 @@ export async function POST(request: Request) {
|
|||||||
|
|
||||||
if (configError || !config) {
|
if (configError || !config) {
|
||||||
return NextResponse.json(
|
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 }
|
{ status: 400 }
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@ -69,12 +116,12 @@ export async function POST(request: Request) {
|
|||||||
let sentCount = 0
|
let sentCount = 0
|
||||||
let failedCount = 0
|
let failedCount = 0
|
||||||
|
|
||||||
for (const phone of phone_numbers) {
|
for (const recipient of recipients) {
|
||||||
const sanitized = sanitizePhoneForMeta(phone)
|
const sanitized = sanitizePhoneForMeta(recipient.phone)
|
||||||
|
|
||||||
if (!isValidE164(sanitized)) {
|
if (!isValidE164(sanitized)) {
|
||||||
results.push({
|
results.push({
|
||||||
phone,
|
phone: recipient.phone,
|
||||||
status: 'failed',
|
status: 'failed',
|
||||||
error: 'Invalid phone number format',
|
error: 'Invalid phone number format',
|
||||||
})
|
})
|
||||||
@ -82,13 +129,8 @@ export async function POST(request: Request) {
|
|||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
|
|
||||||
// Correct sendTemplateMessage signature is:
|
// Retry with phone variants on "not in allowed list" so numbers
|
||||||
// (phoneNumberId, accessToken, to, templateName, language?, params?)
|
// that differ only in a trunk-prefix 0 still reach recipients.
|
||||||
// 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.
|
|
||||||
const variants = phoneVariants(sanitized)
|
const variants = phoneVariants(sanitized)
|
||||||
let sentMessageId: string | null = null
|
let sentMessageId: string | null = null
|
||||||
let lastError: string | null = null
|
let lastError: string | null = null
|
||||||
@ -101,7 +143,7 @@ export async function POST(request: Request) {
|
|||||||
to: variant,
|
to: variant,
|
||||||
templateName: template_name,
|
templateName: template_name,
|
||||||
language: template_language || 'en_US',
|
language: template_language || 'en_US',
|
||||||
params: template_params || [],
|
params: recipient.params ?? [],
|
||||||
})
|
})
|
||||||
sentMessageId = result.messageId
|
sentMessageId = result.messageId
|
||||||
lastError = null
|
lastError = null
|
||||||
@ -111,7 +153,7 @@ export async function POST(request: Request) {
|
|||||||
error instanceof Error ? error.message : 'Unknown error'
|
error instanceof Error ? error.message : 'Unknown error'
|
||||||
if (!isRecipientNotAllowedError(errorMessage)) {
|
if (!isRecipientNotAllowedError(errorMessage)) {
|
||||||
lastError = errorMessage
|
lastError = errorMessage
|
||||||
break // unrelated error — stop retrying
|
break
|
||||||
}
|
}
|
||||||
lastError = errorMessage
|
lastError = errorMessage
|
||||||
// retry with next variant
|
// retry with next variant
|
||||||
@ -120,15 +162,18 @@ export async function POST(request: Request) {
|
|||||||
|
|
||||||
if (sentMessageId) {
|
if (sentMessageId) {
|
||||||
results.push({
|
results.push({
|
||||||
phone,
|
phone: recipient.phone,
|
||||||
status: 'sent',
|
status: 'sent',
|
||||||
whatsapp_message_id: sentMessageId,
|
whatsapp_message_id: sentMessageId,
|
||||||
})
|
})
|
||||||
sentCount++
|
sentCount++
|
||||||
} else {
|
} else {
|
||||||
console.error(`Failed to send broadcast to ${phone}:`, lastError)
|
console.error(
|
||||||
|
`Failed to send broadcast to ${recipient.phone}:`,
|
||||||
|
lastError
|
||||||
|
)
|
||||||
results.push({
|
results.push({
|
||||||
phone,
|
phone: recipient.phone,
|
||||||
status: 'failed',
|
status: 'failed',
|
||||||
error: lastError || 'Unknown error',
|
error: lastError || 'Unknown error',
|
||||||
})
|
})
|
||||||
@ -138,7 +183,7 @@ export async function POST(request: Request) {
|
|||||||
|
|
||||||
return NextResponse.json({
|
return NextResponse.json({
|
||||||
success: true,
|
success: true,
|
||||||
total: phone_numbers.length,
|
total: recipients.length,
|
||||||
sent: sentCount,
|
sent: sentCount,
|
||||||
failed: failedCount,
|
failed: failedCount,
|
||||||
results,
|
results,
|
||||||
|
|||||||
@ -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: {
|
async function handleStatusUpdate(status: {
|
||||||
id: string
|
id: string
|
||||||
status: string
|
status: string
|
||||||
timestamp: string
|
timestamp: string
|
||||||
recipient_id: string
|
recipient_id: string
|
||||||
}) {
|
}) {
|
||||||
// The messages table schema uses `message_id` (not whatsapp_message_id)
|
// 1) Mirror onto messages (legacy behavior) — Meta's status values
|
||||||
// and has no `updated_at` column. Meta's status values (sent/delivered/
|
// already match the CHECK constraint on messages.status.
|
||||||
// read/failed) already match our CHECK constraint.
|
const { error: msgErr } = await supabaseAdmin()
|
||||||
const { error } = await supabaseAdmin()
|
|
||||||
.from('messages')
|
.from('messages')
|
||||||
.update({ status: status.status })
|
.update({ status: status.status })
|
||||||
.eq('message_id', status.id)
|
.eq('message_id', status.id)
|
||||||
|
|
||||||
if (error) {
|
if (msgErr) {
|
||||||
console.error('Error updating message status:', error)
|
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<string, unknown> = { 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) {
|
if (convError) {
|
||||||
console.error('Error updating conversation:', 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(
|
async function parseMessageContent(
|
||||||
|
|||||||
@ -23,6 +23,31 @@ interface UseBroadcastSendingReturn {
|
|||||||
progress: number;
|
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 {
|
export function useBroadcastSending(): UseBroadcastSendingReturn {
|
||||||
const [isProcessing, setIsProcessing] = useState(false);
|
const [isProcessing, setIsProcessing] = useState(false);
|
||||||
const [progress, setProgress] = useState(0);
|
const [progress, setProgress] = useState(0);
|
||||||
@ -74,19 +99,26 @@ export function useBroadcastSending(): UseBroadcastSendingReturn {
|
|||||||
variables: Record<string, { type: 'static' | 'field'; value: string }>,
|
variables: Record<string, { type: 'static' | 'field'; value: string }>,
|
||||||
contact: Contact
|
contact: Contact
|
||||||
): string[] {
|
): string[] {
|
||||||
return Object.keys(variables)
|
// Keys typically are "1","2",... so a numeric-aware sort keeps {{1}}
|
||||||
.sort()
|
// before {{10}}.
|
||||||
.map((key) => {
|
const keys = Object.keys(variables).sort((a, b) => {
|
||||||
const v = variables[key];
|
const an = Number(a);
|
||||||
if (v.type === 'static') return v.value;
|
const bn = Number(b);
|
||||||
const fieldMap: Record<string, string | undefined> = {
|
if (Number.isFinite(an) && Number.isFinite(bn)) return an - bn;
|
||||||
name: contact.name,
|
return a.localeCompare(b);
|
||||||
phone: contact.phone,
|
});
|
||||||
email: contact.email,
|
|
||||||
company: contact.company,
|
return keys.map((key) => {
|
||||||
};
|
const v = variables[key];
|
||||||
return fieldMap[v.value] ?? '';
|
if (v.type === 'static') return v.value;
|
||||||
});
|
const fieldMap: Record<string, string | undefined> = {
|
||||||
|
name: contact.name,
|
||||||
|
phone: contact.phone,
|
||||||
|
email: contact.email,
|
||||||
|
company: contact.company,
|
||||||
|
};
|
||||||
|
return fieldMap[v.value] ?? '';
|
||||||
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
async function createAndSendBroadcast(payload: BroadcastPayload): Promise<string> {
|
async function createAndSendBroadcast(payload: BroadcastPayload): Promise<string> {
|
||||||
@ -96,7 +128,7 @@ export function useBroadcastSending(): UseBroadcastSendingReturn {
|
|||||||
const supabase = createClient();
|
const supabase = createClient();
|
||||||
|
|
||||||
try {
|
try {
|
||||||
// Step 1: Resolve audience contacts
|
// ── Step 1: Resolve audience contacts ─────────────────────────
|
||||||
setProgress(5);
|
setProgress(5);
|
||||||
const contacts = await resolveAudience(payload.audience);
|
const contacts = await resolveAudience(payload.audience);
|
||||||
|
|
||||||
@ -104,7 +136,7 @@ export function useBroadcastSending(): UseBroadcastSendingReturn {
|
|||||||
throw new Error('No contacts found for this audience.');
|
throw new Error('No contacts found for this audience.');
|
||||||
}
|
}
|
||||||
|
|
||||||
// Step 2: Create broadcast record
|
// ── Step 2: Create broadcast row ──────────────────────────────
|
||||||
setProgress(10);
|
setProgress(10);
|
||||||
const { data: broadcast, error: broadcastError } = await supabase
|
const { data: broadcast, error: broadcastError } = await supabase
|
||||||
.from('broadcasts')
|
.from('broadcasts')
|
||||||
@ -132,17 +164,16 @@ export function useBroadcastSending(): UseBroadcastSendingReturn {
|
|||||||
throw new Error(`Failed to create broadcast: ${broadcastError?.message}`);
|
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);
|
setProgress(20);
|
||||||
const BATCH_SIZE = 50;
|
|
||||||
const recipientRows = contacts.map((contact) => ({
|
const recipientRows = contacts.map((contact) => ({
|
||||||
broadcast_id: broadcast.id,
|
broadcast_id: broadcast.id,
|
||||||
contact_id: contact.id,
|
contact_id: contact.id,
|
||||||
status: 'pending' as const,
|
status: 'pending' as const,
|
||||||
}));
|
}));
|
||||||
|
|
||||||
for (let i = 0; i < recipientRows.length; i += BATCH_SIZE) {
|
for (let i = 0; i < recipientRows.length; i += INSERT_BATCH_SIZE) {
|
||||||
const batch = recipientRows.slice(i, i + BATCH_SIZE);
|
const batch = recipientRows.slice(i, i + INSERT_BATCH_SIZE);
|
||||||
const { error: recipientError } = await supabase
|
const { error: recipientError } = await supabase
|
||||||
.from('broadcast_recipients')
|
.from('broadcast_recipients')
|
||||||
.insert(batch);
|
.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);
|
setProgress(30);
|
||||||
const { data: recipients, error: recipientsFetchError } = await supabase
|
const { data: recipients, error: recipientsFetchError } = await supabase
|
||||||
.from('broadcast_recipients')
|
.from('broadcast_recipients')
|
||||||
@ -167,31 +198,29 @@ export function useBroadcastSending(): UseBroadcastSendingReturn {
|
|||||||
let failedCount = 0;
|
let failedCount = 0;
|
||||||
const totalRecipients = recipients.length;
|
const totalRecipients = recipients.length;
|
||||||
|
|
||||||
for (let i = 0; i < recipients.length; i += BATCH_SIZE) {
|
for (let i = 0; i < recipients.length; i += SEND_BATCH_SIZE) {
|
||||||
const batch = recipients.slice(i, i + BATCH_SIZE);
|
const batch = recipients.slice(i, i + SEND_BATCH_SIZE);
|
||||||
const phoneNumbers = batch
|
|
||||||
.map((r) => r.contact?.phone)
|
|
||||||
.filter(Boolean) as string[];
|
|
||||||
|
|
||||||
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
|
if (apiRecipients.length === 0) continue;
|
||||||
const templateParams = batch.map((r) => {
|
|
||||||
if (r.contact) {
|
|
||||||
return resolveVariables(payload.variables, r.contact);
|
|
||||||
}
|
|
||||||
return [];
|
|
||||||
});
|
|
||||||
|
|
||||||
try {
|
try {
|
||||||
const res = await fetch('/api/whatsapp/broadcast', {
|
const res = await fetch('/api/whatsapp/broadcast', {
|
||||||
method: 'POST',
|
method: 'POST',
|
||||||
headers: { 'Content-Type': 'application/json' },
|
headers: { 'Content-Type': 'application/json' },
|
||||||
body: JSON.stringify({
|
body: JSON.stringify({
|
||||||
phone_numbers: phoneNumbers,
|
recipients: apiRecipients,
|
||||||
template_name: payload.template.name,
|
template_name: payload.template.name,
|
||||||
template_language: payload.template.language ?? 'en_US',
|
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');
|
throw new Error(data.error || 'Broadcast API request failed');
|
||||||
}
|
}
|
||||||
|
|
||||||
// Update individual recipient statuses based on results
|
// The API returns one entry per input recipient in order.
|
||||||
if (data.results) {
|
// Map by phone to be defensive against any reordering.
|
||||||
for (let j = 0; j < data.results.length; j++) {
|
const resultsByPhone = new Map<string, BroadcastApiResult>();
|
||||||
const result = data.results[j];
|
for (const r of (data.results ?? []) as BroadcastApiResult[]) {
|
||||||
const recipient = batch[j];
|
resultsByPhone.set(r.phone, r);
|
||||||
if (!recipient) continue;
|
}
|
||||||
|
|
||||||
const newStatus = result.status === 'sent' ? 'sent' : 'failed';
|
for (const recipient of batch) {
|
||||||
if (newStatus === 'sent') sentCount++;
|
const phone = recipient.contact?.phone;
|
||||||
else failedCount++;
|
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
|
await supabase
|
||||||
.from('broadcast_recipients')
|
.from('broadcast_recipients')
|
||||||
.update({
|
.update({
|
||||||
status: newStatus,
|
status: 'failed',
|
||||||
sent_at: newStatus === 'sent' ? new Date().toISOString() : null,
|
error_message: 'No phone number on contact',
|
||||||
error_message: result.error ?? null,
|
})
|
||||||
|
.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);
|
.eq('id', recipient.id);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} catch (err) {
|
} 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) {
|
for (const recipient of batch) {
|
||||||
failedCount++;
|
failedCount++;
|
||||||
await supabase
|
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);
|
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);
|
setProgress(95);
|
||||||
const finalStatus = failedCount === totalRecipients ? 'failed' : 'sent';
|
const finalStatus = failedCount === totalRecipients ? 'failed' : 'sent';
|
||||||
await supabase
|
await supabase
|
||||||
.from('broadcasts')
|
.from('broadcasts')
|
||||||
.update({
|
.update({ status: finalStatus })
|
||||||
status: finalStatus,
|
|
||||||
sent_count: sentCount,
|
|
||||||
failed_count: failedCount,
|
|
||||||
})
|
|
||||||
.eq('id', broadcast.id);
|
.eq('id', broadcast.id);
|
||||||
|
|
||||||
setProgress(100);
|
setProgress(100);
|
||||||
|
// Unused locals intentionally kept for clarity at return site:
|
||||||
|
void sentCount;
|
||||||
|
|
||||||
return broadcast.id;
|
return broadcast.id;
|
||||||
} finally {
|
} finally {
|
||||||
setIsProcessing(false);
|
setIsProcessing(false);
|
||||||
|
|||||||
91
src/lib/broadcast-status.ts
Normal file
91
src/lib/broadcast-status.ts
Normal file
@ -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<BroadcastStatus, StatusDisplay> = {
|
||||||
|
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<RecipientStatus, StatusDisplay> = {
|
||||||
|
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
|
||||||
|
);
|
||||||
|
}
|
||||||
@ -189,6 +189,12 @@ export interface BroadcastRecipient {
|
|||||||
read_at?: string;
|
read_at?: string;
|
||||||
replied_at?: string;
|
replied_at?: string;
|
||||||
error_message?: 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;
|
created_at: string;
|
||||||
contact?: Contact;
|
contact?: Contact;
|
||||||
}
|
}
|
||||||
|
|||||||
84
supabase/migrations/003_broadcast_recipient_wamid.sql
Normal file
84
supabase/migrations/003_broadcast_recipient_wamid.sql
Normal file
@ -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();
|
||||||
Loading…
Reference in New Issue
Block a user