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: 'string', maxlength: 2000, 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/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,7 @@ 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 errorHandler = (error, workerMeta) => {
     logging.info(`Capturing error for worker during execution of job: ${workerMeta.name}`);
@@ -24,7 +25,7 @@ const workerMessageHandler = ({name, message}) => {
 const initTestMode = () => {
     // Output job queue length every 5 seconds
     setInterval(() => {
-        logging.warn(`${jobManager.queue.length()} jobs in the queue. Idle: ${jobManager.queue.idle()}`);
+        logging.warn(`${jobManager.inlineQueue.length()} jobs in the queue. Idle: ${jobManager.inlineQueue.idle()}`);
 
         const runningScheduledjobs = Object.keys(jobManager.bree.workers);
         if (Object.keys(jobManager.bree.workers).length) {
@@ -42,7 +43,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});
 
 module.exports = jobManager;
 module.exports.initTestMode = initTestMode;

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

@@ -24,7 +24,7 @@ const workerMessageHandler = ({name, message}) => {
 const initTestMode = () => {
     // Output job queue length every 5 seconds
     setInterval(() => {
-        logging.warn(`${jobManager.queue.length()} jobs in the queue. Idle: ${jobManager.queue.idle()}`);
+        logging.warn(`${jobManager.inlineQueue.length()} jobs in the queue. Idle: ${jobManager.inlineQueue.idle()}`);
 
         const runningScheduledjobs = Object.keys(jobManager.bree.workers);
         if (Object.keys(jobManager.bree.workers).length) {
@@ -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/core/test/integration/jobs/job-queue.test.js

@@ -0,0 +1,100 @@
+const assert = require('assert/strict');
+const path = require('path');
+const configUtils = require('../../utils/configUtils');
+const dbUtils = require('../../utils/db-utils');
+const models = require('../../../core/server/models');
+
+// Helper function to wait for job completion
+async function waitForJobCompletion(jobName, maxWaitTimeMs = 5000, checkIntervalMs = 50) {
+    return new Promise((resolve, reject) => {
+        const startTime = Date.now();
+        const intervalId = setInterval(async () => {
+            if (Date.now() - startTime >= maxWaitTimeMs) {
+                clearInterval(intervalId);
+                reject(new Error(`Job ${jobName} did not complete within ${maxWaitTimeMs}ms`));
+            }
+            const job = await models.Job.findOne({name: jobName});
+            if (!job) {
+                clearInterval(intervalId);
+                resolve();
+            }
+        }, checkIntervalMs);
+    });
+}
+
+describe('Job Queue', function () {
+    let jobService;
+
+    describe.only('enabled by config', function () {
+        beforeEach(async function () {
+            models.init();
+            configUtils.set('services:jobs:queue:enabled', true);
+            await new Promise((resolve) => {
+                setTimeout(resolve, 500);
+            });
+            jobService = require('../../../core/server/services/jobs/job-service');
+        });
+
+        afterEach(async function () {
+            await configUtils.restore();
+            await dbUtils.teardown();
+        });
+
+        it('should add and execute a job in the queue', async function () {
+            this.timeout(10000); // Increase timeout if needed
+
+            const job = {
+                name: `add-random-numbers-${Date.now()}`,
+                metadata: {
+                    job: path.resolve(__dirname, './test-job.js'),
+                    data: {}
+                }
+            };
+
+            // Add the job to the queue
+            console.log('Adding job to queue:', job.name);
+            const result = await jobService.addQueuedJob(job);
+            console.log('Job added:', result);
+            assert.ok(result);
+
+            // Wait for the job to complete
+            await waitForJobCompletion(job.name, 8000); // Increase wait time
+
+            // Check job status
+            const jobEntry = await models.Job.findOne({name: job.name});
+            console.log('Job status:', jobEntry ? jobEntry.status : 'Not found');
+
+            // Verify that the job no longer exists in the queue
+            assert.equal(jobEntry, null);
+        });
+    });
+
+    describe('not enabled', function () {
+        beforeEach(async function () {
+            models.init();
+            jobService = require('../../../core/server/services/jobs/job-service');
+        });
+
+        afterEach(async function () {
+            await dbUtils.teardown();
+        });
+
+        it('should not add a job to the queue when disabled', async function () {
+            const job = {
+                name: `add-random-numbers-${Date.now()}`,
+                metadata: {
+                    job: path.resolve(__dirname, './test-job.js'),
+                    data: {}
+                }
+            };
+
+            // Attempt to add the job to the queue
+            const result = await jobService.addQueuedJob(job);
+            assert.equal(result, undefined);
+
+            // Verify that the job doesn't exist in the queue
+            const jobEntry = await models.Job.findOne({name: job.name});
+            assert.equal(jobEntry, null);
+        });
+    });
+});

ghost/core/test/integration/jobs/test-job.js

@@ -0,0 +1,7 @@
+module.exports = function testJob() {
+    const num1 = Math.floor(Math.random() * 100);
+    const num2 = Math.floor(Math.random() * 100);
+    const result = num1 + num2;
+
+    return result;
+};

ghost/core/test/unit/server/data/schema/integrity.test.js

@@ -35,7 +35,7 @@ const validateRouteSettings = require('../../../../../core/server/services/route
  */
 describe('DB version integrity', function () {
     // Only these variables should need updating
-    const currentSchemaHash = 'a4f016480ff73c6f52ee4c86482b45a7';
+    const currentSchemaHash = '1110f25f639c22135b9845c72f0be7ef';
     const currentFixturesHash = 'a489d615989eab1023d4b8af0ecee7fd';
     const currentSettingsHash = '051ef2a50e2edb8723e89461448313cb';
     const currentRoutesHash = '3d180d52c663d173a6be791ef411ed01';

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`);