Mongoose Static Method Mod.md

NodeJS static class override example

Mongoose model replacment for .watch()

Just a gist from a larger project. Worker.runonce is a function that will ask the primary process for permission to proceed. The primary process will track what activites are run where (through the unique key) and pass down the approved pid. If the worker's pid doesn't match what prime messages out, it will not run the callback.

The purpose of overriding watch is that having so many watch actions on the prime process was causing OOM Errors in Heroku. Model.watch().on() can be called as normal. If the application has multiple entrypoints, some that do not use clustering, the function simply returns the super.watch payload.

This makes watch streams process aware and will distribute the load.

import mongoose from 'mongoose';

class SomeModel extends mongoose.Model {
	/**
	 * Watch override.
	 * Runs through Worker.runonce before handing off to Model.watch().on().
	 */
	static watch(...args) {
		// Fire off the original watch to get the payload chain.
		const payload = super.watch(...args);

		// If this is the prime process, just return the payload.
		if (Worker.isPrimary) {
			return payload;
		}

		// Unsure if on needs to be bound, but can't hurt.
		const on = payload.on.bind(payload);

		// Generate a unique key from the collection name and md5 of args.
		const key = this.collection.collectionName + md5(args);

		// Return our new object and on function where we inject our Worker.
		return {
			...payload,
			on(event, callback, ...props) {
				// Call the `on` from payload, replacing the callback with our callback.
				on(
					event,
					(data, ...cbargs) => {
						// In the callback, prepend the _id for an extra dose of uniquness.
						let _id = '';

						// If the document doesn't have an _id, this will silently fail.
						try {
							_id = data.documentKey._id;
						} catch (_e) {
							// noop.
						}

						// Up until this point we're running on all processes.
						// Worker.runonce only runs once among all the processes.
						// The callback is supressed on processes that the Worker deems busy or under heavy load.
						Worker.runonce(_id.toString() + key, () => {
							callback(data, ...cbargs);
						});
					},
					...props
				);
			},
		};
	}
}