diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/components/deploy-modal/deploy-modal.tsx b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/components/deploy-modal/deploy-modal.tsx index 3af6d13baf0..aa33d96cc92 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/components/deploy-modal/deploy-modal.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/components/deploy-modal/deploy-modal.tsx @@ -32,7 +32,6 @@ import { releaseDeployAction, tryAcquireDeployAction, } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/deploy-action-lock' -import { syncLocalDraftFromServer } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/sync-local-draft' import type { DeployReadiness } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/use-deploy-readiness' import { runPreDeployChecks } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/use-predeploy-checks' import { normalizeName, startsWithUuid } from '@/executor/constants' @@ -52,6 +51,7 @@ import { useWorkspaceSettings } from '@/hooks/queries/workspace' import { usePermissionConfig } from '@/hooks/use-permission-config' import { useSettingsNavigation } from '@/hooks/use-settings-navigation' import { useWorkflowRegistry } from '@/stores/workflows/registry/store' +import { syncLocalDraftFromServer } from '@/stores/workflows/sync-local-draft' import { mergeSubblockState } from '@/stores/workflows/utils' import { useWorkflowStore } from '@/stores/workflows/workflow/store' import type { WorkflowState } from '@/stores/workflows/workflow/types' diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/sync-local-draft.test.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/sync-local-draft.test.ts deleted file mode 100644 index 6abc9a694d0..00000000000 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/sync-local-draft.test.ts +++ /dev/null @@ -1,246 +0,0 @@ -/** - * @vitest-environment node - */ -import { beforeEach, describe, expect, it, vi } from 'vitest' - -const { - mockRequestJson, - mockApplyWorkflowStateToStores, - mockGetRegistryState, - mockHasPendingOperations, - mockGetOperationQueueState, - mockGetWorkflowDiffState, -} = vi.hoisted(() => ({ - mockRequestJson: vi.fn(), - mockApplyWorkflowStateToStores: vi.fn(), - mockGetRegistryState: vi.fn(() => ({ activeWorkflowId: 'workflow-a' })), - mockHasPendingOperations: vi.fn(() => false), - mockGetOperationQueueState: vi.fn(() => ({ - hasPendingOperations: mockHasPendingOperations, - workflowOperationVersions: {}, - })), - mockGetWorkflowDiffState: vi.fn(() => ({ - hasActiveDiff: false, - pendingExternalUpdates: {}, - reconcilingWorkflows: {}, - reconciliationErrors: {}, - remoteUpdateVersions: {}, - })), -})) - -vi.mock('@/lib/api/client/request', () => ({ - requestJson: mockRequestJson, -})) - -vi.mock('@/lib/api/contracts', () => ({ - getWorkflowStateContract: {}, -})) - -vi.mock('@/stores/workflow-diff/utils', () => ({ - applyWorkflowStateToStores: mockApplyWorkflowStateToStores, -})) - -vi.mock('@/stores/workflow-diff/store', () => ({ - useWorkflowDiffStore: { - getState: mockGetWorkflowDiffState, - }, -})) - -vi.mock('@/stores/operation-queue/store', () => ({ - useOperationQueueStore: { - getState: mockGetOperationQueueState, - }, -})) - -vi.mock('@/stores/workflows/registry/store', () => ({ - useWorkflowRegistry: { - getState: mockGetRegistryState, - }, -})) - -import { syncLocalDraftFromServer } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/sync-local-draft' - -describe('syncLocalDraftFromServer', () => { - beforeEach(() => { - vi.clearAllMocks() - mockGetRegistryState.mockReturnValue({ activeWorkflowId: 'workflow-a' }) - mockHasPendingOperations.mockReturnValue(false) - mockGetOperationQueueState.mockImplementation(() => ({ - hasPendingOperations: mockHasPendingOperations, - workflowOperationVersions: {}, - })) - mockGetWorkflowDiffState.mockReturnValue({ - hasActiveDiff: false, - pendingExternalUpdates: {}, - reconcilingWorkflows: {}, - reconciliationErrors: {}, - remoteUpdateVersions: {}, - }) - }) - - it('hydrates sibling workflow variables into the applied workflow state', async () => { - mockRequestJson.mockResolvedValue({ - data: { - state: { - blocks: {}, - edges: [], - loops: {}, - parallels: {}, - lastSaved: 1, - }, - variables: { - 'variable-a': { - id: 'variable-a', - name: 'API_KEY', - type: 'plain', - value: 'secret', - }, - }, - }, - }) - - await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(true) - - expect(mockApplyWorkflowStateToStores).toHaveBeenCalledWith( - 'workflow-a', - expect.objectContaining({ - variables: { - 'variable-a': { - id: 'variable-a', - name: 'API_KEY', - type: 'plain', - value: 'secret', - }, - }, - }), - { updateLastSaved: true } - ) - }) - - it('does not apply a fetched draft after navigation changes the active workflow', async () => { - mockGetRegistryState - .mockReturnValueOnce({ activeWorkflowId: 'workflow-a' }) - .mockReturnValueOnce({ activeWorkflowId: 'workflow-b' }) - mockRequestJson.mockResolvedValue({ - data: { - state: { - blocks: {}, - edges: [], - loops: {}, - parallels: {}, - lastSaved: 1, - }, - variables: {}, - }, - }) - - await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(false) - - expect(mockApplyWorkflowStateToStores).not.toHaveBeenCalled() - }) - - it('does not synthesize an empty variables object when the server omits variables', async () => { - mockRequestJson.mockResolvedValue({ - data: { - state: { - blocks: {}, - edges: [], - loops: {}, - parallels: {}, - lastSaved: 1, - }, - }, - }) - - await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(true) - - const appliedState = mockApplyWorkflowStateToStores.mock.calls[0][1] - expect(Object.hasOwn(appliedState, 'variables')).toBe(false) - }) - - it('does not apply a fetched draft over newly queued local operations', async () => { - mockHasPendingOperations.mockReturnValueOnce(false).mockReturnValueOnce(true) - mockRequestJson.mockResolvedValue({ - data: { - state: { - blocks: {}, - edges: [], - loops: {}, - parallels: {}, - lastSaved: 1, - }, - variables: {}, - }, - }) - - await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(false) - - expect(mockApplyWorkflowStateToStores).not.toHaveBeenCalled() - }) - - it('does not apply a fetched draft when a newer remote update arrives during fetch', async () => { - mockGetWorkflowDiffState - .mockReturnValueOnce({ - hasActiveDiff: false, - pendingExternalUpdates: {}, - reconcilingWorkflows: {}, - reconciliationErrors: {}, - remoteUpdateVersions: {}, - }) - .mockReturnValueOnce({ - hasActiveDiff: false, - pendingExternalUpdates: {}, - reconcilingWorkflows: {}, - reconciliationErrors: {}, - remoteUpdateVersions: { 'workflow-a': 1 }, - }) - mockRequestJson.mockResolvedValue({ - data: { - state: { - blocks: {}, - edges: [], - loops: {}, - parallels: {}, - lastSaved: 1, - }, - variables: {}, - }, - }) - - await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(false) - - expect(mockApplyWorkflowStateToStores).not.toHaveBeenCalled() - }) - - it('does not apply a fetched draft when local operations queue and drain during fetch', async () => { - mockGetOperationQueueState - .mockReturnValueOnce({ - hasPendingOperations: mockHasPendingOperations, - workflowOperationVersions: {}, - }) - .mockReturnValueOnce({ - hasPendingOperations: mockHasPendingOperations, - workflowOperationVersions: {}, - }) - .mockReturnValueOnce({ - hasPendingOperations: mockHasPendingOperations, - workflowOperationVersions: { 'workflow-a': 1 }, - }) - mockRequestJson.mockResolvedValue({ - data: { - state: { - blocks: {}, - edges: [], - loops: {}, - parallels: {}, - lastSaved: 1, - }, - variables: {}, - }, - }) - - await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(false) - - expect(mockApplyWorkflowStateToStores).not.toHaveBeenCalled() - }) -}) diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/sync-local-draft.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/sync-local-draft.ts deleted file mode 100644 index ad308ee549c..00000000000 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/sync-local-draft.ts +++ /dev/null @@ -1,59 +0,0 @@ -import { requestJson } from '@/lib/api/client/request' -import { getWorkflowStateContract } from '@/lib/api/contracts' -import { useOperationQueueStore } from '@/stores/operation-queue/store' -import { useWorkflowDiffStore } from '@/stores/workflow-diff/store' -import { applyWorkflowStateToStores } from '@/stores/workflow-diff/utils' -import { useWorkflowRegistry } from '@/stores/workflows/registry/store' -import type { WorkflowState } from '@/stores/workflows/workflow/types' - -function canApplyServerSnapshot( - workflowId: string, - remoteVersionAtStart: number, - localOperationVersionAtStart: number -): boolean { - if (useWorkflowRegistry.getState().activeWorkflowId !== workflowId) return false - const operationQueueState = useOperationQueueStore.getState() - if (operationQueueState.hasPendingOperations(workflowId)) return false - if ( - (operationQueueState.workflowOperationVersions[workflowId] ?? 0) !== - localOperationVersionAtStart - ) { - return false - } - - const diffState = useWorkflowDiffStore.getState() - return ( - !diffState.hasActiveDiff && - !diffState.pendingExternalUpdates[workflowId] && - !diffState.reconcilingWorkflows[workflowId] && - !diffState.reconciliationErrors[workflowId] && - (diffState.remoteUpdateVersions[workflowId] ?? 0) === remoteVersionAtStart - ) -} - -export async function syncLocalDraftFromServer(workflowId: string): Promise { - if (useWorkflowRegistry.getState().activeWorkflowId !== workflowId) return false - if (useOperationQueueStore.getState().hasPendingOperations(workflowId)) return false - const localOperationVersionAtStart = - useOperationQueueStore.getState().workflowOperationVersions[workflowId] ?? 0 - const remoteVersionAtStart = useWorkflowDiffStore.getState().remoteUpdateVersions[workflowId] ?? 0 - - const responseData = await requestJson(getWorkflowStateContract, { - params: { id: workflowId }, - }) - const wireState = responseData.data?.state - if (!canApplyServerSnapshot(workflowId, remoteVersionAtStart, localOperationVersionAtStart)) { - return false - } - if (!wireState) { - throw new Error('No workflow state was returned while syncing the local draft') - } - - // double-cast-allowed: workflowStateSchema is a wire supertype; normalized workflow state is persisted in store-compatible shape - const workflowState = wireState as unknown as WorkflowState - if (Object.hasOwn(responseData.data, 'variables')) { - workflowState.variables = responseData.data.variables || {} - } - applyWorkflowStateToStores(workflowId, workflowState, { updateLastSaved: true }) - return true -} diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/use-deployment.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/use-deployment.ts index d0e0b867bfe..4909fd4e467 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/use-deployment.ts +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/use-deployment.ts @@ -5,10 +5,10 @@ import { toError } from '@sim/utils/errors' import { runPreDeployChecks } from '@/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/deploy/hooks/use-predeploy-checks' import { useDeployWorkflow } from '@/hooks/queries/deployments' import { useWorkflowRegistry } from '@/stores/workflows/registry/store' +import { syncLocalDraftFromServer } from '@/stores/workflows/sync-local-draft' import { mergeSubblockState } from '@/stores/workflows/utils' import { useWorkflowStore } from '@/stores/workflows/workflow/store' import { releaseDeployAction, tryAcquireDeployAction } from './deploy-action-lock' -import { syncLocalDraftFromServer } from './sync-local-draft' import type { DeployReadiness } from './use-deploy-readiness' const logger = createLogger('UseDeployment') diff --git a/apps/sim/app/workspace/providers/socket-provider.tsx b/apps/sim/app/workspace/providers/socket-provider.tsx index 9035f519cb9..af6df147547 100644 --- a/apps/sim/app/workspace/providers/socket-provider.tsx +++ b/apps/sim/app/workspace/providers/socket-provider.tsx @@ -610,17 +610,35 @@ export function SocketProvider({ children, user }: SocketProviderProps) { eventHandlers.current.workflowDeployed?.(data) }) - const rehydrateWorkflowStores = async (workflowId: string, workflowState: any) => { + /** + * DEGRADED-PATH fallback: replaces the workflow + subblock stores with + * the raw state pushed by the realtime server. The primary join path is + * `syncLocalDraftFromServer`, which fetches MIGRATED state over HTTP — + * the realtime server loads state raw, without the app's block + * migrations (credential remaps, subblock-id renames, canonicalModes + * backfill), because those need the block registry that apps/realtime + * cannot import. Applying raw state is acceptable only as a fallback + * when the HTTP fetch fails: load-time migrations persist their result + * back to the normalized tables (persistMigratedBlocks), so raw state + * converges with migrated state after the first migrated load. If that + * persistence ever fails repeatedly, raw state diverges from the + * migrated deployed snapshot and change detection reports phantom + * diffs that survive redeploys (see the exact text-precision + * updated_at guard in persistMigratedBlocks). + */ + const applyRawJoinStateFallback = async (workflowId: string, workflowState: any) => { const [ { useOperationQueueStore }, { useWorkflowRegistry }, { useWorkflowStore }, { useSubBlockStore }, + { useWorkflowDiffStore }, ] = await Promise.all([ import('@/stores/operation-queue/store'), import('@/stores/workflows/registry/store'), import('@/stores/workflows/workflow/store'), import('@/stores/workflows/subblock/store'), + import('@/stores/workflow-diff/store'), ]) const { activeWorkflowId } = useWorkflowRegistry.getState() @@ -637,6 +655,11 @@ export function SocketProvider({ children, user }: SocketProviderProps) { return false } + if (useWorkflowDiffStore.getState().hasActiveDiff) { + logger.info('Skipping rehydration - an active diff is in progress') + return false + } + const subblockValues: Record> = {} Object.entries(workflowState.blocks || {}).forEach(([blockId, block]) => { const blockState = block as any @@ -734,6 +757,17 @@ export function SocketProvider({ children, user }: SocketProviderProps) { } }) + /** + * Join-time state sync. The realtime server can only load RAW workflow + * state (no block migrations — see applyRawJoinStateFallback), so the raw + * payload is used purely as a trigger and fallback: the stores are + * synced from the app's MIGRATED HTTP state via + * syncLocalDraftFromServer, which dedupes against an in-flight registry + * hydration on the shared query key and refuses to clobber pending + * local operations, active diffs, or newer remote updates. Only when + * that fetch itself fails does the raw payload get applied, so a + * reconnect on a flaky network still recovers to near-current state. + */ socketInstance.on('workflow-state', async (workflowData) => { logger.info('Received workflow state from server') @@ -751,10 +785,40 @@ export function SocketProvider({ children, user }: SocketProviderProps) { } if (workflowData?.state) { + const { canApplyDraftSnapshot, captureDraftVersions, syncLocalDraftFromServer } = + await import('@/stores/workflows/sync-local-draft') + const versionsAtJoin = captureDraftVersions(workflowData.id) + try { - await rehydrateWorkflowStores(workflowData.id, workflowData.state) + const synced = await syncLocalDraftFromServer(workflowData.id) + if (!synced) { + logger.info('Join-state sync skipped; keeping local state', { + workflowId: workflowData.id, + }) + } } catch (error) { - logger.error('Error rehydrating workflow state:', error) + logger.warn('Join-state sync failed; falling back to raw socket state', { error }) + + /** + * The raw payload was captured at join time. Anything applied to + * the stores while the HTTP sync was failing — local edits, + * remote broadcasts, or an external full reload + * (workflow-updated / revert) — is newer than the payload, so + * applying it would regress those changes. The shared snapshot + * guard covers all of those sources. + */ + if (!canApplyDraftSnapshot(workflowData.id, versionsAtJoin)) { + logger.info('Skipping raw join-state fallback; stores changed since join', { + workflowId: workflowData.id, + }) + return + } + + try { + await applyRawJoinStateFallback(workflowData.id, workflowData.state) + } catch (rehydrateError) { + logger.error('Error rehydrating workflow state:', rehydrateError) + } } } }) diff --git a/apps/sim/ee/workspace-forking/lib/background-work/store.test.ts b/apps/sim/ee/workspace-forking/lib/background-work/store.test.ts index bffc56645c9..04313635abe 100644 --- a/apps/sim/ee/workspace-forking/lib/background-work/store.test.ts +++ b/apps/sim/ee/workspace-forking/lib/background-work/store.test.ts @@ -88,9 +88,21 @@ describe('listSurfacedBackgroundWork', () => { it('trims the over-fetched row and encodes the next cursor from the last returned row', async () => { mockChildrenLookup([]) const rows = [ - { id: 'job-1', updatedAt: new Date('2026-07-01T10:00:00.000Z') }, - { id: 'job-2', updatedAt: new Date('2026-07-01T09:00:00.000Z') }, - { id: 'job-3', updatedAt: new Date('2026-07-01T08:00:00.000Z') }, + { + id: 'job-1', + updatedAt: new Date('2026-07-01T10:00:00.000Z'), + updatedAtCursor: '2026-07-01 10:00:00', + }, + { + id: 'job-2', + updatedAt: new Date('2026-07-01T09:00:00.000Z'), + updatedAtCursor: '2026-07-01 09:00:00', + }, + { + id: 'job-3', + updatedAt: new Date('2026-07-01T08:00:00.000Z'), + updatedAtCursor: '2026-07-01 08:00:00', + }, ] dbChainMockFns.limit.mockResolvedValueOnce(rows as never) @@ -100,14 +112,40 @@ describe('listSurfacedBackgroundWork', () => { expect(result.rows).toEqual(rows.slice(0, 2)) expect(result.nextCursor).not.toBeNull() expect(decodeCursor(result.nextCursor as string)).toEqual({ - updatedAt: '2026-07-01T09:00:00.000Z', + updatedAt: '2026-07-01 09:00:00', id: 'job-2', }) }) + it('encodes the cursor from the exact microsecond timestamp text, not the truncated Date', async () => { + mockChildrenLookup([]) + // The Date column value is millisecond-truncated by Drizzle; the cursor must + // carry the microsecond text so the keyset boundary is exact. + dbChainMockFns.limit.mockResolvedValueOnce([ + { + id: 'job-1', + updatedAt: new Date('2026-07-01T10:00:00.123Z'), + updatedAtCursor: '2026-07-01 10:00:00.123456', + }, + { + id: 'job-2', + updatedAt: new Date('2026-07-01T09:00:00.000Z'), + updatedAtCursor: '2026-07-01 09:00:00', + }, + ] as never) + + const result = await listSurfacedBackgroundWork(executor, 'ws-1', { limit: 1 }) + + expect(decodeCursor(result.nextCursor as string)).toEqual({ + updatedAt: '2026-07-01 10:00:00.123456', + id: 'job-1', + }) + }) + it('applies the cursor as a keyset condition with the id tiebreaker', async () => { mockChildrenLookup([]) - const cursor = encodeCursor({ updatedAt: '2026-07-01T09:00:00.000Z', id: 'job-2' }) + const cursorTimestamp = '2026-07-01 09:00:00.123456' + const cursor = encodeCursor({ updatedAt: cursorTimestamp, id: 'job-2' }) await listSurfacedBackgroundWork(executor, 'ws-1', { cursor }) @@ -115,21 +153,51 @@ describe('listSurfacedBackgroundWork', () => { expect(rowsWhere.type).toBe('and') expect(rowsWhere.conditions).toHaveLength(3) - const cursorDate = new Date('2026-07-01T09:00:00.000Z') + // The cursor timestamp is bound as a `::timestamp`-cast SQL fragment so the + // comparison happens at full microsecond precision in Postgres. + const expectedTimestampFragment = expect.objectContaining({ values: [cursorTimestamp] }) const keyset = (rowsWhere.conditions as MockCondition[])[2] - expect(keyset).toEqual({ - type: 'or', - conditions: [ - { type: 'lt', left: 'updatedAt', right: cursorDate }, - { - type: 'and', - conditions: [ - { type: 'eq', left: 'updatedAt', right: cursorDate }, - { type: 'lt', left: 'id', right: 'job-2' }, - ], - }, - ], - }) + expect(keyset).toEqual( + expect.objectContaining({ + type: 'or', + conditions: [ + expect.objectContaining({ + type: 'lt', + left: 'updatedAt', + right: expectedTimestampFragment, + }), + expect.objectContaining({ + type: 'and', + conditions: [ + expect.objectContaining({ + type: 'eq', + left: 'updatedAt', + right: expectedTimestampFragment, + }), + expect.objectContaining({ type: 'lt', left: 'id', right: 'job-2' }), + ], + }), + ], + }) + ) + }) + + it('accepts a legacy ISO cursor produced before the text-precision format', async () => { + mockChildrenLookup([]) + const cursor = encodeCursor({ updatedAt: '2026-07-01T09:00:00.000Z', id: 'job-2' }) + + await listSurfacedBackgroundWork(executor, 'ws-1', { cursor }) + + const rowsWhere = dbChainMockFns.where.mock.calls[1][0] as MockCondition + expect(rowsWhere.conditions).toHaveLength(3) + const keyset = (rowsWhere.conditions as MockCondition[])[2] + expect((keyset.conditions as MockCondition[])[0]).toEqual( + expect.objectContaining({ + type: 'lt', + left: 'updatedAt', + right: expect.objectContaining({ values: ['2026-07-01T09:00:00.000Z'] }), + }) + ) }) it('pages through rows with identical updatedAt via the id tiebreaker', async () => { @@ -137,21 +205,26 @@ describe('listSurfacedBackgroundWork', () => { // cursor must carry that id so the second page matches the remaining row // through the eq(updatedAt) + lt(id) arm instead of skipping it. const sharedAt = new Date('2026-07-01T09:00:00.000Z') + const sharedAtCursor = '2026-07-01 09:00:00' mockChildrenLookup([]) dbChainMockFns.limit.mockResolvedValueOnce([ - { id: 'job-b', updatedAt: sharedAt }, - { id: 'job-a', updatedAt: sharedAt }, + { id: 'job-b', updatedAt: sharedAt, updatedAtCursor: sharedAtCursor }, + { id: 'job-a', updatedAt: sharedAt, updatedAtCursor: sharedAtCursor }, ] as never) const firstPage = await listSurfacedBackgroundWork(executor, 'ws-1', { limit: 1 }) - expect(firstPage.rows).toEqual([{ id: 'job-b', updatedAt: sharedAt }]) + expect(firstPage.rows).toEqual([ + { id: 'job-b', updatedAt: sharedAt, updatedAtCursor: sharedAtCursor }, + ]) expect(decodeCursor(firstPage.nextCursor as string)).toEqual({ - updatedAt: sharedAt.toISOString(), + updatedAt: sharedAtCursor, id: 'job-b', }) mockChildrenLookup([]) - dbChainMockFns.limit.mockResolvedValueOnce([{ id: 'job-a', updatedAt: sharedAt }] as never) + dbChainMockFns.limit.mockResolvedValueOnce([ + { id: 'job-a', updatedAt: sharedAt, updatedAtCursor: sharedAtCursor }, + ] as never) const secondPage = await listSurfacedBackgroundWork(executor, 'ws-1', { cursor: firstPage.nextCursor as string, @@ -160,17 +233,47 @@ describe('listSurfacedBackgroundWork', () => { const rowsWhere = dbChainMockFns.where.mock.calls[3][0] as MockCondition const keyset = (rowsWhere.conditions as MockCondition[])[2] - expect(keyset.conditions?.[1]).toEqual({ - type: 'and', - conditions: [ - { type: 'eq', left: 'updatedAt', right: sharedAt }, - { type: 'lt', left: 'id', right: 'job-b' }, - ], - }) - expect(secondPage.rows).toEqual([{ id: 'job-a', updatedAt: sharedAt }]) + expect(keyset.conditions?.[1]).toEqual( + expect.objectContaining({ + type: 'and', + conditions: [ + expect.objectContaining({ + type: 'eq', + left: 'updatedAt', + right: expect.objectContaining({ values: [sharedAtCursor] }), + }), + expect.objectContaining({ type: 'lt', left: 'id', right: 'job-b' }), + ], + }) + ) + expect(secondPage.rows).toEqual([ + { id: 'job-a', updatedAt: sharedAt, updatedAtCursor: sharedAtCursor }, + ]) expect(secondPage.nextCursor).toBeNull() }) + it('ignores a cursor whose timestamp Postgres would reject and serves the first page', async () => { + // Feb 30 parses in JS (Date normalizes it to Mar 2) but fails a Postgres + // ::timestamp cast — it must degrade to the first page, not a 500. + mockChildrenLookup([]) + const cursor = encodeCursor({ updatedAt: '2026-02-30 09:00:00', id: 'job-2' }) + + await listSurfacedBackgroundWork(executor, 'ws-1', { cursor }) + + const rowsWhere = dbChainMockFns.where.mock.calls[1][0] as MockCondition + expect(rowsWhere.conditions).toHaveLength(2) + }) + + it('ignores a cursor with an out-of-range time and serves the first page', async () => { + mockChildrenLookup([]) + const cursor = encodeCursor({ updatedAt: '2026-07-01 99:00:00', id: 'job-2' }) + + await listSurfacedBackgroundWork(executor, 'ws-1', { cursor }) + + const rowsWhere = dbChainMockFns.where.mock.calls[1][0] as MockCondition + expect(rowsWhere.conditions).toHaveLength(2) + }) + it('ignores an undecodable cursor and serves the first page', async () => { mockChildrenLookup([]) diff --git a/apps/sim/ee/workspace-forking/lib/background-work/store.ts b/apps/sim/ee/workspace-forking/lib/background-work/store.ts index a0f86376b27..facb4431fbd 100644 --- a/apps/sim/ee/workspace-forking/lib/background-work/store.ts +++ b/apps/sim/ee/workspace-forking/lib/background-work/store.ts @@ -1,7 +1,19 @@ import { backgroundWorkStatus, workspace } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { generateId } from '@sim/utils/id' -import { and, desc, eq, inArray, isNull, lt, lte, or, type SQL, sql } from 'drizzle-orm' +import { + and, + desc, + eq, + getTableColumns, + inArray, + isNull, + lt, + lte, + or, + type SQL, + sql, +} from 'drizzle-orm' import type { DbOrTx } from '@/lib/db/types' const logger = createLogger('ForkBackgroundWork') @@ -185,7 +197,11 @@ function toMetadataRecord(value: unknown): Record { : {} } -/** Keyset position of the last row of a page: `updatedAt` plus the `id` tiebreaker. */ +/** + * Keyset position of the last row of a page: `updatedAt` plus the `id` tiebreaker. + * `updatedAt` carries the row's Postgres text rendering (full microsecond + * precision), not a JS `Date` serialization — see {@link buildCursorCondition}. + */ interface BackgroundWorkCursorData { updatedAt: string id: string @@ -204,22 +220,55 @@ function decodeCursor(cursor: string): BackgroundWorkCursorData | null { } } +/** + * Cursor timestamp formats this store has ever encoded: the Postgres text + * rendering (`YYYY-MM-DD HH:MM:SS[.ffffff]`) and the legacy `toISOString()` + * form (`YYYY-MM-DDTHH:MM:SS.fffZ`). + */ +const CURSOR_TIMESTAMP_PATTERN = + /^\d{4}-(0[1-9]|1[0-2])-(0[1-9]|[12]\d|3[01])[T ]([01]\d|2[0-3]):[0-5]\d:[0-5]\d(\.\d{1,6})?Z?$/ + +/** + * Validates a cursor timestamp before it is bound into a `::timestamp` cast. + * The pattern pins format and field ranges; the calendar check rejects + * impossible dates (e.g. Feb 30) that JS `Date` silently normalizes but + * Postgres rejects with an error — an invalid cursor must degrade to the + * first page, never fail the query. + */ +function isValidCursorTimestamp(value: string): boolean { + if (!CURSOR_TIMESTAMP_PATTERN.test(value)) return false + const [year, month, day] = value.slice(0, 10).split('-').map(Number) + const date = new Date(Date.UTC(year, month - 1, day)) + return date.getUTCMonth() === month - 1 && date.getUTCDate() === day +} + /** * Keyset condition for rows strictly after the cursor position in * `updatedAt DESC, id DESC` order: older `updatedAt`, or the same `updatedAt` * (which is not unique) with a smaller `id`. Null for an invalid cursor, which * degrades to the first page rather than erroring. + * + * The cursor timestamp is compared via a `::timestamp` cast of the captured + * text rather than a JS `Date`: Drizzle dates truncate Postgres microsecond + * timestamps to milliseconds, and a truncated boundary value both skips rows + * sharing the boundary row's millisecond and never matches the `eq` tiebreaker + * arm for microsecond-stamped rows. Legacy cursors that carry an ISO string + * (pre-text format) parse through the same cast — the zone suffix is dropped + * and the value is read as the same UTC wall time the app stores. */ function buildCursorCondition(cursor: string): SQL | null { const cursorData = decodeCursor(cursor) if (!cursorData?.updatedAt || !cursorData.id) return null - const cursorDate = new Date(cursorData.updatedAt) - if (Number.isNaN(cursorDate.getTime())) return null + if (!isValidCursorTimestamp(cursorData.updatedAt)) return null + const cursorTimestamp = sql`${cursorData.updatedAt}::timestamp` return or( - lt(backgroundWorkStatus.updatedAt, cursorDate), - and(eq(backgroundWorkStatus.updatedAt, cursorDate), lt(backgroundWorkStatus.id, cursorData.id)) + lt(backgroundWorkStatus.updatedAt, cursorTimestamp), + and( + eq(backgroundWorkStatus.updatedAt, cursorTimestamp), + lt(backgroundWorkStatus.id, cursorData.id) + ) )! } @@ -286,18 +335,23 @@ export async function listSurfacedBackgroundWork( } // Over-fetch by one row to learn whether another page exists (audit-log pattern). + // `updatedAtCursor` captures the exact microsecond timestamp for the keyset + // cursor; the Date-typed `updatedAt` is millisecond-truncated by Drizzle. const rows = (await executor - .select() + .select({ + ...getTableColumns(backgroundWorkStatus), + updatedAtCursor: sql`${backgroundWorkStatus.updatedAt}::text`, + }) .from(backgroundWorkStatus) .where(and(...conditions)) .orderBy(desc(backgroundWorkStatus.updatedAt), desc(backgroundWorkStatus.id)) - .limit(limit + 1)) as BackgroundWorkRow[] + .limit(limit + 1)) as Array const hasMore = rows.length > limit const page = rows.slice(0, limit) const last = page[page.length - 1] const nextCursor = - hasMore && last ? encodeCursor({ updatedAt: last.updatedAt.toISOString(), id: last.id }) : null + hasMore && last ? encodeCursor({ updatedAt: last.updatedAtCursor, id: last.id }) : null return { rows: page, nextCursor } } diff --git a/apps/sim/hooks/use-collaborative-workflow.ts b/apps/sim/hooks/use-collaborative-workflow.ts index 2f63a73f30a..e0c752bb446 100644 --- a/apps/sim/hooks/use-collaborative-workflow.ts +++ b/apps/sim/hooks/use-collaborative-workflow.ts @@ -527,6 +527,19 @@ export function useCollaborativeWorkflow() { } } } + + /** + * Per-frame `update-position` broadcasts are not applied by this + * handler (positions land via BATCH_UPDATE_POSITIONS commits), so they + * must not count as applied remote state changes — a collaborator's + * drag would otherwise exhaust sync refetch attempts for nothing. + */ + const isUnappliedPositionFrame = + target === OPERATION_TARGETS.BLOCK && operation === BLOCK_OPERATIONS.UPDATE_POSITION + const appliedWorkflowId = metadata?.workflowId || activeWorkflowId + if (!isUnappliedPositionFrame && appliedWorkflowId) { + useOperationQueueStore.getState().markRemoteApplied(appliedWorkflowId) + } } catch (error) { logger.error('Error applying remote operation:', error) } finally { @@ -559,6 +572,11 @@ export function useCollaborativeWorkflow() { if (activeWorkflowId && blockType === 'function' && subblockId === 'code') { useCodeUndoRedoStore.getState().clear(activeWorkflowId, blockId, subblockId) } + + const appliedWorkflowId = workflowId || activeWorkflowId + if (appliedWorkflowId) { + useOperationQueueStore.getState().markRemoteApplied(appliedWorkflowId) + } } catch (error) { logger.error('Error applying remote subblock update:', error) } finally { @@ -585,6 +603,7 @@ export function useCollaborativeWorkflow() { isApplyingRemoteChange.current = true try { + const isKnownField = field === 'name' || field === 'value' || field === 'type' if (field === 'name') { useVariablesStore.getState().updateVariable(variableId, { name: value }) } else if (field === 'value') { @@ -592,6 +611,11 @@ export function useCollaborativeWorkflow() { } else if (field === 'type') { useVariablesStore.getState().updateVariable(variableId, { type: value }) } + + const appliedWorkflowId = workflowId || activeWorkflowId + if (isKnownField && appliedWorkflowId) { + useOperationQueueStore.getState().markRemoteApplied(appliedWorkflowId) + } } catch (error) { logger.error('Error applying remote variable update:', error) } finally { diff --git a/apps/sim/lib/workflows/comparison/compare.test.ts b/apps/sim/lib/workflows/comparison/compare.test.ts index 8b39f481be5..e727b9d1513 100644 --- a/apps/sim/lib/workflows/comparison/compare.test.ts +++ b/apps/sim/lib/workflows/comparison/compare.test.ts @@ -777,6 +777,209 @@ describe('hasWorkflowChanged', () => { }) }) + describe('Table SubBlock Special Handling', () => { + it.concurrent('should ignore non-deterministic table row ids', () => { + const state1 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { + id: 'headers', + type: 'table', + value: [{ id: 'row-a', cells: { Key: 'Authorization', Value: 'Bearer x' } }], + }, + }, + }), + }, + }) + const state2 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { + id: 'headers', + type: 'table', + value: [{ id: 'row-b', cells: { Key: 'Authorization', Value: 'Bearer x' } }], + }, + }, + }), + }, + }) + expect(hasWorkflowChanged(state1, state2)).toBe(false) + }) + + it.concurrent('should treat a null table and a single blank starter row as equal', () => { + const state1 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { id: 'headers', type: 'table', value: null }, + }, + }), + }, + }) + const state2 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { + id: 'headers', + type: 'table', + value: [{ id: 'row-a', cells: { Key: '', Value: '' } }], + }, + }, + }), + }, + }) + expect(hasWorkflowChanged(state1, state2)).toBe(false) + }) + + it.concurrent('should ignore a trailing blank row after a populated row', () => { + const state1 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { + id: 'headers', + type: 'table', + value: [{ id: 'row-a', cells: { Key: 'Accept', Value: 'application/json' } }], + }, + }, + }), + }, + }) + const state2 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { + id: 'headers', + type: 'table', + value: [ + { id: 'row-b', cells: { Key: 'Accept', Value: 'application/json' } }, + { id: 'row-c', cells: { Key: '', Value: '' } }, + ], + }, + }, + }), + }, + }) + expect(hasWorkflowChanged(state1, state2)).toBe(false) + }) + + it.concurrent('should detect real table cell changes', () => { + const state1 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { + id: 'headers', + type: 'table', + value: [{ id: 'row-a', cells: { Key: 'Accept', Value: 'application/json' } }], + }, + }, + }), + }, + }) + const state2 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { + id: 'headers', + type: 'table', + value: [{ id: 'row-a', cells: { Key: 'Accept', Value: 'text/plain' } }], + }, + }, + }), + }, + }) + expect(hasWorkflowChanged(state1, state2)).toBe(true) + }) + + it.concurrent('should preserve flat rows without a cells object', () => { + const state1 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { id: 'headers', type: 'table', value: null }, + }, + }), + }, + }) + const state2 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { + id: 'headers', + type: 'table', + value: [{ name: 'FOO', value: 'bar' }], + }, + }, + }), + }, + }) + expect(hasWorkflowChanged(state1, state2)).toBe(true) + }) + + it.concurrent('should detect edits to flat rows without a cells object', () => { + const state1 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { + id: 'headers', + type: 'table', + value: [{ name: 'FOO', value: 'bar' }], + }, + }, + }), + }, + }) + const state2 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { + id: 'headers', + type: 'table', + value: [{ name: 'FOO', value: 'baz' }], + }, + }, + }), + }, + }) + expect(hasWorkflowChanged(state1, state2)).toBe(true) + }) + + it.concurrent('should detect rows with a partially blank cell', () => { + const state1 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { id: 'headers', type: 'table', value: null }, + }, + }), + }, + }) + const state2 = createWorkflowState({ + blocks: { + block1: createBlock('block1', { + subBlocks: { + headers: { + id: 'headers', + type: 'table', + value: [{ id: 'row-a', cells: { Key: 'Accept', Value: '' } }], + }, + }, + }), + }, + }) + expect(hasWorkflowChanged(state1, state2)).toBe(true) + }) + }) + describe('Loop Changes', () => { it.concurrent('should detect added loops', () => { const state1 = createWorkflowState({ loops: {} }) diff --git a/apps/sim/lib/workflows/comparison/compare.ts b/apps/sim/lib/workflows/comparison/compare.ts index be5e6ab7820..0b402b913f7 100644 --- a/apps/sim/lib/workflows/comparison/compare.ts +++ b/apps/sim/lib/workflows/comparison/compare.ts @@ -244,8 +244,9 @@ export function generateWorkflowDiffSummary( continue } - const currentValue = normalizeSubBlockValue(subId, currentSub.value) - const previousValue = normalizeSubBlockValue(subId, previousSub.value) + const subType = currentSub.type ?? previousSub.type + const currentValue = normalizeSubBlockValue(subId, currentSub.value, subType) + const previousValue = normalizeSubBlockValue(subId, previousSub.value, subType) if (typeof currentValue === 'string' && typeof previousValue === 'string') { if (currentValue !== previousValue) { diff --git a/apps/sim/lib/workflows/comparison/normalize.ts b/apps/sim/lib/workflows/comparison/normalize.ts index 8e4f97af77f..1078d568d15 100644 --- a/apps/sim/lib/workflows/comparison/normalize.ts +++ b/apps/sim/lib/workflows/comparison/normalize.ts @@ -231,6 +231,50 @@ export function normalizeVariables(variables: unknown): Record return variables as Record } +/** Table row shape produced by the table subblock component */ +type TableRowLike = Record & { id?: unknown; cells?: Record } + +function isBlankTableCellValue(value: unknown): boolean { + return value === '' || value === null || value === undefined +} + +/** + * Sanitizes table subblock rows for comparison. Only rows in the editor's + * canonical `{ id, cells }` shape are normalized: + * - The row `id` is dropped — it is a client-generated React key + * (`generateId()`), never read at execution (`transformTable` / response + * `parseHeaders` consume cells only), and non-deterministic: the table + * component materializes an empty starter row with a fresh id on mount. + * - Rows whose cells are all empty are dropped — execution ignores rows + * without a truthy Key, and the editor persists one blank starter row for + * an empty table, so a blank row is presentation, not configuration. + * + * Rows in any other shape (e.g. copilot-written flat rows like + * `{ name, value }`) pass through verbatim — their keys carry data, so + * nothing about them may be assumed blank or UI-only. + * + * @param rows - Array of table row values + * @returns Sanitized rows array + */ +export function sanitizeTableRows(rows: unknown[]): unknown[] { + const sanitized: unknown[] = [] + for (const row of rows) { + if (!row || typeof row !== 'object' || Array.isArray(row)) { + sanitized.push(row) + continue + } + const { cells } = row as TableRowLike + if (!cells || typeof cells !== 'object' || Array.isArray(cells)) { + sanitized.push(row) + continue + } + if (Object.values(cells).every(isBlankTableCellValue)) continue + const { id: _id, ...rest } = row as TableRowLike + sanitized.push(rest) + } + return sanitized +} + /** Input format item with optional UI-only fields */ type InputFormatItem = Record & { collapsed?: boolean } @@ -465,13 +509,19 @@ export function normalizeTriggerConfigValues( /** * Normalizes a subBlock value with sanitization for specific subBlock types. - * Sanitizes: tools (removes isExpanded), inputFormat (removes collapsed) + * Sanitizes: tools (removes isExpanded), inputFormat (removes collapsed), + * table values (removes non-deterministic row ids and all-blank starter rows). * * @param subBlockId - The subBlock ID * @param value - The subBlock value + * @param subBlockType - Optional subBlock type for type-based sanitization * @returns Normalized value */ -export function normalizeSubBlockValue(subBlockId: string, value: unknown): unknown { +export function normalizeSubBlockValue( + subBlockId: string, + value: unknown, + subBlockType?: unknown +): unknown { let normalizedValue = value ?? null if (subBlockId === 'tools' && Array.isArray(normalizedValue)) { @@ -480,6 +530,10 @@ export function normalizeSubBlockValue(subBlockId: string, value: unknown): unkn if (subBlockId === 'inputFormat' && Array.isArray(normalizedValue)) { normalizedValue = sanitizeInputFormat(normalizedValue) } + if (subBlockType === 'table' && Array.isArray(normalizedValue)) { + const sanitizedRows = sanitizeTableRows(normalizedValue) + normalizedValue = sanitizedRows.length > 0 ? sanitizedRows : null + } return normalizedValue } @@ -530,7 +584,7 @@ export function normalizeWorkflowState(state: WorkflowState): NormalizedWorkflow for (const subBlockId of subBlockIds) { const subBlock = blockSubBlocks[subBlockId] as SubBlockWithDiffMarker - const value = normalizeSubBlockValue(subBlockId, subBlock.value) + const value = normalizeSubBlockValue(subBlockId, subBlock.value, subBlock.type) const subBlockRest = extractSubBlockRest(subBlock as Record) normalizedSubBlocks[subBlockId] = { diff --git a/apps/sim/stores/AGENTS.md b/apps/sim/stores/AGENTS.md index bef10dbc6fc..8883b425d8e 100644 --- a/apps/sim/stores/AGENTS.md +++ b/apps/sim/stores/AGENTS.md @@ -3,3 +3,27 @@ Applies to Zustand stores under `apps/sim/**/stores/**` and `apps/sim/**/store.ts`. Store authoring rules — `devtools` middleware, `persist` + `partialize` only for reload-surviving state, immutable updates, `set((state) => ...)` for previous-state-dependent updates, `reset()` action, splitting complex stores into `store.ts` + `types.ts`, and `_hasHydrated` tracking — live in `.claude/rules/sim-stores.md`. + +## Workflow value state invariants + +Workflow state is split across two stores on purpose: `useWorkflowStore` holds block +structure (plus a hydration-time copy of each subblock value) and `useSubBlockStore` +holds live values, so per-keystroke edits don't re-render the canvas. Rules that keep +this split correct: + +- The structure's `subBlocks[*].value` is stale after any edit. Never read it directly + for a current value — merge via `mergeSubblockState`/`mergeSubblockStateWithValues` + (single implementation in `@sim/workflow-persistence/subblocks`) or read the + subblock store. Exception: condition/router dynamic-handle subblocks dual-write the + structure (`syncDynamicHandleSubblockValue`) and may be read from either source. +- Merge semantics are tri-state: a key present in the subblock store wins — including + `null`, which means "explicitly cleared". Absent/`undefined` falls back to the + structure. Do not add merge or precedence logic anywhere else; if a new reader needs + different semantics, extend the shared merge. +- Every subblock-store write must go through `collaborativeSetSubblockValue` (or a + batch equivalent) so the identical value is persisted via the realtime server. A + store write that skips persistence makes the client's merged state diverge from the + DB draft, which deploy snapshots — producing phantom "Update" states on the deploy + button that clear on refresh. Hydration-derived local-only writes are allowed only + when change detection compensates (see `populateTriggerFieldsFromConfig` + + `normalizeTriggerConfigValues`). diff --git a/apps/sim/stores/operation-queue/store.ts b/apps/sim/stores/operation-queue/store.ts index 200abdf2b3f..b3feae6b49b 100644 --- a/apps/sim/stores/operation-queue/store.ts +++ b/apps/sim/stores/operation-queue/store.ts @@ -118,9 +118,19 @@ function dropOperationForMissingTarget(operation: QueuedOperation): boolean { export const useOperationQueueStore = create((set, get) => ({ operations: [], workflowOperationVersions: {}, + remoteApplyVersions: {}, isProcessing: false, hasOperationError: false, + markRemoteApplied: (workflowId) => { + set((state) => ({ + remoteApplyVersions: { + ...state.remoteApplyVersions, + [workflowId]: (state.remoteApplyVersions[workflowId] ?? 0) + 1, + }, + })) + }, + addToQueue: (operation) => { set((state) => ({ workflowOperationVersions: { diff --git a/apps/sim/stores/operation-queue/types.ts b/apps/sim/stores/operation-queue/types.ts index fb56a2a694b..796fc527ae7 100644 --- a/apps/sim/stores/operation-queue/types.ts +++ b/apps/sim/stores/operation-queue/types.ts @@ -44,10 +44,20 @@ export interface QueuedOperation { export interface OperationQueueState { operations: QueuedOperation[] workflowOperationVersions: Record + /** + * Per-workflow counter bumped every time a REMOTE collaborator's operation + * (block op, subblock update, variable update) is applied to the local + * stores. Lets full-state syncs (`syncLocalDraftFromServer`) detect that a + * remote edit landed while their fetch was in flight — the fetched snapshot + * may predate that edit's persist — and refetch instead of clobbering it. + */ + remoteApplyVersions: Record isProcessing: boolean hasOperationError: boolean addToQueue: (operation: Omit) => void + /** Records that a remote collaborator's operation was applied to local stores. */ + markRemoteApplied: (workflowId: string) => void confirmOperation: (operationId: string) => void failOperation: (operationId: string, retryable?: boolean) => void handleOperationTimeout: (operationId: string) => void diff --git a/apps/sim/stores/workflows/subblock/store.ts b/apps/sim/stores/workflows/subblock/store.ts index 7cb745b9fdd..c723f274bca 100644 --- a/apps/sim/stores/workflows/subblock/store.ts +++ b/apps/sim/stores/workflows/subblock/store.ts @@ -24,15 +24,36 @@ export const EMPTY_SUBBLOCK_VALUES: Record export const EMPTY_BLOCK_SUBBLOCK_VALUES: Record = {} /** - * SubBlockState stores values for all subblocks in workflows + * SubBlockState stores values for all subblocks in workflows. * - * Important implementation notes: - * 1. Values are stored per workflow, per block, per subblock - * 2. When workflows are synced to the database, the mergeSubblockState function - * in utils.ts combines the block structure with these values - * 3. If a subblock value exists here but not in the block structure - * (e.g., inputFormat in starter block), the merge function will include it - * in the synchronized state to ensure persistence + * Architecture: values deliberately live here, split from the workflow store's + * block structure, so per-keystroke edits do not re-render the canvas graph. + * The structure keeps a copy of each value from hydration time only — it goes + * stale as soon as a value is edited (except condition/router dynamic-handle + * subblocks, which dual-write the structure because edge handles derive from + * it). Whenever full state is needed (change detection, export, manual runs, + * diffs), mergeSubblockState/mergeSubblockStateWithValues joins this store + * onto the structure. + * + * Value semantics at that join (single source of truth for the contract): + * - Key present with any value, including null: this store wins. Null means + * the user explicitly cleared the field — it must NOT fall back to the + * structure's stale copy, or cleared fields resurrect in comparisons and + * serialization while the DB (correctly) holds null. + * - Key absent or undefined: no value recorded; the structure's value stands. + * - initializeFromWorkflow seeds every structure key (nulls included) at + * hydration, so post-hydration "present with null" is always meaningful. + * + * Persistence: user-authored edits flow through collaborativeSetSubblockValue, + * which updates this store and queues the identical value to the realtime + * server — the client's merged state and the DB draft stay equivalent, which + * is what keeps deploy-time change detection honest (deploy snapshots the DB + * draft). Direct setValue callers do not persist, and each is safe for a + * different reason: remote-broadcast application (already persisted + * server-side), undo/redo (persists via its own queued inverse operations), + * webhook management (writes trigger-runtime ids the comparison excludes), and + * populateTriggerFieldsFromConfig (values derived from the persisted + * triggerConfig aggregate, compensated via normalizeTriggerConfigValues). */ export const useSubBlockStore = create()( @@ -60,16 +81,25 @@ export const useSubBlockStore = create()( } } - if (!row.id) { - row.id = generateId() + const needsId = !row.id + const needsCells = !row.cells || typeof row.cells !== 'object' + if (!needsId && !needsCells) { + return row } - - if (!row.cells || typeof row.cells !== 'object') { + if (needsCells) { logger.warn('Fixing malformed table row cells', { blockId, subBlockId, row }) - row.cells = { Key: '', Value: '' } } - return row + /** + * Repair on a copy: the incoming rows are shared with the caller + * (component state, socket payloads), and mutating them in place + * corrupts state owned elsewhere. Valid rows keep their identity. + */ + return { + ...row, + ...(needsId ? { id: generateId() } : {}), + ...(needsCells ? { cells: { Key: '', Value: '' } } : {}), + } }) } } diff --git a/apps/sim/stores/workflows/subblock/types.ts b/apps/sim/stores/workflows/subblock/types.ts index f9137d52ad8..f62410bd4af 100644 --- a/apps/sim/stores/workflows/subblock/types.ts +++ b/apps/sim/stores/workflows/subblock/types.ts @@ -13,6 +13,15 @@ interface SubBlockStoreState { export interface SubBlockStore extends SubBlockStoreState { setValue: (blockId: string, subBlockId: string, value: SubBlockValue) => void + /** + * Point read for display purposes. Collapses "no value recorded" and + * "explicitly cleared" into `null` (`?? null` on the raw map) — fine for + * rendering, where both states look empty. Consumers that need the + * tri-state distinction (present-null means cleared, absent means unknown; + * see the merge semantics in `@sim/workflow-persistence/subblocks`) must + * read `workflowValues` directly and use key presence, as + * `mergeSubblockStateWithValues` does. + */ getValue: (blockId: string, subBlockId: string) => SubBlockValue | undefined clear: () => void initializeFromWorkflow: (workflowId: string, blocks: Record) => void diff --git a/apps/sim/stores/workflows/sync-local-draft.test.ts b/apps/sim/stores/workflows/sync-local-draft.test.ts new file mode 100644 index 00000000000..fdf0b8831c2 --- /dev/null +++ b/apps/sim/stores/workflows/sync-local-draft.test.ts @@ -0,0 +1,390 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { + mockFetchQuery, + mockApplyWorkflowStateToStores, + mockGetRegistryState, + mockHasPendingOperations, + mockGetOperationQueueState, + mockGetWorkflowDiffState, +} = vi.hoisted(() => ({ + mockFetchQuery: vi.fn(), + mockApplyWorkflowStateToStores: vi.fn(), + mockGetRegistryState: vi.fn(() => ({ activeWorkflowId: 'workflow-a' })), + mockHasPendingOperations: vi.fn(() => false), + mockGetOperationQueueState: vi.fn(() => ({ + hasPendingOperations: mockHasPendingOperations, + workflowOperationVersions: {}, + remoteApplyVersions: {}, + })), + mockGetWorkflowDiffState: vi.fn(() => ({ + hasActiveDiff: false, + pendingExternalUpdates: {}, + reconcilingWorkflows: {}, + reconciliationErrors: {}, + remoteUpdateVersions: {}, + })), +})) + +vi.mock('@/app/_shell/providers/get-query-client', () => ({ + getQueryClient: () => ({ fetchQuery: mockFetchQuery }), +})) + +vi.mock('@/hooks/queries/utils/fetch-workflow-envelope', () => ({ + fetchWorkflowEnvelope: vi.fn(), +})) + +vi.mock('@/hooks/queries/utils/workflow-keys', () => ({ + workflowKeys: { + state: (id: string) => ['workflow', 'state', id], + }, +})) + +vi.mock('@/stores/workflow-diff/utils', () => ({ + applyWorkflowStateToStores: mockApplyWorkflowStateToStores, +})) + +vi.mock('@/stores/workflow-diff/store', () => ({ + useWorkflowDiffStore: { + getState: mockGetWorkflowDiffState, + }, +})) + +vi.mock('@/stores/operation-queue/store', () => ({ + useOperationQueueStore: { + getState: mockGetOperationQueueState, + }, +})) + +vi.mock('@/stores/workflows/registry/store', () => ({ + useWorkflowRegistry: { + getState: mockGetRegistryState, + }, +})) + +import { + canApplyDraftSnapshot, + captureDraftVersions, + syncLocalDraftFromServer, +} from '@/stores/workflows/sync-local-draft' + +function buildEnvelopeState() { + return { + blocks: {}, + edges: [], + loops: {}, + parallels: {}, + lastSaved: 1, + } +} + +describe('syncLocalDraftFromServer', () => { + beforeEach(() => { + vi.clearAllMocks() + mockGetRegistryState.mockReturnValue({ activeWorkflowId: 'workflow-a' }) + mockHasPendingOperations.mockReturnValue(false) + mockGetOperationQueueState.mockImplementation(() => ({ + hasPendingOperations: mockHasPendingOperations, + workflowOperationVersions: {}, + remoteApplyVersions: {}, + })) + mockGetWorkflowDiffState.mockReturnValue({ + hasActiveDiff: false, + pendingExternalUpdates: {}, + reconcilingWorkflows: {}, + reconciliationErrors: {}, + remoteUpdateVersions: {}, + }) + }) + + it('fetches through the shared workflow-state query key with a fresh fetch', async () => { + mockFetchQuery.mockResolvedValue({ state: buildEnvelopeState(), variables: {} }) + + await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(true) + + expect(mockFetchQuery).toHaveBeenCalledWith( + expect.objectContaining({ + queryKey: ['workflow', 'state', 'workflow-a'], + staleTime: 0, + }) + ) + }) + + it('hydrates sibling workflow variables into the applied workflow state', async () => { + mockFetchQuery.mockResolvedValue({ + state: buildEnvelopeState(), + variables: { + 'variable-a': { + id: 'variable-a', + name: 'API_KEY', + type: 'plain', + value: 'secret', + }, + }, + }) + + await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(true) + + expect(mockApplyWorkflowStateToStores).toHaveBeenCalledWith( + 'workflow-a', + expect.objectContaining({ + variables: { + 'variable-a': { + id: 'variable-a', + name: 'API_KEY', + type: 'plain', + value: 'secret', + }, + }, + }), + { updateLastSaved: true } + ) + }) + + it('does not mutate the shared query-cache envelope when stamping variables', async () => { + const envelope = { + state: buildEnvelopeState(), + variables: { 'variable-a': { id: 'variable-a', name: 'X', type: 'plain', value: '1' } }, + } + mockFetchQuery.mockResolvedValue(envelope) + + await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(true) + + expect(Object.hasOwn(envelope.state, 'variables')).toBe(false) + const appliedState = mockApplyWorkflowStateToStores.mock.calls[0][1] + expect(appliedState).not.toBe(envelope.state) + }) + + it('does not apply a fetched draft after navigation changes the active workflow', async () => { + mockGetRegistryState + .mockReturnValueOnce({ activeWorkflowId: 'workflow-a' }) + .mockReturnValueOnce({ activeWorkflowId: 'workflow-b' }) + mockFetchQuery.mockResolvedValue({ state: buildEnvelopeState(), variables: {} }) + + await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(false) + + expect(mockApplyWorkflowStateToStores).not.toHaveBeenCalled() + }) + + it('does not synthesize an empty variables object when the server omits variables', async () => { + mockFetchQuery.mockResolvedValue({ state: buildEnvelopeState() }) + + await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(true) + + const appliedState = mockApplyWorkflowStateToStores.mock.calls[0][1] + expect(Object.hasOwn(appliedState, 'variables')).toBe(false) + }) + + it('does not apply a fetched draft over newly queued local operations', async () => { + mockHasPendingOperations.mockReturnValueOnce(false).mockReturnValueOnce(true) + mockFetchQuery.mockResolvedValue({ state: buildEnvelopeState(), variables: {} }) + + await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(false) + + expect(mockApplyWorkflowStateToStores).not.toHaveBeenCalled() + }) + + it('does not apply a fetched draft when an active diff is showing', async () => { + mockGetWorkflowDiffState.mockReturnValue({ + hasActiveDiff: true, + pendingExternalUpdates: {}, + reconcilingWorkflows: {}, + reconciliationErrors: {}, + remoteUpdateVersions: {}, + }) + mockFetchQuery.mockResolvedValue({ state: buildEnvelopeState(), variables: {} }) + + await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(false) + + expect(mockApplyWorkflowStateToStores).not.toHaveBeenCalled() + }) + + it('does not apply a fetched draft when a newer remote update arrives during fetch', async () => { + mockGetWorkflowDiffState + .mockReturnValueOnce({ + hasActiveDiff: false, + pendingExternalUpdates: {}, + reconcilingWorkflows: {}, + reconciliationErrors: {}, + remoteUpdateVersions: {}, + }) + .mockReturnValueOnce({ + hasActiveDiff: false, + pendingExternalUpdates: {}, + reconcilingWorkflows: {}, + reconciliationErrors: {}, + remoteUpdateVersions: { 'workflow-a': 1 }, + }) + mockFetchQuery.mockResolvedValue({ state: buildEnvelopeState(), variables: {} }) + + await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(false) + + expect(mockApplyWorkflowStateToStores).not.toHaveBeenCalled() + }) + + it('does not apply a fetched draft when local operations queue and drain during fetch', async () => { + mockGetOperationQueueState + .mockReturnValueOnce({ + hasPendingOperations: mockHasPendingOperations, + workflowOperationVersions: {}, + remoteApplyVersions: {}, + }) + .mockReturnValueOnce({ + hasPendingOperations: mockHasPendingOperations, + workflowOperationVersions: {}, + remoteApplyVersions: {}, + }) + .mockReturnValueOnce({ + hasPendingOperations: mockHasPendingOperations, + workflowOperationVersions: {}, + remoteApplyVersions: {}, + }) + .mockReturnValueOnce({ + hasPendingOperations: mockHasPendingOperations, + workflowOperationVersions: { 'workflow-a': 1 }, + remoteApplyVersions: {}, + }) + mockFetchQuery.mockResolvedValue({ state: buildEnvelopeState(), variables: {} }) + + await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(false) + + expect(mockApplyWorkflowStateToStores).not.toHaveBeenCalled() + }) + + it('refetches when a remote op is applied during the fetch, then applies the fresh snapshot', async () => { + const queueState = { + hasPendingOperations: mockHasPendingOperations, + workflowOperationVersions: {} as Record, + remoteApplyVersions: {} as Record, + } + mockGetOperationQueueState.mockImplementation(() => queueState) + + const staleEnvelope = { state: buildEnvelopeState(), variables: {} } + const freshEnvelope = { state: { ...buildEnvelopeState(), lastSaved: 2 }, variables: {} } + mockFetchQuery + .mockImplementationOnce(async () => { + queueState.remoteApplyVersions = { 'workflow-a': 1 } + return staleEnvelope + }) + .mockResolvedValueOnce(freshEnvelope) + + await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(true) + + expect(mockFetchQuery).toHaveBeenCalledTimes(2) + expect(mockApplyWorkflowStateToStores).toHaveBeenCalledWith( + 'workflow-a', + expect.objectContaining({ lastSaved: 2 }), + { updateLastSaved: true } + ) + }) + + it('applies the latest snapshot after exhausting retries during a busy remote session', async () => { + const queueState = { + hasPendingOperations: mockHasPendingOperations, + workflowOperationVersions: {} as Record, + remoteApplyVersions: {} as Record, + } + mockGetOperationQueueState.mockImplementation(() => queueState) + + let remoteVersion = 0 + mockFetchQuery.mockImplementation(async () => { + remoteVersion += 1 + queueState.remoteApplyVersions = { 'workflow-a': remoteVersion } + return { state: { ...buildEnvelopeState(), lastSaved: remoteVersion }, variables: {} } + }) + + await expect(syncLocalDraftFromServer('workflow-a')).resolves.toBe(true) + + expect(mockFetchQuery).toHaveBeenCalledTimes(3) + expect(mockApplyWorkflowStateToStores).toHaveBeenCalledWith( + 'workflow-a', + expect.objectContaining({ lastSaved: 3 }), + { updateLastSaved: true } + ) + }) + + it('propagates fetch failures to the caller', async () => { + mockFetchQuery.mockRejectedValue(new Error('network down')) + + await expect(syncLocalDraftFromServer('workflow-a')).rejects.toThrow('network down') + + expect(mockApplyWorkflowStateToStores).not.toHaveBeenCalled() + }) +}) + +describe('canApplyDraftSnapshot', () => { + beforeEach(() => { + vi.clearAllMocks() + mockGetRegistryState.mockReturnValue({ activeWorkflowId: 'workflow-a' }) + mockHasPendingOperations.mockReturnValue(false) + mockGetOperationQueueState.mockImplementation(() => ({ + hasPendingOperations: mockHasPendingOperations, + workflowOperationVersions: {}, + remoteApplyVersions: {}, + })) + mockGetWorkflowDiffState.mockReturnValue({ + hasActiveDiff: false, + pendingExternalUpdates: {}, + reconcilingWorkflows: {}, + reconciliationErrors: {}, + remoteUpdateVersions: {}, + }) + }) + + it('allows applying when nothing moved since capture', () => { + const versions = captureDraftVersions('workflow-a') + + expect(canApplyDraftSnapshot('workflow-a', versions)).toBe(true) + }) + + it('refuses when an external full reload (workflow-updated) landed since capture', () => { + const versions = captureDraftVersions('workflow-a') + mockGetWorkflowDiffState.mockReturnValue({ + hasActiveDiff: false, + pendingExternalUpdates: {}, + reconcilingWorkflows: {}, + reconciliationErrors: {}, + remoteUpdateVersions: { 'workflow-a': 1 }, + }) + + expect(canApplyDraftSnapshot('workflow-a', versions)).toBe(false) + }) + + it('refuses while an external reload is still reconciling', () => { + const versions = captureDraftVersions('workflow-a') + mockGetWorkflowDiffState.mockReturnValue({ + hasActiveDiff: false, + pendingExternalUpdates: {}, + reconcilingWorkflows: { 'workflow-a': true }, + reconciliationErrors: {}, + remoteUpdateVersions: {}, + }) + + expect(canApplyDraftSnapshot('workflow-a', versions)).toBe(false) + }) + + it('refuses when a remote collaborator op was applied since capture', () => { + const versions = captureDraftVersions('workflow-a') + mockGetOperationQueueState.mockImplementation(() => ({ + hasPendingOperations: mockHasPendingOperations, + workflowOperationVersions: {}, + remoteApplyVersions: { 'workflow-a': 1 }, + })) + + expect(canApplyDraftSnapshot('workflow-a', versions)).toBe(false) + }) + + it('refuses when a local edit was enqueued since capture', () => { + const versions = captureDraftVersions('workflow-a') + mockGetOperationQueueState.mockImplementation(() => ({ + hasPendingOperations: mockHasPendingOperations, + workflowOperationVersions: { 'workflow-a': 1 }, + remoteApplyVersions: {}, + })) + + expect(canApplyDraftSnapshot('workflow-a', versions)).toBe(false) + }) +}) diff --git a/apps/sim/stores/workflows/sync-local-draft.ts b/apps/sim/stores/workflows/sync-local-draft.ts new file mode 100644 index 00000000000..c62e6fb68a6 --- /dev/null +++ b/apps/sim/stores/workflows/sync-local-draft.ts @@ -0,0 +1,159 @@ +import { createLogger } from '@sim/logger' +import { sleep } from '@sim/utils/helpers' +import { getQueryClient } from '@/app/_shell/providers/get-query-client' +import { fetchWorkflowEnvelope } from '@/hooks/queries/utils/fetch-workflow-envelope' +import { workflowKeys } from '@/hooks/queries/utils/workflow-keys' +import { useOperationQueueStore } from '@/stores/operation-queue/store' +import { useWorkflowDiffStore } from '@/stores/workflow-diff/store' +import { applyWorkflowStateToStores } from '@/stores/workflow-diff/utils' +import { useWorkflowRegistry } from '@/stores/workflows/registry/store' +import type { WorkflowState } from '@/stores/workflows/workflow/types' + +const logger = createLogger('SyncLocalDraft') + +/** + * A remote collaborator's edit applied mid-fetch means the fetched snapshot may + * predate that edit's persist (the realtime server debounces writes), so the + * sync refetches. Bounded: a persistently busy session keeps converging through + * op broadcasts anyway, so after the last attempt the latest snapshot is + * applied rather than leaving a reconnecting client on stale state. + */ +const MAX_SYNC_FETCH_ATTEMPTS = 3 +const SYNC_RETRY_DELAY_MS = 50 + +/** + * Version vector describing everything that can move workflow state under a + * full-state snapshot while it is being fetched or held: + * - `localOperation`: local edits enqueued (operation queue) + * - `remoteApply`: remote collaborator ops applied to the stores + * - `remoteUpdate`: external full reloads (workflow-updated / revert) + */ +export interface DraftSyncVersions { + localOperation: number + remoteApply: number + remoteUpdate: number +} + +/** Captures the current {@link DraftSyncVersions} for a workflow. */ +export function captureDraftVersions(workflowId: string): DraftSyncVersions { + const queueState = useOperationQueueStore.getState() + return { + localOperation: queueState.workflowOperationVersions[workflowId] ?? 0, + remoteApply: queueState.remoteApplyVersions[workflowId] ?? 0, + remoteUpdate: useWorkflowDiffStore.getState().remoteUpdateVersions[workflowId] ?? 0, + } +} + +/** + * Hard guards shared by every snapshot application: the workflow is still + * active, nothing local is pending or was enqueued since capture, no diff or + * reconciliation is in progress, and no external full reload + * (workflow-updated / revert) landed since capture. Remote collaborator ops + * (`remoteApply`) are deliberately NOT part of this core — the sync loop + * treats them as a refetch signal rather than a refusal. + */ +function canApplyServerSnapshot( + workflowId: string, + expected: Pick +): boolean { + if (useWorkflowRegistry.getState().activeWorkflowId !== workflowId) return false + const operationQueueState = useOperationQueueStore.getState() + if (operationQueueState.hasPendingOperations(workflowId)) return false + if ( + (operationQueueState.workflowOperationVersions[workflowId] ?? 0) !== expected.localOperation + ) { + return false + } + + const diffState = useWorkflowDiffStore.getState() + return ( + !diffState.hasActiveDiff && + !diffState.pendingExternalUpdates[workflowId] && + !diffState.reconcilingWorkflows[workflowId] && + !diffState.reconciliationErrors[workflowId] && + (diffState.remoteUpdateVersions[workflowId] ?? 0) === expected.remoteUpdate + ) +} + +/** + * Whether a full-state snapshot captured alongside `versions` may still be + * applied. Unlike the sync loop — which refetches when remote collaborator + * ops land mid-fetch — this treats a `remoteApply` movement as a hard + * refusal, because callers holding a fixed snapshot (e.g. the raw join-state + * fallback) cannot refetch a fresher one. + */ +export function canApplyDraftSnapshot(workflowId: string, versions: DraftSyncVersions): boolean { + return ( + canApplyServerSnapshot(workflowId, versions) && + (useOperationQueueStore.getState().remoteApplyVersions[workflowId] ?? 0) === + versions.remoteApply + ) +} + +/** + * Reconciles the local Zustand stores with the server's current draft by + * fetching migrated state over HTTP and applying it — used after deploys and + * on socket join, where the realtime server can only supply raw (unmigrated) + * state. + * + * Fetches through the shared `workflowKeys.state(id)` React Query entry + * (always-fresh, in-flight deduped), so a sync racing the registry hydration + * coalesces into a single request and the cache stays warm. + * + * Refuses to apply (returns false) when the session is busy — pending or + * newly-queued local operations, an active copilot diff, in-progress + * reconciliation, a newer remote update during the fetch, or navigation away — + * so it never clobbers in-flight work. A remote collaborator's op applied + * during the fetch triggers a bounded refetch instead (the snapshot may + * predate that op's persist). Throws when the fetch itself fails. + */ +export async function syncLocalDraftFromServer(workflowId: string): Promise { + if (useWorkflowRegistry.getState().activeWorkflowId !== workflowId) return false + if (useOperationQueueStore.getState().hasPendingOperations(workflowId)) return false + const versionsAtStart = captureDraftVersions(workflowId) + + let envelope: Awaited> | undefined + for (let attempt = 1; ; attempt++) { + const remoteApplyVersionAtStart = + useOperationQueueStore.getState().remoteApplyVersions[workflowId] ?? 0 + + envelope = await getQueryClient().fetchQuery({ + queryKey: workflowKeys.state(workflowId), + queryFn: ({ signal }) => fetchWorkflowEnvelope(workflowId, signal), + staleTime: 0, + }) + + if (!canApplyServerSnapshot(workflowId, versionsAtStart)) { + return false + } + + const remoteOpAppliedDuringFetch = + (useOperationQueueStore.getState().remoteApplyVersions[workflowId] ?? 0) !== + remoteApplyVersionAtStart + if (!remoteOpAppliedDuringFetch) break + + if (attempt >= MAX_SYNC_FETCH_ATTEMPTS) { + logger.info('Applying latest draft snapshot despite concurrent remote ops', { + workflowId, + attempts: attempt, + }) + break + } + await sleep(SYNC_RETRY_DELAY_MS) + } + + const wireState = envelope?.state + if (!envelope || !wireState) { + throw new Error('No workflow state was returned while syncing the local draft') + } + + // Copy before annotating: the envelope is the shared React Query cache entry + // and must not be mutated. + const draftState = Object.hasOwn(envelope, 'variables') + ? { ...wireState, variables: envelope.variables || {} } + : { ...wireState } + // double-cast-allowed: workflowStateSchema is a wire supertype; normalized workflow state is persisted in store-compatible shape + const workflowState = draftState as unknown as WorkflowState + applyWorkflowStateToStores(workflowId, workflowState, { updateLastSaved: true }) + return true +} diff --git a/apps/sim/stores/workflows/utils.ts b/apps/sim/stores/workflows/utils.ts index a57f04b95af..356ba9a274d 100644 --- a/apps/sim/stores/workflows/utils.ts +++ b/apps/sim/stores/workflows/utils.ts @@ -10,6 +10,7 @@ import { buildDefaultCanonicalModes } from '@/lib/workflows/subblocks/visibility import { hasTriggerCapability } from '@/lib/workflows/triggers/trigger-utils' import { getBlock } from '@/blocks' import { normalizeName } from '@/executor/constants' +import { useWorkflowRegistry } from '@/stores/workflows/registry/store' import { useSubBlockStore } from '@/stores/workflows/subblock/store' import { validateEdges } from '@/stores/workflows/workflow/edge-validation' import type { @@ -189,9 +190,13 @@ export function prepareBlockState(options: PrepareBlockStateOptions): BlockState } /** - * Merges workflow block states with subblock values while maintaining block structure + * Merges workflow block states with the sub-block store's values while maintaining + * block structure. Resolves the active workflow when no workflowId is given. + * Value semantics (explicit-null clears, orphaned runtime values such as + * webhookId/triggerPath, undefined fallbacks) are defined by + * {@link mergeSubblockStateWithValues}. * @param blocks - Block configurations from workflow store - * @param workflowId - ID of the workflow to merge values for + * @param workflowId - ID of the workflow to merge values for (defaults to the active workflow) * @param blockId - Optional specific block ID to merge (merges all if not provided) * @returns Merged block states with updated values */ @@ -200,78 +205,12 @@ export function mergeSubblockState( workflowId?: string, blockId?: string ): Record { - const subBlockStore = useSubBlockStore.getState() + const resolvedWorkflowId = workflowId ?? useWorkflowRegistry.getState().activeWorkflowId + const workflowSubblockValues = resolvedWorkflowId + ? useSubBlockStore.getState().workflowValues[resolvedWorkflowId] || {} + : {} - const workflowSubblockValues = workflowId ? subBlockStore.workflowValues[workflowId] || {} : {} - - if (workflowId) { - return mergeSubblockStateWithValues(blocks, workflowSubblockValues, blockId) - } - - const blocksToProcess = blockId ? { [blockId]: blocks[blockId] } : blocks - - return Object.entries(blocksToProcess).reduce( - (acc, [id, block]) => { - if (!block) { - return acc - } - - const blockSubBlocks = block.subBlocks || {} - - const blockValues = workflowSubblockValues[id] || {} - - const mergedSubBlocks = Object.entries(blockSubBlocks).reduce( - (subAcc, [subBlockId, subBlock]) => { - if (!subBlock) { - return subAcc - } - - let storedValue = null - - if (workflowId) { - if (blockValues[subBlockId] !== undefined) { - storedValue = blockValues[subBlockId] - } - } else { - storedValue = subBlockStore.getValue(id, subBlockId) - } - - subAcc[subBlockId] = { - ...subBlock, - value: (storedValue !== undefined && storedValue !== null - ? storedValue - : subBlock.value) as SubBlockState['value'], - } - - return subAcc - }, - {} as Record - ) - - // Add any values that exist in the store but aren't in the block structure - // This handles cases where block config has been updated but values still exist - // IMPORTANT: This includes runtime subblock IDs like webhookId, triggerPath, etc. - Object.entries(blockValues).forEach(([subBlockId, value]) => { - if (!mergedSubBlocks[subBlockId] && value !== null && value !== undefined) { - // Create a minimal subblock structure - mergedSubBlocks[subBlockId] = { - id: subBlockId, - type: 'short-input', // Default type that's safe to use - value: value as SubBlockState['value'], - } - } - }) - - // Return the full block state with updated subBlocks (including orphaned values) - acc[id] = { - ...block, - subBlocks: mergedSubBlocks, - } - - return acc - }, - {} as Record - ) + return mergeSubblockStateWithValues(blocks, workflowSubblockValues, blockId) } function updateValueReferences(value: unknown, nameMap: Map): unknown { @@ -410,6 +349,11 @@ export function regenerateWorkflowIds( /** * Remaps condition/router block IDs within subBlock values when a block is duplicated. * Mutates both `subBlocks` and `subBlockValues` in place (callers must pass cloned data). + * + * The `subBlockValues[id] ?? subBlock.value` fallback is safe here despite the + * structure copy being generally stale: condition/router subblocks are + * dynamic-handle types, which dual-write the structure on every edit + * (syncDynamicHandleSubblockValue), so both sources are current for them. */ export function remapConditionIds( subBlocks: Record, diff --git a/bun.lock b/bun.lock index 18b739ea299..39e51d99b97 100644 --- a/bun.lock +++ b/bun.lock @@ -543,6 +543,7 @@ "@sim/tsconfig": "workspace:*", "@types/node": "24.2.1", "typescript": "^7.0.2", + "vitest": "^4.1.0", }, }, "packages/workflow-renderer": { diff --git a/packages/testing/src/mocks/database.mock.ts b/packages/testing/src/mocks/database.mock.ts index 9a879e4cad8..3c832d79652 100644 --- a/packages/testing/src/mocks/database.mock.ts +++ b/packages/testing/src/mocks/database.mock.ts @@ -349,5 +349,7 @@ export const databaseMock = { */ export const drizzleOrmMock = { sql: createMockSql(), + /** Mirrors drizzle's getTableColumns for schema-mock tables (column-name maps). */ + getTableColumns: vi.fn((table: Record) => ({ ...table })), ...createMockSqlOperators(), } diff --git a/packages/workflow-persistence/package.json b/packages/workflow-persistence/package.json index 3b3275b60c1..bca4c2ff05d 100644 --- a/packages/workflow-persistence/package.json +++ b/packages/workflow-persistence/package.json @@ -40,7 +40,9 @@ "lint": "biome check --write --unsafe .", "lint:check": "biome check .", "format": "biome format --write .", - "format:check": "biome format ." + "format:check": "biome format .", + "test": "vitest run", + "test:watch": "vitest" }, "dependencies": { "@sim/db": "workspace:*", @@ -53,6 +55,7 @@ "devDependencies": { "@sim/tsconfig": "workspace:*", "@types/node": "24.2.1", - "typescript": "^7.0.2" + "typescript": "^7.0.2", + "vitest": "^4.1.0" } } diff --git a/packages/workflow-persistence/src/load.ts b/packages/workflow-persistence/src/load.ts index 288e9217e8a..6d16e41427a 100644 --- a/packages/workflow-persistence/src/load.ts +++ b/packages/workflow-persistence/src/load.ts @@ -2,7 +2,7 @@ import { db, workflow, workflowBlocks, workflowEdges, workflowSubflows } from '@ import { createLogger } from '@sim/logger' import type { BlockState, Loop, Parallel } from '@sim/workflow-types/workflow' import { SUBFLOW_TYPES } from '@sim/workflow-types/workflow' -import { and, eq, isNull } from 'drizzle-orm' +import { and, eq, getTableColumns, isNull, sql } from 'drizzle-orm' import type { Edge } from 'reactflow' import { clampParallelBatchSize } from './subflow-helpers' import type { DbOrTx, NormalizedWorkflowData } from './types' @@ -11,7 +11,14 @@ const logger = createLogger('WorkflowPersistenceLoad') export interface RawNormalizedWorkflow extends NormalizedWorkflowData { workspaceId: string - blockUpdatedAtById: Record + /** + * Each block row's `updated_at` rendered by Postgres as text, preserving the + * full microsecond precision. Kept as a string on purpose: Drizzle's `date` + * mode surfaces timestamps as JS `Date`s, which truncate to milliseconds, so + * a Date-based value can never be used for an exact compare-and-set against + * rows stamped by SQL `now()`/`defaultNow()`. + */ + blockUpdatedAtById: Record } /** @@ -35,7 +42,13 @@ export async function loadWorkflowFromNormalizedTablesRaw( try { const tx = externalTx ?? db const [blocks, edges, subflows, [workflowRow]] = await Promise.all([ - tx.select().from(workflowBlocks).where(eq(workflowBlocks.workflowId, workflowId)), + tx + .select({ + ...getTableColumns(workflowBlocks), + updatedAtText: sql`${workflowBlocks.updatedAt}::text`, + }) + .from(workflowBlocks) + .where(eq(workflowBlocks.workflowId, workflowId)), tx.select().from(workflowEdges).where(eq(workflowEdges.workflowId, workflowId)), tx.select().from(workflowSubflows).where(eq(workflowSubflows.workflowId, workflowId)), tx @@ -54,7 +67,7 @@ export async function loadWorkflowFromNormalizedTablesRaw( } const blocksMap: Record = {} - const blockUpdatedAtById: Record = {} + const blockUpdatedAtById: Record = {} blocks.forEach((block) => { const blockData = (block.data ?? {}) as BlockState['data'] @@ -78,7 +91,7 @@ export async function loadWorkflowFromNormalizedTablesRaw( } blocksMap[block.id] = assembled - blockUpdatedAtById[block.id] = block.updatedAt ?? null + blockUpdatedAtById[block.id] = block.updatedAtText ?? null }) const edgesArray: Edge[] = edges.map((edge) => ({ @@ -180,11 +193,31 @@ export async function loadWorkflowFromNormalizedTablesRaw( } } +/** + * Optimistic-concurrency guard: matches a row's `updated_at` exactly against + * the microsecond-precision text value captured at read time. + * + * The comparison deliberately round-trips through text rather than a JS + * `Date`. Postgres stores `timestamp` columns with microsecond precision, but + * Drizzle's `date` mode truncates to milliseconds on read, so a Date-based + * equality guard can never match rows stamped by SQL `now()`/`defaultNow()` — + * the UPDATE silently no-ops forever and load-time migrations never persist. + * Casting the captured text back to `timestamp` restores an exact + * compare-and-set: a concurrent write stamps a new `updated_at` and the guard + * fails closed (skipped, logged, retried on the next load). Residual, inherent + * limit: a concurrent app write whose JS `new Date()` collides with the stored + * value at exact precision is indistinguishable; a version column would be + * required to close that fully. + */ +function updatedAtMatches(expectedText: string) { + return eq(workflowBlocks.updatedAt, sql`${expectedText}::timestamp`) +} + export async function persistMigratedBlocks( workflowId: string, originalBlocks: Record, migratedBlocks: Record, - blockUpdatedAtById: Record = {} + blockUpdatedAtById: Record = {} ): Promise { try { for (const [blockId, block] of Object.entries(migratedBlocks)) { @@ -197,11 +230,11 @@ export async function persistMigratedBlocks( eq(workflowBlocks.workflowId, workflowId), expectedUpdatedAt === null ? isNull(workflowBlocks.updatedAt) - : eq(workflowBlocks.updatedAt, expectedUpdatedAt) + : updatedAtMatches(expectedUpdatedAt) ) : and(eq(workflowBlocks.id, blockId), eq(workflowBlocks.workflowId, workflowId)) - await db + const persisted = await db .update(workflowBlocks) .set({ subBlocks: block.subBlocks, @@ -209,6 +242,15 @@ export async function persistMigratedBlocks( updatedAt: new Date(), }) .where(whereClause) + .returning({ id: workflowBlocks.id }) + + if (persisted.length === 0) { + logger.warn('Skipped persisting block migration (row changed since read or missing)', { + workflowId, + blockId, + expectedUpdatedAt, + }) + } } } } catch (err) { diff --git a/packages/workflow-persistence/src/subblocks.test.ts b/packages/workflow-persistence/src/subblocks.test.ts new file mode 100644 index 00000000000..4abda21d97d --- /dev/null +++ b/packages/workflow-persistence/src/subblocks.test.ts @@ -0,0 +1,121 @@ +import type { BlockState } from '@sim/workflow-types/workflow' +import { describe, expect, it } from 'vitest' +import { DEFAULT_SUBBLOCK_TYPE, mergeSubblockStateWithValues } from './subblocks' + +function buildBlock(subBlocks: BlockState['subBlocks']): BlockState { + return { + id: 'block-1', + type: 'api', + name: 'API', + position: { x: 0, y: 0 }, + subBlocks, + outputs: {}, + enabled: true, + } as BlockState +} + +function buildBlocks(subBlocks: BlockState['subBlocks']): Record { + return { 'block-1': buildBlock(subBlocks) } +} + +describe('mergeSubblockStateWithValues', () => { + it('overrides structure values with non-null store values', () => { + const blocks = buildBlocks({ + channel: { id: 'channel', type: 'short-input', value: 'old-channel' }, + }) + + const merged = mergeSubblockStateWithValues(blocks, { + 'block-1': { channel: 'new-channel' }, + }) + + expect(merged['block-1'].subBlocks.channel.value).toBe('new-channel') + }) + + it('overrides structure values with explicit null (cleared field)', () => { + const blocks = buildBlocks({ + channel: { id: 'channel', type: 'short-input', value: 'old-channel' }, + }) + + const merged = mergeSubblockStateWithValues(blocks, { + 'block-1': { channel: null }, + }) + + expect(merged['block-1'].subBlocks.channel.value).toBeNull() + }) + + it('keeps structure values when the key is absent from the values map', () => { + const blocks = buildBlocks({ + channel: { id: 'channel', type: 'short-input', value: 'old-channel' }, + }) + + const merged = mergeSubblockStateWithValues(blocks, { 'block-1': {} }) + + expect(merged['block-1'].subBlocks.channel.value).toBe('old-channel') + }) + + it('treats undefined values as absent', () => { + const blocks = buildBlocks({ + channel: { id: 'channel', type: 'short-input', value: 'old-channel' }, + }) + + const merged = mergeSubblockStateWithValues(blocks, { + 'block-1': { channel: undefined }, + }) + + expect(merged['block-1'].subBlocks.channel.value).toBe('old-channel') + }) + + it('does not create entries for null values missing from the structure', () => { + const blocks = buildBlocks({ + channel: { id: 'channel', type: 'short-input', value: 'old-channel' }, + }) + + const merged = mergeSubblockStateWithValues(blocks, { + 'block-1': { webhookId: null }, + }) + + expect(merged['block-1'].subBlocks.webhookId).toBeUndefined() + }) + + it('creates minimal entries for non-null values missing from the structure', () => { + const blocks = buildBlocks({ + channel: { id: 'channel', type: 'short-input', value: 'old-channel' }, + }) + + const merged = mergeSubblockStateWithValues(blocks, { + 'block-1': { webhookId: 'wh-123' }, + }) + + expect(merged['block-1'].subBlocks.webhookId).toEqual({ + id: 'webhookId', + type: DEFAULT_SUBBLOCK_TYPE, + value: 'wh-123', + }) + }) + + it('merges only the requested block when blockId is provided', () => { + const blocks: Record = { + 'block-1': buildBlock({ + channel: { id: 'channel', type: 'short-input', value: 'old-channel' }, + }), + 'block-2': buildBlock({ + channel: { id: 'channel', type: 'short-input', value: 'other' }, + }), + } + + const merged = mergeSubblockStateWithValues( + blocks, + { 'block-1': { channel: null }, 'block-2': { channel: null } }, + 'block-1' + ) + + expect(Object.keys(merged)).toEqual(['block-1']) + expect(merged['block-1'].subBlocks.channel.value).toBeNull() + }) + + it('skips unknown block ids without throwing', () => { + const merged = mergeSubblockStateWithValues(buildBlocks({}), {}, 'missing-block') + + expect(merged).toEqual({}) + }) +}) diff --git a/packages/workflow-persistence/src/subblocks.ts b/packages/workflow-persistence/src/subblocks.ts index 28501bf8f02..0109b4ceeee 100644 --- a/packages/workflow-persistence/src/subblocks.ts +++ b/packages/workflow-persistence/src/subblocks.ts @@ -39,7 +39,19 @@ export function mergeSubBlockValues( /** * Merges workflow block states with explicit subblock values while maintaining block structure. - * Values that are null or undefined do not override existing subblock values. + * + * A value that is present in the map overrides the structure's value — including `null`, + * which represents an explicitly cleared field. The block structure's own copy of a value + * can be stale (it is only rewritten on hydration, while edits land in the values map), so + * skipping nulls here would resurrect the pre-clear value and make the merged state diverge + * from what is actually persisted. + * + * Two softening rules keep sparse maps safe: + * - `undefined` is treated as "no value recorded" and never overrides the structure. + * - `null` never creates an entry for a subblock missing from the structure; only non-null + * structure-less values (e.g. runtime ids like `webhookId`/`triggerPath`) are added, with + * a minimal default shape, so they survive serialization. + * * @param blocks - Block configurations from workflow state * @param subBlockValues - Subblock values keyed by blockId -> subBlockId -> value * @param blockId - Optional specific block ID to merge (merges all if not provided) @@ -59,12 +71,14 @@ export function mergeSubblockStateWithValues( } const blockSubBlocks = block.subBlocks || {} - const blockValues = subBlockValues[id] || {} - const filteredValues = Object.fromEntries( - Object.entries(filterUndefined(blockValues)).filter(([, value]) => value !== null) + const definedValues = filterUndefined(subBlockValues[id] || {}) + const mergeableValues = Object.fromEntries( + Object.entries(definedValues).filter( + ([subBlockId, value]) => value !== null || Object.hasOwn(blockSubBlocks, subBlockId) + ) ) - const mergedSubBlocks = mergeSubBlockValues(blockSubBlocks, filteredValues) as Record< + const mergedSubBlocks = mergeSubBlockValues(blockSubBlocks, mergeableValues) as Record< string, SubBlockState > diff --git a/packages/workflow-persistence/vitest.config.ts b/packages/workflow-persistence/vitest.config.ts new file mode 100644 index 00000000000..471771e48fe --- /dev/null +++ b/packages/workflow-persistence/vitest.config.ts @@ -0,0 +1,9 @@ +import { defineConfig } from 'vitest/config' + +export default defineConfig({ + test: { + globals: false, + environment: 'node', + include: ['src/**/*.test.ts'], + }, +})