ConduitPlatform · kkopanidis · Sep 13, 2026 · Sep 13, 2026 · Sep 13, 2026
diff --git a/modules/database/README.mdx b/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 engines do not need a replica set.
+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 updates
 
@@ -60,7 +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). PostgreSQL wakes the listener with `NOTIFY`; the other SQL dialects poll. The database role needs permission to `CREATE TRIGGER` (and on PostgreSQL, to create a function). SQL live updates do not require a replica set.
+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.
+
+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`.
+
+SQL live updates do not require a replica set. They are not equivalent to Mongo change streams.
 
 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.
 

diff --git a/modules/database/src/realtime/ChangeStreamCoordinator.ts b/modules/database/src/realtime/ChangeStreamCoordinator.ts
@@ -1,3 +1,4 @@
+import { EJSON } from 'bson';
 import { ConduitGrpcSdk } from '@conduitplatform/grpc-sdk';
 import {
   normalizeChangeEvent,
@@ -13,7 +14,7 @@ import type {
   RealtimeStatusCode,
 } from './types.js';
 import type { RealtimeSubscriptionTracker } from './subscriptions.js';
-import { canReadDocument, type AuthorizationSdk } from './authorize.js';
+import { documentReadDecision, type AuthorizationSdk } from './authorize.js';
 
 const LEADER_LOCK = 'realtime:change-stream:leader';
 const RESUME_TOKEN_KEY = 'realtime:resumeToken';
@@ -38,6 +39,8 @@ export type CoordinatorOptions = {
   parseResumeToken?: (token: string | null | undefined) => unknown | undefined;
   prepare?: () => Promise<void>;
   onResumePersisted?: (resumeToken: string) => Promise<void>;
+  leaderLock?: string;
+  resumeTokenKey?: string;
 };
 
 export class ChangeStreamCoordinator {
@@ -54,6 +57,7 @@ export class ChangeStreamCoordinator {
   private watching = false;
   private opening = false;
   private ignoreClose = false;
+  private changeQueue: Promise<void> = Promise.resolve();
 
   constructor(private readonly options: CoordinatorOptions) {}
 
@@ -73,6 +77,10 @@ export class ChangeStreamCoordinator {
     return this.topology;
   }
 
+  async waitForIdle(): Promise<void> {
+    await this.changeQueue;
+  }
+
   async reconcile(): Promise<void> {
     if (this.closed) return;
     if (!this.options.enabled()) {
@@ -120,10 +128,19 @@ export class ChangeStreamCoordinator {
   async shutdown(): Promise<void> {
     this.closed = true;
     this.clearTimers();
+    await this.changeQueue;
     await this.stopStream('idle');
     await this.releaseLeader();
   }
 
+  private get leaderLockName(): string {
+    return this.options.leaderLock ?? LEADER_LOCK;
+  }
+
+  private get resumeTokenName(): string {
+    return this.options.resumeTokenKey ?? RESUME_TOKEN_KEY;
+  }
+
   private async safePrepare(): Promise<void> {
     try {
       await this.options.prepare?.();
@@ -141,7 +158,7 @@ export class ChangeStreamCoordinator {
     }
     try {
       const acquired = await this.options.grpcSdk.state!.tryAcquireLock(
-        LEADER_LOCK,
+        this.leaderLockName,
         LOCK_TTL_MS,
       );
       if (!acquired) {
@@ -185,7 +202,7 @@ export class ChangeStreamCoordinator {
     try {
       const parseToken = this.options.parseResumeToken ?? parseMongoResumeToken;
       const resumeAfter = parseToken(
-        await this.options.grpcSdk.state!.getKey(RESUME_TOKEN_KEY),
+        await this.options.grpcSdk.state!.getKey(this.resumeTokenName),
       );
       if (this.watching || this.closed) return;
       const stream = this.options.watch({ resumeAfter });
@@ -194,7 +211,7 @@ export class ChangeStreamCoordinator {
       this.streamState = 'live';
       this.retryAttempt = 0;
       stream.on('change', (change: unknown) => {
-        void this.handleChange(change as RawChangeEvent);
+        this.enqueueChange(change as RawChangeEvent);
       });
       stream.on('error', (err: unknown) => {
         void this.handleStreamError(err);
@@ -213,31 +230,60 @@ export class ChangeStreamCoordinator {
     }
   }
 
+  private enqueueChange(change: RawChangeEvent) {
+    this.changeQueue = this.changeQueue.then(async () => {
+      if (this.closed || !this.watching) return;
+      try {
+        await this.handleChange(change);
+      } catch (err) {
+        this.lastError = err instanceof Error ? err.message : String(err);
+        ConduitGrpcSdk.Logger.error(err as Error);
+        this.watching = false;
+        await this.stopStream('degraded');
+        this.scheduleRetry();
+      }
+    });
+  }
+
   private async handleChange(change: RawChangeEvent) {
+    const token = resumeTokenOf(change);
     const schema = this.resolveSchema(change.ns?.coll);
-    if (!schema) return;
-    const event = normalizeChangeEvent(change, schema.name);
-    if (!event) return;
+    const event = schema ? normalizeChangeEvent(change, schema.name) : null;
+    if (!event || !schema) {
+      if (token) {
+        await this.persistResumeToken(token);
+      }
+      return;
+    }
     this.lastEventAt = event.occurredAt;
     this.lastError = undefined;
-    await this.options.grpcSdk.state!.setKey(RESUME_TOKEN_KEY, event.resumeToken);
+    await this.emitChange(schema, event);
+    await this.persistResumeToken(event.resumeToken);
+  }
+
+  private async persistResumeToken(token: string) {
+    await this.options.grpcSdk.state!.setKey(this.resumeTokenName, token);
     try {
-      await this.options.onResumePersisted?.(event.resumeToken);
+      await this.options.onResumePersisted?.(token);
     } catch (err) {
       ConduitGrpcSdk.Logger.error(err as Error);
     }
-    this.options.grpcSdk.bus?.publish(
-      `database:change:${schema.name}`,
-      JSON.stringify(event),
-    );
+  }
+
+  private async emitChange(schema: OptedInSchema, event: DatabaseChangeEvent) {
+    const payload = JSON.stringify(event);
+    this.options.grpcSdk.bus?.publish(`database:change:${schema.name}`, payload);
     ConduitGrpcSdk.Metrics?.increment('database_realtime_events_total', 1, {
       operation: event.operation,
     });
-    await this.pushEvent(schema, event);
+    await this.pushEvent(schema, event, payload);
   }
 
-  private async pushEvent(schema: OptedInSchema, event: DatabaseChangeEvent) {
-    const payload = JSON.stringify(event);
+  private async pushEvent(
+    schema: OptedInSchema,
+    event: DatabaseChangeEvent,
+    payload: string,
+  ) {
     const adminRooms = roomsForPublicChange(schema.name, event.documentId);
     await this.safePush('admin', adminRooms, payload);
     if (!schema.authorizationEnabled) {
@@ -250,21 +296,23 @@ export class ChangeStreamCoordinator {
     );
     const allowedRooms: string[] = [];
     for (const userId of userIds) {
-      const allowed = await canReadDocument(
+      const decision = await documentReadDecision(
         this.options.grpcSdk as unknown as AuthorizationSdk,
         schema.name,
         event.documentId,
         userId,
       );
-      if (!allowed) {
+      if (decision === 'allow') {
+        allowedRooms.push(authorizedDocumentRoom(schema.name, event.documentId, userId));
+        continue;
+      }
+      if (decision === 'deny') {
         await this.options.subscriptions.removeUser(
           schema.name,
           event.documentId,
           userId,
         );
-        continue;
       }
-      allowedRooms.push(authorizedDocumentRoom(schema.name, event.documentId, userId));
     }
     if (allowedRooms.length > 0) {
       await this.safePush('router', allowedRooms, payload);
@@ -279,16 +327,12 @@ export class ChangeStreamCoordinator {
     const client =
       target === 'admin' ? this.options.grpcSdk.admin : this.options.grpcSdk.router;
     if (!client?.socketPush) return;
-    try {
-      await client.socketPush({
-        event: 'change',
-        data,
-        rooms,
-        receivers: [],
-      });
-    } catch (err) {
-      ConduitGrpcSdk.Logger.error(err as Error);
-    }
+    await client.socketPush({
+      event: 'change',
+      data,
+      rooms,
+      receivers: [],
+    });
   }
 
   private resolveSchema(collectionName?: string): OptedInSchema | undefined {
@@ -305,7 +349,7 @@ export class ChangeStreamCoordinator {
     ConduitGrpcSdk.Metrics?.increment('database_realtime_stream_errors_total');
     ConduitGrpcSdk.Logger.error(err as Error);
     if (isResumeTokenUnusable(err)) {
-      await this.options.grpcSdk.state!.clearKey(RESUME_TOKEN_KEY);
+      await this.options.grpcSdk.state!.clearKey(this.resumeTokenName);
     }
     await this.stopStream('degraded');
     this.scheduleRetry();
@@ -362,3 +406,10 @@ export class ChangeStreamCoordinator {
     }
   }
 }
+
+function resumeTokenOf(change: RawChangeEvent): string | undefined {
+  if (change._id === undefined || change._id === null) {
+    return undefined;
+  }
+  return EJSON.stringify(change._id);
+}
diff --git a/modules/database/src/realtime/RealtimeService.ts b/modules/database/src/realtime/RealtimeService.ts
@@ -22,6 +22,8 @@ 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';
 
 export class RealtimeService {
   private readonly subscriptions: RealtimeSubscriptionTracker;
@@ -55,13 +57,17 @@ export class RealtimeService {
       this.coordinator = new ChangeStreamCoordinator({
         grpcSdk,
         watch: options => this.sqlSupport!.openWatch(options.resumeAfter),
-        checkTopology: async () => ({ supported: true }),
+        checkTopology: () => this.sqlSupport!.checkTopology(),
         getOptedInSchemas: () => this.getOptedInSchemas(),
         subscriptions: this.subscriptions,
         enabled: () => this.isGloballyEnabled(),
+        parseResumeToken: parseSqlResumeId,
+        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),
       });
@@ -138,13 +144,22 @@ export class RealtimeService {
       const optedIn = toOptedInSchema({
         name: schema.name,
         collectionName: schema.collectionName,
+        documentIdField: this.documentIdField(schema.name),
         modelOptions: schema.modelOptions,
       });
       if (optedIn) schemas.push(optedIn);
     }
     return schemas;
   }
 
+  private documentIdField(schemaName: string): string | undefined {
+    const model = this.adapter.models[schemaName];
+    if (model && 'idField' in model && typeof model.idField === 'string') {
+      return model.idField;
+    }
+    return undefined;
+  }
+
   private openWatch(adapter: MongooseAdapter, resumeAfter?: unknown): ChangeStreamLike {
     const db = adapter.mongoose.connection.db;
     if (!db) {

diff --git a/modules/database/src/realtime/__tests__/authorize.test.ts b/modules/database/src/realtime/__tests__/authorize.test.ts
@@ -3,6 +3,7 @@ import { status } from '@grpc/grpc-js';
 import { GrpcError } from '@conduitplatform/grpc-sdk';
 import {
   assertSchemaAvailable,
+  documentReadDecision,
   optionalDocumentId,
   parseSubscribeRequest,
   requireSchemaName,
@@ -66,4 +67,36 @@ describe('realtime authorization helpers', () => {
       }),
     );
   });
+
+  it('treats missing authorization and can() failures as unavailable', async () => {
+    expect(
+      await documentReadDecision({ isAvailable: () => false }, 'Order', '1', 'user-1'),
+    ).toBe('unavailable');
+    expect(
+      await documentReadDecision(
+        {
+          isAvailable: () => true,
+          authorization: {
+            can: async () => {
+              throw new Error('down');
+            },
+          },
+        },
+        'Order',
+        '1',
+        'user-1',
+      ),
+    ).toBe('unavailable');
+    expect(
+      await documentReadDecision(
+        {
+          isAvailable: () => true,
+          authorization: { can: async () => ({ allow: false }) },
+        },
+        'Order',
+        '1',
+        'user-1',
+      ),
+    ).toBe('deny');
+  });
 });