Sync Anchors & Delta Catch-Up

1// What Flux hands your adapter on every reconnect attempt.
2type SyncAnchorMap = Record<string, number>
3 
4// Example — one entry per subscribed channel:
5{
6  'items': 1718000000000,
7  'notifications': 1718000001234,
8}

1interface RealtimeAdapter {
2  // Called before every reconnect — encode these into your handshake
3  // however your transport supports it (query string, initial frame, headers).
4  notifyReconnect?(anchors: SyncAnchorMap): void
5 
6  // Called once IDB data is confirmed present for a channel — order an
7  // immediate historical backfill over your already-open connection.
8  sync?(channel: string, lastKnownServerTime: number): Promise<void>
9 
10  // Fire this the moment a channel's subscription is confirmed by your
11  // server, so Flux can trigger catch-up and refresh the local TTL clock.
12  onChannelSubscribed?(cb: (channel: string) => void): void
13 
14  // ...connect, disconnect, subscribe, unsubscribe
15}

1-- Flux Delta Sync for Supabase — paste this once in your project's
2-- SQL Editor. Nothing here runs on our servers.
3 
4-- 1. Single-state Changelog table — holds at most ONE row per entity.
5create table if not exists public.flux_changelog (
6  id         bigint generated always as identity primary key,
7  table_name text        not null,
8  row_id     text        not null,
9  op         text        not null check (op in ('INSERT', 'UPDATE', 'DELETE')),
10  row_data   jsonb,
11  changed_at timestamptz not null default now()
12);
13 
14create unique index if not exists flux_changelog_table_row_idx
15  on public.flux_changelog (table_name, row_id);
16 
17create index if not exists flux_changelog_table_changed_idx
18  on public.flux_changelog (table_name, changed_at);
19 
20-- 2. Sparse field-level diff for UPDATE operations.
21create or replace function public.flux_jsonb_diff(p_old jsonb, p_new jsonb)
22returns jsonb
23language sql
24immutable
25as $$
26  select jsonb_object_agg(key, value)
27  from jsonb_each(p_new)
28  where not (p_old ? key)
29     or (p_old -> key) <> value
30     or key = 'id';
31$$;
32 
33-- 3. Single-state Trigger function.
34create or replace function public.flux_log_change()
35returns trigger
36language plpgsql
37security definer
38set search_path = public
39as $$
40declare
41  pk_value      text;
42  v_op          text;
43  v_data        jsonb;
44  v_existing_op text;
45  v_existing    jsonb;
46begin
47  if TG_OP = 'DELETE' then
48    execute 'select ($1).id::text' into pk_value using OLD;
49    v_op   := 'DELETE';
50    v_data := jsonb_build_object('id', pk_value);
51 
52  elsif TG_OP = 'UPDATE' then
53    execute 'select ($1).id::text' into pk_value using NEW;
54    v_op   := 'UPDATE';
55    v_data := public.flux_jsonb_diff(to_jsonb(OLD), to_jsonb(NEW));
56 
57    select op, row_data into v_existing_op, v_existing
58    from public.flux_changelog
59    where table_name = TG_TABLE_NAME and row_id = pk_value;
60 
61    if v_existing_op = 'INSERT' then
62      v_op   := 'INSERT';
63      v_data := to_jsonb(NEW);
64    elsif v_existing_op = 'UPDATE' then
65      v_data := v_existing || v_data;
66    end if;
67 
68  else -- INSERT
69    execute 'select ($1).id::text' into pk_value using NEW;
70    v_op   := 'INSERT';
71    v_data := to_jsonb(NEW);
72  end if;
73 
74  delete from public.flux_changelog
75  where table_name = TG_TABLE_NAME
76    and row_id = pk_value;
77 
78  insert into public.flux_changelog (table_name, row_id, op, row_data)
79  values (TG_TABLE_NAME, pk_value, v_op, v_data);
80 
81  if TG_OP = 'DELETE' then
82    return OLD;
83  else
84    return NEW;
85  end if;
86end;
87$$;
88 
89-- 4. Attach to target table.
90drop trigger if exists flux_changelog_trigger on public.your_table;
91create trigger flux_changelog_trigger
92  after insert or update or delete on public.your_table
93  for each row execute function public.flux_log_change();
94 
95-- 5. RPC Function for delta retrieval over Supabase Realtime broadcast.
96create or replace function public.flux_request_delta(
97  p_channel  text,
98  p_table    text,
99  p_since_ms bigint,
100  p_limit    int default 2000
101)
102returns void
103language plpgsql
104security definer
105set search_path = public
106as $$
107declare
108  v_rows jsonb;
109begin
110  select coalesce(
111           jsonb_agg(
112             jsonb_build_object(
113               'op',        latest.op,
114               'data',      latest.row_data,
115               'timestamp', (extract(epoch from latest.changed_at) * 1000)::bigint
116             )
117             order by latest.changed_at asc
118           ),
119           '[]'::jsonb
120         )
121    into v_rows
122  from (
123    select distinct on (c.row_id)
124      c.row_id, c.op, c.row_data, c.changed_at
125    from public.flux_changelog c
126    where c.table_name = p_table
127      and c.changed_at > to_timestamp(p_since_ms / 1000.0)
128    order by c.row_id, c.changed_at desc
129    limit p_limit
130  ) latest;
131 
132  perform realtime.send(
133    jsonb_build_object('rows', v_rows),
134    'flux_sync_complete',
135    'flux:' || p_channel,
136    false
137  );
138end;
139$$;
140 
141grant execute on function public.flux_request_delta(text, text, bigint, int) to anon, authenticated;
142alter table public.flux_changelog enable row level security;

1// Node/Socket.IO server — queryChangesSince reads single-state changelog
2//
3// Client adapter emits 'flux_sync_request' { channel, last_sync_anchor }
4// Server emits back '${channel}:flux_sync_complete' { rows }
5socket.on('flux_sync_request', async ({ channel, last_sync_anchor }) => {
6  try {
7    const rows = await queryChangesSince(channel, last_sync_anchor)
8    socket.emit(`${channel}:flux_sync_complete`, {
9      rows: rows.map((r) => ({
10        op:        r.op,        // 'CREATE' | 'UPDATE' | 'DELETE'
11        data:      r.data,      // normalized record or sparse diff
12        timestamp: r.timestamp, // unix ms
13      })),
14    })
15  } catch (err) {
16    socket.emit(`${channel}:flux_sync_complete`, { rows: [] })
17  }
18})

1# Python/Socket.IO equivalent (python-socketio)
2@sio.on("flux_sync_request")
3async def on_sync_request(sid, data):
4    channel = data.get("channel")
5    anchor = data.get("last_sync_anchor", 0)
6    try:
7        rows = await query_changes_since(channel, anchor)
8        payload = [{"op": r["op"], "data": r["data"], "timestamp": r["timestamp"]} for r in rows]
9        await sio.emit(f"{channel}:flux_sync_complete", {"rows": payload}, to=sid)
10    except Exception:
11        await sio.emit(f"{channel}:flux_sync_complete", {"rows": []}, to=sid)

1// Node/WebSocket server — queryChangesSince reads single-state changelog
2//
3// Client adapter sends JSON: { action: 'SYNC_REQUEST', channel, last_sync_anchor }
4// Server responds JSON: { action: 'SYNC_COMPLETE', channel, rows }
5ws.on('message', async (raw) => {
6  const frame = JSON.parse(raw.toString())
7  if (frame.action === 'SYNC_REQUEST' && frame.channel) {
8    const rows = await queryChangesSince(frame.channel, frame.last_sync_anchor)
9    ws.send(JSON.stringify({
10      action:  'SYNC_COMPLETE',
11      channel: frame.channel,
12      rows: rows.map((r) => ({
13        op:        r.op,
14        data:      r.data,
15        timestamp: r.timestamp,
16      })),
17    }))
18  }
19})

1// Node/Express SSE route — GET /api/stream/:channel?syncAnchor=<unixMs>
2//
3// SSE backfill is streamed directly down the open response on connect/reconnect
4router.get('/:channel', async (req, res) => {
5  const { channel } = req.params
6  const syncAnchor = req.query.syncAnchor ? Number(req.query.syncAnchor) : 0
7 
8  res.setHeader('Content-Type', 'text/event-stream')
9  res.setHeader('Cache-Control', 'no-cache')
10  res.setHeader('Connection', 'keep-alive')
11  res.flushHeaders()
12 
13  registerSseClient(channel, res)
14 
15  // Unconditional backfill query (runs for syncAnchor >= 0)
16  const rows = await queryChangesSince(channel, syncAnchor)
17  rows.forEach((r) => {
18    const event = {
19      entity: channel,
20      id: r.data?.id ?? '',
21      op: r.op,
22      data: r.data,
23      timestamp: r.timestamp,
24      source: 'flux-sse-backfill',
25    }
26    res.write(`event: UPDATE\ndata: ${JSON.stringify(event)}\n\n`)
27  })
28})

1// Mongo Change Stream watcher creating single-state FluxChangelog
2const changeStream = Item.watch([], { fullDocument: 'updateLookup' })
3 
4changeStream.on('change', async (change) => {
5  const { operationType, documentKey, fullDocument, updateDescription } = change
6  const opMap = { insert: 'CREATE', update: 'UPDATE', replace: 'UPDATE', delete: 'DELETE' }
7  const op = opMap[operationType]
8  if (!op) return
9 
10  const rowId = documentKey._id.toString()
11 
12  let updatedFields = null
13  if (operationType === 'update' && updateDescription?.updatedFields) {
14    updatedFields = { ...updateDescription.updatedFields }
15    delete updatedFields.updatedAt
16    delete updatedFields.__v
17  }
18 
19  const rowData = fullDocument
20    ? { ...fullDocument.toObject?.() ?? fullDocument, id: rowId }
21    : { id: rowId }
22 
23  // logChange executes upsert on { tableName, rowId }, keeping changelog single-state
24  await logChange('items', rowId, op, rowData, updatedFields)
25})

1// queryChangesSince implementation for MongoDB / Mongoose
2async function queryChangesSince(tableName, sinceMs) {
3  const sinceDate = new Date(Number(sinceMs) || 0)
4 
5  const rawRows = await FluxChangelog.find({
6    tableName,
7    changedAt: { $gt: sinceDate },
8  }).sort({ changedAt: -1 }).lean()
9 
10  const latestByRowId = new Map()
11  for (const row of rawRows) {
12    if (!latestByRowId.has(row.rowId)) latestByRowId.set(row.rowId, row)
13  }
14 
15  return Array.from(latestByRowId.values())
16    .sort((a, b) => new Date(a.changedAt).getTime() - new Date(b.changedAt).getTime())
17    .slice(0, 2000)
18    .map((r) => ({
19      op: r.op,
20      data: r.rowData,
21      timestamp: new Date(r.changedAt).getTime(),
22    }))
23}