supabase-realtime

Tools and patterns for an LLM agent to react to database events and coordinate over broadcast channels on Supabase Realtime, deployed as a Supabase Edge Function.

When to reach for this skill

Three triggers, each with what not to do.

1. The agent needs to act on a database event

For example: a new ticket arrives in support_tickets and the agent should triage it. Use watch_table with predicate.event = "INSERT".

Don't use this for state the agent already wrote — watch_table is for changes the agent didn't cause. If the agent just inserted a row and wants to know it was inserted, that's a return value, not a subscription.

2. The agent needs to fan out a result to other agents

For example: triage agent decides routing, then signals a downstream handoff agent. Use broadcast_to_channel.

Don't use broadcast as a queue — Realtime broadcast is fire-and-forget; messages aren't durable. If the receiving agent might be offline, write the work to a real queue (pgmq) and trigger that side via watch_table on the queue table.

3. The agent is the receiving side of a multi-agent workflow

Use subscribe_to_channel. Mirrors watch_table's bounded shape — block until N events or timeout.

Don't subscribe with a high max_events and timeout_ms "just in case" — Edge Function isolates have wall-clock budgets. Spec a tight bound; the pattern is bounded subscription, not persistent.

Core pattern: bounded subscription

The tool blocks for at most timeout_ms or until max_events matching events arrive — whichever first. Then returns the batch. This is the right primitive for agent loops because:

• It maps cleanly to a single MCP tool-call (no streaming protocol)
• It fits Edge Function isolate budgets (caps timeout at 120s, well under the 150s wall-clock limit)
• It composes with normal agent loops: call → process batch → call again

The canonical loop:

while (still_relevant) {
  const { events, closed_reason } = await mcp.call("watch_table", {
    table: "support_tickets",
    predicate: { event: "INSERT" },
    timeout_ms: 60000,
    max_events: 10,
  });
  for (const ev of events) await processEvent(ev);
  if (closed_reason === "timeout" && shouldStop()) break;
}

Why not a persistent WebSocket? The agent's tool-call boundary is the natural checkpoint. Persistent connections fight the Edge Function model and force you into long-lived workers, which is a different deployment shape and a different operational surface.

Tools at a glance

Five tools. No Presence in v1 — see references/presence-deferred.md for why.

Composition modules

These are not MCP tools; they're library-level functions exported from supabase-realtime-skill/server for code that wants the bounded primitives composed with safety nets the agent shouldn't have to re-derive.

boundedQueueDrain — drain a queue/outbox table in one bounded pass

IMPORTANT: at-least-once. Each row may be forwarded more than once if the broadcast succeeds but the ack callback fails. Subscribers MUST be idempotent.

Composes boundedWatch (the watch_table body) + handleBroadcast (with its 3-retry envelope) + a caller-supplied ack callback (canonical: SQL update queue set forwarded_at = now() where id = ...) + an optional dead_letter callback. One drain pass: arrive → broadcast → ack, capped at max_events and timeout_ms to fit Edge Function isolate budgets.

import { boundedQueueDrain, makeSupabaseAdapter } from "supabase-realtime-skill/server";

const result = await boundedQueueDrain({
  adapter: makeSupabaseAdapter("outbox", { supabaseUrl, supabaseKey }),
  table: "outbox",
  read_row: (ev) => {
    const row = ev.new as { destination: string; event_type: string; payload: Record<string, unknown> };
    return { destination: row.destination, event: row.event_type, payload: row.payload };
  },
  ack: async (ev) => sql`update outbox set forwarded_at = now() where id = ${(ev.new as { id: string }).id}`,
  dead_letter: async (ev, _row, err) => sql`insert into dlq (id, payload, error) values (${(ev.new as { id: string }).id}, ${ev.new}::jsonb, ${String(err)})`,
  sender: yourBroadcastSender,
  timeout_ms: 60_000,
  max_events: 25,
});
// result: { forwarded, dead_lettered, failed, closed_reason }

Categories partition the events array — forwarded + dead_lettered + failed === events.length. Full reference: references/queue-drain.md. Filed against ADR-0010 (Proposed): docs/decisions/0010-bounded-queue-drain.md.

Don't use this when ordering across destinations matters (Realtime broadcast is fire-and-forget; per-destination FIFO needs a different shape) or when subscribers can't be made idempotent (run a consumer-side inbox table — out of scope for the current release; revisit if a real consumer flags it).

For multi-tenant deployments, pass private: true to gate the broadcast leg with realtime.messages RLS — see § "Worked example: multi-tenant audit log" below.

Worked example: support-ticket triage

A SaaS app has a support_tickets table. Tickets get auto-embedded via Supabase Automatic Embeddings (writes a halfvec(1536) to embedding). The triage agent watches for new tickets, retrieves the most-similar past resolved tickets via pgvector, decides routing, writes the routing back, and broadcasts a ticket-routed event so a downstream handoff agent picks it up.

End-to-end walkthrough with code in references/worked-example.md. Schema ships at supabase/migrations/20260430000001_support_tickets.sql; this is the eval harness's SUT.

Worked example: multi-tenant audit log

A B2B SaaS app has tenant-scoped audit events; users belong to one or more tenants via a memberships junction table. An agent operating under a forwarded user JWT watches audit_events (Postgres-Changes RLS — table policy admits only the user's own tenants) and broadcasts notable events to tenant:<id>:audit-feed private channels (Broadcast Authorization RLS — realtime.messages policies enforce membership at subscribe + send). Cross-tenant leakage is prevented at the substrate, not in agent code.

Apply with supabase db push; the migration carries memberships + audit_events + the public.user_tenant_ids() SECURITY DEFINER STABLE helper + two realtime.messages RLS policies. Schema ships at supabase/migrations/20260502000001_multi_tenant_audit_demo.sql; operator deep dive at references/multi-tenant-rls.md. The smoke test tests/smoke/multi-tenant-rls.smoke.test.ts loads the same migration as setup, so the demo and the regression gate are byte-equivalent.

Substrate-correctness for this example is filed across ADRs 0011 (JWT-setAuth on the Realtime websocket leg) + 0013 (private: true opt-in on Broadcast Authorization) + 0014 (the worked-example ship + boundedQueueDrain private threading).

References

• predicates.md — supported filter ops, why others are excluded
• replication-identity.md — when to enable REPLICA IDENTITY FULL
• rls-implications.md — RLS + Realtime + broadcast auth (high-level summary)
• multi-tenant-rls.md — multi-tenant deep dive: two RLS layers, setAuth requirement, production schema shape, scaling story
• presence-deferred.md — design questions left open for v2
• pgvector-composition.md — composing CDC + Automatic Embeddings + retrieval
• eval-methodology.md — the 4 metrics, why not LLM-judge
• edge-deployment.md — operator setup
• worked-example.md — support-ticket triage end-to-end (LLM + pgvector)
• queue-drain.md — boundedQueueDrain typed module: contract, schema, drain loop, Edge Function shape
• outbox-forwarder.md — manual-composition predecessor (kept for non-SQL DLQ shapes)