From 223335bf4b1f2cdcf719f1aaee943af08e00eb85 Mon Sep 17 00:00:00 2001 From: John Chantzigoulas Date: Fri, 4 Sep 2026 14:43:54 +0300 Subject: [PATCH] feat(router): add ReBAC-protected event-to-socket relays Forward exact Redis bus events as templated Socket.io messages on /events/, with resource-scoped rooms and local-only HA emission. --- libraries/hermes/src/Socket/Socket.ts | 25 +++- libraries/hermes/src/interfaces/Socket.ts | 9 +- libraries/hermes/src/interfaces/SocketPush.ts | 1 + modules/router/README.mdx | 40 +++++ modules/router/package.json | 1 + modules/router/src/Router.ts | 28 +++- modules/router/src/admin/event-relays.ts | 128 ++++++++++++++++ modules/router/src/admin/index.ts | 96 +++++++++++- .../src/event-relays/EventRelayManager.ts | 138 ++++++++++++++++++ .../event-relays/EventRelaySockets.test.ts | 91 ++++++++++++ .../src/event-relays/EventRelaySockets.ts | 96 ++++++++++++ modules/router/src/event-relays/authorize.ts | 101 +++++++++++++ .../router/src/event-relays/channels.test.ts | 32 ++++ modules/router/src/event-relays/channels.ts | 23 +++ modules/router/src/event-relays/constants.ts | 32 ++++ modules/router/src/event-relays/index.ts | 25 ++++ modules/router/src/event-relays/path.test.ts | 39 +++++ modules/router/src/event-relays/path.ts | 49 +++++++ .../router/src/event-relays/process.test.ts | 85 +++++++++++ modules/router/src/event-relays/process.ts | 75 ++++++++++ modules/router/src/event-relays/push.ts | 28 ++++ modules/router/src/event-relays/rooms.test.ts | 29 ++++ modules/router/src/event-relays/rooms.ts | 6 + .../router/src/event-relays/search.test.ts | 34 +++++ modules/router/src/event-relays/search.ts | 26 ++++ .../router/src/event-relays/template.test.ts | 39 +++++ modules/router/src/event-relays/template.ts | 136 +++++++++++++++++ .../src/event-relays/validation.test.ts | 69 +++++++++ modules/router/src/event-relays/validation.ts | 130 +++++++++++++++++ .../src/event-relays/validationError.ts | 6 + modules/router/src/metrics/index.ts | 21 +++ .../router/src/models/EventRelay.schema.ts | 94 ++++++++++++ modules/router/src/models/index.ts | 1 + modules/router/tsconfig.test.json | 24 +++ 34 files changed, 1750 insertions(+), 7 deletions(-) create mode 100644 modules/router/src/admin/event-relays.ts create mode 100644 modules/router/src/event-relays/EventRelayManager.ts create mode 100644 modules/router/src/event-relays/EventRelaySockets.test.ts create mode 100644 modules/router/src/event-relays/EventRelaySockets.ts create mode 100644 modules/router/src/event-relays/authorize.ts create mode 100644 modules/router/src/event-relays/channels.test.ts create mode 100644 modules/router/src/event-relays/channels.ts create mode 100644 modules/router/src/event-relays/constants.ts create mode 100644 modules/router/src/event-relays/index.ts create mode 100644 modules/router/src/event-relays/path.test.ts create mode 100644 modules/router/src/event-relays/path.ts create mode 100644 modules/router/src/event-relays/process.test.ts create mode 100644 modules/router/src/event-relays/process.ts create mode 100644 modules/router/src/event-relays/push.ts create mode 100644 modules/router/src/event-relays/rooms.test.ts create mode 100644 modules/router/src/event-relays/rooms.ts create mode 100644 modules/router/src/event-relays/search.test.ts create mode 100644 modules/router/src/event-relays/search.ts create mode 100644 modules/router/src/event-relays/template.test.ts create mode 100644 modules/router/src/event-relays/template.ts create mode 100644 modules/router/src/event-relays/validation.test.ts create mode 100644 modules/router/src/event-relays/validation.ts create mode 100644 modules/router/src/event-relays/validationError.ts create mode 100644 modules/router/src/models/EventRelay.schema.ts create mode 100644 modules/router/tsconfig.test.json diff --git a/libraries/hermes/src/Socket/Socket.ts b/libraries/hermes/src/Socket/Socket.ts index 7d9b0b09f..74ced4e2b 100644 --- a/libraries/hermes/src/Socket/Socket.ts +++ b/libraries/hermes/src/Socket/Socket.ts @@ -10,6 +10,7 @@ import { EventResponse, isInstanceOfEventResponse, JoinRoomResponse, + LeaveRoomResponse, SocketPush, } from '../interfaces/index.js'; import ObjectHash from 'object-hash'; @@ -184,11 +185,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 +206,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,7 +225,12 @@ 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( @@ -229,12 +238,18 @@ export class SocketController extends ConduitRouter { ', ', )} 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 +263,7 @@ export class SocketController extends ConduitRouter { } private async handleResponse( - res: EventResponse | JoinRoomResponse, + res: EventResponse | JoinRoomResponse | LeaveRoomResponse, socket: Socket, namespace: string, ) { @@ -307,8 +322,10 @@ export class SocketController extends ConduitRouter { async findAndFilterSockets( userIds: string[], namespace: string, + localOnly: boolean = false, ): Promise[]> { - 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) { diff --git a/libraries/hermes/src/interfaces/Socket.ts b/libraries/hermes/src/interfaces/Socket.ts index a477c5c47..204707624 100644 --- a/libraries/hermes/src/interfaces/Socket.ts +++ b/libraries/hermes/src/interfaces/Socket.ts @@ -40,7 +40,14 @@ export type JoinRoomResponse = { rooms: string[]; }; -export type ConduitSocketHandlerResponse = Promise; +export type LeaveRoomResponse = { + event: 'leave-room'; + rooms: string[]; +}; + +export type ConduitSocketHandlerResponse = Promise< + EventResponse | JoinRoomResponse | LeaveRoomResponse +>; export type ConduitSocketEventHandler = ( request: ConduitSocketParameters, diff --git a/libraries/hermes/src/interfaces/SocketPush.ts b/libraries/hermes/src/interfaces/SocketPush.ts index 2a2ca5e10..8bbabbc85 100644 --- a/libraries/hermes/src/interfaces/SocketPush.ts +++ b/libraries/hermes/src/interfaces/SocketPush.ts @@ -4,4 +4,5 @@ export interface SocketPush { receivers: string[]; rooms: string[]; namespace: string; + localOnly?: boolean; } diff --git a/modules/router/README.mdx b/modules/router/README.mdx index b1dff474d..c896f2cee 100644 --- a/modules/router/README.mdx +++ b/modules/router/README.mdx @@ -77,3 +77,43 @@ If you sent any other string that those provided conduit will not utilize cachin The number that you specify also sets the expiry time of conduit's cache. All caching takes place, AFTER middleware execution - cacheControl?: string; + +## Event Relays + +Event Relays map an **exact** Redis bus channel to a Socket.io event on the dedicated `/events/` namespace. +They are configured through the Admin API (`/router/event-relays`) and are **ephemeral**: Redis pub/sub is not replayed, and missed events are not stored. + +Each relay stores a ReBAC check (`permission` on `resourceType:resourceId`). Clients never choose a raw room name. Optional `notes` describe the relay for operators. + +### Client contract + +```javascript +import { io } from 'socket.io-client'; + +const socket = io(`${SOCKET_BASE_URL}/events/`, { + path: '/realtime', + extraHeaders: { + authorization: `Bearer ${accessToken}`, + }, +}); + +socket.emit('subscribe', relayId, resourceId); +socket.on('order-updated', payload => { + // payload is the rendered JSON template +}); +socket.emit('unsubscribe', relayId, resourceId); +``` + +- Subscribe is allowed only when `User:` has the relay `permission` on `:`. +- The outbound event name is the relay `socketEvent`. +- Placeholders in `messageTemplate` use `{{payload.path}}` against the bus JSON payload. + +### Admin API + +| Method | Path | Description | +|:-------|:-----|:------------| +| `GET` | `/router/event-relays` | Paginated list (`skip`, `limit`, `search`) | +| `GET` | `/router/event-relays/:id` | Single relay | +| `POST` | `/router/event-relays` | Create | +| `PATCH` | `/router/event-relays/:id` | Update | +| `DELETE` | `/router/event-relays/:id` | Delete | diff --git a/modules/router/package.json b/modules/router/package.json index 72670f268..c8d9c12cf 100644 --- a/modules/router/package.json +++ b/modules/router/package.json @@ -23,6 +23,7 @@ "prebuild:bundle": "pnpm --filter @conduitplatform/service-bundle run build", "build:bundle": "rimraf bundle && node ../../libraries/service-bundle/dist/cli.js generate-manifest && tsup && node ../../libraries/service-bundle/dist/cli.js copy-assets && node ../../libraries/service-bundle/dist/cli.js generate-lockfile", "generateTypes": "sh build.sh", + "test": "npx tsc -p tsconfig.test.json && node --test dist-test/event-relays/*.test.js", "build:docker": "docker build -t ghcr.io/conduitplatform/router:latest -f ./Dockerfile ../../ && docker push ghcr.io/conduitplatform/router:latest" }, "license": "ISC", diff --git a/modules/router/src/Router.ts b/modules/router/src/Router.ts index 80f9aa56e..0a3867ed6 100644 --- a/modules/router/src/Router.ts +++ b/modules/router/src/Router.ts @@ -40,6 +40,11 @@ import * as adminRoutes from './admin/routes/index.js'; import metricsSchema from './metrics/index.js'; import { ConfigController, ManagedModule } from '@conduitplatform/module-tools'; import { fileURLToPath } from 'node:url'; +import { + createEventRelayPusher, + createEventsSocket, + EventRelayManager, +} from './event-relays/index.js'; const __filename = fileURLToPath(import.meta.url); const __dirname = path.dirname(__filename); @@ -69,6 +74,8 @@ export default class ConduitDefaultRouter extends ManagedModule { private hasAppliedMiddleware: string[] = []; private _refreshTimeout: NodeJS.Timeout | null = null; private _haInitialized = false; + private eventRelayManager: EventRelayManager; + private eventsSocket?: ConduitSocket; constructor(peerManifestRoot?: string) { super('router', peerManifestRoot); @@ -102,7 +109,16 @@ export default class ConduitDefaultRouter extends ManagedModule { } async onRegister() { - this.adminRouter = new AdminHandlers(this.grpcServer, this.grpcSdk, this); + this.eventRelayManager = new EventRelayManager( + this.grpcSdk, + createEventRelayPusher(data => this._internalRouter.socketPush(data)), + ); + this.adminRouter = new AdminHandlers( + this.grpcServer, + this.grpcSdk, + this, + this.eventRelayManager, + ); this._security = new SecurityModule(this.grpcSdk, this); } @@ -132,8 +148,11 @@ export default class ConduitDefaultRouter extends ManagedModule { } if (config.transports.sockets) { this._internalRouter.initSockets(); + this.registerEventsNamespace(); + await this.eventRelayManager.start(); atLeastOne = true; } else { + await this.eventRelayManager?.stop(); this._internalRouter.stopSockets(); } @@ -336,6 +355,13 @@ export default class ConduitDefaultRouter extends ManagedModule { this._internalRouter.registerConduitRoute(route); } + private registerEventsNamespace() { + if (!this.eventsSocket) { + this.eventsSocket = createEventsSocket(this.grpcSdk, this.eventRelayManager); + } + this._internalRouter.registerConduitSocket(this.eventsSocket); + } + protected registerSchemas(): Promise { const promises = Object.values(models).map(model => { const modelInstance = model.getInstance(this.grpcSdk.database!); diff --git a/modules/router/src/admin/event-relays.ts b/modules/router/src/admin/event-relays.ts new file mode 100644 index 000000000..2e56fbd15 --- /dev/null +++ b/modules/router/src/admin/event-relays.ts @@ -0,0 +1,128 @@ +import { status } from '@grpc/grpc-js'; +import { + GrpcError, + ParsedRouterRequest, + Query, + UnparsedRouterResponse, +} from '@conduitplatform/grpc-sdk'; +import { isNil } from 'lodash-es'; +import { EventRelay } from '../models/index.js'; +import { EventRelayManager } from '../event-relays/EventRelayManager.js'; +import { EventRelayInput, validateEventRelayInput } from '../event-relays/validation.js'; +import { EventRelayValidationError } from '../event-relays/validationError.js'; +import { buildSearchQuery, parsePagination } from '../event-relays/search.js'; + +export class EventRelayAdmin { + constructor(private readonly manager: EventRelayManager) {} + + async listEventRelays(call: ParsedRouterRequest): Promise { + const { skip, limit } = parsePagination( + call.request.params.skip, + call.request.params.limit, + ); + const { search } = call.request.params; + const query = buildSearchQuery(search) as Query; + const relays = await EventRelay.getInstance().findMany(query, { + skip, + limit, + sort: { updatedAt: -1 }, + }); + const count = await EventRelay.getInstance().countDocuments(query); + return { relays, count }; + } + + async getEventRelay(call: ParsedRouterRequest): Promise { + const relay = await EventRelay.getInstance().findOne({ + _id: call.request.params.id, + }); + if (isNil(relay)) { + throw new GrpcError(status.NOT_FOUND, 'Event relay not found'); + } + return relay; + } + + async createEventRelay(call: ParsedRouterRequest): Promise { + const input = parseInput(call.request.params as EventRelayInput); + await assertUniqueName(input.name); + const relay = await EventRelay.getInstance().create({ + ...input, + messageTemplate: input.messageTemplate as EventRelay['messageTemplate'], + }); + await this.manager.notifyChanged(); + return relay; + } + + async patchEventRelay(call: ParsedRouterRequest): Promise { + const existing = await EventRelay.getInstance().findOne({ + _id: call.request.params.id, + }); + if (isNil(existing)) { + throw new GrpcError(status.NOT_FOUND, 'Event relay not found'); + } + + const merged: EventRelayInput = { + name: call.request.params.name ?? existing.name, + notes: + call.request.params.notes === undefined + ? existing.notes + : call.request.params.notes, + active: + call.request.params.active === undefined + ? existing.active + : call.request.params.active, + busEvent: call.request.params.busEvent ?? existing.busEvent, + socketEvent: call.request.params.socketEvent ?? existing.socketEvent, + resourceType: call.request.params.resourceType ?? existing.resourceType, + resourceIdPath: call.request.params.resourceIdPath ?? existing.resourceIdPath, + permission: call.request.params.permission ?? existing.permission, + messageTemplate: + call.request.params.messageTemplate === undefined + ? existing.messageTemplate + : call.request.params.messageTemplate, + }; + const input = parseInput(merged); + if (input.name !== existing.name) { + await assertUniqueName(input.name, existing._id); + } + + const updated = await EventRelay.getInstance().findByIdAndUpdate(existing._id, { + ...input, + messageTemplate: input.messageTemplate as EventRelay['messageTemplate'], + }); + await this.manager.notifyChanged(); + return updated!; + } + + async deleteEventRelay(call: ParsedRouterRequest): Promise { + const existing = await EventRelay.getInstance().findOne({ + _id: call.request.params.id, + }); + if (isNil(existing)) { + throw new GrpcError(status.NOT_FOUND, 'Event relay not found'); + } + await EventRelay.getInstance().deleteOne({ _id: existing._id }); + await this.manager.notifyChanged(); + return { message: 'Event relay deleted' }; + } +} + +function parseInput(params: EventRelayInput): EventRelayInput { + try { + return validateEventRelayInput(params); + } catch (err) { + if (err instanceof EventRelayValidationError) { + throw new GrpcError(status.INVALID_ARGUMENT, err.message); + } + throw err; + } +} + +async function assertUniqueName(name: string, excludeId?: string): Promise { + const existing = await EventRelay.getInstance().findOne({ name }); + if (existing && existing._id !== excludeId) { + throw new GrpcError( + status.ALREADY_EXISTS, + `An event relay named '${name}' already exists`, + ); + } +} diff --git a/modules/router/src/admin/index.ts b/modules/router/src/admin/index.ts index 72065b6f9..0c3231378 100644 --- a/modules/router/src/admin/index.ts +++ b/modules/router/src/admin/index.ts @@ -6,27 +6,34 @@ import { } from '@conduitplatform/grpc-sdk'; import { ConduitBoolean, + ConduitJson, + ConduitNumber, ConduitString, GrpcServer, RoutingManager, } from '@conduitplatform/module-tools'; import { RouterAdmin } from './router.js'; import { SecurityAdmin } from './security.js'; +import { EventRelayAdmin } from './event-relays.js'; import ConduitDefaultRouter from '../Router.js'; -import { Client } from '../models/index.js'; +import { Client, EventRelay } from '../models/index.js'; +import { EventRelayManager } from '../event-relays/EventRelayManager.js'; export class AdminHandlers { private readonly routerAdmin: RouterAdmin; private readonly securityAdmin: SecurityAdmin; + private readonly eventRelayAdmin: EventRelayAdmin; private readonly routingManager: RoutingManager; constructor( private readonly server: GrpcServer, private readonly grpcSdk: ConduitGrpcSdk, private readonly router: ConduitDefaultRouter, + eventRelayManager: EventRelayManager, ) { this.routerAdmin = new RouterAdmin(this.grpcSdk, router); this.securityAdmin = new SecurityAdmin(this.grpcSdk); + this.eventRelayAdmin = new EventRelayAdmin(eventRelayManager); this.routingManager = new RoutingManager(this.grpcSdk.admin, this.server); this.registerAdminRoutes(); } @@ -148,6 +155,93 @@ export class AdminHandlers { new ConduitRouteReturnDefinition('UpdateSecurityClient', Client.name), this.securityAdmin.updateSecurityClient.bind(this.securityAdmin), ); + + this.routingManager.route( + { + path: '/event-relays', + action: ConduitRouteActions.GET, + description: `Returns event-to-socket relays. Delivery is ephemeral Redis pub/sub with no replay.`, + queryParams: { + skip: ConduitNumber.Optional, + limit: ConduitNumber.Optional, + search: ConduitString.Optional, + }, + }, + new ConduitRouteReturnDefinition('GetEventRelays', { + relays: [EventRelay.name], + count: ConduitNumber.Required, + }), + this.eventRelayAdmin.listEventRelays.bind(this.eventRelayAdmin), + ); + this.routingManager.route( + { + path: '/event-relays/:id', + action: ConduitRouteActions.GET, + description: `Returns a single event-to-socket relay.`, + urlParams: { + id: ConduitString.Required, + }, + }, + new ConduitRouteReturnDefinition(EventRelay.name), + this.eventRelayAdmin.getEventRelay.bind(this.eventRelayAdmin), + ); + this.routingManager.route( + { + path: '/event-relays', + action: ConduitRouteActions.POST, + description: `Creates an event-to-socket relay. Clients subscribe on /events/ after a ReBAC check.`, + bodyParams: { + name: ConduitString.Required, + notes: ConduitString.Optional, + active: ConduitBoolean.Optional, + busEvent: ConduitString.Required, + socketEvent: ConduitString.Required, + resourceType: ConduitString.Required, + resourceIdPath: ConduitString.Required, + permission: ConduitString.Required, + messageTemplate: ConduitJson.Required, + }, + }, + new ConduitRouteReturnDefinition('CreateEventRelay', EventRelay.name), + this.eventRelayAdmin.createEventRelay.bind(this.eventRelayAdmin), + ); + this.routingManager.route( + { + path: '/event-relays/:id', + action: ConduitRouteActions.PATCH, + description: `Updates an event-to-socket relay.`, + urlParams: { + id: ConduitString.Required, + }, + bodyParams: { + name: ConduitString.Optional, + notes: ConduitString.Optional, + active: ConduitBoolean.Optional, + busEvent: ConduitString.Optional, + socketEvent: ConduitString.Optional, + resourceType: ConduitString.Optional, + resourceIdPath: ConduitString.Optional, + permission: ConduitString.Optional, + messageTemplate: ConduitJson.Optional, + }, + }, + new ConduitRouteReturnDefinition('PatchEventRelay', EventRelay.name), + this.eventRelayAdmin.patchEventRelay.bind(this.eventRelayAdmin), + ); + this.routingManager.route( + { + path: '/event-relays/:id', + action: ConduitRouteActions.DELETE, + description: `Deletes an event-to-socket relay.`, + urlParams: { + id: ConduitString.Required, + }, + }, + new ConduitRouteReturnDefinition('DeleteEventRelay', { + message: ConduitString.Required, + }), + this.eventRelayAdmin.deleteEventRelay.bind(this.eventRelayAdmin), + ); this.routingManager.registerRoutes(); } } diff --git a/modules/router/src/event-relays/EventRelayManager.ts b/modules/router/src/event-relays/EventRelayManager.ts new file mode 100644 index 000000000..bddae2127 --- /dev/null +++ b/modules/router/src/event-relays/EventRelayManager.ts @@ -0,0 +1,138 @@ +import { ConduitGrpcSdk } from '@conduitplatform/grpc-sdk'; +import { EventRelay } from '../models/index.js'; +import { + EVENT_RELAY_REFRESH_CHANNEL, + EVENT_RELAY_SUBSCRIBER_PREFIX, +} from './constants.js'; +import { groupRelaysByChannel, planChannelSubscriptions } from './channels.js'; +import { buildRelayEmissions, parseBusPayload } from './process.js'; +import { EventRelayPusher } from './push.js'; + +export type { EventRelayPusher } from './push.js'; +export { createEventRelayPusher } from './push.js'; + +export class EventRelayManager { + private readonly relaysByChannel = new Map(); + private readonly subscribedChannels = new Set(); + private started = false; + + constructor( + private readonly grpcSdk: ConduitGrpcSdk, + private readonly push: EventRelayPusher, + ) {} + + async start(): Promise { + if (!this.started) { + this.grpcSdk.bus?.subscribe( + EVENT_RELAY_REFRESH_CHANNEL, + () => { + this.reconcile().catch(err => { + ConduitGrpcSdk.Logger.error(err as Error); + }); + }, + 'router-event-relays-refresh', + ); + this.started = true; + } + await this.reconcile(); + } + + async stop(): Promise { + for (const channel of [...this.subscribedChannels]) { + this.grpcSdk.bus?.unsubscribe(`${EVENT_RELAY_SUBSCRIBER_PREFIX}${channel}`); + this.subscribedChannels.delete(channel); + } + this.relaysByChannel.clear(); + if (this.started) { + this.grpcSdk.bus?.unsubscribe('router-event-relays-refresh'); + this.started = false; + } + } + + async notifyChanged(): Promise { + if (this.started) { + await this.reconcile(); + } + this.grpcSdk.bus?.publish(EVENT_RELAY_REFRESH_CHANNEL, ''); + } + + async reconcile(): Promise { + const relays = await EventRelay.getInstance().findMany({ active: true }); + const next = groupRelaysByChannel(relays); + const { toSubscribe, toUnsubscribe } = planChannelSubscriptions( + this.subscribedChannels, + next.keys(), + ); + + this.relaysByChannel.clear(); + for (const [channel, list] of next) { + this.relaysByChannel.set(channel, list); + } + + for (const channel of toUnsubscribe) { + this.grpcSdk.bus?.unsubscribe(`${EVENT_RELAY_SUBSCRIBER_PREFIX}${channel}`); + this.subscribedChannels.delete(channel); + } + + for (const channel of toSubscribe) { + this.grpcSdk.bus?.subscribe( + channel, + message => this.onBusMessage(channel, message), + `${EVENT_RELAY_SUBSCRIBER_PREFIX}${channel}`, + ); + this.subscribedChannels.add(channel); + } + } + + getActiveRelay(id: string): EventRelay | undefined { + for (const relays of this.relaysByChannel.values()) { + const match = relays.find(relay => relay._id === id); + if (match) return match; + } + return undefined; + } + + private onBusMessage(channel: string, message: string): void { + const relays = this.relaysByChannel.get(channel); + if (!relays || relays.length === 0) { + return; + } + + let payload: unknown; + try { + payload = parseBusPayload(message); + } catch (err) { + ConduitGrpcSdk.Metrics?.increment('event_relays_failed_total'); + ConduitGrpcSdk.Logger.error( + `Event relay failed to parse payload for ${channel}: ${ + err instanceof Error ? err.message : String(err) + }`, + ); + return; + } + + const { emissions, failures } = buildRelayEmissions(relays, payload); + for (const failure of failures) { + ConduitGrpcSdk.Metrics?.increment('event_relays_failed_total'); + ConduitGrpcSdk.Logger.warn( + `Event relay ${failure.relayId} skipped on ${failure.busEvent}: ${failure.reason}`, + ); + } + + for (const emission of emissions) { + this.push(emission.socketEvent, emission.data, [emission.room]).then( + () => { + ConduitGrpcSdk.Metrics?.increment('event_relays_emitted_total'); + }, + err => { + ConduitGrpcSdk.Metrics?.increment('event_relays_failed_total'); + ConduitGrpcSdk.Logger.error( + `Event relay ${emission.relayId} emit failed on ${channel}: ${ + (err as Error).message + }`, + ); + }, + ); + } + } +} diff --git a/modules/router/src/event-relays/EventRelaySockets.test.ts b/modules/router/src/event-relays/EventRelaySockets.test.ts new file mode 100644 index 000000000..19f455b7a --- /dev/null +++ b/modules/router/src/event-relays/EventRelaySockets.test.ts @@ -0,0 +1,91 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { status } from '@grpc/grpc-js'; +import { authorizeRelaySubscription } from './authorize.js'; +import { eventRelayRoom } from './rooms.js'; +import type { RelayLookup } from './authorize.js'; + +function mockSdk(options: { + available?: boolean; + allow?: boolean; + throwOnCan?: boolean; +}) { + return { + isAvailable: () => options.available !== false, + authorization: + options.available === false + ? null + : { + can: async () => { + if (options.throwOnCan) { + throw new Error('authz down'); + } + return { allow: options.allow === true }; + }, + }, + }; +} + +function mockManager(relay: unknown): RelayLookup { + return { + getActiveRelay: () => relay as ReturnType, + }; +} + +const relay = { + _id: 'relay-1', + permission: 'read', + resourceType: 'Order', +}; + +describe('authorizeRelaySubscription', () => { + it('returns the deterministic room when ReBAC allows', async () => { + const room = await authorizeRelaySubscription( + mockSdk({ allow: true }), + mockManager(relay), + 'user-1', + 'relay-1', + 'order-1', + ); + assert.equal(room, eventRelayRoom('relay-1', 'order-1')); + }); + + it('denies when ReBAC returns false', async () => { + await assert.rejects( + () => + authorizeRelaySubscription( + mockSdk({ allow: false }), + mockManager(relay), + 'user-1', + 'relay-1', + 'order-1', + ), + (err: any) => err.code === status.PERMISSION_DENIED, + ); + }); + + it('fails closed when authorization is unavailable', async () => { + await assert.rejects( + () => + authorizeRelaySubscription( + mockSdk({ available: false }), + mockManager(relay), + 'user-1', + 'relay-1', + 'order-1', + ), + (err: any) => err.code === status.UNAVAILABLE, + ); + }); + + it('unsubscribe room matches subscribe room for the same ids', async () => { + const subscribed = await authorizeRelaySubscription( + mockSdk({ allow: true }), + mockManager(relay), + 'user-1', + 'relay-1', + 'order-1', + ); + assert.equal(subscribed, eventRelayRoom('relay-1', 'order-1')); + }); +}); diff --git a/modules/router/src/event-relays/EventRelaySockets.ts b/modules/router/src/event-relays/EventRelaySockets.ts new file mode 100644 index 000000000..32e5fec2a --- /dev/null +++ b/modules/router/src/event-relays/EventRelaySockets.ts @@ -0,0 +1,96 @@ +import { status } from '@grpc/grpc-js'; +import { ConduitGrpcSdk, GrpcError, TYPE } from '@conduitplatform/grpc-sdk'; +import { ConduitSocket, ConduitSocketEvent } from '@conduitplatform/hermes'; +import { EVENTS_NAMESPACE } from './constants.js'; +import { EventRelayManager } from './EventRelayManager.js'; +import { eventRelayRoom } from './rooms.js'; +import { validateResourceId } from './validation.js'; +import { authorizeRelaySubscription, toSubscriptionError } from './authorize.js'; + +export function createEventsSocket( + grpcSdk: ConduitGrpcSdk, + manager: EventRelayManager, +): ConduitSocket { + const events = new Map(); + + events.set('connect', { + name: 'connect', + handler: async () => ({ event: 'join-room', rooms: [] }), + }); + + events.set('disconnect', { + name: 'disconnect', + handler: async () => ({ event: 'leave-room', rooms: [] }), + }); + + events.set('subscribe', { + name: 'subscribe', + params: [TYPE.String, TYPE.String], + handler: async request => { + const userId = request.context?.user?._id as string | undefined; + const [relayId, resourceId] = request.params ?? []; + const room = await authorizeOrThrow(grpcSdk, manager, userId, relayId, resourceId); + return { event: 'join-room', rooms: [room] }; + }, + }); + + events.set('unsubscribe', { + name: 'unsubscribe', + params: [TYPE.String, TYPE.String], + handler: async request => { + const [relayId, resourceId] = request.params ?? []; + if (typeof relayId !== 'string' || relayId.trim() === '') { + throw new GrpcError(status.INVALID_ARGUMENT, 'Relay ID is required'); + } + try { + const validatedResourceId = validateResourceId(resourceId); + return { + event: 'leave-room', + rooms: [eventRelayRoom(relayId, validatedResourceId)], + }; + } catch (err) { + throw toGrpcError(err); + } + }, + }); + + return new ConduitSocket( + { + path: EVENTS_NAMESPACE, + name: 'eventRelays', + description: 'Declarative bus-to-socket event relays', + middlewares: ['authMiddleware'], + }, + events, + ); +} + +async function authorizeOrThrow( + grpcSdk: ConduitGrpcSdk, + manager: EventRelayManager, + userId: string | undefined, + relayId: unknown, + resourceId: unknown, +): Promise { + try { + return await authorizeRelaySubscription( + grpcSdk, + manager, + userId, + relayId, + resourceId, + () => { + ConduitGrpcSdk.Metrics?.increment('event_relay_subscriptions_denied_total'); + }, + ); + } catch (err) { + throw toGrpcError(err); + } +} + +function toGrpcError(err: unknown): GrpcError { + const mapped = toSubscriptionError(err); + return new GrpcError(mapped.code, mapped.message); +} + +export { authorizeRelaySubscription } from './authorize.js'; diff --git a/modules/router/src/event-relays/authorize.ts b/modules/router/src/event-relays/authorize.ts new file mode 100644 index 000000000..852cf5595 --- /dev/null +++ b/modules/router/src/event-relays/authorize.ts @@ -0,0 +1,101 @@ +import { status } from '@grpc/grpc-js'; +import { eventRelayRoom } from './rooms.js'; +import { validateResourceId } from './validation.js'; +import { EventRelayValidationError } from './validationError.js'; + +export class RelaySubscriptionError extends Error { + constructor( + readonly code: number, + message: string, + ) { + super(message); + this.name = 'RelaySubscriptionError'; + } +} + +export type RelaySubscriptionTarget = { + _id: string; + permission: string; + resourceType: string; +}; + +export type RelayAuthorizationSdk = { + isAvailable: (module: string) => boolean; + authorization?: { + can: (request: { + subject: string; + actions: string[]; + resource: string; + }) => Promise<{ allow: boolean }>; + } | null; +}; + +export type RelayLookup = { + getActiveRelay(id: string): RelaySubscriptionTarget | undefined; +}; + +export async function authorizeRelaySubscription( + grpcSdk: RelayAuthorizationSdk, + manager: RelayLookup, + userId: string | undefined, + relayId: unknown, + resourceId: unknown, + onDenied?: () => void, +): Promise { + if (!userId) { + throw new RelaySubscriptionError(status.UNAUTHENTICATED, 'Authentication required'); + } + if (typeof relayId !== 'string' || relayId.trim() === '') { + throw new RelaySubscriptionError(status.INVALID_ARGUMENT, 'Relay ID is required'); + } + + let validatedResourceId: string; + try { + validatedResourceId = validateResourceId(resourceId); + } catch (err) { + throw toSubscriptionError(err); + } + + const relay = manager.getActiveRelay(relayId); + if (!relay) { + throw new RelaySubscriptionError(status.NOT_FOUND, 'Event relay not found'); + } + + if (!grpcSdk.authorization || !grpcSdk.isAvailable('authorization')) { + onDenied?.(); + throw new RelaySubscriptionError(status.UNAVAILABLE, 'Authorization is unavailable'); + } + + let allowed = false; + try { + const decision = await grpcSdk.authorization.can({ + subject: `User:${userId}`, + actions: [relay.permission], + resource: `${relay.resourceType}:${validatedResourceId}`, + }); + allowed = decision.allow; + } catch { + onDenied?.(); + throw new RelaySubscriptionError(status.UNAVAILABLE, 'Authorization check failed'); + } + + if (!allowed) { + onDenied?.(); + throw new RelaySubscriptionError(status.PERMISSION_DENIED, 'Permission denied'); + } + + return eventRelayRoom(relay._id, validatedResourceId); +} + +export function toSubscriptionError(err: unknown): RelaySubscriptionError { + if (err instanceof EventRelayValidationError) { + return new RelaySubscriptionError(status.INVALID_ARGUMENT, err.message); + } + if (err instanceof RelaySubscriptionError) { + return err; + } + return new RelaySubscriptionError( + status.INTERNAL, + err instanceof Error ? err.message : 'Unexpected error', + ); +} diff --git a/modules/router/src/event-relays/channels.test.ts b/modules/router/src/event-relays/channels.test.ts new file mode 100644 index 000000000..9870d0bd9 --- /dev/null +++ b/modules/router/src/event-relays/channels.test.ts @@ -0,0 +1,32 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { groupRelaysByChannel, planChannelSubscriptions } from './channels.js'; + +describe('groupRelaysByChannel', () => { + it('groups multiple relays onto the same exact channel', () => { + const grouped = groupRelaysByChannel([ + { _id: 'a', busEvent: 'orders.paid' }, + { _id: 'b', busEvent: 'orders.paid' }, + { _id: 'c', busEvent: 'orders.shipped' }, + ]); + assert.equal(grouped.get('orders.paid')?.length, 2); + assert.equal(grouped.get('orders.shipped')?.length, 1); + }); +}); + +describe('planChannelSubscriptions', () => { + it('subscribes new channels and unsubscribes removed ones', () => { + const plan = planChannelSubscriptions( + ['orders.paid', 'stale.channel'], + ['orders.paid', 'orders.shipped'], + ); + assert.deepEqual(plan.toSubscribe, ['orders.shipped']); + assert.deepEqual(plan.toUnsubscribe, ['stale.channel']); + }); + + it('is a no-op when the channel set is unchanged', () => { + const plan = planChannelSubscriptions(['orders.paid'], ['orders.paid']); + assert.deepEqual(plan.toSubscribe, []); + assert.deepEqual(plan.toUnsubscribe, []); + }); +}); diff --git a/modules/router/src/event-relays/channels.ts b/modules/router/src/event-relays/channels.ts new file mode 100644 index 000000000..c6ae3d692 --- /dev/null +++ b/modules/router/src/event-relays/channels.ts @@ -0,0 +1,23 @@ +export function groupRelaysByChannel( + relays: T[], +): Map { + const next = new Map(); + for (const relay of relays) { + const list = next.get(relay.busEvent) ?? []; + list.push(relay); + next.set(relay.busEvent, list); + } + return next; +} + +export function planChannelSubscriptions( + currentlySubscribed: Iterable, + nextChannels: Iterable, +): { toSubscribe: string[]; toUnsubscribe: string[] } { + const current = new Set(currentlySubscribed); + const next = new Set(nextChannels); + return { + toUnsubscribe: [...current].filter(channel => !next.has(channel)), + toSubscribe: [...next].filter(channel => !current.has(channel)), + }; +} diff --git a/modules/router/src/event-relays/constants.ts b/modules/router/src/event-relays/constants.ts new file mode 100644 index 000000000..e283da0dd --- /dev/null +++ b/modules/router/src/event-relays/constants.ts @@ -0,0 +1,32 @@ +export const EVENTS_NAMESPACE = '/events/'; +export const EVENT_RELAY_REFRESH_CHANNEL = 'router:event-relays:refresh'; +export const EVENT_RELAY_SUBSCRIBER_PREFIX = 'event-relay:'; + +export const MAX_TEMPLATE_BYTES = 16 * 1024; +export const MAX_OUTPUT_BYTES = 64 * 1024; +export const MAX_TEMPLATE_DEPTH = 10; +export const MAX_PATH_SEGMENTS = 8; +export const MAX_NAME_LENGTH = 64; +export const MAX_DESCRIPTION_LENGTH = 256; +export const MAX_BUS_EVENT_LENGTH = 128; +export const MAX_SOCKET_EVENT_LENGTH = 64; +export const MAX_RESOURCE_TYPE_LENGTH = 64; +export const MAX_RESOURCE_ID_LENGTH = 128; +export const MAX_RESOURCE_ID_PATH_LENGTH = 128; +export const MAX_PERMISSION_LENGTH = 64; + +export const FORBIDDEN_PATH_SEGMENTS = new Set(['__proto__', 'constructor', 'prototype']); + +export const RESERVED_SOCKET_EVENTS = new Set([ + 'connect', + 'disconnect', + 'connect_error', + 'error', + 'join-room', + 'leave-room', + 'conduit_error', + 'subscribe', + 'unsubscribe', + 'ping', + 'pong', +]); diff --git a/modules/router/src/event-relays/index.ts b/modules/router/src/event-relays/index.ts new file mode 100644 index 000000000..caa9a4091 --- /dev/null +++ b/modules/router/src/event-relays/index.ts @@ -0,0 +1,25 @@ +export { EventRelayValidationError } from './validationError.js'; +export { + EVENTS_NAMESPACE, + EVENT_RELAY_REFRESH_CHANNEL, + RESERVED_SOCKET_EVENTS, +} from './constants.js'; +export { lookupOwnPath, parseDotPath, requireOwnPath } from './path.js'; +export { renderMessageTemplate, assertTemplateSize } from './template.js'; +export { + validateEventRelayInput, + validateResourceId, + type EventRelayInput, +} from './validation.js'; +export { eventRelayRoom } from './rooms.js'; +export { + parseBusPayload, + buildRelayEmissions, + type RelayEmission, + type ProcessResult, +} from './process.js'; +export { EventRelayManager } from './EventRelayManager.js'; +export { createEventRelayPusher } from './push.js'; +export { createEventsSocket, authorizeRelaySubscription } from './EventRelaySockets.js'; +export { buildSearchQuery, parsePagination } from './search.js'; +export { groupRelaysByChannel, planChannelSubscriptions } from './channels.js'; diff --git a/modules/router/src/event-relays/path.test.ts b/modules/router/src/event-relays/path.test.ts new file mode 100644 index 000000000..ecd1b0f2b --- /dev/null +++ b/modules/router/src/event-relays/path.test.ts @@ -0,0 +1,39 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { lookupOwnPath, parseDotPath, requireOwnPath } from './path.js'; + +describe('parseDotPath', () => { + it('accepts safe own-property paths', () => { + assert.deepEqual(parseDotPath('_id', 'Path'), ['_id']); + assert.deepEqual(parseDotPath('document.ownerId', 'Path'), ['document', 'ownerId']); + }); + + it('rejects prototype pollution segments', () => { + assert.throws(() => parseDotPath('__proto__.polluted', 'Path'), { + name: 'EventRelayValidationError', + }); + assert.throws(() => parseDotPath('constructor.prototype', 'Path'), { + name: 'EventRelayValidationError', + }); + }); +}); + +describe('lookupOwnPath', () => { + it('reads own properties only', () => { + const payload = { document: { _id: 'abc' } }; + assert.equal(lookupOwnPath(payload, 'document._id'), 'abc'); + }); + + it('does not follow inherited properties', () => { + const payload = Object.create({ leaked: 'nope' }); + payload.own = 'yes'; + assert.equal(lookupOwnPath(payload, 'own'), 'yes'); + assert.equal(lookupOwnPath(payload, 'leaked'), undefined); + }); + + it('requireOwnPath fails closed on missing fields', () => { + assert.throws(() => requireOwnPath({ a: 1 }, 'b', 'Resource ID path'), { + name: 'EventRelayValidationError', + }); + }); +}); diff --git a/modules/router/src/event-relays/path.ts b/modules/router/src/event-relays/path.ts new file mode 100644 index 000000000..e4ba89d7c --- /dev/null +++ b/modules/router/src/event-relays/path.ts @@ -0,0 +1,49 @@ +import { FORBIDDEN_PATH_SEGMENTS, MAX_PATH_SEGMENTS } from './constants.js'; +import { EventRelayValidationError } from './validationError.js'; + +const PATH_SEGMENT = /^[A-Za-z_][A-Za-z0-9_]*$/; + +export function parseDotPath(path: string, label: string): string[] { + const trimmed = path.trim(); + if (!trimmed) { + throw new EventRelayValidationError(`${label} is required`); + } + const segments = trimmed.split('.'); + if (segments.length > MAX_PATH_SEGMENTS) { + throw new EventRelayValidationError(`${label} exceeds ${MAX_PATH_SEGMENTS} segments`); + } + for (const segment of segments) { + if (FORBIDDEN_PATH_SEGMENTS.has(segment) || !PATH_SEGMENT.test(segment)) { + throw new EventRelayValidationError(`${label} contains an invalid path segment`); + } + } + return segments; +} + +export function lookupOwnPath(source: unknown, path: string): unknown { + const segments = parseDotPath(path, 'Path'); + let current: unknown = source; + for (const segment of segments) { + if (current === null || typeof current !== 'object') { + return undefined; + } + if (FORBIDDEN_PATH_SEGMENTS.has(segment)) { + return undefined; + } + if (!Object.prototype.hasOwnProperty.call(current, segment)) { + return undefined; + } + current = (current as Record)[segment]; + } + return current; +} + +export function requireOwnPath(source: unknown, path: string, label: string): unknown { + const value = lookupOwnPath(source, path); + if (value === undefined) { + throw new EventRelayValidationError( + `${label} '${path}' was not found on the payload`, + ); + } + return value; +} diff --git a/modules/router/src/event-relays/process.test.ts b/modules/router/src/event-relays/process.test.ts new file mode 100644 index 000000000..cf3c1c10b --- /dev/null +++ b/modules/router/src/event-relays/process.test.ts @@ -0,0 +1,85 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { buildRelayEmissions, parseBusPayload } from './process.js'; +import { createEventRelayPusher } from './push.js'; +import { eventRelayRoom } from './rooms.js'; +import { EVENTS_NAMESPACE } from './constants.js'; + +const relayA = { + _id: 'relay-a', + busEvent: 'database:update:Order', + socketEvent: 'order-updated', + resourceIdPath: '_id', + messageTemplate: { id: '{{payload._id}}', status: '{{payload.status}}' }, +}; + +const relayB = { + _id: 'relay-b', + busEvent: 'database:update:Order', + socketEvent: 'order-paid', + resourceIdPath: '_id', + messageTemplate: { paid: '{{payload.status}}' }, +}; + +describe('parseBusPayload', () => { + it('parses JSON once', () => { + assert.deepEqual(parseBusPayload('{"_id":"1"}'), { _id: '1' }); + }); + + it('fails closed on malformed JSON', () => { + assert.throws(() => parseBusPayload('{'), { name: 'EventRelayValidationError' }); + assert.throws(() => parseBusPayload(''), { name: 'EventRelayValidationError' }); + }); +}); + +describe('buildRelayEmissions', () => { + it('emits to every mapping on the same channel', () => { + const result = buildRelayEmissions([relayA, relayB], { + _id: 'order-1', + status: 'paid', + }); + assert.equal(result.failures.length, 0); + assert.equal(result.emissions.length, 2); + assert.deepEqual( + result.emissions.map(item => item.socketEvent), + ['order-updated', 'order-paid'], + ); + assert.equal(result.emissions[0].room, eventRelayRoom('relay-a', 'order-1')); + }); + + it('records per-relay failures without dropping siblings', () => { + const result = buildRelayEmissions( + [ + relayA, + { + ...relayB, + resourceIdPath: 'missing', + }, + ], + { _id: 'order-1', status: 'paid' }, + ); + assert.equal(result.emissions.length, 1); + assert.equal(result.failures.length, 1); + assert.equal(result.failures[0].relayId, 'relay-b'); + }); +}); + +describe('createEventRelayPusher', () => { + it('pushes locally to /events/ rooms so HA instances do not duplicate', async () => { + const calls: unknown[] = []; + const push = createEventRelayPusher(async data => { + calls.push(data); + }); + await push('order-updated', { id: '1' }, ['room-1']); + assert.deepEqual(calls, [ + { + event: 'order-updated', + data: { id: '1' }, + receivers: [], + rooms: ['room-1'], + namespace: EVENTS_NAMESPACE, + localOnly: true, + }, + ]); + }); +}); diff --git a/modules/router/src/event-relays/process.ts b/modules/router/src/event-relays/process.ts new file mode 100644 index 000000000..858c9f6d0 --- /dev/null +++ b/modules/router/src/event-relays/process.ts @@ -0,0 +1,75 @@ +import { requireOwnPath } from './path.js'; +import { eventRelayRoom } from './rooms.js'; +import { renderMessageTemplate } from './template.js'; +import { validateResourceId } from './validation.js'; +import { EventRelayValidationError } from './validationError.js'; + +export type RelayProcessInput = { + _id: string; + busEvent: string; + socketEvent: string; + resourceIdPath: string; + messageTemplate: unknown; +}; + +export type RelayEmission = { + relayId: string; + busEvent: string; + socketEvent: string; + room: string; + data: unknown; +}; + +export type RelayFailure = { + relayId: string; + busEvent: string; + reason: string; +}; + +export type ProcessResult = { + emissions: RelayEmission[]; + failures: RelayFailure[]; +}; + +export function parseBusPayload(rawMessage: string): unknown { + if (typeof rawMessage !== 'string' || rawMessage.trim() === '') { + throw new EventRelayValidationError('Bus payload is empty'); + } + try { + return JSON.parse(rawMessage); + } catch { + throw new EventRelayValidationError('Bus payload is not valid JSON'); + } +} + +export function buildRelayEmissions( + relays: RelayProcessInput[], + payload: unknown, +): ProcessResult { + const emissions: RelayEmission[] = []; + const failures: RelayFailure[] = []; + + for (const relay of relays) { + try { + const resourceId = validateResourceId( + requireOwnPath(payload, relay.resourceIdPath, 'Resource ID path'), + ); + const data = renderMessageTemplate(relay.messageTemplate, payload); + emissions.push({ + relayId: relay._id, + busEvent: relay.busEvent, + socketEvent: relay.socketEvent, + room: eventRelayRoom(relay._id, resourceId), + data, + }); + } catch (err) { + failures.push({ + relayId: relay._id, + busEvent: relay.busEvent, + reason: err instanceof Error ? err.message : String(err), + }); + } + } + + return { emissions, failures }; +} diff --git a/modules/router/src/event-relays/push.ts b/modules/router/src/event-relays/push.ts new file mode 100644 index 000000000..79d9911b2 --- /dev/null +++ b/modules/router/src/event-relays/push.ts @@ -0,0 +1,28 @@ +import { EVENTS_NAMESPACE } from './constants.js'; + +export type EventRelayPusher = ( + event: string, + data: unknown, + rooms: string[], +) => Promise; + +export type SocketPushFn = (data: { + event: string; + data?: unknown; + receivers: string[]; + rooms: string[]; + namespace: string; + localOnly?: boolean; +}) => Promise; + +export function createEventRelayPusher(socketPush: SocketPushFn): EventRelayPusher { + return (event, data, rooms) => + socketPush({ + event, + data, + receivers: [], + rooms, + namespace: EVENTS_NAMESPACE, + localOnly: true, + }); +} diff --git a/modules/router/src/event-relays/rooms.test.ts b/modules/router/src/event-relays/rooms.test.ts new file mode 100644 index 000000000..e73b19a77 --- /dev/null +++ b/modules/router/src/event-relays/rooms.test.ts @@ -0,0 +1,29 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { eventRelayRoom } from './rooms.js'; + +describe('eventRelayRoom', () => { + it('is deterministic for the same relay and resource', () => { + assert.equal( + eventRelayRoom('relay-1', 'resource-a'), + eventRelayRoom('relay-1', 'resource-a'), + ); + }); + + it('changes when the resource or relay changes', () => { + assert.notEqual( + eventRelayRoom('relay-1', 'resource-a'), + eventRelayRoom('relay-1', 'resource-b'), + ); + assert.notEqual( + eventRelayRoom('relay-1', 'resource-a'), + eventRelayRoom('relay-2', 'resource-a'), + ); + }); + + it('does not embed the raw resource id', () => { + const room = eventRelayRoom('relay-1', 'secret-resource'); + assert.equal(room.includes('secret-resource'), false); + assert.match(room, /^er:relay-1:[a-f0-9]{64}$/); + }); +}); diff --git a/modules/router/src/event-relays/rooms.ts b/modules/router/src/event-relays/rooms.ts new file mode 100644 index 000000000..fd5dd54ec --- /dev/null +++ b/modules/router/src/event-relays/rooms.ts @@ -0,0 +1,6 @@ +import { createHash } from 'node:crypto'; + +export function eventRelayRoom(relayId: string, resourceId: string): string { + const digest = createHash('sha256').update(resourceId).digest('hex'); + return `er:${relayId}:${digest}`; +} diff --git a/modules/router/src/event-relays/search.test.ts b/modules/router/src/event-relays/search.test.ts new file mode 100644 index 000000000..308ea2890 --- /dev/null +++ b/modules/router/src/event-relays/search.test.ts @@ -0,0 +1,34 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { buildSearchQuery, parsePagination } from './search.js'; + +describe('parsePagination', () => { + it('defaults skip and limit for the admin list', () => { + assert.deepEqual(parsePagination(), { skip: 0, limit: 25 }); + assert.deepEqual(parsePagination(10, 5), { skip: 10, limit: 5 }); + }); +}); + +describe('buildSearchQuery', () => { + it('returns an empty query without search', () => { + assert.deepEqual(buildSearchQuery(), {}); + assert.deepEqual(buildSearchQuery(''), {}); + }); + + it('looks up ObjectIds exactly', () => { + assert.deepEqual(buildSearchQuery('507f1f77bcf86cd799439011'), { + _id: '507f1f77bcf86cd799439011', + }); + }); + + it('escapes regex metacharacters in name/channel search', () => { + const query = buildSearchQuery('order.paid*'); + assert.deepEqual(query, { + $or: [ + { name: { $regex: '.*order\\.paid\\*.*', $options: 'i' } }, + { busEvent: { $regex: '.*order\\.paid\\*.*', $options: 'i' } }, + { socketEvent: { $regex: '.*order\\.paid\\*.*', $options: 'i' } }, + ], + }); + }); +}); diff --git a/modules/router/src/event-relays/search.ts b/modules/router/src/event-relays/search.ts new file mode 100644 index 000000000..c336aa363 --- /dev/null +++ b/modules/router/src/event-relays/search.ts @@ -0,0 +1,26 @@ +export function parsePagination( + skip?: number, + limit?: number, +): { skip: number; limit: number } { + return { + skip: skip ?? 0, + limit: limit ?? 25, + }; +} + +export function buildSearchQuery(search?: string | null): Record { + if (search == null || search === '') { + return {}; + } + if (/^[a-fA-F\d]{24}$/.test(search)) { + return { _id: search }; + } + const identifier = search.replace(/[.*+?^${}()|[\]\\]/g, '\\$&'); + return { + $or: [ + { name: { $regex: `.*${identifier}.*`, $options: 'i' } }, + { busEvent: { $regex: `.*${identifier}.*`, $options: 'i' } }, + { socketEvent: { $regex: `.*${identifier}.*`, $options: 'i' } }, + ], + }; +} diff --git a/modules/router/src/event-relays/template.test.ts b/modules/router/src/event-relays/template.test.ts new file mode 100644 index 000000000..7959a08bd --- /dev/null +++ b/modules/router/src/event-relays/template.test.ts @@ -0,0 +1,39 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { renderMessageTemplate } from './template.js'; + +describe('renderMessageTemplate', () => { + it('substitutes exact placeholders with the raw JSON value', () => { + const rendered = renderMessageTemplate( + { id: '{{payload._id}}', count: '{{payload.count}}' }, + { _id: 'abc', count: 3 }, + ); + assert.deepEqual(rendered, { id: 'abc', count: 3 }); + }); + + it('interpolates placeholders inside strings', () => { + const rendered = renderMessageTemplate( + { label: 'Order {{payload.status}}' }, + { status: 'paid' }, + ); + assert.deepEqual(rendered, { label: 'Order paid' }); + }); + + it('fails closed when a placeholder is missing', () => { + assert.throws( + () => renderMessageTemplate({ id: '{{payload.missing}}' }, { _id: 'abc' }), + { name: 'EventRelayValidationError' }, + ); + }); + + it('rejects prototype paths in placeholders', () => { + assert.throws( + () => + renderMessageTemplate( + { hack: '{{payload.__proto__.polluted}}' }, + { __proto__: { polluted: true } }, + ), + { name: 'EventRelayValidationError' }, + ); + }); +}); diff --git a/modules/router/src/event-relays/template.ts b/modules/router/src/event-relays/template.ts new file mode 100644 index 000000000..f0f70dcc9 --- /dev/null +++ b/modules/router/src/event-relays/template.ts @@ -0,0 +1,136 @@ +import { MAX_OUTPUT_BYTES, MAX_TEMPLATE_BYTES, MAX_TEMPLATE_DEPTH } from './constants.js'; +import { lookupOwnPath, parseDotPath } from './path.js'; +import { EventRelayValidationError } from './validationError.js'; + +const PLACEHOLDER = /\{\{\s*payload\.([A-Za-z_][A-Za-z0-9_.]*)\s*\}\}/g; +const EXACT_PLACEHOLDER = /^\{\{\s*payload\.([A-Za-z_][A-Za-z0-9_.]*)\s*\}\}$/; + +export function assertTemplateSize(template: unknown): void { + let serialized: string; + try { + serialized = JSON.stringify(template); + } catch { + throw new EventRelayValidationError('Message template must be valid JSON'); + } + if (serialized === undefined) { + throw new EventRelayValidationError('Message template must be valid JSON'); + } + if (Buffer.byteLength(serialized, 'utf8') > MAX_TEMPLATE_BYTES) { + throw new EventRelayValidationError( + `Message template exceeds ${MAX_TEMPLATE_BYTES} bytes`, + ); + } + walkTemplate(template, 0, (value, depth) => { + if (depth > MAX_TEMPLATE_DEPTH) { + throw new EventRelayValidationError( + `Message template exceeds ${MAX_TEMPLATE_DEPTH} nested levels`, + ); + } + if (typeof value === 'string') { + validatePlaceholders(value); + } + }); +} + +export function renderMessageTemplate(template: unknown, payload: unknown): unknown { + assertTemplateSize(template); + const rendered = renderValue(template, payload, 0); + const serialized = JSON.stringify(rendered); + if (Buffer.byteLength(serialized, 'utf8') > MAX_OUTPUT_BYTES) { + throw new EventRelayValidationError( + `Rendered message exceeds ${MAX_OUTPUT_BYTES} bytes`, + ); + } + return rendered; +} + +function renderValue(value: unknown, payload: unknown, depth: number): unknown { + if (depth > MAX_TEMPLATE_DEPTH) { + throw new EventRelayValidationError( + `Message template exceeds ${MAX_TEMPLATE_DEPTH} nested levels`, + ); + } + if (typeof value === 'string') { + return interpolateString(value, payload); + } + if (Array.isArray(value)) { + return value.map(item => renderValue(item, payload, depth + 1)); + } + if (value !== null && typeof value === 'object') { + const output: Record = {}; + for (const key of Object.keys(value as Record)) { + output[key] = renderValue( + (value as Record)[key], + payload, + depth + 1, + ); + } + return output; + } + return value; +} + +function interpolateString(value: string, payload: unknown): unknown { + const exact = value.trim().match(EXACT_PLACEHOLDER); + if (exact) { + parseDotPath(exact[1], 'Placeholder path'); + const resolved = lookupOwnPath(payload, exact[1]); + if (resolved === undefined) { + throw new EventRelayValidationError( + `Placeholder payload.${exact[1]} was not found on the payload`, + ); + } + return resolved; + } + + return value.replace(PLACEHOLDER, (_match, path: string) => { + parseDotPath(path, 'Placeholder path'); + const resolved = lookupOwnPath(payload, path); + if (resolved === undefined) { + throw new EventRelayValidationError( + `Placeholder payload.${path} was not found on the payload`, + ); + } + return stringifyPlaceholder(resolved); + }); +} + +function stringifyPlaceholder(value: unknown): string { + if (value === null || typeof value === 'string' || typeof value === 'number') { + return String(value); + } + if (typeof value === 'boolean') { + return value ? 'true' : 'false'; + } + return JSON.stringify(value); +} + +function validatePlaceholders(value: string): void { + const exact = value.trim().match(EXACT_PLACEHOLDER); + if (exact) { + parseDotPath(exact[1], 'Placeholder path'); + return; + } + for (const match of value.matchAll(PLACEHOLDER)) { + parseDotPath(match[1], 'Placeholder path'); + } +} + +function walkTemplate( + value: unknown, + depth: number, + visit: (value: unknown, depth: number) => void, +): void { + visit(value, depth); + if (Array.isArray(value)) { + for (const item of value) { + walkTemplate(item, depth + 1, visit); + } + return; + } + if (value !== null && typeof value === 'object') { + for (const nested of Object.values(value as Record)) { + walkTemplate(nested, depth + 1, visit); + } + } +} diff --git a/modules/router/src/event-relays/validation.test.ts b/modules/router/src/event-relays/validation.test.ts new file mode 100644 index 000000000..d8337a5f1 --- /dev/null +++ b/modules/router/src/event-relays/validation.test.ts @@ -0,0 +1,69 @@ +import { describe, it } from 'node:test'; +import assert from 'node:assert/strict'; +import { validateEventRelayInput, validateResourceId } from './validation.js'; + +const validInput = { + name: 'Order paid', + busEvent: 'database:update:Order', + socketEvent: 'order-updated', + resourceType: 'Order', + resourceIdPath: '_id', + permission: 'read', + messageTemplate: { id: '{{payload._id}}' }, +}; + +describe('validateEventRelayInput', () => { + it('accepts a complete relay definition', () => { + const parsed = validateEventRelayInput(validInput); + assert.equal(parsed.active, true); + assert.equal(parsed.busEvent, 'database:update:Order'); + }); + + it('rejects wildcard bus channels', () => { + assert.throws( + () => validateEventRelayInput({ ...validInput, busEvent: 'database:update:*' }), + { name: 'EventRelayValidationError' }, + ); + }); + + it('rejects reserved socket events', () => { + assert.throws( + () => validateEventRelayInput({ ...validInput, socketEvent: 'subscribe' }), + { name: 'EventRelayValidationError' }, + ); + assert.throws( + () => validateEventRelayInput({ ...validInput, socketEvent: 'join-room' }), + { name: 'EventRelayValidationError' }, + ); + }); + + it('rejects empty required fields', () => { + assert.throws(() => validateEventRelayInput({ ...validInput, name: '' }), { + name: 'EventRelayValidationError', + }); + assert.throws( + () => validateEventRelayInput({ ...validInput, messageTemplate: undefined }), + { name: 'EventRelayValidationError' }, + ); + }); +}); + +describe('validateResourceId', () => { + it('accepts object ids and uuids', () => { + assert.equal( + validateResourceId('507f1f77bcf86cd799439011'), + '507f1f77bcf86cd799439011', + ); + assert.equal( + validateResourceId('3b2c1a90-1111-2222-3333-444444444444'), + '3b2c1a90-1111-2222-3333-444444444444', + ); + }); + + it('rejects empty or colon-containing identifiers', () => { + assert.throws(() => validateResourceId(''), { name: 'EventRelayValidationError' }); + assert.throws(() => validateResourceId('Team:123'), { + name: 'EventRelayValidationError', + }); + }); +}); diff --git a/modules/router/src/event-relays/validation.ts b/modules/router/src/event-relays/validation.ts new file mode 100644 index 000000000..d6c6c5eb6 --- /dev/null +++ b/modules/router/src/event-relays/validation.ts @@ -0,0 +1,130 @@ +import { + MAX_BUS_EVENT_LENGTH, + MAX_DESCRIPTION_LENGTH, + MAX_NAME_LENGTH, + MAX_PERMISSION_LENGTH, + MAX_RESOURCE_ID_LENGTH, + MAX_RESOURCE_ID_PATH_LENGTH, + MAX_RESOURCE_TYPE_LENGTH, + MAX_SOCKET_EVENT_LENGTH, + RESERVED_SOCKET_EVENTS, +} from './constants.js'; +import { parseDotPath } from './path.js'; +import { assertTemplateSize } from './template.js'; +import { EventRelayValidationError } from './validationError.js'; + +export type EventRelayInput = { + name: string; + notes?: string; + active?: boolean; + busEvent: string; + socketEvent: string; + resourceType: string; + resourceIdPath: string; + permission: string; + messageTemplate: unknown; +}; + +const NAME_PATTERN = /^[A-Za-z0-9][A-Za-z0-9 _.-]{0,63}$/; +const BUS_EVENT_PATTERN = /^[A-Za-z0-9][A-Za-z0-9_.:-]{0,127}$/; +const SOCKET_EVENT_PATTERN = /^[A-Za-z][A-Za-z0-9_:-]{0,63}$/; +const RESOURCE_TYPE_PATTERN = /^[A-Za-z][A-Za-z0-9_]{0,63}$/; +const PERMISSION_PATTERN = /^[A-Za-z][A-Za-z0-9_]{0,63}$/; +const RESOURCE_ID_PATTERN = /^[^\s:]{1,128}$/; + +export function validateEventRelayInput(input: EventRelayInput): EventRelayInput { + const name = requireTrimmed(input.name, 'Name'); + if (name.length > MAX_NAME_LENGTH || !NAME_PATTERN.test(name)) { + throw new EventRelayValidationError( + 'Name must be 1-64 characters and start with a letter or number', + ); + } + + const notes = + input.notes === undefined || input.notes === '' + ? undefined + : requireTrimmed(input.notes, 'Notes'); + if (notes && notes.length > MAX_DESCRIPTION_LENGTH) { + throw new EventRelayValidationError( + `Notes must be at most ${MAX_DESCRIPTION_LENGTH} characters`, + ); + } + + const busEvent = requireTrimmed(input.busEvent, 'Bus event'); + if ( + busEvent.length > MAX_BUS_EVENT_LENGTH || + busEvent.includes('*') || + !BUS_EVENT_PATTERN.test(busEvent) + ) { + throw new EventRelayValidationError( + 'Bus event must be an exact channel name with no wildcards', + ); + } + + const socketEvent = requireTrimmed(input.socketEvent, 'Socket event'); + if ( + socketEvent.length > MAX_SOCKET_EVENT_LENGTH || + !SOCKET_EVENT_PATTERN.test(socketEvent) || + RESERVED_SOCKET_EVENTS.has(socketEvent) + ) { + throw new EventRelayValidationError('Socket event must be a non-reserved event name'); + } + + const resourceType = requireTrimmed(input.resourceType, 'Resource type'); + if ( + resourceType.length > MAX_RESOURCE_TYPE_LENGTH || + !RESOURCE_TYPE_PATTERN.test(resourceType) + ) { + throw new EventRelayValidationError('Resource type is invalid'); + } + + const resourceIdPath = requireTrimmed(input.resourceIdPath, 'Resource ID path'); + if (resourceIdPath.length > MAX_RESOURCE_ID_PATH_LENGTH) { + throw new EventRelayValidationError('Resource ID path is too long'); + } + parseDotPath(resourceIdPath, 'Resource ID path'); + + const permission = requireTrimmed(input.permission, 'Permission'); + if (permission.length > MAX_PERMISSION_LENGTH || !PERMISSION_PATTERN.test(permission)) { + throw new EventRelayValidationError('Permission is invalid'); + } + + if (input.messageTemplate === undefined) { + throw new EventRelayValidationError('Message template is required'); + } + assertTemplateSize(input.messageTemplate); + + return { + name, + notes, + active: input.active !== false, + busEvent, + socketEvent, + resourceType, + resourceIdPath, + permission, + messageTemplate: input.messageTemplate, + }; +} + +export function validateResourceId(resourceId: unknown): string { + if (typeof resourceId !== 'string' && typeof resourceId !== 'number') { + throw new EventRelayValidationError('Resource ID must be a string'); + } + const value = String(resourceId).trim(); + if ( + !value || + value.length > MAX_RESOURCE_ID_LENGTH || + !RESOURCE_ID_PATTERN.test(value) + ) { + throw new EventRelayValidationError('Resource ID is invalid'); + } + return value; +} + +function requireTrimmed(value: unknown, label: string): string { + if (typeof value !== 'string' || value.trim() === '') { + throw new EventRelayValidationError(`${label} is required`); + } + return value.trim(); +} diff --git a/modules/router/src/event-relays/validationError.ts b/modules/router/src/event-relays/validationError.ts new file mode 100644 index 000000000..90151b042 --- /dev/null +++ b/modules/router/src/event-relays/validationError.ts @@ -0,0 +1,6 @@ +export class EventRelayValidationError extends Error { + constructor(message: string) { + super(message); + this.name = 'EventRelayValidationError'; + } +} diff --git a/modules/router/src/metrics/index.ts b/modules/router/src/metrics/index.ts index 9dd6d9789..d3461c107 100644 --- a/modules/router/src/metrics/index.ts +++ b/modules/router/src/metrics/index.ts @@ -17,4 +17,25 @@ export default { labelNames: ['platform'], }, }, + eventRelaysEmitted: { + type: MetricType.Counter, + config: { + name: 'event_relays_emitted_total', + help: 'Tracks successfully emitted event-to-socket relays', + }, + }, + eventRelaysFailed: { + type: MetricType.Counter, + config: { + name: 'event_relays_failed_total', + help: 'Tracks event-to-socket relay parse, render, or emit failures', + }, + }, + eventRelaySubscriptionsDenied: { + type: MetricType.Counter, + config: { + name: 'event_relay_subscriptions_denied_total', + help: 'Tracks denied or unavailable event-relay socket subscriptions', + }, + }, }; diff --git a/modules/router/src/models/EventRelay.schema.ts b/modules/router/src/models/EventRelay.schema.ts new file mode 100644 index 000000000..e0c07b1d9 --- /dev/null +++ b/modules/router/src/models/EventRelay.schema.ts @@ -0,0 +1,94 @@ +import { + ConduitModel, + DatabaseProvider, + Indexable, + TYPE, +} from '@conduitplatform/grpc-sdk'; +import { ConduitActiveSchema } from '@conduitplatform/module-tools'; + +const schema: ConduitModel = { + _id: TYPE.ObjectId, + name: { + type: TYPE.String, + unique: true, + required: true, + }, + notes: { + type: TYPE.String, + required: false, + }, + active: { + type: TYPE.Boolean, + required: true, + default: true, + }, + busEvent: { + type: TYPE.String, + required: true, + }, + socketEvent: { + type: TYPE.String, + required: true, + }, + resourceType: { + type: TYPE.String, + required: true, + }, + resourceIdPath: { + type: TYPE.String, + required: true, + }, + permission: { + type: TYPE.String, + required: true, + }, + messageTemplate: { + type: TYPE.JSON, + required: true, + }, + createdAt: TYPE.Date, + updatedAt: TYPE.Date, +}; + +const modelOptions = { + timestamps: true, + conduit: { + permissions: { + extendable: true, + canCreate: false, + canModify: 'ExtensionOnly', + canDelete: false, + }, + }, +} as const; + +const collectionName = undefined; + +export class EventRelay extends ConduitActiveSchema { + private static _instance: EventRelay; + _id!: string; + declare name: string; + notes?: string; + active!: boolean; + busEvent!: string; + socketEvent!: string; + resourceType!: string; + resourceIdPath!: string; + permission!: string; + messageTemplate!: Indexable; + createdAt!: Date; + updatedAt!: Date; + + private constructor(database: DatabaseProvider) { + super(database, EventRelay.name, schema, modelOptions, collectionName); + } + + static getInstance(database?: DatabaseProvider) { + if (EventRelay._instance) return EventRelay._instance; + if (!database) { + throw new Error('No database instance provided!'); + } + EventRelay._instance = new EventRelay(database); + return EventRelay._instance; + } +} diff --git a/modules/router/src/models/index.ts b/modules/router/src/models/index.ts index 400d89cdd..ae11db58e 100644 --- a/modules/router/src/models/index.ts +++ b/modules/router/src/models/index.ts @@ -1,2 +1,3 @@ export * from './Client.schema.js'; export * from './AppMiddleware.schema.js'; +export * from './EventRelay.schema.js'; diff --git a/modules/router/tsconfig.test.json b/modules/router/tsconfig.test.json new file mode 100644 index 000000000..601e5ccee --- /dev/null +++ b/modules/router/tsconfig.test.json @@ -0,0 +1,24 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { + "outDir": "./dist-test", + "rootDir": "./src", + "declaration": false, + "sourceMap": false, + "types": ["node"] + }, + "include": [ + "src/event-relays/constants.ts", + "src/event-relays/validationError.ts", + "src/event-relays/path.ts", + "src/event-relays/template.ts", + "src/event-relays/validation.ts", + "src/event-relays/rooms.ts", + "src/event-relays/process.ts", + "src/event-relays/push.ts", + "src/event-relays/search.ts", + "src/event-relays/channels.ts", + "src/event-relays/authorize.ts", + "src/event-relays/*.test.ts" + ] +}