feat(infisical-pg): completed webhook and integration trigger queue

backend-pg/package.json

@@ -35,6 +35,8 @@
     "@types/jmespath": "^0.15.2",
     "@types/jsonwebtoken": "^9.0.5",
     "@types/jsrp": "^0.2.6",
+    "@types/libsodium-wrappers": "^0.7.13",
+    "@types/lodash.isequal": "^4.5.8",
     "@types/node": "^20.9.5",
     "@types/nodemailer": "^6.4.14",
     "@types/passport-github": "^1.1.12",
@@ -63,6 +65,7 @@
     "vitest": "^1.0.4"
   },
   "dependencies": {
+    "@aws-sdk/client-secrets-manager": "^3.485.0",
     "@casl/ability": "^6.5.0",
     "@fastify/cookie": "^9.2.0",
     "@fastify/cors": "^8.4.1",
@@ -77,6 +80,7 @@
     "@ucast/mongo2js": "^1.3.4",
     "ajv": "^8.12.0",
     "argon2": "^0.31.2",
+    "aws-sdk": "^2.1532.0",
     "axios": "^1.6.2",
     "axios-retry": "^4.0.0",
     "bcrypt": "^5.1.1",
@@ -91,6 +95,8 @@
     "jsonwebtoken": "^9.0.2",
     "jsrp": "^0.2.4",
     "knex": "^3.0.1",
+    "libsodium-wrappers": "^0.7.13",
+    "lodash.isequal": "^4.5.0",
     "mysql2": "^3.6.5",
     "nanoid": "^5.0.4",
     "nodemailer": "^6.9.7",

backend-pg/src/ee/services/secret-rotation/secret-rotation-queue/secret-rotation-queue-fn.ts

@@ -73,7 +73,8 @@ export const secretRotationHttpFn = async (
     url,
     headers,
     data: body,
-    timeout: EXTERNAL_REQUEST_TIMEOUT
+    timeout: EXTERNAL_REQUEST_TIMEOUT,
+    signal: AbortSignal.timeout(EXTERNAL_REQUEST_TIMEOUT)
   });
 };
 

backend-pg/src/lib/fn/index.ts

@@ -3,3 +3,4 @@
 // Code taken to keep in in house and to adjust somethings for our needs
 export * from "./array";
 export * from "./object";
+export * from "./string";

backend-pg/src/lib/fn/string.ts

@@ -0,0 +1,5 @@
+import path from "path";
+
+// given two paths irrespective of ending with / or not
+// this will return true if its equal
+export const isSamePath = async (from: string, to: string) => !path.relative(from, to);

backend-pg/src/lib/knex/index.ts

@@ -117,13 +117,13 @@ export const ormify = <DbOps extends object, Tname extends keyof Tables>(
     }
   },
   update: async (
-    filter: Partial<Tables[Tname]["base"]>,
+    filter: TFindFilter<Tables[Tname]["base"]>,
     data: Tables[Tname]["update"],
     tx?: Knex
   ) => {
     try {
       const res = await (tx || db)(tableName)
-        .where(filter)
+        .where(buildFindFilter(filter))
         .update(data as any)
         .returning("*");
       return res;
@@ -142,9 +142,12 @@ export const ormify = <DbOps extends object, Tname extends keyof Tables>(
       throw new DatabaseError({ error, name: "Delete by id" });
     }
   },
-  delete: async (filter: Partial<Tables[Tname]["base"]>, tx?: Knex) => {
+  delete: async (filter: TFindFilter<Tables[Tname]["base"]>, tx?: Knex) => {
     try {
-      const res = await (tx || db)(tableName).where(filter).delete().returning("*");
+      const res = await (tx || db)(tableName)
+        .where(buildFindFilter(filter))
+        .delete()
+        .returning("*");
       return res;
     } catch (error) {
       throw new DatabaseError({ error, name: "Delete" });

backend-pg/src/queue/queue-service.ts

@@ -5,12 +5,16 @@ import { TCreateAuditLogDTO } from "@app/ee/services/audit-log/audit-log-types";
 
 export enum QueueName {
   SecretRotation = "secret-rotation",
-  AuditLog = "audit-log"
+  AuditLog = "audit-log",
+  IntegrationSync = "sync-integrations",
+  SecretWebhook = "secret-webhook"
 }
 
 export enum QueueJobs {
   SecretRotation = "secret-rotation-job",
-  AuditLog = "audit-log-job"
+  AuditLog = "audit-log-job",
+  SecWebhook = "secret-webhook-trigger",
+  IntegrationSync = "secret-integration-pull"
 }
 
 export type TQueueJobTypes = {
@@ -22,6 +26,14 @@ export type TQueueJobTypes = {
     name: QueueJobs.AuditLog;
     payload: TCreateAuditLogDTO;
   };
+  [QueueName.SecretWebhook]: {
+    name: QueueJobs.SecWebhook;
+    payload: { projectId: string; environment: string; secretPath: string };
+  };
+  [QueueName.IntegrationSync]: {
+    name: QueueJobs.IntegrationSync;
+    payload: { projectId: string; environment: string; secretPath: string };
+  };
 };
 
 export type TQueueServiceFactory = ReturnType<typeof queueServiceFactory>;

backend-pg/src/server/routes/index.ts

@@ -67,6 +67,7 @@ import { projectRoleDalFactory } from "@app/services/project-role/project-role-d
 import { projectRoleServiceFactory } from "@app/services/project-role/project-role-service";
 import { secretBlindIndexDalFactory } from "@app/services/secret/secret-blind-index-dal";
 import { secretDalFactory } from "@app/services/secret/secret-dal";
+import { secretQueueFactory } from "@app/services/secret/secret-queue";
 import { secretServiceFactory } from "@app/services/secret/secret-service";
 import { secretVersionDalFactory } from "@app/services/secret/secret-version-dal";
 import { secretFolderDalFactory } from "@app/services/secret-folder/secret-folder-dal";
@@ -245,16 +246,12 @@ export const registerRoutes = async (
     folderVersionDal,
     permissionService
   });
-
+  const webhookService = webhookServiceFactory({
-  const secretService = secretServiceFactory({
-    folderDal,
-    secretVersionDal,
-    secretBlindIndexDal,
     permissionService,
-    secretDal,
+    webhookDal,
-    secretTagDal,
+    projectEnvDal
-    snapshotService
   });
+
   const secretTagService = secretTagServiceFactory({ secretTagDal, permissionService });
   const folderService = secretFolderServiceFactory({
     permissionService,
@@ -271,7 +268,34 @@ export const registerRoutes = async (
     secretDal
   });
   const projectBotService = projectBotServiceFactory({ permissionService, projectBotDal });
-
+  const integrationAuthService = integrationAuthServiceFactory({
+    integrationAuthDal,
+    integrationDal,
+    permissionService,
+    projectBotDal,
+    projectBotService
+  });
+  const secretQueueService = secretQueueFactory({
+    queueService,
+    webhookService,
+    secretDal,
+    folderDal,
+    secretImportService,
+    integrationAuthService,
+    projectBotService,
+    integrationDal,
+    secretImportDal
+  });
+  const secretService = secretServiceFactory({
+    folderDal,
+    secretVersionDal,
+    secretBlindIndexDal,
+    permissionService,
+    secretDal,
+    secretTagDal,
+    snapshotService,
+    secretQueueService
+  });
   const sarService = secretApprovalRequestServiceFactory({
     permissionService,
     folderDal,
@@ -303,18 +327,6 @@ export const registerRoutes = async (
     integrationDal,
     integrationAuthDal
   });
-  const integrationAuthService = integrationAuthServiceFactory({
-    integrationAuthDal,
-    integrationDal,
-    permissionService,
-    projectBotDal,
-    projectBotService
-  });
-  const webhookService = webhookServiceFactory({
-    permissionService,
-    webhookDal,
-    projectEnvDal
-  });
   const serviceTokenService = serviceTokenServiceFactory({
     projectEnvDal,
     serviceTokenDal,

backend-pg/src/services/integration-auth/integration-auth-service.ts

@@ -1019,6 +1019,7 @@ export const integrationAuthServiceFactory = ({
     getRailwayEnvironments,
     getNorthFlankSecretGroups,
     getTeamcityBuildConfigs,
-    getBitbucketWorkspaces
+    getBitbucketWorkspaces,
+    getIntegrationAccessToken
   };
 };

backend-pg/src/services/integration/integration-dal.ts

@@ -89,5 +89,105 @@ export const integrationDalFactory = (db: TDbClient) => {
     }
   };
 
-  return { ...integrationOrm, find, findOne, findById, findByProjectId };
+  // used for syncing secrets
+  // this will populate integration auth also
+  const findByProjectIdV2 = async (projectId: string, environment: string, tx?: Knex) => {
+    const docs = await (tx || db)(TableName.Integration)
+      .where(`${TableName.Environment}.projectId`, projectId)
+      .where("isActive", true)
+      .where(`${TableName.Environment}.slug`, environment)
+      .join(TableName.Environment, `${TableName.Integration}.envId`, `${TableName.Environment}.id`)
+      .join(
+        TableName.IntegrationAuth,
+        `${TableName.IntegrationAuth}.id`,
+        `${TableName.Integration}.integrationAuthId`
+      )
+      .select(db.ref("name").withSchema(TableName.Environment).as("envName"))
+      .select(db.ref("slug").withSchema(TableName.Environment).as("envSlug"))
+      .select(db.ref("id").withSchema(TableName.Environment).as("envId"))
+      .select(db.ref("projectId").withSchema(TableName.Environment))
+      .select(selectAllTableCols(TableName.Integration))
+      .select(
+        db.ref("id").withSchema(TableName.IntegrationAuth).as("idAu"),
+        db.ref("integration").withSchema(TableName.IntegrationAuth).as("integrationAu"),
+        db.ref("teamId").withSchema(TableName.IntegrationAuth).as("teamIdAu"),
+        db.ref("url").withSchema(TableName.IntegrationAuth).as("urlAu"),
+        db.ref("namespace").withSchema(TableName.IntegrationAuth).as("namespaceAu"),
+        db.ref("accountId").withSchema(TableName.IntegrationAuth).as("accountIdAu"),
+        db.ref("refreshCiphertext").withSchema(TableName.IntegrationAuth).as("refreshCiphertextAu"),
+        db.ref("refreshIV").withSchema(TableName.IntegrationAuth).as("refreshIVAu"),
+        db.ref("refreshTag").withSchema(TableName.IntegrationAuth).as("refreshTagAu"),
+        db
+          .ref("accessIdCiphertext")
+          .withSchema(TableName.IntegrationAuth)
+          .as("accessIdCiphertextAu"),
+        db.ref("accessIdIV").withSchema(TableName.IntegrationAuth).as("accessIdIVAu"),
+        db.ref("accessIdTag").withSchema(TableName.IntegrationAuth).as("accessIdTagAu"),
+        db.ref("accessIV").withSchema(TableName.IntegrationAuth).as("accessIVAu"),
+        db.ref("accessTag").withSchema(TableName.IntegrationAuth).as("accessTagAu"),
+        db.ref("accessCiphertext").withSchema(TableName.IntegrationAuth).as("accessCiphertextAu"),
+        db.ref("accessExpiresAt").withSchema(TableName.IntegrationAuth).as("accessExpiresAtAu"),
+        db.ref("metadata").withSchema(TableName.IntegrationAuth).as("metadataAu"),
+        db.ref("algorithm").withSchema(TableName.IntegrationAuth).as("algorithmAu"),
+        db.ref("keyEncoding").withSchema(TableName.IntegrationAuth).as("keyEncodingAu")
+      );
+    return docs.map(
+      ({
+        envId,
+        envName,
+        envSlug,
+        idAu: id,
+        integrationAu: integration,
+        teamIdAu: teamId,
+        urlAu: url,
+        namespaceAu: namespace,
+        accountIdAu: accountId,
+        refreshIVAu: refreshIV,
+        refreshCiphertextAu: refreshCiphertext,
+        refreshTagAu: refreshTag,
+        accessIVAu: accessIV,
+        accessCiphertextAu: accessCiphertext,
+        accessTagAu: accessTag,
+        accessIdIVAu: accessIdIV,
+        accessIdTagAu: accessIdTag,
+        accessIdCiphertextAu: accessIdCiphertext,
+        metadataAu: metadata,
+        algorithmAu: algorithm,
+        keyEncodingAu: keyEncoding,
+        accessExpiresAtAu: accessExpiresAt,
+        ...el
+      }) => ({
+        ...el,
+        envId,
+        environment: {
+          id: envId,
+          name: envName,
+          slug: envSlug
+        },
+        integrationAuth: {
+          id,
+          integration,
+          teamId,
+          url,
+          namespace,
+          accountId,
+          refreshTag,
+          refreshIV,
+          refreshCiphertext,
+          accessIdCiphertext,
+          accessIdIV,
+          accessIdTag,
+          accessIV,
+          accessCiphertext,
+          accessTag,
+          metadata,
+          algorithm,
+          keyEncoding,
+          accessExpiresAt
+        }
+      })
+    );
+  };
+
+  return { ...integrationOrm, find, findOne, findById, findByProjectId, findByProjectIdV2 };
 };

backend-pg/src/services/secret-import/secret-import-service.ts

@@ -1,5 +1,6 @@
 import { ForbiddenError, subject } from "@casl/ability";
 
+import { SecretType, TSecretImports } from "@app/db/schemas";
 import { TPermissionServiceFactory } from "@app/ee/services/permission/permission-service";
 import {
   ProjectPermissionActions,
@@ -175,6 +176,37 @@ export const secretImportServiceFactory = ({
     return secImports;
   };
 
+  const fnSecretsFromImports = async (
+    allowedImports: (Omit<TSecretImports, "importEnv"> & {
+      importEnv: { id: string; slug: string; name: string };
+    })[]
+  ) => {
+    const importedFolders = await folderDal.findByManySecretPath(
+      allowedImports.map(({ importEnv, importPath }) => ({
+        envId: importEnv.id,
+        secretPath: importPath
+      }))
+    );
+    const folderIds = importedFolders.map((el) => el?.id).filter(Boolean) as string[];
+    if (!folderIds.length) {
+      return [];
+    }
+    const importedSecrets = await secretDal.find({
+      $in: { folderId: folderIds },
+      type: SecretType.Shared
+    });
+
+    const importedSecsGroupByFolderId = groupBy(importedSecrets, (i) => i.folderId);
+    return allowedImports.map(({ importPath, importEnv }, i) => ({
+      secretPath: importPath,
+      environment: importEnv,
+      folderId: importedFolders?.[i]?.id,
+      secrets: importedFolders?.[i]?.id
+        ? importedSecsGroupByFolderId[importedFolders?.[i]?.id as string]
+        : []
+    }));
+  };
+
   const getSecretsFromImports = async ({
     path,
     environment,
@@ -202,29 +234,7 @@ export const secretImportServiceFactory = ({
         })
       )
     );
-    const importedFolders = await folderDal.findByManySecretPath(
+    return fnSecretsFromImports(allowedImports);
-      allowedImports.map(({ importEnv, importPath }) => ({
-        envId: importEnv.id,
-        secretPath: importPath
-      }))
-    );
-    const folderIds = importedFolders.map((el) => el?.id).filter(Boolean) as string[];
-    if (!folderIds.length) {
-      return [];
-    }
-    const importedSecrets = await secretDal.find({
-      $in: { folderId: folderIds }
-    });
-
-    const importedSecsGroupByFolderId = groupBy(importedSecrets, (i) => i.folderId);
-    return allowedImports.map(({ importPath, importEnv }, i) => ({
-      secretPath: importPath,
-      environment: importEnv,
-      folderId: importedFolders?.[i]?.id,
-      secrets: importedFolders?.[i]?.id
-        ? importedSecsGroupByFolderId[importedFolders?.[i]?.id as string]
-        : []
-    }));
   };
 
   return {
@@ -232,6 +242,7 @@ export const secretImportServiceFactory = ({
     updateImport,
     deleteImport,
     getImports,
-    getSecretsFromImports
+    getSecretsFromImports,
+    fnSecretsFromImports
   };
 };

backend-pg/src/services/secret/secret-fns.ts

@@ -0,0 +1,179 @@
+/* eslint-disable no-await-in-loop */
+import path from "path";
+
+import { decryptSymmetric128BitHexKeyUTF8 } from "@app/lib/crypto";
+
+import { TSecretFolderDalFactory } from "../secret-folder/secret-folder-dal";
+import { TSecretDalFactory } from "./secret-dal";
+
+type TInterpolateSecretArg = {
+  projectId: string;
+  secretEncKey: string;
+  secretDal: Pick<TSecretDalFactory, "findByFolderId">;
+  folderDal: Pick<TSecretFolderDalFactory, "findBySecretPath">;
+};
+
+export const interpolateSecrets = ({
+  projectId,
+  secretEncKey,
+  secretDal,
+  folderDal
+}: TInterpolateSecretArg) => {
+  const fetchSecretsCrossEnv = () => {
+    const fetchCache: Record<string, Record<string, string>> = {};
+
+    return async (secRefEnv: string, secRefPath: string[], secRefKey: string) => {
+      const secRefPathUrl = path.join("/", ...secRefPath);
+      const uniqKey = `${secRefEnv}-${secRefPathUrl}`;
+
+      if (fetchCache?.[uniqKey]) {
+        return fetchCache[uniqKey][secRefKey];
+      }
+
+      const folder = await folderDal.findBySecretPath(projectId, secRefEnv, secRefPathUrl);
+      if (!folder) return "";
+      const secrets = await secretDal.findByFolderId(folder.id);
+
+      const decryptedSec = secrets.reduce<Record<string, string>>((prev, secret) => {
+        const secretKey = decryptSymmetric128BitHexKeyUTF8({
+          ciphertext: secret.secretKeyCiphertext,
+          iv: secret.secretKeyIV,
+          tag: secret.secretKeyTag,
+          key: secretEncKey
+        });
+        const secretValue = decryptSymmetric128BitHexKeyUTF8({
+          ciphertext: secret.secretValueCiphertext,
+          iv: secret.secretValueIV,
+          tag: secret.secretValueTag,
+          key: secretEncKey
+        });
+
+        // eslint-disable-next-line
+        prev[secretKey] = secretValue;
+        return prev;
+      }, {});
+
+      fetchCache[uniqKey] = decryptedSec;
+
+      return fetchCache[uniqKey][secRefKey];
+    };
+  };
+
+  const INTERPOLATION_SYNTAX_REG = /\${([^}]+)}/g;
+  const recursivelyExpandSecret = async (
+    expandedSec: Record<string, string>,
+    interpolatedSec: Record<string, string>,
+    fetchCrossEnv: (env: string, secPath: string[], secKey: string) => Promise<string>,
+    recursionChainBreaker: Record<string, boolean>,
+    key: string
+  ) => {
+    if (expandedSec?.[key] !== undefined) {
+      return expandedSec[key];
+    }
+    if (recursionChainBreaker?.[key]) {
+      return "";
+    }
+    // eslint-disable-next-line
+    recursionChainBreaker[key] = true;
+
+    let interpolatedValue = interpolatedSec[key];
+    if (!interpolatedValue) {
+      // eslint-disable-next-line no-console
+      console.error(`Couldn't find referenced value - ${key}`);
+      return "";
+    }
+
+    const refs = interpolatedValue.match(INTERPOLATION_SYNTAX_REG);
+    if (refs) {
+      for (const interpolationSyntax of refs) {
+        const interpolationKey = interpolationSyntax.slice(2, interpolationSyntax.length - 1);
+        const entities = interpolationKey.trim().split(".");
+
+        if (entities.length === 1) {
+          const val = await recursivelyExpandSecret(
+            expandedSec,
+            interpolatedSec,
+            fetchCrossEnv,
+            recursionChainBreaker,
+            interpolationKey
+          );
+          if (val) {
+            interpolatedValue = interpolatedValue.replaceAll(interpolationSyntax, val);
+          }
+          // eslint-disable-next-line
+          continue;
+        }
+
+        if (entities.length > 1) {
+          const secRefEnv = entities[0];
+          const secRefPath = entities.slice(1, entities.length - 1);
+          const secRefKey = entities[entities.length - 1];
+
+          const val = await fetchCrossEnv(secRefEnv, secRefPath, secRefKey);
+          if (val) {
+            interpolatedValue = interpolatedValue.replaceAll(interpolationSyntax, val);
+          }
+        }
+      }
+    }
+
+    // eslint-disable-next-line
+    expandedSec[key] = interpolatedValue;
+    return interpolatedValue;
+  };
+
+  // used to convert multi line ones to quotes ones with \n
+  const formatMultiValueEnv = (val?: string) => {
+    if (!val) return "";
+    if (!val.match("\n")) return val;
+    return `"${val.replace(/\n/g, "\\n")}"`;
+  };
+
+  const expandSecrets = async (
+    secrets: Record<string, { value: string; comment?: string; skipMultilineEncoding?: boolean }>
+  ) => {
+    const expandedSec: Record<string, string> = {};
+    const interpolatedSec: Record<string, string> = {};
+
+    const crossSecEnvFetch = fetchSecretsCrossEnv();
+
+    Object.keys(secrets).forEach((key) => {
+      if (secrets[key].value.match(INTERPOLATION_SYNTAX_REG)) {
+        interpolatedSec[key] = secrets[key].value;
+      } else {
+        expandedSec[key] = secrets[key].value;
+      }
+    });
+
+    for (const key of Object.keys(secrets)) {
+      if (expandedSec?.[key]) {
+        // should not do multi line encoding if user has set it to skip
+        // eslint-disable-next-line
+        secrets[key].value = secrets[key].skipMultilineEncoding
+          ? expandedSec[key]
+          : formatMultiValueEnv(expandedSec[key]);
+        // eslint-disable-next-line
+        continue;
+      }
+
+      // this is to avoid recursion loop. So the graph should be direct graph rather than cyclic
+      // so for any recursion building if there is an entity two times same key meaning it will be looped
+      const recursionChainBreaker: Record<string, boolean> = {};
+      const expandedVal = await recursivelyExpandSecret(
+        expandedSec,
+        interpolatedSec,
+        crossSecEnvFetch,
+        recursionChainBreaker,
+        key
+      );
+
+      // eslint-disable-next-line
+      secrets[key].value = secrets[key].skipMultilineEncoding
+        ? expandedVal
+        : formatMultiValueEnv(expandedVal);
+    }
+
+    return secrets;
+  };
+  return expandSecrets;
+};

backend-pg/src/services/secret/secret-queue.ts

@@ -0,0 +1,226 @@
+/* eslint-disable no-await-in-loop */
+import { decryptSymmetric128BitHexKeyUTF8 } from "@app/lib/crypto";
+import { isSamePath } from "@app/lib/fn";
+import { logger } from "@app/lib/logger";
+import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
+
+import { TIntegrationDalFactory } from "../integration/integration-dal";
+import { TIntegrationAuthServiceFactory } from "../integration-auth/integration-auth-service";
+import { syncIntegrationSecrets } from "../integration-auth/integration-sync-secret";
+import { TProjectBotServiceFactory } from "../project-bot/project-bot-service";
+import { TSecretFolderDalFactory } from "../secret-folder/secret-folder-dal";
+import { TSecretImportDalFactory } from "../secret-import/secret-import-dal";
+import { TSecretImportServiceFactory } from "../secret-import/secret-import-service";
+import { TWebhookServiceFactory } from "../webhook/webhook-service";
+import { TSecretDalFactory } from "./secret-dal";
+import { interpolateSecrets } from "./secret-fns";
+
+export type TSecretQueueFactory = ReturnType<typeof secretQueueFactory>;
+
+type TSecretQueueFactoryDep = {
+  queueService: TQueueServiceFactory;
+  webhookService: Pick<TWebhookServiceFactory, "fnTriggerWebhook">;
+  integrationDal: Pick<TIntegrationDalFactory, "findByProjectIdV2">;
+  projectBotService: Pick<TProjectBotServiceFactory, "getBotKey">;
+  integrationAuthService: Pick<TIntegrationAuthServiceFactory, "getIntegrationAccessToken">;
+  folderDal: Pick<TSecretFolderDalFactory, "findBySecretPath">;
+  secretDal: Pick<TSecretDalFactory, "findByFolderId">;
+  secretImportDal: Pick<TSecretImportDalFactory, "find">;
+  secretImportService: Pick<TSecretImportServiceFactory, "fnSecretsFromImports">;
+};
+
+export type TGetSecrets = {
+  secretPath: string;
+  projectId: string;
+  environment: string;
+};
+
+export const secretQueueFactory = ({
+  queueService,
+  webhookService,
+  integrationDal,
+  projectBotService,
+  integrationAuthService,
+  secretDal,
+  secretImportDal,
+  secretImportService,
+  folderDal
+}: TSecretQueueFactoryDep) => {
+  const syncSecrets = async (dto: TGetSecrets) => {
+    queueService.queue(QueueName.SecretWebhook, QueueJobs.SecWebhook, dto, {
+      jobId: `secret-webhook-${dto.environment}-${dto.projectId}-${dto.secretPath}`,
+      removeOnFail: { count: 5 },
+      removeOnComplete: true,
+      delay: 1000,
+      attempts: 5,
+      backoff: {
+        type: "exponential",
+        delay: 3000
+      }
+    });
+
+    queueService.queue(QueueName.IntegrationSync, QueueJobs.IntegrationSync, dto, {
+      attempts: 5,
+      delay: 1000,
+      backoff: {
+        type: "exponential",
+        delay: 3000
+      },
+      removeOnComplete: true,
+      removeOnFail: {
+        count: 5 // keep the most recent  jobs
+      }
+    });
+  };
+
+  const getIntegrationSecrets = async (dto: TGetSecrets & { folderId: string }, key: string) => {
+    const secrets = await secretDal.findByFolderId(dto.folderId);
+    if (!secrets.length) return {};
+
+    // get imported secrets
+    const secretImport = await secretImportDal.find({ folderId: dto.folderId });
+    const importedSecrets = await secretImportService.fnSecretsFromImports(secretImport);
+    const content: Record<
+      string,
+      { value: string; comment?: string; skipMultilineEncoding?: boolean }
+    > = {};
+
+    importedSecrets.forEach(({ secrets: secs }) => {
+      secs.forEach((secret) => {
+        const secretKey = decryptSymmetric128BitHexKeyUTF8({
+          ciphertext: secret.secretKeyCiphertext,
+          iv: secret.secretKeyIV,
+          tag: secret.secretKeyTag,
+          key
+        });
+        const secretValue = decryptSymmetric128BitHexKeyUTF8({
+          ciphertext: secret.secretValueCiphertext,
+          iv: secret.secretValueIV,
+          tag: secret.secretValueTag,
+          key
+        });
+        content[secretKey] = { value: secretValue };
+        content[secretKey].skipMultilineEncoding = Boolean(secret.skipMultilineEncoding);
+
+        if (secret.secretCommentCiphertext && secret.secretCommentIV && secret.secretCommentTag) {
+          const commentValue = decryptSymmetric128BitHexKeyUTF8({
+            ciphertext: secret.secretCommentCiphertext,
+            iv: secret.secretCommentIV,
+            tag: secret.secretCommentTag,
+            key
+          });
+          content[secretKey].comment = commentValue;
+        }
+      });
+    });
+    console.log(secrets.filter(({ type }) => type === "personal"));
+    secrets.forEach((secret) => {
+      const secretKey = decryptSymmetric128BitHexKeyUTF8({
+        ciphertext: secret.secretKeyCiphertext,
+        iv: secret.secretKeyIV,
+        tag: secret.secretKeyTag,
+        key
+      });
+
+      const secretValue = decryptSymmetric128BitHexKeyUTF8({
+        ciphertext: secret.secretValueCiphertext,
+        iv: secret.secretValueIV,
+        tag: secret.secretValueTag,
+        key
+      });
+
+      content[secretKey] = { value: secretValue };
+
+      if (secret.secretCommentCiphertext && secret.secretCommentIV && secret.secretCommentTag) {
+        const commentValue = decryptSymmetric128BitHexKeyUTF8({
+          ciphertext: secret.secretCommentCiphertext,
+          iv: secret.secretCommentIV,
+          tag: secret.secretCommentTag,
+          key
+        });
+        content[secretKey].comment = commentValue;
+      }
+
+      content[secretKey].skipMultilineEncoding = Boolean(secret.skipMultilineEncoding);
+    });
+    const expandSecrets = interpolateSecrets({
+      projectId: dto.projectId,
+      secretEncKey: key,
+      folderDal,
+      secretDal
+    });
+    await expandSecrets(content);
+    return content;
+  };
+
+  queueService.start(QueueName.IntegrationSync, async (job) => {
+    logger.info("Secret integration sync started", job.data, job.id);
+    const { environment, projectId, secretPath } = job.data;
+    const folder = await folderDal.findBySecretPath(projectId, environment, secretPath);
+    if (!folder) {
+      logger.error("Secret path not found");
+      return;
+    }
+
+    const integrations = await integrationDal.findByProjectIdV2(projectId, environment);
+    const toBeSyncedIntegrations = integrations.filter(
+      ({ secretPath: integrationSecPath, isActive }) =>
+        isActive && isSamePath(secretPath, integrationSecPath)
+    );
+
+    for (const integration of toBeSyncedIntegrations) {
+      const integrationAuth = {
+        ...integration.integrationAuth,
+        createdAt: new Date(),
+        updatedAt: new Date(),
+        projectId: integration.projectId
+      };
+
+      const botKey = await projectBotService.getBotKey(projectId);
+      const { accessToken, accessId } = await integrationAuthService.getIntegrationAccessToken(
+        integrationAuth,
+        botKey
+      );
+      const secrets = await getIntegrationSecrets(
+        { environment, projectId, secretPath, folderId: folder.id },
+        botKey
+      );
+      const suffixedSecrets: typeof secrets = {};
+      const metadata = integration.metadata as Record<string, any>;
+      if (metadata) {
+        Object.keys(secrets).forEach((key) => {
+          const prefix = metadata?.secretPrefix || "";
+          const suffix = metadata?.secretSuffix || "";
+          const newKey = prefix + key + suffix;
+          suffixedSecrets[newKey] = secrets[key];
+        });
+      }
+
+      await syncIntegrationSecrets({
+        integration,
+        integrationAuth,
+        secrets: Object.keys(suffixedSecrets).length !== 0 ? suffixedSecrets : secrets,
+        accessId: accessId as string,
+        accessToken,
+        appendices: {
+          prefix: metadata?.secretPrefix || "",
+          suffix: metadata?.secretSuffix || ""
+        }
+      });
+    }
+
+    logger.info("Secret integration sync ended", job.id);
+  });
+
+  queueService.listen(QueueName.IntegrationSync, "failed", (job, err) => {
+    logger.error("Failed to sync integration", job?.data, err);
+  });
+
+  queueService.start(QueueName.SecretWebhook, async (job) => {
+    logger.info("Secret webhook job started", job.data, job.id);
+    await webhookService.fnTriggerWebhook(job.data);
+    logger.info("Secret webhook job ended", job.id);
+  });
+
+  return { syncSecrets };
+};

backend-pg/src/services/secret/secret-service.ts

@@ -23,6 +23,7 @@ import { TSecretFolderDalFactory } from "../secret-folder/secret-folder-dal";
 import { TSecretTagDalFactory } from "../secret-tag/secret-tag-dal";
 import { TSecretBlindIndexDalFactory } from "./secret-blind-index-dal";
 import { TSecretDalFactory } from "./secret-dal";
+import { TSecretQueueFactory } from "./secret-queue";
 import {
   TCreateBulkSecretDTO,
   TCreateSecretDTO,
@@ -49,6 +50,7 @@ type TSecretServiceFactoryDep = {
   secretBlindIndexDal: TSecretBlindIndexDalFactory;
   permissionService: Pick<TPermissionServiceFactory, "getProjectPermission">;
   snapshotService: Pick<TSecretSnapshotServiceFactory, "performSnapshot">;
+  secretQueueService: Pick<TSecretQueueFactory, "syncSecrets">;
 };
 
 export type TSecretServiceFactory = ReturnType<typeof secretServiceFactory>;
@@ -77,7 +79,8 @@ export const secretServiceFactory = ({
   folderDal,
   secretBlindIndexDal,
   permissionService,
-  snapshotService
+  snapshotService,
+  secretQueueService
 }: TSecretServiceFactoryDep) => {
   // utility function to get secret blind index data
   const interalGenSecBlindIndexByName = async (projectId: string, secretName: string) => {
@@ -329,6 +332,7 @@ export const secretServiceFactory = ({
     );
 
     await snapshotService.performSnapshot(folderId);
+    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
     // TODO(akhilmhdh-pg): licence check, posthog service and snapshot
     return { ...secret[0], tags };
   };
@@ -421,7 +425,7 @@ export const secretServiceFactory = ({
     );
 
     await snapshotService.performSnapshot(folderId);
-
+    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
     // TODO(akhilmhdh-pg): licence check, posthog service and snapshot
     return updatedSecret[0];
   };
@@ -475,6 +479,7 @@ export const secretServiceFactory = ({
     );
 
     await snapshotService.performSnapshot(folderId);
+    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
 
     // TODO(akhilmhdh-pg): licence check, posthog service and snapshot
     return deletedSecret[0];
@@ -576,6 +581,8 @@ export const secretServiceFactory = ({
     );
 
     await snapshotService.performSnapshot(folderId);
+    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
+
     return newSecrets;
   };
 
@@ -649,6 +656,8 @@ export const secretServiceFactory = ({
     );
 
     await snapshotService.performSnapshot(folderId);
+    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
+
     return secrets;
   };
 
@@ -695,6 +704,8 @@ export const secretServiceFactory = ({
     );
 
     await snapshotService.performSnapshot(folderId);
+    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
+
     return secretsDeleted;
   };
 

backend-pg/src/services/webhook/webhook-dal.ts

@@ -1,7 +1,7 @@
 import { Knex } from "knex";
 
 import { TDbClient } from "@app/db";
-import { TableName, TWebhooks } from "@app/db/schemas";
+import { TableName, TWebhooks, TWebhooksUpdate } from "@app/db/schemas";
 import { DatabaseError } from "@app/lib/errors";
 import { ormify, selectAllTableCols } from "@app/lib/knex";
 
@@ -101,5 +101,17 @@ export const webhookDalFactory = (db: TDbClient) => {
     }
   };
 
-  return { ...webhookOrm, findById, findOne, find, findAllWebhooks };
+  const bulkUpdate = async (data: Array<TWebhooksUpdate & { id: string }>, tx?: Knex) => {
+    try {
+      const queries = data.map(({ id, ...el }) =>
+        (tx || db)(TableName.Webhook).where({ id }).update(el)
+      );
+      const docs = await Promise.all(queries);
+      return docs;
+    } catch (error) {
+      throw new DatabaseError({ error, name: "bulk update secret" });
+    }
+  };
+
+  return { ...webhookOrm, findById, findOne, find, findAllWebhooks, bulkUpdate };
 };

backend-pg/src/services/webhook/webhook-fns.ts

@@ -5,6 +5,7 @@ import { getConfig } from "@app/lib/config/env";
 import { request } from "@app/lib/config/request";
 import { decryptSymmetric, decryptSymmetric128BitHexKeyUTF8 } from "@app/lib/crypto";
 
+const WEBHOOK_TRIGGER_TIMEOUT = 15 * 1000;
 export const triggerWebhookRequest = async (
   { url, encryptedSecretKey, iv, tag, keyEncoding }: TWebhooks,
   data: Record<string, unknown>
@@ -39,10 +40,14 @@ export const triggerWebhookRequest = async (
         .createHmac("sha256", secretKey)
         .update(JSON.stringify(payload))
         .digest("hex");
-      headers["x-infisical-signature"] = `t=${data.timestamp};${webhookSign}`;
+      headers["x-infisical-signature"] = `t=${payload.timestamp};${webhookSign}`;
     }
   }
-  const req = await request.post(url, payload, { headers });
+  const req = await request.post(url, payload, {
+    headers,
+    timeout: WEBHOOK_TRIGGER_TIMEOUT,
+    signal: AbortSignal.timeout(WEBHOOK_TRIGGER_TIMEOUT)
+  });
   return req;
 };
 

backend-pg/src/services/webhook/webhook-service.ts

@@ -1,4 +1,5 @@
 import { ForbiddenError } from "@casl/ability";
+import picomatch from "picomatch";
 
 import { SecretEncryptionAlgo, SecretKeyEncoding, TWebhooksInsert } from "@app/db/schemas";
 import { TPermissionServiceFactory } from "@app/ee/services/permission/permission-service";
@@ -16,6 +17,7 @@ import { getWebhookPayload, triggerWebhookRequest } from "./webhook-fns";
 import {
   TCreateWebhookDTO,
   TDeleteWebhookDTO,
+  TFnTriggerWebhookDTO,
   TListWebhookDTO,
   TTestWebhookDTO,
   TUpdateWebhookDTO
@@ -83,7 +85,7 @@ export const webhookServiceFactory = ({
 
     const webhook = await webhookDal.create(insertDoc);
     // TODO(akhilmhdh-pg): add audit log
-    return { ...webhook,projectId, environment: env };
+    return { ...webhook, projectId, environment: env };
   };
 
   const updateWebhook = async ({ actorId, actor, id, isDisabled }: TUpdateWebhookDTO) => {
@@ -101,7 +103,7 @@ export const webhookServiceFactory = ({
     );
 
     const updatedWebhook = await webhookDal.updateById(id, { isDisabled });
-    return { ...webhook,...updatedWebhook };
+    return { ...webhook, ...updatedWebhook };
   };
 
   const deleteWebhook = async ({ id, actor, actorId }: TDeleteWebhookDTO) => {
@@ -119,7 +121,7 @@ export const webhookServiceFactory = ({
     );
 
     const deletedWebhook = await webhookDal.deleteById(id);
-    return { ...webhook,...deletedWebhook };
+    return { ...webhook, ...deletedWebhook };
   };
 
   const testWebhook = async ({ id, actor, actorId }: TTestWebhookDTO) => {
@@ -150,7 +152,7 @@ export const webhookServiceFactory = ({
       lastStatus: isSuccess ? "success" : "failed",
       lastRunErrorMessage: isSuccess ? null : webhookError
     });
-    return {...webhook,...updatedWebhook}
+    return { ...webhook, ...updatedWebhook };
   };
 
   const listWebhooks = async ({
@@ -169,11 +171,63 @@ export const webhookServiceFactory = ({
     return webhookDal.findAllWebhooks(projectId, environment, secretPath);
   };
 
+  // this is reusable function
+  // used in secret queue to trigger webhook and update status when secrets changes
+  const fnTriggerWebhook = async ({ environment, secretPath, projectId }: TFnTriggerWebhookDTO) => {
+    const webhooks = await webhookDal.findAllWebhooks(projectId, environment);
+    const toBeTriggeredHooks = webhooks.filter(
+      ({ secretPath: hookSecretPath, isDisabled }) =>
+        !isDisabled && picomatch.isMatch(secretPath, hookSecretPath, { strictSlashes: false })
+    );
+    if (!toBeTriggeredHooks.length) return;
+    const webhooksTriggered = await Promise.allSettled(
+      toBeTriggeredHooks.map((hook) =>
+        triggerWebhookRequest(
+          hook,
+          getWebhookPayload("secrets.modified", projectId, environment, secretPath)
+        )
+      )
+    );
+    // filter hooks by status
+    const successWebhooks = webhooksTriggered
+      .filter(({ status }) => status === "fulfilled")
+      .map((_, i) => toBeTriggeredHooks[i].id);
+    const failedWebhooks = webhooksTriggered
+      .filter(({ status }) => status === "rejected")
+      .map((data, i) => ({
+        id: toBeTriggeredHooks[i].id,
+        error: data.status === "rejected" && data.reason.message
+      }));
+
+    await webhookDal.transaction(async (tx) => {
+      const env = await projectEnvDal.findOne({ projectId, slug: environment }, tx);
+      if (!env) throw new BadRequestError({ message: "Env not found" });
+      if (successWebhooks.length) {
+        await webhookDal.update(
+          { envId: env.id, $in: { id: successWebhooks } },
+          { lastStatus: "success", lastRunErrorMessage: null },
+          tx
+        );
+      }
+      if (failedWebhooks.length) {
+        await webhookDal.bulkUpdate(
+          failedWebhooks.map(({ id, error }) => ({
+            id,
+            lastRunErrorMessage: error,
+            lastStatus: "failed"
+          })),
+          tx
+        );
+      }
+    });
+  };
+
   return {
     createWebhook,
     deleteWebhook,
     listWebhooks,
     updateWebhook,
-    testWebhook
+    testWebhook,
+    fnTriggerWebhook
   };
 };

backend-pg/src/services/webhook/webhook-types.ts

@@ -24,3 +24,9 @@ export type TListWebhookDTO = {
   environment?: string;
   secretPath?: string;
 } & TProjectPermission;
+
+export type TFnTriggerWebhookDTO = {
+  projectId: string;
+  secretPath: string;
+  environment: string;
+};