Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
89 changes: 89 additions & 0 deletions apps/sim/lib/logs/fetch-log-detail.test.ts
Original file line number Diff line number Diff line change
@@ -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)
})
})
58 changes: 55 additions & 3 deletions apps/sim/lib/logs/fetch-log-detail.ts
Original file line number Diff line number Diff line change
@@ -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 {
Expand All @@ -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'
Expand Down Expand Up @@ -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
Expand Down
130 changes: 130 additions & 0 deletions apps/sim/lib/workflows/application/read-workflow-definition.test.ts
Original file line number Diff line number Diff line change
@@ -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,
})
})
})
23 changes: 18 additions & 5 deletions apps/sim/lib/workflows/application/read-workflow-definition.ts
Original file line number Diff line number Diff line change
@@ -1,15 +1,16 @@
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'
import { assertedWorkflowWorkspaceId } from '@/lib/workflows/application/principal-scope'
import {
type DeployedWorkflowData,
loadDeployedWorkflowState,
loadWorkflowFromNormalizedTables,
NoActiveDeploymentError,
} from '@/lib/workflows/persistence/utils'
import { loadWorkflowReadSnapshot } from '@/lib/workflows/queries'

export interface ReadWorkflowDefinitionInput {
workflowId: string
Expand All @@ -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
Expand All @@ -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),
}
},
})
2 changes: 1 addition & 1 deletion apps/sim/lib/workflows/application/read-workflow.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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')
Expand Down
2 changes: 1 addition & 1 deletion apps/sim/lib/workflows/application/workflow-crud.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 () => {
Expand Down
14 changes: 12 additions & 2 deletions apps/sim/lib/workflows/queries.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 }
})
Expand Down
Loading