Removed job queue manager class (#23167)

ref https://linear.app/ghost/issue/ENG-2028/

We created this persisted queue using a MySQL-based queue to prototype a
more robust queuing system, intending to use Redis, RabbitMQ, or the
like. We didn't find enough benefit from this and found other
constraints, primarily r/w to the db being burdensome, and not enough
benefit to justify the move to RMQ/Redis at this time.

Given that, we're removing the job queue manager. We've previously
removed any calling code, so now we're pulling the class itself and will
follow up with a migration (or two) to clean up the jobs table schema
and entries.

Author: 9larsons

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

@@ -1,12 +1,11 @@
 # Job Manager
 
-A manager for background jobs in Ghost, supporting one-off tasks, recurring jobs, and persistent job queues.
+A manager for background jobs in Ghost, supporting one-off tasks and recurring jobs.
 
 ## Table of Contents
 - [Quick Start](#quick-start)
 - [Job Types](#job-types)
 - [Background Job Requirements](#background-job-requirements)
-- [Configuration](#configuration)
 - [Advanced Usage](#advanced-usage)
 - [Technical Details](#technical-details)
 
@@ -29,20 +28,11 @@ jobManager.addJob({
     at: 'every 5 minutes',
     job: './jobs/check-emails.js'
 });
-
-// Persisted background job
-jobManager.addQueuedJob({
-    name: 'update-member-analytics-123',
-    metadata: {
-        job: './jobs/update-analytics.js',
-        data: { memberId: '123' }
-    }
-});
 ```
 
 ## Job Types
 
-Ghost supports three types of jobs:
+Ghost supports two types of jobs:
 
 1. **Inline Jobs**
    - Run in the main Ghost process
@@ -54,39 +44,16 @@ Ghost supports three types of jobs:
    - Good for CPU-intensive tasks
    - Can be scheduled or recurring
 
-3. **Persisted Queue Jobs**
-   - Stored in database
-   - Survive server restarts
-   - Best for background tasks
-   - Can be monitored and retried
-
 ## Background Job Requirements
 
-Both offloaded and queued jobs must:
+Offloaded jobs must:
 
 - Have a unique name
 - Be idempotent (safe to run multiple times)
 - Use minimal parameters (prefer using IDs rather than full objects)
 - Import their own dependencies
 - Primarily use DB or API calls
 
-## Configuration
-
-The job queue can be configured through Ghost's config:
-
-```js
-jobs: {
-    queue: {
-        enabled: true,
-        maxWorkers: 1,
-        pollMinInterval: 1000,  // 1 sec
-        pollMaxInterval: 60000, // 1 min
-        queueCapacity: 500,    // Max queued jobs
-        fetchCount: 500        // Max jobs per poll
-    }
-}
-```
-
 ## Advanced Usage
 
 Below is a sample code to wire up job manger and initialize jobs. This is the simplest way to interact with the job manager - these jobs do not persist after reboot:
@@ -182,66 +149,3 @@ Offloaded jobs are running on dedicated worker threads which makes their lifecyc
 4. When **exceptions** happen and expected outcome is to terminate current job, leave the exception unhandled allowing it to bubble up to the job manager. Unhandled exceptions [terminate current thread](https://nodejs.org/dist/latest-v14.x/docs/api/worker_threads.html#worker_threads_event_error) and allow for next scheduled job execution to happen.
 
 For more nuances on job structure best practices check [bree documentation](https://github.com/breejs/bree#writing-jobs-with-promises-and-async-await).
-
-### Implementation Notes
-For any persisted tasks, the Job Manager has a queue based on the `jobs` table. This table is polled regularly and processed with a single worker, and is ideal for background tasks, e.g. updating member analytics. (see notes below about job types)
-
-```js
-// the job manager is typically injected into the consumer service via the service wrapper
-//  from there 
-const JobManager = require('../../services/jobs');
-...
-
-// ** job submission should be handled by the wrapper service **
-// this could be via subscription to events, emitted by the wrapped service(s)
-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'),
-            name: 'update-member-email-analytics',
-            data: {
-                memberId
-            }
-        }
-    });
-});
-
-// or it could be passed through as a hook
-const handleAnalyticsJobSubmission = async (memberId) => {
-    await JobManager.addQueuedJob({
-        name: `update-member-email-analytics-${memberId}`,
-        metadata: {
-            job: path.resolve(__dirname, 'jobs/update-member-email-analytics'),
-            name: 'update-member-email-analytics',
-            data: {
-                memberId
-            }
-        }
-    });
-}
-```
-In most cases, jobs should not be submitted by services directly. Because they must import what is needed, it would require too many injected dependencies.
-
-### Adjusting the Job Queue
-The queue manager will poll the `jobs` table every minute unless jobs are being actively or were recently processed, where it will instead poll every second. 
-
-The job queue has a few other config flags for the number of workers, polling rate, max jobs to process, etc.
-
-```
-services: {
-    jobs: {
-        queue: {
-            enabled: true,
-            reportStats: true,
-            maxWorkers: 1,
-            logLevel: 'info' | 'error',
-            pollMinInterval: 1000, // 1 sec
-            pollMaxInterval: 60000, // 1 min
-            queueCapacity: 500, // # of jobs in the process queue at any time
-            fetchCount: 500 // max # of jobs fetched in each poll
-        }
-    }
-}
-```

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

@@ -7,11 +7,9 @@ const Bree = require('bree');
 const pWaitFor = require('p-wait-for');
 const {UnhandledJobError, IncorrectUsageError} = require('@tryghost/errors');
 const logging = require('@tryghost/logging');
-const metrics = require('@tryghost/metrics');
 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 {
@@ -40,7 +38,6 @@ const ALL_STATUSES = {
 class JobManager {
     #domainEvents;
     #completionPromises = new Map();
-    #jobQueueManager = null;
     #config;
     #JobModel;
     #events;
@@ -52,11 +49,9 @@ class JobManager {
      * @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 {JobQueueManager} [options.jobQueueManager] - job queue manager instance (for testing)
      * @param {Object} [options.events] - events instance (for testing)
      */
-    constructor({errorHandler, workerMessageHandler, JobModel, domainEvents, config, isDuplicate = false, jobQueueManager = null, events = null}) {
+    constructor({errorHandler, workerMessageHandler, JobModel, domainEvents, config, events = null}) {
         this.inlineQueue = fastq(this, worker, 3);
         this._jobMessageHandler = this._jobMessageHandler.bind(this);
         this._jobErrorHandler = this._jobErrorHandler.bind(this);
@@ -95,19 +90,6 @@ class JobManager {
         if (JobModel) {
             this._jobsRepository = new JobsRepository({JobModel});
         }
-
-        if (jobQueueManager) {
-            this.#jobQueueManager = jobQueueManager;
-        } else if (!isDuplicate) {
-            this.#initializeJobQueueManager();
-        }
-    }
-
-    #initializeJobQueueManager() {
-        if (this.#config?.get('services:jobs:queue:enabled') === true && !this.#jobQueueManager) {
-            this.#jobQueueManager = new JobQueueManager({JobModel: this.#JobModel, config: this.#config, eventEmitter: this.#events, metricLogger: metrics});
-            this.#jobQueueManager.init();
-        }
     }
 
     inlineJobHandler(jobName) {
@@ -128,35 +110,6 @@ 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 {string} metadata.name - The name of the job. Used for metrics.
-     * @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}) {
-        // Try to initialize JobQueueManager if it's missing
-        if (!this.#jobQueueManager) {
-            this.#initializeJobQueueManager();
-        }
-
-        if (this.#config?.get('services:jobs:queue:enabled') === true && this.#jobQueueManager) {
-            const model = await this.#jobQueueManager.addJob({name, metadata});
-            return model;
-        }
-        return undefined;
-    }
-
     async _jobMessageHandler({name, message}) {
         if (name) {
             if (message === ALL_STATUSES.started) {
@@ -459,10 +412,6 @@ class JobManager {
     async shutdown(options) {
         await this.bree.stop();
 
-        if (this.#jobQueueManager) {
-            await this.#jobQueueManager.shutdown();
-        }
-
         if (this.inlineQueue.idle()) {
             return;
         }