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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions apps/realtime/src/access-revalidation.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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<RoomType> = new Set<RoomType>([
Expand Down
1 change: 1 addition & 0 deletions apps/realtime/src/handlers/index.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
12 changes: 9 additions & 3 deletions apps/realtime/src/handlers/workspace-invalidation-room.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -73,9 +73,15 @@ function createRoomManager(overrides?: Partial<IRoomManager>): 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}`
Expand Down
28 changes: 28 additions & 0 deletions apps/realtime/src/routes/http.test.ts
Original file line number Diff line number Diff line change
@@ -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'

Expand All @@ -11,6 +12,7 @@ function createMocks(req: Partial<IncomingMessage>) {
const roomManager = {
getTotalActiveConnections: vi.fn().mockResolvedValue(0),
isReady: vi.fn().mockReturnValue(true),
emitToRoom: vi.fn(),
} as unknown as IRoomManager

return {
Expand All @@ -20,6 +22,7 @@ function createMocks(req: Partial<IncomingMessage>) {
setHeader,
writeHead,
end,
roomManager,
}
}

Expand Down Expand Up @@ -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' })
})
})
20 changes: 20 additions & 0 deletions apps/realtime/src/routes/http.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
7 changes: 7 additions & 0 deletions apps/sim/app/api/folders/[id]/duplicate/route.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@ const {
mockDuplicateWorkflow,
mockAcquireFolderMutationLock,
mockWithFolderTreeLock,
mockNotifyWorkspace,
} = vi.hoisted(() => ({
mockLogger: {
info: vi.fn(),
Expand All @@ -44,6 +45,7 @@ const {
mockDuplicateWorkflow: vi.fn(),
mockAcquireFolderMutationLock: vi.fn(),
mockWithFolderTreeLock: vi.fn(),
mockNotifyWorkspace: vi.fn(),
}))

vi.mock('@sim/audit', () => auditMock)
Expand All @@ -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'

Expand Down Expand Up @@ -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 () => {
Expand Down
2 changes: 2 additions & 0 deletions apps/sim/app/api/folders/[id]/duplicate/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'

Expand Down Expand Up @@ -208,6 +209,7 @@ export const POST = withRouteHandler(

return { newFolderId, folderMapping, workflowStats }
})
await notifyWorkspaceWorkflowsChanged(targetWorkspaceId)

const elapsed = Date.now() - startTime
logger.info(
Expand Down
5 changes: 4 additions & 1 deletion apps/sim/app/api/folders/reorder/route.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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(),
Expand All @@ -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'
Expand Down Expand Up @@ -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 () => {
Expand Down
5 changes: 4 additions & 1 deletion apps/sim/app/api/folders/reorder/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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')
Expand All @@ -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)
Expand Down Expand Up @@ -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 })
Expand Down
3 changes: 3 additions & 0 deletions apps/sim/app/api/superuser/import-workflow/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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({
Expand Down
3 changes: 3 additions & 0 deletions apps/sim/app/api/v1/admin/workflows/import/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand Down Expand Up @@ -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
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@ const {
mockSaveWorkflowToNormalizedTables,
mockDeduplicateWorkflowName,
mockNormalizeImportedVariables,
mockNotifyWorkspace,
} = vi.hoisted(() => ({
mockLogger: {
info: vi.fn(),
Expand All @@ -37,6 +38,7 @@ const {
mockSaveWorkflowToNormalizedTables: vi.fn(),
mockDeduplicateWorkflowName: vi.fn(),
mockNormalizeImportedVariables: vi.fn(),
mockNotifyWorkspace: vi.fn(),
}))

vi.mock('@sim/logger', () => ({
Expand Down Expand Up @@ -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'

Expand Down Expand Up @@ -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)
})

/**
Expand Down
24 changes: 19 additions & 5 deletions apps/sim/app/api/v1/admin/workspaces/[id]/import/route.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -118,7 +119,8 @@ async function ensureImportFolder(
workspaceId: string,
userId: string,
name: string,
parentId: string | null
parentId: string | null,
onCreated?: () => void
): Promise<string> {
const existing = await findImportFolder(db, workspaceId, name, parentId)
if (existing) return existing
Expand Down Expand Up @@ -148,6 +150,7 @@ async function ensureImportFolder(
createdAt: new Date(),
updatedAt: new Date(),
})
onCreated?.()
return folderId
})
} catch (error) {
Expand All @@ -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)
Expand Down Expand Up @@ -245,7 +252,8 @@ export const POST = withRouteHandler(
workspaceId,
workspaceData.ownerId,
rootFolderName,
null
null,
markWorkspaceTreeChanged
)
}

Expand All @@ -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}`)
Expand All @@ -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
Expand All @@ -302,7 +314,8 @@ async function importSingleWorkflow(
ownerId: string,
createFolders: boolean,
rootFolderId: string | undefined,
folderMap: Map<string, string>
folderMap: Map<string, string>,
onFolderCreated: () => void
): Promise<ImportResult> {
try {
const { data: workflowData, errors } = parseWorkflowJson(wf.content)
Expand Down Expand Up @@ -331,7 +344,8 @@ async function importSingleWorkflow(
workspaceId,
ownerId,
wf.folderPath[i],
parentId
parentId,
onFolderCreated
)
folderMap.set(fullPath, folderId)
parentId = folderId
Expand Down
Loading
Loading