wip: more async of observeChanges and observe

Julusian · Julusian · commit 650ba00f8dfa · 2024-10-25T12:22:18.000+01:00
diff --git a/meteor/server/__tests__/_testEnvironment.test.ts b/meteor/server/__tests__/_testEnvironment.test.ts
@@ -153,7 +153,7 @@ describe('Basic test of test environment', () => {
 		const studios = await Studios.findFetchAsync({})
 		expect(studios).toHaveLength(1)
 
-		const observer = Studios.observeChanges({ _id: protectString('abc') }, {})
+		const observer = await Studios.observeChanges({ _id: protectString('abc') }, {})
 		expect(observer).toBeTruthy()
 
 		await Studios.insertAsync({
diff --git a/meteor/server/publications/lib/__tests__/observerGroup.test.ts b/meteor/server/publications/lib/__tests__/observerGroup.test.ts
@@ -10,7 +10,7 @@ describe('ReactiveMongoObserverGroup', () => {
 
 	test('cleanup on stop', async () => {
 		const handle: LiveQueryHandle = { stop: jest.fn() }
-		const generator = jest.fn(async () => [handle])
+		const generator = jest.fn(async () => [Promise.resolve(handle)])
 
 		const observerGroup = await ReactiveMongoObserverGroup(generator)
 
@@ -39,7 +39,7 @@ describe('ReactiveMongoObserverGroup', () => {
 
 	test('restarting', async () => {
 		const handle: LiveQueryHandle = { stop: jest.fn() }
-		const generator = jest.fn(async () => [handle])
+		const generator = jest.fn(async () => [Promise.resolve(handle)])
 
 		const observerGroup = await ReactiveMongoObserverGroup(generator)
 
@@ -80,7 +80,7 @@ describe('ReactiveMongoObserverGroup', () => {
 
 	test('restart debounce', async () => {
 		const handle: LiveQueryHandle = { stop: jest.fn() }
-		const generator = jest.fn(async () => [handle])
+		const generator = jest.fn(async () => [Promise.resolve(handle)])
 
 		const observerGroup = await ReactiveMongoObserverGroup(generator)
 
diff --git a/meteor/server/publications/lib/__tests__/rundownsObserver.test.ts b/meteor/server/publications/lib/__tests__/rundownsObserver.test.ts
@@ -20,12 +20,12 @@ describe('RundownsObserver', () => {
 		const playlistId = protectString<RundownPlaylistId>('playlist0')
 
 		const onChangedCleanup = jest.fn()
-		const onChanged = jest.fn(() => onChangedCleanup)
+		const onChanged = jest.fn(async () => onChangedCleanup)
 
 		// should not be any observers yet
 		expect(RundownsMock.observers).toHaveLength(0)
 
-		const observer = new RundownsObserver(studioId, playlistId, onChanged)
+		const observer = await RundownsObserver.create(studioId, playlistId, onChanged)
 		try {
 			// should now be an observer
 			expect(RundownsMock.observers).toHaveLength(1)
@@ -73,12 +73,12 @@ describe('RundownsObserver', () => {
 		const playlistId = protectString<RundownPlaylistId>('playlist0')
 
 		const onChangedCleanup = jest.fn()
-		const onChanged = jest.fn<() => void, [RundownId[]]>(() => onChangedCleanup)
+		const onChanged = jest.fn<Promise<() => void>, [RundownId[]]>(async () => onChangedCleanup)
 
 		// should not be any observers yet
 		expect(RundownsMock.observers).toHaveLength(0)
 
-		const observer = new RundownsObserver(studioId, playlistId, onChanged)
+		const observer = await RundownsObserver.create(studioId, playlistId, onChanged)
 		try {
 			// ensure starts correct
 			await waitUntil(async () => {
@@ -127,12 +127,12 @@ describe('RundownsObserver', () => {
 		const playlistId = protectString<RundownPlaylistId>('playlist0')
 
 		const onChangedCleanup = jest.fn()
-		const onChanged = jest.fn<() => void, [RundownId[]]>(() => onChangedCleanup)
+		const onChanged = jest.fn<Promise<() => void>, [RundownId[]]>(async () => onChangedCleanup)
 
 		// should not be any observers yet
 		expect(RundownsMock.observers).toHaveLength(0)
 
-		const observer = new RundownsObserver(studioId, playlistId, onChanged)
+		const observer = await RundownsObserver.create(studioId, playlistId, onChanged)
 		try {
 			// ensure starts correct
 			// ensure starts correct
@@ -181,12 +181,12 @@ describe('RundownsObserver', () => {
 		const playlistId = protectString<RundownPlaylistId>('playlist0')
 
 		const onChangedCleanup = jest.fn()
-		const onChanged = jest.fn<() => void, [RundownId[]]>(() => onChangedCleanup)
+		const onChanged = jest.fn<Promise<() => void>, [RundownId[]]>(async () => onChangedCleanup)
 
 		// should not be any observers yet
 		expect(RundownsMock.observers).toHaveLength(0)
 
-		const observer = new RundownsObserver(studioId, playlistId, onChanged)
+		const observer = await RundownsObserver.create(studioId, playlistId, onChanged)
 		try {
 			// ensure starts correct
 			// ensure starts correct
diff --git a/meteor/server/publications/lib/debounce.ts b/meteor/server/publications/lib/debounce.ts
@@ -0,0 +1,132 @@
+import { Meteor } from 'meteor/meteor'
+
+/**
+ * Based on https://github.com/sindresorhus/p-debounce
+ * With additional features:
+ * - `cancel` method
+ */
+export class PromiseDebounce<TResult> {
+	readonly #fn: () => Promise<TResult>
+	readonly #wait: number
+
+	/** If an execution timeout has passed while  */
+	#isPending = false
+	#timeout: number | undefined
+
+	#isExecuting = false
+	#waitingListeners: Listener<TResult>[] = []
+
+	constructor(fn: () => Promise<TResult>, wait: number) {
+		this.#fn = fn
+		this.#wait = wait
+	}
+
+	/**
+	 * Trigger an execution, and get the result.
+	 * @returns A promise that resolves with the result of the function
+	 */
+	async call(): Promise<TResult> {
+		return new Promise<TResult>((resolve, reject) => {
+			const listener: Listener<TResult> = { resolve, reject }
+			this.#waitingListeners.push(listener)
+
+			// Trigger an execution
+			this.trigger()
+		})
+	}
+
+	/**
+	 * Trigger an execution, but don't report the result.
+	 */
+	trigger(): void {
+		// If an execution is 'imminent', don't do anything
+		if (this.#isPending) return
+
+		// Clear an existing timeout
+		if (this.#timeout) Meteor.clearTimeout(this.#timeout)
+
+		// Start a new one
+		this.#timeout = Meteor.setTimeout(() => {
+			this.#timeout = undefined
+
+			this.#executeFn()
+		}, this.#wait)
+	}
+
+	#executeFn(): void {
+		// If an execution is still in progress, mark as pending and stop
+		if (this.#isExecuting) {
+			this.#isPending = true
+			return
+		}
+
+		// We have the clear to begin executing
+		this.#isExecuting = true
+		this.#isPending = false
+
+		// Collect up the listeners for this execution
+		const listeners = this.#waitingListeners
+		this.#waitingListeners = []
+
+		this.#fn()
+			.then((result) => {
+				for (const listener of listeners) {
+					try {
+						listener.resolve(result)
+					} catch (e) {
+						// TODO - error?
+					}
+				}
+			})
+			.catch((error) => {
+				for (const listener of listeners) {
+					try {
+						listener.reject(error)
+					} catch (e) {
+						// TODO - error?
+					}
+				}
+			})
+			.finally(() => {
+				this.#isExecuting = false
+
+				// If there is a pending execution, run that soon
+				if (this.#isPending) {
+					Meteor.setTimeout(() => this.#executeFn(), 0)
+				}
+			})
+	}
+
+	/**
+	 * Cancel any waiting execution
+	 */
+	cancelWaiting(error?: Error): void {
+		this.#isPending = false
+
+		if (this.#timeout) {
+			Meteor.clearTimeout(this.#timeout)
+			this.#timeout = undefined
+		}
+
+		// Inform any listeners
+		if (this.#waitingListeners.length > 0) {
+			const listeners = this.#waitingListeners
+			this.#waitingListeners = []
+
+			error = error ?? new Error('Cancelled')
+
+			for (const listener of listeners) {
+				try {
+					listener.reject(error)
+				} catch (e) {
+					// TODO - error?
+				}
+			}
+		}
+	}
+}
+
+interface Listener<TResult> {
+	resolve: (value: TResult) => void
+	reject: (reason?: any) => void
+}
diff --git a/meteor/server/publications/lib/rundownsObserver.ts b/meteor/server/publications/lib/rundownsObserver.ts
@@ -1,7 +1,7 @@
 import { Meteor } from 'meteor/meteor'
 import { RundownId, RundownPlaylistId, StudioId } from '@sofie-automation/corelib/dist/dataModel/Ids'
-import _ from 'underscore'
 import { Rundowns } from '../../collections'
+import { PromiseDebounce } from './debounce'
 
 const REACTIVITY_DEBOUNCE = 20
 
@@ -12,30 +12,53 @@ type ChangedHandler = (rundownIds: RundownId[]) => Promise<() => void>
  * Note: Updates are debounced to avoid rapid updates firing
  */
 export class RundownsObserver implements Meteor.LiveQueryHandle {
-	#rundownsLiveQuery: Meteor.LiveQueryHandle
+	#rundownsLiveQuery!: Meteor.LiveQueryHandle
 	#rundownIds: Set<RundownId> = new Set<RundownId>()
 	#changed: ChangedHandler | undefined
 	#cleanup: (() => void) | undefined
 
-	constructor(studioId: StudioId, playlistId: RundownPlaylistId, onChanged: ChangedHandler) {
+	readonly #triggerUpdateRundownContent = new PromiseDebounce<void>(async () => {
+		if (!this.#changed) return
+		this.#cleanup?.()
+
+		const changed = this.#changed
+		this.#cleanup = await changed(this.rundownIds)
+	}, REACTIVITY_DEBOUNCE)
+
+	private constructor(onChanged: ChangedHandler) {
 		this.#changed = onChanged
-		this.#rundownsLiveQuery = Rundowns.observe(
+	}
+
+	static async create(
+		studioId: StudioId,
+		playlistId: RundownPlaylistId,
+		onChanged: ChangedHandler
+	): Promise<RundownsObserver> {
+		const observer = new RundownsObserver(onChanged)
+
+		await observer.init(studioId, playlistId)
+
+		return observer
+	}
+
+	private async init(studioId: StudioId, playlistId: RundownPlaylistId) {
+		this.#rundownsLiveQuery = await Rundowns.observe(
 			{
 				playlistId,
 				studioId,
 			},
 			{
 				added: (doc) => {
 					this.#rundownIds.add(doc._id)
-					this.triggerUpdateRundownContent()
+					this.#triggerUpdateRundownContent.trigger()
 				},
 				changed: (doc) => {
 					this.#rundownIds.add(doc._id)
-					this.triggerUpdateRundownContent()
+					this.#triggerUpdateRundownContent.trigger()
 				},
 				removed: (doc) => {
 					this.#rundownIds.delete(doc._id)
-					this.triggerUpdateRundownContent()
+					this.#triggerUpdateRundownContent.trigger()
 				},
 			},
 			{
@@ -44,28 +67,16 @@ export class RundownsObserver implements Meteor.LiveQueryHandle {
 				},
 			}
 		)
-		this.triggerUpdateRundownContent()
+
+		this.#triggerUpdateRundownContent.trigger()
 	}
 
 	public get rundownIds(): RundownId[] {
 		return Array.from(this.#rundownIds)
 	}
 
-	private innerUpdateRundownContent = () => {
-		if (!this.#changed) return
-		this.#cleanup?.()
-
-		const changed = this.#changed
-		this.#cleanup = changed(this.rundownIds)
-	}
-
-	private triggerUpdateRundownContent = _.debounce(
-		Meteor.bindEnvironment(this.innerUpdateRundownContent),
-		REACTIVITY_DEBOUNCE
-	)
-
 	public stop = (): void => {
-		this.triggerUpdateRundownContent.cancel()
+		this.#triggerUpdateRundownContent.cancelWaiting()
 		this.#rundownsLiveQuery.stop()
 		this.#changed = undefined
 		this.#cleanup?.()
diff --git a/meteor/server/publications/partInstancesUI/publication.ts b/meteor/server/publications/partInstancesUI/publication.ts
@@ -74,7 +74,7 @@ async function setupUIPartInstancesPublicationObservers(
 	)) as Pick<DBRundownPlaylist, RundownPlaylistFields> | undefined
 	if (!playlist) throw new Error(`RundownPlaylist with activationId="${args.playlistActivationId}" not found!`)
 
-	const rundownsObserver = new RundownsObserver(playlist.studioId, playlist._id, async (rundownIds) => {
+	const rundownsObserver = await RundownsObserver.create(playlist.studioId, playlist._id, async (rundownIds) => {
 		logger.silly(`Creating new RundownContentObserver`)
 
 		const cache = createReactiveContentCache()
diff --git a/meteor/server/publications/partsUI/publication.ts b/meteor/server/publications/partsUI/publication.ts
@@ -59,7 +59,7 @@ async function setupUIPartsPublicationObservers(
 	})) as Pick<DBRundownPlaylist, RundownPlaylistFields> | undefined
 	if (!playlist) throw new Error(`RundownPlaylist "${args.playlistId}" not found!`)
 
-	const rundownsObserver = new RundownsObserver(playlist.studioId, playlist._id, async (rundownIds) => {
+	const rundownsObserver = await RundownsObserver.create(playlist.studioId, playlist._id, async (rundownIds) => {
 		logger.silly(`Creating new RundownContentObserver`)
 
 		const cache = createReactiveContentCache()
diff --git a/meteor/server/publications/pieceContentStatusUI/rundown/publication.ts b/meteor/server/publications/pieceContentStatusUI/rundown/publication.ts
@@ -122,7 +122,7 @@ async function setupUIPieceContentStatusesPublicationObservers(
 	})) as Pick<DBRundownPlaylist, RundownPlaylistFields> | undefined
 	if (!playlist) throw new Error(`RundownPlaylist "${args.rundownPlaylistId}" not found!`)
 
-	const rundownsObserver = new RundownsObserver(playlist.studioId, playlist._id, async (rundownIds) => {
+	const rundownsObserver = await RundownsObserver.create(playlist.studioId, playlist._id, async (rundownIds) => {
 		logger.silly(`Creating new RundownContentObserver`)
 
 		// TODO - can this be done cheaper?
diff --git a/meteor/server/publications/segmentPartNotesUI/publication.ts b/meteor/server/publications/segmentPartNotesUI/publication.ts
@@ -66,7 +66,7 @@ async function setupUISegmentPartNotesPublicationObservers(
 	})) as Pick<DBRundownPlaylist, RundownPlaylistFields> | undefined
 	if (!playlist) throw new Error(`RundownPlaylist "${args.playlistId}" not found!`)
 
-	const rundownsObserver = new RundownsObserver(playlist.studioId, playlist._id, async (rundownIds) => {
+	const rundownsObserver = await RundownsObserver.create(playlist.studioId, playlist._id, async (rundownIds) => {
 		logger.silly(`Creating new RundownContentObserver`)
 
 		// TODO - can this be done cheaper?
