fix(workflows): handle realtime invalidation edge cases

Author: Bill Leoutsakos

apps/sim/ee/workspace-forking/lib/promote/rollback.test.ts

@@ -78,23 +78,30 @@ import { rollbackFork } from '@/ee/workspace-forking/lib/promote/rollback'
 
 const EDGE = { childWorkspaceId: 'child-ws', parentWorkspaceId: 'parent-ws' }
 
-/** A fake transaction whose existence query returns the given undeploy ids. */
-function makeTx(existingUndeployIds: string[] = []) {
+/** A fake transaction whose existence/update queries return the supplied workflow ids. */
+function makeTx(existingUndeployIds: string[] = [], unarchivedWorkflowIds: string[] = []) {
   return {
     select: vi.fn(() => ({
       from: vi.fn(() => ({
         where: vi.fn(() => Promise.resolve(existingUndeployIds.map((id) => ({ id })))),
       })),
     })),
     update: vi.fn(() => ({
-      set: vi.fn(() => ({ where: vi.fn(() => Promise.resolve(undefined)) })),
+      set: vi.fn(() => ({
+        where: vi.fn(() =>
+          Object.assign(Promise.resolve(undefined), {
+            returning: vi.fn().mockResolvedValue(unarchivedWorkflowIds.map((id) => ({ id }))),
+          })
+        ),
+      })),
     })),
   }
 }
 
-function setTx(existingUndeployIds: string[] = []) {
+function setTx(existingUndeployIds: string[] = [], unarchivedWorkflowIds: string[] = []) {
   vi.mocked(db.transaction).mockImplementation(
-    async (cb: (tx: unknown) => unknown) => cb(makeTx(existingUndeployIds)) as never
+    async (cb: (tx: unknown) => unknown) =>
+      cb(makeTx(existingUndeployIds, unarchivedWorkflowIds)) as never
   )
 }
 
@@ -165,6 +172,7 @@ describe('rollbackFork', () => {
   })
 
   it('un-archives and reactivates an archived orphan (prior version restored)', async () => {
+    setTx([], ['wf-x'])
     const run = makeRun({
       snapshot: { updated: [], created: [], archived: [{ workflowId: 'wf-x', priorVersion: 2 }] },
     })
@@ -187,6 +195,30 @@ describe('rollbackFork', () => {
     expect(mockNotify).toHaveBeenCalledWith('wf-x')
   })
 
+  it('refreshes workspace lists when an archived orphan had no prior deployment', async () => {
+    setTx([], ['wf-x'])
+    mockGetLatestRun.mockResolvedValue(
+      makeRun({
+        snapshot: {
+          updated: [],
+          created: [],
+          archived: [{ workflowId: 'wf-x', priorVersion: null }],
+        },
+      })
+    )
+
+    const result = await rollbackFork({
+      targetWorkspaceId: 'target-ws',
+      otherWorkspaceId: 'other-ws',
+      userId: 'user-1',
+    })
+
+    expect(result.unarchived).toBe(1)
+    expect(mockReactivate).not.toHaveBeenCalled()
+    expect(mockNotifyWorkspace).toHaveBeenCalledOnce()
+    expect(mockNotifyWorkspace).toHaveBeenCalledWith('target-ws')
+  })
+
   it('aborts with 409 and writes nothing when a newer sync supersedes it mid-flight', async () => {
     const run = makeRun({
       snapshot: { updated: [{ workflowId: 'wf-a', priorVersion: 3 }], created: [], archived: [] },

apps/sim/ee/workspace-forking/lib/promote/rollback.ts

@@ -131,6 +131,7 @@ export async function rollbackFork(params: RollbackForkParams): Promise<Rollback
   const skipped = new Set<string>()
   const outboxEventIds: string[] = []
   const reactivations: Array<{ workflowId: string; operationId: string }> = []
+  const unarchivedWorkflowIds = new Set<string>()
 
   await db.transaction(async (tx) => {
     await setForkLockTimeout(tx)
@@ -151,7 +152,7 @@ export async function rollbackFork(params: RollbackForkParams): Promise<Rollback
 
     // Un-archive the orphans the promote archived BEFORE reactivating them.
     if (archived.length > 0) {
-      await tx
+      const unarchived = await tx
         .update(workflow)
         .set({ archivedAt: null, updatedAt: now })
         .where(
@@ -160,6 +161,8 @@ export async function rollbackFork(params: RollbackForkParams): Promise<Rollback
             archived.map((i) => i.workflowId)
           )
         )
+        .returning({ id: workflow.id })
+      for (const row of unarchived) unarchivedWorkflowIds.add(row.id)
     }
 
     // Which undeploy targets still exist (created targets can be hard-deleted after the
@@ -248,6 +251,12 @@ export async function rollbackFork(params: RollbackForkParams): Promise<Rollback
     }
   })
 
+  // The rollback is already durable. Refresh workspace lists before any downstream deployment
+  // side effect can fail, including archived workflows that had no prior deployment to reactivate.
+  if (unarchivedWorkflowIds.size > 0 || ops.some((op) => !skipped.has(op.workflowId))) {
+    await notifyWorkspaceWorkflowsChanged(targetWorkspaceId)
+  }
+
   // After commit: process the enqueued side-effects (webhooks / schedules / MCP). These
   // are durable outbox rows, so a crash here is recovered by the outbox cron/reaper -
   // failures only warn, they never undo the (committed) restore.
@@ -342,7 +351,6 @@ export async function rollbackFork(params: RollbackForkParams): Promise<Rollback
     if (!skipped.has(op.workflowId)) notifyIds.add(op.workflowId)
   }
   for (const workflowId of notifyIds) void notifyForkWorkflowChanged(workflowId)
-  if (notifyIds.size > 0) await notifyWorkspaceWorkflowsChanged(targetWorkspaceId)
 
   // Attribute each skip to its bucket (a workflow is in exactly one) so the counts
   // reflect what was actually restored, not the snapshot size.

apps/sim/lib/workflows/application/move-workflows-bulk.test.ts

@@ -126,6 +126,28 @@ describe('moveWorkflowsBulk', () => {
     expect(mocks.notifyWorkspace).toHaveBeenCalledWith('workspace-1')
   })
 
+  it('does not refresh workspace lists when every workflow is already in the destination', async () => {
+    queueTableRows(schemaMock.workflow, [{ id: 'workflow-1', name: 'One', folderId: 'folder-1' }])
+    dbChainMockFns.for.mockResolvedValueOnce([
+      { id: 'workflow-1', name: 'One', folderId: 'folder-1' },
+    ])
+    mocks.updateWorkflow.mockResolvedValue({
+      success: true,
+      workflow: { id: 'workflow-1', name: 'One', folderId: 'folder-1' },
+    })
+
+    await moveWorkflowsBulk.execute({
+      principal,
+      input: {
+        workspaceId: 'workspace-1',
+        workflowIds: ['workflow-1'],
+        folderId: 'folder-1',
+      },
+    })
+
+    expect(mocks.notifyWorkspace).not.toHaveBeenCalled()
+  })
+
   it('conceals cross-workspace workflow IDs as failed items', async () => {
     queueTableRows(schemaMock.workflow, [])
 

apps/sim/lib/workflows/application/move-workflows-bulk.ts

@@ -175,7 +175,7 @@ export const moveWorkflowsBulk = defineAuthorizedWorkflowUseCase({
     for (const workflowId of result.moved) {
       await notifyWorkflowUpdated(workflowId)
     }
-    if (result.moved.length > 0) {
+    if (result.changes.some((change) => change.previousFolderId !== result.folderId)) {
       await notifyWorkspaceWorkflowsChanged(context.workspaceId)
     }
   },

apps/sim/lib/workflows/deployment-outbox.test.ts

@@ -391,6 +391,46 @@ describe('versioned deployment preparation outbox', () => {
     expect(mockCleanupRetiredWebhookRegistrations).toHaveBeenCalledTimes(1)
   })
 
+  it('notifies workspace lists even when the fail-fast workflow socket notification fails', async () => {
+    mockIsDeploymentOperationCurrent.mockResolvedValue(true)
+    mockGetDeploymentOperation.mockResolvedValue(operation({ status: 'active', completedAt: NOW }))
+    queueTableRows(schemaMock.workflow, [
+      { id: 'workflow-1', name: 'Workflow', workspaceId: 'workspace-1' },
+    ])
+    const fetchMock = vi.fn(async (input: string | URL | Request) => {
+      if (String(input).endsWith('/api/workflow-deployed')) {
+        throw new Error('workflow socket unavailable')
+      }
+      return new Response(null, { status: 200 })
+    })
+    vi.stubGlobal('fetch', fetchMock)
+    const outboxContext = context()
+
+    await expect(
+      handler()(
+        {
+          ...payload(),
+          checkpoints: {
+            inactiveCleanupCompleted: true,
+            auditEmitted: true,
+            analyticsCaptured: true,
+          },
+        },
+        outboxContext
+      )
+    ).rejects.toThrow('workflow socket unavailable')
+
+    expect(fetchMock.mock.calls.map(([input]) => String(input))).toEqual([
+      expect.stringMatching(/\/api\/workspace-workflows-changed$/),
+      expect.stringMatching(/\/api\/workflow-deployed$/),
+    ])
+    expect(outboxContext.checkpointPayload).toHaveBeenCalledWith(
+      expect.objectContaining({
+        checkpoints: expect.objectContaining({ workspaceListNotified: true }),
+      })
+    )
+  })
+
   it('honors an aborted signal before starting any side effect', async () => {
     const controller = new AbortController()
     controller.abort()

apps/sim/lib/workflows/deployment-outbox.ts

@@ -730,6 +730,13 @@ async function emitPostActivationSideEffects(params: {
     await params.checkpoint({ analyticsCaptured: true })
   }
 
+  const workspaceId = params.workflow.workspaceId as string | null
+  if (workspaceId && !params.checkpoints.workspaceListNotified) {
+    params.context.signal.throwIfAborted()
+    await notifyWorkspaceWorkflowsChanged(workspaceId)
+    await params.checkpoint({ workspaceListNotified: true })
+  }
+
   if (!params.checkpoints.socketNotified) {
     params.context.signal.throwIfAborted()
     await notifySocketDeploymentChanged(params.payload.workflowId, {
@@ -740,13 +747,6 @@ async function emitPostActivationSideEffects(params: {
     await params.checkpoint({ socketNotified: true })
   }
 
-  const workspaceId = params.workflow.workspaceId as string | null
-  if (workspaceId && !params.checkpoints.workspaceListNotified) {
-    params.context.signal.throwIfAborted()
-    await notifyWorkspaceWorkflowsChanged(workspaceId)
-    await params.checkpoint({ workspaceListNotified: true })
-  }
-
   if (workspaceId && !params.checkpoints.workspaceEventEmitted) {
     params.context.signal.throwIfAborted()
     await emitWorkflowDeployedEvent({