feat: refactor ddp publication registration into a PublicationRegistry

Author: Julusian

meteor/__mocks__/helpers/publications.ts

@@ -0,0 +1,17 @@
+import { PublicationRegistry } from '../../server/publicationRegistry'
+import { registerAllPublications } from '../../server/publicationRegistrations'
+
+/**
+ * Test helper: register all DDP publications on a fresh PublicationRegistry and apply them to the
+ * (mock) Meteor server, mirroring what `main.ts` does at startup.
+ *
+ * Call this in suites that exercise Meteor subscriptions. It is needed because the production
+ * registration now only runs explicitly from `main.ts`, rather than as an import-time side effect of
+ * each publication file.
+ */
+export function registerAllPublicationsForTest(): PublicationRegistry {
+	const registry = new PublicationRegistry()
+	registerAllPublications(registry)
+	registry.applyToMeteor()
+	return registry
+}

meteor/server/__tests__/__snapshots__/publicationRegistry.test.ts.snap

@@ -0,0 +1,76 @@
+// Jest Snapshot v1, https://jestjs.io/docs/snapshot-testing
+
+exports[`PublicationRegistry the full set of registered publication names is stable (drift guard) 1`] = `
+[
+  "adLibActions",
+  "adLibActionsForPart",
+  "adLibPieces",
+  "adLibPiecesForPart",
+  "blueprints",
+  "bucketAdLibActions",
+  "bucketAdLibPieces",
+  "buckets",
+  "coreSystem",
+  "deviceTriggersPreview",
+  "evaluations",
+  "expectedPackageWorkStatuses",
+  "expectedPackages",
+  "expectedPlayoutItemsForDevice",
+  "externalEventSubscriptionsForDevice",
+  "externalMessageQueue",
+  "ingestDataCache",
+  "ingestDeviceRundownStatus",
+  "ingestDeviceRundownStatusTestTool",
+  "mappingsForDevice",
+  "mappingsForStudio",
+  "mountedTriggersForDevice",
+  "mountedTriggersForDevicePreview",
+  "notificationsForRundown",
+  "notificationsForRundownPlaylist",
+  "packageContainerStatuses",
+  "packageInfos",
+  "packageManagerExpectedPackages",
+  "packageManagerPackageContainers",
+  "packageManagerPlayoutContext",
+  "partInstances",
+  "partInstancesSimple",
+  "parts",
+  "peripheralDeviceCommands",
+  "peripheralDeviceForDevice",
+  "peripheralDevices",
+  "peripheralDevicesAndSubDevices",
+  "pieceInstances",
+  "pieceInstancesSimple",
+  "pieces",
+  "piecesInfiniteStartingBefore",
+  "rundownBaselineAdLibActions",
+  "rundownBaselineAdLibPieces",
+  "rundownLayouts",
+  "rundownPlaylistForStudio",
+  "rundownPlaylists",
+  "rundownsForDevice",
+  "rundownsInPlaylists",
+  "rundownsWithShowStyleBases",
+  "segments",
+  "showStyleBases",
+  "showStyleVariants",
+  "snapshots",
+  "studios",
+  "timelineDatastore",
+  "timelineDatastoreForDevice",
+  "timelineForDevice",
+  "timelineForStudio",
+  "translationsBundles",
+  "triggeredActions",
+  "uiBlueprintUpgradeStatuses",
+  "uiBucketContentStatuses",
+  "uiPartInstances",
+  "uiParts",
+  "uiPieceContentStatuses",
+  "uiSegmentPartNotes",
+  "uiShowStyleBase",
+  "uiStudio",
+  "uiTriggeredActions",
+  "userActionsLog",
+]
+`;

meteor/server/__tests__/publicationRegistry.test.ts

@@ -0,0 +1,40 @@
+import { PublicationRegistry } from '../publicationRegistry'
+import { registerAllPublications } from '../publicationRegistrations'
+import { AllPubSubNames } from '@sofie-automation/meteor-lib/dist/api/pubsub'
+
+describe('PublicationRegistry', () => {
+	test('registers all publications without error', () => {
+		const registry = new PublicationRegistry()
+		// publishUnsafe throws on a duplicate name, so this is itself an assertion that no two
+		// publications register under the same wire name.
+		expect(() => registerAllPublications(registry)).not.toThrow()
+	})
+
+	test('registered names are unique and only use known publication names', () => {
+		const registry = new PublicationRegistry()
+		registerAllPublications(registry)
+		const names = registry.getAllPublicationNames()
+
+		// No duplicate registrations across all publication modules
+		expect(new Set(names).size).toBe(names.length)
+
+		// Every registered name must be a known PubSub name.
+		const knownNames = new Set<string>(AllPubSubNames)
+		for (const name of names) {
+			expect(knownNames.has(name)).toBe(true)
+		}
+
+		// Conversely, every known PubSub name must be registered (the historical dev-mode check).
+		for (const pubName of AllPubSubNames) {
+			expect(names).toContain(pubName)
+		}
+	})
+
+	test('the full set of registered publication names is stable (drift guard)', () => {
+		const registry = new PublicationRegistry()
+		registerAllPublications(registry)
+		// If this snapshot changes, a publication's wire name was added, removed or renamed. That must
+		// be a deliberate change reviewed against the clients that depend on these exact names.
+		expect([...registry.getAllPublicationNames()].sort()).toMatchSnapshot()
+	})
+})

meteor/server/api/rest/api.ts

@@ -12,6 +12,7 @@ import { createLegacyApiRouter } from './v0/index'
 import { heapSnapshotPrivateApiRouter } from '../heapSnapshot'
 import { getRootSubpath } from '../../lib'
 import type { MethodRegistry } from '../../methodRegistry'
+import type { PublicationRegistry } from '../../publicationRegistry'
 
 const LATEST_REST_API = 'v1.0'
 
@@ -20,7 +21,7 @@ async function redirectToLatest(ctx: koa.ParameterizedContext, _next: koa.Next):
 	ctx.status = 307
 }
 
-export function bindRestApiRouter(methodRegistry: MethodRegistry): void {
+export function bindRestApiRouter(methodRegistry: MethodRegistry, publicationRegistry: PublicationRegistry): void {
 	const apiRouter = new KoaRouter()
 
 	apiRouter.get('/', redirectToLatest)
@@ -45,7 +46,7 @@ export function bindRestApiRouter(methodRegistry: MethodRegistry): void {
 	)
 
 	// Needs to be lazily generated
-	const legacyApiRouter = createLegacyApiRouter(methodRegistry)
+	const legacyApiRouter = createLegacyApiRouter(methodRegistry, publicationRegistry)
 	apiRouter.use('/0', legacyApiRouter.routes(), legacyApiRouter.allowedMethods())
 
 	bindKoaRouter(apiRouter, '/api')

meteor/server/api/rest/v0/__tests__/rest.test.ts

@@ -2,6 +2,7 @@ import { MeteorMock } from '../../../../../__mocks__/meteor'
 import { Meteor } from 'meteor/meteor'
 import { UserActionAPIMethods } from '@sofie-automation/meteor-lib/dist/api/userActions'
 import { MethodRegistry, AnyMethodApiRegistration } from '../../../../methodRegistry'
+import { PublicationRegistry } from '../../../../publicationRegistry'
 import { ClientAPI } from '@sofie-automation/meteor-lib/dist/api/client'
 import { callKoaRoute } from '../../../../../__mocks__/koa-util'
 import { createLegacyApiRouter } from '..'
@@ -25,7 +26,8 @@ describe('REST API', () => {
 		} as unknown as AnyMethodApiRegistration)
 		methodRegistry.applyToMeteor() // register the methods on the (mock) Meteor server
 		const methodSignatures = methodRegistry.getSignatures()
-		const legacyApiRouter = createLegacyApiRouter(methodRegistry)
+		const publicationRegistry = new PublicationRegistry()
+		const legacyApiRouter = createLegacyApiRouter(methodRegistry, publicationRegistry)
 
 		test('calls the UserActionAPI methods, when doing a POST to the endpoint', async () => {
 			for (const [methodName, methodValue] of Object.entries<any>(UserActionAPIMethods)) {

meteor/server/api/rest/v0/index.ts

@@ -7,8 +7,9 @@
 import _ from 'underscore'
 import { Meteor } from 'meteor/meteor'
 import type { MethodRegistry } from '../../../methodRegistry'
+import type { PublicationRegistry } from '../../../publicationRegistry'
+import type { PublicationContext } from '../../../publications/lib/lib'
 import { MeteorPubSub } from '@sofie-automation/meteor-lib/dist/api/pubsub'
-import { MeteorPublications, MeteorPublicationSignatures } from '../../../publications/lib/lib'
 import { UserActionAPIMethods } from '@sofie-automation/meteor-lib/dist/api/userActions'
 import { logger } from '../../../logging'
 import { ClientAPI } from '@sofie-automation/meteor-lib/dist/api/client'
@@ -19,6 +20,20 @@ import { PeripheralDevicePubSub } from '@sofie-automation/shared-lib/dist/pubsub
 
 const LEGACY_API_VERSION = 0
 
+/**
+ * A no-op publication context for the legacy REST path: there is no live subscription, the callback is
+ * only invoked to obtain its cursor, which is then fetched once. There is no connection, so any
+ * publication that requires one will reject (matching the historical behaviour).
+ */
+const legacyRestPublicationContext: PublicationContext = {
+	connection: null,
+	onStop: () => undefined,
+	ready: () => undefined,
+	added: () => undefined,
+	changed: () => undefined,
+	removed: () => undefined,
+}
+
 /**
  * Takes an array of strings and converts them to Null, Boolean, Number, String primitives or Objects, if the string
  * seems like a valid JSON.
@@ -54,10 +69,14 @@ function typeConvertUrlParameters(args: any[]) {
 	return convertedArgs
 }
 
-export function createLegacyApiRouter(methodRegistry: MethodRegistry): KoaRouter {
+export function createLegacyApiRouter(
+	methodRegistry: MethodRegistry,
+	publicationRegistry: PublicationRegistry
+): KoaRouter {
 	const router = new KoaRouter()
 
 	const methodSignatures = methodRegistry.getSignatures()
+	const publicationSignatures = publicationRegistry.getSignatures()
 
 	const index = {
 		version: `${LEGACY_API_VERSION}`,
@@ -87,9 +106,9 @@ export function createLegacyApiRouter(methodRegistry: MethodRegistry): KoaRouter
 	}
 
 	function exposePublication(pubName: string, pubValue: string) {
-		const signature = MeteorPublicationSignatures[pubValue] || []
+		const signature = publicationSignatures[pubValue] || []
 
-		const f = MeteorPublications[pubValue]
+		const f = publicationRegistry.getCursorPublication(pubValue)
 
 		if (f) {
 			let resource = `/publication/${pubName}`
@@ -103,12 +122,10 @@ export function createLegacyApiRouter(methodRegistry: MethodRegistry): KoaRouter
 
 			assignRoute(router, 'GET', resource, signature.length, async (args) => {
 				const convArgs = typeConvertUrlParameters(args)
-				const cursor = await f.apply(
-					{
-						ready: () => null,
-					},
-					convArgs
-				)
+				const cursor = (await f(legacyRestPublicationContext, ...convArgs)) as
+					| { fetch: () => unknown }
+					| null
+					| undefined
 
 				if (cursor) return cursor.fetch()
 				return []

meteor/server/lib/customPublication/index.ts

@@ -2,4 +2,4 @@ export { CustomPublishCollection } from './customPublishCollection'
 export { setUpOptimizedObserverArray } from './optimizedObserverArray'
 export { TriggerUpdate, SetupObserversResult } from './optimizedObserverBase'
 export { setUpCollectionOptimizedObserver } from './optimizedObserverCollection'
-export { meteorCustomPublish, CustomPublish, CustomPublishChanges } from './publish'
+export { CustomPublish, CustomPublishChanges } from './publish'

meteor/server/lib/customPublication/publish.ts

@@ -1,7 +1,6 @@
 import { Meteor } from 'meteor/meteor'
-import { AllPubSubTypes } from '@sofie-automation/meteor-lib/dist/api/pubsub'
 import { ProtectedString, unprotectString } from '@sofie-automation/corelib/dist/protectedString'
-import { PublishDocType, SubscriptionContext, meteorPublishUnsafe } from '../../publications/lib/lib'
+import { PublicationContext } from '../../publications/lib/lib'
 
 export interface CustomPublishChanges<T extends { _id: ProtectedString<any> }> {
 	added: Array<T>
@@ -33,7 +32,7 @@ export class CustomPublishMeteor<DBObj extends { _id: ProtectedString<any> }> {
 	#isReady = false
 
 	constructor(
-		private _meteorSubscription: SubscriptionContext,
+		private _meteorSubscription: PublicationContext,
 		private _collectionName: string
 	) {
 		this._meteorSubscription.onStop(() => {
@@ -86,19 +85,4 @@ export class CustomPublishMeteor<DBObj extends { _id: ProtectedString<any> }> {
 	}
 }
 
-type PublishIfDocument<Doc> = Doc extends { _id: ProtectedString<any> } ? CustomPublish<Doc> : never
-
-/** Wrapping of Meteor.publish to provide types for for custom publications */
-export function meteorCustomPublish<K extends keyof AllPubSubTypes, N extends ReturnType<AllPubSubTypes[K]>>(
-	publicationName: K,
-	customCollectionName: N,
-	cb: (
-		this: SubscriptionContext,
-		publication: PublishIfDocument<PublishDocType<K>>,
-		...args: Parameters<AllPubSubTypes[K]>
-	) => Promise<void>
-): void {
-	meteorPublishUnsafe(publicationName, async function (this: SubscriptionContext, ...args: any[]) {
-		return cb.call(this, new CustomPublishMeteor<any>(this, String(customCollectionName)) as any, ...(args as any))
-	})
-}
+export type PublishIfDocument<Doc> = Doc extends { _id: ProtectedString<any> } ? CustomPublish<Doc> : never

meteor/server/main.ts

@@ -51,22 +51,28 @@ import './logo'
 import './systemTime'
 // import './performanceMonitor' // called above
 
-// Setup publications and security:
-import './publications/_publications'
-import './security/securityVerify'
-
 import { MethodRegistry } from './methodRegistry'
 import { registerAllApiMethods } from './methodRegistrations'
+import { PublicationRegistry } from './publicationRegistry'
+import { registerAllPublications } from './publicationRegistrations'
 import { bindRestApiRouter } from './api/rest/api'
 import { startupVerifyAllMethods } from './security/securityVerify'
 
 // Build and populate the method registry
 const methodRegistry = new MethodRegistry()
 registerAllApiMethods(methodRegistry)
 
-// Apply methods
+// Build and populate the publication registry
+const publicationRegistry = new PublicationRegistry()
+registerAllPublications(publicationRegistry)
+
+// Apply methods and publications
 methodRegistry.applyToMeteor()
+publicationRegistry.applyToMeteor()
 Meteor.startup(() => {
-	bindRestApiRouter(methodRegistry)
+	bindRestApiRouter(methodRegistry, publicationRegistry)
 	startupVerifyAllMethods(methodRegistry)
+
+	// Ensure all the publications were registered at startup
+	if (Meteor.isDevelopment) publicationRegistry.verifyAllPublicationsRegistered()
 })

meteor/server/publicationRegistrations.ts

@@ -0,0 +1,60 @@
+import { PublicationRegistry } from './publicationRegistry'
+
+import { registerBucketsPublications } from './publications/buckets'
+import { registerBlueprintUpgradeStatusPublications } from './publications/blueprintUpgradeStatus/publication'
+import { registerIngestStatusPublications } from './publications/ingestStatus/publication'
+import { registerExternalEventSubscriptionsPublications } from './publications/externalEventSubscriptions'
+import { registerExpectedPackagesPublications } from './publications/packageManager/expectedPackages/publication'
+import { registerPackageContainersPublications } from './publications/packageManager/packageContainers'
+import { registerPlayoutContextPublications } from './publications/packageManager/playoutContext'
+import { registerBucketContentStatusUIPublications } from './publications/pieceContentStatusUI/bucket/publication'
+import { registerRundownContentStatusUIPublications } from './publications/pieceContentStatusUI/rundown/publication'
+import { registerOrganizationPublications } from './publications/organization'
+import { registerPartsUIPublications } from './publications/partsUI/publication'
+import { registerPartInstancesUIPublications } from './publications/partInstancesUI/publication'
+import { registerPeripheralDevicePublications } from './publications/peripheralDevice'
+import { registerPeripheralDeviceForDevicePublications } from './publications/peripheralDeviceForDevice'
+import { registerRundownPublications } from './publications/rundown'
+import { registerRundownPlaylistPublications } from './publications/rundownPlaylist'
+import { registerSegmentPartNotesUIPublications } from './publications/segmentPartNotesUI/publication'
+import { registerShowStylePublications } from './publications/showStyle'
+import { registerShowStyleUIPublications } from './publications/showStyleUI'
+import { registerStudioPublications } from './publications/studio'
+import { registerStudioUIPublications } from './publications/studioUI'
+import { registerSystemPublications } from './publications/system'
+import { registerTimelinePublications } from './publications/timeline'
+import { registerTranslationsBundlesPublications } from './publications/translationsBundles'
+import { registerTriggeredActionsUIPublications } from './publications/triggeredActionsUI'
+import { registerMountedTriggersPublications } from './publications/mountedTriggers'
+import { registerDeviceTriggersPreviewPublications } from './publications/deviceTriggersPreview'
+
+/** Register every DDP publication on the given registry. */
+export function registerAllPublications(registry: PublicationRegistry): void {
+	registerBucketsPublications(registry)
+	registerBlueprintUpgradeStatusPublications(registry)
+	registerIngestStatusPublications(registry)
+	registerExternalEventSubscriptionsPublications(registry)
+	registerExpectedPackagesPublications(registry)
+	registerPackageContainersPublications(registry)
+	registerPlayoutContextPublications(registry)
+	registerBucketContentStatusUIPublications(registry)
+	registerRundownContentStatusUIPublications(registry)
+	registerOrganizationPublications(registry)
+	registerPartsUIPublications(registry)
+	registerPartInstancesUIPublications(registry)
+	registerPeripheralDevicePublications(registry)
+	registerPeripheralDeviceForDevicePublications(registry)
+	registerRundownPublications(registry)
+	registerRundownPlaylistPublications(registry)
+	registerSegmentPartNotesUIPublications(registry)
+	registerShowStylePublications(registry)
+	registerShowStyleUIPublications(registry)
+	registerStudioPublications(registry)
+	registerStudioUIPublications(registry)
+	registerSystemPublications(registry)
+	registerTimelinePublications(registry)
+	registerTranslationsBundlesPublications(registry)
+	registerTriggeredActionsUIPublications(registry)
+	registerMountedTriggersPublications(registry)
+	registerDeviceTriggersPreviewPublications(registry)
+}