fix(logs, workflows): snapshot fetch optionality, restore consistent draft read snapshots (#6594)

* fix(logs): snapshot fetch optionality

* fix(workflows): restore consistent draft read snapshots
This commit is contained in:
Vikhyath Mondreti
2026-08-11 21:27:05 -07:00
committed by GitHub
parent a61cbf7287
commit eb26b42680
7 changed files with 306 additions and 12 deletions
@@ -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<string, unknown>
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)
})
})
+55 -3
View File
@@ -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
@@ -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,
})
})
})
@@ -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<ReadWorkflowDefinitionResult> {
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),
}
},
})
@@ -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')
@@ -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 () => {
+12 -2
View File
@@ -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 }
})