diff --git a/docker/.env b/docker/.env index 66dad89bd..b71166ea3 100644 --- a/docker/.env +++ b/docker/.env @@ -26,7 +26,7 @@ DB_TYPE="mongodb" DB_USER="conduit" DB_PASS="pass" DB_PORT="27017" -DB_CONN_URI="mongodb://conduit:pass@conduit-mongo:27017/conduit?authSource=admin" # profile: mongodb +DB_CONN_URI="mongodb://conduit:pass@conduit-mongo:27017/conduit?authSource=admin&replicaSet=rs0" # profile: mongodb #DB_CONN_URI="postgres://conduit:pass@conduit-postgres:5432/conduit" # profile: postgres # Security diff --git a/docker/docker-compose.standalone.yml b/docker/docker-compose.standalone.yml index 8cd0b0e9b..bd77d5c47 100644 --- a/docker/docker-compose.standalone.yml +++ b/docker/docker-compose.standalone.yml @@ -15,8 +15,12 @@ services: image: 'docker.io/conduitplatform/conduit-standalone:${IMAGE_TAG}' restart: unless-stopped depends_on: - - redis - - mongodb + redis: + condition: service_started + mongodb: + condition: service_started + mongo-init-replica: + condition: service_completed_successfully ports: - '${CORE_GRPC_PORT:-55152}:55152' - '${DB_GRPC_PORT:-55160}:55160' @@ -38,7 +42,7 @@ services: ADMIN_SOCKET_PORT: '${ADMIN_SOCKET_PORT:-3031}' __DEFAULT_HOST_URL: '${ADMIN_DEFAULT_HOST_URL:-http://localhost:3030}' GRPC_KEY: '${GRPC_KEY}' - DB_CONN_URI: '${DB_CONN_URI:-mongodb://conduit:pass@conduit-mongo:27017/conduit?authSource=admin}' + DB_CONN_URI: '${DB_CONN_URI:-mongodb://conduit:pass@conduit-mongo:27017/conduit?authSource=admin&replicaSet=rs0}' networks: default: aliases: @@ -74,12 +78,37 @@ services: MONGO_INITDB_DATABASE: 'conduit' MONGO_INITDB_ROOT_USERNAME: '${DB_USER:-conduit}' MONGO_INITDB_ROOT_PASSWORD: '${DB_PASS:-pass}' + # Existing volumes created before rs0 will not elect a replica set. + # Remove the mongo volume (or start with an empty data dir) if hello stays standalone. + entrypoint: + - bash + - -c + - | + cp /mongo-keyfile /tmp/keyfile + chmod 400 /tmp/keyfile + chown mongodb:mongodb /tmp/keyfile + exec docker-entrypoint.sh mongod --replSet rs0 --bind_ip_all --keyFile /tmp/keyfile networks: default: aliases: - conduit-mongo volumes: - mongo:/data/db + - ./mongo/keyfile:/mongo-keyfile:ro + + mongo-init-replica: + container_name: 'conduit-mongo-init' + image: 'docker.io/library/mongo:4.4.15' + restart: on-failure + depends_on: + - mongodb + environment: + MONGO_HOST: 'conduit-mongo' + MONGO_INITDB_ROOT_USERNAME: '${DB_USER:-conduit}' + MONGO_INITDB_ROOT_PASSWORD: '${DB_PASS:-pass}' + volumes: + - ./mongo/init-replica.sh:/init-replica.sh:ro + command: ['bash', '/init-replica.sh'] # Persistent Volumes volumes: diff --git a/docker/docker-compose.yml b/docker/docker-compose.yml index 83b3e9b55..7567df345 100644 --- a/docker/docker-compose.yml +++ b/docker/docker-compose.yml @@ -71,10 +71,21 @@ services: image: 'docker.io/conduitplatform/database:${IMAGE_TAG}' restart: unless-stopped depends_on: - - core - - ${DB_TYPE:-mongodb} - - prometheus - - loki + core: + condition: service_started + prometheus: + condition: service_started + loki: + condition: service_started + mongodb: + condition: service_started + required: false + postgres: + condition: service_started + required: false + mongo-init-replica: + condition: service_completed_successfully + required: false ports: - '${DB_GRPC_PORT:-55160}:${DB_GRPC_PORT:-55160}' environment: @@ -85,7 +96,7 @@ services: LOKI_URL: 'http://conduit-loki:3100' GRPC_KEY: '${GRPC_KEY}' DB_TYPE: '${DB_TYPE:-mongodb}' - DB_CONN_URI: '${DB_CONN_URI:-mongodb://conduit:pass@conduit-mongo:27017/conduit?authSource=admin}' + DB_CONN_URI: '${DB_CONN_URI:-mongodb://conduit:pass@conduit-mongo:27017/conduit?authSource=admin&replicaSet=rs0}' networks: default: aliases: @@ -271,12 +282,42 @@ services: MONGO_INITDB_ROOT_USERNAME: '${DB_USER:-conduit}' MONGO_INITDB_ROOT_PASSWORD: '${DB_PASS:-pass}' profiles: ['mongodb'] + # Existing volumes created before rs0 will not elect a replica set. + # Remove the mongo volume (or start with an empty data dir) if hello stays standalone. + entrypoint: + - bash + - -c + - | + cp /mongo-keyfile /tmp/keyfile + chmod 400 /tmp/keyfile + chown mongodb:mongodb /tmp/keyfile + exec docker-entrypoint.sh mongod --replSet rs0 --bind_ip_all --keyFile /tmp/keyfile networks: default: aliases: - conduit-mongo volumes: - mongo:/data/db + - ./mongo/keyfile:/mongo-keyfile:ro + + mongo-init-replica: + container_name: 'conduit-mongo-init' + image: 'docker.io/library/mongo:4.4.15' + restart: on-failure + profiles: ['mongodb'] + depends_on: + - mongodb + environment: + MONGO_HOST: 'conduit-mongo' + MONGO_INITDB_ROOT_USERNAME: '${DB_USER:-conduit}' + MONGO_INITDB_ROOT_PASSWORD: '${DB_PASS:-pass}' + volumes: + - ./mongo/init-replica.sh:/init-replica.sh:ro + command: ['bash', '/init-replica.sh'] + networks: + default: + aliases: + - conduit-mongo-init postgres: container_name: 'conduit-postgres' diff --git a/docker/mongo/init-replica.sh b/docker/mongo/init-replica.sh new file mode 100644 index 000000000..f94f5dbf4 --- /dev/null +++ b/docker/mongo/init-replica.sh @@ -0,0 +1,34 @@ +#!/bin/bash +set -euo pipefail +HOST="${MONGO_HOST:-conduit-mongo}" +USER="${MONGO_INITDB_ROOT_USERNAME:-conduit}" +PASS="${MONGO_INITDB_ROOT_PASSWORD:-pass}" + +mongo_eval() { + mongo --host "$HOST" -u "$USER" -p "$PASS" --authenticationDatabase admin --quiet --eval "$1" +} + +until mongo_eval 'db.adminCommand({ ping: 1 })' >/dev/null 2>&1; do + sleep 2 +done + +# rs.status() returns { ok: 0 } before initiate; it does not throw. +mongo_eval ' + var status = rs.status(); + if (status.ok === 1) { + quit(0); + } + var result = rs.initiate({ + _id: "rs0", + members: [{ _id: 0, host: "'"$HOST"':27017" }] + }); + if (result.ok !== 1) { + printjson(result); + quit(1); + } +' + +# Mongoose with replicaSet=rs0 only selects a PRIMARY (myState === 1). +until mongo_eval 'var s = rs.status(); if (s.ok === 1 && s.myState === 1) { quit(0); } quit(1);' >/dev/null 2>&1; do + sleep 1 +done diff --git a/docker/mongo/keyfile b/docker/mongo/keyfile new file mode 100644 index 000000000..056a007b0 --- /dev/null +++ b/docker/mongo/keyfile @@ -0,0 +1 @@ +ConduitLocalDevMongoReplicaSetKeyFileDoNotUseInProductionReplaceBeforeAnyRealDeployment0123456789abcdefghijklmnopqrstuvwxyz diff --git a/libraries/grpc-sdk/src/interfaces/Model.ts b/libraries/grpc-sdk/src/interfaces/Model.ts index 6b7319a9b..b13fd687b 100644 --- a/libraries/grpc-sdk/src/interfaces/Model.ts +++ b/libraries/grpc-sdk/src/interfaces/Model.ts @@ -71,9 +71,7 @@ export interface ConduitArrayValidation { } export type ConduitValidationRules = - | ConduitStringValidation - | ConduitNumberValidation - | ConduitArrayValidation; + ConduitStringValidation | ConduitNumberValidation | ConduitArrayValidation; type BaseConduitModelField = { type?: TYPE | TYPE[] | ConduitModel | ArrayConduitModel[]; @@ -190,6 +188,9 @@ export interface ConduitSchemaOptions { authorization?: { enabled: boolean; }; + realtime?: { + enabled: boolean; + }; /** Mongoose read preference for this schema (ignored by SQL); per-query wins. */ readPreference?: string; }; diff --git a/libraries/grpc-sdk/src/modules/admin/index.ts b/libraries/grpc-sdk/src/modules/admin/index.ts index 889ca0ead..57a6ac7bd 100644 --- a/libraries/grpc-sdk/src/modules/admin/index.ts +++ b/libraries/grpc-sdk/src/modules/admin/index.ts @@ -3,7 +3,8 @@ import { AdminDefinition, RegisterAdminRouteRequest, RegisterAdminRouteRequest_PathDefinition, -} from '../../protoUtils/index.js'; + SocketPushRequest, +} from '../../protoUtils/core.js'; import { ConduitRouteActions } from '../../interfaces/index.js'; export class Admin extends ConduitModule { @@ -28,6 +29,10 @@ export class Admin extends ConduitModule { return this.client!.registerAdminRoute(request); } + socketPush(data: SocketPushRequest) { + return this.client!.socketPush(data); + } + patchRouteMiddlewares( path: string, action: ConduitRouteActions, diff --git a/libraries/hermes/src/Socket/Socket.ts b/libraries/hermes/src/Socket/Socket.ts index 7084fa857..f1c0b47d1 100644 --- a/libraries/hermes/src/Socket/Socket.ts +++ b/libraries/hermes/src/Socket/Socket.ts @@ -62,9 +62,17 @@ export class SocketController extends ConduitRouter { }; this.io = new IOServer(this.httpServer, this.options); this.redisClient = grpcSdk.redisManager.getClient(); + // Admin (e.g. :3031) and Router (e.g. :3001) are separate Socket.IO + // servers that share Redis. The adapter defaults would put both on the + // same stream, so a database change pushed to admin *and* router is + // delivered twice to every socket in those rooms. + const adapterKey = `socket.io:${this.port}`; this.io.adapter( createAdapter(this.redisClient, { onlyPlaintext: true, + streamName: adapterKey, + channelPrefix: adapterKey, + sessionKeyPrefix: `sio:session:${this.port}:`, }), ); this.httpServer.listen(this.port); @@ -144,11 +152,15 @@ export class SocketController extends ConduitRouter { this.io.of(namespace).on('connect', socket => { if (socket.recovered) { + const recoveredRooms = [...socket.rooms].filter( + room => room.startsWith('er:') || room.startsWith('database:'), + ); const recovered = conduitSocket.executeRecovered({ event: 'recovered', socketId: socket.id, context: socket.data, - recoveredRooms: [...socket.rooms].filter(room => room.startsWith('er:')), + params: recoveredRooms, + recoveredRooms, }); if (recovered) { recovered @@ -187,12 +199,13 @@ export class SocketController extends ConduitRouter { }); }); - socket.on('disconnect', () => { + socket.on('disconnect', (reason: string) => { conduitSocket .executeRequest({ event: 'disconnect', socketId: socket.id, context: socket.data, + params: [reason], }) .then(res => this.handleResponse(res, socket, namespace)) .catch(e => { diff --git a/libraries/hermes/src/Socket/isSocketHandshake.test.ts b/libraries/hermes/src/Socket/isSocketHandshake.test.ts index 267de330f..a686bdb76 100644 --- a/libraries/hermes/src/Socket/isSocketHandshake.test.ts +++ b/libraries/hermes/src/Socket/isSocketHandshake.test.ts @@ -16,6 +16,7 @@ describe('isSocketHandshake', () => { isSocketHandshake({ url: '/realtime/ticket?EIO=4&transport=polling' }), false, ); + assert.equal(isSocketHandshake({ url: '/realtime/ticket' }), false); assert.equal(isSocketHandshake({ originalUrl: '/realtime' }), false); assert.equal(isSocketHandshake({ url: '/graphql?EIO=4&transport=polling' }), false); assert.equal( diff --git a/libraries/hermes/src/index.ts b/libraries/hermes/src/index.ts index ac1a9960b..46808add1 100644 --- a/libraries/hermes/src/index.ts +++ b/libraries/hermes/src/index.ts @@ -47,6 +47,15 @@ export class ConduitRoutingController { private _cleanupTimeout: NodeJS.Timeout | null = null; /** Routes registered before MCP starts; replayed in initMCP. */ private readonly _conduitRoutesByKey: Map = new Map(); + /** Sockets registered before Socket.IO starts; replayed in initSockets. */ + private readonly _conduitSocketsByPath: Map = new Map(); + private readonly _socketMiddlewares: Array< + (req: ConduitRequest, res: Response, next: NextFunction) => void + > = []; + private readonly _socketRouteMiddlewares: Array<{ + middleware: ConduitMiddleware; + moduleUrl: string; + }> = []; private readonly routeTrie: RouteTrie = new RouteTrie(); readonly expressApp: Express = express(); readonly server = http.createServer(this.expressApp); @@ -143,6 +152,15 @@ export class ConduitRoutingController { this.expressApp, this.metrics, ); + for (const middleware of this._socketMiddlewares) { + this._socketRouter.registerGlobalMiddleware(middleware); + } + for (const { middleware, moduleUrl } of this._socketRouteMiddlewares) { + this._socketRouter.registerMiddleware(middleware, moduleUrl); + } + for (const socket of this._conduitSocketsByPath.values()) { + this._socketRouter.registerConduitSocket(socket); + } } initMCP(config?: { @@ -231,6 +249,7 @@ export class ConduitRoutingController { ) { this._middlewareRouter.use(middleware); if (socketMiddleware) { + this._socketMiddlewares.push(middleware); this._socketRouter?.registerGlobalMiddleware(middleware); } } @@ -244,6 +263,7 @@ export class ConduitRoutingController { registerRouteMiddleware(middleware: ConduitMiddleware, moduleUrl: string) { this._restRouter?.registerMiddleware(middleware, moduleUrl); this._graphQLRouter?.registerMiddleware(middleware, moduleUrl); + this._socketRouteMiddlewares.push({ middleware, moduleUrl }); this._socketRouter?.registerMiddleware(middleware, moduleUrl); this._mcpRouter?.registerMiddleware(middleware, moduleUrl); } @@ -292,6 +312,7 @@ export class ConduitRoutingController { } registerConduitSocket(socket: ConduitSocket) { + this._conduitSocketsByPath.set(socket.input.path, socket); this._socketRouter?.registerConduitSocket(socket); } diff --git a/libraries/hermes/src/interfaces/Socket.ts b/libraries/hermes/src/interfaces/Socket.ts index 7c7851094..2b9787257 100644 --- a/libraries/hermes/src/interfaces/Socket.ts +++ b/libraries/hermes/src/interfaces/Socket.ts @@ -91,10 +91,13 @@ export class ConduitSocket { executeRecovered( request: ConduitSocketParameters, ): ConduitSocketHandlerResponse | null { - if (!this._input.onRecovered) { - return null; + if (this._input.onRecovered) { + return this._input.onRecovered(request); } - return this._input.onRecovered(request); + if (this._events.has('recovered')) { + return this._events.get('recovered')!.handler(request); + } + return null; } } diff --git a/modules/database/README.mdx b/modules/database/README.mdx index c340b86bd..f573efb1e 100644 --- a/modules/database/README.mdx +++ b/modules/database/README.mdx @@ -47,6 +47,27 @@ since the latter need to go through parsers that are otherwise unnecessary for M When using MongoDB with a replica set (e.g., MongoDB Atlas), the database module supports configuring read preference, write concern, and read concern through the admin panel at `PATCH /config/database`. +Live document updates require a replica set or sharded cluster (Atlas is fine). Helm’s bundled Mongo chart is a **standalone** Deployment (`replicas: 1`, no `--replSet`), so live updates stay `idle` there until you point `DB_CONN_URI` at Atlas or an operator-managed replica set. + +Local Compose files initialize a single-node `rs0`. **Existing Compose Mongo volumes will not become a replica set cleanly** — drop the volume or start from an empty data dir if `hello` still reports standalone. + +### Live updates + +Enable `realtime.enabled` in database module config, then opt a schema in with `modelOptions.conduit.realtime.enabled`. Clients connect to the `/database/` Socket.IO namespace (path `/realtime`) and emit: + +``` +subscribe({ schema: 'Order', documentId?: string }) +unsubscribe({ schema: 'Order', documentId?: string }) +``` + +Events arrive as `change` with `{ version, operation, schema, documentId, occurredAt }` and contain no document fields and no resume token. This is **live-tail, not backfill**: the change stream starts at the end of the oplog. A leader restart or cursor drop does not replay missed events; clients subscribe again and refetch over authorized REST. + +The leader also publishes `database:change:${schema}` on the Redis bus. **Do not also relay `database:change:*` on `/events/` if the same client is on `/database/`** — that duplicates notifications. Keep the bus for other modules; just do not dual-subscribe. + +Client subscribers must authenticate. Schemas with document-level authorization reject schema-wide subscriptions and require a document ID plus a `read` check. Client sockets also require CMS `crudOperations.read.enabled` (checked again at emit: deny skips client delivery and keeps membership; authorization UNAVAILABLE keeps membership without emitting). Admin consumers use `POST /realtime/ticket` for a 30-second handshake token; that token cannot mint another ticket or call REST/GraphQL. 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 | @@ -54,6 +75,7 @@ When using MongoDB with a replica set (e.g., MongoDB Atlas), the database module | `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 | ### Recommended Production Settings diff --git a/modules/database/src/Database.ts b/modules/database/src/Database.ts index 57286cb1e..e2d36e26f 100644 --- a/modules/database/src/Database.ts +++ b/modules/database/src/Database.ts @@ -64,6 +64,7 @@ import { QueueController } from './controllers/queue.controller.js'; import AppConfigSchema, { Config } from './config/index.js'; import { Empty } from './protoTypes/google/protobuf/empty.js'; import { fileURLToPath } from 'node:url'; +import { RealtimeService } from './realtime/index.js'; const __filename = fileURLToPath(import.meta.url); const __dirname = path.dirname(__filename); @@ -107,6 +108,7 @@ export default class DatabaseModule extends ManagedModule { private customEndpointController?: CustomEndpointController; private _authorizationDefinitionsRegistered = false; private _databaseRouterWatchDispose: (() => void) | null = null; + private realtimeService?: RealtimeService; constructor(dbType: string, dbUri: string, peerManifestRoot?: string) { super('database', peerManifestRoot); @@ -164,6 +166,7 @@ export default class DatabaseModule extends ManagedModule { readConcern: config.readConcern ?? 'local', }); } + void this.realtimeService?.reconcile(); if (!config.viewCleanup.enabled) { try { await QueueController.getInstance().drainViewCleanupQueue(); @@ -1015,13 +1018,16 @@ export default class DatabaseModule extends ManagedModule { this.grpcSdk, this._activeAdapter, ); + this.realtimeService = new RealtimeService(this.grpcSdk, this._activeAdapter); this.adminRouter = new AdminHandlers( this.grpcServer, this.grpcSdk, this._activeAdapter, this.schemaController, this.customEndpointController, + this.realtimeService, ); + void this.realtimeService.reconcile(); this._databaseRouterWatchDispose?.(); this._databaseRouterWatchDispose = this.grpcSdk.watchPeer( 'router', @@ -1031,6 +1037,7 @@ export default class DatabaseModule extends ManagedModule { this.grpcServer, this._activeAdapter, this.grpcSdk, + this.realtimeService, ); this.schemaController?.setRouter(this.userRouter); this.customEndpointController?.setRouter(this.userRouter); @@ -1046,4 +1053,8 @@ export default class DatabaseModule extends ManagedModule { ); } } + + async shutdown(): Promise { + await this.realtimeService?.shutdown(); + } } diff --git a/modules/database/src/admin/__tests__/schema.admin.put-patch.test.ts b/modules/database/src/admin/__tests__/schema.admin.put-patch.test.ts index 89a7ae8ea..464bf4462 100644 --- a/modules/database/src/admin/__tests__/schema.admin.put-patch.test.ts +++ b/modules/database/src/admin/__tests__/schema.admin.put-patch.test.ts @@ -185,6 +185,21 @@ describe('SchemaAdmin PUT/PATCH split', () => { warnSpy.mockRestore(); }); + it('persists realtime conduitOptions used by live document updates', async () => { + const requestedSchema = makeRequestedSchema(); + const { admin, createSchema } = setup(requestedSchema); + + await admin.patchSchema( + makeCall({ + id: 'schema-1', + conduitOptions: { realtime: { enabled: true } }, + }), + ); + + const [writtenSchema] = createSchema.mock.calls[0]; + expect(writtenSchema.modelOptions.conduit.realtime.enabled).toBe(true); + }); + it('wipes existing documents when enabling authorization via conduitOptions', async () => { const requestedSchema = makeRequestedSchema(); const { admin, deleteMany } = setup(requestedSchema); diff --git a/modules/database/src/admin/index.ts b/modules/database/src/admin/index.ts index a4c1be91f..9742f31eb 100644 --- a/modules/database/src/admin/index.ts +++ b/modules/database/src/admin/index.ts @@ -23,6 +23,7 @@ import { SchemaController } from '../controllers/cms/schema.controller.js'; import { CustomEndpointController } from '../controllers/customEndpoints/customEndpoint.controller.js'; import { CustomEndpoints, DeclaredSchema, PendingSchemas } from '../models/index.js'; import { ConduitOptions, SchemaFieldsRequired } from '../interfaces/index.js'; +import type { RealtimeService } from '../realtime/index.js'; export class AdminHandlers { private readonly schemaAdmin: SchemaAdmin; @@ -36,6 +37,7 @@ export class AdminHandlers { private readonly _activeAdapter: DatabaseAdapter, private readonly schemaController: SchemaController, private readonly customEndpointController: CustomEndpointController, + private readonly realtimeService: RealtimeService, ) { this.schemaAdmin = new SchemaAdmin( this.grpcSdk, @@ -686,6 +688,7 @@ export class AdminHandlers { new ConduitRouteReturnDefinition('getDatabaseType', 'String'), this.schemaAdmin.getDatabaseType.bind(this.schemaAdmin), ); + this.realtimeService.registerAdmin(this.routingManager); this.routingManager.registerRoutes(); } } diff --git a/modules/database/src/admin/schema.admin.ts b/modules/database/src/admin/schema.admin.ts index 9c149d6db..cc8309664 100644 --- a/modules/database/src/admin/schema.admin.ts +++ b/modules/database/src/admin/schema.admin.ts @@ -250,6 +250,7 @@ export class SchemaAdmin { cms: call.request.params.conduitOptions?.cms, permissions: call.request.params.conduitOptions?.permissions, authorization: call.request.params.conduitOptions?.authorization, + realtime: call.request.params.conduitOptions?.realtime, readPreference: call.request.params.conduitOptions?.readPreference, timestamps: call.request.params.timestamps, }); @@ -343,6 +344,7 @@ export class SchemaAdmin { cms: conduitOptions?.cms, authorization: conduitOptions?.authorization, permissions: conduitOptions?.permissions, + realtime: conduitOptions?.realtime, readPreference: conduitOptions?.readPreference, existingModelOptions: requestedSchema.modelOptions, }); diff --git a/modules/database/src/config/index.ts b/modules/database/src/config/index.ts index 0f5a4fcc8..1ddda80d7 100644 --- a/modules/database/src/config/index.ts +++ b/modules/database/src/config/index.ts @@ -47,6 +47,13 @@ const AppConfigSchema = { default: 0, }, }, + realtime: { + enabled: { + doc: 'Enable MongoDB change-stream live updates for opted-in schemas', + format: 'Boolean', + default: false, + }, + }, }; const config = convict(AppConfigSchema); diff --git a/modules/database/src/index.ts b/modules/database/src/index.ts index 3e9466196..12261992a 100644 --- a/modules/database/src/index.ts +++ b/modules/database/src/index.ts @@ -1,5 +1,6 @@ import path from 'node:path'; import { fileURLToPath } from 'node:url'; +import { ConduitGrpcSdk } from '@conduitplatform/grpc-sdk'; import DatabaseModule from './Database.js'; const dbType = process.env.DB_TYPE ?? 'mongodb'; @@ -7,4 +8,21 @@ const dbUri = process.env.DB_CONN_URI ?? 'mongodb://localhost:27017'; const peerManifestRoot = path.join(path.dirname(fileURLToPath(import.meta.url)), '..'); const database = new DatabaseModule(dbType, dbUri, peerManifestRoot); + +function registerShutdownSignals(): void { + const shutdown = (signal: NodeJS.Signals) => { + void database + .shutdown() + .catch(err => { + ConduitGrpcSdk.Logger.error(err as Error); + }) + .finally(() => { + process.exit(signal === 'SIGINT' ? 130 : 0); + }); + }; + process.once('SIGTERM', () => shutdown('SIGTERM')); + process.once('SIGINT', () => shutdown('SIGINT')); +} + +registerShutdownSignals(); database.start(); diff --git a/modules/database/src/interfaces/ConduitOptions.ts b/modules/database/src/interfaces/ConduitOptions.ts index 82ee970fb..f5f1b3b02 100644 --- a/modules/database/src/interfaces/ConduitOptions.ts +++ b/modules/database/src/interfaces/ConduitOptions.ts @@ -30,5 +30,8 @@ export const ConduitOptions = { authorization: { enabled: ConduitBoolean.Optional, }, + realtime: { + enabled: ConduitBoolean.Optional, + }, readPreference: ConduitString.Optional, }; diff --git a/modules/database/src/metrics/index.ts b/modules/database/src/metrics/index.ts index e00ad3a33..3d3ef0256 100644 --- a/modules/database/src/metrics/index.ts +++ b/modules/database/src/metrics/index.ts @@ -31,4 +31,19 @@ export default { help: 'Tracks the total number of custom endpoints', }, }, + realtimeEvents: { + type: MetricType.Counter, + config: { + name: 'database_realtime_events_total', + help: 'Tracks normalized database change-stream events', + labelNames: ['operation'], + }, + }, + realtimeStreamErrors: { + type: MetricType.Counter, + config: { + name: 'database_realtime_stream_errors_total', + help: 'Tracks change-stream errors', + }, + }, }; diff --git a/modules/database/src/realtime/MongoChangeStreamCoordinator.ts b/modules/database/src/realtime/MongoChangeStreamCoordinator.ts new file mode 100644 index 000000000..b6a8349e2 --- /dev/null +++ b/modules/database/src/realtime/MongoChangeStreamCoordinator.ts @@ -0,0 +1,441 @@ +import { ConduitGrpcSdk } from '@conduitplatform/grpc-sdk'; +import { normalizeChangeEvent, type RawChangeEvent } from './normalize.js'; +import { authorizedDocumentRoom, roomsForPublicChange } from './rooms.js'; +import { topologyFromHello, type TopologyResult } from './topology.js'; +import type { + ChangeStreamLike, + DatabaseChangeEvent, + OptedInSchema, + RealtimeStatusCode, +} from './types.js'; +import type { RealtimeSubscriptionTracker } from './subscriptions.js'; +import { type AuthorizationSdk } from './authorize.js'; +import { checkRebacBatch, RealtimeRebacCache } from './rebacCache.js'; +import { + buildWatchPipeline, + optedInCollectionsKey, + WATCH_RESTART_OPERATIONS, + type WatchPipeline, +} from './watchPipeline.js'; + +const LEADER_LOCK = 'realtime:change-stream:leader'; +const LOCK_TTL_MS = 15_000; +const LOCK_RENEW_MS = 5_000; +const RETRY_BASE_MS = 1_000; +const RETRY_MAX_MS = 30_000; + +type LeaderLock = NonNullable< + Awaited['tryAcquireLock']>> +>; + +export type WatchFactory = (pipeline: WatchPipeline) => ChangeStreamLike; + +export type CoordinatorOptions = { + grpcSdk: ConduitGrpcSdk; + watch: WatchFactory; + hello: () => Promise<{ setName?: string; msg?: string } | null>; + getOptedInSchemas: () => OptedInSchema[]; + subscriptions: RealtimeSubscriptionTracker; + enabled: () => boolean; + engine: () => string; +}; + +export class MongoChangeStreamCoordinator { + private lock: LeaderLock | null = null; + private stream: ChangeStreamLike | null = null; + private renewTimer: NodeJS.Timeout | null = null; + private retryTimer: NodeJS.Timeout | null = null; + private closed = false; + private streamState: RealtimeStatusCode = 'idle'; + private lastEventAt?: string; + private lastError?: string; + private topology: TopologyResult = { supported: false }; + private retryAttempt = 0; + private watching = false; + private opening = false; + private acquiring = false; + private ignoreClose = false; + private lockGeneration = 0; + private changeQueue: Promise = Promise.resolve(); + private watchedCollectionsKey = ''; + private readonly rebacCache = new RealtimeRebacCache(); + + constructor(private readonly options: CoordinatorOptions) {} + + getState(): RealtimeStatusCode { + return this.streamState; + } + + getLastEventAt(): string | undefined { + return this.lastEventAt; + } + + getLastError(): string | undefined { + return this.lastError; + } + + getTopology(): TopologyResult { + 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()) { + await this.stopStream('idle'); + await this.releaseLeader(); + this.streamState = engine !== 'MongoDB' ? 'unsupported' : 'disabled'; + return; + } + this.topology = topologyFromHello(await this.options.hello().catch(() => null)); + if (!this.topology.supported) { + await this.stopStream('idle'); + await this.releaseLeader(); + this.streamState = 'idle'; + this.lastError = this.topology.message; + if ( + !this.topology.message || + this.topology.message.includes('Unable to determine') + ) { + this.scheduleRetry(); + } + return; + } + const collections = this.collectionNames(); + if (collections.length === 0) { + await this.stopStream('idle'); + await this.releaseLeader(); + this.streamState = 'idle'; + return; + } + const nextKey = optedInCollectionsKey(collections); + if (this.watching && nextKey !== this.watchedCollectionsKey) { + await this.stopStream('starting'); + } + await this.ensureLeader(); + } + + async shutdown(): Promise { + this.closed = true; + this.clearTimers(); + await this.changeQueue; + await this.stopStream('idle'); + await this.releaseLeader(); + this.rebacCache.clear(); + } + + private collectionNames(): string[] { + return this.options.getOptedInSchemas().map(schema => schema.collectionName); + } + + private async ensureLeader(): Promise { + if (this.lock) { + if (!this.watching) { + await this.openStream(); + } + return; + } + if (this.acquiring) return; + this.acquiring = true; + try { + if (this.lock) { + if (!this.watching) { + await this.openStream(); + } + return; + } + const acquired = await this.options.grpcSdk.state!.tryAcquireLock( + LEADER_LOCK, + LOCK_TTL_MS, + ); + if (!acquired) { + this.streamState = 'idle'; + this.scheduleRetry(); + return; + } + if (this.lock) { + try { + await this.options.grpcSdk.state!.releaseLock(acquired); + } catch { + // lock may already have expired + } + if (!this.watching) { + await this.openStream(); + } + return; + } + try { + this.lock = await acquired.extend(LOCK_TTL_MS); + } catch { + try { + await this.options.grpcSdk.state!.releaseLock(acquired); + } catch { + // lock may already have expired + } + this.lock = null; + this.streamState = 'idle'; + this.scheduleRetry(); + return; + } + this.bumpLockGeneration(); + this.startRenewal(); + await this.openStream(); + } catch (err) { + this.lastError = err instanceof Error ? err.message : String(err); + this.streamState = 'degraded'; + this.scheduleRetry(); + } finally { + this.acquiring = false; + } + } + + private startRenewal() { + this.clearRenewTimer(); + this.renewTimer = setInterval(() => { + void this.renewLock(); + }, LOCK_RENEW_MS); + } + + private async renewLock() { + if (!this.lock) return; + const generation = this.lockGeneration; + try { + this.lock = await this.lock.extend(LOCK_TTL_MS); + } catch { + if (this.lockGeneration !== generation) return; + await this.fenceLock('idle'); + this.scheduleRetry(); + } + } + + private async openStream() { + if (this.watching || this.closed || this.opening || !this.lock) return; + this.opening = true; + this.streamState = 'starting'; + this.ignoreClose = false; + const generation = this.lockGeneration; + try { + const collections = this.collectionNames(); + const pipeline = buildWatchPipeline(collections); + this.watchedCollectionsKey = optedInCollectionsKey(collections); + const stream = this.options.watch(pipeline); + if (generation !== this.lockGeneration || this.closed) { + try { + await stream.close(); + } catch { + // already closed + } + return; + } + this.stream = stream; + this.watching = true; + this.streamState = 'live'; + this.retryAttempt = 0; + stream.on('change', (change: unknown) => { + this.enqueueChange(change as RawChangeEvent, generation); + }); + stream.on('error', (err: unknown) => { + if (generation !== this.lockGeneration) return; + void this.handleStreamError(err); + }); + stream.on('close', () => { + if (generation !== this.lockGeneration) return; + this.watching = false; + if (!this.closed && !this.ignoreClose && this.lock) { + this.scheduleRetry(); + } + }); + } catch (err) { + this.watching = false; + await this.handleStreamError(err); + } finally { + this.opening = false; + } + } + + private enqueueChange(change: RawChangeEvent, generation: number) { + this.changeQueue = this.changeQueue.then(async () => { + if (this.closed || !this.watching || generation !== this.lockGeneration) return; + try { + await this.handleChange(change, generation); + } catch (err) { + if (generation !== this.lockGeneration) return; + 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, generation: number) { + if (generation !== this.lockGeneration) return; + const schema = this.resolveSchema(change.ns?.coll); + const event = schema ? normalizeChangeEvent(change, schema.name) : null; + if (!event || !schema) { + if (change.operationType && WATCH_RESTART_OPERATIONS.has(change.operationType)) { + await this.stopStream('starting'); + this.scheduleRetry(); + } + return; + } + this.lastEventAt = event.occurredAt; + this.lastError = undefined; + await this.emitChange(schema, event); + } + + private async emitChange(schema: OptedInSchema, event: DatabaseChangeEvent) { + const payload = JSON.stringify(event); + this.options.grpcSdk.bus?.publish(`database:change:${schema.name}`, payload); + ConduitGrpcSdk.Metrics?.increment('database_realtime_events_total', 1, { + operation: event.operation, + }); + await this.pushEvent(schema, event, payload); + } + + private async pushEvent( + schema: OptedInSchema, + event: DatabaseChangeEvent, + payload: string, + ) { + const adminRooms = roomsForPublicChange(schema.name, event.documentId); + await this.safePush('admin', adminRooms, payload); + if (!schema.cmsReadEnabled) { + return; + } + if (!schema.authorizationEnabled) { + await this.safePush('router', adminRooms, payload); + return; + } + const userIds = await this.options.subscriptions.listUsers( + schema.name, + event.documentId, + ); + const decisions = await checkRebacBatch( + this.rebacCache, + this.options.grpcSdk as unknown as AuthorizationSdk, + userIds, + schema.name, + event.documentId, + ); + const allowedRooms: string[] = []; + for (const userId of userIds) { + const decision = decisions.get(userId) ?? 'unavailable'; + 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, + ); + } + } + if (allowedRooms.length > 0) { + await this.safePush('router', allowedRooms, payload); + } + } + + private async safePush( + target: 'admin' | 'router', + rooms: string[], + data: string, + ): Promise { + const client = + target === 'admin' ? this.options.grpcSdk.admin : this.options.grpcSdk.router; + if (!client?.socketPush) return; + await client.socketPush({ + event: 'change', + data, + rooms, + receivers: [], + }); + } + + private resolveSchema(collectionName?: string): OptedInSchema | undefined { + if (!collectionName) return undefined; + return this.options + .getOptedInSchemas() + .find(schema => schema.collectionName === collectionName); + } + + private async handleStreamError(err: unknown) { + this.watching = false; + this.lastError = err instanceof Error ? err.message : String(err); + this.streamState = 'degraded'; + ConduitGrpcSdk.Metrics?.increment('database_realtime_stream_errors_total'); + ConduitGrpcSdk.Logger.error(err as Error); + await this.stopStream('degraded'); + this.scheduleRetry(); + } + + private scheduleRetry() { + if (this.closed || this.retryTimer) return; + const delay = Math.min(RETRY_MAX_MS, RETRY_BASE_MS * 2 ** this.retryAttempt); + this.retryAttempt += 1; + this.retryTimer = setTimeout(() => { + this.retryTimer = null; + void this.reconcile(); + }, delay); + } + + private async stopStream(nextState: RealtimeStatusCode) { + const stream = this.stream; + this.stream = null; + this.watching = false; + this.streamState = nextState; + this.ignoreClose = true; + this.watchedCollectionsKey = ''; + if (stream) { + try { + await stream.close(); + } catch { + // already closed + } + } + } + + private async releaseLeader() { + await this.fenceLock(this.streamState); + } + + private async fenceLock(nextState: RealtimeStatusCode) { + this.bumpLockGeneration(); + this.clearRenewTimer(); + const lock = this.lock; + this.lock = null; + await this.stopStream(nextState); + if (!lock) return; + try { + await this.options.grpcSdk.state!.releaseLock(lock); + } catch { + // lock may already have expired + } + } + + private bumpLockGeneration() { + this.lockGeneration += 1; + } + + private clearTimers() { + this.clearRenewTimer(); + if (this.retryTimer) { + clearTimeout(this.retryTimer); + this.retryTimer = null; + } + } + + private clearRenewTimer() { + if (this.renewTimer) { + clearInterval(this.renewTimer); + this.renewTimer = null; + } + } +} diff --git a/modules/database/src/realtime/RealtimeService.ts b/modules/database/src/realtime/RealtimeService.ts new file mode 100644 index 000000000..90369690e --- /dev/null +++ b/modules/database/src/realtime/RealtimeService.ts @@ -0,0 +1,159 @@ +import { + ConduitGrpcSdk, + ConduitRouteActions, + ConduitRouteReturnDefinition, +} from '@conduitplatform/grpc-sdk'; +import { + ConduitNumber, + ConduitString, + ConfigController, + RoutingManager, +} from '@conduitplatform/module-tools'; +import { DatabaseAdapter } from '../adapters/DatabaseAdapter.js'; +import { MongooseAdapter } from '../adapters/mongoose-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 { registerDatabaseRealtimeSocket } from './sockets.js'; +import { buildRealtimeStatus } from './status.js'; +import { RealtimeSubscriptionTracker } from './subscriptions.js'; +import type { ChangeStreamLike, OptedInSchema, RealtimeStatus } from './types.js'; +import type { WatchPipeline } from './watchPipeline.js'; + +export class RealtimeService { + private readonly subscriptions: RealtimeSubscriptionTracker; + private coordinator?: MongoChangeStreamCoordinator; + + constructor( + private readonly grpcSdk: ConduitGrpcSdk, + private readonly adapter: DatabaseAdapter, + ) { + this.subscriptions = new RealtimeSubscriptionTracker( + grpcSdk.redisManager.getClient(), + ); + this.grpcSdk.bus?.subscribe('database:create:schema', () => { + void this.reconcile(); + }); + this.grpcSdk.bus?.subscribe('database:delete:schema', () => { + void this.reconcile(); + }); + if (adapter instanceof MongooseAdapter) { + this.coordinator = new MongoChangeStreamCoordinator({ + grpcSdk, + watch: pipeline => this.openWatch(adapter, pipeline), + hello: () => this.hello(adapter), + getOptedInSchemas: () => this.getOptedInSchemas(), + subscriptions: this.subscriptions, + enabled: () => this.isGloballyEnabled(), + engine: () => adapter.getDatabaseType(), + }); + } + } + + registerAdmin(routingManager: RoutingManager) { + routingManager.route( + { + path: '/realtime/status', + action: ConduitRouteActions.GET, + description: `Returns live-update capability and runtime status for the database module.`, + }, + new ConduitRouteReturnDefinition('DatabaseRealtimeStatus', { + status: ConduitString.Required, + engine: ConduitString.Required, + activeSchemaCount: ConduitNumber.Required, + lastEventAt: ConduitString.Optional, + message: ConduitString.Optional, + }), + async () => this.getStatus(), + ); + registerDatabaseRealtimeSocket(routingManager, { + mode: 'admin', + grpcSdk: this.grpcSdk, + schemaLookup: this.adapter, + subscriptions: this.subscriptions, + isGloballyEnabled: () => this.isGloballyEnabled(), + }); + } + + registerClient(routingManager: RoutingManager) { + registerDatabaseRealtimeSocket(routingManager, { + mode: 'client', + grpcSdk: this.grpcSdk, + schemaLookup: this.adapter, + subscriptions: this.subscriptions, + isGloballyEnabled: () => this.isGloballyEnabled(), + }); + } + + async reconcile(): Promise { + if (!this.coordinator) return; + await this.coordinator.reconcile(); + } + + async shutdown(): Promise { + await this.coordinator?.shutdown(); + } + + async getStatus(): Promise { + const engine = this.adapter.getDatabaseType(); + const optedIn = this.getOptedInSchemas(); + return buildRealtimeStatus({ + engine, + enabled: this.isGloballyEnabled(), + topologySupported: this.coordinator?.getTopology().supported ?? false, + topologyMessage: this.coordinator?.getTopology().message, + activeSchemaCount: optedIn.length, + streamState: + this.coordinator?.getState() ?? (engine === 'MongoDB' ? 'idle' : 'unsupported'), + lastEventAt: this.coordinator?.getLastEventAt(), + lastError: this.coordinator?.getLastError(), + socketsEnabled: await this.areAdminSocketsEnabled(), + }); + } + + private isGloballyEnabled(): boolean { + return ConfigController.getInstance().config?.realtime?.enabled === true; + } + + private getOptedInSchemas(): OptedInSchema[] { + const schemas: OptedInSchema[] = []; + for (const schema of this.adapter.registeredSchemas.values()) { + const optedIn = toOptedInSchema({ + name: schema.name, + collectionName: schema.collectionName, + modelOptions: schema.modelOptions, + }); + if (optedIn) schemas.push(optedIn); + } + return schemas; + } + + private openWatch( + adapter: MongooseAdapter, + pipeline: WatchPipeline, + ): ChangeStreamLike { + const db = adapter.mongoose.connection.db; + if (!db) { + throw new Error('MongoDB connection is not ready'); + } + return db.watch(pipeline) as unknown as ChangeStreamLike; + } + + private async hello( + adapter: MongooseAdapter, + ): Promise<{ setName?: string; msg?: string } | null> { + const db = adapter.mongoose.connection.db; + if (!db) return null; + return db.admin().command({ hello: 1 }); + } + + private async areAdminSocketsEnabled(): Promise { + try { + const adminConfig = await this.grpcSdk.config.get('admin'); + return adminConfig?.transports?.sockets === true; + } catch { + return true; + } + } +} diff --git a/modules/database/src/realtime/__tests__/authorize.test.ts b/modules/database/src/realtime/__tests__/authorize.test.ts new file mode 100644 index 000000000..9c5eba15b --- /dev/null +++ b/modules/database/src/realtime/__tests__/authorize.test.ts @@ -0,0 +1,69 @@ +import { describe, expect, it } from '@jest/globals'; +import { status } from '@grpc/grpc-js'; +import { GrpcError } from '@conduitplatform/grpc-sdk'; +import { + assertSchemaAvailable, + optionalDocumentId, + parseSubscribeRequest, + requireSchemaName, +} from '../authorize.js'; + +describe('realtime authorization helpers', () => { + it('parses subscribe payloads and required schema names', () => { + expect(parseSubscribeRequest([{ schema: 'Order', documentId: '1' }])).toEqual({ + schema: 'Order', + documentId: '1', + }); + expect(requireSchemaName('Order')).toBe('Order'); + expect(optionalDocumentId(undefined)).toBeUndefined(); + expect(() => requireSchemaName('')).toThrow( + expect.objectContaining({ code: status.INVALID_ARGUMENT }), + ); + }); + + it('rejects missing schemas, disabled realtime, and CMS-read for clients', () => { + const lookup = { + getSchema: (name: string) => { + if (name === 'Missing') return undefined; + if (name === 'Off') { + return { name, modelOptions: { conduit: { realtime: { enabled: false } } } }; + } + return { + name, + modelOptions: { + conduit: { + realtime: { enabled: true }, + cms: { crudOperations: { read: { enabled: false } } }, + authorization: { enabled: true }, + }, + }, + }; + }, + }; + expect(() => assertSchemaAvailable(lookup, 'Missing', true)).toThrow( + expect.objectContaining({ code: status.NOT_FOUND }), + ); + expect(() => assertSchemaAvailable(lookup, 'Off', false)).toThrow( + expect.objectContaining({ code: status.FAILED_PRECONDITION }), + ); + expect(() => assertSchemaAvailable(lookup, 'Order', true)).toThrow( + expect.objectContaining({ code: status.PERMISSION_DENIED }), + ); + const admin = assertSchemaAvailable(lookup, 'Order', false); + expect(admin.authorizationEnabled).toBe(true); + }); + + it('maps adapter NOT_FOUND throws to a subscription error', () => { + const lookup = { + getSchema: () => { + throw new GrpcError(status.NOT_FOUND, 'Schema Missing not defined yet'); + }, + }; + expect(() => assertSchemaAvailable(lookup, 'Missing', false)).toThrow( + expect.objectContaining({ + code: status.NOT_FOUND, + message: 'Schema does not exist', + }), + ); + }); +}); diff --git a/modules/database/src/realtime/__tests__/change-stream.integration.test.ts b/modules/database/src/realtime/__tests__/change-stream.integration.test.ts new file mode 100644 index 000000000..b4e313310 --- /dev/null +++ b/modules/database/src/realtime/__tests__/change-stream.integration.test.ts @@ -0,0 +1,48 @@ +import { describe, expect, it } from '@jest/globals'; +import { MongoClient, ObjectId } from 'mongodb'; +import { normalizeChangeEvent, type RawChangeEvent } from '../normalize.js'; +import { buildWatchPipeline } from '../watchPipeline.js'; + +const replicaSetUri = process.env.DB_CONN_URI; +const integrationEnabled = Boolean(replicaSetUri?.includes('replicaSet')); +const describeIntegration = integrationEnabled ? describe : describe.skip; + +describeIntegration('MongoDB change stream contract', () => { + it('normalizes insert events from a replica-set watch without document fields', async () => { + const client = new MongoClient(replicaSetUri!); + await client.connect(); + const dbName = `conduit_realtime_${Date.now()}`; + const db = client.db(dbName); + try { + const collection = db.collection('orders'); + const stream = db.watch(buildWatchPipeline(['orders'])); + const change = await new Promise((resolve, reject) => { + const timer = setTimeout( + () => reject(new Error('timed out waiting for change')), + 10_000, + ); + stream.on('change', event => { + clearTimeout(timer); + resolve(event as RawChangeEvent); + }); + stream.on('error', err => { + clearTimeout(timer); + reject(err); + }); + void collection.insertOne({ _id: new ObjectId(), secret: 'do-not-leak' }); + }); + await stream.close(); + const event = normalizeChangeEvent(change, 'Order'); + expect(event).toMatchObject({ + version: 1, + operation: 'insert', + schema: 'Order', + }); + expect(event).not.toHaveProperty('fullDocument'); + expect(JSON.stringify(event)).not.toContain('do-not-leak'); + } finally { + await db.dropDatabase().catch(() => undefined); + await client.close(); + } + }); +}); diff --git a/modules/database/src/realtime/__tests__/coordinator.test.ts b/modules/database/src/realtime/__tests__/coordinator.test.ts new file mode 100644 index 000000000..c789ea94d --- /dev/null +++ b/modules/database/src/realtime/__tests__/coordinator.test.ts @@ -0,0 +1,427 @@ +import { EventEmitter } from 'node:events'; +import { afterEach, describe, expect, it, jest } from '@jest/globals'; +import { ObjectId } from 'bson'; +import { MongoChangeStreamCoordinator } from '../MongoChangeStreamCoordinator.js'; +import { RealtimeSubscriptionTracker } from '../subscriptions.js'; +import { roomsForPublicChange } from '../rooms.js'; + +class MemoryStore { + private sets = new Map>(); + readonly ttls = new Map(); + async sadd(key: string, ...members: string[]) { + 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[]) { + 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())]; + } + async scard(key: string) { + return this.sets.get(key)?.size ?? 0; + } + async del(...keys: string[]) { + keys.forEach(key => { + this.sets.delete(key); + this.ttls.delete(key); + }); + return keys.length; + } + async expire(key: string, seconds: number) { + this.ttls.set(key, seconds); + } + async persist(key: string) { + this.ttls.delete(key); + } +} + +function createCoordinator(overrides?: { + allow?: boolean; + authorizationAvailable?: boolean; + schemas?: { + name: string; + collectionName: string; + authorizationEnabled: boolean; + cmsReadEnabled?: boolean; + }[]; +}) { + const stream = new EventEmitter() as EventEmitter & { close: () => Promise }; + stream.close = async () => { + stream.emit('close'); + }; + 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 publish = jest.fn(); + const watch = jest.fn(() => stream as never); + const subscriptions = new RealtimeSubscriptionTracker(new MemoryStore()); + const grpcSdk = { + state: { + tryAcquireLock: jest.fn(async () => lock), + releaseLock: jest.fn(async () => undefined), + }, + bus: { publish }, + router: { socketPush: routerPush }, + admin: { socketPush: adminPush }, + isAvailable: () => overrides?.authorizationAvailable !== false, + authorization: + overrides?.authorizationAvailable === false + ? null + : { + can: async () => ({ allow: overrides?.allow !== false }), + }, + }; + const coordinator = new MongoChangeStreamCoordinator({ + grpcSdk: grpcSdk as never, + watch, + hello: async () => ({ setName: 'rs0' }), + getOptedInSchemas: () => + ( + overrides?.schemas ?? [ + { name: 'Order', collectionName: 'orders', authorizationEnabled: false }, + ] + ).map(schema => ({ + cmsReadEnabled: true, + ...schema, + })), + subscriptions, + enabled: () => true, + engine: () => 'MongoDB', + }); + return { + coordinator, + stream, + watch, + routerPush, + adminPush, + publish, + subscriptions, + grpcSdk, + lock, + }; +} + +function insertChange(collection: string, id: string) { + return { + operationType: 'insert', + ns: { coll: collection }, + documentKey: { _id: new ObjectId(id) }, + }; +} + +function expectWatchFromNow( + watch: { mock: { calls: unknown[][] } }, + callIndex = 0, +) { + expect(watch.mock.calls[callIndex]).toHaveLength(1); + expect(watch.mock.calls[callIndex][0]).not.toHaveProperty('resumeAfter'); +} + +describe('MongoChangeStreamCoordinator', () => { + afterEach(() => { + jest.useRealTimers(); + }); + + it('emits one normalized event to public rooms and ignores other collections', async () => { + const { coordinator, stream, routerPush, adminPush, publish } = createCoordinator(); + await coordinator.reconcile(); + stream.emit('change', { + ...insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c'), + fullDocument: { secret: 'nope' }, + wallTime: new Date('2026-01-02T00:00:00.000Z'), + }); + stream.emit('change', insertChange('other', '64b64c4c4c4c4c4c4c4c4c4d')); + 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); + expect(payload).toMatchObject({ + operation: 'insert', + schema: 'Order', + documentId: '64b64c4c4c4c4c4c4c4c4c4c', + }); + expect(payload).not.toHaveProperty('fullDocument'); + expect(payload).not.toHaveProperty('resumeToken'); + expect(payload).not.toHaveProperty('secret'); + const expectedRooms = roomsForPublicChange('Order', '64b64c4c4c4c4c4c4c4c4c4c'); + expect(routerPush).toHaveBeenCalledWith( + expect.objectContaining({ event: 'change', rooms: expectedRooms }), + ); + expect(adminPush).toHaveBeenCalledWith( + expect.objectContaining({ event: 'change', rooms: expectedRooms }), + ); + expect( + JSON.parse((adminPush.mock.calls[0][0] as { data: string }).data), + ).not.toHaveProperty('resumeToken'); + await coordinator.shutdown(); + }); + + it('serializes overlapping handlers', async () => { + const { coordinator, stream, adminPush } = createCoordinator(); + let release!: () => void; + const gate = new Promise(resolve => { + release = resolve; + }); + let first = true; + adminPush.mockImplementation(async () => { + if (first) { + first = false; + await gate; + } + }); + await coordinator.reconcile(); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4d')); + await Promise.resolve(); + await new Promise(resolve => setImmediate(resolve)); + expect(adminPush).toHaveBeenCalledTimes(1); + release(); + await coordinator.waitForIdle(); + expect(adminPush).toHaveBeenCalledTimes(2); + await coordinator.shutdown(); + }); + + it('watches opted-in collections with $match and $project from now', async () => { + const { coordinator, watch } = createCoordinator(); + await coordinator.reconcile(); + expect(watch).toHaveBeenCalledTimes(1); + const pipeline = watch.mock.calls[0][0] as Record[]; + expectWatchFromNow(watch); + expect(pipeline[0]).toEqual( + expect.objectContaining({ + $match: expect.objectContaining({ + $or: expect.arrayContaining([ + expect.objectContaining({ + 'ns.coll': { $in: ['orders'] }, + }), + ]), + }), + }), + ); + expect(pipeline[1]).toEqual({ + $project: { + fullDocument: 0, + updateDescription: 0, + fullDocumentBeforeChange: 0, + }, + }); + await coordinator.shutdown(); + }); + + it('reopens the watch when the opt-in set changes', async () => { + const schemas = [ + { name: 'Order', collectionName: 'orders', authorizationEnabled: false }, + ]; + const { coordinator, watch } = createCoordinator({ schemas }); + await coordinator.reconcile(); + expect(watch).toHaveBeenCalledTimes(1); + schemas.push({ + name: 'Item', + collectionName: 'items', + authorizationEnabled: false, + }); + await coordinator.reconcile(); + expect(watch).toHaveBeenCalledTimes(2); + const pipeline = watch.mock.calls[1][0] as Record[]; + const match = pipeline[0] as { + $match: { $or: Array<{ 'ns.coll'?: { $in: string[] } }> }; + }; + expect(match.$match.$or[0]['ns.coll']?.$in).toEqual( + expect.arrayContaining(['orders', 'items']), + ); + expectWatchFromNow(watch, 1); + await coordinator.shutdown(); + }); + + it('re-checks ReBAC before emission and drops revoked users', async () => { + const { coordinator, stream, routerPush, subscriptions } = createCoordinator({ + allow: false, + schemas: [{ name: 'Order', collectionName: 'orders', authorizationEnabled: true }], + }); + await subscriptions.addAuthorizedDocument( + 'sock-1', + 'Order', + '64b64c4c4c4c4c4c4c4c4c4c', + 'user-1', + ); + await coordinator.reconcile(); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); + await coordinator.waitForIdle(); + expect(routerPush).not.toHaveBeenCalled(); + expect(await subscriptions.listUsers('Order', '64b64c4c4c4c4c4c4c4c4c4c')).toEqual( + [], + ); + await coordinator.shutdown(); + }); + + it('retries later when the leader lock is held by another instance', async () => { + jest.useFakeTimers(); + const { coordinator, grpcSdk, lock } = createCoordinator(); + grpcSdk.state.tryAcquireLock.mockResolvedValueOnce(null).mockResolvedValue(lock); + await coordinator.reconcile(); + expect(coordinator.getState()).toBe('idle'); + await jest.advanceTimersByTimeAsync(1_000); + expect(coordinator.getState()).toBe('live'); + await coordinator.shutdown(); + jest.useRealTimers(); + }); + + it('opens a single watch when reconcile runs concurrently', async () => { + const { coordinator, watch } = createCoordinator(); + await Promise.all([coordinator.reconcile(), coordinator.reconcile()]); + expect(watch).toHaveBeenCalledTimes(1); + await coordinator.shutdown(); + }); + + it('stops the stream and retries from now when emit fails', async () => { + jest.useFakeTimers(); + const streams: Array Promise }> = []; + const { coordinator, adminPush, watch } = createCoordinator(); + watch.mockImplementation(() => { + const next = new EventEmitter() as EventEmitter & { close: () => Promise }; + next.close = async () => { + next.emit('close'); + }; + streams.push(next); + return next as never; + }); + adminPush + .mockResolvedValueOnce(undefined) + .mockRejectedValueOnce(new Error('push failed')) + .mockResolvedValue(undefined); + await coordinator.reconcile(); + streams[0].emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4b')); + await coordinator.waitForIdle(); + streams[0].emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); + streams[0].emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4d')); + await coordinator.waitForIdle(); + expect(adminPush).toHaveBeenCalledTimes(2); + await jest.advanceTimersByTimeAsync(1_000); + expect(watch).toHaveBeenCalledTimes(2); + expectWatchFromNow(watch, 1); + streams[1].emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); + await coordinator.waitForIdle(); + expect(adminPush).toHaveBeenCalledTimes(3); + await coordinator.shutdown(); + jest.useRealTimers(); + }); + + it.each(['drop', 'rename', 'invalidate', 'dropDatabase'] as const)( + 'reopens the watch on %s from now', + async operationType => { + jest.useFakeTimers(); + const streams: Array Promise }> = []; + const { coordinator, watch } = createCoordinator(); + watch.mockImplementation(() => { + const next = new EventEmitter() as EventEmitter & { close: () => Promise }; + next.close = async () => { + next.emit('close'); + }; + streams.push(next); + return next as never; + }); + await coordinator.reconcile(); + streams[0].emit('change', { + operationType, + ns: { coll: 'orders' }, + }); + await coordinator.waitForIdle(); + await jest.advanceTimersByTimeAsync(1_000); + expect(watch).toHaveBeenCalledTimes(2); + expectWatchFromNow(watch, 1); + await coordinator.shutdown(); + jest.useRealTimers(); + }, + ); + + it('does not remove users when authorization is unavailable', async () => { + const { coordinator, stream, routerPush, subscriptions } = createCoordinator({ + authorizationAvailable: false, + schemas: [{ name: 'Order', collectionName: 'orders', authorizationEnabled: true }], + }); + const removeUser = jest.spyOn(subscriptions, 'removeUser'); + await subscriptions.addAuthorizedDocument( + 'sock-1', + 'Order', + '64b64c4c4c4c4c4c4c4c4c4c', + 'user-1', + ); + expect(await subscriptions.listUsers('Order', '64b64c4c4c4c4c4c4c4c4c4c')).toEqual([ + 'user-1', + ]); + await coordinator.reconcile(); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); + await coordinator.waitForIdle(); + expect(routerPush).not.toHaveBeenCalled(); + expect(removeUser).not.toHaveBeenCalled(); + expect(await subscriptions.listUsers('Order', '64b64c4c4c4c4c4c4c4c4c4c')).toEqual([ + 'user-1', + ]); + await coordinator.shutdown(); + }); + + it('does not open a watch when the lock cannot be extended after acquire', async () => { + const { coordinator, watch, lock } = createCoordinator(); + lock.extend.mockRejectedValueOnce(new Error('extend failed')); + await coordinator.reconcile(); + expect(watch).not.toHaveBeenCalled(); + expect(coordinator.getState()).toBe('idle'); + await coordinator.shutdown(); + }); + + it('ignores draining watch events after lock renew failure', async () => { + jest.useFakeTimers(); + const { coordinator, stream, lock, adminPush } = createCoordinator(); + lock.extend.mockResolvedValueOnce(lock).mockRejectedValueOnce(new Error('lost lock')); + await coordinator.reconcile(); + expect(coordinator.getState()).toBe('live'); + await jest.advanceTimersByTimeAsync(5_000); + expect(coordinator.getState()).toBe('idle'); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); + await coordinator.waitForIdle(); + expect(adminPush).not.toHaveBeenCalled(); + await coordinator.shutdown(); + jest.useRealTimers(); + }); + + it('does not emit to clients when CMS read is denied and keeps membership', async () => { + const can = jest.fn(async () => ({ allow: true })); + const { coordinator, stream, routerPush, adminPush, subscriptions, grpcSdk } = + createCoordinator({ + schemas: [ + { + name: 'Order', + collectionName: 'orders', + authorizationEnabled: true, + cmsReadEnabled: false, + }, + ], + }); + grpcSdk.authorization = { can }; + await subscriptions.addAuthorizedDocument( + 'sock-1', + 'Order', + '64b64c4c4c4c4c4c4c4c4c4c', + 'user-1', + ); + await coordinator.reconcile(); + stream.emit('change', insertChange('orders', '64b64c4c4c4c4c4c4c4c4c4c')); + await coordinator.waitForIdle(); + expect(adminPush).toHaveBeenCalledTimes(1); + expect(routerPush).not.toHaveBeenCalled(); + expect(can).not.toHaveBeenCalled(); + expect(await subscriptions.listUsers('Order', '64b64c4c4c4c4c4c4c4c4c4c')).toEqual([ + 'user-1', + ]); + await coordinator.shutdown(); + }); +}); diff --git a/modules/database/src/realtime/__tests__/normalize.test.ts b/modules/database/src/realtime/__tests__/normalize.test.ts new file mode 100644 index 000000000..223009676 --- /dev/null +++ b/modules/database/src/realtime/__tests__/normalize.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from '@jest/globals'; +import { ObjectId } from 'bson'; +import { normalizeChangeEvent } from '../normalize.js'; + +describe('normalizeChangeEvent', () => { + it('normalizes insert/update/replace/delete into metadata-only events', () => { + const event = normalizeChangeEvent( + { + operationType: 'insert', + documentKey: { _id: new ObjectId('64b64c4c4c4c4c4c4c4c4c4c') }, + fullDocument: { secret: 'nope' }, + wallTime: new Date('2026-01-01T00:00:00.000Z'), + }, + 'Order', + ); + expect(event).toMatchObject({ + version: 1, + operation: 'insert', + schema: 'Order', + documentId: '64b64c4c4c4c4c4c4c4c4c4c', + occurredAt: '2026-01-01T00:00:00.000Z', + }); + expect(event).not.toHaveProperty('resumeToken'); + expect(JSON.parse(JSON.stringify(event))).not.toHaveProperty('fullDocument'); + }); + + it('ignores drop/invalidate and missing document ids', () => { + expect(normalizeChangeEvent({ operationType: 'drop' }, 'Order')).toBeNull(); + expect(normalizeChangeEvent({ operationType: 'insert' }, 'Order')).toBeNull(); + }); +}); diff --git a/modules/database/src/realtime/__tests__/recovery.test.ts b/modules/database/src/realtime/__tests__/recovery.test.ts new file mode 100644 index 000000000..4e5aaee37 --- /dev/null +++ b/modules/database/src/realtime/__tests__/recovery.test.ts @@ -0,0 +1,136 @@ +import { describe, expect, it } from '@jest/globals'; +import { authorizedDocumentRoom } from '../rooms.js'; +import { isRecoverableDisconnect, restoreAuthorizedSubscriptions } from '../recovery.js'; +import { createSocketHandlers } from '../sockets.js'; +import { + RealtimeSubscriptionTracker, + RECOVERY_REDIS_TTL_SECONDS, +} from '../subscriptions.js'; + +class MemoryStore { + private sets = new Map>(); + readonly ttls = new Map(); + async sadd(key: string, ...members: string[]) { + const set = this.sets.get(key) ?? new Set(); + members.forEach(member => set.add(member)); + this.sets.set(key, set); + } + async srem(key: string, ...members: string[]) { + members.forEach(member => this.sets.get(key)?.delete(member)); + } + async smembers(key: string) { + return [...(this.sets.get(key) ?? new Set())]; + } + async scard(key: string) { + return this.sets.get(key)?.size ?? 0; + } + async del(...keys: string[]) { + keys.forEach(key => { + this.sets.delete(key); + this.ttls.delete(key); + }); + } + async expire(key: string, seconds: number) { + this.ttls.set(key, seconds); + } + async persist(key: string) { + this.ttls.delete(key); + } +} + +describe('database socket recovery', () => { + it('does not treat transport close as a wipe', () => { + expect(isRecoverableDisconnect('transport close')).toBe(true); + expect(isRecoverableDisconnect('client namespace disconnect')).toBe(false); + }); + + it('restores authorized Redis membership after a recoverable disconnect', async () => { + const tracker = new RealtimeSubscriptionTracker(new MemoryStore()); + const room = authorizedDocumentRoom('Order', 'doc-1', 'user-1'); + await tracker.addAuthorizedDocument('sock-1', 'Order', 'doc-1', 'user-1'); + await tracker.disconnect('sock-1'); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual([]); + + await restoreAuthorizedSubscriptions({ + socketId: 'sock-1', + rooms: [room], + contextSubs: [], + subscriptions: tracker, + grpcSdk: { + isAvailable: () => true, + authorization: { can: async () => ({ allow: true }) }, + }, + }); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual(['user-1']); + }); + + it('keeps recovered /database/ sockets in the authorized list', async () => { + const tracker = new RealtimeSubscriptionTracker(new MemoryStore()); + await tracker.addAuthorizedDocument('sock-1', 'Order', 'doc-1', 'user-1'); + const handlers = createSocketHandlers({ + mode: 'client', + grpcSdk: { + isAvailable: () => true, + authorization: { can: async () => ({ allow: true }) }, + } as never, + schemaLookup: { getSchema: () => undefined }, + subscriptions: tracker, + isGloballyEnabled: () => true, + }); + await handlers.disconnect({ + request: { socketId: 'sock-1', params: ['transport close'] }, + } as never); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual(['user-1']); + await handlers.recovered({ + request: { + socketId: 'sock-1', + params: [authorizedDocumentRoom('Order', 'doc-1', 'user-1')], + context: {}, + }, + } as never); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual(['user-1']); + }); + + it('keeps membership when recovered authorization is unavailable', async () => { + const tracker = new RealtimeSubscriptionTracker(new MemoryStore()); + const room = authorizedDocumentRoom('Order', 'doc-1', 'user-1'); + await tracker.addAuthorizedDocument('sock-1', 'Order', 'doc-1', 'user-1'); + const { leaveRooms } = await restoreAuthorizedSubscriptions({ + socketId: 'sock-1', + rooms: [room], + contextSubs: [], + subscriptions: tracker, + grpcSdk: { + isAvailable: () => false, + }, + }); + expect(leaveRooms).toEqual([]); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual(['user-1']); + }); + + it('expires Redis membership when a recoverable disconnect never recovers', async () => { + const store = new MemoryStore(); + const tracker = new RealtimeSubscriptionTracker(store); + const handlers = createSocketHandlers({ + mode: 'client', + grpcSdk: { + isAvailable: () => true, + authorization: { can: async () => ({ allow: true }) }, + } as never, + schemaLookup: { getSchema: () => undefined }, + subscriptions: tracker, + isGloballyEnabled: () => true, + }); + await tracker.addAuthorizedDocument('sock-1', 'Order', 'doc-1', 'user-1'); + expect(store.ttls.size).toBe(0); + await handlers.disconnect({ + request: { socketId: 'sock-1', params: ['transport close'] }, + } as never); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual(['user-1']); + expect(store.ttls.get('realtime:socket:sock-1')).toBe(RECOVERY_REDIS_TTL_SECONDS); + expect(store.ttls.get('realtime:doc:Order:doc-1')).toBe(RECOVERY_REDIS_TTL_SECONDS); + expect(store.ttls.get('realtime:userdoc:Order:doc-1:user-1')).toBe( + RECOVERY_REDIS_TTL_SECONDS, + ); + }); +}); diff --git a/modules/database/src/realtime/__tests__/rooms.test.ts b/modules/database/src/realtime/__tests__/rooms.test.ts new file mode 100644 index 000000000..96de67f73 --- /dev/null +++ b/modules/database/src/realtime/__tests__/rooms.test.ts @@ -0,0 +1,38 @@ +import { describe, expect, it } from '@jest/globals'; +import { + authorizedDocumentRoom, + documentRoom, + parseAuthorizedDocumentRoom, + roomsForPublicChange, + schemaRoom, +} from '../rooms.js'; + +describe('realtime rooms', () => { + it('isolates schema, document, and per-user rooms', () => { + expect(schemaRoom('Order')).toBe('database:schema:Order'); + expect(documentRoom('Order', 'abc')).toBe('database:doc:Order:abc'); + expect(authorizedDocumentRoom('Order', 'abc', 'user-1')).toBe( + 'database:doc:Order:abc:user:user-1', + ); + expect(roomsForPublicChange('Order', 'abc')).toEqual([ + 'database:schema:Order', + 'database:doc:Order:abc', + ]); + expect(schemaRoom('Order')).not.toBe(schemaRoom('Orders')); + expect(authorizedDocumentRoom('Order', 'abc', 'u1')).not.toBe( + authorizedDocumentRoom('Order', 'abc', 'u2'), + ); + expect( + parseAuthorizedDocumentRoom(authorizedDocumentRoom('Order', 'abc', 'user-1')), + ).toEqual({ + schema: 'Order', + documentId: 'abc', + userId: 'user-1', + }); + }); + + it('encodes reserved characters so rooms cannot collide', () => { + expect(schemaRoom('a/b')).toBe('database:schema:a%2Fb'); + expect(documentRoom('Order', 'id:1')).toBe('database:doc:Order:id%3A1'); + }); +}); diff --git a/modules/database/src/realtime/__tests__/status.test.ts b/modules/database/src/realtime/__tests__/status.test.ts new file mode 100644 index 000000000..a94790cfb --- /dev/null +++ b/modules/database/src/realtime/__tests__/status.test.ts @@ -0,0 +1,53 @@ +import { describe, expect, it } from '@jest/globals'; +import { buildRealtimeStatus } from '../status.js'; + +describe('buildRealtimeStatus', () => { + it('reports unsupported, disabled, idle, live, and degraded states', () => { + expect( + buildRealtimeStatus({ + engine: 'PostgreSQL', + enabled: true, + topologySupported: true, + activeSchemaCount: 1, + streamState: 'live', + }).status, + ).toBe('unsupported'); + expect( + buildRealtimeStatus({ + engine: 'MongoDB', + enabled: false, + topologySupported: true, + activeSchemaCount: 1, + streamState: 'live', + }).status, + ).toBe('disabled'); + expect( + buildRealtimeStatus({ + engine: 'MongoDB', + enabled: true, + topologySupported: false, + activeSchemaCount: 1, + streamState: 'idle', + }).message, + ).toMatch(/replica set/i); + expect( + buildRealtimeStatus({ + engine: 'MongoDB', + enabled: true, + topologySupported: true, + activeSchemaCount: 2, + streamState: 'live', + }).status, + ).toBe('live'); + expect( + buildRealtimeStatus({ + engine: 'MongoDB', + enabled: true, + topologySupported: true, + activeSchemaCount: 1, + streamState: 'degraded', + lastError: 'boom', + }), + ).toMatchObject({ status: 'degraded', message: 'boom' }); + }); +}); diff --git a/modules/database/src/realtime/__tests__/subscriptions.test.ts b/modules/database/src/realtime/__tests__/subscriptions.test.ts new file mode 100644 index 000000000..13bcc7298 --- /dev/null +++ b/modules/database/src/realtime/__tests__/subscriptions.test.ts @@ -0,0 +1,96 @@ +import { describe, expect, it } from '@jest/globals'; +import { + RealtimeSubscriptionTracker, + RECOVERY_REDIS_TTL_SECONDS, +} from '../subscriptions.js'; + +class MemoryStore { + private sets = new Map>(); + readonly ttls = new Map(); + + async sadd(key: string, ...members: string[]) { + 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[]) { + 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())]; + } + + async scard(key: string) { + return this.sets.get(key)?.size ?? 0; + } + + async del(...keys: string[]) { + keys.forEach(key => { + this.sets.delete(key); + this.ttls.delete(key); + }); + return keys.length; + } + + async expire(key: string, seconds: number) { + this.ttls.set(key, seconds); + } + + async persist(key: string) { + this.ttls.delete(key); + } +} + +describe('RealtimeSubscriptionTracker', () => { + it('tracks users per document and cleans up on disconnect', async () => { + const tracker = new RealtimeSubscriptionTracker(new MemoryStore()); + await tracker.addAuthorizedDocument('s1', 'Order', 'doc-1', 'user-1'); + await tracker.addAuthorizedDocument('s2', 'Order', 'doc-1', 'user-1'); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual(['user-1']); + + await tracker.disconnect('s1'); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual(['user-1']); + + await tracker.disconnect('s2'); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual([]); + }); + + it('does not expire shared document keys while another socket is still live', async () => { + const store = new MemoryStore(); + const tracker = new RealtimeSubscriptionTracker(store); + await tracker.addAuthorizedDocument('s1', 'Order', 'doc-1', 'user-1'); + await tracker.addAuthorizedDocument('s2', 'Order', 'doc-1', 'user-1'); + await tracker.armRecoverableTtl('s1'); + expect(store.ttls.get('realtime:socket:s1')).toBe(RECOVERY_REDIS_TTL_SECONDS); + expect(store.ttls.has('realtime:doc:Order:doc-1')).toBe(false); + expect(store.ttls.has('realtime:userdoc:Order:doc-1:user-1')).toBe(false); + }); + + it('removes revoked users from the document set', async () => { + const tracker = new RealtimeSubscriptionTracker(new MemoryStore()); + await tracker.addAuthorizedDocument('s1', 'Order', 'doc-1', 'user-1'); + await tracker.removeUser('Order', 'doc-1', 'user-1'); + expect(await tracker.listUsers('Order', 'doc-1')).toEqual([]); + }); + + it('arms Redis TTL on recoverable disconnect keys and persists on restore', async () => { + const store = new MemoryStore(); + const tracker = new RealtimeSubscriptionTracker(store); + await tracker.addAuthorizedDocument('s1', 'Order', 'doc-1', 'user-1'); + expect(store.ttls.size).toBe(0); + await tracker.armRecoverableTtl('s1'); + expect(store.ttls.get('realtime:socket:s1')).toBe(RECOVERY_REDIS_TTL_SECONDS); + expect(store.ttls.get('realtime:doc:Order:doc-1')).toBe(RECOVERY_REDIS_TTL_SECONDS); + expect(store.ttls.get('realtime:userdoc:Order:doc-1:user-1')).toBe( + RECOVERY_REDIS_TTL_SECONDS, + ); + await tracker.addAuthorizedDocument('s1', 'Order', 'doc-1', 'user-1'); + expect(store.ttls.size).toBe(0); + }); +}); diff --git a/modules/database/src/realtime/__tests__/topology.test.ts b/modules/database/src/realtime/__tests__/topology.test.ts new file mode 100644 index 000000000..51b52be54 --- /dev/null +++ b/modules/database/src/realtime/__tests__/topology.test.ts @@ -0,0 +1,10 @@ +import { describe, expect, it } from '@jest/globals'; +import { topologyFromHello } from '../topology.js'; + +describe('topology helpers', () => { + it('accepts replica sets and mongos, rejects standalone', () => { + expect(topologyFromHello({ setName: 'rs0' })).toEqual({ supported: true }); + expect(topologyFromHello({ msg: 'isdbgrid' })).toEqual({ supported: true }); + expect(topologyFromHello({}).supported).toBe(false); + }); +}); diff --git a/modules/database/src/realtime/__tests__/watchPipeline.test.ts b/modules/database/src/realtime/__tests__/watchPipeline.test.ts new file mode 100644 index 000000000..7c31a64f5 --- /dev/null +++ b/modules/database/src/realtime/__tests__/watchPipeline.test.ts @@ -0,0 +1,28 @@ +import { describe, expect, it } from '@jest/globals'; +import { buildWatchPipeline } from '../watchPipeline.js'; + +describe('buildWatchPipeline', () => { + it('matches opted-in collections and projects out fullDocument', () => { + const pipeline = buildWatchPipeline(['orders', 'items']); + expect(pipeline[0]).toEqual({ + $match: { + $or: [ + { + operationType: { $in: ['insert', 'update', 'replace', 'delete'] }, + 'ns.coll': { $in: ['orders', 'items'] }, + }, + { + operationType: { $in: ['drop', 'rename', 'invalidate', 'dropDatabase'] }, + }, + ], + }, + }); + expect(pipeline[1]).toEqual({ + $project: { + fullDocument: 0, + updateDescription: 0, + fullDocumentBeforeChange: 0, + }, + }); + }); +}); diff --git a/modules/database/src/realtime/authorize.ts b/modules/database/src/realtime/authorize.ts new file mode 100644 index 000000000..d2a16e1cf --- /dev/null +++ b/modules/database/src/realtime/authorize.ts @@ -0,0 +1,148 @@ +import { status } from '@grpc/grpc-js'; +import { GrpcError } from '@conduitplatform/grpc-sdk'; +import type { OptedInSchema, RebacDecision, SubscribeRequest } from './types.js'; + +export class RealtimeSubscriptionError extends GrpcError { + constructor(code: number, message: string) { + super(code, message); + this.name = 'RealtimeSubscriptionError'; + } +} + +export type AuthorizationSdk = { + isAvailable: (module: string) => boolean; + authorization?: { + can: (request: { + subject: string; + actions: string[]; + resource: string; + }) => Promise<{ allow: boolean }>; + } | null; +}; + +export type SchemaLookup = { + getSchema(name: string): + | { + name: string; + modelOptions?: { + conduit?: { + realtime?: { enabled?: boolean }; + cms?: { crudOperations?: { read?: { enabled?: boolean } } }; + authorization?: { enabled?: boolean }; + }; + }; + } + | undefined; +}; + +export function parseSubscribeRequest(params: unknown[]): SubscribeRequest { + const raw = params[0]; + if (raw && typeof raw === 'object' && !Array.isArray(raw)) { + return raw as SubscribeRequest; + } + return {}; +} + +export function requireSchemaName(value: unknown): string { + if (typeof value !== 'string' || value.trim() === '') { + throw new RealtimeSubscriptionError(status.INVALID_ARGUMENT, 'schema is required'); + } + return value.trim(); +} + +export function optionalDocumentId(value: unknown): string | undefined { + if (value === undefined || value === null || value === '') return undefined; + if (typeof value !== 'string') { + throw new RealtimeSubscriptionError( + status.INVALID_ARGUMENT, + 'documentId must be a string', + ); + } + return value; +} + +export async function readDocumentDecision( + grpcSdk: AuthorizationSdk, + schema: string, + documentId: string, + userId: string, +): Promise { + if (!grpcSdk.authorization || !grpcSdk.isAvailable('authorization')) { + return 'unavailable'; + } + try { + const decision = await grpcSdk.authorization.can({ + subject: `User:${userId}`, + actions: ['read'], + resource: `${schema}:${documentId}`, + }); + return decision.allow === true ? 'allow' : 'deny'; + } catch { + return 'unavailable'; + } +} + +export async function canReadDocument( + grpcSdk: AuthorizationSdk, + schema: string, + documentId: string, + userId: string, +): Promise { + return (await readDocumentDecision(grpcSdk, schema, documentId, userId)) === 'allow'; +} + +export function assertSchemaAvailable( + lookup: SchemaLookup, + schemaName: string, + requireCmsRead: boolean, +): { authorizationEnabled: boolean } { + let schema: ReturnType; + try { + schema = lookup.getSchema(schemaName); + } catch (err) { + if (err instanceof GrpcError && err.code === status.NOT_FOUND) { + throw new RealtimeSubscriptionError(status.NOT_FOUND, 'Schema does not exist'); + } + throw err; + } + if (!schema) { + throw new RealtimeSubscriptionError(status.NOT_FOUND, 'Schema does not exist'); + } + const conduit = schema.modelOptions?.conduit; + if (!conduit?.realtime?.enabled) { + throw new RealtimeSubscriptionError( + status.FAILED_PRECONDITION, + 'Live updates are not enabled for this schema', + ); + } + if (requireCmsRead && conduit.cms?.crudOperations?.read?.enabled !== true) { + throw new RealtimeSubscriptionError( + status.PERMISSION_DENIED, + 'CMS read is not enabled for this schema', + ); + } + return { + authorizationEnabled: conduit.authorization?.enabled === true, + }; +} + +export function toOptedInSchema(schema: { + name: string; + collectionName: string; + modelOptions?: { + conduit?: { + realtime?: { enabled?: boolean }; + cms?: { crudOperations?: { read?: { enabled?: boolean } } }; + authorization?: { enabled?: boolean }; + }; + }; +}): OptedInSchema | null { + if (!schema.modelOptions?.conduit?.realtime?.enabled) return null; + return { + name: schema.name, + collectionName: schema.collectionName, + authorizationEnabled: schema.modelOptions.conduit.authorization?.enabled === true, + cmsReadEnabled: + schema.modelOptions.conduit.cms?.crudOperations?.read?.enabled === true, + }; +} diff --git a/modules/database/src/realtime/index.ts b/modules/database/src/realtime/index.ts new file mode 100644 index 000000000..655c04bba --- /dev/null +++ b/modules/database/src/realtime/index.ts @@ -0,0 +1,11 @@ +export { RealtimeService } from './RealtimeService.js'; +export { buildRealtimeStatus } from './status.js'; +export { normalizeChangeEvent } from './normalize.js'; +export { + schemaRoom, + documentRoom, + authorizedDocumentRoom, + roomsForPublicChange, +} from './rooms.js'; +export { RealtimeSubscriptionTracker } from './subscriptions.js'; +export type { RealtimeStatus, DatabaseChangeEvent } from './types.js'; diff --git a/modules/database/src/realtime/normalize.ts b/modules/database/src/realtime/normalize.ts new file mode 100644 index 000000000..b5055c3d2 --- /dev/null +++ b/modules/database/src/realtime/normalize.ts @@ -0,0 +1,54 @@ +import { + DATABASE_CHANGE_EVENT_VERSION, + DATABASE_CHANGE_OPERATIONS, + type DatabaseChangeEvent, + type DatabaseChangeOperation, +} from './types.js'; + +const OPERATION_SET = new Set(DATABASE_CHANGE_OPERATIONS); + +export type RawChangeEvent = { + operationType?: string; + ns?: { coll?: string }; + documentKey?: { _id?: unknown }; + wallTime?: Date; +}; + +export function normalizeChangeEvent( + change: RawChangeEvent, + schemaName: string, + occurredAt: Date = new Date(), +): DatabaseChangeEvent | null { + const operation = change.operationType; + if (!operation || !OPERATION_SET.has(operation)) { + return null; + } + const documentId = extractDocumentId(change.documentKey?._id); + if (!documentId) { + return null; + } + return { + version: DATABASE_CHANGE_EVENT_VERSION, + operation: operation as DatabaseChangeOperation, + schema: schemaName, + documentId, + occurredAt: (change.wallTime instanceof Date + ? change.wallTime + : occurredAt + ).toISOString(), + }; +} + +function extractDocumentId(id: unknown): string | null { + if (id === undefined || id === null) return null; + if (typeof id === 'string' || typeof id === 'number') return String(id); + if (typeof id === 'object' && id !== null && 'toHexString' in id) { + const hex = (id as { toHexString: () => string }).toHexString(); + return typeof hex === 'string' && hex.length > 0 ? hex : null; + } + if (typeof id === 'object' && id !== null && 'toString' in id) { + const value = String(id); + return value && value !== '[object Object]' ? value : null; + } + return null; +} diff --git a/modules/database/src/realtime/rebacCache.ts b/modules/database/src/realtime/rebacCache.ts new file mode 100644 index 000000000..c57595227 --- /dev/null +++ b/modules/database/src/realtime/rebacCache.ts @@ -0,0 +1,95 @@ +import type { RebacDecision } from './types.js'; +import type { AuthorizationSdk } from './authorize.js'; + +const DEFAULT_TTL_MS = 12_000; +const DEFAULT_MAX_ENTRIES = 10_000; + +type CacheEntry = { + decision: 'allow' | 'deny'; + expiresAt: number; +}; + +export class RealtimeRebacCache { + private readonly entries = new Map(); + + constructor( + private readonly ttlMs: number = DEFAULT_TTL_MS, + private readonly maxEntries: number = DEFAULT_MAX_ENTRIES, + ) {} + + async check( + grpcSdk: AuthorizationSdk, + userId: string, + schema: string, + documentId: string, + ): Promise { + const resource = `${schema}:${documentId}`; + const key = `${userId}:read:${resource}`; + const now = Date.now(); + this.sweepExpired(now); + const cached = this.entries.get(key); + if (cached && cached.expiresAt > now) { + return cached.decision; + } + if (!grpcSdk.authorization || !grpcSdk.isAvailable('authorization')) { + return 'unavailable'; + } + try { + const decision = await grpcSdk.authorization.can({ + subject: `User:${userId}`, + actions: ['read'], + resource, + }); + const value: 'allow' | 'deny' = decision.allow ? 'allow' : 'deny'; + this.set(key, value, now); + return value; + } catch { + return 'unavailable'; + } + } + + clear(): void { + this.entries.clear(); + } + + private set(key: string, decision: 'allow' | 'deny', now: number): void { + if (this.entries.size >= this.maxEntries && !this.entries.has(key)) { + const firstKey = this.entries.keys().next().value; + if (firstKey) { + this.entries.delete(firstKey); + } + } + this.entries.set(key, { decision, expiresAt: now + this.ttlMs }); + } + + private sweepExpired(now: number): void { + for (const [key, entry] of this.entries) { + if (entry.expiresAt <= now) { + this.entries.delete(key); + } + } + } +} + +export async function checkRebacBatch( + cache: RealtimeRebacCache, + grpcSdk: AuthorizationSdk, + userIds: string[], + schema: string, + documentId: string, + concurrency = 8, +): Promise> { + const results = new Map(); + let index = 0; + async function worker(): Promise { + while (index < userIds.length) { + const userId = userIds[index++]; + results.set(userId, await cache.check(grpcSdk, userId, schema, documentId)); + } + } + const workers = Array.from({ length: Math.min(concurrency, userIds.length) }, () => + worker(), + ); + await Promise.all(workers); + return results; +} diff --git a/modules/database/src/realtime/recovery.ts b/modules/database/src/realtime/recovery.ts new file mode 100644 index 000000000..7ff79735b --- /dev/null +++ b/modules/database/src/realtime/recovery.ts @@ -0,0 +1,141 @@ +import type { Indexable, ParsedSocketRequest } from '@conduitplatform/grpc-sdk'; +import { readDocumentDecision, type AuthorizationSdk } from './authorize.js'; +import { authorizedDocumentRoom, parseAuthorizedDocumentRoom } from './rooms.js'; +import type { RealtimeSubscriptionTracker } from './subscriptions.js'; + +const RECOVERABLE_DISCONNECT = new Set([ + 'ping timeout', + 'transport close', + 'transport error', +]); + +const CONTEXT_KEY = 'databaseSubs'; + +export type AuthorizedSub = { + schema: string; + documentId: string; + userId: string; +}; + +export function isRecoverableDisconnect(reason: unknown): boolean { + return typeof reason === 'string' && RECOVERABLE_DISCONNECT.has(reason); +} + +export function authorizedSubsFromContext( + context: Indexable | undefined, +): AuthorizedSub[] { + if (!context) return []; + const raw = context[CONTEXT_KEY]; + if (!Array.isArray(raw)) return []; + const subs: AuthorizedSub[] = []; + for (const item of raw) { + if ( + item && + typeof item === 'object' && + typeof (item as AuthorizedSub).schema === 'string' && + typeof (item as AuthorizedSub).documentId === 'string' && + typeof (item as AuthorizedSub).userId === 'string' + ) { + subs.push({ + schema: (item as AuthorizedSub).schema, + documentId: (item as AuthorizedSub).documentId, + userId: (item as AuthorizedSub).userId, + }); + } + } + return subs; +} + +export function persistAuthorizedSubOnContext( + context: Indexable | undefined, + schema: string, + documentId: string, + userId: string, +): void { + if (!context) return; + const next = authorizedSubsFromContext(context).filter( + sub => + !(sub.schema === schema && sub.documentId === documentId && sub.userId === userId), + ); + next.push({ schema, documentId, userId }); + context[CONTEXT_KEY] = next; +} + +export function removeAuthorizedSubFromContext( + context: Indexable | undefined, + schema: string, + documentId: string, + userId: string, +): void { + if (!context) return; + const next = authorizedSubsFromContext(context).filter( + sub => + !(sub.schema === schema && sub.documentId === documentId && sub.userId === userId), + ); + if (next.length === 0) { + delete context[CONTEXT_KEY]; + } else { + context[CONTEXT_KEY] = next; + } +} + +export function recoveredRoomsFromRequest(call: ParsedSocketRequest): string[] { + const params = call.request.params ?? []; + if (params.every(item => typeof item === 'string')) { + return params as string[]; + } + return []; +} + +export async function restoreAuthorizedSubscriptions(options: { + socketId: string; + rooms: string[]; + contextSubs: AuthorizedSub[]; + subscriptions: RealtimeSubscriptionTracker; + grpcSdk: AuthorizationSdk; +}): Promise<{ leaveRooms: string[] }> { + const seen = new Set(); + const subs: AuthorizedSub[] = []; + for (const room of options.rooms) { + const parsed = parseAuthorizedDocumentRoom(room); + if (!parsed) continue; + const key = `${parsed.schema}:${parsed.documentId}:${parsed.userId}`; + if (seen.has(key)) continue; + seen.add(key); + subs.push(parsed); + } + for (const sub of options.contextSubs) { + const key = `${sub.schema}:${sub.documentId}:${sub.userId}`; + if (seen.has(key)) continue; + seen.add(key); + subs.push(sub); + } + + const leaveRooms: string[] = []; + for (const sub of subs) { + const decision = await readDocumentDecision( + options.grpcSdk, + sub.schema, + sub.documentId, + sub.userId, + ); + const room = authorizedDocumentRoom(sub.schema, sub.documentId, sub.userId); + if (decision === 'deny') { + await options.subscriptions.removeAuthorizedDocument( + options.socketId, + sub.schema, + sub.documentId, + sub.userId, + ); + leaveRooms.push(room); + continue; + } + await options.subscriptions.addAuthorizedDocument( + options.socketId, + sub.schema, + sub.documentId, + sub.userId, + ); + } + return { leaveRooms }; +} diff --git a/modules/database/src/realtime/rooms.ts b/modules/database/src/realtime/rooms.ts new file mode 100644 index 000000000..5792482f9 --- /dev/null +++ b/modules/database/src/realtime/rooms.ts @@ -0,0 +1,47 @@ +const ROOM_PREFIX = 'database'; + +function encodeSegment(value: string): string { + return encodeURIComponent(value); +} + +export function schemaRoom(schema: string): string { + return `${ROOM_PREFIX}:schema:${encodeSegment(schema)}`; +} + +export function documentRoom(schema: string, documentId: string): string { + return `${ROOM_PREFIX}:doc:${encodeSegment(schema)}:${encodeSegment(documentId)}`; +} + +export function authorizedDocumentRoom( + schema: string, + documentId: string, + userId: string, +): string { + return `${ROOM_PREFIX}:doc:${encodeSegment(schema)}:${encodeSegment(documentId)}:user:${encodeSegment(userId)}`; +} + +export function roomsForPublicChange(schema: string, documentId: string): string[] { + return [schemaRoom(schema), documentRoom(schema, documentId)]; +} + +export function parseAuthorizedDocumentRoom(room: string): { + schema: string; + documentId: string; + userId: string; +} | null { + const prefix = `${ROOM_PREFIX}:doc:`; + if (!room.startsWith(prefix)) return null; + const rest = room.slice(prefix.length); + const userMarker = ':user:'; + const userIndex = rest.lastIndexOf(userMarker); + if (userIndex === -1) return null; + const userId = decodeURIComponent(rest.slice(userIndex + userMarker.length)); + const schemaDoc = rest.slice(0, userIndex); + const lastColon = schemaDoc.lastIndexOf(':'); + if (lastColon === -1) return null; + return { + schema: decodeURIComponent(schemaDoc.slice(0, lastColon)), + documentId: decodeURIComponent(schemaDoc.slice(lastColon + 1)), + userId, + }; +} diff --git a/modules/database/src/realtime/sockets.ts b/modules/database/src/realtime/sockets.ts new file mode 100644 index 000000000..3a06ab121 --- /dev/null +++ b/modules/database/src/realtime/sockets.ts @@ -0,0 +1,196 @@ +import { + ConduitGrpcSdk, + ConduitRouteReturnDefinition, + type Indexable, + ParsedSocketRequest, + TYPE, + UnparsedSocketResponse, +} from '@conduitplatform/grpc-sdk'; +import { status } from '@grpc/grpc-js'; +import { RoutingManager } from '@conduitplatform/module-tools'; +import { + assertSchemaAvailable, + canReadDocument, + optionalDocumentId, + parseSubscribeRequest, + RealtimeSubscriptionError, + requireSchemaName, + type AuthorizationSdk, + type SchemaLookup, +} from './authorize.js'; +import { authorizedDocumentRoom, documentRoom, schemaRoom } from './rooms.js'; +import type { RealtimeSubscriptionTracker } from './subscriptions.js'; +import { + authorizedSubsFromContext, + isRecoverableDisconnect, + persistAuthorizedSubOnContext, + recoveredRoomsFromRequest, + removeAuthorizedSubFromContext, + restoreAuthorizedSubscriptions, +} from './recovery.js'; + +type SocketMode = 'client' | 'admin'; + +type RealtimeSocketOptions = { + mode: SocketMode; + grpcSdk: ConduitGrpcSdk; + schemaLookup: SchemaLookup; + subscriptions: RealtimeSubscriptionTracker; + isGloballyEnabled: () => boolean; +}; + +export function registerDatabaseRealtimeSocket( + routingManager: RoutingManager, + options: RealtimeSocketOptions, +) { + const handlers = createSocketHandlers(options); + routingManager.socket( + { + path: '/', + middlewares: options.mode === 'client' ? ['authMiddleware'] : undefined, + }, + { + connect: { handler: handlers.connect }, + disconnect: { handler: handlers.disconnect }, + recovered: { handler: handlers.recovered }, + subscribe: { + params: [TYPE.JSON], + handler: handlers.subscribe, + returnType: new ConduitRouteReturnDefinition('DatabaseRealtimeSubscribe', { + rooms: [TYPE.String], + }), + }, + unsubscribe: { + params: [TYPE.JSON], + handler: handlers.unsubscribe, + returnType: new ConduitRouteReturnDefinition('DatabaseRealtimeUnsubscribe', { + rooms: [TYPE.String], + }), + }, + }, + ); +} + +export function createSocketHandlers(options: RealtimeSocketOptions) { + return { + connect: async (): Promise => { + return { event: 'connected', data: { ok: true } }; + }, + disconnect: async (call: ParsedSocketRequest): Promise => { + const reason = call.request.params?.[0]; + if (isRecoverableDisconnect(reason)) { + await options.subscriptions.armRecoverableTtl(call.request.socketId); + } else { + await options.subscriptions.disconnect(call.request.socketId); + } + return { event: 'disconnected', data: { ok: true } }; + }, + recovered: async (call: ParsedSocketRequest): Promise => { + const { leaveRooms } = await restoreAuthorizedSubscriptions({ + socketId: call.request.socketId, + rooms: recoveredRoomsFromRequest(call), + contextSubs: authorizedSubsFromContext(call.request.context as Indexable), + subscriptions: options.subscriptions, + grpcSdk: options.grpcSdk as unknown as AuthorizationSdk, + }); + if (leaveRooms.length > 0) { + return { event: 'leave-room', rooms: leaveRooms }; + } + return { event: 'join-room', rooms: [] }; + }, + subscribe: async (call: ParsedSocketRequest): Promise => { + const rooms = await resolveSubscription(call, options, 'join'); + return { event: 'join-room', rooms }; + }, + unsubscribe: async (call: ParsedSocketRequest): Promise => { + const rooms = await resolveSubscription(call, options, 'leave'); + return { event: 'leave-room', rooms }; + }, + }; +} + +async function resolveSubscription( + call: ParsedSocketRequest, + options: RealtimeSocketOptions, + action: 'join' | 'leave', +): Promise { + if (!options.isGloballyEnabled()) { + throw new RealtimeSubscriptionError( + status.FAILED_PRECONDITION, + 'Live updates are disabled', + ); + } + const payload = parseSubscribeRequest(call.request.params ?? []); + const schemaName = requireSchemaName(payload.schema); + const documentId = optionalDocumentId(payload.documentId); + const meta = assertSchemaAvailable( + options.schemaLookup, + schemaName, + options.mode === 'client', + ); + + if (options.mode === 'admin') { + if (documentId) return [documentRoom(schemaName, documentId)]; + return [schemaRoom(schemaName)]; + } + + const userId = call.request.context?.user?._id; + if (!userId) { + throw new RealtimeSubscriptionError( + status.UNAUTHENTICATED, + 'Authentication required', + ); + } + + if (meta.authorizationEnabled) { + if (!documentId) { + throw new RealtimeSubscriptionError( + status.PERMISSION_DENIED, + 'Document ID is required for authorized schemas', + ); + } + const allowed = await canReadDocument( + options.grpcSdk as unknown as AuthorizationSdk, + schemaName, + documentId, + userId, + ); + if (!allowed) { + throw new RealtimeSubscriptionError( + status.PERMISSION_DENIED, + 'Read permission denied', + ); + } + if (action === 'join') { + await options.subscriptions.addAuthorizedDocument( + call.request.socketId, + schemaName, + documentId, + userId, + ); + persistAuthorizedSubOnContext( + call.request.context as Indexable, + schemaName, + documentId, + userId, + ); + } else { + await options.subscriptions.removeAuthorizedDocument( + call.request.socketId, + schemaName, + documentId, + userId, + ); + removeAuthorizedSubFromContext( + call.request.context as Indexable, + schemaName, + documentId, + userId, + ); + } + return [authorizedDocumentRoom(schemaName, documentId, userId)]; + } + + if (documentId) return [documentRoom(schemaName, documentId)]; + return [schemaRoom(schemaName)]; +} diff --git a/modules/database/src/realtime/status.ts b/modules/database/src/realtime/status.ts new file mode 100644 index 000000000..b1da21b03 --- /dev/null +++ b/modules/database/src/realtime/status.ts @@ -0,0 +1,74 @@ +import type { RealtimeStatus, RealtimeStatusCode } from './types.js'; + +export type RealtimeStatusInput = { + engine: string; + enabled: boolean; + topologySupported: boolean; + topologyMessage?: string; + activeSchemaCount: number; + streamState: RealtimeStatusCode; + lastEventAt?: string; + lastError?: string; + socketsEnabled?: boolean; +}; + +export function buildRealtimeStatus(input: RealtimeStatusInput): RealtimeStatus { + if (input.engine !== 'MongoDB') { + return { + status: 'unsupported', + engine: input.engine, + activeSchemaCount: 0, + message: 'Live updates require MongoDB', + }; + } + if (!input.enabled) { + return { + status: 'disabled', + engine: input.engine, + activeSchemaCount: input.activeSchemaCount, + }; + } + if (!input.topologySupported) { + return { + status: 'idle', + engine: input.engine, + activeSchemaCount: input.activeSchemaCount, + message: + input.topologyMessage ?? + 'A replica set or sharded MongoDB deployment is required for live updates', + }; + } + if (input.socketsEnabled === false) { + return { + status: 'idle', + engine: input.engine, + activeSchemaCount: input.activeSchemaCount, + message: + 'Enable Admin socket transport to consume live updates in the control panel', + lastEventAt: input.lastEventAt, + }; + } + if (input.activeSchemaCount === 0) { + return { + status: 'idle', + engine: input.engine, + activeSchemaCount: 0, + message: 'No schemas have live updates enabled', + }; + } + if (input.streamState === 'degraded') { + return { + status: 'degraded', + engine: input.engine, + activeSchemaCount: input.activeSchemaCount, + lastEventAt: input.lastEventAt, + message: input.lastError ?? 'Change stream is retrying after an error', + }; + } + return { + status: input.streamState, + engine: input.engine, + activeSchemaCount: input.activeSchemaCount, + lastEventAt: input.lastEventAt, + }; +} diff --git a/modules/database/src/realtime/subscriptions.ts b/modules/database/src/realtime/subscriptions.ts new file mode 100644 index 000000000..bc54e8efd --- /dev/null +++ b/modules/database/src/realtime/subscriptions.ts @@ -0,0 +1,136 @@ +export const RECOVERY_REDIS_TTL_SECONDS = 120; + +export type SubscriptionStore = { + sadd(key: string, ...members: string[]): Promise; + srem(key: string, ...members: string[]): Promise; + smembers(key: string): Promise; + scard(key: string): Promise; + del(...keys: string[]): Promise; + expire(key: string, seconds: number): Promise; + persist(key: string): Promise; +}; + +function socketKey(socketId: string): string { + return `realtime:socket:${socketId}`; +} + +function docUsersKey(schema: string, documentId: string): string { + return `realtime:doc:${schema}:${documentId}`; +} + +function userDocSocketsKey(schema: string, documentId: string, userId: string): string { + return `realtime:userdoc:${schema}:${documentId}:${userId}`; +} + +function subscriptionRecord(schema: string, documentId: string, userId: string): string { + return JSON.stringify({ schema, documentId, userId }); +} + +export class RealtimeSubscriptionTracker { + constructor(private readonly store: SubscriptionStore) {} + + async addAuthorizedDocument( + socketId: string, + schema: string, + documentId: string, + userId: string, + ): Promise { + const socket = socketKey(socketId); + const userDoc = userDocSocketsKey(schema, documentId, userId); + const docUsers = docUsersKey(schema, documentId); + await this.store.sadd(socket, subscriptionRecord(schema, documentId, userId)); + await this.store.sadd(userDoc, socketId); + await this.store.sadd(docUsers, userId); + await this.store.persist(socket); + await this.store.persist(userDoc); + await this.store.persist(docUsers); + } + + async removeAuthorizedDocument( + socketId: string, + schema: string, + documentId: string, + userId: string, + ): Promise { + await this.store.srem( + socketKey(socketId), + subscriptionRecord(schema, documentId, userId), + ); + await this.store.srem(userDocSocketsKey(schema, documentId, userId), socketId); + const remaining = await this.store.scard( + userDocSocketsKey(schema, documentId, userId), + ); + if (remaining === 0) { + await this.store.srem(docUsersKey(schema, documentId), userId); + await this.store.del(userDocSocketsKey(schema, documentId, userId)); + } + } + + async listUsers(schema: string, documentId: string): Promise { + return this.store.smembers(docUsersKey(schema, documentId)); + } + + async removeUser(schema: string, documentId: string, userId: string): Promise { + await this.store.srem(docUsersKey(schema, documentId), userId); + await this.store.del(userDocSocketsKey(schema, documentId, userId)); + } + + async disconnect(socketId: string): Promise { + const records = await this.store.smembers(socketKey(socketId)); + for (const record of records) { + try { + const parsed = JSON.parse(record) as { + schema: string; + documentId: string; + userId: string; + }; + await this.removeAuthorizedDocument( + socketId, + parsed.schema, + parsed.documentId, + parsed.userId, + ); + } catch { + // ignore malformed records + } + } + await this.store.del(socketKey(socketId)); + } + + async armRecoverableTtl( + socketId: string, + ttlSeconds: number = RECOVERY_REDIS_TTL_SECONDS, + ): Promise { + const socket = socketKey(socketId); + const records = await this.store.smembers(socket); + await this.store.expire(socket, ttlSeconds); + for (const record of records) { + try { + const parsed = JSON.parse(record) as { + schema: string; + documentId: string; + userId: string; + }; + const userDoc = userDocSocketsKey( + parsed.schema, + parsed.documentId, + parsed.userId, + ); + const others = (await this.store.smembers(userDoc)).filter(id => id !== socketId); + if (others.length > 0) continue; + await this.store.expire(userDoc, ttlSeconds); + const users = await this.store.smembers( + docUsersKey(parsed.schema, parsed.documentId), + ); + if (users.length <= 1) { + await this.store.expire( + docUsersKey(parsed.schema, parsed.documentId), + ttlSeconds, + ); + } + } catch { + // ignore malformed records + } + } + } +} diff --git a/modules/database/src/realtime/topology.ts b/modules/database/src/realtime/topology.ts new file mode 100644 index 000000000..a1ab72d71 --- /dev/null +++ b/modules/database/src/realtime/topology.ts @@ -0,0 +1,25 @@ +export type TopologyResult = { + supported: boolean; + message?: string; +}; + +export function topologyFromHello( + hello: + | { + setName?: string; + msg?: string; + } + | null + | undefined, +): TopologyResult { + if (!hello) { + return { supported: false, message: 'Unable to determine MongoDB topology' }; + } + if (hello.msg === 'isdbgrid' || Boolean(hello.setName)) { + return { supported: true }; + } + return { + supported: false, + message: 'A replica set or sharded MongoDB deployment is required for live updates', + }; +} diff --git a/modules/database/src/realtime/types.ts b/modules/database/src/realtime/types.ts new file mode 100644 index 000000000..004523198 --- /dev/null +++ b/modules/database/src/realtime/types.ts @@ -0,0 +1,51 @@ +export const DATABASE_CHANGE_EVENT_VERSION = 1 as const; + +export const DATABASE_CHANGE_OPERATIONS = [ + 'insert', + 'update', + 'replace', + 'delete', +] as const; + +export type DatabaseChangeOperation = (typeof DATABASE_CHANGE_OPERATIONS)[number]; + +export type DatabaseChangeEvent = { + version: typeof DATABASE_CHANGE_EVENT_VERSION; + operation: DatabaseChangeOperation; + schema: string; + documentId: string; + occurredAt: string; +}; + +export type RealtimeStatusCode = + 'unsupported' | 'disabled' | 'idle' | 'starting' | 'live' | 'degraded'; + +export type RealtimeStatus = { + status: RealtimeStatusCode; + engine: string; + activeSchemaCount: number; + lastEventAt?: string; + message?: string; +}; + +export type RebacDecision = 'allow' | 'deny' | 'unavailable'; + +export type OptedInSchema = { + name: string; + collectionName: string; + authorizationEnabled: boolean; + cmsReadEnabled: boolean; +}; + +export type SubscribeRequest = { + schema?: unknown; + documentId?: unknown; +}; + +export type ChangeStreamLike = { + on( + event: 'change' | 'error' | 'close' | 'end', + listener: (...args: unknown[]) => void, + ): void; + close(): Promise | void; +}; diff --git a/modules/database/src/realtime/watchPipeline.ts b/modules/database/src/realtime/watchPipeline.ts new file mode 100644 index 000000000..8ca888262 --- /dev/null +++ b/modules/database/src/realtime/watchPipeline.ts @@ -0,0 +1,33 @@ +const DATA_OPERATIONS = ['insert', 'update', 'replace', 'delete'] as const; +const CONTROL_OPERATIONS = ['drop', 'rename', 'invalidate', 'dropDatabase'] as const; + +export const WATCH_RESTART_OPERATIONS = new Set(CONTROL_OPERATIONS); + +export type WatchPipeline = Record[]; + +export function buildWatchPipeline(collectionNames: string[]): WatchPipeline { + return [ + { + $match: { + $or: [ + { + operationType: { $in: [...DATA_OPERATIONS] }, + 'ns.coll': { $in: collectionNames }, + }, + { operationType: { $in: [...CONTROL_OPERATIONS] } }, + ], + }, + }, + { + $project: { + fullDocument: 0, + updateDescription: 0, + fullDocumentBeforeChange: 0, + }, + }, + ]; +} + +export function optedInCollectionsKey(collectionNames: string[]): string { + return [...collectionNames].sort().join('\0'); +} diff --git a/modules/database/src/routes/index.ts b/modules/database/src/routes/index.ts index 2c09cf237..1d4002a74 100644 --- a/modules/database/src/routes/index.ts +++ b/modules/database/src/routes/index.ts @@ -13,6 +13,7 @@ import { import { DatabaseAdapter } from '../adapters/DatabaseAdapter.js'; import { MongooseSchema } from '../adapters/mongoose-adapter/MongooseSchema.js'; import { SequelizeSchema } from '../adapters/sequelize-adapter/SequelizeSchema.js'; +import type { RealtimeService } from '../realtime/index.js'; export class DatabaseRoutes { private readonly handlers: CmsHandlers; @@ -34,6 +35,7 @@ export class DatabaseRoutes { readonly server: GrpcServer, private readonly database: DatabaseAdapter, private readonly grpcSdk: ConduitGrpcSdk, + private readonly realtimeService?: RealtimeService, ) { this.handlers = new CmsHandlers(grpcSdk, database); this._routingManager = new RoutingManager(this.grpcSdk.router!, server); @@ -55,7 +57,6 @@ export class DatabaseRoutes { } requestRefresh() { - if (this.crudRoutes.length === 0 && this.customRoutes.length === 0) return; this._scheduleTimeout(); } @@ -86,6 +87,7 @@ export class DatabaseRoutes { this.crudRoutes.concat(this.customRoutes).forEach(route => { this._routingManager.route(route.input, route.returnType, route.handler); }); + this.realtimeService?.registerClient(this._routingManager); this._routingManager .registerRoutes() .then(() => { diff --git a/modules/database/src/utils/SchemaConverter.ts b/modules/database/src/utils/SchemaConverter.ts index 2e099ed57..f482f4936 100644 --- a/modules/database/src/utils/SchemaConverter.ts +++ b/modules/database/src/utils/SchemaConverter.ts @@ -38,6 +38,9 @@ export namespace SchemaConverter { authorization?: { enabled?: boolean; }; + realtime?: { + enabled?: boolean; + }; permissions?: { extendable?: boolean; canCreate?: boolean; @@ -130,6 +133,9 @@ export namespace SchemaConverter { defaults.conduit!.authorization?.enabled ?? false, }; + modelOptions.conduit.realtime = { + enabled: opts.realtime?.enabled ?? existing?.realtime?.enabled ?? false, + }; modelOptions.conduit.permissions = { extendable: explicit.permissions?.extendable !== undefined diff --git a/modules/database/src/utils/__tests__/validate-model-options.test.ts b/modules/database/src/utils/__tests__/validate-model-options.test.ts new file mode 100644 index 000000000..a832255e5 --- /dev/null +++ b/modules/database/src/utils/__tests__/validate-model-options.test.ts @@ -0,0 +1,37 @@ +import { describe, expect, it } from '@jest/globals'; +import { validateSchemaInput } from '../utilities.js'; + +const baseConduit = { + cms: { enabled: true }, + authorization: { enabled: false }, + permissions: { + extendable: true, + canCreate: true, + canModify: 'Everything' as const, + canDelete: true, + }, +}; + +describe('validateSchemaInput modelOptions.conduit', () => { + it('allows conduit.realtime used by live document updates', () => { + expect(() => + validateSchemaInput('TestSchema', undefined, { + conduit: { + ...baseConduit, + realtime: { enabled: true }, + }, + }), + ).not.toThrow(); + }); + + it('rejects an unknown conduit field', () => { + expect(() => + validateSchemaInput('TestSchema', undefined, { + conduit: { + ...baseConduit, + notARealField: true, + }, + }), + ).toThrow(/fields allowed inside 'conduit' field/); + }); +}); diff --git a/modules/database/src/utils/utilities.ts b/modules/database/src/utils/utilities.ts index ce6cd7328..f3e98a292 100644 --- a/modules/database/src/utils/utilities.ts +++ b/modules/database/src/utils/utilities.ts @@ -183,10 +183,11 @@ function validateModelOptions(modelOptions: ConduitSchemaOptions) { conduitKey !== 'cms' && conduitKey !== 'permissions' && conduitKey !== 'authorization' && - conduitKey !== 'imported' + conduitKey !== 'imported' && + conduitKey !== 'realtime' ) throw new Error( - "Only 'cms', 'permissions', 'authorization', 'imported', and 'readPreference' fields allowed inside 'conduit' field", + "Only 'cms', 'permissions', 'authorization', 'imported', 'readPreference', and 'realtime' fields allowed inside 'conduit' field", ); if (conduitKey === 'imported') { if (!isBoolean(modelOptions.conduit!.imported)) diff --git a/packages/core/package.json b/packages/core/package.json index 0636f4ce2..764e285ab 100644 --- a/packages/core/package.json +++ b/packages/core/package.json @@ -12,6 +12,7 @@ "start": "node dist/bin/www.js", "start:bundle": "node bundle/index.js", "lint": "eslint src", + "test": "npx tsc -p tsconfig.test.json && node --test dist-test/admin/realtime/*.test.js", "prebuild": "npm run generateTypes", "build": "rimraf dist && tsc", "generateTypes": "sh build.sh", diff --git a/packages/core/src/admin/AdminModule.ts b/packages/core/src/admin/AdminModule.ts index 3f18ea069..4f4574571 100644 --- a/packages/core/src/admin/AdminModule.ts +++ b/packages/core/src/admin/AdminModule.ts @@ -14,6 +14,7 @@ import { PatchRouteMiddlewaresRequest, RegisterAdminRouteRequest, RegisterAdminRouteRequest_PathDefinition, + SocketPushRequest, } from '../interfaces/index.js'; import { hashPassword } from './utils/auth.js'; import AdminConfigRawSchema from './config/index.js'; @@ -33,6 +34,7 @@ import { ConduitSocket, grpcToConduitRoute, RouteT, + SocketPush, } from '@conduitplatform/hermes'; import convict from 'convict'; import { NextFunction, Request, Response } from 'express'; @@ -48,6 +50,7 @@ import { getReadinessMiddleware } from '../health/readinessMiddleware.js'; import { ReadinessService } from '../health/ReadinessService.js'; import type { CoreHealthProvider } from '../health/types.js'; import { stripUndeclaredConfigParams } from '../utils/stripUndeclaredConfigParams.js'; +import { adminSocketNamespace } from './realtime/namespace.js'; export default class AdminModule { grpcSdk: ConduitGrpcSdk; @@ -93,6 +96,7 @@ export default class AdminModule { adminRoutes.createApiTokenRoute(), adminRoutes.getApiTokensRoute(), adminRoutes.deleteApiTokenRoute(), + adminRoutes.getRealtimeTicketRoute(), configRoutes.getModulesRoute(), adminRoutes.getStateExportRoute(this), adminRoutes.getStateImportRoute(this), @@ -131,6 +135,7 @@ export default class AdminModule { { registerAdminRoute: this.registerAdminRoute.bind(this), patchRouteMiddlewares: this.patchRouteMiddlewares.bind(this), + socketPush: this.socketPush.bind(this), }, ); } @@ -238,6 +243,27 @@ export default class AdminModule { callback(null, {}); } + async socketPush( + call: GrpcRequest, + callback: GrpcCallback>, + ) { + try { + const moduleName = call.metadata?.get('module-name')?.[0] as string | undefined; + const socketData: SocketPush = { + event: call.request.event, + data: call.request.data ? JSON.parse(call.request.data) : undefined, + receivers: call.request.receivers, + rooms: call.request.rooms, + namespace: adminSocketNamespace(moduleName), + }; + await this._router.socketPush(socketData); + } catch (err) { + ConduitGrpcSdk.Logger.error(err as Error); + return callback({ code: status.INTERNAL, message: 'Well that failed :/' }); + } + callback(null, {}); + } + registerRoute(route: ConduitRoute): void { this._sdkRoutes.push(route); this._router.registerConduitRoute(route); @@ -288,6 +314,16 @@ export default class AdminModule { `New admin route registered: ${r.input.action} ${r.input.path} handler url: ${url}`, ); this._router.registerConduitRoute(r); + } else if (r instanceof ConduitMiddleware) { + ConduitGrpcSdk.Logger.log( + `New admin middleware registered: ${r.input.path} handler url: ${url}`, + ); + this._router.registerRouteMiddleware(r, url); + } else if (r instanceof ConduitSocket) { + ConduitGrpcSdk.Logger.log( + `New admin socket registered: ${r.input.path} handler url: ${url}`, + ); + this._router.registerConduitSocket(r); } }); // @ts-ignore diff --git a/packages/core/src/admin/middleware/Admin.middleware.ts b/packages/core/src/admin/middleware/Admin.middleware.ts index 51c811987..4fb743246 100644 --- a/packages/core/src/admin/middleware/Admin.middleware.ts +++ b/packages/core/src/admin/middleware/Admin.middleware.ts @@ -2,7 +2,7 @@ import { NextFunction, Response } from 'express'; import { isNil } from 'lodash-es'; // Removed ConduitCommons import - now using configManager directly import { isDev } from '../utils/middleware.js'; -import { ConduitRequest } from '@conduitplatform/hermes'; +import { ConduitRequest, isSocketHandshake } from '@conduitplatform/hermes'; import { ConduitGrpcSdk } from '@conduitplatform/grpc-sdk'; export function getAdminMiddleware(configManager: any) { @@ -28,9 +28,14 @@ export function getAdminMiddleware(configManager: any) { ) { return next(); } - // Allow API tokens (cdt_*) to bypass masterkey; Auth.middleware will validate the token + // Allow API tokens (cdt_*) and Socket.IO handshakes with a Bearer token to + // skip masterkey. Auth.middleware still validates the credential. const authHeader = req.headers.authorization; - if (authHeader && authHeader.startsWith('Bearer cdt_')) { + if ( + typeof authHeader === 'string' && + authHeader.startsWith('Bearer ') && + (authHeader.startsWith('Bearer cdt_') || isSocketHandshake(req)) + ) { return next(); } const masterKey = req.headers.masterkey; diff --git a/packages/core/src/admin/middleware/Auth.middleware.ts b/packages/core/src/admin/middleware/Auth.middleware.ts index 4c883244d..9bac68d99 100644 --- a/packages/core/src/admin/middleware/Auth.middleware.ts +++ b/packages/core/src/admin/middleware/Auth.middleware.ts @@ -7,6 +7,7 @@ import { isDev } from '../utils/middleware.js'; import { ConduitRequest } from '@conduitplatform/hermes'; import { gql } from 'graphql-tag'; import { ConfigController } from '@conduitplatform/module-tools'; +import { realtimeTicketForbiddenOnHttp } from '../realtime/ticket.js'; const excludedRestRoutes = ['/ready', '/live', '/login', '/config/modules']; const excludedGqlOperations = [ @@ -113,6 +114,10 @@ async function handleJwtToken( } const { id } = decoded; + if (realtimeTicketForbiddenOnHttp(decoded, req)) { + res.status(401).json({ error: 'Realtime ticket cannot be used for HTTP requests' }); + return; + } if (decoded.twoFaRequired && req.path !== '/verify-twofa') { res.status(401).json({ error: 'Two FA required' }); return; diff --git a/packages/core/src/admin/realtime/namespace.test.ts b/packages/core/src/admin/realtime/namespace.test.ts new file mode 100644 index 000000000..6bfcb3b64 --- /dev/null +++ b/packages/core/src/admin/realtime/namespace.test.ts @@ -0,0 +1,11 @@ +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; +import { adminSocketNamespace } from './namespace.js'; + +describe('adminSocketNamespace', () => { + it('mirrors the Router module namespace contract', () => { + assert.equal(adminSocketNamespace('database'), '/database/'); + assert.throws(() => adminSocketNamespace(''), /module-name/); + assert.throws(() => adminSocketNamespace(undefined), /module-name/); + }); +}); diff --git a/packages/core/src/admin/realtime/namespace.ts b/packages/core/src/admin/realtime/namespace.ts new file mode 100644 index 000000000..ee66e9092 --- /dev/null +++ b/packages/core/src/admin/realtime/namespace.ts @@ -0,0 +1,7 @@ +export function adminSocketNamespace(moduleName: string | undefined | null): string { + const name = typeof moduleName === 'string' ? moduleName.trim() : ''; + if (!name) { + throw new Error('module-name metadata is required for Admin socket push'); + } + return `/${name}/`; +} diff --git a/packages/core/src/admin/realtime/ticket-http.test.ts b/packages/core/src/admin/realtime/ticket-http.test.ts new file mode 100644 index 000000000..589975acc --- /dev/null +++ b/packages/core/src/admin/realtime/ticket-http.test.ts @@ -0,0 +1,97 @@ +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; +import jwt from 'jsonwebtoken'; +import { ConfigController } from '@conduitplatform/module-tools'; +import { isSocketHandshake } from '@conduitplatform/hermes'; +import { getAuthMiddleware } from '../middleware/Auth.middleware.js'; +import { + buildRealtimeTicketClaims, + isRealtimeTicket, + realtimeTicketForbiddenOnHttp, +} from './ticket.js'; + +function mockResponse() { + let statusCode = 0; + let body: unknown; + const res = { + status(code: number) { + statusCode = code; + return this; + }, + json(payload: unknown) { + body = payload; + return this; + }, + }; + return { + res, + get statusCode() { + return statusCode; + }, + get body() { + return body; + }, + }; +} + +describe('realtime ticket HTTP guard', () => { + it('does not treat POST /realtime/ticket as a handshake so a ticket cannot mint another', () => { + const ticket = buildRealtimeTicketClaims('admin-1'); + assert.equal(isRealtimeTicket(ticket), true); + assert.equal(isSocketHandshake({ url: '/realtime/ticket' }), false); + assert.equal( + isSocketHandshake({ url: '/realtime/ticket?EIO=4&transport=polling' }), + false, + ); + assert.equal( + realtimeTicketForbiddenOnHttp(ticket, { url: '/realtime/ticket' }), + true, + ); + assert.equal( + realtimeTicketForbiddenOnHttp(ticket, { + url: '/realtime/ticket?EIO=4&transport=polling', + }), + true, + ); + }); + + it('allows a ticket on Engine.IO handshake polling without sid', () => { + const ticket = buildRealtimeTicketClaims('admin-1'); + const handshake = { url: '/realtime/?EIO=4&transport=polling' }; + assert.equal(isSocketHandshake(handshake), true); + assert.equal(realtimeTicketForbiddenOnHttp(ticket, handshake), false); + }); + + it('Auth middleware returns 401 on POST /realtime/ticket with a realtime ticket', async () => { + const secret = 'ticket-http-test-secret'; + ConfigController.getInstance().config = { auth: { tokenSecret: secret } }; + const token = jwt.sign(buildRealtimeTicketClaims('admin-1'), secret, { + algorithm: 'HS256', + expiresIn: 30, + }); + const middleware = getAuthMiddleware({} as never, { + get: async () => ({ env: 'production' }), + }); + const mock = mockResponse(); + let nextCalled = false; + await middleware( + { + path: '/realtime/ticket', + originalUrl: '/realtime/ticket', + url: '/realtime/ticket', + method: 'POST', + headers: { authorization: `Bearer ${token}` }, + conduit: {}, + } as never, + mock.res as never, + () => { + nextCalled = true; + }, + ); + assert.equal(nextCalled, false); + assert.equal(mock.statusCode, 401); + assert.deepEqual(mock.body, { + error: 'Realtime ticket cannot be used for HTTP requests', + }); + }); +}); diff --git a/packages/core/src/admin/realtime/ticket.test.ts b/packages/core/src/admin/realtime/ticket.test.ts new file mode 100644 index 000000000..e0d66fff1 --- /dev/null +++ b/packages/core/src/admin/realtime/ticket.test.ts @@ -0,0 +1,26 @@ +import assert from 'node:assert/strict'; +import { describe, it } from 'node:test'; +import { + ADMIN_REALTIME_AUDIENCE, + buildRealtimeTicketClaims, + isRealtimeTicket, +} from './ticket.js'; + +describe('admin realtime tickets', () => { + it('scopes claims to the realtime audience', () => { + const claims = buildRealtimeTicketClaims('admin-1'); + assert.equal(claims.id, 'admin-1'); + assert.equal(claims.aud, ADMIN_REALTIME_AUDIENCE); + assert.equal(isRealtimeTicket(claims), true); + }); + + it('rejects ordinary admin JWTs and missing payloads', () => { + assert.equal(isRealtimeTicket({ id: 'admin-1' }), false); + assert.equal(isRealtimeTicket({ id: 'admin-1', aud: 'other' }), false); + assert.equal(isRealtimeTicket(null), false); + assert.equal( + isRealtimeTicket({ id: 'admin-1', aud: [ADMIN_REALTIME_AUDIENCE, 'extra'] }), + true, + ); + }); +}); diff --git a/packages/core/src/admin/realtime/ticket.ts b/packages/core/src/admin/realtime/ticket.ts new file mode 100644 index 000000000..e797f5966 --- /dev/null +++ b/packages/core/src/admin/realtime/ticket.ts @@ -0,0 +1,35 @@ +import { isSocketHandshake } from '@conduitplatform/hermes'; + +export const ADMIN_REALTIME_AUDIENCE = 'admin-realtime'; +export const ADMIN_REALTIME_TICKET_TTL_SECONDS = 30; + +export type RealtimeTicketClaims = { + id: string; + aud?: string | string[]; + twoFaRequired?: boolean; +}; + +export function buildRealtimeTicketClaims(adminId: string): RealtimeTicketClaims { + return { + id: adminId, + aud: ADMIN_REALTIME_AUDIENCE, + }; +} + +export function isRealtimeTicket( + decoded: RealtimeTicketClaims | null | undefined, +): boolean { + if (!decoded) return false; + const audience = decoded.aud; + if (Array.isArray(audience)) { + return audience.includes(ADMIN_REALTIME_AUDIENCE); + } + return audience === ADMIN_REALTIME_AUDIENCE; +} + +export function realtimeTicketForbiddenOnHttp( + decoded: RealtimeTicketClaims | null | undefined, + req: { url?: string; originalUrl?: string }, +): boolean { + return isRealtimeTicket(decoded) && !isSocketHandshake(req); +} diff --git a/packages/core/src/admin/routes/RealtimeTicket.route.ts b/packages/core/src/admin/routes/RealtimeTicket.route.ts new file mode 100644 index 000000000..4a1b8e11e --- /dev/null +++ b/packages/core/src/admin/routes/RealtimeTicket.route.ts @@ -0,0 +1,51 @@ +import { isNil } from 'lodash-es'; +import { ConduitRoute } from '@conduitplatform/hermes'; +import { + ConduitError, + ConduitRouteActions, + ConduitRouteParameters, + ConduitRouteReturnDefinition, +} from '@conduitplatform/grpc-sdk'; +import { + ConduitNumber, + ConduitString, + ConfigController, +} from '@conduitplatform/module-tools'; +import { signToken } from '../utils/auth.js'; +import { + ADMIN_REALTIME_TICKET_TTL_SECONDS, + buildRealtimeTicketClaims, +} from '../realtime/ticket.js'; + +export function getRealtimeTicketRoute() { + return new ConduitRoute( + { + path: '/realtime/ticket', + action: ConduitRouteActions.POST, + mcp: false, + description: + 'Issue a short-lived token for Admin Socket.IO handshakes. The token cannot be used for REST or GraphQL.', + }, + new ConduitRouteReturnDefinition('RealtimeTicket', { + token: ConduitString.Required, + expiresIn: ConduitNumber.Required, + }), + async (req: ConduitRouteParameters) => { + const admin = req.context?.admin; + if (isNil(admin) || isNil(admin._id)) { + throw new ConduitError('UNAUTHORIZED', 401, 'Authentication required'); + } + const adminId = admin._id.toString(); + const authConfig = ConfigController.getInstance().config.auth; + const token = signToken( + buildRealtimeTicketClaims(adminId), + authConfig.tokenSecret, + ADMIN_REALTIME_TICKET_TTL_SECONDS, + ); + return { + token, + expiresIn: ADMIN_REALTIME_TICKET_TTL_SECONDS, + }; + }, + ); +} diff --git a/packages/core/src/admin/routes/index.ts b/packages/core/src/admin/routes/index.ts index 9bed86a70..f7c422e2c 100644 --- a/packages/core/src/admin/routes/index.ts +++ b/packages/core/src/admin/routes/index.ts @@ -20,3 +20,4 @@ export * from './ToggleTwoFa.route.js'; export * from './VerifyQrCode.route.js'; export * from './VerifyTwoFa.route.js'; export * from './ApiTokens.route.js'; +export * from './RealtimeTicket.route.js'; diff --git a/packages/core/src/core.proto b/packages/core/src/core.proto index 527d24f35..858dfa406 100644 --- a/packages/core/src/core.proto +++ b/packages/core/src/core.proto @@ -21,6 +21,14 @@ service Config { service Admin { rpc RegisterAdminRoute (RegisterAdminRouteRequest) returns (google.protobuf.Empty); rpc PatchRouteMiddlewares(PatchRouteMiddlewaresRequest) returns (google.protobuf.Empty); + rpc SocketPush (SocketPushRequest) returns (google.protobuf.Empty); +} + +message SocketPushRequest { + string event = 1; + optional string data = 2; + repeated string receivers = 3; + repeated string rooms = 4; } message RegisterAdminRouteRequest { diff --git a/packages/core/tsconfig.json b/packages/core/tsconfig.json index f4c4ef779..eb15cc5a9 100644 --- a/packages/core/tsconfig.json +++ b/packages/core/tsconfig.json @@ -67,5 +67,5 @@ "forceConsistentCasingInFileNames": true /* Disallow inconsistently-cased references to the same file. */ }, "include": ["src/**/*"], - "exclude": ["tests", "node_modules", "dist", "bundle", "tsup.config.ts"] + "exclude": ["tests", "node_modules", "dist", "bundle", "tsup.config.ts", "**/*.test.ts"] } diff --git a/packages/core/tsconfig.test.json b/packages/core/tsconfig.test.json new file mode 100644 index 000000000..8559c14bf --- /dev/null +++ b/packages/core/tsconfig.test.json @@ -0,0 +1,16 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "outDir": "./dist-test", + "rootDir": "./src", + "declaration": false, + "sourceMap": false, + "types": ["node"] + }, + "include": [ + "src/admin/realtime/ticket.ts", + "src/admin/realtime/namespace.ts", + "src/admin/realtime/*.test.ts" + ], + "exclude": ["node_modules", "dist", "bundle"] +}