From 0a1242db75d16a9ea3dbf3e699d707763622b603 Mon Sep 17 00:00:00 2001 From: Sheen Capadngan Date: Thu, 5 Dec 2024 15:52:17 +0800 Subject: [PATCH] misc: added pg queue init flag --- .../ee/services/audit-log/audit-log-queue.ts | 166 +++++++++--------- backend/src/lib/config/env.ts | 3 +- backend/src/queue/queue-service.ts | 13 +- 3 files changed, 95 insertions(+), 87 deletions(-) diff --git a/backend/src/ee/services/audit-log/audit-log-queue.ts b/backend/src/ee/services/audit-log/audit-log-queue.ts index a1c35bc40..e312c3886 100644 --- a/backend/src/ee/services/audit-log/audit-log-queue.ts +++ b/backend/src/ee/services/audit-log/audit-log-queue.ts @@ -37,7 +37,7 @@ export const auditLogQueueServiceFactory = async ({ const appCfg = getConfig(); const pushToLog = async (data: TCreateAuditLogDTO) => { - if (appCfg.USE_PG_QUEUE) { + if (appCfg.USE_PG_QUEUE && appCfg.SHOULD_INIT_PG_QUEUE) { await queueService.queuePg(QueueJobs.AuditLog, data, { retryLimit: 10, retryBackoff: true @@ -52,96 +52,98 @@ export const auditLogQueueServiceFactory = async ({ } }; - await queueService.startPg( - 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 (appCfg.SHOULD_INIT_PG_QUEUE) { + await queueService.startPg( + 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; - } + 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; - } + 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; + // 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 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 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 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" }; + const headers: RawAxiosRequestHeaders = { "Content-Type": "application/json" }; - if (streamHeaders.length) - streamHeaders.forEach(({ key, value }) => { - headers[key] = value; + 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) }); - - 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 - } - ); + } + ) + ); + }, + { + batchSize: 1, + workerCount: 30, + pollingIntervalSeconds: 0.5 + } + ); + } queueService.start(QueueName.AuditLog, async (job) => { const { actor, event, ipAddress, projectId, userAgent, userAgentType } = job.data; diff --git a/backend/src/lib/config/env.ts b/backend/src/lib/config/env.ts index 8a4cf07b3..66c5f3d98 100644 --- a/backend/src/lib/config/env.ts +++ b/backend/src/lib/config/env.ts @@ -180,7 +180,8 @@ const envSchema = z HSM_KEY_LABEL: zpStr(z.string().optional()), HSM_SLOT: z.coerce.number().optional().default(0), - USE_PG_QUEUE: zodStrBool.default("false") + USE_PG_QUEUE: zodStrBool.default("false"), + SHOULD_INIT_PG_QUEUE: zodStrBool.default("false") }) // To ensure that basic encryption is always possible. .refine( diff --git a/backend/src/queue/queue-service.ts b/backend/src/queue/queue-service.ts index f18562513..8479a249c 100644 --- a/backend/src/queue/queue-service.ts +++ b/backend/src/queue/queue-service.ts @@ -8,6 +8,7 @@ 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, @@ -208,11 +209,15 @@ export const queueServiceFactory = (redisUrl: string, dbConnectionUrl: string) = >; const initialize = async () => { - await pgBoss.start(); + const appCfg = getConfig(); + if (appCfg.SHOULD_INIT_PG_QUEUE) { + logger.info("Initializing pg-queue..."); + 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 = (