feat: poc for secret replication completed

2026-10-15 01:52:22 +00:00 · 2024-05-31 23:10:01 +05:30
parent 520167a8ff
commit 36b7911bcc
21 changed files with 1213 additions and 29 deletions
@@ -50,6 +50,7 @@ export const secretImportServiceFactory = ({
    actorOrgId,
    actorAuthMethod,
    projectId,
+    isReplication,
    path
  }: TCreateSecretImportDTO) => {
    const { permission } = await permissionService.getProjectPermission(
@@ -100,7 +101,8 @@ export const secretImportServiceFactory = ({
          folderId: folder.id,
          position: lastPos + 1,
          importEnv: importEnv.id,
-          importPath: data.path
+          importPath: data.path,
+          isReplication
        },
        tx
      );
@@ -7,6 +7,7 @@ export type TCreateSecretImportDTO = {
    environment: string;
    path: string;
  };
+  isReplication?: boolean;
 } & TProjectPermission;

 export type TUpdateSecretImportDTO = {
@@ -0,0 +1,58 @@
+import { Knex } from "knex";
+
+import { TDbClient } from "@app/db";
+import { TableName, TSecretVersions } from "@app/db/schemas";
+import { ormify, selectAllTableCols } from "@app/lib/knex";
+
+export type TSecretReplicationDALFactory = ReturnType<typeof secretReplicationDALFactory>;
+
+export const secretReplicationDALFactory = (db: TDbClient) => {
+  const orm = ormify(db, TableName.SecretVersion);
+
+  const findSecrets = async (filter: { folderId: string; secrets: { id: string; version: number }[] }, tx?: Knex) => {
+    if (!filter.secrets) return [];
+
+    const sqlRawDocs = await (tx || db)(TableName.SecretVersion)
+      .where({ folderId: filter.folderId })
+      .andWhere((bd) => {
+        filter.secrets.forEach((el) => {
+          void bd.orWhere({
+            [`${TableName.SecretVersion}.secretId` as "secretId"]: el.id,
+            [`${TableName.SecretVersion}.version` as "version"]: el.version
+          });
+        });
+      })
+      .leftJoin<TSecretVersions>(
+        (tx || db)(TableName.SecretVersion)
+          .where("isReplicated", true)
+          .groupBy(["secretId", "version"])
+          .max("version")
+          .select("version", "secretId")
+          .as("latestVersion"),
+        (bd) => {
+          bd.on(`${TableName.SecretVersion}.secretId`, "latestVersion.secretId").andOn(
+            `${TableName.SecretVersion}.version`,
+            "latestVersion.max"
+          );
+        }
+      )
+      // .leftJoin<TSecretVersions>(
+      //   (tx || db)(TableName.SecretVersion).select("isReplicated", "version", "secretId").as("previousVersion"),
+      //   (bd) => {
+      //     bd.on(`${TableName.SecretVersion}.secretId`, "previousVersion.secretId").andOn(
+      //       "previousVersion.version",
+      //       (tx || db).raw(`${TableName.SecretVersion}.version - 1`)
+      //     );
+      //   }
+      // )
+      .select(db.ref("version").withSchema("latestVersion").as("latestReplicatedVersion"))
+      .select(selectAllTableCols(TableName.SecretVersion));
+
+    return sqlRawDocs;
+  };
+
+  return {
+    findSecrets,
+    ...orm
+  };
+};
@@ -0,0 +1,247 @@
+import { TKeyStoreFactory } from "@app/keystore/keystore";
+import { groupBy } from "@app/lib/fn";
+import { logger } from "@app/lib/logger";
+import { QueueJobs, QueueName, TQueueServiceFactory } from "@app/queue";
+
+import { TSecretDALFactory } from "../secret/secret-dal";
+import { fnSecretBulkInsert, fnSecretBulkUpdate } from "../secret/secret-fns";
+import { TSecretVersionDALFactory } from "../secret/secret-version-dal";
+import { TSecretVersionTagDALFactory } from "../secret/secret-version-tag-dal";
+import { TSecretFolderDALFactory } from "../secret-folder/secret-folder-dal";
+import { TSecretImportDALFactory } from "../secret-import/secret-import-dal";
+import { TSecretTagDALFactory } from "../secret-tag/secret-tag-dal";
+import { TSecretReplicationDALFactory } from "./secret-replication-dal";
+import { SecretReplicationOperations, TSyncSecretReplicationDTO } from "./secret-replication-types";
+
+type TSecretReplicationServiceFactoryDep = {
+  secretReplicationDAL: TSecretReplicationDALFactory;
+  secretDAL: Pick<TSecretDALFactory, "find" | "findByBlindIndexes" | "insertMany" | "bulkUpdate" | "delete">;
+  secretImportDAL: Pick<TSecretImportDALFactory, "find">;
+  folderDAL: Pick<TSecretFolderDALFactory, "findSecretPathByFolderIds">;
+  secretVersionDAL: Pick<TSecretVersionDALFactory, "find" | "insertMany" | "update">;
+  secretTagDAL: Pick<TSecretTagDALFactory, "find" | "saveTagsToSecret" | "deleteTagsManySecret">;
+  secretVersionTagDAL: Pick<TSecretVersionTagDALFactory, "find" | "insertMany">;
+  queueService: Pick<TQueueServiceFactory, "start" | "listen" | "queue" | "stopJobById">;
+  keyStore: Pick<TKeyStoreFactory, "acquireLock">;
+};
+
+export type TSecretReplicationServiceFactory = ReturnType<typeof secretReplicationServiceFactory>;
+
+// function getRandomError(): number {
+//   const minCeiled: number = Math.ceil(0);
+//   const maxFloored: number = Math.floor(20);
+//   const val = Math.floor(Math.random() * (maxFloored - minCeiled) + minCeiled); // The maximum is exclusive and the minimum is inclusive
+//   if (val >= 10) throw new Error("Random error point");
+//   return val;
+// }
+
+export const secretReplicationServiceFactory = ({
+  secretReplicationDAL,
+  secretDAL,
+  queueService,
+  secretVersionDAL,
+  secretImportDAL,
+  keyStore,
+  secretVersionTagDAL,
+  secretTagDAL,
+  folderDAL
+}: TSecretReplicationServiceFactoryDep) => {
+  queueService.start(QueueName.SecretReplication, async (job) => {
+    logger.info(job.data, "Replication started");
+    const { secrets, folderId, secretPath, environmentId, projectId } = job.data;
+    const secretImports = await secretImportDAL.find({
+      importPath: secretPath,
+      importEnv: environmentId,
+      isReplication: true
+    });
+    console.log(">>>> Secret Imports replics ", secretImports.length, secretPath, environmentId);
+    if (!secretImports.length) return;
+
+    // unfiltered secrets to be replicated
+    console.log(secrets.length);
+    const toBeReplicatedSecrets = await secretReplicationDAL.findSecrets({ folderId, secrets });
+    const replicatedSecrets = toBeReplicatedSecrets.filter(
+      ({ version, latestReplicatedVersion, secretBlindIndex }) =>
+        secretBlindIndex && (version === 1 || latestReplicatedVersion <= version)
+    );
+
+    const replicatedSecretsGroupBySecretId = groupBy(replicatedSecrets, (i) => i.secretId);
+    console.log("replicated ", replicatedSecretsGroupBySecretId);
+    const lock = await keyStore.acquireLock(
+      replicatedSecrets.map(({ id }) => id),
+      5000
+    );
+
+    try {
+      /*  eslint-disable no-await-in-loop */
+      for (const secretImport of secretImports) {
+        const importFolderId = secretImport.folderId;
+
+        const localSecrets = await secretDAL.find({
+          $in: { secretBlindIndex: replicatedSecrets.map(({ secretBlindIndex }) => secretBlindIndex) },
+          folderId: importFolderId,
+          isReplicated: true
+        });
+        const localSecretsGroupedByBlindIndex = groupBy(localSecrets, (i) => i.secretBlindIndex as string);
+
+        const locallyCreatedSecrets = secrets.filter(({ operation, id }) => {
+          return (
+            (operation === SecretReplicationOperations.Create || operation === SecretReplicationOperations.Update) &&
+            !localSecretsGroupedByBlindIndex[replicatedSecretsGroupBySecretId[id][0].secretBlindIndex as string]?.[0]
+          );
+        });
+
+        const locallyUpdatedSecrets = secrets.filter(
+          ({ operation, id }) =>
+            (operation === SecretReplicationOperations.Create || operation === SecretReplicationOperations.Update) &&
+            localSecretsGroupedByBlindIndex[replicatedSecretsGroupBySecretId[id][0].secretBlindIndex as string]?.[0]
+        );
+
+        console.log(replicatedSecretsGroupBySecretId);
+        console.log(locallyCreatedSecrets);
+        console.log("update", locallyUpdatedSecrets);
+        console.log("local board", localSecrets);
+
+        const locallyDeletedSecrets = secrets
+          .filter(
+            ({ operation, id }) =>
+              operation === SecretReplicationOperations.Delete &&
+              Boolean(replicatedSecretsGroupBySecretId[id]?.[0]?.secretBlindIndex) &&
+              localSecretsGroupedByBlindIndex[replicatedSecretsGroupBySecretId[id][0].secretBlindIndex as string]?.[0]
+          )
+          .map(
+            ({ id }) =>
+              localSecretsGroupedByBlindIndex[replicatedSecretsGroupBySecretId[id][0].secretBlindIndex as string][0]
+          );
+
+        let nestedImportSecrets: TSyncSecretReplicationDTO["secrets"] = [];
+        await secretReplicationDAL.transaction(async (tx) => {
+          if (locallyCreatedSecrets.length) {
+            const newSecrets = await fnSecretBulkInsert({
+              folderId: importFolderId,
+              secretVersionDAL,
+              secretDAL,
+              tx,
+              secretTagDAL,
+              secretVersionTagDAL,
+              inputSecrets: locallyCreatedSecrets.map(({ id }) => {
+                const doc = replicatedSecretsGroupBySecretId[id][0];
+                return {
+                  keyEncoding: doc.keyEncoding,
+                  algorithm: doc.algorithm,
+                  folderId,
+                  type: doc.type,
+                  metadata: doc.metadata,
+                  secretKeyIV: doc.secretKeyIV,
+                  secretKeyTag: doc.secretKeyTag,
+                  secretKeyCiphertext: doc.secretKeyCiphertext,
+                  secretValueIV: doc.secretValueIV,
+                  secretValueTag: doc.secretValueTag,
+                  secretValueCiphertext: doc.secretValueCiphertext,
+                  secretBlindIndex: doc.secretBlindIndex,
+                  secretCommentIV: doc.secretCommentIV,
+                  secretCommentTag: doc.secretCommentTag,
+                  secretCommentCiphertext: doc.secretCommentCiphertext,
+                  isReplicated: true,
+                  skipMultilineEncoding: doc.skipMultilineEncoding
+                };
+              })
+            });
+            nestedImportSecrets = nestedImportSecrets.concat(
+              ...newSecrets.map(({ id, version }) => ({ operation: SecretReplicationOperations.Create, version, id }))
+            );
+          }
+          if (locallyUpdatedSecrets.length) {
+            const newSecrets = await fnSecretBulkUpdate({
+              projectId,
+              folderId: importFolderId,
+              secretVersionDAL,
+              secretDAL,
+              tx,
+              secretTagDAL,
+              secretVersionTagDAL,
+              inputSecrets: locallyUpdatedSecrets.map(({ id }) => {
+                const doc = replicatedSecretsGroupBySecretId[id][0];
+                return {
+                  filter: {
+                    folderId: importFolderId,
+                    id: localSecretsGroupedByBlindIndex[doc.secretBlindIndex as string][0].id
+                  },
+                  data: {
+                    keyEncoding: doc.keyEncoding,
+                    algorithm: doc.algorithm,
+                    type: doc.type,
+                    metadata: doc.metadata,
+                    secretKeyIV: doc.secretKeyIV,
+                    secretKeyTag: doc.secretKeyTag,
+                    secretKeyCiphertext: doc.secretKeyCiphertext,
+                    secretValueIV: doc.secretValueIV,
+                    secretValueTag: doc.secretValueTag,
+                    secretValueCiphertext: doc.secretValueCiphertext,
+                    secretBlindIndex: doc.secretBlindIndex,
+                    secretCommentIV: doc.secretCommentIV,
+                    secretCommentTag: doc.secretCommentTag,
+                    secretCommentCiphertext: doc.secretCommentCiphertext,
+                    isReplicated: true,
+                    skipMultilineEncoding: doc.skipMultilineEncoding
+                  }
+                };
+              })
+            });
+            nestedImportSecrets = nestedImportSecrets.concat(
+              ...newSecrets.map(({ id, version }) => ({ operation: SecretReplicationOperations.Update, version, id }))
+            );
+          }
+          if (locallyDeletedSecrets.length) {
+            const newSecrets = await secretDAL.delete(
+              {
+                $in: {
+                  id: locallyDeletedSecrets.map(({ id }) => id)
+                },
+                isReplicated: true,
+                folderId: importFolderId
+              },
+              tx
+            );
+            nestedImportSecrets = nestedImportSecrets.concat(
+              ...newSecrets.map(({ id, version }) => ({ operation: SecretReplicationOperations.Delete, version, id }))
+            );
+          }
+        });
+        const [folder] = await folderDAL.findSecretPathByFolderIds(projectId, [secretImport.folderId]);
+        console.log("Environment ID -> slug", folder.envId, folder.environmentSlug);
+        await queueService.queue(QueueName.SecretReplication, QueueJobs.SecretReplication, {
+          folderId: folder.id,
+          projectId,
+          secrets: nestedImportSecrets,
+          secretPath: folder.path,
+          environmentId: folder.envId
+        });
+      }
+      await secretVersionDAL.update({ $in: { id: replicatedSecrets.map(({ id }) => id) } }, { isReplicated: true });
+      /*  eslint-enable no-await-in-loop */
+    } finally {
+      await lock.release();
+    }
+  });
+
+  queueService.listen(QueueName.SecretReplication, "failed", async (job, err) => {
+    logger.error(err, "Failed to replicate secret", job?.data);
+  });
+
+  const replicate = async (data: TSyncSecretReplicationDTO) => {
+    await queueService.queue(QueueName.SecretReplication, QueueJobs.SecretReplication, data, {
+      attempts: 3,
+      backoff: {
+        type: "exponential",
+        delay: 1000
+      },
+      removeOnComplete: true,
+      removeOnFail: true
+    });
+  };
+
+  return {
+    replicate
+  };
+};
@@ -0,0 +1,17 @@
+export enum SecretReplicationOperations {
+  Create = "create",
+  Update = "update",
+  Delete = "delete"
+}
+
+export type TSyncSecretReplicationDTO = {
+  secretPath: string;
+  projectId: string;
+  environmentId: string;
+  folderId: string;
+  secrets: {
+    operation: SecretReplicationOperations;
+    id: string;
+    version: number;
+  }[];
+};
@@ -75,7 +75,7 @@ export const secretDALFactory = (db: TDbClient) => {
  };

  const deleteMany = async (
-    data: Array<{ blindIndex: string; type: SecretType }>,
+    data: Array<{ blindIndex: string; type: SecretType; isReplicated?: boolean }>,
    folderId: string,
    userId: string,
    tx?: Knex
@@ -509,7 +509,7 @@ export const fnSecretBulkInsert = async ({
    }))
  );
  const secretVersions = await secretVersionDAL.insertMany(
-    inputSecrets.map(({ tags, references, ...el }) => ({
+    inputSecrets.map(({ tags, references, isReplicated, ...el }) => ({
      ...el,
      folderId,
      secretId: newSecretGroupByBlindIndex[el.secretBlindIndex as string][0].id
@@ -30,6 +30,8 @@ import { TSecretBlindIndexDALFactory } from "../secret-blind-index/secret-blind-
 import { TSecretFolderDALFactory } from "../secret-folder/secret-folder-dal";
 import { TSecretImportDALFactory } from "../secret-import/secret-import-dal";
 import { fnSecretsFromImports } from "../secret-import/secret-import-fns";
+import { TSecretReplicationServiceFactory } from "../secret-replication/secret-replication-service";
+import { SecretReplicationOperations } from "../secret-replication/secret-replication-types";
 import { TSecretTagDALFactory } from "../secret-tag/secret-tag-dal";
 import { TSecretDALFactory } from "./secret-dal";
 import {
@@ -85,6 +87,7 @@ type TSecretServiceFactoryDep = {
  projectBotService: Pick<TProjectBotServiceFactory, "getBotKey">;
  secretImportDAL: Pick<TSecretImportDALFactory, "find" | "findByFolderIds">;
  secretVersionTagDAL: Pick<TSecretVersionTagDALFactory, "insertMany">;
+  secretReplicationService: Pick<TSecretReplicationServiceFactory, "replicate">;
 };

 export type TSecretServiceFactory = ReturnType<typeof secretServiceFactory>;
@@ -101,7 +104,8 @@ export const secretServiceFactory = ({
  projectDAL,
  projectBotService,
  secretImportDAL,
-  secretVersionTagDAL
+  secretVersionTagDAL,
+  secretReplicationService
 }: TSecretServiceFactoryDep) => {
  const getSecretReference = async (projectId: string) => {
    // if bot key missing means e2e still exist
@@ -285,6 +289,19 @@ export const secretServiceFactory = ({
    await snapshotService.performSnapshot(folderId);
    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
    // TODO(akhilmhdh-pg): licence check, posthog service and snapshot
+    await secretReplicationService.replicate({
+      folderId,
+      projectId,
+      environmentId: folder.envId,
+      secretPath: path,
+      secrets: [
+        {
+          operation: SecretReplicationOperations.Create,
+          id: secret[0].id,
+          version: 1
+        }
+      ]
+    });
    return { ...secret[0], environment, workspace: projectId, tags, secretPath: path };
  };

@@ -414,7 +431,19 @@ export const secretServiceFactory = ({

    await snapshotService.performSnapshot(folderId);
    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
-    // TODO(akhilmhdh-pg): licence check, posthog service and snapshot
+    await secretReplicationService.replicate({
+      folderId,
+      projectId,
+      environmentId: folder.envId,
+      secretPath: path,
+      secrets: [
+        {
+          operation: SecretReplicationOperations.Update,
+          id: updatedSecret[0].id,
+          version: updatedSecret[0].version
+        }
+      ]
+    });
    return { ...updatedSecret[0], workspace: projectId, environment, secretPath: path };
  };

@@ -482,7 +511,19 @@ export const secretServiceFactory = ({

    await snapshotService.performSnapshot(folderId);
    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
-
+    await secretReplicationService.replicate({
+      folderId,
+      projectId,
+      environmentId: folder.envId,
+      secretPath: path,
+      secrets: [
+        {
+          operation: SecretReplicationOperations.Delete,
+          id: deletedSecret[0].id,
+          version: deletedSecret[0].version
+        }
+      ]
+    });
    // TODO(akhilmhdh-pg): licence check, posthog service and snapshot
    return { ...deletedSecret[0], _id: deletedSecret[0].id, workspace: projectId, environment, secretPath: path };
  };
@@ -768,6 +809,13 @@ export const secretServiceFactory = ({

    await snapshotService.performSnapshot(folderId);
    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
+    await secretReplicationService.replicate({
+      folderId,
+      projectId,
+      environmentId: folder.envId,
+      secretPath: path,
+      secrets: newSecrets.map(({ id, version }) => ({ id, version, operation: SecretReplicationOperations.Create }))
+    });

    return newSecrets;
  };
@@ -868,6 +916,13 @@ export const secretServiceFactory = ({

    await snapshotService.performSnapshot(folderId);
    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
+    await secretReplicationService.replicate({
+      folderId,
+      projectId,
+      environmentId: folder.envId,
+      secretPath: path,
+      secrets: secrets.map(({ id, version }) => ({ id, version, operation: SecretReplicationOperations.Update }))
+    });

    return secrets;
  };
@@ -930,6 +985,13 @@ export const secretServiceFactory = ({

    await snapshotService.performSnapshot(folderId);
    await secretQueueService.syncSecrets({ secretPath: path, projectId, environment });
+    await secretReplicationService.replicate({
+      folderId,
+      projectId,
+      environmentId: folder.envId,
+      secretPath: path,
+      secrets: secretsDeleted.map(({ id, version }) => ({ id, version, operation: SecretReplicationOperations.Delete }))
+    });

    return secretsDeleted;
  };