import { OPENCLAW_CHAT_VIDEO_UPLOAD_PURPOSE } from "@/lib/openclaw-chat/media-attachments"; const DEFAULT_MAX_RETRIES = 3; const MAX_BACKOFF_MS = 8000; const MAX_CONCURRENT_UPLOADS = 3; const DIRECT_UPLOAD_MIN_BYTES = 1; const DIRECT_UPLOAD_MAX_BYTES = 500 * 1024 * 1024; const SERVER_UPLOAD_MAX_BYTES = 50 * 1024 * 1024; const DIRECT_UPLOAD_PART_CONCURRENCY = 4; const DIRECT_UPLOAD_SUPPRESS_MS = 2 * 60 * 1000; const CLIENT_UPLOAD_ID_FIELD = "clientUploadId"; const DIRECT_UPLOAD_INITIATE_TIMEOUT_MS = 3 * 1000; const DIRECT_UPLOAD_ACTION_TIMEOUT_MS = 15 * 1000; const DIRECT_UPLOAD_ABORT_TIMEOUT_MS = 8 * 1000; const DIRECT_UPLOAD_STALL_TIMEOUT_MS = 10 * 1000; const XHR_UPLOAD_TOTAL_TIMEOUT_MS = 6 * 60 * 1000; const XHR_UPLOAD_STALL_TIMEOUT_MS = 30 * 1000; const RETRYABLE_UPLOAD_STATUSES = new Set([ 408, 425, 429, 500, 502, 503, 504, ]); const DIRECT_UPLOAD_IMAGE_TYPES = new Set([ "image/jpeg", "image/jpg", "image/png", "image/webp", ]); const DIRECT_UPLOAD_OPENCLAW_VIDEO_TYPES = new Set([ "video/mp4", "video/quicktime", "video/webm", "video/x-m4v", "video/x-matroska", ]); const BROWSER_IMAGE_TYPES_BY_EXTENSION: Record = { jpg: "image/jpeg", jpeg: "image/jpeg", png: "image/png", webp: "image/webp", }; let activeUploadCount = 0; const uploadQueue: Array<() => void> = []; let directUploadSuppressedUntil = 0; export type UploadProgressPhase = | "preparing" | "direct" | "multipart" | "server" | "finalizing" | "complete"; export type UploadProgress = { loaded: number; total: number; percent: number; phase: UploadProgressPhase; }; type UploadProgressHandler = (progress: UploadProgress) => void; type UploadFormFile = { blob: Blob; filename: string; contentType: string; purpose: string; clientUploadId: string; }; type DirectUploadPart = { partNumber: number; url: string; }; type DirectUploadInitiateResponse = | { mode: "single"; key: string; url: string; headers?: Record; } | { mode: "multipart"; key: string; uploadId: string; partSize: number; partCount: number; parts: DirectUploadPart[]; }; function createAbortError() { const error = new Error("Upload aborted"); error.name = "AbortError"; return error; } function createTimeoutError(message: string) { const error = new Error(message); error.name = "TimeoutError"; return error; } function sleep(ms: number, signal?: AbortSignal) { if (signal?.aborted) { return Promise.reject(createAbortError()); } return new Promise((resolve, reject) => { const timeout = globalThis.setTimeout(() => { signal?.removeEventListener("abort", onAbort); resolve(); }, ms); const onAbort = () => { globalThis.clearTimeout(timeout); reject(createAbortError()); }; signal?.addEventListener("abort", onAbort, { once: true }); }); } function acquireUploadSlot() { return new Promise((resolve) => { const start = () => { activeUploadCount += 1; resolve(); }; if (activeUploadCount < MAX_CONCURRENT_UPLOADS) { start(); } else { uploadQueue.push(start); } }); } async function withUploadSlot(task: () => Promise): Promise { await acquireUploadSlot(); try { return await task(); } finally { activeUploadCount = Math.max(0, activeUploadCount - 1); uploadQueue.shift()?.(); } } function parseRetryAfterMs(value: string | null): number | null { if (!value) return null; const seconds = Number(value); if (Number.isFinite(seconds) && seconds > 0) { return Math.round(seconds * 1000); } return null; } function computeBackoffMs(attempt: number, retryAfterMs: number | null) { if (retryAfterMs && retryAfterMs > 0) { return Math.min(retryAfterMs, MAX_BACKOFF_MS); } const exp = Math.min(MAX_BACKOFF_MS, 500 * 2 ** attempt); const jitter = Math.floor(Math.random() * 300); return exp + jitter; } function isRetryableUploadStatus(status: number) { return RETRYABLE_UPLOAD_STATUSES.has(status); } function isAbortError(error: unknown, signal?: AbortSignal) { return ( signal?.aborted || (error instanceof Error && error.name === "AbortError") ); } function clampPercent(value: number, max = 99) { if (!Number.isFinite(value)) return 0; return Math.max(0, Math.min(max, Math.round(value))); } function emitUploadProgress( onProgress: UploadProgressHandler | undefined, loaded: number, total: number, phase: UploadProgressPhase, options?: { percent?: number; maxPercent?: number } ) { if (!onProgress) return; const safeLoaded = Math.max(0, Math.round(loaded || 0)); const safeTotal = Math.max(0, Math.round(total || 0)); const computedPercent = typeof options?.percent === "number" ? options.percent : safeTotal > 0 ? (safeLoaded / safeTotal) * 100 : 0; const maxPercent = options?.maxPercent ?? (phase === "complete" ? 100 : 99); onProgress({ loaded: safeLoaded, total: safeTotal, percent: clampPercent(computedPercent, maxPercent), phase, }); } function cloneFormData(formData: FormData) { const copy = new FormData(); formData.forEach((value, key) => { if (typeof value === "string") { copy.append(key, value); return; } const filename = typeof File !== "undefined" && value instanceof File ? value.name : undefined; if (filename) { copy.append(key, value, filename); } else { copy.append(key, value); } }); return copy; } function normalizeBrowserImageType(contentType: string, filename: string) { const normalized = contentType.trim().toLowerCase(); if (normalized && normalized !== "application/octet-stream") { return normalized; } const extension = filename.toLowerCase().split(".").pop() || ""; return BROWSER_IMAGE_TYPES_BY_EXTENSION[extension] || normalized; } function createClientUploadId() { if ( typeof crypto !== "undefined" && typeof crypto.randomUUID === "function" ) { return crypto.randomUUID(); } return `upload-${Date.now()}-${Math.random().toString(36).slice(2, 10)}`; } function ensureClientUploadId(formData: FormData) { const existing = formData.get(CLIENT_UPLOAD_ID_FIELD); if (typeof existing === "string" && existing.trim()) { return existing.trim(); } const clientUploadId = createClientUploadId(); formData.set(CLIENT_UPLOAD_ID_FIELD, clientUploadId); return clientUploadId; } function getUploadFormFile(formData: FormData): UploadFormFile | null { const value = formData.get("file"); if (!(value instanceof Blob)) { return null; } const filename = typeof File !== "undefined" && value instanceof File && value.name ? value.name : `upload-${Date.now()}`; const contentType = normalizeBrowserImageType( value.type || "application/octet-stream", filename ); const purposeValue = formData.get("purpose"); const clientUploadId = ensureClientUploadId(formData); return { blob: value, filename, contentType, purpose: typeof purposeValue === "string" ? purposeValue : "", clientUploadId, }; } function shouldTryDirectUpload(file: UploadFormFile) { if ( Date.now() < directUploadSuppressedUntil && file.blob.size <= SERVER_UPLOAD_MAX_BYTES ) { return false; } if (file.blob.size <= 0 || file.blob.size > DIRECT_UPLOAD_MAX_BYTES) { return false; } const normalizedContentType = file.contentType.toLowerCase(); const supportsDirectUpload = DIRECT_UPLOAD_IMAGE_TYPES.has(normalizedContentType) || (file.purpose === OPENCLAW_CHAT_VIDEO_UPLOAD_PURPOSE && DIRECT_UPLOAD_OPENCLAW_VIDEO_TYPES.has(normalizedContentType)); if (!supportsDirectUpload) { return false; } if (file.purpose === "agent-workflow-preview") { return false; } return ( file.purpose === "canvas-upload" || file.blob.size >= DIRECT_UPLOAD_MIN_BYTES ); } function suppressDirectUploadTemporarily() { directUploadSuppressedUntil = Date.now() + DIRECT_UPLOAD_SUPPRESS_MS; } async function readErrorPayload(response: Response) { try { const payload = await response.clone().json(); return payload?.error || payload?.details || response.statusText; } catch { return response.statusText; } } async function fetchWithControlTimeout( input: RequestInfo | URL, init: RequestInit, timeoutMs: number, signal?: AbortSignal ) { if (signal?.aborted) { throw createAbortError(); } const controller = new AbortController(); let timeout: ReturnType | null = null; let timedOut = false; let settled = false; const onAbort = () => { controller.abort(); }; signal?.addEventListener("abort", onAbort, { once: true }); try { const request = fetch(input, { ...init, signal: controller.signal, }).catch((error) => { if (signal?.aborted) { throw createAbortError(); } if (timedOut) { throw createTimeoutError( `Direct upload control request timed out after ${Math.round( timeoutMs / 1000 )}s` ); } throw error; }); const timeoutPromise = new Promise((_, reject) => { timeout = setTimeout(() => { timedOut = true; controller.abort(); if (!settled) { reject( createTimeoutError( `Direct upload control request timed out after ${Math.round( timeoutMs / 1000 )}s` ) ); } }, timeoutMs); }); return await Promise.race([request, timeoutPromise]); } finally { settled = true; if (timeout) { clearTimeout(timeout); } signal?.removeEventListener("abort", onAbort); } } async function postDirectUploadAction( payload: Record, signal?: AbortSignal, timeoutMs = DIRECT_UPLOAD_ACTION_TIMEOUT_MS ): Promise { const response = await fetchWithControlTimeout( "/api/files/direct-upload", { method: "POST", headers: { "Content-Type": "application/json", }, body: JSON.stringify(payload), }, timeoutMs, signal ); if (!response.ok) { throw new Error(String(await readErrorPayload(response))); } return (await response.json()) as T; } function createResponseFromXhr(xhr: XMLHttpRequest) { const headers = new Headers(); const rawHeaders = xhr.getAllResponseHeaders(); for (const line of rawHeaders.trim().split(/[\r\n]+/)) { if (!line) continue; const separatorIndex = line.indexOf(":"); if (separatorIndex <= 0) continue; headers.append( line.slice(0, separatorIndex).trim(), line.slice(separatorIndex + 1).trim() ); } return new Response(xhr.responseText || null, { status: xhr.status, statusText: xhr.statusText, headers, }); } function createJsonResponse(payload: unknown, init?: ResponseInit) { const headers = new Headers(init?.headers); if (!headers.has("Content-Type")) { headers.set("Content-Type", "application/json"); } return new Response(JSON.stringify(payload), { ...init, headers, }); } function normalizeXhrHeaders(headers?: HeadersInit) { if (!headers) return [] as Array<[string, string]>; if (headers instanceof Headers) { return Array.from(headers.entries()); } if (Array.isArray(headers)) { return headers.map(([key, value]) => [key, String(value)] as [string, string]); } return Object.entries(headers).map( ([key, value]) => [key, String(value)] as [string, string] ); } function xhrUpload( url: string, init: { method: string; headers?: HeadersInit; body: XMLHttpRequestBodyInit; }, options?: { signal?: AbortSignal; onProgress?: (loaded: number, total: number) => void; timeoutMs?: number; stallTimeoutMs?: number; } ) { return new Promise((resolve, reject) => { if (options?.signal?.aborted) { reject(createAbortError()); return; } const xhr = new XMLHttpRequest(); let settled = false; let stallTimer: ReturnType | null = null; const clearStallTimer = () => { if (!stallTimer) return; clearTimeout(stallTimer); stallTimer = null; }; const settle = (callback: () => void) => { if (settled) return; settled = true; clearStallTimer(); options?.signal?.removeEventListener("abort", onAbort); callback(); }; const onAbort = () => { xhr.abort(); settle(() => reject(createAbortError())); }; xhr.open(init.method, url, true); xhr.timeout = options?.timeoutMs ?? XHR_UPLOAD_TOTAL_TIMEOUT_MS; for (const [key, value] of normalizeXhrHeaders(init.headers)) { xhr.setRequestHeader(key, value); } const resetStallTimer = () => { clearStallTimer(); if (settled) return; const stallTimeoutMs = options?.stallTimeoutMs ?? XHR_UPLOAD_STALL_TIMEOUT_MS; if (stallTimeoutMs <= 0) return; stallTimer = setTimeout(() => { const error = new Error( `Upload stalled for ${Math.round(stallTimeoutMs / 1000)}s` ); error.name = "TimeoutError"; settle(() => reject(error)); xhr.abort(); }, stallTimeoutMs); }; xhr.upload.onprogress = (event) => { resetStallTimer(); const total = event.lengthComputable ? event.total : 0; options?.onProgress?.(event.loaded, total); }; xhr.onload = () => { settle(() => resolve(createResponseFromXhr(xhr))); }; xhr.onerror = () => { settle(() => reject(new TypeError("Network request failed"))); }; xhr.ontimeout = () => { settle(() => reject(new Error("Upload request timed out"))); }; xhr.onabort = () => { settle(() => reject(createAbortError())); }; options?.signal?.addEventListener("abort", onAbort, { once: true }); resetStallTimer(); xhr.send(init.body); }); } async function fetchDirectPartWithRetry( url: string, init: { method: string; headers?: HeadersInit; body: XMLHttpRequestBodyInit; }, maxRetries: number, signal?: AbortSignal, onProgress?: (loaded: number, total: number) => void, options?: { stallTimeoutMs?: number } ) { for (let attempt = 0; ; attempt += 1) { try { const response = await xhrUpload( url, init, { onProgress, signal, stallTimeoutMs: options?.stallTimeoutMs, } ); if (response.ok || !isRetryableUploadStatus(response.status) || attempt >= maxRetries) { return response; } await sleep(computeBackoffMs(attempt, null), signal); } catch (error) { if (isAbortError(error, signal) || attempt >= maxRetries) { throw error; } await sleep(computeBackoffMs(attempt, null), signal); } } } async function fetchServerUploadWithRetry( formData: FormData, maxRetries: number, signal?: AbortSignal, onProgress?: UploadProgressHandler ) { for (let attempt = 0; ; attempt++) { try { const response = await xhrUpload( "/api/files/upload", { method: "POST", body: cloneFormData(formData), }, { signal, onProgress: (loaded, total) => { emitUploadProgress(onProgress, loaded, total, "server"); }, } ); if (response.ok || !isRetryableUploadStatus(response.status) || attempt >= maxRetries) { return response; } let retryAfterMs = parseRetryAfterMs(response.headers.get("Retry-After")); if (!retryAfterMs) { try { const payload = await response.clone().json(); const retryAfterSec = Number(payload?.retryAfter || 0); if (Number.isFinite(retryAfterSec) && retryAfterSec > 0) { retryAfterMs = Math.round(retryAfterSec * 1000); } } catch {} } await sleep(computeBackoffMs(attempt, retryAfterMs), signal); } catch (error) { if (isAbortError(error, signal) || attempt >= maxRetries) { throw error; } await sleep(computeBackoffMs(attempt, null), signal); } } } async function uploadMultipartParts(params: { file: UploadFormFile; upload: Extract; maxRetries: number; signal?: AbortSignal; onProgress?: UploadProgressHandler; }) { const uploadedParts: Array<{ ETag: string; PartNumber: number }> = []; const partLoadedBytes = new Map(); let nextIndex = 0; const reportPartProgress = (partNumber: number, loaded: number, total: number) => { partLoadedBytes.set(partNumber, Math.min(loaded, total || loaded)); const uploadedBytes = Array.from(partLoadedBytes.values()).reduce( (sum, value) => sum + value, 0 ); emitUploadProgress( params.onProgress, uploadedBytes, params.file.blob.size, "multipart" ); }; async function worker() { for (;;) { const part = params.upload.parts[nextIndex]; nextIndex += 1; if (!part) return; const start = (part.partNumber - 1) * params.upload.partSize; const end = Math.min(start + params.upload.partSize, params.file.blob.size); const body = params.file.blob.slice(start, end); const response = await fetchDirectPartWithRetry( part.url, { method: "PUT", body, }, params.maxRetries, params.signal, (loaded, total) => { reportPartProgress(part.partNumber, loaded, total || body.size); }, { stallTimeoutMs: DIRECT_UPLOAD_STALL_TIMEOUT_MS } ); if (!response.ok) { throw new Error(`Direct multipart upload failed: ${response.status}`); } reportPartProgress(part.partNumber, body.size, body.size); const etag = response.headers.get("ETag") || response.headers.get("etag"); if (!etag) { throw new Error("Storage did not expose ETag; please allow ETag in bucket CORS"); } uploadedParts.push({ ETag: etag, PartNumber: part.partNumber, }); } } await Promise.all( Array.from( { length: Math.min(DIRECT_UPLOAD_PART_CONCURRENCY, params.upload.parts.length) }, () => worker() ) ); return uploadedParts.sort((a, b) => a.PartNumber - b.PartNumber); } async function fetchDirectUpload( formData: FormData, maxRetries: number, signal?: AbortSignal, onProgress?: UploadProgressHandler ): Promise { const file = getUploadFormFile(formData); if (!file || !shouldTryDirectUpload(file)) { return null; } let multipartUpload: | Extract | null = null; try { emitUploadProgress(onProgress, 0, file.blob.size, "preparing", { percent: 1, }); const upload = await postDirectUploadAction( { action: "initiate", filename: file.filename, contentType: file.contentType, size: file.blob.size, purpose: file.purpose, clientUploadId: file.clientUploadId, }, signal, DIRECT_UPLOAD_INITIATE_TIMEOUT_MS ); if (upload.mode === "single") { emitUploadProgress(onProgress, 0, file.blob.size, "direct", { percent: 2, }); const response = await fetchDirectPartWithRetry( upload.url, { method: "PUT", headers: upload.headers || { "Content-Type": file.contentType, }, body: file.blob, }, maxRetries, signal, (loaded, total) => { emitUploadProgress(onProgress, loaded, total || file.blob.size, "direct"); }, { stallTimeoutMs: DIRECT_UPLOAD_STALL_TIMEOUT_MS } ); if (!response.ok) { throw new Error(`Direct upload failed: ${response.status}`); } emitUploadProgress(onProgress, file.blob.size, file.blob.size, "finalizing", { percent: 99, }); const payload = await postDirectUploadAction>( { action: "complete", key: upload.key, filename: file.filename, contentType: file.contentType, size: file.blob.size, purpose: file.purpose, clientUploadId: file.clientUploadId, }, signal ); emitUploadProgress(onProgress, file.blob.size, file.blob.size, "complete", { percent: 100, }); return createJsonResponse(payload); } multipartUpload = upload; emitUploadProgress(onProgress, 0, file.blob.size, "multipart", { percent: 2, }); const parts = await uploadMultipartParts({ file, upload, maxRetries, signal, onProgress, }); emitUploadProgress(onProgress, file.blob.size, file.blob.size, "finalizing", { percent: 99, }); const payload = await postDirectUploadAction>( { action: "complete", key: upload.key, uploadId: upload.uploadId, filename: file.filename, contentType: file.contentType, size: file.blob.size, parts, purpose: file.purpose, clientUploadId: file.clientUploadId, }, signal ); emitUploadProgress(onProgress, file.blob.size, file.blob.size, "complete", { percent: 100, }); return createJsonResponse(payload); } catch (error) { if (multipartUpload && !isAbortError(error, signal)) { postDirectUploadAction( { action: "abort", key: multipartUpload.key, uploadId: multipartUpload.uploadId, clientUploadId: file.clientUploadId, }, signal, DIRECT_UPLOAD_ABORT_TIMEOUT_MS ).catch(() => {}); } if (!isAbortError(error, signal)) { if (file.blob.size > SERVER_UPLOAD_MAX_BYTES) { return createJsonResponse( { error: "文件超过 50MB,必须使用浏览器直传;当前直传失败,请刷新后重试,或联系管理员检查存储桶 CORS/ETag 配置。", details: error instanceof Error ? error.message : String(error), clientUploadId: file.clientUploadId, }, { status: 502 } ); } console.warn("[Upload] Direct upload failed; falling back to server upload", error); suppressDirectUploadTemporarily(); return null; } throw error; } } export async function fetchUploadWithRetry( formData: FormData, options?: { maxRetries?: number; signal?: AbortSignal; onProgress?: UploadProgressHandler; } ) { const maxRetries = options?.maxRetries ?? DEFAULT_MAX_RETRIES; return withUploadSlot(async () => { const directResponse = await fetchDirectUpload( formData, maxRetries, options?.signal, options?.onProgress ); if (directResponse) { return directResponse; } const response = await fetchServerUploadWithRetry( formData, maxRetries, options?.signal, options?.onProgress ); if (response.ok) { options?.onProgress?.({ loaded: 1, total: 1, percent: 100, phase: "complete", }); } return response; }); }