From a4400b4acd257ea3b46f03bfa4acf54115cede0b Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Tue, 11 Aug 2026 21:14:20 -0700 Subject: [PATCH 1/2] fix(logs): snapshot fetch optionality --- apps/sim/lib/logs/fetch-log-detail.test.ts | 89 ++++++++++++++++++++++ apps/sim/lib/logs/fetch-log-detail.ts | 58 +++++++++++++- 2 files changed, 144 insertions(+), 3 deletions(-) create mode 100644 apps/sim/lib/logs/fetch-log-detail.test.ts diff --git a/apps/sim/lib/logs/fetch-log-detail.test.ts b/apps/sim/lib/logs/fetch-log-detail.test.ts new file mode 100644 index 00000000000..7eb064ae99e --- /dev/null +++ b/apps/sim/lib/logs/fetch-log-detail.test.ts @@ -0,0 +1,89 @@ +/** + * @vitest-environment node + */ + +import { usageLog, user, workflowExecutionLogs, workflowExecutionSnapshots } from '@sim/db/schema' +import { dbChainMockFns, queueTableRows, resetDbChainMock } from '@sim/testing' +import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest' + +const mocks = vi.hoisted(() => ({ + checkWorkspaceAccess: vi.fn(), + materializeExecutionData: vi.fn(), +})) + +vi.mock('@/lib/workspaces/permissions/utils', () => ({ + checkWorkspaceAccess: mocks.checkWorkspaceAccess, +})) + +vi.mock('@/lib/logs/execution/trace-store', () => ({ + materializeExecutionDataForDisplay: mocks.materializeExecutionData, +})) + +vi.mock('@/lib/logs/execution-origin', () => ({ + workflowExecutionOriginSql: () => ({ as: () => ({}) }), +})) + +import { fetchLogDetail } from '@/lib/logs/fetch-log-detail' + +describe('fetchLogDetail', () => { + beforeEach(() => { + vi.clearAllMocks() + resetDbChainMock() + mocks.checkWorkspaceAccess.mockResolvedValue({ hasAccess: true }) + mocks.materializeExecutionData.mockResolvedValue({}) + }) + + afterAll(resetDbChainMock) + + it('loads workflow detail without materializing its execution snapshot', async () => { + queueTableRows(workflowExecutionLogs, [ + { + id: 'log-1', + workflowId: 'workflow-1', + executionId: 'execution-1', + deploymentVersionId: null, + deploymentVersion: null, + deploymentVersionName: null, + level: 'info', + status: 'completed', + trigger: 'manual', + startedAt: new Date('2026-01-01T00:00:00.000Z'), + endedAt: new Date('2026-01-01T00:00:01.000Z'), + totalDurationMs: 1000, + executionData: {}, + costTotal: null, + files: null, + createdAt: new Date('2026-01-01T00:00:00.000Z'), + workflowName: 'Workflow', + workflowDescription: null, + workflowFolderId: null, + workflowUserId: 'user-1', + workflowWorkspaceId: 'workspace-1', + workflowCreatedAt: new Date('2026-01-01T00:00:00.000Z'), + workflowUpdatedAt: new Date('2026-01-01T00:00:00.000Z'), + pausedStatus: null, + pausedTotalPauseCount: 0, + pausedResumedCount: 0, + executionOrigin: null, + }, + ]) + queueTableRows(usageLog, []) + + const result = await fetchLogDetail({ + userId: 'user-1', + workspaceId: 'workspace-1', + lookupColumn: 'id', + lookupValue: 'log-1', + }) + + expect(result).toMatchObject({ id: 'log-1', executionId: 'execution-1' }) + + const workflowSelection = dbChainMockFns.select.mock.calls[0]?.[0] as Record + expect(workflowSelection).not.toHaveProperty('workflowState') + expect(Object.values(workflowSelection)).not.toContain(workflowExecutionSnapshots.stateData) + + const joinedTables = dbChainMockFns.leftJoin.mock.calls.map(([table]) => table) + expect(joinedTables).not.toContain(workflowExecutionSnapshots) + expect(joinedTables).not.toContain(user) + }) +}) diff --git a/apps/sim/lib/logs/fetch-log-detail.ts b/apps/sim/lib/logs/fetch-log-detail.ts index 64931f50fda..5c7f5b75824 100644 --- a/apps/sim/lib/logs/fetch-log-detail.ts +++ b/apps/sim/lib/logs/fetch-log-detail.ts @@ -1,5 +1,12 @@ import { db } from '@sim/db' -import { jobExecutionLogs, usageLog } from '@sim/db/schema' +import { + jobExecutionLogs, + pausedExecutions, + usageLog, + workflow, + workflowDeploymentVersion, + workflowExecutionLogs, +} from '@sim/db/schema' import { and, eq, type SQL } from 'drizzle-orm' import type { CostLedger } from '@/lib/api/contracts/logs' import { @@ -9,7 +16,7 @@ import { pickLatestStartedMarker, } from '@/lib/logs/execution/progress-markers' import { materializeExecutionDataForDisplay } from '@/lib/logs/execution/trace-store' -import { getPublicWorkflowLog } from '@/lib/logs/public-queries' +import { workflowExecutionOriginSql } from '@/lib/logs/execution-origin' import { checkWorkspaceAccess } from '@/lib/workspaces/permissions/utils' type LookupColumn = 'id' | 'executionId' @@ -91,7 +98,52 @@ export async function fetchLogDetail({ const access = await checkWorkspaceAccess(workspaceId, userId) if (!access.hasAccess) return null - const log = await getPublicWorkflowLog({ column: lookupColumn, value: lookupValue }, workspaceId) + const workflowMatch: SQL = + lookupColumn === 'id' + ? eq(workflowExecutionLogs.id, lookupValue) + : eq(workflowExecutionLogs.executionId, lookupValue) + + const rows = await db + .select({ + id: workflowExecutionLogs.id, + workflowId: workflowExecutionLogs.workflowId, + executionId: workflowExecutionLogs.executionId, + deploymentVersionId: workflowExecutionLogs.deploymentVersionId, + level: workflowExecutionLogs.level, + status: workflowExecutionLogs.status, + trigger: workflowExecutionLogs.trigger, + startedAt: workflowExecutionLogs.startedAt, + endedAt: workflowExecutionLogs.endedAt, + totalDurationMs: workflowExecutionLogs.totalDurationMs, + executionData: workflowExecutionLogs.executionData, + costTotal: workflowExecutionLogs.costTotal, + files: workflowExecutionLogs.files, + createdAt: workflowExecutionLogs.createdAt, + workflowName: workflow.name, + workflowDescription: workflow.description, + workflowFolderId: workflow.folderId, + workflowUserId: workflow.userId, + workflowWorkspaceId: workflow.workspaceId, + workflowCreatedAt: workflow.createdAt, + workflowUpdatedAt: workflow.updatedAt, + deploymentVersion: workflowDeploymentVersion.version, + deploymentVersionName: workflowDeploymentVersion.name, + pausedStatus: pausedExecutions.status, + pausedTotalPauseCount: pausedExecutions.totalPauseCount, + pausedResumedCount: pausedExecutions.resumedCount, + executionOrigin: workflowExecutionOriginSql().as('execution_origin'), + }) + .from(workflowExecutionLogs) + .leftJoin(workflow, eq(workflowExecutionLogs.workflowId, workflow.id)) + .leftJoin( + workflowDeploymentVersion, + eq(workflowDeploymentVersion.id, workflowExecutionLogs.deploymentVersionId) + ) + .leftJoin(pausedExecutions, eq(pausedExecutions.executionId, workflowExecutionLogs.executionId)) + .where(and(workflowMatch, eq(workflowExecutionLogs.workspaceId, workspaceId))) + .limit(1) + + const log = rows[0] if (log) { const workflowSummary = log.workflowId From 3125237f79a77cf7eab47b344b0d50a745fb8731 Mon Sep 17 00:00:00 2001 From: Vikhyath Mondreti Date: Tue, 11 Aug 2026 21:18:47 -0700 Subject: [PATCH 2/2] fix(workflows): restore consistent draft read snapshots --- .../read-workflow-definition.test.ts | 130 ++++++++++++++++++ .../application/read-workflow-definition.ts | 23 +++- .../workflows/application/read-workflow.ts | 2 +- .../application/workflow-crud.test.ts | 2 +- apps/sim/lib/workflows/queries.ts | 14 +- 5 files changed, 162 insertions(+), 9 deletions(-) create mode 100644 apps/sim/lib/workflows/application/read-workflow-definition.test.ts diff --git a/apps/sim/lib/workflows/application/read-workflow-definition.test.ts b/apps/sim/lib/workflows/application/read-workflow-definition.test.ts new file mode 100644 index 00000000000..62f5be0265f --- /dev/null +++ b/apps/sim/lib/workflows/application/read-workflow-definition.test.ts @@ -0,0 +1,130 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const mocks = vi.hoisted(() => ({ + loadDeployed: vi.fn(), + loadSnapshot: vi.fn(), + resolveContext: vi.fn(), + resolvePermission: vi.fn(), +})) + +vi.mock('@sim/platform-authz/workspace', () => ({ + permissionSatisfies: (actual: string | null, required: string) => { + const rank = { read: 1, write: 2, admin: 3 } as const + return ( + actual !== null && rank[actual as keyof typeof rank] >= rank[required as keyof typeof rank] + ) + }, + resolveEffectiveWorkspacePermission: mocks.resolvePermission, +})) + +vi.mock('@/lib/workflows/application/context', () => ({ + resolveActiveWorkflowApplicationContext: mocks.resolveContext, +})) + +vi.mock('@/lib/workflows/queries', () => ({ + loadWorkflowReadSnapshot: mocks.loadSnapshot, +})) + +vi.mock('@/lib/workflows/persistence/utils', () => ({ + loadDeployedWorkflowState: mocks.loadDeployed, + NoActiveDeploymentError: class NoActiveDeploymentError extends Error {}, +})) + +import { readWorkflowDefinition } from '@/lib/workflows/application/read-workflow-definition' + +const WORKFLOW_ID = 'workflow-1' +const WORKSPACE_ID = 'workspace-1' +const principal = { + kind: 'session' as const, + userId: 'user-1', + sessionId: 'session-1', +} +const contextWorkflow = { + id: WORKFLOW_ID, + workspaceId: WORKSPACE_ID, + archivedAt: null, + name: 'Context workflow', +} +const snapshotWorkflow = { + ...contextWorkflow, + name: 'Snapshot workflow', +} +const context = { + workflowId: WORKFLOW_ID, + workflow: contextWorkflow, + workspaceId: WORKSPACE_ID, + workspaceOrganizationId: null, + allowPersonalApiKeys: true, + billedAccountUserId: 'billing-owner-1', +} +const draftState = { + blocks: {}, + edges: [], + loops: {}, + parallels: {}, + isFromNormalizedTables: true, +} + +describe('readWorkflowDefinition', () => { + beforeEach(() => { + vi.clearAllMocks() + mocks.resolveContext.mockResolvedValue(context) + mocks.resolvePermission.mockResolvedValue('read') + mocks.loadSnapshot.mockResolvedValue({ + workflowRecord: snapshotWorkflow, + normalizedData: draftState, + }) + mocks.loadDeployed.mockResolvedValue({ ...draftState, deploymentVersionId: 'version-1' }) + }) + + it('returns the workflow row and draft state from one canonical snapshot', async () => { + const result = await readWorkflowDefinition.execute({ + principal, + input: { workflowId: WORKFLOW_ID, state: 'draft' }, + }) + + expect(mocks.loadSnapshot).toHaveBeenCalledWith(WORKFLOW_ID, WORKSPACE_ID) + expect(mocks.loadDeployed).not.toHaveBeenCalled() + expect(result).toEqual({ + workflow: snapshotWorkflow, + workspaceId: WORKSPACE_ID, + state: draftState, + }) + }) + + it.each([ + ['missing', null], + ['archived', { ...snapshotWorkflow, archivedAt: new Date('2026-08-11T00:00:00Z') }], + ['cross-workspace', { ...snapshotWorkflow, workspaceId: 'workspace-other' }], + ])('rejects a %s workflow row returned by the draft snapshot', async (_label, workflowRecord) => { + mocks.loadSnapshot.mockResolvedValueOnce({ workflowRecord, normalizedData: draftState }) + + await expect( + readWorkflowDefinition.execute({ + principal, + input: { workflowId: WORKFLOW_ID, state: 'draft' }, + }) + ).rejects.toMatchObject({ code: 'not_found' }) + }) + + it('keeps deployed reads on the immutable deployment-state path', async () => { + const deployedState = { ...draftState, deploymentVersionId: 'version-1' } + mocks.loadDeployed.mockResolvedValueOnce(deployedState) + + const result = await readWorkflowDefinition.execute({ + principal, + input: { workflowId: WORKFLOW_ID, state: 'deployed' }, + }) + + expect(mocks.loadSnapshot).not.toHaveBeenCalled() + expect(mocks.loadDeployed).toHaveBeenCalledWith(WORKFLOW_ID, WORKSPACE_ID) + expect(result).toEqual({ + workflow: contextWorkflow, + workspaceId: WORKSPACE_ID, + state: deployedState, + }) + }) +}) diff --git a/apps/sim/lib/workflows/application/read-workflow-definition.ts b/apps/sim/lib/workflows/application/read-workflow-definition.ts index 0969b0ad563..5c1557ff52d 100644 --- a/apps/sim/lib/workflows/application/read-workflow-definition.ts +++ b/apps/sim/lib/workflows/application/read-workflow-definition.ts @@ -1,5 +1,6 @@ import type { Principal } from '@sim/auth/principal' import type { NormalizedWorkflowData } from '@sim/workflow-persistence/types' +import { OrchestrationError } from '@/lib/core/orchestration/types' import { defineAuthorizedWorkflowUseCase } from '@/lib/workflows/application/authorized-workflow-use-case' import { resolveActiveWorkflowApplicationContext } from '@/lib/workflows/application/context' import { workflowOperations } from '@/lib/workflows/application/operations' @@ -7,9 +8,9 @@ import { assertedWorkflowWorkspaceId } from '@/lib/workflows/application/princip import { type DeployedWorkflowData, loadDeployedWorkflowState, - loadWorkflowFromNormalizedTables, NoActiveDeploymentError, } from '@/lib/workflows/persistence/utils' +import { loadWorkflowReadSnapshot } from '@/lib/workflows/queries' export interface ReadWorkflowDefinitionInput { workflowId: string @@ -23,10 +24,9 @@ export interface ReadWorkflowDefinitionResult { state: NormalizedWorkflowData | DeployedWorkflowData | null } -async function loadDefinition(input: ReadWorkflowDefinitionInput, workspaceId: string) { - if (input.state === 'draft') return loadWorkflowFromNormalizedTables(input.workflowId) +async function loadDeployedDefinition(workflowId: string, workspaceId: string) { try { - return await loadDeployedWorkflowState(input.workflowId, workspaceId) + return await loadDeployedWorkflowState(workflowId, workspaceId) } catch (error) { if (error instanceof NoActiveDeploymentError) return null throw error @@ -47,10 +47,23 @@ export const readWorkflowDefinition = defineAuthorizedWorkflowUseCase({ assertedWorkspaceId: assertedWorkflowWorkspaceId(principal, input.assertedWorkspaceId), }), async execute({ input, context }): Promise { + if (input.state === 'draft') { + const snapshot = await loadWorkflowReadSnapshot(context.workflowId, context.workspaceId) + const workflow = snapshot.workflowRecord + if (!workflow || workflow.archivedAt || workflow.workspaceId !== context.workspaceId) { + throw new OrchestrationError('not_found', 'Workflow not found') + } + return { + workflow, + workspaceId: context.workspaceId, + state: snapshot.normalizedData, + } + } + return { workflow: context.workflow, workspaceId: context.workspaceId, - state: await loadDefinition(input, context.workspaceId), + state: await loadDeployedDefinition(context.workflowId, context.workspaceId), } }, }) diff --git a/apps/sim/lib/workflows/application/read-workflow.ts b/apps/sim/lib/workflows/application/read-workflow.ts index 984fd49a782..3adaa524418 100644 --- a/apps/sim/lib/workflows/application/read-workflow.ts +++ b/apps/sim/lib/workflows/application/read-workflow.ts @@ -26,7 +26,7 @@ export const readWorkflow = defineAuthorizedWorkflowUseCase({ assertedWorkspaceId: assertedWorkflowWorkspaceId(principal, input.assertedWorkspaceId), }), async execute({ principal, context }) { - const snapshot = await loadWorkflowReadSnapshot(context.workflowId) + const snapshot = await loadWorkflowReadSnapshot(context.workflowId, context.workspaceId) const workflow = snapshot.workflowRecord if (!workflow || workflow.archivedAt || workflow.workspaceId !== context.workspaceId) { throw new OrchestrationError('not_found', 'Workflow not found') diff --git a/apps/sim/lib/workflows/application/workflow-crud.test.ts b/apps/sim/lib/workflows/application/workflow-crud.test.ts index f8911354272..f71e3546d9b 100644 --- a/apps/sim/lib/workflows/application/workflow-crud.test.ts +++ b/apps/sim/lib/workflows/application/workflow-crud.test.ts @@ -331,7 +331,7 @@ describe('authorized workflow CRUD and version reads', () => { expect(mocks.resolvePermission).toHaveBeenCalledWith('user-1', WORKSPACE_ID, null, undefined, { forUpdate: undefined, }) - expect(mocks.loadSnapshot).toHaveBeenCalledWith(WORKFLOW_ID) + expect(mocks.loadSnapshot).toHaveBeenCalledWith(WORKFLOW_ID, WORKSPACE_ID) }) it('rejects executor reads whose canonical target is outside the signed origin workspace', async () => { diff --git a/apps/sim/lib/workflows/queries.ts b/apps/sim/lib/workflows/queries.ts index e53aae0f052..f4a4a11c198 100644 --- a/apps/sim/lib/workflows/queries.ts +++ b/apps/sim/lib/workflows/queries.ts @@ -138,12 +138,22 @@ export async function listWorkspaceWorkflows(input: ListWorkspaceWorkflowsInput) * the editor route and public metadata route derive their own response from * this read rather than issuing independent block/workflow queries. */ -export async function loadWorkflowReadSnapshot(workflowId: string) { +export async function loadWorkflowReadSnapshot(workflowId: string, workspaceId: string) { return db.transaction(async (tx) => { await tx.execute(sql`SET TRANSACTION ISOLATION LEVEL REPEATABLE READ`) const [normalizedData, [workflowRecord]] = await Promise.all([ loadWorkflowFromNormalizedTables(workflowId, tx), - tx.select().from(workflow).where(eq(workflow.id, workflowId)).limit(1), + tx + .select() + .from(workflow) + .where( + and( + eq(workflow.id, workflowId), + eq(workflow.workspaceId, workspaceId), + isNull(workflow.archivedAt) + ) + ) + .limit(1), ]) return { normalizedData, workflowRecord: workflowRecord ?? null } })