misc: updated processes to run after db migration

backend/src/ee/services/secret-rotation-v2/secret-rotation-v2-queue.ts

@@ -42,6 +42,7 @@ export const secretRotationV2QueueServiceFactory = async ({
   smtpService,
   notificationService
 }: TSecretRotationV2QueueServiceFactoryDep) => {
+  const init = async () => {
     const appCfg = getConfig();
 
     if (appCfg.isRotationDevelopmentMode) {
@@ -206,4 +207,7 @@ export const secretRotationV2QueueServiceFactory = async ({
       undefined,
       { tz: "UTC" }
     );
+  };
+
+  return { init };
 };

backend/src/ee/services/secret-scanning-v2/secret-scanning-v2-queue.ts

@@ -141,6 +141,61 @@ export const secretScanningV2QueueServiceFactory = async ({
     }
   };
 
+  const queueResourceDiffScan = async ({
+    payload,
+    dataSourceId,
+    dataSourceType
+  }: Pick<TQueueSecretScanningResourceDiffScan, "payload" | "dataSourceId" | "dataSourceType">) => {
+    const factory = SECRET_SCANNING_FACTORY_MAP[dataSourceType as SecretScanningDataSource]({
+      kmsService,
+      appConnectionDAL
+    });
+
+    const resourcePayload = factory.getDiffScanResourcePayload(payload);
+
+    try {
+      const { resourceId, scanId } = await secretScanningV2DAL.resources.transaction(async (tx) => {
+        const [resource] = await secretScanningV2DAL.resources.upsert(
+          [
+            {
+              ...resourcePayload,
+              dataSourceId
+            }
+          ],
+          ["externalId", "dataSourceId"],
+          tx
+        );
+
+        const scan = await secretScanningV2DAL.scans.create(
+          {
+            resourceId: resource.id,
+            type: SecretScanningScanType.DiffScan
+          },
+          tx
+        );
+
+        return {
+          resourceId: resource.id,
+          scanId: scan.id
+        };
+      });
+
+      await queueService.queuePg(QueueJobs.SecretScanningV2DiffScan, {
+        payload,
+        dataSourceId,
+        dataSourceType,
+        scanId,
+        resourceId
+      });
+    } catch (error) {
+      logger.error(
+        error,
+        `secretScanningV2Queue: Failed to queue diff scan [dataSourceId=${dataSourceId}] [resourceExternalId=${resourcePayload.externalId}]`
+      );
+    }
+  };
+
+  const init = async () => {
     await queueService.startPg<QueueName.SecretScanningV2>(
       QueueJobs.SecretScanningV2FullScan,
       async ([job]) => {
@@ -337,60 +392,6 @@ export const secretScanningV2QueueServiceFactory = async ({
       }
     );
 
-  const queueResourceDiffScan = async ({
-    payload,
-    dataSourceId,
-    dataSourceType
-  }: Pick<TQueueSecretScanningResourceDiffScan, "payload" | "dataSourceId" | "dataSourceType">) => {
-    const factory = SECRET_SCANNING_FACTORY_MAP[dataSourceType as SecretScanningDataSource]({
-      kmsService,
-      appConnectionDAL
-    });
-
-    const resourcePayload = factory.getDiffScanResourcePayload(payload);
-
-    try {
-      const { resourceId, scanId } = await secretScanningV2DAL.resources.transaction(async (tx) => {
-        const [resource] = await secretScanningV2DAL.resources.upsert(
-          [
-            {
-              ...resourcePayload,
-              dataSourceId
-            }
-          ],
-          ["externalId", "dataSourceId"],
-          tx
-        );
-
-        const scan = await secretScanningV2DAL.scans.create(
-          {
-            resourceId: resource.id,
-            type: SecretScanningScanType.DiffScan
-          },
-          tx
-        );
-
-        return {
-          resourceId: resource.id,
-          scanId: scan.id
-        };
-      });
-
-      await queueService.queuePg(QueueJobs.SecretScanningV2DiffScan, {
-        payload,
-        dataSourceId,
-        dataSourceType,
-        scanId,
-        resourceId
-      });
-    } catch (error) {
-      logger.error(
-        error,
-        `secretScanningV2Queue: Failed to queue diff scan [dataSourceId=${dataSourceId}] [resourceExternalId=${resourcePayload.externalId}]`
-      );
-    }
-  };
-
     await queueService.startPg<QueueName.SecretScanningV2>(
       QueueJobs.SecretScanningV2DiffScan,
       async ([job]) => {
@@ -666,9 +667,11 @@ export const secretScanningV2QueueServiceFactory = async ({
         pollingIntervalSeconds: 1
       }
     );
+  };
 
   return {
     queueDataSourceFullScan,
-    queueResourceDiffScan
+    queueResourceDiffScan,
+    init
   };
 };

backend/src/main.ts

@@ -72,7 +72,7 @@ const run = async () => {
   const keyStore = keyStoreFactory(envConfig, keyValueStoreDAL);
   const redis = buildRedisFromConfig(envConfig);
 
-  const server = await main({
+  const { server, completeServerInitialization } = await main({
     db,
     auditLogDb,
     superAdminDAL,
@@ -140,7 +140,18 @@ const run = async () => {
     }
   });
 
-  logger.info("Migrations complete. Marking server as READY...");
+  logger.info("Migrations complete. Completing server initialization...");
+
+  try {
+    await completeServerInitialization();
+  } catch (error) {
+    logger.error(error, "Failed to complete server initialization");
+    await server.close();
+    await queue.shutdown();
+    process.exit(1);
+  }
+
+  logger.info("Server initialization complete. Marking server as READY...");
 
   markServerReady();
 

backend/src/server/app.ts

@@ -221,7 +221,7 @@ export const main = async ({
       };
     });
 
-    await server.register(registerRoutes, {
+    const completeServerInitialization = await registerRoutes(server, {
       smtp,
       queue,
       db,
@@ -240,7 +240,7 @@ export const main = async ({
 
     await server.ready();
     server.swagger();
-    return server;
+    return { server, completeServerInitialization };
   } catch (err) {
     server.log.error(err);
     await queue.shutdown();

backend/src/server/routes/index.ts

@@ -2206,7 +2206,7 @@ export const registerRoutes = async (
     internalCaFns
   });
 
-  await secretRotationV2QueueServiceFactory({
+  const secretRotationV2Queue = await secretRotationV2QueueServiceFactory({
     secretRotationV2Service,
     secretRotationV2DAL,
     queueService,
@@ -2300,11 +2300,12 @@ export const registerRoutes = async (
   // setup the communication with license key server
   await licenseService.init();
 
+  const completeServerInitialization = async () => {
+    await superAdminService.initServerCfg();
+
     // If FIPS is enabled, we check to ensure that the users license includes FIPS mode.
     crypto.verifyFipsLicense(licenseService);
 
-  await superAdminService.initServerCfg();
-
     // Start HSM service if it's configured/enabled.
     await hsmService.startService();
 
@@ -2343,8 +2344,44 @@ export const registerRoutes = async (
     await kmsService.startService(hsmStatus);
     await microsoftTeamsService.start();
     await dynamicSecretQueueService.init();
+    await secretScanningV2Queue.init();
+    await secretRotationV2Queue.init();
+    await notificationQueue.init();
     await eventBusService.init();
 
+    const cronJobs: CronJob[] = [];
+    if (appCfg.isProductionMode) {
+      const rateLimitSyncJob = await rateLimitService.initializeBackgroundSync();
+      if (rateLimitSyncJob) {
+        cronJobs.push(rateLimitSyncJob);
+      }
+      const licenseSyncJob = await licenseService.initializeBackgroundSync();
+      if (licenseSyncJob) {
+        cronJobs.push(licenseSyncJob);
+      }
+
+      const microsoftTeamsSyncJob = await microsoftTeamsService.initializeBackgroundSync();
+      if (microsoftTeamsSyncJob) {
+        cronJobs.push(microsoftTeamsSyncJob);
+      }
+
+      const adminIntegrationsSyncJob = await superAdminService.initializeAdminIntegrationConfigSync();
+      if (adminIntegrationsSyncJob) {
+        cronJobs.push(adminIntegrationsSyncJob);
+      }
+    }
+
+    const configSyncJob = await superAdminService.initializeEnvConfigSync();
+    if (configSyncJob) {
+      cronJobs.push(configSyncJob);
+    }
+
+    const oauthConfigSyncJob = await initializeOauthConfigSync();
+    if (oauthConfigSyncJob) {
+      cronJobs.push(oauthConfigSyncJob);
+    }
+  };
+
   // inject all services
   server.decorate<FastifyZodProvider["services"]>("services", {
     login: loginService,
@@ -2473,38 +2510,6 @@ export const registerRoutes = async (
     convertor: convertorService
   });
 
-  const cronJobs: CronJob[] = [];
-  if (appCfg.isProductionMode) {
-    const rateLimitSyncJob = await rateLimitService.initializeBackgroundSync();
-    if (rateLimitSyncJob) {
-      cronJobs.push(rateLimitSyncJob);
-    }
-    const licenseSyncJob = await licenseService.initializeBackgroundSync();
-    if (licenseSyncJob) {
-      cronJobs.push(licenseSyncJob);
-    }
-
-    const microsoftTeamsSyncJob = await microsoftTeamsService.initializeBackgroundSync();
-    if (microsoftTeamsSyncJob) {
-      cronJobs.push(microsoftTeamsSyncJob);
-    }
-
-    const adminIntegrationsSyncJob = await superAdminService.initializeAdminIntegrationConfigSync();
-    if (adminIntegrationsSyncJob) {
-      cronJobs.push(adminIntegrationsSyncJob);
-    }
-  }
-
-  const configSyncJob = await superAdminService.initializeEnvConfigSync();
-  if (configSyncJob) {
-    cronJobs.push(configSyncJob);
-  }
-
-  const oauthConfigSyncJob = await initializeOauthConfigSync();
-  if (oauthConfigSyncJob) {
-    cronJobs.push(oauthConfigSyncJob);
-  }
-
   server.decorate<FastifyZodProvider["store"]>("store", {
     user: userDAL,
     kmipClient: kmipClientDAL
@@ -2593,9 +2598,10 @@ export const registerRoutes = async (
   await server.register(registerV4Routes, { prefix: "/api/v4" });
 
   server.addHook("onClose", async () => {
-    cronJobs.forEach((job) => job.stop());
     await telemetryService.flushAll();
     await eventBusService.close();
     sseService.close();
   });
+
+  return completeServerInitialization;
 };

backend/src/services/notification/notification-queue.ts

@@ -10,6 +10,7 @@ type TNotificationQueueServiceFactoryDep = {
 
 export type TNotificationQueueServiceFactory = {
   pushUserNotifications: (data: TCreateUserNotificationDTO[]) => Promise<void>;
+  init: () => Promise<void>;
 };
 
 export const notificationQueueServiceFactory = async ({
@@ -20,6 +21,7 @@ export const notificationQueueServiceFactory = async ({
     await queueService.queuePg(QueueJobs.UserNotification, { notifications: data });
   };
 
+  const init = async () => {
     await queueService.startPg(
       QueueJobs.UserNotification,
       async ([job]) => {
@@ -32,8 +34,10 @@ export const notificationQueueServiceFactory = async ({
         pollingIntervalSeconds: 1
       }
     );
+  };
 
   return {
-    pushUserNotifications
+    pushUserNotifications,
+    init
   };
 };