feat: restructure ExpectedPackages collection in preparation for multiple owners

.github/workflows/node.yaml

@@ -502,16 +502,19 @@ jobs:
           - node-version: 22.x
             package-name: job-worker
             send-coverage: true
-          # No tests for the gateways yet
+          # No tests for some gateways yet
           # - node-version: 22.x
           #   package-name: playout-gateway
-          # - node-version: 22.x
-          #   package-name: mos-gateway
+          #   send-coverage: true
+          - node-version: 22.x
+            package-name: mos-gateway
+            send-coverage: true
           - node-version: 22.x
             package-name: live-status-gateway
             send-coverage: true
           - node-version: 22.x
             package-name: webui
+            send-coverage: true
           # manual meteor-lib as it only needs a couple of versions
           - node-version: 22.x
             package-name: meteor-lib

meteor/__mocks__/helpers/database.ts

@@ -476,6 +476,7 @@ export async function setupMockShowStyleBlueprint(
 						rundown,
 						globalAdLibPieces: [],
 						globalActions: [],
+						globalPieces: [],
 						baseline: { timelineObjects: [] },
 					}
 				},

meteor/server/api/__tests__/cleanup.test.ts

@@ -193,7 +193,7 @@ async function setDefaultDatatoDB(env: DefaultEnvironment, now: number) {
 		startSegmentId: segmentId,
 		timelineObjectsString: '' as any,
 	}
-	const pieceId = await Pieces.mutableCollection.insertAsync(piece)
+	await Pieces.mutableCollection.insertAsync(piece)
 
 	await AdLibActions.mutableCollection.insertAsync({
 		_id: getRandomId(),
@@ -256,22 +256,12 @@ async function setDefaultDatatoDB(env: DefaultEnvironment, now: number) {
 	})
 	const packageId = await ExpectedPackages.mutableCollection.insertAsync({
 		_id: getRandomId(),
-		blueprintPackageId: '',
-		// @ts-expect-error bucketId is not a part of all ExpectedPackageDBs
-		bucketId,
-		content: {} as any,
-		contentVersionHash: '',
-		created: 0,
-		fromPieceType: '' as any,
-		layers: [],
-		pieceId,
-		rundownId,
-		segmentId,
-		sideEffect: {} as any,
 		studioId,
-		sources: {} as any,
-		type: '' as any,
-		version: {} as any,
+		rundownId,
+		bucketId: null,
+		created: 0,
+		package: {} as any,
+		ingestSources: [] as any,
 	})
 	await ExpectedPackageWorkStatuses.insertAsync({
 		_id: getRandomId(),

meteor/server/api/deviceTriggers/TagsService.ts

@@ -4,12 +4,14 @@ import { PieceInstance } from '@sofie-automation/corelib/dist/dataModel/PieceIns
 import { PieceInstanceFields, ContentCache } from './reactiveContentCacheForPieceInstances'
 import { SourceLayers } from '@sofie-automation/corelib/dist/dataModel/ShowStyleBase'
 import {
+	createPartCurrentTimes,
 	PieceInstanceWithTimings,
 	processAndPrunePieceInstanceTimings,
 } from '@sofie-automation/corelib/dist/playout/processAndPrune'
 import { applyAndValidateOverrides } from '@sofie-automation/corelib/dist/settings/objectWithOverrides'
 import { IWrappedAdLib } from '@sofie-automation/meteor-lib/dist/triggers/actionFilterChainCompilers'
 import { areSetsEqual, doSetsIntersect } from '@sofie-automation/corelib/dist/lib'
+import { getCurrentTime } from '../../lib/lib'
 
 export class TagsService {
 	protected onAirPiecesTags: Set<string> = new Set()
@@ -130,12 +132,11 @@ export class TagsService {
 	): PieceInstanceWithTimings[] {
 		// Approximate when 'now' is in the PartInstance, so that any adlibbed Pieces will be timed roughly correctly
 		const partStarted = partInstanceTimings?.plannedStartedPlayback
-		const nowInPart = partStarted === undefined ? 0 : Date.now() - partStarted
 
 		return processAndPrunePieceInstanceTimings(
 			sourceLayers,
 			pieceInstances as PieceInstance[],
-			nowInPart,
+			createPartCurrentTimes(getCurrentTime(), partStarted),
 			false,
 			false
 		)

meteor/server/api/ingest/debug.ts

@@ -8,7 +8,6 @@ import { QueueStudioJob } from '../../worker/worker'
 import { StudioJobs } from '@sofie-automation/corelib/dist/worker/studio'
 import { RundownPlaylistId, SegmentId } from '@sofie-automation/corelib/dist/dataModel/Ids'
 import { MeteorDebugMethods } from '../../methods'
-import { DBRundown } from '@sofie-automation/corelib/dist/dataModel/Rundown'
 
 MeteorDebugMethods({
 	/**
@@ -47,31 +46,4 @@ MeteorDebugMethods({
 			segmentExternalId: segment.externalId,
 		})
 	},
-	/**
-	 * Regenerate all the expected packages for all rundowns in the system.
-	 * Additionally it will recreate any expectedMediaItems and expectedPlayoutItems.
-	 * This shouldn't be necessary as ingest will do this for each rundown as part of its workflow
-	 */
-	debug_recreateExpectedPackages: async () => {
-		const rundowns = (await Rundowns.findFetchAsync(
-			{},
-			{
-				projection: {
-					_id: 1,
-					studioId: 1,
-					source: 1,
-				},
-			}
-		)) as Array<Pick<DBRundown, '_id' | 'studioId' | 'source'>>
-
-		await Promise.all(
-			rundowns
-				.filter((rundown) => rundown.source.type !== 'snapshot')
-				.map(async (rundown) =>
-					runIngestOperation(rundown.studioId, IngestJobs.ExpectedPackagesRegenerate, {
-						rundownId: rundown._id,
-					})
-				)
-		)
-	},
 })

meteor/server/api/ingest/packageInfo.ts

@@ -1,10 +1,7 @@
 import {
-	ExpectedPackageDBFromBucketAdLib,
-	ExpectedPackageDBFromBucketAdLibAction,
-	ExpectedPackageDBFromStudioBaselineObjects,
 	ExpectedPackageDBType,
-	ExpectedPackageFromRundown,
-	ExpectedPackageFromRundownBaseline,
+	ExpectedPackageDB,
+	ExpectedPackageIngestSourceBucket,
 } from '@sofie-automation/corelib/dist/dataModel/ExpectedPackages'
 import { PackageInfoDB } from '@sofie-automation/corelib/dist/dataModel/PackageInfos'
 import { ExpectedPackages, Rundowns } from '../../collections'
@@ -28,51 +25,59 @@ export async function onUpdatedPackageInfo(packageId: ExpectedPackageId, _doc: P
 		return
 	}
 
-	if (pkg.listenToPackageInfoUpdates) {
-		switch (pkg.fromPieceType) {
-			case ExpectedPackageDBType.PIECE:
-			case ExpectedPackageDBType.ADLIB_PIECE:
-			case ExpectedPackageDBType.ADLIB_ACTION:
-			case ExpectedPackageDBType.BASELINE_ADLIB_PIECE:
-			case ExpectedPackageDBType.BASELINE_ADLIB_ACTION:
-			case ExpectedPackageDBType.RUNDOWN_BASELINE_OBJECTS:
-				onUpdatedPackageInfoForRundownDebounce(pkg)
-				break
-			case ExpectedPackageDBType.BUCKET_ADLIB:
-			case ExpectedPackageDBType.BUCKET_ADLIB_ACTION:
-				onUpdatedPackageInfoForBucketItemDebounce(pkg)
-				break
-			case ExpectedPackageDBType.STUDIO_BASELINE_OBJECTS:
-				onUpdatedPackageInfoForStudioBaselineDebounce(pkg)
-				break
-			default:
-				assertNever(pkg)
-				break
+	if (pkg.package.listenToPackageInfoUpdates) {
+		for (const source of pkg.ingestSources) {
+			switch (source.fromPieceType) {
+				case ExpectedPackageDBType.PIECE:
+				case ExpectedPackageDBType.ADLIB_PIECE:
+				case ExpectedPackageDBType.ADLIB_ACTION:
+				case ExpectedPackageDBType.BASELINE_ADLIB_PIECE:
+				case ExpectedPackageDBType.BASELINE_ADLIB_ACTION:
+				case ExpectedPackageDBType.BASELINE_PIECE:
+				case ExpectedPackageDBType.RUNDOWN_BASELINE_OBJECTS:
+					onUpdatedPackageInfoForRundownDebounce(pkg)
+					break
+				case ExpectedPackageDBType.BUCKET_ADLIB:
+				case ExpectedPackageDBType.BUCKET_ADLIB_ACTION:
+					onUpdatedPackageInfoForBucketItemDebounce(pkg, source)
+					break
+				case ExpectedPackageDBType.STUDIO_BASELINE_OBJECTS:
+					onUpdatedPackageInfoForStudioBaselineDebounce(pkg)
+					break
+				default:
+					assertNever(source)
+					break
+			}
 		}
 	}
 }
 
 const pendingRundownPackageUpdates = new Map<RundownId, Array<ExpectedPackageId>>()
-function onUpdatedPackageInfoForRundownDebounce(pkg: ExpectedPackageFromRundown | ExpectedPackageFromRundownBaseline) {
-	const existingEntry = pendingRundownPackageUpdates.get(pkg.rundownId)
+function onUpdatedPackageInfoForRundownDebounce(pkg: ExpectedPackageDB) {
+	if (!pkg.rundownId) {
+		logger.error(`Updating ExpectedPackage "${pkg._id}" for Rundown "${pkg.rundownId}" not possible`)
+		return
+	}
+
+	const rundownId = pkg.rundownId
+
+	const existingEntry = pendingRundownPackageUpdates.get(rundownId)
 	if (existingEntry) {
 		// already queued, add to the batch
 		existingEntry.push(pkg._id)
 	} else {
-		pendingRundownPackageUpdates.set(pkg.rundownId, [pkg._id])
+		pendingRundownPackageUpdates.set(rundownId, [pkg._id])
 	}
 
 	// TODO: Scaling - this won't batch correctly if package manager directs calls to multiple instances
 	lazyIgnore(
-		`onUpdatedPackageInfoForRundown_${pkg.rundownId}`,
+		`onUpdatedPackageInfoForRundown_${rundownId}`,
 		() => {
-			const packageIds = pendingRundownPackageUpdates.get(pkg.rundownId)
+			const packageIds = pendingRundownPackageUpdates.get(rundownId)
 			if (packageIds) {
-				pendingRundownPackageUpdates.delete(pkg.rundownId)
-				onUpdatedPackageInfoForRundown(pkg.rundownId, packageIds).catch((e) => {
-					logger.error(
-						`Updating ExpectedPackages for Rundown "${pkg.rundownId}" failed: ${stringifyError(e)}`
-					)
+				pendingRundownPackageUpdates.delete(rundownId)
+				onUpdatedPackageInfoForRundown(rundownId, packageIds).catch((e) => {
+					logger.error(`Updating ExpectedPackages for Rundown "${rundownId}" failed: ${stringifyError(e)}`)
 				})
 			}
 		},
@@ -107,19 +112,24 @@ async function onUpdatedPackageInfoForRundown(
 	})
 }
 
-function onUpdatedPackageInfoForBucketItemDebounce(
-	pkg: ExpectedPackageDBFromBucketAdLib | ExpectedPackageDBFromBucketAdLibAction
-) {
+function onUpdatedPackageInfoForBucketItemDebounce(pkg: ExpectedPackageDB, source: ExpectedPackageIngestSourceBucket) {
+	if (!pkg.bucketId) {
+		logger.error(`Updating ExpectedPackage "${pkg._id}" for Bucket "${pkg.bucketId}" not possible`)
+		return
+	}
+
+	const bucketId = pkg.bucketId
+
 	lazyIgnore(
-		`onUpdatedPackageInfoForBucket_${pkg.studioId}_${pkg.bucketId}_${pkg.pieceExternalId}`,
+		`onUpdatedPackageInfoForBucket_${pkg.studioId}_${bucketId}_${source.pieceExternalId}`,
 		() => {
 			runIngestOperation(pkg.studioId, IngestJobs.BucketItemRegenerate, {
-				bucketId: pkg.bucketId,
-				externalId: pkg.pieceExternalId,
+				bucketId: bucketId,
+				externalId: source.pieceExternalId,
 			}).catch((err) => {
 				logger.error(
-					`Updating ExpectedPackages for Bucket "${pkg.bucketId}" Item "${
-						pkg.pieceExternalId
+					`Updating ExpectedPackages for Bucket "${bucketId}" Item "${
+						source.pieceExternalId
 					}" failed: ${stringifyError(err)}`
 				)
 			})
@@ -128,7 +138,7 @@ function onUpdatedPackageInfoForBucketItemDebounce(
 	)
 }
 
-function onUpdatedPackageInfoForStudioBaselineDebounce(pkg: ExpectedPackageDBFromStudioBaselineObjects) {
+function onUpdatedPackageInfoForStudioBaselineDebounce(pkg: ExpectedPackageDB) {
 	lazyIgnore(
 		`onUpdatedPackageInfoForStudioBaseline_${pkg.studioId}`,
 		() => {

meteor/server/api/integration/expectedPackages.ts

@@ -33,6 +33,7 @@ import {
 } from '../../collections'
 import { logger } from '../../logging'
 import _ from 'underscore'
+import { ExpectedPackageDB } from '@sofie-automation/corelib/dist/dataModel/ExpectedPackages'
 
 export namespace PackageManagerIntegration {
 	export async function updateExpectedPackageWorkStatuses(
@@ -98,9 +99,17 @@ export namespace PackageManagerIntegration {
 					const fromPackageIds = workStatus.fromPackages.map((p) => p.id)
 					if (fromPackageIds.length) {
 						ps.push(
-							ExpectedPackages.findOneAsync({
-								_id: { $in: fromPackageIds },
-							}).then((expPackage) => {
+							ExpectedPackages.findOneAsync(
+								{
+									_id: { $in: fromPackageIds },
+								},
+								{
+									projection: {
+										_id: 1,
+										studioId: 1,
+									},
+								}
+							).then((expPackage: Pick<ExpectedPackageDB, '_id' | 'studioId'> | undefined) => {
 								if (!expPackage)
 									throw new Meteor.Error(404, `ExpectedPackages "${fromPackageIds}" not found`)
 

meteor/server/api/rest/v1/typeConversion.ts

@@ -396,6 +396,7 @@ export function studioSettingsFrom(apiStudioSettings: APIStudioSettings): Comple
 		enableBuckets: apiStudioSettings.enableBuckets ?? true, // Backwards compatible
 		enableEvaluationForm: apiStudioSettings.enableEvaluationForm ?? true, // Backwards compatible
 		mockPieceContentStatus: apiStudioSettings.mockPieceContentStatus,
+		rundownGlobalPiecesPrepareTime: apiStudioSettings.rundownGlobalPiecesPrepareTime,
 	}
 }
 
@@ -423,6 +424,7 @@ export function APIStudioSettingsFrom(settings: IStudioSettings): Complete<APISt
 		enableBuckets: settings.enableBuckets,
 		enableEvaluationForm: settings.enableEvaluationForm,
 		mockPieceContentStatus: settings.mockPieceContentStatus,
+		rundownGlobalPiecesPrepareTime: settings.rundownGlobalPiecesPrepareTime,
 	}
 }
 

meteor/server/collections/collection.ts

@@ -234,7 +234,8 @@ export interface AsyncOnlyReadOnlyMongoCollection<DBInterface extends { _id: Pro
 	observeChanges(
 		selector: MongoQuery<DBInterface> | DBInterface['_id'],
 		callbacks: PromisifyCallbacks<ObserveChangesCallbacks<DBInterface>>,
-		options?: Omit<FindOptions<DBInterface>, 'fields'>
+		findOptions?: Omit<FindOptions<DBInterface>, 'fields'>,
+		callbackOptions?: { nonMutatingCallbacks?: boolean | undefined }
 	): Promise<Meteor.LiveQueryHandle>
 
 	/**

meteor/server/collections/implementations/asyncCollection.ts

@@ -141,7 +141,8 @@ export class WrappedAsyncMongoCollection<DBInterface extends { _id: ProtectedStr
 	async observeChanges(
 		selector: MongoQuery<DBInterface> | DBInterface['_id'],
 		callbacks: PromisifyCallbacks<ObserveChangesCallbacks<DBInterface>>,
-		options?: FindOptions<DBInterface>
+		findOptions?: FindOptions<DBInterface>,
+		callbackOptions?: { nonMutatingCallbacks?: boolean | undefined }
 	): Promise<Meteor.LiveQueryHandle> {
 		const span = profiler.startSpan(`MongoCollection.${this.name}.observeChanges`)
 		if (span) {
@@ -152,8 +153,8 @@ export class WrappedAsyncMongoCollection<DBInterface extends { _id: ProtectedStr
 		}
 		try {
 			const res = await this._collection
-				.find((selector ?? {}) as any, options as any)
-				.observeChangesAsync(callbacks)
+				.find((selector ?? {}) as any, findOptions as any)
+				.observeChangesAsync(callbacks, callbackOptions)
 			if (span) span.end()
 			return res
 		} catch (e) {

meteor/server/collections/packages-media.ts

@@ -32,15 +32,17 @@ export const ExpectedPackages = createAsyncOnlyReadOnlyMongoCollection<ExpectedP
 )
 registerIndex(ExpectedPackages, {
 	studioId: 1,
-	fromPieceType: 1,
 })
 registerIndex(ExpectedPackages, {
 	studioId: 1,
-	pieceId: 1,
+	rundownId: 1,
+})
+registerIndex(ExpectedPackages, {
+	studioId: 1,
+	bucketId: 1,
 })
 registerIndex(ExpectedPackages, {
 	rundownId: 1,
-	pieceId: 1,
 })
 
 export const ExpectedPackageWorkStatuses = createAsyncOnlyMongoCollection<ExpectedPackageWorkStatus>(

meteor/server/lib/rest/v1/studios.ts

@@ -224,4 +224,5 @@ export interface APIStudioSettings {
 	enableBuckets?: boolean
 	enableEvaluationForm?: boolean
 	mockPieceContentStatus?: boolean
+	rundownGlobalPiecesPrepareTime?: number
 }