feat(database): stream live document updates over admin sockets

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

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:

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'

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

docker/mongo/keyfile

@@ -0,0 +1 @@
+ConduitLocalDevMongoReplicaSetKeyFileDoNotUseInProductionReplaceBeforeAnyRealDeployment0123456789abcdefghijklmnopqrstuvwxyz

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;
   };

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<typeof AdminDefinition> {
@@ -28,6 +29,10 @@ export class Admin extends ConduitModule<typeof AdminDefinition> {
     return this.client!.registerAdminRoute(request);
   }
 
+  socketPush(data: SocketPushRequest) {
+    return this.client!.socketPush(data);
+  }
+
   patchRouteMiddlewares(
     path: string,
     action: ConduitRouteActions,

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 => {

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(

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<string, ConduitRoute> = new Map();
+  /** Sockets registered before Socket.IO starts; replayed in initSockets. */
+  private readonly _conduitSocketsByPath: Map<string, ConduitSocket> = 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);
   }
 

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;
   }
 }
 

modules/database/README.mdx

@@ -47,13 +47,35 @@ 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                                              |
 | :---------------: | :------------------------------------------------------------------------ | :-------: | :------------------------------------------------------- |
 | `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