Connected connectors can be deleted and reconnected; mirror sweeps stale rows and ops serialize per org

src/main/services/composio/composio-client.ts

@@ -93,19 +93,34 @@ export class ComposioClient {
   }
 
   async listConnectedAccounts(userId: string): Promise<ComposioConnectedAccount[]> {
-    const params = new URLSearchParams({ user_ids: userId, limit: "50" });
-    const response = await this.request("GET", `${BASE_URL}/connected_accounts?${params.toString()}`);
-    if (!Array.isArray(response.items)) return [];
-    return (response.items as Array<Record<string, unknown>>).map((item) => {
-      const toolkit = (item.toolkit ?? {}) as Record<string, unknown>;
-      const authConfig = (item.auth_config ?? {}) as Record<string, unknown>;
-      return {
-        id: String(item.id ?? ""),
-        toolkitSlug: String(toolkit.slug ?? "").toLowerCase(),
-        status: String(item.status ?? ""),
-        authConfigId: String(authConfig.id ?? ""),
-        createdAt: String(item.created_at ?? ""),
-      };
+    const accounts: ComposioConnectedAccount[] = [];
+    let cursor: string | null = null;
+    do {
+      const params = new URLSearchParams({ user_ids: userId, limit: "100" });
+      if (cursor) params.set("cursor", cursor);
+      const response = await this.request("GET", `${BASE_URL}/connected_accounts?${params.toString()}`);
+      const items = Array.isArray(response.items) ? response.items as Array<Record<string, unknown>> : [];
+      for (const item of items) {
+        const toolkit = (item.toolkit ?? {}) as Record<string, unknown>;
+        const authConfig = (item.auth_config ?? {}) as Record<string, unknown>;
+        accounts.push({
+          id: String(item.id ?? ""),
+          toolkitSlug: String(toolkit.slug ?? "").toLowerCase(),
+          status: String(item.status ?? ""),
+          authConfigId: String(authConfig.id ?? ""),
+          createdAt: String(item.created_at ?? ""),
+        });
+      }
+      cursor = typeof response.next_cursor === "string" && response.next_cursor ? response.next_cursor : null;
+      if (items.length === 0) break;
+    } while (cursor);
+    return accounts;
+  }
+
+  /** Idempotent: an account already gone in Composio (404) counts as deleted. */
+  async deleteConnectedAccount(accountId: string): Promise<void> {
+    await this.request("DELETE", `${BASE_URL}/connected_accounts/${encodeURIComponent(accountId)}`, undefined, {
+      tolerateNotFound: true,
     });
   }
 
@@ -128,7 +143,12 @@ export class ComposioClient {
     };
   }
 
-  private async request(method: string, url: string, body?: Record<string, unknown>): Promise<Record<string, unknown>> {
+  private async request(
+    method: string,
+    url: string,
+    body?: Record<string, unknown>,
+    options?: { tolerateNotFound?: boolean },
+  ): Promise<Record<string, unknown>> {
     return new Promise((resolve, reject) => {
       const request = net.request({ method, url });
       request.setHeader("x-api-key", this.apiKey);
@@ -139,6 +159,10 @@ export class ComposioClient {
       request.on("response", (response) => {
         response.on("data", (chunk) => { responseData += chunk.toString(); });
         response.on("end", () => {
+          if (response.statusCode === 404 && options?.tolerateNotFound) {
+            resolve({});
+            return;
+          }
           if (response.statusCode && response.statusCode >= 400) {
             reject(new Error(`Composio API ${method} ${url} returned ${response.statusCode}: ${responseData.slice(0, 200)}`));
             return;

src/main/services/composio/composio-connection-service.ts

@@ -17,8 +17,14 @@ let catalogCache: { fetchedAt: number; entries: ConnectorCatalogEntry[] } | null
  * accounts against them. One connected account per toolkit per organization,
  * mirrored into connector_accounts so agent runs and the Connectors page
  * read local state instead of calling out on every frame.
+ *
+ * Mutations and refreshes are serialized per organization: a refresh whose
+ * server read predates a delete or connect would otherwise land afterwards
+ * and resurrect (or sweep) rows from its stale snapshot.
  */
 export class ComposioConnectionService {
+  private static queues = new Map<string, Promise<unknown>>();
+
   constructor(private readonly db: DatabaseSync) {}
 
   listCatalog(): Promise<ConnectorCatalogEntry[]> {
@@ -29,14 +35,32 @@ export class ComposioConnectionService {
    * Starts (or restarts) the company's connection for a toolkit: initiates a
    * connected account against the toolkit's configured auth config via the
    * link flow and records the pending account locally as INITIALIZING — the
-   * poll that follows adopts Composio's real status. The returned redirect
-   * URL completes authorization (or API-key entry) in the browser.
+   * poll that follows adopts Composio's real status. Any previous account
+   * for the toolkit is deleted first so reconnecting replaces instead of
+   * accumulating accounts in Composio. The returned redirect URL completes
+   * authorization (or API-key entry) in the browser.
    */
-  async connectToolkit(orgId: string, toolkitSlug: string): Promise<ConnectComposioToolkitResult> {
+  connectToolkit(orgId: string, toolkitSlug: string): Promise<ConnectComposioToolkitResult> {
+    return this.serialize(orgId, () => this.connectToolkitNow(orgId, toolkitSlug));
+  }
+
+  /** See class docs: linearizes connect/refresh/disconnect for one org. */
+  private serialize<T>(orgId: string, operation: () => Promise<T>): Promise<T> {
+    const tail = ComposioConnectionService.queues.get(orgId) ?? Promise.resolve();
+    const next = tail.then(operation, operation);
+    ComposioConnectionService.queues.set(orgId, next.then(() => undefined, () => undefined));
+    return next;
+  }
+
+  private async connectToolkitNow(orgId: string, toolkitSlug: string): Promise<ConnectComposioToolkitResult> {
     const slug = normalizeSlug(toolkitSlug);
     const client = this.requireClient();
     const entry = (await this.catalog(client)).find((item) => item.slug === slug);
     if (!entry) throw new Error(`No auth config for "${slug}" exists in the Composio project.`);
+    const previous = this.db.prepare(
+      "select id from connector_accounts where org_id = ? and toolkit_slug = ?",
+    ).get(orgId, slug) as { id: string } | undefined;
+    if (previous) await client.deleteConnectedAccount(previous.id);
     const account = await client.createConnectedAccount({
       authConfigId: entry.authConfigId,
       userId: getComposioUserId(this.db, orgId),
@@ -56,11 +80,21 @@ export class ComposioConnectionService {
     return { accountId: account.id, redirectUrl: account.redirectUrl };
   }
 
-  /** Pulls account statuses from Composio and updates the local mirror. */
-  async refreshAccounts(orgId: string): Promise<void> {
+  /**
+   * Pulls account statuses from Composio and updates the local mirror, in
+   * both directions: statuses adopt Composio's truth, and local rows whose
+   * account no longer exists in Composio are removed along with their
+   * grants — deleting an account in the dashboard cleans up here too.
+   */
+  refreshAccounts(orgId: string): Promise<void> {
+    return this.serialize(orgId, () => this.refreshAccountsNow(orgId));
+  }
+
+  private async refreshAccountsNow(orgId: string): Promise<void> {
     const client = this.requireClient();
     const accounts = await client.listConnectedAccounts(getComposioUserId(this.db, orgId));
     const now = new Date().toISOString();
+    const liveIds = new Set(accounts.map((account) => account.id));
     for (const account of accounts) {
       if (!account.toolkitSlug) continue;
       this.db.prepare(`
@@ -82,6 +116,36 @@ export class ComposioConnectionService {
         now,
       );
     }
+    const localRows = this.db.prepare("select id, toolkit_slug from connector_accounts where org_id = ?")
+      .all(orgId) as Array<{ id: string; toolkit_slug: string }>;
+    for (const row of localRows) {
+      if (liveIds.has(row.id)) continue;
+      this.removeLocalAccount(orgId, row.toolkit_slug);
+    }
+  }
+
+  /**
+   * Deletes the company's connection for a toolkit: removes the connected
+   * account in Composio (already-gone counts as deleted), then the local
+   * mirror row and every agent grant that pointed at it.
+   */
+  disconnectToolkit(orgId: string, toolkitSlug: string): Promise<void> {
+    return this.serialize(orgId, () => this.disconnectToolkitNow(orgId, toolkitSlug));
+  }
+
+  private async disconnectToolkitNow(orgId: string, toolkitSlug: string): Promise<void> {
+    const slug = normalizeSlug(toolkitSlug);
+    const row = this.db.prepare(
+      "select id from connector_accounts where org_id = ? and toolkit_slug = ?",
+    ).get(orgId, slug) as { id: string } | undefined;
+    if (!row) return;
+    await this.requireClient().deleteConnectedAccount(row.id);
+    this.removeLocalAccount(orgId, slug);
+  }
+
+  private removeLocalAccount(orgId: string, slug: string): void {
+    this.db.prepare("delete from connector_accounts where org_id = ? and toolkit_slug = ?").run(orgId, slug);
+    this.db.prepare("delete from agent_connector_grants where org_id = ? and toolkit_slug = ?").run(orgId, slug);
   }
 
   private requireClient(): ComposioClient {