fix(integ): bound credential scans and classify disabled TTL

Author: TheodoreSpeaks

apps/sim/lib/credentials/deletion.test.ts

@@ -15,13 +15,16 @@
  * which cannot express a subquery), and `@sim/db` is a `drizzle-orm/pg-proxy`
  * client whose driver captures the compiled statement and replays the rows
  * Postgres would return for the scenario under test.
+ *
+ * Credential-reference scans also use the real query builder to verify that
+ * workspace filtering stays inside a materialized boundary before JSON search.
  */
 import { drizzle } from 'drizzle-orm/pg-proxy'
 import { beforeEach, describe, expect, it, vi } from 'vitest'
 
 const { capturedQueries, driverRows, mockLogger } = vi.hoisted(() => ({
   capturedQueries: [] as { sql: string; params: unknown[] }[],
-  driverRows: { value: [] as unknown[] },
+  driverRows: { value: [] as unknown[], error: null as Error | null },
   mockLogger: {
     info: vi.fn(),
     warn: vi.fn(),
@@ -38,11 +41,12 @@ vi.mock('@sim/logger', () => ({ createLogger: () => mockLogger }))
 vi.mock('@sim/db', () => ({
   db: drizzle(async (sql: string, params: unknown[]) => {
     capturedQueries.push({ sql, params })
+    if (driverRows.error) throw driverRows.error
     return { rows: driverRows.value }
   }),
 }))
 
-import { deleteOrphanedOAuthAccount } from '@/lib/credentials/deletion'
+import { clearCredentialRefs, deleteOrphanedOAuthAccount } from '@/lib/credentials/deletion'
 
 const ACCOUNT_ID = 'acct-bob-google'
 
@@ -64,13 +68,14 @@ function guardSubquery(sql: string): string {
   return match[1]
 }
 
-describe('deleteOrphanedOAuthAccount', () => {
-  beforeEach(() => {
-    capturedQueries.length = 0
-    driverRows.value = []
-    vi.clearAllMocks()
-  })
+beforeEach(() => {
+  capturedQueries.length = 0
+  driverRows.value = []
+  driverRows.error = null
+  vi.clearAllMocks()
+})
 
+describe('deleteOrphanedOAuthAccount', () => {
   it('guards the account delete with a reference check against the credential table', async () => {
     await deleteOrphanedOAuthAccount(ACCOUNT_ID)
 
@@ -137,3 +142,65 @@ describe('deleteOrphanedOAuthAccount', () => {
     expect(sql).not.toContain('provider_id')
   })
 })
+
+describe('clearCredentialRefs', () => {
+  const sources = [
+    ['workflow_blocks', 'sub_blocks'],
+    ['workflow_deployment_version', 'state'],
+    ['paused_executions', 'execution_snapshot'],
+    ['workflow_checkpoints', 'workflow_state'],
+  ] as const
+
+  it('scopes every snapshot scan before converting JSON to text, including archived workflows', async () => {
+    await clearCredentialRefs('credential-target', 'workspace-target')
+
+    const reads = capturedQueries.filter((query) => normalizeSql(query.sql).startsWith('WITH'))
+    expect(reads).toHaveLength(sources.length)
+    for (const [table, column] of sources) {
+      const query = reads.find((query) => query.sql.includes(`FROM "${table}"`))
+      expect(query).toBeDefined()
+      const statement = normalizeSql(query!.sql)
+      expect(statement).toContain(
+        `WITH workspace_credential_refs AS MATERIALIZED ( SELECT "${table}"."id" AS id, "${table}"."${column}" AS value FROM "${table}" INNER JOIN "workflow" ON "workflow"."id" = "${table}"."workflow_id" WHERE "workflow"."workspace_id" = $1 ) SELECT id, value FROM workspace_credential_refs WHERE value::text LIKE $2`
+      )
+      expect(statement).not.toContain('deleted_at')
+      expect(query!.params).toEqual(['workspace-target', '%credential-target%'])
+    }
+  })
+
+  it('clears matching references returned by every raw scan and preserves other values', async () => {
+    driverRows.value = [
+      {
+        id: 'snapshot-target',
+        value: {
+          blocks: [{ id: 'credential', value: 'credential-target' }],
+          params: { credential: 'credential-target', other: 'credential-other' },
+          name: 'credential-target',
+        },
+      },
+      { id: 'substring-only', value: { name: 'credential-target' } },
+    ]
+
+    await clearCredentialRefs('credential-target', 'workspace-target')
+
+    for (const [table] of sources) {
+      const updates = capturedQueries.filter((query) => query.sql.startsWith(`update "${table}"`))
+      expect(updates).toHaveLength(1)
+      expect(JSON.parse(updates[0].params[0] as string)).toEqual({
+        blocks: [{ id: 'credential', value: '' }],
+        params: { credential: '', other: 'credential-other' },
+        name: 'credential-target',
+      })
+      expect(updates[0].params.at(-1)).toBe('snapshot-target')
+    }
+  })
+
+  it('propagates database failures', async () => {
+    driverRows.error = new Error('database unavailable')
+    await expect(
+      clearCredentialRefs('credential-target', 'workspace-target')
+    ).rejects.toMatchObject({
+      cause: driverRows.error,
+    })
+  })
+})

apps/sim/lib/credentials/deletion.ts

@@ -3,6 +3,7 @@ import { db } from '@sim/db'
 import * as schema from '@sim/db/schema'
 import { createLogger } from '@sim/logger'
 import { and, eq, notExists, or, sql } from 'drizzle-orm'
+import type { AnyPgColumn, PgTable } from 'drizzle-orm/pg-core'
 import type { NextRequest } from 'next/server'
 import {
   type ResourceOwner,
@@ -216,23 +217,16 @@ async function clearInWorkflowBlocks(
   workspaceId: string,
   needle: string
 ): Promise<void> {
-  const rows = await db
-    .select({
-      id: schema.workflowBlocks.id,
-      subBlocks: schema.workflowBlocks.subBlocks,
-    })
-    .from(schema.workflowBlocks)
-    .innerJoin(schema.workflow, eq(schema.workflow.id, schema.workflowBlocks.workflowId))
-    .where(
-      and(
-        eq(schema.workflow.workspaceId, workspaceId),
-        sql`${schema.workflowBlocks.subBlocks}::text LIKE ${needle}`
-      )
-    )
+  const rows = await readWorkspaceCredentialRefs(workspaceId, needle, {
+    table: schema.workflowBlocks,
+    id: schema.workflowBlocks.id,
+    workflowId: schema.workflowBlocks.workflowId,
+    value: schema.workflowBlocks.subBlocks,
+  })
 
   let updated = 0
   for (const row of rows) {
-    const next = clearCredentialInValue(row.subBlocks, credentialId)
+    const next = clearCredentialInValue(row.value, credentialId)
     if (next.changed) {
       await db
         .update(schema.workflowBlocks)
@@ -255,22 +249,15 @@ async function clearInDeploymentVersions(
   workspaceId: string,
   needle: string
 ): Promise<void> {
-  const rows = await db
-    .select({
-      id: schema.workflowDeploymentVersion.id,
-      state: schema.workflowDeploymentVersion.state,
-    })
-    .from(schema.workflowDeploymentVersion)
-    .innerJoin(schema.workflow, eq(schema.workflow.id, schema.workflowDeploymentVersion.workflowId))
-    .where(
-      and(
-        eq(schema.workflow.workspaceId, workspaceId),
-        sql`${schema.workflowDeploymentVersion.state}::text LIKE ${needle}`
-      )
-    )
+  const rows = await readWorkspaceCredentialRefs(workspaceId, needle, {
+    table: schema.workflowDeploymentVersion,
+    id: schema.workflowDeploymentVersion.id,
+    workflowId: schema.workflowDeploymentVersion.workflowId,
+    value: schema.workflowDeploymentVersion.state,
+  })
 
   for (const row of rows) {
-    const next = clearCredentialInValue(row.state, credentialId)
+    const next = clearCredentialInValue(row.value, credentialId)
     if (next.changed) {
       await db
         .update(schema.workflowDeploymentVersion)
@@ -285,22 +272,15 @@ async function clearInPausedExecutions(
   workspaceId: string,
   needle: string
 ): Promise<void> {
-  const rows = await db
-    .select({
-      id: schema.pausedExecutions.id,
-      executionSnapshot: schema.pausedExecutions.executionSnapshot,
-    })
-    .from(schema.pausedExecutions)
-    .innerJoin(schema.workflow, eq(schema.workflow.id, schema.pausedExecutions.workflowId))
-    .where(
-      and(
-        eq(schema.workflow.workspaceId, workspaceId),
-        sql`${schema.pausedExecutions.executionSnapshot}::text LIKE ${needle}`
-      )
-    )
+  const rows = await readWorkspaceCredentialRefs(workspaceId, needle, {
+    table: schema.pausedExecutions,
+    id: schema.pausedExecutions.id,
+    workflowId: schema.pausedExecutions.workflowId,
+    value: schema.pausedExecutions.executionSnapshot,
+  })
 
   for (const row of rows) {
-    const next = clearCredentialInValue(row.executionSnapshot, credentialId)
+    const next = clearCredentialInValue(row.value, credentialId)
     if (next.changed) {
       await db
         .update(schema.pausedExecutions)
@@ -315,22 +295,15 @@ async function clearInWorkflowCheckpoints(
   workspaceId: string,
   needle: string
 ): Promise<void> {
-  const rows = await db
-    .select({
-      id: schema.workflowCheckpoints.id,
-      workflowState: schema.workflowCheckpoints.workflowState,
-    })
-    .from(schema.workflowCheckpoints)
-    .innerJoin(schema.workflow, eq(schema.workflow.id, schema.workflowCheckpoints.workflowId))
-    .where(
-      and(
-        eq(schema.workflow.workspaceId, workspaceId),
-        sql`${schema.workflowCheckpoints.workflowState}::text LIKE ${needle}`
-      )
-    )
+  const rows = await readWorkspaceCredentialRefs(workspaceId, needle, {
+    table: schema.workflowCheckpoints,
+    id: schema.workflowCheckpoints.id,
+    workflowId: schema.workflowCheckpoints.workflowId,
+    value: schema.workflowCheckpoints.workflowState,
+  })
 
   for (const row of rows) {
-    const next = clearCredentialInValue(row.workflowState, credentialId)
+    const next = clearCredentialInValue(row.value, credentialId)
     if (next.changed) {
       await db
         .update(schema.workflowCheckpoints)
@@ -340,6 +313,29 @@ async function clearInWorkflowCheckpoints(
   }
 }
 
+/**
+ * Restrict the rows before inspecting their JSON. With a plain join, Postgres can push
+ * the text predicate below the workspace join and detoast every tenant's snapshots.
+ * This query has reached 46s in production. Materializing the workspace selection
+ * keeps the expensive scan local, including archived workflows
+ * whose frozen snapshots still need their credential references removed.
+ */
+async function readWorkspaceCredentialRefs(
+  workspaceId: string,
+  needle: string,
+  source: { table: PgTable; id: AnyPgColumn; workflowId: AnyPgColumn; value: AnyPgColumn }
+): Promise<Array<{ id: string; value: unknown }>> {
+  return db.execute<{ id: string; value: unknown }>(sql`
+    WITH workspace_credential_refs AS MATERIALIZED (
+      SELECT ${source.id} AS id, ${source.value} AS value
+      FROM ${source.table}
+      INNER JOIN ${schema.workflow} ON ${schema.workflow.id} = ${source.workflowId}
+      WHERE ${schema.workflow.workspaceId} = ${workspaceId}
+    )
+    SELECT id, value FROM workspace_credential_refs WHERE value::text LIKE ${needle}
+  `)
+}
+
 async function clearInKnowledgeConnectors(credentialId: string): Promise<void> {
   await db
     .update(schema.knowledgeConnector)

apps/sim/lib/table/api/route-policies.test.ts

@@ -30,6 +30,7 @@ import { generateInternalDelegationToken, generateInternalToken } from '@/lib/au
 import { OrchestrationError } from '@/lib/core/orchestration/types'
 import { internalTableSessionOrExecutorAuth } from '@/lib/table/api'
 import { v2TableErrorPolicies } from '@/lib/table/api/route-policies'
+import { TableRowTtlDisabledError } from '@/lib/table/errors'
 
 afterAll(resetEnvMock)
 
@@ -187,4 +188,24 @@ describe('internal Table route authentication', () => {
       error: { code: 'BAD_REQUEST', message: 'Invalid workflow ID' },
     })
   })
+
+  it.each([false, true])(
+    'preserves the TTL-disabled reason code (wrapped: %s)',
+    async (wrapped) => {
+      const error = new TableRowTtlDisabledError()
+      error.message = 'TTL support is turned off'
+      const response = v2TableErrorPolicies.default.render(
+        wrapped ? new Error('operation failed', { cause: error }) : error
+      )
+
+      expect(response.status).toBe(400)
+      await expect(response.json()).resolves.toEqual({
+        error: {
+          code: 'BAD_REQUEST',
+          message: 'TTL support is turned off',
+          details: { code: 'TABLE_ROW_TTL_DISABLED' },
+        },
+      })
+    }
+  )
 })

apps/sim/lib/table/api/route-policies.ts

@@ -7,8 +7,10 @@ import {
   internalOrchestrationErrorPolicy,
   type V2ErrorPolicy,
 } from '@/lib/api/server/routes'
+import { asOrchestrationError } from '@/lib/core/orchestration/types'
 import { TABLE_DELEGATION_AUDIENCE } from '@/lib/table/application/authorization'
 import { TableOperationError } from '@/lib/table/application/errors'
+import { TableRowTtlDisabledError } from '@/lib/table/errors'
 import { TableLockedError } from '@/lib/table/mutation-locks'
 import {
   v2CaughtOrchestrationError,
@@ -25,6 +27,12 @@ export const internalTableSessionOrExecutorAuth = createInternalSessionOrExecuto
 })
 
 function renderTableError(error: unknown) {
+  const classified = asOrchestrationError(error)
+  if (classified instanceof TableRowTtlDisabledError) {
+    return v2Error('BAD_REQUEST', classified.message, {
+      details: { code: classified.detailCode },
+    })
+  }
   if (error instanceof TableOperationError) {
     return v2ErrorForOrchestration(
       error.code,

apps/sim/lib/table/errors.ts

@@ -1,3 +1,15 @@
+import { OrchestrationError } from '@/lib/core/orchestration/types'
+
+/** A disabled TTL feature, distinct from malformed column input. */
+export class TableRowTtlDisabledError extends OrchestrationError {
+  readonly detailCode = 'TABLE_ROW_TTL_DISABLED'
+
+  constructor() {
+    super('validation', 'Expiration columns are not enabled')
+    this.name = 'TableRowTtlDisabledError'
+  }
+}
+
 /**
  * Stable, machine-readable codes for table query failures. SDKs and clients
  * branch on these instead of string-matching human-facing messages.

apps/sim/lib/table/ttl-availability.test.ts

@@ -28,7 +28,20 @@ describe('table row TTL availability', () => {
 
     await expect(assertTableRowTtlEnabled()).rejects.toMatchObject({
       code: 'validation',
-      message: 'Expiration columns are not enabled',
+      detailCode: 'TABLE_ROW_TTL_DISABLED',
     })
   })
+
+  it('allows TTL column creation while the flag is enabled', async () => {
+    mockIsFeatureEnabled.mockResolvedValue(true)
+
+    await expect(assertTableRowTtlEnabled()).resolves.toBeUndefined()
+  })
+
+  it('propagates flag lookup failures instead of reporting the feature as disabled', async () => {
+    const error = new Error('flag service unavailable')
+    mockIsFeatureEnabled.mockRejectedValue(error)
+
+    await expect(assertTableRowTtlEnabled()).rejects.toBe(error)
+  })
 })

apps/sim/lib/table/ttl-availability.ts

@@ -1,5 +1,5 @@
 import { isFeatureEnabled } from '@/lib/core/config/feature-flags'
-import { OrchestrationError } from '@/lib/core/orchestration/types'
+import { TableRowTtlDisabledError } from '@/lib/table/errors'
 
 /** Whether TTL columns and their cleanup behavior are enabled globally. */
 export function isTableRowTtlEnabled(): Promise<boolean> {
@@ -9,5 +9,5 @@ export function isTableRowTtlEnabled(): Promise<boolean> {
 /** Rejects attempts to introduce a TTL column while the feature is disabled. */
 export async function assertTableRowTtlEnabled(): Promise<void> {
   if (await isTableRowTtlEnabled()) return
-  throw new OrchestrationError('validation', 'Expiration columns are not enabled')
+  throw new TableRowTtlDisabledError()
 }