diff --git a/.github/workflows/test-build.yml b/.github/workflows/test-build.yml index 5fccc030659..d7122f1e889 100644 --- a/.github/workflows/test-build.yml +++ b/.github/workflows/test-build.yml @@ -87,8 +87,9 @@ jobs: - name: Verify OAuth lifecycle and SCIM membership guards in PostgreSQL working-directory: apps/sim + # These suites share a schema and install triggers; parallel files can deadlock DDL against cleanup. run: >- - bunx vitest run + bunx vitest run --no-file-parallelism lib/auth/oauth-token-family.postgres.test.ts lib/auth/oauth-provider-lifecycle.postgres.test.ts app/api/auth/oauth2/token/route.postgres.test.ts diff --git a/apps/sim/lib/credentials/deletion.test.ts b/apps/sim/lib/credentials/deletion.test.ts index 1a175f2b3e9..ec24a495e21 100644 --- a/apps/sim/lib/credentials/deletion.test.ts +++ b/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, + }) + }) +}) diff --git a/apps/sim/lib/credentials/deletion.ts b/apps/sim/lib/credentials/deletion.ts index 0206d68457b..9884f0faf39 100644 --- a/apps/sim/lib/credentials/deletion.ts +++ b/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 { - 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 { - 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 { - 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 { - 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> { + 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 { await db .update(schema.knowledgeConnector) diff --git a/apps/sim/lib/table/api/route-policies.test.ts b/apps/sim/lib/table/api/route-policies.test.ts index f691f55db8e..035653975b8 100644 --- a/apps/sim/lib/table/api/route-policies.test.ts +++ b/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' }, + }, + }) + } + ) }) diff --git a/apps/sim/lib/table/api/route-policies.ts b/apps/sim/lib/table/api/route-policies.ts index bd8402a0809..f0715d24536 100644 --- a/apps/sim/lib/table/api/route-policies.ts +++ b/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, diff --git a/apps/sim/lib/table/errors.ts b/apps/sim/lib/table/errors.ts index d54cf0b4a69..a9ce6aa78f1 100644 --- a/apps/sim/lib/table/errors.ts +++ b/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. diff --git a/apps/sim/lib/table/ttl-availability.test.ts b/apps/sim/lib/table/ttl-availability.test.ts index 4710ce7d0b9..f6b87f2e357 100644 --- a/apps/sim/lib/table/ttl-availability.test.ts +++ b/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) + }) }) diff --git a/apps/sim/lib/table/ttl-availability.ts b/apps/sim/lib/table/ttl-availability.ts index f5442b35975..a04002f12c4 100644 --- a/apps/sim/lib/table/ttl-availability.ts +++ b/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 { @@ -9,5 +9,5 @@ export function isTableRowTtlEnabled(): Promise { /** Rejects attempts to introduce a TTL column while the feature is disabled. */ export async function assertTableRowTtlEnabled(): Promise { if (await isTableRowTtlEnabled()) return - throw new OrchestrationError('validation', 'Expiration columns are not enabled') + throw new TableRowTtlDisabledError() }