wip: fix types

Author: Julusian

packages/job-worker/src/ingest/generationRundown.ts

@@ -357,8 +357,6 @@ function generateExpectedPackagesForBaseline(
 
 			contentVersionHash: getContentVersionHash(expectedPackage),
 
-			created: null,
-
 			package: expectedPackage,
 
 			ingestSources: [source],

packages/job-worker/src/ingest/model/IngestExpectedPackage.ts

@@ -1,4 +1,4 @@
-import type { ExpectedPackage, Time } from '@sofie-automation/blueprints-integration'
+import type { ExpectedPackage } from '@sofie-automation/blueprints-integration'
 import type {
 	ExpectedPackageDBNew,
 	ExpectedPackageDBType,
@@ -21,9 +21,6 @@ export interface IngestExpectedPackage<
 	/** Hash that changes whenever the content or version changes. See getContentVersionHash() */
 	contentVersionHash: string
 
-	/** The time this expectedPackage was created. This can be null when it has not yet been written to mongodb */
-	created: Time | null
-
 	package: ReadonlyDeep<ExpectedPackage.Any>
 
 	// HACK: Temporary single item
@@ -36,7 +33,6 @@ export function stripExpectedPackageDBToIngestExpectedPackage(
 	return {
 		_id: expectedPackage._id,
 		contentVersionHash: expectedPackage.contentVersionHash,
-		created: expectedPackage.created,
 		package: expectedPackage.package,
 		ingestSources: expectedPackage.ingestSources as any, // nocommit - avoid this cast?
 	}

packages/job-worker/src/ingest/model/implementation/DocumentChangeTracker.ts

@@ -98,6 +98,10 @@ export class DocumentChangeTracker<TDoc extends { _id: ProtectedString<any> }> {
 		return Array.from(this.#deletedIds.values())
 	}
 
+	getDocumentsToSave(): ReadonlyMap<TDoc['_id'], TDoc> {
+		return this.#documentsToSave
+	}
+
 	/**
 	 * Generate the mongodb BulkWrite operations for the documents known to this tracker
 	 * @returns mongodb BulkWrite operations
@@ -125,35 +129,4 @@ export class DocumentChangeTracker<TDoc extends { _id: ProtectedString<any> }> {
 
 		return ops
 	}
-
-	/**
-	 * Generate the mongodb BulkWrite operations for the documents known to this tracker
-	 * @returns mongodb BulkWrite operations
-	 */
-	generateWriteOpsWithTransform<TDoc2 extends { _id: ProtectedString<any> } = TDoc>(
-		transformDocs: (doc: TDoc[]) => TDoc2[]
-	): AnyBulkWriteOperation<TDoc2>[] {
-		const ops: AnyBulkWriteOperation<TDoc2>[] = []
-
-		const transformedDocs = transformDocs(Array.from(this.#documentsToSave.values()))
-		for (const doc of transformedDocs) {
-			ops.push({
-				replaceOne: {
-					filter: { _id: doc._id },
-					replacement: doc,
-					upsert: true,
-				},
-			})
-		}
-
-		if (this.#deletedIds.size > 0) {
-			ops.push({
-				deleteMany: {
-					filter: { _id: { $in: Array.from(this.#deletedIds) as any } },
-				},
-			})
-		}
-
-		return ops
-	}
 }

packages/job-worker/src/ingest/model/implementation/ExpectedPackagesStore.ts

@@ -18,17 +18,6 @@ import {
 import type { IngestExpectedPackage } from '../IngestExpectedPackage'
 import { ExpectedPackageDBType } from '@sofie-automation/corelib/dist/dataModel/ExpectedPackages'
 
-function mutateExpectedPackage<TPackageSource extends { fromPieceType: ExpectedPackageDBType }>(
-	oldObj: IngestExpectedPackage<TPackageSource>,
-	newObj: IngestExpectedPackage<TPackageSource>
-): IngestExpectedPackage<TPackageSource> {
-	return {
-		...newObj,
-		// Retain the created property
-		created: oldObj.created,
-	}
-}
-
 export class ExpectedPackagesStore<TPackageSource extends { fromPieceType: ExpectedPackageDBType }> {
 	#expectedMediaItems: ExpectedMediaItemRundown[]
 	#expectedPlayoutItems: ExpectedPlayoutItemRundown[]
@@ -148,8 +137,7 @@ export class ExpectedPackagesStore<TPackageSource extends { fromPieceType: Expec
 		diffAndReturnLatestObjects(
 			this.#expectedPackagesWithChanges,
 			oldStore.#expectedPackages,
-			this.#expectedPackages,
-			mutateExpectedPackage
+			this.#expectedPackages
 		)
 	}
 
@@ -183,8 +171,7 @@ export class ExpectedPackagesStore<TPackageSource extends { fromPieceType: Expec
 		this.#expectedPackages = diffAndReturnLatestObjects(
 			this.#expectedPackagesWithChanges,
 			this.#expectedPackages,
-			expectedPackages,
-			mutateExpectedPackage
+			expectedPackages
 		)
 	}
 }

packages/job-worker/src/ingest/model/implementation/IngestModelImpl.ts

@@ -659,7 +659,7 @@ export class IngestModelImpl implements IngestModel, IngestDatabasePersistedMode
 			partIds.add(part.part._id)
 		}
 
-		const saveHelper = new SaveIngestModelHelper()
+		const saveHelper = new SaveIngestModelHelper(this.rundownId)
 		for (const [segmentId, segment] of this.segmentsImpl.entries()) {
 			saveHelper.addSegment(segment.segmentModel, segment.deleted)
 			if (segment.deleted) {

packages/job-worker/src/ingest/model/implementation/IngestSegmentModelImpl.ts

@@ -280,8 +280,6 @@ function generateExpectedPackagesForPart(
 
 			contentVersionHash: getContentVersionHash(expectedPackage),
 
-			created: null,
-
 			package: expectedPackage,
 
 			ingestSources: [source],

packages/job-worker/src/ingest/model/implementation/SaveIngestModel.ts

@@ -13,8 +13,13 @@ import { DocumentChangeTracker } from './DocumentChangeTracker'
 import { logger } from '../../../logging'
 import { ProtectedString } from '@sofie-automation/corelib/dist/protectedString'
 import { IngestExpectedPackage } from '../IngestExpectedPackage'
+import { AnyBulkWriteOperation } from 'mongodb'
+import { ExpectedPackageId, RundownId } from '@sofie-automation/corelib/dist/dataModel/Ids'
+import { Complete, normalizeArrayToMap } from '@sofie-automation/corelib/dist/lib'
 
 export class SaveIngestModelHelper {
+	readonly #rundownId: RundownId
+
 	#expectedPackages = new DocumentChangeTracker<IngestExpectedPackage<any>>()
 	#expectedPlayoutItems = new DocumentChangeTracker<ExpectedPlayoutItem>()
 	#expectedMediaItems = new DocumentChangeTracker<ExpectedMediaItem>()
@@ -25,6 +30,10 @@ export class SaveIngestModelHelper {
 	#adLibPieces = new DocumentChangeTracker<AdLibPiece>()
 	#adLibActions = new DocumentChangeTracker<AdLibAction>()
 
+	constructor(rundownId: RundownId) {
+		this.#rundownId = rundownId
+	}
+
 	addExpectedPackagesStore<TPackageSource extends { fromPieceType: ExpectedPackageDBType }>(
 		store: ExpectedPackagesStore<TPackageSource>,
 		deleteAll?: boolean
@@ -75,10 +84,10 @@ export class SaveIngestModelHelper {
 		}
 
 		return [
-			context.directCollections.ExpectedPackages.bulkWrite(
-				this.#expectedPackages.generateWriteOpsWithTransform((docs) =>
-					transformIngestExpectedPackageToDb(context, docs)
-				)
+			writeExpectedPackagesChangesForRundown(
+				context,
+				this.#rundownId,
+				Array.from(this.#expectedPackages.getDocumentsToSave().values())
 			),
 			context.directCollections.ExpectedPlayoutItems.bulkWrite(this.#expectedPlayoutItems.generateWriteOps()),
 			context.directCollections.ExpectedMediaItems.bulkWrite(this.#expectedMediaItems.generateWriteOps()),
@@ -92,14 +101,87 @@ export class SaveIngestModelHelper {
 	}
 }
 
-async function transformIngestExpectedPackageToDb(
+export async function writeExpectedPackagesChangesForRundown(
 	context: JobContext,
-	docs: IngestExpectedPackage<any>[]
-): Promise<ExpectedPackageDBNew[]> {
-	return docs.map((doc) => ({
-		...doc,
-		studioId: context.studioId,
-		rundownId: null,
-		bucketId: null,
-	}))
+	rundownId: RundownId | null,
+	documentsToSave: IngestExpectedPackage<any>[]
+): Promise<void> {
+	const existingDocs = (await context.directCollections.ExpectedPackages.findFetch(
+		{
+			studioId: context.studioId,
+			rundownId: rundownId,
+			bucketId: null,
+		},
+		{
+			projection: {
+				_id: 1,
+				// Future: playoutSources
+			},
+		}
+	)) as Pick<ExpectedPackageDBNew, '_id' | 'created'>[]
+	const existingDocsMap = normalizeArrayToMap(existingDocs, '_id')
+
+	// Generate any insert and update operations
+	const ops: AnyBulkWriteOperation<ExpectedPackageDBNew>[] = []
+	for (const doc of documentsToSave) {
+		const newDbDoc: Complete<Omit<ExpectedPackageDBNew, '_id'>> = {
+			// Future: omit 'playoutSources from this doc
+			studioId: context.studioId,
+			rundownId: rundownId,
+			bucketId: null,
+			created: Date.now(),
+			contentVersionHash: doc.contentVersionHash,
+			package: doc.package,
+			ingestSources: doc.ingestSources,
+		}
+
+		const existingDoc = existingDocsMap.get(doc._id)
+		if (existingDoc) {
+			// Document already exists, perform an update to preserve other fields
+			ops.push({
+				updateOne: {
+					filter: { _id: doc._id },
+					update: {
+						$set: {
+							// Update every field that we want to define
+							...newDbDoc,
+						},
+					},
+				},
+			})
+		} else {
+			// Insert this new document
+			ops.push({
+				insertOne: {
+					document: {
+						...newDbDoc,
+						_id: doc._id,
+					},
+				},
+			})
+		}
+	}
+
+	// Look over the existing documents, and see is no longer referenced
+	const documentsToSaveMap = normalizeArrayToMap(documentsToSave, '_id')
+	const idsToDelete: ExpectedPackageId[] = []
+
+	for (const doc of existingDocs) {
+		// Skip if this document is in the list of documents to save
+		if (documentsToSaveMap.has(doc._id)) continue
+
+		// Future: check for playoutSources
+		idsToDelete.push(doc._id)
+	}
+
+	// const idsToDelete = changeTracker.getDeletedIds()
+	if (idsToDelete.length > 0) {
+		ops.push({
+			deleteMany: {
+				filter: { _id: { $in: idsToDelete as any } },
+			},
+		})
+	}
+
+	if (ops.length > 0) await context.directCollections.ExpectedPackages.bulkWrite(ops)
 }

packages/job-worker/src/studio/model/StudioBaselineHelper.ts

@@ -1,20 +1,22 @@
 import { JobContext } from '../../jobs'
 import {
-	ExpectedPackageDBNew,
 	ExpectedPackageDBType,
 	ExpectedPackageIngestSource,
+	ExpectedPackageIngestSourceStudioBaseline,
 	getContentVersionHash,
 	getExpectedPackageIdFromIngestSource,
 } from '@sofie-automation/corelib/dist/dataModel/ExpectedPackages'
 import { ExpectedPlayoutItemStudio } from '@sofie-automation/corelib/dist/dataModel/ExpectedPlayoutItem'
 import { saveIntoDb } from '../../db/changes'
 import { ReadonlyDeep } from 'type-fest'
 import { ExpectedPackage } from '@sofie-automation/blueprints-integration'
+import type { IngestExpectedPackage } from '../../ingest/model/IngestExpectedPackage'
+import { writeExpectedPackagesChangesForRundown } from '../../ingest/model/implementation/SaveIngestModel'
 
 export class StudioBaselineHelper {
 	readonly #context: JobContext
 
-	#pendingExpectedPackages: ExpectedPackageDBNew[] | undefined
+	#pendingExpectedPackages: IngestExpectedPackage<ExpectedPackageIngestSourceStudioBaseline>[] | undefined
 	#pendingExpectedPlayoutItems: ExpectedPlayoutItemStudio[] | undefined
 
 	constructor(context: JobContext) {
@@ -27,16 +29,12 @@ export class StudioBaselineHelper {
 
 	setExpectedPackages(packages: ReadonlyDeep<ExpectedPackage.Any>[]): void {
 		const source: ExpectedPackageIngestSource = { fromPieceType: ExpectedPackageDBType.STUDIO_BASELINE_OBJECTS }
+
 		this.#pendingExpectedPackages = packages.map((expectedPackage) => ({
 			_id: getExpectedPackageIdFromIngestSource(this.#context.studioId, source, expectedPackage._id),
 
-			studioId: this.#context.studioId,
-			rundownId: null,
-
 			contentVersionHash: getContentVersionHash(expectedPackage),
 
-			created: Date.now(), // nocommit - avoid churn on this?
-
 			package: expectedPackage,
 
 			ingestSources: [source],
@@ -57,19 +55,7 @@ export class StudioBaselineHelper {
 				  )
 				: undefined,
 			this.#pendingExpectedPackages
-				? saveIntoDb<ExpectedPackageDBNew>(
-						this.#context,
-						this.#context.directCollections.ExpectedPackages,
-						{
-							studioId: this.#context.studioId,
-							rundownId: null,
-							bucketId: null,
-						},
-						this.#pendingExpectedPackages,
-						{
-							// nocommit - preserve created timestamps
-						}
-				  )
+				? writeExpectedPackagesChangesForRundown(this.#context, null, this.#pendingExpectedPackages)
 				: undefined,
 		])