fix(broadcasts): senior-review hardening pass

Fixes 8 real issues found during a thorough review of the broadcast
feature (PRs #21-#29).

Security
  * Webhook POSTs had zero spoof protection. Anyone could forge a
    status update and drift broadcast counts arbitrarily. Added
    HMAC-SHA256 signature verification via the X-Hub-Signature-256
    header (Meta's standard). New helper src/lib/whatsapp/webhook-
    signature.ts with timing-safe comparison. Rollout fails open
    with a warning when META_APP_SECRET is unset so existing
    deploys don't break at rollout; set the env var to enforce.

Correctness
  * CSV audience was completely broken — synthesized "csv-N" strings
    as contact_id, which failed the UUID cast on insert. Every CSV
    broadcast silently created zero recipients. Replaced with a
    real find-or-create against the contacts table (batched lookup
    by phone, single INSERT for missing rows, de-duplicates CSV
    inputs). CSV contacts now end up as real contacts.rows with
    real UUIDs flowing through the rest of the pipeline.

  * Recipient insert errors were logged with console.error and the
    loop marched on — broadcasts then ran with incomplete recipient
    sets, webhook updates couldn't find rows, and aggregate counts
    drifted. Now: on batch insert failure, flip the broadcast to
    failed + failed_count=total + throw so handleSend's toast fires.

  * Webhook status guard ordered "failed" above "replied", which
    meant a webhook replay delivering "failed" for an already-
    replied recipient would upgrade replied → failed. Split the
    ladder: forward-only ordering across pending → sent →
    delivered → read → replied; failed is a terminal side branch
    that can only be reached from pending or sent (never from any
    success state). Net effect: a message that was read can no
    longer regress to failed on a replay.

  * Template sync only pulled page 1 (ignored paging.next). Users
    with >100 templates silently missed some, and broadcasts with
    those names failed with Meta's "template not found". Now
    follows paging.next with a 20-page safety cap, reports
    truncated=true in the response if hit.

  * Template sync reported success:true even when individual
    upserts failed. Now accumulates per-template errors and
    surfaces them via a separate toast.error in the UI — green
    toast no longer hides silent drift.

Performance
  * Migration 005: replace the broadcast_recipients aggregate
    trigger installed in migration 003. Old version ran
    COUNT(*) FILTER on every row change — O(n²) across a send
    loop (10k recipients × 10k index scans). New version
    increments / decrements by ±1 per status transition: O(1)
    per change. Kept the old recompute_broadcast_counts() as a
    callable safety-net in case of drift.

  * List page poll kept running while the tab was hidden. Added
    visibilitychange handler — polling pauses when hidden,
    resumes (and does an immediate fetch) on refocus.

UX
  * Step 3 (Personalize) let users skip past unmapped placeholders
    → broadcasts then sent to Meta with empty-string params.
    Next is now disabled until every {{N}} has a value; an amber
    warning lists the missing placeholders.

Documented accepted risks in the PR body.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

src/app/(dashboard)/broadcasts/page.tsx

@@ -90,18 +90,38 @@ export default function BroadcastsPage() {
   );
 
   useEffect(() => {
-    // Start / stop the poller based on whether any row is sending.
+    function startPolling() {
-    if (anySending && !pollTimer.current) {
+      if (pollTimer.current) return;
       pollTimer.current = setInterval(fetchBroadcasts, POLL_INTERVAL_MS);
-    } else if (!anySending && pollTimer.current) {
+    }
+    function stopPolling() {
+      if (!pollTimer.current) return;
       clearInterval(pollTimer.current);
       pollTimer.current = null;
     }
-    return () => {
+
-      if (pollTimer.current) {
+    // Pause polling while the tab is hidden — keeps Supabase cold when
-        clearInterval(pollTimer.current);
+    // the user is away, and ensures a fresh fetch the moment they
-        pollTimer.current = null;
+    // refocus so they don't see stale data on return.
+    function handleVisibilityChange() {
+      if (!anySending) return;
+      if (document.visibilityState === 'hidden') {
+        stopPolling();
+      } else {
+        fetchBroadcasts();
+        startPolling();
       }
+    }
+
+    if (anySending && document.visibilityState === 'visible') {
+      startPolling();
+    } else {
+      stopPolling();
+    }
+    document.addEventListener('visibilitychange', handleVisibilityChange);
+    return () => {
+      stopPolling();
+      document.removeEventListener('visibilitychange', handleVisibilityChange);
     };
   }, [anySending]);
 

src/app/api/whatsapp/templates/sync/route.ts

@@ -125,33 +125,47 @@ export async function POST() {
 
     const accessToken = decrypt(config.access_token)
 
-    // Pull up to 100 templates. Meta paginates via `paging.next`; for
+    // Paginate through every template Meta has for this WABA. Meta
-    // now we take page 1, which covers realistic small-business use.
+    // returns at most 100 per page; `paging.next` is a full URL. Cap
-    // A follow-up can loop through paging.next if someone has more
+    // at 20 pages (2k templates) as a safety against infinite loops
-    // than 100 templates (rare — Meta caps new templates per account).
+    // from a misbehaving upstream.
-    const url = `${META_API_BASE}/${config.waba_id}/message_templates?limit=100&fields=id,name,language,status,category,components`
+    const metaTemplates: MetaTemplate[] = []
-    const metaRes = await fetch(url, {
+    let nextUrl:
-      headers: { Authorization: `Bearer ${accessToken}` },
+      | string
-    })
+      | null = `${META_API_BASE}/${config.waba_id}/message_templates?limit=100&fields=id,name,language,status,category,components`
+    const PAGE_CAP = 20
+    let pageCount = 0
 
-    if (!metaRes.ok) {
+    while (nextUrl && pageCount < PAGE_CAP) {
-      let metaErr = `Meta API error: ${metaRes.status}`
+      pageCount++
-      try {
+      const metaRes: Response = await fetch(nextUrl, {
-        const body = await metaRes.json()
+        headers: { Authorization: `Bearer ${accessToken}` },
-        if (body?.error?.message) metaErr = body.error.message
+      })
-      } catch {
+
-        // response wasn't JSON — keep the fallback
+      if (!metaRes.ok) {
+        let metaErr = `Meta API error: ${metaRes.status}`
+        try {
+          const body = await metaRes.json()
+          if (body?.error?.message) metaErr = body.error.message
+        } catch {
+          // response wasn't JSON — keep the fallback
+        }
+        return NextResponse.json({ error: metaErr }, { status: 502 })
       }
-      return NextResponse.json({ error: metaErr }, { status: 502 })
-    }
 
-    const metaBody: { data?: MetaTemplate[] } = await metaRes.json()
+      const metaBody: {
-    const metaTemplates = metaBody.data ?? []
+        data?: MetaTemplate[]
+        paging?: { next?: string }
+      } = await metaRes.json()
+      if (metaBody.data) metaTemplates.push(...metaBody.data)
+      nextUrl = metaBody.paging?.next ?? null
+    }
 
     // For each Meta template: upsert by (user_id, name, language).
     // No UNIQUE constraint on that triple, so we match manually.
     let inserted = 0
     let updated = 0
+    const errors: { name: string; language: string; message: string }[] = []
 
     for (const t of metaTemplates) {
       const body = (t.components ?? []).find((c) => c.type === 'BODY')
@@ -171,7 +185,7 @@ export async function POST() {
         updated_at: new Date().toISOString(),
       }
 
-      const { data: existing } = await supabase
+      const { data: existing, error: lookupErr } = await supabase
         .from('message_templates')
         .select('id')
         .eq('user_id', user.id)
@@ -179,23 +193,52 @@ export async function POST() {
         .eq('language', t.language)
         .maybeSingle()
 
+      if (lookupErr) {
+        errors.push({
+          name: t.name,
+          language: t.language,
+          message: lookupErr.message,
+        })
+        continue
+      }
+
       if (existing?.id) {
-        await supabase
+        const { error: updErr } = await supabase
           .from('message_templates')
           .update(row)
           .eq('id', existing.id)
-        updated++
+        if (updErr) {
+          errors.push({
+            name: t.name,
+            language: t.language,
+            message: updErr.message,
+          })
+        } else {
+          updated++
+        }
       } else {
-        await supabase.from('message_templates').insert(row)
+        const { error: insErr } = await supabase
-        inserted++
+          .from('message_templates')
+          .insert(row)
+        if (insErr) {
+          errors.push({
+            name: t.name,
+            language: t.language,
+            message: insErr.message,
+          })
+        } else {
+          inserted++
+        }
       }
     }
 
     return NextResponse.json({
-      success: true,
+      success: errors.length === 0,
       total: metaTemplates.length,
       inserted,
       updated,
+      errors,
+      truncated: pageCount >= PAGE_CAP && nextUrl !== null,
     })
   } catch (error) {
     console.error('Error syncing WhatsApp templates:', error)

src/app/api/whatsapp/webhook/route.ts

@@ -3,6 +3,7 @@ import { createClient } from '@supabase/supabase-js'
 import { decrypt } from '@/lib/whatsapp/encryption'
 import { getMediaUrl, downloadMedia } from '@/lib/whatsapp/meta-api'
 import { normalizePhone, phonesMatch } from '@/lib/whatsapp/phone-utils'
+import { verifyMetaWebhookSignature } from '@/lib/whatsapp/webhook-signature'
 
 // Lazy-initialized to avoid build-time crash when env vars are missing
 // eslint-disable-next-line @typescript-eslint/no-explicit-any
@@ -120,10 +121,27 @@ export async function GET(request: Request) {
 
 // POST - Receive messages
 export async function POST(request: Request) {
-  // Always return 200 immediately to acknowledge receipt
+  // Read raw body first so we can HMAC-verify the exact bytes Meta
-  const body = await request.json()
+  // signed. request.json() would re-encode and break the signature.
+  const rawBody = await request.text()
+  const signature = request.headers.get('x-hub-signature-256')
 
-  // Process asynchronously
+  if (!verifyMetaWebhookSignature(rawBody, signature)) {
+    // 401 (not 200) — we want Meta's delivery dashboard to show failures
+    // loudly if a misconfiguration causes signatures to stop matching,
+    // rather than silently eating events.
+    console.warn('[webhook] rejected request with invalid signature')
+    return NextResponse.json({ error: 'Invalid signature' }, { status: 401 })
+  }
+
+  let body: { entry?: WhatsAppWebhookEntry[] }
+  try {
+    body = JSON.parse(rawBody)
+  } catch {
+    return NextResponse.json({ error: 'Invalid JSON' }, { status: 400 })
+  }
+
+  // Process asynchronously so we can ack Meta within their timeout.
   processWebhook(body).catch((error) => {
     console.error('Error processing webhook:', error)
   })
@@ -179,20 +197,46 @@ async function processWebhook(body: { entry?: WhatsAppWebhookEntry[] }) {
   }
 }
 
-// Ordered forward-only: a webhook replay must never regress a recipient
+// The happy-path status ladder — pending → sent → delivered → read →
-// from `read` back to `sent`. Index into this array gives the "level".
+// replied. Webhook replays must never regress a recipient back down
-const RECIPIENT_STATUS_ORDER = [
+// this ladder.
+//
+// `failed` is NOT on this ladder. It's a terminal side branch that is
+// only valid from the early states (pending / sent) — once Meta has
+// delivered or the user has read or replied, a later "failed" status
+// event is a bug in Meta's pipeline or a spoof attempt and must be
+// ignored.
+const RECIPIENT_STATUS_LADDER = [
   'pending',
   'sent',
   'delivered',
   'read',
   'replied',
-  'failed',
 ] as const
 
-function recipientStatusLevel(s: string): number {
+function ladderLevel(s: string): number {
-  const idx = (RECIPIENT_STATUS_ORDER as readonly string[]).indexOf(s)
+  const idx = (RECIPIENT_STATUS_LADDER as readonly string[]).indexOf(s)
-  return idx < 0 ? 0 : idx
+  return idx < 0 ? -1 : idx
+}
+
+/**
+ * Can a recipient transition from `current` to `incoming`?
+ *   - Along the ladder, only forward moves are allowed.
+ *   - `failed` is accepted only from `pending` or `sent`; it's refused
+ *     once the recipient has reached any of the success states.
+ */
+function isValidStatusTransition(current: string, incoming: string): boolean {
+  if (incoming === 'failed') {
+    return current === 'pending' || current === 'sent'
+  }
+  if (current === 'failed') {
+    return false // failed is terminal
+  }
+  const ci = ladderLevel(current)
+  const ii = ladderLevel(incoming)
+  if (ii < 0) return false // unknown incoming status
+  if (ci < 0) return true // unknown current — accept anything on the ladder
+  return ii > ci
 }
 
 async function handleStatusUpdate(status: {
@@ -230,10 +274,9 @@ async function handleStatusUpdate(status: {
   }
   if (!recipient) return // message wasn't part of a broadcast — fine
 
-  // Forward-only status guard: only advance.
+  // Guard transitions — forward-only on the success ladder, and
-  const currentLevel = recipientStatusLevel(recipient.status)
+  // `failed` only from pre-delivered states.
-  const incomingLevel = recipientStatusLevel(status.status)
+  if (!isValidStatusTransition(recipient.status, status.status)) return
-  if (incomingLevel <= currentLevel) return
 
   const update: Record<string, unknown> = { status: status.status }
   if (status.status === 'sent' && !('sent_at' in update)) update.sent_at = tsIso

src/components/broadcasts/step3-personalize.tsx

@@ -111,6 +111,24 @@ export function Step3Personalize({
     return [...new Set(matches)].sort();
   }, [template.body_text]);
 
+  /**
+   * A placeholder is "unmapped" if the user hasn't picked either a
+   * static value or a field/custom-field source. Blocks Next until
+   * every placeholder has something — otherwise the broadcast would
+   * ship with empty strings and confuse recipients.
+   */
+  const unmappedKeys = useMemo(() => {
+    const missing: string[] = [];
+    for (const placeholder of placeholders) {
+      const key = placeholder.replace(/^\{\{|\}\}$/g, '');
+      const mapping = variables[key];
+      if (!mapping || !mapping.value?.trim()) {
+        missing.push(placeholder);
+      }
+    }
+    return missing;
+  }, [placeholders, variables]);
+
   function updateVariable(key: string, patch: Partial<VariableMapping>) {
     const current = variables[key] ?? { type: 'static' as VariableType, value: '' };
     onUpdate({
@@ -311,6 +329,16 @@ export function Step3Personalize({
         </div>
       </div>
 
+      {unmappedKeys.length > 0 && (
+        <div className="rounded-md border border-amber-500/30 bg-amber-500/10 px-3 py-2 text-xs text-amber-300">
+          Map every placeholder before continuing — still missing{' '}
+          <span className="font-mono font-semibold">
+            {unmappedKeys.join(', ')}
+          </span>
+          . Otherwise those placeholders will ship to Meta as empty strings.
+        </div>
+      )}
+
       <div className="flex items-center justify-between border-t border-slate-800 pt-4">
         <Button
           variant="outline"
@@ -322,7 +350,8 @@ export function Step3Personalize({
         </Button>
         <Button
           onClick={onNext}
-          className="bg-emerald-600 text-white hover:bg-emerald-700"
+          disabled={unmappedKeys.length > 0}
+          className="bg-emerald-600 text-white hover:bg-emerald-700 disabled:opacity-50"
         >
           Next
           <ArrowRight className="h-4 w-4" />

src/components/settings/template-manager.tsx

@@ -200,6 +200,22 @@ export function TemplateManager() {
             ? ` (${data.inserted} new, ${data.updated} updated)`
             : ''),
       );
+      if (Array.isArray(data.errors) && data.errors.length > 0) {
+        // Surface per-template failures so users don't trust a green
+        // toast that hides silent drift.
+        const preview = data.errors.slice(0, 3).map(
+          (e: { name: string; language: string; message: string }) =>
+            `${e.name} (${e.language})`,
+        );
+        const suffix =
+          data.errors.length > 3 ? `, +${data.errors.length - 3} more` : '';
+        toast.error(`Failed to sync: ${preview.join(', ')}${suffix}`);
+      }
+      if (data.truncated) {
+        toast.warning(
+          'Hit Meta pagination cap — more templates may exist. Contact support if this persists.',
+        );
+      }
       await fetchTemplates(user.id);
     } catch (err) {
       console.error('Template sync error:', err);

src/hooks/use-broadcast-sending.ts

@@ -179,20 +179,7 @@ export function useBroadcastSending(): UseBroadcastSendingReturn {
     } 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
+      contacts = await upsertCsvContacts(supabase, audience.csvContacts);
-      // 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
@@ -209,6 +196,84 @@ export function useBroadcastSending(): UseBroadcastSendingReturn {
     return contacts;
   }
 
+  /**
+   * CSV uploads arrive as raw phone/name pairs, not DB rows. Before we
+   * can insert broadcast_recipients (whose contact_id FKs contacts.id),
+   * we need real contacts.id UUIDs. So: look up each CSV phone in the
+   * caller's contacts table; insert any that don't exist; return the
+   * resolved set.
+   *
+   * Pre-existing implementation synthesized `csv-N` strings as
+   * contact_id, which failed the UUID cast on insert — every CSV
+   * broadcast silently created zero recipients.
+   */
+  async function upsertCsvContacts(
+    supabase: ReturnType<typeof createClient>,
+    csvRows: { phone: string; name?: string }[],
+  ): Promise<Contact[]> {
+    if (csvRows.length === 0) return [];
+
+    const {
+      data: { session },
+    } = await supabase.auth.getSession();
+    const user = session?.user;
+    if (!user) {
+      throw new Error('You are not signed in.');
+    }
+
+    // De-duplicate by phone within the CSV (users can paste duplicates).
+    const uniqueByPhone = new Map<string, { phone: string; name?: string }>();
+    for (const row of csvRows) {
+      if (row.phone) uniqueByPhone.set(row.phone, row);
+    }
+    const phones = [...uniqueByPhone.keys()];
+
+    // Single round-trip lookup of existing contacts by phone.
+    const { data: existing, error: lookupErr } = await supabase
+      .from('contacts')
+      .select('*')
+      .eq('user_id', user.id)
+      .in('phone', phones);
+    if (lookupErr) {
+      throw new Error(`Failed to look up CSV contacts: ${lookupErr.message}`);
+    }
+
+    const byPhone = new Map<string, Contact>();
+    for (const c of (existing ?? []) as Contact[]) {
+      if (c.phone) byPhone.set(c.phone, c);
+    }
+
+    // Insert only missing contacts, in one batch per 200 rows (PostgREST
+    // has a default payload cap — 200 keeps individual requests small).
+    const missing = phones
+      .filter((p) => !byPhone.has(p))
+      .map((phone) => ({
+        user_id: user.id,
+        phone,
+        name: uniqueByPhone.get(phone)?.name ?? null,
+      }));
+
+    const INSERT_CHUNK = 200;
+    for (let i = 0; i < missing.length; i += INSERT_CHUNK) {
+      const chunk = missing.slice(i, i + INSERT_CHUNK);
+      const { data: inserted, error: insertErr } = await supabase
+        .from('contacts')
+        .insert(chunk)
+        .select();
+      if (insertErr) {
+        throw new Error(`Failed to create CSV contacts: ${insertErr.message}`);
+      }
+      for (const c of (inserted ?? []) as Contact[]) {
+        if (c.phone) byPhone.set(c.phone, c);
+      }
+    }
+
+    // Preserve input order so analytics roughly matches the CSV order.
+    return phones
+      .map((p) => byPhone.get(p))
+      .filter((c): c is Contact => Boolean(c));
+  }
+
   async function resolveCustomFieldAudience(
     supabase: ReturnType<typeof createClient>,
     filter: CustomFieldFilter,
@@ -317,7 +382,21 @@ export function useBroadcastSending(): UseBroadcastSendingReturn {
           .from('broadcast_recipients')
           .insert(batch);
         if (recipientError) {
-          console.error('Failed to insert recipient batch:', recipientError);
+          // Previous impl logged and marched on — the broadcast then ran
+          // with an incomplete recipient set, so webhook status updates
+          // couldn't find some rows and the aggregate counts drifted.
+          // Flip the broadcast to failed so the user sees the problem
+          // immediately, then throw to abort the send loop.
+          await supabase
+            .from('broadcasts')
+            .update({
+              status: 'failed',
+              failed_count: contacts.length,
+            })
+            .eq('id', broadcast.id);
+          throw new Error(
+            `Failed to insert recipient batch ${i / INSERT_BATCH_SIZE + 1}: ${recipientError.message}`,
+          );
         }
       }
 

src/lib/whatsapp/webhook-signature.ts

@@ -0,0 +1,45 @@
+import crypto from 'node:crypto'
+
+/**
+ * Verify the HMAC-SHA256 signature Meta attaches to webhook POSTs.
+ *
+ * Meta signs the raw request body with your App Secret and sends the
+ * result in the `x-hub-signature-256: sha256=<hex>` header. Without
+ * verification, anyone who knows our webhook URL can POST fabricated
+ * status updates and drift broadcast counts arbitrarily.
+ *
+ * Reference:
+ *   https://developers.facebook.com/docs/graph-api/webhooks/getting-started#verify-payloads
+ *
+ * Rollout behavior:
+ *   If `META_APP_SECRET` is unset, we **fail open** with a loud warning
+ *   log. This avoids breaking existing deployments at the moment this
+ *   code ships — operators can set the env var on their next deploy.
+ *   Once the env var is set, verification is strict.
+ */
+export function verifyMetaWebhookSignature(
+  rawBody: string,
+  signatureHeader: string | null,
+): boolean {
+  const secret = process.env.META_APP_SECRET
+  if (!secret) {
+    console.warn(
+      '[webhook] META_APP_SECRET not set — skipping signature verification. ' +
+        'Set the env var to harden against spoofed webhook POSTs.',
+    )
+    return true
+  }
+
+  if (!signatureHeader) return false
+  if (!signatureHeader.startsWith('sha256=')) return false
+
+  const expected =
+    'sha256=' +
+    crypto.createHmac('sha256', secret).update(rawBody).digest('hex')
+
+  const a = Buffer.from(signatureHeader)
+  const b = Buffer.from(expected)
+  // Bail if lengths differ — timingSafeEqual throws otherwise.
+  if (a.length !== b.length) return false
+  return crypto.timingSafeEqual(a, b)
+}

supabase/migrations/005_broadcast_counts_incremental.sql

@@ -0,0 +1,129 @@
+-- ============================================================
+-- 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;