diff --git a/apps/realtime/src/access-revalidation.ts b/apps/realtime/src/access-revalidation.ts index 8e0ca514812..1563c0d5f32 100644 --- a/apps/realtime/src/access-revalidation.ts +++ b/apps/realtime/src/access-revalidation.ts @@ -49,8 +49,8 @@ function fallbackRoleFor(type: RoomType): string { /** * Room types whose membership is mirrored in the room manager's (Redis) presence * state, and therefore need a presence removal + rebroadcast after an eviction. - * The workspace-files / workspace-tables invalidation rooms carry no presence at - * all, and a file-doc room's roster is pod-local in-memory state reconciled by its + * The workspace-files / workspace-tables / workspace-workflows invalidation rooms carry no + * presence at all, and a file-doc room's roster is pod-local in-memory state reconciled by its * registered eviction handler — neither has anything for the cleanup lane to do. */ const PRESENCE_ROOM_TYPES: ReadonlySet = new Set([ diff --git a/apps/realtime/src/handlers/index.ts b/apps/realtime/src/handlers/index.ts index 8dd71093673..3f81b0f8f35 100644 --- a/apps/realtime/src/handlers/index.ts +++ b/apps/realtime/src/handlers/index.ts @@ -20,6 +20,7 @@ export function setupAllHandlers(socket: AuthenticatedSocket, roomManager: IRoom // Presence-free, workspace-scoped live-list rooms (share one implementation). setupWorkspaceInvalidationRoom(socket, roomManager, ROOM_TYPES.WORKSPACE_FILES) setupWorkspaceInvalidationRoom(socket, roomManager, ROOM_TYPES.WORKSPACE_TABLES) + setupWorkspaceInvalidationRoom(socket, roomManager, ROOM_TYPES.WORKSPACE_WORKFLOWS) setupWorkspaceFileDocHandlers(socket, roomManager) setupTablesHandlers(socket, roomManager) setupConnectionHandlers(socket, roomManager) diff --git a/apps/realtime/src/handlers/workspace-invalidation-room.test.ts b/apps/realtime/src/handlers/workspace-invalidation-room.test.ts index e487edd0a8e..789ccbc0f88 100644 --- a/apps/realtime/src/handlers/workspace-invalidation-room.test.ts +++ b/apps/realtime/src/handlers/workspace-invalidation-room.test.ts @@ -73,9 +73,15 @@ function createRoomManager(overrides?: Partial): IRoomManager { } as unknown as IRoomManager } -// The two presence-free live-list rooms share one implementation; run the whole suite against both -// so files and tables can never drift. Event names and room names derive from the room type. -describe.each([ROOM_TYPES.WORKSPACE_FILES, ROOM_TYPES.WORKSPACE_TABLES] as const)( +// The presence-free live-list rooms share one implementation; run the whole suite against all of +// them so their authorization and lifecycle behavior cannot drift. +const workspaceInvalidationRoomTypesCoveredBySharedLifecycle = [ + ROOM_TYPES.WORKSPACE_FILES, + ROOM_TYPES.WORKSPACE_TABLES, + ROOM_TYPES.WORKSPACE_WORKFLOWS, +] as const + +describe.each(workspaceInvalidationRoomTypesCoveredBySharedLifecycle)( 'setupWorkspaceInvalidationRoom(%s)', (roomType) => { const joinEvent = `join-${roomType}` diff --git a/apps/realtime/src/routes/http.test.ts b/apps/realtime/src/routes/http.test.ts index 725341deac9..ba334a1e114 100644 --- a/apps/realtime/src/routes/http.test.ts +++ b/apps/realtime/src/routes/http.test.ts @@ -1,5 +1,6 @@ import type { IncomingMessage, ServerResponse } from 'http' import { describe, expect, it, vi } from 'vitest' +import { env } from '@/env' import type { IRoomManager } from '@/rooms' import { createHttpHandler } from '@/routes/http' @@ -11,6 +12,7 @@ function createMocks(req: Partial) { const roomManager = { getTotalActiveConnections: vi.fn().mockResolvedValue(0), isReady: vi.fn().mockReturnValue(true), + emitToRoom: vi.fn(), } as unknown as IRoomManager return { @@ -20,6 +22,7 @@ function createMocks(req: Partial) { setHeader, writeHead, end, + roomManager, } } @@ -58,4 +61,29 @@ describe('createHttpHandler', () => { expect(writeHead).toHaveBeenCalledWith(200, { 'Content-Type': 'application/json' }) }) + + it('fans workflow-tree changes out to the workspace workflows room', async () => { + const workspaceId = 'workspace-1' + const body = JSON.stringify({ workspaceId }) + const request = { + method: 'POST', + url: '/api/workspace-workflows-changed', + headers: { 'x-api-key': env.INTERNAL_API_SECRET }, + on(event: string, callback: (chunk?: Buffer) => void) { + if (event === 'data') callback(Buffer.from(body)) + if (event === 'end') callback() + return this + }, + } as unknown as IncomingMessage + const { handler, res, roomManager, writeHead } = createMocks(request) + + await handler(request, res) + + expect(roomManager.emitToRoom).toHaveBeenCalledWith( + { type: 'workspace-workflows', id: workspaceId }, + 'workspace-workflows-changed', + { workspaceId, timestamp: expect.any(Number) } + ) + expect(writeHead).toHaveBeenCalledWith(200, { 'Content-Type': 'application/json' }) + }) }) diff --git a/apps/realtime/src/routes/http.ts b/apps/realtime/src/routes/http.ts index aed7d1a58a9..7ad8f11e05f 100644 --- a/apps/realtime/src/routes/http.ts +++ b/apps/realtime/src/routes/http.ts @@ -205,6 +205,26 @@ export function createHttpHandler(roomManager: IRoomManager, logger: Logger) { return } + // Fan out a workflow-tree change to everyone viewing a workspace's persistent sidebar. + // One signal invalidates both workflow and workflow-folder lists. + if (req.method === 'POST' && req.url === '/api/workspace-workflows-changed') { + try { + const body = await readRequestBody(req) + const { workspaceId } = JSON.parse(body) + if (!isNonEmptyString(workspaceId)) return sendError(res, 'Invalid workspaceId', 400) + roomManager.emitToRoom( + { type: ROOM_TYPES.WORKSPACE_WORKFLOWS, id: workspaceId }, + 'workspace-workflows-changed', + { workspaceId, timestamp: Date.now() } + ) + sendSuccess(res) + } catch (error) { + logger.error('Error handling workspace workflows changed notification:', error) + sendError(res, 'Failed to process workflows change notification') + } + return + } + // Merge a durable file write into a file's LIVE collaborative document so open editors reconcile to // it (Stage C) — this is the stream-end/durable reconcile, not token-by-token streaming (that is now // applied client-side by the open editor). Returns `{ applied }`: when false, no seeded live room diff --git a/apps/sim/app/api/folders/[id]/duplicate/route.test.ts b/apps/sim/app/api/folders/[id]/duplicate/route.test.ts index 210ca44f054..d0e484d2c8b 100644 --- a/apps/sim/app/api/folders/[id]/duplicate/route.test.ts +++ b/apps/sim/app/api/folders/[id]/duplicate/route.test.ts @@ -29,6 +29,7 @@ const { mockDuplicateWorkflow, mockAcquireFolderMutationLock, mockWithFolderTreeLock, + mockNotifyWorkspace, } = vi.hoisted(() => ({ mockLogger: { info: vi.fn(), @@ -44,6 +45,7 @@ const { mockDuplicateWorkflow: vi.fn(), mockAcquireFolderMutationLock: vi.fn(), mockWithFolderTreeLock: vi.fn(), + mockNotifyWorkspace: vi.fn(), })) vi.mock('@sim/audit', () => auditMock) @@ -62,6 +64,9 @@ vi.mock('@/lib/folders/naming', () => ({ deduplicateFolderName: mockDeduplicateF vi.mock('@/lib/workflows/persistence/duplicate', () => ({ duplicateWorkflow: mockDuplicateWorkflow, })) +vi.mock('@/lib/realtime/notify', () => ({ + notifyWorkspaceWorkflowsChanged: mockNotifyWorkspace, +})) import { POST } from '@/app/api/folders/[id]/duplicate/route' @@ -141,6 +146,8 @@ describe('POST /api/folders/[id]/duplicate', () => { expect(response.status).toBe(201) await expect(response.json()).resolves.toMatchObject({ folder: { name: 'Copy' } }) + expect(mockNotifyWorkspace).toHaveBeenCalledOnce() + expect(mockNotifyWorkspace).toHaveBeenCalledWith(WORKSPACE_ID) }) it('refuses a single-folder duplicate once the workspace is at the ceiling', async () => { diff --git a/apps/sim/app/api/folders/[id]/duplicate/route.ts b/apps/sim/app/api/folders/[id]/duplicate/route.ts index cd0e6fea92c..52ebc7ad786 100644 --- a/apps/sim/app/api/folders/[id]/duplicate/route.ts +++ b/apps/sim/app/api/folders/[id]/duplicate/route.ts @@ -19,6 +19,7 @@ import { nextFolderSortOrder } from '@/lib/folders/orchestration' import { assertFolderCollectionHasRoom, toFolderApi } from '@/lib/folders/queries' import { folderMutationStatus } from '@/lib/folders/status' import { collectDescendantFolderIds } from '@/lib/folders/subtree' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { duplicateWorkflow } from '@/lib/workflows/persistence/duplicate' import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils' @@ -208,6 +209,7 @@ export const POST = withRouteHandler( return { newFolderId, folderMapping, workflowStats } }) + await notifyWorkspaceWorkflowsChanged(targetWorkspaceId) const elapsed = Date.now() - startTime logger.info( diff --git a/apps/sim/app/api/folders/reorder/route.test.ts b/apps/sim/app/api/folders/reorder/route.test.ts index c9803b66fbb..f00e0c39930 100644 --- a/apps/sim/app/api/folders/reorder/route.test.ts +++ b/apps/sim/app/api/folders/reorder/route.test.ts @@ -6,7 +6,7 @@ import { authMockFns, createMockRequest, permissionsMock, permissionsMockFns } from '@sim/testing' import { beforeEach, describe, expect, it, vi } from 'vitest' -const { mockLogger } = vi.hoisted(() => ({ +const { mockLogger, mockNotifyFolder } = vi.hoisted(() => ({ mockLogger: { info: vi.fn(), warn: vi.fn(), @@ -16,11 +16,13 @@ const { mockLogger } = vi.hoisted(() => ({ fatal: vi.fn(), child: vi.fn(), }, + mockNotifyFolder: vi.fn(), })) const mockGetUserEntityPermissions = permissionsMockFns.mockGetUserEntityPermissions vi.mock('@/lib/workspaces/permissions/utils', () => permissionsMock) +vi.mock('@/lib/realtime/notify', () => ({ notifyFolderResourceChanged: mockNotifyFolder })) import { db } from '@sim/db' import { PUT } from '@/app/api/folders/reorder/route' @@ -73,6 +75,7 @@ describe('PUT /api/folders/reorder', () => { expect(response.status).toBe(200) const data = await response.json() expect(data).toMatchObject({ success: true, updated: 1 }) + expect(mockNotifyFolder).toHaveBeenCalledWith('workflow', 'workspace-123') }) it('maps a sibling-name collision from a reparent to a 409', async () => { diff --git a/apps/sim/app/api/folders/reorder/route.ts b/apps/sim/app/api/folders/reorder/route.ts index bd7f89d5e68..a3083251cd5 100644 --- a/apps/sim/app/api/folders/reorder/route.ts +++ b/apps/sim/app/api/folders/reorder/route.ts @@ -12,6 +12,7 @@ import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { withTransactionRetry } from '@/lib/db/transaction' import { acquireFolderMutationLock } from '@/lib/folders/locks' import { folderResourceSupportsLocking } from '@/lib/folders/resource-traits' +import { notifyFolderResourceChanged } from '@/lib/realtime/notify' import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils' const logger = createLogger('FolderReorderAPI') @@ -38,7 +39,7 @@ export const PUT = withRouteHandler(async (req: NextRequest) => { return NextResponse.json({ error: 'Write access required' }, { status: 403 }) } - return await withTransactionRetry( + const response = await withTransactionRetry( async (tx) => { await acquireFolderMutationLock(tx, workspaceId, resourceType) const folderIds = updates.map((u) => u.id) @@ -178,6 +179,8 @@ export const PUT = withRouteHandler(async (req: NextRequest) => { }, { label: 'reorder-folders' } ) + if (response.ok) await notifyFolderResourceChanged(resourceType, workspaceId) + return response } catch (error) { if (error instanceof FolderLockedError) { return NextResponse.json({ error: error.message }, { status: error.status }) diff --git a/apps/sim/app/api/superuser/import-workflow/route.ts b/apps/sim/app/api/superuser/import-workflow/route.ts index 5bc5b4bf5ee..c250056ca2a 100644 --- a/apps/sim/app/api/superuser/import-workflow/route.ts +++ b/apps/sim/app/api/superuser/import-workflow/route.ts @@ -11,6 +11,7 @@ import { loadCopilotChatMessages } from '@/lib/copilot/chat/lifecycle' import { appendCopilotChatMessages } from '@/lib/copilot/chat/messages-store' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { verifyEffectiveSuperUser } from '@/lib/permissions/super-user' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { parseWorkflowJson } from '@/lib/workflows/operations/import-export' import { loadWorkflowFromNormalizedTables, @@ -171,6 +172,8 @@ export const POST = withRouteHandler(async (request: NextRequest) => { ) } + await notifyWorkspaceWorkflowsChanged(targetWorkspaceId) + // Copy copilot chats associated with the source workflow const sourceCopilotChats = await db .select({ diff --git a/apps/sim/app/api/v1/admin/workflows/import/route.ts b/apps/sim/app/api/v1/admin/workflows/import/route.ts index 3f4290a8b0d..28b5441d740 100644 --- a/apps/sim/app/api/v1/admin/workflows/import/route.ts +++ b/apps/sim/app/api/v1/admin/workflows/import/route.ts @@ -29,6 +29,7 @@ import { NextResponse } from 'next/server' import { adminV1ImportWorkflowContract } from '@/lib/api/contracts/v1/admin' import { parseRequest } from '@/lib/api/server' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { parseWorkflowJson } from '@/lib/workflows/operations/import-export' import { prepareWorkflowStateForPersistence } from '@/lib/workflows/persistence/prepare-state' import { saveWorkflowToNormalizedTables } from '@/lib/workflows/persistence/utils' @@ -164,6 +165,8 @@ export const POST = withRouteHandler( return internalErrorResponse(`Failed to save workflow state: ${saveResult.error}`) } + await notifyWorkspaceWorkflowsChanged(workspaceId) + const variablesRecord = normalizeImportedVariables(workflowData.variables) if (Object.keys(variablesRecord).length > 0) { await db diff --git a/apps/sim/app/api/v1/admin/workspaces/[id]/import/route.test.ts b/apps/sim/app/api/v1/admin/workspaces/[id]/import/route.test.ts index 9b6b562fc6e..0d93e86922a 100644 --- a/apps/sim/app/api/v1/admin/workspaces/[id]/import/route.test.ts +++ b/apps/sim/app/api/v1/admin/workspaces/[id]/import/route.test.ts @@ -20,6 +20,7 @@ const { mockSaveWorkflowToNormalizedTables, mockDeduplicateWorkflowName, mockNormalizeImportedVariables, + mockNotifyWorkspace, } = vi.hoisted(() => ({ mockLogger: { info: vi.fn(), @@ -37,6 +38,7 @@ const { mockSaveWorkflowToNormalizedTables: vi.fn(), mockDeduplicateWorkflowName: vi.fn(), mockNormalizeImportedVariables: vi.fn(), + mockNotifyWorkspace: vi.fn(), })) vi.mock('@sim/logger', () => ({ @@ -65,6 +67,9 @@ vi.mock('@/lib/workflows/utils', () => ({ deduplicateWorkflowName: mockDeduplica vi.mock('@/lib/workflows/variables/parse', () => ({ normalizeImportedVariables: mockNormalizeImportedVariables, })) +vi.mock('@/lib/realtime/notify', () => ({ + notifyWorkspaceWorkflowsChanged: mockNotifyWorkspace, +})) import { POST } from '@/app/api/v1/admin/workspaces/[id]/import/route' @@ -113,6 +118,8 @@ describe('admin workspace import POST', () => { expect(response.status).toBe(200) await expect(response.json()).resolves.toMatchObject({ imported: 1, failed: 0 }) + expect(mockNotifyWorkspace).toHaveBeenCalledOnce() + expect(mockNotifyWorkspace).toHaveBeenCalledWith(WORKSPACE_ID) }) /** diff --git a/apps/sim/app/api/v1/admin/workspaces/[id]/import/route.ts b/apps/sim/app/api/v1/admin/workspaces/[id]/import/route.ts index 33f94c17a08..396af4dd704 100644 --- a/apps/sim/app/api/v1/admin/workspaces/[id]/import/route.ts +++ b/apps/sim/app/api/v1/admin/workspaces/[id]/import/route.ts @@ -40,6 +40,7 @@ import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import type { DbOrTx } from '@/lib/db/types' import { withFolderTreeLock } from '@/lib/folders/locks' import { assertFolderCollectionHasRoom } from '@/lib/folders/queries' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { extractWorkflowName, extractWorkflowsFromZip, @@ -118,7 +119,8 @@ async function ensureImportFolder( workspaceId: string, userId: string, name: string, - parentId: string | null + parentId: string | null, + onCreated?: () => void ): Promise { const existing = await findImportFolder(db, workspaceId, name, parentId) if (existing) return existing @@ -148,6 +150,7 @@ async function ensureImportFolder( createdAt: new Date(), updatedAt: new Date(), }) + onCreated?.() return folderId }) } catch (error) { @@ -173,6 +176,10 @@ export const POST = withRouteHandler( const { id: workspaceId } = parsed.data.params const { createFolders, rootFolderName } = parsed.data.query + let workspaceTreeChanged = false + const markWorkspaceTreeChanged = () => { + workspaceTreeChanged = true + } try { const workspaceData = await getWorkspaceWithOwner(workspaceId) @@ -245,7 +252,8 @@ export const POST = withRouteHandler( workspaceId, workspaceData.ownerId, rootFolderName, - null + null, + markWorkspaceTreeChanged ) } @@ -259,11 +267,13 @@ export const POST = withRouteHandler( workspaceData.ownerId, createFolders, rootFolderId, - folderMap + folderMap, + markWorkspaceTreeChanged ) results.push(result) if (result.success) { + workspaceTreeChanged = true logger.info(`Admin API: Imported workflow ${result.workflowId} (${result.name})`) } else { logger.warn(`Admin API: Failed to import workflow ${result.name}: ${result.error}`) @@ -276,8 +286,10 @@ export const POST = withRouteHandler( logger.info(`Admin API: Import complete - ${imported} succeeded, ${failed} failed`) const response: WorkspaceImportResponse = { imported, failed, results } + if (workspaceTreeChanged) await notifyWorkspaceWorkflowsChanged(workspaceId) return NextResponse.json(response) } catch (error) { + if (workspaceTreeChanged) await notifyWorkspaceWorkflowsChanged(workspaceId) /** * The workspace folder ceiling refuses the import as a classified `conflict`. It is a * whole-import failure rather than a per-workflow one: once the tree is full every @@ -302,7 +314,8 @@ async function importSingleWorkflow( ownerId: string, createFolders: boolean, rootFolderId: string | undefined, - folderMap: Map + folderMap: Map, + onFolderCreated: () => void ): Promise { try { const { data: workflowData, errors } = parseWorkflowJson(wf.content) @@ -331,7 +344,8 @@ async function importSingleWorkflow( workspaceId, ownerId, wf.folderPath[i], - parentId + parentId, + onFolderCreated ) folderMap.set(fullPath, folderId) parentId = folderId diff --git a/apps/sim/app/api/v1/workflows/import/route.test.ts b/apps/sim/app/api/v1/workflows/import/route.test.ts index 99e68fb508d..a28b1ca120d 100644 --- a/apps/sim/app/api/v1/workflows/import/route.test.ts +++ b/apps/sim/app/api/v1/workflows/import/route.test.ts @@ -24,6 +24,7 @@ const { mockDbDelete, mockDbUpdate, mockWorkspaceRows, + mockNotifyWorkspace, } = vi.hoisted(() => ({ mockCheckRateLimit: vi.fn(), mockValidateWorkspaceAccess: vi.fn(), @@ -37,6 +38,7 @@ const { mockDbDelete: vi.fn(), mockDbUpdate: vi.fn(), mockWorkspaceRows: { value: [{ id: 'ws-1' }] as Array<{ id: string }> }, + mockNotifyWorkspace: vi.fn(), })) vi.mock('@/app/api/v1/middleware', () => ({ @@ -59,6 +61,9 @@ vi.mock('@/lib/workflows/orchestration', () => ({ vi.mock('@/lib/workflows/persistence/utils', () => ({ saveWorkflowToNormalizedTables: mockSaveWorkflowToNormalizedTables, })) +vi.mock('@/lib/realtime/notify', () => ({ + notifyWorkspaceWorkflowsChanged: mockNotifyWorkspace, +})) vi.mock('@/lib/workflows/operations/import-export', () => ({ parseWorkflowJson: mockParseWorkflowJson, @@ -275,6 +280,8 @@ describe('POST /api/v1/workflows/import', () => { { workspaceId: WORKSPACE_ID, subjectUserId: null }, expect.anything() ) + expect(mockNotifyWorkspace).toHaveBeenCalledOnce() + expect(mockNotifyWorkspace).toHaveBeenCalledWith(WORKSPACE_ID) }) it('derives the name from the export envelope and deduplicates it', async () => { @@ -395,6 +402,7 @@ describe('POST /api/v1/workflows/import', () => { expect(response.status).toBe(500) expect(mockDbDelete).toHaveBeenCalled() expect(whereSpy).toHaveBeenCalled() + expect(mockNotifyWorkspace).not.toHaveBeenCalled() }) it('rolls back the created workflow when the variables write throws', async () => { diff --git a/apps/sim/app/api/workflows/[id]/duplicate/route.ts b/apps/sim/app/api/workflows/[id]/duplicate/route.ts index beba4a3f3ab..68e143b2cd5 100644 --- a/apps/sim/app/api/workflows/[id]/duplicate/route.ts +++ b/apps/sim/app/api/workflows/[id]/duplicate/route.ts @@ -9,6 +9,7 @@ import { PlatformEvents } from '@/lib/core/telemetry' import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { captureServerEvent } from '@/lib/posthog/server' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { duplicateWorkflow } from '@/lib/workflows/persistence/duplicate' const logger = createLogger('WorkflowDuplicateAPI') @@ -46,6 +47,7 @@ export const POST = withRouteHandler( requestId, newWorkflowId: newId, }) + await notifyWorkspaceWorkflowsChanged(result.workspaceId) try { PlatformEvents.workflowDuplicated({ diff --git a/apps/sim/app/api/workflows/reorder/route.ts b/apps/sim/app/api/workflows/reorder/route.ts index adb1b5416e5..45ab9f9addd 100644 --- a/apps/sim/app/api/workflows/reorder/route.ts +++ b/apps/sim/app/api/workflows/reorder/route.ts @@ -16,6 +16,7 @@ import { parseRequest } from '@/lib/api/server' import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid' import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { getUserEntityPermissions } from '@/lib/workspaces/permissions/utils' const logger = createLogger('WorkflowReorderAPI') @@ -83,6 +84,8 @@ export const PUT = withRouteHandler(async (req: NextRequest) => { `[${requestId}] Reordered ${validUpdates.length} workflows in workspace ${workspaceId}` ) + await notifyWorkspaceWorkflowsChanged(workspaceId) + return NextResponse.json({ success: true, updated: validUpdates.length }) } catch (error) { if ( diff --git a/apps/sim/app/api/workflows/route.test.ts b/apps/sim/app/api/workflows/route.test.ts index fa7c349267d..15e8839e314 100644 --- a/apps/sim/app/api/workflows/route.test.ts +++ b/apps/sim/app/api/workflows/route.test.ts @@ -18,8 +18,9 @@ import { } from '@sim/testing' import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest' -const { mockWorkflowCreated } = vi.hoisted(() => ({ +const { mockWorkflowCreated, mockNotifyWorkspace } = vi.hoisted(() => ({ mockWorkflowCreated: vi.fn(), + mockNotifyWorkspace: vi.fn(), })) const mockGetUserEntityPermissions = permissionsMockFns.mockGetUserEntityPermissions @@ -45,6 +46,9 @@ vi.mock('@/lib/workflows/defaults', () => ({ })) vi.mock('@/lib/workflows/persistence/utils', () => workflowsPersistenceUtilsMock) +vi.mock('@/lib/realtime/notify', () => ({ + notifyWorkspaceWorkflowsChanged: mockNotifyWorkspace, +})) import { POST } from '@/app/api/workflows/route' @@ -109,6 +113,7 @@ describe('Workflows API Route - POST ordering', () => { expect(response.status).toBe(200) expect(data.sortOrder).toBe(1) expect(dbChainMockFns.values).toHaveBeenCalledWith(expect.objectContaining({ sortOrder: 1 })) + expect(mockNotifyWorkspace).toHaveBeenCalledWith('workspace-123') }) it('defaults to sortOrder 0 when there are no siblings', async () => { diff --git a/apps/sim/app/api/workflows/route.ts b/apps/sim/app/api/workflows/route.ts index 05d94a31b1d..580064b5a85 100644 --- a/apps/sim/app/api/workflows/route.ts +++ b/apps/sim/app/api/workflows/route.ts @@ -7,6 +7,7 @@ import { checkSessionOrInternalAuth } from '@/lib/auth/hybrid' import { generateRequestId } from '@/lib/core/utils/request' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { captureServerEvent } from '@/lib/posthog/server' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { performCreateWorkflow } from '@/lib/workflows/orchestration' import { listWorkflowsForUser } from '@/lib/workflows/queries' import { getUserEntityPermissions, workspaceExists } from '@/lib/workspaces/permissions/utils' @@ -138,6 +139,7 @@ export const POST = withRouteHandler(async (req: NextRequest) => { } const createdWorkflow = result.workflow + await notifyWorkspaceWorkflowsChanged(workspaceId) import('@/lib/core/telemetry') .then(({ PlatformEvents }) => { diff --git a/apps/sim/app/api/workspaces/[id]/fork/excluded-workflows/route.test.ts b/apps/sim/app/api/workspaces/[id]/fork/excluded-workflows/route.test.ts index f31acbbc923..8793f0305dd 100644 --- a/apps/sim/app/api/workspaces/[id]/fork/excluded-workflows/route.test.ts +++ b/apps/sim/app/api/workspaces/[id]/fork/excluded-workflows/route.test.ts @@ -11,10 +11,13 @@ import { } from '@sim/testing' import { afterAll, beforeEach, describe, expect, it, vi } from 'vitest' -const { mockAssertWorkspaceAdminAccess, mockCaptureServerEvent } = vi.hoisted(() => ({ - mockAssertWorkspaceAdminAccess: vi.fn(), - mockCaptureServerEvent: vi.fn(), -})) +const { mockAssertWorkspaceAdminAccess, mockCaptureServerEvent, mockNotifyWorkspace } = vi.hoisted( + () => ({ + mockAssertWorkspaceAdminAccess: vi.fn(), + mockCaptureServerEvent: vi.fn(), + mockNotifyWorkspace: vi.fn(), + }) +) vi.mock('@/ee/workspace-forking/lib/lineage/authz', () => ({ assertWorkspaceAdminAccess: mockAssertWorkspaceAdminAccess, @@ -25,6 +28,9 @@ vi.mock('@sim/audit', () => auditMock) vi.mock('@/lib/posthog/server', () => ({ captureServerEvent: mockCaptureServerEvent, })) +vi.mock('@/lib/realtime/notify', () => ({ + notifyWorkspaceWorkflowsChanged: mockNotifyWorkspace, +})) import { PUT } from '@/app/api/workspaces/[id]/fork/excluded-workflows/route' @@ -117,6 +123,8 @@ describe('fork excluded-workflows route', () => { expect.objectContaining({ workflow_count: 2, fork_sync_excluded: true }), { groups: { workspace: WORKSPACE_ID } } ) + expect(mockNotifyWorkspace).toHaveBeenCalledOnce() + expect(mockNotifyWorkspace).toHaveBeenCalledWith(WORKSPACE_ID) }) it('records the inclusion action when unmarking workflows', async () => { diff --git a/apps/sim/app/api/workspaces/[id]/fork/excluded-workflows/route.ts b/apps/sim/app/api/workspaces/[id]/fork/excluded-workflows/route.ts index 842713be6d0..8d8050654d3 100644 --- a/apps/sim/app/api/workspaces/[id]/fork/excluded-workflows/route.ts +++ b/apps/sim/app/api/workspaces/[id]/fork/excluded-workflows/route.ts @@ -10,6 +10,7 @@ import { parseRequest } from '@/lib/api/server' import { getSession } from '@/lib/auth' import { withRouteHandler } from '@/lib/core/utils/with-route-handler' import { captureServerEvent } from '@/lib/posthog/server' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { assertWorkspaceAdminAccess } from '@/ee/workspace-forking/lib/lineage/authz' const logger = createLogger('ForkExcludedWorkflowsAPI') @@ -80,6 +81,8 @@ export const PUT = withRouteHandler( }, { groups: { workspace: workspaceId } } ) + + await notifyWorkspaceWorkflowsChanged(workspaceId) } logger.info('Updated fork-sync exclusion', { diff --git a/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.test.tsx b/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.test.tsx new file mode 100644 index 00000000000..434fcce907e --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.test.tsx @@ -0,0 +1,54 @@ +/** + * @vitest-environment jsdom + */ +import { act } from 'react' +import { ROOM_TYPES } from '@sim/realtime-protocol/rooms' +import { createRoot } from 'react-dom/client' +import { afterEach, describe, expect, it, vi } from 'vitest' + +const handlers = new Map void>() +const socket = { + connected: true, + emit: vi.fn(), + on: vi.fn((event: string, handler: (data?: { workspaceId: string }) => void) => { + handlers.set(event, handler) + }), + off: vi.fn(), +} + +vi.mock('@/app/workspace/providers/socket-provider', () => ({ + useSocket: () => ({ socket }), +})) + +import { useWorkspaceInvalidationRoom } from './use-workspace-invalidation-room' + +describe('useWorkspaceInvalidationRoom', () => { + afterEach(() => { + handlers.clear() + vi.clearAllMocks() + }) + + it('runs catch-up invalidation after a successful initial join and rejoin', () => { + ;(globalThis as { IS_REACT_ACT_ENVIRONMENT?: boolean }).IS_REACT_ACT_ENVIRONMENT = true + const container = document.createElement('div') + const root = createRoot(container) + const onChanged = vi.fn() + + function Probe() { + useWorkspaceInvalidationRoom('workspace-1', ROOM_TYPES.WORKSPACE_WORKFLOWS, onChanged) + return null + } + + act(() => root.render()) + expect(socket.emit).toHaveBeenCalledWith('join-workspace-workflows', { + workspaceId: 'workspace-1', + }) + + act(() => handlers.get('join-workspace-workflows-success')?.({ workspaceId: 'workspace-1' })) + act(() => handlers.get('connect')?.()) + act(() => handlers.get('join-workspace-workflows-success')?.({ workspaceId: 'workspace-1' })) + + expect(onChanged).toHaveBeenCalledTimes(2) + act(() => root.unmount()) + }) +}) diff --git a/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.ts b/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.ts index 0c652fcb8d6..32c706a83e9 100644 --- a/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.ts +++ b/apps/sim/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room.ts @@ -21,8 +21,8 @@ interface JoinErrorPayload { /** * Joins a workspace-scoped, presence-free "invalidation room" over the shared socket and runs * `onChanged` whenever the server broadcasts `${roomType}-changed` for this workspace, so the list - * refetches without waiting for staleness. Shared core behind {@link useWorkspaceFilesRoom} and - * {@link useWorkspaceTablesRoom}; event names derive from `roomType`. + * refetches without waiting for staleness. Shared core behind the files, tables, and workflows + * workspace rooms; event names derive from `roomType`. * * These rooms carry no presence — "who's in a resource" comes from the per-resource room, not from * who's browsing the section. Mutations happen server-side (HTTP + copilot) and fan out this signal. @@ -72,6 +72,9 @@ export function useWorkspaceInvalidationRoom( clearTimeout(retryTimer) retryTimer = null } + // The server's invalidation broadcast is intentionally lossy. Refetch after every successful + // initial join or reconnect so mutations completed before/during a disconnect are recovered. + onChangedRef.current() } const handleJoinError = (data: JoinErrorPayload) => { if (data.workspaceId !== workspaceId) return diff --git a/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/hooks/index.ts b/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/hooks/index.ts index cee9841d8e6..53cfc2026d2 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/hooks/index.ts +++ b/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/hooks/index.ts @@ -18,4 +18,5 @@ export { useWorkflowOperations } from './use-workflow-operations' export { useWorkflowSelection } from './use-workflow-selection' export { useWorkspaceLogoUpload } from './use-workspace-logo-upload' export { useWorkspaceManagement } from './use-workspace-management' +export { useWorkspaceWorkflowsRoom } from './use-workspace-workflows-room' export { WORKSPACE_LOGO_ACCEPT_ATTRIBUTE } from './workspace-logo-file' diff --git a/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/hooks/use-workspace-workflows-room.test.ts b/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/hooks/use-workspace-workflows-room.test.ts new file mode 100644 index 00000000000..f7b3a71bf0f --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/hooks/use-workspace-workflows-room.test.ts @@ -0,0 +1,47 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import { folderKeys } from '@/hooks/queries/utils/folder-keys' +import { selectorKeys } from '@/hooks/queries/utils/selector-keys' +import { workflowKeys } from '@/hooks/queries/utils/workflow-keys' + +const { mockInvalidateQueries, mockUseWorkspaceInvalidationRoom } = vi.hoisted(() => ({ + mockInvalidateQueries: vi.fn().mockResolvedValue(undefined), + mockUseWorkspaceInvalidationRoom: vi.fn(), +})) + +vi.mock('@tanstack/react-query', () => ({ + useQueryClient: () => ({ invalidateQueries: mockInvalidateQueries }), +})) + +vi.mock('@/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room', () => ({ + useWorkspaceInvalidationRoom: mockUseWorkspaceInvalidationRoom, +})) + +import { useWorkspaceWorkflowsRoom } from './use-workspace-workflows-room' + +describe('useWorkspaceWorkflowsRoom', () => { + beforeEach(() => { + vi.clearAllMocks() + }) + + it('invalidates workflow, selector, and workflow-folder caches for the current workspace', () => { + useWorkspaceWorkflowsRoom('workspace-1') + const onChanged = mockUseWorkspaceInvalidationRoom.mock.calls[0][2] + + onChanged() + + expect(mockInvalidateQueries.mock.calls.map(([options]) => options.queryKey)).toEqual( + expect.arrayContaining([ + workflowKeys.list('workspace-1', 'active'), + workflowKeys.list('workspace-1', 'archived'), + workflowKeys.list('workspace-1', 'all'), + selectorKeys.all, + folderKeys.list('workspace-1', 'active', 'workflow'), + folderKeys.list('workspace-1', 'archived', 'workflow'), + ]) + ) + expect(mockInvalidateQueries).toHaveBeenCalledTimes(6) + }) +}) diff --git a/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/hooks/use-workspace-workflows-room.ts b/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/hooks/use-workspace-workflows-room.ts new file mode 100644 index 00000000000..6183e0ce849 --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/hooks/use-workspace-workflows-room.ts @@ -0,0 +1,22 @@ +'use client' + +import { ROOM_TYPES } from '@sim/realtime-protocol/rooms' +import { useQueryClient } from '@tanstack/react-query' +import { useWorkspaceInvalidationRoom } from '@/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room' +import { folderKeys } from '@/hooks/queries/utils/folder-keys' +import { invalidateWorkflowLists } from '@/hooks/queries/utils/invalidate-workflow-lists' + +/** Keeps the persistent workflow sidebar and its folder tree live across external mutations. */ +export function useWorkspaceWorkflowsRoom(workspaceId: string): void { + const queryClient = useQueryClient() + + useWorkspaceInvalidationRoom(workspaceId, ROOM_TYPES.WORKSPACE_WORKFLOWS, () => { + void invalidateWorkflowLists(queryClient, workspaceId, ['active', 'archived', 'all']) + void queryClient.invalidateQueries({ + queryKey: folderKeys.list(workspaceId, 'active', 'workflow'), + }) + void queryClient.invalidateQueries({ + queryKey: folderKeys.list(workspaceId, 'archived', 'workflow'), + }) + }) +} diff --git a/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/sidebar.tsx b/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/sidebar.tsx index efe8b0c769e..0212a71ff7f 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/sidebar.tsx +++ b/apps/sim/app/workspace/[workspaceId]/w/components/sidebar/sidebar.tsx @@ -93,6 +93,7 @@ import { useWorkflowOperations, useWorkspaceLogoUpload, useWorkspaceManagement, + useWorkspaceWorkflowsRoom, WORKSPACE_LOGO_ACCEPT_ATTRIBUTE, } from '@/app/workspace/[workspaceId]/w/components/sidebar/hooks' import { @@ -411,6 +412,7 @@ export const Sidebar = memo(function Sidebar({ const params = useParams() const workspaceId = params.workspaceId as string const workflowId = params.workflowId as string | undefined + useWorkspaceWorkflowsRoom(workspaceId) const router = useRouter() const pathname = usePathname() diff --git a/apps/sim/ee/workspace-forking/lib/promote/promote.test.ts b/apps/sim/ee/workspace-forking/lib/promote/promote.test.ts index c44dd9c9e2a..b9be2c80d5f 100644 --- a/apps/sim/ee/workspace-forking/lib/promote/promote.test.ts +++ b/apps/sim/ee/workspace-forking/lib/promote/promote.test.ts @@ -22,6 +22,7 @@ const { mockAssertForkStorageHeadroom, mockLoadTargetWebhookPaths, mockVerifyDrops, + mockNotifyWorkspace, } = vi.hoisted(() => ({ mockComputePlan: vi.fn(), mockBuildCopySelection: vi.fn(), @@ -40,6 +41,7 @@ const { mockAssertForkStorageHeadroom: vi.fn(), mockLoadTargetWebhookPaths: vi.fn(), mockVerifyDrops: vi.fn(), + mockNotifyWorkspace: vi.fn(), })) vi.mock('@/lib/workflows/deployment-outbox', () => ({ @@ -52,6 +54,9 @@ vi.mock('@/lib/workflows/orchestration/deploy', () => ({ vi.mock('@/lib/workflows/persistence/utils', () => ({ undeployWorkflow: vi.fn(async () => ({ success: true })), })) +vi.mock('@/lib/realtime/notify', () => ({ + notifyWorkspaceWorkflowsChanged: mockNotifyWorkspace, +})) vi.mock('@/ee/workspace-forking/lib/background-work/store', () => ({ startBackgroundWork: vi.fn(), })) @@ -600,6 +605,8 @@ describe('promoteFork dependent values', () => { }) expect(result.blocked).toBeNull() + expect(mockNotifyWorkspace).toHaveBeenCalledOnce() + expect(mockNotifyWorkspace).toHaveBeenCalledWith('tgt-ws') // The apply map the workflow write receives carries the COPIED id: the dependent-value // apply runs AFTER the reference remap and wins for its subblock, so a raw source id // would clobber the remapped value in the written state. diff --git a/apps/sim/ee/workspace-forking/lib/promote/promote.ts b/apps/sim/ee/workspace-forking/lib/promote/promote.ts index 0356790c4b7..2dcaa13a68b 100644 --- a/apps/sim/ee/workspace-forking/lib/promote/promote.ts +++ b/apps/sim/ee/workspace-forking/lib/promote/promote.ts @@ -7,6 +7,7 @@ import { and, eq, inArray, isNull } from 'drizzle-orm' import type { ForkSyncBlocker, PromoteCopyResources } from '@/lib/api/contracts/workspace-fork' import type { DbOrTx } from '@/lib/db/types' import { notifyMcpToolServers } from '@/lib/mcp/workflow-mcp-sync' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { enqueueWorkflowUndeploySideEffects, processWorkflowDeploymentOutboxEvent, @@ -1008,6 +1009,10 @@ export async function promoteFork(params: PromoteForkParams): Promise 0) { + await notifyWorkspaceWorkflowsChanged(targetWorkspaceId) + } + // Process archived orphans' undeploy side-effects after commit (durably retried by the // outbox cron if this dies first), so the locked transaction never held a network call. for (const eventId of txResult.undeployEventIds) { diff --git a/apps/sim/ee/workspace-forking/lib/promote/rollback.test.ts b/apps/sim/ee/workspace-forking/lib/promote/rollback.test.ts index 340280ec150..6a99e578b1e 100644 --- a/apps/sim/ee/workspace-forking/lib/promote/rollback.test.ts +++ b/apps/sim/ee/workspace-forking/lib/promote/rollback.test.ts @@ -16,6 +16,7 @@ const { mockProcessOutbox, mockNotify, mockGetDeploymentStatus, + mockNotifyWorkspace, } = vi.hoisted(() => ({ mockResolveForkEdge: vi.fn(), mockAcquireTargetLock: vi.fn(), @@ -29,6 +30,7 @@ const { mockProcessOutbox: vi.fn(), mockNotify: vi.fn(), mockGetDeploymentStatus: vi.fn(), + mockNotifyWorkspace: vi.fn(), })) vi.mock('@/ee/workspace-forking/lib/lineage/lineage', () => ({ @@ -67,14 +69,17 @@ vi.mock('@/lib/workflows/deployment-outbox', () => ({ vi.mock('@/ee/workspace-forking/lib/socket', () => ({ notifyForkWorkflowChanged: mockNotify, })) +vi.mock('@/lib/realtime/notify', () => ({ + notifyWorkspaceWorkflowsChanged: mockNotifyWorkspace, +})) import { db } from '@sim/db' 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(() => ({ @@ -82,14 +87,21 @@ function makeTx(existingUndeployIds: string[] = []) { })), })), 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 ) } @@ -155,9 +167,12 @@ describe('rollbackFork', () => { expect(mockDeleteAllRuns).toHaveBeenCalledTimes(1) expect(mockNotify).toHaveBeenCalledWith('wf-a') expect(mockNotify).toHaveBeenCalledWith('wf-b') + expect(mockNotifyWorkspace).toHaveBeenCalledOnce() + expect(mockNotifyWorkspace).toHaveBeenCalledWith('target-ws') }) 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 }] }, }) @@ -180,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: [] }, diff --git a/apps/sim/ee/workspace-forking/lib/promote/rollback.ts b/apps/sim/ee/workspace-forking/lib/promote/rollback.ts index c2981d67cda..f1f1e2dee48 100644 --- a/apps/sim/ee/workspace-forking/lib/promote/rollback.ts +++ b/apps/sim/ee/workspace-forking/lib/promote/rollback.ts @@ -3,6 +3,7 @@ import { chat, workflow } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { and, inArray, isNull } from 'drizzle-orm' import { generateRequestId } from '@/lib/core/utils/request' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { enqueueWorkflowUndeploySideEffects, processWorkflowDeploymentOutboxEvent, @@ -130,6 +131,7 @@ export async function rollbackFork(params: RollbackForkParams): Promise() const outboxEventIds: string[] = [] const reactivations: Array<{ workflowId: string; operationId: string }> = [] + const unarchivedWorkflowIds = new Set() await db.transaction(async (tx) => { await setForkLockTimeout(tx) @@ -150,7 +152,7 @@ export async function rollbackFork(params: RollbackForkParams): Promise 0) { - await tx + const unarchived = await tx .update(workflow) .set({ archivedAt: null, updatedAt: now }) .where( @@ -159,6 +161,8 @@ export async function rollbackFork(params: RollbackForkParams): Promise 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 @@ -247,6 +251,12 @@ export async function rollbackFork(params: RollbackForkParams): Promise 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. diff --git a/apps/sim/lib/copilot/tools/handlers/materialize-file.test.ts b/apps/sim/lib/copilot/tools/handlers/materialize-file.test.ts index ae1b44dab75..cf41a59b921 100644 --- a/apps/sim/lib/copilot/tools/handlers/materialize-file.test.ts +++ b/apps/sim/lib/copilot/tools/handlers/materialize-file.test.ts @@ -20,6 +20,7 @@ const { mockMaybeNotifyStorageLimitForBillingContext, mockReadWorkspaceFileMetadata, mockResolveStorageBillingContext, + mockNotifyWorkspace, } = vi.hoisted(() => ({ mockAllocateUniqueWorkspaceFileName: vi.fn(), mockAdmitCreateWorkspaceFile: vi.fn(), @@ -36,6 +37,7 @@ const { mockMaybeNotifyStorageLimitForBillingContext: vi.fn(), mockReadWorkspaceFileMetadata: vi.fn(), mockResolveStorageBillingContext: vi.fn(), + mockNotifyWorkspace: vi.fn(), })) vi.mock('@/lib/copilot/tools/handlers/access', () => ({ @@ -111,6 +113,9 @@ vi.mock('@/lib/workflows/operations/import-export', () => ({ parseWorkflowJson: vi.mock('@/lib/workflows/persistence/utils', () => ({ saveWorkflowToNormalizedTables: vi.fn() })) vi.mock('@/lib/workflows/utils', () => ({ deduplicateWorkflowName: vi.fn() })) vi.mock('@/app/api/v1/admin/types', () => ({ extractWorkflowMetadata: vi.fn() })) +vi.mock('@/lib/realtime/notify', () => ({ + notifyWorkspaceWorkflowsChanged: mockNotifyWorkspace, +})) import type { ExecutionContext } from '@/lib/copilot/request/types' import { executeMaterializeFile } from '@/lib/copilot/tools/handlers/materialize-file' @@ -272,6 +277,7 @@ describe('executeMaterializeFile - workflow import', () => { ) expect(result.success).toBe(true) + expect(mockNotifyWorkspace).toHaveBeenCalledWith('ws-1') const insertedWorkflow = dbChainMockFns.values.mock.calls[0]?.[0] as Record expect(insertedWorkflow).toMatchObject({ name: 'Imported Workflow' }) expect(insertedWorkflow).not.toHaveProperty('description') diff --git a/apps/sim/lib/copilot/tools/handlers/materialize-file.ts b/apps/sim/lib/copilot/tools/handlers/materialize-file.ts index aefdec1dabc..b1bac6bb4a4 100644 --- a/apps/sim/lib/copilot/tools/handlers/materialize-file.ts +++ b/apps/sim/lib/copilot/tools/handlers/materialize-file.ts @@ -28,6 +28,7 @@ import { ensureWorkspaceAccess } from '@/lib/copilot/tools/handlers/access' import { findMothershipUploadRowByChatAndName } from '@/lib/copilot/tools/handlers/upload-file-reader' import { canonicalWorkspaceFilePath, encodeVfsPathSegments } from '@/lib/copilot/vfs/path-utils' import { asOrchestrationError } from '@/lib/core/orchestration/types' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { getServePathPrefix } from '@/lib/uploads' import { ArchiveError, @@ -354,6 +355,8 @@ async function executeImport( .where(eq(workflow.id, workflowId)) } + await notifyWorkspaceWorkflowsChanged(workspaceId) + logger.info('Imported workflow from upload', { fileName, workflowId, diff --git a/apps/sim/lib/realtime/notify.test.ts b/apps/sim/lib/realtime/notify.test.ts index 98196492257..17cb2cb1fac 100644 --- a/apps/sim/lib/realtime/notify.test.ts +++ b/apps/sim/lib/realtime/notify.test.ts @@ -6,7 +6,28 @@ import { afterEach, describe, expect, it, vi } from 'vitest' vi.mock('@/lib/core/utils/urls', () => ({ getSocketServerUrl: () => 'http://realtime' })) vi.mock('@/lib/core/config/env', () => ({ env: { INTERNAL_API_SECRET: 'secret' } })) -import { mergeEditIntoLiveFileDoc } from './notify' +import { mergeEditIntoLiveFileDoc, notifyFolderResourceChanged } from './notify' + +describe('notifyFolderResourceChanged', () => { + afterEach(() => { + vi.unstubAllGlobals() + }) + + it('routes workflow folder changes to the workspace workflows endpoint', async () => { + const fetchMock = vi.fn().mockResolvedValue({ ok: true }) + vi.stubGlobal('fetch', fetchMock) + + await notifyFolderResourceChanged('workflow', 'workspace-1') + + expect(fetchMock).toHaveBeenCalledWith( + 'http://realtime/api/workspace-workflows-changed', + expect.objectContaining({ + method: 'POST', + body: JSON.stringify({ workspaceId: 'workspace-1' }), + }) + ) + }) +}) describe('mergeEditIntoLiveFileDoc', () => { afterEach(() => { diff --git a/apps/sim/lib/realtime/notify.ts b/apps/sim/lib/realtime/notify.ts index f1d9846919b..1ba7cda633a 100644 --- a/apps/sim/lib/realtime/notify.ts +++ b/apps/sim/lib/realtime/notify.ts @@ -84,6 +84,29 @@ export async function notifyWorkspaceTablesChanged(workspaceId: string): Promise } } +/** Best-effort fan-out that refreshes workflow and workflow-folder lists for a workspace. */ +export async function notifyWorkspaceWorkflowsChanged(workspaceId: string): Promise { + try { + const response = await fetch(`${getSocketServerUrl()}/api/workspace-workflows-changed`, { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'x-api-key': env.INTERNAL_API_SECRET }, + body: JSON.stringify({ workspaceId }), + signal: AbortSignal.timeout(NOTIFY_TIMEOUT_MS), + }) + if (!response.ok) { + logger.warn('workspace-workflows-changed notify failed', { + workspaceId, + status: response.status, + }) + } + } catch (error) { + logger.warn('workspace-workflows-changed notify error', { + workspaceId, + error: getErrorMessage(error), + }) + } +} + /** Best-effort fan-out that invalidates open editors for one durably changed workflow. */ export async function notifyWorkflowUpdated(workflowId: string): Promise { try { @@ -149,12 +172,13 @@ export async function notifyWorkflowReverted(workflowId: string, timestamp: numb * (create/rename/move/delete/restore) for one of these must fan out the same list-changed signal as a * direct resource mutation, because a new/renamed/removed folder changes what that resource's browser * shows. Extend this map as more resource lists adopt an invalidation room — `file` and - * `knowledge_base` currently refetch through their own paths, and `workflow` has no such list room. + * `knowledge_base` currently refetch through their own paths. */ const FOLDER_RESOURCE_NOTIFIERS: Partial< Record Promise> > = { table: notifyWorkspaceTablesChanged, + workflow: notifyWorkspaceWorkflowsChanged, } /** diff --git a/apps/sim/lib/workflows/application/create-workflow.ts b/apps/sim/lib/workflows/application/create-workflow.ts index 53580cd0e21..31f1f57eb88 100644 --- a/apps/sim/lib/workflows/application/create-workflow.ts +++ b/apps/sim/lib/workflows/application/create-workflow.ts @@ -6,7 +6,7 @@ import { OrchestrationError } from '@/lib/core/orchestration/types' import { PlatformEvents } from '@/lib/core/telemetry' import { MAX_FOLDERS_PER_WORKSPACE } from '@/lib/folders/constants' import { loadActiveFolderPathIndex } from '@/lib/folders/queries' -import { notifyWorkflowUpdated } from '@/lib/realtime/notify' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { defineAuthorizedWorkflowUseCase } from '@/lib/workflows/application/authorized-workflow-use-case' import { workflowOperations } from '@/lib/workflows/application/operations' import { requireWorkflowTransition } from '@/lib/workflows/application/transition-result' @@ -98,7 +98,7 @@ export const createWorkflow = defineAuthorizedWorkflowUseCase({ }, }), async afterSuccess({ result }) { - await notifyWorkflowUpdated(result.workflow.id) + await notifyWorkspaceWorkflowsChanged(result.workflow.workspaceId) try { PlatformEvents.workflowCreated({ workflowId: result.workflow.id, diff --git a/apps/sim/lib/workflows/application/delete-workflow.ts b/apps/sim/lib/workflows/application/delete-workflow.ts index cfa01b44119..99d98f05b60 100644 --- a/apps/sim/lib/workflows/application/delete-workflow.ts +++ b/apps/sim/lib/workflows/application/delete-workflow.ts @@ -3,7 +3,7 @@ import { type Principal, resolvePrincipalAttribution } from '@sim/auth/principal import { createLogger } from '@sim/logger' import { assertWorkflowMutable, WorkflowLockedError } from '@sim/platform-authz/workflow' import { OrchestrationError } from '@/lib/core/orchestration/types' -import { notifyWorkflowDeleted } from '@/lib/realtime/notify' +import { notifyWorkflowDeleted, notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { defineAuthorizedWorkflowUseCase } from '@/lib/workflows/application/authorized-workflow-use-case' import { resolveActiveWorkflowApplicationContext } from '@/lib/workflows/application/context' import { workflowOperations } from '@/lib/workflows/application/operations' @@ -69,6 +69,9 @@ export const deleteWorkflow = defineAuthorizedWorkflowUseCase({ metadata: { archived: true }, } : [], - afterSuccess: ({ context, result }) => - result.archived ? notifyWorkflowDeleted(context.workflowId) : undefined, + afterSuccess: async ({ context, result }) => { + if (!result.archived) return + await notifyWorkflowDeleted(context.workflowId) + await notifyWorkspaceWorkflowsChanged(context.workspaceId) + }, }) diff --git a/apps/sim/lib/workflows/application/duplicate-workflow.ts b/apps/sim/lib/workflows/application/duplicate-workflow.ts index a1a3abe0902..a62fd6acb72 100644 --- a/apps/sim/lib/workflows/application/duplicate-workflow.ts +++ b/apps/sim/lib/workflows/application/duplicate-workflow.ts @@ -7,7 +7,7 @@ import { OrchestrationError } from '@/lib/core/orchestration/types' import { generateRequestId } from '@/lib/core/utils/request' import { MAX_FOLDERS_PER_WORKSPACE } from '@/lib/folders/constants' import { loadActiveFolderPathIndex } from '@/lib/folders/queries' -import { notifyWorkflowUpdated } from '@/lib/realtime/notify' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { defineAuthorizedWorkflowUseCase } from '@/lib/workflows/application/authorized-workflow-use-case' import { resolveActiveWorkflowApplicationContext } from '@/lib/workflows/application/context' import { workflowOperations } from '@/lib/workflows/application/operations' @@ -100,5 +100,5 @@ export const duplicateWorkflow = defineAuthorizedWorkflowUseCase({ source: principalAuditSource(principal), }, }), - afterSuccess: ({ result }) => notifyWorkflowUpdated(result.id), + afterSuccess: ({ context }) => notifyWorkspaceWorkflowsChanged(context.workspaceId), }) diff --git a/apps/sim/lib/workflows/application/move-workflows-bulk.test.ts b/apps/sim/lib/workflows/application/move-workflows-bulk.test.ts index 20260e59dfd..3a7ee0336b5 100644 --- a/apps/sim/lib/workflows/application/move-workflows-bulk.test.ts +++ b/apps/sim/lib/workflows/application/move-workflows-bulk.test.ts @@ -15,6 +15,7 @@ const { FolderLockedError, WorkflowLockedError, mocks } = vi.hoisted(() => { assertWorkflowMutable: vi.fn(), audit: vi.fn(), notify: vi.fn(), + notifyWorkspace: vi.fn(), permission: vi.fn(), resolveContext: vi.fn(), updateWorkflow: vi.fn(), @@ -48,7 +49,10 @@ vi.mock('@/lib/workflows/orchestration', () => ({ updateWorkflowRecord: mocks.updateWorkflow, })) -vi.mock('@/lib/realtime/notify', () => ({ notifyWorkflowUpdated: mocks.notify })) +vi.mock('@/lib/realtime/notify', () => ({ + notifyWorkflowUpdated: mocks.notify, + notifyWorkspaceWorkflowsChanged: mocks.notifyWorkspace, +})) import { moveWorkflowsBulk } from '@/lib/workflows/application/move-workflows-bulk' @@ -118,6 +122,30 @@ describe('moveWorkflowsBulk', () => { ) expect(mocks.notify).toHaveBeenCalledWith('workflow-1') expect(mocks.notify).not.toHaveBeenCalledWith('workflow-2') + expect(mocks.notifyWorkspace).toHaveBeenCalledOnce() + 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 () => { @@ -139,6 +167,7 @@ describe('moveWorkflowsBulk', () => { expect(mocks.updateWorkflow).not.toHaveBeenCalled() expect(mocks.audit).not.toHaveBeenCalled() + expect(mocks.notifyWorkspace).not.toHaveBeenCalled() }) it('rejects a delegated service the operation does not accept, before canonical loading', async () => { diff --git a/apps/sim/lib/workflows/application/move-workflows-bulk.ts b/apps/sim/lib/workflows/application/move-workflows-bulk.ts index 2232a78150a..74ca726110e 100644 --- a/apps/sim/lib/workflows/application/move-workflows-bulk.ts +++ b/apps/sim/lib/workflows/application/move-workflows-bulk.ts @@ -11,7 +11,7 @@ import { import { and, eq, inArray, isNull } from 'drizzle-orm' import { principalAuditSource } from '@/lib/core/application' import { asOrchestrationError, OrchestrationError } from '@/lib/core/orchestration/types' -import { notifyWorkflowUpdated } from '@/lib/realtime/notify' +import { notifyWorkflowUpdated, notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { defineAuthorizedWorkflowUseCase } from '@/lib/workflows/application/authorized-workflow-use-case' import { workflowOperations } from '@/lib/workflows/application/operations' import { requireWorkflowTransition } from '@/lib/workflows/application/transition-result' @@ -171,9 +171,12 @@ export const moveWorkflowsBulk = defineAuthorizedWorkflowUseCase({ source: principalAuditSource(principal), }, })), - afterSuccess: async ({ result }) => { + afterSuccess: async ({ context, result }) => { for (const workflowId of result.moved) { await notifyWorkflowUpdated(workflowId) } + if (result.changes.some((change) => change.previousFolderId !== result.folderId)) { + await notifyWorkspaceWorkflowsChanged(context.workspaceId) + } }, }) diff --git a/apps/sim/lib/workflows/application/restore-workflow.test.ts b/apps/sim/lib/workflows/application/restore-workflow.test.ts index def7c6931a9..7953f96d3a4 100644 --- a/apps/sim/lib/workflows/application/restore-workflow.test.ts +++ b/apps/sim/lib/workflows/application/restore-workflow.test.ts @@ -10,6 +10,7 @@ const mocks = vi.hoisted(() => ({ resolveContext: vi.fn(), resolvePermission: vi.fn(), notify: vi.fn(), + notifyWorkspace: vi.fn(), restoreRecord: vi.fn(), folderIndex: vi.fn(), })) @@ -33,7 +34,10 @@ vi.mock('@sim/platform-authz/workspace', () => ({ vi.mock('@/lib/workflows/application/context', () => ({ resolveArchivedWorkflowApplicationContext: mocks.resolveContext, })) -vi.mock('@/lib/realtime/notify', () => ({ notifyWorkflowUpdated: mocks.notify })) +vi.mock('@/lib/realtime/notify', () => ({ + notifyWorkflowUpdated: mocks.notify, + notifyWorkspaceWorkflowsChanged: mocks.notifyWorkspace, +})) vi.mock('@/lib/workflows/lifecycle', () => ({ restoreWorkflow: mocks.restoreRecord })) vi.mock('@/lib/folders/queries', () => ({ loadActiveFolderPathIndex: mocks.folderIndex })) @@ -87,6 +91,7 @@ describe('restoreWorkflow', () => { }) ) expect(mocks.recordAudit).toHaveBeenCalledBefore(mocks.notify) + expect(mocks.notifyWorkspace).toHaveBeenCalledWith('workspace-1') }) it('refuses a workflow that is not archived as a conflict', async () => { diff --git a/apps/sim/lib/workflows/application/restore-workflow.ts b/apps/sim/lib/workflows/application/restore-workflow.ts index 44573566e88..46a976c57ac 100644 --- a/apps/sim/lib/workflows/application/restore-workflow.ts +++ b/apps/sim/lib/workflows/application/restore-workflow.ts @@ -11,7 +11,7 @@ import { OrchestrationError } from '@/lib/core/orchestration/types' import { generateRequestId } from '@/lib/core/utils/request' import { MAX_FOLDERS_PER_WORKSPACE } from '@/lib/folders/constants' import { loadActiveFolderPathIndex } from '@/lib/folders/queries' -import { notifyWorkflowUpdated } from '@/lib/realtime/notify' +import { notifyWorkflowUpdated, notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { defineAuthorizedWorkflowUseCase } from '@/lib/workflows/application/authorized-workflow-use-case' import { resolveArchivedWorkflowApplicationContext } from '@/lib/workflows/application/context' import { workflowOperations } from '@/lib/workflows/application/operations' @@ -94,5 +94,8 @@ export const restoreWorkflow = defineAuthorizedWorkflowUseCase({ source: principalAuditSource(principal), }, }), - afterSuccess: ({ context }) => notifyWorkflowUpdated(context.workflowId), + afterSuccess: async ({ context }) => { + await notifyWorkflowUpdated(context.workflowId) + await notifyWorkspaceWorkflowsChanged(context.workspaceId) + }, }) diff --git a/apps/sim/lib/workflows/application/update-workflow.ts b/apps/sim/lib/workflows/application/update-workflow.ts index 9002d9839df..b0c11266eae 100644 --- a/apps/sim/lib/workflows/application/update-workflow.ts +++ b/apps/sim/lib/workflows/application/update-workflow.ts @@ -11,7 +11,7 @@ import type { WorkspaceUseCaseAuditEntry } from '@/lib/core/application' import { OrchestrationError } from '@/lib/core/orchestration/types' import { MAX_FOLDERS_PER_WORKSPACE } from '@/lib/folders/constants' import { loadActiveFolderPathIndex } from '@/lib/folders/queries' -import { notifyWorkflowUpdated } from '@/lib/realtime/notify' +import { notifyWorkflowUpdated, notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { defineAuthorizedWorkflowUseCase } from '@/lib/workflows/application/authorized-workflow-use-case' import { type ActiveWorkflowApplicationContext, @@ -249,11 +249,13 @@ function projectWorkflowUpdateAudit(args: { return entries } -function notifyAfterWorkflowUpdate(args: { +async function notifyAfterWorkflowUpdate(args: { context: ActiveWorkflowApplicationContext result: WorkflowUpdateResult }) { - return args.result.changes.length > 0 ? notifyWorkflowUpdated(args.context.workflowId) : undefined + if (args.result.changes.length === 0) return + await notifyWorkflowUpdated(args.context.workflowId) + await notifyWorkspaceWorkflowsChanged(args.context.workspaceId) } export const updateWorkflow = defineAuthorizedWorkflowUseCase({ diff --git a/apps/sim/lib/workflows/application/workflow-crud.test.ts b/apps/sim/lib/workflows/application/workflow-crud.test.ts index 8c7168eab0d..78ad8d6d355 100644 --- a/apps/sim/lib/workflows/application/workflow-crud.test.ts +++ b/apps/sim/lib/workflows/application/workflow-crud.test.ts @@ -22,6 +22,8 @@ const mocks = vi.hoisted(() => ({ readVersion: vi.fn(), loadNormalized: vi.fn(), notifyWorkflowUpdated: vi.fn(), + notifyWorkflowDeleted: vi.fn(), + notifyWorkspaceWorkflowsChanged: vi.fn(), workflowCreated: vi.fn(), })) @@ -91,6 +93,8 @@ vi.mock('@/lib/workflows/persistence/utils', () => ({ vi.mock('@/lib/realtime/notify', () => ({ notifyWorkflowUpdated: mocks.notifyWorkflowUpdated, + notifyWorkflowDeleted: mocks.notifyWorkflowDeleted, + notifyWorkspaceWorkflowsChanged: mocks.notifyWorkspaceWorkflowsChanged, })) vi.mock('@/lib/core/telemetry', () => ({ @@ -236,7 +240,7 @@ describe('authorized workflow CRUD and version reads', () => { }), }) ) - expect(mocks.notifyWorkflowUpdated).toHaveBeenCalledWith(WORKFLOW_ID) + expect(mocks.notifyWorkspaceWorkflowsChanged).toHaveBeenCalledWith(WORKSPACE_ID) expect(mocks.workflowCreated).toHaveBeenCalledWith( expect.objectContaining({ workflowId: WORKFLOW_ID, workspaceId: WORKSPACE_ID }) ) @@ -277,6 +281,8 @@ describe('authorized workflow CRUD and version reads', () => { }) ).rejects.toBe(failure) expect(mocks.recordAudit).not.toHaveBeenCalled() + expect(mocks.notifyWorkflowDeleted).not.toHaveBeenCalled() + expect(mocks.notifyWorkspaceWorkflowsChanged).not.toHaveBeenCalled() }) it('returns forbidden when a workspace key does not match canonical workflow scope', async () => { diff --git a/apps/sim/lib/workflows/application/workflow-vfs.test.ts b/apps/sim/lib/workflows/application/workflow-vfs.test.ts index bb6303f2a26..b48f4466ff8 100644 --- a/apps/sim/lib/workflows/application/workflow-vfs.test.ts +++ b/apps/sim/lib/workflows/application/workflow-vfs.test.ts @@ -22,6 +22,8 @@ const { FolderLockedError, WorkflowLockedError, mocks } = vi.hoisted(() => { logError: vi.fn(), notifyFolder: vi.fn(), notifyWorkflow: vi.fn(), + notifyWorkflowDeleted: vi.fn(), + notifyWorkspace: vi.fn(), permission: vi.fn(), relocateFolder: vi.fn(), resolveContext: vi.fn(), @@ -94,7 +96,9 @@ vi.mock('@/lib/workflows/persistence/duplicate', () => ({ vi.mock('@/lib/realtime/notify', () => ({ notifyFolderResourceChanged: mocks.notifyFolder, + notifyWorkflowDeleted: mocks.notifyWorkflowDeleted, notifyWorkflowUpdated: mocks.notifyWorkflow, + notifyWorkspaceWorkflowsChanged: mocks.notifyWorkspace, })) import { @@ -171,15 +175,21 @@ describe('workflow VFS application commands', () => { queueTableRows(schemaMock.workflow, [ { id: 'workflow-1', name: 'One', folderId: null }, { id: 'workflow-2', name: 'Two', folderId: null }, + { id: 'workflow-3', name: 'Three', folderId: null }, ]) queueTableRows(schemaMock.workflow, [{ id: 'workflow-1', name: 'One', folderId: null }]) queueTableRows(schemaMock.workflow, [{ id: 'workflow-2', name: 'Two', folderId: null }]) + queueTableRows(schemaMock.workflow, [{ id: 'workflow-3', name: 'Three', folderId: null }]) mocks.updateWorkflow .mockResolvedValueOnce({ success: true, workflow: { id: 'workflow-1', name: 'One', folderId: null }, }) .mockResolvedValueOnce({ success: false, error: 'Workflow is locked', errorCode: 'locked' }) + .mockResolvedValueOnce({ + success: true, + workflow: { id: 'workflow-3', name: 'Three', folderId: null }, + }) const result = await moveWorkflowVfsItems.execute({ principal, @@ -188,6 +198,7 @@ describe('workflow VFS application commands', () => { sources: [ { source: 'workflows/One', segments: ['One'] }, { source: 'workflows/Two', segments: ['Two'] }, + { source: 'workflows/Three', segments: ['Three'] }, ], destination: { segments: [], trailingSlash: true }, }, @@ -197,8 +208,9 @@ describe('workflow VFS application commands', () => { expect(result.outcomes).toEqual([ expect.objectContaining({ source: 'workflows/One', resourceId: 'workflow-1' }), expect.objectContaining({ source: 'workflows/Two', error: 'Workflow is locked' }), + expect.objectContaining({ source: 'workflows/Three', resourceId: 'workflow-3' }), ]) - expect(mocks.audit).toHaveBeenCalledOnce() + expect(mocks.audit).toHaveBeenCalledTimes(2) expect(mocks.audit).toHaveBeenCalledWith( expect.objectContaining({ action: 'workflow.updated', @@ -208,6 +220,9 @@ describe('workflow VFS application commands', () => { ) expect(mocks.notifyWorkflow).toHaveBeenCalledWith('workflow-1') expect(mocks.notifyWorkflow).not.toHaveBeenCalledWith('workflow-2') + expect(mocks.notifyWorkflow).toHaveBeenCalledWith('workflow-3') + expect(mocks.notifyWorkspace).toHaveBeenCalledOnce() + expect(mocks.notifyWorkspace).toHaveBeenCalledWith('workspace-1') }) it('propagates an unexpected mutation failure without projecting a partial outcome', async () => { @@ -229,6 +244,7 @@ describe('workflow VFS application commands', () => { expect(mocks.audit).not.toHaveBeenCalled() expect(mocks.notifyWorkflow).not.toHaveBeenCalled() expect(mocks.notifyFolder).not.toHaveBeenCalled() + expect(mocks.notifyWorkspace).not.toHaveBeenCalled() }) it('owns mkdir path planning and audits only the folder it creates', async () => { @@ -274,6 +290,6 @@ describe('workflow VFS application commands', () => { metadata: expect.objectContaining({ operation: 'workflows.vfs.folders.create' }), }) ) - expect(mocks.notifyFolder).toHaveBeenCalledWith('workflow', 'workspace-1') + expect(mocks.notifyWorkspace).toHaveBeenCalledWith('workspace-1') }) }) diff --git a/apps/sim/lib/workflows/application/workflow-vfs.ts b/apps/sim/lib/workflows/application/workflow-vfs.ts index a313729984b..8d2059bc0fa 100644 --- a/apps/sim/lib/workflows/application/workflow-vfs.ts +++ b/apps/sim/lib/workflows/application/workflow-vfs.ts @@ -29,9 +29,9 @@ import { } from '@/lib/folders/paths' import { loadActiveFolderPathIndex } from '@/lib/folders/queries' import { - notifyFolderResourceChanged, notifyWorkflowDeleted, notifyWorkflowUpdated, + notifyWorkspaceWorkflowsChanged, } from '@/lib/realtime/notify' import { VfsPathLimitError, validateVfsPathSegments } from '@/lib/vfs/limits' import { encodeVfsPathSegments } from '@/lib/vfs/path' @@ -462,7 +462,7 @@ export const createWorkflowVfsFolders = defineAuthorizedWorkflowUseCase({ projectAudit: ({ result }) => createdFolderAuditEntries(result.createdFolders), afterSuccess: ({ context, result }) => result.createdFolders.length > 0 - ? notifyFolderResourceChanged('workflow', context.workspaceId) + ? notifyWorkspaceWorkflowsChanged(context.workspaceId) : undefined, }) @@ -606,8 +606,12 @@ export const moveWorkflowVfsItems = defineAuthorizedWorkflowUseCase({ for (const change of result.movedWorkflows) { await notifyWorkflowUpdated(change.id) } - if (result.createdFolders.length > 0 || result.movedFolders.length > 0) { - await notifyFolderResourceChanged('workflow', context.workspaceId) + if ( + result.createdFolders.length > 0 || + result.movedWorkflows.length > 0 || + result.movedFolders.length > 0 + ) { + await notifyWorkspaceWorkflowsChanged(context.workspaceId) } }, }) @@ -717,8 +721,8 @@ export const copyWorkflowVfsItems = defineAuthorizedWorkflowUseCase({ for (const change of result.duplicatedWorkflows) { await notifyWorkflowUpdated(change.id) } - if (result.createdFolders.length > 0) { - await notifyFolderResourceChanged('workflow', context.workspaceId) + if (result.createdFolders.length > 0 || result.duplicatedWorkflows.length > 0) { + await notifyWorkspaceWorkflowsChanged(context.workspaceId) } }, }) @@ -840,8 +844,8 @@ export const deleteWorkflowVfsItems = defineAuthorizedWorkflowUseCase({ for (const workflow of result.deletedWorkflows) { await notifyWorkflowDeleted(workflow.id) } - if (result.deletedFolders.length > 0) { - await notifyFolderResourceChanged('workflow', context.workspaceId) + if (result.deletedWorkflows.length > 0 || result.deletedFolders.length > 0) { + await notifyWorkspaceWorkflowsChanged(context.workspaceId) } }, }) diff --git a/apps/sim/lib/workflows/deployment-outbox.test.ts b/apps/sim/lib/workflows/deployment-outbox.test.ts index 4483575af30..3a3b108f5e0 100644 --- a/apps/sim/lib/workflows/deployment-outbox.test.ts +++ b/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() diff --git a/apps/sim/lib/workflows/deployment-outbox.ts b/apps/sim/lib/workflows/deployment-outbox.ts index 62875063496..50a34c94ab7 100644 --- a/apps/sim/lib/workflows/deployment-outbox.ts +++ b/apps/sim/lib/workflows/deployment-outbox.ts @@ -23,6 +23,7 @@ import { syncMcpToolsForWorkflow, } from '@/lib/mcp/workflow-mcp-sync' import { captureServerEvent } from '@/lib/posthog/server' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { cleanupWebhooksForWorkflow, prepareStableTriggerWebhooksForDeploy, @@ -85,6 +86,7 @@ interface DeploymentPreparationCheckpoints { auditEmitted?: boolean analyticsCaptured?: boolean socketNotified?: boolean + workspaceListNotified?: boolean workspaceEventEmitted?: boolean } @@ -728,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, { @@ -738,7 +747,6 @@ async function emitPostActivationSideEffects(params: { await params.checkpoint({ socketNotified: true }) } - const workspaceId = params.workflow.workspaceId as string | null if (workspaceId && !params.checkpoints.workspaceEventEmitted) { params.context.signal.throwIfAborted() await emitWorkflowDeployedEvent({ @@ -1448,6 +1456,7 @@ function parseDeploymentPreparationCheckpoints(value: unknown): DeploymentPrepar ...(record.auditEmitted === true ? { auditEmitted: true } : {}), ...(record.analyticsCaptured === true ? { analyticsCaptured: true } : {}), ...(record.socketNotified === true ? { socketNotified: true } : {}), + ...(record.workspaceListNotified === true ? { workspaceListNotified: true } : {}), ...(record.workspaceEventEmitted === true ? { workspaceEventEmitted: true } : {}), } } diff --git a/apps/sim/lib/workflows/operations/import-workflow.ts b/apps/sim/lib/workflows/operations/import-workflow.ts index a6c42d03dbf..d3d6ecea66b 100644 --- a/apps/sim/lib/workflows/operations/import-workflow.ts +++ b/apps/sim/lib/workflows/operations/import-workflow.ts @@ -16,6 +16,7 @@ import { } from '@/lib/api/contracts/v1/workflows' import { workflowStateSchema } from '@/lib/api/contracts/workflows' import { serializeZodIssues } from '@/lib/api/server' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { parseWorkflowJson } from '@/lib/workflows/operations/import-export' import { type PerformCreateWorkflowParams, @@ -412,6 +413,8 @@ async function executeImportWorkflowIntoWorkspace( blocksCount: Object.keys(workflowState.blocks).length, }) + await notifyWorkspaceWorkflowsChanged(workspaceId) + return { success: true, workflow: { diff --git a/apps/sim/lib/workflows/orchestration/deploy.ts b/apps/sim/lib/workflows/orchestration/deploy.ts index 097a296a83e..1ef0e71e1cc 100644 --- a/apps/sim/lib/workflows/orchestration/deploy.ts +++ b/apps/sim/lib/workflows/orchestration/deploy.ts @@ -13,6 +13,7 @@ import type { OrchestrationErrorCode } from '@/lib/core/orchestration/types' import { generateRequestId } from '@/lib/core/utils/request' import { getSocketServerUrl } from '@/lib/core/utils/urls' import { captureServerEvent } from '@/lib/posthog/server' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { validateTriggerWebhookConfigForDeploy } from '@/lib/webhooks/deploy' import { normalizedStringify } from '@/lib/workflows/comparison/normalize' import { @@ -579,9 +580,12 @@ export async function performFullUndeploy( } await notifySocketDeploymentChanged(workflowId) + const undeployWorkspaceId = workflowData.workspaceId as string | null + if (undeployWorkspaceId) { + await notifyWorkspaceWorkflowsChanged(undeployWorkspaceId) + } const sideEffectWarning = await processDeploymentSideEffectsNow(outboxEventId, requestId) - const undeployWorkspaceId = workflowData.workspaceId as string | null if (undeployWorkspaceId) { void emitWorkflowUndeployedEvent({ workflowId, diff --git a/apps/sim/lib/workflows/orchestration/workflow-lifecycle.ts b/apps/sim/lib/workflows/orchestration/workflow-lifecycle.ts index d2098cbd418..4552dc9d2c1 100644 --- a/apps/sim/lib/workflows/orchestration/workflow-lifecycle.ts +++ b/apps/sim/lib/workflows/orchestration/workflow-lifecycle.ts @@ -9,6 +9,7 @@ import { and, eq, isNull, ne } from 'drizzle-orm' import type { OrchestrationErrorCode } from '@/lib/core/orchestration/types' import { generateRequestId } from '@/lib/core/utils/request' import type { DbOrTx } from '@/lib/db/types' +import { notifyWorkspaceWorkflowsChanged } from '@/lib/realtime/notify' import { buildDefaultWorkflowArtifacts } from '@/lib/workflows/defaults' import { archiveWorkflow, restoreWorkflow } from '@/lib/workflows/lifecycle' import { saveWorkflowToNormalizedTables } from '@/lib/workflows/persistence/utils' @@ -486,6 +487,10 @@ export async function performDeleteWorkflow( metadata: { archived: true }, }) + if (result.workflow.workspaceId) { + await notifyWorkspaceWorkflowsChanged(result.workflow.workspaceId) + } + return result } @@ -525,6 +530,10 @@ export async function performRestoreWorkflow( }, }) + if (restoreResult.workflow.workspaceId) { + await notifyWorkspaceWorkflowsChanged(restoreResult.workflow.workspaceId) + } + return { success: true, workflow: restoreResult.workflow } } catch (error) { logger.error(`[${requestId}] Failed to restore workflow ${workflowId}`, { error }) diff --git a/packages/platform-authz/src/room-policy.ts b/packages/platform-authz/src/room-policy.ts index 0d22ded1a3c..73ea28105ed 100644 --- a/packages/platform-authz/src/room-policy.ts +++ b/packages/platform-authz/src/room-policy.ts @@ -22,6 +22,7 @@ export const ROOM_MEMBERSHIP_ACTIONS = { [ROOM_TYPES.WORKFLOW]: 'read', [ROOM_TYPES.WORKSPACE_FILES]: 'read', [ROOM_TYPES.WORKSPACE_TABLES]: 'read', + [ROOM_TYPES.WORKSPACE_WORKFLOWS]: 'read', [ROOM_TYPES.WORKSPACE_FILE_DOC]: 'write', [ROOM_TYPES.TABLE]: 'read', } as const satisfies Record diff --git a/packages/platform-authz/src/rooms.ts b/packages/platform-authz/src/rooms.ts index 6d76fdceb4f..cf5ded87b83 100644 --- a/packages/platform-authz/src/rooms.ts +++ b/packages/platform-authz/src/rooms.ts @@ -87,6 +87,8 @@ const ROOM_WORKSPACE_RESOLVERS: Partial> [ROOM_TYPES.WORKSPACE_FILES]: resolveWorkspaceRoomWorkspace, // A workspace-tables room is addressed directly by its workspace id. [ROOM_TYPES.WORKSPACE_TABLES]: resolveWorkspaceRoomWorkspace, + // A workspace-workflows room is addressed directly by its workspace id. + [ROOM_TYPES.WORKSPACE_WORKFLOWS]: resolveWorkspaceRoomWorkspace, // A file-doc room is addressed by file id; resolve it to its workspace. [ROOM_TYPES.WORKSPACE_FILE_DOC]: resolveFileDocWorkspace, // A table room is addressed by table id; resolve it to its workspace. @@ -103,9 +105,9 @@ export interface RoomAuthorizationResult { /** * Authorizes a user against a workspace-scoped realtime room (workspace-files, - * file-doc, table). Mirrors `authorizeWorkflowByWorkspacePermission` (the - * exemplary workflow authorizer) but generalized over room type: resolve the - * room's workspace, then gate on the user's effective workspace permission under + * workspace-tables, workspace-workflows, file-doc, table). Mirrors + * `authorizeWorkflowByWorkspacePermission` (the exemplary workflow authorizer) but generalized + * over room type: resolve the room's workspace, then gate on the user's effective permission under * the read < write < admin ordering. Workflow rooms use their own authorizer and * do not pass through here (see {@link ROOM_WORKSPACE_RESOLVERS}). * diff --git a/packages/realtime-protocol/src/rooms.test.ts b/packages/realtime-protocol/src/rooms.test.ts index bb3651e9978..c0e35f7bc1d 100644 --- a/packages/realtime-protocol/src/rooms.test.ts +++ b/packages/realtime-protocol/src/rooms.test.ts @@ -51,6 +51,7 @@ describe('parseRoomName', () => { { type: ROOM_TYPES.WORKSPACE_FILES, id: 'ws-456' }, { type: ROOM_TYPES.WORKSPACE_FILE_DOC, id: 'file-789' }, { type: ROOM_TYPES.TABLE, id: 'table-abc' }, + { type: ROOM_TYPES.WORKSPACE_WORKFLOWS, id: 'ws-workflows' }, ] for (const ref of refs) { expect(parseRoomName(roomName(ref))).toEqual(ref) diff --git a/packages/realtime-protocol/src/rooms.ts b/packages/realtime-protocol/src/rooms.ts index afc3edde5a8..1ebb75889c7 100644 --- a/packages/realtime-protocol/src/rooms.ts +++ b/packages/realtime-protocol/src/rooms.ts @@ -43,6 +43,11 @@ export const ROOM_TYPES = { * space is the workspace id, mirroring {@link ROOM_TYPES.WORKSPACE_FILES}. */ WORKSPACE_TABLES: 'workspace-tables', + /** + * The workspace workflow sidebar (one room per workspace). Carries no presence, + * only a lossy invalidation signal for workflow and workflow-folder lists. + */ + WORKSPACE_WORKFLOWS: 'workspace-workflows', } as const export type RoomType = (typeof ROOM_TYPES)[keyof typeof ROOM_TYPES]