chore: split job-worker context.ts into multiple files

Author: Julusian

packages/job-worker/src/workers/caches.ts

@@ -16,7 +16,7 @@ import { clone, deepFreeze } from '@sofie-automation/corelib/dist/lib'
 import { logger } from '../logging'
 import deepmerge = require('deepmerge')
 import { ProcessedShowStyleBase, ProcessedShowStyleVariant, StudioCacheContext } from '../jobs'
-import { StudioCacheContextImpl } from './context'
+import { StudioCacheContextImpl } from './context/StudioCacheContextImpl'
 
 /**
  * A Wrapper to maintain a cache and provide a context using the cache when appropriate

packages/job-worker/src/workers/context/JobContextImpl.ts

@@ -0,0 +1,141 @@
+import { IDirectCollections } from '../../db'
+import { JobContext } from '../../jobs'
+import { WorkerDataCache } from '../caches'
+import { RundownId, RundownPlaylistId } from '@sofie-automation/corelib/dist/dataModel/Ids'
+import { getIngestQueueName, IngestJobFunc } from '@sofie-automation/corelib/dist/worker/ingest'
+import { ApmSpan, ApmTransaction } from '../../profiler'
+import { getRandomString } from '@sofie-automation/corelib/dist/lib'
+import { stringifyError } from '@sofie-automation/shared-lib/dist/lib/stringifyError'
+import { getStudioQueueName, StudioJobFunc } from '@sofie-automation/corelib/dist/worker/studio'
+import { LockBase, PlaylistLock, RundownLock } from '../../jobs/lock'
+import { logger } from '../../logging'
+import { BaseModel } from '../../modelBase'
+import { LocksManager } from '../locks'
+import { unprotectString } from '@sofie-automation/corelib/dist/protectedString'
+import { EventsJobFunc, getEventsQueueName } from '@sofie-automation/corelib/dist/worker/events'
+import { FastTrackTimelineFunc } from '../../main'
+import { TimelineComplete } from '@sofie-automation/corelib/dist/dataModel/Timeline'
+import type { QueueJobFunc } from './util'
+import { StudioCacheContextImpl } from './StudioCacheContextImpl'
+import { PlaylistLockImpl, RundownLockImpl } from './Locks'
+
+export class JobContextImpl extends StudioCacheContextImpl implements JobContext {
+	private readonly locks: Array<LockBase> = []
+	private readonly caches: Array<BaseModel> = []
+
+	constructor(
+		directCollections: Readonly<IDirectCollections>,
+		cacheData: WorkerDataCache,
+		private readonly locksManager: LocksManager,
+		private readonly transaction: ApmTransaction | undefined,
+		private readonly queueJob: QueueJobFunc,
+		private readonly fastTrackTimeline: FastTrackTimelineFunc | null
+	) {
+		super(directCollections, cacheData)
+	}
+
+	trackCache(cache: BaseModel): void {
+		this.caches.push(cache)
+	}
+
+	async lockPlaylist(playlistId: RundownPlaylistId): Promise<PlaylistLock> {
+		const span = this.startSpan('lockPlaylist')
+		if (span) span.setLabel('playlistId', unprotectString(playlistId))
+
+		const lockId = getRandomString()
+		logger.silly(`PlaylistLock: Locking "${playlistId}"`)
+
+		const resourceId = `playlist:${playlistId}`
+		await this.locksManager.aquire(lockId, resourceId)
+
+		const doRelease = async () => {
+			const span = this.startSpan('unlockPlaylist')
+			if (span) span.setLabel('playlistId', unprotectString(playlistId))
+
+			await this.locksManager.release(lockId, resourceId)
+
+			if (span) span.end()
+		}
+		const lock = new PlaylistLockImpl(playlistId, doRelease)
+		this.locks.push(lock)
+
+		logger.silly(`PlaylistLock: Locked "${playlistId}"`)
+
+		if (span) span.end()
+
+		return lock
+	}
+
+	async lockRundown(rundownId: RundownId): Promise<RundownLock> {
+		const span = this.startSpan('lockRundown')
+		if (span) span.setLabel('rundownId', unprotectString(rundownId))
+
+		const lockId = getRandomString()
+		logger.silly(`RundownLock: Locking "${rundownId}"`)
+
+		const resourceId = `rundown:${rundownId}`
+		await this.locksManager.aquire(lockId, resourceId)
+
+		const doRelease = async () => {
+			const span = this.startSpan('unlockRundown')
+			if (span) span.setLabel('rundownId', unprotectString(rundownId))
+
+			await this.locksManager.release(lockId, resourceId)
+
+			if (span) span.end()
+		}
+		const lock = new RundownLockImpl(rundownId, doRelease)
+		this.locks.push(lock)
+
+		logger.silly(`RundownLock: Locked "${rundownId}"`)
+
+		if (span) span.end()
+
+		return lock
+	}
+
+	/** Ensure resources are cleaned up after the job completes */
+	async cleanupResources(): Promise<void> {
+		// Ensure all locks are freed
+		for (const lock of this.locks) {
+			if (lock.isLocked) {
+				logger.warn(`Lock never freed: ${lock}`)
+				await lock.release().catch((e) => {
+					logger.error(`Lock free failed: ${stringifyError(e)}`)
+				})
+			}
+		}
+
+		// Ensure all caches were saved/aborted
+		for (const cache of this.caches) {
+			try {
+				cache.assertNoChanges()
+			} catch (e) {
+				logger.warn(`${cache.displayName} has unsaved changes: ${stringifyError(e)}`)
+			}
+		}
+	}
+
+	startSpan(spanName: string): ApmSpan | null {
+		if (this.transaction) return this.transaction.startSpan(spanName)
+		return null
+	}
+
+	async queueIngestJob<T extends keyof IngestJobFunc>(name: T, data: Parameters<IngestJobFunc[T]>[0]): Promise<void> {
+		await this.queueJob(getIngestQueueName(this.studioId), name, data)
+	}
+	async queueStudioJob<T extends keyof StudioJobFunc>(name: T, data: Parameters<StudioJobFunc[T]>[0]): Promise<void> {
+		await this.queueJob(getStudioQueueName(this.studioId), name, data)
+	}
+	async queueEventJob<T extends keyof EventsJobFunc>(name: T, data: Parameters<EventsJobFunc[T]>[0]): Promise<void> {
+		await this.queueJob(getEventsQueueName(this.studioId), name, data)
+	}
+
+	hackPublishTimelineToFastTrack(newTimeline: TimelineComplete): void {
+		if (this.fastTrackTimeline) {
+			this.fastTrackTimeline(newTimeline).catch((e) => {
+				logger.error(`Failed to publish timeline to fast track: ${stringifyError(e)}`)
+			})
+		}
+	}
+}

packages/job-worker/src/workers/context/Locks.ts

@@ -0,0 +1,67 @@
+import type { RundownId, RundownPlaylistId } from '@sofie-automation/corelib/dist/dataModel/Ids'
+import { PlaylistLock, RundownLock } from '../../jobs/lock'
+import { logger } from '../../logging'
+
+export class PlaylistLockImpl extends PlaylistLock {
+	#isLocked = true
+
+	public constructor(playlistId: RundownPlaylistId, private readonly doRelease: () => Promise<void>) {
+		super(playlistId)
+	}
+
+	get isLocked(): boolean {
+		return this.#isLocked
+	}
+
+	async release(): Promise<void> {
+		if (!this.#isLocked) {
+			logger.warn(`PlaylistLock: Already released "${this.playlistId}"`)
+		} else {
+			logger.silly(`PlaylistLock: Releasing "${this.playlistId}"`)
+
+			this.#isLocked = false
+
+			await this.doRelease()
+
+			logger.silly(`PlaylistLock: Released "${this.playlistId}"`)
+
+			if (this.deferedFunctions.length > 0) {
+				for (const fcn of this.deferedFunctions) {
+					await fcn()
+				}
+			}
+		}
+	}
+}
+
+export class RundownLockImpl extends RundownLock {
+	#isLocked = true
+
+	public constructor(rundownId: RundownId, private readonly doRelease: () => Promise<void>) {
+		super(rundownId)
+	}
+
+	get isLocked(): boolean {
+		return this.#isLocked
+	}
+
+	async release(): Promise<void> {
+		if (!this.#isLocked) {
+			logger.warn(`RundownLock: Already released "${this.rundownId}"`)
+		} else {
+			logger.silly(`RundownLock: Releasing "${this.rundownId}"`)
+
+			this.#isLocked = false
+
+			await this.doRelease()
+
+			logger.silly(`RundownLock: Released "${this.rundownId}"`)
+
+			if (this.deferedFunctions.length > 0) {
+				for (const fcn of this.deferedFunctions) {
+					await fcn()
+				}
+			}
+		}
+	}
+}