Moved gift subscription cron jobs to main thread via domain events (TryGhost#27639)

closes TryGhost#27638

- gift cleanup and reminder jobs were running in Bree worker threads
with no service init, forcing each worker to manually re-init every
transitive dependency (settings, stripe, members, tiers, email-address)
before calling into the gift service
- the two job files had drifted apart with different and incomplete init
lists, encoding boot order by hand
- replaced both worker bodies with a thin trigger that posts a domain
event back to the main thread, where the `GiftServiceWrapper` subscribes
and runs the work in the already-initialised process
- this is the same pattern used by `email-analytics/fetch-latest` and
`outbox-job` since 2023, and unifies the cron path with the existing
admin API flush endpoint which already used
`StartGiftReminderFlushEvent`
- added a sibling `StartGiftCleanupEvent` for the `processConsumed` +
`processExpired` flow

Author: mike182uk

ghost/core/core/server/services/gifts/events/start-gift-cleanup-event.js

@@ -0,0 +1,18 @@
+module.exports = class StartGiftCleanupEvent {
+    /**
+     * @param {any} data
+     * @param {Date} timestamp
+     */
+    constructor(data, timestamp) {
+        this.data = data;
+        this.timestamp = timestamp;
+    }
+
+    /**
+     * @param {any} [data]
+     * @param {Date} [timestamp]
+     */
+    static create(data, timestamp) {
+        return new StartGiftCleanupEvent(data, timestamp ?? new Date());
+    }
+};

ghost/core/core/server/services/gifts/events/start-gift-reminder-flush-event.js

@@ -1,16 +1,18 @@
 module.exports = class StartGiftReminderFlushEvent {
     /**
+     * @param {any} data
      * @param {Date} timestamp
      */
-    constructor(timestamp) {
-        this.data = null;
+    constructor(data, timestamp) {
+        this.data = data;
         this.timestamp = timestamp;
     }
 
     /**
-     * @returns {StartGiftReminderFlushEvent}
+     * @param {any} [data]
+     * @param {Date} [timestamp]
      */
-    static create() {
-        return new StartGiftReminderFlushEvent(new Date());
+    static create(data, timestamp) {
+        return new StartGiftReminderFlushEvent(data, timestamp ?? new Date());
     }
 };

ghost/core/core/server/services/gifts/gift-service-wrapper.js

@@ -41,6 +41,8 @@ class GiftServiceWrapper {
         const logging = require('@tryghost/logging');
         const {SubscriptionActivatedEvent} = require('../../../shared/events');
         const StartGiftReminderFlushEvent = require('./events/start-gift-reminder-flush-event');
+        const StartGiftCleanupEvent = require('./events/start-gift-cleanup-event');
+        const jobs = require('./jobs');
 
         const {GhostMailer} = require('../mail');
         const settingsCache = require('../../../shared/settings-cache');
@@ -100,13 +102,41 @@ class GiftServiceWrapper {
         });
 
         DomainEvents.subscribe(StartGiftReminderFlushEvent, async () => {
+            const start = Date.now();
             try {
-                await this.service.processReminders();
+                const {remindedCount, skippedCount, failedCount} = await this.service.processReminders();
+
+                logging.info(`Sent ${remindedCount} gift reminders, skipped ${skippedCount}, failed ${failedCount} in ${Date.now() - start}ms`);
+            } catch (err) {
+                logging.error(err, 'Failed to process gift reminders');
+            }
+        });
+
+        DomainEvents.subscribe(StartGiftCleanupEvent, async () => {
+            const consumedStart = Date.now();
+            try {
+                const {consumedCount, updatedMemberCount} = await this.service.processConsumed();
+
+                logging.info(`Consumed ${consumedCount} gifts, updated ${updatedMemberCount} members in ${Date.now() - consumedStart}ms`);
             } catch (err) {
-                logging.error(err, 'Failed to flush gift reminders');
+                logging.error(err, 'Failed to process consumed gifts');
+            }
+
+            const expiredStart = Date.now();
+            try {
+                const {expiredCount} = await this.service.processExpired();
+
+                logging.info(`Expired ${expiredCount} gifts in ${Date.now() - expiredStart}ms`);
+            } catch (err) {
+                logging.error(err, 'Failed to process expired gifts');
             }
         });
 
+        if (labsService.isSet('giftSubscriptions')) {
+            jobs.scheduleGiftCleanupJob();
+            jobs.scheduleGiftReminderJob();
+        }
+
         this.#initialized = true;
     }
 }

ghost/core/core/server/services/gifts/jobs/clean-gifts-job.js

@@ -0,0 +1,46 @@
+const {parentPort} = require('node:worker_threads');
+const StartGiftCleanupEvent = require('../events/start-gift-cleanup-event');
+
+// Recurring job to clean consumed and expired gifts. The actual work runs on
+// the main thread via a domain event so we don't have to re-initialize every
+// service dependency inside this worker
+
+// Exit early when cancelled to prevent stalling shutdown. No cleanup needed
+// when cancelling as everything is idempotent and will pick up where it left
+// off on next run
+function cancel() {
+    if (parentPort) {
+        parentPort.postMessage('Gift cleanup job cancelled before completion');
+        parentPort.postMessage('cancelled');
+    } else {
+        setTimeout(() => {
+            process.exit(0);
+        }, 1000);
+    }
+}
+
+if (parentPort) {
+    parentPort.once('message', (message) => {
+        if (message === 'cancel') {
+            return cancel();
+        }
+    });
+}
+
+(async () => {
+    if (parentPort) {
+        // Bounce to the main thread via the JobManager's event bridge - the
+        // GiftServiceWrapper subscribes to this event and runs the work in the
+        // already-initialised main process
+        parentPort.postMessage({
+            event: {
+                type: StartGiftCleanupEvent.name
+            }
+        });
+        parentPort.postMessage('done');
+    } else {
+        setTimeout(() => {
+            process.exit(0);
+        }, 1000);
+    }
+})();

ghost/core/core/server/services/gifts/jobs/index.js

@@ -0,0 +1,40 @@
+const path = require('path');
+const jobsService = require('../../jobs');
+
+let hasScheduled = {
+    cleanup: false,
+    reminders: false
+};
+
+function scheduleJob(key, name, jobFile) {
+    if (hasScheduled[key] || process.env.NODE_ENV?.startsWith('test')) {
+        return hasScheduled[key];
+    }
+
+    // randomise the schedule so the job doesn't fire at the same instant
+    // across every Ghost instance - spreads load across the day and avoids
+    // DB spikes on the hour. Hour is bounded to a 0-5am off-peak window.
+    const s = Math.floor(Math.random() * 60);
+    const m = Math.floor(Math.random() * 60);
+    const h = Math.floor(Math.random() * 6);
+
+    jobsService.addJob({
+        at: `${s} ${m} ${h} * * *`,
+        job: path.resolve(__dirname, jobFile),
+        name
+    });
+
+    hasScheduled[key] = true;
+
+    return true;
+}
+
+module.exports = {
+    scheduleGiftCleanupJob() {
+        return scheduleJob('cleanup', 'clean-gifts', 'clean-gifts-job.js');
+    },
+
+    scheduleGiftReminderJob() {
+        return scheduleJob('reminders', 'send-gift-reminders', 'send-gift-reminders-job.js');
+    }
+};

ghost/core/core/server/services/gifts/jobs/send-gift-reminders-job.js

@@ -0,0 +1,46 @@
+const {parentPort} = require('node:worker_threads');
+const StartGiftReminderFlushEvent = require('../events/start-gift-reminder-flush-event');
+
+// Recurring job to send gift reminder emails. The actual work runs on the
+// main thread via a domain event so we don't have to re-initialize every
+// service dependency inside this worker
+
+// Exit early when cancelled to prevent stalling shutdown. No cleanup needed
+// when cancelling as everything is idempotent - any gift that hasn't yet had
+// its reminder recorded will be picked up on the next run
+function cancel() {
+    if (parentPort) {
+        parentPort.postMessage('Gift reminder job cancelled before completion');
+        parentPort.postMessage('cancelled');
+    } else {
+        setTimeout(() => {
+            process.exit(0);
+        }, 1000);
+    }
+}
+
+if (parentPort) {
+    parentPort.once('message', (message) => {
+        if (message === 'cancel') {
+            return cancel();
+        }
+    });
+}
+
+(async () => {
+    if (parentPort) {
+        // Bounce to the main thread via the JobManager's event bridge - the
+        // GiftServiceWrapper subscribes to this event and runs the work in the
+        // already-initialised main process
+        parentPort.postMessage({
+            event: {
+                type: StartGiftReminderFlushEvent.name
+            }
+        });
+        parentPort.postMessage('done');
+    } else {
+        setTimeout(() => {
+            process.exit(0);
+        }, 1000);
+    }
+})();

ghost/core/core/server/services/members/jobs/clean-gifts.js

@@ -3,8 +3,6 @@ const jobsService = require('../../jobs');
 
 let hasScheduled = {
     expiredComped: false,
-    giftCleanup: false,
-    giftReminders: false,
     tokens: false
 };
 
@@ -33,14 +31,6 @@ module.exports = {
         return scheduleJob('expiredComped', 'clean-expired-comped', 'clean-expired-comped.js');
     },
 
-    async scheduleGiftCleanupJob() {
-        return scheduleJob('giftCleanup', 'clean-gifts', 'clean-gifts.js');
-    },
-
-    async scheduleGiftReminderJob() {
-        return scheduleJob('giftReminders', 'send-gift-reminders', 'send-gift-reminders.js');
-    },
-
     async scheduleTokenCleanupJob() {
         return scheduleJob('tokens', 'clean-tokens', 'clean-tokens.js', 24);
     }

ghost/core/core/server/services/members/jobs/index.js

@@ -168,13 +168,6 @@ module.exports = {
 
         // Schedule daily cron job to clean expired tokens
         memberJobs.scheduleTokenCleanupJob();
-
-        // Schedule daily cron jobs to clean up consumed/expired gifts and to
-        // send gift reminder emails.
-        if (labsService.isSet('giftSubscriptions')) {
-            memberJobs.scheduleGiftCleanupJob();
-            memberJobs.scheduleGiftReminderJob();
-        }
     },
     contentGating: require('./content-gating'),