Replace Cursor#observeChanges calls with MongoDB change streams

codecov.yml

@@ -2,7 +2,7 @@ coverage:
   status:
     project:
       default:
-        target: 67%
+        target: 68%
         threshold: 1%
     patch:
       default:

imports/_test/fixtures.ts

@@ -3,10 +3,14 @@ import 'regenerator-runtime/runtime.js';
 
 import {assert, expect} from 'chai';
 
+import {Meteor} from 'meteor/meteor';
+
 import {cleanup as unmount} from '@testing-library/react';
 import totalOrder from 'total-order';
 import {sorted} from '@iterable-iterator/sorted';
 
+// eslint-disable-next-line import/no-unassigned-import
+import '../api/endpoint/_dev/_disableRateLimiting';
 import logout from '../api/user/logout';
 import invoke from '../api/endpoint/invoke';
 import call from '../api/endpoint/call';
@@ -174,19 +178,36 @@ export const dropId = ({_id, ...rest}) => {
 
 export const dropIds = (x) => x.map(dropId);
 
-export const create = (template, extra) => {
-	if (typeof template === 'function') return extra ?? template();
+export const dropOwner = ({owner, ...rest}) => {
+	assert(typeof owner === 'string');
+	return rest;
+};
+
+export const dropOwners = (x) => x.map(dropOwner);
+
+export const create = (template, extra, hasExtra: boolean) => {
+	if (typeof template === 'function') return hasExtra ? extra : template();
 	if (Array.isArray(template)) {
 		return template
-			.map((x, i) => create(x, extra?.[i]))
+			.map((x, i) =>
+				create(
+					x,
+					extra?.[i],
+					Object.prototype.hasOwnProperty.call(extra ?? [], i),
+				),
+			)
 			.concat(extra?.slice(template.length) ?? []);
 	}
 
 	return Object.fromEntries(
 		(extra === undefined ? [] : Object.entries(extra)).concat(
 			Object.entries(template).map(([key, value]) => [
 				key,
-				create(value, extra?.[key]),
+				create(
+					value,
+					extra?.[key],
+					Object.prototype.hasOwnProperty.call(extra ?? {}, key),
+				),
 			]),
 		),
 	);
@@ -204,4 +225,5 @@ export const findOneOrThrow = async <T extends Document, U = T>(
 	return result!;
 };
 
-export const makeTemplate = (template) => (extra?) => create(template, extra);
+export const makeTemplate = (template) => (extra?) =>
+	create(template, extra, extra !== undefined);

imports/api/ObserveSequenceChangesCallbacks.ts

@@ -0,0 +1,8 @@
+import {type Mongo} from 'meteor/mongo';
+
+type ObserveSequenceChangesCallbacks<T> = Pick<
+	Mongo.ObserveChangesCallbacks<T>,
+	'addedBefore' | 'movedBefore' | 'changed' | 'removed'
+>;
+
+export default ObserveSequenceChangesCallbacks;

imports/api/ObserveSetChangesCallbacks.ts

@@ -0,0 +1,8 @@
+import {type Mongo} from 'meteor/mongo';
+
+type ObserveSetChangesCallbacks<T> = Pick<
+	Mongo.ObserveChangesCallbacks<T>,
+	'added' | 'changed' | 'removed'
+>;
+
+export default ObserveSetChangesCallbacks;

imports/api/appointments.ts

@@ -61,8 +61,8 @@ export const appointmentUpdate = schema.union([
 	schema.object({
 		patient: schema.object({
 			_id: schema.string(),
-			firstname: schema.string(),
-			lastname: schema.string(),
+			firstname: schema.string().optional(),
+			lastname: schema.string().optional(),
 		}),
 		phone: schema.string().optional(),
 		datetime: schema.date().optional(),

imports/api/collection/define.ts

@@ -1,11 +1,21 @@
+import assert from 'assert';
+
 import type Document from '../Document';
 import Collection from '../Collection';
 
+import {hasCollection, addCollection} from './registry';
+
 const define = <T extends Document, U = T>(name: string) => {
-	return new Collection<T, U>(name, {
+	assert(!hasCollection(name));
+
+	const collection = new Collection<T, U>(name, {
 		idGeneration: 'STRING',
 		defineMutationMethods: false,
 	});
+
+	addCollection(name, collection);
+
+	return collection;
 };
 
 export default define;

imports/api/collection/registry.ts

@@ -0,0 +1,26 @@
+import assert from 'assert';
+
+import type Document from '../Document';
+import type Collection from '../Collection';
+
+const _registry = new Map<string, Collection<any, any>>();
+
+export const getCollection = <T extends Document, U = T>(
+	name: string,
+): Collection<T, U> => {
+	const collection = _registry.get(name);
+	assert(collection !== undefined);
+	return collection;
+};
+
+export const hasCollection = (name: string) => {
+	return _registry.has(name);
+};
+
+export const addCollection = <T extends Document, U = T>(
+	name: string,
+	collection: Collection<T, U>,
+) => {
+	assert(!hasCollection(name));
+	_registry.set(name, collection);
+};

imports/api/consultations.ts

@@ -32,6 +32,7 @@
 import type Selector from './query/Selector';
 import {type AuthenticatedContext} from './publication/Context';
 import {type DocumentUpdate} from './DocumentUpdate';
+import observeSetChanges from './query/observeSetChanges';
 
 export const DEFAULT_DURATION_IN_MINUTES = 15;
 export const DEFAULT_DURATION_IN_SECONDS = DEFAULT_DURATION_IN_MINUTES * 60;
@@ -127,18 +128,18 @@
 	},
 });
 
-export function setupConsultationsStatsPublication(
+export async function setupConsultationsStatsPublication(
 	this: AuthenticatedContext,
 	collectionName: string,
 	filter: Filter<ConsultationDocument>,
 ) {
 	// Generate unique key depending on parameters
 	const key = statsKey(filter);
-	const selector = {
+	const scopedFilter = {
 		...filter,
 		isDone: true,
 		owner: this.userId,
-	} as Selector<ConsultationDocument>;
+	} as Filter<ConsultationDocument>;
 	const options = {fields: {_id: 1, price: 1, datetime: 1}};
 
 	const minHeap = new PairingHeap(increasing);
@@ -158,48 +159,53 @@
 	// Until then, we don't want to send a lot of `changed` messages—hence
 	// tracking the `initializing` state.
 	let initializing = true;
-	const handle = Consultations.find(selector, options).observeChanges({
-		added: (_id, {price, datetime}) => {
-			count += 1;
-			if (price) total += price;
-			const minRef = minHeap.push(datetime);
-			const maxRef = maxHeap.push(datetime);
-			refs.set(_id, [price, minRef, maxRef]);
-
-			if (!initializing) {
-				this.changed(collectionName, key, state());
-			}
-		},
-
-		changed: (_id, fields) => {
-			const [oldPrice, minRef, maxRef] = refs.get(_id);
-			let newPrice: number = oldPrice;
-			if (Object.prototype.hasOwnProperty.call(fields, 'price')) {
-				newPrice = fields.price!;
-				if (oldPrice) total -= oldPrice;
-				if (newPrice) total += newPrice;
-				refs.set(_id, [newPrice, minRef, maxRef]);
-			}
-
-			if (Object.prototype.hasOwnProperty.call(fields, 'datetime')) {
-				const datetime = fields.datetime;
-				minHeap.update(minRef, datetime);
-				maxHeap.update(maxRef, datetime);
-			}
-
-			this.changed(collectionName, key, state());
-		},
+	const handle = await observeSetChanges(
+		Consultations,
+		scopedFilter,
+		options,
+		{
+			added: (_id, {price, datetime}) => {
+				count += 1;
+				if (price) total += price;
+				const minRef = minHeap.push(datetime);
+				const maxRef = maxHeap.push(datetime);
+				refs.set(_id, [price, minRef, maxRef]);
+
+				if (!initializing) {
+					this.changed(collectionName, key, state());
+				}
+			},
+
+			changed: (_id, fields) => {
+				const [oldPrice, minRef, maxRef] = refs.get(_id);
+				if (Object.prototype.hasOwnProperty.call(fields, 'price')) {
+					const newPrice = fields.price;
+					if (oldPrice) total -= oldPrice;
+					if (newPrice) total += newPrice;
+					refs.set(_id, [newPrice, minRef, maxRef]);
+				}
+
+				if (Object.prototype.hasOwnProperty.call(fields, 'datetime')) {
+					const datetime = fields.datetime;
+					minHeap.update(minRef, datetime);
+					maxHeap.update(maxRef, datetime);
+				}
 
-		removed: (_id) => {
-			count -= 1;
-			const [price, minRef, maxRef] = refs.get(_id);
-			if (price) total -= price;
-			minHeap.delete(minRef);
-			maxHeap.delete(maxRef);
-			refs.delete(_id);
-			this.changed(collectionName, key, state());
+				this.changed(collectionName, key, state());
+			},
+
+			removed: (_id) => {
+				count -= 1;
+				const [price, minRef, maxRef] = refs.get(_id);
+				if (price) total -= price;
+				minHeap.delete(minRef);
+				maxHeap.delete(maxRef);
+				refs.delete(_id);
+				this.changed(collectionName, key, state());
+			},
 		},
-	});
+		{projectionFn: ({price, datetime}) => ({price, datetime})},
+	);
 
 	// Instead, we'll send one `added` message right after `observeChanges` has
 	// returned, and mark the subscription as ready.

imports/api/createTagCollection.ts

@@ -43,6 +43,7 @@
 import type UserFilter from './query/UserFilter';
 import type Options from './query/Options';
 import type Projection from './query/Projection';
+import observeSetChanges from './query/observeSetChanges';
 
 export const STATS_SUFFIX = '.stats';
 export const FIND_CACHE_SUFFIX = '.find.cache';
@@ -190,12 +191,12 @@
 		name: statsPublication,
 		authentication: AuthenticationLoggedIn,
 		schema: schema.tuple([schema.string()]),
-		handle(name) {
+		async handle(name) {
 			const uid = JSON.stringify({name, owner: this.userId});
 			const query = {
 				[key]: {$elemMatch: {name}},
 				owner: this.userId,
-			} as Selector<P>;
+			} as Filter<P>;
 			// We only include relevant fields
 			const options = {fields: {_id: 1, [key]: 1}};
 
@@ -205,22 +206,28 @@
 			// `observeChanges` only returns after the initial `added` callbacks have run.
 			// Until then, we don't want to send a lot of `changed` messages—hence
 			// tracking the `initializing` state.
-			const handle = Parent.find(query, options).observeChanges({
-				added: () => {
-					count += 1;
+			const handle = await observeSetChanges(
+				Parent,
+				query,
+				options,
+				{
+					added: () => {
+						count += 1;
 
-					if (!initializing) {
+						if (!initializing) {
+							this.changed(stats, uid, {count});
+						}
+					},
+
+					removed: () => {
+						count -= 1;
 						this.changed(stats, uid, {count});
-					}
-				},
+					},
 
-				removed: () => {
-					count -= 1;
-					this.changed(stats, uid, {count});
+					// We don't care about `changed` events.
 				},
-
-				// We don't care about `changed` events.
-			});
+				{projectionFn: (_fields) => ({})},
+			);
 
 			// Instead, we'll send one `added` message right after `observeChanges` has
 			// returned, and mark the subscription as ready.
@@ -231,8 +238,8 @@
 			// Stop observing the cursor when the client unsubscribes. Stopping a
 			// subscription automatically takes care of sending the client any `removed`
 			// messages.
-			this.onStop(() => {
-				handle.stop();
+			this.onStop(async () => {
+				await handle.emit('stop');
 			});
 		},
 	});

imports/api/endpoint/_dev/_disableRateLimiting.ts

@@ -0,0 +1,9 @@
+import {Meteor} from 'meteor/meteor';
+import {Accounts} from 'meteor/accounts-base';
+
+import isTest from '../../../app/isTest';
+
+if (Meteor.isServer && isTest()) {
+	// @ts-expect-error Missing from type definitions.
+	Accounts.removeDefaultRateLimit();
+}