wip

Author: Julusian

meteor/server/publications/packageManager/expectedPackages/contentCache.ts

@@ -2,6 +2,31 @@ import { ReactiveCacheCollection } from '../../lib/ReactiveCacheCollection'
 import { literal } from '@sofie-automation/corelib/dist/lib'
 import { MongoFieldSpecifierOnesStrict } from '@sofie-automation/corelib/dist/mongo'
 import { ExpectedPackageDB } from '@sofie-automation/corelib/dist/dataModel/ExpectedPackages'
+import { DBRundownPlaylist } from '@sofie-automation/corelib/dist/dataModel/RundownPlaylist'
+import { PieceInstance } from '@sofie-automation/corelib/dist/dataModel/PieceInstance'
+
+export type RundownPlaylistCompact = Pick<
+	DBRundownPlaylist,
+	'_id' | 'activationId' | 'currentPartInfo' | 'nextPartInfo'
+>
+export const rundownPlaylistFieldSpecifier = literal<MongoFieldSpecifierOnesStrict<RundownPlaylistCompact>>({
+	_id: 1,
+	activationId: 1,
+	currentPartInfo: 1, // So that it invalidates when the current changes
+	nextPartInfo: 1, // So that it invalidates when the next changes
+})
+
+export type PieceInstanceCompact = Pick<
+	PieceInstance,
+	'_id' | 'rundownId' | 'partInstanceId' | 'neededExpectedPackageIds'
+>
+
+export const pieceInstanceFieldsSpecifier = literal<MongoFieldSpecifierOnesStrict<PieceInstanceCompact>>({
+	_id: 1,
+	rundownId: 1,
+	partInstanceId: 1,
+	neededExpectedPackageIds: 1,
+})
 
 export type ExpectedPackageDBCompact = Pick<ExpectedPackageDB, '_id' | 'package'>
 
@@ -12,11 +37,15 @@ export const expectedPackageDBFieldsSpecifier = literal<MongoFieldSpecifierOnesS
 
 export interface ExpectedPackagesContentCache {
 	ExpectedPackages: ReactiveCacheCollection<ExpectedPackageDBCompact>
+	RundownPlaylists: ReactiveCacheCollection<RundownPlaylistCompact>
+	PieceInstances: ReactiveCacheCollection<PieceInstanceCompact>
 }
 
 export function createReactiveContentCache(): ExpectedPackagesContentCache {
 	const cache: ExpectedPackagesContentCache = {
 		ExpectedPackages: new ReactiveCacheCollection<ExpectedPackageDBCompact>('expectedPackages'),
+		RundownPlaylists: new ReactiveCacheCollection<RundownPlaylistCompact>('rundownPlaylists'),
+		PieceInstances: new ReactiveCacheCollection<PieceInstanceCompact>('pieceInstances'),
 	}
 
 	return cache

meteor/server/publications/packageManager/expectedPackages/contentObserver.ts

@@ -1,14 +1,29 @@
 import { Meteor } from 'meteor/meteor'
-import { StudioId } from '@sofie-automation/corelib/dist/dataModel/Ids'
+import { PartInstanceId, StudioId } from '@sofie-automation/corelib/dist/dataModel/Ids'
 import { logger } from '../../../logging'
-import { ExpectedPackagesContentCache, expectedPackageDBFieldsSpecifier } from './contentCache'
-import { ExpectedPackages } from '../../../collections'
+import {
+	ExpectedPackagesContentCache,
+	expectedPackageDBFieldsSpecifier,
+	pieceInstanceFieldsSpecifier,
+	rundownPlaylistFieldSpecifier,
+} from './contentCache'
+import { ExpectedPackages, PieceInstances, RundownPlaylists } from '../../../collections'
 import { waitForAllObserversReady } from '../../lib/lib'
+import { ReactiveMongoObserverGroup, ReactiveMongoObserverGroupHandle } from '../../lib/observerGroup'
+import { equivalentArrays } from '@sofie-automation/shared-lib/dist/lib/lib'
+import _ from 'underscore'
+
+const REACTIVITY_DEBOUNCE = 20
 
 export class ExpectedPackagesContentObserver implements Meteor.LiveQueryHandle {
 	#observers: Meteor.LiveQueryHandle[] = []
 	#cache: ExpectedPackagesContentCache
 
+	#partInstanceIds: PartInstanceId[] = []
+	#partInstanceIdObserver!: ReactiveMongoObserverGroupHandle
+
+	#disposed = false
+
 	private constructor(cache: ExpectedPackagesContentCache) {
 		this.#cache = cache
 	}
@@ -21,7 +36,27 @@ export class ExpectedPackagesContentObserver implements Meteor.LiveQueryHandle {
 
 		const observer = new ExpectedPackagesContentObserver(cache)
 
+		// Run the ShowStyleBase query in a ReactiveMongoObserverGroup, so that it can be restarted whenever
+		observer.#partInstanceIdObserver = await ReactiveMongoObserverGroup(async () => {
+			// Clear already cached data
+			cache.PieceInstances.remove({})
+
+			return [
+				PieceInstances.observeChanges(
+					{
+						// We can use the `this.#partInstanceIds` here, as this is restarted every time that property changes
+						partInstanceId: { $in: observer.#partInstanceIds },
+					},
+					cache.PieceInstances.link(),
+					{
+						projection: pieceInstanceFieldsSpecifier,
+					}
+				),
+			]
+		})
+
 		// Subscribe to the database, and pipe any updates into the ReactiveCacheCollections
+		// This takes ownership of the #partInstanceIdObserver, and will stop it if this throws
 		observer.#observers = await waitForAllObserversReady([
 			ExpectedPackages.observeChanges(
 				{
@@ -32,16 +67,60 @@ export class ExpectedPackagesContentObserver implements Meteor.LiveQueryHandle {
 					projection: expectedPackageDBFieldsSpecifier,
 				}
 			),
+
+			RundownPlaylists.observeChanges(
+				{
+					studioId: studioId,
+				},
+				cache.RundownPlaylists.link(() => {
+					observer.updatePartInstanceIds()
+				}),
+				{
+					fields: rundownPlaylistFieldSpecifier,
+				}
+			),
+
+			observer.#partInstanceIdObserver,
 		])
 
 		return observer
 	}
 
+	private updatePartInstanceIds = _.debounce(
+		Meteor.bindEnvironment(() => {
+			if (this.#disposed) return
+
+			const newPartInstanceIdsSet = new Set<PartInstanceId>()
+
+			this.#cache.RundownPlaylists.find({}).forEach((playlist) => {
+				if (playlist.activationId) {
+					if (playlist.nextPartInfo) {
+						newPartInstanceIdsSet.add(playlist.nextPartInfo.partInstanceId)
+					}
+					if (playlist.currentPartInfo) {
+						newPartInstanceIdsSet.add(playlist.currentPartInfo.partInstanceId)
+					}
+				}
+			})
+
+			const newPartInstanceIds = Array.from(newPartInstanceIdsSet)
+
+			if (!equivalentArrays(newPartInstanceIds, this.#partInstanceIds)) {
+				this.#partInstanceIds = newPartInstanceIds
+				// trigger the rundown group to restart
+				this.#partInstanceIdObserver.restart()
+			}
+		}),
+		REACTIVITY_DEBOUNCE
+	)
+
 	public get cache(): ExpectedPackagesContentCache {
 		return this.#cache
 	}
 
 	public stop = (): void => {
+		this.#disposed = true
+
 		this.#observers.forEach((observer) => observer.stop())
 	}
 }

meteor/server/publications/packageManager/expectedPackages/generate.ts

@@ -14,7 +14,7 @@ import { DBStudio, StudioLight, StudioPackageContainer } from '@sofie-automation
 import { clone, omit } from '../../../lib/tempLib'
 import { CustomPublishCollection } from '../../../lib/customPublication'
 import { logger } from '../../../logging'
-import { ExpectedPackagesContentCache } from './contentCache'
+import { ExpectedPackageDBCompact, ExpectedPackagesContentCache } from './contentCache'
 import type { StudioFields } from './publication'
 
 /**
@@ -60,14 +60,9 @@ export async function updateCollectionForExpectedPackageIds(
 
 			const routedPackage = generateExpectedPackageForDevice(
 				studio,
-				{
-					...packageDoc.package,
-					_id: unprotectString(packageDoc._id),
-				},
+				packageDoc,
 				deviceId,
 				null,
-				// nocommit - how to handle this priority in the new flow?
-				Priorities.OTHER, // low priority
 				packageContainers
 			)
 
@@ -88,7 +83,7 @@ export async function updateCollectionForExpectedPackageIds(
 	})
 }
 
-enum Priorities {
+export enum ExpectedPackagePriorities {
 	// Lower priorities are done first
 
 	/** Highest priority */
@@ -103,16 +98,15 @@ function generateExpectedPackageForDevice(
 		StudioLight,
 		'_id' | 'packageContainersWithOverrides' | 'previewContainerIds' | 'thumbnailContainerIds'
 	>,
-	expectedPackage: PackageManagerExpectedPackageBase,
+	expectedPackage: ExpectedPackageDBCompact,
 	deviceId: PeripheralDeviceId,
 	pieceInstanceId: PieceInstanceId | null,
-	priority: Priorities,
 	packageContainers: Record<string, StudioPackageContainer>
 ): PackageManagerExpectedPackage {
 	// Lookup Package sources:
 	const combinedSources: PackageContainerOnPackage[] = []
 
-	for (const packageSource of expectedPackage.sources) {
+	for (const packageSource of expectedPackage.package.sources) {
 		const lookedUpSource = packageContainers[packageSource.containerId]
 		if (lookedUpSource) {
 			combinedSources.push(calculateCombinedSource(packageSource, lookedUpSource))
@@ -130,25 +124,26 @@ function generateExpectedPackageForDevice(
 	}
 
 	// Lookup Package targets:
-	const combinedTargets = calculateCombinedTargets(expectedPackage, deviceId, packageContainers)
+	const combinedTargets = calculateCombinedTargets(expectedPackage.package, deviceId, packageContainers)
 
-	if (!combinedSources.length && expectedPackage.sources.length !== 0) {
+	if (!combinedSources.length && expectedPackage.package.sources.length !== 0) {
 		logger.warn(`Pub.expectedPackagesForDevice: No sources found for "${expectedPackage._id}"`)
 	}
 	if (!combinedTargets.length) {
 		logger.warn(`Pub.expectedPackagesForDevice: No targets found for "${expectedPackage._id}"`)
 	}
-	const packageSideEffect = getSideEffect(expectedPackage, studio)
+	const packageSideEffect = getSideEffect(expectedPackage.package, studio)
 
 	return {
 		_id: protectString(`${expectedPackage._id}_${deviceId}_${pieceInstanceId}`),
+		expectedPackageId: expectedPackage._id,
 		expectedPackage: {
-			...expectedPackage,
+			...expectedPackage.package,
 			sideEffect: packageSideEffect,
 		},
 		sources: combinedSources,
 		targets: combinedTargets,
-		priority: priority,
+		priority: ExpectedPackagePriorities.OTHER, // This gets overriden later if needed
 		playoutDeviceId: deviceId,
 		pieceInstanceId,
 	}

meteor/server/publications/packageManager/expectedPackages/publication.ts

@@ -11,14 +11,19 @@ import { logger } from '../../../logging'
 import { ReadonlyDeep } from 'type-fest'
 import { applyAndValidateOverrides } from '@sofie-automation/corelib/dist/settings/objectWithOverrides'
 import { MongoFieldSpecifierOnesStrict } from '@sofie-automation/corelib/dist/mongo'
-import { ExpectedPackageId, PeripheralDeviceId, StudioId } from '@sofie-automation/corelib/dist/dataModel/Ids'
+import {
+	ExpectedPackageId,
+	PeripheralDeviceId,
+	PieceInstanceId,
+	StudioId,
+} from '@sofie-automation/corelib/dist/dataModel/Ids'
 import { Studios } from '../../../collections'
 import { check, Match } from 'meteor/check'
 import { PackageManagerExpectedPackage } from '@sofie-automation/shared-lib/dist/package-manager/publications'
 import { ExpectedPackagesContentObserver } from './contentObserver'
 import { createReactiveContentCache, ExpectedPackagesContentCache } from './contentCache'
 import { buildMappingsToDeviceIdMap } from './util'
-import { updateCollectionForExpectedPackageIds } from './generate'
+import { ExpectedPackagePriorities, updateCollectionForExpectedPackageIds } from './generate'
 import {
 	PeripheralDevicePubSub,
 	PeripheralDevicePubSubCollectionsNames,
@@ -37,6 +42,7 @@ interface ExpectedPackagesPublicationUpdateProps {
 	newCache: ExpectedPackagesContentCache
 
 	invalidateExpectedPackageIds?: ExpectedPackageId[]
+	invalidatePieceInstanceIds?: PieceInstanceId[]
 }
 
 interface ExpectedPackagesPublicationState {
@@ -81,6 +87,11 @@ async function setupExpectedPackagesPublicationObservers(
 			changed: (id) => triggerUpdate({ invalidateExpectedPackageIds: [protectString<ExpectedPackageId>(id)] }),
 			removed: (id) => triggerUpdate({ invalidateExpectedPackageIds: [protectString<ExpectedPackageId>(id)] }),
 		}),
+		contentCache.PieceInstances.find({}).observeChanges({
+			added: (id) => triggerUpdate({ invalidatePieceInstanceIds: [protectString<PieceInstanceId>(id)] }),
+			changed: (id) => triggerUpdate({ invalidatePieceInstanceIds: [protectString<PieceInstanceId>(id)] }),
+			removed: (id) => triggerUpdate({ invalidatePieceInstanceIds: [protectString<PieceInstanceId>(id)] }),
+		}),
 
 		Studios.observeChanges(
 			args.studioId,
@@ -151,7 +162,7 @@ async function manipulateExpectedPackagesPublicationData(
 
 	let regenerateExpectedPackageIds: Set<ExpectedPackageId>
 	if (invalidateAllItems) {
-		// force every piece to be regenerated
+		// force every package to be regenerated
 		collection.remove(null)
 		regenerateExpectedPackageIds = new Set(state.contentCache.ExpectedPackages.find({}).map((p) => p._id))
 	} else {
@@ -168,6 +179,49 @@ async function manipulateExpectedPackagesPublicationData(
 		args.filterPlayoutDeviceIds,
 		regenerateExpectedPackageIds
 	)
+
+	// Ensure the priorities are correct for the packages
+	// We can do this as a post-step, as it means we can generate the packages solely based on the content
+	// If one gets regenerated, its priority will be reset to OTHER. But as it has already changed, this fixup is 'free'
+	// For those not regenerated, we can set the priority to the correct value if it has changed, without any deeper checks
+	updatePackagePriorities(state.contentCache, collection)
+}
+
+function updatePackagePriorities(
+	contentCache: ReadonlyDeep<ExpectedPackagesContentCache>,
+	collection: CustomPublishCollection<PackageManagerExpectedPackage>
+) {
+	const highPriorityPackages = new Map<ExpectedPackageId, ExpectedPackagePriorities>()
+
+	// Compile the map of the expected priority of each package
+	const knownPieceInstances = contentCache.PieceInstances.find({})
+	const playlist = contentCache.RundownPlaylists.findOne({})
+	const currentPartInstanceId = playlist?.currentPartInfo?.partInstanceId
+	for (const pieceInstance of knownPieceInstances) {
+		const packageIds = pieceInstance.neededExpectedPackageIds
+		if (!packageIds) continue
+
+		const packagePriority =
+			pieceInstance.partInstanceId === currentPartInstanceId
+				? ExpectedPackagePriorities.PLAYOUT_CURRENT
+				: ExpectedPackagePriorities.PLAYOUT_NEXT
+
+		for (const packageId of packageIds) {
+			const existingPriority = highPriorityPackages.get(packageId) ?? ExpectedPackagePriorities.OTHER
+			highPriorityPackages.set(packageId, Math.min(existingPriority, packagePriority))
+		}
+	}
+
+	// Iterate through and update each package
+	collection.updateAll((pkg) => {
+		const expectedPriority = highPriorityPackages.get(pkg.expectedPackageId) ?? ExpectedPackagePriorities.OTHER
+		if (pkg.priority === expectedPriority) return false
+
+		return {
+			...pkg,
+			priority: expectedPriority,
+		}
+	})
 }
 
 meteorCustomPublish(

packages/shared-lib/src/package-manager/publications.ts

@@ -1,5 +1,5 @@
 import { ExpectedPackage, PackageContainer, PackageContainerOnPackage } from './package'
-import { PeripheralDeviceId, PieceInstanceId, RundownId, RundownPlaylistId } from '../core/model/Ids'
+import { ExpectedPackageId, PeripheralDeviceId, PieceInstanceId, RundownId, RundownPlaylistId } from '../core/model/Ids'
 import { ProtectedString } from '../lib/protectedString'
 import { ReadonlyDeep } from 'type-fest'
 
@@ -34,6 +34,8 @@ export interface PackageManagerExpectedPackage {
 	/** Unique id of the expectedPackage */
 	_id: PackageManagerExpectedPackageId
 
+	expectedPackageId: ExpectedPackageId
+
 	expectedPackage: PackageManagerExpectedPackageBase
 	/** Lower should be done first */
 	priority: number