wacrm/supabase/migrations/003_broadcast_recipient_wamid.sql
Arnas Donauskas 5d5c5f0a20
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>
2026-04-17 11:22:38 +03:00

85 lines
3.6 KiB
PL/PgSQL

-- ============================================================
-- 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();