Use PG instead of bullMQ on the new certificates renew queue

backend/src/server/routes/index.ts

@@ -2339,7 +2339,7 @@ export const registerRoutes = async (
   await dailyReminderQueueService.startSecretReminderMigrationJob();
   await dailyExpiringPkiItemAlert.startSendingAlerts();
   await pkiSubscriberQueue.startDailyAutoRenewalJob();
-  await certificateV3Queue.startDailyAutoRenewalJob();
+  await certificateV3Queue.init();
   await kmsService.startService(hsmStatus);
   await microsoftTeamsService.start();
   await dynamicSecretQueueService.init();

backend/src/services/certificate-v3/certificate-v3-queue.ts

@@ -1,5 +1,6 @@
 /* eslint-disable no-await-in-loop */
 import { EventType, TAuditLogServiceFactory } from "@app/ee/services/audit-log/audit-log-types";
+import { getConfig } from "@app/lib/config/env";
 import { logger } from "@app/lib/logger";
 import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
 
@@ -21,130 +22,142 @@ export const certificateV3QueueServiceFactory = ({
   certificateV3Service,
   auditLogService
 }: TCertificateV3QueueServiceFactoryDep) => {
-  queueService.start(QueueName.CertificateV3AutoRenewal, async (job) => {
-    if (job.name === QueueJobs.CertificateV3DailyAutoRenewal) {
-      logger.info(`${QueueJobs.CertificateV3DailyAutoRenewal}: queue task started`);
+  const appCfg = getConfig();
 
-      const { QUEUE_BATCH_SIZE } = CERTIFICATE_RENEWAL_CONFIG;
-      let offset = 0;
-      let hasMore = true;
-      let totalCertificatesFound = 0;
-      let totalCertificatesRenewed = 0;
-
-      while (hasMore) {
-        const certificates = await certificateDAL.findCertificatesEligibleForRenewal({
-          limit: QUEUE_BATCH_SIZE,
-          offset
-        });
-
-        if (certificates.length === 0) {
-          hasMore = false;
-          break;
-        }
-
-        totalCertificatesFound += certificates.length;
-        logger.info(
-          `${QueueJobs.CertificateV3DailyAutoRenewal}: found ${certificates.length} certificates eligible for renewal (batch ${Math.floor(offset / QUEUE_BATCH_SIZE) + 1}, total found so far: ${totalCertificatesFound})`
-        );
-
-        for (const certificate of certificates) {
-          try {
-            if (certificate.renewBeforeDays) {
-              const { MIN_RENEW_BEFORE_DAYS, MAX_RENEW_BEFORE_DAYS } = CERTIFICATE_RENEWAL_CONFIG;
-              if (
-                certificate.renewBeforeDays < MIN_RENEW_BEFORE_DAYS ||
-                certificate.renewBeforeDays > MAX_RENEW_BEFORE_DAYS
-              ) {
-                // eslint-disable-next-line no-continue
-                continue;
-              }
-            }
-
-            await certificateV3Service.renewCertificate({
-              actor: ActorType.PLATFORM,
-              actorId: "",
-              actorAuthMethod: null,
-              actorOrgId: "",
-              certificateId: certificate.id,
-              internal: true
-            });
-
-            totalCertificatesRenewed += 1;
-
-            await auditLogService.createAuditLog({
-              projectId: certificate.projectId,
-              actor: {
-                type: ActorType.PLATFORM,
-                metadata: {}
-              },
-              event: {
-                type: EventType.AUTOMATED_RENEW_CERTIFICATE,
-                metadata: {
-                  certificateId: certificate.id,
-                  commonName: certificate.commonName || "",
-                  profileId: certificate.profileId!,
-                  renewBeforeDays: certificate.renewBeforeDays?.toString() || "",
-                  profileName: certificate.profileName || ""
-                }
-              }
-            });
-          } catch (error) {
-            const errorMessage = error instanceof Error ? error.message : String(error);
-            logger.error(error, `Failed to renew certificate ${certificate.id}: ${errorMessage}`);
-            await auditLogService.createAuditLog({
-              projectId: certificate.projectId,
-              actor: {
-                type: ActorType.PLATFORM,
-                metadata: {}
-              },
-              event: {
-                type: EventType.AUTOMATED_RENEW_CERTIFICATE_FAILED,
-                metadata: {
-                  certificateId: certificate.id,
-                  commonName: certificate.commonName || "",
-                  profileId: certificate.profileId || "",
-                  renewBeforeDays: certificate.renewBeforeDays?.toString() || "",
-                  profileName: certificate.profileName || "",
-                  error: errorMessage
-                }
-              }
-            });
-          }
-        }
-
-        offset += QUEUE_BATCH_SIZE;
-      }
-
-      logger.info(
-        `${QueueJobs.CertificateV3DailyAutoRenewal}: queue task completed. Renewed ${totalCertificatesRenewed} certificates out of ${totalCertificatesFound}`
-      );
+  const init = async () => {
+    if (appCfg.isSecondaryInstance) {
+      return;
     }
-  });
-
-  const startDailyAutoRenewalJob = async () => {
-    const { DAILY_CRON_SCHEDULE, QUEUE_START_DELAY_MS } = CERTIFICATE_RENEWAL_CONFIG;
 
     await queueService.stopRepeatableJob(
       QueueName.CertificateV3AutoRenewal,
       QueueJobs.CertificateV3DailyAutoRenewal,
-      { pattern: DAILY_CRON_SCHEDULE, utc: true },
+      { pattern: CERTIFICATE_RENEWAL_CONFIG.DAILY_CRON_SCHEDULE, utc: true },
       QueueName.CertificateV3AutoRenewal
     );
 
-    await queueService.queue(QueueName.CertificateV3AutoRenewal, QueueJobs.CertificateV3DailyAutoRenewal, undefined, {
-      delay: QUEUE_START_DELAY_MS,
-      jobId: QueueName.CertificateV3AutoRenewal,
-      repeat: { pattern: DAILY_CRON_SCHEDULE, utc: true }
-    });
+    await queueService.startPg<QueueName.CertificateV3AutoRenewal>(
+      QueueJobs.CertificateV3DailyAutoRenewal,
+      async () => {
+        try {
+          logger.info(`${QueueJobs.CertificateV3DailyAutoRenewal}: queue task started`);
+
+          const { QUEUE_BATCH_SIZE } = CERTIFICATE_RENEWAL_CONFIG;
+          let offset = 0;
+          let hasMore = true;
+          let totalCertificatesFound = 0;
+          let totalCertificatesRenewed = 0;
+
+          while (hasMore) {
+            const certificates = await certificateDAL.findCertificatesEligibleForRenewal({
+              limit: QUEUE_BATCH_SIZE,
+              offset
+            });
+
+            if (certificates.length === 0) {
+              hasMore = false;
+              break;
+            }
+
+            totalCertificatesFound += certificates.length;
+            logger.info(
+              `${QueueJobs.CertificateV3DailyAutoRenewal}: found ${certificates.length} certificates eligible for renewal (batch ${Math.floor(offset / QUEUE_BATCH_SIZE) + 1}, total found so far: ${totalCertificatesFound})`
+            );
+
+            for (const certificate of certificates) {
+              try {
+                if (certificate.renewBeforeDays) {
+                  const { MIN_RENEW_BEFORE_DAYS, MAX_RENEW_BEFORE_DAYS } = CERTIFICATE_RENEWAL_CONFIG;
+                  if (
+                    certificate.renewBeforeDays < MIN_RENEW_BEFORE_DAYS ||
+                    certificate.renewBeforeDays > MAX_RENEW_BEFORE_DAYS
+                  ) {
+                    // eslint-disable-next-line no-continue
+                    continue;
+                  }
+                }
+
+                await certificateV3Service.renewCertificate({
+                  actor: ActorType.PLATFORM,
+                  actorId: "",
+                  actorAuthMethod: null,
+                  actorOrgId: "",
+                  certificateId: certificate.id,
+                  internal: true
+                });
+
+                totalCertificatesRenewed += 1;
+
+                await auditLogService.createAuditLog({
+                  projectId: certificate.projectId,
+                  actor: {
+                    type: ActorType.PLATFORM,
+                    metadata: {}
+                  },
+                  event: {
+                    type: EventType.AUTOMATED_RENEW_CERTIFICATE,
+                    metadata: {
+                      certificateId: certificate.id,
+                      commonName: certificate.commonName || "",
+                      profileId: certificate.profileId!,
+                      renewBeforeDays: certificate.renewBeforeDays?.toString() || "",
+                      profileName: certificate.profileName || ""
+                    }
+                  }
+                });
+              } catch (error) {
+                const errorMessage = error instanceof Error ? error.message : String(error);
+                logger.error(error, `Failed to renew certificate ${certificate.id}: ${errorMessage}`);
+                await auditLogService.createAuditLog({
+                  projectId: certificate.projectId,
+                  actor: {
+                    type: ActorType.PLATFORM,
+                    metadata: {}
+                  },
+                  event: {
+                    type: EventType.AUTOMATED_RENEW_CERTIFICATE_FAILED,
+                    metadata: {
+                      certificateId: certificate.id,
+                      commonName: certificate.commonName || "",
+                      profileId: certificate.profileId || "",
+                      renewBeforeDays: certificate.renewBeforeDays?.toString() || "",
+                      profileName: certificate.profileName || "",
+                      error: errorMessage
+                    }
+                  }
+                });
+              }
+            }
+
+            offset += QUEUE_BATCH_SIZE;
+          }
+
+          logger.info(
+            `${QueueJobs.CertificateV3DailyAutoRenewal}: queue task completed. Renewed ${totalCertificatesRenewed} certificates out of ${totalCertificatesFound}`
+          );
+        } catch (error) {
+          logger.error(error, `${QueueJobs.CertificateV3DailyAutoRenewal}: certificate renewal failed`);
+          throw error;
+        }
+      },
+      {
+        batchSize: 1,
+        workerCount: 1,
+        pollingIntervalSeconds: 60
+      }
+    );
+
+    await queueService.schedulePg(
+      QueueJobs.CertificateV3DailyAutoRenewal,
+      CERTIFICATE_RENEWAL_CONFIG.DAILY_CRON_SCHEDULE,
+      undefined,
+      { tz: "UTC" }
+    );
   };
 
-  queueService.listen(QueueName.CertificateV3AutoRenewal, "failed", (_, err) => {
-    logger.error(err, `${QueueName.CertificateV3AutoRenewal}: failed`);
-  });
-
   return {
-    startDailyAutoRenewalJob
+    init
   };
 };
 
-export type TCertificateV3QueueFactory = ReturnType<typeof certificateV3QueueServiceFactory>;
+export type TCertificateV3QueueServiceFactory = ReturnType<typeof certificateV3QueueServiceFactory>;