[NEW][APPS] New Scheduler API

app/apps/server/bridges/bridges.js

@@ -16,6 +16,7 @@ import { AppUserBridge } from './users';
 import { AppLivechatBridge } from './livechat';
 import { AppUploadBridge } from './uploads';
 import { UiInteractionBridge } from './uiInteraction';
+import { AppSchedulerBridge } from './scheduler';
 
 export class RealAppBridges extends AppBridges {
 	constructor(orch) {
@@ -37,6 +38,7 @@ export class RealAppBridges extends AppBridges {
 		this._livechatBridge = new AppLivechatBridge(orch);
 		this._uploadBridge = new AppUploadBridge(orch);
 		this._uiInteractionBridge = new UiInteractionBridge(orch);
+		this._schedulerBridge = new AppSchedulerBridge(orch);
 	}
 
 	getCommandBridge() {
@@ -102,4 +104,8 @@ export class RealAppBridges extends AppBridges {
 	getUiInteractionBridge() {
 		return this._uiInteractionBridge;
 	}
+
+	getSchedulerBridge() {
+		return this._schedulerBridge;
+	}
 }

app/apps/server/bridges/index.js

@@ -10,6 +10,7 @@ import { AppRoomBridge } from './rooms';
 import { AppInternalBridge } from './internal';
 import { AppSettingBridge } from './settings';
 import { AppUserBridge } from './users';
+import { AppSchedulerBridge } from './scheduler';
 
 export {
 	RealAppBridges,
@@ -24,4 +25,5 @@ export {
 	AppSettingBridge,
 	AppUserBridge,
 	AppInternalBridge,
+	AppSchedulerBridge,
 };

app/apps/server/bridges/scheduler.js

@@ -0,0 +1,62 @@
+import Agenda from 'agenda';
+import { MongoInternals } from 'meteor/mongo';
+
+function _callProcessor(processor) {
+	return (job) => processor(job.attrs.data);
+}
+
+export class AppSchedulerBridge {
+	constructor(orch) {
+		this.orch = orch;
+		this.scheduler = new Agenda({
+			mongo: MongoInternals.defaultRemoteCollectionDriver().mongo.client.db(),
+			collection: 'rocketchat_agenda_jobs',
+		});
+		this.isConnected = false;
+	}
+
+	async registerProcessors(processors = [], appId) {
+		this.orch.debugLog(`The App ${ appId } is registering job processors`, processors);
+		processors.forEach(({ id, processor }) => this.scheduler.define(id, _callProcessor(processor)));
+	}
+
+	async scheduleOnce(job, appId) {
+		this.orch.debugLog(`The App ${ appId } is scheduling an onetime job`, job);
+		await this.startScheduler();
+		await this.scheduler.schedule(job.when, job.id, job.data || {});
+	}
+
+	async scheduleRecurring(job, appId) {
+		this.orch.debugLog(`The App ${ appId } is scheduling a recurring job`, job);
+		await this.startScheduler();
+		await this.scheduler.every(job.cron, job.id, job.data || {});
+	}
+
+	async cancelJob(jobId, appId) {
+		this.orch.debugLog(`The App ${ appId } is canceling a job`, jobId);
+		await this.startScheduler();
+		try {
+			await this.scheduler.cancel({ name: jobId });
+		} catch (e) {
+			console.error(e);
+		}
+	}
+
+	async cancelAllJobs(appId) {
+		this.orch.debugLog(`Canceling all jobs of App ${ appId }`);
+		await this.startScheduler();
+		const matcher = new RegExp(`^_${ appId }`);
+		try {
+			await this.scheduler.cancel({ name: { $regex: matcher } });
+		} catch (e) {
+			console.error(e);
+		}
+	}
+
+	async startScheduler() {
+		if (!this.isConnected) {
+			await this.scheduler.start();
+			this.isConnected = true;
+		}
+	}
+}

package.json

@@ -62,6 +62,7 @@
 		"@storybook/addon-viewport": "^5.3.19",
 		"@storybook/addons": "^5.3.19",
 		"@storybook/react": "^5.3.19",
+		"@types/agenda": "^2.0.9",
 		"@types/bcrypt": "^3.0.0",
 		"@types/chai": "^4.2.12",
 		"@types/chai-spies": "^1.0.1",
@@ -133,7 +134,7 @@
 		"@nivo/heatmap": "^0.61.0",
 		"@nivo/line": "^0.61.1",
 		"@nivo/pie": "^0.61.1",
-		"@rocket.chat/apps-engine": "1.19.0-alpha.3984",
+		"@rocket.chat/apps-engine": "1.19.0-alpha.4006",
 		"@rocket.chat/css-in-js": "^0.17.0",
 		"@rocket.chat/fuselage": "^0.17.0",
 		"@rocket.chat/fuselage-hooks": "^0.17.0",
@@ -151,6 +152,7 @@
 		"@types/xml-crypto": "^1.4.1",
 		"@types/xmldom": "^0.1.30",
 		"adm-zip": "RocketChat/adm-zip",
+		"agenda": "^3.1.0",
 		"apn": "2.2.0",
 		"archiver": "^3.1.1",
 		"arraybuffer-to-string": "^1.0.2",