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