diff --git a/apps/docs/openapi-v2-knowledge.json b/apps/docs/openapi-v2-knowledge.json index 38a6d2ed2f5..f9c45723d60 100644 --- a/apps/docs/openapi-v2-knowledge.json +++ b/apps/docs/openapi-v2-knowledge.json @@ -781,6 +781,179 @@ ] } }, + "/api/v2/knowledge/{id}/documents/uploads": { + "parameters": [ + { + "$ref": "#/components/parameters/KnowledgeBaseId" + } + ], + "post": { + "operationId": "createKnowledgeDocumentUpload", + "summary": "Create Document Upload", + "description": "Create a stateless multipart upload session for a knowledge document. Write access, billing, usage, file type, file size, and workspace storage are checked before provider storage is allocated. The signed upload token binds the caller, workspace, knowledge base, filename, content type, byte size, provider, and knowledge-document purpose. Files may be up to 100 MB.", + "tags": ["Knowledge Bases"], + "x-codeSamples": [ + { + "label": "cURL", + "lang": "bash", + "source": "curl -X POST \\\n \"https://www.sim.ai/api/v2/knowledge/{id}/documents/uploads\" \\\n -H \"X-API-Key: YOUR_API_KEY\" \\\n -H \"Content-Type: application/json\" \\\n -d '{\"workspaceId\":\"YOUR_WORKSPACE_ID\",\"name\":\"guide.pdf\",\"contentType\":\"application/pdf\",\"size\":248913}'" + } + ], + "requestBody": { + "required": true, + "description": "Metadata for the document that will be uploaded through signed part URLs.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/CreateDocumentUploadBody" + } + } + } + }, + "responses": { + "201": { + "description": "The multipart upload session and its signed control-plane token.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/DocumentUploadEnvelope" + } + } + } + }, + "400": { "$ref": "#/components/responses/BadRequest" }, + "401": { "$ref": "#/components/responses/Unauthorized" }, + "402": { "$ref": "#/components/responses/UsageLimitExceeded" }, + "403": { "$ref": "#/components/responses/Forbidden" }, + "404": { "$ref": "#/components/responses/NotFound" }, + "413": { "$ref": "#/components/responses/PayloadTooLarge" }, + "415": { "$ref": "#/components/responses/UnsupportedMediaType" }, + "429": { "$ref": "#/components/responses/RateLimited" }, + "500": { "$ref": "#/components/responses/InternalError" } + } + } + }, + "/api/v2/knowledge/{id}/documents/uploads/{uploadId}": { + "parameters": [ + { "$ref": "#/components/parameters/KnowledgeBaseId" }, + { "$ref": "#/components/parameters/UploadId" }, + { "$ref": "#/components/parameters/UploadTokenHeader" } + ], + "delete": { + "operationId": "abortKnowledgeDocumentUpload", + "summary": "Abort Document Upload", + "description": "Abort an incomplete knowledge-document upload and discard its provider parts. Aborting an already aborted session is safe.", + "tags": ["Knowledge Bases"], + "parameters": [{ "$ref": "#/components/parameters/WorkspaceIdQuery" }], + "responses": { + "200": { + "description": "The aborted upload session.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/DocumentUploadEnvelope" + } + } + } + }, + "400": { "$ref": "#/components/responses/BadRequest" }, + "401": { "$ref": "#/components/responses/Unauthorized" }, + "403": { "$ref": "#/components/responses/Forbidden" }, + "404": { "$ref": "#/components/responses/NotFound" }, + "409": { "$ref": "#/components/responses/Conflict" }, + "429": { "$ref": "#/components/responses/RateLimited" }, + "500": { "$ref": "#/components/responses/InternalError" } + } + } + }, + "/api/v2/knowledge/{id}/documents/uploads/{uploadId}/parts": { + "parameters": [ + { "$ref": "#/components/parameters/KnowledgeBaseId" }, + { "$ref": "#/components/parameters/UploadId" }, + { "$ref": "#/components/parameters/UploadTokenHeader" } + ], + "post": { + "operationId": "createKnowledgeDocumentUploadPartUrls", + "summary": "Create Document Upload Part URLs", + "description": "Issue short-lived signed PUT URLs for up to 100 part numbers. PUT each byte range directly to the returned URL with the returned headers.", + "tags": ["Knowledge Bases"], + "parameters": [{ "$ref": "#/components/parameters/WorkspaceIdQuery" }], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/CreatePartUrlsBody" + } + } + } + }, + "responses": { + "200": { + "description": "Signed URLs for the requested parts.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/PartUrlsEnvelope" + } + } + } + }, + "400": { "$ref": "#/components/responses/BadRequest" }, + "401": { "$ref": "#/components/responses/Unauthorized" }, + "403": { "$ref": "#/components/responses/Forbidden" }, + "404": { "$ref": "#/components/responses/NotFound" }, + "409": { "$ref": "#/components/responses/Conflict" }, + "429": { "$ref": "#/components/responses/RateLimited" }, + "500": { "$ref": "#/components/responses/InternalError" } + } + } + }, + "/api/v2/knowledge/{id}/documents/uploads/{uploadId}/complete": { + "parameters": [ + { "$ref": "#/components/parameters/KnowledgeBaseId" }, + { "$ref": "#/components/parameters/UploadId" }, + { "$ref": "#/components/parameters/UploadTokenHeader" } + ], + "post": { + "operationId": "completeKnowledgeDocumentUpload", + "summary": "Complete Document Upload", + "description": "Verify and assemble all parts, record knowledge-base storage ownership, create the knowledge document, and queue asynchronous processing. Repeating the same completion is idempotent and returns the same document. It never registers a general workspace file.", + "tags": ["Knowledge Bases"], + "parameters": [{ "$ref": "#/components/parameters/WorkspaceIdQuery" }], + "requestBody": { + "required": true, + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/CompleteUploadBody" + } + } + } + }, + "responses": { + "200": { + "description": "The completed upload and queued knowledge document.", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/DocumentUploadEnvelope" + } + } + } + }, + "400": { "$ref": "#/components/responses/BadRequest" }, + "401": { "$ref": "#/components/responses/Unauthorized" }, + "402": { "$ref": "#/components/responses/UsageLimitExceeded" }, + "403": { "$ref": "#/components/responses/Forbidden" }, + "404": { "$ref": "#/components/responses/NotFound" }, + "409": { "$ref": "#/components/responses/Conflict" }, + "413": { "$ref": "#/components/responses/PayloadTooLarge" }, + "429": { "$ref": "#/components/responses/RateLimited" }, + "500": { "$ref": "#/components/responses/InternalError" } + } + } + }, "/api/v2/knowledge/{id}/documents/{documentId}": { "parameters": [ { @@ -956,6 +1129,17 @@ "example": "b2d4f8a0-1c3e-4a5b-9d7c-2e6f0a8b4c12" } }, + "UploadId": { + "name": "uploadId", + "in": "path", + "required": true, + "description": "The upload session identifier returned when the upload was created.", + "schema": { + "type": "string", + "minLength": 1, + "example": "upload_01K0M9J4W6K4J3T73Q8W2NYR9P" + } + }, "WorkspaceIdQuery": { "name": "workspaceId", "in": "query", @@ -966,6 +1150,16 @@ "minLength": 1, "example": "a91c4b2e-6d3f-4e8a-b5c7-0d9e2f1a8c64" } + }, + "UploadTokenHeader": { + "name": "upload-token", + "in": "header", + "required": true, + "description": "The signed token returned when this upload was created. It is bound to the caller and all upload metadata.", + "schema": { + "type": "string", + "minLength": 1 + } } }, "headers": { @@ -1231,6 +1425,212 @@ } } }, + "CreateDocumentUploadBody": { + "type": "object", + "additionalProperties": false, + "required": ["workspaceId", "name", "contentType", "size"], + "properties": { + "workspaceId": { + "type": "string", + "minLength": 1, + "description": "Workspace that owns the knowledge base." + }, + "name": { + "type": "string", + "minLength": 1, + "maxLength": 255, + "description": "Filename recorded on the knowledge document." + }, + "contentType": { + "type": "string", + "minLength": 1, + "maxLength": 255, + "description": "Supported MIME type for the document." + }, + "size": { + "type": "integer", + "minimum": 1, + "maximum": 104857600, + "description": "Exact file size in bytes." + }, + "tag1": { "type": "string", "maxLength": 1000 }, + "tag2": { "type": "string", "maxLength": 1000 }, + "tag3": { "type": "string", "maxLength": 1000 }, + "tag4": { "type": "string", "maxLength": 1000 }, + "tag5": { "type": "string", "maxLength": 1000 }, + "tag6": { "type": "string", "maxLength": 1000 }, + "tag7": { "type": "string", "maxLength": 1000 }, + "processingOptions": { + "type": "object", + "additionalProperties": false, + "properties": { + "recipe": { "type": "string", "maxLength": 255 }, + "lang": { "type": "string", "maxLength": 35 } + }, + "description": "Optional processing recipe and language, bound into the signed upload state." + } + } + }, + "DocumentUpload": { + "type": "object", + "required": [ + "id", + "knowledgeBaseId", + "status", + "name", + "contentType", + "size", + "partSize", + "partCount", + "uploadToken", + "expiresAt", + "error", + "document" + ], + "properties": { + "id": { + "type": "string", + "description": "Upload session identifier." + }, + "knowledgeBaseId": { + "type": "string", + "description": "Knowledge base that will own the document." + }, + "status": { + "type": "string", + "enum": ["uploading", "finalizing", "completed", "failed", "aborted", "expired"] + }, + "name": { + "type": "string" + }, + "contentType": { + "type": "string" + }, + "size": { + "type": "integer", + "minimum": 1 + }, + "partSize": { + "type": "integer", + "minimum": 1 + }, + "partCount": { + "type": "integer", + "minimum": 1 + }, + "uploadToken": { + "type": "string", + "minLength": 1, + "description": "Signed token required for part URLs, completion, and abort." + }, + "expiresAt": { + "type": "string", + "format": "date-time" + }, + "error": { + "type": ["string", "null"] + }, + "document": { + "oneOf": [{ "$ref": "#/components/schemas/DocumentSummary" }, { "type": "null" }], + "description": "The queued document after completion; null while uploading or after abort." + } + } + }, + "DocumentUploadEnvelope": { + "type": "object", + "required": ["data"], + "properties": { + "data": { + "$ref": "#/components/schemas/DocumentUpload" + } + } + }, + "CreatePartUrlsBody": { + "type": "object", + "additionalProperties": false, + "required": ["partNumbers"], + "properties": { + "partNumbers": { + "type": "array", + "minItems": 1, + "maxItems": 100, + "items": { + "type": "integer", + "minimum": 1 + } + } + } + }, + "UploadPartUrl": { + "type": "object", + "required": ["partNumber", "url", "headers", "expiresAt"], + "properties": { + "partNumber": { + "type": "integer", + "minimum": 1 + }, + "url": { + "type": "string", + "format": "uri" + }, + "headers": { + "type": "object", + "additionalProperties": { + "type": "string" + } + }, + "expiresAt": { + "type": "string", + "format": "date-time" + } + } + }, + "PartUrlsEnvelope": { + "type": "object", + "required": ["data"], + "properties": { + "data": { + "type": "object", + "required": ["parts"], + "properties": { + "parts": { + "type": "array", + "maxItems": 100, + "items": { + "$ref": "#/components/schemas/UploadPartUrl" + } + } + } + } + } + }, + "CompleteUploadBody": { + "type": "object", + "additionalProperties": false, + "required": ["parts"], + "properties": { + "parts": { + "type": "array", + "minItems": 1, + "maxItems": 640, + "items": { + "type": "object", + "additionalProperties": false, + "required": ["partNumber"], + "properties": { + "partNumber": { + "type": "integer", + "minimum": 1 + }, + "etag": { + "type": "string", + "minLength": 1 + } + } + } + } + } + }, "DocumentSummary": { "type": "object", "description": "Summary representation of a document, returned in list operations and as the upload acknowledgement.", diff --git a/apps/sim/app/api/files/uploads/[uploadId]/complete/route.ts b/apps/sim/app/api/files/uploads/[uploadId]/complete/route.ts index 560357745da..ce66561f854 100644 --- a/apps/sim/app/api/files/uploads/[uploadId]/complete/route.ts +++ b/apps/sim/app/api/files/uploads/[uploadId]/complete/route.ts @@ -32,6 +32,7 @@ export const POST = withRouteHandler(async (request: NextRequest, context: Uploa uploadId: parsed.data.params.uploadId, workspaceId, userId: user, + purpose: 'workspace_file', uploadToken: parsed.data.headers['upload-token'], }) const metadata = upload.metadata as { folderId?: string | null } diff --git a/apps/sim/app/api/files/uploads/[uploadId]/parts/route.ts b/apps/sim/app/api/files/uploads/[uploadId]/parts/route.ts index a01a9c4ca9d..80e0edd9e12 100644 --- a/apps/sim/app/api/files/uploads/[uploadId]/parts/route.ts +++ b/apps/sim/app/api/files/uploads/[uploadId]/parts/route.ts @@ -29,6 +29,7 @@ export const POST = withRouteHandler(async (request: NextRequest, context: Uploa uploadId: parsed.data.params.uploadId, workspaceId, userId: user, + purpose: 'workspace_file', uploadToken: parsed.data.headers['upload-token'], }) const parts = await createUploadPartUrls({ diff --git a/apps/sim/app/api/files/uploads/[uploadId]/route.ts b/apps/sim/app/api/files/uploads/[uploadId]/route.ts index ffda91c0c44..f33f3bb3004 100644 --- a/apps/sim/app/api/files/uploads/[uploadId]/route.ts +++ b/apps/sim/app/api/files/uploads/[uploadId]/route.ts @@ -27,6 +27,7 @@ export const DELETE = withRouteHandler(async (request: NextRequest, context: Upl uploadId: parsed.data.params.uploadId, workspaceId, userId: user, + purpose: 'workspace_file', uploadToken: parsed.data.headers['upload-token'], }) return NextResponse.json({ data: toV2FileUpload(await abortUploadSession(upload), null) }) diff --git a/apps/sim/app/api/knowledge/[id]/documents/uploads/[uploadId]/complete/route.ts b/apps/sim/app/api/knowledge/[id]/documents/uploads/[uploadId]/complete/route.ts new file mode 100644 index 00000000000..8d0cf3e8bf6 --- /dev/null +++ b/apps/sim/app/api/knowledge/[id]/documents/uploads/[uploadId]/complete/route.ts @@ -0,0 +1,74 @@ +import { type NextRequest, NextResponse } from 'next/server' +import { completeKnowledgeDocumentUploadContract } from '@/lib/api/contracts/knowledge/upload-sessions' +import { parseRequest } from '@/lib/api/server' +import { generateRequestId } from '@/lib/core/utils/request' +import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { completeUploadSession } from '@/lib/uploads/multipart-session/service' +import { uploadSessionErrorResponse } from '@/app/api/files/uploads/utils' +import { + requireKnowledgeDocumentUploadAccess, + requireKnowledgeDocumentUploadActor, + resolveKnowledgeDocumentUploadAttribution, +} from '@/app/api/knowledge/[id]/documents/uploads/utils' +import { + finalizeKnowledgeDocumentUpload, + getOwnedKnowledgeDocumentUpload, + toV2KnowledgeDocumentUpload, +} from '@/app/api/v2/knowledge/[id]/documents/uploads/utils' + +interface KnowledgeDocumentUploadRouteParams { + params: Promise<{ id: string; uploadId: string }> +} + +export const POST = withRouteHandler( + async (request: NextRequest, context: KnowledgeDocumentUploadRouteParams) => { + const actor = await requireKnowledgeDocumentUploadActor() + if (actor instanceof NextResponse) return actor + const parsed = await parseRequest(completeKnowledgeDocumentUploadContract, request, context) + if (!parsed.success) return parsed.response + const { id: knowledgeBaseId, uploadId } = parsed.data.params + const { workspaceId } = parsed.data.query + const access = await requireKnowledgeDocumentUploadAccess({ + knowledgeBaseId, + workspaceId, + userId: actor.id, + }) + if (access instanceof NextResponse) return access + const requestId = generateRequestId() + try { + const upload = getOwnedKnowledgeDocumentUpload({ + knowledgeBaseId, + uploadId, + workspaceId, + userId: actor.id, + uploadToken: parsed.data.headers['upload-token'], + }) + const completed = await completeUploadSession({ + session: upload, + parts: parsed.data.body.parts, + finalize: (claimed) => + finalizeKnowledgeDocumentUpload({ + claimed, + knowledgeBaseId, + knowledgeBaseName: access.knowledgeBase.name, + workspaceId, + userId: actor.id, + resolveAttribution: () => + resolveKnowledgeDocumentUploadAttribution({ workspaceId, userId: actor.id }), + source: 'ui', + requestId, + request, + actorName: actor.name, + actorEmail: actor.email, + }), + }) + return NextResponse.json({ + data: toV2KnowledgeDocumentUpload(completed.session, completed.value), + }) + } catch (error) { + const classified = uploadSessionErrorResponse(error) + if (classified) return classified + throw error + } + } +) diff --git a/apps/sim/app/api/knowledge/[id]/documents/uploads/[uploadId]/parts/route.ts b/apps/sim/app/api/knowledge/[id]/documents/uploads/[uploadId]/parts/route.ts new file mode 100644 index 00000000000..482705e5b62 --- /dev/null +++ b/apps/sim/app/api/knowledge/[id]/documents/uploads/[uploadId]/parts/route.ts @@ -0,0 +1,55 @@ +import { type NextRequest, NextResponse } from 'next/server' +import { createKnowledgeDocumentUploadPartUrlsContract } from '@/lib/api/contracts/knowledge/upload-sessions' +import { parseRequest } from '@/lib/api/server' +import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { createUploadPartUrls } from '@/lib/uploads/multipart-session/service' +import { uploadSessionErrorResponse } from '@/app/api/files/uploads/utils' +import { + requireKnowledgeDocumentUploadAccess, + requireKnowledgeDocumentUploadActor, +} from '@/app/api/knowledge/[id]/documents/uploads/utils' +import { getOwnedKnowledgeDocumentUpload } from '@/app/api/v2/knowledge/[id]/documents/uploads/utils' + +interface KnowledgeDocumentUploadRouteParams { + params: Promise<{ id: string; uploadId: string }> +} + +export const POST = withRouteHandler( + async (request: NextRequest, context: KnowledgeDocumentUploadRouteParams) => { + const actor = await requireKnowledgeDocumentUploadActor() + if (actor instanceof NextResponse) return actor + const parsed = await parseRequest( + createKnowledgeDocumentUploadPartUrlsContract, + request, + context + ) + if (!parsed.success) return parsed.response + const { id: knowledgeBaseId, uploadId } = parsed.data.params + const { workspaceId } = parsed.data.query + const access = await requireKnowledgeDocumentUploadAccess({ + knowledgeBaseId, + workspaceId, + userId: actor.id, + }) + if (access instanceof NextResponse) return access + try { + const upload = getOwnedKnowledgeDocumentUpload({ + knowledgeBaseId, + uploadId, + workspaceId, + userId: actor.id, + uploadToken: parsed.data.headers['upload-token'], + }) + const parts = await createUploadPartUrls({ + session: upload, + partNumbers: parsed.data.body.partNumbers, + localOrigin: request.nextUrl.origin, + }) + return NextResponse.json({ data: { parts } }) + } catch (error) { + const classified = uploadSessionErrorResponse(error) + if (classified) return classified + throw error + } + } +) diff --git a/apps/sim/app/api/knowledge/[id]/documents/uploads/[uploadId]/route.ts b/apps/sim/app/api/knowledge/[id]/documents/uploads/[uploadId]/route.ts new file mode 100644 index 00000000000..bc37c1026c2 --- /dev/null +++ b/apps/sim/app/api/knowledge/[id]/documents/uploads/[uploadId]/route.ts @@ -0,0 +1,50 @@ +import { type NextRequest, NextResponse } from 'next/server' +import { abortKnowledgeDocumentUploadContract } from '@/lib/api/contracts/knowledge/upload-sessions' +import { parseRequest } from '@/lib/api/server' +import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { uploadSessionErrorResponse } from '@/app/api/files/uploads/utils' +import { + requireKnowledgeDocumentUploadAccess, + requireKnowledgeDocumentUploadActor, +} from '@/app/api/knowledge/[id]/documents/uploads/utils' +import { + abortKnowledgeDocumentUpload, + getOwnedKnowledgeDocumentUpload, + toV2KnowledgeDocumentUpload, +} from '@/app/api/v2/knowledge/[id]/documents/uploads/utils' + +interface KnowledgeDocumentUploadRouteParams { + params: Promise<{ id: string; uploadId: string }> +} + +export const DELETE = withRouteHandler( + async (request: NextRequest, context: KnowledgeDocumentUploadRouteParams) => { + const actor = await requireKnowledgeDocumentUploadActor() + if (actor instanceof NextResponse) return actor + const parsed = await parseRequest(abortKnowledgeDocumentUploadContract, request, context) + if (!parsed.success) return parsed.response + const { id: knowledgeBaseId, uploadId } = parsed.data.params + const { workspaceId } = parsed.data.query + const access = await requireKnowledgeDocumentUploadAccess({ + knowledgeBaseId, + workspaceId, + userId: actor.id, + }) + if (access instanceof NextResponse) return access + try { + const upload = getOwnedKnowledgeDocumentUpload({ + knowledgeBaseId, + uploadId, + workspaceId, + userId: actor.id, + uploadToken: parsed.data.headers['upload-token'], + }) + const aborted = await abortKnowledgeDocumentUpload(upload, knowledgeBaseId) + return NextResponse.json({ data: toV2KnowledgeDocumentUpload(aborted, null) }) + } catch (error) { + const classified = uploadSessionErrorResponse(error) + if (classified) return classified + throw error + } + } +) diff --git a/apps/sim/app/api/knowledge/[id]/documents/uploads/route.test.ts b/apps/sim/app/api/knowledge/[id]/documents/uploads/route.test.ts new file mode 100644 index 00000000000..df28c35218f --- /dev/null +++ b/apps/sim/app/api/knowledge/[id]/documents/uploads/route.test.ts @@ -0,0 +1,118 @@ +/** + * @vitest-environment node + */ +import { NextRequest, NextResponse } from 'next/server' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { + mockCreateKnowledgeDocumentUploadSession, + mockRequireKnowledgeDocumentUploadAccess, + mockRequireKnowledgeDocumentUploadActor, + mockRequireKnowledgeDocumentUploadBilling, +} = vi.hoisted(() => ({ + mockCreateKnowledgeDocumentUploadSession: vi.fn(), + mockRequireKnowledgeDocumentUploadAccess: vi.fn(), + mockRequireKnowledgeDocumentUploadActor: vi.fn(), + mockRequireKnowledgeDocumentUploadBilling: vi.fn(), +})) + +vi.mock('@/app/api/knowledge/[id]/documents/uploads/utils', () => ({ + requireKnowledgeDocumentUploadAccess: mockRequireKnowledgeDocumentUploadAccess, + requireKnowledgeDocumentUploadActor: mockRequireKnowledgeDocumentUploadActor, + requireKnowledgeDocumentUploadBilling: mockRequireKnowledgeDocumentUploadBilling, +})) +vi.mock('@/app/api/files/uploads/utils', () => ({ uploadSessionErrorResponse: vi.fn() })) +vi.mock('@/app/api/v2/knowledge/[id]/documents/uploads/utils', () => ({ + createKnowledgeDocumentUploadSession: mockCreateKnowledgeDocumentUploadSession, + toV2KnowledgeDocumentUpload: (session: Record) => ({ + ...session, + name: session.fileName, + contentType: session.contentType, + size: session.fileSize, + expiresAt: '2026-08-05T00:00:00.000Z', + document: null, + }), +})) + +import { POST } from '@/app/api/knowledge/[id]/documents/uploads/route' + +const WORKSPACE_ID = '6fc7631d-88cd-46f8-9f0a-d4764daef7f8' + +function request() { + return POST( + new NextRequest('http://localhost:3000/api/knowledge/kb-1/documents/uploads', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + workspaceId: WORKSPACE_ID, + name: 'guide.pdf', + contentType: 'application/pdf', + size: 1024, + tag1: 'product', + processingOptions: { recipe: 'default', lang: 'en' }, + }), + }), + { params: Promise.resolve({ id: 'kb-1' }) } + ) +} + +describe('POST /api/knowledge/[id]/documents/uploads', () => { + beforeEach(() => { + vi.clearAllMocks() + mockRequireKnowledgeDocumentUploadActor.mockResolvedValue({ id: 'user-1' }) + mockRequireKnowledgeDocumentUploadAccess.mockResolvedValue({ + knowledgeBase: { id: 'kb-1', name: 'Docs', workspaceId: WORKSPACE_ID }, + }) + mockRequireKnowledgeDocumentUploadBilling.mockResolvedValue({ actorUserId: 'user-1' }) + mockCreateKnowledgeDocumentUploadSession.mockResolvedValue({ + id: 'upload-1', + knowledgeBaseId: 'kb-1', + status: 'uploading', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + partSize: 8 * 1024 * 1024, + partCount: 1, + uploadToken: 'token', + error: null, + }) + }) + + it('authorizes and bills before allocating a first-party upload session', async () => { + const response = await request() + + expect(response.status).toBe(201) + expect(mockRequireKnowledgeDocumentUploadAccess).toHaveBeenCalledWith({ + knowledgeBaseId: 'kb-1', + workspaceId: WORKSPACE_ID, + userId: 'user-1', + }) + expect(mockCreateKnowledgeDocumentUploadSession).toHaveBeenCalledWith({ + workspaceId: WORKSPACE_ID, + userId: 'user-1', + knowledgeBaseId: 'kb-1', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + metadata: { + tag1: 'product', + processingOptions: { recipe: 'default', lang: 'en' }, + }, + }) + expect(mockRequireKnowledgeDocumentUploadBilling.mock.invocationCallOrder[0]).toBeLessThan( + mockCreateKnowledgeDocumentUploadSession.mock.invocationCallOrder[0] + ) + }) + + it('does not bill or allocate storage when write access is denied', async () => { + mockRequireKnowledgeDocumentUploadAccess.mockResolvedValue( + NextResponse.json({ error: 'Forbidden' }, { status: 403 }) + ) + + const response = await request() + + expect(response.status).toBe(403) + expect(mockRequireKnowledgeDocumentUploadBilling).not.toHaveBeenCalled() + expect(mockCreateKnowledgeDocumentUploadSession).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/app/api/knowledge/[id]/documents/uploads/route.ts b/apps/sim/app/api/knowledge/[id]/documents/uploads/route.ts new file mode 100644 index 00000000000..09b7aec9e59 --- /dev/null +++ b/apps/sim/app/api/knowledge/[id]/documents/uploads/route.ts @@ -0,0 +1,61 @@ +import { type NextRequest, NextResponse } from 'next/server' +import { createKnowledgeDocumentUploadContract } from '@/lib/api/contracts/knowledge/upload-sessions' +import { parseRequest } from '@/lib/api/server' +import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { validateFileType } from '@/lib/uploads/utils/validation' +import { uploadSessionErrorResponse } from '@/app/api/files/uploads/utils' +import { + requireKnowledgeDocumentUploadAccess, + requireKnowledgeDocumentUploadActor, + requireKnowledgeDocumentUploadBilling, +} from '@/app/api/knowledge/[id]/documents/uploads/utils' +import { + createKnowledgeDocumentUploadSession, + toV2KnowledgeDocumentUpload, +} from '@/app/api/v2/knowledge/[id]/documents/uploads/utils' + +interface KnowledgeDocumentUploadsRouteParams { + params: Promise<{ id: string }> +} + +export const POST = withRouteHandler( + async (request: NextRequest, context: KnowledgeDocumentUploadsRouteParams) => { + const actor = await requireKnowledgeDocumentUploadActor() + if (actor instanceof NextResponse) return actor + const parsed = await parseRequest(createKnowledgeDocumentUploadContract, request, context) + if (!parsed.success) return parsed.response + const { id: knowledgeBaseId } = parsed.data.params + const { workspaceId, name, contentType, size, ...metadata } = parsed.data.body + const access = await requireKnowledgeDocumentUploadAccess({ + knowledgeBaseId, + workspaceId, + userId: actor.id, + }) + if (access instanceof NextResponse) return access + const billing = await requireKnowledgeDocumentUploadBilling({ + workspaceId, + userId: actor.id, + }) + if (billing instanceof NextResponse) return billing + const fileTypeError = validateFileType(name, contentType) + if (fileTypeError) { + return NextResponse.json({ error: fileTypeError.message }, { status: 415 }) + } + try { + const upload = await createKnowledgeDocumentUploadSession({ + workspaceId, + userId: actor.id, + knowledgeBaseId, + fileName: name, + contentType, + fileSize: size, + metadata, + }) + return NextResponse.json({ data: toV2KnowledgeDocumentUpload(upload, null) }, { status: 201 }) + } catch (error) { + const classified = uploadSessionErrorResponse(error) + if (classified) return classified + throw error + } + } +) diff --git a/apps/sim/app/api/knowledge/[id]/documents/uploads/utils.ts b/apps/sim/app/api/knowledge/[id]/documents/uploads/utils.ts new file mode 100644 index 00000000000..450b17ecd0b --- /dev/null +++ b/apps/sim/app/api/knowledge/[id]/documents/uploads/utils.ts @@ -0,0 +1,73 @@ +import { NextResponse } from 'next/server' +import { getSession } from '@/lib/auth' +import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution' +import { + checkAttributedUsageLimits, + resolveBillingAttribution, +} from '@/lib/billing/core/billing-attribution' +import type { KnowledgeBaseAccessResult } from '@/app/api/knowledge/utils' +import { checkKnowledgeBaseWriteAccess } from '@/app/api/knowledge/utils' + +export interface KnowledgeDocumentUploadActor { + id: string + name?: string | null + email?: string | null +} + +export async function requireKnowledgeDocumentUploadActor(): Promise< + KnowledgeDocumentUploadActor | NextResponse +> { + const session = await getSession() + if (!session?.user?.id) { + return NextResponse.json({ error: 'Unauthorized' }, { status: 401 }) + } + return { + id: session.user.id, + name: session.user.name, + email: session.user.email, + } +} + +export async function requireKnowledgeDocumentUploadAccess(params: { + knowledgeBaseId: string + workspaceId: string + userId: string +}): Promise<{ knowledgeBase: KnowledgeBaseAccessResult['knowledgeBase'] } | NextResponse> { + const access = await checkKnowledgeBaseWriteAccess(params.knowledgeBaseId, params.userId) + if (!access.hasAccess) { + return 'notFound' in access && access.notFound + ? NextResponse.json({ error: 'Knowledge base not found' }, { status: 404 }) + : NextResponse.json({ error: 'Forbidden' }, { status: 403 }) + } + if (access.knowledgeBase.workspaceId !== params.workspaceId) { + return NextResponse.json({ error: 'Knowledge base not found' }, { status: 404 }) + } + return { knowledgeBase: access.knowledgeBase } +} + +export async function requireKnowledgeDocumentUploadBilling(params: { + workspaceId: string + userId: string +}): Promise { + const attribution = await resolveKnowledgeDocumentUploadAttribution(params) + const usage = await checkAttributedUsageLimits(attribution) + if (usage.isExceeded) { + return NextResponse.json( + { + error: usage.message || 'Usage limit exceeded. Please upgrade your plan to continue.', + }, + { status: 402 } + ) + } + return attribution +} + +export function resolveKnowledgeDocumentUploadAttribution(params: { + workspaceId: string + userId: string +}): Promise { + return resolveBillingAttribution({ + actorUserId: params.userId, + workspaceId: params.workspaceId, + }) +} diff --git a/apps/sim/app/api/v2/files/uploads/[uploadId]/complete/route.ts b/apps/sim/app/api/v2/files/uploads/[uploadId]/complete/route.ts index 3dfca6ca127..e4504fd952a 100644 --- a/apps/sim/app/api/v2/files/uploads/[uploadId]/complete/route.ts +++ b/apps/sim/app/api/v2/files/uploads/[uploadId]/complete/route.ts @@ -48,6 +48,7 @@ export const POST = withRouteHandler( uploadId, workspaceId, userId, + purpose: 'workspace_file', uploadToken: parsed.data.headers['upload-token'], }) const metadata = session.metadata as { folderId?: string | null } diff --git a/apps/sim/app/api/v2/files/uploads/[uploadId]/parts/route.ts b/apps/sim/app/api/v2/files/uploads/[uploadId]/parts/route.ts index 4272e75796f..c3148698c7b 100644 --- a/apps/sim/app/api/v2/files/uploads/[uploadId]/parts/route.ts +++ b/apps/sim/app/api/v2/files/uploads/[uploadId]/parts/route.ts @@ -45,6 +45,7 @@ export const POST = withRouteHandler( uploadId, workspaceId, userId, + purpose: 'workspace_file', uploadToken: parsed.data.headers['upload-token'], }) const parts = await createUploadPartUrls({ diff --git a/apps/sim/app/api/v2/files/uploads/[uploadId]/route.ts b/apps/sim/app/api/v2/files/uploads/[uploadId]/route.ts index cccb93f3524..dd07baa1d72 100644 --- a/apps/sim/app/api/v2/files/uploads/[uploadId]/route.ts +++ b/apps/sim/app/api/v2/files/uploads/[uploadId]/route.ts @@ -43,6 +43,7 @@ export const DELETE = withRouteHandler( uploadId, workspaceId, userId, + purpose: 'workspace_file', uploadToken: parsed.data.headers['upload-token'], }) const aborted = await abortUploadSession(session) diff --git a/apps/sim/app/api/v2/files/uploads/route.test.ts b/apps/sim/app/api/v2/files/uploads/route.test.ts index 34c934f0183..b70ac9aba3d 100644 --- a/apps/sim/app/api/v2/files/uploads/route.test.ts +++ b/apps/sim/app/api/v2/files/uploads/route.test.ts @@ -65,6 +65,7 @@ describe('POST /api/v2/files/uploads', () => { id: 'upload-1', workspaceId: WORKSPACE_ID, userId: 'user-1', + knowledgeBaseId: null, purpose: 'workspace_file', storageContext: 'workspace', storageKey: `${WORKSPACE_ID}/file.csv`, diff --git a/apps/sim/app/api/v2/knowledge/[id]/documents/route.ts b/apps/sim/app/api/v2/knowledge/[id]/documents/route.ts index 1f513d5f2ed..b6adfff3c07 100644 --- a/apps/sim/app/api/v2/knowledge/[id]/documents/route.ts +++ b/apps/sim/app/api/v2/knowledge/[id]/documents/route.ts @@ -24,6 +24,7 @@ import type { DocumentSortField, SortOrder } from '@/lib/knowledge/documents/typ import { performUploadKnowledgeDocument } from '@/lib/knowledge/orchestration' import type { KnowledgeBaseWithCounts } from '@/lib/knowledge/types' 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 { resolveKnowledgeBase, serializeDate } from '@/app/api/v1/knowledge/utils' import { checkRateLimit, type RateLimitResult } from '@/app/api/v1/middleware' @@ -44,7 +45,7 @@ const logger = createLogger('V2KnowledgeDocumentsAPI') export const dynamic = 'force-dynamic' export const revalidate = 0 -const MAX_FILE_SIZE = 100 * 1024 * 1024 +const MAX_FILE_SIZE = MAX_KNOWLEDGE_DOCUMENT_FILE_SIZE const MAX_MULTIPART_OVERHEAD_BYTES = 1024 * 1024 interface DocumentsRouteParams { diff --git a/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/complete/route.test.ts b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/complete/route.test.ts new file mode 100644 index 00000000000..b923f82a39c --- /dev/null +++ b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/complete/route.test.ts @@ -0,0 +1,175 @@ +/** + * @vitest-environment node + */ +import { NextRequest } from 'next/server' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { + mockCheckRateLimit, + mockCompleteUploadSession, + mockFinalizeKnowledgeDocumentUpload, + mockResolveKnowledgeDocumentUploadAccess, + mockResolveKnowledgeDocumentUploadAttribution, +} = vi.hoisted(() => ({ + mockCheckRateLimit: vi.fn(), + mockCompleteUploadSession: vi.fn(), + mockFinalizeKnowledgeDocumentUpload: vi.fn(), + mockResolveKnowledgeDocumentUploadAccess: vi.fn(), + mockResolveKnowledgeDocumentUploadAttribution: vi.fn(), +})) + +vi.mock('@/app/api/v1/middleware', () => ({ checkRateLimit: mockCheckRateLimit })) +vi.mock('@/app/api/v2/lib/gate', () => ({ + v2ApiGateError: vi.fn().mockResolvedValue(null), +})) +vi.mock('@/lib/uploads/multipart-session/service', () => ({ + completeUploadSession: mockCompleteUploadSession, +})) +vi.mock('@/app/api/v2/knowledge/[id]/documents/uploads/utils', () => ({ + finalizeKnowledgeDocumentUpload: mockFinalizeKnowledgeDocumentUpload, + getOwnedKnowledgeDocumentUpload: vi.fn(() => SESSION), + resolveKnowledgeDocumentUploadAccess: mockResolveKnowledgeDocumentUploadAccess, + resolveKnowledgeDocumentUploadAttribution: mockResolveKnowledgeDocumentUploadAttribution, + toV2KnowledgeDocumentUpload: (session: Record, document: unknown) => ({ + ...session, + name: session.fileName, + contentType: session.contentType, + size: session.fileSize, + expiresAt: '2026-08-04T21:00:00.000Z', + document, + }), +})) + +import { OrchestrationError } from '@/lib/core/orchestration/types' +import { POST } from '@/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/complete/route' + +const WORKSPACE_ID = '6fc7631d-88cd-46f8-9f0a-d4764daef7f8' +const FILE_URL = '/api/files/serve/s3/kb%2Fguide.pdf?context=knowledge-base' +const SESSION = { + id: 'upload-1', + workspaceId: WORKSPACE_ID, + userId: 'user-1', + knowledgeBaseId: 'kb-1', + purpose: 'knowledge_document', + storageContext: 'knowledge-base', + storageKey: 'kb/guide.pdf', + storageProvider: 's3', + providerUploadId: 'provider-1', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + partSize: 8 * 1024 * 1024, + partCount: 1, + status: 'uploading', + metadata: { + tag1: 'product', + processingOptions: { recipe: 'default', lang: 'en' }, + }, + uploadToken: 'token', + createdAt: new Date('2026-08-03T21:00:00.000Z'), + expiresAt: new Date('2026-08-04T21:00:00.000Z'), + completedFileId: null, + error: null, + completedAt: null, + updatedAt: new Date('2026-08-03T21:00:00.000Z'), +} as const +const DOCUMENT = { + id: 'upload-1', + knowledgeBaseId: 'kb-1', + filename: 'guide.pdf', + fileUrl: FILE_URL, + fileSize: 1024, + mimeType: 'application/pdf', + chunkCount: 0, + tokenCount: 0, + characterCount: 0, + enabled: true, + uploadedAt: new Date('2026-08-03T21:01:00.000Z'), +} +const RATE_LIMIT = { + allowed: true, + userId: 'user-1', + keyType: 'workspace', + limit: 100, + remaining: 99, + resetAt: new Date('2026-08-03T22:00:00.000Z'), +} + +function request() { + return POST( + new NextRequest( + `http://localhost:3000/api/v2/knowledge/kb-1/documents/uploads/upload-1/complete?workspaceId=${WORKSPACE_ID}`, + { + method: 'POST', + headers: { 'Content-Type': 'application/json', 'upload-token': 'token' }, + body: JSON.stringify({ parts: [{ partNumber: 1, etag: 'etag-1' }] }), + } + ), + { params: Promise.resolve({ id: 'kb-1', uploadId: 'upload-1' }) } + ) +} + +describe('POST knowledge-document multipart completion', () => { + beforeEach(() => { + vi.clearAllMocks() + mockCheckRateLimit.mockResolvedValue(RATE_LIMIT) + mockResolveKnowledgeDocumentUploadAccess.mockResolvedValue({ + kb: { id: 'kb-1', name: 'Docs' }, + }) + mockResolveKnowledgeDocumentUploadAttribution.mockResolvedValue({ actorUserId: 'payer-1' }) + mockFinalizeKnowledgeDocumentUpload.mockResolvedValue({ + value: DOCUMENT, + completedFileId: DOCUMENT.id, + }) + mockCompleteUploadSession.mockImplementation(async ({ session, finalize }) => { + const finalized = await finalize(session) + return { + session: { ...session, status: 'completed', completedFileId: finalized.completedFileId }, + value: finalized.value, + alreadyCompleted: false, + } + }) + }) + + it('delegates completion to the shared finalizer and returns the bound document', async () => { + const response = await request() + + expect(response.status).toBe(200) + expect(await response.json()).toMatchObject({ data: { document: { id: 'upload-1' } } }) + expect(mockFinalizeKnowledgeDocumentUpload).toHaveBeenCalledWith( + expect.objectContaining({ + claimed: SESSION, + knowledgeBaseId: 'kb-1', + knowledgeBaseName: 'Docs', + workspaceId: WORKSPACE_ID, + userId: 'user-1', + source: 'api', + }) + ) + }) + + it('resolves the payer lazily, only when the finalizer asks for one', async () => { + await request() + + expect(mockResolveKnowledgeDocumentUploadAttribution).not.toHaveBeenCalled() + + const { resolveAttribution } = mockFinalizeKnowledgeDocumentUpload.mock.calls[0][0] + await resolveAttribution() + + expect(mockResolveKnowledgeDocumentUploadAttribution).toHaveBeenCalledWith({ + workspaceId: WORKSPACE_ID, + userId: 'user-1', + rateLimit: RATE_LIMIT, + }) + }) + + it('maps an orchestration failure from the finalizer onto its v2 status', async () => { + mockFinalizeKnowledgeDocumentUpload.mockRejectedValue( + new OrchestrationError('payload_too_large', 'Storage limit exceeded') + ) + + const response = await request() + + expect(response.status).toBe(413) + }) +}) diff --git a/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/complete/route.ts b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/complete/route.ts new file mode 100644 index 00000000000..b4835d232ea --- /dev/null +++ b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/complete/route.ts @@ -0,0 +1,97 @@ +import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' +import type { NextRequest } from 'next/server' +import { NextResponse } from 'next/server' +import { v2CompleteKnowledgeDocumentUploadContract } from '@/lib/api/contracts/v2/knowledge' +import { parseRequest } from '@/lib/api/server' +import { generateRequestId } from '@/lib/core/utils/request' +import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { completeUploadSession } from '@/lib/uploads/multipart-session/service' +import { checkRateLimit } from '@/app/api/v1/middleware' +import { + finalizeKnowledgeDocumentUpload, + getOwnedKnowledgeDocumentUpload, + resolveKnowledgeDocumentUploadAccess, + resolveKnowledgeDocumentUploadAttribution, + toV2KnowledgeDocumentUpload, +} from '@/app/api/v2/knowledge/[id]/documents/uploads/utils' +import { v2ApiGateError } from '@/app/api/v2/lib/gate' +import { + v2CaughtOrchestrationError, + v2Data, + v2Error, + v2RateLimitError, + v2ValidationError, +} from '@/app/api/v2/lib/response' + +const logger = createLogger('V2CompleteKnowledgeDocumentUploadAPI') + +interface KnowledgeDocumentUploadRouteParams { + params: Promise<{ id: string; uploadId: string }> +} + +export const POST = withRouteHandler( + async (request: NextRequest, context: KnowledgeDocumentUploadRouteParams) => { + const requestId = generateRequestId() + + try { + const rateLimit = await checkRateLimit(request, 'knowledge-detail') + if (!rateLimit.allowed) return v2RateLimitError(rateLimit) + const userId = rateLimit.userId! + const gate = await v2ApiGateError(userId) + if (gate) return gate + + const parsed = await parseRequest( + v2CompleteKnowledgeDocumentUploadContract, + request, + context, + { validationErrorResponse: v2ValidationError } + ) + if (!parsed.success) return parsed.response + const { id: knowledgeBaseId, uploadId } = parsed.data.params + const { workspaceId } = parsed.data.query + + const access = await resolveKnowledgeDocumentUploadAccess({ + knowledgeBaseId, + workspaceId, + userId, + rateLimit, + }) + if (access instanceof NextResponse) return access + + const session = getOwnedKnowledgeDocumentUpload({ + knowledgeBaseId, + uploadId, + workspaceId, + userId, + uploadToken: parsed.data.headers['upload-token'], + }) + const result = await completeUploadSession({ + session, + parts: parsed.data.body.parts, + finalize: (claimed) => + finalizeKnowledgeDocumentUpload({ + claimed, + knowledgeBaseId, + knowledgeBaseName: access.kb.name, + workspaceId, + userId, + resolveAttribution: () => + resolveKnowledgeDocumentUploadAttribution({ workspaceId, userId, rateLimit }), + source: 'api', + requestId, + request, + }), + }) + + return v2Data(toV2KnowledgeDocumentUpload(result.session, result.value), { rateLimit }) + } catch (error) { + const classified = v2CaughtOrchestrationError(error) + if (classified) return classified + logger.error(`[${requestId}] Failed to complete knowledge-document upload`, { + error: getErrorMessage(error), + }) + return v2Error('INTERNAL_ERROR', 'Internal server error') + } + } +) diff --git a/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/parts/route.ts b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/parts/route.ts new file mode 100644 index 00000000000..80b079ad985 --- /dev/null +++ b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/parts/route.ts @@ -0,0 +1,78 @@ +import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' +import type { NextRequest } from 'next/server' +import { NextResponse } from 'next/server' +import { v2CreateKnowledgeDocumentUploadPartUrlsContract } from '@/lib/api/contracts/v2/knowledge' +import { parseRequest } from '@/lib/api/server' +import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { createUploadPartUrls } from '@/lib/uploads/multipart-session/service' +import { checkRateLimit } from '@/app/api/v1/middleware' +import { + getOwnedKnowledgeDocumentUpload, + resolveKnowledgeDocumentUploadAccess, +} from '@/app/api/v2/knowledge/[id]/documents/uploads/utils' +import { v2ApiGateError } from '@/app/api/v2/lib/gate' +import { + v2CaughtOrchestrationError, + v2Data, + v2Error, + v2RateLimitError, + v2ValidationError, +} from '@/app/api/v2/lib/response' + +const logger = createLogger('V2KnowledgeDocumentUploadPartsAPI') + +interface KnowledgeDocumentUploadRouteParams { + params: Promise<{ id: string; uploadId: string }> +} + +export const POST = withRouteHandler( + async (request: NextRequest, context: KnowledgeDocumentUploadRouteParams) => { + try { + const rateLimit = await checkRateLimit(request, 'knowledge-detail') + if (!rateLimit.allowed) return v2RateLimitError(rateLimit) + const userId = rateLimit.userId! + const gate = await v2ApiGateError(userId) + if (gate) return gate + + const parsed = await parseRequest( + v2CreateKnowledgeDocumentUploadPartUrlsContract, + request, + context, + { validationErrorResponse: v2ValidationError } + ) + if (!parsed.success) return parsed.response + const { id: knowledgeBaseId, uploadId } = parsed.data.params + const { workspaceId } = parsed.data.query + + const access = await resolveKnowledgeDocumentUploadAccess({ + knowledgeBaseId, + workspaceId, + userId, + rateLimit, + }) + if (access instanceof NextResponse) return access + + const session = getOwnedKnowledgeDocumentUpload({ + knowledgeBaseId, + uploadId, + workspaceId, + userId, + uploadToken: parsed.data.headers['upload-token'], + }) + const parts = await createUploadPartUrls({ + session, + partNumbers: parsed.data.body.partNumbers, + localOrigin: request.nextUrl.origin, + }) + return v2Data({ parts }, { rateLimit }) + } catch (error) { + const classified = v2CaughtOrchestrationError(error) + if (classified) return classified + logger.error('Failed to create knowledge-document upload part URLs', { + error: getErrorMessage(error), + }) + return v2Error('INTERNAL_ERROR', 'Internal server error') + } + } +) diff --git a/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/route.ts b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/route.ts new file mode 100644 index 00000000000..108f4137573 --- /dev/null +++ b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/[uploadId]/route.ts @@ -0,0 +1,72 @@ +import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' +import type { NextRequest } from 'next/server' +import { NextResponse } from 'next/server' +import { v2AbortKnowledgeDocumentUploadContract } from '@/lib/api/contracts/v2/knowledge' +import { parseRequest } from '@/lib/api/server' +import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { checkRateLimit } from '@/app/api/v1/middleware' +import { + abortKnowledgeDocumentUpload, + getOwnedKnowledgeDocumentUpload, + resolveKnowledgeDocumentUploadAccess, + toV2KnowledgeDocumentUpload, +} from '@/app/api/v2/knowledge/[id]/documents/uploads/utils' +import { v2ApiGateError } from '@/app/api/v2/lib/gate' +import { + v2CaughtOrchestrationError, + v2Data, + v2Error, + v2RateLimitError, + v2ValidationError, +} from '@/app/api/v2/lib/response' + +const logger = createLogger('V2KnowledgeDocumentUploadAPI') + +interface KnowledgeDocumentUploadRouteParams { + params: Promise<{ id: string; uploadId: string }> +} + +export const DELETE = withRouteHandler( + async (request: NextRequest, context: KnowledgeDocumentUploadRouteParams) => { + try { + const rateLimit = await checkRateLimit(request, 'knowledge-detail') + if (!rateLimit.allowed) return v2RateLimitError(rateLimit) + const userId = rateLimit.userId! + const gate = await v2ApiGateError(userId) + if (gate) return gate + + const parsed = await parseRequest(v2AbortKnowledgeDocumentUploadContract, request, context, { + validationErrorResponse: v2ValidationError, + }) + if (!parsed.success) return parsed.response + const { id: knowledgeBaseId, uploadId } = parsed.data.params + const { workspaceId } = parsed.data.query + + const access = await resolveKnowledgeDocumentUploadAccess({ + knowledgeBaseId, + workspaceId, + userId, + rateLimit, + }) + if (access instanceof NextResponse) return access + + const session = getOwnedKnowledgeDocumentUpload({ + knowledgeBaseId, + uploadId, + workspaceId, + userId, + uploadToken: parsed.data.headers['upload-token'], + }) + const aborted = await abortKnowledgeDocumentUpload(session, knowledgeBaseId) + return v2Data(toV2KnowledgeDocumentUpload(aborted, null), { rateLimit }) + } catch (error) { + const classified = v2CaughtOrchestrationError(error) + if (classified) return classified + logger.error('Failed to abort knowledge-document upload session', { + error: getErrorMessage(error), + }) + return v2Error('INTERNAL_ERROR', 'Internal server error') + } + } +) diff --git a/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/route.test.ts b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/route.test.ts new file mode 100644 index 00000000000..cf3ea905882 --- /dev/null +++ b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/route.test.ts @@ -0,0 +1,129 @@ +/** + * @vitest-environment node + */ +import { NextRequest, NextResponse } from 'next/server' +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { + mockCheckRateLimit, + mockCreateKnowledgeDocumentUploadSession, + mockResolveKnowledgeDocumentUploadAccess, + mockResolveKnowledgeDocumentUploadBilling, +} = vi.hoisted(() => ({ + mockCheckRateLimit: vi.fn(), + mockCreateKnowledgeDocumentUploadSession: vi.fn(), + mockResolveKnowledgeDocumentUploadAccess: vi.fn(), + mockResolveKnowledgeDocumentUploadBilling: vi.fn(), +})) + +vi.mock('@/app/api/v1/middleware', () => ({ checkRateLimit: mockCheckRateLimit })) +vi.mock('@/app/api/v2/lib/gate', () => ({ + v2ApiGateError: vi.fn().mockResolvedValue(null), +})) +vi.mock('@/app/api/v2/knowledge/[id]/documents/uploads/utils', () => ({ + createKnowledgeDocumentUploadSession: mockCreateKnowledgeDocumentUploadSession, + resolveKnowledgeDocumentUploadAccess: mockResolveKnowledgeDocumentUploadAccess, + resolveKnowledgeDocumentUploadBilling: mockResolveKnowledgeDocumentUploadBilling, + toV2KnowledgeDocumentUpload: (session: Record) => ({ + ...session, + name: session.fileName, + contentType: session.contentType, + size: session.fileSize, + expiresAt: '2026-08-04T21:00:00.000Z', + document: null, + }), +})) + +import { POST } from '@/app/api/v2/knowledge/[id]/documents/uploads/route' + +const WORKSPACE_ID = '6fc7631d-88cd-46f8-9f0a-d4764daef7f8' +const RATE_LIMIT = { + allowed: true, + userId: 'user-1', + keyType: 'workspace', + limit: 100, + remaining: 99, + resetAt: new Date('2026-08-03T22:00:00.000Z'), +} + +function request() { + return POST( + new NextRequest('http://localhost:3000/api/v2/knowledge/kb-1/documents/uploads', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + workspaceId: WORKSPACE_ID, + name: 'guide.pdf', + contentType: 'application/pdf', + size: 1024, + tag1: 'product', + processingOptions: { recipe: 'default', lang: 'en' }, + }), + }), + { params: Promise.resolve({ id: 'kb-1' }) } + ) +} + +describe('POST /api/v2/knowledge/[id]/documents/uploads', () => { + beforeEach(() => { + vi.clearAllMocks() + mockCheckRateLimit.mockResolvedValue(RATE_LIMIT) + mockResolveKnowledgeDocumentUploadAccess.mockResolvedValue({ + kb: { id: 'kb-1', name: 'Docs' }, + }) + mockResolveKnowledgeDocumentUploadBilling.mockResolvedValue({ actorUserId: 'user-1' }) + mockCreateKnowledgeDocumentUploadSession.mockResolvedValue({ + id: 'upload-1', + knowledgeBaseId: 'kb-1', + status: 'uploading', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + partSize: 8 * 1024 * 1024, + partCount: 1, + uploadToken: 'token', + error: null, + }) + }) + + it('authorizes the knowledge base and runs usage billing before accepting storage', async () => { + const response = await request() + + expect(response.status).toBe(201) + expect(mockResolveKnowledgeDocumentUploadAccess).toHaveBeenCalledWith( + expect.objectContaining({ + knowledgeBaseId: 'kb-1', + workspaceId: WORKSPACE_ID, + userId: 'user-1', + }) + ) + expect(mockResolveKnowledgeDocumentUploadBilling).toHaveBeenCalled() + expect(mockCreateKnowledgeDocumentUploadSession).toHaveBeenCalledWith({ + workspaceId: WORKSPACE_ID, + userId: 'user-1', + knowledgeBaseId: 'kb-1', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + metadata: { + tag1: 'product', + processingOptions: { recipe: 'default', lang: 'en' }, + }, + }) + expect(mockResolveKnowledgeDocumentUploadBilling.mock.invocationCallOrder[0]).toBeLessThan( + mockCreateKnowledgeDocumentUploadSession.mock.invocationCallOrder[0] + ) + }) + + it('does not run billing or create provider state when knowledge write access is denied', async () => { + mockResolveKnowledgeDocumentUploadAccess.mockResolvedValue( + NextResponse.json({ error: { code: 'FORBIDDEN', message: 'Access denied' } }, { status: 403 }) + ) + + const response = await request() + + expect(response.status).toBe(403) + expect(mockResolveKnowledgeDocumentUploadBilling).not.toHaveBeenCalled() + expect(mockCreateKnowledgeDocumentUploadSession).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/route.ts b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/route.ts new file mode 100644 index 00000000000..fb15afddeea --- /dev/null +++ b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/route.ts @@ -0,0 +1,86 @@ +import { createLogger } from '@sim/logger' +import { getErrorMessage } from '@sim/utils/errors' +import type { NextRequest } from 'next/server' +import { NextResponse } from 'next/server' +import { v2CreateKnowledgeDocumentUploadContract } from '@/lib/api/contracts/v2/knowledge' +import { parseRequest } from '@/lib/api/server' +import { withRouteHandler } from '@/lib/core/utils/with-route-handler' +import { validateFileType } from '@/lib/uploads/utils/validation' +import { checkRateLimit } from '@/app/api/v1/middleware' +import { + createKnowledgeDocumentUploadSession, + resolveKnowledgeDocumentUploadAccess, + resolveKnowledgeDocumentUploadBilling, + toV2KnowledgeDocumentUpload, +} from '@/app/api/v2/knowledge/[id]/documents/uploads/utils' +import { v2ApiGateError } from '@/app/api/v2/lib/gate' +import { + v2CaughtOrchestrationError, + v2Data, + v2Error, + v2RateLimitError, + v2ValidationError, +} from '@/app/api/v2/lib/response' + +const logger = createLogger('V2KnowledgeDocumentUploadsAPI') + +interface KnowledgeDocumentUploadsRouteParams { + params: Promise<{ id: string }> +} + +export const POST = withRouteHandler( + async (request: NextRequest, context: KnowledgeDocumentUploadsRouteParams) => { + try { + const rateLimit = await checkRateLimit(request, 'knowledge-detail') + if (!rateLimit.allowed) return v2RateLimitError(rateLimit) + const userId = rateLimit.userId! + const gate = await v2ApiGateError(userId) + if (gate) return gate + + const parsed = await parseRequest(v2CreateKnowledgeDocumentUploadContract, request, context, { + validationErrorResponse: v2ValidationError, + }) + if (!parsed.success) return parsed.response + const { id: knowledgeBaseId } = parsed.data.params + const { workspaceId, name, contentType, size, ...metadata } = parsed.data.body + + const access = await resolveKnowledgeDocumentUploadAccess({ + knowledgeBaseId, + workspaceId, + userId, + rateLimit, + }) + if (access instanceof NextResponse) return access + + const billing = await resolveKnowledgeDocumentUploadBilling({ + workspaceId, + userId, + rateLimit, + }) + if (billing instanceof NextResponse) return billing + + const fileTypeError = validateFileType(name, contentType) + if (fileTypeError) { + return v2Error('UNSUPPORTED_MEDIA_TYPE', fileTypeError.message) + } + + const session = await createKnowledgeDocumentUploadSession({ + workspaceId, + userId, + knowledgeBaseId, + fileName: name, + contentType, + fileSize: size, + metadata, + }) + return v2Data(toV2KnowledgeDocumentUpload(session, null), { rateLimit, status: 201 }) + } catch (error) { + const classified = v2CaughtOrchestrationError(error) + if (classified) return classified + logger.error('Failed to create knowledge-document upload session', { + error: getErrorMessage(error), + }) + return v2Error('INTERNAL_ERROR', 'Internal server error') + } + } +) diff --git a/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/utils.test.ts b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/utils.test.ts new file mode 100644 index 00000000000..708f666b2a9 --- /dev/null +++ b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/utils.test.ts @@ -0,0 +1,235 @@ +/** + * @vitest-environment node + */ +import { NextRequest } from 'next/server' +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type { UploadSessionRecord } from '@/lib/uploads/multipart-session/service' + +const { + mockAbortUploadSession, + mockCreateUploadSession, + mockFindBoundKnowledgeDocument, + mockPerformUploadKnowledgeDocument, + mockRecordKnowledgeBaseFileOwnership, +} = vi.hoisted(() => ({ + mockAbortUploadSession: vi.fn(), + mockCreateUploadSession: vi.fn(), + mockFindBoundKnowledgeDocument: vi.fn(), + mockPerformUploadKnowledgeDocument: vi.fn(), + mockRecordKnowledgeBaseFileOwnership: vi.fn(), +})) + +vi.mock('@/lib/knowledge/orchestration', () => ({ + performUploadKnowledgeDocument: mockPerformUploadKnowledgeDocument, +})) +vi.mock('@/lib/knowledge/orchestration/documents', () => ({ + findBoundKnowledgeDocument: mockFindBoundKnowledgeDocument, +})) +vi.mock('@/lib/uploads/multipart-session/service', () => ({ + abortUploadSession: mockAbortUploadSession, + createUploadSession: mockCreateUploadSession, + getOwnedUploadSession: vi.fn(), +})) +vi.mock('@/lib/uploads/server/metadata', () => ({ + recordKnowledgeBaseFileOwnership: mockRecordKnowledgeBaseFileOwnership, +})) + +import { + abortKnowledgeDocumentUpload, + createKnowledgeDocumentUploadSession, + finalizeKnowledgeDocumentUpload, +} from '@/app/api/v2/knowledge/[id]/documents/uploads/utils' + +const WORKSPACE_ID = '6fc7631d-88cd-46f8-9f0a-d4764daef7f8' +const CLAIMED: UploadSessionRecord = { + id: 'upload-1', + workspaceId: WORKSPACE_ID, + userId: 'user-1', + knowledgeBaseId: 'kb-1', + purpose: 'knowledge_document', + storageContext: 'knowledge-base', + storageKey: 'kb/guide.pdf', + storageProvider: 's3', + providerUploadId: 'provider-1', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + partSize: 8 * 1024 * 1024, + partCount: 1, + status: 'uploading', + metadata: { tag1: 'product', processingOptions: { recipe: 'default', lang: 'en' } }, + uploadToken: 'token', + createdAt: new Date('2026-08-03T21:00:00.000Z'), + expiresAt: new Date('2026-08-04T21:00:00.000Z'), + completedFileId: null, + error: null, + completedAt: null, + updatedAt: new Date('2026-08-03T21:00:00.000Z'), +} +const DOCUMENT = { id: 'upload-1', knowledgeBaseId: 'kb-1', filename: 'guide.pdf' } + +function finalize(resolveAttribution = vi.fn().mockResolvedValue({ actorUserId: 'payer-1' })) { + return finalizeKnowledgeDocumentUpload({ + claimed: CLAIMED, + knowledgeBaseId: 'kb-1', + knowledgeBaseName: 'Docs', + workspaceId: WORKSPACE_ID, + userId: 'user-1', + resolveAttribution, + source: 'api', + requestId: 'req-1', + request: new NextRequest('http://localhost:3000/api/v2/knowledge/kb-1'), + }) +} + +function createSession() { + return createKnowledgeDocumentUploadSession({ + workspaceId: WORKSPACE_ID, + userId: 'user-1', + knowledgeBaseId: 'kb-1', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + metadata: { tag1: 'product' }, + }) +} + +describe('createKnowledgeDocumentUploadSession', () => { + beforeEach(() => { + vi.clearAllMocks() + mockCreateUploadSession.mockResolvedValue(CLAIMED) + mockRecordKnowledgeBaseFileOwnership.mockResolvedValue(undefined) + mockAbortUploadSession.mockResolvedValue({ ...CLAIMED, status: 'aborted' }) + }) + + it('records the ownership binding before returning the upload token', async () => { + await expect(createSession()).resolves.toBe(CLAIMED) + + expect(mockCreateUploadSession).toHaveBeenCalledWith({ + workspaceId: WORKSPACE_ID, + userId: 'user-1', + knowledgeBaseId: 'kb-1', + purpose: 'knowledge_document', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + metadata: { tag1: 'product' }, + }) + expect(mockRecordKnowledgeBaseFileOwnership).toHaveBeenCalledWith({ + key: 'kb/guide.pdf', + userId: 'user-1', + workspaceId: WORKSPACE_ID, + originalName: 'guide.pdf', + contentType: 'application/pdf', + size: 1024, + }) + expect(mockCreateUploadSession.mock.invocationCallOrder[0]).toBeLessThan( + mockRecordKnowledgeBaseFileOwnership.mock.invocationCallOrder[0] + ) + }) + + it('aborts provider state when the ownership binding cannot be recorded', async () => { + mockRecordKnowledgeBaseFileOwnership.mockRejectedValue(new Error('database unavailable')) + + await expect(createSession()).rejects.toThrow('database unavailable') + expect(mockAbortUploadSession).toHaveBeenCalledWith(CLAIMED) + }) +}) + +describe('abortKnowledgeDocumentUpload', () => { + beforeEach(() => { + vi.clearAllMocks() + mockAbortUploadSession.mockResolvedValue({ ...CLAIMED, status: 'aborted' }) + }) + + it('aborts an upload that no document is bound to', async () => { + mockFindBoundKnowledgeDocument.mockResolvedValue({ status: 'absent' }) + + await expect(abortKnowledgeDocumentUpload(CLAIMED, 'kb-1')).resolves.toMatchObject({ + status: 'aborted', + }) + expect(mockAbortUploadSession).toHaveBeenCalledWith(CLAIMED) + }) + + it('refuses to abort once a document is bound, so committed bytes survive', async () => { + mockFindBoundKnowledgeDocument.mockResolvedValue({ status: 'bound', document: DOCUMENT }) + + await expect(abortKnowledgeDocumentUpload(CLAIMED, 'kb-1')).rejects.toThrow( + 'Upload has already been completed' + ) + expect(mockAbortUploadSession).not.toHaveBeenCalled() + }) +}) + +describe('finalizeKnowledgeDocumentUpload', () => { + beforeEach(() => { + vi.clearAllMocks() + mockFindBoundKnowledgeDocument.mockResolvedValue({ status: 'absent' }) + mockPerformUploadKnowledgeDocument.mockResolvedValue({ + success: true, + document: DOCUMENT, + created: true, + }) + }) + + it('creates the document, carrying session tags and processing options through', async () => { + const result = await finalize() + + expect(result).toEqual({ value: DOCUMENT, completedFileId: 'upload-1' }) + expect(mockPerformUploadKnowledgeDocument).toHaveBeenCalledWith( + expect.objectContaining({ + documentId: 'upload-1', + startProcessing: 'queue', + uploadedBy: 'payer-1', + processingOptions: { recipe: 'default', lang: 'en' }, + document: expect.objectContaining({ filename: 'guide.pdf', tag1: 'product' }), + }) + ) + }) + + it('answers a retry from the bound document without resolving a payer', async () => { + mockFindBoundKnowledgeDocument.mockResolvedValue({ status: 'bound', document: DOCUMENT }) + const resolveAttribution = vi.fn() + + const result = await finalize(resolveAttribution) + + expect(result).toEqual({ value: DOCUMENT, completedFileId: 'upload-1' }) + expect(resolveAttribution).not.toHaveBeenCalled() + expect(mockPerformUploadKnowledgeDocument).not.toHaveBeenCalled() + }) + + it('retains completed bytes for retry when document creation fails', async () => { + mockPerformUploadKnowledgeDocument.mockResolvedValue({ + success: false, + errorCode: 'payload_too_large', + error: 'Storage limit exceeded', + }) + + await expect(finalize()).rejects.toThrow('Storage limit exceeded') + expect(mockFindBoundKnowledgeDocument).toHaveBeenCalledTimes(1) + }) + + it('lets a retry converge when the first response fails after the document binds', async () => { + mockFindBoundKnowledgeDocument + .mockResolvedValueOnce({ status: 'absent' }) + .mockResolvedValueOnce({ status: 'bound', document: DOCUMENT }) + mockPerformUploadKnowledgeDocument.mockRejectedValue(new Error('audit sink exploded')) + + await expect(finalize()).rejects.toThrow('audit sink exploded') + await expect(finalize()).resolves.toEqual({ + value: DOCUMENT, + completedFileId: 'upload-1', + }) + expect(mockPerformUploadKnowledgeDocument).toHaveBeenCalledTimes(1) + }) + + it('rejects an upload id already bound to a different document without deleting anything', async () => { + mockFindBoundKnowledgeDocument.mockResolvedValue({ status: 'conflict' }) + const resolveAttribution = vi.fn() + + await expect(finalize(resolveAttribution)).rejects.toThrow( + 'Upload id is already bound to a different document' + ) + expect(resolveAttribution).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/utils.ts b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/utils.ts new file mode 100644 index 00000000000..2844b5245dc --- /dev/null +++ b/apps/sim/app/api/v2/knowledge/[id]/documents/uploads/utils.ts @@ -0,0 +1,277 @@ +import type { NextRequest } from 'next/server' +import { NextResponse } from 'next/server' +import type { + V2KnowledgeDocumentSummary, + V2KnowledgeDocumentUpload, +} from '@/lib/api/contracts/v2/knowledge' +import { v2KnowledgeDocumentUploadMetadataSchema } from '@/lib/api/contracts/v2/knowledge' +import type { BillingAttributionSnapshot } from '@/lib/billing/core/billing-attribution' +import { + checkAttributedUsageLimits, + resolveBillingAttribution, + resolveSystemBillingAttribution, +} from '@/lib/billing/core/billing-attribution' +import { OrchestrationError } from '@/lib/core/orchestration/types' +import { performUploadKnowledgeDocument } from '@/lib/knowledge/orchestration' +import type { CreatedKnowledgeDocument } from '@/lib/knowledge/orchestration/documents' +import { findBoundKnowledgeDocument } from '@/lib/knowledge/orchestration/documents' +import type { KnowledgeBaseWithCounts } from '@/lib/knowledge/types' +import { + abortUploadSession, + createUploadSession, + getOwnedUploadSession, + type UploadSessionRecord, +} from '@/lib/uploads/multipart-session/service' +import { recordKnowledgeBaseFileOwnership } from '@/lib/uploads/server/metadata' +import { resolveKnowledgeBase, serializeDate } from '@/app/api/v1/knowledge/utils' +import type { RateLimitResult } from '@/app/api/v1/middleware' +import { v2Error } from '@/app/api/v2/lib/response' + +export async function resolveKnowledgeDocumentUploadAccess(params: { + knowledgeBaseId: string + workspaceId: string + userId: string + rateLimit: RateLimitResult +}): Promise<{ kb: KnowledgeBaseWithCounts } | NextResponse> { + const result = await resolveKnowledgeBase( + params.knowledgeBaseId, + params.workspaceId, + params.userId, + params.rateLimit, + 'write' + ) + if (!(result instanceof NextResponse)) return result + if (result.status === 404) return v2Error('NOT_FOUND', 'Knowledge base not found') + return v2Error('FORBIDDEN', 'Access denied') +} + +/** + * Resolves the payer for an upload without enforcing usage limits. Completion uses this + * because its bytes were already admitted when the session was created; re-running + * admission there would strand uploaded parts and fail idempotent completion retries. + */ +export async function resolveKnowledgeDocumentUploadAttribution(params: { + workspaceId: string + userId: string + rateLimit: RateLimitResult +}): Promise { + return params.rateLimit.keyType === 'workspace' + ? resolveSystemBillingAttribution(params.workspaceId) + : resolveBillingAttribution({ + actorUserId: params.userId, + workspaceId: params.workspaceId, + }) +} + +/** Admission check for a new upload session. Enforced only at session creation. */ +export async function resolveKnowledgeDocumentUploadBilling(params: { + workspaceId: string + userId: string + rateLimit: RateLimitResult +}): Promise { + const attribution = await resolveKnowledgeDocumentUploadAttribution(params) + const usage = await checkAttributedUsageLimits(attribution) + if (usage.isExceeded) { + return v2Error( + 'USAGE_LIMIT_EXCEEDED', + usage.message || 'Usage limit exceeded. Please upgrade your plan to continue.' + ) + } + return attribution +} + +export function getOwnedKnowledgeDocumentUpload(params: { + knowledgeBaseId: string + uploadId: string + workspaceId: string + userId: string + uploadToken: string +}): UploadSessionRecord { + return getOwnedUploadSession({ + uploadId: params.uploadId, + workspaceId: params.workspaceId, + userId: params.userId, + purpose: 'knowledge_document', + knowledgeBaseId: params.knowledgeBaseId, + uploadToken: params.uploadToken, + }) +} + +/** + * Creates a knowledge-document upload and records its ownership binding before the token is + * returned. Failed or abandoned sessions can then be reclaimed by the knowledge-base orphan + * sweeper without racing a later document insert. + */ +export async function createKnowledgeDocumentUploadSession(params: { + workspaceId: string + userId: string + knowledgeBaseId: string + fileName: string + contentType: string + fileSize: number + metadata: Record +}): Promise { + const session = await createUploadSession({ + ...params, + purpose: 'knowledge_document', + }) + try { + await recordKnowledgeBaseFileOwnership({ + key: session.storageKey, + userId: params.userId, + workspaceId: params.workspaceId, + originalName: params.fileName, + contentType: params.contentType, + size: params.fileSize, + }) + } catch (error) { + await abortUploadSession(session) + throw error + } + return session +} + +export function toV2KnowledgeDocumentSummary( + document: CreatedKnowledgeDocument +): V2KnowledgeDocumentSummary { + return { + id: document.id, + knowledgeBaseId: document.knowledgeBaseId, + filename: document.filename, + fileSize: document.fileSize, + mimeType: document.mimeType, + processingStatus: document.processingStatus ?? 'pending', + chunkCount: document.chunkCount, + tokenCount: document.tokenCount, + characterCount: document.characterCount, + enabled: document.enabled, + createdAt: serializeDate(document.uploadedAt), + } +} + +export function toV2KnowledgeDocumentUpload( + session: UploadSessionRecord, + document: CreatedKnowledgeDocument | null +): V2KnowledgeDocumentUpload { + if (!session.knowledgeBaseId) { + throw new Error('Knowledge-document upload session is missing its knowledge base') + } + return { + id: session.id, + knowledgeBaseId: session.knowledgeBaseId, + status: session.status, + name: session.fileName, + contentType: session.contentType, + size: session.fileSize, + partSize: session.partSize, + partCount: session.partCount, + uploadToken: session.uploadToken, + expiresAt: session.expiresAt.toISOString(), + error: session.error, + document: document ? toV2KnowledgeDocumentSummary(document) : null, + } +} + +export function knowledgeDocumentFileUrl(session: UploadSessionRecord): string { + if (session.storageContext !== 'knowledge-base') { + throw new Error('Knowledge-document upload has an invalid storage context') + } + const providerPrefix = session.storageProvider === 'local' ? '' : `${session.storageProvider}/` + return `/api/files/serve/${providerPrefix}${encodeURIComponent(session.storageKey)}?context=knowledge-base` +} + +function knowledgeDocumentInputFor(session: UploadSessionRecord) { + const { processingOptions: _processingOptions, ...documentTags } = + v2KnowledgeDocumentUploadMetadataSchema.parse(session.metadata) + return { + filename: session.fileName, + fileUrl: knowledgeDocumentFileUrl(session), + fileSize: session.fileSize, + mimeType: session.contentType, + ...documentTags, + } +} + +/** + * Aborts an upload session, refusing once a document is bound to it. + * + * Upload sessions are stateless — the signed token always reconstructs as `uploading`, so this + * guard preserves the completed state exposed by the document binding. Provider aborts must + * also remain non-destructive after commit because an in-flight completion is not visible here + * until its document transaction commits. + */ +export async function abortKnowledgeDocumentUpload( + session: UploadSessionRecord, + knowledgeBaseId: string +): Promise { + const bound = await findBoundKnowledgeDocument({ + documentId: session.id, + knowledgeBaseId, + document: knowledgeDocumentInputFor(session), + }) + if (bound.status !== 'absent') { + throw new OrchestrationError('conflict', 'Upload has already been completed') + } + return abortUploadSession(session) +} + +/** + * Binds a completed multipart session to its knowledge document. Shared by the public v2 + * and session-authenticated routes so both get identical completion semantics. + * + * Ordering is load-bearing. A retry is answered from the already-bound document before any + * work that can fail independently of the upload runs, so a payer that became unresolvable + * after the session was created cannot turn a valid retry into an error. The ownership binding + * is recorded before the upload token is issued, so failures retain retriable state and the + * delayed orphan sweeper reclaims sessions that never bind to a document. + */ +export async function finalizeKnowledgeDocumentUpload(params: { + claimed: UploadSessionRecord + knowledgeBaseId: string + knowledgeBaseName: string | null + workspaceId: string + userId: string + resolveAttribution: () => Promise + source: 'api' | 'ui' + requestId: string + request: NextRequest + actorName?: string | null + actorEmail?: string | null +}): Promise<{ value: CreatedKnowledgeDocument; completedFileId: string }> { + const { claimed, knowledgeBaseId, workspaceId, requestId } = params + const { processingOptions } = v2KnowledgeDocumentUploadMetadataSchema.parse(claimed.metadata) + const document = knowledgeDocumentInputFor(claimed) + + const bound = await findBoundKnowledgeDocument({ + documentId: claimed.id, + knowledgeBaseId, + document, + }) + if (bound.status === 'bound') { + return { value: bound.document, completedFileId: bound.document.id } + } + if (bound.status === 'conflict') { + throw new OrchestrationError('conflict', 'Upload id is already bound to a different document') + } + + const billingAttribution = await params.resolveAttribution() + const outcome = await performUploadKnowledgeDocument({ + knowledgeBase: { id: knowledgeBaseId, name: params.knowledgeBaseName, workspaceId }, + document, + documentId: claimed.id, + startProcessing: 'queue', + processingOptions, + billingAttribution, + uploadedBy: billingAttribution.actorUserId, + userId: params.userId, + ...(params.actorName ? { actorName: params.actorName } : {}), + ...(params.actorEmail ? { actorEmail: params.actorEmail } : {}), + source: params.source, + requestId, + request: params.request, + }) + if (!outcome.success) { + throw new OrchestrationError(outcome.errorCode, outcome.error) + } + return { value: outcome.document, completedFileId: outcome.document.id } +} diff --git a/apps/sim/app/workspace/[workspaceId]/knowledge/hooks/use-knowledge-upload.ts b/apps/sim/app/workspace/[workspaceId]/knowledge/hooks/use-knowledge-upload.ts index 3d096f53535..d4270b5ae34 100644 --- a/apps/sim/app/workspace/[workspaceId]/knowledge/hooks/use-knowledge-upload.ts +++ b/apps/sim/app/workspace/[workspaceId]/knowledge/hooks/use-knowledge-upload.ts @@ -1,38 +1,19 @@ import { useCallback, useState } from 'react' import { createLogger } from '@sim/logger' import { getErrorMessage } from '@sim/utils/errors' -import { sleep } from '@sim/utils/helpers' import { useQueryClient } from '@tanstack/react-query' +import type { V2KnowledgeDocumentSummary } from '@/lib/api/contracts/v2/knowledge' import { - calculateUploadTimeoutMs, - DirectUploadError, - isTransientUploadError, - LARGE_FILE_THRESHOLD, - MULTIPART_MAX_RETRIES, - MULTIPART_RETRY_BACKOFF, - MULTIPART_RETRY_DELAY_MS, - normalizePresignedData, - type PresignedUploadInfo, - runUploadStrategy, runWithConcurrency, type UploadProgressEvent, WHOLE_FILE_PARALLEL_UPLOADS, } from '@/lib/uploads/client/direct-upload' -import { getFileContentType, isAbortError, isNetworkError } from '@/lib/uploads/utils/file-utils' +import { uploadKnowledgeDocumentSession } from '@/lib/uploads/client/session-upload' import { knowledgeKeys } from '@/hooks/queries/utils/knowledge-keys' const logger = createLogger('KnowledgeUpload') -const KB_BATCH_PRESIGNED_ENDPOINT = '/api/files/presigned/batch?type=knowledge-base' -const KB_API_UPLOAD_ENDPOINT = '/api/files/upload' - -const BATCH_REQUEST_SIZE = 50 - -export interface UploadedFile { - filename: string - fileUrl: string - fileSize: number - mimeType: string +interface KnowledgeDocumentUploadFile extends File { tag1?: string tag2?: string tag3?: string @@ -86,153 +67,6 @@ class KnowledgeUploadError extends Error { } } -class ProcessingError extends KnowledgeUploadError { - constructor(message: string, details?: unknown) { - super(message, 'PROCESSING_ERROR', details) - } -} - -interface BatchPresignedFile { - fileName: string - contentType: string - fileSize: number -} - -/** - * Fetch presigned upload data for the small files in `files`. Returns a sparse - * array aligned with the input: entries for files >= LARGE_FILE_THRESHOLD are - * `undefined` because those uploads use multipart and never consume a presigned - * single-PUT URL. - */ -const fetchBatchPresignedData = async ( - files: File[], - workspaceId: string -): Promise<(PresignedUploadInfo | undefined)[]> => { - const result: (PresignedUploadInfo | undefined)[] = new Array(files.length).fill(undefined) - const smallFileIndices: number[] = [] - for (let i = 0; i < files.length; i++) { - if (files[i].size <= LARGE_FILE_THRESHOLD) smallFileIndices.push(i) - } - if (smallFileIndices.length === 0) return result - - const batchEndpoint = `${KB_BATCH_PRESIGNED_ENDPOINT}&workspaceId=${encodeURIComponent(workspaceId)}` - - for (let start = 0; start < smallFileIndices.length; start += BATCH_REQUEST_SIZE) { - const batchIndices = smallFileIndices.slice(start, start + BATCH_REQUEST_SIZE) - const batchFiles = batchIndices.map((i) => files[i]) - const body: { files: BatchPresignedFile[] } = { - files: batchFiles.map((file) => ({ - fileName: file.name, - contentType: getFileContentType(file), - fileSize: file.size, - })), - } - - const response = await fetch(batchEndpoint, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify(body), - }) - - if (!response.ok) { - throw new Error(`Batch presigned URL generation failed: ${response.statusText}`) - } - - const { files: presignedItems } = (await response.json()) as { files: unknown[] } - batchIndices.forEach((fileIdx, batchPos) => { - result[fileIdx] = normalizePresignedData(presignedItems[batchPos], batchFiles[batchPos].name) - }) - } - - return result -} - -/** - * Server-proxied fallback used when cloud storage isn't configured. - */ -const uploadFileThroughAPI = async ( - file: File, - workspaceId: string | undefined -): Promise<{ filePath: string }> => { - const formData = new FormData() - formData.append('file', file) - formData.append('context', 'knowledge-base') - if (workspaceId) formData.append('workspaceId', workspaceId) - - const controller = new AbortController() - const timeoutId = setTimeout(() => controller.abort(), calculateUploadTimeoutMs(file.size)) - - try { - const response = await fetch(KB_API_UPLOAD_ENDPOINT, { - method: 'POST', - body: formData, - signal: controller.signal, - }) - - if (!response.ok) { - let errorData: { message?: string; error?: string } | null = null - try { - errorData = (await response.json()) as { message?: string; error?: string } - } catch {} - throw new KnowledgeUploadError( - `Failed to upload ${file.name}: ${errorData?.message || errorData?.error || response.statusText}`, - 'API_UPLOAD_ERROR', - errorData - ) - } - - const result = (await response.json()) as { - fileInfo?: { path?: string } - path?: string - } - const filePath = result.fileInfo?.path ?? result.path - if (!filePath) { - throw new KnowledgeUploadError( - `Invalid upload response for ${file.name}: missing file path`, - 'API_UPLOAD_ERROR', - result - ) - } - - return { filePath } - } finally { - clearTimeout(timeoutId) - } -} - -const toAbsoluteUrl = (path: string): string => - path.startsWith('http') ? path : `${window.location.origin}${path}` - -/** - * Build the {@link UploadedFile} payload from a `File`, carrying through any - * `tagN` fields the caller attached to it. Pure — kept at module scope so it - * isn't rebuilt on every render of the hook. - */ -const buildUploadedFile = (file: File, fileUrl: string): UploadedFile => { - const f = file as File & { - tag1?: string - tag2?: string - tag3?: string - tag4?: string - tag5?: string - tag6?: string - tag7?: string - } - return { - filename: file.name, - fileUrl, - fileSize: file.size, - mimeType: getFileContentType(file), - tag1: f.tag1, - tag2: f.tag2, - tag3: f.tag3, - tag4: f.tag4, - tag5: f.tag5, - tag6: f.tag6, - tag7: f.tag7, - } -} - export function useKnowledgeUpload(options: UseKnowledgeUploadOptions = {}) { const queryClient = useQueryClient() const [isUploading, setIsUploading] = useState(false) @@ -255,51 +89,40 @@ export function useKnowledgeUpload(options: UseKnowledgeUploadOptions = {}) { const uploadOneFile = async ( file: File, fileIndex: number, - presigned: PresignedUploadInfo | undefined - ): Promise => { + knowledgeBaseId: string, + processingOptions: ProcessingOptions + ): Promise => { if (!options.workspaceId) { throw new KnowledgeUploadError('workspaceId is required for upload', 'MISSING_WORKSPACE_ID') } - const onProgress = (event: UploadProgressEvent) => { updateFileStatus(fileIndex, { progress: event.percent, status: 'uploading' }) } - - let attempt = 0 - while (true) { - try { - const result = await runUploadStrategy({ - file, - workspaceId: options.workspaceId, - context: 'knowledge-base', - presignedEndpoint: `/api/files/presigned?type=knowledge-base&workspaceId=${encodeURIComponent(options.workspaceId)}`, - presignedOverride: presigned, - onProgress, - }) - return buildUploadedFile(file, toAbsoluteUrl(result.path)) - } catch (error) { - if (error instanceof DirectUploadError && error.code === 'FALLBACK_REQUIRED') { - const { filePath } = await uploadFileThroughAPI(file, options.workspaceId) - return buildUploadedFile(file, toAbsoluteUrl(filePath)) - } - - const retryable = isNetworkError(error) || isTransientUploadError(error) - if (isAbortError(error) || !retryable || attempt >= MULTIPART_MAX_RETRIES) { - throw error - } - - const delay = MULTIPART_RETRY_DELAY_MS * MULTIPART_RETRY_BACKOFF ** attempt - attempt++ - logger.warn( - `Upload retry ${attempt}/${MULTIPART_MAX_RETRIES} for ${file.name} in ${Math.round(delay / 1000)}s` - ) - updateFileStatus(fileIndex, { progress: 0, status: 'uploading' }) - await sleep(delay) - } - } + const taggedFile = file as KnowledgeDocumentUploadFile + return uploadKnowledgeDocumentSession({ + workspaceId: options.workspaceId, + knowledgeBaseId, + file, + onProgress, + tag1: taggedFile.tag1, + tag2: taggedFile.tag2, + tag3: taggedFile.tag3, + tag4: taggedFile.tag4, + tag5: taggedFile.tag5, + tag6: taggedFile.tag6, + tag7: taggedFile.tag7, + processingOptions: { + recipe: processingOptions.recipe ?? 'default', + lang: 'en', + }, + }) } - const uploadFilesInBatches = async (files: File[]): Promise => { + const uploadFilesInBatches = async ( + files: File[], + knowledgeBaseId: string, + processingOptions: ProcessingOptions + ): Promise => { if (!options.workspaceId) { throw new KnowledgeUploadError('workspaceId is required for upload', 'MISSING_WORKSPACE_ID') } @@ -313,9 +136,7 @@ export function useKnowledgeUpload(options: UseKnowledgeUploadOptions = {}) { setUploadProgress((prev) => ({ ...prev, fileStatuses })) - logger.info(`Starting batch upload of ${files.length} files`) - - const presignedData = await fetchBatchPresignedData(files, options.workspaceId) + logger.info(`Starting signed session upload of ${files.length} files`) const settled = await runWithConcurrency( files, @@ -323,7 +144,7 @@ export function useKnowledgeUpload(options: UseKnowledgeUploadOptions = {}) { async (file, index) => { updateFileStatus(index, { status: 'uploading' }) try { - const uploaded = await uploadOneFile(file, index, presignedData[index]) + const uploaded = await uploadOneFile(file, index, knowledgeBaseId, processingOptions) setUploadProgress((prev) => ({ ...prev, filesCompleted: prev.filesCompleted + 1, @@ -337,7 +158,7 @@ export function useKnowledgeUpload(options: UseKnowledgeUploadOptions = {}) { } ) - const succeeded: UploadedFile[] = [] + const succeeded: V2KnowledgeDocumentSummary[] = [] const failed: Array<{ file: File; error: Error }> = [] settled.forEach((result, idx) => { if (result?.status === 'fulfilled') { @@ -354,7 +175,7 @@ export function useKnowledgeUpload(options: UseKnowledgeUploadOptions = {}) { throw new KnowledgeUploadError( `Failed to upload ${failed.length} file(s)`, 'PARTIAL_UPLOAD_FAILURE', - { failedFiles: failed, uploadedFiles: succeeded } + { failedFiles: failed, uploadedDocuments: succeeded } ) } @@ -365,7 +186,7 @@ export function useKnowledgeUpload(options: UseKnowledgeUploadOptions = {}) { files: File[], knowledgeBaseId: string, processingOptions: ProcessingOptions = {} - ): Promise => { + ): Promise => { if (files.length === 0) { throw new KnowledgeUploadError('No files provided for upload', 'NO_FILES') } @@ -378,76 +199,27 @@ export function useKnowledgeUpload(options: UseKnowledgeUploadOptions = {}) { setUploadError(null) setUploadProgress({ stage: 'uploading', filesCompleted: 0, totalFiles: files.length }) - const uploadedFiles = await uploadFilesInBatches(files) + const uploadedDocuments = await uploadFilesInBatches( + files, + knowledgeBaseId, + processingOptions + ) setUploadProgress((prev) => ({ ...prev, stage: 'processing' })) - - // boundary-raw-fetch: bulk document-processing kickoff with dynamic recipe payload; response is consumed alongside the upload progress lifecycle and not modeled by a single contract - const processResponse = await fetch(`/api/knowledge/${knowledgeBaseId}/documents`, { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - documents: uploadedFiles.map((f) => ({ ...f })), - processingOptions: { - recipe: processingOptions.recipe ?? 'default', - lang: 'en', - }, - bulk: true, - }), - }) - - if (!processResponse.ok) { - let errorData: { error?: string; message?: string } | null = null - try { - errorData = (await processResponse.json()) as { error?: string; message?: string } - } catch {} - logger.error('Document processing failed:', { - status: processResponse.status, - error: errorData, - }) - throw new ProcessingError( - `Failed to start document processing: ${errorData?.error || errorData?.message || 'Unknown error'}`, - errorData - ) - } - - const processResult = (await processResponse.json()) as { - success?: boolean - error?: string - data?: { documentsCreated?: unknown } - } - - if (!processResult.success) { - throw new ProcessingError( - `Document processing failed: ${processResult.error || 'Unknown error'}`, - processResult - ) - } - - if (!processResult.data?.documentsCreated) { - throw new ProcessingError( - 'Invalid processing response: missing document data', - processResult - ) - } - - setUploadProgress((prev) => ({ ...prev, stage: 'completing' })) - logger.info(`Successfully started processing ${uploadedFiles.length} documents`) + logger.info(`Successfully started processing ${uploadedDocuments.length} documents`) await queryClient.invalidateQueries({ queryKey: knowledgeKeys.detail(knowledgeBaseId) }) - return uploadedFiles + return uploadedDocuments } catch (err) { logger.error('Error uploading documents:', err) const error: UploadError = err instanceof KnowledgeUploadError ? { message: err.message, code: err.code, details: err.details, timestamp: Date.now() } - : err instanceof DirectUploadError - ? { message: err.message, code: err.code, details: err.details, timestamp: Date.now() } - : err instanceof Error - ? { message: err.message, timestamp: Date.now() } - : { message: 'Unknown error occurred during upload', timestamp: Date.now() } + : err instanceof Error + ? { message: err.message, timestamp: Date.now() } + : { message: 'Unknown error occurred during upload', timestamp: Date.now() } setUploadError(error) options.onError?.(error) diff --git a/apps/sim/background/cleanup-soft-deletes.ts b/apps/sim/background/cleanup-soft-deletes.ts index b9ffd3c932b..3227d82e92c 100644 --- a/apps/sim/background/cleanup-soft-deletes.ts +++ b/apps/sim/background/cleanup-soft-deletes.ts @@ -702,14 +702,13 @@ const CLEANUP_TARGETS = [ ] as const /** - * Sweep abandoned knowledge-base ownership bindings. The presigned upload flow - * writes a `workspace_files` binding when it hands out an upload URL, before the - * object is stored and before any document is created. If the upload is never - * completed, that binding is orphaned — no `document.storageKey` ever references - * its key. Such bindings are inert (read access requires a live document, and - * the move re-point only follows referenced keys), but they accumulate, so we - * drop the best-effort object and soft-delete the binding once they are older - * than the grace window. + * Sweep abandoned knowledge-base ownership bindings. Presigned and multipart upload flows + * write a `workspace_files` binding before the object is stored and before any document is + * created. If the upload is never completed, that binding is orphaned — no + * `document.storageKey` ever references its key. Such bindings are inert (read access requires + * a live document, and the move re-point only follows referenced keys), but they accumulate, + * so we drop the best-effort object and soft-delete the binding once they are older than the + * grace window. */ async function cleanupOrphanedKnowledgeBaseBindings( workspaceIds: string[], diff --git a/apps/sim/lib/api/contracts/knowledge/upload-sessions.ts b/apps/sim/lib/api/contracts/knowledge/upload-sessions.ts new file mode 100644 index 00000000000..7a15defe3fd --- /dev/null +++ b/apps/sim/lib/api/contracts/knowledge/upload-sessions.ts @@ -0,0 +1,51 @@ +import { defineRouteContract } from '@/lib/api/contracts/types' +import { + v2CreateKnowledgeDocumentUploadBodySchema, + v2KnowledgeDocumentUploadParamsSchema, + v2KnowledgeDocumentUploadSchema, + v2UploadKnowledgeDocumentQuerySchema, +} from '@/lib/api/contracts/v2/knowledge' +import { v2DataResponse } from '@/lib/api/contracts/v2/shared' +import { + v2CompleteUploadBodySchema, + v2PartUrlsBodySchema, + v2PartUrlsDataSchema, + v2UploadTokenHeadersSchema, +} from '@/lib/api/contracts/v2/uploads' + +export const createKnowledgeDocumentUploadContract = defineRouteContract({ + method: 'POST', + path: '/api/knowledge/[id]/documents/uploads', + params: v2KnowledgeDocumentUploadParamsSchema.omit({ uploadId: true }), + body: v2CreateKnowledgeDocumentUploadBodySchema, + response: { mode: 'json', schema: v2DataResponse(v2KnowledgeDocumentUploadSchema) }, +}) + +export const abortKnowledgeDocumentUploadContract = defineRouteContract({ + method: 'DELETE', + path: '/api/knowledge/[id]/documents/uploads/[uploadId]', + params: v2KnowledgeDocumentUploadParamsSchema, + query: v2UploadKnowledgeDocumentQuerySchema, + headers: v2UploadTokenHeadersSchema, + response: { mode: 'json', schema: v2DataResponse(v2KnowledgeDocumentUploadSchema) }, +}) + +export const createKnowledgeDocumentUploadPartUrlsContract = defineRouteContract({ + method: 'POST', + path: '/api/knowledge/[id]/documents/uploads/[uploadId]/parts', + params: v2KnowledgeDocumentUploadParamsSchema, + query: v2UploadKnowledgeDocumentQuerySchema, + headers: v2UploadTokenHeadersSchema, + body: v2PartUrlsBodySchema, + response: { mode: 'json', schema: v2DataResponse(v2PartUrlsDataSchema) }, +}) + +export const completeKnowledgeDocumentUploadContract = defineRouteContract({ + method: 'POST', + path: '/api/knowledge/[id]/documents/uploads/[uploadId]/complete', + params: v2KnowledgeDocumentUploadParamsSchema, + query: v2UploadKnowledgeDocumentQuerySchema, + headers: v2UploadTokenHeadersSchema, + body: v2CompleteUploadBodySchema, + response: { mode: 'json', schema: v2DataResponse(v2KnowledgeDocumentUploadSchema) }, +}) diff --git a/apps/sim/lib/api/contracts/v2/knowledge.ts b/apps/sim/lib/api/contracts/v2/knowledge.ts index d92f30c20a1..bdca6fe4f01 100644 --- a/apps/sim/lib/api/contracts/v2/knowledge.ts +++ b/apps/sim/lib/api/contracts/v2/knowledge.ts @@ -22,6 +22,14 @@ import { v2SearchSchema, v2SortFields, } from '@/lib/api/contracts/v2/shared' +import { + v2CompleteUploadBodySchema, + v2PartUrlsBodySchema, + v2PartUrlsDataSchema, + v2UploadStatusSchema, + v2UploadTokenHeadersSchema, +} from '@/lib/api/contracts/v2/uploads' +import { MAX_KNOWLEDGE_DOCUMENT_FILE_SIZE } from '@/lib/uploads/shared/types' /** * v2 knowledge contracts. @@ -151,6 +159,67 @@ export type V2KnowledgeSearchData = z.output export const v2UploadKnowledgeDocumentQuerySchema = z.object({ workspaceId: workspaceIdSchema }) export type V2UploadKnowledgeDocumentQuery = z.output +export const v2KnowledgeDocumentUploadParamsSchema = knowledgeBaseParamsSchema.extend({ + uploadId: z.string().min(1, 'uploadId is required'), +}) +export type V2KnowledgeDocumentUploadParams = z.output + +const knowledgeDocumentUploadTagSchema = z + .string() + .max(1000, 'Knowledge document tag values cannot exceed 1000 characters') + .optional() + +export const v2KnowledgeDocumentUploadMetadataSchema = z + .object({ + tag1: knowledgeDocumentUploadTagSchema, + tag2: knowledgeDocumentUploadTagSchema, + tag3: knowledgeDocumentUploadTagSchema, + tag4: knowledgeDocumentUploadTagSchema, + tag5: knowledgeDocumentUploadTagSchema, + tag6: knowledgeDocumentUploadTagSchema, + tag7: knowledgeDocumentUploadTagSchema, + processingOptions: z + .object({ + recipe: z.string().max(255, 'recipe cannot exceed 255 characters').optional(), + lang: z.string().max(35, 'lang cannot exceed 35 characters').optional(), + }) + .strict() + .optional(), + }) + .strict() +export type V2KnowledgeDocumentUploadMetadata = z.output< + typeof v2KnowledgeDocumentUploadMetadataSchema +> + +export const v2CreateKnowledgeDocumentUploadBodySchema = z + .object({ + workspaceId: workspaceIdSchema, + name: z.string().trim().min(1, 'name is required').max(255, 'name is too long'), + contentType: z.string().trim().min(1, 'contentType is required').max(255), + size: z.number().int().min(1).max(MAX_KNOWLEDGE_DOCUMENT_FILE_SIZE), + ...v2KnowledgeDocumentUploadMetadataSchema.shape, + }) + .strict() +export type V2CreateKnowledgeDocumentUploadBody = z.input< + typeof v2CreateKnowledgeDocumentUploadBodySchema +> + +export const v2KnowledgeDocumentUploadSchema = z.object({ + id: z.string(), + knowledgeBaseId: z.string(), + status: v2UploadStatusSchema, + name: z.string(), + contentType: z.string(), + size: z.number().int().positive(), + partSize: z.number().int().positive(), + partCount: z.number().int().positive(), + uploadToken: z.string().min(1), + expiresAt: z.string().datetime(), + error: z.string().nullable(), + document: v2KnowledgeDocumentSummarySchema.nullable(), +}) +export type V2KnowledgeDocumentUpload = z.output + export const v2KnowledgeBaseSortFields = ['name', 'createdAt', 'updatedAt'] as const export type V2KnowledgeBaseSortBy = (typeof v2KnowledgeBaseSortFields)[number] @@ -271,6 +340,43 @@ export const v2UploadKnowledgeDocumentContract = defineRouteContract({ }, }) +export const v2CreateKnowledgeDocumentUploadContract = defineRouteContract({ + method: 'POST', + path: '/api/v2/knowledge/[id]/documents/uploads', + params: knowledgeBaseParamsSchema, + body: v2CreateKnowledgeDocumentUploadBodySchema, + response: { mode: 'json', schema: v2DataResponse(v2KnowledgeDocumentUploadSchema) }, +}) + +export const v2AbortKnowledgeDocumentUploadContract = defineRouteContract({ + method: 'DELETE', + path: '/api/v2/knowledge/[id]/documents/uploads/[uploadId]', + params: v2KnowledgeDocumentUploadParamsSchema, + query: v2UploadKnowledgeDocumentQuerySchema, + headers: v2UploadTokenHeadersSchema, + response: { mode: 'json', schema: v2DataResponse(v2KnowledgeDocumentUploadSchema) }, +}) + +export const v2CreateKnowledgeDocumentUploadPartUrlsContract = defineRouteContract({ + method: 'POST', + path: '/api/v2/knowledge/[id]/documents/uploads/[uploadId]/parts', + params: v2KnowledgeDocumentUploadParamsSchema, + query: v2UploadKnowledgeDocumentQuerySchema, + headers: v2UploadTokenHeadersSchema, + body: v2PartUrlsBodySchema, + response: { mode: 'json', schema: v2DataResponse(v2PartUrlsDataSchema) }, +}) + +export const v2CompleteKnowledgeDocumentUploadContract = defineRouteContract({ + method: 'POST', + path: '/api/v2/knowledge/[id]/documents/uploads/[uploadId]/complete', + params: v2KnowledgeDocumentUploadParamsSchema, + query: v2UploadKnowledgeDocumentQuerySchema, + headers: v2UploadTokenHeadersSchema, + body: v2CompleteUploadBodySchema, + response: { mode: 'json', schema: v2DataResponse(v2KnowledgeDocumentUploadSchema) }, +}) + export const v2GetKnowledgeDocumentContract = defineRouteContract({ method: 'GET', path: '/api/v2/knowledge/[id]/documents/[documentId]', diff --git a/apps/sim/lib/knowledge/documents/service.ts b/apps/sim/lib/knowledge/documents/service.ts index 9ca64bb6d4b..70b6dc6c691 100644 --- a/apps/sim/lib/knowledge/documents/service.ts +++ b/apps/sim/lib/knowledge/documents/service.ts @@ -1521,7 +1521,8 @@ export async function createSingleDocument( }, knowledgeBaseId: string, requestId: string, - uploadedBy: string | null = null + uploadedBy: string | null = null, + documentId = generateId() ): Promise<{ id: string knowledgeBaseId: string @@ -1542,7 +1543,6 @@ export async function createSingleDocument( tag6: string | null tag7: string | null }> { - const documentId = generateId() const now = new Date() const [resolvedDocumentData] = await resolveServerKnownDocumentSizes([documentData]) const admission = await resolveDocumentStorageAdmission( @@ -1713,6 +1713,60 @@ export async function createSingleDocument( } } +/** Returns one active document by its deterministic upload id. */ +export async function getDocumentByUploadId( + documentId: string, + knowledgeBaseId: string +): Promise< + | (Awaited> & { + processingStatus: 'pending' | 'processing' | 'completed' | 'failed' + }) + | null +> { + const [existing] = await db + .select({ + id: document.id, + knowledgeBaseId: document.knowledgeBaseId, + filename: document.filename, + fileUrl: document.fileUrl, + fileSize: document.fileSize, + mimeType: document.mimeType, + chunkCount: document.chunkCount, + tokenCount: document.tokenCount, + characterCount: document.characterCount, + enabled: document.enabled, + uploadedAt: document.uploadedAt, + tag1: document.tag1, + tag2: document.tag2, + tag3: document.tag3, + tag4: document.tag4, + tag5: document.tag5, + tag6: document.tag6, + tag7: document.tag7, + processingStatus: document.processingStatus, + }) + .from(document) + .where( + and( + eq(document.id, documentId), + eq(document.knowledgeBaseId, knowledgeBaseId), + isNull(document.deletedAt) + ) + ) + .limit(1) + if (!existing) return null + const processingStatus = existing.processingStatus + if ( + processingStatus !== 'pending' && + processingStatus !== 'processing' && + processingStatus !== 'completed' && + processingStatus !== 'failed' + ) { + throw new Error(`Document ${existing.id} has invalid processing status`) + } + return { ...existing, processingStatus } +} + export async function bulkDocumentOperation( knowledgeBaseId: string, operation: 'enable' | 'disable' | 'delete', diff --git a/apps/sim/lib/knowledge/orchestration/documents.test.ts b/apps/sim/lib/knowledge/orchestration/documents.test.ts index bfbe1bdc403..42a8b250e05 100644 --- a/apps/sim/lib/knowledge/orchestration/documents.test.ts +++ b/apps/sim/lib/knowledge/orchestration/documents.test.ts @@ -8,6 +8,7 @@ const { mockCreateDocumentRecords, mockCreateSingleDocument, mockDeleteDocument, + mockGetDocumentByUploadId, mockMarkDocumentAsFailedTimeout, mockProcessDocumentAsync, mockProcessDocumentsWithQueue, @@ -19,6 +20,7 @@ const { mockCreateDocumentRecords: vi.fn(), mockCreateSingleDocument: vi.fn(), mockDeleteDocument: vi.fn(), + mockGetDocumentByUploadId: vi.fn(), mockMarkDocumentAsFailedTimeout: vi.fn(), mockProcessDocumentAsync: vi.fn(), mockProcessDocumentsWithQueue: vi.fn(), @@ -43,6 +45,7 @@ vi.mock('@/lib/knowledge/documents/service', () => ({ createDocumentRecords: mockCreateDocumentRecords, createSingleDocument: mockCreateSingleDocument, deleteDocument: mockDeleteDocument, + getDocumentByUploadId: mockGetDocumentByUploadId, markDocumentAsFailedTimeout: mockMarkDocumentAsFailedTimeout, processDocumentAsync: mockProcessDocumentAsync, processDocumentsWithQueue: mockProcessDocumentsWithQueue, @@ -74,6 +77,7 @@ describe('performUploadKnowledgeDocument', () => { beforeEach(() => { vi.clearAllMocks() mockCreateSingleDocument.mockResolvedValue({ id: 'doc-1', filename: 'report.pdf' }) + mockGetDocumentByUploadId.mockResolvedValue(null) mockProcessDocumentsWithQueue.mockResolvedValue(undefined) mockProcessDocumentAsync.mockResolvedValue(undefined) }) @@ -155,6 +159,71 @@ describe('performUploadKnowledgeDocument', () => { .errorCode ).toBe('forbidden') }) + + it('returns the document already bound to a stateless upload id without duplicating work', async () => { + const existing = { + id: 'upload-1', + knowledgeBaseId: 'kb-1', + filename: FILE.filename, + fileUrl: FILE.fileUrl, + fileSize: FILE.fileSize, + mimeType: FILE.mimeType, + chunkCount: 0, + tokenCount: 0, + characterCount: 0, + enabled: true, + uploadedAt: new Date(), + } + mockGetDocumentByUploadId.mockResolvedValue(existing) + + const outcome = await performUploadKnowledgeDocument({ + ...ACTOR, + knowledgeBase: KB, + document: FILE, + documentId: 'upload-1', + startProcessing: 'queue', + }) + + expect(outcome).toMatchObject({ success: true, created: false, document: existing }) + expect(mockCreateSingleDocument).not.toHaveBeenCalled() + expect(mockProcessDocumentsWithQueue).not.toHaveBeenCalled() + expect(mockRecordAudit).not.toHaveBeenCalled() + }) + + it('converges on the existing document when concurrent completions race to insert', async () => { + const existing = { + id: 'upload-1', + knowledgeBaseId: 'kb-1', + ...FILE, + chunkCount: 0, + tokenCount: 0, + characterCount: 0, + enabled: true, + uploadedAt: new Date(), + processingStatus: 'pending', + } + mockGetDocumentByUploadId.mockResolvedValueOnce(null).mockResolvedValueOnce(existing) + mockCreateSingleDocument.mockRejectedValue(new Error('duplicate key')) + + const outcome = await performUploadKnowledgeDocument({ + ...ACTOR, + knowledgeBase: KB, + document: FILE, + documentId: 'upload-1', + startProcessing: 'queue', + }) + + expect(outcome).toMatchObject({ success: true, created: false, document: existing }) + expect(mockCreateSingleDocument).toHaveBeenCalledWith( + FILE, + 'kb-1', + 'req-1', + 'user-1', + 'upload-1' + ) + expect(mockProcessDocumentsWithQueue).not.toHaveBeenCalled() + expect(mockRecordAudit).not.toHaveBeenCalled() + }) }) describe('performUploadKnowledgeDocuments', () => { diff --git a/apps/sim/lib/knowledge/orchestration/documents.ts b/apps/sim/lib/knowledge/orchestration/documents.ts index d0111d53289..de91b1a6ee0 100644 --- a/apps/sim/lib/knowledge/orchestration/documents.ts +++ b/apps/sim/lib/knowledge/orchestration/documents.ts @@ -10,6 +10,7 @@ import { createSingleDocument, type DocumentData, deleteDocument, + getDocumentByUploadId, markDocumentAsFailedTimeout, type ProcessingOptions, processDocumentAsync, @@ -59,7 +60,10 @@ export interface KnowledgeDocumentInput { */ export type KnowledgeDocumentProcessing = 'queue' | 'async' -export type CreatedKnowledgeDocument = Awaited> +export type CreatedKnowledgeDocument = Awaited> & { + /** Present when an idempotent completion returns an already-processing document. */ + processingStatus?: 'pending' | 'processing' | 'completed' | 'failed' +} export interface PerformUploadKnowledgeDocumentParams extends KnowledgeOperationContext { knowledgeBase: KnowledgeBaseTarget @@ -69,12 +73,49 @@ export interface PerformUploadKnowledgeDocumentParams extends KnowledgeOperation billingAttribution?: BillingAttributionSnapshot /** Row owner recorded on the document; defaults to the acting user. */ uploadedBy?: string | null + /** Deterministic id carried by a stateless upload token for completion retries. */ + documentId?: string } export type PerformUploadKnowledgeDocumentResult = KnowledgeOrchestrationResult<{ document: CreatedKnowledgeDocument + created: boolean }> +function isSameKnowledgeDocumentUpload( + existing: CreatedKnowledgeDocument, + document: KnowledgeDocumentInput +): boolean { + return ( + existing.filename === document.filename && + existing.fileUrl === document.fileUrl && + existing.fileSize === document.fileSize && + existing.mimeType === document.mimeType + ) +} + +export type BoundKnowledgeDocument = + | { status: 'absent' } + | { status: 'bound'; document: CreatedKnowledgeDocument } + | { status: 'conflict' } + +/** + * Resolves what a stateless upload id is already bound to. Callers use this to answer a + * completion retry from existing state before doing any work that can fail independently + * of the upload — resolving a billing payer, or deleting the uploaded object. + */ +export async function findBoundKnowledgeDocument(params: { + documentId: string + knowledgeBaseId: string + document: KnowledgeDocumentInput +}): Promise { + const existing = await getDocumentByUploadId(params.documentId, params.knowledgeBaseId) + if (!existing) return { status: 'absent' } + return isSameKnowledgeDocumentUpload(existing, params.document) + ? { status: 'bound', document: existing } + : { status: 'conflict' } +} + function auditUpload( params: KnowledgeOperationContext & { knowledgeBase: KnowledgeBaseTarget }, entry: { resourceId: string; resourceName: string; description: string; metadata: object } @@ -137,14 +178,49 @@ export async function performUploadKnowledgeDocument( const requestId = params.requestId ?? generateRequestId() let created: CreatedKnowledgeDocument - try { - created = await createSingleDocument( + if (params.documentId) { + const bound = await findBoundKnowledgeDocument({ + documentId: params.documentId, + knowledgeBaseId: knowledgeBase.id, document, - knowledgeBase.id, - requestId, - params.uploadedBy ?? params.userId - ) + }) + if (bound.status === 'conflict') { + return fail('Upload id is already bound to a different document', 'conflict') + } + if (bound.status === 'bound') { + return { success: true, document: bound.document, created: false } + } + } + + try { + created = params.documentId + ? await createSingleDocument( + document, + knowledgeBase.id, + requestId, + params.uploadedBy ?? params.userId, + params.documentId + ) + : await createSingleDocument( + document, + knowledgeBase.id, + requestId, + params.uploadedBy ?? params.userId + ) } catch (error) { + if (params.documentId) { + const bound = await findBoundKnowledgeDocument({ + documentId: params.documentId, + knowledgeBaseId: knowledgeBase.id, + document, + }) + if (bound.status === 'conflict') { + return fail('Upload id is already bound to a different document', 'conflict') + } + if (bound.status === 'bound') { + return { success: true, document: bound.document, created: false } + } + } return classifyKnowledgeFailure( error, requestId, @@ -205,7 +281,7 @@ export async function performUploadKnowledgeDocument( }, }) - return { success: true, document: created } + return { success: true, document: created, created: true } } export interface PerformUploadKnowledgeDocumentsParams extends KnowledgeOperationContext { diff --git a/apps/sim/lib/table/orchestration/import-resource.ts b/apps/sim/lib/table/orchestration/import-resource.ts index 4e0902dc9db..699ac1ead3f 100644 --- a/apps/sim/lib/table/orchestration/import-resource.ts +++ b/apps/sim/lib/table/orchestration/import-resource.ts @@ -134,6 +134,7 @@ export function getOwnedTableImportUpload(params: { uploadId: params.importId, workspaceId: params.workspaceId, userId: params.userId, + purpose: 'table_import', uploadToken: params.uploadToken, }) tableImportBodyFromUpload(upload) diff --git a/apps/sim/lib/uploads/client/session-upload.test.ts b/apps/sim/lib/uploads/client/session-upload.test.ts new file mode 100644 index 00000000000..790f15ae108 --- /dev/null +++ b/apps/sim/lib/uploads/client/session-upload.test.ts @@ -0,0 +1,94 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' +import type { V2CompletedPart, V2UploadPartUrl } from '@/lib/api/contracts/v2/uploads' + +interface MultipartMockParams { + getPartUrls: (partNumbers: number[]) => Promise + complete: (parts: V2CompletedPart[]) => Promise +} + +const { mockRequestJson, mockUploadMultipartSession } = vi.hoisted(() => ({ + mockRequestJson: vi.fn(), + mockUploadMultipartSession: vi.fn(), +})) + +vi.mock('@/lib/api/client/request', () => ({ requestJson: mockRequestJson })) +vi.mock('@/lib/uploads/client/multipart-session', () => ({ + uploadMultipartSession: mockUploadMultipartSession, +})) + +import { uploadKnowledgeDocumentSession } from '@/lib/uploads/client/session-upload' + +const DOCUMENT = { + id: 'upload-1', + knowledgeBaseId: 'kb-1', + filename: 'guide.pdf', + fileSize: 1024, + mimeType: 'application/pdf', + processingStatus: 'pending', + chunkCount: 0, + tokenCount: 0, + characterCount: 0, + enabled: true, + createdAt: '2026-08-04T21:00:00.000Z', +} as const + +describe('uploadKnowledgeDocumentSession', () => { + beforeEach(() => { + vi.clearAllMocks() + mockRequestJson + .mockResolvedValueOnce({ + data: { + id: 'upload-1', + partSize: 8 * 1024 * 1024, + partCount: 1, + uploadToken: 'token', + }, + }) + .mockResolvedValueOnce({ + data: { parts: [{ partNumber: 1, url: 'https://storage.example/part-1', headers: {} }] }, + }) + .mockResolvedValueOnce({ data: { document: DOCUMENT } }) + mockUploadMultipartSession.mockImplementation( + async (params: MultipartMockParams) => { + await params.getPartUrls([1]) + return params.complete([{ partNumber: 1, etag: 'etag-1' }]) + } + ) + }) + + it('uses the first-party session routes and preserves signed processing metadata', async () => { + const file = { + name: 'guide.pdf', + type: 'application/pdf', + size: 1024, + } as File + + await expect( + uploadKnowledgeDocumentSession({ + workspaceId: '6fc7631d-88cd-46f8-9f0a-d4764daef7f8', + knowledgeBaseId: 'kb-1', + file, + tag1: 'product', + processingOptions: { recipe: 'default', lang: 'en' }, + }) + ).resolves.toEqual(DOCUMENT) + + expect(mockRequestJson.mock.calls[0][0].path).toBe('/api/knowledge/[id]/documents/uploads') + expect(mockRequestJson.mock.calls[0][1].body).toMatchObject({ + name: 'guide.pdf', + contentType: 'application/pdf', + size: 1024, + tag1: 'product', + processingOptions: { recipe: 'default', lang: 'en' }, + }) + expect(mockRequestJson.mock.calls[1][0].path).toBe( + '/api/knowledge/[id]/documents/uploads/[uploadId]/parts' + ) + expect(mockRequestJson.mock.calls[2][0].path).toBe( + '/api/knowledge/[id]/documents/uploads/[uploadId]/complete' + ) + }) +}) diff --git a/apps/sim/lib/uploads/client/session-upload.ts b/apps/sim/lib/uploads/client/session-upload.ts index eecc4237419..e06d9c57860 100644 --- a/apps/sim/lib/uploads/client/session-upload.ts +++ b/apps/sim/lib/uploads/client/session-upload.ts @@ -1,10 +1,20 @@ import { requestJson } from '@/lib/api/client/request' +import { + abortKnowledgeDocumentUploadContract, + completeKnowledgeDocumentUploadContract, + createKnowledgeDocumentUploadContract, + createKnowledgeDocumentUploadPartUrlsContract, +} from '@/lib/api/contracts/knowledge/upload-sessions' import { abortWorkspaceFileUploadContract, completeWorkspaceFileUploadContract, createWorkspaceFileUploadContract, createWorkspaceFileUploadPartUrlsContract, } from '@/lib/api/contracts/upload-sessions' +import type { + V2KnowledgeDocumentSummary, + V2KnowledgeDocumentUploadMetadata, +} from '@/lib/api/contracts/v2/knowledge' import type { UploadProgressEvent } from '@/lib/uploads/client/direct-upload' import { uploadMultipartSession } from '@/lib/uploads/client/multipart-session' import { getFileContentType } from '@/lib/uploads/utils/file-utils' @@ -17,6 +27,14 @@ interface UploadWorkspaceFileSessionParams { onProgress?: (event: UploadProgressEvent) => void } +interface UploadKnowledgeDocumentSessionParams extends V2KnowledgeDocumentUploadMetadata { + workspaceId: string + knowledgeBaseId: string + file: File + signal?: AbortSignal + onProgress?: (event: UploadProgressEvent) => void +} + export async function uploadWorkspaceFileSession(params: UploadWorkspaceFileSessionParams) { const { workspaceId, folderId, file, signal, onProgress } = params const created = await requestJson(createWorkspaceFileUploadContract, { @@ -66,3 +84,58 @@ export async function uploadWorkspaceFileSession(params: UploadWorkspaceFileSess }, }) } + +export async function uploadKnowledgeDocumentSession( + params: UploadKnowledgeDocumentSessionParams +): Promise { + const { workspaceId, knowledgeBaseId, file, signal, onProgress, ...metadata } = params + const created = await requestJson(createKnowledgeDocumentUploadContract, { + params: { id: knowledgeBaseId }, + body: { + workspaceId, + name: file.name, + contentType: getFileContentType(file), + size: file.size, + ...metadata, + }, + signal, + }) + const upload = created.data + return uploadMultipartSession({ + file, + partSize: upload.partSize, + partCount: upload.partCount, + signal, + onProgress, + getPartUrls: async (partNumbers) => { + const batch = await requestJson(createKnowledgeDocumentUploadPartUrlsContract, { + params: { id: knowledgeBaseId, uploadId: upload.id }, + query: { workspaceId }, + headers: { 'upload-token': upload.uploadToken }, + body: { partNumbers }, + signal, + }) + return batch.data.parts + }, + complete: async (parts) => { + const completed = await requestJson(completeKnowledgeDocumentUploadContract, { + params: { id: knowledgeBaseId, uploadId: upload.id }, + query: { workspaceId }, + headers: { 'upload-token': upload.uploadToken }, + body: { parts }, + signal, + }) + if (!completed.data.document) { + throw new Error('Completed upload returned no knowledge document') + } + return completed.data.document + }, + abort: async () => { + await requestJson(abortKnowledgeDocumentUploadContract, { + params: { id: knowledgeBaseId, uploadId: upload.id }, + query: { workspaceId }, + headers: { 'upload-token': upload.uploadToken }, + }) + }, + }) +} diff --git a/apps/sim/lib/uploads/core/upload-token.ts b/apps/sim/lib/uploads/core/upload-token.ts index d030655b58b..da058708536 100644 --- a/apps/sim/lib/uploads/core/upload-token.ts +++ b/apps/sim/lib/uploads/core/upload-token.ts @@ -9,6 +9,8 @@ export interface UploadTokenPayload { userId: string workspaceId: string context: StorageContext + /** Knowledge base bound to a knowledge-document multipart session. */ + knowledgeBaseId?: string /** Original file name, carried so the completion handler can record ownership metadata. */ fileName?: string /** File MIME type, carried for ownership metadata at completion. */ @@ -16,7 +18,7 @@ export interface UploadTokenPayload { /** File size in bytes, carried for ownership metadata at completion. */ fileSize?: number /** Multipart-session purpose. Omitted by the legacy multipart endpoint. */ - purpose?: 'workspace_file' | 'table_import' + purpose?: 'workspace_file' | 'table_import' | 'knowledge_document' /** Storage provider that owns the multipart upload state. */ provider?: 's3' | 'blob' | 'gcs' | 'local' /** Provider-issued multipart upload id. Local and block-blob uploads do not need one. */ @@ -44,8 +46,11 @@ const fromBase64Url = (input: string): string => Buffer.from(input, 'base64url') const sign = (payload: string): string => hmacSha256Base64(payload, env.INTERNAL_API_SECRET) /** - * Sign an upload session token binding (uploadId, key, userId, workspaceId, context). - * Used to prevent IDOR on multipart upload follow-up calls (get-part-urls, complete, abort). + * Sign an upload session token binding every supplied field to its signature. + * Multipart sessions include the caller, workspace, storage context and key, + * purpose, provider state, file metadata, part geometry, and—for knowledge + * documents—the target knowledge base. Follow-up calls reconstruct their + * complete trusted session exclusively from this signed state. */ export function signUploadToken(payload: UploadTokenPayload, expiresInSeconds = 60 * 60): string { const signed: SignedPayload = { @@ -103,10 +108,15 @@ export function verifyUploadToken(token: string): UploadTokenVerification { userId: parsed.userId, workspaceId: parsed.workspaceId, context: parsed.context as StorageContext, + ...(typeof parsed.knowledgeBaseId === 'string' + ? { knowledgeBaseId: parsed.knowledgeBaseId } + : {}), ...(typeof parsed.fileName === 'string' ? { fileName: parsed.fileName } : {}), ...(typeof parsed.contentType === 'string' ? { contentType: parsed.contentType } : {}), ...(typeof parsed.fileSize === 'number' ? { fileSize: parsed.fileSize } : {}), - ...(parsed.purpose === 'workspace_file' || parsed.purpose === 'table_import' + ...(parsed.purpose === 'workspace_file' || + parsed.purpose === 'table_import' || + parsed.purpose === 'knowledge_document' ? { purpose: parsed.purpose } : {}), ...(parsed.provider === 's3' || diff --git a/apps/sim/lib/uploads/multipart-session/service.test.ts b/apps/sim/lib/uploads/multipart-session/service.test.ts new file mode 100644 index 00000000000..3c030a9028b --- /dev/null +++ b/apps/sim/lib/uploads/multipart-session/service.test.ts @@ -0,0 +1,129 @@ +/** + * @vitest-environment node + */ +import { beforeEach, describe, expect, it, vi } from 'vitest' + +const { + mockCheckStorageQuotaForBillingContext, + mockInitiateMultipartProviderUpload, + mockResolveStorageBillingContext, +} = vi.hoisted(() => ({ + mockCheckStorageQuotaForBillingContext: vi.fn(), + mockInitiateMultipartProviderUpload: vi.fn(), + mockResolveStorageBillingContext: vi.fn(), +})) + +vi.mock('@/lib/billing/storage', () => ({ + checkStorageQuotaForBillingContext: mockCheckStorageQuotaForBillingContext, + resolveStorageBillingContext: mockResolveStorageBillingContext, +})) + +vi.mock('@/lib/uploads/core/storage-service', () => ({ headObject: vi.fn() })) + +vi.mock('@/lib/uploads/contexts/workspace', () => ({ + generateWorkspaceFileKey: vi.fn( + (workspaceId: string, fileName: string) => `workspace/${workspaceId}/${fileName}` + ), +})) + +vi.mock('@/lib/uploads/multipart-session/provider', () => ({ + abortMultipartProviderUpload: vi.fn(), + completeMultipartProviderUpload: vi.fn(), + getMultipartProviderPartUrls: vi.fn(), + initiateMultipartProviderUpload: mockInitiateMultipartProviderUpload, +})) + +import { + createUploadSession, + getOwnedUploadSession, + verifyUploadSessionToken, +} from '@/lib/uploads/multipart-session/service' + +const WORKSPACE_ID = '6fc7631d-88cd-46f8-9f0a-d4764daef7f8' + +describe('knowledge-document multipart sessions', () => { + beforeEach(() => { + vi.clearAllMocks() + mockResolveStorageBillingContext.mockResolvedValue({ workspaceId: WORKSPACE_ID }) + mockCheckStorageQuotaForBillingContext.mockResolvedValue({ allowed: true }) + mockInitiateMultipartProviderUpload.mockResolvedValue({ + provider: 's3', + providerUploadId: 'provider-upload-1', + }) + }) + + it('binds knowledge ownership and all storage state into the signed token', async () => { + const created = await createUploadSession({ + id: 'upload-1', + workspaceId: WORKSPACE_ID, + userId: 'user-1', + knowledgeBaseId: 'kb-1', + purpose: 'knowledge_document', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + }) + + const verified = verifyUploadSessionToken(created.uploadToken) + expect(verified).toMatchObject({ + id: 'upload-1', + workspaceId: WORKSPACE_ID, + userId: 'user-1', + knowledgeBaseId: 'kb-1', + purpose: 'knowledge_document', + storageContext: 'knowledge-base', + storageProvider: 's3', + providerUploadId: 'provider-upload-1', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + }) + expect(verified.storageKey).toMatch(/^kb\/.*-guide\.pdf$/) + }) + + it.each([ + { userId: 'other-user', knowledgeBaseId: 'kb-1', purpose: 'knowledge_document' as const }, + { userId: 'user-1', knowledgeBaseId: 'kb-2', purpose: 'knowledge_document' as const }, + { userId: 'user-1', knowledgeBaseId: 'kb-1', purpose: 'workspace_file' as const }, + ])('rejects a session whose signed scope does not match $purpose', async (scope) => { + const created = await createUploadSession({ + id: 'upload-1', + workspaceId: WORKSPACE_ID, + userId: 'user-1', + knowledgeBaseId: 'kb-1', + purpose: 'knowledge_document', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + }) + + expect(() => + getOwnedUploadSession({ + uploadId: 'upload-1', + workspaceId: WORKSPACE_ID, + uploadToken: created.uploadToken, + ...scope, + }) + ).toThrow('Upload session not found') + }) + + it('runs the storage quota gate before creating provider state', async () => { + mockCheckStorageQuotaForBillingContext.mockResolvedValue({ + allowed: false, + error: 'Storage limit exceeded', + }) + + await expect( + createUploadSession({ + workspaceId: WORKSPACE_ID, + userId: 'user-1', + knowledgeBaseId: 'kb-1', + purpose: 'knowledge_document', + fileName: 'guide.pdf', + contentType: 'application/pdf', + fileSize: 1024, + }) + ).rejects.toMatchObject({ code: 'payload_too_large' }) + expect(mockInitiateMultipartProviderUpload).not.toHaveBeenCalled() + }) +}) diff --git a/apps/sim/lib/uploads/multipart-session/service.ts b/apps/sim/lib/uploads/multipart-session/service.ts index fba4eb353ef..2157091cb73 100644 --- a/apps/sim/lib/uploads/multipart-session/service.ts +++ b/apps/sim/lib/uploads/multipart-session/service.ts @@ -4,6 +4,7 @@ import { resolveStorageBillingContext, } from '@/lib/billing/storage' import { OrchestrationError } from '@/lib/core/orchestration/types' +import { generateKnowledgeBaseFileKey } from '@/lib/uploads/contexts/knowledge-base/knowledge-base-file-manager' import { generateWorkspaceFileKey } from '@/lib/uploads/contexts/workspace' import { headObject } from '@/lib/uploads/core/storage-service' import { signUploadToken, verifyUploadToken } from '@/lib/uploads/core/upload-token' @@ -16,20 +17,25 @@ import { type MultipartPartUrl, type MultipartStorageProvider, } from '@/lib/uploads/multipart-session/provider' -import { MAX_WORKSPACE_FILE_SIZE, type StorageContext } from '@/lib/uploads/shared/types' +import { + MAX_KNOWLEDGE_DOCUMENT_FILE_SIZE, + MAX_WORKSPACE_FILE_SIZE, + type StorageContext, +} from '@/lib/uploads/shared/types' import { sanitizeFileName } from '@/executor/constants' export const MULTIPART_SESSION_PART_SIZE = 8 * 1024 * 1024 export const MULTIPART_SESSION_MAX_PART_URLS = 100 export const MULTIPART_SESSION_TTL_MS = 24 * 60 * 60 * 1000 -export type UploadSessionPurpose = 'workspace_file' | 'table_import' +export type UploadSessionPurpose = 'workspace_file' | 'table_import' | 'knowledge_document' export type UploadSessionStatus = 'uploading' | 'completed' | 'aborted' export interface UploadSessionRecord { id: string workspaceId: string userId: string + knowledgeBaseId: string | null purpose: UploadSessionPurpose storageContext: StorageContext storageKey: string @@ -61,31 +67,31 @@ export class UploadSessionError extends OrchestrationError { } } -interface CreateUploadSessionParams { +interface CreateUploadSessionBaseParams { id?: string workspaceId: string userId: string - purpose: UploadSessionPurpose fileName: string contentType: string fileSize: number metadata?: Record } +type CreateUploadSessionParams = CreateUploadSessionBaseParams & + ( + | { purpose: 'workspace_file' | 'table_import'; knowledgeBaseId?: never } + | { purpose: 'knowledge_document'; knowledgeBaseId: string } + ) + export async function createUploadSession( params: CreateUploadSessionParams ): Promise { - validateFileSize(params.fileSize) + validateFile(params) const id = params.id ?? generateId() - const storageContext: StorageContext = - params.purpose === 'workspace_file' ? 'workspace' : 'table-import' - const storageKey = - params.purpose === 'workspace_file' - ? generateWorkspaceFileKey(params.workspaceId, params.fileName) - : `table-import/${params.workspaceId}/${id}/${sanitizeFileName(params.fileName)}` + const { storageContext, storageKey } = resolveUploadStorage(params, id) const partCount = Math.ceil(params.fileSize / MULTIPART_SESSION_PART_SIZE) - if (params.purpose === 'workspace_file') { + if (params.purpose === 'workspace_file' || params.purpose === 'knowledge_document') { const billingContext = await resolveStorageBillingContext(params.workspaceId) const quota = await checkStorageQuotaForBillingContext(billingContext, params.fileSize) if (!quota.allowed) { @@ -111,6 +117,9 @@ export async function createUploadSession( userId: params.userId, workspaceId: params.workspaceId, context: storageContext, + ...(params.purpose === 'knowledge_document' + ? { knowledgeBaseId: params.knowledgeBaseId } + : {}), fileName: params.fileName, contentType: params.contentType, fileSize: params.fileSize, @@ -130,6 +139,7 @@ export async function createUploadSession( id, workspaceId: params.workspaceId, userId: params.userId, + knowledgeBaseId: params.purpose === 'knowledge_document' ? params.knowledgeBaseId : null, purpose: params.purpose, storageContext, storageKey, @@ -156,6 +166,8 @@ export function getOwnedUploadSession(params: { uploadId: string workspaceId: string userId?: string + purpose: UploadSessionPurpose + knowledgeBaseId?: string uploadToken: string }): UploadSessionRecord { const session = verifyUploadSessionToken(params.uploadToken) @@ -165,6 +177,12 @@ export function getOwnedUploadSession(params: { if (params.userId && session.userId !== params.userId) { throw new UploadSessionError('not_found', 'Upload session not found') } + if (session.purpose !== params.purpose) { + throw new UploadSessionError('not_found', 'Upload session not found') + } + if (params.knowledgeBaseId !== undefined && session.knowledgeBaseId !== params.knowledgeBaseId) { + throw new UploadSessionError('not_found', 'Upload session not found') + } return session } @@ -188,9 +206,25 @@ export function verifyUploadSessionToken(uploadToken: string): UploadSessionReco ) { throw new UploadSessionError('forbidden', 'Upload token is not a multipart session token') } - if (payload.context !== 'workspace' && payload.context !== 'table-import') { + if ( + payload.context !== 'workspace' && + payload.context !== 'table-import' && + payload.context !== 'knowledge-base' + ) { throw new UploadSessionError('forbidden', 'Upload token has an invalid storage context') } + const knowledgeBaseId = payload.knowledgeBaseId?.trim() || null + if ( + (payload.purpose === 'workspace_file' && payload.context !== 'workspace') || + (payload.purpose === 'table_import' && payload.context !== 'table-import') || + (payload.purpose === 'knowledge_document' && + (payload.context !== 'knowledge-base' || !knowledgeBaseId || !payload.key.startsWith('kb/'))) + ) { + throw new UploadSessionError('forbidden', 'Upload token purpose does not match its storage') + } + if (payload.purpose !== 'knowledge_document' && knowledgeBaseId) { + throw new UploadSessionError('forbidden', 'Upload token has unexpected knowledge-base state') + } const createdAt = new Date(payload.createdAt) const expiresAt = new Date(payload.expiresAt) if (!Number.isFinite(createdAt.getTime()) || !Number.isFinite(expiresAt.getTime())) { @@ -201,6 +235,7 @@ export function verifyUploadSessionToken(uploadToken: string): UploadSessionReco id: payload.uploadId, workspaceId: payload.workspaceId, userId: payload.userId, + knowledgeBaseId, purpose: payload.purpose, storageContext: payload.context, storageKey: payload.key, @@ -373,14 +408,47 @@ function validateCompletedParts(session: UploadSessionRecord, parts: CompletedUp } } -function validateFileSize(fileSize: number): void { - if (!Number.isSafeInteger(fileSize) || fileSize < 1) { +function validateFile(params: CreateUploadSessionParams): void { + if (!params.fileName.trim()) { + throw new UploadSessionError('validation', 'fileName must not be empty') + } + if (!params.contentType.trim()) { + throw new UploadSessionError('validation', 'contentType must not be empty') + } + if (!Number.isSafeInteger(params.fileSize) || params.fileSize < 1) { throw new UploadSessionError('validation', 'fileSize must be a positive integer') } - if (fileSize > MAX_WORKSPACE_FILE_SIZE) { - throw new UploadSessionError( - 'validation', - `File size exceeds maximum of ${MAX_WORKSPACE_FILE_SIZE} bytes` - ) + const maximum = + params.purpose === 'knowledge_document' + ? MAX_KNOWLEDGE_DOCUMENT_FILE_SIZE + : MAX_WORKSPACE_FILE_SIZE + if (params.fileSize > maximum) { + throw new UploadSessionError('validation', `File size exceeds maximum of ${maximum} bytes`) + } + if (params.purpose === 'knowledge_document' && !params.knowledgeBaseId.trim()) { + throw new UploadSessionError('validation', 'knowledgeBaseId must not be empty') + } +} + +function resolveUploadStorage( + params: CreateUploadSessionParams, + id: string +): { storageContext: StorageContext; storageKey: string } { + switch (params.purpose) { + case 'workspace_file': + return { + storageContext: 'workspace', + storageKey: generateWorkspaceFileKey(params.workspaceId, params.fileName), + } + case 'table_import': + return { + storageContext: 'table-import', + storageKey: `table-import/${params.workspaceId}/${id}/${sanitizeFileName(params.fileName)}`, + } + case 'knowledge_document': + return { + storageContext: 'knowledge-base', + storageKey: generateKnowledgeBaseFileKey(params.fileName), + } } } diff --git a/apps/sim/lib/uploads/providers/blob/client.test.ts b/apps/sim/lib/uploads/providers/blob/client.test.ts index 19b06c93a91..f31cf572108 100644 --- a/apps/sim/lib/uploads/providers/blob/client.test.ts +++ b/apps/sim/lib/uploads/providers/blob/client.test.ts @@ -51,6 +51,7 @@ vi.mock('@/lib/uploads/config', () => ({ })) import { + abortMultipartUpload, deleteFromBlob, downloadFromBlob, getPresignedUrl, @@ -194,6 +195,15 @@ describe('Azure Blob Storage Client', () => { }) }) + describe('abortMultipartUpload', () => { + it('leaves the blob key untouched while Azure garbage-collects uncommitted blocks', async () => { + await abortMultipartUpload('test-file-key') + + expect(mockGetBlockBlobClient).not.toHaveBeenCalled() + expect(mockDeleteIfExists).not.toHaveBeenCalled() + }) + }) + describe('getPresignedUrl', () => { it('should generate a presigned URL for Azure Blob Storage', async () => { const testKey = 'test-file-key' diff --git a/apps/sim/lib/uploads/providers/blob/client.ts b/apps/sim/lib/uploads/providers/blob/client.ts index 076933db181..7f2faf821e3 100644 --- a/apps/sim/lib/uploads/providers/blob/client.ts +++ b/apps/sim/lib/uploads/providers/blob/client.ts @@ -745,40 +745,12 @@ export async function completeMultipartUpload( } /** - * Abort multipart upload by deleting the blob if it exists + * Abandons an Azure multipart upload without deleting its key. + * + * Azure has no cancellation operation for staged blocks and garbage-collects uncommitted blocks + * after a week. Deleting by key is unsafe because a concurrent completion may already have + * committed the final blob at that key. */ -export async function abortMultipartUpload(key: string, customConfig?: BlobConfig): Promise { - const { BlobServiceClient, StorageSharedKeyCredential } = await import('@azure/storage-blob') - let blobServiceClient: BlobServiceClientType - let containerName: string - - if (customConfig) { - if (customConfig.connectionString) { - blobServiceClient = BlobServiceClient.fromConnectionString(customConfig.connectionString) - } else if (customConfig.accountName && customConfig.accountKey) { - const credential = new StorageSharedKeyCredential( - customConfig.accountName, - customConfig.accountKey - ) - blobServiceClient = new BlobServiceClient( - `https://${customConfig.accountName}.blob.core.windows.net`, - credential - ) - } else { - throw new Error('Invalid custom blob configuration') - } - containerName = customConfig.containerName - } else { - blobServiceClient = await getBlobServiceClient() - containerName = BLOB_CONFIG.containerName - } - - const containerClient = blobServiceClient.getContainerClient(containerName) - const blockBlobClient = containerClient.getBlockBlobClient(key) - - try { - await blockBlobClient.deleteIfExists() - } catch (error) { - logger.warn('Error cleaning up multipart upload:', error) - } +export function abortMultipartUpload(_key: string, _customConfig?: BlobConfig): Promise { + return Promise.resolve() } diff --git a/apps/sim/lib/uploads/shared/types.ts b/apps/sim/lib/uploads/shared/types.ts index 0c24f770444..d7de7961ee5 100644 --- a/apps/sim/lib/uploads/shared/types.ts +++ b/apps/sim/lib/uploads/shared/types.ts @@ -22,6 +22,9 @@ export function toLegacyWorkspaceFileSize(size: number): number { */ export const MAX_WORKSPACE_FORMDATA_FILE_SIZE = 100 * 1024 * 1024 +/** Maximum size accepted by the knowledge-document parsing pipeline. */ +export const MAX_KNOWLEDGE_DOCUMENT_FILE_SIZE = 100 * 1024 * 1024 + export type StorageContext = | 'knowledge-base' | 'chat' diff --git a/scripts/check-api-validation-contracts.ts b/scripts/check-api-validation-contracts.ts index 0a8378abdae..001621c4f3a 100644 --- a/scripts/check-api-validation-contracts.ts +++ b/scripts/check-api-validation-contracts.ts @@ -9,8 +9,8 @@ const QUERY_HOOKS_DIR = path.join(ROOT, 'apps/sim/hooks/queries') const SELECTOR_HOOKS_DIR = path.join(ROOT, 'apps/sim/hooks/selectors') const BASELINE = { - totalRoutes: 1083, - zodRoutes: 1083, + totalRoutes: 1091, + zodRoutes: 1091, nonZodRoutes: 0, } as const