diff --git a/modules/database/README.mdx b/modules/database/README.mdx index 751111223..aef5b52d5 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 live updates do not use a replica set; they use an internal trigger-backed change queue (not native WAL/binlog CDC). +Live document updates on MongoDB also require a replica set or sharded cluster. Local Compose files initialize a single-node replica set so change streams can be exercised. PostgreSQL live updates require logical replication (`wal_level=logical`, a `pgoutput` publication, and a replication slot) — the same class of topology tax as a Mongo replica set. MySQL, MariaDB, and SQLite live updates are out of v1. ### Live updates @@ -60,11 +60,11 @@ unsubscribe({ schema: 'Order', documentId?: string }) Events arrive as `change` with `{ version, operation, schema, documentId, occurredAt, resumeToken }` and contain no document fields. Consumers should refetch through their authorized REST or custom-endpoint path. -MongoDB uses native change streams. PostgreSQL, MySQL, MariaDB, and SQLite capture the same metadata through table triggers and an internal `_cnd_DatabaseChange` log (not a CMS schema). That log is a **queue**, not native CDC: each opted-in write does an extra insert (and the leader later deletes acked rows). `TRUNCATE`, `COPY`, and table-rewrite DDL are not captured. Delivery is at-least-once; duplicates are possible. PostgreSQL and MySQL/MariaDB only consume rows whose `occurred_at` is at least 750ms old so a later autoincrement id is less likely to become visible before an earlier one; that is a timestamp lag, not a commit watermark. A transaction that stays open longer than 750ms after a later row’s insert can still leave a hole (`WHERE id > cursor` never sees the earlier id, and trim then deletes it). SQLite writers are serialized, so lag is 0. +MongoDB uses native change streams. PostgreSQL uses in-process WAL CDC (`pgoutput` publication + a **temporary** logical slot). That is WAL CDC, Postgres-only — not a changelog table, not triggers, not Debezium. There is **no catch-up**: re-subscribe, leader restart, or slot drop does not replay missed events; clients fetch current data. `LISTEN`/`NOTIFY` is not the capture path. -PostgreSQL wakes the listener with `LISTEN`/`NOTIFY` on a dedicated session connection. A transaction-mode pooler (PgBouncer default, many serverless poolers) cannot keep `LISTEN` and will surface as unsupported/degraded — use a direct/session URI. MySQL, MariaDB, and SQLite poll. The database role needs permission to `CREATE TABLE` / `CREATE TRIGGER` (PostgreSQL also needs `CREATE FUNCTION`, `LISTEN`, and `pg_notify`). MySQL with binary logging often needs `log_bin_trust_function_creators`. Triggers use the schema’s physical primary key (`idField`), not a virtual `_id`. +The database role needs permission to `CREATE PUBLICATION`, `ALTER PUBLICATION`, and to create a logical replication slot (`REPLICATION` / managed-Postgres logical-replication grants). A transaction-mode pooler cannot speak the replication protocol; use a direct/session URI. Tables without a primary key get `REPLICA IDENTITY FULL` so UPDATE/DELETE can be published. `TRUNCATE` is not emitted as document events. Custom PKs use the schema’s physical primary key (`idField`), not a virtual `_id`. -SQL live updates do not require a replica set. They are not equivalent to Mongo change streams. +MySQL, MariaDB, and SQLite are unsupported for live updates. Do not enable `realtime` on those engines. Client subscribers must authenticate. Schemas with document-level authorization reject schema-wide subscriptions and require a document ID plus a `read` check. Admin consumers use `POST /realtime/ticket` for a 30-second handshake token; session JWTs and masterkeys must not be sent from browser code. diff --git a/modules/database/src/realtime/ChangeStreamCoordinator.ts b/modules/database/src/realtime/ChangeStreamCoordinator.ts index fb6dd48a2..6e99d791b 100644 --- a/modules/database/src/realtime/ChangeStreamCoordinator.ts +++ b/modules/database/src/realtime/ChangeStreamCoordinator.ts @@ -39,6 +39,7 @@ export type CoordinatorOptions = { parseResumeToken?: (token: string | null | undefined) => unknown | undefined; prepare?: () => Promise; onResumePersisted?: (resumeToken: string) => Promise; + persistResume?: boolean; leaderLock?: string; resumeTokenKey?: string; }; @@ -141,6 +142,10 @@ export class ChangeStreamCoordinator { return this.options.resumeTokenKey ?? RESUME_TOKEN_KEY; } + private get persistResume(): boolean { + return this.options.persistResume !== false; + } + private async safePrepare(): Promise { try { await this.options.prepare?.(); @@ -200,16 +205,15 @@ export class ChangeStreamCoordinator { this.streamState = 'starting'; this.ignoreClose = false; try { - const parseToken = this.options.parseResumeToken ?? parseMongoResumeToken; - const resumeAfter = parseToken( - await this.options.grpcSdk.state!.getKey(this.resumeTokenName), - ); + const resumeAfter = this.persistResume + ? (this.options.parseResumeToken ?? parseMongoResumeToken)( + await this.options.grpcSdk.state!.getKey(this.resumeTokenName), + ) + : undefined; if (this.watching || this.closed) return; const stream = this.options.watch({ resumeAfter }); this.stream = stream; this.watching = true; - this.streamState = 'live'; - this.retryAttempt = 0; stream.on('change', (change: unknown) => { this.enqueueChange(change as RawChangeEvent); }); @@ -222,6 +226,12 @@ export class ChangeStreamCoordinator { this.scheduleRetry(); } }); + if (stream.ready) { + await stream.ready; + } + if (this.closed || !this.watching) return; + this.streamState = 'live'; + this.retryAttempt = 0; } catch (err) { this.watching = false; await this.handleStreamError(err); @@ -250,7 +260,7 @@ export class ChangeStreamCoordinator { const schema = this.resolveSchema(change.ns?.coll); const event = schema ? normalizeChangeEvent(change, schema.name) : null; if (!event || !schema) { - if (token) { + if (this.persistResume && token) { await this.persistResumeToken(token); } return; @@ -258,7 +268,9 @@ export class ChangeStreamCoordinator { this.lastEventAt = event.occurredAt; this.lastError = undefined; await this.emitChange(schema, event); - await this.persistResumeToken(event.resumeToken); + if (this.persistResume) { + await this.persistResumeToken(event.resumeToken); + } } private async persistResumeToken(token: string) { @@ -348,7 +360,7 @@ export class ChangeStreamCoordinator { this.streamState = 'degraded'; ConduitGrpcSdk.Metrics?.increment('database_realtime_stream_errors_total'); ConduitGrpcSdk.Logger.error(err as Error); - if (isResumeTokenUnusable(err)) { + if (this.persistResume && isResumeTokenUnusable(err)) { await this.options.grpcSdk.state!.clearKey(this.resumeTokenName); } await this.stopStream('degraded'); diff --git a/modules/database/src/realtime/RealtimeService.ts b/modules/database/src/realtime/RealtimeService.ts index ab3e227af..005d8288c 100644 --- a/modules/database/src/realtime/RealtimeService.ts +++ b/modules/database/src/realtime/RealtimeService.ts @@ -22,8 +22,7 @@ import { RealtimeSubscriptionTracker } from './subscriptions.js'; import type { ChangeStreamLike, OptedInSchema, RealtimeStatus } from './types.js'; import { topologyFromHello } from './topology.js'; import { SqlRealtimeSupport } from './sql/SqlRealtimeSupport.js'; -import { parseSqlResumeId } from './sql/resume.js'; -import { SQL_LEADER_LOCK, SQL_RESUME_TOKEN_KEY } from './sql/constants.js'; +import { SQL_LEADER_LOCK } from './sql/constants.js'; export class RealtimeService { private readonly subscriptions: RealtimeSubscriptionTracker; @@ -56,20 +55,17 @@ export class RealtimeService { this.sqlSupport = new SqlRealtimeSupport(adapter); this.coordinator = new ChangeStreamCoordinator({ grpcSdk, - watch: options => this.sqlSupport!.openWatch(options.resumeAfter), + watch: () => this.sqlSupport!.openWatch(), checkTopology: () => this.sqlSupport!.checkTopology(), getOptedInSchemas: () => this.getOptedInSchemas(), subscriptions: this.subscriptions, enabled: () => this.isGloballyEnabled(), - parseResumeToken: parseSqlResumeId, + persistResume: false, leaderLock: SQL_LEADER_LOCK, - resumeTokenKey: SQL_RESUME_TOKEN_KEY, prepare: () => this.sqlSupport!.prepare( this.isGloballyEnabled() ? this.getOptedInSchemas() : [], - { ensureLog: this.isGloballyEnabled() }, ), - onResumePersisted: token => this.sqlSupport!.trimThrough(token), }); } } diff --git a/modules/database/src/realtime/__tests__/coordinator.test.ts b/modules/database/src/realtime/__tests__/coordinator.test.ts index f17784985..3bd823034 100644 --- a/modules/database/src/realtime/__tests__/coordinator.test.ts +++ b/modules/database/src/realtime/__tests__/coordinator.test.ts @@ -4,8 +4,7 @@ import { EJSON, ObjectId } from 'bson'; import { ChangeStreamCoordinator } from '../ChangeStreamCoordinator.js'; import { RealtimeSubscriptionTracker } from '../subscriptions.js'; import { roomsForPublicChange } from '../rooms.js'; -import { parseSqlResumeId } from '../sql/resume.js'; -import { SQL_LEADER_LOCK, SQL_RESUME_TOKEN_KEY } from '../sql/constants.js'; +import { SQL_LEADER_LOCK } from '../sql/constants.js'; class MemoryStore { private sets = new Map>(); @@ -46,14 +45,22 @@ function createCoordinator(overrides?: { getKeyDelayMs?: number; onResumePersisted?: (token: string) => Promise; parseResumeToken?: (token: string | null | undefined) => unknown | undefined; + persistResume?: boolean; leaderLock?: string; resumeTokenKey?: string; adminPush?: () => Promise; + watchReady?: Promise; }) { - const stream = new EventEmitter() as EventEmitter & { close: () => Promise }; + const stream = new EventEmitter() as EventEmitter & { + close: () => Promise; + ready?: Promise; + }; stream.close = async () => { stream.emit('close'); }; + if (overrides?.watchReady) { + stream.ready = overrides.watchReady; + } const state = new Map(); const lock = { extend: jest.fn(async () => lock), @@ -106,6 +113,7 @@ function createCoordinator(overrides?: { enabled: () => true, onResumePersisted: overrides?.onResumePersisted, parseResumeToken: overrides?.parseResumeToken, + persistResume: overrides?.persistResume, leaderLock: overrides?.leaderLock, resumeTokenKey: overrides?.resumeTokenKey, }); @@ -320,14 +328,14 @@ describe('ChangeStreamCoordinator', () => { await coordinator.shutdown(); }); - it('fans out SQL-shaped log events without document fields', async () => { - const { coordinator, stream, publish } = createCoordinator(); + it('fans out SQL-shaped WAL events without document fields', async () => { + const { coordinator, stream, publish } = createCoordinator({ persistResume: false }); await coordinator.reconcile(); stream.emit('change', { operationType: 'update', ns: { coll: 'orders' }, documentKey: { _id: 'order-1' }, - _id: '1842', + _id: '0/16B3748:12:1', wallTime: new Date('2026-03-01T00:00:00.000Z'), fullDocument: { secret: 'nope' }, }); @@ -344,15 +352,79 @@ describe('ChangeStreamCoordinator', () => { await coordinator.shutdown(); }); - it('ignores leftover Mongo tokens on the SQL resume key', async () => { - const { coordinator, watch, state } = createCoordinator({ - parseResumeToken: parseSqlResumeId, + it('opens a SQL watch without resume catch-up', async () => { + const { coordinator, watch, grpcSdk } = createCoordinator({ + persistResume: false, leaderLock: SQL_LEADER_LOCK, - resumeTokenKey: SQL_RESUME_TOKEN_KEY, }); - state.set(SQL_RESUME_TOKEN_KEY, EJSON.stringify({ _data: 'mongo' })); await coordinator.reconcile(); expect(watch).toHaveBeenCalledWith({ resumeAfter: undefined }); + expect(grpcSdk.state.tryAcquireLock).toHaveBeenCalledWith( + SQL_LEADER_LOCK, + expect.any(Number), + ); + expect(grpcSdk.state.getKey).not.toHaveBeenCalled(); + await coordinator.shutdown(); + }); + + it('does not persist resume tokens when persistResume is false', async () => { + const { coordinator, stream, grpcSdk } = createCoordinator({ persistResume: false }); + await coordinator.reconcile(); + stream.emit('change', { + operationType: 'insert', + ns: { coll: 'orders' }, + documentKey: { _id: 'order-1' }, + _id: '0/1:1:1', + wallTime: new Date('2026-03-01T00:00:00.000Z'), + }); + await coordinator.waitForIdle(); + expect(grpcSdk.state.setKey).not.toHaveBeenCalled(); + await coordinator.shutdown(); + }); + + it('stays starting until the watch is ready', async () => { + let resolveReady: () => void = () => undefined; + const watchReady = new Promise(resolve => { + resolveReady = resolve; + }); + const { coordinator } = createCoordinator({ + persistResume: false, + watchReady, + }); + const reconcile = coordinator.reconcile(); + await waitFor(() => coordinator.getState() === 'starting'); + expect(coordinator.getState()).toBe('starting'); + resolveReady(); + await reconcile; + expect(coordinator.getState()).toBe('live'); + await coordinator.shutdown(); + }); + + it('retries as degraded when the watch errors before it is live', async () => { + let resolveReady: () => void = () => undefined; + const watchReady = new Promise(resolve => { + resolveReady = resolve; + }); + const { coordinator, stream } = createCoordinator({ + persistResume: false, + watchReady, + }); + const reconcile = coordinator.reconcile(); + await waitFor(() => coordinator.getState() === 'starting'); + stream.emit('error', new Error('all replication slots are in use')); + resolveReady(); + await reconcile; + await waitFor(() => coordinator.getState() === 'degraded'); + expect(coordinator.getState()).toBe('degraded'); await coordinator.shutdown(); }); }); + +async function waitFor(predicate: () => boolean, timeoutMs = 2000): Promise { + const start = Date.now(); + while (Date.now() - start < timeoutMs) { + if (predicate()) return; + await new Promise(resolve => setImmediate(resolve)); + } + throw new Error('timed out waiting for condition'); +} diff --git a/modules/database/src/realtime/__tests__/sql-builders.test.ts b/modules/database/src/realtime/__tests__/sql-builders.test.ts index 6f7b35fae..663a65d90 100644 --- a/modules/database/src/realtime/__tests__/sql-builders.test.ts +++ b/modules/database/src/realtime/__tests__/sql-builders.test.ts @@ -1,104 +1,260 @@ import { describe, expect, it } from '@jest/globals'; -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, PK_COLUMN } from '../sql/constants.js'; -import { fetchChangeLogSql } from '../sql/changelog.js'; -import { - captureFunctionName, - fitIdentifier, - quoteIdent, - rowTriggerName, - triggerBaseName, -} from '../sql/identifiers.js'; -import { toRawChangeEvent } from '../sql/mapEvent.js'; import { normalizeChangeEvent } from '../normalize.js'; +import { + createPublicationSql, + addPublicationTableSql, + dropPublicationTableSql, + replicaIdentityFullSql, +} from '../sql/publication.js'; +import { PUBLICATION_NAME } from '../sql/constants.js'; +import { quoteIdent, quoteQualified } from '../sql/identifiers.js'; +import { documentIdFromChange, toRawChangeEvent } from '../sql/mapEvent.js'; +import { + PgoutputDecoder, + formatLsn, + parseLsn, + postgresTimeToDate, +} from '../sql/pgoutput.js'; -describe('SQL realtime builders', () => { - it('quotes identifiers per dialect', () => { - expect(quoteIdent('postgres', 'orders')).toBe('"orders"'); - expect(quoteIdent('mysql', 'orders')).toBe('`orders`'); - expect(quoteIdent('sqlite', 'weird"name')).toBe('"weird""name"'); +describe('PostgreSQL WAL publication SQL', () => { + it('creates a pgoutput publication for DML only', () => { + const sql = createPublicationSql(); + expect(sql).toContain(quoteIdent(PUBLICATION_NAME)); + expect(sql).toContain("publish = 'insert,update,delete'"); + expect(sql).not.toMatch(/TRIGGER|_cnd_DatabaseChange|pg_notify|LISTEN/i); }); - it('fits trigger names into dialect identifier limits', () => { - const long = 'c'.repeat(80); - expect(fitIdentifier(long, 64).length).toBeLessThanOrEqual(64); - expect(triggerBaseName(long, 'postgres').length).toBeLessThanOrEqual(63); - expect(rowTriggerName(long, 'i', 'mysql').length).toBeLessThanOrEqual(64); + it('adds and drops qualified tables', () => { + expect(addPublicationTableSql(PUBLICATION_NAME, 'public', 'orders')).toBe( + `ALTER PUBLICATION ${quoteIdent(PUBLICATION_NAME)} ADD TABLE ${quoteQualified( + 'public', + 'orders', + )}`, + ); + expect(dropPublicationTableSql(PUBLICATION_NAME, 'public', 'orders')).toContain( + 'DROP TABLE', + ); + expect(replicaIdentityFullSql('public', 'orders')).toBe( + `ALTER TABLE ${quoteQualified('public', 'orders')} REPLICA IDENTITY FULL`, + ); }); +}); - 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', '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"'); +describe('pgoutput decoder', () => { + it('decodes relation + insert/update/delete without leaking extra columns into the mapped id', () => { + const decoder = new PgoutputDecoder(); + expect( + decoder.decodeMessage(encodeRelation(42, 'public', 'orders', ['_id', 'secret'])), + ).toBeUndefined(); + const insert = decoder.decodeMessage(encodeInsert(42, ['order-1', 'do-not-leak'])); + expect(insert).toMatchObject({ + tag: 'insert', + relation: { name: 'orders' }, + newRow: { _id: 'order-1', secret: 'do-not-leak' }, + }); + const update = decoder.decodeMessage( + encodeUpdate(42, ['order-1'], ['order-1', 'still-secret']), + ); + expect(update?.tag).toBe('update'); + expect(update && 'newRow' in update ? update.newRow : undefined).toMatchObject({ + _id: 'order-1', + secret: 'still-secret', + }); + const del = decoder.decodeMessage(encodeDelete(42, ['order-1'])); + expect(del?.tag).toBe('delete'); + expect(documentIdFromChange(del!)).toBe('order-1'); + expect(documentIdFromChange(insert!, 'sku')).toBeUndefined(); }); - 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'); - expect(sqlite.map(t => t.triggerName)).toEqual([ - rowTriggerName('orders', 'i', 'sqlite'), - rowTriggerName('orders', 'u', 'sqlite'), - 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('decodes begin commit timestamps from the postgres epoch', () => { + const decoder = new PgoutputDecoder(); + const micros = 1_000_000n; + const begin = decoder.decodeMessage(encodeBegin(0x10n, micros, 9)); + expect(begin).toMatchObject({ tag: 'begin', xid: 9 }); + expect(begin && 'commitTime' in begin ? begin.commitTime : undefined).toEqual( + postgresTimeToDate(micros), + ); }); - it('builds dialect-specific change-log tables', () => { - expect(createChangeLogTableSql('postgres')).toMatch(/BIGSERIAL/); - expect(createChangeLogTableSql('mysql')).toMatch(/AUTO_INCREMENT/); - expect(createChangeLogTableSql('sqlite')).toMatch(/AUTOINCREMENT/); + it('round-trips LSN formatting', () => { + expect(formatLsn(parseLsn('0/16B3748'))).toBe('0/016B3748'); }); - 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'); + it('consumes unchanged TOAST and binary columns without desyncing later text ids', () => { + const decoder = new PgoutputDecoder(); + expect( + decoder.decodeMessage( + encodeRelation(42, 'public', 'orders', ['blob', 'toast', '_id']), + ), + ).toBeUndefined(); + const insert = decoder.decodeMessage( + encodeInsertKinds(42, [ + { kind: 'b', bytes: Buffer.from([1, 2, 3, 4]) }, + { kind: 'u' }, + { kind: 't', value: 'order-1' }, + ]), + ); + expect(insert).toMatchObject({ + tag: 'insert', + newRow: { _id: 'order-1' }, + }); + expect(insert && 'newRow' in insert ? insert.newRow : {}).not.toHaveProperty('blob'); + expect(insert && 'newRow' in insert ? insert.newRow : {}).not.toHaveProperty('toast'); }); -}); -describe('SQL resume tokens', () => { - it('accepts decimal ids and ignores leftover Mongo tokens', () => { - expect(parseSqlResumeId('1842')).toBe('1842'); - expect(parseSqlResumeId(EJSON.stringify('1842'))).toBe('1842'); - expect(parseSqlResumeId(EJSON.stringify(1842))).toBe('1842'); - 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(); + it('throws on a short buffer instead of reading past the end', () => { + const decoder = new PgoutputDecoder(); + expect(() => decoder.decodeMessage(Buffer.from('B'))).toThrow(/underflow/); }); }); -describe('SQL change-log event mapping', () => { - it('maps log rows to metadata-only change events', () => { +describe('WAL event mapping', () => { + it('maps metadata-only change events and uses a custom PK', () => { const raw = toRawChangeEvent({ - id: '1842', - collection_name: 'orders', - document_id: 'order-1', operation: 'insert', - occurred_at: '2026-03-01T00:00:00.000Z', + table: 'orders', + documentId: 'sku-1', + lsn: '0/1:1:1', + occurredAt: new Date('2026-03-01T00:00:00.000Z'), }); const event = normalizeChangeEvent(raw, 'Order'); expect(event).toMatchObject({ operation: 'insert', schema: 'Order', - documentId: 'order-1', + documentId: 'sku-1', }); + expect(event).not.toHaveProperty('fullDocument'); + expect( + documentIdFromChange( + { + tag: 'insert', + newRow: { sku: 'sku-1', secret: 'hidden' }, + }, + 'sku', + ), + ).toBe('sku-1'); }); }); + +function encodeRelation( + oid: number, + schema: string, + name: string, + columns: string[], +): Buffer { + const parts = [ + Buffer.from('R'), + i32(oid), + cstring(schema), + cstring(name), + Buffer.from([100]), + i16(columns.length), + ]; + for (const column of columns) { + parts.push(Buffer.from([1]), cstring(column), i32(25), i32(-1)); + } + return Buffer.concat(parts); +} + +function encodeInsert(oid: number, values: (string | null)[]): Buffer { + return Buffer.concat([ + Buffer.from('I'), + i32(oid), + Buffer.from('N'), + encodeTuple(values), + ]); +} + +function encodeInsertKinds(oid: number, values: TupleColumn[]): Buffer { + return Buffer.concat([ + Buffer.from('I'), + i32(oid), + Buffer.from('N'), + encodeTypedTuple(values), + ]); +} + +function encodeUpdate( + oid: number, + key: (string | null)[], + values: (string | null)[], +): Buffer { + return Buffer.concat([ + Buffer.from('U'), + i32(oid), + Buffer.from('K'), + encodeTuple(key), + Buffer.from('N'), + encodeTuple(values), + ]); +} + +function encodeDelete(oid: number, key: (string | null)[]): Buffer { + return Buffer.concat([Buffer.from('D'), i32(oid), Buffer.from('K'), encodeTuple(key)]); +} + +function encodeBegin(finalLsn: bigint, micros: bigint, xid: number): Buffer { + const buf = Buffer.alloc(1 + 8 + 8 + 4); + buf[0] = 'B'.charCodeAt(0); + buf.writeBigUInt64BE(finalLsn, 1); + buf.writeBigInt64BE(micros, 9); + buf.writeInt32BE(xid, 17); + return buf; +} + +function encodeTuple(values: (string | null)[]): Buffer { + return encodeTypedTuple( + values.map(value => + value == null ? { kind: 'n' as const } : { kind: 't' as const, value }, + ), + ); +} + +type TupleColumn = + | { kind: 'n' } + | { kind: 'u' } + | { kind: 't'; value: string } + | { kind: 'b'; bytes: Buffer }; + +function encodeTypedTuple(values: TupleColumn[]): Buffer { + const parts = [i16(values.length)]; + for (const value of values) { + switch (value.kind) { + case 'n': + parts.push(Buffer.from('n')); + break; + case 'u': + parts.push(Buffer.from('u')); + break; + case 't': { + const bytes = Buffer.from(value.value, 'utf8'); + parts.push(Buffer.from('t'), i32(bytes.length), bytes); + break; + } + case 'b': + parts.push(Buffer.from('b'), i32(value.bytes.length), value.bytes); + break; + default: { + const _exhaustive: never = value; + return _exhaustive; + } + } + } + return Buffer.concat(parts); +} + +function cstring(value: string): Buffer { + return Buffer.concat([Buffer.from(value, 'utf8'), Buffer.from([0])]); +} + +function i16(value: number): Buffer { + const buf = Buffer.alloc(2); + buf.writeInt16BE(value); + return buf; +} + +function i32(value: number): Buffer { + const buf = Buffer.alloc(4); + buf.writeInt32BE(value); + return buf; +} 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 be5483f82..8d6868fa9 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,330 +1,85 @@ +import { EventEmitter } from 'node:events'; 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'; +import { SqlChangeStream } from '../sql/SqlChangeStream.js'; +import type { ReplicationChange, ReplicationFeed } from '../sql/replication.js'; -function run(db: sqlite3.Database, sql: string): Promise { - return new Promise((resolve, reject) => { - db.exec(sql, err => { - if (err) reject(err); - else resolve(); +describe('SqlChangeStream WAL contract', () => { + it('captures insert/update/delete without document fields', async () => { + const feed = new FakeFeed(); + const stream = new SqlChangeStream({ + connectionUri: 'postgres://localhost/db', + idFieldByTable: { orders: '_id' }, + createFeed: () => feed, }); - }); -} - -function all(db: sqlite3.Database, sql: string): Promise { - return new Promise((resolve, reject) => { - db.all(sql, (err, rows) => { - if (err) reject(err); - else resolve(rows as T[]); + const received: unknown[] = []; + stream.on('change', change => received.push(change)); + await stream.ready; + feed.push(row('insert', 'order-1', 'do-not-leak')); + feed.push(row('update', 'order-1', 'still-secret')); + feed.push({ + tag: 'delete', + table: 'orders', + keyRow: { _id: 'order-1' }, + lsn: '0/3:1:3', + occurredAt: new Date('2026-03-01T00:00:03.000Z'), }); - }); -} - -describe('SQLite change-log contract', () => { - it('captures insert/update/delete including raw SQL without document fields', async () => { - const db = new sqlite3.Database(':memory:'); - try { - await run(db, `CREATE TABLE "orders" (_id TEXT PRIMARY KEY, 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 ('order-1', 'do-not-leak')`, - ); - await run(db, `UPDATE "orders" SET secret = 'still-secret' WHERE _id = 'order-1'`); - await run(db, `DELETE FROM "orders" WHERE _id = 'order-1'`); - const rows = await all<{ - id: number; - collection_name: string; - document_id: string; - operation: string; - occurred_at: string; - }>( - db, - `SELECT id, collection_name, document_id, operation, occurred_at FROM "${CHANGE_LOG_TABLE}" ORDER BY id ASC`, - ); - expect(rows.map(row => row.operation)).toEqual(['insert', 'update', 'delete']); - const events = rows.map(row => - normalizeChangeEvent( - toRawChangeEvent({ - id: String(row.id), - collection_name: row.collection_name, - document_id: row.document_id, - operation: row.operation, - occurred_at: row.occurred_at, - }), - 'Order', - ), - ); - expect(events.map(event => event?.operation)).toEqual([ - 'insert', - 'update', - 'delete', - ]); - for (const event of events) { - expect(event).toMatchObject({ schema: 'Order', documentId: 'order-1' }); - expect(event).not.toHaveProperty('fullDocument'); - expect(JSON.stringify(event)).not.toContain('do-not-leak'); - expect(JSON.stringify(event)).not.toContain('still-secret'); - } - } finally { - await new Promise(resolve => { - db.close(() => resolve()); - }); - } - }); - - 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(resolve => { - db.close(() => resolve()); - }); + const events = (received as Parameters[0][]).map( + change => normalizeChangeEvent(change, 'Order'), + ); + expect(events.map(event => event?.operation)).toEqual(['insert', 'update', 'delete']); + for (const event of events) { + expect(event).toMatchObject({ schema: 'Order', documentId: 'order-1' }); + expect(event).not.toHaveProperty('fullDocument'); + expect(JSON.stringify(event)).not.toContain('do-not-leak'); + expect(JSON.stringify(event)).not.toContain('still-secret'); } + await stream.close(); }); +}); - 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(resolve => { - db.close(() => resolve()); - }); - } - }); +const logicalUri = process.env.SQL_LOGICAL_URI; +const describeLivePgoutput = logicalUri ? describe : describe.skip; - it('retargets leftover #1602 _id triggers onto the physical PK', async () => { - const sequelize = new Sequelize({ - dialect: 'sqlite', - storage: ':memory:', - logging: false, +describeLivePgoutput( + 'SqlChangeStream live pgoutput (set SQL_LOGICAL_URI; skipped in CI)', + () => { + it('requires a Postgres URI with wal_level=logical', () => { + expect(logicalUri).toMatch(/^postgres/); }); - 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(); - } - }); +function row(tag: 'insert' | 'update', id: string, secret: string): ReplicationChange { + const seq = tag === 'insert' ? '1' : '2'; + return { + tag, + table: 'orders', + newRow: { _id: id, secret }, + lsn: `0/${seq}:1:${seq}`, + occurredAt: new Date(`2026-03-01T00:00:0${seq}.000Z`), + }; +} - 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(); - } - }); +class FakeFeed implements ReplicationFeed { + readonly emitter = new EventEmitter(); + started = false; - 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(); - } - }); + on(event: 'change', listener: (change: ReplicationChange) => void): void; + on(event: 'error', listener: (err: Error) => void): void; + on(event: string, listener: (...args: never[]) => void): void { + this.emitter.on(event, listener); + } - 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(); - } - }); -}); + async start(): Promise { + this.started = true; + } -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 stop(): Promise { + return; + } -async function waitUntil(predicate: () => boolean, timeoutMs = 2000): Promise { - const start = Date.now(); - while (Date.now() - start < timeoutMs) { - if (predicate()) return; - await new Promise(resolve => setTimeout(resolve, 10)); + push(change: ReplicationChange): void { + this.emitter.emit('change', change); } - 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 index 9b48331b2..28ca233f6 100644 --- a/modules/database/src/realtime/__tests__/sql-realtime-support.test.ts +++ b/modules/database/src/realtime/__tests__/sql-realtime-support.test.ts @@ -1,56 +1,126 @@ +import { EventEmitter } from 'node:events'; 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'; +import { LEGACY_CHANGE_LOG_TABLE, PUBLICATION_NAME } from '../sql/constants.js'; +import { quoteIdent } from '../sql/identifiers.js'; +import type { ReplicationChange, ReplicationFeed } from '../sql/replication.js'; describe('SqlRealtimeSupport', () => { - it('does not CREATE TABLE when ensureLog is false', async () => { - const sequelize = new Sequelize({ - dialect: 'sqlite', - storage: ':memory:', - logging: false, - }); - try { + it('reports mysql and sqlite as unsupported', async () => { + for (const dialect of ['mysql', 'mariadb', 'sqlite'] as const) { + const sequelize = { + getDialect: () => dialect, + query: async () => [], + }; const support = new SqlRealtimeSupport({ sequelize, - connectionUri: 'sqlite://', + connectionUri: `${dialect}://localhost/db`, } 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([]); + const result = await support.checkTopology(); + expect(result.supported).toBe(false); + expect(result.message).toMatch(/PostgreSQL WAL CDC only/i); + } + }); + + it('fails topology when wal_level is not logical', async () => { + const support = new SqlRealtimeSupport({ + sequelize: { + getDialect: () => 'postgres', + query: async (sql: string) => { + if (sql.includes('pg_settings')) { + return [ + { name: 'wal_level', setting: 'replica' }, + { name: 'max_replication_slots', setting: '10' }, + { name: 'max_wal_senders', setting: '10' }, + ]; + } + return []; + }, + }, + connectionUri: 'postgres://localhost/db', + } as never); + const result = await support.checkTopology(); + expect(result.supported).toBe(false); + expect(result.message).toMatch(/wal_level=logical/); + }); + + it('does not create a probe replication slot during topology checks', async () => { + const connect = jest + .spyOn(pg.Client.prototype, 'connect') + .mockResolvedValue(undefined); + const query = jest + .spyOn(pg.Client.prototype, 'query') + .mockRejectedValue(new Error('all replication slots are in use')); + try { + const support = new SqlRealtimeSupport({ + sequelize: { + getDialect: () => 'postgres', + query: async (sql: string) => { + if (sql.includes('pg_settings')) { + return [ + { name: 'wal_level', setting: 'logical' }, + { name: 'max_replication_slots', setting: '1' }, + { name: 'max_wal_senders', setting: '1' }, + ]; + } + return []; + }, + }, + connectionUri: 'postgres://localhost/db', + } as never); + const result = await support.checkTopology(); + expect(result).toEqual({ supported: true }); + expect(connect).not.toHaveBeenCalled(); + expect(query).not.toHaveBeenCalled(); } finally { - await sequelize.close(); + connect.mockRestore(); + query.mockRestore(); } }); - it('retargets leftover postgres triggers that still call conduit_realtime_capture', async () => { - const triggerName = triggerBaseName('orders', 'postgres'); - const functionName = captureFunctionName('orders'); + it('fails topology when replication slots or wal senders are zero', async () => { + const support = new SqlRealtimeSupport({ + sequelize: { + getDialect: () => 'postgres', + query: async () => [ + { name: 'wal_level', setting: 'logical' }, + { name: 'max_replication_slots', setting: '0' }, + { name: 'max_wal_senders', setting: '10' }, + ], + }, + connectionUri: 'postgres://localhost/db', + } as never); + const result = await support.checkTopology(); + expect(result.supported).toBe(false); + expect(result.message).toMatch(/max_replication_slots/); + }); + + it('syncs publication tables and replica identity without changelog DDL', async () => { 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()', - }, - ]; + if (sql.includes('FROM pg_publication ') && sql.includes('pubname')) { + return []; + } + if (sql.includes('pg_publication_tables')) { + return []; + } + if (sql.includes('relreplident')) { + return [{ ident: 'd', has_pk: true }]; } return []; }, }; - await syncTriggers(sequelize as never, [ + const support = new SqlRealtimeSupport({ + sequelize, + connectionUri: 'postgres://localhost/db', + } as never); + await support.prepare([ { name: 'Order', collectionName: 'orders', @@ -58,143 +128,171 @@ describe('SqlRealtimeSupport', () => { documentIdField: 'sku', }, ]); - expect(queries.some(sql => sql.includes('DROP TRIGGER'))).toBe(true); + expect(queries.some(sql => sql.includes('CREATE PUBLICATION'))).toBe(true); + expect(queries.some(sql => sql.includes('ADD TABLE'))).toBe(true); + expect(queries.some(sql => sql.includes(quoteIdent(PUBLICATION_NAME)))).toBe(true); expect( queries.some( - sql => - sql.includes('CREATE TRIGGER') && - sql.includes(functionName) && - !sql.includes('conduit_realtime_capture'), + sql => sql.includes('_cnd_DatabaseChange') && sql.includes('CREATE TABLE'), ), - ).toBe(true); + ).toBe(false); + expect(queries.some(sql => sql.includes('CREATE TRIGGER'))).toBe(false); + expect(queries.some(sql => sql.includes('LISTEN'))).toBe(false); }); +}); - 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, [ +describe('SqlChangeStream', () => { + it('emits metadata-only WAL changes and uses the physical PK', async () => { + const feed = new FakeFeed(); + const stream = new SqlChangeStream({ + connectionUri: 'postgres://localhost/db', + idFieldByTable: { orders: 'sku' }, + createFeed: () => feed, + }); + const received: unknown[] = []; + stream.on('change', change => { + received.push(change); + }); + await stream.ready; + feed.push({ + tag: 'insert', + table: 'orders', + newRow: { sku: 'sku-1', secret: 'hidden' }, + lsn: '0/1:1:1', + occurredAt: new Date('2026-03-01T00:00:00.000Z'), + }); + expect(received).toEqual([ { - name: 'Order', - collectionName: 'orders', - authorizationEnabled: false, - documentIdField: 'sku', + operationType: 'insert', + ns: { coll: 'orders' }, + documentKey: { _id: 'sku-1' }, + wallTime: new Date('2026-03-01T00:00:00.000Z'), + _id: '0/1:1:1', }, ]); - 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); + expect(JSON.stringify(received)).not.toContain('hidden'); + await stream.close(); + expect(feed.stopped).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(); - } + it('skips rows with a NULL document id', async () => { + const feed = new FakeFeed(); + const stream = new SqlChangeStream({ + connectionUri: 'postgres://localhost/db', + createFeed: () => feed, + }); + const received: unknown[] = []; + stream.on('change', change => received.push(change)); + await stream.ready; + feed.push({ + tag: 'insert', + table: 'orders', + newRow: { _id: null, secret: 'ok' }, + lsn: '0/1:1:1', + occurredAt: new Date(), + }); + expect(received).toEqual([]); + await stream.close(); }); -}); -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[]) => void; - const firstFetch = new Promise[]>(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 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); + it('emits error when the replication feed cannot create a slot', async () => { const stream = new SqlChangeStream({ - sequelize: sequelize as never, connectionUri: 'postgres://localhost/db', - defaultCursor: '0', + createFeed: () => new FailingFeed('all replication slots are in use'), + }); + const err = await new Promise(resolve => { + stream.on('error', resolve); + }); + await stream.ready; + expect(err).toBeInstanceOf(Error); + expect((err as Error).message).toMatch(/slots are in use/); + await stream.close(); + }); +}); + +describe('legacy changelog cleanup', () => { + it('drops leftover sqlite changelog objects without breaking DML', async () => { + const sequelize = new Sequelize({ + dialect: 'sqlite', + storage: ':memory:', + logging: false, }); 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' }, - }); + await sequelize.query(`CREATE TABLE "orders" (_id TEXT PRIMARY KEY, secret TEXT)`); + await sequelize.query( + `CREATE TABLE "${LEGACY_CHANGE_LOG_TABLE}" ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + collection_name TEXT NOT NULL, + document_id TEXT NOT NULL, + operation TEXT NOT NULL, + occurred_at TEXT NOT NULL + )`, + ); + await sequelize.query( + `CREATE TRIGGER "cnd_rt_i_orders" AFTER INSERT ON "orders" + BEGIN + INSERT INTO "${LEGACY_CHANGE_LOG_TABLE}" (collection_name, document_id, operation, occurred_at) + VALUES ('orders', NEW."_id", 'insert', datetime('now')); + END`, + ); + const support = new SqlRealtimeSupport({ + sequelize, + connectionUri: 'sqlite://', + } as never); + await support.prepare([]); + await sequelize.query(`INSERT INTO "orders" (_id, secret) VALUES ('order-1', 'x')`); + const tables = await sequelize.query( + `SELECT name FROM sqlite_master WHERE type = 'table' AND name = :name`, + { type: QueryTypes.SELECT, replacements: { name: LEGACY_CHANGE_LOG_TABLE } }, + ); + const triggers = await sequelize.query( + `SELECT name FROM sqlite_master WHERE type = 'trigger' AND name LIKE 'cnd_rt_%'`, + { type: QueryTypes.SELECT }, + ); + expect(tables).toEqual([]); + expect(triggers).toEqual([]); } finally { - await stream.close(); - Client.mockRestore(); + await sequelize.close(); } }); }); -async function waitUntil(predicate: () => boolean, timeoutMs = 2000): Promise { - const start = Date.now(); - while (Date.now() - start < timeoutMs) { - if (predicate()) return; - await new Promise(resolve => setTimeout(resolve, 10)); +class FakeFeed implements ReplicationFeed { + readonly emitter = new EventEmitter(); + started = false; + stopped = false; + + on(event: 'change', listener: (change: ReplicationChange) => void): void; + on(event: 'error', listener: (err: Error) => void): void; + on(event: string, listener: (...args: never[]) => void): void { + this.emitter.on(event, listener); + } + + async start(): Promise { + this.started = true; + } + + async stop(): Promise { + this.stopped = true; + } + + push(change: ReplicationChange): void { + this.emitter.emit('change', change); + } +} + +class FailingFeed implements ReplicationFeed { + constructor(private readonly message: string) {} + + on(): void { + return; + } + + async start(): Promise { + throw new Error(this.message); + } + + async stop(): Promise { + return; } - throw new Error('timed out waiting for condition'); } diff --git a/modules/database/src/realtime/__tests__/sql-replication.test.ts b/modules/database/src/realtime/__tests__/sql-replication.test.ts new file mode 100644 index 000000000..ca0b4189f --- /dev/null +++ b/modules/database/src/realtime/__tests__/sql-replication.test.ts @@ -0,0 +1,203 @@ +import { describe, expect, it, jest } from '@jest/globals'; +import pg from 'pg'; +import { PgoutputDecoder } from '../sql/pgoutput.js'; +import { createPgoutputFeed } from '../sql/replication.js'; +import type { ReplicationChange } from '../sql/replication.js'; + +describe('pgoutput CopyData fixture', () => { + it('decodes a recorded XLogData insert after a relation message', () => { + const decoder = new PgoutputDecoder(); + const relation = decoder.decodeMessage( + encodeRelation(7, 'public', 'orders', ['_id', 'secret']), + ); + expect(relation).toBeUndefined(); + const insert = decoder.decodeMessage(encodeInsert(7, ['order-1', 'do-not-leak'])); + expect(insert).toMatchObject({ + tag: 'insert', + relation: { name: 'orders' }, + newRow: { _id: 'order-1', secret: 'do-not-leak' }, + }); + }); + + it('starts a temp slot, speaks CopyData w/k, and emits a metadata-only change', async () => { + const copyListeners: Array<(msg: { chunk: Buffer }) => void> = []; + const fakeConnection = { + on: (event: string, listener: (msg: { chunk: Buffer }) => void) => { + if (event === 'copyData') copyListeners.push(listener); + }, + sendCopyFromChunk: jest.fn(), + }; + const connect = jest + .spyOn(pg.Client.prototype, 'connect') + .mockImplementation(async function (this: pg.Client) { + Object.defineProperty(this, 'connection', { + value: fakeConnection, + configurable: true, + }); + }); + const query = jest + .spyOn(pg.Client.prototype, 'query') + .mockImplementation((sql: unknown) => { + const text = String(sql); + if (text.includes('CREATE_REPLICATION_SLOT')) { + return Promise.resolve({ + rows: [{ consistent_point: '0/16B3748' }], + }) as never; + } + if (text.includes('START_REPLICATION')) { + return new Promise(() => undefined) as never; + } + return Promise.reject(new Error(`unexpected query: ${text}`)) as never; + }); + const end = jest.spyOn(pg.Client.prototype, 'end').mockResolvedValue(undefined); + const feed = createPgoutputFeed({ + connectionUri: 'postgres://localhost/db', + publicationName: 'cnd_realtime', + }); + const changes: ReplicationChange[] = []; + const errors: Error[] = []; + feed.on('change', change => changes.push(change)); + feed.on('error', err => errors.push(err)); + try { + await feed.start(); + expect( + query.mock.calls.some(call => + String(call[0]).includes('CREATE_REPLICATION_SLOT'), + ), + ).toBe(true); + expect( + query.mock.calls.some(call => String(call[0]).includes('START_REPLICATION')), + ).toBe(true); + expect(copyListeners).toHaveLength(1); + copyListeners[0]({ + chunk: xlogData(encodeRelation(7, 'public', 'orders', ['_id', 'secret'])), + }); + copyListeners[0]({ + chunk: xlogData(encodeInsert(7, ['order-1', 'do-not-leak'])), + }); + copyListeners[0]({ chunk: keepalive(0x16b3748n, true) }); + expect(errors).toEqual([]); + expect(changes).toHaveLength(1); + expect(changes[0]).toMatchObject({ + tag: 'insert', + table: 'orders', + newRow: { _id: 'order-1', secret: 'do-not-leak' }, + }); + expect(fakeConnection.sendCopyFromChunk).toHaveBeenCalled(); + } finally { + await feed.stop(); + connect.mockRestore(); + query.mockRestore(); + end.mockRestore(); + } + }); + + it('emits error when CopyData is truncated', async () => { + const copyListeners: Array<(msg: { chunk: Buffer }) => void> = []; + const fakeConnection = { + on: (event: string, listener: (msg: { chunk: Buffer }) => void) => { + if (event === 'copyData') copyListeners.push(listener); + }, + sendCopyFromChunk: jest.fn(), + }; + const connect = jest + .spyOn(pg.Client.prototype, 'connect') + .mockImplementation(async function (this: pg.Client) { + Object.defineProperty(this, 'connection', { + value: fakeConnection, + configurable: true, + }); + }); + const query = jest + .spyOn(pg.Client.prototype, 'query') + .mockImplementation((sql: unknown) => { + const text = String(sql); + if (text.includes('CREATE_REPLICATION_SLOT')) { + return Promise.resolve({ + rows: [{ consistent_point: '0/1' }], + }) as never; + } + return new Promise(() => undefined) as never; + }); + const end = jest.spyOn(pg.Client.prototype, 'end').mockResolvedValue(undefined); + const feed = createPgoutputFeed({ connectionUri: 'postgres://localhost/db' }); + const errors: Error[] = []; + feed.on('error', err => errors.push(err)); + try { + await feed.start(); + copyListeners[0]({ + chunk: Buffer.concat([Buffer.from('w'), Buffer.alloc(24), Buffer.from('B')]), + }); + expect(errors[0]?.message).toMatch(/underflow/); + } finally { + await feed.stop(); + connect.mockRestore(); + query.mockRestore(); + end.mockRestore(); + } + }); +}); + +function xlogData(payload: Buffer, walStart = 0x16b3748n): Buffer { + const buf = Buffer.alloc(25 + payload.length); + buf[0] = 'w'.charCodeAt(0); + buf.writeBigUInt64BE(walStart, 1); + buf.writeBigUInt64BE(walStart, 9); + buf.writeBigInt64BE(0n, 17); + payload.copy(buf, 25); + return buf; +} + +function keepalive(walEnd: bigint, replyRequested: boolean): Buffer { + const buf = Buffer.alloc(1 + 8 + 8 + 1); + buf[0] = 'k'.charCodeAt(0); + buf.writeBigUInt64BE(walEnd, 1); + buf.writeBigInt64BE(0n, 9); + buf[17] = replyRequested ? 1 : 0; + return buf; +} + +function encodeRelation( + oid: number, + schema: string, + name: string, + columns: string[], +): Buffer { + const parts = [ + Buffer.from('R'), + i32(oid), + cstring(schema), + cstring(name), + Buffer.from([100]), + i16(columns.length), + ]; + for (const column of columns) { + parts.push(Buffer.from([1]), cstring(column), i32(25), i32(-1)); + } + return Buffer.concat(parts); +} + +function encodeInsert(oid: number, values: string[]): Buffer { + const parts = [Buffer.from('I'), i32(oid), Buffer.from('N'), i16(values.length)]; + for (const value of values) { + const bytes = Buffer.from(value, 'utf8'); + parts.push(Buffer.from('t'), i32(bytes.length), bytes); + } + return Buffer.concat(parts); +} + +function cstring(value: string): Buffer { + return Buffer.concat([Buffer.from(value, 'utf8'), Buffer.from([0])]); +} + +function i16(value: number): Buffer { + const buf = Buffer.alloc(2); + buf.writeInt16BE(value); + return buf; +} + +function i32(value: number): Buffer { + const buf = Buffer.alloc(4); + buf.writeInt32BE(value); + return buf; +} diff --git a/modules/database/src/realtime/__tests__/status.test.ts b/modules/database/src/realtime/__tests__/status.test.ts index 8eb3560af..2b95d4e1c 100644 --- a/modules/database/src/realtime/__tests__/status.test.ts +++ b/modules/database/src/realtime/__tests__/status.test.ts @@ -24,12 +24,24 @@ describe('buildRealtimeStatus', () => { expect( buildRealtimeStatus({ engine: 'mysql', - enabled: false, + enabled: true, + topologySupported: true, + activeSchemaCount: 1, + streamState: 'live', + }), + ).toMatchObject({ + status: 'unsupported', + message: 'Live updates are not supported for this database engine', + }); + expect( + buildRealtimeStatus({ + engine: 'sqlite', + enabled: true, topologySupported: true, activeSchemaCount: 1, streamState: 'live', }).status, - ).toBe('disabled'); + ).toBe('unsupported'); expect( buildRealtimeStatus({ engine: 'MongoDB', @@ -54,11 +66,20 @@ describe('buildRealtimeStatus', () => { enabled: true, topologySupported: false, topologyMessage: - 'PostgreSQL live updates need a session-mode connection that can LISTEN (not a transaction-mode pooler)', + 'PostgreSQL live updates require wal_level=logical (managed Postgres: enable logical replication / rds.logical_replication).', + activeSchemaCount: 1, + streamState: 'idle', + }).message, + ).toMatch(/wal_level=logical/); + expect( + buildRealtimeStatus({ + engine: 'PostgreSQL', + enabled: true, + topologySupported: false, activeSchemaCount: 1, streamState: 'idle', }).message, - ).toMatch(/LISTEN/); + ).toMatch(/logical replication/i); expect( buildRealtimeStatus({ engine: 'PostgreSQL', @@ -67,7 +88,7 @@ describe('buildRealtimeStatus', () => { activeSchemaCount: 1, streamState: 'idle', }).message, - ).toMatch(/internal change queue/i); + ).not.toMatch(/change queue|LISTEN|trigger/i); expect( buildRealtimeStatus({ engine: 'MongoDB', diff --git a/modules/database/src/realtime/sql/SqlChangeStream.ts b/modules/database/src/realtime/sql/SqlChangeStream.ts index 9861bb4b1..d4ad66933 100644 --- a/modules/database/src/realtime/sql/SqlChangeStream.ts +++ b/modules/database/src/realtime/sql/SqlChangeStream.ts @@ -1,51 +1,38 @@ import { EventEmitter } from 'node:events'; -import pg from 'pg'; -import type { Sequelize } from 'sequelize'; import type { ChangeStreamLike } from '../types.js'; +import { DEFAULT_ID_FIELD, PUBLICATION_NAME } from './constants.js'; +import { documentIdFromChange, toRawChangeEvent } from './mapEvent.js'; import { - CHANGE_LOG_BATCH_SIZE, - NOTIFY_CHANNEL, - POSTGRES_FALLBACK_POLL_MS, - SQL_POLL_INTERVAL_MS, - assertSqlDialect, - type SqlDialect, -} from './constants.js'; -import { changeLogLagMs, fetchChangeLogBatch } from './changelog.js'; -import { toRawChangeEvent } from './mapEvent.js'; -import { sqlCursorFromResumeAfter } from './resume.js'; + createPgoutputFeed, + type ReplicationChange, + type ReplicationFeed, + type ReplicationFeedFactory, +} from './replication.js'; export type SqlChangeStreamOptions = { - sequelize: Sequelize; connectionUri: string; - resumeAfter?: unknown; - defaultCursor?: string; + publicationName?: string; + idFieldByTable?: Record; + createFeed?: ReplicationFeedFactory; }; export class SqlChangeStream implements ChangeStreamLike { + readonly ready: Promise; private readonly emitter = new EventEmitter(); - 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 readonly feed: ReplicationFeed; + private readonly idFieldByTable: Record; 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.lagMs = changeLogLagMs(this.dialect); - this.cursor = sqlCursorFromResumeAfter(options.resumeAfter) ?? options.defaultCursor; - queueMicrotask(() => { - if (!this.closed) { - void this.start(); - } + this.idFieldByTable = options.idFieldByTable ?? {}; + this.feed = (options.createFeed ?? createPgoutputFeed)({ + connectionUri: options.connectionUri, + publicationName: options.publicationName ?? PUBLICATION_NAME, }); + this.feed.on('change', change => this.onChange(change)); + this.feed.on('error', err => this.emitError(err)); + this.ready = this.start(); } on( @@ -58,23 +45,10 @@ export class SqlChangeStream implements ChangeStreamLike { async close(): Promise { if (this.closed) return; this.closed = true; - if (this.pollTimer) { - clearInterval(this.pollTimer); - this.pollTimer = null; - } - const client = this.listenClient; - this.listenClient = null; - if (client) { - try { - await client.query(`UNLISTEN ${NOTIFY_CHANNEL}`); - } catch { - // ignore - } - try { - await client.end(); - } catch { - // ignore - } + try { + await this.feed.stop(); + } catch { + // already closed } this.emitter.emit('close'); } @@ -83,72 +57,27 @@ export class SqlChangeStream implements ChangeStreamLike { if (this.closed || this.started) return; this.started = true; try { - if (this.cursor === undefined) { - this.cursor = '0'; - } - if (this.dialect === 'postgres') { - await this.startPostgresListen(); - this.pollTimer = setInterval(() => { - this.requestDrain(); - }, POSTGRES_FALLBACK_POLL_MS); - } else { - this.pollTimer = setInterval(() => { - this.requestDrain(); - }, SQL_POLL_INTERVAL_MS); - } - this.requestDrain(); + await this.feed.start(); } catch (err) { this.emitError(err); } } - private async startPostgresListen(): Promise { - const client = new pg.Client({ connectionString: this.connectionUri }); - this.listenClient = client; - client.on('notification', () => { - this.requestDrain(); - }); - client.on('error', (err: Error) => { - this.emitError(err); - }); - await client.connect(); - await client.query(`LISTEN ${NOTIFY_CHANNEL}`); - } - - private requestDrain(): void { - this.drainRequested = true; - void this.drain(); - } - - private async drain(): Promise { - if (this.draining || this.closed) return; - this.draining = true; - try { - 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(); - } - } + private onChange(change: ReplicationChange): void { + if (this.closed) return; + const idField = this.idFieldByTable[change.table] ?? DEFAULT_ID_FIELD; + const documentId = documentIdFromChange(change, idField); + if (!documentId) return; + this.emitter.emit( + 'change', + toRawChangeEvent({ + operation: change.tag, + table: change.table, + documentId, + lsn: change.lsn, + occurredAt: change.occurredAt, + }), + ); } private emitError(err: unknown): void { diff --git a/modules/database/src/realtime/sql/SqlRealtimeSupport.ts b/modules/database/src/realtime/sql/SqlRealtimeSupport.ts index e2c45a112..b42ea54c5 100644 --- a/modules/database/src/realtime/sql/SqlRealtimeSupport.ts +++ b/modules/database/src/realtime/sql/SqlRealtimeSupport.ts @@ -1,79 +1,100 @@ -import pg from 'pg'; +import { QueryTypes } from 'sequelize'; import type { SequelizeAdapter } from '../../adapters/sequelize-adapter/index.js'; -import type { OptedInSchema } from '../types.js'; -import type { ChangeStreamLike } from '../types.js'; +import type { ChangeStreamLike, OptedInSchema } from '../types.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 { + DEFAULT_ID_FIELD, + PUBLICATION_NAME, + SQL_ENGINE_UNSUPPORTED, + sqlSchemaName, +} from './constants.js'; +import { dropLegacyCapture } from './leftover.js'; +import { syncPublication } from './publication.js'; import { SqlChangeStream } from './SqlChangeStream.js'; -import { syncTriggers } from './triggers.js'; +import type { ReplicationFeedFactory } from './replication.js'; export class SqlRealtimeSupport { - private watchFromId = '0'; + private schemas: OptedInSchema[] = []; - constructor(private readonly adapter: SequelizeAdapter) {} + constructor( + private readonly adapter: SequelizeAdapter, + private readonly createFeed?: ReplicationFeedFactory, + ) {} async checkTopology(): Promise { - 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)}`, - }; - } + await dropLegacyCapture(this.adapter.sequelize).catch(() => undefined); + const dialect = this.adapter.sequelize.getDialect(); if (dialect !== 'postgres') { - return { supported: true }; + return { supported: false, message: SQL_ENGINE_UNSUPPORTED }; } - 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 }; + await this.adapter.sequelize.query('SELECT 1'); } catch (err) { return { supported: false, - message: - 'PostgreSQL live updates need a session-mode connection that can LISTEN (not a transaction-mode pooler): ' + - errorMessage(err), + message: `SQL live updates cannot reach the database: ${errorMessage(err)}`, }; - } finally { - try { - await client.end(); - } catch { - // ignore - } } + return this.probeLogicalReplication(); } - async prepare( - schemas: OptedInSchema[], - options?: { ensureLog?: boolean }, - ): Promise { - const ensureLog = options?.ensureLog !== false; - if (ensureLog) { - await ensureChangeLog(this.adapter.sequelize); - this.watchFromId = await maxChangeLogId(this.adapter.sequelize); + async prepare(schemas: OptedInSchema[]): Promise { + this.schemas = schemas; + await dropLegacyCapture(this.adapter.sequelize); + if (this.adapter.sequelize.getDialect() !== 'postgres') { + return; } - await syncTriggers(this.adapter.sequelize, schemas); + await syncPublication(this.adapter.sequelize, schemas, { + schemaName: sqlSchemaName(), + publicationName: PUBLICATION_NAME, + }); } - openWatch(resumeAfter?: unknown): ChangeStreamLike { + openWatch(): ChangeStreamLike { return new SqlChangeStream({ - sequelize: this.adapter.sequelize, connectionUri: this.adapter.connectionUri, - resumeAfter, - defaultCursor: this.watchFromId, + publicationName: PUBLICATION_NAME, + idFieldByTable: Object.fromEntries( + this.schemas.map(schema => [ + schema.collectionName, + schema.documentIdField ?? DEFAULT_ID_FIELD, + ]), + ), + createFeed: this.createFeed, }); } - async trimThrough(resumeToken: string): Promise { - const id = parseSqlResumeId(resumeToken); - if (!id) return; - await trimChangeLog(this.adapter.sequelize, id); + private async probeLogicalReplication(): Promise { + const settings = await this.adapter.sequelize.query( + `SELECT name, setting + FROM pg_settings + WHERE name IN ('wal_level', 'max_replication_slots', 'max_wal_senders')`, + { type: QueryTypes.SELECT }, + ); + const map = new Map( + (settings as { name: string; setting: string }[]).map(row => [ + String(row.name), + String(row.setting), + ]), + ); + if (map.get('wal_level') !== 'logical') { + return { + supported: false, + message: + 'PostgreSQL live updates require wal_level=logical (managed Postgres: enable logical replication / rds.logical_replication).', + }; + } + if (map.get('max_replication_slots') === '0' || map.get('max_wal_senders') === '0') { + return { + supported: false, + message: + 'PostgreSQL live updates need max_replication_slots and max_wal_senders greater than 0.', + }; + } + // Settings only: do not CREATE_REPLICATION_SLOT here. Every pod reconciles; + // a probe slot would compete with the leader's live temp slot and fail-close + // the feed. Slot create belongs in PgoutputReplicationFeed.start() (degraded + retry). + return { supported: true }; } } diff --git a/modules/database/src/realtime/sql/changelog.ts b/modules/database/src/realtime/sql/changelog.ts deleted file mode 100644 index 177463781..000000000 --- a/modules/database/src/realtime/sql/changelog.ts +++ /dev/null @@ -1,169 +0,0 @@ -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 { 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 { - const dialect = assertSqlDialect(sequelize.getDialect()); - await sequelize.query(createChangeLogTableSql(dialect)); -} - -export async function fetchChangeLogBatch( - sequelize: Sequelize, - resumeId: string, - limit: number = CHANGE_LOG_BATCH_SIZE, - lagMs?: number, -): Promise { - const dialect = assertSqlDialect(sequelize.getDialect()); - 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[]).map(row => ({ - id: String(row.id), - collection_name: String(row.collection_name), - document_id: String(row.document_id), - operation: String(row.operation), - occurred_at: occurredAtValue(row.occurred_at), - })); -} - -export async function maxChangeLogId(sequelize: Sequelize): Promise { - const dialect = assertSqlDialect(sequelize.getDialect()); - const table = quoteIdent(dialect, CHANGE_LOG_TABLE); - const rows = await sequelize.query(`SELECT MAX(id) AS max_id FROM ${table}`, { - type: QueryTypes.SELECT, - }); - const maxId = (rows[0] as { max_id?: unknown } | undefined)?.max_id; - if (maxId === undefined || maxId === null) { - return '0'; - } - return String(maxId); -} - -export async function trimChangeLog( - sequelize: Sequelize, - throughId: string, -): Promise { - if (!/^\d+$/.test(throughId)) { - return; - } - const dialect = assertSqlDialect(sequelize.getDialect()); - const table = quoteIdent(dialect, CHANGE_LOG_TABLE); - for (let i = 0; i < 50; i++) { - const sql = trimSql(dialect, table); - const [, metadata] = await sequelize.query(sql, { - replacements: { id: throughId, limit: CHANGE_LOG_BATCH_SIZE }, - }); - const affected = affectedRows(metadata); - if (affected === 0) { - return; - } - } -} - -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 { - const replacements: Record = { 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': - case 'mariadb': - return `DELETE FROM ${table} WHERE id <= :id ORDER BY id ASC LIMIT :limit`; - case 'postgres': - case 'sqlite': - return `DELETE FROM ${table} WHERE id IN ( - SELECT id FROM ${table} WHERE id <= :id ORDER BY id ASC LIMIT :limit - )`; - default: { - const _exhaustive: never = dialect; - return _exhaustive; - } - } -} - -function occurredAtValue(value: unknown): Date | string | number { - if (value instanceof Date || typeof value === 'string' || typeof value === 'number') { - return value; - } - return new Date().toISOString(); -} - -function affectedRows(metadata: unknown): number { - if (!metadata || typeof metadata !== 'object') { - return 0; - } - const record = metadata as { rowCount?: unknown; affectedRows?: unknown }; - if (typeof record.rowCount === 'number') { - return record.rowCount; - } - if (typeof record.affectedRows === 'number') { - return record.affectedRows; - } - return 0; -} diff --git a/modules/database/src/realtime/sql/constants.ts b/modules/database/src/realtime/sql/constants.ts index 2f99444bd..5fca15e80 100644 --- a/modules/database/src/realtime/sql/constants.ts +++ b/modules/database/src/realtime/sql/constants.ts @@ -1,27 +1,26 @@ -export const CHANGE_LOG_TABLE = '_cnd_DatabaseChange'; -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'; - -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 PUBLICATION_NAME = 'cnd_realtime'; export const SQL_LEADER_LOCK = 'realtime:sql:change-stream:leader'; -export const SQL_RESUME_TOKEN_KEY = 'realtime:sql:resumeToken'; +export const DEFAULT_ID_FIELD = '_id'; +export const DEFAULT_SQL_SCHEMA = 'public'; + +export const LEGACY_CHANGE_LOG_TABLE = '_cnd_DatabaseChange'; +export const LEGACY_TRIGGER_PREFIX = 'cnd_rt_'; +export const LEGACY_CAPTURE_FUNCTION_PREFIX = 'cnd_rt_fn_'; +export const LEGACY_SHARED_CAPTURE_FUNCTION = 'conduit_realtime_capture'; export const SQL_DIALECTS = ['postgres', 'mysql', 'mariadb', 'sqlite'] as const; export type SqlDialect = (typeof SQL_DIALECTS)[number]; +export const LOGICAL_REPLICATION_UNAVAILABLE = + 'PostgreSQL live updates require logical replication (wal_level=logical, a pgoutput publication, and a replication slot). Leader restart or slot drop skips missed events; clients refetch.'; + +export const SQL_ENGINE_UNSUPPORTED = + 'Live updates are PostgreSQL WAL CDC only. MySQL, MariaDB, and SQLite are out of v1.'; + export function isSqlDialect(dialect: string): dialect is SqlDialect { return (SQL_DIALECTS as readonly string[]).includes(dialect); } -export function assertSqlDialect(dialect: string): SqlDialect { - if (isSqlDialect(dialect)) { - return dialect; - } - throw new Error(`Unsupported SQL dialect for live updates: ${dialect}`); +export function sqlSchemaName(): string { + return process.env.SQL_SCHEMA ?? DEFAULT_SQL_SCHEMA; } diff --git a/modules/database/src/realtime/sql/ddl.ts b/modules/database/src/realtime/sql/ddl.ts deleted file mode 100644 index 10eacccd9..000000000 --- a/modules/database/src/realtime/sql/ddl.ts +++ /dev/null @@ -1,86 +0,0 @@ -import { - CHANGE_LOG_TABLE, - NOTIFY_CHANNEL, - PK_COLUMN, - type SqlDialect, -} from './constants.js'; -import { quoteIdent } from './identifiers.js'; - -export function createChangeLogTableSql(dialect: SqlDialect): string { - const table = quoteIdent(dialect, CHANGE_LOG_TABLE); - switch (dialect) { - case 'postgres': - return `CREATE TABLE IF NOT EXISTS ${table} ( - id BIGSERIAL PRIMARY KEY, - collection_name TEXT NOT NULL, - document_id TEXT NOT NULL, - operation TEXT NOT NULL, - occurred_at TIMESTAMPTZ NOT NULL DEFAULT NOW() - )`; - case 'mysql': - case 'mariadb': - return `CREATE TABLE IF NOT EXISTS ${table} ( - id BIGINT NOT NULL AUTO_INCREMENT PRIMARY KEY, - collection_name VARCHAR(255) NOT NULL, - document_id VARCHAR(255) NOT NULL, - operation VARCHAR(16) NOT NULL, - occurred_at TIMESTAMP(6) NOT NULL DEFAULT CURRENT_TIMESTAMP(6) - )`; - case 'sqlite': - return `CREATE TABLE IF NOT EXISTS ${table} ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - collection_name TEXT NOT NULL, - document_id TEXT NOT NULL, - operation TEXT NOT NULL, - occurred_at TEXT NOT NULL DEFAULT (datetime('now')) - )`; - default: { - const _exhaustive: never = dialect; - return _exhaustive; - } - } -} - -export function createCaptureFunctionSql( - pkColumn: string = PK_COLUMN, - functionName: string, -): string { - const table = quoteIdent('postgres', CHANGE_LOG_TABLE); - const pk = quoteIdent('postgres', pkColumn); - const fn = quoteIdent('postgres', functionName); - const channel = NOTIFY_CHANNEL.replace(/'/g, "''"); - return `CREATE OR REPLACE FUNCTION ${fn}() RETURNS trigger AS $$ -DECLARE - doc_id text; - op text; -BEGIN - IF TG_OP = 'DELETE' THEN - doc_id := OLD.${pk}::text; - op := 'delete'; - ELSIF TG_OP = 'INSERT' THEN - doc_id := NEW.${pk}::text; - op := 'insert'; - ELSE - doc_id := NEW.${pk}::text; - op := 'update'; - END IF; - IF doc_id IS NULL THEN - IF TG_OP = 'DELETE' THEN - RETURN OLD; - END IF; - RETURN NEW; - END IF; - INSERT INTO ${table} (collection_name, document_id, operation, occurred_at) - VALUES (TG_TABLE_NAME, doc_id, op, NOW()); - PERFORM pg_notify('${channel}', ''); - IF TG_OP = 'DELETE' THEN - RETURN OLD; - END IF; - RETURN NEW; -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 69b83ba5e..312b06ecf 100644 --- a/modules/database/src/realtime/sql/identifiers.ts +++ b/modules/database/src/realtime/sql/identifiers.ts @@ -1,68 +1,15 @@ -import { createHash } from 'node:crypto'; -import type { SqlDialect } from './constants.js'; -import { CAPTURE_FUNCTION_PREFIX, TRIGGER_NAME_PREFIX } from './constants.js'; - -const MYSQL_IDENT_LIMIT = 64; -const POSTGRES_IDENT_LIMIT = 63; -const SQLITE_IDENT_LIMIT = 128; - -export function quoteIdent(dialect: SqlDialect, name: string): string { - if (dialect === 'mysql' || dialect === 'mariadb') { - return `\`${name.replace(/`/g, '``')}\``; - } +export function quoteIdent(name: string): string { return `"${name.replace(/"/g, '""')}"`; } -export function sqlStringLiteral(value: string): string { - return `'${value.replace(/'/g, "''")}'`; -} - -export function identifierLimit(dialect: SqlDialect): number { - switch (dialect) { - case 'postgres': - return POSTGRES_IDENT_LIMIT; - case 'mysql': - case 'mariadb': - return MYSQL_IDENT_LIMIT; - case 'sqlite': - return SQLITE_IDENT_LIMIT; - default: { - const _exhaustive: never = dialect; - return _exhaustive; - } - } +export function quoteQualified(schema: string, table: string): string { + return `${quoteIdent(schema)}.${quoteIdent(table)}`; } -export function triggerBaseName(collectionName: string, dialect: SqlDialect): string { - return fitIdentifier( - `${TRIGGER_NAME_PREFIX}${collectionName}`, - identifierLimit(dialect), - ); -} - -export function captureFunctionName(collectionName: string): string { - return fitIdentifier( - `${CAPTURE_FUNCTION_PREFIX}${collectionName}`, - identifierLimit('postgres'), - ); -} - -export function rowTriggerName( - collectionName: string, - operation: 'i' | 'u' | 'd', - dialect: SqlDialect, -): string { - return fitIdentifier( - `${TRIGGER_NAME_PREFIX}${operation}_${collectionName}`, - identifierLimit(dialect), - ); +export function quoteLiteral(value: string): string { + return `'${value.replace(/'/g, "''")}'`; } -export function fitIdentifier(raw: string, maxLength: number): string { - if (raw.length <= maxLength) { - return raw; - } - const hash = createHash('sha1').update(raw).digest('hex').slice(0, 8); - const keep = Math.max(0, maxLength - hash.length - 1); - return `${raw.slice(0, keep)}_${hash}`; +export function mysqlQuoteIdent(name: string): string { + return `\`${name.replace(/`/g, '``')}\``; } diff --git a/modules/database/src/realtime/sql/index.ts b/modules/database/src/realtime/sql/index.ts index 75cf9ee99..8812a322f 100644 --- a/modules/database/src/realtime/sql/index.ts +++ b/modules/database/src/realtime/sql/index.ts @@ -1,20 +1,4 @@ -export { - CHANGE_LOG_TABLE, - SQL_DIALECTS, - SQL_LEADER_LOCK, - SQL_RESUME_TOKEN_KEY, -} from './constants.js'; +export { PUBLICATION_NAME, SQL_LEADER_LOCK } from './constants.js'; export { SqlChangeStream } from './SqlChangeStream.js'; export { SqlRealtimeSupport } from './SqlRealtimeSupport.js'; -export { parseSqlResumeId, sqlCursorFromResumeAfter } from './resume.js'; -export { desiredTriggers, syncTriggers } from './triggers.js'; -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'; +export { dropLegacyCapture } from './leftover.js'; diff --git a/modules/database/src/realtime/sql/leftover.ts b/modules/database/src/realtime/sql/leftover.ts new file mode 100644 index 000000000..bf0d45a44 --- /dev/null +++ b/modules/database/src/realtime/sql/leftover.ts @@ -0,0 +1,117 @@ +import { QueryTypes, Sequelize } from 'sequelize'; +import { + LEGACY_CAPTURE_FUNCTION_PREFIX, + LEGACY_CHANGE_LOG_TABLE, + LEGACY_SHARED_CAPTURE_FUNCTION, + LEGACY_TRIGGER_PREFIX, + type SqlDialect, + isSqlDialect, +} from './constants.js'; +import { mysqlQuoteIdent, quoteIdent } from './identifiers.js'; + +export async function dropLegacyCapture(sequelize: Sequelize): Promise { + const dialect = sequelize.getDialect(); + if (!isSqlDialect(dialect)) return; + switch (dialect) { + case 'postgres': + await dropPostgresLegacy(sequelize); + return; + case 'mysql': + case 'mariadb': + await dropMysqlLegacy(sequelize); + return; + case 'sqlite': + await dropSqliteLegacy(sequelize); + return; + default: { + const _exhaustive: never = dialect; + return _exhaustive; + } + } +} + +async function dropPostgresLegacy(sequelize: Sequelize): Promise { + const triggers = await sequelize.query( + `SELECT event_object_schema AS table_schema, + event_object_table AS table_name, + trigger_name AS trigger_name + FROM information_schema.triggers + WHERE trigger_name LIKE :prefix`, + { + type: QueryTypes.SELECT, + replacements: { prefix: `${LEGACY_TRIGGER_PREFIX}%` }, + }, + ); + const seen = new Set(); + for (const row of triggers as { + table_schema: string; + table_name: string; + trigger_name: string; + }[]) { + const key = `${row.table_schema}.${row.table_name}.${row.trigger_name}`; + if (seen.has(key)) continue; + seen.add(key); + await sequelize.query( + `DROP TRIGGER IF EXISTS ${quoteIdent(row.trigger_name)} ON ${quoteIdent( + row.table_schema, + )}.${quoteIdent(row.table_name)}`, + ); + } + const functions = await sequelize.query( + `SELECT p.proname AS function_name + FROM pg_proc p + JOIN pg_namespace n ON n.oid = p.pronamespace + WHERE n.nspname = current_schema() + AND (p.proname LIKE :prefix OR p.proname = :shared)`, + { + type: QueryTypes.SELECT, + replacements: { + prefix: `${LEGACY_CAPTURE_FUNCTION_PREFIX}%`, + shared: LEGACY_SHARED_CAPTURE_FUNCTION, + }, + }, + ); + for (const row of functions as { function_name: string }[]) { + await sequelize.query(`DROP FUNCTION IF EXISTS ${quoteIdent(row.function_name)}()`); + } + await sequelize.query(`DROP TABLE IF EXISTS ${quoteIdent(LEGACY_CHANGE_LOG_TABLE)}`); +} + +async function dropMysqlLegacy(sequelize: Sequelize): Promise { + const triggers = await sequelize.query( + `SELECT trigger_name AS trigger_name + FROM information_schema.triggers + WHERE trigger_schema = DATABASE() + AND trigger_name LIKE :prefix`, + { + type: QueryTypes.SELECT, + replacements: { prefix: `${LEGACY_TRIGGER_PREFIX}%` }, + }, + ); + const seen = new Set(); + for (const row of triggers as { trigger_name: string }[]) { + const name = String(row.trigger_name); + if (seen.has(name)) continue; + seen.add(name); + await sequelize.query(`DROP TRIGGER IF EXISTS ${mysqlQuoteIdent(name)}`); + } + await sequelize.query( + `DROP TABLE IF EXISTS ${mysqlQuoteIdent(LEGACY_CHANGE_LOG_TABLE)}`, + ); +} + +async function dropSqliteLegacy(sequelize: Sequelize): Promise { + const triggers = await sequelize.query( + `SELECT name AS trigger_name + FROM sqlite_master + WHERE type = 'trigger' AND name LIKE :prefix`, + { + type: QueryTypes.SELECT, + replacements: { prefix: `${LEGACY_TRIGGER_PREFIX}%` }, + }, + ); + for (const row of triggers as { trigger_name: string }[]) { + await sequelize.query(`DROP TRIGGER IF EXISTS ${quoteIdent(row.trigger_name)}`); + } + await sequelize.query(`DROP TABLE IF EXISTS ${quoteIdent(LEGACY_CHANGE_LOG_TABLE)}`); +} diff --git a/modules/database/src/realtime/sql/mapEvent.ts b/modules/database/src/realtime/sql/mapEvent.ts index 0a569beed..0dfae6b51 100644 --- a/modules/database/src/realtime/sql/mapEvent.ts +++ b/modules/database/src/realtime/sql/mapEvent.ts @@ -1,21 +1,42 @@ import type { RawChangeEvent } from '../normalize.js'; +import { DEFAULT_ID_FIELD } from './constants.js'; -export type ChangeLogRow = { - id: string; - collection_name: string; - document_id: string; - operation: string; - occurred_at: Date | string | number; +export type ChangeRows = { + tag: 'insert' | 'update' | 'delete'; + newRow?: Record; + oldRow?: Record; + keyRow?: Record; }; -export function toRawChangeEvent(row: ChangeLogRow): RawChangeEvent { - const occurredAt = - row.occurred_at instanceof Date ? row.occurred_at : new Date(row.occurred_at); +export type MappedWalChange = { + operation: 'insert' | 'update' | 'delete'; + table: string; + documentId: string; + lsn: string; + occurredAt: Date; +}; + +export function documentIdFromChange( + change: ChangeRows, + idField: string = DEFAULT_ID_FIELD, +): string | undefined { + const row = + change.tag === 'delete' + ? (change.keyRow ?? change.oldRow) + : (change.newRow ?? change.keyRow ?? change.oldRow); + const value = row?.[idField]; + if (value == null || value === '') { + return undefined; + } + return String(value); +} + +export function toRawChangeEvent(change: MappedWalChange): RawChangeEvent { return { - operationType: row.operation, - ns: { coll: row.collection_name }, - documentKey: { _id: row.document_id }, - wallTime: Number.isNaN(occurredAt.getTime()) ? new Date() : occurredAt, - _id: row.id, + operationType: change.operation, + ns: { coll: change.table }, + documentKey: { _id: change.documentId }, + wallTime: change.occurredAt, + _id: change.lsn, }; } diff --git a/modules/database/src/realtime/sql/pgoutput.ts b/modules/database/src/realtime/sql/pgoutput.ts new file mode 100644 index 000000000..1f925cb59 --- /dev/null +++ b/modules/database/src/realtime/sql/pgoutput.ts @@ -0,0 +1,261 @@ +export type PgoutputRelation = { + oid: number; + schema: string; + name: string; + columns: string[]; +}; + +export type PgoutputChange = { + tag: 'insert' | 'update' | 'delete'; + relation: PgoutputRelation; + newRow?: Record; + oldRow?: Record; + keyRow?: Record; +}; + +export type PgoutputBegin = { + tag: 'begin'; + finalLsn: bigint; + commitTime: Date; + xid: number; +}; + +const POSTGRES_EPOCH_MS = Date.UTC(2000, 0, 1); + +export class BufferReader { + constructor( + private readonly buf: Buffer, + private offset = 0, + ) {} + + remaining(): number { + return this.buf.length - this.offset; + } + + need(n: number): void { + if (n < 0 || this.remaining() < n) { + throw new Error(`pgoutput buffer underflow: need ${n}, have ${this.remaining()}`); + } + } + + u8(): number { + this.need(1); + const value = this.buf[this.offset]; + this.offset += 1; + return value; + } + + i16(): number { + this.need(2); + const value = this.buf.readInt16BE(this.offset); + this.offset += 2; + return value; + } + + i32(): number { + this.need(4); + const value = this.buf.readInt32BE(this.offset); + this.offset += 4; + return value; + } + + i64(): bigint { + this.need(8); + const value = this.buf.readBigInt64BE(this.offset); + this.offset += 8; + return value; + } + + u64(): bigint { + this.need(8); + const value = this.buf.readBigUInt64BE(this.offset); + this.offset += 8; + return value; + } + + cstring(): string { + const start = this.offset; + while (this.offset < this.buf.length && this.buf[this.offset] !== 0) { + this.offset += 1; + } + if (this.offset >= this.buf.length) { + throw new Error('pgoutput buffer underflow: unterminated cstring'); + } + const value = this.buf.subarray(start, this.offset).toString('utf8'); + this.offset += 1; + return value; + } + + bytes(length: number): Buffer { + this.need(length); + const value = this.buf.subarray(this.offset, this.offset + length); + this.offset += length; + return value; + } + + char(): string { + return String.fromCharCode(this.u8()); + } +} + +export function postgresTimeToDate(microseconds: bigint): Date { + return new Date(POSTGRES_EPOCH_MS + Number(microseconds / 1000n)); +} + +export function nowPostgresMicros(): bigint { + return BigInt(Date.now() - POSTGRES_EPOCH_MS) * 1000n; +} + +export function formatLsn(lsn: bigint): string { + const hi = Number(lsn >> 32n) >>> 0; + const lo = Number(lsn & 0xffffffffn) >>> 0; + return `${hi.toString(16).toUpperCase()}/${lo.toString(16).toUpperCase().padStart(8, '0')}`; +} + +export function parseLsn(value: string): bigint { + const [hi, lo] = value.split('/'); + if (!hi || !lo) { + throw new Error(`Invalid LSN: ${value}`); + } + return (BigInt(parseInt(hi, 16)) << 32n) + BigInt(parseInt(lo, 16)); +} + +export class PgoutputDecoder { + private readonly relations = new Map(); + + decodeMessage(payload: Buffer): PgoutputBegin | PgoutputChange | undefined { + if (payload.length === 0) return undefined; + const reader = new BufferReader(payload); + const tag = reader.char(); + switch (tag) { + case 'B': + return this.begin(reader); + case 'R': + this.relation(reader); + return undefined; + case 'I': + return this.insert(reader); + case 'U': + return this.update(reader); + case 'D': + return this.delete(reader); + default: + return undefined; + } + } + + private begin(reader: BufferReader): PgoutputBegin { + const finalLsn = reader.u64(); + const commitTime = postgresTimeToDate(reader.i64()); + const xid = reader.i32(); + return { tag: 'begin', finalLsn, commitTime, xid }; + } + + private relation(reader: BufferReader): void { + const oid = reader.i32(); + const schema = reader.cstring(); + const name = reader.cstring(); + reader.u8(); + const columnCount = reader.i16(); + const columns: string[] = []; + for (let i = 0; i < columnCount; i++) { + reader.u8(); + columns.push(reader.cstring()); + reader.i32(); + reader.i32(); + } + this.relations.set(oid, { oid, schema, name, columns }); + } + + private insert(reader: BufferReader): PgoutputChange | undefined { + const relation = this.relations.get(reader.i32()); + if (!relation) return undefined; + if (reader.char() !== 'N') return undefined; + return { + tag: 'insert', + relation, + newRow: readTuple(reader, relation.columns), + }; + } + + private update(reader: BufferReader): PgoutputChange | undefined { + const relation = this.relations.get(reader.i32()); + if (!relation) return undefined; + let keyRow: Record | undefined; + let oldRow: Record | undefined; + let kind = reader.char(); + if (kind === 'K' || kind === 'O') { + const row = readTuple(reader, relation.columns); + if (kind === 'K') keyRow = row; + else oldRow = row; + kind = reader.char(); + } + if (kind !== 'N') return undefined; + return { + tag: 'update', + relation, + newRow: readTuple(reader, relation.columns), + oldRow, + keyRow, + }; + } + + private delete(reader: BufferReader): PgoutputChange | undefined { + const relation = this.relations.get(reader.i32()); + if (!relation) return undefined; + const kind = reader.char(); + if (kind !== 'K' && kind !== 'O') return undefined; + const row = readTuple(reader, relation.columns); + return { + tag: 'delete', + relation, + oldRow: kind === 'O' ? row : undefined, + keyRow: kind === 'K' ? row : undefined, + }; + } +} + +function readTuple( + reader: BufferReader, + columns: string[], +): Record { + const count = reader.i16(); + const row: Record = {}; + for (let i = 0; i < count; i++) { + const name = columns[i]; + const value = readTupleColumn(reader); + if (name && value.set) { + row[name] = value.value; + } + } + return row; +} + +function readTupleColumn(reader: BufferReader): { + set: boolean; + value: string | null; +} { + const kind = reader.char(); + if (kind !== 'n' && kind !== 'u' && kind !== 't' && kind !== 'b') { + throw new Error(`Unsupported pgoutput tuple kind '${kind}'`); + } + switch (kind) { + case 'n': + return { set: true, value: null }; + case 'u': + return { set: false, value: null }; + case 't': { + const length = reader.i32(); + return { set: true, value: reader.bytes(length).toString('utf8') }; + } + case 'b': { + const length = reader.i32(); + reader.bytes(length); + return { set: false, value: null }; + } + default: { + const _exhaustive: never = kind; + return _exhaustive; + } + } +} diff --git a/modules/database/src/realtime/sql/publication.ts b/modules/database/src/realtime/sql/publication.ts new file mode 100644 index 000000000..8f334932d --- /dev/null +++ b/modules/database/src/realtime/sql/publication.ts @@ -0,0 +1,148 @@ +import { QueryTypes, Sequelize } from 'sequelize'; +import type { OptedInSchema } from '../types.js'; +import { PUBLICATION_NAME } from './constants.js'; +import { quoteIdent, quoteQualified } from './identifiers.js'; + +export type PublicationTable = { + schema: string; + table: string; +}; + +export function createPublicationSql(publicationName: string = PUBLICATION_NAME): string { + return `CREATE PUBLICATION ${quoteIdent(publicationName)} WITH (publish = 'insert,update,delete')`; +} + +export function addPublicationTableSql( + publicationName: string, + schema: string, + table: string, +): string { + return `ALTER PUBLICATION ${quoteIdent(publicationName)} ADD TABLE ${quoteQualified(schema, table)}`; +} + +export function dropPublicationTableSql( + publicationName: string, + schema: string, + table: string, +): string { + return `ALTER PUBLICATION ${quoteIdent(publicationName)} DROP TABLE ${quoteQualified(schema, table)}`; +} + +export function replicaIdentityFullSql(schema: string, table: string): string { + return `ALTER TABLE ${quoteQualified(schema, table)} REPLICA IDENTITY FULL`; +} + +export async function ensurePublication( + sequelize: Sequelize, + publicationName: string = PUBLICATION_NAME, +): Promise { + const rows = await sequelize.query( + `SELECT pubname FROM pg_publication WHERE pubname = :name`, + { type: QueryTypes.SELECT, replacements: { name: publicationName } }, + ); + if (rows.length > 0) return; + try { + await sequelize.query(createPublicationSql(publicationName)); + } catch (err) { + if (!isAlreadyPresent(err)) throw err; + } +} + +export async function listPublicationTables( + sequelize: Sequelize, + publicationName: string = PUBLICATION_NAME, +): Promise { + const rows = await sequelize.query( + `SELECT schemaname AS schema_name, tablename AS table_name + FROM pg_publication_tables + WHERE pubname = :name`, + { type: QueryTypes.SELECT, replacements: { name: publicationName } }, + ); + return (rows as { schema_name: string; table_name: string }[]).map(row => ({ + schema: String(row.schema_name), + table: String(row.table_name), + })); +} + +export async function syncPublication( + sequelize: Sequelize, + schemas: OptedInSchema[], + options: { schemaName: string; publicationName?: string }, +): Promise { + const publicationName = options.publicationName ?? PUBLICATION_NAME; + await ensurePublication(sequelize, publicationName); + const desired = new Map(); + for (const schema of schemas) { + desired.set(tableKey(options.schemaName, schema.collectionName), { + schema: options.schemaName, + table: schema.collectionName, + }); + } + const existing = await listPublicationTables(sequelize, publicationName); + for (const current of existing) { + if (desired.has(tableKey(current.schema, current.table))) continue; + await sequelize.query( + dropPublicationTableSql(publicationName, current.schema, current.table), + ); + } + const afterDrop = new Set( + (await listPublicationTables(sequelize, publicationName)).map(table => + tableKey(table.schema, table.table), + ), + ); + for (const table of desired.values()) { + const key = tableKey(table.schema, table.table); + await ensureReplicaIdentity(sequelize, table.schema, table.table); + if (afterDrop.has(key)) continue; + try { + await sequelize.query( + addPublicationTableSql(publicationName, table.schema, table.table), + ); + } catch (err) { + if (!isAlreadyPresent(err)) throw err; + } + } +} + +async function ensureReplicaIdentity( + sequelize: Sequelize, + schema: string, + table: string, +): Promise { + const rows = await sequelize.query( + `SELECT c.relreplident AS ident, + EXISTS ( + SELECT 1 FROM pg_index i + WHERE i.indrelid = c.oid AND i.indisprimary + ) AS has_pk + FROM pg_class c + JOIN pg_namespace n ON n.oid = c.relnamespace + WHERE n.nspname = :schema AND c.relname = :table AND c.relkind = 'r'`, + { type: QueryTypes.SELECT, replacements: { schema, table } }, + ); + const row = rows[0] as { ident?: string; has_pk?: unknown } | undefined; + if (!row) { + throw new Error( + `PostgreSQL live updates cannot publish ${quoteQualified(schema, table)}: table not found`, + ); + } + if (truthy(row.has_pk) || row.ident === 'f' || row.ident === 'i') { + return; + } + await sequelize.query(replicaIdentityFullSql(schema, table)); +} + +function tableKey(schema: string, table: string): string { + return `${schema}.${table}`; +} + +function truthy(value: unknown): boolean { + return ( + value === true || value === 't' || value === 'true' || value === 1 || value === '1' + ); +} + +function isAlreadyPresent(err: unknown): boolean { + const message = err instanceof Error ? err.message : String(err); + return /already member|already exists/i.test(message); +} diff --git a/modules/database/src/realtime/sql/replication.ts b/modules/database/src/realtime/sql/replication.ts new file mode 100644 index 000000000..beafb1222 --- /dev/null +++ b/modules/database/src/realtime/sql/replication.ts @@ -0,0 +1,219 @@ +import { EventEmitter } from 'node:events'; +import pg from 'pg'; +import { PUBLICATION_NAME } from './constants.js'; +import { quoteLiteral } from './identifiers.js'; +import { + BufferReader, + PgoutputDecoder, + formatLsn, + nowPostgresMicros, + parseLsn, + type PgoutputBegin, + type PgoutputChange, +} from './pgoutput.js'; + +export type ReplicationChange = { + tag: 'insert' | 'update' | 'delete'; + table: string; + newRow?: Record; + oldRow?: Record; + keyRow?: Record; + lsn: string; + occurredAt: Date; +}; + +export type ReplicationFeed = { + on(event: 'change', listener: (change: ReplicationChange) => void): void; + on(event: 'error', listener: (err: Error) => void): void; + start(): Promise; + stop(): Promise; +}; + +export type ReplicationFeedFactory = (options: { + connectionUri: string; + publicationName?: string; +}) => ReplicationFeed; + +type PgReplicationConnection = { + on(event: 'copyData', listener: (msg: { chunk: Buffer }) => void): void; + on(event: 'error', listener: (err: Error) => void): void; + sendCopyFromChunk?(chunk: Buffer): void; +}; + +const XLOG_HEADER_BYTES = 25; +const STANDBY_STATUS_INTERVAL_MS = 10_000; + +export class PgoutputReplicationFeed implements ReplicationFeed { + private readonly emitter = new EventEmitter(); + private readonly connectionUri: string; + private readonly publicationName: string; + private readonly decoder = new PgoutputDecoder(); + private client: pg.Client | null = null; + private ackTimer: NodeJS.Timeout | null = null; + private closed = false; + private started = false; + private lastBegin: PgoutputBegin | undefined; + private changeSeq = 0; + private flushedLsn = 0n; + + constructor(options: { connectionUri: string; publicationName?: string }) { + this.connectionUri = options.connectionUri; + this.publicationName = options.publicationName ?? PUBLICATION_NAME; + } + + on(event: 'change', listener: (change: ReplicationChange) => void): void; + on(event: 'error', listener: (err: Error) => void): void; + on( + event: 'change' | 'error', + listener: ((change: ReplicationChange) => void) | ((err: Error) => void), + ): void { + this.emitter.on(event, listener); + } + + async start(): Promise { + if (this.closed || this.started) return; + this.started = true; + const slotName = replicationSlotName(); + const client = createReplicationClient(this.connectionUri); + this.client = client; + client.on('error', err => this.emitError(err)); + await client.connect(); + if (this.closed) { + await this.stop(); + return; + } + const slot = await client.query( + `CREATE_REPLICATION_SLOT ${slotName} TEMPORARY LOGICAL pgoutput`, + ); + const consistentPoint = String(slot.rows[0]?.consistent_point ?? '0/0'); + this.flushedLsn = parseLsn(consistentPoint); + const connection = replicationConnection(client); + connection.on('copyData', msg => { + try { + this.onCopyData(msg.chunk, connection); + } catch (err) { + this.emitError(err); + } + }); + this.ackTimer = setInterval(() => { + sendStandbyStatus(connection, this.flushedLsn); + }, STANDBY_STATUS_INTERVAL_MS); + const startSql = + `START_REPLICATION SLOT ${slotName} LOGICAL ${consistentPoint} (` + + `proto_version '1', publication_names ${quoteLiteral(this.publicationName)})`; + void client.query(startSql).catch(err => this.emitError(err)); + } + + async stop(): Promise { + this.closed = true; + if (this.ackTimer) { + clearInterval(this.ackTimer); + this.ackTimer = null; + } + const client = this.client; + this.client = null; + if (!client) return; + try { + await client.end(); + } catch { + // already closed + } + } + + private onCopyData(chunk: Buffer, connection: PgReplicationConnection): void { + if (this.closed || chunk.length === 0) return; + const type = String.fromCharCode(chunk[0]); + if (type === 'k') { + this.onKeepalive(chunk, connection); + return; + } + if (type !== 'w' || chunk.length < XLOG_HEADER_BYTES) return; + const reader = new BufferReader(chunk, 1); + const walStart = reader.u64(); + reader.u64(); + reader.i64(); + const message = this.decoder.decodeMessage(chunk.subarray(XLOG_HEADER_BYTES)); + this.flushedLsn = walStart > this.flushedLsn ? walStart : this.flushedLsn; + if (!message) return; + if (message.tag === 'begin') { + this.lastBegin = message; + this.changeSeq = 0; + return; + } + this.emitChange(message, walStart); + } + + private onKeepalive(chunk: Buffer, connection: PgReplicationConnection): void { + if (chunk.length < 18) return; + const reader = new BufferReader(chunk, 1); + const walEnd = reader.u64(); + reader.i64(); + const replyRequested = reader.u8() === 1; + if (walEnd > this.flushedLsn) { + this.flushedLsn = walEnd; + } + if (replyRequested) { + sendStandbyStatus(connection, this.flushedLsn); + } + } + + private emitChange(change: PgoutputChange, walStart: bigint): void { + this.changeSeq += 1; + const xid = this.lastBegin?.xid ?? 0; + const occurredAt = this.lastBegin?.commitTime ?? new Date(); + this.emitter.emit('change', { + tag: change.tag, + table: change.relation.name, + newRow: change.newRow, + oldRow: change.oldRow, + keyRow: change.keyRow, + lsn: `${formatLsn(walStart)}:${xid}:${this.changeSeq}`, + occurredAt, + }); + } + + private emitError(err: unknown): void { + if (this.closed) return; + this.emitter.emit('error', err instanceof Error ? err : new Error(String(err))); + } +} + +export function createReplicationClient(connectionString: string): pg.Client { + const config: pg.ClientConfig & { replication: 'database' } = { + connectionString, + replication: 'database', + }; + return new pg.Client(config); +} + +export function createPgoutputFeed(options: { + connectionUri: string; + publicationName?: string; +}): ReplicationFeed { + return new PgoutputReplicationFeed(options); +} + +function replicationSlotName(): string { + return `cnd_rt_${process.pid}_${Math.random().toString(36).slice(2, 10)}`; +} + +function replicationConnection(client: pg.Client): PgReplicationConnection { + const connection = (client as unknown as { connection?: PgReplicationConnection }) + .connection; + if (!connection) { + throw new Error('PostgreSQL client is missing the replication connection'); + } + return connection; +} + +function sendStandbyStatus(connection: PgReplicationConnection, lsn: bigint): void { + if (!connection.sendCopyFromChunk) return; + const buf = Buffer.alloc(1 + 8 + 8 + 8 + 8 + 1); + buf[0] = 0x72; + buf.writeBigUInt64BE(lsn, 1); + buf.writeBigUInt64BE(lsn, 9); + buf.writeBigUInt64BE(lsn, 17); + buf.writeBigInt64BE(nowPostgresMicros(), 25); + buf[33] = 0; + connection.sendCopyFromChunk(buf); +} diff --git a/modules/database/src/realtime/sql/resume.ts b/modules/database/src/realtime/sql/resume.ts deleted file mode 100644 index 04ce0168e..000000000 --- a/modules/database/src/realtime/sql/resume.ts +++ /dev/null @@ -1,27 +0,0 @@ -import { parseResumeToken } from '../normalize.js'; - -export function parseSqlResumeId(token: string | null | undefined): string | undefined { - if (!token) return undefined; - if (/^\d+$/.test(token)) { - return token; - } - const parsed = parseResumeToken(token); - return decimalId(parsed); -} - -export function sqlCursorFromResumeAfter(resumeAfter: unknown): string | undefined { - return decimalId(resumeAfter); -} - -function decimalId(value: unknown): string | undefined { - if (typeof value === 'bigint' && value >= 0n) { - return value.toString(); - } - if (typeof value === 'string' && /^\d+$/.test(value)) { - return value; - } - 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 deleted file mode 100644 index 0ac0e9fa8..000000000 --- a/modules/database/src/realtime/sql/triggerSql.ts +++ /dev/null @@ -1,186 +0,0 @@ -import { 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, 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 ${quoteIdent(dialect, functionName)}()`, - dropSql: `DROP TRIGGER IF EXISTS ${quotedTrigger} ON ${table}`, - }, - ]; - } - case 'mysql': - case 'mariadb': - return [ - 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', - 'NEW', - table, - pk, - logTable, - collectionLiteral, - ), - sqliteRowTrigger( - collectionName, - pkColumn, - 'u', - 'UPDATE', - 'update', - 'NEW', - table, - pk, - logTable, - collectionLiteral, - ), - sqliteRowTrigger( - collectionName, - pkColumn, - 'd', - 'DELETE', - 'delete', - 'OLD', - table, - pk, - logTable, - collectionLiteral, - ), - ]; - default: { - const _exhaustive: never = dialect; - return _exhaustive; - } - } -} - -function mysqlRowTrigger( - dialect: SqlDialect, - collectionName: string, - pkColumn: string, - opKey: 'i' | 'u' | 'd', - timing: 'INSERT' | 'UPDATE' | 'DELETE', - operation: 'insert' | 'update' | 'delete', - row: 'NEW' | 'OLD', -): DesiredTrigger { - const triggerName = rowTriggerName(collectionName, opKey, dialect); - const quotedTrigger = quoteIdent(dialect, triggerName); - const table = quoteIdent(dialect, collectionName); - 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 - 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}`, - }; -} - -function sqliteRowTrigger( - collectionName: string, - pkColumn: string, - opKey: 'i' | 'u' | 'd', - timing: 'INSERT' | 'UPDATE' | 'DELETE', - operation: 'insert' | 'update' | 'delete', - row: 'NEW' | 'OLD', - table: string, - pk: string, - logTable: string, - collectionLiteral: string, -): DesiredTrigger { - const triggerName = rowTriggerName(collectionName, opKey, 'sqlite'); - const quotedTrigger = quoteIdent('sqlite', triggerName); - const opLiteral = sqlStringLiteral(operation); - return { - triggerName, - collectionName, - pkColumn, - sql: `CREATE TRIGGER ${quotedTrigger} AFTER ${timing} ON ${table} -BEGIN - INSERT INTO ${logTable} (collection_name, document_id, operation, occurred_at) - 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 deleted file mode 100644 index ce18b2f67..000000000 --- a/modules/database/src/realtime/sql/triggers.ts +++ /dev/null @@ -1,151 +0,0 @@ -import { QueryTypes, Sequelize } from 'sequelize'; -import type { OptedInSchema } from '../types.js'; -import { - CHANGE_LOG_TABLE, - PK_COLUMN, - TRIGGER_NAME_PREFIX, - type SqlDialect, - assertSqlDialect, -} from './constants.js'; -import { quoteIdent } from './identifiers.js'; -import { desiredTriggers, type DesiredTrigger } from './triggerSql.js'; - -export type { DesiredTrigger }; -export { desiredTriggers }; - -export type ExistingTrigger = { - triggerName: string; - tableName: string; - definition?: string; -}; - -export async function listExistingTriggers( - sequelize: Sequelize, -): Promise { - const dialect = assertSqlDialect(sequelize.getDialect()); - const prefix = `${TRIGGER_NAME_PREFIX}%`; - switch (dialect) { - case 'postgres': { - const rows = await sequelize.query( - `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; definition?: string }[], - ); - } - case 'mysql': - case 'mariadb': { - const rows = await sequelize.query( - `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; definition?: string }[], - ); - } - case 'sqlite': { - const rows = await sequelize.query( - `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; definition?: string }[], - ); - } - default: { - const _exhaustive: never = dialect; - return _exhaustive; - } - } -} - -function uniqueTriggers( - rows: { trigger_name: string; table_name: string; definition?: string }[], -): ExistingTrigger[] { - const seen = new Set(); - const result: ExistingTrigger[] = []; - for (const row of rows) { - const triggerName = String(row.trigger_name); - if (seen.has(triggerName)) continue; - seen.add(triggerName); - 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[], -): Promise { - const dialect = assertSqlDialect(sequelize.getDialect()); - const desired = new Map(); - for (const schema of schemas) { - if (schema.collectionName === CHANGE_LOG_TABLE) continue; - 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; - 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()) { - 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); - } -} - -function dropExistingSql(dialect: SqlDialect, trigger: ExistingTrigger): string { - const quotedTrigger = quoteIdent(dialect, trigger.triggerName); - if (dialect === 'postgres') { - return `DROP TRIGGER IF EXISTS ${quotedTrigger} ON ${quoteIdent(dialect, trigger.tableName)}`; - } - return `DROP TRIGGER IF EXISTS ${quotedTrigger}`; -} diff --git a/modules/database/src/realtime/status.ts b/modules/database/src/realtime/status.ts index 4e05c0d45..0f271e604 100644 --- a/modules/database/src/realtime/status.ts +++ b/modules/database/src/realtime/status.ts @@ -1,12 +1,7 @@ import type { RealtimeStatus, RealtimeStatusCode } from './types.js'; +import { LOGICAL_REPLICATION_UNAVAILABLE } from './sql/constants.js'; -const SUPPORTED_REALTIME_ENGINES = new Set([ - 'MongoDB', - 'PostgreSQL', - 'mysql', - 'mariadb', - 'sqlite', -]); +const SUPPORTED_REALTIME_ENGINES = new Set(['MongoDB', 'PostgreSQL']); export type RealtimeStatusInput = { engine: string; @@ -45,7 +40,7 @@ export function buildRealtimeStatus(input: RealtimeStatusInput): RealtimeStatus input.topologyMessage ?? (input.engine === 'MongoDB' ? 'A replica set or sharded MongoDB deployment is required for live updates' - : 'SQL live updates use an internal change queue (triggers), not native CDC. Database topology does not support live updates'), + : LOGICAL_REPLICATION_UNAVAILABLE), }; } if (input.socketsEnabled === false) { diff --git a/modules/database/src/realtime/types.ts b/modules/database/src/realtime/types.ts index be326cad1..9ca9b5767 100644 --- a/modules/database/src/realtime/types.ts +++ b/modules/database/src/realtime/types.ts @@ -47,4 +47,5 @@ export type ChangeStreamLike = { listener: (...args: unknown[]) => void, ): void; close(): Promise | void; + ready?: Promise; };