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 a9e322962..a1c35bc40 100644 --- a/backend/src/ee/services/audit-log/audit-log-queue.ts +++ b/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( - 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( + 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; diff --git a/backend/src/queue/queue-service.ts b/backend/src/queue/queue-service.ts index 3205c8c94..f18562513 100644 --- a/backend/src/queue/queue-service.ts +++ b/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 = (