feat(api): process maddo pgboss jobs in maddo server

Co-authored-by: Aurélie Crouillebois <[email protected]> Co-authored-by: Vincent Hardouin <[email protected]>
1024pix · Feb 28, 2025 · c0f272e · c0f272e
1 parent e3cdafb
commit c0f272e
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 17 deletions.
diff --git a/api/index.js b/api/index.js
@@ -64,7 +64,7 @@ process.on('SIGINT', () => {
   try {
     await start();
     if (config.infra.startJobInWebProcess) {
-      registerJobs({ jobGroup: JobGroup.DEFAULT });
+      registerJobs({ jobGroups: [JobGroup.DEFAULT, JobGroup.FAST] });
     }
   } catch (error) {
     logger.error(error);

diff --git a/api/index.maddo.js b/api/index.maddo.js
@@ -2,11 +2,13 @@ import 'dotenv/config';
 
 import { databaseConnections } from './db/database-connections.js';
 import { createMaddoServer } from './server.maddo.js';
-import { schema as configSchema } from './src/shared/config.js';
+import { JobGroup } from './src/shared/application/jobs/job-controller.js';
+import { config, schema as configSchema } from './src/shared/config.js';
 import { quitAllStorages } from './src/shared/infrastructure/key-value-storages/index.js';
 import { logger } from './src/shared/infrastructure/utils/logger.js';
 import { redisMonitor } from './src/shared/infrastructure/utils/redis-monitor.js';
 import { validateEnvironmentVariables } from './src/shared/infrastructure/validate-environment-variables.js';
+import { registerJobs } from './worker.js';
 
 validateEnvironmentVariables(configSchema);
 
@@ -45,6 +47,9 @@ process.on('SIGINT', () => {
 (async () => {
   try {
     await start();
+    if (config.infra.startJobInWebProcess) {
+      registerJobs({ jobGroups: [JobGroup.MADDO] });
+    }
   } catch (error) {
     logger.error(error);
     throw error;

diff --git a/api/tests/unit/worker_test.js b/api/tests/unit/worker_test.js
@@ -29,7 +29,7 @@ describe('#registerJobs', function () {
   it('should register UserAnonymizedEventLoggingJob', async function () {
     // when
     await registerJobs({
-      jobGroup: JobGroup.DEFAULT,
+      jobGroups: [JobGroup.DEFAULT],
       dependencies: {
         startPgBoss: startPgBossStub,
         createJobQueue: createJobQueueStub,
@@ -50,7 +50,7 @@ describe('#registerJobs', function () {
       .stub(UserAnonymizedEventLoggingJobController.prototype, 'legacyName')
       .get(() => 'legyNameForUserAnonymizedEventLoggingJobController');
     await registerJobs({
-      jobGroup: JobGroup.DEFAULT,
+      jobGroups: [JobGroup.DEFAULT],
       dependencies: {
         startPgBoss: startPgBossStub,
         createJobQueue: createJobQueueStub,
@@ -71,7 +71,7 @@ describe('#registerJobs', function () {
 
     // when
     await registerJobs({
-      jobGroup: JobGroup.DEFAULT,
+      jobGroups: [JobGroup.DEFAULT],
       dependencies: {
         startPgBoss: startPgBossStub,
         createJobQueue: createJobQueueStub,
@@ -92,7 +92,7 @@ describe('#registerJobs', function () {
 
     // when
     await registerJobs({
-      jobGroup: JobGroup.DEFAULT,
+      jobGroups: [JobGroup.DEFAULT],
       dependencies: {
         startPgBoss: startPgBossStub,
         createJobQueue: createJobQueueStub,
@@ -106,9 +106,24 @@ describe('#registerJobs', function () {
     );
   });
 
+  it('should throws an error when no groups is invalid', async function () {
+    // given
+    const error = await catchErr(registerJobs)({
+      dependencies: {
+        startPgBoss: startPgBossStub,
+        createJobQueue: createJobQueueStub,
+      },
+    });
+
+    // then
+    expect(error).to.be.instanceOf(Error);
+    expect(error.message).to.equal('Job groups are mandatory');
+  });
+
   it('should throws an error when group is invalid', async function () {
     // given
     const error = await catchErr(registerJobs)({
+      jobGroups: ['pouet'],
       dependencies: {
         startPgBoss: startPgBossStub,
         createJobQueue: createJobQueueStub,
@@ -126,7 +141,7 @@ describe('#registerJobs', function () {
       sinon.stub(config.features.scheduleComputeOrganizationLearnersCertificability, 'cron').value('0 21 * * *');
 
       await registerJobs({
-        jobGroup: JobGroup.DEFAULT,
+        jobGroups: [JobGroup.DEFAULT],
         dependencies: {
           startPgBoss: startPgBossStub,
           createJobQueue: createJobQueueStub,
@@ -150,7 +165,7 @@ describe('#registerJobs', function () {
       sinon.stub(config.features.scheduleComputeOrganizationLearnersCertificability, 'cron').value('0 21 * * *');
 
       await registerJobs({
-        jobGroup: JobGroup.DEFAULT,
+        jobGroups: [JobGroup.DEFAULT],
         dependencies: {
           startPgBoss: startPgBossStub,
           createJobQueue: createJobQueueStub,
@@ -170,7 +185,7 @@ describe('#registerJobs', function () {
         sinon.stub(config.pgBoss, 'exportSenderJobEnabled').value(false);
 
         await registerJobs({
-          jobGroup: JobGroup.DEFAULT,
+          jobGroups: [JobGroup.DEFAULT],
           dependencies: {
             startPgBoss: startPgBossStub,
             createJobQueue: createJobQueueStub,

diff --git a/api/worker.js b/api/worker.js
@@ -19,7 +19,6 @@ import { child } from './src/shared/infrastructure/utils/logger.js';
 
 const logger = child('worker', { event: 'worker' });
 
-const isJobInWebProcess = process.env.START_JOB_IN_WEB_PROCESS === 'true';
 const workerDirPath = dirname(fileURLToPath(import.meta.url));
 
 const metrics = new Metrics({ config });
@@ -57,15 +56,19 @@ function createJobQueue(pgBoss) {
   return jobQueue;
 }
 
+function checkJobGroups(jobGroups) {
+  if (!jobGroups) throw new Error('Job groups are mandatory');
+  jobGroups.forEach((jobGroup) => checkJobGroup(jobGroup));
+}
+
 function checkJobGroup(jobGroup) {
   if (!Object.values(JobGroup).includes(jobGroup)) {
     throw new Error(`Job group invalid, allowed Job groups are [${Object.values(JobGroup)}]`);
   }
-  logger.info(`Job group "${jobGroup}"`);
 }
 
-export async function registerJobs({ jobGroup, dependencies = { startPgBoss, createJobQueue } }) {
-  checkJobGroup(jobGroup);
+export async function registerJobs({ jobGroups, dependencies = { startPgBoss, createJobQueue } }) {
+  checkJobGroups(jobGroups);
 
   const pgBoss = await dependencies.startPgBoss();
 
@@ -88,7 +91,7 @@ export async function registerJobs({ jobGroup, dependencies = { startPgBoss, cre
   for (const [moduleName, ModuleClass] of Object.entries(jobModules)) {
     const job = new ModuleClass();
 
-    if (!isJobInWebProcess && job.jobGroup !== jobGroup) continue;
+    if (!jobGroups.includes(job.jobGroup)) continue;
 
     if (job.isJobEnabled) {
       logger.info(`Job "${job.jobName}" registered from module "${moduleName}."`);
@@ -127,15 +130,15 @@ export async function registerJobs({ jobGroup, dependencies = { startPgBoss, cre
     }
   }
 
-  logger.info(`${jobRegisteredCount} jobs registered for group "${jobGroup}".`);
-  logger.info(`${cronJobCount} cron jobs scheduled for group "${jobGroup}".`);
+  logger.info(`${jobRegisteredCount} jobs registered for groups "${jobGroups}".`);
+  logger.info(`${cronJobCount} cron jobs scheduled for groups "${jobGroups}".`);
 }
 
 const isRunningFromCli = import.meta.filename === process.argv[1];
 
 async function main() {
   const jobGroup = process.argv[2] ? JobGroup[process.argv[2]?.toUpperCase()] : JobGroup.DEFAULT;
-  await registerJobs({ jobGroup });
+  await registerJobs({ jobGroups: [jobGroup] });
   process.on('SIGINT', async () => {
     await quitAllStorages();
     await metrics.clearMetrics();