wip: theoretical ingest flow

Author: Julusian

packages/corelib/src/dataModel/ExpectedPackages.ts

@@ -32,6 +32,7 @@ export type ExpectedPackageFromRundownBaseline =
 	| ExpectedPackageDBFromBaselineAdLibAction
 	| ExpectedPackageDBFromBaselineAdLibPiece
 	| ExpectedPackageDBFromRundownBaselineObjects
+	| ExpectedPackageDBFromBaselinePiece
 
 export type ExpectedPackageDBFromBucket = ExpectedPackageDBFromBucketAdLib | ExpectedPackageDBFromBucketAdLibAction
 
@@ -47,6 +48,7 @@ export enum ExpectedPackageDBType {
 	ADLIB_ACTION = 'adlib_action',
 	BASELINE_ADLIB_PIECE = 'baseline_adlib_piece',
 	BASELINE_ADLIB_ACTION = 'baseline_adlib_action',
+	BASELINE_PIECE = 'baseline_piece',
 	BUCKET_ADLIB = 'bucket_adlib',
 	BUCKET_ADLIB_ACTION = 'bucket_adlib_action',
 	RUNDOWN_BASELINE_OBJECTS = 'rundown_baseline_objects',
@@ -79,6 +81,13 @@ export interface ExpectedPackageDBFromPiece extends ExpectedPackageDBBase {
 	/** The rundown of the Piece this package belongs to */
 	rundownId: RundownId
 }
+export interface ExpectedPackageDBFromBaselinePiece extends ExpectedPackageDBBase {
+	fromPieceType: ExpectedPackageDBType.BASELINE_PIECE
+	/** The Piece this package belongs to */
+	pieceId: PieceId
+	/** The rundown of the Piece this package belongs to */
+	rundownId: RundownId
+}
 
 export interface ExpectedPackageDBFromBaselineAdLibPiece extends ExpectedPackageDBBase {
 	fromPieceType: ExpectedPackageDBType.BASELINE_ADLIB_PIECE

packages/corelib/src/dataModel/Piece.ts

@@ -53,6 +53,15 @@ export interface PieceGeneric extends Omit<IBlueprintPieceGeneric, 'content'> {
 export interface Piece
 	extends PieceGeneric,
 		Omit<IBlueprintPieceDB, '_id' | 'content' | 'userEditOperations' | 'userEditProperties'> {
+	/** Timeline enabler. When the piece should be active on the timeline. */
+	enable: {
+		start: number | 'now' // TODO - now will be removed from this eventually, but as it is not an acceptable value 99% of the time, that is not really breaking
+		duration?: number
+
+		// Pieces owned by the Rundown should always be absolute
+		isAbsolute?: boolean
+	}
+
 	/**
 	 * This is the id of the rundown this piece starts playing in.
 	 * Currently this is the only rundown the piece could be playing in

packages/job-worker/src/blueprints/postProcess.ts

@@ -13,6 +13,7 @@ import {
 	PieceLifespan,
 	IBlueprintPieceType,
 	ITranslatableMessage,
+	IBlueprintRundownPiece,
 } from '@sofie-automation/blueprints-integration'
 import {
 	AdLibActionId,
@@ -357,6 +358,85 @@ export function postProcessAdLibActions(
 	})
 }
 
+/**
+ * Process and validate some IBlueprintRundownPiece into Piece
+ * @param context Context from the job queue
+ * @param pieces IBlueprintPiece to process
+ * @param blueprintId Id of the Blueprint the Pieces are from
+ * @param rundownId Id of the Rundown the Pieces belong to
+ * @param setInvalid If true all Pieces will be marked as `invalid`, this should be set to match the owning Part
+ */
+export function postProcessGlobalPieces(
+	context: JobContext,
+	pieces: Array<IBlueprintRundownPiece>,
+	blueprintId: BlueprintId,
+	rundownId: RundownId,
+	setInvalid?: boolean
+): Piece[] {
+	const span = context.startSpan('blueprints.postProcess.postProcessPieces')
+
+	const uniqueIds = new Map<string, number>()
+	const timelineUniqueIds = new Set<string>()
+
+	const processedPieces = pieces.map((orgPiece: IBlueprintRundownPiece) => {
+		if (!orgPiece.externalId)
+			throw new Error(
+				`Error in blueprint "${blueprintId}" externalId not set for rundown piece ("${orgPiece.name}")`
+			)
+
+		const docId = getIdHash(
+			'Piece',
+			uniqueIds,
+			`${rundownId}_${blueprintId}_rundown_piece_${orgPiece.sourceLayerId}_${orgPiece.externalId}`
+		)
+
+		const piece: Piece = {
+			...orgPiece,
+			content: omit(orgPiece.content, 'timelineObjects'),
+
+			pieceType: IBlueprintPieceType.Normal,
+			lifespan: PieceLifespan.OutOnRundownChange,
+
+			_id: protectString(docId),
+			startRundownId: rundownId,
+			startSegmentId: null,
+			startPartId: null,
+			invalid: setInvalid ?? false,
+			timelineObjectsString: EmptyPieceTimelineObjectsBlob,
+		}
+
+		if (piece.pieceType !== IBlueprintPieceType.Normal) {
+			// transition pieces must not be infinite, lets enforce that
+			piece.lifespan = PieceLifespan.WithinPart
+		}
+		if (piece.extendOnHold) {
+			// HOLD pieces must not be infinite, as they become that when being held
+			piece.lifespan = PieceLifespan.WithinPart
+		}
+
+		if (piece.enable.start === 'now')
+			throw new Error(
+				`Error in blueprint "${blueprintId}" rundown piece cannot have a start of 'now'! ("${piece.name}")`
+			)
+
+		const timelineObjects = postProcessTimelineObjects(
+			piece._id,
+			blueprintId,
+			orgPiece.content.timelineObjects,
+			timelineUniqueIds
+		)
+		piece.timelineObjectsString = serializePieceTimelineObjectsBlob(timelineObjects)
+
+		// Fill in ids of unnamed expectedPackages
+		setDefaultIdOnExpectedPackages(piece.expectedPackages)
+
+		return piece
+	})
+
+	span?.end()
+	return processedPieces
+}
+
 /**
  * Process and validate TSRTimelineObj for the StudioBaseline into TimelineObjRundown
  * @param blueprintId Id of the Blueprint the TSRTimelineObj are from

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

@@ -254,7 +254,7 @@ export async function updateExpectedMediaItemsForRundownBaseline(
 	const expectedMediaItems = generateExpectedMediaItemsFull(
 		context.studio._id,
 		ingestModel.rundownId,
-		[],
+		ingestModel.getGlobalPieces(),
 		baselineAdlibPieces,
 		baselineAdlibActions
 	)

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

@@ -148,6 +148,21 @@ export async function updateExpectedPackagesForRundownBaseline(
 		preserveTypesDuringSave.add(ExpectedPackageDBType.RUNDOWN_BASELINE_OBJECTS)
 	}
 
+	// Add expected packages for global pieces
+	for (const piece of ingestModel.getGlobalPieces()) {
+		if (piece.expectedPackages) {
+			const bases = generateExpectedPackageBases(context.studio, piece._id, piece.expectedPackages)
+			for (const base of bases) {
+				expectedPackages.push({
+					...base,
+					rundownId: ingestModel.rundownId,
+					pieceId: piece._id,
+					fromPieceType: ExpectedPackageDBType.BASELINE_PIECE,
+				})
+			}
+		}
+	}
+
 	// Preserve anything existing
 	for (const expectedPackage of ingestModel.expectedPackagesForRundownBaseline) {
 		if (preserveTypesDuringSave.has(expectedPackage.fromPieceType)) {

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

@@ -62,6 +62,9 @@ export async function updateExpectedPlayoutItemsForRundownBaseline(
 	for (const action of baselineAdlibActions) {
 		baselineExpectedPlayoutItems.push(...extractExpectedPlayoutItems(studioId, rundownId, undefined, action))
 	}
+	for (const piece of ingestModel.getGlobalPieces()) {
+		baselineExpectedPlayoutItems.push(...extractExpectedPlayoutItems(studioId, rundownId, undefined, piece))
+	}
 
 	if (baseline) {
 		for (const item of baseline.expectedPlayoutItems ?? []) {

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

@@ -11,6 +11,7 @@ import { WatchedPackagesHelper } from '../blueprints/context/watchedPackages'
 import {
 	postProcessAdLibPieces,
 	postProcessGlobalAdLibActions,
+	postProcessGlobalPieces,
 	postProcessRundownBaselineItems,
 } from '../blueprints/postProcess'
 import { logger } from '../logging'
@@ -313,10 +314,9 @@ export async function regenerateRundownAndBaselineFromIngestData(
 	)
 	const globalPieces = postProcessGlobalPieces(
 		context,
+		rundownRes.globalPieces || [],
 		showStyle.base.blueprintId,
-		dbRundown._id,
-		undefined,
-		rundownRes.globalPieces || []
+		dbRundown._id
 	)
 
 	await ingestModel.setRundownBaseline(timelineObjectsBlob, adlibPieces, adlibActions, globalPieces)

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

@@ -2,6 +2,7 @@ import { ExpectedMediaItemRundown } from '@sofie-automation/corelib/dist/dataMod
 import {
 	ExpectedPackageDBFromBaselineAdLibAction,
 	ExpectedPackageDBFromBaselineAdLibPiece,
+	ExpectedPackageDBFromBaselinePiece,
 	ExpectedPackageDBFromRundownBaselineObjects,
 	ExpectedPackageFromRundown,
 } from '@sofie-automation/corelib/dist/dataModel/ExpectedPackages'
@@ -37,6 +38,7 @@ export type ExpectedPackageForIngestModelBaseline =
 	| ExpectedPackageDBFromBaselineAdLibAction
 	| ExpectedPackageDBFromBaselineAdLibPiece
 	| ExpectedPackageDBFromRundownBaselineObjects
+	| ExpectedPackageDBFromBaselinePiece
 export type ExpectedPackageForIngestModel = ExpectedPackageFromRundown | ExpectedPackageForIngestModelBaseline
 
 export interface IngestModelReadonly {
@@ -130,6 +132,11 @@ export interface IngestModelReadonly {
 	 */
 	getAllPieces(): ReadonlyDeep<Piece>[]
 
+	/**
+	 * Get the Pieces which belong to the Rundown, not a Part
+	 */
+	getGlobalPieces(): ReadonlyDeep<Piece>[]
+
 	/**
 	 * Search for a Part through the whole Rundown
 	 * @param id Id of the Part
@@ -251,7 +258,7 @@ export interface IngestModel extends IngestModelReadonly, BaseModel, INotificati
 		timelineObjectsBlob: PieceTimelineObjectsBlob,
 		adlibPieces: RundownBaselineAdLibItem[],
 		adlibActions: RundownBaselineAdLibAction[],
-		pieces: never[]
+		pieces: Piece[]
 	): Promise<void>
 
 	/**

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

@@ -116,6 +116,8 @@ export class IngestModelImpl implements IngestModel, DatabasePersistedModel {
 	}
 
 	protected readonly segmentsImpl: Map<SegmentId, SegmentWrapper>
+	readonly #piecesWithChanges = new Set<PieceId>()
+	#piecesImpl: ReadonlyArray<Piece>
 
 	readonly #rundownBaselineExpectedPackagesStore: ExpectedPackagesStore<ExpectedPackageForIngestModelBaseline>
 
@@ -224,6 +226,8 @@ export class IngestModelImpl implements IngestModel, DatabasePersistedModel {
 				})
 			}
 
+			this.#piecesImpl = groupedPieces.get(null) ?? []
+
 			this.#rundownBaselineObjs = new LazyInitialise(async () =>
 				context.directCollections.RundownBaselineObjects.findFetch({
 					rundownId: this.rundownId,
@@ -253,6 +257,7 @@ export class IngestModelImpl implements IngestModel, DatabasePersistedModel {
 			)
 
 			this.segmentsImpl = new Map()
+			this.#piecesImpl = []
 
 			this.#rundownBaselineObjs = new LazyInitialise(async () => [])
 			this.#rundownBaselineAdLibPieces = new LazyInitialise(async () => [])
@@ -334,6 +339,10 @@ export class IngestModelImpl implements IngestModel, DatabasePersistedModel {
 		return this.getAllOrderedParts().flatMap((part) => part.pieces)
 	}
 
+	getGlobalPieces(): ReadonlyDeep<Piece>[] {
+		return [...this.#piecesImpl]
+	}
+
 	findPart(partId: PartId): IngestPartModel | undefined {
 		for (const segment of this.segmentsImpl.values()) {
 			if (!segment || segment.deleted) continue
@@ -477,7 +486,8 @@ export class IngestModelImpl implements IngestModel, DatabasePersistedModel {
 	async setRundownBaseline(
 		timelineObjectsBlob: PieceTimelineObjectsBlob,
 		adlibPieces: RundownBaselineAdLibItem[],
-		adlibActions: RundownBaselineAdLibAction[]
+		adlibActions: RundownBaselineAdLibAction[],
+		pieces: Piece[]
 	): Promise<void> {
 		const [loadedRundownBaselineObjs, loadedRundownBaselineAdLibPieces, loadedRundownBaselineAdLibActions] =
 			await Promise.all([
@@ -499,11 +509,13 @@ export class IngestModelImpl implements IngestModel, DatabasePersistedModel {
 		)
 
 		// Compare and update the adlibPieces
-		const newAdlibPieces = adlibPieces.map((piece) => ({
-			...clone(piece),
-			partId: undefined,
-			rundownId: this.rundownId,
-		}))
+		const newAdlibPieces = adlibPieces.map(
+			(piece): AdLibPiece => ({
+				...clone(piece),
+				partId: undefined,
+				rundownId: this.rundownId,
+			})
+		)
 		this.#rundownBaselineAdLibPieces.setValue(
 			diffAndReturnLatestObjects(
 				this.#rundownBaselineAdLibPiecesWithChanges,
@@ -513,18 +525,31 @@ export class IngestModelImpl implements IngestModel, DatabasePersistedModel {
 		)
 
 		// Compare and update the adlibActions
-		const newAdlibActions = adlibActions.map((action) => ({
-			...clone(action),
-			partId: undefined,
-			rundownId: this.rundownId,
-		}))
+		const newAdlibActions = adlibActions.map(
+			(action): RundownBaselineAdLibAction => ({
+				...clone(action),
+				partId: undefined,
+				rundownId: this.rundownId,
+			})
+		)
 		this.#rundownBaselineAdLibActions.setValue(
 			diffAndReturnLatestObjects(
 				this.#rundownBaselineAdLibActionsWithChanges,
 				loadedRundownBaselineAdLibActions,
 				newAdlibActions
 			)
 		)
+
+		// Compare and update the rundown pieces
+		const newPieces = pieces.map(
+			(piece): Piece => ({
+				...clone(piece),
+				startRundownId: this.rundownId,
+				startPartId: null,
+				startSegmentId: null,
+			})
+		)
+		this.#piecesImpl = diffAndReturnLatestObjects(this.#piecesWithChanges, this.#piecesImpl, newPieces)
 	}
 
 	setRundownOrphaned(orphaned: RundownOrphanedReason | undefined): void {
@@ -628,9 +653,26 @@ export class IngestModelImpl implements IngestModel, DatabasePersistedModel {
 			for (const segment of this.segmentsImpl.values()) {
 				if (segment.deleted) {
 					logOrThrowError(new Error(`Failed no changes in model assertion, Segment has been changed`))
+					break
 				} else {
 					const err = segment.segmentModel.checkNoChanges()
-					if (err) logOrThrowError(err)
+					if (err) {
+						logOrThrowError(err)
+						break
+					}
+				}
+			}
+
+			if (this.#piecesWithChanges.size) {
+				logOrThrowError(new Error(`Failed no changes in model assertion, Rundown Pieces have been changed`))
+			} else {
+				for (const piece of this.#piecesImpl.values()) {
+					if (!piece) {
+						logOrThrowError(
+							new Error(`Failed no changes in model assertion, Rundown Pieces have been changed`)
+						)
+						break
+					}
 				}
 			}
 		} finally {
@@ -688,6 +730,8 @@ export class IngestModelImpl implements IngestModel, DatabasePersistedModel {
 		saveHelper.addExpectedPackagesStore(this.#rundownBaselineExpectedPackagesStore)
 		this.#rundownBaselineExpectedPackagesStore.clearChangedFlags()
 
+		saveHelper.addChangedPieces(this.#piecesImpl, this.#piecesWithChanges)
+
 		await Promise.all([
 			this.#rundownHasChanged && this.#rundownImpl
 				? this.context.directCollections.Rundowns.replace(this.#rundownImpl)