Postgres & Supabase Backend Setup

1-- Run once. On Supabase paste it into the SQL Editor, everywhere else use your
2-- normal migration tooling. Nothing in this block is Supabase specific.
3 
4-- One row per entity. A new mutation on a row REPLACES its changelog entry
5-- instead of appending another, so the table stays small on a live app.
6-- Deletions are caught here because a plain SELECT can never tell "row was
7-- deleted" from "row never existed."
8--
9-- user_id decides who may read the entry:
10--   null      a GLOBAL row, readable by every client (public tables)
11--   a uuid    a PER USER row, readable only by that user
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  user_id    uuid,
17  op         text        not null check (op in ('INSERT', 'UPDATE', 'DELETE')),
18  row_data   jsonb,
19  changed_at timestamptz not null default now()
20);
21 
22-- Already running the older table that had no user_id? This adds the column.
23alter table public.flux_changelog add column if not exists user_id uuid;
24 
25-- The trigger looks an entry up by table and row when it replaces it.
26create index if not exists flux_changelog_table_row_idx
27  on public.flux_changelog (table_name, row_id);
28 
29-- The range scan every catch up runs: one table, one audience, newer than an anchor.
30create index if not exists flux_changelog_table_changed_idx
31  on public.flux_changelog (table_name, changed_at);
32 
33create index if not exists flux_changelog_table_user_changed_idx
34  on public.flux_changelog (table_name, user_id, changed_at);
35 
36alter table public.flux_changelog enable row level security;
37-- No SELECT policy on purpose. Entries hold full row snapshots. They are read
38-- only through the SECURITY DEFINER functions in the delta RPC step, which
39-- filter by caller. A non Supabase listener reads them over a trusted
40-- database connection.

1-- Decides, per table, whether a row is GLOBAL or belongs to ONE USER.
2-- The trigger calls this for every change. Adapt the cases to your schema.
3--
4-- Return null    the row is global and every client may read it
5-- Return a uuid  the row belongs to that user and nobody else may read it
6--
7-- A table with no rule raises an error on purpose. A table you forgot can
8-- never become public by accident, the first write to it fails loudly.
9create or replace function public.flux_resolve_owner(p_table text, p_row jsonb)
10returns uuid
11language plpgsql
12stable
13security definer
14set search_path = public
15as $$
16begin
17  case p_table
18    -- Per user tables: the owner is stored on the row.
19    when 'projects' then
20      return (p_row ->> 'user_id')::uuid;
21 
22    -- The row is the user's own record.
23    when 'profiles' then
24      return (p_row ->> 'id')::uuid;
25 
26    -- Child tables: the owner is the owner of the parent row. If a cascade
27    -- already removed the parent, this returns null and the deleted row
28    -- becomes a global tombstone that carries only an id.
29    when 'project_domains' then
30      return (
31        select pr.user_id
32        from public.projects pr
33        where pr.id = (p_row ->> 'project_id')::uuid
34      );
35 
36    -- Global tables: the same rows for everyone.
37    when 'blog_posts', 'plan_pricing', 'currency_rates' then
38      return null;
39 
40    else
41      raise exception 'flux_resolve_owner: no owner rule for table %', p_table;
42  end case;
43end;
44$$;

1-- The trigger function. Every entry stores the FULL row, never a diff, so a
2-- client that applies it always receives a complete record.
3--
4--   INSERT  full row
5--   UPDATE  full row (the newest state)
6--   DELETE  just {"id": ...}, which is all a client needs to remove a record
7--
8-- Assumes each tracked table has a primary key column named "id".
9create or replace function public.flux_log_change()
10returns trigger
11language plpgsql
12security definer
13set search_path to 'public'
14as $function$
15declare
16  v_row         jsonb;
17  pk_value      text;
18  v_op          text;
19  v_data        jsonb;
20  v_owner       uuid;
21  v_id          bigint;
22  v_existing_op text;
23begin
24  if TG_OP = 'DELETE' then
25    v_row := to_jsonb(OLD);
26  else
27    v_row := to_jsonb(NEW);
28  end if;
29 
30  pk_value := v_row ->> 'id';
31  v_owner  := public.flux_resolve_owner(TG_TABLE_NAME, v_row);
32 
33  if TG_OP = 'DELETE' then
34    v_op   := 'DELETE';
35    v_data := jsonb_build_object('id', pk_value);
36 
37  elsif TG_OP = 'UPDATE' then
38    v_op   := 'UPDATE';
39    v_data := v_row;
40 
41    -- A row the client has not seen yet (a pending INSERT) stays an INSERT.
42    select c.op into v_existing_op
43    from public.flux_changelog c
44    where c.table_name = TG_TABLE_NAME
45      and c.row_id     = pk_value
46    order by c.id desc
47    limit 1;
48 
49    if v_existing_op = 'INSERT' then
50      v_op := 'INSERT';
51    end if;
52 
53  else
54    v_op   := 'INSERT';
55    v_data := v_row;
56  end if;
57 
58  -- Single state: only the newest entry per row survives.
59  delete from public.flux_changelog
60  where table_name = TG_TABLE_NAME
61    and row_id     = pk_value;
62 
63  insert into public.flux_changelog (table_name, row_id, user_id, op, row_data)
64  values (TG_TABLE_NAME, pk_value, v_owner, v_op, v_data)
65  returning id into v_id;
66 
67  -- NOTIFY payloads are capped at 8000 bytes, so only the changelog id, the
68  -- table and the owner travel. A listener looks the real data up by id.
69  perform pg_notify('flux_changes', json_build_object(
70    'id', v_id, 'table', TG_TABLE_NAME, 'user_id', v_owner
71  )::text);
72 
73  if TG_OP = 'DELETE' then
74    return OLD;
75  end if;
76  return NEW;
77end;
78$function$;
79 
80-- Attach to every table you want realtime and catch up on. Copy this block
81-- per table and swap the table name in both lines. The trigger name matters,
82-- flux_reconcile only answers for tables that carry flux_changelog_trigger.
83drop trigger if exists flux_changelog_trigger on public.your_table;
84create trigger flux_changelog_trigger
85  after insert or update or delete on public.your_table
86  for each row execute function public.flux_log_change();

1-- Turns on Supabase's managed Realtime broadcast for this table. The dashboard
2-- toggle under Database, Replication does the same thing.
3alter publication supabase_realtime add table public.your_table;

1-- Delta catch up, batch catch up and reconcile. Run after Steps 1 to 3.
2 
3-- The reply travels as a Realtime broadcast, so the return type stays void.
4-- Rows are filtered by the CALLER: their own per user rows plus global rows
5-- (user_id is null), never another user's. SECURITY DEFINER is required
6-- because flux_changelog has RLS enabled and no SELECT policy.
7drop function if exists public.flux_request_delta(text, text, bigint, integer);
8 
9create or replace function public.flux_request_delta(
10  p_channel  text,
11  p_table    text,
12  p_since_ms bigint,
13  p_limit    integer default 2000
14)
15returns void
16language plpgsql
17security definer
18set search_path to 'public'
19as $function$
20declare
21  -- Must stay below the cleanup retention (14 days) so a gap can never
22  -- straddle a purge.
23  c_retention_ms constant bigint := 13::bigint * 24 * 60 * 60 * 1000;
24 
25  -- server_now is reported BEHIND the real clock. changed_at is transaction
26  -- start time, so a transaction that began before this read can still commit
27  -- after it with an older changed_at. The overlap re delivers such rows next
28  -- time, and applying a row twice is harmless.
29  c_overlap_ms   constant bigint := 5000;
30 
31  v_caller  uuid    := auth.uid();   -- null for logged out callers: global rows only
32  v_now_ms  bigint  := (extract(epoch from clock_timestamp()) * 1000)::bigint;
33  v_rows    jsonb   := '[]'::jsonb;
34  v_count   integer := 0;
35  v_too_old boolean := false;
36begin
37  if p_since_ms < v_now_ms - c_retention_ms then
38    v_too_old := true;
39  else
40    with page as (
41      select c.id, c.row_id, c.op, c.row_data, c.changed_at
42      from public.flux_changelog c
43      where c.table_name = p_table
44        and c.changed_at > to_timestamp(p_since_ms / 1000.0)
45        and (c.user_id is null or c.user_id = v_caller)
46      order by c.changed_at asc, c.id asc
47      limit p_limit + 1
48    )
49    select
50      count(*),
51      coalesce(
52        jsonb_agg(
53          jsonb_build_object(
54            'op',        p.op,
55            'row_id',    p.row_id,
56            'data',      jsonb_build_object('id', p.row_id) || coalesce(p.row_data, '{}'::jsonb),
57            'timestamp', (extract(epoch from p.changed_at) * 1000)::bigint
58          )
59          order by p.changed_at asc, p.id asc
60        ),
61        '[]'::jsonb
62      )
63    into v_count, v_rows
64    from page p;
65 
66    -- More rows than we are willing to send: refuse a partial delta so the
67    -- client refetches instead of claiming coverage over missing rows.
68    if v_count > p_limit then
69      v_too_old := true;
70      v_rows    := '[]'::jsonb;
71    end if;
72  end if;
73 
74  perform realtime.send(
75    jsonb_build_object(
76      'rows',       v_rows,
77      'server_now', v_now_ms - c_overlap_ms,
78      'too_old',    v_too_old
79    ),
80    'flux_sync_complete',
81    'flux:' || p_channel,
82    false
83  );
84end;
85$function$;
86 
87-- One HTTP request for every table. since_ms is parsed as numeric and floored,
88-- so a fractional epoch value can never break the bigint cast.
89create or replace function public.flux_request_delta_batch(p_requests jsonb)
90returns void
91language plpgsql
92security definer
93set search_path to 'public'
94as $function$
95declare
96  r jsonb;
97begin
98  if jsonb_typeof(p_requests) <> 'array' then
99    raise exception 'flux_request_delta_batch: p_requests must be a json array';
100  end if;
101  if jsonb_array_length(p_requests) > 50 then
102    raise exception 'flux_request_delta_batch: too many requests (max 50)';
103  end if;
104 
105  for r in select * from jsonb_array_elements(p_requests) loop
106    perform public.flux_request_delta(
107      r ->> 'channel',
108      r ->> 'table',
109      floor((r ->> 'since_ms')::numeric)::bigint
110    );
111  end loop;
112end;
113$function$;
114 
115-- Which of these local ids no longer exist on the server? SECURITY INVOKER on
116-- purpose: RLS applies, so an id the caller can no longer see is reported as
117-- deleted for THEM. Only tables that carry the flux trigger are allowed.
118create or replace function public.flux_reconcile(p_table text, p_local_ids text[])
119returns text[]
120language plpgsql
121security invoker
122set search_path to 'public'
123as $function$
124declare
125  v_missing text[];
126begin
127  if not exists (
128    select 1
129    from pg_trigger t
130    join pg_class c     on c.oid = t.tgrelid
131    join pg_namespace n on n.oid = c.relnamespace
132    where not t.tgisinternal
133      and t.tgname  = 'flux_changelog_trigger'
134      and n.nspname = 'public'
135      and c.relname = p_table
136  ) then
137    raise exception 'flux_reconcile: "%" is not a flux tracked table', p_table;
138  end if;
139 
140  execute format(
141    'select coalesce(
142       array(
143         select unnest($1)
144         except
145         select t.id::text from public.%I t where t.id::text = any($1)
146       ),
147       ''{}''::text[]
148     )',
149    p_table
150  )
151  into v_missing
152  using p_local_ids;
153 
154  return v_missing;
155end;
156$function$;
157 
158-- PostgREST callers need explicit EXECUTE. Logged in users only by default.
159revoke all on function public.flux_request_delta(text, text, bigint, integer) from public, anon;
160revoke all on function public.flux_request_delta_batch(jsonb)                 from public, anon;
161revoke all on function public.flux_reconcile(text, text[])                    from public, anon;
162 
163grant execute on function public.flux_request_delta(text, text, bigint, integer) to authenticated;
164grant execute on function public.flux_request_delta_batch(jsonb)                 to authenticated;
165grant execute on function public.flux_reconcile(text, text[])                    to authenticated;
166 
167-- Logged out visitors catching up on PUBLIC tables (blog_posts, plan_pricing,
168-- currency_rates): uncomment to allow. They only ever receive global rows.
169-- grant execute on function public.flux_request_delta(text, text, bigint, integer) to anon;
170-- grant execute on function public.flux_request_delta_batch(jsonb)                 to anon;

1// What flux_request_delta broadcasts on 'flux:' || p_channel, event
2// 'flux_sync_complete'. The adapter turns it into one outcome per channel.
3{
4  rows: [
5    {
6      op:        'INSERT' | 'UPDATE' | 'DELETE',
7      row_id:    '662f...',
8      data:      { id: '662f...', ...fullRow },   // DELETE: { id } only
9      timestamp: 1718000001234                    // epoch ms, server clock
10    }
11  ],
12  server_now: 1718000099000,   // becomes the new sync anchor (already 5 seconds behind)
13  too_old:    false            // true: the server will not send a partial delta
14}
15 
16// Outcome the adapter reports:
17//   rows applied               applied
18//   no rows                    confirmed current (a quiet table is a valid answer)
19//   too_old = true             too old, the client refetches the channel
20//   RPC error or no reply      failed, the gap stays open and the client refetches

1// server/postgres_listener.js
2//
3// The bridge a non Supabase Postgres needs and Supabase does not. LISTEN and
4// NOTIFY only reach OTHER DATABASE CONNECTIONS, never a browser, so one
5// dedicated connection is held open, hears the NOTIFY fired by the trigger,
6// and forwards it into the registry your transport page already built.
7//
8// Every row forwarded here is a FULL row (INSERT and UPDATE) or { id } (DELETE).
9// There are no partial updates, so a client can replace or merge safely.
10//
11// The NOTIFY carries user_id. A null owner means a GLOBAL row, sent to every
12// subscriber. A uuid owner means a PER USER row, sent only to that user's
13// sockets. The registry on the transport page does that filtering.
14 
15const { Client } = require('pg')
16const { broadcastToWs } = require('./broadcast') // swap for io.to(room).emit or broadcastToSse
17 
18function tableToChannel(table) {
19  // However you map table to Flux channel name. Simplest case: identical.
20  return table
21}
22 
23async function startPostgresListener() {
24  // A dedicated client, NOT a pooled one. See the callout below this block.
25  const listenerClient = new Client({ connectionString: process.env.DATABASE_URL })
26  await listenerClient.connect()
27  await listenerClient.query('LISTEN flux_changes')
28 
29  console.log('[pg] Listening for flux_changes notifications')
30 
31  listenerClient.on('notification', async (msg) => {
32    let payload
33    try {
34      payload = JSON.parse(msg.payload)
35    } catch {
36      return
37    }
38 
39    // Single state: this always fetches the row's CURRENT logged entry, even
40    // if several edits landed before the listener got to this one.
41    const { rows } = await listenerClient.query(
42      'SELECT row_id, op, row_data, user_id, changed_at FROM flux_changelog WHERE id = $1',
43      [payload.id]
44    )
45    const row = rows[0]
46    if (!row) return
47 
48    const op = row.op === 'INSERT' ? 'CREATE' : row.op
49 
50    broadcastToWs(
51      tableToChannel(payload.table),
52      {
53        entity:    payload.table,
54        id:        row.row_id,
55        op,
56        data:      row.op === 'DELETE' ? { id: row.row_id } : row.row_data,
57        timestamp: new Date(row.changed_at).getTime(),
58        source:    'my_postgres_listener',
59      },
60      row.user_id   // null: everyone, uuid: only that user
61    )
62  })
63 
64  listenerClient.on('error', (err) => {
65    console.error('[pg] Listener connection error:', err)
66    // A dropped LISTEN connection means live updates silently stop until this
67    // reconnects. Add real reconnect with backoff before shipping.
68  })
69}
70 
71module.exports = { startPostgresListener }

1# server/postgres_listener.py
2#
3# Same bridge using asyncpg's native LISTEN and NOTIFY support. Forwards into
4# the registry your chosen transport already built.
5#
6# Rows are FULL rows (INSERT and UPDATE) or {"id": ...} (DELETE). The NOTIFY
7# carries user_id: None means a GLOBAL row sent to everyone, a uuid means a
8# PER USER row sent only to that user's connections.
9 
10import asyncio
11import json
12import asyncpg
13from broadcast import broadcast_to_ws  # or sse_registry or socketio's sio
14 
15def table_to_channel(table: str) -> str:
16    return table
17 
18async def start_postgres_listener(dsn: str):
19    # A dedicated, unpooled connection. LISTEN needs one connection held open
20    # for the whole process lifetime.
21    conn = await asyncpg.connect(dsn)
22 
23    async def on_notify(connection, pid, channel, payload):
24        data = json.loads(payload)
25 
26        row = await conn.fetchrow(
27            "SELECT row_id, op, row_data, user_id, changed_at FROM flux_changelog WHERE id = $1",
28            data["id"],
29        )
30        if not row:
31            return
32 
33        op = "CREATE" if row["op"] == "INSERT" else row["op"]
34        row_data = json.loads(row["row_data"]) if row["row_data"] else {}
35        owner = str(row["user_id"]) if row["user_id"] else None
36 
37        await broadcast_to_ws(
38            table_to_channel(data["table"]),
39            {
40                "entity": data["table"],
41                "id": row["row_id"],
42                "op": op,
43                "data": {"id": row["row_id"]} if row["op"] == "DELETE" else row_data,
44                "timestamp": int(row["changed_at"].timestamp() * 1000),
45                "source": "my_postgres_listener",
46            },
47            owner,
48        )
49 
50    await conn.add_listener("flux_changes", on_notify)
51    print("[pg] Listening for flux_changes notifications")
52 
53    # This connection only listens. Keep the coroutine alive.
54    while True:
55        await asyncio.sleep(3600)

1-- Optional. With single state dedup flux_changelog grows only with the number
2-- of distinct rows ever touched, plus one tombstone per deletion that is never
3-- replaced. Cleanup mostly matters for those tombstones.
4--
5-- Retention is a pair: this cleanup purges entries older than 14 days, and
6-- flux_request_delta refuses (too_old) any anchor older than 13 days. The one
7-- day gap means a catch up can never be answered from a half purged window.
8-- If you change one number, change the other, and the client setting
9-- deltaRetentionMs to match the smaller one.
10--
11-- On Supabase pg_cron is available under Database, Extensions. Elsewhere you
12-- may need to install it, or call the same DELETE from an external scheduler.
13create or replace function public.flux_changelog_cleanup(retain_days int default 14)
14returns void
15language sql
16security definer
17set search_path = public
18as $$
19  delete from public.flux_changelog
20  where changed_at < now() - (retain_days || ' days')::interval;
21$$;
22 
23select cron.schedule('flux_changelog_cleanup', '0 3 * * *',
24  $$select public.flux_changelog_cleanup(14)$$);

1-- Only if you ran the OLDER version of this schema that stored sparse diffs.
2-- Run once, after Steps 1 to 3.
3 
4-- 1. The diff function is no longer used.
5drop function if exists public.flux_jsonb_diff(jsonb, jsonb);
6 
7-- 2. Replace every pending UPDATE entry with the current full row, so no
8--    client ever receives a partial record. Repeat per tracked table.
9update public.flux_changelog c
10set    row_data = to_jsonb(t)
11from   public.your_table t
12where  c.table_name = 'your_table'
13  and  c.op         = 'UPDATE'
14  and  c.row_id     = t.id::text;
15 
16-- 3. Fill the owner on existing entries. Deletes carry no row to resolve an
17--    owner from, so old tombstones stay global and expose only an id.
18update public.flux_changelog
19set    user_id = public.flux_resolve_owner(table_name, row_data)
20where  op <> 'DELETE';

1-- a) Every INSERT or UPDATE entry carries a full record with its id.
2select table_name, row_id, op
3from public.flux_changelog
4where op in ('INSERT', 'UPDATE') and not (row_data ? 'id');
5 
6-- b) Per user tables have owners, global tables do not.
7select table_name,
8       count(*) filter (where user_id is null)     as global_rows,
9       count(*) filter (where user_id is not null) as user_rows
10from public.flux_changelog
11group by table_name
12order by table_name;
13 
14-- c) The delta RPC does not error for a caller with no auth.uid(). The SQL
15--    Editor has none, so this only sees global rows.
16select public.flux_request_delta(
17  'tbl:profiles', 'profiles',
18  (extract(epoch from now() - interval '2 days') * 1000)::bigint
19);
20 
21-- d) Reconcile reports ids the server does not have.
22select public.flux_reconcile('projects', array['definitely not an id']);

1-- In a second psql session or the SQL Editor: fire a manual NOTIFY to confirm
2-- your listener process receives them, independent of any real trigger.
3select pg_notify('flux_changes', '{"id": 0, "table": "test", "user_id": null}');