Merge branch 'main' into feat/point-in-time-revamp

backend/src/queue/queue-service.ts

@@ -11,9 +11,15 @@ import {
   TScanFullRepoEventPayload,
   TScanPushEventPayload
 } from "@app/ee/services/secret-scanning/secret-scanning-queue/secret-scanning-queue-types";
+import {
+  TQueueSecretScanningDataSourceFullScan,
+  TQueueSecretScanningResourceDiffScan,
+  TQueueSecretScanningSendNotification
+} from "@app/ee/services/secret-scanning-v2/secret-scanning-v2-types";
 import { getConfig } from "@app/lib/config/env";
 import { buildRedisFromConfig, TRedisConfigKeys } from "@app/lib/config/redis";
 import { logger } from "@app/lib/logger";
+import { QueueWorkerProfile } from "@app/lib/types";
 import { CaType } from "@app/services/certificate-authority/certificate-authority-enums";
 import {
   TFailedIntegrationSyncEmailsPayload,
@@ -55,7 +61,8 @@ export enum QueueName {
   AppConnectionSecretSync = "app-connection-secret-sync",
   SecretRotationV2 = "secret-rotation-v2",
   FolderTreeCheckpoint = "folder-tree-checkpoint",
-  InvalidateCache = "invalidate-cache"
+  InvalidateCache = "invalidate-cache",
+  SecretScanningV2 = "secret-scanning-v2"
 }
 
 export enum QueueJobs {
@@ -90,6 +97,9 @@ export enum QueueJobs {
   SecretRotationV2SendNotification = "secret-rotation-v2-send-notification",
   CreateFolderTreeCheckpoint = "create-folder-tree-checkpoint",
   InvalidateCache = "invalidate-cache",
+  SecretScanningV2FullScan = "secret-scanning-v2-full-scan",
+  SecretScanningV2DiffScan = "secret-scanning-v2-diff-scan",
+  SecretScanningV2SendNotification = "secret-scanning-v2-notification",
   CaOrderCertificateForSubscriber = "ca-order-certificate-for-subscriber",
   PkiSubscriberDailyAutoRenewal = "pki-subscriber-daily-auto-renewal"
 }
@@ -258,6 +268,19 @@ export type TQueueJobTypes = {
       };
     };
   };
+  [QueueName.SecretScanningV2]:
+    | {
+        name: QueueJobs.SecretScanningV2FullScan;
+        payload: TQueueSecretScanningDataSourceFullScan;
+      }
+    | {
+        name: QueueJobs.SecretScanningV2DiffScan;
+        payload: TQueueSecretScanningResourceDiffScan;
+      }
+    | {
+        name: QueueJobs.SecretScanningV2SendNotification;
+        payload: TQueueSecretScanningSendNotification;
+      };
   [QueueName.CaLifecycle]: {
     name: QueueJobs.CaOrderCertificateForSubscriber;
     payload: {
@@ -271,6 +294,37 @@ export type TQueueJobTypes = {
   };
 };
 
+const SECRET_SCANNING_JOBS = [
+  QueueJobs.SecretScanningV2FullScan,
+  QueueJobs.SecretScanningV2DiffScan,
+  QueueJobs.SecretScanningV2SendNotification,
+  QueueJobs.SecretScan
+];
+
+const NON_STANDARD_JOBS = [...SECRET_SCANNING_JOBS];
+
+const SECRET_SCANNING_QUEUES = [
+  QueueName.SecretScanningV2,
+  QueueName.SecretFullRepoScan,
+  QueueName.SecretPushEventScan
+];
+
+const NON_STANDARD_QUEUES = [...SECRET_SCANNING_QUEUES];
+
+const isQueueEnabled = (name: QueueName) => {
+  const appCfg = getConfig();
+  switch (appCfg.QUEUE_WORKER_PROFILE) {
+    case QueueWorkerProfile.Standard:
+      return !NON_STANDARD_QUEUES.includes(name);
+    case QueueWorkerProfile.SecretScanning:
+      return SECRET_SCANNING_QUEUES.includes(name);
+    case QueueWorkerProfile.All:
+    default:
+      // allow all
+      return true;
+  }
+};
+
 export type TQueueServiceFactory = ReturnType<typeof queueServiceFactory>;
 export const queueServiceFactory = (
   redisCfg: TRedisConfigKeys,
@@ -327,7 +381,7 @@ export const queueServiceFactory = (
     });
 
     const appCfg = getConfig();
-    if (appCfg.QUEUE_WORKERS_ENABLED) {
+    if (appCfg.QUEUE_WORKERS_ENABLED && isQueueEnabled(name)) {
       workerContainer[name] = new Worker<TQueueJobTypes[T]["payload"], void, TQueueJobTypes[T]["name"]>(name, jobFn, {
         ...queueSettings,
         connection
@@ -346,6 +400,30 @@ export const queueServiceFactory = (
       throw new Error(`${jobName} queue is already initialized`);
     }
 
+    const appCfg = getConfig();
+
+    if (!appCfg.QUEUE_WORKERS_ENABLED) return;
+
+    switch (appCfg.QUEUE_WORKER_PROFILE) {
+      case QueueWorkerProfile.Standard:
+        if (NON_STANDARD_JOBS.includes(jobName)) {
+          // only process standard jobs
+          return;
+        }
+
+        break;
+      case QueueWorkerProfile.SecretScanning:
+        if (!SECRET_SCANNING_JOBS.includes(jobName)) {
+          // only process secret scanning jobs
+          return;
+        }
+
+        break;
+      case QueueWorkerProfile.All:
+      default:
+      // allow all
+    }
+
     await pgBoss.createQueue(jobName);
     queueContainerPg[jobName] = true;
 
@@ -365,7 +443,7 @@ export const queueServiceFactory = (
     listener: WorkerListener<TQueueJobTypes[T]["payload"], void, TQueueJobTypes[T]["name"]>[U]
   ) => {
     const appCfg = getConfig();
-    if (!appCfg.QUEUE_WORKERS_ENABLED) {
+    if (!appCfg.QUEUE_WORKERS_ENABLED || !isQueueEnabled(name)) {
       return;
     }