From d55cd011b76896eea9b455e127d71b70b327d2a8 Mon Sep 17 00:00:00 2001 From: yokowu <18836617@qq.com> Date: Mon, 19 Oct 2026 19:27:14 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81=E8=B5=84=E6=BA=90=E5=8C=85?= =?UTF-8?q?=E5=88=86=E5=8D=B7=E5=AF=BC=E5=85=A5=E4=B8=8E=E6=96=AD=E7=82=B9?= =?UTF-8?q?=E7=BB=AD=E4=BC=A0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- monkeyai/admin/nginx/app.conf | 15 + .../src/components/resource-import-dialog.tsx | 298 ++++++++++++ monkeyai/admin/src/lib/resource-upload.ts | 172 +++++++ .../admin/src/pages/resource-imports-page.tsx | 428 +++-------------- monkeyai/admin/test/resource-api.test.mjs | 10 + monkeyai/admin/test/resource-upload.test.mjs | 87 ++++ monkeyai/backend/api/admin.yaml | 145 +++++- monkeyai/backend/internal/app/app.go | 15 +- monkeyai/backend/internal/app/billing.go | 22 +- monkeyai/backend/internal/app/billing_test.go | 2 +- .../backend/internal/resourceimport/load.go | 130 ++++++ .../internal/resourceimport/preview.go | 10 +- .../backend/internal/resourceimport/query.sql | 54 ++- .../backend/internal/resourceimport/reader.go | 92 ++++ .../internal/resourceimport/service.go | 80 +--- .../internal/resourceimport/sqlc/query.sql.go | 201 +++++++- .../backend/internal/resourceimport/upload.go | 430 ++++++++++++++++++ .../internal/resourceimport/upload_test.go | 302 ++++++++++++ .../000027_resource_import_uploads.down.sql | 2 + .../000027_resource_import_uploads.up.sql | 22 + monkeyai/backend/schema/schema.sql | 20 + .../design/resource-package-import-plan.md | 2 + .../design/resource-package-volumes-plan.md | 74 +++ 23 files changed, 2149 insertions(+), 464 deletions(-) create mode 100644 monkeyai/admin/src/components/resource-import-dialog.tsx create mode 100644 monkeyai/admin/src/lib/resource-upload.ts create mode 100644 monkeyai/admin/test/resource-upload.test.mjs create mode 100644 monkeyai/backend/internal/resourceimport/load.go create mode 100644 monkeyai/backend/internal/resourceimport/reader.go create mode 100644 monkeyai/backend/internal/resourceimport/upload.go create mode 100644 monkeyai/backend/internal/resourceimport/upload_test.go create mode 100644 monkeyai/backend/migrations/000027_resource_import_uploads.down.sql create mode 100644 monkeyai/backend/migrations/000027_resource_import_uploads.up.sql create mode 100644 monkeyai/design/resource-package-volumes-plan.md diff --git a/monkeyai/admin/nginx/app.conf b/monkeyai/admin/nginx/app.conf index fcd42cdd8..202c0426b 100644 --- a/monkeyai/admin/nginx/app.conf +++ b/monkeyai/admin/nginx/app.conf @@ -78,6 +78,21 @@ location = /api/v1/endpoints/connect { proxy_buffering off; } +location ~ ^/api/admin/v1/resource-import-uploads(?:/|$) { + client_max_body_size 64m; + proxy_pass http://monkeyai_backend; + proxy_http_version 1.1; + proxy_set_header Host $http_host; + proxy_set_header X-Real-IP $remote_addr; + proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for; + proxy_set_header X-Forwarded-Host $http_host; + proxy_set_header X-Forwarded-Proto $scheme; + proxy_request_buffering off; + proxy_read_timeout 30m; + proxy_send_timeout 30m; + proxy_buffering off; +} + location ~ ^/api/admin/v1/resource-imports(?:/preview)?$ { client_max_body_size 2064m; proxy_pass http://monkeyai_backend; diff --git a/monkeyai/admin/src/components/resource-import-dialog.tsx b/monkeyai/admin/src/components/resource-import-dialog.tsx new file mode 100644 index 000000000..7f7a0193f --- /dev/null +++ b/monkeyai/admin/src/components/resource-import-dialog.tsx @@ -0,0 +1,298 @@ +import { useEffect, useMemo, useRef, useState, type InputHTMLAttributes } from "react" +import { api } from "@/lib/api" +import { archiveDirectory, createUpload, matchArchives, sortArchives, uploadArchives, uploadBase, uploadedFileBytes, validateArchives, type ResourceUpload, type StagedArchive, type UploadProgress } from "@/lib/resource-upload" +import { Badge } from "@/components/ui/badge" +import { Button } from "@/components/ui/button" +import { Checkbox } from "@/components/ui/checkbox" +import { Dialog, DialogContent, DialogFooter, DialogHeader, DialogTitle } from "@/components/ui/dialog" +import { Input } from "@/components/ui/input" +import { Select, SelectContent, SelectGroup, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select" + +const base = "/api/admin/v1/resource-imports" +const actions = { create: "新增", update: "更新", skip: "跳过", retire: "停用", restore: "恢复" } as const +const types = { skill: "技能", rule: "规则", connector: "连接器", expert: "专家" } as const + +type Change = { + type: string + slug: string + name: string + action: keyof typeof actions + warnings?: string[] + source_sha256?: string + storage_sha256?: string +} +type Preview = { + publisher: string + release_id: string + version: number + package_sha256: string + plan_digest: string + changes: Change[] + ignored_static: string[] + already_imported: boolean +} +type Phase = "idle" | "packing" | "uploading" | "checking" | "previewing" | "applying" + +function sizeLabel(size: number) { + if (size < 1024) return `${size} B` + if (size < 1024 * 1024) return `${(size / 1024).toFixed(1)} KiB` + return `${(size / 1024 / 1024).toFixed(2)} MiB` +} +function message(reason: unknown) { return reason instanceof Error ? reason.message : "导入失败，请重试" } + +export function ResourceImportDialog({ initialUpload, onClose, onChanged, onImported }: { + initialUpload: ResourceUpload | null + onClose: () => void + onChanged: () => void + onImported: () => void +}) { + const [files, setFiles] = useState([]) + const [upload, setUpload] = useState(initialUpload) + const [preview, setPreview] = useState(null) + const [stale, setStale] = useState(false) + const [phase, setPhase] = useState(initialUpload?.status === "ready" ? "previewing" : "idle") + const [error, setError] = useState("") + const [notice, setNotice] = useState(initialUpload?.status === "uploading" ? "请重新选择原来的 ZIP 或全部分卷，已上传的部分会自动跳过。" : "") + const [progress, setProgress] = useState({ chunks: initialUpload?.chunks ?? [], index: -1, loaded: 0 }) + const [modes, setModes] = useState>({}) + const [confirmed, setConfirmed] = useState(false) + const [normalization, setNormalization] = useState(false) + const [typeFilter, setTypeFilter] = useState("all") + const [actionFilter, setActionFilter] = useState("changed") + const archiveInputRef = useRef(null) + const directoryInputRef = useRef(null) + const cleanupRef = useRef(undefined) + const controllerRef = useRef(null) + const busy = phase !== "idle" + + useEffect(() => () => { + controllerRef.current?.abort() + if (cleanupRef.current) void cleanupRef.current().catch(() => {}) + }, []) + + useEffect(() => { + if (initialUpload?.status !== "ready") return + const controller = new AbortController() + controllerRef.current = controller + void api(`${base}/preview`, { method: "POST", signal: controller.signal, body: JSON.stringify({ upload_id: initialUpload.id, options: {} }) }) + .then(setPreview) + .catch((reason: unknown) => { if (!controller.signal.aborted) setError(message(reason)) }) + .finally(() => { if (!controller.signal.aborted) setPhase("idle") }) + return () => controller.abort() + }, [initialUpload]) + + const selectionError = useMemo(() => { + if (!files.length) return "" + try { + validateArchives(files) + if (upload && !matchArchives(files, upload.files)) return "文件与当前上传任务不一致，请重新选择原来的 ZIP 或全部分卷" + return "" + } catch (reason) { return message(reason) } + }, [files, upload]) + const listedFiles = upload?.files ?? files + const total = upload?.total_size ?? files.reduce((sum, file) => sum + file.size, 0) + const fileBytes = uploadedFileBytes(listedFiles, upload?.chunk_size ?? 64 * 1024 * 1024, progress) + const uploaded = fileBytes.reduce((sum, size) => sum + size, 0) + const percent = total ? Math.min(100, Math.floor(uploaded / total * 100)) : 0 + const step = preview ? 2 : upload || busy ? 1 : 0 + const needsNormalization = preview?.changes.some((item) => item.type === "skill" && (item.warnings?.length ?? 0) > 0) + const filteredChanges = preview?.changes.filter((item) => (typeFilter === "all" || item.type === typeFilter) && (actionFilter === "all" || (actionFilter === "changed" ? item.action !== "skip" : item.action === actionFilter))) ?? [] + const retired = preview?.changes.filter((item) => item.action === "retire").length ?? 0 + + const cleanupStaged = () => { + const cleanup = cleanupRef.current + cleanupRef.current = undefined + if (cleanup) void cleanup().catch(() => {}) + } + const close = () => { + if (phase === "applying" || phase === "packing") return + controllerRef.current?.abort() + onChanged() + onClose() + } + const selectFiles = (selected: File[], append = false) => { + if (busy || (upload && upload.status !== "uploading")) return + cleanupStaged() + const next = append ? [...new Map([...files, ...selected].map((file) => [file.name, file])).values()] : selected + setFiles(sortArchives(next)) + setError("") + setNotice("") + } + const selectDirectory = async (list: FileList) => { + setPhase("packing") + setError("") + setNotice("") + cleanupStaged() + try { + const staged = await archiveDirectory(list) + cleanupRef.current = staged.cleanup + setFiles([staged.file]) + } catch (reason) { setError(message(reason)) } + finally { setPhase("idle") } + } + const options = (confirm = false) => ({ authorization_modes: modes, accept_normalization: normalization, confirm_final: confirm }) + + const run = async (apply: boolean) => { + if (busy) return + const controller = new AbortController() + controllerRef.current = controller + setError("") + setNotice("") + try { + let current = upload + if (apply) { + if (!current || !preview || stale) return + setPhase("applying") + await api(base, { method: "POST", signal: controller.signal, body: JSON.stringify({ upload_id: current.id, options: options(true), package_sha256: preview.package_sha256, plan_digest: preview.plan_digest }) }) + onImported() + return + } + if (!current || current.status === "uploading") { + setPhase("uploading") + if (!current) { + current = await createUpload(files, controller.signal) + setUpload(current) + onChanged() + } else { + current = await api(`${uploadBase}/${current.id}`, { signal: controller.signal }) + if (current.status === "imported") { onImported(); return } + setUpload(current) + } + setProgress({ chunks: current.chunks, index: -1, loaded: 0 }) + if (current.status === "uploading") { + current = await uploadArchives(current, files, controller.signal, setProgress) + setUpload(current) + setPhase("checking") + current = await api(`${uploadBase}/${current.id}/complete`, { method: "POST", signal: controller.signal }) + setUpload(current) + onChanged() + } + } + setPhase("previewing") + const result = await api(`${base}/preview`, { method: "POST", signal: controller.signal, body: JSON.stringify({ upload_id: current.id, options: options() }) }) + setPreview(result) + setStale(false) + setConfirmed(false) + } catch (reason) { + if (controller.signal.aborted) setNotice("上传已暂停，已上传的部分会保留，可稍后继续。") + else setError(message(reason)) + if (apply) { setStale(true); setConfirmed(false) } + onChanged() + } finally { + setProgress((value) => ({ ...value, index: -1, loaded: 0 })) + setPhase("idle") + } + } + + const restart = async () => { + if (busy) return + setError("") + try { + if (upload) await api(`${uploadBase}/${upload.id}`, { method: "DELETE" }) + cleanupStaged() + setUpload(null) + setFiles([]) + setPreview(null) + setStale(false) + setProgress({ chunks: [], index: -1, loaded: 0 }) + setConfirmed(false) + setNormalization(false) + setModes({}) + setNotice("") + onChanged() + } catch (reason) { setError(message(reason)) } + } + + return ( + { if (!open) close() }}> + + + 导入资源包 +

支持普通 ZIP、按字节拆分的 .zip.001 等全部分卷，或包含 manifest.json 的解压目录。总大小不超过 2 GiB。

+ +
    + {["选择文件", "上传解析", "确认导入"].map((label, index) =>
  1. {index + 1}{label}
    2. )} +
+ {!preview && <> + { + const selected = Array.from(event.currentTarget.files ?? []) + if (selected.length) selectFiles(selected, !upload) + event.currentTarget.value = "" + }} /> + )} onChange={(event) => { + const list = event.currentTarget.files + if (list?.length) void selectDirectory(list) + event.currentTarget.value = "" + }} /> + {(!upload || upload.status === "uploading") &&
event.preventDefault()} onDrop={(event) => { + event.preventDefault() + if (event.dataTransfer.files.length) selectFiles(Array.from(event.dataTransfer.files), !upload) + }}> +

{upload ? "重新选择原来的 ZIP 或全部分卷，继续上传" : "拖入 ZIP 或全部分卷，自动按编号排序"}

+
+ + +
+
} + {listedFiles.length > 0 &&
+ {listedFiles.map((file, index) =>
+ {file.name} + {sizeLabel(file.size)} + {upload ? {fileBytes[index] >= file.size ? "已上传" : fileBytes[index] > 0 ? `${Math.floor(fileBytes[index] / file.size * 100)}%` : "等待上传"} : } +
)} +
} + {listedFiles.length > 0 &&

共 {listedFiles.length} 个文件 · {sizeLabel(total)}

} + {upload &&
+
{phase === "checking" ? "正在检查资源包完整性…" : phase === "previewing" ? "正在解析资源并计算变更…" : phase === "uploading" ? "正在上传并核对文件…" : upload.status === "ready" ? "上传完成，可继续预览" : "已保留上传进度"}{percent}%
+ +

{sizeLabel(uploaded)} / {sizeLabel(total)} · 未完成任务保留 24 小时

+
} + {phase === "packing" &&

正在打包目录…

} + } + {(error || selectionError) &&

{error || selectionError}

} + {notice &&

{notice}

} + {!preview && + {upload && } + {!upload && files.length > 0 && } + + {phase === "uploading" ? : } + } + {preview &&
+

{preview.publisher} / v{preview.version}

共 {preview.changes.length} 项资源
+
{Object.entries(actions).map(([action, label]) => {label} {preview.changes.filter((item) => item.action === action).length})}
+ {preview.already_imported &&

此发布版本已经导入。

} + {preview.ignored_static.length > 0 &&

已忽略 {preview.ignored_static.length} 个静态资源

} +
+ + +
+
+ {!filteredChanges.length &&

当前筛选下没有资源变更

} + {filteredChanges.map((item) =>
+
{item.name} · {types[item.type as keyof typeof types] ?? item.type} / {item.slug}{actions[item.action]}
+ {item.warnings?.map((warning) =>

{warning}

)} + {(item.warnings?.length ?? 0) > 0 && item.storage_sha256 &&

原产物：{item.source_sha256} → 存储包：{item.storage_sha256}

} + {item.type === "connector" && item.action !== "retire" &&
认证归属：
} +
)} +
+ {stale &&

选项或资源状态已变化，请重新预览后确认。

} + {retired > 0 &&

本次将停用 {retired} 项资源，请核对上方停用明细。恢复资源时，仍有效的旧授权或凭证可能重新生效。

} + {!preview.already_imported && <> + {needsNormalization &&
} +
+ } + + + + {preview.already_imported ? : <>} + +
} + + + ) +} diff --git a/monkeyai/admin/src/lib/resource-upload.ts b/monkeyai/admin/src/lib/resource-upload.ts new file mode 100644 index 000000000..e11282902 --- /dev/null +++ b/monkeyai/admin/src/lib/resource-upload.ts @@ -0,0 +1,172 @@ +import { api, ApiError } from "./api.ts" +import { stageDirectoryZip } from "./stream-zip.ts" + +export const maxPackage = 2 * 1024 * 1024 * 1024 +export const uploadBase = "/api/admin/v1/resource-import-uploads" + +export type UploadFile = { name: string; size: number } +export type UploadChunk = { index: number; size: number; sha256: string } +export type ResourceUpload = { + id: string + files: UploadFile[] + total_size: number + status: "uploading" | "ready" | "imported" + package_sha256: string + created_at: string + expires_at: string + chunk_size: number + chunks: UploadChunk[] +} +export type UploadProgress = { chunks: UploadChunk[]; index: number; loaded: number } +export type StagedArchive = { file: File; cleanup?: () => Promise } + +const volumeName = /^(.+\.zip)\.([0-9]{3,6})$/i + +export function sortArchives(files: File[]): File[] { + return [...files].sort((a, b) => { + const left = volumeName.exec(a.name) + const right = volumeName.exec(b.name) + if (left && right && left[1].toLowerCase() === right[1].toLowerCase()) return Number(left[2]) - Number(right[2]) + return a.name.localeCompare(b.name) + }) +} + +export function validateArchives(files: UploadFile[]): void { + if (!files.length) throw new Error("请选择 ZIP 或全部分卷") + if (files.length > 256) throw new Error("最多选择 256 个分卷") + if (files.some((file) => !file.size)) throw new Error("存在空文件，请重新选择") + if (files.reduce((sum, file) => sum + file.size, 0) > maxPackage) throw new Error("资源包总大小超过 2 GiB") + if (files.length === 1 && /\.zip$/i.test(files[0].name)) return + let base = "" + files.forEach((file, index) => { + const match = volumeName.exec(file.name) + if (!match) throw new Error("请选择普通 ZIP 或 .zip.001、.002 等全部分卷；不支持 .z01 或 7z 分卷") + if (base && base !== match[1].toLowerCase()) throw new Error("所选分卷来自不同资源包，请重新选择") + if (Number(match[2]) !== index + 1) throw new Error(`分卷缺失或编号重复：需要 ${match[1]}.${String(index + 1).padStart(3, "0")}`) + base = match[1].toLowerCase() + }) +} + +export function matchArchives(files: UploadFile[], expected: UploadFile[]): boolean { + return files.length === expected.length && files.every((file, index) => file.name === expected[index].name && file.size === expected[index].size) +} + +export function archiveChunk(files: Blob[], index: number, chunkSize: number): Blob { + let offset = index * chunkSize + let remaining = chunkSize + const parts: Blob[] = [] + for (const file of files) { + if (offset >= file.size) { + offset -= file.size + continue + } + const size = Math.min(file.size - offset, remaining) + parts.push(file.slice(offset, offset + size)) + remaining -= size + if (remaining === 0) break + offset = 0 + } + return new Blob(parts, { type: "application/octet-stream" }) +} + +export function uploadedFileBytes(files: UploadFile[], chunkSize: number, progress: UploadProgress): number[] { + const ranges = progress.chunks.map((chunk) => [chunk.index * chunkSize, chunk.index * chunkSize + chunk.size]) + if (progress.loaded > 0) ranges.push([progress.index * chunkSize, progress.index * chunkSize + progress.loaded]) + let offset = 0 + return files.map((file) => { + const size = Math.min(file.size, ranges.reduce((sum, [start, end]) => sum + Math.max(0, Math.min(end, offset + file.size) - Math.max(start, offset)), 0)) + offset += file.size + return size + }) +} + +async function chunkSHA(blob: Blob): Promise { + const hash = await crypto.subtle.digest("SHA-256", await blob.arrayBuffer()) + return "sha256:" + Array.from(new Uint8Array(hash), (byte) => byte.toString(16).padStart(2, "0")).join("") +} + +function putChunk(path: string, blob: Blob, sha: string, signal: AbortSignal, onProgress: (loaded: number) => void): Promise { + return new Promise((resolve, reject) => { + signal.throwIfAborted() + const xhr = new XMLHttpRequest() + const abort = () => xhr.abort() + const finish = (error?: Error) => { + signal.removeEventListener("abort", abort) + if (error) reject(error) + else resolve() + } + xhr.open("PUT", path) + xhr.withCredentials = true + xhr.setRequestHeader("Content-Type", "application/octet-stream") + xhr.setRequestHeader("X-Content-SHA256", sha) + xhr.upload.onprogress = (event) => onProgress(event.loaded) + xhr.onload = () => { + if (xhr.status >= 200 && xhr.status < 300) return finish() + let message = "上传失败，请重试" + try { message = JSON.parse(xhr.responseText)?.error?.message ?? message } catch { /* 保留网络错误提示。 */ } + finish(new ApiError(xhr.status, message)) + } + xhr.onerror = () => finish(new Error("网络中断，已保留上传进度，请重试")) + xhr.onabort = () => finish(new DOMException("上传已暂停", "AbortError")) + signal.addEventListener("abort", abort, { once: true }) + xhr.send(blob) + }) +} + +export async function uploadArchives( + upload: ResourceUpload, + files: File[], + signal: AbortSignal, + onProgress: (progress: UploadProgress) => void, +): Promise { + validateArchives(files) + if (!matchArchives(files, upload.files)) throw new Error("文件与当前上传任务不一致，请重新选择原来的全部分卷") + const chunks = [...upload.chunks] + const count = Math.ceil(upload.total_size / upload.chunk_size) + for (let index = 0; index < count; index++) { + signal.throwIfAborted() + const blob = archiveChunk(files, index, upload.chunk_size) + const sha = await chunkSHA(blob) + signal.throwIfAborted() + const existing = chunks.find((chunk) => chunk.index === index) + if (existing) { + if (existing.sha256 !== sha) throw new Error("本地文件内容与已上传部分不同，请选择原来的资源包，或放弃任务后重新上传") + continue + } + onProgress({ chunks: [...chunks], index, loaded: 0 }) + await putChunk(`${uploadBase}/${upload.id}/chunks/${index}`, blob, sha, signal, (loaded) => onProgress({ chunks: [...chunks], index, loaded })) + chunks.push({ index, size: blob.size, sha256: sha }) + onProgress({ chunks: [...chunks], index: -1, loaded: 0 }) + } + return { ...upload, chunks } +} + +export async function createUpload(files: File[], signal: AbortSignal): Promise { + validateArchives(files) + return api(uploadBase, { method: "POST", signal, body: JSON.stringify({ files: files.map(({ name, size }) => ({ name, size })) }) }) +} + +export async function archiveDirectory(list: FileList): Promise { + if (list.length === 0) throw new Error("目录没有文件") + const entries: [string, File][] = [] + for (const file of Array.from(list)) { + const relativePath = file.webkitRelativePath.split("/").slice(1).join("/") + if (!relativePath || relativePath.split("/").some((part) => part === ".." || !part)) throw new Error("目录内存在无效路径") + entries.push([relativePath, file]) + } + const byPath = new Map(entries.sort(([a], [b]) => a.localeCompare(b))) + if (byPath.size > 5000) throw new Error("目录文件数量超限") + const size = [...byPath.values()].reduce((sum, file) => sum + file.size, 0) + if (size > maxPackage) throw new Error("目录文件总大小超过 2 GiB，请选择现成 ZIP") + if (!navigator.storage?.getDirectory) { + if (size > 200 * 1024 * 1024) throw new Error("当前浏览器不支持大目录磁盘暂存，请选择现成 ZIP") + const { default: JSZip } = await import("jszip") + const zip = new JSZip() + for (const [path, file] of byPath) zip.file(path, file, { date: new Date("1980-01-01T00:00:00Z") }) + const blob = await zip.generateAsync({ type: "blob", compression: "DEFLATE" }) + if (blob.size > maxPackage) throw new Error("资源包超过 2 GiB") + return { file: new File([blob], "release.zip", { type: "application/zip" }) } + } + const staged = await stageDirectoryZip(byPath, maxPackage) + return { ...staged, file: new File([staged.file], "release.zip", { type: "application/zip" }) } +} diff --git a/monkeyai/admin/src/pages/resource-imports-page.tsx b/monkeyai/admin/src/pages/resource-imports-page.tsx index d1648f0ff..2f4a7303f 100644 --- a/monkeyai/admin/src/pages/resource-imports-page.tsx +++ b/monkeyai/admin/src/pages/resource-imports-page.tsx @@ -1,143 +1,40 @@ -import { useEffect, useRef, useState, type InputHTMLAttributes } from "react" -import { ArrowDown01Icon, FileArchiveIcon, FolderOpenIcon, Upload04Icon } from "@hugeicons/core-free-icons" -import { HugeiconsIcon } from "@hugeicons/react" +import { useEffect, useState } from "react" import { useTranslation } from "react-i18next" -import { TablePagination } from "@/components/table-pagination" - import { api } from "@/lib/api" -import { stageDirectoryZip } from "@/lib/stream-zip" +import { uploadBase, type ResourceUpload } from "@/lib/resource-upload" +import { ResourceImportDialog } from "@/components/resource-import-dialog" +import { TablePagination } from "@/components/table-pagination" import { Badge } from "@/components/ui/badge" import { Button } from "@/components/ui/button" -import { Checkbox } from "@/components/ui/checkbox" import { Card, CardContent } from "@/components/ui/card" -import { Dialog, DialogContent, DialogFooter, DialogHeader, DialogTitle } from "@/components/ui/dialog" -import { DropdownMenu, DropdownMenuContent, DropdownMenuGroup, DropdownMenuItem, DropdownMenuTrigger } from "@/components/ui/dropdown-menu" -import { Input } from "@/components/ui/input" -import { Item, ItemActions, ItemContent, ItemDescription, ItemMedia, ItemTitle } from "@/components/ui/item" import { ScrollArea } from "@/components/ui/scroll-area" -import { Select, SelectContent, SelectGroup, SelectItem, SelectTrigger, SelectValue } from "@/components/ui/select" import { Table, TableBody, TableCell, TableHead, TableHeader, TableRow } from "@/components/ui/table" const base = "/api/admin/v1/resource-imports" -const maxPackage = 2 * 1024 * 1024 * 1024 - -const counts = [ - ["create", "新增"], - ["update", "更新"], - ["skip", "跳过"], - ["retire", "退役"], - ["restore", "恢复"], - ["ignored_static", "忽略静态资源"], -] as const - -const actionLabels = { - create: "新增", - update: "更新", - skip: "跳过", - retire: "停用", - restore: "恢复", -} as const - -type Change = { - type: string - slug: string - name: string - action: "create" | "update" | "skip" | "retire" | "restore" - warnings?: string[] - source_sha256?: string - storage_sha256?: string -} - -type Preview = { - publisher: string - release_id: string - version: number - package_sha256: string - plan_digest: string - changes: Change[] - ignored_static: string[] - requires_confirmation: boolean - already_imported: boolean -} - -type Batch = { - id: string - publisher: string - release_id: string - version: number - status: string - result_counts: Record - created_at: string -} - -type HistoryPage = { - items: Batch[] - next_cursor: string | null -} - -type HistoryRequest = { - key: string - page?: HistoryPage - error?: string -} +const counts = [["create", "新增"], ["update", "更新"], ["skip", "跳过"], ["retire", "停用"], ["restore", "恢复"], ["ignored_static", "忽略静态资源"]] as const -type StagedArchive = { file: File; cleanup?: () => Promise } - -async function archiveDirectory(list: FileList): Promise { - if (list.length === 0) throw new Error("目录没有文件") - const byPath = new Map() - for (const file of Array.from(list)) { - const parts = file.webkitRelativePath.split("/") - const relativePath = parts.slice(1).join("/") - if (!relativePath || relativePath.split("/").some((part) => part === ".." || !part)) { - throw new Error("目录内存在无效路径") - } - byPath.set(relativePath, file) - } - if (byPath.size > 5000) throw new Error("目录文件数量超限") - let size = 0 - for (const file of byPath.values()) { - size += file.size - if (size > maxPackage) throw new Error("目录文件总大小超过 2 GiB，请选择现成 ZIP") - } - if (!navigator.storage?.getDirectory) { - if (size > 200 * 1024 * 1024) throw new Error("当前浏览器不支持大目录磁盘暂存，请选择现成 ZIP") - const { default: JSZip } = await import("jszip") - const zip = new JSZip() - for (const [path, file] of byPath) zip.file(path, file) - const blob = await zip.generateAsync({ type: "blob", compression: "DEFLATE" }) - if (blob.size > maxPackage) throw new Error("资源包超过 2 GiB") - return { file: new File([blob], "release.zip", { type: "application/zip" }) } - } - return stageDirectoryZip(byPath, maxPackage) -} +type Batch = { id: string; publisher: string; release_id: string; version: number; status: string; result_counts: Record; created_at: string } +type HistoryPage = { items: Batch[]; next_cursor: string | null } +type HistoryRequest = { key: string; page?: HistoryPage; error?: string } export function ResourceImportsPage() { const { i18n } = useTranslation() - const [dialogOpen, setDialogOpen] = useState(false) - const [file, setFile] = useState(null) - const [sourceName, setSourceName] = useState("") - const archiveInputRef = useRef(null) - const directoryInputRef = useRef(null) - const stagedCleanupRef = useRef(undefined) - const [preview, setPreview] = useState(null) + const [dialog, setDialog] = useState<{ upload: ResourceUpload | null } | null>(null) const [request, setRequest] = useState(null) const [cursors, setCursors] = useState<(string | undefined)[]>([undefined]) const [active, setActive] = useState(0) const [pageSize, setPageSize] = useState(50) const [refresh, setRefresh] = useState(0) - const [modes, setModes] = useState>({}) - const [confirmed, setConfirmed] = useState(false) - const [normalization, setNormalization] = useState(false) - const [busy, setBusy] = useState(false) - const [error, setError] = useState("") - + const [pending, setPending] = useState([]) + const [pendingError, setPendingError] = useState("") + const [discarding, setDiscarding] = useState("") const cursor = cursors[active] const historyKey = `${refresh}:${pageSize}:${cursor ?? ""}` const current = request?.key === historyKey ? request : null const history = current?.page ?? null const historyError = current?.error ?? "" const historyLoading = current === null + const dateFormat = new Intl.DateTimeFormat(i18n.resolvedLanguage ?? i18n.language, { dateStyle: "short", timeStyle: "short" }) useEffect(() => { let activeRequest = true @@ -147,276 +44,79 @@ export function ResourceImportsPage() { if (cursor) query.set("cursor", cursor) void api(`${base}?${query}`, { signal: controller.signal }) .then((page) => { if (activeRequest) setRequest({ key, page }) }) - .catch((reason: unknown) => { - if (activeRequest) setRequest({ key, error: reason instanceof Error ? reason.message : "读取记录失败" }) - }) + .catch((reason: unknown) => { if (activeRequest) setRequest({ key, error: reason instanceof Error ? reason.message : "读取记录失败" }) }) return () => { activeRequest = false; controller.abort() } }, [cursor, refresh, pageSize]) - useEffect(() => () => { - if (stagedCleanupRef.current) void stagedCleanupRef.current().catch(() => {}) - }, []) - - const cleanupStaged = () => { - const cleanup = stagedCleanupRef.current - stagedCleanupRef.current = undefined - if (cleanup) void cleanup().catch(() => {}) - } - - const closeDialog = (open: boolean) => { - if (busy && !open) return - setDialogOpen(open) - if (!open) { - cleanupStaged() - setFile(null) - setSourceName("") - setPreview(null) - setModes({}) - setConfirmed(false) - setNormalization(false) - setError("") - } - } - - const options = (confirmFinal: boolean) => ({ - authorization_modes: modes, - accept_normalization: normalization, - confirm_final: confirmFinal, - }) - - const upload = async (apply: boolean) => { - if (!file || busy) return - setBusy(true) - setError("") + useEffect(() => { + const controller = new AbortController() + void api<{ items: ResourceUpload[] }>(uploadBase, { signal: controller.signal }) + .then(({ items }) => { if (!controller.signal.aborted) { setPending(items); setPendingError("") } }) + .catch((reason: unknown) => { if (!controller.signal.aborted) setPendingError(reason instanceof Error ? reason.message : "读取未完成任务失败") }) + return () => controller.abort() + }, [refresh]) + + const discard = async (upload: ResourceUpload) => { + setDiscarding(upload.id) + setPendingError("") try { - const form = new FormData() - form.set("package", file) - form.set("options", JSON.stringify(options(apply))) - if (apply && preview) { - form.set("package_sha256", preview.package_sha256) - form.set("plan_digest", preview.plan_digest) - } - const result = await api(apply ? base : `${base}/preview`, { method: "POST", body: form }) - if (apply) { - cleanupStaged() - setDialogOpen(false) - setFile(null) - setPreview(null) - setModes({}) - setConfirmed(false) - setNormalization(false) - setCursors([undefined]) - setActive(0) - setRefresh((value) => value + 1) - } else { - setPreview(result) - } - } catch (reason) { - setError(reason instanceof Error ? reason.message : "导入失败") - if (apply) setPreview(null) - } finally { - setBusy(false) - } + await api(`${uploadBase}/${upload.id}`, { method: "DELETE" }) + setRefresh((value) => value + 1) + } catch (reason) { setPendingError(reason instanceof Error ? reason.message : "放弃任务失败") } + finally { setDiscarding("") } } - const needsNormalization = preview?.changes.some((item) => item.type === "skill" && (item.warnings?.length ?? 0) > 0) - const changed = preview?.changes.filter((item) => item.action !== "skip") ?? [] - const dateFormat = new Intl.DateTimeFormat(i18n.resolvedLanguage ?? i18n.language, { dateStyle: "short", timeStyle: "short" }) - return (

资源包导入记录

- +
+ {pendingError &&

未完成任务：{pendingError}

} + {pending.length > 0 &&
+

未完成任务 · 上传后保留 24 小时

+ {pending.map((upload) => { + const uploaded = upload.chunks.reduce((sum, chunk) => sum + chunk.size, 0) + return
+

{upload.files[0]?.name}{upload.files.length > 1 ? ` 等 ${upload.files.length} 卷` : ""}

{upload.status === "ready" ? "待确认导入" : `已上传 ${Math.floor(uploaded / upload.total_size * 100)}%`} · 有效期至 {dateFormat.format(new Date(upload.expires_at))}

+ + +
+ })} +
} - {historyError && ( -
- {historyError} - -
- )} + {historyError &&
{historyError}
} {historyLoading && !history &&

加载中…

} {history && history.items.length === 0 && !historyLoading &&

暂无导入记录

} - {history && history.items.length > 0 && ( - - - - - {["导入时间", "发布者", "版本", "发布 ID", "状态", "资源变更"].map((column) => {column})} - - - - {history.items.map((batch) => ( - - {dateFormat.format(new Date(batch.created_at))} - {batch.publisher} - v{batch.version} - {batch.release_id.slice(0, 8)}… - {batch.status === "succeeded" ? "成功" : "失败"} - - {counts.map(([key, label]) => `${label} ${batch.result_counts[key] ?? 0}`).join(" · ")} - - - ))} - -
- - )} - 0 && + + {["导入时间", "发布者", "版本", "发布 ID", "状态", "资源变更"].map((column) => {column})} + {history.items.map((batch) => + {dateFormat.format(new Date(batch.created_at))} + {batch.publisher} + v{batch.version} + {batch.release_id.slice(0, 8)}… + {batch.status === "succeeded" ? "成功" : "失败"} + {counts.map(([key, label]) => `${label} ${batch.result_counts[key] ?? 0}`).join(" · ")} + )} +
+ } + { setPageSize(size); setCursors([undefined]); setActive(0) }} onPrevious={() => setActive((value) => Math.max(0, value - 1))} - onNext={() => { - const next = history?.next_cursor - if (!next) return - setCursors((current) => [...current.slice(0, active + 1), next]) - setActive((value) => value + 1) - }} - /> + onNext={() => { if (history?.next_cursor) { setCursors((current) => [...current.slice(0, active + 1), history.next_cursor!]); setActive((value) => value + 1) } }} /> - - - - 导入资源包 -

选择发布包 ZIP，或选择包含 manifest.json 的解压目录。

- -
- 全量同步：本次资源包是该发布者最终的 Agent 资源集合，缺失资源将在确认后停用。独立静态资源和设计资源跳过导入。 -
- { - const selected = event.currentTarget.files?.[0] - if (!selected) return - cleanupStaged() - if (selected.size > maxPackage) { - setFile(null) - setError("资源包超过 2 GiB") - event.currentTarget.value = "" - return - } - setFile(selected) - setSourceName(selected.name) - setPreview(null) - setConfirmed(false) - setModes({}) - setNormalization(false) - setError("") - event.currentTarget.value = "" - if (directoryInputRef.current) directoryInputRef.current.value = "" - }} /> - )} onChange={async (event) => { - const input = event.currentTarget - const selected = input.files - if (!selected?.length) return - const directoryName = selected[0].webkitRelativePath.split("/")[0] - setBusy(true) - cleanupStaged() - setFile(null) - setPreview(null) - setError("") - try { - const archive = await archiveDirectory(selected) - stagedCleanupRef.current = archive.cleanup - setFile(archive.file) - setSourceName(directoryName) - setConfirmed(false) - setModes({}) - setNormalization(false) - if (archiveInputRef.current) archiveInputRef.current.value = "" - } catch (reason) { - setSourceName("") - setError(reason instanceof Error ? reason.message : "目录打包失败") - } finally { - input.value = "" - setBusy(false) - } - }} /> - - - - - - {sourceName || "尚未选择资源包"} - {file ? `${sourceName === file.name ? "ZIP 文件" : "解压目录"} · ${(file.size / 1024 / 1024).toFixed(2)} MiB` : "支持 ZIP 文件或解压目录，包体不超过 2 GiB"} - - - - }> - - 选择来源 - - - - - archiveInputRef.current?.click()}>选择 ZIP - directoryInputRef.current?.click()}>选择解压目录 - - - - - - {error &&

{error}

} - {!preview && - - - } - {preview &&
-
-

预览 · {preview.publisher} / v{preview.version}

- 共 {preview.changes.length} 项资源 -
- {preview.already_imported &&

此发布版本已经导入。

} - {preview.ignored_static.length > 0 &&

已忽略 {preview.ignored_static.length} 个静态资源：{preview.ignored_static.join("、")}

} -
- {preview.changes.map((item) =>
-
- {item.name} · {item.type} / {item.slug} - {actionLabels[item.action]} -
- {item.warnings?.map((warning) =>

{warning}

)} - {(item.warnings?.length ?? 0) > 0 && item.storage_sha256 &&

原产物：{item.source_sha256} → 存储包：{item.storage_sha256}

} - {item.type === "connector" && item.action !== "retire" &&
- 认证归属： - -
} -
)} -
- {changed.some((item) => item.action === "retire") &&

此操作将停用上表中的移除项；恢复项的旧授权或凭证可能重新生效。

} - {needsNormalization &&
- - -
} -
- - -
- - - - -
} - - + {dialog && setDialog(null)} onChanged={() => setRefresh((value) => value + 1)} onImported={() => { + setDialog(null) + setCursors([undefined]) + setActive(0) + setRefresh((value) => value + 1) + }} />}
) } diff --git a/monkeyai/admin/test/resource-api.test.mjs b/monkeyai/admin/test/resource-api.test.mjs index d52cf0fa3..a0e71af51 100644 --- a/monkeyai/admin/test/resource-api.test.mjs +++ b/monkeyai/admin/test/resource-api.test.mjs @@ -42,6 +42,16 @@ test("技能上传代理允许 500 MiB 包并保留 multipart 开销", () => { for (const path of ["/api/admin/v1/rules", "/api/v1/skills/owned/manifest"]) { assert.doesNotMatch(matchingLocation(path), /client_max_body_size/, path) } + for (const path of [ + "/api/admin/v1/resource-import-uploads", + "/api/admin/v1/resource-import-uploads/upload/chunks/0", + "/api/admin/v1/resource-import-uploads/upload/complete", + ]) { + const body = matchingLocation(path) + assert.match(body, /client_max_body_size 64m;/, path) + assert.match(body, /proxy_request_buffering off;/, path) + assert.match(body, /proxy_read_timeout 30m;/, path) + } for (const path of [ "/api/admin/v1/resource-imports", "/api/admin/v1/resource-imports/preview", diff --git a/monkeyai/admin/test/resource-upload.test.mjs b/monkeyai/admin/test/resource-upload.test.mjs new file mode 100644 index 000000000..e4f906ed0 --- /dev/null +++ b/monkeyai/admin/test/resource-upload.test.mjs @@ -0,0 +1,87 @@ +import assert from "node:assert/strict" +import { File } from "node:buffer" +import { test } from "node:test" +import { archiveChunk, matchArchives, sortArchives, uploadArchives, uploadedFileBytes, validateArchives } from "../src/lib/resource-upload.ts" + +const sha = async (text) => "sha256:" + Buffer.from(await crypto.subtle.digest("SHA-256", new TextEncoder().encode(text))).toString("hex") + +function upload(files, chunks = []) { + return { id: "test-upload", files, total_size: files.reduce((sum, file) => sum + file.size, 0), status: "uploading", package_sha256: "", created_at: "", expires_at: "", chunk_size: 4, chunks } +} + +test("分卷按数字排序并识别缺卷、重卷及混包", () => { + const files = sortArchives([new File(["b"], "release.zip.002"), new File(["a"], "release.zip.001")]) + assert.deepEqual(files.map((file) => file.name), ["release.zip.001", "release.zip.002"]) + assert.doesNotThrow(() => validateArchives(files)) + for (const names of [["release.zip.002"], ["release.zip.001", "release.zip.001"], ["release.zip.001", "other.zip.002"], ["release.z01"], ["release.7z.001"]]) { + assert.throws(() => validateArchives(names.map((name) => ({ name, size: 1 })))) + } + assert.throws(() => validateArchives([{ name: "release.zip", size: 2 ** 31 + 1 }]), /2 GiB/) + assert.throws(() => validateArchives([{ name: "release.zip", size: 0 }]), /空文件/) +}) + +test("传输块跨越本地分卷边界，保持原始 ZIP 字节顺序", async () => { + const files = [new Blob(["abc"]), new Blob(["defgh"]), new Blob(["ij"])] + assert.equal(await archiveChunk(files, 0, 4).text(), "abcd") + assert.equal(await archiveChunk(files, 1, 4).text(), "efgh") + assert.equal(await archiveChunk(files, 2, 4).text(), "ij") + assert.equal(archiveChunk(files, 3, 4).size, 0) +}) + +test("每卷进度根据传输块区间计算，支持非连续已上传块", () => { + const files = [{ name: "a", size: 3 }, { name: "b", size: 5 }, { name: "c", size: 2 }] + assert.deepEqual(uploadedFileBytes(files, 4, { chunks: [{ index: 0, size: 4 }, { index: 2, size: 2 }], index: 1, loaded: 2 }), [3, 3, 2]) + assert.equal(matchArchives(files, files.map((file) => ({ ...file }))), true) + assert.equal(matchArchives(files, files.slice(1)), false) + assert.equal(matchArchives(files, [{ name: "a", size: 4 }, ...files.slice(1)]), false) +}) + +test("续传核对已上传内容，只发送缺失块并更新进度", async (t) => { + const files = [new File(["abc"], "release.zip.001"), new File(["defgh"], "release.zip.002")] + const existing = { index: 0, size: 4, sha256: await sha("abcd") } + const requests = [] + class XHR { + upload = {} + status = 204 + open(method, path) { requests.push({ method, path, headers: {} }); this.request = requests.at(-1) } + setRequestHeader(name, value) { this.request.headers[name] = value } + send(blob) { + this.request.blob = blob + this.upload.onprogress({ loaded: blob.size }) + queueMicrotask(() => this.onload()) + } + abort() { this.onabort() } + } + const original = globalThis.XMLHttpRequest + globalThis.XMLHttpRequest = XHR + t.after(() => { if (original) globalThis.XMLHttpRequest = original; else delete globalThis.XMLHttpRequest }) + const progress = [] + const result = await uploadArchives(upload(files, [existing]), files, new AbortController().signal, (value) => progress.push(value)) + assert.equal(requests.length, 1) + assert.equal(requests[0].path, "/api/admin/v1/resource-import-uploads/test-upload/chunks/1") + assert.equal(await requests[0].blob.text(), "efgh") + assert.equal(requests[0].headers["X-Content-SHA256"], await sha("efgh")) + assert.equal(result.chunks.length, 2) + assert.equal(progress.at(-1).chunks.length, 2) + await assert.rejects(uploadArchives(upload(files, [{ ...existing, sha256: await sha("xxxx") }]), files, new AbortController().signal, () => {}), /本地文件内容/) + assert.equal(requests.length, 1) +}) + +test("暂停上传不会继续发送后续传输块", async (t) => { + const files = [new File(["abcdefghij"], "release.zip")] + const controller = new AbortController() + let count = 0 + class XHR { + upload = {} + status = 204 + open() { count++ } + setRequestHeader() {} + send() { queueMicrotask(() => controller.abort()) } + abort() { this.onabort() } + } + const original = globalThis.XMLHttpRequest + globalThis.XMLHttpRequest = XHR + t.after(() => { if (original) globalThis.XMLHttpRequest = original; else delete globalThis.XMLHttpRequest }) + await assert.rejects(uploadArchives(upload(files), files, controller.signal, () => {}), { name: "AbortError" }) + assert.equal(count, 1) +}) diff --git a/monkeyai/backend/api/admin.yaml b/monkeyai/backend/api/admin.yaml index 95fc78b3c..7502a3669 100644 --- a/monkeyai/backend/api/admin.yaml +++ b/monkeyai/backend/api/admin.yaml @@ -7,13 +7,105 @@ servers: security: - AdminSession: [] paths: + /api/admin/v1/resource-import-uploads: + post: + summary: 创建资源包上传会话 + description: 仅管理员可用，会话与创建者绑定，有效期 24 小时，每人最多 5 个未完成任务。普通 ZIP 或按字节拆分的 .zip.001 等分卷按顺序组成一个 ZIP，总大小不超过 2 GiB，最多 256 卷；不支持 .z01 或 7z 分卷。网络上传统一使用 64 MiB 传输块。 + requestBody: + required: true + content: + application/json: + schema: + type: object + required: [files] + properties: + files: + type: array + minItems: 1 + maxItems: 256 + items: {$ref: '#/components/schemas/ResourceImportUploadFile'} + responses: + '201': + description: 返回上传会话 + content: + application/json: + schema: {$ref: '#/components/schemas/ResourceImportUpload'} + '400': {$ref: '#/components/responses/ResourceError400'} + '409': {description: 未完成任务数已达上限} + get: + summary: 查询当前管理员未完成的资源包上传任务 + responses: + '200': + description: 返回尚未过期的 uploading 和 ready 任务 + content: + application/json: + schema: + type: object + required: [items] + properties: + items: + type: array + items: {$ref: '#/components/schemas/ResourceImportUpload'} + /api/admin/v1/resource-import-uploads/{id}: + parameters: + - {name: id, in: path, required: true, schema: {type: string, format: uuid}} + get: + summary: 查询上传进度及已保存的传输块 + responses: + '200': + description: 刷新页面或重启服务后可据此续传；本地文件须重新选择并核对已上传块摘要 + content: + application/json: + schema: {$ref: '#/components/schemas/ResourceImportUpload'} + '404': {$ref: '#/components/responses/ResourceError404'} + delete: + summary: 放弃未完成上传任务 + description: 使任务立即失效，后台定期清理暂存对象；已导入任务不能取消。 + responses: + '204': {description: 任务已取消} + '404': {$ref: '#/components/responses/ResourceError404'} + '409': {description: 任务已导入} + /api/admin/v1/resource-import-uploads/{id}/chunks/{index}: + put: + summary: 上传一个资源包传输块 + description: index 从 0 开始，对应逻辑 ZIP 的 index × chunk_size 偏移；除末块外必须恰好 64 MiB。服务端核对实际大小和 SHA-256，同内容重复上传幂等，不同内容或已封存返回 409。分块可乱序上传。 + parameters: + - {name: id, in: path, required: true, schema: {type: string, format: uuid}} + - {name: index, in: path, required: true, schema: {type: integer, minimum: 0, maximum: 31}} + - {name: X-Content-SHA256, in: header, required: true, schema: {type: string, pattern: '^sha256:[0-9a-f]{64}$'}} + requestBody: + required: true + content: + application/octet-stream: + schema: {type: string, format: binary} + responses: + '204': {description: 传输块已保存或同内容已存在} + '400': {$ref: '#/components/responses/ResourceError400'} + '404': {$ref: '#/components/responses/ResourceError404'} + '409': {description: 块内容冲突或上传已封存} + /api/admin/v1/resource-import-uploads/{id}/complete: + post: + summary: 检查完整 ZIP 并封存上传内容 + description: 收齐所有传输块后检查 ZIP 路径、展开限制及外层文件 CRC，计算完整 ZIP 的 package_sha256。成功后 ready 状态不可修改，重复完成幂等。本接口不导入资源，也不校验包内声明的发布协议摘要。 + parameters: + - {name: id, in: path, required: true, schema: {type: string, format: uuid}} + responses: + '200': + description: 完整上传会话 + content: + application/json: + schema: {$ref: '#/components/schemas/ResourceImportUpload'} + '400': {$ref: '#/components/responses/ResourceError400'} + '404': {$ref: '#/components/responses/ResourceError404'} /api/admin/v1/resource-imports/preview: post: summary: 解析资源发布包并预览变更 - description: ZIP 包体最大 2 GiB，磁盘暂存并解析 Agent 资源，不写入资源。JSON 仅检查语法，不校验发布端摘要、字段格式或元数据一致性；缺省字段使用默认值，CHECKSUMS.sha256 和 release.json 可省略。返回新增/更新/跳过/退役/恢复与 plan_digest。 + description: 使用已封存的 upload_id，或直接上传不超过 2 GiB 的 ZIP，磁盘暂存并解析完整 Agent 清单，不写入资源。JSON 元数据仅检查语法，不校验发布端摘要、字段格式或元数据一致性；缺省字段使用默认值，CHECKSUMS.sha256 和 release.json 可省略。返回新增/更新/跳过/退役/恢复与 plan_digest。缺失资源若仍被其他启用的 Expert 引用，将返回包含 Expert 名称和 ID 的 409 冲突。 requestBody: required: true content: + application/json: + schema: {$ref: '#/components/schemas/ResourceImportInput'} multipart/form-data: schema: type: object @@ -28,10 +120,15 @@ paths: /api/admin/v1/resource-imports: post: summary: 按预览确认原子导入完整 Agent 清单 - description: 重新上传同一 ZIP（最大 2 GiB），校验 package_sha256 和 plan_digest；需在 options 中确认最终资源集合。 + description: 使用预览时相同的 upload_id，无需再次上传；兼容重新上传同一 ZIP（最大 2 GiB）。校验 package_sha256 和 plan_digest；需在 options 中确认最终资源集合。全部分卷只产生一个导入批次，保留发布者锁与整包事务。 requestBody: required: true content: + application/json: + schema: + allOf: + - {$ref: '#/components/schemas/ResourceImportInput'} + - {required: [options, package_sha256, plan_digest]} multipart/form-data: schema: type: object @@ -4909,6 +5006,50 @@ components: schema: $ref: "#/components/schemas/ErrorResponse" schemas: + ResourceImportUploadFile: + type: object + required: [name, size] + properties: + name: {type: string, description: ZIP 文件名或从 001 连续递增的分卷文件名} + size: {type: integer, format: int64, minimum: 1, maximum: 2147483648} + ResourceImportUpload: + type: object + required: [id, files, total_size, status, package_sha256, chunk_size, chunks, created_at, expires_at] + properties: + id: {type: string, format: uuid} + files: + type: array + items: {$ref: '#/components/schemas/ResourceImportUploadFile'} + total_size: {type: integer, format: int64} + status: {type: string, enum: [uploading, ready, imported]} + package_sha256: {type: string, description: 封存后由服务端计算的完整 ZIP 摘要，上传期间为空} + chunk_size: {type: integer, const: 67108864} + chunks: + type: array + items: + type: object + required: [index, size, sha256] + properties: + index: {type: integer, minimum: 0, maximum: 31} + size: {type: integer, format: int64} + sha256: {type: string} + created_at: {type: string, format: date-time} + expires_at: {type: string, format: date-time} + ResourceImportInput: + type: object + required: [upload_id] + properties: + upload_id: {type: string, format: uuid} + package_sha256: {type: string} + plan_digest: {type: string} + options: + type: object + properties: + confirm_final: {type: boolean, description: 确认导入时必须为 true} + accept_normalization: {type: boolean} + authorization_modes: + type: object + additionalProperties: {type: string, enum: [independent, centralized]} FeedbackPage: type: object additionalProperties: false diff --git a/monkeyai/backend/internal/app/app.go b/monkeyai/backend/internal/app/app.go index 63d4ed149..d3962bfb9 100644 --- a/monkeyai/backend/internal/app/app.go +++ b/monkeyai/backend/internal/app/app.go @@ -62,6 +62,7 @@ type App struct { endpoints *endpoint.Service feedback *feedback.Service connectors *mcp.Service + resourceImports *resourceimport.Service } type MemberWriters func(*pgxpool.Pool) (member.UserWriter, member.GroupWriter, error) @@ -136,6 +137,7 @@ func newApp(ctx context.Context, cfg config.Config, logger *slog.Logger, factory endpoints: handler.(*applicationHandler).endpoints, feedback: handler.(*applicationHandler).feedback, connectors: handler.(*applicationHandler).connectors, + resourceImports: handler.(*applicationHandler).resourceImports, database: pool, shutdownTimeout: cfg.ShutdownTimeout, }, nil @@ -237,7 +239,8 @@ func newApplicationHandlerWithMembers(ctx context.Context, logger *slog.Logger, models.RegisterAdmin(admin) store.RegisterAdmin(admin) store.RegisterGrants(admin, map[string]*resource.CRUD{"rule": rules, "skill": skills.CRUD, "expert": experts.CRUD, "connector": connectors.Connectors}) - resourceimport.NewService(pool, storage).RegisterAdmin(admin) + resourceImports := resourceimport.NewService(pool, storage) + resourceImports.RegisterAdmin(admin) rules.Register(admin) skills.RegisterAdmin(admin) connectors.RegisterAdmin(admin) @@ -320,7 +323,7 @@ func newApplicationHandlerWithMembers(ctx context.Context, logger *slog.Logger, } })(identities.AuthRouter()) router.Mount("/", httpapi.New(logger, readiness{pool: pool, storage: storage, endpoints: endpoints}, admin, agent, auth)) - return &applicationHandler{Handler: router, billing: charges, proxy: modelProxy, images: imageService, videos: videoService, inputs: imageInputs, endpoints: endpoints, relay: runtimeRelay, runtimeKeys: keys, feedback: feedbacks, connectors: connectors}, nil + return &applicationHandler{Handler: router, billing: charges, proxy: modelProxy, images: imageService, videos: videoService, inputs: imageInputs, endpoints: endpoints, relay: runtimeRelay, runtimeKeys: keys, feedback: feedbacks, connectors: connectors, resourceImports: resourceImports}, nil } type sessionAdapter struct{ service *sessionreporting.Service } @@ -395,6 +398,13 @@ func (a *App) Run(ctx context.Context) error { go func() { defer close(cleanupDone); runImageCleanup(workerCtx, a.inputs) }() feedbackCleanupDone := make(chan struct{}) go func() { defer close(feedbackCleanupDone); _ = a.feedback.Run(workerCtx, time.Hour) }() + uploadCleanupDone := make(chan struct{}) + go func() { + defer close(uploadCleanupDone) + if a.resourceImports != nil { + a.resourceImports.Run(workerCtx) + } + }() defer func() { stopWorker() <-workerDone @@ -405,6 +415,7 @@ func (a *App) Run(ctx context.Context) error { <-videoDone <-cleanupDone <-feedbackCleanupDone + <-uploadCleanupDone }() result := make(chan error, len(a.servers)) diff --git a/monkeyai/backend/internal/app/billing.go b/monkeyai/backend/internal/app/billing.go index 67cea0b1f..2dad5ae0b 100644 --- a/monkeyai/backend/internal/app/billing.go +++ b/monkeyai/backend/internal/app/billing.go @@ -17,22 +17,24 @@ import ( "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/proxy" "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/relay" "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/resource" + "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/resourceimport" "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/setting" "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/videogen" ) type applicationHandler struct { http.Handler - billing *billing.Service - proxy *proxy.Proxy - images *imagegen.Service - videos *videogen.Service - inputs *imagegen.Inputs - endpoints *endpoint.Service - runtimeKeys *apikey.Service - relay *relay.Service - feedback *feedback.Service - connectors *mcp.Service + billing *billing.Service + proxy *proxy.Proxy + images *imagegen.Service + videos *videogen.Service + inputs *imagegen.Inputs + endpoints *endpoint.Service + runtimeKeys *apikey.Service + relay *relay.Service + feedback *feedback.Service + connectors *mcp.Service + resourceImports *resourceimport.Service } type modelBilling struct{ service *billing.Service } type modelUsageRecorder struct{ models *model.Postgres } diff --git a/monkeyai/backend/internal/app/billing_test.go b/monkeyai/backend/internal/app/billing_test.go index 646c7743f..a0046bc09 100644 --- a/monkeyai/backend/internal/app/billing_test.go +++ b/monkeyai/backend/internal/app/billing_test.go @@ -516,7 +516,7 @@ func TestBillingIntegration(t *testing.T) { if modelStats["last_24h"].(map[string]any)["model_calls"].(float64) == 0 { t.Fatal("真实模型调用未进入统计") } - for _, path := range []string{"realtime", "tasks", "history"} { + for _, path := range []string{"realtime", "tasks/overview", "history"} { must("GET", "/api/admin/v1/statistics/"+path, nil, "") } savedCookie := cookie diff --git a/monkeyai/backend/internal/resourceimport/load.go b/monkeyai/backend/internal/resourceimport/load.go new file mode 100644 index 000000000..e18acfeeb --- /dev/null +++ b/monkeyai/backend/internal/resourceimport/load.go @@ -0,0 +1,130 @@ +package resourceimport + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "io" + "mime/multipart" + "net/http" + "strings" + + "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/identity" + "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/resource" + "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/resourceimport/sqlc" + "github.com/jackc/pgx/v5" +) + +type importInput struct { + UploadID string `json:"upload_id"` + Options Options `json:"options"` + PackageSHA256 string `json:"package_sha256"` + PlanDigest string `json:"plan_digest"` +} + +type loadedImport struct { + input importInput + pkg Package + items Resources + sha string + cleanup func() + tx pgx.Tx +} + +func (s *Service) loadImport(w http.ResponseWriter, r *http.Request) (loaded loadedImport, err error) { + var reader io.ReaderAt + var size int64 + loaded.cleanup = func() {} + defer func() { + if err != nil { + loaded.cleanup() + } + }() + if strings.HasPrefix(r.Header.Get("Content-Type"), "application/json") { + if err = resource.Decode(w, r, &loaded.input); err != nil { + return + } + actor, ok := identity.UserFromContext(r.Context()) + if !ok || actor.Role != "admin" { + err = resource.NotFound + return + } + var upload Upload + upload, loaded.tx, err = s.lockUpload(r.Context(), loaded.input.UploadID, actor.ID) + if err != nil { + return + } + loaded.cleanup = func() { _ = loaded.tx.Rollback(context.WithoutCancel(r.Context())) } + if upload.Status != "ready" && upload.Status != "imported" { + err = resource.Invalid("请先完成资源包上传和完整性检查") + return + } + var cleanup func() + reader, loaded.sha, cleanup, err = s.openUpload(r.Context(), upload) + if err != nil { + return + } + loaded.cleanup = func() { + cleanup() + _ = loaded.tx.Rollback(context.WithoutCancel(r.Context())) + } + size = upload.TotalSize + } else { + r.Body = http.MaxBytesReader(w, r.Body, MaxPackage+(16<<20)) + var file multipart.File + loaded.cleanup = func() { + if file != nil { + _ = file.Close() + } + if r.MultipartForm != nil { + _ = r.MultipartForm.RemoveAll() + } + } + if err = r.ParseMultipartForm(8 << 20); err != nil { + err = resource.Invalid("资源包上传格式或大小无效") + return + } + var header *multipart.FileHeader + file, header, err = r.FormFile("package") + if err != nil { + err = resource.Invalid("缺少资源包文件") + return + } + if header.Size <= 0 || header.Size > MaxPackage { + err = resource.Invalid("资源包超过大小限制") + return + } + if value := r.FormValue("options"); value != "" { + if json.Unmarshal([]byte(value), &loaded.input.Options) != nil { + err = resource.Invalid("导入选项无效") + return + } + } + loaded.input.PackageSHA256, loaded.input.PlanDigest = r.FormValue("package_sha256"), r.FormValue("plan_digest") + hash := sha256.New() + if _, err = io.Copy(hash, io.NewSectionReader(file, 0, header.Size)); err != nil { + return + } + loaded.sha = "sha256:" + hex.EncodeToString(hash.Sum(nil)) + reader, size = file, header.Size + } + loaded.pkg, err = ParseReader(reader, size) + if err == nil { + loaded.items, err = ParseResources(loaded.pkg) + } + if err != nil { + err = resource.Invalid(err.Error()) + } + return +} + +func (loaded loadedImport) finish(ctx context.Context) error { + if loaded.tx == nil { + return nil + } + if err := sqlc.New(loaded.tx).FinishImportUpload(ctx, loaded.input.UploadID); err != nil { + return err + } + return loaded.tx.Commit(ctx) +} diff --git a/monkeyai/backend/internal/resourceimport/preview.go b/monkeyai/backend/internal/resourceimport/preview.go index 651dc41ff..79945507f 100644 --- a/monkeyai/backend/internal/resourceimport/preview.go +++ b/monkeyai/backend/internal/resourceimport/preview.go @@ -171,7 +171,8 @@ func checkReferences(ctx context.Context, q resource.Queryer, b binding, present if err != nil { return err } - for _, id := range ids { + for _, reference := range ids { + id := reference.ID needed := false for slug, expert := range r.Experts { bound, ok := bindings[slug] @@ -192,7 +193,7 @@ func checkReferences(ctx context.Context, q resource.Queryer, b binding, present break } if needed { - return fmt.Errorf("资源 %s 仍被保留的 Expert %s 引用", b.Slug, id) + return fmt.Errorf("资源 %s（%s）仍被本次保留的 Expert「%s」（%s）引用，请在完整资源包中保留此资源或解除引用", b.Slug, b.Type, reference.Name, id) } local := true for _, old := range bindings { @@ -202,13 +203,16 @@ func checkReferences(ctx context.Context, q resource.Queryer, b binding, present } } if local { - return fmt.Errorf("资源 %s 仍被其他 Expert %s 引用", b.Slug, id) + return fmt.Errorf("资源 %s（%s）仍被其他 Expert「%s」（%s）引用；该 Expert 不属于本发布者，请先解除引用或在完整资源包中保留此资源", b.Slug, b.Type, reference.Name, id) } } return nil } func (s *Service) Evaluate(ctx context.Context, q resource.Queryer, p Package, r Resources, opts Options, packageSHA string) (Preview, error) { + if opts.AuthorizationModes == nil { + opts.AuthorizationModes = map[string]string{} + } result := Preview{Publisher: p.Release.Publisher.Name, ReleaseID: p.Release.ReleaseID, Version: p.Release.Version, PackageSHA256: packageSHA, Changes: []Change{}, IgnoredStatic: []string{}} for _, name := range p.Names() { diff --git a/monkeyai/backend/internal/resourceimport/query.sql b/monkeyai/backend/internal/resourceimport/query.sql index ac5b84191..fe841a801 100644 --- a/monkeyai/backend/internal/resourceimport/query.sql +++ b/monkeyai/backend/internal/resourceimport/query.sql @@ -43,14 +43,14 @@ SELECT data FROM ( ) t LIMIT 1; -- name: ListImportReferences :many -SELECT id FROM ( - SELECT e.id::text AS id FROM expert_skills l JOIN experts e ON e.id=l.expert_id +SELECT id, name FROM ( + SELECT e.id::text AS id, e.name FROM expert_skills l JOIN experts e ON e.id=l.expert_id WHERE $1::text='skill' AND l.skill_id=$2::uuid AND e.deleted_at IS NULL AND e.enabled UNION ALL - SELECT e.id::text AS id FROM expert_rules l JOIN experts e ON e.id=l.expert_id + SELECT e.id::text AS id, e.name FROM expert_rules l JOIN experts e ON e.id=l.expert_id WHERE $1::text='rule' AND l.rule_id=$2::uuid AND e.deleted_at IS NULL AND e.enabled UNION ALL - SELECT e.id::text AS id FROM expert_connectors l JOIN experts e ON e.id=l.expert_id + SELECT e.id::text AS id, e.name FROM expert_connectors l JOIN experts e ON e.id=l.expert_id WHERE $1::text='connector' AND l.connector_id=$2::uuid AND e.deleted_at IS NULL AND e.enabled ) refs; @@ -166,3 +166,49 @@ UPDATE connectors SET enabled=false,revision=revision+1,updated_at=now() WHERE i UPDATE resource_import_bindings SET retired_at=now(),retired_import_id=sqlc.arg(import_id)::uuid, last_import_id=sqlc.arg(import_id)::uuid,updated_at=now() WHERE publisher=sqlc.arg(publisher) AND slug=sqlc.arg(slug); + +-- name: CreateImportUpload :one +INSERT INTO resource_import_uploads (id, actor_user_id, files, total_size) +SELECT (d->>'id')::uuid, (d->>'actor_user_id')::uuid, d->'files', (d->>'total_size')::bigint +FROM (SELECT sqlc.arg(data)::jsonb AS d) input +WHERE (SELECT count(*) FROM resource_import_uploads + WHERE actor_user_id=(d->>'actor_user_id')::uuid AND status IN ('uploading','ready') AND expires_at>now()) < 5 +RETURNING to_jsonb(resource_import_uploads); + +-- name: ListImportUploads :many +SELECT to_jsonb(u) FROM resource_import_uploads u +WHERE actor_user_id=$1::uuid AND status IN ('uploading','ready') AND expires_at>now() +ORDER BY created_at DESC; + +-- name: GetImportUpload :one +SELECT to_jsonb(u) FROM resource_import_uploads u +WHERE id=sqlc.arg(id)::uuid AND actor_user_id=sqlc.arg(actor_user_id)::uuid AND status<>'cancelled' AND expires_at>now(); + +-- name: LockImportUpload :one +SELECT to_jsonb(u) FROM resource_import_uploads u +WHERE id=sqlc.arg(id)::uuid AND actor_user_id=sqlc.arg(actor_user_id)::uuid AND status<>'cancelled' AND expires_at>now() +FOR UPDATE; + +-- name: ListImportChunks :many +SELECT to_jsonb(c) FROM resource_import_chunks c WHERE upload_id=$1::uuid ORDER BY index; + +-- name: CreateImportChunk :exec +INSERT INTO resource_import_chunks (upload_id,index,size,sha256) +SELECT (d->>'upload_id')::uuid,(d->>'index')::integer,(d->>'size')::bigint,d->>'sha256' +FROM (SELECT sqlc.arg(data)::jsonb AS d) input; + +-- name: ReadyImportUpload :exec +UPDATE resource_import_uploads SET status='ready',package_sha256=sqlc.arg(package_sha256) WHERE id=sqlc.arg(id)::uuid; + +-- name: FinishImportUpload :exec +UPDATE resource_import_uploads SET status='imported' WHERE id=$1::uuid; + +-- name: CancelImportUpload :exec +UPDATE resource_import_uploads SET status='cancelled',package_sha256='',expires_at=now() WHERE id=$1::uuid; + +-- name: ListExpiredImportUploads :many +SELECT to_jsonb(u) FROM resource_import_uploads u WHERE expires_at<=now() +ORDER BY expires_at LIMIT 10 FOR UPDATE SKIP LOCKED; + +-- name: DeleteImportUpload :exec +DELETE FROM resource_import_uploads WHERE id=$1::uuid; diff --git a/monkeyai/backend/internal/resourceimport/reader.go b/monkeyai/backend/internal/resourceimport/reader.go new file mode 100644 index 000000000..3960e9c9f --- /dev/null +++ b/monkeyai/backend/internal/resourceimport/reader.go @@ -0,0 +1,92 @@ +package resourceimport + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "os" + + "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/resource" +) + +type archiveReader struct { + files []*os.File + sizes []int64 +} + +func (r *archiveReader) ReadAt(buffer []byte, offset int64) (int, error) { + if offset < 0 { + return 0, fmt.Errorf("读取偏移量无效") + } + if len(buffer) == 0 { + return 0, nil + } + read := 0 + for index, size := range r.sizes { + if offset >= size { + offset -= size + continue + } + length := min(int64(len(buffer)-read), size-offset) + n, err := r.files[index].ReadAt(buffer[read:read+int(length)], offset) + read += n + if err != nil { + return read, err + } + if read == len(buffer) { + return read, nil + } + offset = 0 + } + return read, io.EOF +} + +func (s *Service) openUpload(ctx context.Context, upload Upload) (*archiveReader, string, func(), error) { + if len(upload.Chunks) != upload.chunkCount() { + return nil, "", nil, resource.Invalid("资源包尚未上传完整，请继续上传缺失部分") + } + reader := &archiveReader{} + cleanup := func() { + for _, file := range reader.files { + _ = file.Close() + _ = os.Remove(file.Name()) + } + } + fail := func(err error) (*archiveReader, string, func(), error) { + cleanup() + return nil, "", nil, err + } + hash := sha256.New() + for index, chunk := range upload.Chunks { + if chunk.Index != index || chunk.Size != upload.chunkSize(index) { + return fail(resource.Invalid("资源包传输块不完整，请重新上传")) + } + if err := ctx.Err(); err != nil { + return fail(err) + } + stream, err := s.Storage.Get(ctx, upload.chunkKey(index)) + if err != nil { + return fail(fmt.Errorf("读取资源包传输块 %d: %w", index+1, err)) + } + file, err := os.CreateTemp("", "resource-import-*") + if err != nil { + _ = stream.Close() + return fail(err) + } + reader.files = append(reader.files, file) + reader.sizes = append(reader.sizes, chunk.Size) + chunkHash := sha256.New() + n, readErr := io.Copy(io.MultiWriter(file, hash, chunkHash), io.LimitReader(stream, chunk.Size+1)) + closeErr := stream.Close() + if readErr != nil || closeErr != nil || n != chunk.Size || "sha256:"+hex.EncodeToString(chunkHash.Sum(nil)) != chunk.SHA256 { + return fail(resource.Invalid(fmt.Sprintf("资源包传输块 %d 损坏，请重新上传", index+1))) + } + } + sha := "sha256:" + hex.EncodeToString(hash.Sum(nil)) + if upload.PackageSHA256 != "" && upload.PackageSHA256 != sha { + return fail(resource.Invalid("资源包内容已变化，请重新上传")) + } + return reader, sha, cleanup, nil +} diff --git a/monkeyai/backend/internal/resourceimport/service.go b/monkeyai/backend/internal/resourceimport/service.go index 7c246fe83..36a416f31 100644 --- a/monkeyai/backend/internal/resourceimport/service.go +++ b/monkeyai/backend/internal/resourceimport/service.go @@ -2,16 +2,12 @@ package resourceimport import ( "context" - "crypto/sha256" "encoding/base64" - "encoding/hex" "encoding/json" "errors" "fmt" "hash/fnv" - "io" "log/slog" - "mime/multipart" "net/http" "path" "slices" @@ -54,72 +50,15 @@ func decodeHistoryCursor(value string) ([]byte, error) { } func (s *Service) RegisterAdmin(router chi.Router) { - load := func(w http.ResponseWriter, r *http.Request) (pkg Package, items Resources, options Options, sha string, cleanup func(), err error) { - r.Body = http.MaxBytesReader(w, r.Body, MaxPackage+(16<<20)) - if err = r.ParseMultipartForm(8 << 20); err != nil { - if r.MultipartForm != nil { - r.MultipartForm.RemoveAll() - } - err = resource.Invalid("资源包上传格式或大小无效") - return - } - var file multipart.File - cleanup = func() { - if file != nil { - file.Close() - } - if r.MultipartForm != nil { - if removeErr := r.MultipartForm.RemoveAll(); removeErr != nil { - slog.WarnContext(r.Context(), "清理导入临时文件失败", "error", removeErr) - } - } - } - defer func() { - if err != nil { - cleanup() - cleanup = nil - } - }() - var header *multipart.FileHeader - file, header, err = r.FormFile("package") - if err != nil { - err = resource.Invalid("缺少资源包文件") - return - } - if header.Size <= 0 || header.Size > MaxPackage { - err = resource.Invalid("资源包超过大小限制") - return - } - if value := r.FormValue("options"); value != "" { - if json.Unmarshal([]byte(value), &options) != nil { - err = resource.Invalid("导入选项无效") - return - } - } - hash := sha256.New() - if _, err = io.Copy(hash, io.NewSectionReader(file, 0, header.Size)); err != nil { - return - } - sha = "sha256:" + hex.EncodeToString(hash.Sum(nil)) - pkg, err = ParseReader(file, header.Size) - if err != nil { - err = resource.Invalid(err.Error()) - return - } - items, err = ParseResources(pkg) - if err != nil { - err = resource.Invalid(err.Error()) - } - return - } + s.RegisterUploads(router) router.Post("/resource-imports/preview", func(w http.ResponseWriter, r *http.Request) { - pkg, items, options, sha, cleanup, err := load(w, r) + loaded, err := s.loadImport(w, r) if err != nil { resource.Fail(w, err) return } - defer cleanup() - out, err := s.Evaluate(r.Context(), s.Pool, pkg, items, options, sha) + defer loaded.cleanup() + out, err := s.Evaluate(r.Context(), s.Pool, loaded.pkg, loaded.items, loaded.input.Options, loaded.sha) if err != nil { resource.Fail(w, &resource.Error{Status: 409, Code: "import_conflict", Message: err.Error()}) return @@ -127,17 +66,20 @@ func (s *Service) RegisterAdmin(router chi.Router) { resource.JSON(w, http.StatusOK, out) }) router.Post("/resource-imports", func(w http.ResponseWriter, r *http.Request) { - pkg, items, options, sha, cleanup, err := load(w, r) + loaded, err := s.loadImport(w, r) if err != nil { resource.Fail(w, err) return } - defer cleanup() - if r.FormValue("package_sha256") != sha || r.FormValue("plan_digest") == "" || !options.ConfirmFinal { + defer loaded.cleanup() + if loaded.input.PackageSHA256 != loaded.sha || loaded.input.PlanDigest == "" || !loaded.input.Options.ConfirmFinal { resource.Fail(w, resource.Invalid("必须先预览并确认本包是最终 Agent 资源集合")) return } - out, err := s.Apply(r.Context(), pkg, items, options, sha, r.FormValue("plan_digest")) + out, err := s.Apply(r.Context(), loaded.pkg, loaded.items, loaded.input.Options, loaded.sha, loaded.input.PlanDigest) + if err == nil { + err = loaded.finish(r.Context()) + } if err != nil { resource.Fail(w, err) return diff --git a/monkeyai/backend/internal/resourceimport/sqlc/query.sql.go b/monkeyai/backend/internal/resourceimport/sqlc/query.sql.go index e957ac593..b8808fe28 100644 --- a/monkeyai/backend/internal/resourceimport/sqlc/query.sql.go +++ b/monkeyai/backend/internal/resourceimport/sqlc/query.sql.go @@ -9,6 +9,15 @@ import ( "context" ) +const cancelImportUpload = `-- name: CancelImportUpload :exec +UPDATE resource_import_uploads SET status='cancelled',package_sha256='',expires_at=now() WHERE id=$1::uuid +` + +func (q *Queries) CancelImportUpload(ctx context.Context, dollar_1 string) error { + _, err := q.db.Exec(ctx, cancelImportUpload, dollar_1) + return err +} + const clearImportExpertConnectors = `-- name: ClearImportExpertConnectors :exec DELETE FROM expert_connectors WHERE expert_id=$1::uuid ` @@ -49,6 +58,17 @@ func (q *Queries) CreateImportBatch(ctx context.Context, data []byte) error { return err } +const createImportChunk = `-- name: CreateImportChunk :exec +INSERT INTO resource_import_chunks (upload_id,index,size,sha256) +SELECT (d->>'upload_id')::uuid,(d->>'index')::integer,(d->>'size')::bigint,d->>'sha256' +FROM (SELECT $1::jsonb AS d) input +` + +func (q *Queries) CreateImportChunk(ctx context.Context, data []byte) error { + _, err := q.db.Exec(ctx, createImportChunk, data) + return err +} + const createImportConnector = `-- name: CreateImportConnector :exec INSERT INTO connectors (id,owner_user_id,ownership_type,origin,name,description,url, authorization_mode,authorization_method,oauth_config,timeout_ms,auth_header_name,icon_s3_key,name_i18n,description_i18n) @@ -99,6 +119,40 @@ func (q *Queries) CreateImportSkill(ctx context.Context, data []byte) error { return err } +const createImportUpload = `-- name: CreateImportUpload :one +INSERT INTO resource_import_uploads (id, actor_user_id, files, total_size) +SELECT (d->>'id')::uuid, (d->>'actor_user_id')::uuid, d->'files', (d->>'total_size')::bigint +FROM (SELECT $1::jsonb AS d) input +WHERE (SELECT count(*) FROM resource_import_uploads + WHERE actor_user_id=(d->>'actor_user_id')::uuid AND status IN ('uploading','ready') AND expires_at>now()) < 5 +RETURNING to_jsonb(resource_import_uploads) +` + +func (q *Queries) CreateImportUpload(ctx context.Context, data []byte) ([]byte, error) { + row := q.db.QueryRow(ctx, createImportUpload, data) + var to_jsonb []byte + err := row.Scan(&to_jsonb) + return to_jsonb, err +} + +const deleteImportUpload = `-- name: DeleteImportUpload :exec +DELETE FROM resource_import_uploads WHERE id=$1::uuid +` + +func (q *Queries) DeleteImportUpload(ctx context.Context, dollar_1 string) error { + _, err := q.db.Exec(ctx, deleteImportUpload, dollar_1) + return err +} + +const finishImportUpload = `-- name: FinishImportUpload :exec +UPDATE resource_import_uploads SET status='imported' WHERE id=$1::uuid +` + +func (q *Queries) FinishImportUpload(ctx context.Context, dollar_1 string) error { + _, err := q.db.Exec(ctx, finishImportUpload, dollar_1) + return err +} + const getImportHistory = `-- name: GetImportHistory :one SELECT to_jsonb(i) FROM resource_imports i WHERE id = $1::uuid ` @@ -145,6 +199,23 @@ func (q *Queries) GetImportTarget(ctx context.Context, arg GetImportTargetParams return data, err } +const getImportUpload = `-- name: GetImportUpload :one +SELECT to_jsonb(u) FROM resource_import_uploads u +WHERE id=$1::uuid AND actor_user_id=$2::uuid AND status<>'cancelled' AND expires_at>now() +` + +type GetImportUploadParams struct { + ID string + ActorUserID string +} + +func (q *Queries) GetImportUpload(ctx context.Context, arg GetImportUploadParams) ([]byte, error) { + row := q.db.QueryRow(ctx, getImportUpload, arg.ID, arg.ActorUserID) + var to_jsonb []byte + err := row.Scan(&to_jsonb) + return to_jsonb, err +} + const getLatestImport = `-- name: GetLatestImport :one SELECT version, release_id::text AS release_id, manifest_sha256 FROM resource_imports WHERE publisher = $1 AND status = 'succeeded' @@ -215,6 +286,31 @@ func (q *Queries) LinkImportExpertSkill(ctx context.Context, arg LinkImportExper return err } +const listExpiredImportUploads = `-- name: ListExpiredImportUploads :many +SELECT to_jsonb(u) FROM resource_import_uploads u WHERE expires_at<=now() +ORDER BY expires_at LIMIT 10 FOR UPDATE SKIP LOCKED +` + +func (q *Queries) ListExpiredImportUploads(ctx context.Context) ([][]byte, error) { + rows, err := q.db.Query(ctx, listExpiredImportUploads) + if err != nil { + return nil, err + } + defer rows.Close() + items := [][]byte{} + for rows.Next() { + var to_jsonb []byte + if err := rows.Scan(&to_jsonb); err != nil { + return nil, err + } + items = append(items, to_jsonb) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listImportBindings = `-- name: ListImportBindings :many SELECT to_jsonb(b) FROM resource_import_bindings b WHERE publisher = $1 ORDER BY slug ` @@ -239,6 +335,30 @@ func (q *Queries) ListImportBindings(ctx context.Context, publisher string) ([][ return items, nil } +const listImportChunks = `-- name: ListImportChunks :many +SELECT to_jsonb(c) FROM resource_import_chunks c WHERE upload_id=$1::uuid ORDER BY index +` + +func (q *Queries) ListImportChunks(ctx context.Context, dollar_1 string) ([][]byte, error) { + rows, err := q.db.Query(ctx, listImportChunks, dollar_1) + if err != nil { + return nil, err + } + defer rows.Close() + items := [][]byte{} + for rows.Next() { + var to_jsonb []byte + if err := rows.Scan(&to_jsonb); err != nil { + return nil, err + } + items = append(items, to_jsonb) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const listImportHistory = `-- name: ListImportHistory :many SELECT to_jsonb(i) FROM resource_imports i WHERE $1::jsonb = '{}'::jsonb @@ -273,14 +393,14 @@ func (q *Queries) ListImportHistory(ctx context.Context, arg ListImportHistoryPa } const listImportReferences = `-- name: ListImportReferences :many -SELECT id FROM ( - SELECT e.id::text AS id FROM expert_skills l JOIN experts e ON e.id=l.expert_id +SELECT id, name FROM ( + SELECT e.id::text AS id, e.name FROM expert_skills l JOIN experts e ON e.id=l.expert_id WHERE $1::text='skill' AND l.skill_id=$2::uuid AND e.deleted_at IS NULL AND e.enabled UNION ALL - SELECT e.id::text AS id FROM expert_rules l JOIN experts e ON e.id=l.expert_id + SELECT e.id::text AS id, e.name FROM expert_rules l JOIN experts e ON e.id=l.expert_id WHERE $1::text='rule' AND l.rule_id=$2::uuid AND e.deleted_at IS NULL AND e.enabled UNION ALL - SELECT e.id::text AS id FROM expert_connectors l JOIN experts e ON e.id=l.expert_id + SELECT e.id::text AS id, e.name FROM expert_connectors l JOIN experts e ON e.id=l.expert_id WHERE $1::text='connector' AND l.connector_id=$2::uuid AND e.deleted_at IS NULL AND e.enabled ) refs ` @@ -290,19 +410,50 @@ type ListImportReferencesParams struct { Column2 string } -func (q *Queries) ListImportReferences(ctx context.Context, arg ListImportReferencesParams) ([]string, error) { +type ListImportReferencesRow struct { + ID string + Name string +} + +func (q *Queries) ListImportReferences(ctx context.Context, arg ListImportReferencesParams) ([]ListImportReferencesRow, error) { rows, err := q.db.Query(ctx, listImportReferences, arg.Column1, arg.Column2) if err != nil { return nil, err } defer rows.Close() - items := []string{} + items := []ListImportReferencesRow{} + for rows.Next() { + var i ListImportReferencesRow + if err := rows.Scan(&i.ID, &i.Name); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + +const listImportUploads = `-- name: ListImportUploads :many +SELECT to_jsonb(u) FROM resource_import_uploads u +WHERE actor_user_id=$1::uuid AND status IN ('uploading','ready') AND expires_at>now() +ORDER BY created_at DESC +` + +func (q *Queries) ListImportUploads(ctx context.Context, dollar_1 string) ([][]byte, error) { + rows, err := q.db.Query(ctx, listImportUploads, dollar_1) + if err != nil { + return nil, err + } + defer rows.Close() + items := [][]byte{} for rows.Next() { - var id string - if err := rows.Scan(&id); err != nil { + var to_jsonb []byte + if err := rows.Scan(&to_jsonb); err != nil { return nil, err } - items = append(items, id) + items = append(items, to_jsonb) } if err := rows.Err(); err != nil { return nil, err @@ -365,6 +516,38 @@ func (q *Queries) LockImportSkill(ctx context.Context, dollar_1 string) (string, return id, err } +const lockImportUpload = `-- name: LockImportUpload :one +SELECT to_jsonb(u) FROM resource_import_uploads u +WHERE id=$1::uuid AND actor_user_id=$2::uuid AND status<>'cancelled' AND expires_at>now() +FOR UPDATE +` + +type LockImportUploadParams struct { + ID string + ActorUserID string +} + +func (q *Queries) LockImportUpload(ctx context.Context, arg LockImportUploadParams) ([]byte, error) { + row := q.db.QueryRow(ctx, lockImportUpload, arg.ID, arg.ActorUserID) + var to_jsonb []byte + err := row.Scan(&to_jsonb) + return to_jsonb, err +} + +const readyImportUpload = `-- name: ReadyImportUpload :exec +UPDATE resource_import_uploads SET status='ready',package_sha256=$1 WHERE id=$2::uuid +` + +type ReadyImportUploadParams struct { + PackageSha256 string + ID string +} + +func (q *Queries) ReadyImportUpload(ctx context.Context, arg ReadyImportUploadParams) error { + _, err := q.db.Exec(ctx, readyImportUpload, arg.PackageSha256, arg.ID) + return err +} + const retireImportBinding = `-- name: RetireImportBinding :exec UPDATE resource_import_bindings SET retired_at=now(),retired_import_id=$1::uuid, last_import_id=$1::uuid,updated_at=now() diff --git a/monkeyai/backend/internal/resourceimport/upload.go b/monkeyai/backend/internal/resourceimport/upload.go new file mode 100644 index 000000000..4d2d0c1e3 --- /dev/null +++ b/monkeyai/backend/internal/resourceimport/upload.go @@ -0,0 +1,430 @@ +package resourceimport + +import ( + "context" + "crypto/sha256" + "encoding/hex" + "errors" + "fmt" + "io" + "log/slog" + "net/http" + "os" + "regexp" + "strconv" + "strings" + "time" + "unicode/utf8" + "uuid" + + "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/identity" + "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/resource" + "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/resourceimport/sqlc" + "github.com/go-chi/chi/v5" + "github.com/jackc/pgx/v5" +) + +const ChunkSize int64 = 64 << 20 + +var volumeName = regexp.MustCompile(`(?i)^(.+\.zip)\.([0-9]{3,6})$`) + +type UploadFile struct { + Name string `json:"name"` + Size int64 `json:"size"` +} + +type Chunk struct { + Index int `json:"index"` + Size int64 `json:"size"` + SHA256 string `json:"sha256"` +} + +type Upload struct { + ID string `json:"id"` + Files []UploadFile `json:"files"` + TotalSize int64 `json:"total_size"` + Status string `json:"status"` + PackageSHA256 string `json:"package_sha256"` + CreatedAt time.Time `json:"created_at"` + ExpiresAt time.Time `json:"expires_at"` + ChunkSize int64 `json:"chunk_size"` + Chunks []Chunk `json:"chunks"` +} + +func validateUpload(files []UploadFile) (int64, error) { + if len(files) == 0 || len(files) > 256 { + return 0, resource.Invalid("请选择 ZIP 或全部分卷，最多 256 卷") + } + var total int64 + base := "" + for i, file := range files { + if !safePath(file.Name) || strings.Contains(file.Name, "/") || !utf8.ValidString(file.Name) || len(file.Name) > 255 || file.Size <= 0 || file.Size > MaxPackage-total { + return 0, resource.Invalid("文件名或大小无效，资源包总大小不得超过 2 GiB") + } + total += file.Size + if len(files) == 1 && strings.HasSuffix(strings.ToLower(file.Name), ".zip") { + continue + } + match := volumeName.FindStringSubmatch(file.Name) + if match == nil { + return 0, resource.Invalid("请选择普通 ZIP，或按字节拆分的 .zip.001、.002 等全部分卷；不支持 .z01 或 7z 分卷") + } + n, _ := strconv.Atoi(match[2]) + if n != i+1 || (base != "" && !strings.EqualFold(base, match[1])) { + return 0, resource.Invalid("分卷编号必须从 001 连续递增，且属于同一资源包") + } + base = match[1] + } + return total, nil +} + +func (u Upload) chunkCount() int { return int((u.TotalSize + ChunkSize - 1) / ChunkSize) } +func (u Upload) chunkSize(index int) int64 { + if index < 0 || index >= u.chunkCount() { + return 0 + } + return min(ChunkSize, u.TotalSize-int64(index)*ChunkSize) +} +func (u Upload) chunkKey(index int) string { + return fmt.Sprintf("resource-import/uploads/%s/%d", u.ID, index) +} + +func decodeUpload(data []byte) (Upload, error) { + var upload Upload + err := decodeJSON(data, &upload) + upload.ChunkSize, upload.Chunks = ChunkSize, []Chunk{} + return upload, err +} + +func (s *Service) uploadChunks(ctx context.Context, q resource.Queryer, upload *Upload) error { + rows, err := sqlc.New(q).ListImportChunks(ctx, upload.ID) + if err != nil { + return err + } + for _, data := range rows { + var chunk Chunk + if err := decodeJSON(data, &chunk); err != nil { + return err + } + upload.Chunks = append(upload.Chunks, chunk) + } + return nil +} + +func (s *Service) lockUpload(ctx context.Context, id, actor string) (Upload, pgx.Tx, error) { + if _, err := uuid.Parse(id); err != nil { + return Upload{}, nil, resource.NotFound + } + tx, err := s.Pool.Begin(ctx) + if err != nil { + return Upload{}, nil, err + } + data, err := sqlc.New(tx).LockImportUpload(ctx, sqlc.LockImportUploadParams{ID: id, ActorUserID: actor}) + if err != nil { + _ = tx.Rollback(context.WithoutCancel(ctx)) + return Upload{}, nil, err + } + upload, err := decodeUpload(data) + if err == nil { + err = s.uploadChunks(ctx, tx, &upload) + } + if err != nil { + _ = tx.Rollback(context.WithoutCancel(ctx)) + return Upload{}, nil, err + } + return upload, tx, nil +} + +func (s *Service) RegisterUploads(router chi.Router) { + router.Route("/resource-import-uploads", func(router chi.Router) { + router.Use(func(next http.Handler) http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + actor, ok := identity.UserFromContext(r.Context()) + if !ok || actor.Role != "admin" { + resource.Fail(w, resource.NotFound) + return + } + next.ServeHTTP(w, r) + }) + }) + router.Post("/", s.createUpload) + router.Get("/", s.listUploads) + router.Get("/{id}", s.getUpload) + router.Put("/{id}/chunks/{index}", s.putChunk) + router.Post("/{id}/complete", s.completeUpload) + router.Delete("/{id}", s.cancelUpload) + }) +} + +func (s *Service) createUpload(w http.ResponseWriter, r *http.Request) { + var input struct { + Files []UploadFile `json:"files"` + } + if err := resource.Decode(w, r, &input); err != nil { + resource.Fail(w, err) + return + } + total, err := validateUpload(input.Files) + if err != nil { + resource.Fail(w, err) + return + } + actor, _ := identity.UserFromContext(r.Context()) + data, err := sqlc.New(s.Pool).CreateImportUpload(r.Context(), payload(resource.Object{ + "id": resource.ID(), "actor_user_id": actor.ID, "files": input.Files, "total_size": total, + })) + if errors.Is(err, pgx.ErrNoRows) { + err = conflict("最多保留 5 个未完成上传任务，请先继续或放弃已有任务") + } + if err != nil { + resource.Fail(w, err) + return + } + upload, err := decodeUpload(data) + if err != nil { + resource.Fail(w, err) + return + } + resource.JSON(w, http.StatusCreated, upload) +} + +func (s *Service) listUploads(w http.ResponseWriter, r *http.Request) { + actor, _ := identity.UserFromContext(r.Context()) + rows, err := sqlc.New(s.Pool).ListImportUploads(r.Context(), actor.ID) + if err != nil { + resource.Fail(w, err) + return + } + uploads := []Upload{} + for _, data := range rows { + upload, err := decodeUpload(data) + if err == nil { + err = s.uploadChunks(r.Context(), s.Pool, &upload) + } + if err != nil { + resource.Fail(w, err) + return + } + uploads = append(uploads, upload) + } + resource.JSON(w, http.StatusOK, resource.Object{"items": uploads}) +} + +func (s *Service) getUpload(w http.ResponseWriter, r *http.Request) { + id := chi.URLParam(r, "id") + if _, err := uuid.Parse(id); err != nil { + resource.Fail(w, resource.NotFound) + return + } + actor, _ := identity.UserFromContext(r.Context()) + data, err := sqlc.New(s.Pool).GetImportUpload(r.Context(), sqlc.GetImportUploadParams{ID: id, ActorUserID: actor.ID}) + if err != nil { + resource.Fail(w, err) + return + } + upload, err := decodeUpload(data) + if err == nil { + err = s.uploadChunks(r.Context(), s.Pool, &upload) + } + if err != nil { + resource.Fail(w, err) + return + } + resource.JSON(w, http.StatusOK, upload) +} + +func (s *Service) putChunk(w http.ResponseWriter, r *http.Request) { + actor, _ := identity.UserFromContext(r.Context()) + upload, tx, err := s.lockUpload(r.Context(), chi.URLParam(r, "id"), actor.ID) + if err != nil { + resource.Fail(w, err) + return + } + defer tx.Rollback(context.WithoutCancel(r.Context())) + index, err := strconv.Atoi(chi.URLParam(r, "index")) + size := upload.chunkSize(index) + if err != nil || size == 0 { + resource.Fail(w, resource.Invalid("传输块编号无效")) + return + } + if upload.Status != "uploading" { + resource.Fail(w, conflict("上传内容已封存，不能再修改")) + return + } + expected := r.Header.Get("X-Content-SHA256") + if len(expected) != 71 || !strings.HasPrefix(expected, "sha256:") { + resource.Fail(w, resource.Invalid("缺少传输块 SHA-256")) + return + } + file, err := os.CreateTemp("", "resource-import-chunk-*") + if err != nil { + resource.Fail(w, err) + return + } + defer func() { _ = file.Close(); _ = os.Remove(file.Name()) }() + hash := sha256.New() + length, err := io.Copy(io.MultiWriter(file, hash), http.MaxBytesReader(w, r.Body, size)) + if err != nil || length != size || "sha256:"+hex.EncodeToString(hash.Sum(nil)) != expected { + resource.Fail(w, resource.Invalid("传输块大小或校验值不匹配，请重试")) + return + } + for _, chunk := range upload.Chunks { + if chunk.Index == index { + if chunk.SHA256 != expected { + resource.Fail(w, conflict("该传输块已有不同内容，请重新选择资源包创建任务")) + return + } + w.WriteHeader(http.StatusNoContent) + return + } + } + if _, err := file.Seek(0, io.SeekStart); err != nil { + resource.Fail(w, err) + return + } + if writer, ok := s.Storage.(interface { + PutReader(context.Context, string, io.ReadSeeker, int64, string) error + }); ok { + err = writer.PutReader(r.Context(), upload.chunkKey(index), file, size, "application/octet-stream") + } else { + var body []byte + body, err = io.ReadAll(file) + if err == nil { + err = s.Storage.Put(r.Context(), upload.chunkKey(index), body, "application/octet-stream") + } + } + if err != nil { + resource.Fail(w, err) + return + } + if err := sqlc.New(tx).CreateImportChunk(r.Context(), payload(resource.Object{ + "upload_id": upload.ID, "index": index, "size": size, "sha256": expected, + })); err != nil { + resource.Fail(w, err) + return + } + if err := tx.Commit(r.Context()); err != nil { + resource.Fail(w, err) + return + } + w.WriteHeader(http.StatusNoContent) +} + +func (s *Service) completeUpload(w http.ResponseWriter, r *http.Request) { + actor, _ := identity.UserFromContext(r.Context()) + upload, tx, err := s.lockUpload(r.Context(), chi.URLParam(r, "id"), actor.ID) + if err != nil { + resource.Fail(w, err) + return + } + defer tx.Rollback(context.WithoutCancel(r.Context())) + if upload.Status != "uploading" { + resource.JSON(w, http.StatusOK, upload) + return + } + reader, sha, cleanup, err := s.openUpload(r.Context(), upload) + if err != nil { + resource.Fail(w, err) + return + } + defer cleanup() + pkg, err := ParseReader(reader, upload.TotalSize) + if err == nil { + for name, entry := range pkg.entries { + stream, openErr := entry.Open() + if openErr != nil { + err = fmt.Errorf("打开文件 %s: %w", name, openErr) + break + } + n, readErr := io.Copy(io.Discard, io.LimitReader(stream, int64(entry.UncompressedSize64)+1)) + closeErr := stream.Close() + if readErr != nil || closeErr != nil || uint64(n) != entry.UncompressedSize64 { + err = fmt.Errorf("文件 %s 损坏，请检查分卷是否完整且来自同一资源包", name) + break + } + } + } + if err != nil { + resource.Fail(w, resource.Invalid("资源包不完整或格式无效："+err.Error())) + return + } + if err := sqlc.New(tx).ReadyImportUpload(r.Context(), sqlc.ReadyImportUploadParams{ID: upload.ID, PackageSha256: sha}); err != nil { + resource.Fail(w, err) + return + } + if err := tx.Commit(r.Context()); err != nil { + resource.Fail(w, err) + return + } + upload.Status, upload.PackageSHA256 = "ready", sha + resource.JSON(w, http.StatusOK, upload) +} + +func (s *Service) cancelUpload(w http.ResponseWriter, r *http.Request) { + actor, _ := identity.UserFromContext(r.Context()) + upload, tx, err := s.lockUpload(r.Context(), chi.URLParam(r, "id"), actor.ID) + if err != nil { + resource.Fail(w, err) + return + } + defer tx.Rollback(context.WithoutCancel(r.Context())) + if upload.Status == "imported" { + resource.Fail(w, conflict("该任务已完成导入")) + return + } + if err := sqlc.New(tx).CancelImportUpload(r.Context(), upload.ID); err != nil { + resource.Fail(w, err) + return + } + if err := tx.Commit(r.Context()); err != nil { + resource.Fail(w, err) + return + } + w.WriteHeader(http.StatusNoContent) +} + +func (s *Service) CleanupUploads(ctx context.Context) error { + tx, err := s.Pool.Begin(ctx) + if err != nil { + return err + } + defer tx.Rollback(context.WithoutCancel(ctx)) + rows, err := sqlc.New(tx).ListExpiredImportUploads(ctx) + if err != nil { + return err + } + for _, data := range rows { + upload, err := decodeUpload(data) + if err != nil { + return err + } + // 同时删除可能在数据库提交失败时遗留的对象。 + for i := range upload.chunkCount() { + if err := s.Storage.Delete(ctx, upload.chunkKey(i)); err != nil { + return err + } + } + if err := sqlc.New(tx).DeleteImportUpload(ctx, upload.ID); err != nil { + return err + } + } + return tx.Commit(ctx) +} + +func (s *Service) Run(ctx context.Context) { + ticker := time.NewTicker(time.Minute) + defer ticker.Stop() + for { + cleanupCtx, cancel := context.WithTimeout(ctx, 50*time.Second) + err := s.CleanupUploads(cleanupCtx) + cancel() + if err != nil && ctx.Err() == nil { + slog.WarnContext(ctx, "清理资源包上传失败", "error", err) + } + select { + case <-ctx.Done(): + return + case <-ticker.C: + } + } +} diff --git a/monkeyai/backend/internal/resourceimport/upload_test.go b/monkeyai/backend/internal/resourceimport/upload_test.go new file mode 100644 index 000000000..723389fd4 --- /dev/null +++ b/monkeyai/backend/internal/resourceimport/upload_test.go @@ -0,0 +1,302 @@ +package resourceimport + +import ( + "bytes" + "context" + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "io" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "reflect" + "strings" + "sync" + "testing" + + "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/identity" + "github.com/chaitin/MonkeyCode/monkeyai/backend/internal/resource" + "github.com/go-chi/chi/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +func TestValidateUpload(t *testing.T) { + for _, files := range [][]UploadFile{ + {{"release.zip", 5}}, + {{"release.zip.001", 3}, {"release.zip.002", 2}}, + {{"release.ZIP.001", 5}}, + } { + if total, err := validateUpload(files); err != nil || total != 5 { + t.Fatalf("有效分卷被拒绝: %v %d %v", files, total, err) + } + } + for _, files := range [][]UploadFile{ + nil, {{"../release.zip", 1}}, {{"release.zip", 0}}, {{"release.zip", MaxPackage + 1}}, + {{"release.zip.002", 1}}, {{"release.zip.001", 1}, {"release.zip.003", 1}}, + {{"release.zip.001", 1}, {"other.zip.002", 1}}, {{"release.zip", 1}, {"other.zip", 1}}, + {{"release.z01", 1}}, {{"release.7z.001", 1}}, {{"release.zip.001", MaxPackage}, {"release.zip.002", 1}}, + } { + if _, err := validateUpload(files); err == nil { + t.Fatalf("无效分卷被接受: %v", files) + } + } +} + +func TestArchiveReader(t *testing.T) { + reader := &archiveReader{} + for _, content := range []string{"abc", "de", "fghij"} { + file, err := os.CreateTemp(t.TempDir(), "part") + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = file.Close() }) + if _, err := file.WriteString(content); err != nil { + t.Fatal(err) + } + reader.files = append(reader.files, file) + reader.sizes = append(reader.sizes, int64(len(content))) + } + for _, tc := range []struct { + offset int64 + size int + want string + eof bool + }{{0, 10, "abcdefghij", false}, {2, 6, "cdefgh", false}, {3, 2, "de", false}, {8, 5, "ij", true}, {10, 1, "", true}, {10, 0, "", false}} { + buffer := make([]byte, tc.size) + n, err := reader.ReadAt(buffer, tc.offset) + if string(buffer[:n]) != tc.want || (err == io.EOF) != tc.eof || (err != nil && err != io.EOF) { + t.Fatalf("跨卷读取错误: %+v: %q %v", tc, buffer[:n], err) + } + } +} + +type uploadStorage struct { + mu sync.Mutex + objects map[string][]byte +} + +func (s *uploadStorage) Put(_ context.Context, key string, data []byte, _ string) error { + s.mu.Lock() + defer s.mu.Unlock() + s.objects[key] = bytes.Clone(data) + return nil +} +func (s *uploadStorage) PutReader(ctx context.Context, key string, reader io.ReadSeeker, size int64, mime string) error { + body, err := io.ReadAll(io.LimitReader(reader, size+1)) + if err != nil || int64(len(body)) != size { + return fmt.Errorf("流式上传大小无效: %d %v", len(body), err) + } + return s.Put(ctx, key, body, mime) +} +func (s *uploadStorage) Get(_ context.Context, key string) (io.ReadCloser, error) { + s.mu.Lock() + defer s.mu.Unlock() + data, ok := s.objects[key] + if !ok { + return nil, fmt.Errorf("对象不存在: %s", key) + } + return io.NopCloser(bytes.NewReader(bytes.Clone(data))), nil +} +func (s *uploadStorage) Delete(_ context.Context, key string) error { + s.mu.Lock() + defer s.mu.Unlock() + delete(s.objects, key) + return nil +} +func (s *uploadStorage) Ping(context.Context) error { return nil } + +func uploadDatabase(t *testing.T) *pgxpool.Pool { + t.Helper() + dsn := os.Getenv("MONKEYAI_TEST_DATABASE_URL") + if dsn == "" { + t.Skip("设置 MONKEYAI_TEST_DATABASE_URL 运行分卷上传集成测试") + } + root, err := pgxpool.New(t.Context(), dsn) + if err != nil { + t.Fatal(err) + } + schema := "import_uploads_" + strings.ReplaceAll(resource.ID(), "-", "") + if _, err := root.Exec(t.Context(), "CREATE SCHEMA "+schema); err != nil { + root.Close() + t.Fatal(err) + } + t.Cleanup(func() { _, _ = root.Exec(context.Background(), "DROP SCHEMA "+schema+" CASCADE"); root.Close() }) + config, err := pgxpool.ParseConfig(dsn) + if err != nil { + t.Fatal(err) + } + config.ConnConfig.RuntimeParams["search_path"] = schema + pool, err := pgxpool.NewWithConfig(t.Context(), config) + if err != nil { + t.Fatal(err) + } + t.Cleanup(pool.Close) + files, err := filepath.Glob("../../migrations/*.up.sql") + if err != nil { + t.Fatal(err) + } + for _, file := range files { + data, err := os.ReadFile(file) + if err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(t.Context(), string(data)); err != nil { + t.Fatal(err) + } + } + return pool +} + +func TestUploadIntegration(t *testing.T) { + pool := uploadDatabase(t) + storage := &uploadStorage{objects: map[string][]byte{}} + service := NewService(pool, storage) + identities := identity.NewService(pool, nil, "http://localhost") + cookies := map[string]*http.Cookie{} + for _, name := range []string{"admin", "other"} { + id := resource.ID() + if _, err := pool.Exec(t.Context(), "INSERT INTO users(id,name,email,role) VALUES($1,$2,$3,'admin')", id, name, name+"@example.com"); err != nil { + t.Fatal(err) + } + hash := sha256.Sum256([]byte(name)) + if _, err := pool.Exec(t.Context(), "INSERT INTO browser_sessions(token_hash,user_id,authentication_method,expires_at) VALUES($1,$2,'password',now()+interval '1 hour')", hex.EncodeToString(hash[:]), id); err != nil { + t.Fatal(err) + } + cookies[name] = &http.Cookie{Name: "monkeyai_session", Value: name} + } + newRouter := func() http.Handler { + router := chi.NewRouter() + router.Use(identities.RequireAdmin) + NewService(pool, storage).RegisterAdmin(router) + return router + } + router := newRouter() + call := func(actor, method, path string, body any, status int) []byte { + t.Helper() + var raw []byte + binary, ok := body.([]byte) + if ok { + raw = binary + } else if body != nil { + raw, _ = json.Marshal(body) + } + req := httptest.NewRequest(method, path, bytes.NewReader(raw)) + req.AddCookie(cookies[actor]) + req.Header.Set("Content-Type", "application/json") + if ok { + req.Header.Set("Content-Type", "application/octet-stream") + req.Header.Set("X-Content-SHA256", digest(raw)) + } + w := httptest.NewRecorder() + router.ServeHTTP(w, req) + if w.Code != status { + t.Fatalf("%s %s: 状态 %d，期望 %d: %s", method, path, w.Code, status, w.Body.String()) + } + return w.Body.Bytes() + } + archive := packageFixture(t, nil) + input := resource.Object{"files": []UploadFile{{"release.zip.001", int64(len(archive) / 2)}, {"release.zip.002", int64(len(archive) - len(archive)/2)}}} + var upload Upload + if err := json.Unmarshal(call("admin", "POST", "/resource-import-uploads", input, 201), &upload); err != nil { + t.Fatal(err) + } + path := "/resource-import-uploads/" + upload.ID + call("other", "GET", path, nil, 404) + call("other", "PUT", path+"/chunks/0", archive, 404) + call("admin", "POST", path+"/complete", nil, 400) + call("admin", "PUT", path+"/chunks/1", archive, 400) + call("admin", "PUT", path+"/chunks/0", archive, 204) + call("admin", "PUT", path+"/chunks/0", archive, 204) + call("admin", "PUT", path+"/chunks/0", bytes.Repeat([]byte("x"), len(archive)), 409) + // 新建处理器模拟进程重启，上传内容和进度不依赖进程内存。 + router = newRouter() + var resumed Upload + if err := json.Unmarshal(call("admin", "GET", path, nil, 200), &resumed); err != nil || len(resumed.Chunks) != 1 { + t.Fatalf("上传进度未恢复: %+v %v", resumed, err) + } + call("admin", "POST", "/resource-imports/preview", importInput{UploadID: upload.ID}, 400) + call("admin", "POST", path+"/complete", nil, 200) + call("admin", "POST", path+"/complete", nil, 200) + call("admin", "PUT", path+"/chunks/0", archive, 409) + var preview Preview + if err := json.Unmarshal(call("admin", "POST", "/resource-imports/preview", importInput{UploadID: upload.ID}, 200), &preview); err != nil { + t.Fatal(err) + } + if preview.PackageSHA256 != digest(archive) || len(preview.Changes) != 1 || preview.Changes[0].Action != "create" { + t.Fatalf("分卷预览与完整 ZIP 不符: %+v", preview) + } + call("other", "POST", "/resource-imports/preview", importInput{UploadID: upload.ID}, 404) + apply := importInput{UploadID: upload.ID, Options: Options{ConfirmFinal: true, AuthorizationModes: map[string]string{}}, PackageSHA256: preview.PackageSHA256, PlanDigest: "stale"} + call("admin", "POST", "/resource-imports", apply, 409) + apply.PlanDigest = preview.PlanDigest + apply.Options.ConfirmFinal = false + call("admin", "POST", "/resource-imports", apply, 400) + apply.Options.ConfirmFinal = true + call("admin", "POST", "/resource-imports", apply, 200) + var imported Upload + _ = json.Unmarshal(call("admin", "GET", path, nil, 200), &imported) + if imported.Status != "imported" { + t.Fatalf("导入任务状态未更新: %+v", imported) + } + var count int + if err := pool.QueryRow(t.Context(), "SELECT count(*) FROM resource_imports").Scan(&count); err != nil || count != 1 { + t.Fatalf("分卷应只产生一个批次: %d %v", count, err) + } + t.Run("外部Expert阻止停用并返回可读名称", func(t *testing.T) { + expertID := "d02112c9-1a1a-43bb-9757-5c354eefe4e7" + if _, err := pool.Exec(t.Context(), `INSERT INTO experts(id,name,description,prompt,created_by_user_id,owner_user_id) + SELECT $1,'外部研究助手','','',owner_user_id,owner_user_id FROM skills LIMIT 1`, expertID); err != nil { + t.Fatal(err) + } + if _, err := pool.Exec(t.Context(), "INSERT INTO expert_skills(expert_id,skill_id) SELECT $1,id FROM skills LIMIT 1", expertID); err != nil { + t.Fatal(err) + } + removed := packageFixture(t, func(manifest *Manifest, _ map[string][]byte) { manifest.AgentResources.Skills = nil }) + pkg, err := ParseReader(bytes.NewReader(removed), int64(len(removed))) + if err != nil { + t.Fatal(err) + } + items, err := ParseResources(pkg) + if err != nil { + t.Fatal(err) + } + _, err = service.Evaluate(t.Context(), pool, pkg, items, Options{}, digest(removed)) + if err == nil || !strings.Contains(err.Error(), "外部研究助手") || !strings.Contains(err.Error(), expertID) || !strings.Contains(err.Error(), "不属于本发布者") { + t.Fatalf("未说明外部引用冲突: %v", err) + } + if _, err := pool.Exec(t.Context(), "UPDATE experts SET enabled=false WHERE id=$1", expertID); err != nil { + t.Fatal(err) + } + preview, err := service.Evaluate(t.Context(), pool, pkg, items, Options{}, digest(removed)) + if err != nil || len(preview.Changes) != 1 || preview.Changes[0].Action != "retire" { + t.Fatalf("解除有效引用后应允许预览停用: %+v %v", preview, err) + } + }) + if _, err := pool.Exec(t.Context(), "UPDATE resource_import_uploads SET expires_at=now()-interval '1 second' WHERE id=$1", upload.ID); err != nil { + t.Fatal(err) + } + call("admin", "GET", path, nil, 404) + if err := service.CleanupUploads(t.Context()); err != nil { + t.Fatal(err) + } + if _, found := storage.objects[upload.chunkKey(0)]; found { + t.Fatal("过期上传块未清理") + } + var cancelled Upload + _ = json.Unmarshal(call("admin", "POST", "/resource-import-uploads", input, 201), &cancelled) + call("admin", "DELETE", "/resource-import-uploads/"+cancelled.ID, nil, 204) + call("admin", "GET", "/resource-import-uploads/"+cancelled.ID, nil, 404) + if err := service.CleanupUploads(t.Context()); err != nil { + t.Fatal(err) + } + var list struct { + Items []Upload `json:"items"` + } + _ = json.Unmarshal(call("admin", "GET", "/resource-import-uploads", nil, 200), &list) + if !reflect.DeepEqual(list.Items, []Upload{}) { + t.Fatalf("已完成和已取消任务不应显示为未完成: %+v", list) + } +} diff --git a/monkeyai/backend/migrations/000027_resource_import_uploads.down.sql b/monkeyai/backend/migrations/000027_resource_import_uploads.down.sql new file mode 100644 index 000000000..48f1d9f4c --- /dev/null +++ b/monkeyai/backend/migrations/000027_resource_import_uploads.down.sql @@ -0,0 +1,2 @@ +DROP TABLE resource_import_chunks; +DROP TABLE resource_import_uploads; diff --git a/monkeyai/backend/migrations/000027_resource_import_uploads.up.sql b/monkeyai/backend/migrations/000027_resource_import_uploads.up.sql new file mode 100644 index 000000000..e33a39cc2 --- /dev/null +++ b/monkeyai/backend/migrations/000027_resource_import_uploads.up.sql @@ -0,0 +1,22 @@ +CREATE TABLE resource_import_uploads ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + actor_user_id uuid NOT NULL REFERENCES users(id), + files jsonb NOT NULL CHECK (jsonb_typeof(files) = 'array'), + total_size bigint NOT NULL CHECK (total_size BETWEEN 1 AND 2147483648), + status text NOT NULL DEFAULT 'uploading' CHECK (status IN ('uploading', 'ready', 'imported', 'cancelled')), + package_sha256 text NOT NULL DEFAULT '', + created_at timestamptz NOT NULL DEFAULT now(), + expires_at timestamptz NOT NULL DEFAULT now() + interval '24 hours', + CHECK ((status IN ('ready', 'imported')) = (package_sha256 ~ '^sha256:[0-9a-f]{64}$')) +); + +CREATE INDEX resource_import_uploads_actor_created_idx ON resource_import_uploads (actor_user_id, created_at DESC); +CREATE INDEX resource_import_uploads_expires_idx ON resource_import_uploads (expires_at); + +CREATE TABLE resource_import_chunks ( + upload_id uuid NOT NULL REFERENCES resource_import_uploads(id) ON DELETE CASCADE, + index integer NOT NULL CHECK (index BETWEEN 0 AND 31), + size bigint NOT NULL CHECK (size BETWEEN 1 AND 67108864), + sha256 text NOT NULL CHECK (sha256 ~ '^sha256:[0-9a-f]{64}$'), + PRIMARY KEY (upload_id, index) +); diff --git a/monkeyai/backend/schema/schema.sql b/monkeyai/backend/schema/schema.sql index 0d32163d9..a7528ecb5 100644 --- a/monkeyai/backend/schema/schema.sql +++ b/monkeyai/backend/schema/schema.sql @@ -1423,3 +1423,23 @@ ALTER TABLE device_client_connections ALTER TABLE endpoints DROP CONSTRAINT endpoints_platform_check, ADD CONSTRAINT endpoints_platform_check CHECK (octet_length(platform) BETWEEN 1 AND 64); + +CREATE TABLE resource_import_uploads ( + id uuid PRIMARY KEY DEFAULT gen_random_uuid(), + actor_user_id uuid NOT NULL REFERENCES users(id), + files jsonb NOT NULL CHECK (jsonb_typeof(files) = 'array'), + total_size bigint NOT NULL CHECK (total_size BETWEEN 1 AND 2147483648), + status text NOT NULL DEFAULT 'uploading' CHECK (status IN ('uploading', 'ready', 'imported', 'cancelled')), + package_sha256 text NOT NULL DEFAULT '', + created_at timestamptz NOT NULL DEFAULT now(), + expires_at timestamptz NOT NULL DEFAULT now() + interval '24 hours', + CHECK ((status IN ('ready', 'imported')) = (package_sha256 ~ '^sha256:[0-9a-f]{64}$')) +); + +CREATE TABLE resource_import_chunks ( + upload_id uuid NOT NULL REFERENCES resource_import_uploads(id) ON DELETE CASCADE, + index integer NOT NULL CHECK (index BETWEEN 0 AND 31), + size bigint NOT NULL CHECK (size BETWEEN 1 AND 67108864), + sha256 text NOT NULL CHECK (sha256 ~ '^sha256:[0-9a-f]{64}$'), + PRIMARY KEY (upload_id, index) +); diff --git a/monkeyai/design/resource-package-import-plan.md b/monkeyai/design/resource-package-import-plan.md index 21262233e..6a93176c8 100644 --- a/monkeyai/design/resource-package-import-plan.md +++ b/monkeyai/design/resource-package-import-plan.md @@ -1,5 +1,7 @@ # 管理端资源包导入方案 +分卷上传与续传的实现见 [资源包分卷导入实现](resource-package-volumes-plan.md)。原 multipart 接口保持兼容，管理端改用上传会话统一预览和确认。 + > 状态：2026-10-14 已按用户最新要求改为宽松导入，优先让资源包可导入。JSON 仅检查语法，取消发布协议摘要、字段格式和产物一致性检查。五个真实发布 ZIP 的解析、后端测试及静态检查、管理端构建和测试通过。数据库/S3 实际导入与线上部署尚未验收。 ## 1. 当前约定 diff --git a/monkeyai/design/resource-package-volumes-plan.md b/monkeyai/design/resource-package-volumes-plan.md new file mode 100644 index 000000000..a28d9cbc8 --- /dev/null +++ b/monkeyai/design/resource-package-volumes-plan.md @@ -0,0 +1,74 @@ +# 资源包分卷导入实现 + +状态：已实现并完成验证，尚未部署。分支 `feat-resource-import-volumes`。 + +## 范围 + +- 支持普通 ZIP、普通 ZIP 按字节切分的 `.zip.001`、`.zip.002` 等全部分卷，保留解压目录入口。 +- 不支持原生 `.z01 + .zip` 分卷或 7z 压缩格式。文件扩展名符合要求仍须通过完整 ZIP 检查。 +- 一组分卷始终作为发布者的一份完整 Agent 清单，统一预览、确认、导入；只产生一个导入批次。 +- 整包仍不超过 2 GiB、展开不超过 4 GiB、外层文件不超过 5000 个。分卷与网络传输块分别处理，最多 256 卷，每个传输块 64 MiB，末块可小于该大小。 +- 保留现有宽松元数据解析、同源更新、缺失资源停用、外部引用保护、发布者锁和事务补偿。 + +## 管理端 + +沿用「资源包导入记录」页面。导入弹窗分为选择文件、上传解析、确认导入三个步骤。 + +1. 拖入或多选全部分卷，按编号排序；支持补充、移除、清空。检查缺卷、重复编号、混包、空文件与总大小。 +2. 显示总体及每卷上传进度。暂停、网络中断或刷新后保留服务端进度，续传时重新选择原文件并核对已上传块的实际摘要，只传缺失部分。 +3. 完成完整性检查后自动预览；展示各类变更汇总，可按资源类型和动作筛选。明确列出停用项并要求确认完整清单。 +4. 认证归属变化或并发冲突后要求重新预览，复用已上传内容；确认阶段只提交 `upload_id`、选项、包摘要和预览摘要。 +5. 列表顶部显示当前管理员的未完成任务，支持继续上传、继续确认、放弃；保留 24 小时，完成后移出未完成列表并刷新导入记录。 + +## 服务端 + +新增迁移 `000027_resource_import_uploads`，包含上传会话和传输块记录。 + +- 会话绑定创建管理员，每人最多保留 5 个未过期任务；其他管理员不可读取或修改。 +- 传输块保存到已有对象存储，服务端检查实际大小与 SHA-256。同内容重试幂等，不同内容返回冲突。 +- 上传、封存、预览、确认和取消通过会话行锁协调，避免封存后改写或导入期间删除。 +- `complete` 检查传输块齐全、实际摘要、ZIP 路径及展开限制、外层文件 CRC，计算完整 ZIP 的服务端摘要，然后封存为 `ready`。 +- 从对象存储逐块下载至请求级临时文件，以跨文件 `io.ReaderAt` 提供统一 ZIP 读取，不在内存或磁盘生成额外的合并 ZIP。请求结束关闭并清理临时文件。 +- 生产 S3 上传先落临时文件并流式写入对象存储；整包读取也使用磁盘暂存，仍受现有资源正文内存限制约束。 +- 复用 `ParseReader → ParseResources → Evaluate → Apply`，保留旧 multipart 接口，同时接受 JSON `upload_id` 请求。 +- 任务有效期 24 小时，后台每分钟回收过期及取消任务的暂存对象和数据库记录；多实例使用 `SKIP LOCKED` 协调清理。对象删除失败保留记录等待重试。 +- Nginx 新增上传会话路由，允许 64 MiB 二进制请求，关闭请求缓冲并使用 30 分钟代理超时。 + +## 接口 + +| 接口 | 作用 | +| --- | --- | +| `POST /api/admin/v1/resource-import-uploads` | 创建会话并声明按序排列的本地文件及大小 | +| `GET /api/admin/v1/resource-import-uploads` | 当前管理员未完成任务 | +| `GET /api/admin/v1/resource-import-uploads/{id}` | 查询已上传块及任务状态 | +| `PUT /api/admin/v1/resource-import-uploads/{id}/chunks/{index}` | 从零编号，上传逻辑 ZIP 对应的传输块 | +| `POST /api/admin/v1/resource-import-uploads/{id}/complete` | 完整性检查、计算包摘要并封存 | +| `DELETE /api/admin/v1/resource-import-uploads/{id}` | 放弃未完成任务 | +| `POST /api/admin/v1/resource-imports/preview` | 使用 upload_id 统一预览，兼容原 multipart | +| `POST /api/admin/v1/resource-imports` | 使用同一 upload_id 确认导入，兼容原 multipart | + +## 外部 Expert 引用冲突 + +报错「资源 research 仍被其他 Expert … 引用」来自完整清单同步的停用保护：旧绑定中的资源不在本次解析结果里，但仍有启用且未删除的 Expert 使用它，而该 Expert 不属于本发布者的导入绑定。 + +可能原因包括新包移除或改名资源、只上传了部分资源包。处理方式为在完整包中保留资源，或解除外部 Expert 引用。本次保留该保护，错误提示增加资源类型、Expert 名称、ID 和处理建议。 + +同时规范化缺省和空认证归属选项，避免恢复任务时预览请求省略字段、确认请求传空对象导致错误的摘要冲突。 + +## 验证 + +- [x] 后端分卷文件名、顺序、路径、数量及大小限制测试。 +- [x] 跨文件 ReaderAt 边界、偏移和 EOF 测试。 +- [x] PostgreSQL 集成：缺块拦截、重复与冲突上传、管理员隔离、重启恢复进度、封存不可修改、整包预览与单批次导入、取消与过期清理。 +- [x] PostgreSQL 集成复现外部 Expert 阻止资源停用，解除有效引用后允许预览停用。 +- [x] 前端分卷排序、缺卷/重卷/混包、跨卷传输块、各卷进度、内容核对、续传仅上传缺块和暂停测试。 +- [x] 浏览器连接本地 PostgreSQL 与 RustFS：乱序选卷、缺卷补齐、预览、刷新后继续确认、成功导入记录；桌面及 390 像素窄屏布局。 +- [x] 65 MiB 真实 ZIP 续传：预先保存 64 MiB 首块，页面恢复后仅请求末块 `/chunks/1`，随后成功预览及导入。 +- [x] 浏览器复现 research 外部 Expert 冲突，解除引用后可重新预览新增和停用项。 +- [x] 完整 Go 测试（启用 PostgreSQL/S3 集成）与 `go vet ./...`。 +- [x] 管理端构建、全部 225 项测试及修改源文件 ESLint。 +- [x] OpenAPI YAML、SQLC/schema/迁移一致性、Nginx 配置语法和 `git diff --check`。 + +全量回归时，将计费集成测试中的旧统计地址 `/statistics/tasks` 更新为现有 `/statistics/tasks/overview`，未修改统计业务行为。 + +尚未部署，也未自动修改已有数据库中的 Expert 引用。