feat: added audit log prune, resolved env update and pino file transport on prod

backend/e2e-test/mocks/queue.ts

@@ -16,9 +16,11 @@ export const mockQueue = (): TQueueServiceFactory => {
       queues[name] = jobFn;
       workers[name] = jobFn;
     },
-    listen: async (name, event) => {
+    listen: (name, event) => {
       events[name] = event;
     },
+    clearQueue: async () => {},
+    stopJobById: async () => {},
     stopRepeatableJobByJobId: async () => true
   };
 };

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

@@ -2,6 +2,7 @@ import { Knex } from "knex";
 
 import { TDbClient } from "@app/db";
 import { TableName } from "@app/db/schemas";
+import { DatabaseError } from "@app/lib/errors";
 import { ormify, stripUndefinedInWhere } from "@app/lib/knex";
 
 export type TAuditLogDALFactory = ReturnType<typeof auditLogDALFactory>;
@@ -25,27 +26,42 @@ export const auditLogDALFactory = (db: TDbClient) => {
     { orgId, projectId, userAgentType, startDate, endDate, limit = 20, offset = 0, actor, eventType }: TFindQuery,
     tx?: Knex
   ) => {
-    const sqlQuery = (tx || db)(TableName.AuditLog)
+    try {
-      .where(
+      const sqlQuery = (tx || db)(TableName.AuditLog)
-        stripUndefinedInWhere({
+        .where(
-          projectId,
+          stripUndefinedInWhere({
-          orgId,
+            projectId,
-          eventType,
+            orgId,
-          actor,
+            eventType,
-          userAgentType
+            actor,
-        })
+            userAgentType
-      )
+          })
-      .limit(limit)
+        )
-      .offset(offset);
+        .limit(limit)
-    if (startDate) {
+        .offset(offset);
-      void sqlQuery.where("createdAt", ">=", startDate);
+      if (startDate) {
+        void sqlQuery.where("createdAt", ">=", startDate);
+      }
+      if (endDate) {
+        void sqlQuery.where("createdAt", "<=", endDate);
+      }
+      const docs = await sqlQuery;
+      return docs;
+    } catch (error) {
+      throw new DatabaseError({ error });
     }
-    if (endDate) {
-      void sqlQuery.where("createdAt", "<=", endDate);
-    }
-    const docs = await sqlQuery;
-    return docs;
   };
 
-  return { ...auditLogOrm, find };
+  // delete all audit log that have expired
+  const pruneAuditLog = async (tx?: Knex) => {
+    try {
+      const today = new Date();
+      const docs = await (tx || db)(TableName.AuditLog).where("expiresAt", "<", today).del();
+      return docs;
+    } catch (error) {
+      throw new DatabaseError({ error, name: "PruneAuditLog" });
+    }
+  };
+
+  return { ...auditLogOrm, pruneAuditLog, find };
 };

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

@@ -1,3 +1,4 @@
+import { logger } from "@app/lib/logger";
 import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
 import { TProjectDALFactory } from "@app/services/project/project-dal";
 
@@ -14,6 +15,19 @@ type TAuditLogQueueServiceFactoryDep = {
 
 export type TAuditLogQueueServiceFactory = ReturnType<typeof auditLogQueueServiceFactory>;
 
+const getTimeDiffForNextAuditLogPrune = (mills: number) => {
+  const today = new Date(mills);
+  // Get UTC midnight timestamp for today
+  const nextUtcMidnight = new Date(Date.UTC(today.getFullYear(), today.getMonth(), today.getDate(), 0, 0, 0));
+
+  // Check if we have already passed UTC midnight today
+  if (today.getTime() >= nextUtcMidnight.getTime()) {
+    // Add one day to get the timestamp for tomorrow's UTC midnight
+    nextUtcMidnight.setDate(nextUtcMidnight.getDate() + 1);
+  }
+  return nextUtcMidnight.getTime();
+};
+
 export const auditLogQueueServiceFactory = ({
   auditLogDAL,
   queueService,
@@ -43,6 +57,8 @@ export const auditLogQueueServiceFactory = ({
 
     const plan = await licenseService.getPlan(orgId);
     const ttl = plan.auditLogsRetentionDays * MS_IN_DAY;
+    // skip inserting if audit log retension is 0 meaning its not supported
+    if (ttl === 0) return;
     await auditLogDAL.create({
       actor: actor.type,
       actorMetadata: actor.metadata,
@@ -57,7 +73,47 @@ export const auditLogQueueServiceFactory = ({
     });
   });
 
+  queueService.start(
+    QueueName.AuditLogPrune,
+    async () => {
+      logger.info("Started audit log pruning");
+      await auditLogDAL.pruneAuditLog();
+      // calculate next utc time delay
+      // const nextPruneTime = getTimeDiffForNextAuditLogPrune();
+      // await queueService.stopJobById(QueueName.AuditLogPrune, "audit-log-prune");
+      logger.info("Finished audit log pruning");
+    },
+    {
+      settings: {
+        repeatStrategy: getTimeDiffForNextAuditLogPrune
+      }
+    }
+  );
+
+  // we are not using repeat because we want to run the in a predictable time of midnight UTC
+  // repeat has only cron job and every so we do the repeat manually
+  const startAuditLogPruneJob = async () => {
+    // clear previous job
+    await queueService.stopRepeatableJob(
+      QueueName.AuditLogPrune,
+      QueueJobs.AuditLogPrune,
+      {},
+      QueueName.AuditLogPrune // just a job id
+    );
+    await queueService.queue(QueueName.AuditLogPrune, QueueJobs.AuditLogPrune, undefined, {
+      delay: 5000,
+      jobId: QueueName.AuditLogPrune,
+      repeat: {}
+    });
+  };
+
+  queueService.listen(QueueName.AuditLogPrune, "error", (err) => {
+    logger.error("Audit log pruning failed");
+    logger.error(err);
+  });
+
   return {
-    pushToLog
+    pushToLog,
+    startAuditLogPruneJob
   };
 };

backend/src/lib/logger/logger.ts

@@ -26,14 +26,23 @@ const loggerConfig = z.object({
   AWS_CLOUDWATCH_LOG_REGION: z.string().default("us-east-1"),
   AWS_CLOUDWATCH_LOG_ACCESS_KEY_ID: z.string().min(1).optional(),
   AWS_CLOUDWATCH_LOG_ACCESS_KEY_SECRET: z.string().min(1).optional(),
-  AWS_CLOUDWATCH_LOG_INTERVAL: z.coerce.number().default(1000)
+  AWS_CLOUDWATCH_LOG_INTERVAL: z.coerce.number().default(1000),
+  NODE_ENV: z.enum(["development", "test", "production"]).default("production")
 });
 
 export const initLogger = async () => {
-  const targets: pino.TransportMultiOptions["targets"][number][] = [
-    { level: "info", target: "pino/file", options: {} }
-  ];
   const cfg = loggerConfig.parse(process.env);
+  const targets: pino.TransportMultiOptions["targets"][number][] = [
+    {
+      level: "info",
+      target: "pino/file",
+      options: {
+        destination: cfg.NODE_ENV === "development" ? 1 : "/var/log/infisical",
+        mkdir: true
+      }
+    }
+  ];
+
   if (cfg.AWS_CLOUDWATCH_LOG_ACCESS_KEY_ID && cfg.AWS_CLOUDWATCH_LOG_ACCESS_KEY_SECRET) {
     targets.push({
       target: "@serdnam/pino-cloudwatch-transport",

backend/src/queue/queue-service.ts

@@ -1,4 +1,4 @@
-import { Job, JobsOptions, Queue, RepeatOptions, Worker, WorkerListener } from "bullmq";
+import { Job, JobsOptions, Queue, QueueOptions, RepeatOptions, Worker, WorkerListener } from "bullmq";
 import Redis from "ioredis";
 
 import { TCreateAuditLogDTO } from "@app/ee/services/audit-log/audit-log-types";
@@ -11,6 +11,7 @@ export enum QueueName {
   SecretRotation = "secret-rotation",
   SecretReminder = "secret-reminder",
   AuditLog = "audit-log",
+  AuditLogPrune = "audit-log-prune",
   IntegrationSync = "sync-integrations",
   SecretWebhook = "secret-webhook",
   SecretFullRepoScan = "secret-full-repo-scan",
@@ -21,6 +22,7 @@ export enum QueueJobs {
   SecretReminder = "secret-reminder-job",
   SecretRotation = "secret-rotation-job",
   AuditLog = "audit-log-job",
+  AuditLogPrune = "audit-log-prune-job",
   SecWebhook = "secret-webhook-trigger",
   IntegrationSync = "secret-integration-pull",
   SecretScan = "secret-scan"
@@ -45,6 +47,10 @@ export type TQueueJobTypes = {
     name: QueueJobs.AuditLog;
     payload: TCreateAuditLogDTO;
   };
+  [QueueName.AuditLogPrune]: {
+    name: QueueJobs.AuditLogPrune;
+    payload: undefined;
+  };
   [QueueName.SecretWebhook]: {
     name: QueueJobs.SecWebhook;
     payload: { projectId: string; environment: string; secretPath: string };
@@ -74,16 +80,20 @@ export const queueServiceFactory = (redisUrl: string) => {
 
   const start = <T extends QueueName>(
     name: T,
-    jobFn: (job: Job<TQueueJobTypes[T]["payload"], void, TQueueJobTypes[T]["name"]>) => Promise<void>
+    jobFn: (job: Job<TQueueJobTypes[T]["payload"], void, TQueueJobTypes[T]["name"]>) => Promise<void>,
+    queueSettings: Omit<QueueOptions, "connection"> = {}
   ) => {
     if (queueContainer[name]) {
       throw new Error(`${name} queue is already initialized`);
     }
 
     queueContainer[name] = new Queue<TQueueJobTypes[T]["payload"], void, TQueueJobTypes[T]["name"]>(name as string, {
+      ...queueSettings,
       connection
     });
+
     workerContainer[name] = new Worker<TQueueJobTypes[T]["payload"], void, TQueueJobTypes[T]["name"]>(name, jobFn, {
+      ...queueSettings,
       connection
     });
   };
@@ -129,9 +139,20 @@ export const queueServiceFactory = (redisUrl: string) => {
     return q.removeRepeatableByKey(job.repeatJobKey);
   };
 
+  const stopJobById = async <T extends QueueName>(name: T, jobId: string) => {
+    const q = queueContainer[name];
+    const job = await q.getJob(jobId);
+    return job?.remove().catch(() => undefined);
+  };
+
+  const clearQueue = async (name: QueueName) => {
+    const q = queueContainer[name];
+    await q.drain();
+  };
+
   const shutdown = async () => {
     await Promise.all(Object.values(workerContainer).map((worker) => worker.close()));
   };
 
-  return { start, listen, queue, shutdown, stopRepeatableJob, stopRepeatableJobByJobId };
+  return { start, listen, queue, shutdown, stopRepeatableJob, stopRepeatableJobByJobId, clearQueue, stopJobById };
 };

backend/src/server/routes/index.ts

@@ -444,6 +444,7 @@ export const registerRoutes = async (
   });
 
   await superAdminService.initServerCfg();
+  await auditLogQueue.startAuditLogPruneJob();
   // setup the communication with license key server
   await licenseService.init();
   // inject all services

backend/src/services/project-env/project-env-service.ts

@@ -67,7 +67,7 @@ export const projectEnvServiceFactory = ({
     if (!oldEnv) throw new BadRequestError({ message: "Environment not found" });
 
     if (slug) {
-      const existingEnv = await projectEnvDAL.findOne({ slug });
+      const existingEnv = await projectEnvDAL.findOne({ slug, projectId });
       if (existingEnv && existingEnv.id !== id) {
         throw new BadRequestError({
           message: "Environment with slug already exist",

docker-compose.pg.yml

@@ -96,32 +96,32 @@ services:
       - 1025:1025 # SMTP server
       - 8025:8025 # Web UI
 
-  mongo:
+  # mongo:
-    image: mongo
+  #   image: mongo
-    container_name: infisical-dev-mongo
+  #   container_name: infisical-dev-mongo
-    restart: always
+  #   restart: always
-    env_file: .env
+  #   env_file: .env
-    environment:
+  #   environment:
-      - MONGO_INITDB_ROOT_USERNAME=root
+  #     - MONGO_INITDB_ROOT_USERNAME=root
-      - MONGO_INITDB_ROOT_PASSWORD=example
+  #     - MONGO_INITDB_ROOT_PASSWORD=example
-    volumes:
+  #   volumes:
-      - mongo-data:/data/db
+  #     - mongo-data:/data/db
-    ports:
+  #   ports:
-      - 27017:27017
+  #     - 27017:27017
-
+  #
-  mongo-express:
+  # mongo-express:
-    container_name: infisical-dev-mongo-express
+  #   container_name: infisical-dev-mongo-express
-    image: mongo-express
+  #   image: mongo-express
-    restart: always
+  #   restart: always
-    depends_on:
+  #   depends_on:
-      - mongo
+  #     - mongo
-    env_file: .env
+  #   env_file: .env
-    environment:
+  #   environment:
-      - ME_CONFIG_MONGODB_ADMINUSERNAME=root
+  #     - ME_CONFIG_MONGODB_ADMINUSERNAME=root
-      - ME_CONFIG_MONGODB_ADMINPASSWORD=example
+  #     - ME_CONFIG_MONGODB_ADMINPASSWORD=example
-      - ME_CONFIG_MONGODB_URL=mongodb://root:example@mongo:27017/
+  #     - ME_CONFIG_MONGODB_URL=mongodb://root:example@mongo:27017/
-    ports:
+  #   ports:
-      - 8081:8081
+  #     - 8081:8081
 
 volumes:
   postgres-data: