feat: standalone ddp server implementation

for now intended to run at a secondary path off the meteor ddp server

Author: Julusian

meteor/.meteor/packages

@@ -10,7 +10,6 @@
 
 meteor@2.3.0
 webapp@2.1.2
-ddp@1.4.2
 
 mongo@2.3.0 # The database Meteor supports right now
 

meteor/__mocks__/_setupMocks.ts

@@ -14,7 +14,7 @@ jest.mock('meteor/meteor', (...args) => require('./meteor').setup(args), { virtu
 jest.mock('meteor/random', (...args) => require('./random').setup(args), { virtual: true })
 jest.mock('meteor/check', (...args) => require('./check').setup(args), { virtual: true })
 jest.mock('meteor/tracker', (...args) => require('./tracker').setup(args), { virtual: true })
-jest.mock('meteor/ejson', (...args) => require('./ejson').setup(args), { virtual: true })
+jest.mock('meteor/ejson', (...args) => require('./meteor-ejson').setup(args), { virtual: true })
 
 jest.mock('meteor/mdg:validated-method', (...args) => require('./validated-method').setup(args), { virtual: true })
 jest.mock('meteor/julusian:meteor-elastic-apm', (...args) => require('./meteor-elastic-apm').setup(args), {

meteor/__mocks__/ejson.ts meteor/__mocks__/meteor-ejson.tsmeteor/__mocks__/ejson.ts renamed to meteor/__mocks__/meteor-ejson.ts

@@ -1,4 +1,3 @@
-// @ts-expect-error No types available
 import EJSON from 'ejson'
 
 export function setup(): any {

meteor/__mocks__/meteor.ts

@@ -1,5 +1,4 @@
 /* eslint-disable @typescript-eslint/no-unsafe-function-type, @typescript-eslint/only-throw-error */
-import { USER_PERMISSIONS_HEADER } from '@sofie-automation/meteor-lib/dist/userPermissions'
 
 let controllableDefer = false
 
@@ -96,6 +95,8 @@ export namespace MeteorMock {
 	export const absolutePath = process.cwd()
 
 	function getMethodContext() {
+		const { USER_PERMISSIONS_HEADER } = require('../server/security/auth')
+
 		return {
 			connection: {
 				clientAddress: '1.1.1.1',
@@ -113,7 +114,8 @@ export namespace MeteorMock {
 		private _stack?: string
 		constructor(
 			public error: number,
-			public reason?: string
+			public reason?: string,
+			public details?: unknown
 		) {
 			const e = new $.Error('')
 			let stack: string = e.stack || ''
@@ -132,9 +134,6 @@ export namespace MeteorMock {
 		get message(): string {
 			return this.toString()
 		}
-		get details(): any {
-			return undefined
-		}
 		get errorType(): string {
 			return 'Meteor.Error'
 		}
@@ -231,6 +230,10 @@ export namespace MeteorMock {
 		mockStartupFunctions.push(fcn)
 	}
 
+	export function onConnection(_callback: (connection: any) => void): void {
+		// no-op in tests; the standalone DDP server / Connections.ts register a handler here at import time
+	}
+
 	export function publish(publicationName: string, handler: Function): any {
 		publications[publicationName] = handler
 	}

meteor/__mocks__/mongo.ts

@@ -96,6 +96,7 @@ export namespace MongoMock {
 			}
 
 			return {
+				collectionName: this._name,
 				_fetchRaw: () => {
 					return docs
 				},

meteor/package.json

@@ -50,6 +50,7 @@
 		"bcrypt": "^6.0.0",
 		"deep-extend": "0.6.0",
 		"deepmerge": "^4.3.1",
+		"ejson": "^2.2.3",
 		"elastic-apm-node": "^4.15.0",
 		"i18next": "^26.3.1",
 		"indexof": "0.0.1",
@@ -68,18 +69,22 @@
 		"threadedclass": "^1.4.0",
 		"type-fest": "^4.41.0",
 		"underscore": "^1.13.8",
-		"winston": "^3.19.0"
+		"winston": "^3.19.0",
+		"ws": "^8.21.0"
 	},
 	"devDependencies": {
 		"@babel/core": "^7.29.7",
+		"@babel/plugin-transform-export-namespace-from": "^7.27.1",
 		"@babel/plugin-transform-modules-commonjs": "^7.29.7",
 		"@meteorjs/rspack": "2.0.1",
 		"@rsdoctor/rspack-plugin": "1.5.7",
 		"@rspack/cli": "1.7.1",
 		"@rspack/core": "1.7.1",
 		"@shopify/jest-koa-mocks": "^5.3.1",
 		"@sofie-automation/code-standard-preset": "^3.2.2",
+		"@sofie-automation/server-core-integration": "portal:../packages/server-core-integration",
 		"@types/deep-extend": "^0.6.2",
+		"@types/ejson": "^2.2.2",
 		"@types/jest": "^30.0.0",
 		"@types/koa": "^3.0.3",
 		"@types/koa-bodyparser": "^4.3.13",
@@ -89,9 +94,9 @@
 		"@types/node": "^22.19.21",
 		"@types/semver": "^7.7.1",
 		"@types/underscore": "^1.13.0",
+		"@types/ws": "^8.18.1",
 		"babel-jest": "^30.4.1",
 		"commit-and-tag-version": "^12.7.3",
-		"ejson": "^2.2.3",
 		"eslint": "^9.39.4",
 		"fast-clone": "^1.5.13",
 		"glob": "^13.0.6",

meteor/scripts/babel-jest.js

@@ -1,7 +1,12 @@
 const babelJest = require('babel-jest')
 
 module.exports = babelJest.default.createTransformer({
-	plugins: ['@babel/plugin-transform-modules-commonjs'],
+	plugins: [
+		'@babel/plugin-transform-modules-commonjs',
+		// Some workspace deps (e.g. server-core-integration) ship ESM that uses `export * as ns from`;
+		// transform-modules-commonjs alone can't lower that, so include the namespace-from transform.
+		'@babel/plugin-transform-export-namespace-from',
+	],
 	babelrc: false,
 	configFile: false,
 })

meteor/server/Connections.ts

@@ -11,55 +11,63 @@ const connectionsGauge = new MetricsGauge({
 	help: 'Number of open ddp connections',
 })
 
-Meteor.onConnection((conn: Meteor.Connection) => {
+/**
+ * Track a newly-opened DDP connection (from either Meteor's built-in server or the standalone DDP
+ * server). Updates the open-connection metrics.
+ */
+export function trackConnectionOpen(connectionId: string, clientAddress: string): void {
 	// This is called whenever a new ddp-connection is opened (ie a web-client or a peripheral-device)
+	connections.add(connectionId)
+	logger.debug(`Client connected: "${connectionId}", "${clientAddress}"`)
+	traceConnections()
+}
 
-	const connectionId: string = conn.id
-	// var clientAddress = conn.clientAddress; // ip-adress
-
-	connections.add(conn.id)
-	logger.debug(`Client connected: "${conn.id}", "${conn.clientAddress}"`)
+/**
+ * Track a closed DDP connection. Updates metrics, and marks any PeripheralDevices that were bound to this
+ * connection (and their children) as offline.
+ */
+export function trackConnectionClose(connectionId: string, clientAddress: string): void {
+	connections.delete(connectionId)
+	logger.debug(`Client disconnected: "${connectionId}", "${clientAddress}"`)
 	traceConnections()
 
-	conn.onClose(() => {
-		// called when a connection is closed
-		connections.delete(conn.id)
-		logger.debug(`Client disconnected: "${conn.id}", "${conn.clientAddress}"`)
-		traceConnections()
+	if (!connectionId) return
 
-		if (connectionId) {
-			deferAsync(async () => {
-				const devices = await PeripheralDevices.findFetchAsync({
-					connectionId: connectionId,
-				})
+	deferAsync(async () => {
+		const devices = await PeripheralDevices.findFetchAsync({
+			connectionId: connectionId,
+		})
 
-				for (const device of devices) {
-					// set the status of the machine to offline:
+		for (const device of devices) {
+			// set the status of the machine to offline:
 
-					await PeripheralDevices.updateAsync(device._id, {
-						$set: {
-							lastSeen: getCurrentTime(),
-							connected: false,
-							// connectionId: ''
-						},
-					})
-					await PeripheralDevices.updateAsync(
-						{
-							parentDeviceId: device._id,
-						},
-						{
-							$set: {
-								lastSeen: getCurrentTime(),
-								connected: false,
-								// connectionId: ''
-							},
-						},
-						{ multi: true }
-					)
-				}
+			await PeripheralDevices.updateAsync(device._id, {
+				$set: {
+					lastSeen: getCurrentTime(),
+					connected: false,
+					// connectionId: ''
+				},
 			})
+			await PeripheralDevices.updateAsync(
+				{
+					parentDeviceId: device._id,
+				},
+				{
+					$set: {
+						lastSeen: getCurrentTime(),
+						connected: false,
+						// connectionId: ''
+					},
+				},
+				{ multi: true }
+			)
 		}
 	})
+}
+
+Meteor.onConnection((conn: Meteor.Connection) => {
+	trackConnectionOpen(conn.id, conn.clientAddress)
+	conn.onClose(() => trackConnectionClose(conn.id, conn.clientAddress))
 })
 
 let logTimeout: number | undefined = undefined

meteor/server/__tests__/Connections.test.ts

@@ -0,0 +1,57 @@
+jest.mock('../api/integration/influx')
+
+import { trackConnectionClose } from '../Connections'
+import { PeripheralDevices } from '../collections'
+import { protectString } from '@sofie-automation/corelib/dist/protectedString'
+import { PeripheralDeviceId } from '@sofie-automation/corelib/dist/dataModel/Ids'
+import { waitUntil } from '../../__mocks__/helpers/jest'
+
+describe('Connections.trackConnectionClose', () => {
+	beforeEach(async () => {
+		await PeripheralDevices.removeAsync({})
+	})
+
+	test('marks the device and its parentDeviceId children offline for the closed connection', async () => {
+		const parentId = protectString<PeripheralDeviceId>('parent1')
+		const childId = protectString<PeripheralDeviceId>('child1')
+		const otherId = protectString<PeripheralDeviceId>('other1')
+		await PeripheralDevices.insertAsync({
+			_id: parentId,
+			connectionId: 'conn1',
+			connected: true,
+			lastSeen: 1,
+		} as any)
+		await PeripheralDevices.insertAsync({
+			_id: childId,
+			parentDeviceId: parentId,
+			connected: true,
+			lastSeen: 1,
+		} as any)
+		await PeripheralDevices.insertAsync({
+			_id: otherId,
+			connectionId: 'conn2',
+			connected: true,
+			lastSeen: 1,
+		} as any)
+
+		trackConnectionClose('conn1', '127.0.0.1')
+
+		await waitUntil(async () => {
+			expect((await PeripheralDevices.findOneAsync(parentId))?.connected).toBe(false)
+			expect((await PeripheralDevices.findOneAsync(childId))?.connected).toBe(false)
+		}, 1000)
+
+		// A device on a different connection is untouched.
+		expect((await PeripheralDevices.findOneAsync(otherId))?.connected).toBe(true)
+	})
+
+	test('does nothing when no connectionId is given', async () => {
+		const id = protectString<PeripheralDeviceId>('d1')
+		await PeripheralDevices.insertAsync({ _id: id, connectionId: 'conn1', connected: true, lastSeen: 1 } as any)
+
+		trackConnectionClose('', '127.0.0.1')
+		await new Promise((r) => setTimeout(r, 50)) // give any (incorrect) deferred work a chance to run
+
+		expect((await PeripheralDevices.findOneAsync(id))?.connected).toBe(true)
+	})
+})

meteor/server/api/rest/koa.ts

@@ -6,8 +6,9 @@ import { WebApp } from 'meteor/webapp'
 import { Meteor } from 'meteor/meteor'
 import { getRandomString } from '@sofie-automation/corelib/dist/lib'
 import { stringifyError } from '@sofie-automation/shared-lib/dist/lib/stringifyError'
-import _ from 'underscore'
 import { getRootSubpath, public_dir } from '../../lib'
+import { getClientAddress } from '../../lib/clientAddress'
+import { STANDALONE_DDP_SERVER_ENABLED, STANDALONE_DDP_SERVER_PATH } from '../../ddp-server/config'
 import staticServe from 'koa-static'
 import { logger } from '../../logging'
 import { PackageInfo } from '../../coreSystem'
@@ -124,6 +125,12 @@ function getExtendedMeteorRuntimeConfig() {
 		// @ts-expect-error missing types for internal meteor detail
 		...__meteor_runtime_config__,
 		sofieVersionExtended: versionExtended,
+		// The webui's DDP connection endpoint is defined here — point it at the standalone DDP server when
+		// enabled, otherwise Meteor's own DDP endpoint. This is a site-root-relative path (no prefix): the
+		// vendored client resolves it against ROOT_URL_PATH_PREFIX (it no longer appends `/websocket` — see
+		// packages/webui/src/meteor/socket-stream-client/urls.js), and the server matches the prefixed path
+		// the same way (`getRootSubpath() + path` in ddp-server/index.ts).
+		DDP_DEFAULT_CONNECTION_URL: STANDALONE_DDP_SERVER_ENABLED ? STANDALONE_DDP_SERVER_PATH : '/websocket',
 	})})`
 }
 
@@ -164,37 +171,6 @@ export function bindKoaRouter(koaRouter: KoaRouter, bindPath: string): void {
 	rootRouter.use(bindPathWithPrefix, koaRouter.routes()).use(bindPathWithPrefix, koaRouter.allowedMethods())
 }
 
-const REVERSE_PROXY_COUNT = process.env.HTTP_FORWARDED_COUNT ? parseInt(process.env.HTTP_FORWARDED_COUNT) : 0
-
-// X-Forwarded-For (a de-facto standard) has the following syntax by convention
-// X-Forwarded-For: 203.0.113.195, 2001:db8:85a3:8d3:1319:8a2e:370:7348
-// X-Forwarded-For: 203.0.113.195,2001:db8:85a3:8d3:1319:8a2e:370:7348,198.51.100.178
-function getClientAddrFromXForwarded(headerVal: undefined | string | string[]): string | undefined {
-	if (headerVal === undefined) return undefined
-	if (typeof headerVal !== 'string') {
-		headerVal = _.last(headerVal) as string
-	}
-	const remoteAddresses = headerVal.split(',')
-	return remoteAddresses[remoteAddresses.length - REVERSE_PROXY_COUNT]?.trim() ?? remoteAddresses[0]?.trim()
-}
-
-// Forwarded uses the following syntax:
-// Forwarded: for=192.0.2.60;proto=http;by=203.0.113.43
-// Forwarded: for=192.0.2.43, for="[2001:db8:cafe::17]"
-function getClientAddrFromForwarded(forwardedVal: string | undefined): string | undefined {
-	if (forwardedVal === undefined) return undefined
-	const allProxies = forwardedVal.split(',')
-	const proxyInfo = allProxies[allProxies.length - REVERSE_PROXY_COUNT] ?? allProxies[0]
-	const directives = proxyInfo?.trim().split(';')
-	for (const directive of directives) {
-		let match: RegExpMatchArray | null
-		if ((match = directive.trim().match(/^for=("\[)?([\w.:])+(\]")?/))) {
-			return match[2]
-		}
-	}
-	return undefined
-}
-
 export const makeMeteorConnectionFromKoa = (
 	ctx: Koa.ParameterizedContext<Koa.DefaultState, Koa.DefaultContext, unknown>
 ): Meteor.Connection => {
@@ -206,13 +182,7 @@ export const makeMeteorConnectionFromKoa = (
 		onClose: () => {
 			/* no-op */
 		},
-		clientAddress:
-			// This replicates Meteor behavior which uses the HTTP_FORWARDED_COUNT to extract the "world-facing"
-			// IP address of the Client User Agent
-			getClientAddrFromForwarded(ctx.req.headers.forwarded) ||
-			getClientAddrFromXForwarded(ctx.req.headers['x-forwarded-for']) ||
-			ctx.req.socket.remoteAddress ||
-			'unknown',
+		clientAddress: getClientAddress(ctx.req.headers, ctx.req.socket.remoteAddress),
 		httpHeaders: ctx.req.headers as Record<string, string>,
 	}
 }