Added experimental background job queue

ghost/core/core/server/data/schema/schema.js

@@ -985,7 +985,9 @@ module.exports = {
         started_at: {type: 'dateTime', nullable: true},
         finished_at: {type: 'dateTime', nullable: true},
         created_at: {type: 'dateTime', nullable: false},
-        updated_at: {type: 'dateTime', nullable: true}
+        updated_at: {type: 'dateTime', nullable: true},
+        metadata: {type: 'text', maxlength: 1000000000, fieldtype: 'long', nullable: true},
+        queue_entry: {type: 'integer', nullable: true, unsigned: true}
     },
     redirects: {
         id: {type: 'string', maxlength: 24, nullable: false, primary: true},

ghost/core/core/server/services/email-analytics/EmailAnalyticsServiceWrapper.js

@@ -1,4 +1,6 @@
 const logging = require('@tryghost/logging');
+const JobManager = require('../../services/jobs');
+const path = require('path');
 
 class EmailAnalyticsServiceWrapper {
     init() {
@@ -11,7 +13,7 @@ class EmailAnalyticsServiceWrapper {
         const MailgunProvider = require('@tryghost/email-analytics-provider-mailgun');
         const {EmailRecipientFailure, EmailSpamComplaintEvent, Email} = require('../../models');
         const StartEmailAnalyticsJobEvent = require('./events/StartEmailAnalyticsJobEvent');
-
+        const {MemberEmailAnalyticsUpdateEvent} = require('@tryghost/member-events');
         const domainEvents = require('@tryghost/domain-events');
         const config = require('../../../shared/config');
         const settings = require('../../../shared/settings-cache');
@@ -47,14 +49,28 @@ class EmailAnalyticsServiceWrapper {
             providers: [
                 new MailgunProvider({config, settings})
             ],
-            queries
+            queries,
+            domainEvents
         });
 
         // We currently cannot trigger a non-offloaded job from the job manager
         // So the email analytics jobs simply emits an event.
         domainEvents.subscribe(StartEmailAnalyticsJobEvent, async () => {
             await this.startFetch();
         });
+
+        domainEvents.subscribe(MemberEmailAnalyticsUpdateEvent, async (event) => {
+            const memberId = event.data.memberId;
+            await JobManager.addQueuedJob({
+                name: `update-member-email-analytics-${memberId}`,
+                metadata: {
+                    job: path.resolve(__dirname, 'jobs/update-member-email-analytics'),
+                    data: {
+                        memberId
+                    }
+                }
+            });
+        });
     }
 
     async fetchLatestOpenedEvents({maxEvents} = {maxEvents: Infinity}) {

ghost/core/core/server/services/email-analytics/jobs/index.js

@@ -29,7 +29,7 @@ module.exports = {
                 const m = Math.floor(Math.random() * 5); // 0-4
 
                 jobsService.addJob({
-                    at: `${s} ${m}/5 * * * *`,
+                    at: `${s} ${m}/1 * * * *`,
                     job: path.resolve(__dirname, 'fetch-latest/index.js'),
                     name: 'email-analytics-fetch-latest'
                 });

ghost/core/core/server/services/email-analytics/jobs/update-member-email-analytics/index.js

@@ -0,0 +1,13 @@
+const queries = require('../../lib/queries');
+
+/**
+ * Updates email analytics for a specific member
+ * 
+ * @param {Object} options - The options object
+ * @param {string} options.memberId - The ID of the member to update analytics for
+ * @returns {Promise<Object>} The result of the aggregation query (1/0)
+ */
+module.exports = async function updateMemberEmailAnalytics({memberId}) {
+    const result = await queries.aggregateMemberStats(memberId);
+    return result;
+};

ghost/core/core/server/services/jobs/job-service.js

@@ -8,6 +8,8 @@ const logging = require('@tryghost/logging');
 const models = require('../../models');
 const sentry = require('../../../shared/sentry');
 const domainEvents = require('@tryghost/domain-events');
+const config = require('../../../shared/config');
+const db = require('../../data/db');
 
 const errorHandler = (error, workerMeta) => {
     logging.info(`Capturing error for worker during execution of job: ${workerMeta.name}`);
@@ -42,7 +44,7 @@ const initTestMode = () => {
     }, 5000);
 };
 
-const jobManager = new JobManager({errorHandler, workerMessageHandler, JobModel: models.Job, domainEvents});
+const jobManager = new JobManager({errorHandler, workerMessageHandler, JobModel: models.Job, domainEvents, config, db: db});
 
 module.exports = jobManager;
 module.exports.initTestMode = initTestMode;

ghost/core/core/server/services/mentions-jobs/job-service.js

@@ -42,7 +42,7 @@ const initTestMode = () => {
     }, 5000);
 };
 
-const jobManager = new JobManager({errorHandler, workerMessageHandler, JobModel: models.Job, domainEvents});
+const jobManager = new JobManager({errorHandler, workerMessageHandler, JobModel: models.Job, domainEvents, isDuplicate: true});
 
 module.exports = jobManager;
 module.exports.initTestMode = initTestMode;

ghost/email-analytics-service/lib/EmailAnalyticsService.js

@@ -1,6 +1,7 @@
 const EventProcessingResult = require('./EventProcessingResult');
 const logging = require('@tryghost/logging');
 const errors = require('@tryghost/errors');
+const {MemberEmailAnalyticsUpdateEvent} = require('@tryghost/member-events');
 
 /**
  * @typedef {import('@tryghost/email-service').EmailEventProcessor} EmailEventProcessor
@@ -73,13 +74,15 @@ module.exports = class EmailAnalyticsService {
      * @param {object} dependencies.queries
      * @param {EmailEventProcessor} dependencies.eventProcessor
      * @param {object} dependencies.providers
+     * @param {import('@tryghost/domain-events')} dependencies.domainEvents
      */
-    constructor({config, settings, queries, eventProcessor, providers}) {
+    constructor({config, settings, queries, eventProcessor, providers, domainEvents}) {
         this.config = config;
         this.settings = settings;
         this.queries = queries;
         this.eventProcessor = eventProcessor;
         this.providers = providers;
+        this.domainEvents = domainEvents;
     }
 
     getStatus() {
@@ -511,7 +514,12 @@ module.exports = class EmailAnalyticsService {
         startTime = Date.now();
         logging.info(`[EmailAnalytics] Aggregating for ${memberIds.length} members`);
         for (const memberId of memberIds) {
-            await this.aggregateMemberStats(memberId);
+            if (this.config.get('services:jobs:queue:enabled')) {
+                // With the queue enabled we will dispatch an event to update the member email analytics on the background queue (multithreaded :))
+                await this.domainEvents.dispatch(MemberEmailAnalyticsUpdateEvent.create({memberId}));
+            } else {
+                await this.aggregateMemberStats(memberId);
+            }
         }
         endTime = Date.now() - startTime;
         logging.info(`[EmailAnalytics] Aggregating for ${memberIds.length} members took ${endTime}ms`);

ghost/job-manager/lib/JobManager.js

@@ -10,6 +10,7 @@ const logging = require('@tryghost/logging');
 const isCronExpression = require('./is-cron-expression');
 const assembleBreeJob = require('./assemble-bree-job');
 const JobsRepository = require('./JobsRepository');
+const JobQueueManager = require('./JobQueueManager');
 
 const worker = async (task, callback) => {
     try {
@@ -27,22 +28,36 @@ const ALL_STATUSES = {
     queued: 'queued'
 };
 
+/**
+ * @typedef {Object} ScheduledJob
+ * @property {Function | string} job - Function or path to a module defining a job
+ * @property {string} [name] - Unique job name, if not provided takes function name or job script filename
+ * @property {string | Date} [at] - Date, cron or human readable schedule format
+ * @property {Object} [data] - Data to be passed into the job
+ * @property {boolean} [offloaded=true] - If true, creates an "offloaded" job running in a worker thread. If false, runs an "inline" job on the same event loop
+ */
 class JobManager {
     #domainEvents;
     #completionPromises = new Map();
+    #jobQueueManager = null;
+    #config;
 
     /**
      * @param {Object} options
      * @param {Function} [options.errorHandler] - custom job error handler
      * @param {Function} [options.workerMessageHandler] - custom message handler coming from workers
      * @param {Object} [options.JobModel] - a model which can persist job data in the storage
      * @param {Object} [options.domainEvents] - domain events emitter
+     * @param {Object} [options.config] - config
+     * @param {boolean} [options.isDuplicate] - if true, the job manager will not initialize the job queue
+     * @param {Object} [options.db] - the database object
      */
-    constructor({errorHandler, workerMessageHandler, JobModel, domainEvents}) {
-        this.queue = fastq(this, worker, 3);
+    constructor({errorHandler, workerMessageHandler, JobModel, domainEvents, config, isDuplicate = false, db}) {
+        this.inlineQueue = fastq(this, worker, 3);
         this._jobMessageHandler = this._jobMessageHandler.bind(this);
         this._jobErrorHandler = this._jobErrorHandler.bind(this);
         this.#domainEvents = domainEvents;
+        this.#config = config;
 
         const combinedMessageHandler = workerMessageHandler
             ? ({name, message}) => {
@@ -72,7 +87,15 @@ class JobManager {
         });
 
         if (JobModel) {
-            this._jobsRepository = new JobsRepository({JobModel});
+            this._jobsRepository = new JobsRepository({JobModel, db});
+        }
+
+        // We have a duplicate job manager in Ghost core for the mentions job service that should be
+        //  refactored to use the job queue when we're able.
+        if (!isDuplicate && this.#config.get('services:jobs:queue:enabled')) {
+            logging.info(`[JobManager] Initializing job queue based on config`);
+            this.#jobQueueManager = new JobQueueManager({JobModel, config, db});
+            this.#jobQueueManager.init();
         }
     }
 
@@ -94,6 +117,31 @@ class JobManager {
         };
     }
 
+    /**
+     * @typedef {Object} QueuedJob
+     * @property {string} name - The name or identifier of the job.
+     * @property {Object} metadata - Metadata associated with the job.
+     * @property {string} metadata.job - The absolute path to the job to execute.
+     * @property {Object} metadata.data - The data associated with the job.
+     */
+
+    /**
+     * @method addQueuedJob
+     * @async
+     * @description Adds a new job to the job repository, which will be polled and executed by the job queue manager.
+     * @param {QueuedJob} job - The job to be added to the queue.
+     * @returns {Promise<Object>} The added job model.
+     */
+    async addQueuedJob({name, metadata}) {
+        // Adding some extra security so we don't add jobs when the queue is disabled from callers.
+        if (this.#config.get('services:jobs:queue:enabled')) {
+            const model = await this.#jobQueueManager.addJob({name, metadata});
+            return model;
+        }
+        logging.warn('[JobManager] Job queue is disabled but job was attempted to be added. job: ', name);
+        return Promise.reject();
+    }
+
     async _jobMessageHandler({name, message}) {
         if (name) {
             if (message === ALL_STATUSES.started) {
@@ -128,7 +176,7 @@ class JobManager {
                     this.#completionPromises.delete(name);
                 }
 
-                if (this.queue.length() <= 1) {
+                if (this.inlineQueue.length() <= 1) {
                     if (this.#completionPromises.has('all')) {
                         for (const listeners of this.#completionPromises.get('all')) {
                             listeners.resolve();
@@ -168,7 +216,7 @@ class JobManager {
             this.#completionPromises.delete(jobMeta.name);
         }
 
-        if (this.queue.length() <= 1) {
+        if (this.inlineQueue.length() <= 1) {
             if (this.#completionPromises.has('all')) {
                 for (const listeners of this.#completionPromises.get('all')) {
                     listeners.reject(error);
@@ -181,7 +229,7 @@ class JobManager {
 
     /**
      * By default schedules an "offloaded" job. If `offloaded: true` parameter is set,
-     * puts an "inline" immediate job into the queue.
+     * puts an "inline" immediate job into the inlineQueue.
      *
      * @param {Object} GhostJob - job options
      * @prop {Function | String} GhostJob.job - function or path to a module defining a job
@@ -192,7 +240,7 @@ class JobManager {
      */
     addJob({name, at, job, data, offloaded = true}) {
         if (offloaded) {
-            logging.info('Adding offloaded job to the queue');
+            logging.info('Adding offloaded job to the inline job queue');
             let schedule;
 
             if (!name) {
@@ -229,9 +277,9 @@ class JobManager {
             this.bree.add(breeJob);
             return this.bree.start(name);
         } else {
-            logging.info(`Adding one-off job to queue with current length = ${this.queue.length()} called '${name || 'anonymous'}'`);
+            logging.info(`Adding one-off job to inlineQueue with current length = ${this.inlineQueue.length()} called '${name || 'anonymous'}'`);
 
-            this.queue.push(async () => {
+            this.inlineQueue.push(async () => {
                 try {
                     // NOTE: setting the status here otherwise it is impossible to
                     //       distinguish between states when the job fails immediately
@@ -325,13 +373,11 @@ class JobManager {
     /**
      * Awaits completion of the offloaded one-off job.
      * CAUTION: it might take a long time to resolve!
-     * @param {String} name one-off job name
+     * @param {string} name one-off job name
      * @returns resolves with a Job model at current state
      */
     async awaitOneOffCompletion(name) {
-        const persistedJob = await this._jobsRepository.read({
-            name
-        });
+        const persistedJob = await this._jobsRepository.read(name);
 
         if (!persistedJob || ![ALL_STATUSES.finished, ALL_STATUSES.failed].includes(persistedJob.get('status'))) {
             // NOTE: can implement exponential backoff here if that's ever needed
@@ -366,7 +412,7 @@ class JobManager {
         const name = 'all';
 
         return new Promise((resolve, reject) => {
-            if (this.queue.idle()) {
+            if (this.inlineQueue.idle()) {
                 resolve();
                 return;
             }
@@ -379,7 +425,7 @@ class JobManager {
     }
 
     /**
-     * Removes an "offloaded" job from scheduled jobs queue.
+     * Removes an "offloaded" job from scheduled jobs inlineQueue.
      * It's NOT yet possible to remove "inline" jobs (will be possible when scheduling is added https://github.com/breejs/bree/issues/68).
      * The method will throw an Error if job with provided name does not exist.
      *
@@ -398,15 +444,15 @@ class JobManager {
     async shutdown(options) {
         await this.bree.stop();
 
-        if (this.queue.idle()) {
+        if (this.inlineQueue.idle()) {
             return;
         }
 
-        logging.warn('Waiting for busy job queue');
+        logging.warn('Waiting for busy job in inline job queue');
 
-        await pWaitFor(() => this.queue.idle() === true, options);
+        await pWaitFor(() => this.inlineQueue.idle() === true, options);
 
-        logging.warn('Job queue finished');
+        logging.warn('Inline job queue finished');
     }
 }