complex job operations from oplog

jobServer/Job.js

@@ -4,6 +4,7 @@ const defaultLogger = {
     e: console.error,
     i: console.info
 };
+const { JOB_PRIORITIES } = require('./constants/JobPriorities');
 /**
  * Base class for creating jobs.
  * 
@@ -274,6 +275,46 @@ class Job {
 
         this.logger?.d(`Progress reported for job "${this.jobName}":`, progressData);
     }
+
+    /**
+     * Get job retry configuration
+     * @typedef {Object} RetryConfig
+     * @property {boolean} enabled - Whether retries are enabled
+     * @property {number} attempts - Number of retry attempts
+     * @property {number} delay - Delay between retry attempts in milliseconds, delay is by default exponentially increasing after each attempt
+     * @returns {RetryConfig|null} Retry configuration or null for default
+     */
+    getRetryConfig() {
+        return {
+            enabled: true,
+            attempts: 3,
+            delay: 2000 // 2 seconds
+        };
+    }
+
+    /**
+     * Get job priority
+     * @returns {string} Priority level from JOB_PRIORITIES
+     */
+    getPriority() {
+        return JOB_PRIORITIES.NORMAL;
+    }
+
+    /**
+     * Get job concurrency
+     * @returns {number|null} Maximum concurrent instances or null for default
+     */
+    getConcurrency() {
+        return 1;
+    }
+
+    /**
+     * Get job lock lifetime in milliseconds
+     * @returns {number|null} Lock lifetime or null for default
+     */
+    getLockLifetime() {
+        return 55 * 60 * 1000; // 55 minutes
+    }
 }
 
 module.exports = Job;

jobServer/JobManager.js

@@ -1,5 +1,6 @@
-const {RUNNER_TYPES, createJobRunner} = require('./JobRunner');
+const {RUNNER_TYPES, createJobRunner} = require('./JobRunner/index');
 const config = require("./config");
+const JobUtils = require('./JobUtils');
 
 /**
  * Manages job configurations and initialization.
@@ -25,6 +26,9 @@ class JobManager {
      */
     #jobRunner = null;
 
+
+    #jobConfigsCollection;
+
     /**
      * Creates a new JobManager instance
      * @param {Object} db Database connection
@@ -40,6 +44,199 @@ class JobManager {
         const pulseConfig = config.PULSE;
 
         this.#jobRunner = createJobRunner(this.#db, runnerType, pulseConfig, Logger);
+        this.#jobConfigsCollection = db.collection('jobConfigs');
+        this.#watchConfigs();
+    }
+
+    /**
+     * Watches for changes in job configurations
+     * @private
+     * @returns {Promise<void>} A promise that resolves once the watcher is started
+     */
+    async #watchConfigs() {
+        const changeStream = this.#jobConfigsCollection.watch();
+
+        changeStream.on('change', async(change) => {
+            if (change.operationType === 'update' || change.operationType === 'insert') {
+                const jobConfig = change.fullDocument;
+                await this.#applyConfig(jobConfig);
+            }
+        });
+    }
+
+    /**
+     * Applies job configuration changes
+     * @private
+     * @param {Object} jobConfig The job configuration to apply
+     * @param {string} jobConfig.jobName The name of the job
+     * @param {boolean} [jobConfig.runNow] Whether to run the job immediately
+     * @param {Object} [jobConfig.schedule] Schedule configuration
+     * @param {Object} [jobConfig.retry] Retry configuration
+     * @param {boolean} [jobConfig.enabled] Whether the job is enabled
+     * @returns {Promise<void>} A promise that resolves once the configuration is applied
+     */
+    async #applyConfig(jobConfig) {
+        try {
+            if (!this.#jobRunner) {
+                throw new Error('Job runner not initialized');
+            }
+
+            const { jobName } = jobConfig;
+
+            if (jobConfig.runNow === true) {
+                await this.#jobRunner.runJobNow(jobName);
+                await this.#jobConfigsCollection.updateOne(
+                    { jobName },
+                    { $unset: { runNow: "" } }
+                );
+            }
+
+            if (jobConfig.schedule) {
+                await this.#jobRunner.updateSchedule(jobName, jobConfig.schedule);
+            }
+
+            if (jobConfig.retry) {
+                await this.#jobRunner.configureRetry(jobName, jobConfig.retry);
+            }
+
+            if (typeof jobConfig.enabled === 'boolean') {
+                if (jobConfig.enabled) {
+                    await this.#jobRunner.enableJob(jobName);
+                    this.#log.i(`Job ${jobName} enabled via config`);
+                }
+                else {
+                    await this.#jobRunner.disableJob(jobName);
+                    this.#log.i(`Job ${jobName} disabled via config`);
+                }
+            }
+        }
+        catch (error) {
+            this.#log.e('Error applying job configuration:', error);
+        }
+    }
+
+    /**
+     * Loads job classes and manages their configurations
+     * @private
+     * @param {Object.<string, Function>} jobClasses - Object containing job class implementations keyed by job name
+     * @returns {Promise<void>} A promise that resolves once all jobs are loaded and configured
+     * @throws {Error} If job loading or configuration fails
+     */
+    async #loadJobs(jobClasses) {
+        // Calculate checksums for all job definitions
+        const jobDefinitionChecksums = Object.entries(jobClasses).reduce((acc, [name, JobClass]) => {
+            acc[name] = JobUtils.calculateJobChecksum(JobClass);
+            return acc;
+        }, {});
+
+        // Initialize or update job configurations
+        await this.#initializeJobConfigs(jobClasses, jobDefinitionChecksums);
+
+        // Load and apply configurations
+        await this.#applyJobConfigurations(jobClasses, jobDefinitionChecksums);
+    }
+
+    /**
+     * Initializes or updates job configurations in the database
+     * @private
+     * @param {Object.<string, Function>} jobClasses - Job class implementations
+     * @param {Object.<string, string>} jobDefinitionChecksums - Checksums of job definitions
+     */
+    async #initializeJobConfigs(jobClasses, jobDefinitionChecksums) {
+        await Promise.all(
+            Object.entries(jobClasses).map(async([jobName, JobClass ]) => {
+                this.#log.d(`Initializing job config for ${jobName}`);
+                const currentChecksum = jobDefinitionChecksums[jobName];
+                const existingConfigOverride = await this.#jobConfigsCollection.findOne({ jobName });
+
+                if (!existingConfigOverride) {
+                    // Create new configuration for new job
+                    await this.#createDefaultJobConfig(jobName, currentChecksum, JobClass);
+                }
+                else if (existingConfigOverride.checksum !== currentChecksum) {
+                    // Reset configuration if job implementation has changed
+                    await this.#resetJobConfig(jobName, currentChecksum);
+                }
+            })
+        );
+    }
+
+    /**
+     * Creates a default configuration for a new job
+     * @private
+     * @param {string} jobName - Name of the job
+     * @param {string} checksum - Checksum of the job definition
+     * @param {Function} JobClass - Job class implementation
+     */
+    async #createDefaultJobConfig(jobName, checksum, JobClass) {
+        const instance = new JobClass(jobName);
+
+        await this.#jobConfigsCollection.insertOne({
+            jobName,
+            enabled: true,
+            checksum,
+            createdAt: new Date(),
+            defaultConfig: {
+                schedule: instance.getSchedule(),
+                retry: instance.getRetryConfig(),
+                priority: instance.getPriority(),
+                concurrency: instance.getConcurrency(),
+                lockLifetime: instance.getLockLifetime()
+            }
+        });
+        this.#log.d(`Created default configuration for new job: ${jobName}`);
+    }
+
+    /**
+     * Resets job configuration when implementation changes
+     * @private
+     * @param {string} jobName - Name of the job
+     * @param {string} newChecksum - New checksum of the job definition
+     */
+    async #resetJobConfig(jobName, newChecksum) {
+        this.#log.w(`Job ${jobName} implementation changed, resetting configuration`);
+        await this.#jobConfigsCollection.updateOne(
+            { jobName },
+            {
+                $set: {
+                    enabled: true,
+                    checksum: newChecksum,
+                    updatedAt: new Date()
+                },
+                $unset: {
+                    schedule: "",
+                    retry: "",
+                    runNow: ""
+                }
+            }
+        );
+    }
+
+    /**
+     * Applies configurations to jobs
+     * @private
+     * @param {Object.<string, Function>} jobClasses - Job class implementations
+     * @param {Object.<string, string>} jobDefinitionChecksums - Checksums of job definitions
+     */
+    async #applyJobConfigurations(jobClasses, jobDefinitionChecksums) {
+        // Load all existing configuration overrides
+        const configOverrides = await this.#jobConfigsCollection.find({}).toArray();
+        const configOverridesMap = new Map(configOverrides.map(conf => [conf.jobName, conf]));
+
+        // Create and configure jobs
+        await Promise.all(
+            Object.entries(jobClasses).map(async([jobName, JobClass]) => {
+                // Create the job with default settings
+                await this.#jobRunner.createJob(jobName, JobClass);
+
+                // Apply configuration override if valid
+                const configOverride = configOverridesMap.get(jobName);
+                if (configOverride && configOverride.checksum === jobDefinitionChecksums[jobName]) {
+                    await this.#applyConfig(configOverride);
+                    this.#log.d(`Applied configuration override for job: ${jobName}`);
+                }
+            })
+        );
     }
 
     /**
@@ -82,20 +279,6 @@ class JobManager {
         await this.#jobRunner.disableJob(jobName);
     }
 
-    /**
-     * Loads the job classes into the job runner
-     * @param {Object.<string, Function>} jobClasses Object containing job classes keyed by job name
-     * @returns {Promise<void>} A promise that resolves once the jobs are loaded
-     */
-    #loadJobs(jobClasses) {
-        return Promise.all(
-            Object.entries(jobClasses)
-                .map(([name, JobClass]) => {
-                    return this.#jobRunner.createJob(name, JobClass);
-                })
-        );
-    }
-
     /**
      * Closes the JobManager and cleans up resources
      * @returns {Promise<void>} A promise that resolves once cleanup is complete

jobServer/JobServer.js

@@ -109,8 +109,6 @@ class JobServer {
             // await this.#jobConfigsCollection.createIndex({ jobName: 1 }, /*{ unique: true }*/);
 
             this.#setupSignalHandlers();
-            // Watch for changes in job configurations
-            this.#watchJobConfigs();
 
             this.#log.i('Job process init successfully');
         }
@@ -176,40 +174,6 @@ class JobServer {
         });
     }
 
-    /**
-     * Watch for changes in job configurations
-     * @private
-     */
-    async #watchJobConfigs() {
-        const changeStream = this.#jobConfigsCollection.watch();
-
-        changeStream.on('change', async(change) => {
-            try {
-                if (change.operationType === 'update' || change.operationType === 'insert') {
-                    const jobName = change.fullDocument.jobName;
-                    const enabled = change.fullDocument.enabled;
-
-                    if (enabled) {
-                        await this.#jobManager.enableJob(jobName);
-                    }
-                    else {
-                        await this.#jobManager.disableJob(jobName);
-                    }
-
-                    this.#log.i(`Job ${jobName} ${enabled ? 'enabled' : 'disabled'}`);
-                }
-            }
-            catch (error) {
-                this.#log.e('Error processing job config change:', error);
-            }
-        });
-
-        changeStream.on('error', (error) => {
-            this.#log.e('Error in job configs change stream:', error);
-            // Implement reconnection logic here
-        });
-    }
-
     /**
      * Shuts down the job process.
      * @param {number} [exitCode=0] - The exit code to use when shutting down the process.

jobServer/JobUtils.js

@@ -1,6 +1,7 @@
 const Job = require('./Job');
 const {isValidCron} = require('cron-validator');
 const later = require('@breejs/later');
+const crypto = require('crypto');
 
 /**
  * Class responsible for validating job classes.
@@ -235,6 +236,19 @@ class JobUtils {
         return `${minutePart} ${hourPart} ${dayPart} ${monthPart} ${dayOfWeekPart}`;
     }
 
+    /**
+     * Calculates checksum for a job class
+     * @param {Function} JobClass The job class to calculate checksum for
+     * @returns {string} The calculated checksum
+     */
+    static calculateJobChecksum(JobClass) {
+        const jobString = JobClass.toString();
+        return crypto
+            .createHash('sha256')
+            .update(jobString)
+            .digest('hex');
+    }
+
 }
 
 module.exports = JobUtils;