wip: tidying and potential perf

Julusian · Julusian · commit 5067d6040ec6 · 2024-12-03T15:49:45.000Z
diff --git a/meteor/server/collections/collection.ts b/meteor/server/collections/collection.ts
@@ -249,7 +249,8 @@ export interface AsyncOnlyReadOnlyMongoCollection<DBInterface extends { _id: Pro
 	observeChanges(
 		selector: MongoQuery<DBInterface> | DBInterface['_id'],
 		callbacks: PromisifyCallbacks<ObserveChangesCallbacks<DBInterface>>,
-		options?: FindOptions<DBInterface>
+		findOptions?: FindOptions<DBInterface>,
+		callbackOptions?: { nonMutatingCallbacks?: boolean | undefined }
 	): Promise<Meteor.LiveQueryHandle>
 
 	/**
diff --git a/meteor/server/collections/implementations/asyncCollection.ts b/meteor/server/collections/implementations/asyncCollection.ts
@@ -137,7 +137,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) {
@@ -148,8 +149,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) {
diff --git a/meteor/server/publications/ingestStatus/publication.ts b/meteor/server/publications/ingestStatus/publication.ts
@@ -54,35 +54,36 @@ async function setupIngestRundownStatusPublicationObservers(
 		const contentObserver = await RundownContentObserver.create(rundownIds, cache)
 
 		const innerQueries = [
-			cache.Playlists.find({}).observeChanges({
-				added: (docId) => triggerUpdate({ invalidatePlaylistIds: [protectString(docId)] }),
-				changed: (docId) => triggerUpdate({ invalidatePlaylistIds: [protectString(docId)] }),
-				removed: (docId) => triggerUpdate({ invalidatePlaylistIds: [protectString(docId)] }),
-			}),
-			cache.Rundowns.find({}).observe({
-				added: (doc) => {
-					triggerUpdate({ invalidateRundownIds: [doc._id] })
-					contentObserver.checkPlaylistIds()
-				},
-				changed: (doc) => {
-					triggerUpdate({ invalidateRundownIds: [doc._id] })
-					contentObserver.checkPlaylistIds()
+			cache.Playlists.find({}).observeChanges(
+				{
+					added: (docId) => triggerUpdate({ invalidatePlaylistIds: [protectString(docId)] }),
+					changed: (docId) => triggerUpdate({ invalidatePlaylistIds: [protectString(docId)] }),
+					removed: (docId) => triggerUpdate({ invalidatePlaylistIds: [protectString(docId)] }),
 				},
-				removed: (doc) => {
-					triggerUpdate({ invalidateRundownIds: [doc._id] })
-					contentObserver.checkPlaylistIds()
+				{ nonMutatingCallbacks: true }
+			),
+			cache.Rundowns.find({}).observeChanges(
+				{
+					added: (docId) => {
+						triggerUpdate({ invalidateRundownIds: [protectString(docId)] })
+						contentObserver.checkPlaylistIds()
+					},
+					changed: (docId) => {
+						triggerUpdate({ invalidateRundownIds: [protectString(docId)] })
+						contentObserver.checkPlaylistIds()
+					},
+					removed: (docId) => {
+						triggerUpdate({ invalidateRundownIds: [protectString(docId)] })
+						contentObserver.checkPlaylistIds()
+					},
 				},
-			}),
+				{ nonMutatingCallbacks: true }
+			),
 			cache.Parts.find({}).observe({
 				added: (doc) => triggerUpdate({ invalidateRundownIds: [doc.rundownId] }),
 				changed: (doc, oldDoc) => triggerUpdate({ invalidateRundownIds: [doc.rundownId, oldDoc.rundownId] }),
 				removed: (doc) => triggerUpdate({ invalidateRundownIds: [doc.rundownId] }),
 			}),
-			// cache.Segments.find({}).observe({
-			// 	added: (doc) => triggerUpdate({ invalidateRundownIds: [doc.rundownId] }),
-			// 	changed: (doc, oldDoc) => triggerUpdate({ invalidateRundownIds: [doc.rundownId, oldDoc.rundownId] }),
-			// 	removed: (doc) => triggerUpdate({ invalidateRundownIds: [doc.rundownId] }),
-			// }),
 			cache.PartInstances.find({}).observe({
 				added: (doc) => triggerUpdate({ invalidateRundownIds: [doc.rundownId] }),
 				changed: (doc, oldDoc) => triggerUpdate({ invalidateRundownIds: [doc.rundownId, oldDoc.rundownId] }),
@@ -116,9 +117,6 @@ async function manipulateIngestRundownStatusPublicationData(
 ): Promise<void> {
 	// Prepare data for publication:
 
-	// We know that `collection` does diffing when 'commiting' all of the changes we have made
-	// meaning that for anything we will call `replace()` on, we can `remove()` it first for no extra cost
-
 	if (updateProps?.newCache !== undefined) {
 		state.contentCache = updateProps.newCache ?? undefined
 	}
diff --git a/meteor/server/publications/ingestStatus/reactiveContentCache.ts b/meteor/server/publications/ingestStatus/reactiveContentCache.ts
@@ -5,7 +5,6 @@ import type { MongoFieldSpecifierOnesStrict } from '@sofie-automation/corelib/di
 import type { PartInstance } from '@sofie-automation/meteor-lib/dist/collections/PartInstances'
 import type { NrcsIngestDataCacheObj } from '@sofie-automation/corelib/dist/dataModel/NrcsIngestDataCache'
 import type { RundownId } from '@sofie-automation/corelib/dist/dataModel/Ids'
-// import type { DBSegment } from '@sofie-automation/corelib/dist/dataModel/Segment'
 import type { DBRundown } from '@sofie-automation/corelib/dist/dataModel/Rundown'
 import type { DBRundownPlaylist } from '@sofie-automation/corelib/dist/dataModel/RundownPlaylist'
 
@@ -41,13 +40,6 @@ export const partFieldSpecifier = literal<MongoFieldSpecifierOnesStrict<Pick<DBP
 	ingestNotifyPartReady: 1,
 })
 
-// export type SegmentFields = '_id' | 'rundownId' | 'externalId'
-// export const segmentFieldSpecifier = literal<MongoFieldSpecifierOnesStrict<Pick<DBSegment, SegmentFields>>>({
-// 	_id: 1,
-// 	rundownId: 1,
-// 	externalId: 1,
-// })
-
 export type PartInstanceFields = '_id' | 'rundownId' | 'segmentId' | 'part'
 export const partInstanceFieldSpecifier = literal<
 	MongoFieldSpecifierOnesStrict<Pick<PartInstance, PartInstanceFields>>
@@ -65,7 +57,6 @@ export interface ContentCache {
 	Rundowns: ReactiveCacheCollection<Pick<DBRundown, RundownFields>>
 	NrcsIngestData: ReactiveCacheCollection<NrcsIngestDataCacheObj>
 	Parts: ReactiveCacheCollection<Pick<DBPart, PartFields>>
-	// Segments: ReactiveCacheCollection<Pick<DBSegment, SegmentFields>>
 	PartInstances: ReactiveCacheCollection<Pick<PartInstance, PartInstanceFields>>
 }
 
@@ -77,7 +68,6 @@ export function createReactiveContentCache(rundownIds: RundownId[]): ContentCach
 		Rundowns: new ReactiveCacheCollection<Pick<DBRundown, RundownFields>>('rundowns'),
 		NrcsIngestData: new ReactiveCacheCollection<NrcsIngestDataCacheObj>('nrcsIngestData'), // TODO - is this needed?
 		Parts: new ReactiveCacheCollection<Pick<DBPart, PartFields>>('parts'),
-		// Segments: new ReactiveCacheCollection<Pick<DBSegment, SegmentFields>>('segments'),
 		PartInstances: new ReactiveCacheCollection<Pick<PartInstance, PartInstanceFields>>('partInstances'),
 	}
 
diff --git a/meteor/server/publications/ingestStatus/rundownContentObserver.ts b/meteor/server/publications/ingestStatus/rundownContentObserver.ts
@@ -73,19 +73,11 @@ export class RundownContentObserver {
 				cache.Rundowns.link(),
 				{
 					projection: rundownFieldSpecifier,
+				},
+				{
+					nonMutatingCallbacks: true,
 				}
 			),
-			// Segments.observeChanges(
-			// 	{
-			// 		rundownId: {
-			// 			$in: rundownIds,
-			// 		},
-			// 	},
-			// 	cache.Segments.link(),
-			// 	{
-			// 		projection: segmentFieldSpecifier,
-			// 	}
-			// ),
 			Parts.observeChanges(
 				{
 					rundownId: {
@@ -95,6 +87,9 @@ export class RundownContentObserver {
 				cache.Parts.link(),
 				{
 					projection: partFieldSpecifier,
+				},
+				{
+					nonMutatingCallbacks: true,
 				}
 			),
 			PartInstances.observeChanges(
@@ -104,15 +99,22 @@ export class RundownContentObserver {
 					orphaned: { $exists: false },
 				},
 				cache.PartInstances.link(),
-				{ fields: partInstanceFieldSpecifier }
+				{ fields: partInstanceFieldSpecifier },
+				{
+					nonMutatingCallbacks: true,
+				}
 			),
 			NrcsIngestDataCache.observeChanges(
 				{
 					rundownId: {
 						$in: rundownIds,
 					},
 				},
-				cache.NrcsIngestData.link()
+				cache.NrcsIngestData.link(),
+				{},
+				{
+					nonMutatingCallbacks: true,
+				}
 			),
 
 			observer.#playlistIdObserver,
