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 index 20de27440d5..868a41cc5b4 100644 --- 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 @@ -29,7 +29,7 @@ const roots: Root[] = [] function mount( workspaceId: string, roomType: RoomType, - onChanged: () => unknown = () => {}, + onChanged: () => Promise = async () => {}, dedupeKey?: string ): Root { function Probe() { @@ -119,7 +119,7 @@ describe('useWorkspaceInvalidationRoom', () => { }) it('delivers a burst of broadcasts once at once and once after it, not once per broadcast', async () => { - const onChanged = vi.fn() + const onChanged = vi.fn(async () => {}) mount('ws-1', 'workspace-files', onChanged) for (let index = 0; index < 20; index++) broadcast('workspace-files-changed', 'ws-1') 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 47e00885723..e76be4113d8 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 @@ -36,10 +36,11 @@ interface JoinErrorPayload { } /** - * Runs when the room's resources changed. A returned promise is awaited, up to - * {@link DELIVERY_SETTLE_CAP_MS}, before the next delivery starts. + * Runs when the room's resources changed and returns the invalidations it started. The promise is + * awaited, up to {@link DELIVERY_SETTLE_CAP_MS}, before the next delivery starts, so a callback + * that drops it lets the next broadcast cancel and restart refetches still in flight. */ -type RoomChangedCallback = () => unknown +type RoomChangedCallback = () => Promise interface SharedRoomSubscription { callbacks: Map> diff --git a/apps/sim/app/workspace/[workspaceId]/tables/hooks/use-workspace-tables-room.test.tsx b/apps/sim/app/workspace/[workspaceId]/tables/hooks/use-workspace-tables-room.test.tsx new file mode 100644 index 00000000000..8aed18094b7 --- /dev/null +++ b/apps/sim/app/workspace/[workspaceId]/tables/hooks/use-workspace-tables-room.test.tsx @@ -0,0 +1,89 @@ +/** + * @vitest-environment jsdom + */ + +import { act } from 'react' +import { QueryClient, QueryClientProvider, useQuery } from '@tanstack/react-query' +import { createRoot, type Root } from 'react-dom/client' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockUseSocket } = vi.hoisted(() => ({ mockUseSocket: vi.fn() })) + +vi.mock('@/app/workspace/providers/socket-provider', () => ({ + useSocket: () => mockUseSocket(), +})) + +import { INVALIDATION_COALESCE_MS } from '@/app/workspace/[workspaceId]/hooks/use-workspace-invalidation-room' +import { useWorkspaceTablesRoom } from '@/app/workspace/[workspaceId]/tables/hooks/use-workspace-tables-room' +import { tableKeys } from '@/hooks/queries/utils/table-keys' + +function fakeSocket() { + return { connected: true, emit: vi.fn(), on: vi.fn(), off: vi.fn() } +} + +let socket: ReturnType +let root: Root | null = null + +function broadcast(workspaceId: string) { + for (const [name, handler] of socket.on.mock.calls) { + if (name === 'workspace-tables-changed') handler({ workspaceId }) + } +} + +describe('useWorkspaceTablesRoom', () => { + beforeEach(() => { + vi.useFakeTimers() + socket = fakeSocket() + mockUseSocket.mockImplementation(() => ({ socket })) + }) + + afterEach(() => { + if (root) act(() => root?.unmount()) + root = null + vi.useRealTimers() + }) + + it('does not restart a list refetch still in flight when the next broadcast arrives', async () => { + const pending: Array<() => void> = [] + const fetchList = vi.fn( + () => + new Promise((resolve) => { + pending.push(() => resolve([])) + }) + ) + const queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } }) + queryClient.setQueryData(tableKeys.list('ws-1'), []) + + function Probe() { + useQuery({ + queryKey: tableKeys.list('ws-1'), + queryFn: fetchList, + staleTime: Number.POSITIVE_INFINITY, + }) + useWorkspaceTablesRoom('ws-1') + return null + } + const container = document.createElement('div') + document.body.appendChild(container) + root = createRoot(container) + act(() => + root?.render( + + + + ) + ) + + broadcast('ws-1') + await vi.advanceTimersByTimeAsync(0) + expect(fetchList).toHaveBeenCalledTimes(1) + + broadcast('ws-1') + await vi.advanceTimersByTimeAsync(5 * INVALIDATION_COALESCE_MS) + expect(fetchList).toHaveBeenCalledTimes(1) + + for (const resolve of pending.splice(0)) resolve() + await vi.advanceTimersByTimeAsync(INVALIDATION_COALESCE_MS) + expect(fetchList).toHaveBeenCalledTimes(2) + }) +}) diff --git a/apps/sim/app/workspace/[workspaceId]/tables/hooks/use-workspace-tables-room.ts b/apps/sim/app/workspace/[workspaceId]/tables/hooks/use-workspace-tables-room.ts index 3387ede2b5f..6af4e60993f 100644 --- a/apps/sim/app/workspace/[workspaceId]/tables/hooks/use-workspace-tables-room.ts +++ b/apps/sim/app/workspace/[workspaceId]/tables/hooks/use-workspace-tables-room.ts @@ -16,8 +16,10 @@ import { tableKeys } from '@/hooks/queries/utils/table-keys' */ export function useWorkspaceTablesRoom(workspaceId: string): void { const queryClient = useQueryClient() - useWorkspaceInvalidationRoom(workspaceId, ROOM_TYPES.WORKSPACE_TABLES, () => { - queryClient.invalidateQueries({ queryKey: tableKeys.lists() }) - queryClient.invalidateQueries({ queryKey: folderKeys.resource('table') }) - }) + useWorkspaceInvalidationRoom(workspaceId, ROOM_TYPES.WORKSPACE_TABLES, () => + Promise.all([ + queryClient.invalidateQueries({ queryKey: tableKeys.lists() }), + queryClient.invalidateQueries({ queryKey: folderKeys.resource('table') }), + ]) + ) } diff --git a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/hooks/use-resource-folders.ts b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/hooks/use-resource-folders.ts index d3dde26b7ba..72560e2bb58 100644 --- a/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/hooks/use-resource-folders.ts +++ b/apps/sim/app/workspace/[workspaceId]/w/[workflowId]/components/panel/components/editor/components/sub-block/hooks/use-resource-folders.ts @@ -77,7 +77,8 @@ export function useResourceFolders( useWorkspaceInvalidationRoom( isFileResource ? (workspaceId ?? '') : '', ROOM_TYPES.WORKSPACE_FILES, - () => (workspaceId ? invalidateWorkspaceFileBrowsers(queryClient, workspaceId) : undefined), + () => + workspaceId ? invalidateWorkspaceFileBrowsers(queryClient, workspaceId) : Promise.resolve(), WORKSPACE_FILE_BROWSER_INVALIDATION_KEY ) 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 index 60c679120bf..8fa07ebc54a 100644 --- 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 @@ -21,9 +21,11 @@ import { invalidateWorkflowLists } from '@/hooks/queries/utils/invalidate-workfl */ export function useWorkspaceWorkflowsRoom(workspaceId: string): void { const queryClient = useQueryClient() - useWorkspaceInvalidationRoom(workspaceId, ROOM_TYPES.WORKSPACE_WORKFLOWS, () => { - invalidateWorkflowLists(queryClient, workspaceId, ['active', 'archived']) - queryClient.invalidateQueries({ queryKey: folderKeys.list(workspaceId, 'active') }) - queryClient.invalidateQueries({ queryKey: folderKeys.list(workspaceId, 'archived') }) - }) + useWorkspaceInvalidationRoom(workspaceId, ROOM_TYPES.WORKSPACE_WORKFLOWS, () => + Promise.all([ + invalidateWorkflowLists(queryClient, workspaceId, ['active', 'archived']), + queryClient.invalidateQueries({ queryKey: folderKeys.list(workspaceId, 'active') }), + queryClient.invalidateQueries({ queryKey: folderKeys.list(workspaceId, 'archived') }), + ]) + ) } diff --git a/apps/sim/lib/auth/admin-removal-chat-cleanup.ts b/apps/sim/lib/auth/admin-removal-chat-cleanup.ts index 21a0f1b91b9..fea143a940e 100644 --- a/apps/sim/lib/auth/admin-removal-chat-cleanup.ts +++ b/apps/sim/lib/auth/admin-removal-chat-cleanup.ts @@ -14,14 +14,9 @@ import { const ADMIN_REMOVAL_CLEANUP_DELAY_MS = 60 * 60_000 /** - * The user delete's `before` hook: `/admin/remove-user` cascades the account's chats, and those in - * the workspaces it owns, away with no transaction to queue their cleanup in, so the cleanup is - * queued now, before the delete, and delayed past it. The handler purges only chats whose rows - * are gone, so a delete that fails or is refused purges nothing. A failed read or enqueue throws, - * which aborts the removal rather than erasing chats whose cleanup nothing recorded. Better Auth's - * delete runs outside any transaction this hook can join, so a chat or attachment written between - * this read and that delete is not covered; account deletion, which owns its transaction, re-reads - * under the account lock instead. + * The user delete's `before` hook: queues, delayed past the delete, the cleanup of the chats + * `/admin/remove-user` cascades away; a failed read or enqueue throws and aborts the removal. A + * chat or attachment written between this read and Better Auth's delete is not covered. */ export async function enqueueAdminRemovedChatCleanup(user: { id: string }): Promise { const targets = await readAccountRemovalChats({ diff --git a/apps/sim/lib/cleanup/chat-backend-cleanup-event.ts b/apps/sim/lib/cleanup/chat-backend-cleanup-event.ts index d538c2de1fa..168ad9ebcc3 100644 --- a/apps/sim/lib/cleanup/chat-backend-cleanup-event.ts +++ b/apps/sim/lib/cleanup/chat-backend-cleanup-event.ts @@ -1 +1,2 @@ export const CHAT_BACKEND_CLEANUP_EVENT = 'copilot.chat.backend-cleanup' +export const CHAT_INLINE_IMAGE_CLEANUP_EVENT = 'copilot.chat.inline-image-cleanup' diff --git a/apps/sim/lib/cleanup/chat-backend-cleanup-outbox.test.ts b/apps/sim/lib/cleanup/chat-backend-cleanup-outbox.test.ts new file mode 100644 index 00000000000..878c36fd6f7 --- /dev/null +++ b/apps/sim/lib/cleanup/chat-backend-cleanup-outbox.test.ts @@ -0,0 +1,124 @@ +import { copilotChats } from '@sim/db/schema' +import { queueTableRows, resetDbChainMock } from '@sim/testing/mocks/database.mock' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockSweepStoragePrefix, deadline } = vi.hoisted(() => ({ + mockSweepStoragePrefix: vi.fn(), + deadline: { near: false }, +})) + +vi.mock('@/lib/uploads/core/prefix-sweep', () => ({ + sweepStoragePrefix: mockSweepStoragePrefix, + outboxDeadlineNear: () => deadline.near, +})) + +import { CHAT_INLINE_IMAGE_CLEANUP_EVENT } from '@/lib/cleanup/chat-backend-cleanup-event' +import { chatBackendCleanupOutboxHandlers } from '@/lib/cleanup/chat-backend-cleanup-outbox' +import type { OutboxEventContext } from '@/lib/core/outbox/service' + +const sweepImages = chatBackendCleanupOutboxHandlers[CHAT_INLINE_IMAGE_CLEANUP_EVENT] +const chatIds = Array.from({ length: 25 }, (_, index) => `chat-${String(index).padStart(2, '0')}`) +const prefixOf = (chatId: string) => `chat-images/${chatId}/` + +function outboxContext() { + return { + eventId: 'event-1', + eventType: CHAT_INLINE_IMAGE_CLEANUP_EVENT, + attempts: 0, + maxAttempts: 10, + signal: new AbortController().signal, + checkpointPayload: vi.fn(async (_patch: Record) => {}), + } satisfies OutboxEventContext +} + +const sweptPrefixes = () => mockSweepStoragePrefix.mock.calls.map(([prefix]) => prefix) + +describe('chat inline image cleanup', () => { + beforeEach(() => { + resetDbChainMock() + deadline.near = false + mockSweepStoragePrefix.mockReset().mockResolvedValue('done') + }) + + it('sweeps several chats at once, never more than ten', async () => { + queueTableRows(copilotChats, []) + let inFlight = 0 + let peak = 0 + mockSweepStoragePrefix.mockImplementation(async () => { + peak = Math.max(peak, ++inFlight) + await new Promise((resolve) => setImmediate(resolve)) + inFlight-- + return 'done' + }) + + await sweepImages({ chatIds }, outboxContext()) + + expect(peak).toBe(10) + expect(new Set(sweptPrefixes())).toEqual(new Set(chatIds.map(prefixOf))) + }) + + it('checkpoints the chats finished before the lowest unfinished one and resumes after them', async () => { + queueTableRows(copilotChats, []) + const slow = chatIds[3] + mockSweepStoragePrefix.mockImplementation(async (prefix: string) => { + if (prefix !== prefixOf(slow)) return 'done' + // Finishes after the chats behind it, so only "lowest unfinished" can yield 3 here. + await new Promise((resolve) => setImmediate(resolve)) + deadline.near = true + return 'continue' + }) + const context = outboxContext() + + const result = await sweepImages({ chatIds: chatIds.slice(0, 6) }, context) + + expect(result).toMatchObject({ outcome: 'deferred', consumeAttempt: false }) + expect(context.checkpointPayload).toHaveBeenCalledWith({ swept: 3 }) + + queueTableRows(copilotChats, []) + deadline.near = false + mockSweepStoragePrefix.mockReset().mockResolvedValue('done') + await expect( + sweepImages({ chatIds: chatIds.slice(0, 6), swept: 3 }, outboxContext()) + ).resolves.toBe(undefined) + expect(sweptPrefixes().sort()).toEqual(chatIds.slice(3, 6).map(prefixOf)) + }) + + it('skips a chat id no image key can hold instead of failing the event on every retry', async () => { + queueTableRows(copilotChats, []) + + await expect( + sweepImages({ chatIds: ['../escape', chatIds[0]] }, outboxContext()) + ).resolves.toBe(undefined) + expect(sweptPrefixes()).toEqual([prefixOf(chatIds[0])]) + }) + + it('spends an attempt when a chat’s listing stops advancing', async () => { + queueTableRows(copilotChats, []) + mockSweepStoragePrefix.mockImplementation(async (prefix: string) => + prefix === prefixOf(chatIds[0]) ? 'stalled' : 'done' + ) + const context = outboxContext() + + const result = await sweepImages({ chatIds: chatIds.slice(0, 2) }, context) + + expect(result).toMatchObject({ outcome: 'deferred' }) + expect(result).not.toMatchObject({ consumeAttempt: false }) + expect(context.checkpointPayload).not.toHaveBeenCalled() + }) + + it('spends an attempt when the lowest unfinished chat stalls, though a later one continues', async () => { + queueTableRows(copilotChats, []) + mockSweepStoragePrefix.mockImplementation(async (prefix: string) => { + if (prefix === prefixOf(chatIds[0])) return 'stalled' + deadline.near = true + return 'continue' + }) + const context = outboxContext() + + const result = await sweepImages({ chatIds: chatIds.slice(0, 2) }, context) + + expect(result).toMatchObject({ outcome: 'deferred' }) + expect(result).not.toMatchObject({ consumeAttempt: false }) + expect(context.checkpointPayload).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/cleanup/chat-backend-cleanup-outbox.ts b/apps/sim/lib/cleanup/chat-backend-cleanup-outbox.ts index 0b2dfd6d086..6b09559e97d 100644 --- a/apps/sim/lib/cleanup/chat-backend-cleanup-outbox.ts +++ b/apps/sim/lib/cleanup/chat-backend-cleanup-outbox.ts @@ -1,37 +1,81 @@ import { db } from '@sim/db' import { copilotChats } from '@sim/db/schema' +import { createLogger } from '@sim/logger' import { describeError, toError } from '@sim/utils/errors' import { inArray } from 'drizzle-orm' -import { chatBackendCleanupPayloadSchema } from '@/lib/cleanup/chat-backend-cleanup' -import { CHAT_BACKEND_CLEANUP_EVENT } from '@/lib/cleanup/chat-backend-cleanup-event' import { - cleanupCopilotBackend, - organizationAttachmentsStillReferenced, -} from '@/lib/cleanup/chat-cleanup' + chatBackendCleanupPayloadSchema, + chatInlineImageCleanupPayloadSchema, +} from '@/lib/cleanup/chat-backend-cleanup' +import { + CHAT_BACKEND_CLEANUP_EVENT, + CHAT_INLINE_IMAGE_CLEANUP_EVENT, +} from '@/lib/cleanup/chat-backend-cleanup-event' +import { cleanupCopilotBackend, deletableChatFiles } from '@/lib/cleanup/chat-cleanup' import { env } from '@/lib/core/config/env' -import type { OutboxHandlerRegistry } from '@/lib/core/outbox/service' +import { + continueOutboxHandler, + deferOutboxHandler, + type OutboxEventContext, + type OutboxHandlerRegistry, +} from '@/lib/core/outbox/service' import { mapWithConcurrency } from '@/lib/core/utils/concurrency' +import { inlineChatImagePrefix } from '@/lib/mothership/chat/inline-image-key' +import { outboxDeadlineNear, sweepStoragePrefix } from '@/lib/uploads/core/prefix-sweep' import { deleteFile } from '@/lib/uploads/core/storage-service' -const FILE_DELETE_CONCURRENCY = 10 +const logger = createLogger('ChatBackendCleanupOutbox') + +/** Most object deletes, or chat prefix sweeps, one handler run has in flight at once. */ +const STORAGE_CONCURRENCY = 10 + +/** The ids among `chatIds` that still have a `copilot_chats` row. */ +async function survivingChatIds(chatIds: readonly string[]): Promise> { + if (chatIds.length === 0) return new Set() + const rows = await db + .select({ id: copilotChats.id }) + .from(copilotChats) + .where(inArray(copilotChats.id, [...chatIds])) + return new Set(rows.map(({ id }) => id)) +} + +/** + * Sweeps `chatId`'s inline image prefix until it is empty, the deadline nears (`continue`), or a + * listing stops advancing (`stalled`). + */ +async function sweepChatImages( + chatId: string, + context: OutboxEventContext +): Promise<'done' | 'continue' | 'stalled'> { + const prefix = inlineChatImagePrefix(chatId) + if (!prefix) { + logger.warn('Skipping an inline image sweep for a chat id no image key can hold', { + eventId: context.eventId, + chatId, + }) + return 'done' + } + for (;;) { + context.signal.throwIfAborted() + if (outboxDeadlineNear(context)) return 'continue' + const outcome = await sweepStoragePrefix(prefix, 'mothership', context) + if (outcome !== 'continue') return outcome + } +} export const chatBackendCleanupOutboxHandlers = { /** - * Purges the copilot backend data and stored files of chats whose rows are gone. A chat whose - * row survived (its delete was spared, rolled back, or never happened) keeps both. Any failure - * throws so the outbox retries it: a failed backend call, a failed object delete, or chat ids - * queued while `COPILOT_API_KEY` was set reaching a processor without it. An object already - * gone counts as deleted, so a retry repeats only what is left. + * Purges the copilot backend data and stored files of chats whose rows are gone; a surviving + * chat keeps both, and a file another chat of its scope still references is kept + * ({@link deletableChatFiles}). Any failure throws so the outbox retries; an object already gone + * counts as deleted. */ [CHAT_BACKEND_CLEANUP_EVENT]: async (rawPayload, context) => { const { chatIds, files } = chatBackendCleanupPayloadSchema.parse(rawPayload) context.signal.throwIfAborted() - const named = [...new Set([...chatIds, ...files.map((file) => file.chatId)])] - const surviving = await db - .select({ id: copilotChats.id }) - .from(copilotChats) - .where(inArray(copilotChats.id, named)) - const survivingIds = new Set(surviving.map(({ id }) => id)) + const survivingIds = await survivingChatIds([ + ...new Set([...chatIds, ...files.map((file) => file.chatId)]), + ]) const deletedChatIds = chatIds.filter((id) => !survivingIds.has(id)) const deletedFiles = files.filter((file) => !survivingIds.has(file.chatId)) @@ -46,10 +90,9 @@ export const chatBackendCleanupOutboxHandlers = { } if (deletedFiles.length === 0) return - const stillReferenced = await organizationAttachmentsStillReferenced(deletedFiles) const failures = await mapWithConcurrency( - deletedFiles.filter((file) => !stillReferenced.has(file.key)), - FILE_DELETE_CONCURRENCY, + await deletableChatFiles(deletedFiles), + STORAGE_CONCURRENCY, async ({ key, context: storageContext }) => { context.signal.throwIfAborted() try { @@ -62,4 +105,28 @@ export const chatBackendCleanupOutboxHandlers = { const failure = failures.find((error) => error !== undefined) if (failure) throw failure }, + + /** + * Deletes everything under the `chat-images//` prefix of each chat whose row is gone, so + * an image a turn published after the delete read the transcript goes too. Sweeps several chats + * at once, checkpoints how many leading chats are finished, and continues in a later run once + * the deadline nears. + */ + [CHAT_INLINE_IMAGE_CLEANUP_EVENT]: async (rawPayload, context) => { + const { chatIds, swept } = chatInlineImageCleanupPayloadSchema.parse(rawPayload) + context.signal.throwIfAborted() + const pending = chatIds.slice(swept) + const survivingIds = await survivingChatIds(pending) + const outcomes = await mapWithConcurrency(pending, STORAGE_CONCURRENCY, (chatId) => + survivingIds.has(chatId) ? Promise.resolve('done' as const) : sweepChatImages(chatId, context) + ) + const unfinished = outcomes.findIndex((outcome) => outcome !== 'done') + if (unfinished === -1) return + if (unfinished > 0) await context.checkpointPayload({ swept: swept + unfinished }) + // A free rerun only after durable progress, or while the lowest unfinished chat itself advances; + // a stalled one behind a continuing chat would otherwise rerun without end. + if (unfinished > 0 || outcomes[unfinished] === 'continue') + return continueOutboxHandler('Continuing chat inline image cleanup') + return deferOutboxHandler('Chat inline image cleanup removed nothing from its listing') + }, } satisfies OutboxHandlerRegistry diff --git a/apps/sim/lib/cleanup/chat-backend-cleanup.integration.ts b/apps/sim/lib/cleanup/chat-backend-cleanup.integration.ts index afa2668a52f..d5a59b3b7af 100644 --- a/apps/sim/lib/cleanup/chat-backend-cleanup.integration.ts +++ b/apps/sim/lib/cleanup/chat-backend-cleanup.integration.ts @@ -76,11 +76,16 @@ const { backend, storageRoot } = await vi.hoisted(async () => { vi.mock('@/lib/uploads/core/setup.server', () => uploadsSetupMock) import { enqueueAdminRemovedChatCleanup } from '@/lib/auth/admin-removal-chat-cleanup' -import { CHAT_BACKEND_CLEANUP_EVENT } from '@/lib/cleanup/chat-backend-cleanup-event' +import { + CHAT_BACKEND_CLEANUP_EVENT, + CHAT_INLINE_IMAGE_CLEANUP_EVENT, +} from '@/lib/cleanup/chat-backend-cleanup-event' import { chatBackendCleanupOutboxHandlers } from '@/lib/cleanup/chat-backend-cleanup-outbox' +import { prepareChatCleanup } from '@/lib/cleanup/chat-cleanup' import { env } from '@/lib/core/config/env' import { processOutboxEventById } from '@/lib/core/outbox/service' import { deleteCopilotChat } from '@/lib/mothership/chat/deletion' +import { inlineChatImageKey } from '@/lib/mothership/chat/inline-image-key' import { deleteUserAccount } from '@/lib/users/account-deletion' setUploadDirServer(storageRoot) @@ -202,7 +207,7 @@ async function seedChatWithFiles(userId: string, workspaceId: string) { const queuedBySuite = () => and( - eq(outboxEvent.eventType, CHAT_BACKEND_CLEANUP_EVENT), + inArray(outboxEvent.eventType, [CHAT_BACKEND_CLEANUP_EVENT, CHAT_INLINE_IMAGE_CLEANUP_EVENT]), sql`${outboxEvent.createdAt} >= ${suiteStartedAt}::timestamp` ) @@ -303,7 +308,7 @@ describe('cleanup of a permanently deleted chat', () => { expect(await existingChats([chatId])).toEqual(new Set()) const delivered = await deliverCleanups() - expect(delivered.outcomes).toEqual(['completed', 'completed']) + expect(delivered.outcomes).toEqual(['completed', 'completed', 'completed']) expect(delivered.purged).toEqual(new Set([chatId])) expect(await objectExists(keys.upload)).toBe(false) expect(await objectExists(keys.attachment)).toBe(false) @@ -312,6 +317,237 @@ describe('cleanup of a permanently deleted chat', () => { expect(backend.rejected).toBe(0) }) + /** + * A fork whose copy of an upload failed keeps naming the source's key in its messages, while + * the upload's `workspace_files` row cascades away with the source chat. + */ + it('keeps a deleted chat’s upload that a fork in its workspace still references', async () => { + const ownerId = await seedUser() + const workspaceId = await seedWorkspace(ownerId) + const uploadIn = async (chatId: string) => { + const key = `workspace/${workspaceId}/1700000000000-${generateId().slice(0, 8)}-shot.png` + await db.insert(workspaceFiles).values({ + id: generateId(), + key, + userId: ownerId, + workspaceId, + context: 'mothership', + chatId, + originalName: 'shot.png', + contentType: 'image/png', + sizeBytes: 1, + }) + await putObject(key) + return key + } + const forkedSource = await seedChat(ownerId, workspaceId) + const sharedKey = await uploadIn(forkedSource) + const fork = await seedChat(ownerId, workspaceId) + await db.insert(copilotMessages).values({ + chatId: fork, + messageId: generateId(), + role: 'user', + content: { + role: 'user', + content: 'see attached', + fileAttachments: [{ key: sharedKey, filename: 'shot.png' }], + }, + }) + const unforkedSource = await seedChat(ownerId, workspaceId) + const ownKey = await uploadIn(unforkedSource) + + await deleteCopilotChat({ chatId: forkedSource, userId: ownerId }) + await deleteCopilotChat({ chatId: unforkedSource, userId: ownerId }) + + expect(new Set((await deliverCleanups()).outcomes)).toEqual(new Set(['completed'])) + expect(await objectExists(sharedKey)).toBe(true) + expect(await objectExists(ownKey)).toBe(false) + }) + + it('deletes a forked upload once the last chat attaching it is deleted', async () => { + const ownerId = await seedUser() + const workspaceId = await seedWorkspace(ownerId) + const source = await seedChat(ownerId, workspaceId) + const key = `workspace/${workspaceId}/1700000000000-${generateId().slice(0, 8)}-shot.png` + await db.insert(workspaceFiles).values({ + id: generateId(), + key, + userId: ownerId, + workspaceId, + context: 'mothership', + chatId: source, + originalName: 'shot.png', + contentType: 'image/png', + sizeBytes: 1, + }) + await putObject(key) + const fork = await seedChat(ownerId, workspaceId) + await db.insert(copilotMessages).values({ + chatId: fork, + messageId: generateId(), + role: 'user', + content: { role: 'user', content: 'see attached', fileAttachments: [{ key }] }, + }) + + await deleteCopilotChat({ chatId: source, userId: ownerId }) + await deliverCleanups() + expect(await objectExists(key)).toBe(true) + + await deleteCopilotChat({ chatId: fork, userId: ownerId }) + await deliverCleanups() + expect(await objectExists(key)).toBe(false) + }) + + it('never deletes an attached workspace upload whose row is live or only soft-deleted', async () => { + const ownerId = await seedUser() + const workspaceId = await seedWorkspace(ownerId) + const chatId = await seedChat(ownerId, workspaceId) + const upload = (name: string) => + `workspace/${workspaceId}/1700000000000-${generateId().slice(0, 8)}-${name}` + const live = upload('live.png') + const restorable = upload('restorable.png') + await db.insert(workspaceFiles).values( + [live, restorable].map((key) => ({ + id: generateId(), + key, + userId: ownerId, + workspaceId, + context: 'workspace', + originalName: 'file.png', + contentType: 'image/png', + sizeBytes: 1, + ...(key === restorable ? { deletedAt: new Date() } : {}), + })) + ) + for (const key of [live, restorable]) await putObject(key) + await db.insert(copilotMessages).values({ + chatId, + messageId: generateId(), + role: 'user', + content: { + role: 'user', + content: 'x', + fileAttachments: [{ key: live }, { key: restorable }], + }, + }) + + await deleteCopilotChat({ chatId, userId: ownerId }) + await deliverCleanups() + + expect(await objectExists(live)).toBe(true) + expect(await objectExists(restorable)).toBe(true) + }) + + it('places a legacy upload ref by its workspace key, keeping one a remaining chat attaches', async () => { + const ownerId = await seedUser() + const workspaceId = await seedWorkspace(ownerId) + const deletedChat = await seedChat(ownerId, workspaceId) + const remainingChat = await seedChat(ownerId, workspaceId) + const referenced = `workspace/${workspaceId}/1700000000000-abcd1234-kept.png` + const unreferenced = `workspace/${workspaceId}/1700000000000-abcd1234-gone.png` + for (const key of [referenced, unreferenced]) await putObject(key) + await db.insert(copilotMessages).values({ + chatId: remainingChat, + messageId: generateId(), + role: 'user', + content: { role: 'user', content: 'x', fileAttachments: [{ key: referenced }] }, + }) + await db.delete(copilotChats).where(eq(copilotChats.id, deletedChat)) + await db.insert(outboxEvent).values({ + id: generateId(), + eventType: CHAT_BACKEND_CLEANUP_EVENT, + payload: { + chatIds: [], + files: [referenced, unreferenced].map((key) => ({ + key, + context: 'mothership', + chatId: deletedChat, + })), + }, + }) + + expect((await deliverCleanups()).outcomes).toEqual(['completed']) + expect(await objectExists(referenced)).toBe(true) + expect(await objectExists(unreferenced)).toBe(false) + }) + + it('accepts the payloads queued before refs carried their scope', async () => { + const ownerId = await seedUser() + const workspaceId = await seedWorkspace(ownerId) + const deletedChat = await seedChat(ownerId, workspaceId) + const organizationAttachment = `copilot/${generateId()}/org-attachment.png` + await putObject(organizationAttachment) + await db.delete(copilotChats).where(eq(copilotChats.id, deletedChat)) + const organizationId = generateId() + await db.insert(outboxEvent).values([ + { + id: generateId(), + eventType: CHAT_BACKEND_CLEANUP_EVENT, + payload: { chatIds: [deletedChat], files: [] }, + }, + { + id: generateId(), + eventType: CHAT_BACKEND_CLEANUP_EVENT, + payload: { + chatIds: [], + files: [ + { + key: organizationAttachment, + context: 'copilot', + chatId: deletedChat, + organizationId, + }, + ], + }, + }, + ]) + + const delivered = await deliverCleanups() + expect(delivered.outcomes).toEqual(['completed', 'completed']) + expect(delivered.purged).toEqual(new Set([deletedChat])) + // Its legacy `organizationId` places it, and no chat of that organization still names it. + expect(await objectExists(organizationAttachment)).toBe(false) + }) + + it('sweeps the inline images of the chats whose rows are gone, never a surviving chat’s', async () => { + const ownerId = await seedUser() + const workspaceId = await seedWorkspace(ownerId) + const gone = await seedChat(ownerId, workspaceId) + const survivor = await seedChat(ownerId, workspaceId) + const goneImages = [ + inlineChatImageKey(gone, 'req-1', 'files/a.png'), + inlineChatImageKey(gone, 'req-2', 'files/b.png'), + ] + const survivorImage = inlineChatImageKey(survivor, 'req-1', 'files/a.png') + for (const key of [...goneImages, survivorImage]) await putObject(key) + await db.delete(copilotChats).where(eq(copilotChats.id, gone)) + await db.insert(outboxEvent).values({ + id: generateId(), + eventType: CHAT_INLINE_IMAGE_CLEANUP_EVENT, + payload: { chatIds: [gone, survivor] }, + }) + + expect((await deliverCleanups()).outcomes).toEqual(['completed']) + for (const key of goneImages) expect(await objectExists(key)).toBe(false) + expect(await objectExists(survivorImage)).toBe(true) + }) + + it('queues the inline image sweep of the chats a cleanup job purges', async () => { + const ownerId = await seedUser() + const workspaceId = await seedWorkspace(ownerId) + const chatId = await seedChat(ownerId, workspaceId) + const image = inlineChatImageKey(chatId, 'req-1', 'files/a.png') + await putObject(image) + + const cleanup = await prepareChatCleanup([chatId], 'test') + await db.delete(copilotChats).where(eq(copilotChats.id, chatId)) + await cleanup.execute() + + expect(backend.purged).toEqual([[chatId]]) + expect((await deliverCleanups()).outcomes).toEqual(['completed']) + expect(await objectExists(image)).toBe(false) + }) + it('records no cleanup when the delete rolls back', async () => { const ownerId = await seedUser() const workspaceId = await seedWorkspace(ownerId) @@ -337,14 +573,14 @@ describe('cleanup of a permanently deleted chat', () => { await rm(join(storageRoot, keys.upload)) await mkdir(join(storageRoot, keys.upload, 'blocker'), { recursive: true }) - expect((await deliverCleanups()).outcomes).toEqual(['pending', 'pending']) + expect((await deliverCleanups()).outcomes).toEqual(['completed', 'pending', 'pending']) expect(await objectExists(keys.attachment)).toBe(false) backend.unavailable = false await rm(join(storageRoot, keys.upload), { recursive: true }) await putObject(keys.upload) const retried = await deliverCleanups() - expect(retried.outcomes).toEqual(['completed', 'completed']) + expect(retried.outcomes).toEqual(['completed', 'completed', 'completed']) expect(retried.purged).toEqual(new Set([chatId])) expect(await objectExists(keys.upload)).toBe(false) }) @@ -356,7 +592,7 @@ describe('cleanup of a permanently deleted chat', () => { await deleteCopilotChat({ chatId, userId: ownerId }) Object.assign(env, { COPILOT_API_KEY: undefined }) - expect((await deliverCleanups()).outcomes).toEqual(['completed', 'pending']) + expect((await deliverCleanups()).outcomes).toEqual(['completed', 'completed', 'pending']) expect(backend.purged).toEqual([]) expect(await objectExists(keys.upload)).toBe(false) expect(await objectExists(keys.attachment)).toBe(false) @@ -368,7 +604,7 @@ describe('cleanup of a permanently deleted chat', () => { Object.assign(env, { COPILOT_API_KEY: backend.apiKey }) const retried = await deliverCleanups() - expect(retried.outcomes).toEqual(['completed', 'completed']) + expect(retried.outcomes).toEqual(['completed', 'completed', 'completed']) expect(retried.purged).toEqual(new Set([chatId])) }) }) @@ -434,6 +670,58 @@ describe('cleanup of the chats a removal cascades away', () => { expect(await objectExists(ownChat.keys.attachment)).toBe(false) }) + /** + * Inline images have no row, and the hook reads the transcript before Better Auth deletes the + * user, so an image a turn publishes in between was never collected by key. + */ + it('purges inline images a removed user’s chat published after the cleanup read it', async () => { + const removedId = await seedUser() + const bystanderId = await seedUser() + const workspaceId = await seedWorkspace(bystanderId, [removedId]) + const chatId = await seedChat(removedId, workspaceId) + const bystandersChat = await seedChat(bystanderId, workspaceId) + const read = inlineChatImageKey(chatId, 'req-1', 'files/chart.png') + await db.insert(copilotMessages).values({ + chatId, + messageId: generateId(), + role: 'assistant', + content: { role: 'assistant', requestId: 'req-1', content: '![chart](files/chart.png)' }, + }) + await putObject(read) + const bystanders = inlineChatImageKey(bystandersChat, 'req-1', 'files/chart.png') + await putObject(bystanders) + + await enqueueAdminRemovedChatCleanup({ id: removedId }) + const late = [ + inlineChatImageKey(chatId, 'req-2', 'files/later.png'), + inlineChatImageKey(chatId, 'req-2', 'files/another.png'), + ] + for (const key of late) await putObject(key) + await db.delete(user).where(eq(user.id, removedId)) + + const delivered = await deliverCleanups() + expect(new Set(delivered.outcomes)).toEqual(new Set(['completed'])) + for (const key of [read, ...late]) expect(await objectExists(key)).toBe(false) + expect(await objectExists(bystanders)).toBe(true) + }) + + it('still deletes the inline image keys an event queued before prefix sweeps names', async () => { + const ownerId = await seedUser() + const workspaceId = await seedWorkspace(ownerId) + const chatId = await seedChat(ownerId, workspaceId) + const key = inlineChatImageKey(chatId, 'req-1', 'files/chart.png') + await putObject(key) + await db.delete(copilotChats).where(eq(copilotChats.id, chatId)) + await db.insert(outboxEvent).values({ + id: generateId(), + eventType: CHAT_BACKEND_CLEANUP_EVENT, + payload: { chatIds: [], files: [{ key, context: 'mothership', chatId }] }, + }) + + expect((await deliverCleanups()).outcomes).toEqual(['completed']) + expect(await objectExists(key)).toBe(false) + }) + it('purges nothing for a direct user delete that never happened', async () => { const keptId = await seedUser() const workspaceId = await seedWorkspace(keptId) diff --git a/apps/sim/lib/cleanup/chat-backend-cleanup.ts b/apps/sim/lib/cleanup/chat-backend-cleanup.ts index 4b142ab9681..0ec7683de6f 100644 --- a/apps/sim/lib/cleanup/chat-backend-cleanup.ts +++ b/apps/sim/lib/cleanup/chat-backend-cleanup.ts @@ -3,7 +3,10 @@ import { copilotChats, workflow, workspace } from '@sim/db/schema' import { chunkArray } from '@sim/utils/helpers' import { eq, inArray } from 'drizzle-orm' import { z } from 'zod' -import { CHAT_BACKEND_CLEANUP_EVENT } from '@/lib/cleanup/chat-backend-cleanup-event' +import { + CHAT_BACKEND_CLEANUP_EVENT, + CHAT_INLINE_IMAGE_CLEANUP_EVENT, +} from '@/lib/cleanup/chat-backend-cleanup-event' import { type ChatFileRef, type ChatScopedContext, @@ -26,9 +29,25 @@ const chatFileSchema = z ...ChatScopedContext[], ]), chatId: chatIdSchema, + scope: z + .discriminatedUnion('kind', [ + z.object({ kind: z.literal('organization'), id: chatIdSchema }).strict(), + z.object({ kind: z.literal('workspace'), id: chatIdSchema }).strict(), + ]) + .nullable() + .optional(), + /** + * Legacy: an organization attachment's scope on events queued before refs carried `scope`. + * Remove once those events have drained. + */ organizationId: chatIdSchema.optional(), }) .strict() + .transform(({ organizationId, ...file }): ChatFileRef => { + if (file.scope === undefined && organizationId) + return { ...file, scope: { kind: 'organization', id: organizationId } } + return file + }) export const chatBackendCleanupPayloadSchema = z .object({ @@ -40,7 +59,17 @@ export const chatBackendCleanupPayloadSchema = z message: 'A chat cleanup names at least one chat or stored file', }) -type ChatBackendCleanupPayload = z.infer +type ChatBackendCleanupPayload = z.input + +export const chatInlineImageCleanupPayloadSchema = z + .object({ + chatIds: z.array(chatIdSchema).min(1).max(CHAT_BACKEND_CLEANUP_BATCH_SIZE), + /** A checkpoint: how many leading `chatIds` already had their prefix swept. */ + swept: z.number().int().min(0).max(CHAT_BACKEND_CLEANUP_BATCH_SIZE).default(0), + }) + .strict() + +type ChatInlineImageCleanupPayload = z.input /** What a chat deletion leaves outside Sim's rows: the chats' copilot backend data and stored files. */ export interface ChatCleanupTargets { @@ -49,43 +78,58 @@ export interface ChatCleanupTargets { files: readonly ChatFileRef[] } +/** + * Queues the sweep of each chat's `chat-images//` prefix, one event per + * {@link CHAT_BACKEND_CLEANUP_BATCH_SIZE} chats. Inline images have no row naming them, and a turn + * can publish one after any read of the transcript, so they go by prefix once the chat's row is + * gone; the handler skips a chat whose row survived. + */ +export async function enqueueChatInlineImageCleanup( + executor: Pick, + chatIds: readonly string[], + options: { availableAt?: Date } = {} +): Promise { + const payloads: ChatInlineImageCleanupPayload[] = chunkArray( + [...new Set(chatIds)], + CHAT_BACKEND_CLEANUP_BATCH_SIZE + ).map((batch) => ({ chatIds: batch })) + for (const events of chunkArray(payloads, MAX_BULK_ENQUEUE_EVENTS)) { + await enqueueOutboxEvents(executor, CHAT_INLINE_IMAGE_CLEANUP_EVENT, events, options) + } +} + /** * Queues the purge of what the given chats leave outside Sim's rows once a delete cascades them * away: one bounded event per {@link CHAT_BACKEND_CLEANUP_BATCH_SIZE} chats for the copilot backend, - * and one per as many stored files. Call it in the deleting transaction so the cleanup commits or - * rolls back with the delete; the handler purges only chats whose rows are gone, so a chat the - * delete ends up sparing keeps its data. A delete with no transaction to join queues it first, - * with an `availableAt` past the moment the delete commits. + * one per as many stored files, and the chats' inline image sweep + * ({@link enqueueChatInlineImageCleanup}). Call it in the deleting transaction so the cleanup + * commits or rolls back with the delete; the handlers purge only chats whose rows are gone, so a + * chat the delete ends up sparing keeps its data. A delete with no transaction to join queues it + * first, with an `availableAt` past the moment the delete commits. * * Without `COPILOT_API_KEY` this Sim never reached a copilot backend, so no chat ids are queued for - * it; the stored files are queued either way. + * it; the stored files and inline images are queued either way. */ export async function enqueueChatBackendCleanup( executor: Pick, targets: ChatCleanupTargets, options: { availableAt?: Date } = {} ): Promise { - const chatIds = env.COPILOT_API_KEY ? [...new Set(targets.chatIds)] : [] + const backendChatIds = env.COPILOT_API_KEY ? [...new Set(targets.chatIds)] : [] const filesByKey = new Map() for (const file of targets.files) if (!filesByKey.has(file.key)) filesByKey.set(file.key, file) const payloads: ChatBackendCleanupPayload[] = [ - ...chunkArray(chatIds, CHAT_BACKEND_CLEANUP_BATCH_SIZE).map((batch) => ({ + ...chunkArray(backendChatIds, CHAT_BACKEND_CLEANUP_BATCH_SIZE).map((batch) => ({ chatIds: batch, - files: [], })), ...chunkArray([...filesByKey.values()], CHAT_BACKEND_CLEANUP_BATCH_SIZE).map((batch) => ({ - chatIds: [], - files: batch.map(({ key, context, chatId, organizationId }) => ({ - key, - context, - chatId, - ...(organizationId ? { organizationId } : {}), - })), + files: batch, })), ] for (const events of chunkArray(payloads, MAX_BULK_ENQUEUE_EVENTS)) { await enqueueOutboxEvents(executor, CHAT_BACKEND_CLEANUP_EVENT, events, options) } + await enqueueChatInlineImageCleanup(executor, targets.chatIds, options) } /** @@ -104,9 +148,11 @@ export async function readAccountRemovalChats(params: { /** * Queues the cleanup of every chat an account removal cascades away, inside its transaction: * the chat ids are re-read there, after the account row is locked, and the files of any chat - * created since {@link readAccountRemovalChats} ran are read in the transaction. A key a turn adds - * to an already-read chat between that read and the removal's run stops is not collected here; - * when it names one of the account's uploads, the account storage sweep still reclaims it. + * created since {@link readAccountRemovalChats} ran are read in the transaction. Every chat's inline + * images go with its prefix sweep, however late they were written. An attachment a turn adds to + * an already-read chat between that read and the removal's run stops is not collected here; when + * it names one of the account's uploads, the account storage sweep still reclaims it, and + * otherwise it stays behind. */ export async function enqueueAccountRemovalChatCleanupInTx( tx: DbTransaction, diff --git a/apps/sim/lib/cleanup/chat-cleanup.integration.ts b/apps/sim/lib/cleanup/chat-cleanup.integration.ts index fe6d7294d5b..eca076a0522 100644 --- a/apps/sim/lib/cleanup/chat-cleanup.integration.ts +++ b/apps/sim/lib/cleanup/chat-cleanup.integration.ts @@ -1,57 +1,78 @@ /** Exercises chat cleanup's storage-key collection against real PostgreSQL message rows. */ import { db, dbFor } from '@sim/db' -import { copilotChats, copilotMessages, user } from '@sim/db/schema' +import { + copilotChats, + copilotMessages, + organization, + user, + workspace, + workspaceFiles, +} from '@sim/db/schema' +import { deleteWorkspaceFixture, insertWorkspaceFixture } from '@sim/db/testing/workspace-fixtures' import { generateId } from '@sim/utils/id' import { compareStrings } from '@sim/utils/string' import { eq } from 'drizzle-orm' import { afterAll, beforeAll, describe, expect, it } from 'vitest' -import { collectChatFiles } from '@/lib/cleanup/chat-files' -import { inlineChatImageKey } from '@/lib/mothership/chat/inline-image-key' +import { deletableChatFiles } from '@/lib/cleanup/chat-cleanup' +import { type ChatFileRef, collectChatFiles } from '@/lib/cleanup/chat-files' function attachment(key: string) { return { id: generateId(), key, filename: 'x.png', media_type: 'image/png', size: 1 } } -describe('chat cleanup file collection in PostgreSQL', () => { - const userId = generateId() +const userId = generateId() +const organizationId = generateId() - beforeAll(async () => { - const now = new Date() - await db.insert(user).values({ - id: userId, - name: 'Chat cleanup fixture', - email: `${userId}@fixture.test`, - emailVerified: true, - createdAt: now, - updatedAt: now, - }) +beforeAll(async () => { + const now = new Date() + await db.insert(user).values({ + id: userId, + name: 'Chat cleanup fixture', + email: `${userId}@fixture.test`, + emailVerified: true, + createdAt: now, + updatedAt: now, }) - - afterAll(async () => { - await db.delete(copilotChats).where(eq(copilotChats.userId, userId)) - await db.delete(user).where(eq(user.id, userId)) - await Promise.all([db.$client.end(), dbFor('cleanup').$client.end()]) + await db.insert(organization).values({ + id: organizationId, + name: 'Chat cleanup fixture', + slug: organizationId, + createdAt: now, + updatedAt: now, }) +}) - async function seedChat(contents: unknown[], deletedSeqs: number[] = []): Promise { - const [chat] = await db - .insert(copilotChats) - .values({ userId, type: 'mothership' }) - .returning({ id: copilotChats.id }) - await db.insert(copilotMessages).values( - contents.map((content, seq) => ({ - chatId: chat.id, - messageId: generateId(), - role: 'user', - seq, - content, - ...(deletedSeqs.includes(seq) ? { deletedAt: new Date() } : {}), - })) - ) - return chat.id - } +afterAll(async () => { + await db.delete(copilotChats).where(eq(copilotChats.userId, userId)) + await db.delete(organization).where(eq(organization.id, organizationId)) + await db.delete(user).where(eq(user.id, userId)) + await Promise.all([db.$client.end(), dbFor('cleanup').$client.end()]) +}) - it('collects attachment and inline image keys from every row, never from tool payloads', async () => { +async function seedChat( + contents: unknown[], + deletedSeqs: number[] = [], + organizationId?: string +): Promise { + const [chat] = await db + .insert(copilotChats) + .values({ userId, type: 'mothership', organizationId }) + .returning({ id: copilotChats.id }) + await db.insert(copilotMessages).values( + contents.map((content, seq) => ({ + chatId: chat.id, + messageId: generateId(), + role: 'user', + seq, + content, + ...(deletedSeqs.includes(seq) ? { deletedAt: new Date() } : {}), + })) + ) + return chat.id +} + +describe('chat cleanup file collection in PostgreSQL', () => { + it('collects attachment keys from every row, never from tool payloads or inline images', async () => { const chatId = await seedChat( [ { @@ -64,6 +85,7 @@ describe('chat cleanup file collection in PostgreSQL', () => { 'not an attachment', null, { key: '' }, + { key: 42 }, ], }, { @@ -108,18 +130,101 @@ describe('chat cleanup file collection in PostgreSQL', () => { key: 'assistant/org-1/user-1/u1/shared.png', context: 'mothership', chatId, - organizationId: 'org-1', + scope: { kind: 'organization', id: 'org-1' }, }, - { - key: inlineChatImageKey(chatId, 'req-1', 'files/chart.png'), - context: 'mothership', - chatId, - }, - { key: inlineChatImageKey(chatId, 'req-1', '/tmp/out.png'), context: 'mothership', chatId }, - { key: 'copilot/1234/legacy.png', context: 'copilot', chatId }, - { key: 'copilot/5/other.png', context: 'copilot', chatId: otherChatId }, - { key: 'copilot/9/removed.png', context: 'copilot', chatId }, + { key: 'copilot/1234/legacy.png', context: 'copilot', chatId, scope: null }, + { key: 'copilot/5/other.png', context: 'copilot', chatId: otherChatId, scope: null }, + { key: 'copilot/9/removed.png', context: 'copilot', chatId, scope: null }, ].sort((a, b) => compareStrings(a.key, b.key)) ) }) }) + +describe('deletable chat files against PostgreSQL message rows', () => { + const ref = (key: string, scope: ChatFileRef['scope']): ChatFileRef => ({ + key, + context: 'copilot', + chatId: generateId(), + ...(scope === undefined ? {} : { scope }), + }) + + it('keeps a key another chat of its scope still attaches, and deletes the rest', async () => { + const shared = `assistant/${organizationId}/${userId}/u1/shared.png` + const own = `assistant/${organizationId}/${userId}/u2/own.png` + const elsewhere = `copilot/${generateId()}/elsewhere.png` + const source = await seedChat( + [{ role: 'user', content: 'look', fileAttachments: [attachment(shared), attachment(own)] }], + [], + organizationId + ) + await seedChat( + [{ role: 'user', content: 'fork', fileAttachments: [attachment(shared), 'x', { key: 7 }] }], + [], + organizationId + ) + // Attached only by a chat outside the scope, which cannot be a fork of the deleted one. + await seedChat([{ role: 'user', content: 'other', fileAttachments: [attachment(elsewhere)] }]) + const files = await collectChatFiles([source]) + await db.delete(copilotChats).where(eq(copilotChats.id, source)) + + const deletable = await deletableChatFiles([ + ...files, + ref(elsewhere, { kind: 'organization', id: organizationId }), + ]) + + expect(deletable.map((file) => file.key).sort(compareStrings)).toEqual( + [elsewhere, own].sort(compareStrings) + ) + }) + + it("scopes a chat's file row by the row's own workspace, not the chat's organization", async () => { + const workspaceId = generateId() + await insertWorkspaceFixture(db, { + id: workspaceId, + name: 'Chat cleanup fixture', + ownerId: userId, + billedAccountUserId: userId, + }) + try { + const chatId = await seedChat([{ role: 'user', content: 'look' }], [], organizationId) + const key = `workspace/${workspaceId}/${generateId()}/linked.png` + await db.insert(workspaceFiles).values({ + id: generateId(), + key, + userId, + workspaceId, + context: 'mothership', + chatId, + originalName: 'linked.png', + contentType: 'image/png', + sizeBytes: 1, + }) + + const files = await collectChatFiles([chatId]) + + expect(files.find((file) => file.key === key)?.scope).toEqual({ + kind: 'workspace', + id: workspaceId, + }) + } finally { + await deleteWorkspaceFixture(db, eq(workspace.id, workspaceId)) + } + }) + + it('places a legacy ref by its key, keeping one it cannot place', async () => { + const referenced = `assistant/${organizationId}/${userId}/u1/kept.png` + const unreferenced = `assistant/${organizationId}/${userId}/u2/gone.png` + const unplaceable = `copilot/${generateId()}/legacy.png` + await seedChat( + [{ role: 'user', content: 'fork', fileAttachments: [attachment(referenced)] }], + [], + organizationId + ) + + const deletable = await deletableChatFiles( + [referenced, unreferenced, unplaceable].map((key) => ref(key, undefined)) + ) + + expect(deletable.map((file) => file.key)).toEqual([unreferenced]) + }) +}) diff --git a/apps/sim/lib/cleanup/chat-cleanup.test.ts b/apps/sim/lib/cleanup/chat-cleanup.test.ts index d6a8596661a..e9a43ac3802 100644 --- a/apps/sim/lib/cleanup/chat-cleanup.test.ts +++ b/apps/sim/lib/cleanup/chat-cleanup.test.ts @@ -7,7 +7,6 @@ import { beforeEach, describe, expect, it, vi } from 'vitest' vi.mock('@/lib/uploads', () => uploadsMock) import { prepareChatCleanup } from '@/lib/cleanup/chat-cleanup' -import { inlineChatImageKey } from '@/lib/mothership/chat/inline-image-key' const chatId = '3f0c2a52-8a43-4d4b-9b5f-0b3c7a1e2d10' @@ -15,25 +14,15 @@ function attachment(key: string) { return { id: 'wf_x', key, filename: 'x.png', media_type: 'image/png', size: 1 } } -/** - * Purges one deleted chat whose message rows are `messages`, while `remaining` are the message - * rows other chats of the organization still hold; returns the deleted keys by context. - */ -async function purge( - messages: Record[], - remaining: Record[] = [] -) { +/** Purges one deleted chat whose message rows are `messages`; returns the deleted keys by context. */ +async function purge(messages: Record[]) { queueTableRows(workspaceFiles, []) queueTableRows( copilotMessages, - messages.map((content) => ({ chatId, content })) + messages.map((content) => ({ chatId, fileAttachments: content.fileAttachments ?? null })) ) const cleanup = await prepareChatCleanup([chatId], 'test') queueTableRows(copilotChats, []) - queueTableRows( - copilotMessages, - remaining.map((content) => ({ fileAttachments: content.fileAttachments ?? null })) - ) await cleanup.execute() return Object.fromEntries( storageServiceMockFns.mockDeleteFiles.mock.calls.map(([keys, context]) => [context, keys]) @@ -63,39 +52,4 @@ describe('chat purge storage', () => { ]) expect(deleted).toEqual({ copilot: ['copilot/1234/legacy.png'] }) }) - - it('deletes an organization attachment only once no remaining chat references it', async () => { - const shared = 'assistant/org-1/user-1/u1/shared.png' - const own = 'assistant/org-1/user-1/u2/own.png' - const message = { - role: 'user', - content: 'look', - fileAttachments: [attachment(shared), attachment(own)], - } - // A fork of this chat still holds the shared upload. - const deleted = await purge( - [message], - [{ role: 'user', content: 'fork', fileAttachments: [attachment(shared)] }] - ) - expect(deleted).toEqual({ mothership: [own] }) - }) - - it('deletes the inline images an assistant message published', async () => { - const deleted = await purge([ - { - role: 'assistant', - requestId: 'req-1', - content: 'Here: ![chart](files/chart.png) and `![code](files/no.png)`', - contentBlocks: [{ type: 'text', content: '![other](/tmp/out.png)' }], - }, - { role: 'user', requestId: 'req-2', content: '![u](files/u.png)' }, - { role: 'assistant', content: '![x](files/x.png)' }, - ]) - expect(deleted).toEqual({ - mothership: [ - inlineChatImageKey(chatId, 'req-1', 'files/chart.png'), - inlineChatImageKey(chatId, 'req-1', '/tmp/out.png'), - ], - }) - }) }) diff --git a/apps/sim/lib/cleanup/chat-cleanup.ts b/apps/sim/lib/cleanup/chat-cleanup.ts index 5045ece59ce..8de4de6ed36 100644 --- a/apps/sim/lib/cleanup/chat-cleanup.ts +++ b/apps/sim/lib/cleanup/chat-cleanup.ts @@ -2,11 +2,14 @@ import { dbFor } from '@sim/db' import { copilotChats, copilotMessages } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { chunkArray } from '@sim/utils/helpers' -import { isRecordLike } from '@sim/utils/object' -import { and, eq, inArray, or, sql } from 'drizzle-orm' +import { eq, inArray, sql } from 'drizzle-orm' +import { enqueueChatInlineImageCleanup } from '@/lib/cleanup/chat-backend-cleanup' import { + CHAT_FILE_COLLECT_CHUNK_SIZE, type ChatFileRef, + type ChatFileReferenceScope, type ChatScopedContext, + chatFileReferenceScope, collectChatFiles, } from '@/lib/cleanup/chat-files' import { env } from '@/lib/core/config/env' @@ -19,57 +22,72 @@ const logger = createLogger('ChatCleanup') const cleanupDb = dbFor('cleanup') const COPILOT_CLEANUP_BATCH_SIZE = 1000 -/** Bounds how many keys or chat ids one query names. */ -const CHAT_FILE_COLLECT_CHUNK_SIZE = 500 /** - * Organization attachment keys that a remaining chat still references, so they outlive the - * chats being purged. Runs after the caller deleted those chats' rows, so any match is + * The keys among `keys` that some chat of `scope` names in its messages' `fileAttachments`, in one + * scan of the scope's messages. The keys travel as one JSON parameter, so their number never + * meets the bind limit. A scope with no chat left is not scanned. + */ +async function keysReferencedInScope( + scope: ChatFileReferenceScope, + keys: readonly string[] +): Promise { + const inScope = + scope.kind === 'organization' + ? eq(copilotChats.organizationId, scope.id) + : eq(copilotChats.workspaceId, scope.id) + const [anyChat] = await cleanupDb + .select({ id: copilotChats.id }) + .from(copilotChats) + .where(inScope) + .limit(1) + if (!anyChat) return [] + const attachments = sql`${copilotMessages.content} -> 'fileAttachments'` + const rows = await cleanupDb.execute<{ key: string }>(sql` + SELECT DISTINCT attachment ->> 'key' AS key + FROM ${copilotMessages} + JOIN ${copilotChats} ON ${copilotChats.id} = ${copilotMessages.chatId} + CROSS JOIN LATERAL jsonb_array_elements( + CASE WHEN jsonb_typeof(${attachments}) = 'array' THEN ${attachments} ELSE '[]'::jsonb END + ) AS attachment + WHERE ${inScope} + AND jsonb_typeof(attachment -> 'key') = 'string' + AND attachment ->> 'key' IN (SELECT jsonb_array_elements_text(${JSON.stringify(keys)}::jsonb)) + `) + return rows.map((row) => row.key) +} + +/** + * The files among `files` that may be deleted: those no remaining chat of their scope still + * references ({@link ChatFileRef}), checked with one query per scope. A key whose scope is unknown + * is kept rather than risked. Runs after the caller deleted those chats' rows, so any match is * another chat: a fork, or a chat that is only soft-deleted and may be restored. */ -export async function organizationAttachmentsStillReferenced( - files: readonly ChatFileRef[] -): Promise> { - const keysByOrganization = new Map() +export async function deletableChatFiles(files: readonly ChatFileRef[]): Promise { + const keysByScope = new Map() + const unknown = new Set() for (const file of files) { - if (!file.organizationId) continue - const keys = keysByOrganization.get(file.organizationId) - if (keys) keys.push(file.key) - else keysByOrganization.set(file.organizationId, [file.key]) - } - const referenced = new Set() - for (const [organizationId, keys] of keysByOrganization) { - for (const chunk of chunkArray(keys, CHAT_FILE_COLLECT_CHUNK_SIZE)) { - const wanted = new Set(chunk) - const rows = await cleanupDb - .select({ fileAttachments: sql`${copilotMessages.content} -> 'fileAttachments'` }) - .from(copilotMessages) - .innerJoin(copilotChats, eq(copilotChats.id, copilotMessages.chatId)) - .where( - and( - eq(copilotChats.organizationId, organizationId), - or( - ...chunk.map( - (key) => - sql`${copilotMessages.content} @> ${JSON.stringify({ fileAttachments: [{ key }] })}::jsonb` - ) - ) - ) - ) - for (const { fileAttachments } of rows) { - if (!Array.isArray(fileAttachments)) continue - for (const attachment of fileAttachments) { - if ( - isRecordLike(attachment) && - typeof attachment.key === 'string' && - wanted.has(attachment.key) - ) - referenced.add(attachment.key) - } - } + const scope = chatFileReferenceScope(file) + if (scope === 'unshared') continue + if (scope === 'unknown') { + unknown.add(file.key) + continue } + const scopeKey = `${scope.kind}:${scope.id}` + const group = keysByScope.get(scopeKey) + if (group) group.keys.push(file.key) + else keysByScope.set(scopeKey, { scope, keys: [file.key] }) + } + if (unknown.size > 0) { + logger.warn(`Keeping ${unknown.size} chat files whose reference scope is unknown`, { + keys: [...unknown], + }) } - return referenced + const kept = new Set(unknown) + for (const { scope, keys } of keysByScope.values()) { + for (const key of await keysReferencedInScope(scope, [...new Set(keys)])) kept.add(key) + } + return files.filter((file) => !kept.has(file.key)) } /** Groups files by storage context so each context can use one batch DELETE call. */ @@ -156,7 +174,7 @@ export async function cleanupCopilotBackend( /** * Full chat cleanup: collect file refs, then (after DB deletion by caller) - * call copilot backend and delete storage files. + * call copilot backend, delete storage files, and queue the chats' inline image sweep. * * Usage: * const cleanup = await prepareChatCleanup(chatIds, label) @@ -194,12 +212,9 @@ export async function prepareChatCleanup( ) } const confirmedChatIds = chatIds.filter((id) => !survivors.has(id)) - const sharedKeys = await organizationAttachmentsStillReferenced( + const confirmedFiles = await deletableChatFiles( files.filter((file) => !survivors.has(file.chatId)) ) - const confirmedFiles = files.filter( - (file) => !survivors.has(file.chatId) && !sharedKeys.has(file.key) - ) // Call copilot backend if (confirmedChatIds.length > 0) { @@ -207,6 +222,7 @@ export async function prepareChatCleanup( logger.info( `[${label}] Copilot backend: ${copilotResult.deleted} deleted, ${copilotResult.failed} failed` ) + await enqueueChatInlineImageCleanup(cleanupDb, confirmedChatIds) } // Delete storage files with correct context per file diff --git a/apps/sim/lib/cleanup/chat-files.ts b/apps/sim/lib/cleanup/chat-files.ts index 9c09dfa7fb2..a576ae7289b 100644 --- a/apps/sim/lib/cleanup/chat-files.ts +++ b/apps/sim/lib/cleanup/chat-files.ts @@ -1,22 +1,24 @@ import { dbFor } from '@sim/db' -import { copilotMessages, workspaceFiles } from '@sim/db/schema' +import { + copilotChats, + copilotMessages, + workspaceFile, + workspaceFiles, + workspaceFileVersion, +} from '@sim/db/schema' import { chunkArray } from '@sim/utils/helpers' import { isRecordLike } from '@sim/utils/object' -import { and, inArray, isNull } from 'drizzle-orm' +import { and, inArray, isNull, sql } from 'drizzle-orm' import type { DbOrTx } from '@/lib/db/types' -import { - inlineChatImageKey, - inlineChatImageReferences, -} from '@/lib/mothership/chat/inline-image-key' -import { detoastedMessageContent, textMessageContent } from '@/lib/mothership/chat/text-transcript' +import { inlineChatImagePrefix } from '@/lib/mothership/chat/inline-image-key' import type { StorageContext } from '@/lib/uploads' -import { tryInferContextFromKey } from '@/lib/uploads/utils/file' +import { extractWorkspaceIdFromStorageKey, tryInferContextFromKey } from '@/lib/uploads/utils/file' /** File collection reads use the cleanup pool so a purge never waits on the request pool. */ const cleanupDb = dbFor('cleanup') -/** Bounds how many chats' `copilot_messages` rows are scanned per query. */ -const CHAT_FILE_COLLECT_CHUNK_SIZE = 500 +/** Bounds how many chat ids or keys one chat cleanup query names. */ +export const CHAT_FILE_COLLECT_CHUNK_SIZE = 500 /** * Only storage in these contexts is tied to chat/task lifecycle. Workspace @@ -27,17 +29,29 @@ const CHAT_FILE_COLLECT_CHUNK_SIZE = 500 const CHAT_SCOPED_CONTEXTS = ['copilot', 'mothership'] as const satisfies readonly StorageContext[] export type ChatScopedContext = (typeof CHAT_SCOPED_CONTEXTS)[number] -/** A stored object that lives and dies with the chat named by `chatId`. */ +/** The chats a {@link ChatFileRef}'s key must be checked against before it is deleted. */ +export type ChatFileReferenceScope = + | { kind: 'organization'; id: string } + | { kind: 'workspace'; id: string } + +/** + * A stored object that lives and dies with the chat named by `chatId`. A fork keeps naming its + * source's key wherever it did not copy the file (an organization attachment, a message-only + * copilot attachment, or an upload whose copy failed), and a fork stays in its source's workspace + * or organization, so the key is deleted only once no remaining chat of that scope names it in + * its `fileAttachments`. + */ export interface ChatFileRef { key: string context: ChatScopedContext chatId: string /** - * An organization Chat attachment (`assistant//…`): no `workspace_files` row owns it, - * and a fork carries the same key, so it is deleted only once no remaining chat of that - * organization references it. + * The organization or workspace whose chats may still reference the key: an organization + * attachment's organization, else the chat's organization, else the file row's or chat's + * workspace. `null` when no other chat can reference it (the chat had neither, so no fork was + * made from it). Absent only on events queued before refs carried their scope. */ - organizationId?: string + scope?: ChatFileReferenceScope | null } /** The organization an `assistant//…` attachment key was uploaded under. */ @@ -48,47 +62,75 @@ function organizationAttachmentOwner(key: string): string | undefined { } /** - * The chat images an assistant message published under its request id, keyed by chat id, - * so they are purged with the chat. A row this cannot read as such a message has none. + * Which chats must be checked before `file`'s key is deleted: a scope, `unshared` when no other + * chat can reference it, or `unknown`, in which case the key is kept rather than risk deleting a + * file a fork still shows. */ -function inlineChatImageKeys(chatId: string, content: Record): string[] { - if (content.role !== 'assistant' || typeof content.requestId !== 'string') return [] - const published = inlineChatImageReferences({ - role: 'assistant', - requestId: content.requestId, - content: typeof content.content === 'string' ? content.content : '', - contentBlocks: Array.isArray(content.contentBlocks) - ? content.contentBlocks.flatMap((block) => - isRecordLike(block) && block.type === 'text' && typeof block.content === 'string' - ? [{ type: 'text' as const, content: block.content }] - : [] - ) - : undefined, - }) - if (!published) return [] - const keys: string[] = [] - for (const reference of published.references) { - try { - keys.push(inlineChatImageKey(chatId, published.requestId, reference)) - } catch { - // A chat or request id outside the key grammar never had an image stored under it. - } - } - return keys +export function chatFileReferenceScope( + file: ChatFileRef +): ChatFileReferenceScope | 'unshared' | 'unknown' { + if (file.scope !== undefined) return file.scope ?? 'unshared' + // Legacy: refs queued before they carried their scope recover it from the key. Remove once + // those events have drained. + const organizationId = organizationAttachmentOwner(file.key) + if (organizationId) return { kind: 'organization', id: organizationId } + const workspaceId = extractWorkspaceIdFromStorageKey(file.key) + if (workspaceId) return { kind: 'workspace', id: workspaceId } + // A fork copies the source's inline images under its own id, so the chat's own are unshared. + const ownImages = inlineChatImagePrefix(file.chatId) + if (ownImages && file.key.startsWith(ownImages)) return 'unshared' + return 'unknown' +} + +/** The string keys of a message's `fileAttachments`; anything else names no stored file. */ +function fileAttachmentKeys(value: unknown): string[] { + if (!Array.isArray(value)) return [] + return value.flatMap((attachment) => + isRecordLike(attachment) && typeof attachment.key === 'string' && attachment.key + ? [attachment.key] + : [] + ) } /** - * Collect all file storage keys for the given chat IDs from three sources: + * The keys among `keys` that no workspace file row, legacy file row or file version names, live + * or soft-deleted, in one query. + */ +async function keysWithoutFileRow(executor: DbOrTx, keys: readonly string[]): Promise { + if (keys.length === 0) return [] + const rows = await executor.execute<{ key: string }>(sql` + SELECT candidate AS key + FROM jsonb_array_elements_text(${JSON.stringify(keys)}::jsonb) AS candidate + WHERE NOT EXISTS (SELECT 1 FROM ${workspaceFiles} WHERE ${workspaceFiles.key} = candidate) + AND NOT EXISTS (SELECT 1 FROM ${workspaceFile} WHERE ${workspaceFile.key} = candidate) + AND NOT EXISTS ( + SELECT 1 FROM ${workspaceFileVersion} WHERE ${workspaceFileVersion.key} = candidate + ) + `) + return rows.map((row) => row.key) +} + +/** + * Collect all file storage keys for the given chat IDs from two sources: * 1. workspaceFiles rows with chatId FK (chat-scoped contexts only) - * 2. fileAttachments[].key inside each copilot_messages.content: copilot keys, and - * organization attachments under their own context (see {@link ChatFileRef.organizationId}) - * 3. the chat-scoped inline images each assistant message published + * 2. fileAttachments[].key inside each copilot_messages.content: copilot keys, organization + * attachments (`assistant//…`) under their own context, and `workspace//…` uploads + * no file row names at all + * + * Each ref carries the scope whose chats may still reference it (see {@link ChatFileRef}). + * + * Inline chat images are not collected: they live under the chat's own + * `chat-images//` prefix with no row naming them, and a turn can publish one after this + * read, so the chat cleanup handler deletes that whole prefix once the chat's row is gone. * * A workspace attachment is owned by its `workspace_files` row (source 1, or the workspace * file lifecycle), never by the message: a fork carries the same attachment when its copy * failed or its file was deleted, and the copilot bucket falls back to the workspace bucket * on GCS (and may be configured to it on S3), so deleting such a key as copilot storage - * could delete a file another chat or the workspace still uses. + * could delete a file another chat or the workspace still uses. Only once no row names the key + * at all (a soft-deleted row can still be restored) does the message own it: a fork kept the + * upload past its source chat, whose row cascaded away, so the last chat attaching it collects + * it, and the reference check keeps it while another chat of that workspace still attaches it. * * Reads through the cleanup pool unless `executor` is given, which a deletion passes to read the * keys inside its transaction once it holds the chat rows, so no row naming a key can commit @@ -104,12 +146,21 @@ export async function collectChatFiles( const seen = new Set() for (const chunk of chunkArray([...chatIds], CHAT_FILE_COLLECT_CHUNK_SIZE)) { - const [linkedFiles, messageRows] = await Promise.all([ + const [chats, linkedFiles, messageRows] = await Promise.all([ + executor + .select({ + id: copilotChats.id, + workspaceId: copilotChats.workspaceId, + organizationId: copilotChats.organizationId, + }) + .from(copilotChats) + .where(inArray(copilotChats.id, chunk)), executor .select({ key: workspaceFiles.key, context: workspaceFiles.context, chatId: workspaceFiles.chatId, + workspaceId: workspaceFiles.workspaceId, }) .from(workspaceFiles) .where( @@ -123,51 +174,77 @@ export async function collectChatFiles( // deletion path collecting blob keys, so attachments on any row count. executor .select({ - content: textMessageContent(['role', 'requestId', 'content', 'fileAttachments']), + fileAttachments: sql`${copilotMessages.content} -> 'fileAttachments'`, chatId: copilotMessages.chatId, }) .from(copilotMessages) - .crossJoinLateral(detoastedMessageContent) .where(inArray(copilotMessages.chatId, chunk)), ]) + const chatScopes = new Map(chats.map((chat) => [chat.id, chat])) + /** The scope a key collected from `chatId` is checked in, preferring the file row's workspace. */ + const scopeOf = ( + chatId: string, + fileWorkspaceId?: string | null + ): ChatFileReferenceScope | null => { + // A file row's own workspace owns it, whatever the chat it was attached in belongs to. + if (fileWorkspaceId) return { kind: 'workspace', id: fileWorkspaceId } + const chat = chatScopes.get(chatId) + if (chat?.organizationId) return { kind: 'organization', id: chat.organizationId } + return chat?.workspaceId ? { kind: 'workspace', id: chat.workspaceId } : null + } + for (const f of linkedFiles) { if (f.chatId && !seen.has(f.key)) { seen.add(f.key) - files.push({ key: f.key, context: f.context as ChatScopedContext, chatId: f.chatId }) + files.push({ + key: f.key, + context: f.context as ChatScopedContext, + chatId: f.chatId, + scope: scopeOf(f.chatId, f.workspaceId), + }) } } + const unownedCandidates = new Map() for (const row of messageRows) { - const msg = row.content - if (!msg || typeof msg !== 'object') continue - for (const key of inlineChatImageKeys(row.chatId, msg as Record)) { - if (!seen.has(key)) { - seen.add(key) - files.push({ key, context: 'mothership', chatId: row.chatId }) + for (const key of fileAttachmentKeys(row.fileAttachments)) { + if (seen.has(key)) continue + const uploadWorkspaceId = key.startsWith('workspace/') + ? extractWorkspaceIdFromStorageKey(key) + : null + if (uploadWorkspaceId) { + if (!unownedCandidates.has(key)) + unownedCandidates.set(key, { chatId: row.chatId, workspaceId: uploadWorkspaceId }) + continue } - } - const attachments = (msg as Record).fileAttachments - if (!Array.isArray(attachments)) continue - for (const attachment of attachments) { - if ( - attachment && - typeof attachment === 'object' && - (attachment as Record).key - ) { - const key = (attachment as Record).key as string - if (seen.has(key)) continue - const organizationId = organizationAttachmentOwner(key) - if (organizationId) { - seen.add(key) - files.push({ key, context: 'mothership', chatId: row.chatId, organizationId }) - } else if (tryInferContextFromKey(key) === 'copilot') { - seen.add(key) - files.push({ key, context: 'copilot', chatId: row.chatId }) - } + const organizationId = organizationAttachmentOwner(key) + if (organizationId) { + seen.add(key) + files.push({ + key, + context: 'mothership', + chatId: row.chatId, + scope: { kind: 'organization', id: organizationId }, + }) + } else if (tryInferContextFromKey(key) === 'copilot') { + seen.add(key) + files.push({ key, context: 'copilot', chatId: row.chatId, scope: scopeOf(row.chatId) }) } } } + + for (const key of await keysWithoutFileRow(executor, [...unownedCandidates.keys()])) { + const candidate = unownedCandidates.get(key) + if (!candidate || seen.has(key)) continue + seen.add(key) + files.push({ + key, + context: 'mothership', + chatId: candidate.chatId, + scope: { kind: 'workspace', id: candidate.workspaceId }, + }) + } } return files diff --git a/apps/sim/lib/core/outbox/handlers.ts b/apps/sim/lib/core/outbox/handlers.ts index cdecc40f1c8..62f9094fe03 100644 --- a/apps/sim/lib/core/outbox/handlers.ts +++ b/apps/sim/lib/core/outbox/handlers.ts @@ -13,7 +13,10 @@ import { } from '@/lib/billing/enterprise-owner-claim-events' import { MEMBER_BILLING_RECONCILIATION_EVENT_TYPE } from '@/lib/billing/organizations/membership-reconciliation-event' import { OUTBOX_EVENT_TYPES } from '@/lib/billing/webhooks/outbox-events' -import { CHAT_BACKEND_CLEANUP_EVENT } from '@/lib/cleanup/chat-backend-cleanup-event' +import { + CHAT_BACKEND_CLEANUP_EVENT, + CHAT_INLINE_IMAGE_CLEANUP_EVENT, +} from '@/lib/cleanup/chat-backend-cleanup-event' import type { LazyOutboxHandlerGroup } from '@/lib/core/outbox/service' import { EXECUTION_REMOVAL_STOP_EVENT } from '@/lib/execution/removal-stop-event' import { DIRECT_GRANT_EMAIL_EVENT_TYPE } from '@/lib/invitations/direct-grant-event' @@ -193,7 +196,7 @@ export const OUTBOX_HANDLER_GROUPS: readonly LazyOutboxHandlerGroup[] = [ }, }, { - events: [CHAT_BACKEND_CLEANUP_EVENT], + events: [CHAT_BACKEND_CLEANUP_EVENT, CHAT_INLINE_IMAGE_CLEANUP_EVENT], load: async () => { const { chatBackendCleanupOutboxHandlers } = await import( '@/lib/cleanup/chat-backend-cleanup-outbox' diff --git a/apps/sim/lib/core/outbox/retention.ts b/apps/sim/lib/core/outbox/retention.ts index 3af4bdf01e0..2bc40c8e28d 100644 --- a/apps/sim/lib/core/outbox/retention.ts +++ b/apps/sim/lib/core/outbox/retention.ts @@ -2,7 +2,10 @@ import { db } from '@sim/db' import { outboxEvent } from '@sim/db/schema' import { createLogger } from '@sim/logger' import { and, asc, eq, inArray, lt, sql } from 'drizzle-orm' -import { CHAT_BACKEND_CLEANUP_EVENT } from '@/lib/cleanup/chat-backend-cleanup-event' +import { + CHAT_BACKEND_CLEANUP_EVENT, + CHAT_INLINE_IMAGE_CLEANUP_EVENT, +} from '@/lib/cleanup/chat-backend-cleanup-event' import { EMBEDDING_CHECKPOINT_CLEANUP_EVENT, OCR_CHECKPOINT_CLEANUP_OUTBOX_EVENT, @@ -51,6 +54,7 @@ const PRUNABLE_OUTBOX_EVENT_TYPES = [ WORKSPACE_FILE_STORAGE_CLEANUP_OUTBOX_EVENT, MEMORY_CLEANUP_EVENT, CHAT_BACKEND_CLEANUP_EVENT, + CHAT_INLINE_IMAGE_CLEANUP_EVENT, ] as const async function pruneCompletedBatch(eventType: string, cutoff: Date): Promise { diff --git a/apps/sim/lib/mothership/chat/application/fork.ts b/apps/sim/lib/mothership/chat/application/fork.ts index def38f220b2..507df151b56 100644 --- a/apps/sim/lib/mothership/chat/application/fork.ts +++ b/apps/sim/lib/mothership/chat/application/fork.ts @@ -166,11 +166,10 @@ export const forkChat = defineAuthorizedChatUseCase({ ) } /** - * The fork can share keys with its source (organization attachments, files whose copy - * failed), and chat cleanup deletes a shared key once no remaining chat references it, - * checking only after it deletes the source row. Holding the source row until commit - * orders the two: a purge that already removed it refuses this fork, and one that has - * not waits for this commit and then sees the fork's references. + * The fork can share keys with its source (`ChatFileRef` in `lib/cleanup/chat-files.ts`), + * and chat cleanup checks for other references only after the source row is deleted. + * Holding the source row until commit orders the two: a purge that already removed it + * refuses this fork, and one that has not waits for this commit and sees its references. */ const [source] = await tx .select({ id: copilotChats.id, memorySpaceId: copilotChats.memorySpaceId }) diff --git a/apps/sim/lib/mothership/chat/deletion.ts b/apps/sim/lib/mothership/chat/deletion.ts index e3a6d68ba4b..e3e0bc650d7 100644 --- a/apps/sim/lib/mothership/chat/deletion.ts +++ b/apps/sim/lib/mothership/chat/deletion.ts @@ -46,8 +46,9 @@ export async function softDeleteMothershipChat(params: { /** * Permanently deletes a copilot chat and stops its live runs in the same transaction, which also - * queues the purge of what lives outside Sim's rows: the copilot backend's chat data and the - * chat's stored files. Their keys are read under the chat row's lock, so no message or file + * queues the purge of what lives outside Sim's rows: the copilot backend's chat data, the chat's + * stored files, and its inline images (swept by prefix when the purge runs, so one published after + * this read goes too). The file keys are read under the chat row's lock, so no message or file * naming the chat can commit between that read and the delete that cascades them away, and the * purge is recorded exactly when the delete commits. Null when the chat was not the user's. */ diff --git a/apps/sim/lib/mothership/chat/inline-image-key.ts b/apps/sim/lib/mothership/chat/inline-image-key.ts index aa35c449011..19757e59920 100644 --- a/apps/sim/lib/mothership/chat/inline-image-key.ts +++ b/apps/sim/lib/mothership/chat/inline-image-key.ts @@ -8,12 +8,24 @@ import { import { collectMarkdownImageSources } from '@/lib/mothership/chat/markdown-images' import type { PersistedMessage } from '@/lib/mothership/chat/persisted-message' +function chatImagePrefix(validChatId: string): string { + return `${INLINE_CHAT_IMAGE_PREFIX}${validChatId}/` +} + +/** + * The `chat-images//` prefix every inline image a chat published is stored under, or null + * for an id outside the key grammar, under which no image can have been stored. + */ +export function inlineChatImagePrefix(chatId: string): string | null { + return inlineImageRequestIdSchema.safeParse(chatId).success ? chatImagePrefix(chatId) : null +} + /** Stable per-message source identity makes replay independent of sandbox lifetime. */ export function inlineChatImageKey(chatId: string, requestId: string, reference: string): string { - inlineImageRequestIdSchema.parse(chatId) + const prefix = chatImagePrefix(inlineImageRequestIdSchema.parse(chatId)) inlineImageRequestIdSchema.parse(requestId) const digest = createHash('sha256').update(normalizeInlineFileReference(reference)).digest('hex') - return `${INLINE_CHAT_IMAGE_PREFIX}${chatId}/${requestId}/${digest}.webp` + return `${prefix}${requestId}/${digest}.webp` } /** diff --git a/apps/sim/lib/projects/files/prefix-cleanup.test.ts b/apps/sim/lib/projects/files/prefix-cleanup.test.ts index 9993c3e22b3..cb535a234ca 100644 --- a/apps/sim/lib/projects/files/prefix-cleanup.test.ts +++ b/apps/sim/lib/projects/files/prefix-cleanup.test.ts @@ -10,15 +10,10 @@ import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' vi.mock('@/lib/uploads/core/setup.server', () => uploadsSetupMock) vi.mock('@/lib/uploads/config', () => uploadsConfigMock) vi.mock('@/lib/uploads/server/metadata', () => uploadsMetadataMock) -vi.mock('@/lib/uploads/core/prefix-sweep', async (importOriginal) => { - const actual = await importOriginal() - return { ...actual, listStorageKeyPage: vi.fn(actual.listStorageKeyPage) } -}) import type { OutboxEventContext } from '@/lib/core/outbox/service' import { PROJECT_FILE_PREFIX_CLEANUP_EVENT } from '@/lib/projects/files/outbox-events' import { projectFilePrefixCleanupOutboxHandlers } from '@/lib/projects/files/prefix-cleanup' -import { listStorageKeyPage } from '@/lib/uploads/core/prefix-sweep' const cleanup = projectFilePrefixCleanupOutboxHandlers[PROJECT_FILE_PREFIX_CLEANUP_EVENT] @@ -83,46 +78,4 @@ describe('Project prefix cleanup on local storage', () => { expect(await exists('project/p1/zz.txt')).toBe(false) }) - - it('keeps sweeping after a short page that is not the last', async () => { - await put('project/p1/a.txt') - await put('project/p1/b.txt') - vi.mocked(listStorageKeyPage).mockResolvedValueOnce({ - keys: ['project/p1/a.txt'], - done: false, - nextPageToken: 'next', - }) - - const result = await cleanup({ projectId: 'p1' }, outboxContext()) - - expect(result).toMatchObject({ outcome: 'deferred', consumeAttempt: false }) - expect(await exists('project/p1/a.txt')).toBe(false) - expect(await exists('project/p1/b.txt')).toBe(true) - }) - - it('follows an empty page that is not the last to the objects after it', async () => { - await put('project/p1/a.txt') - vi.mocked(listStorageKeyPage) - .mockResolvedValueOnce({ keys: [], done: false, nextPageToken: 'next' }) - .mockResolvedValueOnce({ keys: ['project/p1/a.txt'], done: true }) - - await expect(cleanup({ projectId: 'p1' }, outboxContext())).resolves.toBeUndefined() - - expect(vi.mocked(listStorageKeyPage).mock.calls[1]?.[3]).toBe('next') - expect(await exists('project/p1/a.txt')).toBe(false) - }) - - it('spends an attempt when its listing stops advancing without removing anything', async () => { - vi.mocked(listStorageKeyPage).mockResolvedValue({ - keys: [], - done: false, - nextPageToken: 'stuck', - }) - - const result = await cleanup({ projectId: 'p1' }, outboxContext()) - - expect(result).toMatchObject({ outcome: 'deferred' }) - expect(result).not.toMatchObject({ consumeAttempt: false }) - expect(listStorageKeyPage).toHaveBeenCalledTimes(2) - }) }) diff --git a/apps/sim/lib/projects/files/prefix-cleanup.ts b/apps/sim/lib/projects/files/prefix-cleanup.ts index 2b39b2eac8d..d47d2fc7bb9 100644 --- a/apps/sim/lib/projects/files/prefix-cleanup.ts +++ b/apps/sim/lib/projects/files/prefix-cleanup.ts @@ -7,7 +7,6 @@ import { continueOutboxHandler, deferOutboxHandler, enqueueOutboxEvent, - type OutboxEventContext, type OutboxHandler, type OutboxHandlerRegistry, } from '@/lib/core/outbox/service' @@ -18,10 +17,9 @@ import { } from '@/lib/projects/files/outbox-events' import { isObjectNotFoundError } from '@/lib/uploads/core/errors' import { - deleteStorageKeysUnderPrefix, - listStorageKeyPage, outboxDeadlineNear, storageCleanupAvailableAt, + sweepStoragePrefix, } from '@/lib/uploads/core/prefix-sweep' import { abortProviderUpload } from '@/lib/uploads/upload-session/provider' import type { UploadSessionRecord } from '@/lib/uploads/upload-session/service' @@ -201,35 +199,6 @@ export async function queueRetiredProjectUploadCleanup( }) } -/** - * Deletes one listing page's objects per run, re-listing from the prefix start next time. A - * provider (Azure Blob, GCS, S3-compatibles) may return a short or empty page that is not the last, - * so only `page.done` ends the sweep; a non-final page that removes nothing is followed to the next - * one within the run, so objects behind it are still reached. `stalled` when no page removed - * anything before the listing stopped advancing or the deadline neared. - */ -async function sweepPrefix( - projectId: string, - context: OutboxEventContext -): Promise<'done' | 'continue' | 'stalled'> { - const prefix = prefixFor(projectId) - let pageToken: string | undefined - for (;;) { - const page = await listStorageKeyPage(prefix, 'project', context.signal, pageToken) - const { deleted, complete } = await deleteStorageKeysUnderPrefix( - prefix, - 'project', - page.keys, - context - ) - if (complete && page.done) return 'done' - if (deleted > 0) return 'continue' - const advanced = page.nextPageToken !== undefined && page.nextPageToken !== pageToken - if (!complete || !advanced || outboxDeadlineNear(context)) return 'stalled' - pageToken = page.nextPageToken - } -} - const cleanupPrefix: OutboxHandler = async (raw, context) => { const payload = parsePayload(raw) const [owner] = await db @@ -267,7 +236,7 @@ const cleanupPrefix: OutboxHandler = async (raw, context) => { return continueOutboxHandler('Continuing retired Project upload cleanup') } } - const sweep = await sweepPrefix(payload.projectId, context) + const sweep = await sweepStoragePrefix(prefixFor(payload.projectId), 'project', context) if (sweep === 'continue') return continueOutboxHandler('Continuing Project storage cleanup') if (sweep === 'stalled') return deferOutboxHandler('Project storage cleanup removed nothing from its listing') diff --git a/apps/sim/lib/uploads/core/prefix-sweep.test.ts b/apps/sim/lib/uploads/core/prefix-sweep.test.ts new file mode 100644 index 00000000000..fb189222a0a --- /dev/null +++ b/apps/sim/lib/uploads/core/prefix-sweep.test.ts @@ -0,0 +1,53 @@ +import { storageServiceMock, storageServiceMockFns } from '@sim/testing/mocks/storage-service.mock' +import { setUploadsConfig, uploadsConfigMock } from '@sim/testing/mocks/uploads-config.mock' +import { uploadsSetupMock } from '@sim/testing/mocks/uploads-setup.mock' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { mockListS3ObjectKeyPage } = vi.hoisted(() => ({ mockListS3ObjectKeyPage: vi.fn() })) + +vi.mock('@/lib/uploads/config', () => uploadsConfigMock) +vi.mock('@/lib/uploads/core/setup.server', () => uploadsSetupMock) +vi.mock('@/lib/uploads/core/storage-service', () => storageServiceMock) +vi.mock('@/lib/uploads/providers/s3/client', () => ({ + listS3ObjectKeyPage: mockListS3ObjectKeyPage, +})) + +import { sweepStoragePrefix } from '@/lib/uploads/core/prefix-sweep' + +const outbox = { signal: new AbortController().signal } +const deletedKeys = () => storageServiceMockFns.mockDeleteFile.mock.calls.map(([{ key }]) => key) + +describe('sweepStoragePrefix over a provider listing', () => { + beforeEach(() => { + setUploadsConfig({ USE_S3_STORAGE: true, S3_CONFIG: { bucket: 'bucket', region: 'r' } }) + mockListS3ObjectKeyPage.mockReset() + storageServiceMockFns.mockDeleteFile.mockReset().mockResolvedValue(undefined) + }) + + it('continues after a short page that is not the last', async () => { + mockListS3ObjectKeyPage.mockResolvedValueOnce({ + keys: ['project/p1/a.txt'], + nextContinuationToken: 'next', + }) + + expect(await sweepStoragePrefix('project/p1/', 'project', outbox)).toBe('continue') + expect(deletedKeys()).toEqual(['project/p1/a.txt']) + }) + + it('follows an empty page that is not the last to the objects after it', async () => { + mockListS3ObjectKeyPage + .mockResolvedValueOnce({ keys: [], nextContinuationToken: 'next' }) + .mockResolvedValueOnce({ keys: ['project/p1/a.txt'] }) + + expect(await sweepStoragePrefix('project/p1/', 'project', outbox)).toBe('done') + expect(mockListS3ObjectKeyPage.mock.calls[1]?.[0]).toMatchObject({ continuationToken: 'next' }) + expect(deletedKeys()).toEqual(['project/p1/a.txt']) + }) + + it('stalls when its listing stops advancing without removing anything', async () => { + mockListS3ObjectKeyPage.mockResolvedValue({ keys: [], nextContinuationToken: 'stuck' }) + + expect(await sweepStoragePrefix('project/p1/', 'project', outbox)).toBe('stalled') + expect(mockListS3ObjectKeyPage).toHaveBeenCalledTimes(2) + }) +}) diff --git a/apps/sim/lib/uploads/core/prefix-sweep.ts b/apps/sim/lib/uploads/core/prefix-sweep.ts index 662d65103b9..ef0d7cfab4d 100644 --- a/apps/sim/lib/uploads/core/prefix-sweep.ts +++ b/apps/sim/lib/uploads/core/prefix-sweep.ts @@ -215,3 +215,34 @@ export async function deleteStorageKeysUnderPrefix( } return { deleted, complete: true } } + +/** + * One outbox run's step of deleting every object under `prefix`: deletes one listing page's + * objects, re-listing from the prefix start on the next step. A provider (Azure Blob, GCS, + * S3-compatibles) may return a short or empty page that is not the last, so only `page.done` ends + * the sweep; a non-final page that removes nothing is followed to the next one within the step, so + * objects behind it are still reached. `continue` after a step that removed something with more + * left; `stalled` when no page removed anything before the listing stopped advancing or the + * deadline neared. + */ +export async function sweepStoragePrefix( + prefix: string, + context: StorageContext, + outbox: Pick +): Promise<'done' | 'continue' | 'stalled'> { + let pageToken: string | undefined + for (;;) { + const page = await listStorageKeyPage(prefix, context, outbox.signal, pageToken) + const { deleted, complete } = await deleteStorageKeysUnderPrefix( + prefix, + context, + page.keys, + outbox + ) + if (complete && page.done) return 'done' + if (deleted > 0) return 'continue' + const advanced = page.nextPageToken !== undefined && page.nextPageToken !== pageToken + if (!complete || !advanced || outboxDeadlineNear(outbox)) return 'stalled' + pageToken = page.nextPageToken + } +} diff --git a/apps/sim/lib/workflows/persistence/utils.test.ts b/apps/sim/lib/workflows/persistence/utils.test.ts index 9ac1eb02bb0..20c8bc8cd8f 100644 --- a/apps/sim/lib/workflows/persistence/utils.test.ts +++ b/apps/sim/lib/workflows/persistence/utils.test.ts @@ -817,6 +817,38 @@ describe('Database Helpers', () => { expect(dbChainMockFns.where).toHaveBeenCalledTimes(1) }) + // `state: null` stands in for the row Postgres returns when it withholds the blob for the hinted version. + it('keeps withholding the active state after a by-id read of another version', async () => { + queueActiveVersion('dv-active', buildDeployedState()) + await dbHelpers.loadDeployedWorkflowState('wf-hint', 'workspace-1') + + queueTableRows(schemaMock.workflowDeploymentVersion, [ + { id: 'dv-older', state: buildDeployedState() }, + ]) + await dbHelpers.loadWorkflowDeploymentVersionState('wf-hint', 'dv-older', 'workspace-1') + + queueActiveVersion('dv-active', null) + const deployed = await dbHelpers.loadDeployedWorkflowState('wf-hint', 'workspace-1') + + expect(deployed.deploymentVersionId).toBe('dv-active') + expect(mockSanitizeAgentToolsInBlocks).toHaveBeenCalledTimes(2) + }) + + it('withholds the active state again after a rollback to a version still cached', async () => { + queueActiveVersion('dv-a', buildDeployedState()) + await dbHelpers.loadDeployedWorkflowState('wf-rollback', 'workspace-1') + queueActiveVersion('dv-b', buildDeployedState()) + await dbHelpers.loadDeployedWorkflowState('wf-rollback', 'workspace-1') + queueActiveVersion('dv-a', buildDeployedState()) + await dbHelpers.loadDeployedWorkflowState('wf-rollback', 'workspace-1') + + queueActiveVersion('dv-a', null) + const deployed = await dbHelpers.loadDeployedWorkflowState('wf-rollback', 'workspace-1') + + expect(deployed.deploymentVersionId).toBe('dv-a') + expect(mockSanitizeAgentToolsInBlocks).toHaveBeenCalledTimes(2) + }) + it('invalidateDeployedStateCache(id) forces a rebuild on the next call', async () => { queueActiveVersion('dv-inv', buildDeployedState()) queueActiveVersion('dv-inv', buildDeployedState()) diff --git a/apps/sim/lib/workflows/persistence/utils.ts b/apps/sim/lib/workflows/persistence/utils.ts index fda242c5d20..86e0862dfab 100644 --- a/apps/sim/lib/workflows/persistence/utils.ts +++ b/apps/sim/lib/workflows/persistence/utils.ts @@ -146,9 +146,12 @@ const deployedStateCache = new LRUCache< }) /** - * The version id each workflow last cached, so the active-version read can ask - * Postgres to withhold a `state` this process already holds. Only a hint: a - * stale or evicted entry costs a re-read, never a wrong answer. + * The active version id each workflow last resolved to, so the active-version + * read can ask Postgres to withhold a `state` this process already holds. Only + * {@link loadDeployedWorkflowState} writes it: a by-id read of an older version + * (an execution or webhook admitted before a cutover) must not repoint it away + * from the active one. Only a hint: a stale or evicted entry costs a re-read, + * never a wrong answer. */ const cachedVersionByWorkflow = new LRUCache({ max: DEPLOYED_STATE_CACHE_MAX_ENTRIES, @@ -272,7 +275,6 @@ export async function materializeDeploymentState( if (options.cache === false) return deployedState /** The cached entry is shared, so each caller gets its own copy. */ deployedStateCache.set(version.id, { workflowId, state: deployedState }) - cachedVersionByWorkflow.set(workflowId, version.id) return structuredClone(deployedState) } @@ -311,20 +313,19 @@ export async function loadDeployedWorkflowState( throw new NoActiveDeploymentError(workflowId) } - if (active.id === cachedVersionId && active.state == null) { - // Served from the cache, or re-read by id if the entry was evicted since the hint. - return loadWorkflowDeploymentVersionState(workflowId, active.id, providedWorkspaceId) - } - - if (!active.state) { - throw new NoActiveDeploymentError(workflowId) - } - - return materializeDeploymentState( - workflowId, - active, - await resolveWorkspaceId(workflowId, providedWorkspaceId) - ) + /** Postgres withheld the state this process already caches for the active version. */ + const withheld = active.id === cachedVersionId && active.state == null + if (!withheld && !active.state) throw new NoActiveDeploymentError(workflowId) + const deployedState = withheld + ? // Served from the cache, or re-read by id if the entry was evicted since the hint. + await loadWorkflowDeploymentVersionState(workflowId, active.id, providedWorkspaceId) + : await materializeDeploymentState( + workflowId, + active, + await resolveWorkspaceId(workflowId, providedWorkspaceId) + ) + cachedVersionByWorkflow.set(workflowId, active.id) + return deployedState } catch (error) { // An undeployed workflow is an outcome each caller handles, not a load failure. if (!(error instanceof NoActiveDeploymentError)) {