fix: Correctly cleanup existing slack integrations

control-plane/src/modules/integrations/integrations.ts

@@ -9,19 +9,19 @@ import { valtown } from "./valtown";
 import { slack } from "./slack";
 import { InstallableIntegration } from "./types";
 
-const installableIntegrations: Record<string, InstallableIntegration> = {
+const installables: Record<string, InstallableIntegration> = {
   [toolhouseIntegration]: toolhouse,
   [tavilyIntegration]: tavily,
   [valtownIntegration]: valtown,
   [slackIntegration]: slack,
 };
 
 export function getInstallables(tool: string) {
-  if (!installableIntegrations[tool as keyof typeof installableIntegrations]) {
+  if (!installables[tool as keyof typeof installables]) {
     throw new Error(`Unknown tool provider integration requested: ${tool}`);
   }
 
-  return installableIntegrations[tool as keyof typeof installableIntegrations];
+  return installables[tool as keyof typeof installables];
 }
 
 export const getIntegrations = async ({
@@ -78,7 +78,7 @@ export const upsertIntegrations = async ({
 
   await Promise.all(
     Object.entries(config)
-      .filter(([key]) => installableIntegrations[key as keyof typeof installableIntegrations])
+      .filter(([key]) => installables[key as keyof typeof installables])
       .map(([key, value]) => {
       if (value) {
         return getInstallables(key)?.onActivate?.(clusterId, config);

control-plane/src/modules/integrations/slack/index.ts

@@ -13,7 +13,6 @@ import { nango } from "../nango";
 import { InstallableIntegration } from "../types";
 import { integrationSchema } from "../schema";
 import { z } from "zod";
-import { upsertIntegrations } from "../integrations";
 
 let app: App | undefined;
 
@@ -36,12 +35,13 @@ export const slack: InstallableIntegration = {
     logger.info("Deactivating Slack integration", {
       clusterId
     })
+
     if (!prevConfig.slack) {
       logger.warn("Can not deactivate Slack integration with no config")
       return
     }
     // Cleanup the Nango connection
-    await deleteNangoConnection(clusterId, prevConfig.slack.nangoConnectionId);
+    await deleteNangoConnection(prevConfig.slack.nangoConnectionId);
   },
   onActivate: async (clusterId: string, config: z.infer<typeof integrationSchema>) => {
     logger.info("Activating Slack integration", {
@@ -54,30 +54,7 @@ export const slack: InstallableIntegration = {
     }
 
     // If another cluster is connected with this teamId, remove it
-    const [conflict] = await db.select({
-      cluster_id: integrations.cluster_id,
-      slack: integrations.slack,
-    })
-      .from(integrations)
-      .where(
-        and(
-          sql`slack->>'teamId' = ${config.slack.teamId}`,
-          ne(integrations.cluster_id, clusterId)
-        ));
-
-    if (conflict) {
-      logger.info("Removing conflicting Slack integration", {
-        teamId: config.slack.teamId,
-        conflictClusterId: conflict.cluster_id,
-      });
-
-      await upsertIntegrations({
-        clusterId: conflict.cluster_id,
-        config: {
-          slack: null,
-        }
-      })
-    }
+    await cleanupConflictingConnections(clusterId, config);
   },
   handleCall: async () => {
     logger.warn("Slack integration does not support calls");
@@ -297,16 +274,58 @@ const getAccessToken = async (connectionId: string) => {
   return result;
 };
 
-const deleteNangoConnection = async (clusterId: string, connectionId: string) => {
+const cleanupConflictingConnections = async (clusterId: string, config: z.infer<typeof integrationSchema>) => {
+  if (!config.slack) {
+    return;
+  }
+
+  const conflicts = await db
+    .select({
+      cluster_id: integrations.cluster_id,
+      slack: integrations.slack,
+    })
+    .from(integrations)
+    .where(
+      and(
+        sql`slack->>'teamId' = ${config.slack.teamId}`,
+        ne(integrations.cluster_id, clusterId)
+      )
+    );
+
+  // Cleanup Nango connections
+  await Promise.all(conflicts.map((conflict) => {
+    if (conflict.slack) {
+      logger.info("Removing conflicting Slack Nango connections", {
+        clusterId: conflict.cluster_id,
+        connectionId: conflict.slack.nangoConnectionId
+      })
+      return deleteNangoConnection(conflict.slack.nangoConnectionId);
+    }
+  }));
+
+  if (conflicts.length) {
+    logger.info("Removed conflicting Slack integrations from DB", {
+      conflicts: conflicts.map((conflict) => conflict.cluster_id)
+    })
+
+    // Cleanup Slack integrations
+    await db
+    .delete(integrations)
+    .where(
+      and(
+        sql`slack->>'teamId' = ${config.slack.teamId}`,
+        ne(integrations.cluster_id, clusterId)
+      )
+    );
+
+  }
+}
+
+const deleteNangoConnection = async (connectionId: string) => {
   if (!nango) {
     throw new Error("Nango is not configured");
   }
 
-  logger.info("Removing Slack integration", {
-    connectionId,
-    clusterId
-  });
-
   await nango.deleteConnection(
     env.NANGO_SLACK_INTEGRATION_ID,
     connectionId