fix: ensure cleanup of some observer helpers

Author: Julusian

meteor/server/api/deviceTriggers/RundownsObserver.ts

@@ -21,12 +21,18 @@ export class RundownsObserver {
 	#changed: ChangedHandler | undefined
 	#cleanup: (() => void) | undefined
 
+	#disposed = false
+
 	readonly #triggerUpdateRundownContent = new PromiseDebounce(async () => {
+		if (this.#disposed) return
+
 		if (!this.#changed) return
 		this.#cleanup?.()
 
 		const changed = this.#changed
 		this.#cleanup = await changed(this.rundownIds)
+
+		if (this.#disposed) this.#cleanup?.()
 	}, REACTIVITY_DEBOUNCE)
 
 	private constructor(onChanged: ChangedHandler) {
@@ -69,6 +75,8 @@ export class RundownsObserver {
 	}
 
 	public stop = (): void => {
+		this.#disposed = true
+
 		this.#triggerUpdateRundownContent.cancelWaiting()
 		this.#rundownsLiveQuery.stop()
 		this.#changed = undefined

meteor/server/api/deviceTriggers/StudioObserver.ts

@@ -67,6 +67,8 @@ export class StudioObserver extends EventEmitter {
 
 	#changed: ChangedHandler
 
+	#disposed = false
+
 	constructor(studioId: StudioId, onChanged: ChangedHandler) {
 		super()
 		this.#changed = onChanged
@@ -94,6 +96,8 @@ export class StudioObserver extends EventEmitter {
 					activePlaylist: Pick<DBRundownPlaylist, RundownPlaylistFields>
 				} | null
 			): void => {
+				if (this.#disposed) return
+
 				const activePlaylistId = state?.activePlaylist?._id
 				const activationId = state?.activePlaylist?.activationId
 				const currentRundownId =
@@ -159,6 +163,8 @@ export class StudioObserver extends EventEmitter {
 			} | null
 		]
 	>(async (state): Promise<void> => {
+		if (this.#disposed) return
+
 		const showStyleBaseId = state?.showStyleBase._id
 
 		if (showStyleBaseId === undefined || !this.nextProps?.activePlaylistId || !this.nextProps?.activationId) {
@@ -187,28 +193,30 @@ export class StudioObserver extends EventEmitter {
 
 		this.showStyleBaseId = showStyleBaseId
 
-		let cleanupChanges: (() => void) | undefined = undefined
-
 		this.#rundownsLiveQuery = await RundownsObserver.create(activePlaylistId, async (rundownIds) => {
 			logger.silly(`Creating new RundownContentObserver`)
-			const obs1 = await RundownContentObserver.create(activePlaylistId, showStyleBaseId, rundownIds, (cache) => {
-				cleanupChanges = this.#changed(showStyleBaseId, cache)
 
-				return () => {
-					void 0
-				}
+			const obs1 = await RundownContentObserver.create(activePlaylistId, showStyleBaseId, rundownIds, (cache) => {
+				return this.#changed(showStyleBaseId, cache)
 			})
 
 			return () => {
 				obs1.stop()
-				cleanupChanges?.()
 			}
 		})
+
+		if (this.#disposed) {
+			// If we were disposed of while waiting for the observer to be created, stop it immediately
+			this.#rundownsLiveQuery.stop()
+		}
 	}, REACTIVITY_DEBOUNCE)
 
 	public stop = (): void => {
+		this.#disposed = true
+
 		this.updateShowStyle.cancelWaiting()
 		this.#playlistInStudioLiveQuery.stop()
 		this.updatePlaylistInStudio.cancel()
+		this.#rundownsLiveQuery?.stop()
 	}
 }

meteor/server/publications/lib/rundownsObserver.ts

@@ -17,12 +17,17 @@ export class RundownsObserver implements Meteor.LiveQueryHandle {
 	#changed: ChangedHandler | undefined
 	#cleanup: (() => void) | undefined
 
+	#disposed = false
+
 	readonly #triggerUpdateRundownContent = new PromiseDebounce(async () => {
+		if (this.#disposed) return
 		if (!this.#changed) return
 		this.#cleanup?.()
 
 		const changed = this.#changed
 		this.#cleanup = await changed(this.rundownIds)
+
+		if (this.#disposed) this.#cleanup?.()
 	}, REACTIVITY_DEBOUNCE)
 
 	private constructor(onChanged: ChangedHandler) {
@@ -76,6 +81,8 @@ export class RundownsObserver implements Meteor.LiveQueryHandle {
 	}
 
 	public stop = (): void => {
+		this.#disposed = true
+
 		this.#triggerUpdateRundownContent.cancelWaiting()
 		this.#rundownsLiveQuery.stop()
 		this.#changed = undefined

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

@@ -21,6 +21,8 @@ export class ExpectedPackagesContentObserver implements Meteor.LiveQueryHandle {
 	#partInstanceIds: PartInstanceId[] = []
 	#partInstanceIdObserver!: ReactiveMongoObserverGroupHandle
 
+	#disposed = false
+
 	private constructor(cache: ExpectedPackagesContentCache) {
 		this.#cache = cache
 	}
@@ -82,6 +84,8 @@ export class ExpectedPackagesContentObserver implements Meteor.LiveQueryHandle {
 
 	private updatePartInstanceIds = _.debounce(
 		Meteor.bindEnvironment(() => {
+			if (this.#disposed) return
+
 			const newPartInstanceIdsSet = new Set<PartInstanceId>()
 
 			this.#cache.RundownPlaylists.find({}).forEach((playlist) => {
@@ -111,6 +115,8 @@ export class ExpectedPackagesContentObserver implements Meteor.LiveQueryHandle {
 	}
 
 	public stop = (): void => {
+		this.#disposed = true
+
 		this.#observers.forEach((observer) => observer.stop())
 	}
 }

meteor/server/publications/pieceContentStatusUI/bucket/bucketContentObserver.ts

@@ -33,6 +33,8 @@ export class BucketContentObserver implements Meteor.LiveQueryHandle {
 	#showStyleBaseIds: ShowStyleBaseId[] = []
 	#showStyleBaseIdObserver!: ReactiveMongoObserverGroupHandle
 
+	#disposed = false
+
 	private constructor(cache: BucketContentCache) {
 		this.#cache = cache
 	}
@@ -111,6 +113,8 @@ export class BucketContentObserver implements Meteor.LiveQueryHandle {
 
 	private updateShowStyleBaseIds = _.debounce(
 		Meteor.bindEnvironment(() => {
+			if (this.#disposed) return
+
 			const newShowStyleBaseIdsSet = new Set<ShowStyleBaseId>()
 			this.#cache.BucketAdLibs.find({}).forEach((adlib) => newShowStyleBaseIdsSet.add(adlib.showStyleBaseId))
 			this.#cache.BucketAdLibActions.find({}).forEach((action) =>
@@ -133,6 +137,8 @@ export class BucketContentObserver implements Meteor.LiveQueryHandle {
 	}
 
 	public stop = (): void => {
+		this.#disposed = true
+
 		this.#observers.forEach((observer) => observer.stop())
 	}
 }