feat(database): capture live updates via Postgres pgoutput

modules/database/README.mdx

@@ -47,7 +47,7 @@ since the latter need to go through parsers that are otherwise unnecessary for M
 
 When using MongoDB with a replica set (e.g., MongoDB Atlas), the database module supports configuring read preference, write concern, and read concern through the admin panel at `PATCH /config/database`.
 
-Live document updates on MongoDB also require a replica set or sharded cluster. Local Compose files initialize a single-node replica set so change streams can be exercised. SQL live updates do not use a replica set; they use an internal trigger-backed change queue (not native WAL/binlog CDC).
+Live document updates on MongoDB also require a replica set or sharded cluster. Local Compose files initialize a single-node replica set so change streams can be exercised. PostgreSQL live updates require logical replication (`wal_level=logical`, a `pgoutput` publication, and a replication slot) — the same class of topology tax as a Mongo replica set. MySQL, MariaDB, and SQLite live updates are out of v1.
 
 ### Live updates
 
@@ -60,11 +60,11 @@ unsubscribe({ schema: 'Order', documentId?: string })
 
 Events arrive as `change` with `{ version, operation, schema, documentId, occurredAt, resumeToken }` and contain no document fields. Consumers should refetch through their authorized REST or custom-endpoint path.
 
-MongoDB uses native change streams. PostgreSQL, MySQL, MariaDB, and SQLite capture the same metadata through table triggers and an internal `_cnd_DatabaseChange` log (not a CMS schema). That log is a **queue**, not native CDC: each opted-in write does an extra insert (and the leader later deletes acked rows). `TRUNCATE`, `COPY`, and table-rewrite DDL are not captured. Delivery is at-least-once; duplicates are possible. PostgreSQL and MySQL/MariaDB only consume rows whose `occurred_at` is at least 750ms old so a later autoincrement id is less likely to become visible before an earlier one; that is a timestamp lag, not a commit watermark. A transaction that stays open longer than 750ms after a later row’s insert can still leave a hole (`WHERE id > cursor` never sees the earlier id, and trim then deletes it). SQLite writers are serialized, so lag is 0.
+MongoDB uses native change streams. PostgreSQL uses in-process WAL CDC (`pgoutput` publication + a **temporary** logical slot). That is WAL CDC, Postgres-only — not a changelog table, not triggers, not Debezium. There is **no catch-up**: re-subscribe, leader restart, or slot drop does not replay missed events; clients fetch current data. `LISTEN`/`NOTIFY` is not the capture path.
 
-PostgreSQL wakes the listener with `LISTEN`/`NOTIFY` on a dedicated session connection. A transaction-mode pooler (PgBouncer default, many serverless poolers) cannot keep `LISTEN` and will surface as unsupported/degraded — use a direct/session URI. MySQL, MariaDB, and SQLite poll. The database role needs permission to `CREATE TABLE` / `CREATE TRIGGER` (PostgreSQL also needs `CREATE FUNCTION`, `LISTEN`, and `pg_notify`). MySQL with binary logging often needs `log_bin_trust_function_creators`. Triggers use the schema’s physical primary key (`idField`), not a virtual `_id`.
+The database role needs permission to `CREATE PUBLICATION`, `ALTER PUBLICATION`, and to create a logical replication slot (`REPLICATION` / managed-Postgres logical-replication grants). A transaction-mode pooler cannot speak the replication protocol; use a direct/session URI. Tables without a primary key get `REPLICA IDENTITY FULL` so UPDATE/DELETE can be published. `TRUNCATE` is not emitted as document events. Custom PKs use the schema’s physical primary key (`idField`), not a virtual `_id`.
 
-SQL live updates do not require a replica set. They are not equivalent to Mongo change streams.
+MySQL, MariaDB, and SQLite are unsupported for live updates. Do not enable `realtime` on those engines.
 
 Client subscribers must authenticate. Schemas with document-level authorization reject schema-wide subscriptions and require a document ID plus a `read` check. Admin consumers use `POST /realtime/ticket` for a 30-second handshake token; session JWTs and masterkeys must not be sent from browser code.
 

modules/database/src/realtime/ChangeStreamCoordinator.ts

@@ -39,6 +39,7 @@ export type CoordinatorOptions = {
   parseResumeToken?: (token: string | null | undefined) => unknown | undefined;
   prepare?: () => Promise<void>;
   onResumePersisted?: (resumeToken: string) => Promise<void>;
+  persistResume?: boolean;
   leaderLock?: string;
   resumeTokenKey?: string;
 };
@@ -141,6 +142,10 @@ export class ChangeStreamCoordinator {
     return this.options.resumeTokenKey ?? RESUME_TOKEN_KEY;
   }
 
+  private get persistResume(): boolean {
+    return this.options.persistResume !== false;
+  }
+
   private async safePrepare(): Promise<void> {
     try {
       await this.options.prepare?.();
@@ -200,16 +205,15 @@ export class ChangeStreamCoordinator {
     this.streamState = 'starting';
     this.ignoreClose = false;
     try {
-      const parseToken = this.options.parseResumeToken ?? parseMongoResumeToken;
-      const resumeAfter = parseToken(
-        await this.options.grpcSdk.state!.getKey(this.resumeTokenName),
-      );
+      const resumeAfter = this.persistResume
+        ? (this.options.parseResumeToken ?? parseMongoResumeToken)(
+            await this.options.grpcSdk.state!.getKey(this.resumeTokenName),
+          )
+        : undefined;
       if (this.watching || this.closed) return;
       const stream = this.options.watch({ resumeAfter });
       this.stream = stream;
       this.watching = true;
-      this.streamState = 'live';
-      this.retryAttempt = 0;
       stream.on('change', (change: unknown) => {
         this.enqueueChange(change as RawChangeEvent);
       });
@@ -222,6 +226,12 @@ export class ChangeStreamCoordinator {
           this.scheduleRetry();
         }
       });
+      if (stream.ready) {
+        await stream.ready;
+      }
+      if (this.closed || !this.watching) return;
+      this.streamState = 'live';
+      this.retryAttempt = 0;
     } catch (err) {
       this.watching = false;
       await this.handleStreamError(err);
@@ -250,15 +260,17 @@ export class ChangeStreamCoordinator {
     const schema = this.resolveSchema(change.ns?.coll);
     const event = schema ? normalizeChangeEvent(change, schema.name) : null;
     if (!event || !schema) {
-      if (token) {
+      if (this.persistResume && token) {
         await this.persistResumeToken(token);
       }
       return;
     }
     this.lastEventAt = event.occurredAt;
     this.lastError = undefined;
     await this.emitChange(schema, event);
-    await this.persistResumeToken(event.resumeToken);
+    if (this.persistResume) {
+      await this.persistResumeToken(event.resumeToken);
+    }
   }
 
   private async persistResumeToken(token: string) {
@@ -348,7 +360,7 @@ export class ChangeStreamCoordinator {
     this.streamState = 'degraded';
     ConduitGrpcSdk.Metrics?.increment('database_realtime_stream_errors_total');
     ConduitGrpcSdk.Logger.error(err as Error);
-    if (isResumeTokenUnusable(err)) {
+    if (this.persistResume && isResumeTokenUnusable(err)) {
       await this.options.grpcSdk.state!.clearKey(this.resumeTokenName);
     }
     await this.stopStream('degraded');

modules/database/src/realtime/RealtimeService.ts

@@ -22,8 +22,7 @@ import { RealtimeSubscriptionTracker } from './subscriptions.js';
 import type { ChangeStreamLike, OptedInSchema, RealtimeStatus } from './types.js';
 import { topologyFromHello } from './topology.js';
 import { SqlRealtimeSupport } from './sql/SqlRealtimeSupport.js';
-import { parseSqlResumeId } from './sql/resume.js';
-import { SQL_LEADER_LOCK, SQL_RESUME_TOKEN_KEY } from './sql/constants.js';
+import { SQL_LEADER_LOCK } from './sql/constants.js';
 
 export class RealtimeService {
   private readonly subscriptions: RealtimeSubscriptionTracker;
@@ -56,20 +55,17 @@ export class RealtimeService {
       this.sqlSupport = new SqlRealtimeSupport(adapter);
       this.coordinator = new ChangeStreamCoordinator({
         grpcSdk,
-        watch: options => this.sqlSupport!.openWatch(options.resumeAfter),
+        watch: () => this.sqlSupport!.openWatch(),
         checkTopology: () => this.sqlSupport!.checkTopology(),
         getOptedInSchemas: () => this.getOptedInSchemas(),
         subscriptions: this.subscriptions,
         enabled: () => this.isGloballyEnabled(),
-        parseResumeToken: parseSqlResumeId,
+        persistResume: false,
         leaderLock: SQL_LEADER_LOCK,
-        resumeTokenKey: SQL_RESUME_TOKEN_KEY,
         prepare: () =>
           this.sqlSupport!.prepare(
             this.isGloballyEnabled() ? this.getOptedInSchemas() : [],
-            { ensureLog: this.isGloballyEnabled() },
           ),
-        onResumePersisted: token => this.sqlSupport!.trimThrough(token),
       });
     }
   }

modules/database/src/realtime/__tests__/coordinator.test.ts

@@ -4,8 +4,7 @@ import { EJSON, ObjectId } from 'bson';
 import { ChangeStreamCoordinator } from '../ChangeStreamCoordinator.js';
 import { RealtimeSubscriptionTracker } from '../subscriptions.js';
 import { roomsForPublicChange } from '../rooms.js';
-import { parseSqlResumeId } from '../sql/resume.js';
-import { SQL_LEADER_LOCK, SQL_RESUME_TOKEN_KEY } from '../sql/constants.js';
+import { SQL_LEADER_LOCK } from '../sql/constants.js';
 
 class MemoryStore {
   private sets = new Map<string, Set<string>>();
@@ -46,14 +45,22 @@ function createCoordinator(overrides?: {
   getKeyDelayMs?: number;
   onResumePersisted?: (token: string) => Promise<void>;
   parseResumeToken?: (token: string | null | undefined) => unknown | undefined;
+  persistResume?: boolean;
   leaderLock?: string;
   resumeTokenKey?: string;
   adminPush?: () => Promise<void>;
+  watchReady?: Promise<void>;
 }) {
-  const stream = new EventEmitter() as EventEmitter & { close: () => Promise<void> };
+  const stream = new EventEmitter() as EventEmitter & {
+    close: () => Promise<void>;
+    ready?: Promise<void>;
+  };
   stream.close = async () => {
     stream.emit('close');
   };
+  if (overrides?.watchReady) {
+    stream.ready = overrides.watchReady;
+  }
   const state = new Map<string, string>();
   const lock = {
     extend: jest.fn(async () => lock),
@@ -106,6 +113,7 @@ function createCoordinator(overrides?: {
     enabled: () => true,
     onResumePersisted: overrides?.onResumePersisted,
     parseResumeToken: overrides?.parseResumeToken,
+    persistResume: overrides?.persistResume,
     leaderLock: overrides?.leaderLock,
     resumeTokenKey: overrides?.resumeTokenKey,
   });
@@ -320,14 +328,14 @@ describe('ChangeStreamCoordinator', () => {
     await coordinator.shutdown();
   });
 
-  it('fans out SQL-shaped log events without document fields', async () => {
-    const { coordinator, stream, publish } = createCoordinator();
+  it('fans out SQL-shaped WAL events without document fields', async () => {
+    const { coordinator, stream, publish } = createCoordinator({ persistResume: false });
     await coordinator.reconcile();
     stream.emit('change', {
       operationType: 'update',
       ns: { coll: 'orders' },
       documentKey: { _id: 'order-1' },
-      _id: '1842',
+      _id: '0/16B3748:12:1',
       wallTime: new Date('2026-03-01T00:00:00.000Z'),
       fullDocument: { secret: 'nope' },
     });
@@ -344,15 +352,79 @@ describe('ChangeStreamCoordinator', () => {
     await coordinator.shutdown();
   });
 
-  it('ignores leftover Mongo tokens on the SQL resume key', async () => {
-    const { coordinator, watch, state } = createCoordinator({
-      parseResumeToken: parseSqlResumeId,
+  it('opens a SQL watch without resume catch-up', async () => {
+    const { coordinator, watch, grpcSdk } = createCoordinator({
+      persistResume: false,
       leaderLock: SQL_LEADER_LOCK,
-      resumeTokenKey: SQL_RESUME_TOKEN_KEY,
     });
-    state.set(SQL_RESUME_TOKEN_KEY, EJSON.stringify({ _data: 'mongo' }));
     await coordinator.reconcile();
     expect(watch).toHaveBeenCalledWith({ resumeAfter: undefined });
+    expect(grpcSdk.state.tryAcquireLock).toHaveBeenCalledWith(
+      SQL_LEADER_LOCK,
+      expect.any(Number),
+    );
+    expect(grpcSdk.state.getKey).not.toHaveBeenCalled();
+    await coordinator.shutdown();
+  });
+
+  it('does not persist resume tokens when persistResume is false', async () => {
+    const { coordinator, stream, grpcSdk } = createCoordinator({ persistResume: false });
+    await coordinator.reconcile();
+    stream.emit('change', {
+      operationType: 'insert',
+      ns: { coll: 'orders' },
+      documentKey: { _id: 'order-1' },
+      _id: '0/1:1:1',
+      wallTime: new Date('2026-03-01T00:00:00.000Z'),
+    });
+    await coordinator.waitForIdle();
+    expect(grpcSdk.state.setKey).not.toHaveBeenCalled();
+    await coordinator.shutdown();
+  });
+
+  it('stays starting until the watch is ready', async () => {
+    let resolveReady: () => void = () => undefined;
+    const watchReady = new Promise<void>(resolve => {
+      resolveReady = resolve;
+    });
+    const { coordinator } = createCoordinator({
+      persistResume: false,
+      watchReady,
+    });
+    const reconcile = coordinator.reconcile();
+    await waitFor(() => coordinator.getState() === 'starting');
+    expect(coordinator.getState()).toBe('starting');
+    resolveReady();
+    await reconcile;
+    expect(coordinator.getState()).toBe('live');
+    await coordinator.shutdown();
+  });
+
+  it('retries as degraded when the watch errors before it is live', async () => {
+    let resolveReady: () => void = () => undefined;
+    const watchReady = new Promise<void>(resolve => {
+      resolveReady = resolve;
+    });
+    const { coordinator, stream } = createCoordinator({
+      persistResume: false,
+      watchReady,
+    });
+    const reconcile = coordinator.reconcile();
+    await waitFor(() => coordinator.getState() === 'starting');
+    stream.emit('error', new Error('all replication slots are in use'));
+    resolveReady();
+    await reconcile;
+    await waitFor(() => coordinator.getState() === 'degraded');
+    expect(coordinator.getState()).toBe('degraded');
     await coordinator.shutdown();
   });
 });
+
+async function waitFor(predicate: () => boolean, timeoutMs = 2000): Promise<void> {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    if (predicate()) return;
+    await new Promise(resolve => setImmediate(resolve));
+  }
+  throw new Error('timed out waiting for condition');
+}