fix(knowledge): make document uploads durable

Author: TheodoreSpeaks

apps/sim/app/api/v2/knowledge/[id]/documents/route.test.ts

@@ -17,7 +17,6 @@ const {
   mockUploadDocument,
   mockReadFormData,
   mockReadFile,
-  mockUploadWorkspaceFile,
   mockPlatformUploaded,
   mockCapture,
   mockIsPayloadSizeLimitError,
@@ -26,7 +25,6 @@ const {
   mockUploadDocument: vi.fn(),
   mockReadFormData: vi.fn(),
   mockReadFile: vi.fn(),
-  mockUploadWorkspaceFile: vi.fn(),
   mockPlatformUploaded: vi.fn(),
   mockCapture: vi.fn(),
   mockIsPayloadSizeLimitError: vi.fn(),
@@ -58,10 +56,6 @@ vi.mock('@/lib/core/utils/stream-limits', () => ({
   readFileToBufferWithLimit: mockReadFile,
 }))
 
-vi.mock('@/lib/uploads/contexts/workspace', () => ({
-  uploadWorkspaceFile: mockUploadWorkspaceFile,
-}))
-
 vi.mock('@/lib/core/telemetry', () => ({
   PlatformEvents: { knowledgeBaseDocumentsUploaded: mockPlatformUploaded },
 }))
@@ -102,13 +96,11 @@ describe('POST /api/v2/knowledge/[id]/documents', () => {
       knowledgeBaseId: 'kb-1',
       knowledgeBaseName: 'Support docs',
       workspaceId: WORKSPACE_ID,
-      storageActorUserId: 'user-1',
     })
     const formData = new FormData()
     formData.set('file', new File(['hello'], 'support.txt', { type: 'text/plain' }))
     mockReadFormData.mockResolvedValue(formData)
     mockReadFile.mockResolvedValue(Buffer.from('hello'))
-    mockUploadWorkspaceFile.mockResolvedValue({ url: 's3://workspace/support.txt' })
     mockUploadDocument.mockResolvedValue({
       created: true,
       document: {
@@ -141,21 +133,14 @@ describe('POST /api/v2/knowledge/[id]/documents', () => {
       input: { knowledgeBaseId: 'kb-1', assertedWorkspaceId: WORKSPACE_ID },
       request,
     })
-    expect(mockUploadWorkspaceFile).toHaveBeenCalledWith(
-      WORKSPACE_ID,
-      'user-1',
-      Buffer.from('hello'),
-      'support.txt',
-      'text/plain'
-    )
     expect(mockUploadDocument).toHaveBeenCalledWith({
       principal: PRINCIPAL,
       input: {
         knowledgeBaseId: 'kb-1',
         assertedWorkspaceId: WORKSPACE_ID,
-        document: {
+        file: {
+          buffer: Buffer.from('hello'),
           filename: 'support.txt',
-          fileUrl: 's3://workspace/support.txt',
           fileSize: 5,
           mimeType: 'text/plain',
         },
@@ -194,7 +179,6 @@ describe('POST /api/v2/knowledge/[id]/documents', () => {
       error: { code: 'USAGE_LIMIT_EXCEEDED', message: 'Upgrade required' },
     })
     expect(mockReadFormData).not.toHaveBeenCalled()
-    expect(mockUploadWorkspaceFile).not.toHaveBeenCalled()
     expect(mockUploadDocument).not.toHaveBeenCalled()
   })
 
@@ -214,7 +198,7 @@ describe('POST /api/v2/knowledge/[id]/documents', () => {
     expect(mockCapture).not.toHaveBeenCalled()
   })
 
-  it('preserves the malformed multipart envelope without transferring storage', async () => {
+  it('preserves the malformed multipart envelope without entering the upload operation', async () => {
     mockReadFormData.mockRejectedValueOnce(new Error('multipart boundary missing'))
 
     const response = await POST(buildRequest(), { params: Promise.resolve({ id: 'kb-1' }) })
@@ -223,12 +207,11 @@ describe('POST /api/v2/knowledge/[id]/documents', () => {
     expect(await response.json()).toEqual({
       error: { code: 'BAD_REQUEST', message: 'Request body must be valid multipart form data' },
     })
-    expect(mockUploadWorkspaceFile).not.toHaveBeenCalled()
     expect(mockUploadDocument).not.toHaveBeenCalled()
     expect(mockPlatformUploaded).not.toHaveBeenCalled()
   })
 
-  it('preserves bounded multipart rejection and stops before storage transfer', async () => {
+  it('preserves bounded multipart rejection and stops before the upload operation', async () => {
     const error = new Error('knowledge document upload body exceeds maximum size')
     mockReadFormData.mockRejectedValueOnce(error)
     mockIsPayloadSizeLimitError.mockImplementation((candidate: unknown) => candidate === error)
@@ -239,11 +222,10 @@ describe('POST /api/v2/knowledge/[id]/documents', () => {
     expect(await response.json()).toEqual({
       error: { code: 'PAYLOAD_TOO_LARGE', message: error.message },
     })
-    expect(mockUploadWorkspaceFile).not.toHaveBeenCalled()
     expect(mockUploadDocument).not.toHaveBeenCalled()
   })
 
-  it('requires a file form field before storage transfer', async () => {
+  it('requires a file form field before the upload operation', async () => {
     mockReadFormData.mockResolvedValueOnce(new FormData())
 
     const response = await POST(buildRequest(), { params: Promise.resolve({ id: 'kb-1' }) })
@@ -252,7 +234,7 @@ describe('POST /api/v2/knowledge/[id]/documents', () => {
     expect(await response.json()).toEqual({
       error: { code: 'BAD_REQUEST', message: 'file form field is required' },
     })
-    expect(mockUploadWorkspaceFile).not.toHaveBeenCalled()
+    expect(mockUploadDocument).not.toHaveBeenCalled()
   })
 
   it('preserves the exact file-size rejection before reading file bytes', async () => {
@@ -269,7 +251,7 @@ describe('POST /api/v2/knowledge/[id]/documents', () => {
       error: { code: 'PAYLOAD_TOO_LARGE', message: 'File size exceeds 100MB limit (100.00MB)' },
     })
     expect(mockReadFile).not.toHaveBeenCalled()
-    expect(mockUploadWorkspaceFile).not.toHaveBeenCalled()
+    expect(mockUploadDocument).not.toHaveBeenCalled()
   })
 
   it('preserves unsupported file-type validation before reading file bytes', async () => {
@@ -286,24 +268,24 @@ describe('POST /api/v2/knowledge/[id]/documents', () => {
       error: { code: 'UNSUPPORTED_MEDIA_TYPE', message: expectedMessage },
     })
     expect(mockReadFile).not.toHaveBeenCalled()
-    expect(mockUploadWorkspaceFile).not.toHaveBeenCalled()
+    expect(mockUploadDocument).not.toHaveBeenCalled()
   })
 
-  it('does not register or emit effects when storage transfer fails', async () => {
-    mockUploadWorkspaceFile.mockRejectedValueOnce(new Error('storage unavailable'))
+  it('does not emit effects when the upload operation fails', async () => {
+    mockUploadDocument.mockRejectedValueOnce(new Error('storage unavailable'))
 
     const response = await POST(buildRequest(), { params: Promise.resolve({ id: 'kb-1' }) })
 
     expect(response.status).toBe(500)
     expect(await response.json()).toEqual({
       error: { code: 'INTERNAL_ERROR', message: 'Internal server error' },
     })
-    expect(mockUploadDocument).not.toHaveBeenCalled()
+    expect(mockUploadDocument).toHaveBeenCalledOnce()
     expect(mockPlatformUploaded).not.toHaveBeenCalled()
     expect(mockCapture).not.toHaveBeenCalled()
   })
 
-  it('preserves application authorization errors after storage transfer', async () => {
+  it('preserves final application authorization errors', async () => {
     mockUploadDocument.mockRejectedValueOnce(
       new OrchestrationError('forbidden', 'Insufficient workspace permissions')
     )
@@ -314,7 +296,7 @@ describe('POST /api/v2/knowledge/[id]/documents', () => {
     expect(await response.json()).toEqual({
       error: { code: 'FORBIDDEN', message: 'Insufficient workspace permissions' },
     })
-    expect(mockUploadWorkspaceFile).toHaveBeenCalledOnce()
+    expect(mockUploadDocument).toHaveBeenCalledOnce()
     expect(mockPlatformUploaded).not.toHaveBeenCalled()
     expect(mockCapture).not.toHaveBeenCalled()
   })

apps/sim/app/api/v2/knowledge/[id]/documents/route.ts

@@ -26,7 +26,6 @@ import {
 import { knowledgeOperations } from '@/lib/knowledge/application/operations'
 import { KnowledgeDocumentUnsupportedMediaTypeError } from '@/lib/knowledge/application/upload-sessions'
 import { captureServerEvent } from '@/lib/posthog/server'
-import { uploadWorkspaceFile } from '@/lib/uploads/contexts/workspace'
 import { MAX_KNOWLEDGE_DOCUMENT_FILE_SIZE } from '@/lib/uploads/shared/types'
 import { validateFileType } from '@/lib/uploads/utils/validation'
 import { serializeDate } from '@/app/api/v1/knowledge/utils'
@@ -138,20 +137,12 @@ export const POST = defineV2BodyLifecycleRoute({
     })
     return { file: rawFile, buffer, contentType }
   },
-  transfer: ({ admission, body }) =>
-    uploadWorkspaceFile(
-      admission.workspaceId,
-      admission.storageActorUserId,
-      body.buffer,
-      body.file.name,
-      body.contentType
-    ),
-  mapInput: ({ parsed, body, transfer }) => ({
+  mapInput: ({ parsed, body }) => ({
     knowledgeBaseId: parsed.params.id,
     assertedWorkspaceId: parsed.query.workspaceId,
-    document: {
+    file: {
+      buffer: body.buffer,
       filename: body.file.name,
-      fileUrl: transfer.url,
       fileSize: body.file.size,
       mimeType: body.contentType,
     },

apps/sim/app/api/webhooks/outbox/process/route.ts

@@ -9,6 +9,7 @@ import { billingOutboxHandlers } from '@/lib/billing/webhooks/outbox-handlers'
 import { processOutboxEvents } from '@/lib/core/outbox/service'
 import { generateRequestId } from '@/lib/core/utils/request'
 import { withRouteHandler } from '@/lib/core/utils/with-route-handler'
+import { knowledgeDocumentProcessingOutboxHandlers } from '@/lib/knowledge/documents/processing-outbox-handler'
 import { workflowDeploymentOutboxHandlers } from '@/lib/workflows/deployment-outbox'
 import { invitationMigrationOutboxHandlers } from '@/lib/workspaces/admin-move'
 import { reapStaleBackgroundWork } from '@/ee/workspace-forking/lib/background-work/store'
@@ -23,6 +24,7 @@ const handlers = {
   ...membershipBillingOutboxHandlers,
   ...enterpriseIssuanceOutboxHandlers,
   ...invitationMigrationOutboxHandlers,
+  ...knowledgeDocumentProcessingOutboxHandlers,
   ...workflowDeploymentOutboxHandlers,
 } as const
 

apps/sim/lib/api/server/routes/v2-body-lifecycle-route.test.ts

@@ -43,7 +43,7 @@ const contract = defineRouteContract({
 
 class StageRejection extends Error {}
 
-type RejectableStage = 'admission' | 'body' | 'transfer' | 'application' | 'presenter' | 'effects'
+type RejectableStage = 'admission' | 'body' | 'application' | 'presenter' | 'effects'
 
 let rejectedStage: RejectableStage | null = null
 
@@ -80,11 +80,10 @@ function buildHandler() {
       rejectAt('body')
       return { bytes: Buffer.from('body') }
     },
-    async transfer({ admission }) {
-      rejectAt('transfer')
-      return { url: `stored://${admission.canonicalWorkspaceId}` }
-    },
-    mapInput: ({ parsed, transfer }) => ({ id: parsed.params.id, url: transfer.url }),
+    mapInput: ({ parsed, admission }) => ({
+      id: parsed.params.id,
+      url: `stored://${admission.canonicalWorkspaceId}`,
+    }),
     useCase: {
       operation,
       async execute({ input }) {
@@ -161,7 +160,6 @@ describe('defineV2BodyLifecycleRoute', () => {
         errorPolicy: { render: () => null },
         admission: { mapInput: () => ({}), useCase },
         readBody: async () => Buffer.alloc(0),
-        transfer: async () => ({ url: 'stored://item-1' }),
         mapInput: () => ({}),
         useCase,
         present: () => ({ data: { id: 'item-1' } }),
@@ -183,7 +181,6 @@ describe('defineV2BodyLifecycleRoute', () => {
       'contract',
       'admission',
       'body',
-      'transfer',
       'application',
       'presenter',
       'effects',
@@ -260,24 +257,20 @@ describe('defineV2BodyLifecycleRoute', () => {
     ])
   })
 
-  it.each<RejectableStage>([
-    'admission',
-    'body',
-    'transfer',
-    'application',
-    'presenter',
-    'effects',
-  ])('renders typed %s rejection without entering later phases', async (stage) => {
-    rejectedStage = stage
+  it.each<RejectableStage>(['admission', 'body', 'application', 'presenter', 'effects'])(
+    'renders typed %s rejection without entering later phases',
+    async (stage) => {
+      rejectedStage = stage
 
-    const response = await buildHandler()(buildRequest(), context())
+      const response = await buildHandler()(buildRequest(), context())
 
-    expect(response.status).toBe(409)
-    expect(await response.json()).toEqual({
-      error: { code: 'CONFLICT', message: `${stage} rejected` },
-    })
-    expect(mocks.order.at(-1)).toBe(stage)
-  })
+      expect(response.status).toBe(409)
+      expect(await response.json()).toEqual({
+        error: { code: 'CONFLICT', message: `${stage} rejected` },
+      })
+      expect(mocks.order.at(-1)).toBe(stage)
+    }
+  )
 
   it.each(['authentication', 'rollout', 'rate_limit'] as const)(
     'maps %s infrastructure failures to service unavailable',

apps/sim/lib/api/server/routes/v2-body-lifecycle-route.ts

@@ -41,18 +41,13 @@ interface V2BodyLifecycleContext<C extends JsonApiRouteContract, A> {
   admission: A
 }
 
-interface V2BodyLifecycleTransferContext<C extends JsonApiRouteContract, A, B>
+interface V2BodyLifecycleInputContext<C extends JsonApiRouteContract, A, B>
   extends V2BodyLifecycleContext<C, A> {
   body: B
 }
 
-interface V2BodyLifecycleInputContext<C extends JsonApiRouteContract, A, B, T>
-  extends V2BodyLifecycleTransferContext<C, A, B> {
-  transfer: T
-}
-
-interface V2BodyLifecycleSuccessContext<C extends JsonApiRouteContract, A, B, T, I, R>
-  extends V2BodyLifecycleInputContext<C, A, B, T> {
+interface V2BodyLifecycleSuccessContext<C extends JsonApiRouteContract, A, B, I, R>
+  extends V2BodyLifecycleInputContext<C, A, B> {
   input: I
   result: R
 }
@@ -63,7 +58,6 @@ interface V2BodyLifecycleRouteOptions<
   AI,
   A,
   B,
-  T,
   I,
   R,
 > {
@@ -75,12 +69,11 @@ interface V2BodyLifecycleRouteOptions<
   parseOptions?: Omit<ParseRequestOptions, 'validationErrorResponse'>
   admission: V2BodyLifecycleAdmission<O, C, AI, A>
   readBody(context: V2BodyLifecycleContext<C, A>): Promise<B>
-  transfer(context: V2BodyLifecycleTransferContext<C, A, B>): Promise<T>
-  mapInput(context: V2BodyLifecycleInputContext<C, A, B, T>): I
+  mapInput(context: V2BodyLifecycleInputContext<C, A, B>): I
   useCase: OperationUseCase<NoInfer<O>, I, R>
   present(result: R): ContractJsonResponse<C> | Promise<ContractJsonResponse<C>>
   onSuccess?(
-    context: V2BodyLifecycleSuccessContext<C, A, B, T, NoInfer<I>, NoInfer<R>>
+    context: V2BodyLifecycleSuccessContext<C, A, B, NoInfer<I>, NoInfer<R>>
   ): void | Promise<void>
 }
 
@@ -95,10 +88,9 @@ export function defineV2BodyLifecycleRoute<
   AI,
   A,
   B,
-  T,
   I,
   R,
->(options: V2BodyLifecycleRouteOptions<C, O, AI, A, B, T, I, R>): JsonNextRouteHandler {
+>(options: V2BodyLifecycleRouteOptions<C, O, AI, A, B, I, R>): JsonNextRouteHandler {
   if (options.contract.body) {
     throw new Error(
       `${options.contract.method} ${options.contract.path} must omit its body schema so admission precedes body reads`
@@ -153,8 +145,7 @@ export function defineV2BodyLifecycleRoute<
         })
         const lifecycleContext = { request, principal, parsed: parsed.data, admission }
         const body = await options.readBody(lifecycleContext)
-        const transfer = await options.transfer({ ...lifecycleContext, body })
-        const inputContext = { ...lifecycleContext, body, transfer }
+        const inputContext = { ...lifecycleContext, body }
         const input = options.mapInput(inputContext)
         const result = await options.useCase.execute({ principal, input, request })
         const responseBody = options.contract.response.schema.parse(await options.present(result))