Postgres & Supabase Backend Setup

1-- Works unmodified on Supabase, Neon, RDS, CockroachDB, or a self-hosted
2-- Postgres instance — run this once, in the SQL Editor (Supabase) or via
3-- your normal migration tooling (everyone else).
4 
5-- 1. Changelog table — holds at most ONE row per entity. A new mutation
6--    on a row replaces its changelog entry rather than appending another,
7--    which is what keeps this table small on a live app instead of
8--    growing per-edit forever. Deletions are still caught here, since a
9--    plain SELECT can never tell "row was deleted" from "row never
10--    existed." This is also what backs delta catch-up on reconnect — see
11--    core-concepts/sync-anchors.
12create table if not exists public.flux_changelog (
13  id         bigint generated always as identity primary key,
14  table_name text        not null,
15  row_id     text        not null,
16  op         text        not null check (op in ('INSERT', 'UPDATE', 'DELETE')),
17  row_data   jsonb,
18  changed_at timestamptz not null default now()
19);
20 
21create index if not exists flux_changelog_table_row_idx
22  on public.flux_changelog (table_name, row_id);
23 
24create index if not exists flux_changelog_table_changed_idx
25  on public.flux_changelog (table_name, changed_at);
26 
27-- 2. Sparse field-level diff. row_data does NOT always hold a full row —
28--    see the trigger function below for exactly which ops get a full row
29--    vs a partial one, and why. This function only computes the diff; it
30--    always keeps 'id' even when unchanged, so a partial payload still
31--    carries something to key a client-side merge on.
32create or replace function public.flux_jsonb_diff(p_old jsonb, p_new jsonb)
33returns jsonb
34language sql
35immutable
36as $$
37  select jsonb_object_agg(key, value)
38  from jsonb_each(p_new)
39  where not (p_old ? key)
40     or (p_old -> key) <> value
41     or key = 'id';
42$$;
43 
44-- 3. Trigger function — assumes each table's primary key column is named
45--    "id". Adjust if a table you attach this to differs.
46--
47--    row_data shape by op:
48--      INSERT — full row (to_jsonb(NEW)). No prior state to diff against.
49--      UPDATE — sparse diff: only the fields that actually changed, plus
50--               id. Cuts payload size sharply on wide tables where a
51--               mutation typically touches a couple of columns.
52--      DELETE — just {"id": ...}. Nothing else is needed to remove a
53--               record client-side.
54--
55--    Single-state dedup (delete-then-insert below) means only the most
56--    recent changelog entry per row survives. Combined with sparse
57--    UPDATE diffs, that creates a real gap if handled naively: a row
58--    updated twice before a client reconnects would have its second
59--    diff computed against the FIRST update's post-state, not against
60--    what the client has cached — so overwriting the changelog entry
61--    outright would silently drop the first update's changed fields
62--    from catch-up, even though they genuinely differ from the client's
63--    view. The block below closes that gap by merging into whatever
64--    changelog entry is still pending for this row, instead of blindly
65--    replacing it:
66--      - if the pending entry is still an INSERT (client hasn't caught
67--        up to the row's creation at all yet), stay an INSERT with the
68--        newest values — equivalent to to_jsonb(NEW).
69--      - if the pending entry is an UPDATE, merge the new diff on top
70--        of the old one (jsonb || — newer keys win, untouched earlier
71--        keys survive), so the client eventually receives every field
72--        that changed since its last sync, not just the last edit's.
73create or replace function public.flux_log_change()
74returns trigger
75language plpgsql
76security definer
77set search_path = public
78as $$
79declare
80  pk_value      text;
81  v_op          text;
82  v_data        jsonb;
83  v_id          bigint;
84  v_existing_op text;
85  v_existing    jsonb;
86begin
87  if TG_OP = 'DELETE' then
88    execute 'select ($1).id::text' into pk_value using OLD;
89    v_op   := 'DELETE';
90    v_data := jsonb_build_object('id', pk_value);
91 
92  elsif TG_OP = 'UPDATE' then
93    execute 'select ($1).id::text' into pk_value using NEW;
94    v_op   := 'UPDATE';
95    v_data := public.flux_jsonb_diff(to_jsonb(OLD), to_jsonb(NEW));
96 
97    select op, row_data into v_existing_op, v_existing
98    from public.flux_changelog
99    where table_name = TG_TABLE_NAME and row_id = pk_value;
100 
101    if v_existing_op = 'INSERT' then
102      v_op   := 'INSERT';
103      v_data := to_jsonb(NEW);
104    elsif v_existing_op = 'UPDATE' then
105      v_data := v_existing || v_data;
106    end if;
107    -- v_existing_op = 'DELETE' or null: nothing pending to merge into,
108    -- this diff stands alone.
109 
110  else -- INSERT
111    execute 'select ($1).id::text' into pk_value using NEW;
112    v_op   := 'INSERT';
113    v_data := to_jsonb(NEW);
114  end if;
115 
116  -- Delete any previous changelog entry for this row so only the newest
117  -- (now merge-aware) state remains — this is what keeps the table
118  -- single-state instead of accumulating one row per edit.
119  delete from public.flux_changelog
120  where table_name = TG_TABLE_NAME
121    and row_id = pk_value;
122 
123  insert into public.flux_changelog (table_name, row_id, op, row_data)
124  values (TG_TABLE_NAME, pk_value, v_op, v_data)
125  returning id into v_id;
126 
127  -- Postgres caps a NOTIFY payload at 8000 bytes — a full row (or a large
128  -- jsonb column) can silently exceed that. Sending just the changelog
129  -- row's own id sidesteps the limit entirely; a listener looks the real
130  -- data up by id (Step 4).
131  perform pg_notify('flux_changes', json_build_object(
132    'id', v_id, 'table', TG_TABLE_NAME
133  )::text);
134 
135  if TG_OP = 'DELETE' then
136    return OLD;
137  else
138    return NEW;
139  end if;
140end;
141$$;
142 
143-- 4. Attach to every table you want realtime + catch-up on — copy this
144--    block per table, swapping the table name in both lines.
145drop trigger if exists flux_changelog_trigger on public.your_table;
146create trigger flux_changelog_trigger
147  after insert or update or delete on public.your_table
148  for each row execute function public.flux_log_change();
149 
150alter table public.flux_changelog enable row level security;
151-- No public SELECT policy on flux_changelog — inserts hold full row
152-- snapshots, updates hold changed-field diffs, deletes hold just an id,
153-- but none of that should be directly queryable. Supabase reads it only
154-- through the SECURITY DEFINER RPC in Step 3 (which bypasses RLS on
155-- purpose, since that's the one sanctioned read path); a non-Supabase
156-- listener reads it via a direct, trusted DB connection.

1-- Enables Supabase's managed Realtime broadcast for this table — the
2-- dashboard toggle (Database > Replication) does the same thing; this is
3-- the SQL equivalent if you'd rather have it in a migration.
4alter publication supabase_realtime add table public.your_table;

1-- The function Flux calls automatically on reconnect for catch-up —
2-- covered in full, with the RLS caveats, in core-concepts/sync-anchors
3-- Step 5. Included here only so this page's schema is complete on its
4-- own; don't paste this twice if you've already run the sync-anchors SQL.
5--
6-- SECURITY DEFINER, not INVOKER: flux_changelog has RLS enabled with no
7-- SELECT policy (see the callout at the end of Step 1), so a function
8-- running as the calling anon/authenticated role would see zero rows —
9-- the RPC call itself succeeds, but silently returns an empty rows array
10-- every single time, indistinguishable from "nothing changed." DEFINER
11-- runs the function as its owner instead, which bypasses RLS for this
12-- one sanctioned read path while the table itself stays locked down from
13-- direct PostgREST access. The explicit GRANT EXECUTE below is required
14-- once this is DEFINER — anon/authenticated still need permission to
15-- call the function itself, RLS bypass or not.
16--
17-- LIMIT placement matters here: it's applied INSIDE the DISTINCT ON
18-- subquery, bounding how many distinct entities get pulled per call.
19-- Applying it to the outer query instead would do nothing useful — that
20-- query returns exactly one row (the jsonb_agg'd array), so a LIMIT out
21-- there just limits a single-row result set.
22--
23-- The DISTINCT ON here is a safety net, not the primary source of dedup
24-- — the trigger in Step 1 already keeps at most one row per entity. It
25-- only matters if two overlapping transactions manage to log the same
26-- row before either commits; in the normal case it's a no-op.
27--
28-- 'data' in the returned payload is NOT always a full row — see Step 1's
29-- trigger comment. INSERT rows are full, UPDATE rows are a merged sparse
30-- diff (changed fields since the client's last sync, plus id), DELETE
31-- rows are just {id}. The client-side adapter's normalizer expects this
32-- and treats UPDATE payloads as a patch to merge, not a row to replace.
33create or replace function public.flux_request_delta(
34  p_channel  text,
35  p_table    text,
36  p_since_ms bigint,
37  p_limit    int default 2000
38)
39returns void
40language plpgsql
41security definer
42set search_path = public
43as $$
44declare
45  v_rows jsonb;
46begin
47  select coalesce(
48           jsonb_agg(
49             jsonb_build_object(
50               'op',        latest.op,
51               'data',      latest.row_data,
52               'timestamp', (extract(epoch from latest.changed_at) * 1000)::bigint
53             )
54             order by latest.changed_at asc
55           ),
56           '[]'::jsonb
57         )
58    into v_rows
59  from (
60    select distinct on (c.row_id)
61      c.row_id, c.op, c.row_data, c.changed_at
62    from public.flux_changelog c
63    where c.table_name = p_table
64      and c.changed_at > to_timestamp(p_since_ms / 1000.0)
65    order by c.row_id, c.changed_at desc
66    limit p_limit
67  ) latest;
68 
69  perform realtime.send(
70    jsonb_build_object('rows', v_rows),
71    'flux_sync_complete',
72    'flux:' || p_channel,
73    false
74  );
75end;
76$$;
77 
78-- Required because the function above is SECURITY DEFINER — PostgREST
79-- callers (anon/authenticated) still need explicit EXECUTE permission to
80-- invoke it at all, independent of the RLS bypass DEFINER grants them
81-- once inside.
82grant execute on function public.flux_request_delta(text, text, bigint, int) to anon, authenticated;

1// server/postgres-listener.js
2//
3// The bridge a non-Supabase Postgres needs and Supabase doesn't: LISTEN/
4// NOTIFY only reaches OTHER DATABASE CONNECTIONS, never a browser, so
5// something has to hold one dedicated connection open, hear the NOTIFY
6// fired by the trigger in Step 1, and forward it into whichever transport
7// registry your app already built — the exact same broadcastToWs()/
8// io.emit() this project's WebSocket and Socket.IO getting-started pages
9// already wire up. This is new code to run, not new infrastructure to
10// stand up — it lives inside your existing backend process.
11//
12// IMPORTANT: row_data forwarded here is NOT always a full row. Per the
13// Step 1 trigger, INSERT rows are full, UPDATE rows are a merged sparse
14// diff (changed fields since this row's changelog entry was last
15// consumed, plus id), DELETE rows are just { id }. This applies to LIVE
16// push here exactly as it does to catch-up on the Supabase RPC path,
17// since both read the same flux_changelog table. Whatever consumes these
18// events on the client (or downstream in your own app) needs to MERGE an
19// 'UPDATE' payload into existing state by id, not replace the record
20// wholesale — replacing would wipe out every field this payload didn't
21// happen to include.
22 
23const { Client } = require('pg')
24const { broadcastToWs } = require('./ws-registry') // swap for io.emit if using Socket.IO
25 
26function tableToChannel(table) {
27  // However you map table -> Flux channel name. Simplest case: identical.
28  return table
29}
30 
31async function startPostgresListener() {
32  // A dedicated client, NOT a pooled one — see the callout below this
33  // block for why pgbouncer (or any transaction-mode pooler) breaks this.
34  const listenerClient = new Client({ connectionString: process.env.DATABASE_URL })
35  await listenerClient.connect()
36  await listenerClient.query('LISTEN flux_changes')
37 
38  console.log('[pg] Listening for flux_changes notifications')
39 
40  listenerClient.on('notification', async (msg) => {
41    let payload
42    try {
43      payload = JSON.parse(msg.payload)
44    } catch {
45      return
46    }
47 
48    const { id, table } = payload
49 
50    // The NOTIFY payload only ever carries the changelog row's own id —
51    // see the trigger function comment in Step 1 for why. Look the real
52    // row up here, over the same connection. Because the changelog is
53    // single-state, this always fetches the row's CURRENT logged state —
54    // even if several edits landed before the listener got to this one,
55    // there's only ever one row to find, not a backlog to drain (and per
56    // the trigger's merge logic, that one row already reflects every
57    // field changed since it was last consumed, not just the latest edit).
58    const { rows } = await listenerClient.query(
59      'SELECT row_id, op, row_data, changed_at FROM flux_changelog WHERE id = $1',
60      [id]
61    )
62    const row = rows[0]
63    if (!row) return
64 
65    const channel = tableToChannel(table)
66 
67    broadcastToWs(channel, {
68      entity:    table,
69      id:        row.row_id,
70      op:        row.op,          // 'INSERT' | 'UPDATE' | 'DELETE'
71      data:      row.row_data,    // full row (INSERT) / sparse diff (UPDATE) / {id} (DELETE)
72      timestamp: new Date(row.changed_at).getTime(),
73      source:    'my-postgres-listener',
74    })
75 
76    // Socket.IO instead of WebSocket? Same idea, different call:
77    // io.emit(`${channel}:UPDATE`, { entity: table, id: row.row_id, ... })
78  })
79 
80  listenerClient.on('error', (err) => {
81    console.error('[pg] Listener connection error:', err)
82    // A dropped LISTEN connection means live updates silently stop until
83    // this reconnects — add real reconnect/backoff logic here before
84    // shipping this to production, this is intentionally left minimal.
85  })
86}
87 
88module.exports = { startPostgresListener }

1# server/postgres_listener.py
2#
3# Same bridge, asyncpg's native LISTEN/NOTIFY support. Forwards into
4# whichever registry your chosen transport already built — sse_registry,
5# ws_registry, or a Socket.IO AsyncServer's sio.emit().
6#
7# Same caveat as the Node version: row_data is a full row for INSERT, a
8# merged sparse diff for UPDATE, and just {"id": ...} for DELETE. Merge
9# UPDATE payloads into existing client-side state by id — don't replace.
10 
11import asyncio
12import json
13import asyncpg
14from ws_registry import broadcast_to_ws  # or sse_registry / socketio's sio
15 
16def table_to_channel(table: str) -> str:
17    return table
18 
19async def start_postgres_listener(dsn: str):
20    # A dedicated, unpooled connection — same requirement as the Node
21    # version. asyncpg's own connection pool is fine for normal queries,
22    # but LISTEN needs one connection held open for the process lifetime.
23    conn = await asyncpg.connect(dsn)
24 
25    async def on_notify(connection, pid, channel, payload):
26        data = json.loads(payload)
27        row_id, table = data["id"], data["table"]
28 
29        row = await conn.fetchrow(
30            "SELECT row_id, op, row_data, changed_at FROM flux_changelog WHERE id = $1",
31            row_id,
32        )
33        if not row:
34            return
35 
36        target_channel = table_to_channel(table)
37 
38        await broadcast_to_ws(target_channel, {
39            "entity": table,
40            "id": row["row_id"],
41            "op": row["op"],
42            "data": json.loads(row["row_data"]),  # full row / sparse diff / {id}
43            "timestamp": int(row["changed_at"].timestamp() * 1000),
44            "source": "my-postgres-listener",
45        })
46 
47    await conn.add_listener("flux_changes", on_notify)
48    print("[pg] Listening for flux_changes notifications")
49 
50    # This connection does nothing but listen — keep the coroutine alive
51    # for the lifetime of the process.
52    while True:
53        await asyncio.sleep(3600)

1-- Optional. With single-state dedup, flux_changelog no longer grows per
2-- edit — it grows only with the number of distinct rows you've ever
3-- touched, plus one tombstone row per deletion that never gets replaced.
4-- Cleanup mainly matters for those tombstones and for entities whose
5-- parent table gets truncated or dropped without matching deletes. On
6-- Supabase, pg_cron ships enabled by default (Database > Extensions). On
7-- Neon, RDS, CockroachDB, or self-hosted, you may need to install
8-- pg_cron yourself, or skip this and hit the same cleanup via an
9-- external cron calling a small admin endpoint instead — the DELETE
10-- statement is what matters, not the scheduler wrapped around it.
11create or replace function public.flux_changelog_cleanup(retain_days int default 14)
12returns void
13language sql
14security definer
15set search_path = public
16as $$
17  delete from public.flux_changelog
18  where changed_at < now() - (retain_days || ' days')::interval;
19$$;
20 
21select cron.schedule('flux-changelog-cleanup', '0 3 * * *',
22  $$select public.flux_changelog_cleanup(14)$$);

1-- In a second psql session, or the SQL Editor — fire a manual NOTIFY to
2-- confirm your listener process is actually receiving them, independent
3-- of whether a real trigger has fired yet:
4select pg_notify('flux_changes', '{"id": 0, "table": "test"}');