diff --git a/modules/database/README.mdx b/modules/database/README.mdx index 61bc195b0..aef5b52d5 100644 --- a/modules/database/README.mdx +++ b/modules/database/README.mdx @@ -34,20 +34,20 @@ since the latter need to go through parsers that are otherwise unnecessary for M ## Environment Variables πŸ“ƒ -| Variable | Description | Required | Example | Default | -|:--------------------:|:-----------------------------------------------------| :------: | :----------------: | :------: | -| `CONDUIT_SERVER` | Conduit Core's address and port | True | `0.0.0.0:55152` | - | -| `SERVICE_URL` | This should be where this service listens on. If behind a LB it should point to the LB's IP/DNS | False | `0.0.0.0:55190` | -| `GRPC_PORT` | The port number the gRPC server will listen to | False | `55190` | -| `GRPC_KEY` | Specifying a secret enables gRPC signed request protection (**use across modules**) | False | `someRandomSecret` | - | -| `DB_CONN_URI` | DB Connection URI | False | `postgres://conduit:pass@localhost:5432/conduit` | `mongodb://localhost:27017` | -| `DB_TYPE` | DB Engine Type | False | `postgres` | `mongodb` | +| Variable | Description | Required | Example | Default | +| :--------------: | :---------------------------------------------------------------------------------------------- | :------: | :----------------------------------------------: | :-------------------------: | +| `CONDUIT_SERVER` | Conduit Core's address and port | True | `0.0.0.0:55152` | - | +| `SERVICE_URL` | This should be where this service listens on. If behind a LB it should point to the LB's IP/DNS | False | `0.0.0.0:55190` | +| `GRPC_PORT` | The port number the gRPC server will listen to | False | `55190` | +| `GRPC_KEY` | Specifying a secret enables gRPC signed request protection (**use across modules**) | False | `someRandomSecret` | - | +| `DB_CONN_URI` | DB Connection URI | False | `postgres://conduit:pass@localhost:5432/conduit` | `mongodb://localhost:27017` | +| `DB_TYPE` | DB Engine Type | False | `postgres` | `mongodb` | ## Replica Set Configuration 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 also require a replica set or sharded cluster. Local Compose files initialize a single-node replica set so change streams can be exercised. +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,22 +60,29 @@ 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 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. + +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`. + +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. Admin sockets must be enabled (`admin.transports.sockets`) and the Admin socket port (`ADMIN_SOCKET_PORT`, default 3031) reachable from the UI. ### Configuration Options -| Setting | Values | Default | Description | -| :---------------: | :------------------------------------------------------------------------ | :-------: | :------------------------------------------------------- | -| `readPreference` | `primary`, `primaryPreferred`, `secondary`, `secondaryPreferred`, `nearest` | `primary` | Controls which replica set members receive read queries | -| `writeConcern` | `1`, `majority` | `1` | How many members must acknowledge a write | -| `readConcern` | `local`, `available`, `majority`, `linearizable`, `snapshot` | `local` | Consistency level for read operations | -| `realtime.enabled` | `true`, `false` | `false` | Enable MongoDB change-stream live updates for opted-in schemas | +| Setting | Values | Default | Description | +| :----------------: | :-------------------------------------------------------------------------- | :-------: | :------------------------------------------------------ | +| `readPreference` | `primary`, `primaryPreferred`, `secondary`, `secondaryPreferred`, `nearest` | `primary` | Controls which replica set members receive read queries | +| `writeConcern` | `1`, `majority` | `1` | How many members must acknowledge a write | +| `readConcern` | `local`, `available`, `majority`, `linearizable`, `snapshot` | `local` | Consistency level for read operations | +| `realtime.enabled` | `true`, `false` | `false` | Enable live updates for opted-in schemas | ### Recommended Production Settings For MongoDB Atlas deployments with read replicas: + - **readPreference**: `secondaryPreferred` β€” distributes reads across replicas, falls back to primary - **writeConcern**: `majority` β€” ensures writes survive replica set elections - **readConcern**: `local` β€” suitable for most workloads @@ -87,7 +94,7 @@ Modules can override the configured readPreference on individual queries when th ```typescript const doc = await MySchema.getInstance().findOne( { _id: someId }, - { readPreference: 'primary' } + { readPreference: 'primary' }, ); ``` @@ -100,19 +107,19 @@ Standalone deployments (without replicas) work identically regardless of these s | Operator | Description | | :--------: | :----------------------------------------------------------------------------------------------------------------------- | -| `in` | Matches any of the values specified in an array | +| `in` | Matches any of the values specified in an array | | `contains` | Checks if a value is contained in an array or not. | -| `nin` | Selects the documents where the value of a field is not equal any value in the specified array. | -| `eq` | Matches documents where the value of a field equals the specified value. | -| `ne` | Selects the documents where the value of the field is not equal to the specified value | -| `lt` | Selects the documents where the value of the field is less than (i.e. <) the specified value. | -| `gt` | Selects the documents where the value of the field is greater than (i.e. >) the specified value. | -| `lte` | Selects the documents where the value of the field is less or equal than (i.e. <=) the specified value. | -| `gte` | Selects the documents where the value of the field is greater or equal than (i.e. >=) the specified value. | -| `or` | Performs OR operation on an array of two or more expressions and selects the documents that satisfy at least one | -| `and` | Performs AND operation on an array of two or more expressions and selects the documents that satisfy at least one | -| `not` | Performs NOT operation on an array of two or more expressions and selects the documents that do not match the expression | -| `regex` | Select the documents where the value of the field matches the regex. | +| `nin` | Selects the documents where the value of a field is not equal any value in the specified array. | +| `eq` | Matches documents where the value of a field equals the specified value. | +| `ne` | Selects the documents where the value of the field is not equal to the specified value | +| `lt` | Selects the documents where the value of the field is less than (i.e. <) the specified value. | +| `gt` | Selects the documents where the value of the field is greater than (i.e. >) the specified value. | +| `lte` | Selects the documents where the value of the field is less or equal than (i.e. <=) the specified value. | +| `gte` | Selects the documents where the value of the field is greater or equal than (i.e. >=) the specified value. | +| `or` | Performs OR operation on an array of two or more expressions and selects the documents that satisfy at least one | +| `and` | Performs AND operation on an array of two or more expressions and selects the documents that satisfy at least one | +| `not` | Performs NOT operation on an array of two or more expressions and selects the documents that do not match the expression | +| `regex` | Select the documents where the value of the field matches the regex. | These operators have been tested thoroughly in varying levels of complexity. @@ -137,6 +144,7 @@ add its id in place of the data. ## Sequelize Caveats Currently, the Sequelize implementation has the following limitations: + - No index creation - Update queries only update provided fields without the option to replace entire rows, unless all columns are provided. - Like operations do not work diff --git a/modules/database/package.json b/modules/database/package.json index 656bee595..c5a30675a 100644 --- a/modules/database/package.json +++ b/modules/database/package.json @@ -72,9 +72,10 @@ "@jest/globals": "^30.5.1", "@types/convict": "^6.1.6", "@types/jest": "^30.0.0", - "@types/object-hash": "^3.0.6", "@types/lodash-es": "^4.17.12", "@types/node": "24.13.3", + "@types/object-hash": "^3.0.6", + "@types/pg": "^8.23.1", "copyfiles": "^2.4.1", "jest": "^30.5.1", "rimraf": "^6.1.3", diff --git a/modules/database/src/adapters/sequelize-adapter/index.ts b/modules/database/src/adapters/sequelize-adapter/index.ts index 30df75904..e4a172219 100644 --- a/modules/database/src/adapters/sequelize-adapter/index.ts +++ b/modules/database/src/adapters/sequelize-adapter/index.ts @@ -140,6 +140,7 @@ export abstract class SequelizeAdapter extends DatabaseAdapter this.models['_DeclaredSchema'].originalSchema.collectionName; for (const table of tableNames) { if (table === declaredSchemaTableName) continue; + if (table.startsWith('_cnd_')) continue; const tableInDeclaredSchemas = declaredSchemas.some( (declaredSchema: ConduitSchema) => { if (declaredSchema.collectionName && declaredSchema.collectionName !== '') { diff --git a/modules/database/src/config/index.ts b/modules/database/src/config/index.ts index 1ddda80d7..9fda0bb6b 100644 --- a/modules/database/src/config/index.ts +++ b/modules/database/src/config/index.ts @@ -49,7 +49,7 @@ const AppConfigSchema = { }, realtime: { enabled: { - doc: 'Enable MongoDB change-stream live updates for opted-in schemas', + doc: 'Enable live updates for opted-in schemas', format: 'Boolean', default: false, }, diff --git a/modules/database/src/realtime/MongoChangeStreamCoordinator.ts b/modules/database/src/realtime/ChangeStreamCoordinator.ts similarity index 65% rename from modules/database/src/realtime/MongoChangeStreamCoordinator.ts rename to modules/database/src/realtime/ChangeStreamCoordinator.ts index b41cc2f89..6e99d791b 100644 --- a/modules/database/src/realtime/MongoChangeStreamCoordinator.ts +++ b/modules/database/src/realtime/ChangeStreamCoordinator.ts @@ -1,15 +1,12 @@ +import { EJSON } from 'bson'; import { ConduitGrpcSdk } from '@conduitplatform/grpc-sdk'; import { normalizeChangeEvent, - parseResumeToken, + parseResumeToken as parseMongoResumeToken, type RawChangeEvent, } from './normalize.js'; import { authorizedDocumentRoom, roomsForPublicChange } from './rooms.js'; -import { - isResumeTokenUnusable, - topologyFromHello, - type TopologyResult, -} from './topology.js'; +import { isResumeTokenUnusable, type TopologyResult } from './topology.js'; import type { ChangeStreamLike, DatabaseChangeEvent, @@ -17,7 +14,7 @@ import type { RealtimeStatusCode, } from './types.js'; import type { RealtimeSubscriptionTracker } from './subscriptions.js'; -import { canReadDocument, type AuthorizationSdk } from './authorize.js'; +import { documentReadDecision, type AuthorizationSdk } from './authorize.js'; const LEADER_LOCK = 'realtime:change-stream:leader'; const RESUME_TOKEN_KEY = 'realtime:resumeToken'; @@ -35,14 +32,19 @@ export type WatchFactory = (options: { resumeAfter?: unknown }) => ChangeStreamL export type CoordinatorOptions = { grpcSdk: ConduitGrpcSdk; watch: WatchFactory; - hello: () => Promise<{ setName?: string; msg?: string } | null>; + checkTopology: () => Promise; getOptedInSchemas: () => OptedInSchema[]; subscriptions: RealtimeSubscriptionTracker; enabled: () => boolean; - engine: () => string; + parseResumeToken?: (token: string | null | undefined) => unknown | undefined; + prepare?: () => Promise; + onResumePersisted?: (resumeToken: string) => Promise; + persistResume?: boolean; + leaderLock?: string; + resumeTokenKey?: string; }; -export class MongoChangeStreamCoordinator { +export class ChangeStreamCoordinator { private lock: LeaderLock | null = null; private stream: ChangeStreamLike | null = null; private renewTimer: NodeJS.Timeout | null = null; @@ -56,6 +58,7 @@ export class MongoChangeStreamCoordinator { private watching = false; private opening = false; private ignoreClose = false; + private changeQueue: Promise = Promise.resolve(); constructor(private readonly options: CoordinatorOptions) {} @@ -75,23 +78,28 @@ export class MongoChangeStreamCoordinator { return this.topology; } + async waitForIdle(): Promise { + await this.changeQueue; + } + async reconcile(): Promise { if (this.closed) return; - const engine = this.options.engine(); - if (engine !== 'MongoDB' || !this.options.enabled()) { + if (!this.options.enabled()) { + await this.safePrepare(); await this.stopStream('idle'); await this.releaseLeader(); - this.streamState = engine !== 'MongoDB' ? 'unsupported' : 'disabled'; + this.streamState = 'disabled'; return; } - this.topology = topologyFromHello(await this.options.hello().catch(() => null)); + this.topology = await this.options.checkTopology().catch(() => ({ + supported: false, + message: 'Unable to determine database topology', + })); if (!this.topology.supported) { await this.stopStream('idle'); await this.releaseLeader(); this.streamState = 'idle'; this.lastError = this.topology.message; - // Hello can fail during startup before Mongo is ready. Keep retrying - // that case; a confirmed standalone topology will not recover. if ( !this.topology.message || this.topology.message.includes('Unable to determine') @@ -100,6 +108,15 @@ export class MongoChangeStreamCoordinator { } return; } + try { + await this.options.prepare?.(); + } catch (err) { + this.lastError = err instanceof Error ? err.message : String(err); + this.streamState = 'degraded'; + ConduitGrpcSdk.Logger.error(err as Error); + this.scheduleRetry(); + return; + } if (this.options.getOptedInSchemas().length === 0) { await this.stopStream('idle'); await this.releaseLeader(); @@ -112,10 +129,31 @@ export class MongoChangeStreamCoordinator { async shutdown(): Promise { this.closed = true; this.clearTimers(); + await this.changeQueue; await this.stopStream('idle'); await this.releaseLeader(); } + private get leaderLockName(): string { + return this.options.leaderLock ?? LEADER_LOCK; + } + + private get resumeTokenName(): string { + return this.options.resumeTokenKey ?? RESUME_TOKEN_KEY; + } + + private get persistResume(): boolean { + return this.options.persistResume !== false; + } + + private async safePrepare(): Promise { + try { + await this.options.prepare?.(); + } catch (err) { + ConduitGrpcSdk.Logger.error(err as Error); + } + } + private async ensureLeader(): Promise { if (this.lock) { if (!this.watching) { @@ -125,13 +163,10 @@ export class MongoChangeStreamCoordinator { } try { const acquired = await this.options.grpcSdk.state!.tryAcquireLock( - LEADER_LOCK, + this.leaderLockName, LOCK_TTL_MS, ); if (!acquired) { - // Another instance holds the lock, or a crashed holder has not - // expired yet. Without a retry, a standalone process stays idle - // forever after a restart races the previous TTL. this.streamState = 'idle'; this.scheduleRetry(); return; @@ -170,17 +205,17 @@ export class MongoChangeStreamCoordinator { this.streamState = 'starting'; this.ignoreClose = false; try { - const resumeAfter = parseResumeToken( - await this.options.grpcSdk.state!.getKey(RESUME_TOKEN_KEY), - ); + 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) => { - void this.handleChange(change as RawChangeEvent); + this.enqueueChange(change as RawChangeEvent); }); stream.on('error', (err: unknown) => { void this.handleStreamError(err); @@ -191,6 +226,12 @@ export class MongoChangeStreamCoordinator { 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); @@ -199,26 +240,62 @@ export class MongoChangeStreamCoordinator { } } + private enqueueChange(change: RawChangeEvent) { + this.changeQueue = this.changeQueue.then(async () => { + if (this.closed || !this.watching) return; + try { + await this.handleChange(change); + } catch (err) { + this.lastError = err instanceof Error ? err.message : String(err); + ConduitGrpcSdk.Logger.error(err as Error); + this.watching = false; + await this.stopStream('degraded'); + this.scheduleRetry(); + } + }); + } + private async handleChange(change: RawChangeEvent) { + const token = resumeTokenOf(change); const schema = this.resolveSchema(change.ns?.coll); - if (!schema) return; - const event = normalizeChangeEvent(change, schema.name); - if (!event) return; + const event = schema ? normalizeChangeEvent(change, schema.name) : null; + if (!event || !schema) { + if (this.persistResume && token) { + await this.persistResumeToken(token); + } + return; + } this.lastEventAt = event.occurredAt; this.lastError = undefined; - await this.options.grpcSdk.state!.setKey(RESUME_TOKEN_KEY, event.resumeToken); - this.options.grpcSdk.bus?.publish( - `database:change:${schema.name}`, - JSON.stringify(event), - ); + await this.emitChange(schema, event); + if (this.persistResume) { + await this.persistResumeToken(event.resumeToken); + } + } + + private async persistResumeToken(token: string) { + await this.options.grpcSdk.state!.setKey(this.resumeTokenName, token); + try { + await this.options.onResumePersisted?.(token); + } catch (err) { + ConduitGrpcSdk.Logger.error(err as Error); + } + } + + private async emitChange(schema: OptedInSchema, event: DatabaseChangeEvent) { + const payload = JSON.stringify(event); + this.options.grpcSdk.bus?.publish(`database:change:${schema.name}`, payload); ConduitGrpcSdk.Metrics?.increment('database_realtime_events_total', 1, { operation: event.operation, }); - await this.pushEvent(schema, event); + await this.pushEvent(schema, event, payload); } - private async pushEvent(schema: OptedInSchema, event: DatabaseChangeEvent) { - const payload = JSON.stringify(event); + private async pushEvent( + schema: OptedInSchema, + event: DatabaseChangeEvent, + payload: string, + ) { const adminRooms = roomsForPublicChange(schema.name, event.documentId); await this.safePush('admin', adminRooms, payload); if (!schema.authorizationEnabled) { @@ -231,21 +308,23 @@ export class MongoChangeStreamCoordinator { ); const allowedRooms: string[] = []; for (const userId of userIds) { - const allowed = await canReadDocument( + const decision = await documentReadDecision( this.options.grpcSdk as unknown as AuthorizationSdk, schema.name, event.documentId, userId, ); - if (!allowed) { + if (decision === 'allow') { + allowedRooms.push(authorizedDocumentRoom(schema.name, event.documentId, userId)); + continue; + } + if (decision === 'deny') { await this.options.subscriptions.removeUser( schema.name, event.documentId, userId, ); - continue; } - allowedRooms.push(authorizedDocumentRoom(schema.name, event.documentId, userId)); } if (allowedRooms.length > 0) { await this.safePush('router', allowedRooms, payload); @@ -260,16 +339,12 @@ export class MongoChangeStreamCoordinator { const client = target === 'admin' ? this.options.grpcSdk.admin : this.options.grpcSdk.router; if (!client?.socketPush) return; - try { - await client.socketPush({ - event: 'change', - data, - rooms, - receivers: [], - }); - } catch (err) { - ConduitGrpcSdk.Logger.error(err as Error); - } + await client.socketPush({ + event: 'change', + data, + rooms, + receivers: [], + }); } private resolveSchema(collectionName?: string): OptedInSchema | undefined { @@ -285,8 +360,8 @@ export class MongoChangeStreamCoordinator { this.streamState = 'degraded'; ConduitGrpcSdk.Metrics?.increment('database_realtime_stream_errors_total'); ConduitGrpcSdk.Logger.error(err as Error); - if (isResumeTokenUnusable(err)) { - await this.options.grpcSdk.state!.clearKey(RESUME_TOKEN_KEY); + if (this.persistResume && isResumeTokenUnusable(err)) { + await this.options.grpcSdk.state!.clearKey(this.resumeTokenName); } await this.stopStream('degraded'); this.scheduleRetry(); @@ -343,3 +418,10 @@ export class MongoChangeStreamCoordinator { } } } + +function resumeTokenOf(change: RawChangeEvent): string | undefined { + if (change._id === undefined || change._id === null) { + return undefined; + } + return EJSON.stringify(change._id); +} diff --git a/modules/database/src/realtime/RealtimeService.ts b/modules/database/src/realtime/RealtimeService.ts index e17e320ea..005d8288c 100644 --- a/modules/database/src/realtime/RealtimeService.ts +++ b/modules/database/src/realtime/RealtimeService.ts @@ -11,18 +11,23 @@ import { } from '@conduitplatform/module-tools'; import { DatabaseAdapter } from '../adapters/DatabaseAdapter.js'; import { MongooseAdapter } from '../adapters/mongoose-adapter/index.js'; +import { SequelizeAdapter } from '../adapters/sequelize-adapter/index.js'; import { MongooseSchema } from '../adapters/mongoose-adapter/MongooseSchema.js'; import { SequelizeSchema } from '../adapters/sequelize-adapter/SequelizeSchema.js'; import { toOptedInSchema } from './authorize.js'; -import { MongoChangeStreamCoordinator } from './MongoChangeStreamCoordinator.js'; +import { ChangeStreamCoordinator } from './ChangeStreamCoordinator.js'; import { registerDatabaseRealtimeSocket } from './sockets.js'; import { buildRealtimeStatus } from './status.js'; 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 { SQL_LEADER_LOCK } from './sql/constants.js'; export class RealtimeService { private readonly subscriptions: RealtimeSubscriptionTracker; - private coordinator?: MongoChangeStreamCoordinator; + private coordinator?: ChangeStreamCoordinator; + private sqlSupport?: SqlRealtimeSupport; constructor( private readonly grpcSdk: ConduitGrpcSdk, @@ -38,14 +43,29 @@ export class RealtimeService { void this.reconcile(); }); if (adapter instanceof MongooseAdapter) { - this.coordinator = new MongoChangeStreamCoordinator({ + this.coordinator = new ChangeStreamCoordinator({ grpcSdk, watch: options => this.openWatch(adapter, options.resumeAfter), - hello: () => this.hello(adapter), + checkTopology: () => this.checkMongoTopology(adapter), getOptedInSchemas: () => this.getOptedInSchemas(), subscriptions: this.subscriptions, enabled: () => this.isGloballyEnabled(), - engine: () => adapter.getDatabaseType(), + }); + } else if (adapter instanceof SequelizeAdapter) { + this.sqlSupport = new SqlRealtimeSupport(adapter); + this.coordinator = new ChangeStreamCoordinator({ + grpcSdk, + watch: () => this.sqlSupport!.openWatch(), + checkTopology: () => this.sqlSupport!.checkTopology(), + getOptedInSchemas: () => this.getOptedInSchemas(), + subscriptions: this.subscriptions, + enabled: () => this.isGloballyEnabled(), + persistResume: false, + leaderLock: SQL_LEADER_LOCK, + prepare: () => + this.sqlSupport!.prepare( + this.isGloballyEnabled() ? this.getOptedInSchemas() : [], + ), }); } } @@ -103,8 +123,7 @@ export class RealtimeService { topologySupported: this.coordinator?.getTopology().supported ?? false, topologyMessage: this.coordinator?.getTopology().message, activeSchemaCount: optedIn.length, - streamState: - this.coordinator?.getState() ?? (engine === 'MongoDB' ? 'idle' : 'unsupported'), + streamState: this.coordinator?.getState() ?? 'unsupported', lastEventAt: this.coordinator?.getLastEventAt(), lastError: this.coordinator?.getLastError(), socketsEnabled: await this.areAdminSocketsEnabled(), @@ -121,6 +140,7 @@ export class RealtimeService { const optedIn = toOptedInSchema({ name: schema.name, collectionName: schema.collectionName, + documentIdField: this.documentIdField(schema.name), modelOptions: schema.modelOptions, }); if (optedIn) schemas.push(optedIn); @@ -128,6 +148,14 @@ export class RealtimeService { return schemas; } + private documentIdField(schemaName: string): string | undefined { + const model = this.adapter.models[schemaName]; + if (model && 'idField' in model && typeof model.idField === 'string') { + return model.idField; + } + return undefined; + } + private openWatch(adapter: MongooseAdapter, resumeAfter?: unknown): ChangeStreamLike { const db = adapter.mongoose.connection.db; if (!db) { @@ -139,6 +167,10 @@ export class RealtimeService { ) as unknown as ChangeStreamLike; } + private async checkMongoTopology(adapter: MongooseAdapter) { + return topologyFromHello(await this.hello(adapter).catch(() => null)); + } + private async hello( adapter: MongooseAdapter, ): Promise<{ setName?: string; msg?: string } | null> { diff --git a/modules/database/src/realtime/__tests__/authorize.test.ts b/modules/database/src/realtime/__tests__/authorize.test.ts index 9c5eba15b..36f32563f 100644 --- a/modules/database/src/realtime/__tests__/authorize.test.ts +++ b/modules/database/src/realtime/__tests__/authorize.test.ts @@ -3,6 +3,7 @@ import { status } from '@grpc/grpc-js'; import { GrpcError } from '@conduitplatform/grpc-sdk'; import { assertSchemaAvailable, + documentReadDecision, optionalDocumentId, parseSubscribeRequest, requireSchemaName, @@ -66,4 +67,36 @@ describe('realtime authorization helpers', () => { }), ); }); + + it('treats missing authorization and can() failures as unavailable', async () => { + expect( + await documentReadDecision({ isAvailable: () => false }, 'Order', '1', 'user-1'), + ).toBe('unavailable'); + expect( + await documentReadDecision( + { + isAvailable: () => true, + authorization: { + can: async () => { + throw new Error('down'); + }, + }, + }, + 'Order', + '1', + 'user-1', + ), + ).toBe('unavailable'); + expect( + await documentReadDecision( + { + isAvailable: () => true, + authorization: { can: async () => ({ allow: false }) }, + }, + 'Order', + '1', + 'user-1', + ), + ).toBe('deny'); + }); }); diff --git a/modules/database/src/realtime/__tests__/coordinator.test.ts b/modules/database/src/realtime/__tests__/coordinator.test.ts index efddfca70..3bd823034 100644 --- a/modules/database/src/realtime/__tests__/coordinator.test.ts +++ b/modules/database/src/realtime/__tests__/coordinator.test.ts @@ -1,9 +1,10 @@ import { EventEmitter } from 'node:events'; import { describe, expect, it, jest } from '@jest/globals'; -import { ObjectId } from 'bson'; -import { MongoChangeStreamCoordinator } from '../MongoChangeStreamCoordinator.js'; +import { EJSON, ObjectId } from 'bson'; +import { ChangeStreamCoordinator } from '../ChangeStreamCoordinator.js'; import { RealtimeSubscriptionTracker } from '../subscriptions.js'; import { roomsForPublicChange } from '../rooms.js'; +import { SQL_LEADER_LOCK } from '../sql/constants.js'; class MemoryStore { private sets = new Map>(); @@ -11,9 +12,13 @@ class MemoryStore { const set = this.sets.get(key) ?? new Set(); members.forEach(member => set.add(member)); this.sets.set(key, set); + return members.length; } async srem(key: string, ...members: string[]) { - members.forEach(member => this.sets.get(key)?.delete(member)); + const set = this.sets.get(key); + if (!set) return 0; + members.forEach(member => set.delete(member)); + return members.length; } async smembers(key: string) { return [...(this.sets.get(key) ?? new Set())]; @@ -23,25 +28,46 @@ class MemoryStore { } async del(...keys: string[]) { keys.forEach(key => this.sets.delete(key)); + return keys.length; } } function createCoordinator(overrides?: { allow?: boolean; - schemas?: { name: string; collectionName: string; authorizationEnabled: boolean }[]; + authorizationAvailable?: boolean; + canThrows?: boolean; + schemas?: { + name: string; + collectionName: string; + authorizationEnabled: boolean; + documentIdField?: string; + }[]; getKeyDelayMs?: number; + onResumePersisted?: (token: string) => Promise; + 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), release: jest.fn(async () => undefined), }; const routerPush = jest.fn(async () => undefined); - const adminPush = jest.fn(async () => undefined); + const adminPush = jest.fn(overrides?.adminPush ?? (async () => undefined)); const publish = jest.fn(); const watch = jest.fn(() => stream as never); const subscriptions = new RealtimeSubscriptionTracker(new MemoryStore()); @@ -65,22 +91,31 @@ function createCoordinator(overrides?: { bus: { publish }, router: { socketPush: routerPush }, admin: { socketPush: adminPush }, - isAvailable: () => true, + isAvailable: () => overrides?.authorizationAvailable !== false, authorization: { - can: async () => ({ allow: overrides?.allow !== false }), + can: async () => { + if (overrides?.canThrows) { + throw new Error('authorization unavailable'); + } + return { allow: overrides?.allow !== false }; + }, }, }; - const coordinator = new MongoChangeStreamCoordinator({ + const coordinator = new ChangeStreamCoordinator({ grpcSdk: grpcSdk as never, watch, - hello: async () => ({ setName: 'rs0' }), + checkTopology: async () => ({ supported: true }), getOptedInSchemas: () => overrides?.schemas ?? [ { name: 'Order', collectionName: 'orders', authorizationEnabled: false }, ], subscriptions, enabled: () => true, - engine: () => 'MongoDB', + onResumePersisted: overrides?.onResumePersisted, + parseResumeToken: overrides?.parseResumeToken, + persistResume: overrides?.persistResume, + leaderLock: overrides?.leaderLock, + resumeTokenKey: overrides?.resumeTokenKey, }); return { coordinator, @@ -96,7 +131,7 @@ function createCoordinator(overrides?: { }; } -describe('MongoChangeStreamCoordinator', () => { +describe('ChangeStreamCoordinator', () => { it('emits one normalized event to public rooms and ignores other collections', async () => { const { coordinator, stream, routerPush, adminPush, publish } = createCoordinator(); await coordinator.reconcile(); @@ -115,7 +150,7 @@ describe('MongoChangeStreamCoordinator', () => { documentKey: { _id: new ObjectId('64b64c4c4c4c4c4c4c4c4c4d') }, _id: resume, }); - await new Promise(resolve => setImmediate(resolve)); + await coordinator.waitForIdle(); expect(publish).toHaveBeenCalledTimes(1); expect(publish.mock.calls[0][0]).toBe('database:change:Order'); const payload = JSON.parse(publish.mock.calls[0][1] as string); @@ -136,6 +171,86 @@ describe('MongoChangeStreamCoordinator', () => { await coordinator.shutdown(); }); + it('persists skipped collection tokens after the opted-in emit', async () => { + const order: string[] = []; + const { coordinator, stream, grpcSdk } = createCoordinator({ + onResumePersisted: async () => { + order.push('trim'); + }, + }); + grpcSdk.state.setKey.mockImplementation(async (key: string, value: string) => { + order.push(`setKey:${value}`); + }); + await coordinator.reconcile(); + stream.emit('change', { + operationType: 'insert', + ns: { coll: 'orders' }, + documentKey: { _id: new ObjectId('64b64c4c4c4c4c4c4c4c4c4c') }, + _id: { _data: 'token-a' }, + wallTime: new Date('2026-01-02T00:00:00.000Z'), + }); + stream.emit('change', { + operationType: 'insert', + ns: { coll: 'other' }, + documentKey: { _id: new ObjectId('64b64c4c4c4c4c4c4c4c4c4d') }, + _id: { _data: 'token-b' }, + }); + await coordinator.waitForIdle(); + expect(order).toEqual([ + `setKey:${EJSON.stringify({ _data: 'token-a' })}`, + 'trim', + `setKey:${EJSON.stringify({ _data: 'token-b' })}`, + 'trim', + ]); + await coordinator.shutdown(); + }); + + it('persists resume and trims only after a successful emit', async () => { + const order: string[] = []; + const { coordinator, stream, grpcSdk, publish } = createCoordinator({ + onResumePersisted: async () => { + order.push('trim'); + }, + }); + publish.mockImplementation(() => { + order.push('publish'); + }); + grpcSdk.state.setKey.mockImplementation(async () => { + order.push('setKey'); + }); + await coordinator.reconcile(); + stream.emit('change', { + operationType: 'update', + ns: { coll: 'orders' }, + documentKey: { _id: 'order-1' }, + _id: '1842', + wallTime: new Date('2026-03-01T00:00:00.000Z'), + }); + await coordinator.waitForIdle(); + expect(order).toEqual(['publish', 'setKey', 'trim']); + await coordinator.shutdown(); + }); + + it('does not persist a later token when emit fails', async () => { + const { coordinator, stream, grpcSdk, adminPush } = createCoordinator({ + adminPush: async () => { + throw new Error('socket down'); + }, + }); + await coordinator.reconcile(); + stream.emit('change', { + operationType: 'insert', + ns: { coll: 'orders' }, + documentKey: { _id: 'order-1' }, + _id: '10', + wallTime: new Date('2026-03-01T00:00:00.000Z'), + }); + await coordinator.waitForIdle(); + expect(grpcSdk.state.setKey).not.toHaveBeenCalled(); + expect(coordinator.getState()).toBe('degraded'); + await coordinator.shutdown(); + }); + it('re-checks ReBAC before emission and drops revoked users', async () => { const { coordinator, stream, routerPush, subscriptions } = createCoordinator({ allow: false, @@ -154,12 +269,37 @@ describe('MongoChangeStreamCoordinator', () => { documentKey: { _id: new ObjectId('64b64c4c4c4c4c4c4c4c4c4c') }, _id: { _data: 'token' }, }); - await new Promise(resolve => setImmediate(resolve)); + await coordinator.waitForIdle(); expect(routerPush).not.toHaveBeenCalled(); expect(await subscriptions.listUsers('Order', '64b64c4c4c4c4c4c4c4c4c')).toEqual([]); await coordinator.shutdown(); }); + it('does not remove users when authorization is unavailable', async () => { + const { coordinator, stream, routerPush, subscriptions, grpcSdk } = createCoordinator( + { + authorizationAvailable: false, + schemas: [ + { name: 'Order', collectionName: 'orders', authorizationEnabled: true }, + ], + }, + ); + expect(grpcSdk.isAvailable('authorization')).toBe(false); + await subscriptions.addAuthorizedDocument('sock-1', 'Order', 'doc-1', 'user-1'); + expect(await subscriptions.listUsers('Order', 'doc-1')).toEqual(['user-1']); + await coordinator.reconcile(); + stream.emit('change', { + operationType: 'update', + ns: { coll: 'orders' }, + documentKey: { _id: 'doc-1' }, + _id: { _data: 'token' }, + }); + await coordinator.waitForIdle(); + expect(routerPush).not.toHaveBeenCalled(); + expect(await subscriptions.listUsers('Order', 'doc-1')).toEqual(['user-1']); + await coordinator.shutdown(); + }); + it('retries later when the leader lock is held by another instance', async () => { jest.useFakeTimers(); const { coordinator, grpcSdk, lock } = createCoordinator(); @@ -187,4 +327,104 @@ describe('MongoChangeStreamCoordinator', () => { expect(grpcSdk.state.clearKey).toHaveBeenCalled(); await coordinator.shutdown(); }); + + 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: '0/16B3748:12:1', + wallTime: new Date('2026-03-01T00:00:00.000Z'), + fullDocument: { secret: 'nope' }, + }); + await coordinator.waitForIdle(); + expect(publish).toHaveBeenCalledTimes(1); + const payload = JSON.parse(publish.mock.calls[0][1] as string); + expect(payload).toMatchObject({ + operation: 'update', + schema: 'Order', + documentId: 'order-1', + }); + expect(payload).not.toHaveProperty('fullDocument'); + expect(JSON.stringify(payload)).not.toContain('nope'); + await coordinator.shutdown(); + }); + + it('opens a SQL watch without resume catch-up', async () => { + const { coordinator, watch, grpcSdk } = createCoordinator({ + persistResume: false, + leaderLock: SQL_LEADER_LOCK, + }); + 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 new file mode 100644 index 000000000..663a65d90 --- /dev/null +++ b/modules/database/src/realtime/__tests__/sql-builders.test.ts @@ -0,0 +1,260 @@ +import { describe, expect, it } from '@jest/globals'; +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('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('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`, + ); + }); +}); + +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('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('round-trips LSN formatting', () => { + expect(formatLsn(parseLsn('0/16B3748'))).toBe('0/016B3748'); + }); + + 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'); + }); + + 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('WAL event mapping', () => { + it('maps metadata-only change events and uses a custom PK', () => { + const raw = toRawChangeEvent({ + operation: 'insert', + 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: '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 new file mode 100644 index 000000000..8d6868fa9 --- /dev/null +++ b/modules/database/src/realtime/__tests__/sql-change-stream.integration.test.ts @@ -0,0 +1,85 @@ +import { EventEmitter } from 'node:events'; +import { describe, expect, it } from '@jest/globals'; +import { normalizeChangeEvent } from '../normalize.js'; +import { SqlChangeStream } from '../sql/SqlChangeStream.js'; +import type { ReplicationChange, ReplicationFeed } from '../sql/replication.js'; + +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, + }); + 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'), + }); + 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(); + }); +}); + +const logicalUri = process.env.SQL_LOGICAL_URI; +const describeLivePgoutput = logicalUri ? describe : describe.skip; + +describeLivePgoutput( + 'SqlChangeStream live pgoutput (set SQL_LOGICAL_URI; skipped in CI)', + () => { + it('requires a Postgres URI with wal_level=logical', () => { + expect(logicalUri).toMatch(/^postgres/); + }); + }, +); + +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`), + }; +} + +class FakeFeed implements ReplicationFeed { + readonly emitter = new EventEmitter(); + started = 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 { + return; + } + + push(change: ReplicationChange): void { + this.emitter.emit('change', change); + } +} diff --git a/modules/database/src/realtime/__tests__/sql-realtime-support.test.ts b/modules/database/src/realtime/__tests__/sql-realtime-support.test.ts new file mode 100644 index 000000000..28ca233f6 --- /dev/null +++ b/modules/database/src/realtime/__tests__/sql-realtime-support.test.ts @@ -0,0 +1,298 @@ +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 { 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('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: `${dialect}://localhost/db`, + } as never); + 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 { + connect.mockRestore(); + query.mockRestore(); + } + }); + + 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('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 []; + }, + }; + const support = new SqlRealtimeSupport({ + sequelize, + connectionUri: 'postgres://localhost/db', + } as never); + await support.prepare([ + { + name: 'Order', + collectionName: 'orders', + authorizationEnabled: false, + documentIdField: 'sku', + }, + ]); + 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('_cnd_DatabaseChange') && sql.includes('CREATE TABLE'), + ), + ).toBe(false); + expect(queries.some(sql => sql.includes('CREATE TRIGGER'))).toBe(false); + expect(queries.some(sql => sql.includes('LISTEN'))).toBe(false); + }); +}); + +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([ + { + operationType: 'insert', + ns: { coll: 'orders' }, + documentKey: { _id: 'sku-1' }, + wallTime: new Date('2026-03-01T00:00:00.000Z'), + _id: '0/1:1:1', + }, + ]); + expect(JSON.stringify(received)).not.toContain('hidden'); + await stream.close(); + expect(feed.stopped).toBe(true); + }); + + 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(); + }); + + it('emits error when the replication feed cannot create a slot', async () => { + const stream = new SqlChangeStream({ + connectionUri: 'postgres://localhost/db', + 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 { + 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 sequelize.close(); + } + }); +}); + +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; + } +} 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 a94790cfb..2b95d4e1c 100644 --- a/modules/database/src/realtime/__tests__/status.test.ts +++ b/modules/database/src/realtime/__tests__/status.test.ts @@ -3,6 +3,15 @@ import { buildRealtimeStatus } from '../status.js'; describe('buildRealtimeStatus', () => { it('reports unsupported, disabled, idle, live, and degraded states', () => { + expect( + buildRealtimeStatus({ + engine: 'oracle', + enabled: true, + topologySupported: true, + activeSchemaCount: 1, + streamState: 'live', + }).status, + ).toBe('unsupported'); expect( buildRealtimeStatus({ engine: 'PostgreSQL', @@ -11,6 +20,27 @@ describe('buildRealtimeStatus', () => { activeSchemaCount: 1, streamState: 'live', }).status, + ).toBe('live'); + expect( + buildRealtimeStatus({ + engine: 'mysql', + 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('unsupported'); expect( buildRealtimeStatus({ @@ -30,6 +60,35 @@ describe('buildRealtimeStatus', () => { streamState: 'idle', }).message, ).toMatch(/replica set/i); + expect( + buildRealtimeStatus({ + engine: 'PostgreSQL', + enabled: true, + topologySupported: false, + topologyMessage: + '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(/logical replication/i); + expect( + buildRealtimeStatus({ + engine: 'PostgreSQL', + enabled: true, + topologySupported: false, + activeSchemaCount: 1, + streamState: 'idle', + }).message, + ).not.toMatch(/change queue|LISTEN|trigger/i); expect( buildRealtimeStatus({ engine: 'MongoDB', diff --git a/modules/database/src/realtime/authorize.ts b/modules/database/src/realtime/authorize.ts index cf7ed899c..2d9b3e410 100644 --- a/modules/database/src/realtime/authorize.ts +++ b/modules/database/src/realtime/authorize.ts @@ -61,27 +61,39 @@ export function optionalDocumentId(value: unknown): string | undefined { return value; } -export async function canReadDocument( +export type DocumentReadDecision = 'allow' | 'deny' | 'unavailable'; + +export async function documentReadDecision( grpcSdk: AuthorizationSdk, schema: string, documentId: string, userId: string, -): Promise { - if (!grpcSdk.authorization || !grpcSdk.isAvailable('authorization')) { - return false; +): Promise { + const auth = grpcSdk.authorization; + if (!auth || !grpcSdk.isAvailable('authorization')) { + return 'unavailable'; } try { - const decision = await grpcSdk.authorization.can({ + const decision = await auth.can({ subject: `User:${userId}`, actions: ['read'], resource: `${schema}:${documentId}`, }); - return decision.allow === true; + return decision.allow === true ? 'allow' : 'deny'; } catch { - return false; + return 'unavailable'; } } +export async function canReadDocument( + grpcSdk: AuthorizationSdk, + schema: string, + documentId: string, + userId: string, +): Promise { + return (await documentReadDecision(grpcSdk, schema, documentId, userId)) === 'allow'; +} + export function assertSchemaAvailable( lookup: SchemaLookup, schemaName: string, @@ -120,6 +132,7 @@ export function assertSchemaAvailable( export function toOptedInSchema(schema: { name: string; collectionName: string; + documentIdField?: string; modelOptions?: { conduit?: { realtime?: { enabled?: boolean }; @@ -132,5 +145,6 @@ export function toOptedInSchema(schema: { name: schema.name, collectionName: schema.collectionName, authorizationEnabled: schema.modelOptions.conduit.authorization?.enabled === true, + documentIdField: schema.documentIdField, }; } diff --git a/modules/database/src/realtime/index.ts b/modules/database/src/realtime/index.ts index 655c04bba..dab82831a 100644 --- a/modules/database/src/realtime/index.ts +++ b/modules/database/src/realtime/index.ts @@ -1,4 +1,5 @@ export { RealtimeService } from './RealtimeService.js'; +export { ChangeStreamCoordinator } from './ChangeStreamCoordinator.js'; export { buildRealtimeStatus } from './status.js'; export { normalizeChangeEvent } from './normalize.js'; export { diff --git a/modules/database/src/realtime/sql/SqlChangeStream.ts b/modules/database/src/realtime/sql/SqlChangeStream.ts new file mode 100644 index 000000000..d4ad66933 --- /dev/null +++ b/modules/database/src/realtime/sql/SqlChangeStream.ts @@ -0,0 +1,87 @@ +import { EventEmitter } from 'node:events'; +import type { ChangeStreamLike } from '../types.js'; +import { DEFAULT_ID_FIELD, PUBLICATION_NAME } from './constants.js'; +import { documentIdFromChange, toRawChangeEvent } from './mapEvent.js'; +import { + createPgoutputFeed, + type ReplicationChange, + type ReplicationFeed, + type ReplicationFeedFactory, +} from './replication.js'; + +export type SqlChangeStreamOptions = { + connectionUri: string; + publicationName?: string; + idFieldByTable?: Record; + createFeed?: ReplicationFeedFactory; +}; + +export class SqlChangeStream implements ChangeStreamLike { + readonly ready: Promise; + private readonly emitter = new EventEmitter(); + private readonly feed: ReplicationFeed; + private readonly idFieldByTable: Record; + private closed = false; + private started = false; + + constructor(options: SqlChangeStreamOptions) { + 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( + event: 'change' | 'error' | 'close' | 'end', + listener: (...args: unknown[]) => void, + ): void { + this.emitter.on(event, listener); + } + + async close(): Promise { + if (this.closed) return; + this.closed = true; + try { + await this.feed.stop(); + } catch { + // already closed + } + this.emitter.emit('close'); + } + + private async start(): Promise { + if (this.closed || this.started) return; + this.started = true; + try { + await this.feed.start(); + } catch (err) { + this.emitError(err); + } + } + + 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 { + if (this.closed) return; + this.emitter.emit('error', err); + } +} diff --git a/modules/database/src/realtime/sql/SqlRealtimeSupport.ts b/modules/database/src/realtime/sql/SqlRealtimeSupport.ts new file mode 100644 index 000000000..b42ea54c5 --- /dev/null +++ b/modules/database/src/realtime/sql/SqlRealtimeSupport.ts @@ -0,0 +1,103 @@ +import { QueryTypes } from 'sequelize'; +import type { SequelizeAdapter } from '../../adapters/sequelize-adapter/index.js'; +import type { ChangeStreamLike, OptedInSchema } from '../types.js'; +import type { TopologyResult } from '../topology.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 type { ReplicationFeedFactory } from './replication.js'; + +export class SqlRealtimeSupport { + private schemas: OptedInSchema[] = []; + + constructor( + private readonly adapter: SequelizeAdapter, + private readonly createFeed?: ReplicationFeedFactory, + ) {} + + async checkTopology(): Promise { + await dropLegacyCapture(this.adapter.sequelize).catch(() => undefined); + const dialect = this.adapter.sequelize.getDialect(); + if (dialect !== 'postgres') { + return { supported: false, message: SQL_ENGINE_UNSUPPORTED }; + } + try { + await this.adapter.sequelize.query('SELECT 1'); + } catch (err) { + return { + supported: false, + message: `SQL live updates cannot reach the database: ${errorMessage(err)}`, + }; + } + return this.probeLogicalReplication(); + } + + async prepare(schemas: OptedInSchema[]): Promise { + this.schemas = schemas; + await dropLegacyCapture(this.adapter.sequelize); + if (this.adapter.sequelize.getDialect() !== 'postgres') { + return; + } + await syncPublication(this.adapter.sequelize, schemas, { + schemaName: sqlSchemaName(), + publicationName: PUBLICATION_NAME, + }); + } + + openWatch(): ChangeStreamLike { + return new SqlChangeStream({ + connectionUri: this.adapter.connectionUri, + publicationName: PUBLICATION_NAME, + idFieldByTable: Object.fromEntries( + this.schemas.map(schema => [ + schema.collectionName, + schema.documentIdField ?? DEFAULT_ID_FIELD, + ]), + ), + createFeed: this.createFeed, + }); + } + + 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 }; + } +} + +function errorMessage(err: unknown): string { + return err instanceof Error ? err.message : String(err); +} diff --git a/modules/database/src/realtime/sql/constants.ts b/modules/database/src/realtime/sql/constants.ts new file mode 100644 index 000000000..5fca15e80 --- /dev/null +++ b/modules/database/src/realtime/sql/constants.ts @@ -0,0 +1,26 @@ +export const PUBLICATION_NAME = 'cnd_realtime'; +export const SQL_LEADER_LOCK = 'realtime:sql:change-stream:leader'; +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 sqlSchemaName(): string { + return process.env.SQL_SCHEMA ?? DEFAULT_SQL_SCHEMA; +} diff --git a/modules/database/src/realtime/sql/identifiers.ts b/modules/database/src/realtime/sql/identifiers.ts new file mode 100644 index 000000000..312b06ecf --- /dev/null +++ b/modules/database/src/realtime/sql/identifiers.ts @@ -0,0 +1,15 @@ +export function quoteIdent(name: string): string { + return `"${name.replace(/"/g, '""')}"`; +} + +export function quoteQualified(schema: string, table: string): string { + return `${quoteIdent(schema)}.${quoteIdent(table)}`; +} + +export function quoteLiteral(value: string): string { + return `'${value.replace(/'/g, "''")}'`; +} + +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 new file mode 100644 index 000000000..8812a322f --- /dev/null +++ b/modules/database/src/realtime/sql/index.ts @@ -0,0 +1,4 @@ +export { PUBLICATION_NAME, SQL_LEADER_LOCK } from './constants.js'; +export { SqlChangeStream } from './SqlChangeStream.js'; +export { SqlRealtimeSupport } from './SqlRealtimeSupport.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 new file mode 100644 index 000000000..0dfae6b51 --- /dev/null +++ b/modules/database/src/realtime/sql/mapEvent.ts @@ -0,0 +1,42 @@ +import type { RawChangeEvent } from '../normalize.js'; +import { DEFAULT_ID_FIELD } from './constants.js'; + +export type ChangeRows = { + tag: 'insert' | 'update' | 'delete'; + newRow?: Record; + oldRow?: Record; + keyRow?: Record; +}; + +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: 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/status.ts b/modules/database/src/realtime/status.ts index b1da21b03..0f271e604 100644 --- a/modules/database/src/realtime/status.ts +++ b/modules/database/src/realtime/status.ts @@ -1,4 +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']); export type RealtimeStatusInput = { engine: string; @@ -13,12 +16,12 @@ export type RealtimeStatusInput = { }; export function buildRealtimeStatus(input: RealtimeStatusInput): RealtimeStatus { - if (input.engine !== 'MongoDB') { + if (!SUPPORTED_REALTIME_ENGINES.has(input.engine)) { return { status: 'unsupported', engine: input.engine, activeSchemaCount: 0, - message: 'Live updates require MongoDB', + message: 'Live updates are not supported for this database engine', }; } if (!input.enabled) { @@ -35,7 +38,9 @@ export function buildRealtimeStatus(input: RealtimeStatusInput): RealtimeStatus activeSchemaCount: input.activeSchemaCount, message: input.topologyMessage ?? - 'A replica set or sharded MongoDB deployment is required for live updates', + (input.engine === 'MongoDB' + ? 'A replica set or sharded MongoDB deployment is required for 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 9901c2d86..9ca9b5767 100644 --- a/modules/database/src/realtime/types.ts +++ b/modules/database/src/realtime/types.ts @@ -33,6 +33,7 @@ export type OptedInSchema = { name: string; collectionName: string; authorizationEnabled: boolean; + documentIdField?: string; }; export type SubscribeRequest = { @@ -46,4 +47,5 @@ export type ChangeStreamLike = { listener: (...args: unknown[]) => void, ): void; close(): Promise | void; + ready?: Promise; }; diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index cbd5f6ae6..3f414b6ad 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -427,7 +427,7 @@ importers: version: 6.1.3 ts-jest: specifier: ^29.4.12 - version: 29.4.12(@babel/core@7.29.7)(@jest/transform@30.5.1)(@jest/types@30.5.1)(babel-jest@30.5.1(@babel/core@7.29.7))(jest-util@30.5.1)(jest@30.5.1(@types/node@24.13.3)(ts-node@10.9.2(@types/node@24.13.3)(typescript@6.0.3)))(typescript@6.0.3) + version: 29.4.12(@babel/core@7.29.7)(@jest/transform@30.5.1)(@jest/types@30.5.1)(babel-jest@30.5.1(@babel/core@7.29.7))(esbuild@0.28.1)(jest-util@30.5.1)(jest@30.5.1(@types/node@24.13.3)(ts-node@10.9.2(@types/node@24.13.3)(typescript@6.0.3)))(typescript@6.0.3) typescript: specifier: ~6.0.2 version: 6.0.3 @@ -914,6 +914,9 @@ importers: '@types/object-hash': specifier: ^3.0.6 version: 3.0.6 + '@types/pg': + specifier: ^8.23.1 + version: 8.23.1 copyfiles: specifier: ^2.4.1 version: 2.4.1 @@ -925,7 +928,7 @@ importers: version: 6.1.3 ts-jest: specifier: ^29.4.12 - version: 29.4.12(@babel/core@7.29.7)(@jest/transform@30.5.1)(@jest/types@30.5.1)(babel-jest@30.5.1(@babel/core@7.29.7))(jest-util@30.5.1)(jest@30.5.1(@types/node@24.13.3)(ts-node@10.9.2(@types/node@24.13.3)(typescript@6.0.3)))(typescript@6.0.3) + version: 29.4.12(@babel/core@7.29.7)(@jest/transform@30.5.1)(@jest/types@30.5.1)(babel-jest@30.5.1(@babel/core@7.29.7))(esbuild@0.28.1)(jest-util@30.5.1)(jest@30.5.1(@types/node@24.13.3)(ts-node@10.9.2(@types/node@24.13.3)(typescript@6.0.3)))(typescript@6.0.3) ts-proto: specifier: ^2.12.1 version: 2.12.1 @@ -1292,7 +1295,7 @@ importers: version: 6.1.3 ts-jest: specifier: ^29.4.12 - version: 29.4.12(@babel/core@7.29.7)(@jest/transform@30.5.1)(@jest/types@30.5.1)(babel-jest@30.5.1(@babel/core@7.29.7))(jest-util@30.5.1)(jest@30.5.1(@types/node@24.13.3)(ts-node@10.9.2(@types/node@24.13.3)(typescript@6.0.3)))(typescript@6.0.3) + version: 29.4.12(@babel/core@7.29.7)(@jest/transform@30.5.1)(@jest/types@30.5.1)(babel-jest@30.5.1(@babel/core@7.29.7))(esbuild@0.28.1)(jest-util@30.5.1)(jest@30.5.1(@types/node@24.13.3)(ts-node@10.9.2(@types/node@24.13.3)(typescript@6.0.3)))(typescript@6.0.3) ts-node: specifier: ^10.9.2 version: 10.9.2(@types/node@24.13.3)(typescript@6.0.3) @@ -3393,6 +3396,9 @@ packages: '@types/otp-generator@4.0.2': resolution: {integrity: sha512-9+qqWzuFb332hXPbLgjUyOXlbcaTQkmkmqQjTduvNuOmPV5fW+iLv70JsVEhdUy0DWi4kY34++HDCaWl6N0AYg==} + '@types/pg@8.23.1': + resolution: {integrity: sha512-fKVHpikPdg4GKks3JuLEhvwSyvwzF23hnabPy6DD8ljVbC7+6J5dQzdv4arV6jqq57djnMgs1HKBxX4P8aBI3A==} + '@types/qs@6.15.1': resolution: {integrity: sha512-GZHUBZR9hckSUhrxmp1nG6NwdpM9fCunJwyThLW1X3AyHgd9IlHb6VANpQQqDr2o/qQp6McZ3y/IA2rVzKzSbw==} @@ -11058,6 +11064,12 @@ snapshots: '@types/otp-generator@4.0.2': {} + '@types/pg@8.23.1': + dependencies: + '@types/node': 24.13.3 + pg-protocol: 1.15.0 + pg-types: 2.2.0 + '@types/qs@6.15.1': {} '@types/range-parser@1.2.7': {} @@ -16191,7 +16203,7 @@ snapshots: ts-interface-checker@0.1.13: {} - ts-jest@29.4.12(@babel/core@7.29.7)(@jest/transform@30.5.1)(@jest/types@30.5.1)(babel-jest@30.5.1(@babel/core@7.29.7))(jest-util@30.5.1)(jest@30.5.1(@types/node@24.13.3)(ts-node@10.9.2(@types/node@24.13.3)(typescript@6.0.3)))(typescript@6.0.3): + ts-jest@29.4.12(@babel/core@7.29.7)(@jest/transform@30.5.1)(@jest/types@30.5.1)(babel-jest@30.5.1(@babel/core@7.29.7))(esbuild@0.28.1)(jest-util@30.5.1)(jest@30.5.1(@types/node@24.13.3)(ts-node@10.9.2(@types/node@24.13.3)(typescript@6.0.3)))(typescript@6.0.3): dependencies: bs-logger: 0.2.6 fast-json-stable-stringify: 2.1.0 @@ -16209,6 +16221,7 @@ snapshots: '@jest/transform': 30.5.1 '@jest/types': 30.5.1 babel-jest: 30.5.1(@babel/core@7.29.7) + esbuild: 0.28.1 jest-util: 30.5.1 ts-node@10.9.2(@types/node@24.13.3)(typescript@6.0.3):