diff --git a/modules/database/README.mdx b/modules/database/README.mdx
index 8b3d1aa45..751111223 100644
--- 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
index c6fa87b73..fb6dd48a2 100644
--- 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
index 0786801ed..ab3e227af 100644
--- 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,6 +144,7 @@ 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);
@@ -145,6 +152,14 @@ export class RealtimeService {
     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
index 9c5eba15b..36f32563f 100644
--- 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');
+  });
 });
diff --git a/modules/database/src/realtime/__tests__/coordinator.test.ts b/modules/database/src/realtime/__tests__/coordinator.test.ts
index 55fa5baa9..f17784985 100644
--- a/modules/database/src/realtime/__tests__/coordinator.test.ts
+++ b/modules/database/src/realtime/__tests__/coordinator.test.ts
@@ -1,9 +1,11 @@
 import { EventEmitter } from 'node:events';
 import { describe, expect, it, jest } from '@jest/globals';
-import { ObjectId } from 'bson';
+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';
 
 class MemoryStore {
   private sets = new Map<string, Set<string>>();
@@ -11,9 +13,13 @@ class MemoryStore {
     const set = this.sets.get(key) ?? new Set<string>();
     members.forEach(member => set.add(member));
     this.sets.set(key, set);
+    return members.length;
   }
   async srem(key: string, ...members: string[]) {
-    members.forEach(member => this.sets.get(key)?.delete(member));
+    const set = this.sets.get(key);
+    if (!set) return 0;
+    members.forEach(member => set.delete(member));
+    return members.length;
   }
   async smembers(key: string) {
     return [...(this.sets.get(key) ?? new Set())];
@@ -23,13 +29,26 @@ class MemoryStore {
   }
   async del(...keys: string[]) {
     keys.forEach(key => this.sets.delete(key));
+    return keys.length;
   }
 }
 
 function createCoordinator(overrides?: {
   allow?: boolean;
-  schemas?: { name: string; collectionName: string; authorizationEnabled: boolean }[];
+  authorizationAvailable?: boolean;
+  canThrows?: boolean;
+  schemas?: {
+    name: string;
+    collectionName: string;
+    authorizationEnabled: boolean;
+    documentIdField?: string;
+  }[];
   getKeyDelayMs?: number;
+  onResumePersisted?: (token: string) => Promise<void>;
+  parseResumeToken?: (token: string | null | undefined) => unknown | undefined;
+  leaderLock?: string;
+  resumeTokenKey?: string;
+  adminPush?: () => Promise<void>;
 }) {
   const stream = new EventEmitter() as EventEmitter & { close: () => Promise<void> };
   stream.close = async () => {
@@ -41,7 +60,7 @@ function createCoordinator(overrides?: {
     release: jest.fn(async () => undefined),
   };
   const routerPush = jest.fn(async () => undefined);
-  const adminPush = jest.fn(async () => undefined);
+  const adminPush = jest.fn(overrides?.adminPush ?? (async () => undefined));
   const publish = jest.fn();
   const watch = jest.fn(() => stream as never);
   const subscriptions = new RealtimeSubscriptionTracker(new MemoryStore());
@@ -65,9 +84,14 @@ function createCoordinator(overrides?: {
     bus: { publish },
     router: { socketPush: routerPush },
     admin: { socketPush: adminPush },
-    isAvailable: () => true,
+    isAvailable: () => overrides?.authorizationAvailable !== false,
     authorization: {
-      can: async () => ({ allow: overrides?.allow !== false }),
+      can: async () => {
+        if (overrides?.canThrows) {
+          throw new Error('authorization unavailable');
+        }
+        return { allow: overrides?.allow !== false };
+      },
     },
   };
   const coordinator = new ChangeStreamCoordinator({
@@ -80,6 +104,10 @@ function createCoordinator(overrides?: {
       ],
     subscriptions,
     enabled: () => true,
+    onResumePersisted: overrides?.onResumePersisted,
+    parseResumeToken: overrides?.parseResumeToken,
+    leaderLock: overrides?.leaderLock,
+    resumeTokenKey: overrides?.resumeTokenKey,
   });
   return {
     coordinator,
@@ -114,7 +142,7 @@ describe('ChangeStreamCoordinator', () => {
       documentKey: { _id: new ObjectId('64b64c4c4c4c4c4c4c4c4c4d') },
       _id: resume,
     });
-    await new Promise(resolve => setImmediate(resolve));
+    await coordinator.waitForIdle();
     expect(publish).toHaveBeenCalledTimes(1);
     expect(publish.mock.calls[0][0]).toBe('database:change:Order');
     const payload = JSON.parse(publish.mock.calls[0][1] as string);
@@ -135,6 +163,86 @@ describe('ChangeStreamCoordinator', () => {
     await coordinator.shutdown();
   });
 
+  it('persists skipped collection tokens after the opted-in emit', async () => {
+    const order: string[] = [];
+    const { coordinator, stream, grpcSdk } = createCoordinator({
+      onResumePersisted: async () => {
+        order.push('trim');
+      },
+    });
+    grpcSdk.state.setKey.mockImplementation(async (key: string, value: string) => {
+      order.push(`setKey:${value}`);
+    });
+    await coordinator.reconcile();
+    stream.emit('change', {
+      operationType: 'insert',
+      ns: { coll: 'orders' },
+      documentKey: { _id: new ObjectId('64b64c4c4c4c4c4c4c4c4c4c') },
+      _id: { _data: 'token-a' },
+      wallTime: new Date('2026-01-02T00:00:00.000Z'),
+    });
+    stream.emit('change', {
+      operationType: 'insert',
+      ns: { coll: 'other' },
+      documentKey: { _id: new ObjectId('64b64c4c4c4c4c4c4c4c4c4d') },
+      _id: { _data: 'token-b' },
+    });
+    await coordinator.waitForIdle();
+    expect(order).toEqual([
+      `setKey:${EJSON.stringify({ _data: 'token-a' })}`,
+      'trim',
+      `setKey:${EJSON.stringify({ _data: 'token-b' })}`,
+      'trim',
+    ]);
+    await coordinator.shutdown();
+  });
+
+  it('persists resume and trims only after a successful emit', async () => {
+    const order: string[] = [];
+    const { coordinator, stream, grpcSdk, publish } = createCoordinator({
+      onResumePersisted: async () => {
+        order.push('trim');
+      },
+    });
+    publish.mockImplementation(() => {
+      order.push('publish');
+    });
+    grpcSdk.state.setKey.mockImplementation(async () => {
+      order.push('setKey');
+    });
+    await coordinator.reconcile();
+    stream.emit('change', {
+      operationType: 'update',
+      ns: { coll: 'orders' },
+      documentKey: { _id: 'order-1' },
+      _id: '1842',
+      wallTime: new Date('2026-03-01T00:00:00.000Z'),
+    });
+    await coordinator.waitForIdle();
+    expect(order).toEqual(['publish', 'setKey', 'trim']);
+    await coordinator.shutdown();
+  });
+
+  it('does not persist a later token when emit fails', async () => {
+    const { coordinator, stream, grpcSdk, adminPush } = createCoordinator({
+      adminPush: async () => {
+        throw new Error('socket down');
+      },
+    });
+    await coordinator.reconcile();
+    stream.emit('change', {
+      operationType: 'insert',
+      ns: { coll: 'orders' },
+      documentKey: { _id: 'order-1' },
+      _id: '10',
+      wallTime: new Date('2026-03-01T00:00:00.000Z'),
+    });
+    await coordinator.waitForIdle();
+    expect(grpcSdk.state.setKey).not.toHaveBeenCalled();
+    expect(coordinator.getState()).toBe('degraded');
+    await coordinator.shutdown();
+  });
+
   it('re-checks ReBAC before emission and drops revoked users', async () => {
     const { coordinator, stream, routerPush, subscriptions } = createCoordinator({
       allow: false,
@@ -153,12 +261,37 @@ describe('ChangeStreamCoordinator', () => {
       documentKey: { _id: new ObjectId('64b64c4c4c4c4c4c4c4c4c4c') },
       _id: { _data: 'token' },
     });
-    await new Promise(resolve => setImmediate(resolve));
+    await coordinator.waitForIdle();
     expect(routerPush).not.toHaveBeenCalled();
     expect(await subscriptions.listUsers('Order', '64b64c4c4c4c4c4c4c4c4c')).toEqual([]);
     await coordinator.shutdown();
   });
 
+  it('does not remove users when authorization is unavailable', async () => {
+    const { coordinator, stream, routerPush, subscriptions, grpcSdk } = createCoordinator(
+      {
+        authorizationAvailable: false,
+        schemas: [
+          { name: 'Order', collectionName: 'orders', authorizationEnabled: true },
+        ],
+      },
+    );
+    expect(grpcSdk.isAvailable('authorization')).toBe(false);
+    await subscriptions.addAuthorizedDocument('sock-1', 'Order', 'doc-1', 'user-1');
+    expect(await subscriptions.listUsers('Order', 'doc-1')).toEqual(['user-1']);
+    await coordinator.reconcile();
+    stream.emit('change', {
+      operationType: 'update',
+      ns: { coll: 'orders' },
+      documentKey: { _id: 'doc-1' },
+      _id: { _data: 'token' },
+    });
+    await coordinator.waitForIdle();
+    expect(routerPush).not.toHaveBeenCalled();
+    expect(await subscriptions.listUsers('Order', 'doc-1')).toEqual(['user-1']);
+    await coordinator.shutdown();
+  });
+
   it('retries later when the leader lock is held by another instance', async () => {
     jest.useFakeTimers();
     const { coordinator, grpcSdk, lock } = createCoordinator();
@@ -198,7 +331,7 @@ describe('ChangeStreamCoordinator', () => {
       wallTime: new Date('2026-03-01T00:00:00.000Z'),
       fullDocument: { secret: 'nope' },
     });
-    await new Promise(resolve => setImmediate(resolve));
+    await coordinator.waitForIdle();
     expect(publish).toHaveBeenCalledTimes(1);
     const payload = JSON.parse(publish.mock.calls[0][1] as string);
     expect(payload).toMatchObject({
@@ -210,4 +343,16 @@ describe('ChangeStreamCoordinator', () => {
     expect(JSON.stringify(payload)).not.toContain('nope');
     await coordinator.shutdown();
   });
+
+  it('ignores leftover Mongo tokens on the SQL resume key', async () => {
+    const { coordinator, watch, state } = createCoordinator({
+      parseResumeToken: parseSqlResumeId,
+      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 });
+    await coordinator.shutdown();
+  });
 });
diff --git a/modules/database/src/realtime/__tests__/sql-builders.test.ts b/modules/database/src/realtime/__tests__/sql-builders.test.ts
index 37692e2d1..6f7b35fae 100644
--- a/modules/database/src/realtime/__tests__/sql-builders.test.ts
+++ b/modules/database/src/realtime/__tests__/sql-builders.test.ts
@@ -3,8 +3,10 @@ import { EJSON } from 'bson';
 import { parseSqlResumeId, sqlCursorFromResumeAfter } from '../sql/resume.js';
 import { createCaptureFunctionSql, createChangeLogTableSql } from '../sql/ddl.js';
 import { desiredTriggers } from '../sql/triggerSql.js';
-import { CHANGE_LOG_TABLE, CHANGE_LOG_FUNCTION } from '../sql/constants.js';
+import { CHANGE_LOG_TABLE, PK_COLUMN } from '../sql/constants.js';
+import { fetchChangeLogSql } from '../sql/changelog.js';
 import {
+  captureFunctionName,
   fitIdentifier,
   quoteIdent,
   rowTriggerName,
@@ -27,19 +29,24 @@ describe('SQL realtime builders', () => {
     expect(rowTriggerName(long, 'i', 'mysql').length).toBeLessThanOrEqual(64);
   });
 
-  it('builds a postgres capture function and per-table trigger', () => {
-    const fn = createCaptureFunctionSql();
-    expect(fn).toContain(CHANGE_LOG_FUNCTION);
+  it('builds a postgres capture function and per-table trigger using the physical PK', () => {
+    const functionName = captureFunctionName('orders');
+    const fn = createCaptureFunctionSql('sku', functionName);
+    expect(fn).toContain(quoteIdent('postgres', functionName));
     expect(fn).toContain(CHANGE_LOG_TABLE);
+    expect(fn).toContain('"sku"');
+    expect(fn).not.toContain('"_id"');
     expect(fn).toContain('pg_notify');
-    const triggers = desiredTriggers('postgres', 'orders');
+    const triggers = desiredTriggers('postgres', 'orders', 'sku');
     expect(triggers).toHaveLength(1);
     expect(triggers[0].sql).toMatch(/AFTER INSERT OR UPDATE OR DELETE/);
     expect(triggers[0].sql).toContain('EXECUTE PROCEDURE');
+    expect(triggers[0].sql).toContain(quoteIdent('postgres', functionName));
     expect(triggers[0].dropSql).toMatch(/DROP TRIGGER IF EXISTS/);
+    expect(triggers[0].functionSql).toContain('"sku"');
   });
 
-  it('builds three row triggers for mysql and sqlite', () => {
+  it('builds three row triggers for mysql and sqlite that skip NULL PKs', () => {
     expect(desiredTriggers('mysql', 'orders')).toHaveLength(3);
     expect(desiredTriggers('mariadb', 'cnd_User')).toHaveLength(3);
     const sqlite = desiredTriggers('sqlite', 'orders');
@@ -49,6 +56,8 @@ describe('SQL realtime builders', () => {
       rowTriggerName('orders', 'd', 'sqlite'),
     ]);
     expect(sqlite[0].sql).toContain(CHANGE_LOG_TABLE);
+    expect(sqlite[0].sql).toContain('IS NOT NULL');
+    expect(desiredTriggers('mysql', 'orders', PK_COLUMN)[0].sql).toContain('IS NOT NULL');
   });
 
   it('builds dialect-specific change-log tables', () => {
@@ -56,6 +65,12 @@ describe('SQL realtime builders', () => {
     expect(createChangeLogTableSql('mysql')).toMatch(/AUTO_INCREMENT/);
     expect(createChangeLogTableSql('sqlite')).toMatch(/AUTOINCREMENT/);
   });
+
+  it('adds a commit-visibility lag predicate for postgres and mysql', () => {
+    expect(fetchChangeLogSql('postgres', 750)).toContain('make_interval');
+    expect(fetchChangeLogSql('mysql', 750)).toContain('DATE_SUB');
+    expect(fetchChangeLogSql('sqlite', 0)).not.toContain('datetime');
+  });
 });
 
 describe('SQL resume tokens', () => {
@@ -66,6 +81,7 @@ describe('SQL resume tokens', () => {
     expect(parseSqlResumeId(EJSON.stringify({ _data: 'mongo' }))).toBeUndefined();
     expect(sqlCursorFromResumeAfter(1842)).toBe('1842');
     expect(sqlCursorFromResumeAfter({ _data: 'mongo' })).toBeUndefined();
+    expect(sqlCursorFromResumeAfter(Number.MAX_SAFE_INTEGER + 1)).toBeUndefined();
   });
 });
 
diff --git a/modules/database/src/realtime/__tests__/sql-change-stream.integration.test.ts b/modules/database/src/realtime/__tests__/sql-change-stream.integration.test.ts
index 706bb15d8..be5483f82 100644
--- a/modules/database/src/realtime/__tests__/sql-change-stream.integration.test.ts
+++ b/modules/database/src/realtime/__tests__/sql-change-stream.integration.test.ts
@@ -1,10 +1,15 @@
 import { describe, expect, it } from '@jest/globals';
 import sqlite3 from 'sqlite3';
+import { QueryTypes, Sequelize } from 'sequelize';
 import { normalizeChangeEvent } from '../normalize.js';
 import { CHANGE_LOG_TABLE } from '../sql/constants.js';
 import { createChangeLogTableSql } from '../sql/ddl.js';
+import { quoteIdent, rowTriggerName } from '../sql/identifiers.js';
 import { toRawChangeEvent } from '../sql/mapEvent.js';
 import { desiredTriggers } from '../sql/triggerSql.js';
+import { ensureChangeLog, fetchChangeLogBatch, trimChangeLog } from '../sql/changelog.js';
+import { syncTriggers } from '../sql/triggers.js';
+import { SqlRealtimeSupport } from '../sql/SqlRealtimeSupport.js';
 
 function run(db: sqlite3.Database, sql: string): Promise<void> {
   return new Promise((resolve, reject) => {
@@ -79,4 +84,247 @@ describe('SQLite change-log contract', () => {
       });
     }
   });
+
+  it('uses the physical primary key for custom-PK tables', async () => {
+    const db = new sqlite3.Database(':memory:');
+    try {
+      await run(db, `CREATE TABLE "orders" (sku TEXT PRIMARY KEY, secret TEXT)`);
+      await run(db, createChangeLogTableSql('sqlite'));
+      for (const trigger of desiredTriggers('sqlite', 'orders', 'sku')) {
+        await run(db, trigger.sql);
+      }
+      await run(db, `INSERT INTO "orders" (sku, secret) VALUES ('sku-1', 'hidden')`);
+      const rows = await all<{ document_id: string }>(
+        db,
+        `SELECT document_id FROM "${CHANGE_LOG_TABLE}"`,
+      );
+      expect(rows).toEqual([{ document_id: 'sku-1' }]);
+    } finally {
+      await new Promise<void>(resolve => {
+        db.close(() => resolve());
+      });
+    }
+  });
+
+  it('skips NULL document ids without aborting the user write', async () => {
+    const db = new sqlite3.Database(':memory:');
+    try {
+      await run(db, `CREATE TABLE "orders" (_id TEXT, secret TEXT)`);
+      await run(db, createChangeLogTableSql('sqlite'));
+      for (const trigger of desiredTriggers('sqlite', 'orders')) {
+        await run(db, trigger.sql);
+      }
+      await run(db, `INSERT INTO "orders" (_id, secret) VALUES (NULL, 'ok')`);
+      const orders = await all<{ secret: string }>(db, `SELECT secret FROM "orders"`);
+      const log = await all<{ id: number }>(db, `SELECT id FROM "${CHANGE_LOG_TABLE}"`);
+      expect(orders).toEqual([{ secret: 'ok' }]);
+      expect(log).toEqual([]);
+    } finally {
+      await new Promise<void>(resolve => {
+        db.close(() => resolve());
+      });
+    }
+  });
+
+  it('retargets leftover #1602 _id triggers onto the physical PK', async () => {
+    const sequelize = new Sequelize({
+      dialect: 'sqlite',
+      storage: ':memory:',
+      logging: false,
+    });
+    try {
+      await sequelize.query(`CREATE TABLE "orders" (sku TEXT PRIMARY KEY, secret TEXT)`);
+      await ensureChangeLog(sequelize);
+      for (const sql of leftoverSqliteIdTriggerSql('orders')) {
+        await sequelize.query(sql);
+      }
+      await expect(
+        sequelize.query(`INSERT INTO "orders" (sku, secret) VALUES ('sku-1', 'hidden')`),
+      ).rejects.toThrow();
+      await syncTriggers(sequelize, [
+        {
+          name: 'Order',
+          collectionName: 'orders',
+          authorizationEnabled: false,
+          documentIdField: 'sku',
+        },
+      ]);
+      await sequelize.query(
+        `INSERT INTO "orders" (sku, secret) VALUES ('sku-1', 'hidden')`,
+      );
+      const rows = await fetchChangeLogBatch(sequelize, '0', 200, 0);
+      expect(rows.map(row => row.document_id)).toEqual(['sku-1']);
+    } finally {
+      await sequelize.close();
+    }
+  });
+
+  it('retargets leftover _id bodies so NULL PKs skip without aborting DML', async () => {
+    const sequelize = new Sequelize({
+      dialect: 'sqlite',
+      storage: ':memory:',
+      logging: false,
+    });
+    try {
+      await sequelize.query(`CREATE TABLE "orders" (_id TEXT, secret TEXT)`);
+      await ensureChangeLog(sequelize);
+      for (const sql of leftoverSqliteIdTriggerSql('orders')) {
+        await sequelize.query(sql);
+      }
+      await expect(
+        sequelize.query(`INSERT INTO "orders" (_id, secret) VALUES (NULL, 'ok')`),
+      ).rejects.toThrow();
+      await syncTriggers(sequelize, [
+        {
+          name: 'Order',
+          collectionName: 'orders',
+          authorizationEnabled: false,
+          documentIdField: '_id',
+        },
+      ]);
+      await sequelize.query(`INSERT INTO "orders" (_id, secret) VALUES (NULL, 'ok')`);
+      const orders = await sequelize.query(`SELECT secret FROM "orders"`, {
+        type: QueryTypes.SELECT,
+      });
+      expect(orders).toEqual([{ secret: 'ok' }]);
+      expect(await fetchChangeLogBatch(sequelize, '0', 200, 0)).toEqual([]);
+    } finally {
+      await sequelize.close();
+    }
+  });
+
+  it('drops triggers on opt-out without breaking DML', async () => {
+    const sequelize = new Sequelize({
+      dialect: 'sqlite',
+      storage: ':memory:',
+      logging: false,
+    });
+    try {
+      await sequelize.query(`CREATE TABLE "orders" (_id TEXT PRIMARY KEY, secret TEXT)`);
+      await ensureChangeLog(sequelize);
+      await syncTriggers(sequelize, [
+        {
+          name: 'Order',
+          collectionName: 'orders',
+          authorizationEnabled: false,
+          documentIdField: '_id',
+        },
+      ]);
+      await sequelize.query(`INSERT INTO "orders" (_id, secret) VALUES ('order-1', 'x')`);
+      expect((await fetchChangeLogBatch(sequelize, '0', 200, 0)).length).toBe(1);
+      await syncTriggers(sequelize, []);
+      await sequelize.query(`INSERT INTO "orders" (_id, secret) VALUES ('order-2', 'y')`);
+      expect((await fetchChangeLogBatch(sequelize, '0', 200, 0)).length).toBe(1);
+    } finally {
+      await sequelize.close();
+    }
+  });
+
+  it('leaves an existing same-name trigger in place on reconcile', async () => {
+    const sequelize = new Sequelize({
+      dialect: 'sqlite',
+      storage: ':memory:',
+      logging: false,
+    });
+    try {
+      await sequelize.query(`CREATE TABLE "orders" (_id TEXT PRIMARY KEY, secret TEXT)`);
+      await ensureChangeLog(sequelize);
+      const schemas = [
+        {
+          name: 'Order',
+          collectionName: 'orders',
+          authorizationEnabled: false,
+          documentIdField: '_id',
+        },
+      ];
+      await syncTriggers(sequelize, schemas);
+      const before = await sequelize.query(
+        `SELECT sql FROM sqlite_master WHERE type = 'trigger' AND name LIKE 'cnd_rt_%' ORDER BY name`,
+        { raw: true },
+      );
+      await syncTriggers(sequelize, schemas);
+      const after = await sequelize.query(
+        `SELECT sql FROM sqlite_master WHERE type = 'trigger' AND name LIKE 'cnd_rt_%' ORDER BY name`,
+        { raw: true },
+      );
+      expect(after).toEqual(before);
+    } finally {
+      await sequelize.close();
+    }
+  });
+
+  it('trims acked ids and starts SqlChangeStream from the prepare watermark', async () => {
+    const sequelize = new Sequelize({
+      dialect: 'sqlite',
+      storage: ':memory:',
+      logging: false,
+    });
+    try {
+      const support = new SqlRealtimeSupport({
+        sequelize,
+        connectionUri: 'sqlite://',
+      } as never);
+      await ensureChangeLog(sequelize);
+      await sequelize.query(
+        `INSERT INTO "${CHANGE_LOG_TABLE}" (collection_name, document_id, operation, occurred_at)
+         VALUES ('orders', 'old', 'insert', datetime('now'))`,
+      );
+      await trimChangeLog(sequelize, '1');
+      expect(await fetchChangeLogBatch(sequelize, '0', 200, 0)).toEqual([]);
+      await sequelize.query(
+        `INSERT INTO "${CHANGE_LOG_TABLE}" (collection_name, document_id, operation, occurred_at)
+         VALUES ('orders', 'old-after-trim', 'insert', datetime('now'))`,
+      );
+      await support.prepare([], { ensureLog: true });
+      await sequelize.query(
+        `INSERT INTO "${CHANGE_LOG_TABLE}" (collection_name, document_id, operation, occurred_at)
+         VALUES ('orders', 'new', 'insert', datetime('now'))`,
+      );
+      const received: unknown[] = [];
+      const stream = support.openWatch();
+      stream.on('change', change => {
+        received.push(change);
+      });
+      await waitUntil(() => received.length >= 1);
+      expect(received).toHaveLength(1);
+      expect(received[0]).toMatchObject({
+        operationType: 'insert',
+        documentKey: { _id: 'new' },
+      });
+      await stream.close();
+    } finally {
+      await sequelize.close();
+    }
+  });
 });
+
+function leftoverSqliteIdTriggerSql(collectionName: string): string[] {
+  const table = quoteIdent('sqlite', collectionName);
+  const logTable = quoteIdent('sqlite', CHANGE_LOG_TABLE);
+  return (
+    [
+      ['i', 'INSERT', 'insert', 'NEW'],
+      ['u', 'UPDATE', 'update', 'NEW'],
+      ['d', 'DELETE', 'delete', 'OLD'],
+    ] as const
+  ).map(([opKey, timing, operation, row]) => {
+    const quotedTrigger = quoteIdent(
+      'sqlite',
+      rowTriggerName(collectionName, opKey, 'sqlite'),
+    );
+    return `CREATE TRIGGER ${quotedTrigger} AFTER ${timing} ON ${table}
+BEGIN
+  INSERT INTO ${logTable} (collection_name, document_id, operation, occurred_at)
+  VALUES ('${collectionName}', ${row}."_id", '${operation}', datetime('now'));
+END`;
+  });
+}
+
+async function waitUntil(predicate: () => boolean, timeoutMs = 2000): Promise<void> {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    if (predicate()) return;
+    await new Promise(resolve => setTimeout(resolve, 10));
+  }
+  throw new Error('timed out waiting for condition');
+}
diff --git a/modules/database/src/realtime/__tests__/sql-realtime-support.test.ts b/modules/database/src/realtime/__tests__/sql-realtime-support.test.ts
new file mode 100644
index 000000000..9b48331b2
--- /dev/null
+++ b/modules/database/src/realtime/__tests__/sql-realtime-support.test.ts
@@ -0,0 +1,200 @@
+import { describe, expect, it, jest } from '@jest/globals';
+import pg from 'pg';
+import { QueryTypes, Sequelize } from 'sequelize';
+import { SqlChangeStream } from '../sql/SqlChangeStream.js';
+import { SqlRealtimeSupport } from '../sql/SqlRealtimeSupport.js';
+import { CHANGE_LOG_TABLE } from '../sql/constants.js';
+import { captureFunctionName, triggerBaseName } from '../sql/identifiers.js';
+import { syncTriggers } from '../sql/triggers.js';
+
+describe('SqlRealtimeSupport', () => {
+  it('does not CREATE TABLE when ensureLog is false', async () => {
+    const sequelize = new Sequelize({
+      dialect: 'sqlite',
+      storage: ':memory:',
+      logging: false,
+    });
+    try {
+      const support = new SqlRealtimeSupport({
+        sequelize,
+        connectionUri: 'sqlite://',
+      } as never);
+      await support.prepare([], { ensureLog: false });
+      const tables = await sequelize.query(
+        `SELECT name FROM sqlite_master WHERE type = 'table' AND name = :name`,
+        { type: QueryTypes.SELECT, replacements: { name: CHANGE_LOG_TABLE } },
+      );
+      expect(tables).toEqual([]);
+    } finally {
+      await sequelize.close();
+    }
+  });
+
+  it('retargets leftover postgres triggers that still call conduit_realtime_capture', async () => {
+    const triggerName = triggerBaseName('orders', 'postgres');
+    const functionName = captureFunctionName('orders');
+    const queries: string[] = [];
+    const sequelize = {
+      getDialect: () => 'postgres',
+      query: async (sql: string) => {
+        queries.push(sql);
+        if (sql.includes('information_schema.triggers')) {
+          return [
+            {
+              trigger_name: triggerName,
+              table_name: 'orders',
+              definition: 'EXECUTE PROCEDURE conduit_realtime_capture()',
+            },
+          ];
+        }
+        return [];
+      },
+    };
+    await syncTriggers(sequelize as never, [
+      {
+        name: 'Order',
+        collectionName: 'orders',
+        authorizationEnabled: false,
+        documentIdField: 'sku',
+      },
+    ]);
+    expect(queries.some(sql => sql.includes('DROP TRIGGER'))).toBe(true);
+    expect(
+      queries.some(
+        sql =>
+          sql.includes('CREATE TRIGGER') &&
+          sql.includes(functionName) &&
+          !sql.includes('conduit_realtime_capture'),
+      ),
+    ).toBe(true);
+  });
+
+  it('leaves a matching postgres trigger in place', async () => {
+    const triggerName = triggerBaseName('orders', 'postgres');
+    const functionName = captureFunctionName('orders');
+    const queries: string[] = [];
+    const sequelize = {
+      getDialect: () => 'postgres',
+      query: async (sql: string) => {
+        queries.push(sql);
+        if (sql.includes('information_schema.triggers')) {
+          return [
+            {
+              trigger_name: triggerName,
+              table_name: 'orders',
+              definition: `EXECUTE PROCEDURE ${functionName}()`,
+            },
+          ];
+        }
+        return [];
+      },
+    };
+    await syncTriggers(sequelize as never, [
+      {
+        name: 'Order',
+        collectionName: 'orders',
+        authorizationEnabled: false,
+        documentIdField: 'sku',
+      },
+    ]);
+    expect(queries.some(sql => sql.includes('DROP TRIGGER'))).toBe(false);
+    expect(queries.some(sql => sql.includes('CREATE TRIGGER'))).toBe(false);
+    expect(queries.some(sql => sql.includes('CREATE OR REPLACE FUNCTION'))).toBe(true);
+  });
+
+  it('reports postgres live updates unsupported when LISTEN fails', async () => {
+    const connect = jest
+      .spyOn(pg.Client.prototype, 'connect')
+      .mockRejectedValue(new Error('LISTEN not allowed'));
+    try {
+      const support = new SqlRealtimeSupport({
+        sequelize: {
+          getDialect: () => 'postgres',
+          query: async () => [[]],
+        },
+        connectionUri: 'postgres://localhost/db',
+      } as never);
+      const result = await support.checkTopology();
+      expect(result.supported).toBe(false);
+      expect(result.message).toMatch(/LISTEN/);
+    } finally {
+      connect.mockRestore();
+    }
+  });
+});
+
+describe('SqlChangeStream drain coalesce', () => {
+  it('fetches again when a notify arrives while a drain is in flight', async () => {
+    let fetches = 0;
+    let releaseFirst!: (rows: Record<string, unknown>[]) => void;
+    const firstFetch = new Promise<Record<string, unknown>[]>(resolve => {
+      releaseFirst = resolve;
+    });
+    const sequelize = {
+      getDialect: () => 'postgres',
+      query: async (sql: string) => {
+        if (typeof sql === 'string' && sql.includes('WHERE id >')) {
+          fetches += 1;
+          if (fetches === 1) {
+            return firstFetch;
+          }
+          if (fetches === 2) {
+            return [
+              {
+                id: 2,
+                collection_name: 'orders',
+                document_id: 'second',
+                operation: 'insert',
+                occurred_at: new Date().toISOString(),
+              },
+            ];
+          }
+          return [];
+        }
+        return [];
+      },
+    };
+    const listeners: Record<string, () => void> = {};
+    const fakeClient = {
+      on(event: string, cb: () => void) {
+        listeners[event] = cb;
+      },
+      connect: async () => undefined,
+      query: async () => undefined,
+      end: async () => undefined,
+    };
+    const Client = jest.spyOn(pg, 'Client').mockImplementation(() => fakeClient as never);
+    const stream = new SqlChangeStream({
+      sequelize: sequelize as never,
+      connectionUri: 'postgres://localhost/db',
+      defaultCursor: '0',
+    });
+    try {
+      const received: unknown[] = [];
+      stream.on('change', change => {
+        received.push(change);
+      });
+      await waitUntil(() => fetches === 1);
+      listeners.notification?.();
+      releaseFirst([]);
+      await waitUntil(() => received.length >= 1);
+      expect(fetches).toBeGreaterThanOrEqual(2);
+      expect(received).toHaveLength(1);
+      expect(received[0]).toMatchObject({
+        documentKey: { _id: 'second' },
+      });
+    } finally {
+      await stream.close();
+      Client.mockRestore();
+    }
+  });
+});
+
+async function waitUntil(predicate: () => boolean, timeoutMs = 2000): Promise<void> {
+  const start = Date.now();
+  while (Date.now() - start < timeoutMs) {
+    if (predicate()) return;
+    await new Promise(resolve => setTimeout(resolve, 10));
+  }
+  throw new Error('timed out waiting for condition');
+}
diff --git a/modules/database/src/realtime/__tests__/status.test.ts b/modules/database/src/realtime/__tests__/status.test.ts
index 542793178..8eb3560af 100644
--- a/modules/database/src/realtime/__tests__/status.test.ts
+++ b/modules/database/src/realtime/__tests__/status.test.ts
@@ -48,6 +48,26 @@ describe('buildRealtimeStatus', () => {
         streamState: 'idle',
       }).message,
     ).toMatch(/replica set/i);
+    expect(
+      buildRealtimeStatus({
+        engine: 'PostgreSQL',
+        enabled: true,
+        topologySupported: false,
+        topologyMessage:
+          'PostgreSQL live updates need a session-mode connection that can LISTEN (not a transaction-mode pooler)',
+        activeSchemaCount: 1,
+        streamState: 'idle',
+      }).message,
+    ).toMatch(/LISTEN/);
+    expect(
+      buildRealtimeStatus({
+        engine: 'PostgreSQL',
+        enabled: true,
+        topologySupported: false,
+        activeSchemaCount: 1,
+        streamState: 'idle',
+      }).message,
+    ).toMatch(/internal change queue/i);
     expect(
       buildRealtimeStatus({
         engine: 'MongoDB',
diff --git a/modules/database/src/realtime/authorize.ts b/modules/database/src/realtime/authorize.ts
index cf7ed899c..2d9b3e410 100644
--- a/modules/database/src/realtime/authorize.ts
+++ b/modules/database/src/realtime/authorize.ts
@@ -61,27 +61,39 @@ export function optionalDocumentId(value: unknown): string | undefined {
   return value;
 }
 
-export async function canReadDocument(
+export type DocumentReadDecision = 'allow' | 'deny' | 'unavailable';
+
+export async function documentReadDecision(
   grpcSdk: AuthorizationSdk,
   schema: string,
   documentId: string,
   userId: string,
-): Promise<boolean> {
-  if (!grpcSdk.authorization || !grpcSdk.isAvailable('authorization')) {
-    return false;
+): Promise<DocumentReadDecision> {
+  const auth = grpcSdk.authorization;
+  if (!auth || !grpcSdk.isAvailable('authorization')) {
+    return 'unavailable';
   }
   try {
-    const decision = await grpcSdk.authorization.can({
+    const decision = await auth.can({
       subject: `User:${userId}`,
       actions: ['read'],
       resource: `${schema}:${documentId}`,
     });
-    return decision.allow === true;
+    return decision.allow === true ? 'allow' : 'deny';
   } catch {
-    return false;
+    return 'unavailable';
   }
 }
 
+export async function canReadDocument(
+  grpcSdk: AuthorizationSdk,
+  schema: string,
+  documentId: string,
+  userId: string,
+): Promise<boolean> {
+  return (await documentReadDecision(grpcSdk, schema, documentId, userId)) === 'allow';
+}
+
 export function assertSchemaAvailable(
   lookup: SchemaLookup,
   schemaName: string,
@@ -120,6 +132,7 @@ export function assertSchemaAvailable(
 export function toOptedInSchema(schema: {
   name: string;
   collectionName: string;
+  documentIdField?: string;
   modelOptions?: {
     conduit?: {
       realtime?: { enabled?: boolean };
@@ -132,5 +145,6 @@ export function toOptedInSchema(schema: {
     name: schema.name,
     collectionName: schema.collectionName,
     authorizationEnabled: schema.modelOptions.conduit.authorization?.enabled === true,
+    documentIdField: schema.documentIdField,
   };
 }
diff --git a/modules/database/src/realtime/sql/SqlChangeStream.ts b/modules/database/src/realtime/sql/SqlChangeStream.ts
index 9ad6eccdb..9861bb4b1 100644
--- a/modules/database/src/realtime/sql/SqlChangeStream.ts
+++ b/modules/database/src/realtime/sql/SqlChangeStream.ts
@@ -10,7 +10,7 @@ import {
   assertSqlDialect,
   type SqlDialect,
 } from './constants.js';
-import { fetchChangeLogBatch, maxChangeLogId } from './changelog.js';
+import { changeLogLagMs, fetchChangeLogBatch } from './changelog.js';
 import { toRawChangeEvent } from './mapEvent.js';
 import { sqlCursorFromResumeAfter } from './resume.js';
 
@@ -18,6 +18,7 @@ export type SqlChangeStreamOptions = {
   sequelize: Sequelize;
   connectionUri: string;
   resumeAfter?: unknown;
+  defaultCursor?: string;
 };
 
 export class SqlChangeStream implements ChangeStreamLike {
@@ -25,18 +26,21 @@ export class SqlChangeStream implements ChangeStreamLike {
   private readonly sequelize: Sequelize;
   private readonly connectionUri: string;
   private readonly dialect: SqlDialect;
+  private readonly lagMs: number;
   private cursor: string | undefined;
   private listenClient: pg.Client | null = null;
   private pollTimer: NodeJS.Timeout | null = null;
   private closed = false;
   private draining = false;
+  private drainRequested = false;
   private started = false;
 
   constructor(options: SqlChangeStreamOptions) {
     this.sequelize = options.sequelize;
     this.connectionUri = options.connectionUri;
     this.dialect = assertSqlDialect(options.sequelize.getDialect());
-    this.cursor = sqlCursorFromResumeAfter(options.resumeAfter);
+    this.lagMs = changeLogLagMs(this.dialect);
+    this.cursor = sqlCursorFromResumeAfter(options.resumeAfter) ?? options.defaultCursor;
     queueMicrotask(() => {
       if (!this.closed) {
         void this.start();
@@ -80,19 +84,19 @@ export class SqlChangeStream implements ChangeStreamLike {
     this.started = true;
     try {
       if (this.cursor === undefined) {
-        this.cursor = await maxChangeLogId(this.sequelize);
+        this.cursor = '0';
       }
       if (this.dialect === 'postgres') {
         await this.startPostgresListen();
         this.pollTimer = setInterval(() => {
-          void this.drain();
+          this.requestDrain();
         }, POSTGRES_FALLBACK_POLL_MS);
       } else {
         this.pollTimer = setInterval(() => {
-          void this.drain();
+          this.requestDrain();
         }, SQL_POLL_INTERVAL_MS);
       }
-      await this.drain();
+      this.requestDrain();
     } catch (err) {
       this.emitError(err);
     }
@@ -102,7 +106,7 @@ export class SqlChangeStream implements ChangeStreamLike {
     const client = new pg.Client({ connectionString: this.connectionUri });
     this.listenClient = client;
     client.on('notification', () => {
-      void this.drain();
+      this.requestDrain();
     });
     client.on('error', (err: Error) => {
       this.emitError(err);
@@ -111,27 +115,39 @@ export class SqlChangeStream implements ChangeStreamLike {
     await client.query(`LISTEN ${NOTIFY_CHANNEL}`);
   }
 
+  private requestDrain(): void {
+    this.drainRequested = true;
+    void this.drain();
+  }
+
   private async drain(): Promise<void> {
     if (this.draining || this.closed) return;
     this.draining = true;
     try {
-      while (!this.closed) {
-        const rows = await fetchChangeLogBatch(
-          this.sequelize,
-          this.cursor ?? '0',
-          CHANGE_LOG_BATCH_SIZE,
-        );
-        if (rows.length === 0) break;
-        for (const row of rows) {
-          if (this.closed) return;
-          this.emitter.emit('change', toRawChangeEvent(row));
-          this.cursor = row.id;
+      while (this.drainRequested && !this.closed) {
+        this.drainRequested = false;
+        while (!this.closed) {
+          const rows = await fetchChangeLogBatch(
+            this.sequelize,
+            this.cursor ?? '0',
+            CHANGE_LOG_BATCH_SIZE,
+            this.lagMs,
+          );
+          if (rows.length === 0) break;
+          for (const row of rows) {
+            if (this.closed) return;
+            this.emitter.emit('change', toRawChangeEvent(row));
+            this.cursor = row.id;
+          }
         }
       }
     } catch (err) {
       this.emitError(err);
     } finally {
       this.draining = false;
+      if (this.drainRequested && !this.closed) {
+        void this.drain();
+      }
     }
   }
 
diff --git a/modules/database/src/realtime/sql/SqlRealtimeSupport.ts b/modules/database/src/realtime/sql/SqlRealtimeSupport.ts
index 2163a1242..e2c45a112 100644
--- a/modules/database/src/realtime/sql/SqlRealtimeSupport.ts
+++ b/modules/database/src/realtime/sql/SqlRealtimeSupport.ts
@@ -1,16 +1,63 @@
+import pg from 'pg';
 import type { SequelizeAdapter } from '../../adapters/sequelize-adapter/index.js';
 import type { OptedInSchema } from '../types.js';
 import type { ChangeStreamLike } from '../types.js';
-import { ensureChangeLog, trimChangeLog } from './changelog.js';
+import type { TopologyResult } from '../topology.js';
+import { NOTIFY_CHANNEL, assertSqlDialect } from './constants.js';
+import { ensureChangeLog, maxChangeLogId, trimChangeLog } from './changelog.js';
 import { parseSqlResumeId } from './resume.js';
 import { SqlChangeStream } from './SqlChangeStream.js';
 import { syncTriggers } from './triggers.js';
 
 export class SqlRealtimeSupport {
+  private watchFromId = '0';
+
   constructor(private readonly adapter: SequelizeAdapter) {}
 
-  async prepare(schemas: OptedInSchema[]): Promise<void> {
-    await ensureChangeLog(this.adapter.sequelize);
+  async checkTopology(): Promise<TopologyResult> {
+    const dialect = assertSqlDialect(this.adapter.sequelize.getDialect());
+    try {
+      await this.adapter.sequelize.query('SELECT 1');
+    } catch (err) {
+      return {
+        supported: false,
+        message: `SQL live updates cannot reach the database: ${errorMessage(err)}`,
+      };
+    }
+    if (dialect !== 'postgres') {
+      return { supported: true };
+    }
+    const client = new pg.Client({ connectionString: this.adapter.connectionUri });
+    try {
+      await client.connect();
+      await client.query(`LISTEN ${NOTIFY_CHANNEL}`);
+      await client.query(`UNLISTEN ${NOTIFY_CHANNEL}`);
+      return { supported: true };
+    } catch (err) {
+      return {
+        supported: false,
+        message:
+          'PostgreSQL live updates need a session-mode connection that can LISTEN (not a transaction-mode pooler): ' +
+          errorMessage(err),
+      };
+    } finally {
+      try {
+        await client.end();
+      } catch {
+        // ignore
+      }
+    }
+  }
+
+  async prepare(
+    schemas: OptedInSchema[],
+    options?: { ensureLog?: boolean },
+  ): Promise<void> {
+    const ensureLog = options?.ensureLog !== false;
+    if (ensureLog) {
+      await ensureChangeLog(this.adapter.sequelize);
+      this.watchFromId = await maxChangeLogId(this.adapter.sequelize);
+    }
     await syncTriggers(this.adapter.sequelize, schemas);
   }
 
@@ -19,6 +66,7 @@ export class SqlRealtimeSupport {
       sequelize: this.adapter.sequelize,
       connectionUri: this.adapter.connectionUri,
       resumeAfter,
+      defaultCursor: this.watchFromId,
     });
   }
 
@@ -28,3 +76,7 @@ export class SqlRealtimeSupport {
     await trimChangeLog(this.adapter.sequelize, id);
   }
 }
+
+function errorMessage(err: unknown): string {
+  return err instanceof Error ? err.message : String(err);
+}
diff --git a/modules/database/src/realtime/sql/changelog.ts b/modules/database/src/realtime/sql/changelog.ts
index 1596d4c5b..177463781 100644
--- a/modules/database/src/realtime/sql/changelog.ts
+++ b/modules/database/src/realtime/sql/changelog.ts
@@ -1,40 +1,45 @@
 import { QueryTypes, Sequelize } from 'sequelize';
 import {
   CHANGE_LOG_BATCH_SIZE,
+  CHANGE_LOG_LAG_MS,
   CHANGE_LOG_TABLE,
   type SqlDialect,
   assertSqlDialect,
 } from './constants.js';
 import { quoteIdent } from './identifiers.js';
-import { createCaptureFunctionSql, createChangeLogTableSql } from './ddl.js';
+import { createChangeLogTableSql } from './ddl.js';
 import type { ChangeLogRow } from './mapEvent.js';
 
+export function changeLogLagMs(dialect: SqlDialect): number {
+  return dialect === 'sqlite' ? 0 : CHANGE_LOG_LAG_MS;
+}
+
+export function fetchChangeLogSql(dialect: SqlDialect, lagMs: number): string {
+  const table = quoteIdent(dialect, CHANGE_LOG_TABLE);
+  return `SELECT id, collection_name, document_id, operation, occurred_at
+     FROM ${table}
+     WHERE id > :resumeId${lagPredicate(dialect, lagMs)}
+     ORDER BY id ASC
+     LIMIT :limit`;
+}
+
 export async function ensureChangeLog(sequelize: Sequelize): Promise<void> {
   const dialect = assertSqlDialect(sequelize.getDialect());
   await sequelize.query(createChangeLogTableSql(dialect));
-  if (dialect === 'postgres') {
-    await sequelize.query(createCaptureFunctionSql());
-  }
 }
 
 export async function fetchChangeLogBatch(
   sequelize: Sequelize,
   resumeId: string,
   limit: number = CHANGE_LOG_BATCH_SIZE,
+  lagMs?: number,
 ): Promise<ChangeLogRow[]> {
   const dialect = assertSqlDialect(sequelize.getDialect());
-  const table = quoteIdent(dialect, CHANGE_LOG_TABLE);
-  const rows = await sequelize.query(
-    `SELECT id, collection_name, document_id, operation, occurred_at
-     FROM ${table}
-     WHERE id > :resumeId
-     ORDER BY id ASC
-     LIMIT :limit`,
-    {
-      type: QueryTypes.SELECT,
-      replacements: { resumeId, limit },
-    },
-  );
+  const resolvedLag = lagMs ?? changeLogLagMs(dialect);
+  const rows = await sequelize.query(fetchChangeLogSql(dialect, resolvedLag), {
+    type: QueryTypes.SELECT,
+    replacements: fetchReplacements(dialect, resumeId, limit, resolvedLag),
+  });
   return (rows as Record<string, unknown>[]).map(row => ({
     id: String(row.id),
     collection_name: String(row.collection_name),
@@ -78,6 +83,53 @@ export async function trimChangeLog(
   }
 }
 
+function lagPredicate(dialect: SqlDialect, lagMs: number): string {
+  if (lagMs <= 0) {
+    return '';
+  }
+  switch (dialect) {
+    case 'postgres':
+      return ' AND occurred_at <= NOW() - make_interval(secs => :lagSeconds)';
+    case 'mysql':
+    case 'mariadb':
+      return ' AND occurred_at <= DATE_SUB(NOW(6), INTERVAL :lagMicrosecond MICROSECOND)';
+    case 'sqlite':
+      return ` AND occurred_at <= datetime('now', :lagModifier)`;
+    default: {
+      const _exhaustive: never = dialect;
+      return _exhaustive;
+    }
+  }
+}
+
+function fetchReplacements(
+  dialect: SqlDialect,
+  resumeId: string,
+  limit: number,
+  lagMs: number,
+): Record<string, string | number> {
+  const replacements: Record<string, string | number> = { resumeId, limit };
+  if (lagMs <= 0) {
+    return replacements;
+  }
+  switch (dialect) {
+    case 'postgres':
+      replacements.lagSeconds = lagMs / 1000;
+      return replacements;
+    case 'mysql':
+    case 'mariadb':
+      replacements.lagMicrosecond = lagMs * 1000;
+      return replacements;
+    case 'sqlite':
+      replacements.lagModifier = `-${lagMs / 1000} seconds`;
+      return replacements;
+    default: {
+      const _exhaustive: never = dialect;
+      return _exhaustive;
+    }
+  }
+}
+
 function trimSql(dialect: SqlDialect, table: string): string {
   switch (dialect) {
     case 'mysql':
diff --git a/modules/database/src/realtime/sql/constants.ts b/modules/database/src/realtime/sql/constants.ts
index 8073df1b1..2f99444bd 100644
--- a/modules/database/src/realtime/sql/constants.ts
+++ b/modules/database/src/realtime/sql/constants.ts
@@ -1,5 +1,5 @@
 export const CHANGE_LOG_TABLE = '_cnd_DatabaseChange';
-export const CHANGE_LOG_FUNCTION = 'conduit_realtime_capture';
+export const CAPTURE_FUNCTION_PREFIX = 'cnd_rt_fn_';
 export const NOTIFY_CHANNEL = 'conduit_realtime';
 export const TRIGGER_NAME_PREFIX = 'cnd_rt_';
 export const PK_COLUMN = '_id';
@@ -7,6 +7,10 @@ export const PK_COLUMN = '_id';
 export const SQL_POLL_INTERVAL_MS = 250;
 export const POSTGRES_FALLBACK_POLL_MS = 2_000;
 export const CHANGE_LOG_BATCH_SIZE = 200;
+export const CHANGE_LOG_LAG_MS = 750;
+
+export const SQL_LEADER_LOCK = 'realtime:sql:change-stream:leader';
+export const SQL_RESUME_TOKEN_KEY = 'realtime:sql:resumeToken';
 
 export const SQL_DIALECTS = ['postgres', 'mysql', 'mariadb', 'sqlite'] as const;
 export type SqlDialect = (typeof SQL_DIALECTS)[number];
diff --git a/modules/database/src/realtime/sql/ddl.ts b/modules/database/src/realtime/sql/ddl.ts
index 9dc03e239..10eacccd9 100644
--- a/modules/database/src/realtime/sql/ddl.ts
+++ b/modules/database/src/realtime/sql/ddl.ts
@@ -1,5 +1,4 @@
 import {
-  CHANGE_LOG_FUNCTION,
   CHANGE_LOG_TABLE,
   NOTIFY_CHANNEL,
   PK_COLUMN,
@@ -25,7 +24,7 @@ export function createChangeLogTableSql(dialect: SqlDialect): string {
         collection_name VARCHAR(255) NOT NULL,
         document_id VARCHAR(255) NOT NULL,
         operation VARCHAR(16) NOT NULL,
-        occurred_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP
+        occurred_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6)
       )`;
     case 'sqlite':
       return `CREATE TABLE IF NOT EXISTS ${table} (
@@ -42,11 +41,15 @@ export function createChangeLogTableSql(dialect: SqlDialect): string {
   }
 }
 
-export function createCaptureFunctionSql(): string {
+export function createCaptureFunctionSql(
+  pkColumn: string = PK_COLUMN,
+  functionName: string,
+): string {
   const table = quoteIdent('postgres', CHANGE_LOG_TABLE);
-  const pk = quoteIdent('postgres', PK_COLUMN);
+  const pk = quoteIdent('postgres', pkColumn);
+  const fn = quoteIdent('postgres', functionName);
   const channel = NOTIFY_CHANNEL.replace(/'/g, "''");
-  return `CREATE OR REPLACE FUNCTION ${CHANGE_LOG_FUNCTION}() RETURNS trigger AS $$
+  return `CREATE OR REPLACE FUNCTION ${fn}() RETURNS trigger AS $$
 DECLARE
   doc_id text;
   op text;
@@ -77,3 +80,7 @@ BEGIN
 END;
 $$ LANGUAGE plpgsql`;
 }
+
+export function dropCaptureFunctionSql(functionName: string): string {
+  return `DROP FUNCTION IF EXISTS ${quoteIdent('postgres', functionName)}()`;
+}
diff --git a/modules/database/src/realtime/sql/identifiers.ts b/modules/database/src/realtime/sql/identifiers.ts
index 5e6a89349..69b83ba5e 100644
--- a/modules/database/src/realtime/sql/identifiers.ts
+++ b/modules/database/src/realtime/sql/identifiers.ts
@@ -1,6 +1,6 @@
 import { createHash } from 'node:crypto';
 import type { SqlDialect } from './constants.js';
-import { TRIGGER_NAME_PREFIX } from './constants.js';
+import { CAPTURE_FUNCTION_PREFIX, TRIGGER_NAME_PREFIX } from './constants.js';
 
 const MYSQL_IDENT_LIMIT = 64;
 const POSTGRES_IDENT_LIMIT = 63;
@@ -40,6 +40,13 @@ export function triggerBaseName(collectionName: string, dialect: SqlDialect): st
   );
 }
 
+export function captureFunctionName(collectionName: string): string {
+  return fitIdentifier(
+    `${CAPTURE_FUNCTION_PREFIX}${collectionName}`,
+    identifierLimit('postgres'),
+  );
+}
+
 export function rowTriggerName(
   collectionName: string,
   operation: 'i' | 'u' | 'd',
diff --git a/modules/database/src/realtime/sql/index.ts b/modules/database/src/realtime/sql/index.ts
index d0f7fd4e5..75cf9ee99 100644
--- a/modules/database/src/realtime/sql/index.ts
+++ b/modules/database/src/realtime/sql/index.ts
@@ -1,4 +1,9 @@
-export { CHANGE_LOG_TABLE, SQL_DIALECTS } from './constants.js';
+export {
+  CHANGE_LOG_TABLE,
+  SQL_DIALECTS,
+  SQL_LEADER_LOCK,
+  SQL_RESUME_TOKEN_KEY,
+} from './constants.js';
 export { SqlChangeStream } from './SqlChangeStream.js';
 export { SqlRealtimeSupport } from './SqlRealtimeSupport.js';
 export { parseSqlResumeId, sqlCursorFromResumeAfter } from './resume.js';
@@ -7,7 +12,9 @@ export { createChangeLogTableSql, createCaptureFunctionSql } from './ddl.js';
 export {
   quoteIdent,
   triggerBaseName,
+  captureFunctionName,
   rowTriggerName,
   fitIdentifier,
 } from './identifiers.js';
 export { toRawChangeEvent } from './mapEvent.js';
+export { fetchChangeLogSql, changeLogLagMs } from './changelog.js';
diff --git a/modules/database/src/realtime/sql/resume.ts b/modules/database/src/realtime/sql/resume.ts
index 42c58d52f..04ce0168e 100644
--- a/modules/database/src/realtime/sql/resume.ts
+++ b/modules/database/src/realtime/sql/resume.ts
@@ -6,31 +6,22 @@ export function parseSqlResumeId(token: string | null | undefined): string | und
     return token;
   }
   const parsed = parseResumeToken(token);
-  if (typeof parsed === 'number' && Number.isInteger(parsed) && parsed >= 0) {
-    return String(parsed);
-  }
-  if (typeof parsed === 'bigint' && parsed >= 0n) {
-    return parsed.toString();
-  }
-  if (typeof parsed === 'string' && /^\d+$/.test(parsed)) {
-    return parsed;
-  }
-  return undefined;
+  return decimalId(parsed);
 }
 
 export function sqlCursorFromResumeAfter(resumeAfter: unknown): string | undefined {
-  if (
-    typeof resumeAfter === 'number' &&
-    Number.isInteger(resumeAfter) &&
-    resumeAfter >= 0
-  ) {
-    return String(resumeAfter);
+  return decimalId(resumeAfter);
+}
+
+function decimalId(value: unknown): string | undefined {
+  if (typeof value === 'bigint' && value >= 0n) {
+    return value.toString();
   }
-  if (typeof resumeAfter === 'bigint' && resumeAfter >= 0n) {
-    return resumeAfter.toString();
+  if (typeof value === 'string' && /^\d+$/.test(value)) {
+    return value;
   }
-  if (typeof resumeAfter === 'string' && /^\d+$/.test(resumeAfter)) {
-    return resumeAfter;
+  if (typeof value === 'number' && Number.isSafeInteger(value) && value >= 0) {
+    return String(value);
   }
   return undefined;
 }
diff --git a/modules/database/src/realtime/sql/triggerSql.ts b/modules/database/src/realtime/sql/triggerSql.ts
index 809928007..0ac0e9fa8 100644
--- a/modules/database/src/realtime/sql/triggerSql.ts
+++ b/modules/database/src/realtime/sql/triggerSql.ts
@@ -1,42 +1,49 @@
+import { CHANGE_LOG_TABLE, PK_COLUMN, type SqlDialect } from './constants.js';
 import {
-  CHANGE_LOG_FUNCTION,
-  CHANGE_LOG_TABLE,
-  PK_COLUMN,
-  type SqlDialect,
-} from './constants.js';
-import {
+  captureFunctionName,
   quoteIdent,
   rowTriggerName,
   sqlStringLiteral,
   triggerBaseName,
 } from './identifiers.js';
+import { createCaptureFunctionSql, dropCaptureFunctionSql } from './ddl.js';
 
 export type DesiredTrigger = {
   triggerName: string;
   collectionName: string;
+  pkColumn: string;
   sql: string;
   dropSql: string;
+  functionName?: string;
+  functionSql?: string;
+  dropFunctionSql?: string;
 };
 
 export function desiredTriggers(
   dialect: SqlDialect,
   collectionName: string,
+  pkColumn: string = PK_COLUMN,
 ): DesiredTrigger[] {
   const table = quoteIdent(dialect, collectionName);
-  const pk = quoteIdent(dialect, PK_COLUMN);
+  const pk = quoteIdent(dialect, pkColumn);
   const logTable = quoteIdent(dialect, CHANGE_LOG_TABLE);
   const collectionLiteral = sqlStringLiteral(collectionName);
   switch (dialect) {
     case 'postgres': {
       const triggerName = triggerBaseName(collectionName, dialect);
       const quotedTrigger = quoteIdent(dialect, triggerName);
+      const functionName = captureFunctionName(collectionName);
       return [
         {
           triggerName,
           collectionName,
+          pkColumn,
+          functionName,
+          functionSql: createCaptureFunctionSql(pkColumn, functionName),
+          dropFunctionSql: dropCaptureFunctionSql(functionName),
           sql: `CREATE TRIGGER ${quotedTrigger}
 AFTER INSERT OR UPDATE OR DELETE ON ${table}
-FOR EACH ROW EXECUTE PROCEDURE ${CHANGE_LOG_FUNCTION}()`,
+FOR EACH ROW EXECUTE PROCEDURE ${quoteIdent(dialect, functionName)}()`,
           dropSql: `DROP TRIGGER IF EXISTS ${quotedTrigger} ON ${table}`,
         },
       ];
@@ -44,14 +51,39 @@ FOR EACH ROW EXECUTE PROCEDURE ${CHANGE_LOG_FUNCTION}()`,
     case 'mysql':
     case 'mariadb':
       return [
-        mysqlRowTrigger(dialect, collectionName, 'i', 'INSERT', 'insert', 'NEW'),
-        mysqlRowTrigger(dialect, collectionName, 'u', 'UPDATE', 'update', 'NEW'),
-        mysqlRowTrigger(dialect, collectionName, 'd', 'DELETE', 'delete', 'OLD'),
+        mysqlRowTrigger(
+          dialect,
+          collectionName,
+          pkColumn,
+          'i',
+          'INSERT',
+          'insert',
+          'NEW',
+        ),
+        mysqlRowTrigger(
+          dialect,
+          collectionName,
+          pkColumn,
+          'u',
+          'UPDATE',
+          'update',
+          'NEW',
+        ),
+        mysqlRowTrigger(
+          dialect,
+          collectionName,
+          pkColumn,
+          'd',
+          'DELETE',
+          'delete',
+          'OLD',
+        ),
       ];
     case 'sqlite':
       return [
         sqliteRowTrigger(
           collectionName,
+          pkColumn,
           'i',
           'INSERT',
           'insert',
@@ -63,6 +95,7 @@ FOR EACH ROW EXECUTE PROCEDURE ${CHANGE_LOG_FUNCTION}()`,
         ),
         sqliteRowTrigger(
           collectionName,
+          pkColumn,
           'u',
           'UPDATE',
           'update',
@@ -74,6 +107,7 @@ FOR EACH ROW EXECUTE PROCEDURE ${CHANGE_LOG_FUNCTION}()`,
         ),
         sqliteRowTrigger(
           collectionName,
+          pkColumn,
           'd',
           'DELETE',
           'delete',
@@ -94,6 +128,7 @@ FOR EACH ROW EXECUTE PROCEDURE ${CHANGE_LOG_FUNCTION}()`,
 function mysqlRowTrigger(
   dialect: SqlDialect,
   collectionName: string,
+  pkColumn: string,
   opKey: 'i' | 'u' | 'd',
   timing: 'INSERT' | 'UPDATE' | 'DELETE',
   operation: 'insert' | 'update' | 'delete',
@@ -102,17 +137,20 @@ function mysqlRowTrigger(
   const triggerName = rowTriggerName(collectionName, opKey, dialect);
   const quotedTrigger = quoteIdent(dialect, triggerName);
   const table = quoteIdent(dialect, collectionName);
-  const pk = quoteIdent(dialect, PK_COLUMN);
+  const pk = quoteIdent(dialect, pkColumn);
   const logTable = quoteIdent(dialect, CHANGE_LOG_TABLE);
   const collectionLiteral = sqlStringLiteral(collectionName);
   const opLiteral = sqlStringLiteral(operation);
   return {
     triggerName,
     collectionName,
+    pkColumn,
     sql: `CREATE TRIGGER ${quotedTrigger} AFTER ${timing} ON ${table}
 FOR EACH ROW BEGIN
-  INSERT INTO ${logTable} (collection_name, document_id, operation, occurred_at)
-  VALUES (${collectionLiteral}, CAST(${row}.${pk} AS CHAR), ${opLiteral}, CURRENT_TIMESTAMP);
+  IF ${row}.${pk} IS NOT NULL THEN
+    INSERT INTO ${logTable} (collection_name, document_id, operation, occurred_at)
+    VALUES (${collectionLiteral}, CAST(${row}.${pk} AS CHAR), ${opLiteral}, CURRENT_TIMESTAMP(6));
+  END IF;
 END`,
     dropSql: `DROP TRIGGER IF EXISTS ${quotedTrigger}`,
   };
@@ -120,6 +158,7 @@ END`,
 
 function sqliteRowTrigger(
   collectionName: string,
+  pkColumn: string,
   opKey: 'i' | 'u' | 'd',
   timing: 'INSERT' | 'UPDATE' | 'DELETE',
   operation: 'insert' | 'update' | 'delete',
@@ -135,10 +174,12 @@ function sqliteRowTrigger(
   return {
     triggerName,
     collectionName,
+    pkColumn,
     sql: `CREATE TRIGGER ${quotedTrigger} AFTER ${timing} ON ${table}
 BEGIN
   INSERT INTO ${logTable} (collection_name, document_id, operation, occurred_at)
-  VALUES (${collectionLiteral}, ${row}.${pk}, ${opLiteral}, datetime('now'));
+  SELECT ${collectionLiteral}, ${row}.${pk}, ${opLiteral}, datetime('now')
+  WHERE ${row}.${pk} IS NOT NULL;
 END`,
     dropSql: `DROP TRIGGER IF EXISTS ${quotedTrigger}`,
   };
diff --git a/modules/database/src/realtime/sql/triggers.ts b/modules/database/src/realtime/sql/triggers.ts
index 1bd7f096c..ce18b2f67 100644
--- a/modules/database/src/realtime/sql/triggers.ts
+++ b/modules/database/src/realtime/sql/triggers.ts
@@ -2,6 +2,7 @@ import { QueryTypes, Sequelize } from 'sequelize';
 import type { OptedInSchema } from '../types.js';
 import {
   CHANGE_LOG_TABLE,
+  PK_COLUMN,
   TRIGGER_NAME_PREFIX,
   type SqlDialect,
   assertSqlDialect,
@@ -15,6 +16,7 @@ export { desiredTriggers };
 export type ExistingTrigger = {
   triggerName: string;
   tableName: string;
+  definition?: string;
 };
 
 export async function listExistingTriggers(
@@ -25,32 +27,40 @@ export async function listExistingTriggers(
   switch (dialect) {
     case 'postgres': {
       const rows = await sequelize.query(
-        `SELECT trigger_name AS trigger_name, event_object_table AS table_name
+        `SELECT trigger_name AS trigger_name, event_object_table AS table_name,
+                action_statement AS definition
          FROM information_schema.triggers
          WHERE trigger_name LIKE :prefix`,
         { type: QueryTypes.SELECT, replacements: { prefix } },
       );
-      return uniqueTriggers(rows as { trigger_name: string; table_name: string }[]);
+      return uniqueTriggers(
+        rows as { trigger_name: string; table_name: string; definition?: string }[],
+      );
     }
     case 'mysql':
     case 'mariadb': {
       const rows = await sequelize.query(
-        `SELECT trigger_name AS trigger_name, event_object_table AS table_name
+        `SELECT trigger_name AS trigger_name, event_object_table AS table_name,
+                action_statement AS definition
          FROM information_schema.triggers
          WHERE trigger_schema = DATABASE()
          AND trigger_name LIKE :prefix`,
         { type: QueryTypes.SELECT, replacements: { prefix } },
       );
-      return uniqueTriggers(rows as { trigger_name: string; table_name: string }[]);
+      return uniqueTriggers(
+        rows as { trigger_name: string; table_name: string; definition?: string }[],
+      );
     }
     case 'sqlite': {
       const rows = await sequelize.query(
-        `SELECT name AS trigger_name, tbl_name AS table_name
+        `SELECT name AS trigger_name, tbl_name AS table_name, sql AS definition
          FROM sqlite_master
          WHERE type = 'trigger' AND name LIKE :prefix`,
         { type: QueryTypes.SELECT, replacements: { prefix } },
       );
-      return uniqueTriggers(rows as { trigger_name: string; table_name: string }[]);
+      return uniqueTriggers(
+        rows as { trigger_name: string; table_name: string; definition?: string }[],
+      );
     }
     default: {
       const _exhaustive: never = dialect;
@@ -60,7 +70,7 @@ export async function listExistingTriggers(
 }
 
 function uniqueTriggers(
-  rows: { trigger_name: string; table_name: string }[],
+  rows: { trigger_name: string; table_name: string; definition?: string }[],
 ): ExistingTrigger[] {
   const seen = new Set<string>();
   const result: ExistingTrigger[] = [];
@@ -68,11 +78,28 @@ function uniqueTriggers(
     const triggerName = String(row.trigger_name);
     if (seen.has(triggerName)) continue;
     seen.add(triggerName);
-    result.push({ triggerName, tableName: String(row.table_name) });
+    result.push({
+      triggerName,
+      tableName: String(row.table_name),
+      definition: row.definition == null ? undefined : String(row.definition),
+    });
   }
   return result;
 }
 
+function triggerMatches(
+  dialect: SqlDialect,
+  trigger: DesiredTrigger,
+  definition: string | undefined,
+): boolean {
+  if (!definition) return false;
+  if (trigger.functionName) {
+    return definition.includes(trigger.functionName);
+  }
+  const pk = quoteIdent(dialect, trigger.pkColumn);
+  return definition.includes(pk) && /IS NOT NULL/i.test(definition);
+}
+
 export async function syncTriggers(
   sequelize: Sequelize,
   schemas: OptedInSchema[],
@@ -81,18 +108,36 @@ export async function syncTriggers(
   const desired = new Map<string, DesiredTrigger>();
   for (const schema of schemas) {
     if (schema.collectionName === CHANGE_LOG_TABLE) continue;
-    for (const trigger of desiredTriggers(dialect, schema.collectionName)) {
+    const pkColumn = schema.documentIdField ?? PK_COLUMN;
+    for (const trigger of desiredTriggers(dialect, schema.collectionName, pkColumn)) {
       desired.set(trigger.triggerName, trigger);
     }
   }
   const existing = await listExistingTriggers(sequelize);
+  const existingByName = new Map(
+    existing.map(trigger => [trigger.triggerName, trigger] as const),
+  );
   for (const current of existing) {
     if (desired.has(current.triggerName)) continue;
-    const dropSql = dropExistingSql(dialect, current);
-    await sequelize.query(dropSql);
+    await sequelize.query(dropExistingSql(dialect, current));
+    const leftover = desiredTriggers(dialect, current.tableName, PK_COLUMN).find(
+      trigger => trigger.triggerName === current.triggerName,
+    );
+    if (leftover?.dropFunctionSql) {
+      await sequelize.query(leftover.dropFunctionSql);
+    }
   }
   for (const trigger of desired.values()) {
-    await sequelize.query(trigger.dropSql);
+    if (trigger.functionSql) {
+      await sequelize.query(trigger.functionSql);
+    }
+    const current = existingByName.get(trigger.triggerName);
+    if (current && triggerMatches(dialect, trigger, current.definition)) {
+      continue;
+    }
+    if (current) {
+      await sequelize.query(dropExistingSql(dialect, current));
+    }
     await sequelize.query(trigger.sql);
   }
 }
diff --git a/modules/database/src/realtime/status.ts b/modules/database/src/realtime/status.ts
index 436523498..4e05c0d45 100644
--- a/modules/database/src/realtime/status.ts
+++ b/modules/database/src/realtime/status.ts
@@ -45,7 +45,7 @@ export function buildRealtimeStatus(input: RealtimeStatusInput): RealtimeStatus
         input.topologyMessage ??
         (input.engine === 'MongoDB'
           ? 'A replica set or sharded MongoDB deployment is required for live updates'
-          : 'Database topology does not support live updates'),
+          : 'SQL live updates use an internal change queue (triggers), not native CDC. Database topology does not support live updates'),
     };
   }
   if (input.socketsEnabled === false) {
diff --git a/modules/database/src/realtime/types.ts b/modules/database/src/realtime/types.ts
index 9901c2d86..be326cad1 100644
--- a/modules/database/src/realtime/types.ts
+++ b/modules/database/src/realtime/types.ts
@@ -33,6 +33,7 @@ export type OptedInSchema = {
   name: string;
   collectionName: string;
   authorizationEnabled: boolean;
+  documentIdField?: string;
 };
 
 export type SubscribeRequest = {