fix(db): recover failed credential group index on migration retry (#7595)

* fix(db): recover failed credential group index on migration retry

* fix(test): compare connector identifiers exactly

Author: TheodoreSpeaks

apps/sim/lib/knowledge/orchestration/connectors.test.ts

@@ -1396,9 +1396,10 @@ describe('members-mode connectors', () => {
     expect(mockDispatchMemberSync).not.toHaveBeenCalled()
     expect(mockRecordAudit).not.toHaveBeenCalled()
     expect(mockRevoke).toHaveBeenCalledWith(
-      expect.objectContaining({ connectorId: expect.not.stringMatching(MEMBERS_CONNECTOR.id) }),
+      expect.objectContaining({ connectorId: expect.any(String) }),
       ACTOR.userId
     )
+    expect(mockRevoke.mock.calls[0][0].connectorId).not.toBe(MEMBERS_CONNECTOR.id)
   })
 
   it('does not reuse matching settings bound to a different account option', async () => {

packages/db/migrations/0326_enterprise_organization_search.sql

@@ -384,6 +384,31 @@ CREATE INDEX CONCURRENTLY IF NOT EXISTS "credential_group_organization_id_idx" O
 --> statement-breakpoint
 CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "credential_group_organization_unique" ON "credential_group" USING btree ("organization_id");
 --> statement-breakpoint
+-- A failed concurrent build leaves an INVALID index that IF NOT EXISTS skips.
+-- Rename only that failed index so it can be dropped concurrently outside this block.
+-- The recovery name also survives interruption between the rename and drop.
+DO $$
+BEGIN
+  IF EXISTS (
+    SELECT 1 FROM pg_index
+    WHERE indexrelid = to_regclass('"public"."credential_group_workspace_unique_failed_0326"')
+      AND (indisvalid OR indrelid <> '"public"."credential_group"'::regclass)
+  ) THEN
+    RAISE EXCEPTION 'Refusing to drop unexpected index credential_group_workspace_unique_failed_0326';
+  END IF;
+  IF EXISTS (
+    SELECT 1 FROM pg_index
+    WHERE indexrelid = to_regclass('"public"."credential_group_workspace_unique"')
+      AND indrelid = '"public"."credential_group"'::regclass
+      AND NOT indisvalid
+  ) THEN
+    ALTER INDEX "public"."credential_group_workspace_unique" RENAME TO "credential_group_workspace_unique_failed_0326";
+  END IF;
+END $$;
+--> statement-breakpoint
+-- migration-safe: Only the invalid workspace index left by 0326 is renamed above. Valid indexes and legacy uniqueness remain intact; the following statement rebuilds the failed index without removing rows.
+DROP INDEX CONCURRENTLY IF EXISTS "public"."credential_group_workspace_unique_failed_0326";
+--> statement-breakpoint
 CREATE UNIQUE INDEX CONCURRENTLY IF NOT EXISTS "credential_group_workspace_unique" ON "credential_group" USING btree ("workspace_id");
 --> statement-breakpoint
 CREATE INDEX CONCURRENTLY IF NOT EXISTS "doc_connector_source_lookup_idx" ON "document" USING btree ("connector_id","external_id");

packages/db/organization-search-migration.postgres.test.ts

@@ -147,7 +147,7 @@ async function createMigrationFixture() {
 }
 
 describe.skipIf(!databaseUrl)('Organization Search PostgreSQL migration replay', () => {
-  it('preserves old uniqueness and data when a concurrent replacement fails, then replays after repair', async () => {
+  it('preserves data on duplicate failures and rebuilds the invalid index after duplicates are removed', async () => {
     const fixture = await createMigrationFixture()
     const { sql, schema } = fixture
     try {
@@ -158,20 +158,20 @@ describe.skipIf(!databaseUrl)('Organization Search PostgreSQL migration replay',
         await sql`SELECT indisvalid FROM pg_index
         WHERE indexrelid = ${`"${schema}"."credential_group_workspace_unique"`}::regclass`
       ).toEqual([{ indisvalid: false }])
-      await expect(fixture.migrate()).rejects.toMatchObject({
-        code: 'P0001',
-        message: expect.stringContaining('credential_group_workspace_unique'),
-        hint: expect.stringContaining('Repair the listed indexes'),
-      })
+      await expect(fixture.migrate()).rejects.toMatchObject({ code: '23505' })
       expect(await sql`SELECT id FROM credential_group`).toHaveLength(2)
       await expect(sql`INSERT INTO credential_group (id, workspace_id, name)
         VALUES ('third', 'workspace-a', 'First')`).rejects.toMatchObject({
         code: '23505',
         constraint_name: 'credential_group_workspace_name_unique',
       })
       await sql`DELETE FROM credential_group WHERE id = 'duplicate'`
-      await sql.unsafe('DROP INDEX CONCURRENTLY credential_group_workspace_unique')
       await fixture.migrate()
+      expect(
+        await sql`SELECT indisvalid, indisready FROM pg_index
+        WHERE indexrelid = ${`"${schema}"."credential_group_workspace_unique"`}::regclass`
+      ).toEqual([{ indisvalid: true, indisready: true }])
+      expect(await sql`SELECT id FROM credential_group`).toEqual([{ id: 'first' }])
       await expect(sql`INSERT INTO credential_group (id, workspace_id, name)
         VALUES ('third', 'workspace-a', 'Different')`).rejects.toMatchObject({
         code: '23505',
@@ -182,6 +182,69 @@ describe.skipIf(!databaseUrl)('Organization Search PostgreSQL migration replay',
     }
   })
 
+  it('resumes index recovery after interruption between renaming and dropping the failed index', async () => {
+    const fixture = await createMigrationFixture()
+    const { sql, schema } = fixture
+    try {
+      await sql`INSERT INTO credential_group (id, workspace_id, name)
+        VALUES ('first', 'workspace-a', 'First'), ('duplicate', 'workspace-a', 'Second')`
+      await expect(fixture.migrate()).rejects.toMatchObject({ code: '23505' })
+      await sql`DELETE FROM credential_group WHERE id = 'duplicate'`
+      const recovery = fixture.statements.find((statement) =>
+        statement.includes('RENAME TO "credential_group_workspace_unique_failed_0326"')
+      )
+      expect(recovery).toBeDefined()
+      await sql.unsafe(recovery!)
+      await fixture.migrate()
+      expect(
+        await sql`SELECT indisvalid FROM pg_index
+        WHERE indexrelid = ${`"${schema}"."credential_group_workspace_unique"`}::regclass`
+      ).toEqual([{ indisvalid: true }])
+      expect(
+        await sql`SELECT to_regclass(${`"${schema}"."credential_group_workspace_unique_failed_0326"`}) AS recovery`
+      ).toEqual([{ recovery: null }])
+    } finally {
+      await fixture.cleanup()
+    }
+  })
+
+  it('does not drop a healthy index occupying the recovery name', async () => {
+    const fixture = await createMigrationFixture()
+    const { sql, schema } = fixture
+    try {
+      await sql.unsafe(
+        'CREATE INDEX credential_group_workspace_unique_failed_0326 ON credential_group (workspace_id)'
+      )
+      await expect(fixture.migrate()).rejects.toMatchObject({
+        code: 'P0001',
+        message: expect.stringContaining('Refusing to drop unexpected index'),
+      })
+      expect(
+        await sql`SELECT indisvalid FROM pg_index
+        WHERE indexrelid = ${`"${schema}"."credential_group_workspace_unique_failed_0326"`}::regclass`
+      ).toEqual([{ indisvalid: true }])
+    } finally {
+      await fixture.cleanup()
+    }
+  })
+
+  it('preserves the healthy workspace index when the migration is replayed', async () => {
+    const fixture = await createMigrationFixture()
+    const { sql, schema } = fixture
+    try {
+      await fixture.migrate()
+      const before = await sql`SELECT indexrelid::oid AS oid FROM pg_index
+        WHERE indexrelid = ${`"${schema}"."credential_group_workspace_unique"`}::regclass`
+      await fixture.migrate()
+      expect(
+        await sql`SELECT indexrelid::oid AS oid FROM pg_index
+        WHERE indexrelid = ${`"${schema}"."credential_group_workspace_unique"`}::regclass`
+      ).toEqual(before)
+    } finally {
+      await fixture.cleanup()
+    }
+  })
+
   it.each(['complete migration', 'committed pre-index phase'] as const)(
     'replays after a %s without losing owner constraints or workspace behavior',
     async (interruption) => {