-- ============================================================ -- Incremental broadcast aggregate trigger. -- -- Migration 003 installed a trigger that recomputed every counter -- (sent/delivered/read/replied/failed) via COUNT(*) FILTER on every -- row change. For a 10k-recipient broadcast, the send loop produces -- 10k INSERTs + 10k UPDATEs = 20k full aggregate scans, each walking -- the (broadcast_id, status) index. Workable at small scale, but -- O(n²) overall. -- -- This migration replaces that with an incremental trigger that -- adjusts the parent broadcast's counts by ±1 based on the OLD → -- NEW.status delta. O(1) per recipient change; no scans at all. -- -- Semantic model (same as the lib/broadcast-status.ts "forward-only -- ladder" in the webhook): -- sent_count = recipients whose status is at or past 'sent' -- delivered_count = ... at or past 'delivered' -- read_count = ... at or past 'read' -- replied_count = status = 'replied' -- failed_count = status = 'failed' -- -- A webhook that advances a recipient pending → sent → delivered → -- read → replied bumps every rung it crosses by 1. Going to 'failed' -- only bumps failed_count (and can only happen from pending / sent, -- enforced in the webhook). -- -- Keeps the safety net: a public recompute_broadcast_counts() SQL -- function is retained so ops can run it manually if counts ever -- drift (e.g. after bulk DB surgery). -- -- Idempotent — safe to run multiple times. -- ============================================================ -- Delta a single column by +1 / -1. CREATE OR REPLACE FUNCTION public._bcast_bump(bid UUID, col TEXT, delta INT) RETURNS VOID AS $$ BEGIN EXECUTE format( 'UPDATE broadcasts SET %I = GREATEST(0, %I + $1), updated_at = NOW() WHERE id = $2', col, col ) USING delta, bid; END; $$ LANGUAGE plpgsql SECURITY DEFINER SET search_path = public; -- Columns this recipient's status contributes to. CREATE OR REPLACE FUNCTION public._bcast_cols_for_status(s TEXT) RETURNS TEXT[] AS $$ BEGIN -- 'pending' contributes to nothing. IF s = 'pending' THEN RETURN ARRAY[]::TEXT[]; END IF; IF s = 'sent' THEN RETURN ARRAY['sent_count']; END IF; IF s = 'delivered' THEN RETURN ARRAY['sent_count','delivered_count']; END IF; IF s = 'read' THEN RETURN ARRAY['sent_count','delivered_count','read_count']; END IF; IF s = 'replied' THEN RETURN ARRAY['sent_count','delivered_count','read_count','replied_count']; END IF; IF s = 'failed' THEN RETURN ARRAY['failed_count']; END IF; RETURN ARRAY[]::TEXT[]; END; $$ LANGUAGE plpgsql IMMUTABLE; -- Replace the trigger body with the incremental version. CREATE OR REPLACE FUNCTION public.broadcast_recipient_aggregate_trigger() RETURNS TRIGGER AS $$ DECLARE old_cols TEXT[]; new_cols TEXT[]; c TEXT; BEGIN IF TG_OP = 'INSERT' THEN new_cols := _bcast_cols_for_status(NEW.status); FOREACH c IN ARRAY new_cols LOOP PERFORM _bcast_bump(NEW.broadcast_id, c, 1); END LOOP; RETURN NEW; END IF; IF TG_OP = 'DELETE' THEN old_cols := _bcast_cols_for_status(OLD.status); FOREACH c IN ARRAY old_cols LOOP PERFORM _bcast_bump(OLD.broadcast_id, c, -1); END LOOP; RETURN OLD; END IF; -- UPDATE: only care if status changed. IF OLD.status IS DISTINCT FROM NEW.status THEN old_cols := _bcast_cols_for_status(OLD.status); new_cols := _bcast_cols_for_status(NEW.status); -- Subtract the old contributions, add the new. FOREACH c IN ARRAY old_cols LOOP PERFORM _bcast_bump(NEW.broadcast_id, c, -1); END LOOP; FOREACH c IN ARRAY new_cols LOOP PERFORM _bcast_bump(NEW.broadcast_id, c, 1); END LOOP; END IF; RETURN NEW; END; $$ LANGUAGE plpgsql SECURITY DEFINER SET search_path = public; -- Trigger itself remains the same (INSERT/UPDATE/DELETE) — just its -- body has been replaced. -- Safety net — rebuild counts from scratch. Retained as-is so ops can -- run it on demand if something ever drifts. Matches the incremental -- trigger's semantic model exactly. 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;