feat: add ReBAC event relays and live document updates

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,35 @@ services:
       MONGO_INITDB_DATABASE: 'conduit'
       MONGO_INITDB_ROOT_USERNAME: '${DB_USER:-conduit}'
       MONGO_INITDB_ROOT_PASSWORD: '${DB_PASS:-pass}'
+    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,40 @@ services:
       MONGO_INITDB_ROOT_USERNAME: '${DB_USER:-conduit}'
       MONGO_INITDB_ROOT_PASSWORD: '${DB_PASS:-pass}'
     profiles: ['mongodb']
+    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/package.json

@@ -11,6 +11,7 @@
   "scripts": {
     "prepublish": "npm run build",
     "build": "rimraf dist && tsc",
+    "test": "tsc -p tsconfig.test.json && node --test dist-test/Socket/applySocketGlobalMiddlewares.test.js",
     "publish": "npm publish",
     "postbuild": "copyfiles -u 1 src/*.proto src/**/*.json ./dist/"
   },

libraries/hermes/src/Socket/Socket.ts

@@ -10,10 +10,12 @@ import {
   EventResponse,
   isInstanceOfEventResponse,
   JoinRoomResponse,
+  LeaveRoomResponse,
   SocketPush,
 } from '../interfaces/index.js';
 import ObjectHash from 'object-hash';
 import { ConduitError, ConduitGrpcSdk } from '@conduitplatform/grpc-sdk';
+import { applySocketGlobalMiddlewares } from './applySocketGlobalMiddlewares.js';
 import { buildSocketMiddlewareParams } from './buildSocketMiddlewareParams.js';
 
 export class SocketController extends ConduitRouter {
@@ -56,9 +58,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);
@@ -106,20 +116,18 @@ export class SocketController extends ConduitRouter {
     this._registeredNamespaces.set(namespace, conduitSocket);
 
     const self = this;
-    this.globalMiddlewares.forEach(middleware => {
-      self.io.engine.use((req: any, res: any, next: any) => {
-        req.path = namespace;
-        middleware(req, res, next);
-      });
-    });
     this.io.of(namespace).use((socket, next) => {
-      const context = buildSocketMiddlewareParams(socket);
-      self
-        .checkMiddlewares(context, conduitSocket.input.middlewares)
-        .then(r => {
-          Object.assign(context.context, r);
-          socket.data = context.context;
-          next();
+      applySocketGlobalMiddlewares(socket, self.globalMiddlewares)
+        .then(() => {
+          const context = buildSocketMiddlewareParams(socket);
+          Object.assign(context.context, socket.data);
+          return self
+            .checkMiddlewares(context, conduitSocket.input.middlewares)
+            .then(r => {
+              Object.assign(context.context, r);
+              socket.data = context.context;
+              next();
+            });
         })
         .catch((err: Error | ConduitError) => {
           next(err);
@@ -184,11 +192,13 @@ export class SocketController extends ConduitRouter {
   }
 
   async handleSocketPush(push: SocketPush) {
+    const localOnly = push.localOnly === true;
     if (push.event === 'join-room') {
       if (push.rooms.length === 0) return;
       const filteredSockets = await this.findAndFilterSockets(
         push.receivers,
         push.namespace,
+        localOnly,
       );
       for (const socket of filteredSockets) {
         ConduitGrpcSdk.Logger.info(
@@ -203,6 +213,7 @@ export class SocketController extends ConduitRouter {
         const filteredSockets = await this.findAndFilterSockets(
           push.receivers,
           push.namespace,
+          localOnly,
         );
         for (const socket of filteredSockets) {
           for (const room of push.rooms) {
@@ -221,20 +232,31 @@ export class SocketController extends ConduitRouter {
         ConduitGrpcSdk.Logger.info(
           `Emitting event: ${push.event} to all sockets in namespace: ${push.namespace}`,
         );
-        this.io.of(push.namespace).emit(push.event, push.data);
+        const nsp = this.io.of(push.namespace);
+        if (localOnly) {
+          nsp.local.emit(push.event, push.data);
+        } else {
+          nsp.emit(push.event, push.data);
+        }
       } else {
         if (push.rooms.length !== 0) {
           ConduitGrpcSdk.Logger.info(
             `Emitting event: ${push.event} to rooms: ${push.rooms.join(
               ', ',
             )} in namespace: ${push.namespace}`,
           );
-          this.io.of(push.namespace).to(push.rooms).emit(push.event, push.data);
+          const target = this.io.of(push.namespace).to(push.rooms);
+          if (localOnly) {
+            target.local.emit(push.event, push.data);
+          } else {
+            target.emit(push.event, push.data);
+          }
         }
         if (push.receivers.length !== 0) {
           const filteredSockets = await this.findAndFilterSockets(
             push.receivers,
             push.namespace,
+            localOnly,
           );
           for (const socket of filteredSockets) {
             ConduitGrpcSdk.Logger.info(
@@ -248,7 +270,7 @@ export class SocketController extends ConduitRouter {
   }
 
   private async handleResponse(
-    res: EventResponse | JoinRoomResponse,
+    res: EventResponse | JoinRoomResponse | LeaveRoomResponse,
     socket: Socket,
     namespace: string,
   ) {
@@ -307,8 +329,10 @@ export class SocketController extends ConduitRouter {
   async findAndFilterSockets(
     userIds: string[],
     namespace: string,
+    localOnly: boolean = false,
   ): Promise<RemoteSocket<any, any>[]> {
-    const sockets = await this.io.of(namespace).fetchSockets();
+    const nsp = this.io.of(namespace);
+    const sockets = localOnly ? await nsp.local.fetchSockets() : await nsp.fetchSockets();
     const userIdSet = new Set(userIds);
     return sockets.filter(socket => {
       if (socket.data && socket.data.user) {