wip: simplify mongo mock types

Author: Julusian

meteor/__mocks__/mongo.ts

@@ -1,7 +1,6 @@
 /* eslint-disable @typescript-eslint/explicit-module-boundary-types */
 import * as _ from 'underscore'
 import { literal, ProtectedString, unprotectString, protectString, getRandomString } from '../server/lib/tempLib'
-import { sleep } from '../server/lib/lib'
 import { RandomMock } from './random'
 import { MeteorMock } from './meteor'
 import { Random } from 'meteor/random'
@@ -18,7 +17,10 @@ import {
 } from '@sofie-automation/meteor-lib/dist/collections/lib'
 import { mongoWhere, mongoFindOptions, mongoModify, MongoQuery } from '@sofie-automation/corelib/dist/mongo'
 import { AsyncOnlyMongoCollection, AsyncOnlyReadOnlyMongoCollection } from '../server/collections/collection'
-import type { MinimalMeteorMongoCollection, MinimalMongoCursor } from '../server/collections/implementations/base'
+import type {
+	MinimalMeteorMongoCollection,
+	MinimalMongoCursor,
+} from '../server/collections/implementations/asyncCollection'
 const clone = require('fast-clone')
 
 export namespace MongoMock {
@@ -90,12 +92,21 @@ export namespace MongoMock {
 					return docs
 				},
 				fetchAsync: async () => {
+					// Force this to be performed async
+					await MeteorMock.sleepNoFakeTimers(0)
+
 					return clone(docs)
 				},
 				countAsync: async () => {
+					// Force this to be performed async
+					await MeteorMock.sleepNoFakeTimers(0)
+
 					return docs.length
 				},
 				async observeAsync(clbs: ObserveCallbacks<T>): Promise<Meteor.LiveQueryHandle> {
+					// Force this to be performed async
+					await MeteorMock.sleepNoFakeTimers(0)
+
 					const id = Random.id(5)
 					observers.push(
 						literal<ObserverEntry<T>>({
@@ -111,6 +122,9 @@ export namespace MongoMock {
 					}
 				},
 				async observeChangesAsync(clbs: ObserveChangesCallbacks<T>): Promise<Meteor.LiveQueryHandle> {
+					// Force this to be performed async
+					await MeteorMock.sleepNoFakeTimers(0)
+
 					// todo - finish implementing uses of callbacks
 					const id = Random.id(5)
 					observers.push(
@@ -139,6 +153,9 @@ export namespace MongoMock {
 			return docs[0]
 		}
 		async updateAsync(query: any, modifier: any, options?: UpdateOptions): Promise<number> {
+			// Force this to be performed async
+			await MeteorMock.sleepNoFakeTimers(0)
+
 			const unimplementedUsedOptions = _.without(_.keys(options), 'multi')
 			if (unimplementedUsedOptions.length > 0) {
 				throw new Error(`update being performed using unimplemented options: ${unimplementedUsedOptions}`)
@@ -173,6 +190,9 @@ export namespace MongoMock {
 			return docs.length
 		}
 		async insertAsync(doc: any): Promise<string> {
+			// Force this to be performed async
+			await MeteorMock.sleepNoFakeTimers(0)
+
 			const d = _.clone(doc)
 			if (!d._id) d._id = protectString(RandomMock.id())
 
@@ -206,6 +226,9 @@ export namespace MongoMock {
 			modifier: any,
 			options?: UpsertOptions
 		): Promise<{ numberAffected: number | undefined; insertedId: string | undefined }> {
+			// Force this to be performed async
+			await MeteorMock.sleepNoFakeTimers(0)
+
 			const id = _.isString(query) ? query : query._id
 
 			const docs = this.find(id)._fetchRaw()
@@ -220,6 +243,9 @@ export namespace MongoMock {
 			}
 		}
 		async removeAsync(query: any): Promise<number> {
+			// Force this to be performed async
+			await MeteorMock.sleepNoFakeTimers(0)
+
 			const docs = this.find(query)._fetchRaw()
 
 			_.each(docs, (doc) => {
@@ -254,7 +280,7 @@ export namespace MongoMock {
 		rawCollection(): any {
 			return {
 				bulkWrite: async (updates: AnyBulkWriteOperation<any>[], _options: unknown) => {
-					await sleep(this.asyncBulkWriteDelay)
+					await MeteorMock.sleepNoFakeTimers(this.asyncBulkWriteDelay)
 
 					for (const update of updates) {
 						if ('insertOne' in update) {

meteor/server/__tests__/_testEnvironment.test.ts

@@ -31,7 +31,7 @@ import {
 import { DBStudio } from '@sofie-automation/corelib/dist/dataModel/Studio'
 import { Mongo } from 'meteor/mongo'
 import { defaultStudio } from '../../__mocks__/defaultCollectionObjects'
-import { MinimalMeteorMongoCollection } from '../collections/implementations/base'
+import { MinimalMeteorMongoCollection } from '../collections/implementations/asyncCollection'
 
 describe('Basic test of test environment', () => {
 	test('Meteor Random mock', () => {

meteor/server/api/deviceTriggers/StudioObserver.ts

@@ -20,7 +20,7 @@ import { RundownContentObserver } from './RundownContentObserver'
 import { RundownsObserver } from './RundownsObserver'
 import { RundownPlaylists, Rundowns, ShowStyleBases } from '../../collections'
 import { PromiseDebounce } from '../../publications/lib/debounce'
-import { MinimalMongoCursor } from '../../collections/implementations/base'
+import { MinimalMongoCursor } from '../../collections/implementations/asyncCollection'
 
 type ChangedHandler = (showStyleBaseId: ShowStyleBaseId, cache: ContentCache) => () => void
 

meteor/server/collections/collection.ts

@@ -19,7 +19,7 @@ import {
 	UpdateOptions,
 	UpsertOptions,
 } from '@sofie-automation/meteor-lib/dist/collections/lib'
-import { MinimalMongoCursor } from './implementations/base'
+import { MinimalMongoCursor } from './implementations/asyncCollection'
 
 export interface MongoAllowRules<DBInterface> {
 	insert?: (userId: UserId, doc: DBInterface) => Promise<boolean> | boolean
@@ -116,8 +116,7 @@ function wrapMeteorCollectionIntoAsyncCollection<DBInterface extends { _id: Prot
 	name: CollectionName
 ) {
 	if ((collection as any)._isMock) {
-		// We use a special one in tests, to reduce the amount of hops between fibers and promises
-		// nocommit - is this still needed?
+		// We use a special one in tests, to add some async which naturally doesn't happen in the collection
 		return new WrappedMockCollection<DBInterface>(collection, name)
 	} else {
 		// Override the default mongodb methods, because the errors thrown by them doesn't contain the proper call stack

meteor/server/collections/implementations/asyncCollection.ts

@@ -1,25 +1,74 @@
-import { MongoQuery } from '@sofie-automation/corelib/dist/mongo'
-import { ProtectedString } from '@sofie-automation/corelib/dist/protectedString'
+import { MongoModifier, MongoQuery } from '@sofie-automation/corelib/dist/mongo'
+import { ProtectedString, protectString, unprotectString } from '@sofie-automation/corelib/dist/protectedString'
 import { Meteor } from 'meteor/meteor'
+import { Mongo } from 'meteor/mongo'
 import {
+	UpdateOptions,
+	UpsertOptions,
+	IndexSpecifier,
+	MongoCursor,
 	FindOptions,
 	ObserveChangesCallbacks,
 	ObserveCallbacks,
 } from '@sofie-automation/meteor-lib/dist/collections/lib'
+import type { AnyBulkWriteOperation, Collection as RawCollection, Db as RawDb } from 'mongodb'
+import { stringifyError } from '@sofie-automation/shared-lib/dist/lib/stringifyError'
+import { NpmModuleMongodb } from 'meteor/npm-mongo'
+import { profiler } from '../../api/profiler'
 import { PromisifyCallbacks } from '@sofie-automation/shared-lib/dist/lib/types'
-import type { AnyBulkWriteOperation } from 'mongodb'
-import { MinimalMongoCursor, WrappedMongoCollectionBase } from './base'
 import { AsyncOnlyMongoCollection } from '../collection'
-import { profiler } from '../../api/profiler'
+
+export type MinimalMongoCursor<T extends { _id: ProtectedString<any> }> = Pick<
+	MongoCursor<T>,
+	'fetchAsync' | 'observeChangesAsync' | 'observeAsync' | 'countAsync'
+	// | 'forEach' | 'map' |
+>
+
+export type MinimalMeteorMongoCollection<T extends { _id: ProtectedString<any> }> = Pick<
+	Mongo.Collection<T>,
+	// | 'find'
+	'insertAsync' | 'removeAsync' | 'updateAsync' | 'upsertAsync' | 'rawCollection' | 'rawDatabase' | 'createIndex'
+> & {
+	find: (...args: Parameters<Mongo.Collection<T>['find']>) => MinimalMongoCursor<T>
+}
 
 export class WrappedAsyncMongoCollection<DBInterface extends { _id: ProtectedString<any> }>
-	extends WrappedMongoCollectionBase<DBInterface>
 	implements AsyncOnlyMongoCollection<DBInterface>
 {
+	protected readonly _collection: MinimalMeteorMongoCollection<DBInterface>
+
+	public readonly name: string | null
+
+	constructor(collection: Mongo.Collection<DBInterface>, name: string | null) {
+		this._collection = collection as any
+		this.name = name
+	}
+
+	protected get _isMock(): boolean {
+		// @ts-expect-error re-export private property
+		return this._collection._isMock
+	}
+
+	public get mockCollection(): MinimalMeteorMongoCollection<DBInterface> {
+		return this._collection
+	}
+
 	get mutableCollection(): AsyncOnlyMongoCollection<DBInterface> {
 		return this
 	}
 
+	protected wrapMongoError(e: unknown): never {
+		const str = stringifyError(e) || 'Unknown MongoDB Error'
+		throw new Meteor.Error(e instanceof Meteor.Error ? e.error : 500, `Collection "${this.name}": ${str}`)
+	}
+
+	rawCollection(): RawCollection<DBInterface> {
+		return this._collection.rawCollection() as any
+	}
+	protected rawDatabase(): RawDb {
+		return this._collection.rawDatabase() as any
+	}
+
 	async findFetchAsync(
 		selector: MongoQuery<DBInterface> | DBInterface['_id'],
 		options?: FindOptions<DBInterface>
@@ -131,10 +180,115 @@ export class WrappedAsyncMongoCollection<DBInterface extends { _id: ProtectedStr
 		}
 	}
 
+	public async countDocuments(
+		selector?: MongoQuery<DBInterface>,
+		options?: FindOptions<DBInterface>
+	): Promise<number> {
+		const span = profiler.startSpan(`MongoCollection.${this.name}.countDocuments`)
+		if (span) {
+			span.addLabels({
+				collection: this.name,
+				query: JSON.stringify(selector),
+			})
+		}
+		try {
+			const res = await this._collection.find((selector ?? {}) as any, options as any).countAsync()
+			if (span) span.end()
+			return res
+		} catch (e) {
+			if (span) span.end()
+			this.wrapMongoError(e)
+		}
+	}
+
+	public async insertAsync(doc: DBInterface): Promise<DBInterface['_id']> {
+		const span = profiler.startSpan(`MongoCollection.${this.name}.insert`)
+		if (span) {
+			span.addLabels({
+				collection: this.name,
+				id: unprotectString(doc._id),
+			})
+		}
+		try {
+			const resultId = await this._collection.insertAsync(doc as unknown as Mongo.OptionalId<DBInterface>)
+			if (span) span.end()
+			return protectString(resultId)
+		} catch (e) {
+			if (span) span.end()
+			this.wrapMongoError(e)
+		}
+	}
+
 	async insertManyAsync(docs: DBInterface[]): Promise<Array<DBInterface['_id']>> {
 		return Promise.all(docs.map(async (doc) => this.insertAsync(doc)))
 	}
 
+	public async removeAsync(selector: MongoQuery<DBInterface> | DBInterface['_id']): Promise<number> {
+		const span = profiler.startSpan(`MongoCollection.${this.name}.remove`)
+		if (span) {
+			span.addLabels({
+				collection: this.name,
+				query: JSON.stringify(selector),
+			})
+		}
+		try {
+			const res = await this._collection.removeAsync(selector as any)
+			if (span) span.end()
+			return res
+		} catch (e) {
+			if (span) span.end()
+			this.wrapMongoError(e)
+		}
+	}
+	public async updateAsync(
+		selector: MongoQuery<DBInterface> | DBInterface['_id'] | { _id: DBInterface['_id'] },
+		modifier: MongoModifier<DBInterface>,
+		options?: UpdateOptions
+	): Promise<number> {
+		const span = profiler.startSpan(`MongoCollection.${this.name}.update`)
+		if (span) {
+			span.addLabels({
+				collection: this.name,
+				query: JSON.stringify(selector),
+			})
+		}
+		try {
+			const res = await this._collection.updateAsync(selector as any, modifier as any, options)
+			if (span) span.end()
+			return res
+		} catch (e) {
+			if (span) span.end()
+			this.wrapMongoError(e)
+		}
+	}
+	public async upsertAsync(
+		selector: MongoQuery<DBInterface> | DBInterface['_id'] | { _id: DBInterface['_id'] },
+		modifier: MongoModifier<DBInterface>,
+		options?: UpsertOptions
+	): Promise<{
+		numberAffected?: number
+		insertedId?: DBInterface['_id']
+	}> {
+		const span = profiler.startSpan(`MongoCollection.${this.name}.upsert`)
+		if (span) {
+			span.addLabels({
+				collection: this.name,
+				query: JSON.stringify(selector),
+			})
+		}
+		try {
+			const result = await this._collection.upsertAsync(selector as any, modifier as any, options)
+			if (span) span.end()
+			return {
+				numberAffected: result.numberAffected,
+				insertedId: protectString(result.insertedId),
+			}
+		} catch (e) {
+			if (span) span.end()
+			this.wrapMongoError(e)
+		}
+	}
+
 	async upsertManyAsync(docs: DBInterface[]): Promise<{ numberAffected: number; insertedIds: DBInterface['_id'][] }> {
 		const result: {
 			numberAffected: number
@@ -178,16 +332,16 @@ export class WrappedAsyncMongoCollection<DBInterface extends { _id: ProtectedStr
 		if (span) span.end()
 	}
 
-	async countDocuments(selector?: MongoQuery<DBInterface>, options?: FindOptions<DBInterface>): Promise<number> {
-		const span = profiler.startSpan(`MongoCollection.${this.name}.countDocuments`)
+	createIndex(keys: IndexSpecifier<DBInterface> | string, options?: NpmModuleMongodb.CreateIndexesOptions): void {
+		const span = profiler.startSpan(`MongoCollection.${this.name}.createIndex`)
 		if (span) {
 			span.addLabels({
 				collection: this.name,
-				query: JSON.stringify(selector),
+				keys: JSON.stringify(keys),
 			})
 		}
 		try {
-			const res = await this._collection.find((selector ?? {}) as any, options as any).countAsync()
+			const res = this._collection.createIndex(keys as any, options)
 			if (span) span.end()
 			return res
 		} catch (e) {