feat: failed integration sync emails debouncer

backend/src/keystore/keystore.ts

@@ -22,12 +22,15 @@ export const KeyStorePrefixes = {
     `sync-integration-last-run-${projectId}-${environmentSlug}-${secretPath}` as const,
   IdentityAccessTokenStatusUpdate: (identityAccessTokenId: string) =>
     `identity-access-token-status:${identityAccessTokenId}`,
-  ServiceTokenStatusUpdate: (serviceTokenId: string) => `service-token-status:${serviceTokenId}`
+  ServiceTokenStatusUpdate: (serviceTokenId: string) => `service-token-status:${serviceTokenId}`,
+  SendFailedIntegrationSyncEmails: (projectId: string, secretPath: string, environmentSlug: string) =>
+    `send-failed-integration-sync-emails-${projectId}-${secretPath}-${environmentSlug}` as const
 };
 
 export const KeyStoreTtls = {
   SetSyncSecretIntegrationLastRunTimestampInSeconds: 10,
-  AccessTokenStatusUpdateInSeconds: 120
+  AccessTokenStatusUpdateInSeconds: 120,
+  SendFailedIntegrationSyncEmailsInSeconds: 60
 };
 
 type TWaitTillReady = {

backend/src/queue/queue-service.ts

@@ -7,7 +7,7 @@ import {
   TScanFullRepoEventPayload,
   TScanPushEventPayload
 } from "@app/ee/services/secret-scanning/secret-scanning-queue/secret-scanning-queue-types";
-import { TSyncSecretsDTO } from "@app/services/secret/secret-types";
+import { TIntegrationSyncPayload, TSyncSecretsDTO } from "@app/services/secret/secret-types";
 
 export enum QueueName {
   SecretRotation = "secret-rotation",
@@ -42,6 +42,7 @@ export enum QueueJobs {
   SecWebhook = "secret-webhook-trigger",
   TelemetryInstanceStats = "telemetry-self-hosted-stats",
   IntegrationSync = "secret-integration-pull",
+  SendFailedIntegrationSyncEmails = "send-failed-integration-sync-emails",
   SecretScan = "secret-scan",
   UpgradeProjectToGhost = "upgrade-project-to-ghost-job",
   DynamicSecretRevocation = "dynamic-secret-revocation",
@@ -88,16 +89,28 @@ export type TQueueJobTypes = {
     name: QueueJobs.SecWebhook;
     payload: { projectId: string; environment: string; secretPath: string; depth?: number };
   };
-  [QueueName.IntegrationSync]: {
+
+  [QueueName.AccessTokenStatusUpdate]:
+    | {
+        name: QueueJobs.IdentityAccessTokenStatusUpdate;
+        payload: { identityAccessTokenId: string; numberOfUses: number };
+      }
+    | {
+        name: QueueJobs.ServiceTokenStatusUpdate;
+        payload: { serviceTokenId: string };
+      };
+
+  [QueueName.IntegrationSync]:
+    | {
         name: QueueJobs.IntegrationSync;
+        payload: TIntegrationSyncPayload;
+      }
+    | {
+        name: QueueJobs.SendFailedIntegrationSyncEmails;
         payload: {
-      isManual?: boolean;
-      actorId?: string;
           projectId: string;
-      environment: string;
+          environmentSlug: string;
           secretPath: string;
-      depth?: number;
-      deDupeQueue?: Record<string, boolean>;
         };
       };
   [QueueName.SecretFullRepoScan]: {
@@ -153,15 +166,6 @@ export type TQueueJobTypes = {
     name: QueueJobs.ProjectV3Migration;
     payload: { projectId: string };
   };
-  [QueueName.AccessTokenStatusUpdate]:
-    | {
-        name: QueueJobs.IdentityAccessTokenStatusUpdate;
-        payload: { identityAccessTokenId: string; numberOfUses: number };
-      }
-    | {
-        name: QueueJobs.ServiceTokenStatusUpdate;
-        payload: { serviceTokenId: string };
-      };
 };
 
 export type TQueueServiceFactory = ReturnType<typeof queueServiceFactory>;

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

@@ -17,7 +17,7 @@ import { TSnapshotSecretV2DALFactory } from "@app/ee/services/secret-snapshot/sn
 import { KeyStorePrefixes, KeyStoreTtls, TKeyStoreFactory } from "@app/keystore/keystore";
 import { getConfig } from "@app/lib/config/env";
 import { decryptSymmetric128BitHexKeyUTF8 } from "@app/lib/crypto";
-import { daysToMillisecond, secondsToMillis } from "@app/lib/dates";
+import { applyJitter, daysToMillisecond, secondsToMillis } from "@app/lib/dates";
 import { BadRequestError } from "@app/lib/errors";
 import { getTimeDifferenceInSeconds, groupBy, isSamePath, unique } from "@app/lib/fn";
 import { logger } from "@app/lib/logger";
@@ -55,8 +55,11 @@ import { fnTriggerWebhook } from "../webhook/webhook-fns";
 import { TSecretDALFactory } from "./secret-dal";
 import { interpolateSecrets } from "./secret-fns";
 import {
+  FailedIntegrationSyncEmailsPayloadSchema,
   TCreateSecretReminderDTO,
+  TFailedIntegrationSyncEmailsPayload,
   THandleReminderDTO,
+  TIntegrationSyncPayload,
   TRemoveSecretReminderDTO,
   TSyncSecretsDTO
 } from "./secret-types";
@@ -515,6 +518,38 @@ export const secretQueueFactory = ({
     );
   };
 
+  const sendFailedIntegrationSyncEmails = async (payload: TFailedIntegrationSyncEmailsPayload) => {
+    const key = KeyStorePrefixes.SendFailedIntegrationSyncEmails(
+      payload.projectId,
+      payload.secretPath,
+      payload.environmentSlug
+    );
+
+    await keyStore.setItemWithExpiry(
+      key,
+      KeyStoreTtls.SendFailedIntegrationSyncEmailsInSeconds,
+      JSON.stringify(payload)
+    );
+    await queueService.queue(
+      QueueName.IntegrationSync,
+      QueueJobs.SendFailedIntegrationSyncEmails,
+      {
+        secretPath: payload.secretPath,
+        projectId: payload.projectId,
+        environmentSlug: payload.environmentSlug
+      },
+      {
+        delay: applyJitter(
+          secondsToMillis(KeyStoreTtls.SendFailedIntegrationSyncEmailsInSeconds / 2),
+          secondsToMillis(10)
+        ),
+        jobId: key,
+        removeOnFail: true,
+        removeOnComplete: true
+      }
+    );
+  };
+
   queueService.start(QueueName.SecretSync, async (job) => {
     const {
       _deDupeQueue: deDupeQueue,
@@ -560,42 +595,72 @@ export const secretQueueFactory = ({
   });
 
   queueService.start(QueueName.IntegrationSync, async (job) => {
-    const { environment, actorId, isManual, projectId, secretPath, depth = 1, deDupeQueue = {} } = job.data;
+    if (job.name === QueueJobs.SendFailedIntegrationSyncEmails) {
-    if (depth > MAX_SYNC_SECRET_DEPTH) return;
-
-    const folder = await folderDAL.findBySecretPath(projectId, environment, secretPath);
-    if (!folder) {
-      throw new Error("Secret path not found");
-    }
-
-    const sendIntegrationSyncFailedMail = async (integrations: { integrationId: string; syncMessage?: string }[]) => {
       const appCfg = getConfig();
 
-      // If smtp is not configured, we can return early without having to fetch the project members.
+      // If smtp is not configured, we can return early
       if (!appCfg.isSmtpConfigured) return;
 
-      const projectMembers = await projectMembershipDAL.findAllProjectMembers(projectId);
+      const jobPayload = job.data as Pick<
-      const project = await projectDAL.findById(projectId);
+        TFailedIntegrationSyncEmailsPayload,
+        "projectId" | "secretPath" | "environmentSlug"
+      >;
+
+      const failedIntegrationsDetails = await keyStore.getItem(
+        KeyStorePrefixes.SendFailedIntegrationSyncEmails(
+          jobPayload.projectId,
+          jobPayload.secretPath,
+          jobPayload.environmentSlug
+        )
+      );
+
+      if (!failedIntegrationsDetails) return;
+
+      const failedSyncKeyStore = FailedIntegrationSyncEmailsPayloadSchema.parse(JSON.parse(failedIntegrationsDetails));
+
+      const projectMembers = await projectMembershipDAL.findAllProjectMembers(failedSyncKeyStore.projectId);
+      const project = await projectDAL.findById(failedSyncKeyStore.projectId);
 
       // Only send emails to admins, and if its a manual trigger, only send it to the person who triggered it (if actor is admin as well)
       const filteredProjectMembers = projectMembers
         .filter((member) => member.roles.some((role) => role.role === ProjectMembershipRole.Admin))
-        .filter((member) => (isManual && actorId ? member.userId === actorId : true));
+        .filter((member) =>
+          failedSyncKeyStore.manuallyTriggeredByUserId
+            ? member.userId === failedSyncKeyStore.manuallyTriggeredByUserId
+            : true
+        );
 
       await smtpService.sendMail({
         recipients: filteredProjectMembers.map((member) => member.user.email!),
         template: SmtpTemplates.IntegrationSyncFailed,
         subjectLine: `Integration Sync Failed`,
         substitutions: {
-          syncMessage: integrations[0]?.syncMessage, // We are only displaying the sync message if its a singular integration, so we can just grab the first one in the array.
+          syncMessage: failedSyncKeyStore.count === 1 ? failedSyncKeyStore.syncMessage : undefined, // We are only displaying the sync message if its a singular integration, so we can just grab the first one in the array.
-          secretPath,
+          secretPath: failedSyncKeyStore.secretPath,
-          environment: folder.environment.name,
+          environment: failedSyncKeyStore.environmentName,
-          count: integrations.length,
+          count: failedSyncKeyStore.count,
           projectName: project.name,
           integrationUrl: `${appCfg.SITE_URL}/integrations/${project.id}`
         }
       });
-    };
+    }
+
+    if (job.name === QueueJobs.IntegrationSync) {
+      const {
+        environment,
+        actorId,
+        isManual,
+        projectId,
+        secretPath,
+        depth = 1,
+        deDupeQueue = {}
+      } = job.data as TIntegrationSyncPayload;
+      if (depth > MAX_SYNC_SECRET_DEPTH) return;
+
+      const folder = await folderDAL.findBySecretPath(projectId, environment, secretPath);
+      if (!folder) {
+        throw new Error("Secret path not found");
+      }
 
       // find all imports made with the given environment and secret path
       const linkSourceDto = {
@@ -612,7 +677,7 @@ export const secretQueueFactory = ({
         const importedFolders = await folderDAL.findSecretPathByFolderIds(projectId, importedFolderIds);
         const foldersGroupedById = groupBy(importedFolders.filter(Boolean), (i) => i?.id as string);
         logger.info(
-        `getIntegrationSecrets: Syncing secret due to link change [jobId=${job.id}] [projectId=${job.data.projectId}] [environment=${job.data.environment}]  [secretPath=${job.data.secretPath}] [depth=${depth}]`
+          `getIntegrationSecrets: Syncing secret due to link change [jobId=${job.id}] [projectId=${job.data.projectId}] [environment=${environment}]  [secretPath=${job.data.secretPath}] [depth=${depth}]`
         );
         await Promise.all(
           imports
@@ -664,7 +729,7 @@ export const secretQueueFactory = ({
         const referencedFolders = await folderDAL.findSecretPathByFolderIds(projectId, referencedFolderIds);
         const referencedFoldersGroupedById = groupBy(referencedFolders.filter(Boolean), (i) => i?.id as string);
         logger.info(
-        `getIntegrationSecrets: Syncing secret due to reference change [jobId=${job.id}] [projectId=${job.data.projectId}] [environment=${job.data.environment}]  [secretPath=${job.data.secretPath}] [depth=${depth}]`
+          `getIntegrationSecrets: Syncing secret due to reference change [jobId=${job.id}] [projectId=${job.data.projectId}] [environment=${environment}]  [secretPath=${job.data.secretPath}] [depth=${depth}]`
         );
         await Promise.all(
           referencedFolderIds
@@ -698,11 +763,11 @@ export const secretQueueFactory = ({
         ({ secretPath: integrationSecPath, isActive }) => isActive && isSamePath(secretPath, integrationSecPath)
       );
 
-    const integrationsFailedToSync: { integrationId: string }[] = [];
+      const integrationsFailedToSync: { integrationId: string; syncMessage?: string }[] = [];
 
       if (!integrations.length) return;
       logger.info(
-      `getIntegrationSecrets: secret integration sync started [jobId=${job.id}] [jobId=${job.id}] [projectId=${job.data.projectId}] [environment=${job.data.environment}]  [secretPath=${job.data.secretPath}] [depth=${job.data.depth}]`
+        `getIntegrationSecrets: secret integration sync started [jobId=${job.id}] [jobId=${job.id}] [projectId=${job.data.projectId}] [environment=${environment}]  [secretPath=${job.data.secretPath}] [depth=${depth}]`
       );
 
       const lock = await keyStore.acquireLock(
@@ -725,7 +790,7 @@ export const secretQueueFactory = ({
         const isStaleSyncIntegration = new Date(job.timestamp) < new Date(lastRunSyncIntegrationTimestamp);
         if (isStaleSyncIntegration) {
           logger.info(
-          `getIntegrationSecrets: secret integration sync stale [jobId=${job.id}] [jobId=${job.id}] [projectId=${job.data.projectId}] [environment=${job.data.environment}]  [secretPath=${job.data.secretPath}] [depth=${job.data.depth}]`
+            `getIntegrationSecrets: secret integration sync stale [jobId=${job.id}] [jobId=${job.id}] [projectId=${job.data.projectId}] [environment=${environment}]  [secretPath=${job.data.secretPath}] [depth=${depth}]`
           );
           return;
         }
@@ -874,13 +939,14 @@ export const secretQueueFactory = ({
             // May be undefined, if it's undefined we assume the sync was successful, hence the strict equality type check.
             if (response?.isSynced === false) {
               integrationsFailedToSync.push({
-              integrationId: integration.id
+                integrationId: integration.id,
+                syncMessage: response.syncMessage
               });
             }
           } catch (err) {
             logger.error(
               err,
-            `Secret integration sync error [projectId=${job.data.projectId}] [environment=${job.data.environment}]  [secretPath=${job.data.secretPath}]`
+              `Secret integration sync error [projectId=${job.data.projectId}] [environment=${environment}]  [secretPath=${job.data.secretPath}]`
             );
 
             const message =
@@ -909,14 +975,27 @@ export const secretQueueFactory = ({
             });
 
             integrationsFailedToSync.push({
-            integrationId: integration.id
+              integrationId: integration.id,
+              syncMessage: message
             });
           }
         }
       } finally {
         await lock.release();
-
+        if (integrationsFailedToSync.length) {
-      await sendIntegrationSyncFailedMail(integrationsFailedToSync);
+          await sendFailedIntegrationSyncEmails({
+            count: integrationsFailedToSync.length,
+            environmentName: folder.environment.name,
+            environmentSlug: environment,
+            ...(isManual &&
+              actorId && {
+                manuallyTriggeredByUserId: actorId
+              }),
+            projectId,
+            secretPath,
+            syncMessage: integrationsFailedToSync[0].syncMessage
+          });
+        }
       }
 
       await keyStore.setItemWithExpiry(
@@ -925,6 +1004,7 @@ export const secretQueueFactory = ({
         lockAcquiredTime.toISOString()
       );
       logger.info("Secret integration sync ended: %s", job.id);
+    }
   });
 
   queueService.start(QueueName.SecretReminder, async ({ data }) => {

backend/src/services/secret/secret-types.ts

@@ -1,4 +1,5 @@
 import { Knex } from "knex";
+import { z } from "zod";
 
 import { SecretType, TSecretBlindIndexes, TSecrets, TSecretsInsert, TSecretsUpdate } from "@app/db/schemas";
 import { TProjectPermission } from "@app/lib/types";
@@ -21,6 +22,29 @@ type TPartialSecret = Pick<TSecrets, "id" | "secretReminderRepeatDays" | "secret
 
 type TPartialInputSecret = Pick<TSecrets, "type" | "secretReminderNote" | "secretReminderRepeatDays" | "id">;
 
+export const FailedIntegrationSyncEmailsPayloadSchema = z.object({
+  projectId: z.string(),
+  secretPath: z.string(),
+  environmentName: z.string(),
+  environmentSlug: z.string(),
+
+  count: z.number(),
+  syncMessage: z.string().optional(),
+  manuallyTriggeredByUserId: z.string().optional()
+});
+
+export type TFailedIntegrationSyncEmailsPayload = z.infer<typeof FailedIntegrationSyncEmailsPayloadSchema>;
+
+export type TIntegrationSyncPayload = {
+  isManual?: boolean;
+  actorId?: string;
+  projectId: string;
+  environment: string;
+  secretPath: string;
+  depth?: number;
+  deDupeQueue?: Record<string, boolean>;
+};
+
 export type TCreateSecretDTO = {
   secretName: string;
   path: string;