wip: save under new ids for bucket items

Julusian · Julusian · commit b05d92b23ec4 · 2025-04-08T11:04:35.000+01:00
diff --git a/packages/corelib/src/dataModel/ExpectedPackages.ts b/packages/corelib/src/dataModel/ExpectedPackages.ts
@@ -1,6 +1,6 @@
 import { ExpectedPackage, Time } from '@sofie-automation/blueprints-integration'
 import { protectString, unprotectString } from '../protectedString'
-import { assertNever, getHash } from '../lib'
+import { assertNever, getHash, hashObj } from '../lib'
 import {
 	AdLibActionId,
 	BucketAdLibActionId,
@@ -53,8 +53,10 @@ export interface ExpectedPackageDB {
 
 	package: ReadonlyDeep<ExpectedPackage.Any>
 
-	// HACK: This should be ExpectedPackageIngestSource[], but for the first iteration this is limited to a single source
-	ingestSources: [ExpectedPackageIngestSource]
+	/**
+	 * The ingest sources that generated this package.
+	 */
+	ingestSources: ExpectedPackageIngestSource[]
 
 	// playoutSources: {
 	// 	/** Any playout PieceInstance. This is limited to the current and next partInstances */
@@ -183,18 +185,20 @@ export function getExpectedPackageIdFromIngestSource(
 	return protectString(`${parentId}_${ownerId}_${getHash(localExpectedPackageId)}`)
 }
 
-// Future implementation of id generation, once shared ownership is implemented
-// export function getExpectedPackageIdNew(
-// 	/** _id of the rundown*/
-// 	rundownId: RundownId,
-// 	/** The locally unique id of the expectedPackage */
-// 	expectedPackage: ReadonlyDeep<ExpectedPackage.Any>
-// ): ExpectedPackageId {
-// 	// This may be too agressive, but we don't know how to merge some of the properties
-// 	const objHash = hashObj({
-// 		...expectedPackage,
-// 		listenToPackageInfoUpdates: false, // Not relevant for the hash
-// 	} satisfies ReadonlyDeep<ExpectedPackage.Any>)
-
-// 	return protectString(`${rundownId}_${getHash(objHash)}`)
-// }
+/**
+ * Generate the expectedPackageId for the given expectedPackage.
+ */
+export function getExpectedPackageIdNew(
+	/** Preferably a RundownId or BucketId, but StudioId is allowed when not owned by a rundown or bucket */
+	parentId: RundownId | StudioId | BucketId,
+	/** The locally unique id of the expectedPackage */
+	expectedPackage: ReadonlyDeep<ExpectedPackage.Any>
+): ExpectedPackageId {
+	// This may be too agressive, but we don't know how to merge some of the properties
+	const objHash = hashObj({
+		...expectedPackage,
+		listenToPackageInfoUpdates: false, // Not relevant for the hash
+	} satisfies ReadonlyDeep<ExpectedPackage.Any>)
+
+	return protectString(`${parentId}_${getHash(objHash)}`)
+}
diff --git a/packages/job-worker/src/ingest/expectedPackages.ts b/packages/job-worker/src/ingest/expectedPackages.ts
@@ -3,11 +3,10 @@ import { BucketAdLib } from '@sofie-automation/corelib/dist/dataModel/BucketAdLi
 import {
 	ExpectedPackageDBType,
 	ExpectedPackageDB,
-	getExpectedPackageIdFromIngestSource,
 	ExpectedPackageIngestSource,
+	getExpectedPackageIdNew,
 } from '@sofie-automation/corelib/dist/dataModel/ExpectedPackages'
 import { AdLibActionId, PieceId, BucketId } from '@sofie-automation/corelib/dist/dataModel/Ids'
-import { saveIntoDb } from '../db/changes'
 import { PlayoutModel } from '../playout/model/PlayoutModel'
 import { StudioPlayoutModel } from '../studio/model/StudioPlayoutModel'
 import { ReadonlyDeep } from 'type-fest'
@@ -21,7 +20,8 @@ import {
 import { JobContext, JobStudio } from '../jobs'
 import { IngestModel } from './model/IngestModel'
 import { IngestPartModel } from './model/IngestPartModel'
-import { clone, hashObj } from '@sofie-automation/corelib/dist/lib'
+import { hashObj } from '@sofie-automation/corelib/dist/lib'
+import { AnyBulkWriteOperation } from 'mongodb'
 
 export function updateExpectedMediaAndPlayoutItemsForPartModel(context: JobContext, part: IngestPartModel): void {
 	updateExpectedMediaItemsForPartModel(context, part)
@@ -87,89 +87,134 @@ function generateBucketExpectedPackages(
 
 	for (let i = 0; i < expectedPackages.length; i++) {
 		const expectedPackage = expectedPackages[i]
-		const id = expectedPackage._id || '__unnamed' + i
+
+		const fullPackage: ReadonlyDeep<ExpectedPackage.Any> = {
+			...expectedPackage,
+			_id: expectedPackage._id || '__unnamed' + i,
+		}
 
 		bases.push({
-			_id: getExpectedPackageIdFromIngestSource(bucketId, source, id),
-			package: {
-				...clone<ExpectedPackage.Any>(expectedPackage),
-				_id: id,
-			},
+			_id: getExpectedPackageIdNew(bucketId, fullPackage),
+			package: fullPackage,
 			studioId: studio._id,
 			rundownId: null,
 			bucketId: bucketId,
-			created: Date.now(), // This will be preserved during the `saveIntoDb`
+			created: Date.now(), // This will be preserved during the save if needed
 			ingestSources: [source],
 		})
 	}
 
 	return bases
 }
 
-export async function updateExpectedPackagesForBucketAdLibPiece(
+async function writeUpdatedExpectedPackages(
 	context: JobContext,
-	adlib: BucketAdLib
+	bucketId: BucketId,
+	documentsToSave: ExpectedPackageDB[],
+	matchSource: Partial<ExpectedPackageIngestSource>
 ): Promise<void> {
-	const packages = generateExpectedPackagesForBucketAdlib(context.studio, adlib)
-
-	await saveIntoDb(
-		context,
-		context.directCollections.ExpectedPackages,
-		{
-			studioId: context.studioId,
-			bucketId: adlib.bucketId,
-			// Note: This assumes that there is only one ingest source for each piece
-			ingestSources: {
-				$elemMatch: {
-					fromPieceType: ExpectedPackageDBType.BUCKET_ADLIB,
-					pieceId: adlib._id,
+	const writeOps: AnyBulkWriteOperation<ExpectedPackageDB>[] = []
+
+	const documentIdsToSave = documentsToSave.map((doc) => doc._id)
+
+	// Find which documents already exist in the database
+	// It would be nice to avoid this, but that would make the update operation incredibly complex
+	// There is no risk of race conditions, as bucket packages are only modified in the ingest job worker
+	const existingDocIds = new Set(
+		(
+			await context.directCollections.ExpectedPackages.findFetch(
+				{
+					_id: { $in: documentIdsToSave },
+					studioId: context.studioId,
+					bucketId: bucketId,
 				},
-			},
-		},
-		packages,
-		{
-			beforeDiff: (obj, oldObj) => {
-				return {
-					...obj,
-					// Preserve old created timestamp
-					created: oldObj.created,
+				{
+					projection: {
+						_id: 1,
+					},
 				}
-			},
-		}
+			)
+		).map((doc) => doc._id)
 	)
+
+	for (const doc of documentsToSave) {
+		if (existingDocIds.has(doc._id)) {
+			// Document already exists, perform an update to merge the source into the existing document
+			writeOps.push({
+				updateOne: {
+					filter: {
+						_id: doc._id,
+						ingestSources: {
+							// This is pretty messy, but we need to make sure that we don't add the same source twice
+							// nocommit - does this work?
+							$not: {
+								elemMatch: matchSource,
+							},
+						},
+					},
+					update: {
+						$push: {
+							ingestSources: doc.ingestSources[0],
+						},
+					},
+				},
+			})
+		} else {
+			// Perform a simple insert
+			writeOps.push({
+				insertOne: {
+					document: doc,
+				},
+			})
+		}
+	}
+
+	// Remove any old references from this source
+	writeOps.push({
+		updateMany: {
+			filter: {
+				studioId: context.studioId,
+				bucketId: bucketId,
+				_id: { $nin: documentIdsToSave },
+			},
+			update: {
+				$pull: {
+					ingestSources: matchSource,
+				},
+			},
+		},
+	})
+
+	await context.directCollections.ExpectedPackages.bulkWrite(writeOps)
+
+	// Check for any packages that no longer have any sources
+	await cleanUpUnusedPackagesInBucket(context, bucketId)
+}
+
+export async function updateExpectedPackagesForBucketAdLibPiece(
+	context: JobContext,
+	adlib: BucketAdLib
+): Promise<void> {
+	const documentsToSave = generateExpectedPackagesForBucketAdlib(context.studio, adlib)
+
+	await writeUpdatedExpectedPackages(context, adlib.bucketId, documentsToSave, {
+		fromPieceType: ExpectedPackageDBType.BUCKET_ADLIB,
+		pieceId: adlib._id,
+		// nocommit - does this work?
+	})
 }
 
 export async function updateExpectedPackagesForBucketAdLibAction(
 	context: JobContext,
 	action: BucketAdLibAction
 ): Promise<void> {
-	const packages = generateExpectedPackagesForBucketAdlibAction(context.studio, action)
-
-	await saveIntoDb(
-		context,
-		context.directCollections.ExpectedPackages,
-		{
-			studioId: context.studioId,
-			bucketId: action.bucketId,
-			// Note: This assumes that there is only one ingest source for each piece
-			ingestSources: {
-				$elemMatch: {
-					fromPieceType: ExpectedPackageDBType.BUCKET_ADLIB_ACTION,
-					pieceId: action._id,
-				},
-			},
-		},
-		packages,
-		{
-			beforeDiff: (obj, oldObj) => {
-				return {
-					...obj,
-					// Preserve old created timestamp
-					created: oldObj.created,
-				}
-			},
-		}
-	)
+	const documentsToSave = generateExpectedPackagesForBucketAdlibAction(context.studio, action)
+
+	await writeUpdatedExpectedPackages(context, action.bucketId, documentsToSave, {
+		fromPieceType: ExpectedPackageDBType.BUCKET_ADLIB_ACTION,
+		pieceId: action._id,
+		// nocommit - does this work?
+	})
 }
 
 export async function cleanUpExpectedPackagesForBucketAdLibs(
@@ -205,15 +250,19 @@ export async function cleanUpExpectedPackagesForBucketAdLibs(
 		)
 
 		// Remove any expected packages that have now have no owners
-		await context.directCollections.ExpectedPackages.remove({
-			studioId: context.studioId,
-			bucketId: bucketId,
-			ingestSources: { $size: 0 },
-			// Future: these currently can't be referenced by playoutSources, but they could be in the future
-		})
+		await cleanUpUnusedPackagesInBucket(context, bucketId)
 	}
 }
 
+async function cleanUpUnusedPackagesInBucket(context: JobContext, bucketId: BucketId) {
+	await context.directCollections.ExpectedPackages.remove({
+		studioId: context.studioId,
+		bucketId: bucketId,
+		ingestSources: { $size: 0 },
+		// Future: these currently can't be referenced by playoutSources, but they could be in the future
+	})
+}
+
 export function updateBaselineExpectedPackagesOnStudio(
 	context: JobContext,
 	playoutModel: StudioPlayoutModel | PlayoutModel,
