misc: always initialize pg-boss

backend/src/ee/services/audit-log/audit-log-queue.ts

@@ -52,98 +52,96 @@ export const auditLogQueueServiceFactory = async ({
     }
   };
 
-  if (appCfg.USE_PG_QUEUE) {
-    await queueService.startPg<QueueName.AuditLog>(
-      QueueJobs.AuditLog,
-      async ([job]) => {
-        const { actor, event, ipAddress, projectId, userAgent, userAgentType } = job.data;
-        let { orgId } = job.data;
-        const MS_IN_DAY = 24 * 60 * 60 * 1000;
-        let project;
+  await queueService.startPg<QueueName.AuditLog>(
+    QueueJobs.AuditLog,
+    async ([job]) => {
+      const { actor, event, ipAddress, projectId, userAgent, userAgentType } = job.data;
+      let { orgId } = job.data;
+      const MS_IN_DAY = 24 * 60 * 60 * 1000;
+      let project;
 
-        if (!orgId) {
-          // it will never be undefined for both org and project id
-          // TODO(akhilmhdh): use caching here in dal to avoid db calls
-          project = await projectDAL.findById(projectId as string);
-          orgId = project.orgId;
-        }
-
-        const plan = await licenseService.getPlan(orgId);
-        if (plan.auditLogsRetentionDays === 0) {
-          // skip inserting if audit log retention is 0 meaning its not supported
-          return;
-        }
-
-        // For project actions, set TTL to project-level audit log retention config
-        // This condition ensures that the plan's audit log retention days cannot be bypassed
-        const ttlInDays =
-          project?.auditLogsRetentionDays && project.auditLogsRetentionDays < plan.auditLogsRetentionDays
-            ? project.auditLogsRetentionDays
-            : plan.auditLogsRetentionDays;
-
-        const ttl = ttlInDays * MS_IN_DAY;
-
-        const auditLog = await auditLogDAL.create({
-          actor: actor.type,
-          actorMetadata: actor.metadata,
-          userAgent,
-          projectId,
-          projectName: project?.name,
-          ipAddress,
-          orgId,
-          eventType: event.type,
-          expiresAt: new Date(Date.now() + ttl),
-          eventMetadata: event.metadata,
-          userAgentType
-        });
-
-        const logStreams = orgId ? await auditLogStreamDAL.find({ orgId }) : [];
-        await Promise.allSettled(
-          logStreams.map(
-            async ({
-              url,
-              encryptedHeadersTag,
-              encryptedHeadersIV,
-              encryptedHeadersKeyEncoding,
-              encryptedHeadersCiphertext
-            }) => {
-              const streamHeaders =
-                encryptedHeadersIV && encryptedHeadersCiphertext && encryptedHeadersTag
-                  ? (JSON.parse(
-                      infisicalSymmetricDecrypt({
-                        keyEncoding: encryptedHeadersKeyEncoding as SecretKeyEncoding,
-                        iv: encryptedHeadersIV,
-                        tag: encryptedHeadersTag,
-                        ciphertext: encryptedHeadersCiphertext
-                      })
-                    ) as LogStreamHeaders[])
-                  : [];
-
-              const headers: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
-
-              if (streamHeaders.length)
-                streamHeaders.forEach(({ key, value }) => {
-                  headers[key] = value;
-                });
-
-              return request.post(url, auditLog, {
-                headers,
-                // request timeout
-                timeout: AUDIT_LOG_STREAM_TIMEOUT,
-                // connection timeout
-                signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
-              });
-            }
-          )
-        );
-      },
-      {
-        batchSize: 1,
-        workerCount: 30,
-        pollingIntervalSeconds: 0.5
+      if (!orgId) {
+        // it will never be undefined for both org and project id
+        // TODO(akhilmhdh): use caching here in dal to avoid db calls
+        project = await projectDAL.findById(projectId as string);
+        orgId = project.orgId;
       }
-    );
-  }
+
+      const plan = await licenseService.getPlan(orgId);
+      if (plan.auditLogsRetentionDays === 0) {
+        // skip inserting if audit log retention is 0 meaning its not supported
+        return;
+      }
+
+      // For project actions, set TTL to project-level audit log retention config
+      // This condition ensures that the plan's audit log retention days cannot be bypassed
+      const ttlInDays =
+        project?.auditLogsRetentionDays && project.auditLogsRetentionDays < plan.auditLogsRetentionDays
+          ? project.auditLogsRetentionDays
+          : plan.auditLogsRetentionDays;
+
+      const ttl = ttlInDays * MS_IN_DAY;
+
+      const auditLog = await auditLogDAL.create({
+        actor: actor.type,
+        actorMetadata: actor.metadata,
+        userAgent,
+        projectId,
+        projectName: project?.name,
+        ipAddress,
+        orgId,
+        eventType: event.type,
+        expiresAt: new Date(Date.now() + ttl),
+        eventMetadata: event.metadata,
+        userAgentType
+      });
+
+      const logStreams = orgId ? await auditLogStreamDAL.find({ orgId }) : [];
+      await Promise.allSettled(
+        logStreams.map(
+          async ({
+            url,
+            encryptedHeadersTag,
+            encryptedHeadersIV,
+            encryptedHeadersKeyEncoding,
+            encryptedHeadersCiphertext
+          }) => {
+            const streamHeaders =
+              encryptedHeadersIV && encryptedHeadersCiphertext && encryptedHeadersTag
+                ? (JSON.parse(
+                    infisicalSymmetricDecrypt({
+                      keyEncoding: encryptedHeadersKeyEncoding as SecretKeyEncoding,
+                      iv: encryptedHeadersIV,
+                      tag: encryptedHeadersTag,
+                      ciphertext: encryptedHeadersCiphertext
+                    })
+                  ) as LogStreamHeaders[])
+                : [];
+
+            const headers: RawAxiosRequestHeaders = { "Content-Type": "application/json" };
+
+            if (streamHeaders.length)
+              streamHeaders.forEach(({ key, value }) => {
+                headers[key] = value;
+              });
+
+            return request.post(url, auditLog, {
+              headers,
+              // request timeout
+              timeout: AUDIT_LOG_STREAM_TIMEOUT,
+              // connection timeout
+              signal: AbortSignal.timeout(AUDIT_LOG_STREAM_TIMEOUT)
+            });
+          }
+        )
+      );
+    },
+    {
+      batchSize: 1,
+      workerCount: 30,
+      pollingIntervalSeconds: 0.5
+    }
+  );
 
   queueService.start(QueueName.AuditLog, async (job) => {
     const { actor, event, ipAddress, projectId, userAgent, userAgentType } = job.data;

backend/src/queue/queue-service.ts

@@ -8,7 +8,6 @@ import {
   TScanFullRepoEventPayload,
   TScanPushEventPayload
 } from "@app/ee/services/secret-scanning/secret-scanning-queue/secret-scanning-queue-types";
-import { getConfig } from "@app/lib/config/env";
 import { logger } from "@app/lib/logger";
 import {
   TFailedIntegrationSyncEmailsPayload,
@@ -209,15 +208,11 @@ export const queueServiceFactory = (redisUrl: string, dbConnectionUrl: string) =
   >;
 
   const initialize = async () => {
-    const appCfg = getConfig();
-    if (appCfg.USE_PG_QUEUE) {
-      logger.info("Initializing PG queue...");
-      await pgBoss.start();
+    await pgBoss.start();
 
-      pgBoss.on("error", (error) => {
-        logger.error(error, "pg-queue error");
-      });
-    }
+    pgBoss.on("error", (error) => {
+      logger.error(error, "pg-queue error");
+    });
   };
 
   const start = <T extends QueueName>(