feat: add top-level agenda.disable and agenda.enable methods

src/JobDbRepository.ts

@@ -88,12 +88,20 @@
 	 * Internal method to unlock jobs so that they can be re-run
 	 */
 	async unlockJobs(jobIds: ObjectId[]): Promise<void> {
-		await this.collection.updateMany(
+		await this.updateMany(
 			{ _id: { $in: jobIds }, nextRunAt: { $ne: null } },
 			{ $unset: { lockedAt: true } }
 		);
 	}
 
+	/**
+	 * Internal method to toggle the `disabled` flag on jobs in the DB
+	 */
+	async setJobsDisabled(query: Filter<IJobParameters>, disabled: boolean): Promise<number> {
+		const result = await this.updateMany(query, { $set: { disabled }});
+		return result || 0;
+	}
+
 	async lockJob(job: JobWithId): Promise<IJobParameters | undefined> {
 		// Query to run against collection to see if we need to lock it
 		const criteria: Filter<Omit<IJobParameters, 'lockedAt'> & { lockedAt?: Date | null }> = {
@@ -136,7 +144,7 @@
 				disabled: { $ne: true },
 				$or: [
 					{
 						lockedAt: { $eq: null as any },
 						nextRunAt: { $lte: nextScanAt }
 					},
 					{
@@ -234,8 +242,8 @@
 		// We have a result from the above calls
 		if (res) {
 			// Grab ID and nextRunAt from MongoDB and store it as an attribute on Job
 			job.attrs._id = res._id;
 			job.attrs.nextRunAt = res.nextRunAt;
 
 			// check if we should process the job immediately
 			this.agenda.emit('processJob', job);
@@ -397,4 +405,25 @@
 			throw error;
 		}
 	}
+
+	/**
+	 * Internal method used to update multiple jobs with one call
+	 * @name JobDbRepository#updateMany
+	 * @function
+	 * @returns {Promise} the number of jobs updated in the DB
+	 */
+	private async updateMany(
+		query: Filter<IJobParameters>,
+		update: UpdateFilter<IJobParameters>
+	): Promise<number> {
+		const { modifiedCount, matchedCount } = await this.collection.updateMany(
+			query,
+			update
+		);
+		if(matchedCount !== modifiedCount) {
+		  log(`WARN: ${matchedCount} jobs matched query but ${modifiedCount} were modified`);
+		}
+
+		return modifiedCount;
+	}
 }

src/index.ts

@@ -1,7 +1,7 @@
 import { EventEmitter } from 'events';
 import * as debug from 'debug';
 
-import type { Db, Filter, MongoClientOptions, Sort } from 'mongodb';
+import type { Db, Filter, MongoClientOptions, Sort, UpdateResult } from 'mongodb';
 import { SortDirection } from 'mongodb';
 import { ForkOptions } from 'child_process';
 import type { IJobDefinition } from './types/JobDefinition';
@@ -200,6 +200,52 @@
 		}
 	}
 
+  /**
+ * Disables any jobs matching the passed MongoDB query by setting the `disabled` flag to `true`
+ * @name Agenda#disable
+ * @function
+ * @param query MongoDB query to use when enabling
+ * @returns {Promise<number>} Resolved with the number of disabled job instances.
+ */
+async disable (
+  this: Agenda,
+  query: Filter<IJobParameters>
+): Promise<number> {
+  log('attempting to disable all jobs matching query: %o', query);
+  try {
+    const modifiedCount = await this.db.setJobsDisabled(query, true);
+    log(`${modifiedCount} jobs disabled`);
+    return modifiedCount;
+  } catch (error) {
+    log('error trying to mark jobs as `disabled`');
+    throw error;
+  }
+};
+
+/**
+ * Enables any jobs matching the passed MongoDB query by setting the `disabled` flag to `false`
+ * @name Agenda#enable
+ * @function
+ * @param query MongoDB query to use when enabling
+ * @caller client code, Agenda.purge(), Job.remove()
+ * @returns {Promise<Number>} A promise that contains the number of removed documents when fulfilled.
+ */
+async enable (
+  this: Agenda,
+  query:  Filter<IJobParameters> = {}
+): Promise<number> {
+  log('attempting to enable all jobs matching query %o', query);
+  try {
+    const modifiedCount = await this.db.setJobsDisabled(query, false);
+    log(`${modifiedCount} jobs enabled`);
+    return modifiedCount;
+  } catch (error) {
+    log('error trying to mark jobs as `enabled`');
+    throw error;
+  }
+};
+
+
 	/**
 	 * Set name of queue
 	 * @param name

test/agenda.test.ts

@@ -566,6 +566,105 @@ describe('Agenda', () => {
 		});
 	});
 
+  describe('disable', () => {
+    beforeEach(async() => {
+      await Promise.all([
+        globalAgenda.create('sendEmail', {to: 'some guy'}).schedule('1 minute').save(),
+        globalAgenda.create('sendEmail', {from: 'some guy'}).schedule('1 minute').save(),
+        globalAgenda.create('some job').schedule('30 seconds').save()
+      ]);
+    });
+
+    it('disables all jobs', async() => {
+      const ct = await globalAgenda.disable({});
+
+      expect(ct).to.be(3);
+      const disabledJobs = await globalAgenda.jobs({});
+
+      expect(disabledJobs).to.have.length(3);
+      disabledJobs.map(x => expect(x.attrs.disabled).to.be(true));
+    });
+
+    it('disables jobs when queried by name', async() => {
+      const ct = await globalAgenda.disable({name: 'sendEmail'});
+
+      expect(ct).to.be(2);
+      const disabledJobs = await globalAgenda.jobs({name: 'sendEmail'});
+
+      expect(disabledJobs).to.have.length(2);
+      disabledJobs.map(x => expect(x.attrs.disabled).to.be(true));
+    });
+
+    it('disables jobs when queried by data', async() => {
+      const ct = await globalAgenda.disable({'data.from': 'some guy'});
+
+      expect(ct).to.be(1);
+      const disabledJobs = await globalAgenda.jobs({'data.from': 'some guy', disabled: true});
+
+      expect(disabledJobs).to.have.length(1);
+    });
+
+    it('does not modify `nextRunAt`', async() => {
+      const js = await globalAgenda.jobs({name: 'some job'});
+      const ct = await globalAgenda.disable({name: 'some job'});
+
+      expect(ct).to.be(1);
+      const disabledJobs = await globalAgenda.jobs({name: 'some job', disabled: true});
+
+      expect(disabledJobs[0].attrs.nextRunAt.toString()).to.be(js[0].attrs.nextRunAt.toString());
+    });
+  });
+
+  describe('enable', () => {
+    beforeEach(async() => {
+      await Promise.all([
+        globalAgenda.create('sendEmail', {to: 'some guy'}).schedule('1 minute').save(),
+        globalAgenda.create('sendEmail', {from: 'some guy'}).schedule('1 minute').save(),
+        globalAgenda.create('some job').schedule('30 seconds').save()
+      ]);
+    });
+
+    it('enables all jobs', async() => {
+      const ct = await globalAgenda.enable({});
+
+      expect(ct).to.be(3);
+      const enabledJobs = await globalAgenda.jobs({});
+
+      expect(enabledJobs).to.have.length(3);
+      enabledJobs.map(x => expect(x.attrs.disabled).to.be(false));
+    });
+
+    it('enables jobs when queried by name', async() => {
+      const ct = await globalAgenda.enable({name: 'sendEmail'});
+
+      expect(ct).to.be(2);
+      const enabledJobs = await globalAgenda.jobs({name: 'sendEmail'});
+
+      expect(enabledJobs).to.have.length(2);
+      enabledJobs.map(x => expect(x.attrs.disabled).to.be(false));
+    });
+
+    it('enables jobs when queried by data', async() => {
+      const ct = await globalAgenda.enable({'data.from': 'some guy'});
+
+      expect(ct).to.be(1);
+      const enabledJobs = await globalAgenda.jobs({'data.from': 'some guy', disabled: false});
+
+      expect(enabledJobs).to.have.length(1);
+    });
+
+    it('does not modify `nextRunAt`', async() => {
+      const js = await globalAgenda.jobs({name: 'some job'});
+      const ct = await globalAgenda.enable({name: 'some job'});
+
+      expect(ct).to.be(1);
+      const enabledJobs = await globalAgenda.jobs({name: 'some job', disabled: false});
+
+      expect(enabledJobs[0].attrs.nextRunAt.toString()).to.be(js[0].attrs.nextRunAt.toString());
+    });
+  });
+
+
 	describe('search', () => {
 		beforeEach(async () => {
 			await globalAgenda.create('jobA', 1).save();