import "server-only"; import { createReadStream } from "node:fs"; import { open } from "node:fs/promises"; import { AbortMultipartUploadCommand, CompleteMultipartUploadCommand, CreateMultipartUploadCommand, HeadObjectCommand, PutObjectCommand, UploadPartCommand, type S3Client } from "@aws-sdk/client-s3"; import { getDefaultStorageConfig, getStorageConfigById } from "@/lib/db/queries"; import { createS3Client, toDbS3UploadConfig, buildS3PublicUrl } from "./service"; type ZipPublication = { id: string; path: string; bytes: number; disposition: string; onProgress?: (bytes: number) => Promise }; const PART_BYTES = 8 * 1024 ** 2; async function uploadZip(client: S3Client, bucket: string, key: string, params: ZipPublication) { const metadata = { Bucket: bucket, Key: key, ContentType: "application/zip", ContentDisposition: params.disposition, CacheControl: "public, max-age=1800" }; if (params.bytes <= 2 * PART_BYTES) { const body = createReadStream(params.path); try { await client.send(new PutObjectCommand({ ...metadata, Body: body, ContentLength: params.bytes }), { abortSignal: AbortSignal.timeout(90_000) }); await params.onProgress?.(params.bytes); } finally { body.destroy(); } return; } const initiated = await client.send(new CreateMultipartUploadCommand(metadata), { abortSignal: AbortSignal.timeout(15_000) }); if (!initiated.UploadId) throw new Error("七牛未返回分片上传编号"); const upload = { Bucket: bucket, Key: key, UploadId: initiated.UploadId }; const handle = await open(params.path, "r").catch(async error => { await client.send(new AbortMultipartUploadCommand(upload), { abortSignal: AbortSignal.timeout(10_000) }).catch(() => undefined); throw error; }); const parts: Array<{ PartNumber: number; ETag: string }> = []; let cursor = 0; let uploaded = 0; let progress = Promise.resolve(); const controller = new AbortController(); try { const workers = await Promise.allSettled(Array.from({ length: 3 }, async () => { try { while (!controller.signal.aborted) { const index = cursor++; const start = index * PART_BYTES; if (start >= params.bytes) return; const body = Buffer.allocUnsafe(Math.min(PART_BYTES, params.bytes - start)); let read = 0; while (read < body.length) { const result = await handle.read(body, read, body.length - read, start + read); if (!result.bytesRead) throw new Error("完整 ZIP 文件读取中断,请重试下载"); read += result.bytesRead; } const part = await client.send(new UploadPartCommand({ ...upload, PartNumber: index + 1, Body: body, ContentLength: body.length }), { abortSignal: AbortSignal.any([controller.signal, AbortSignal.timeout(180_000)]), }); if (!part.ETag) throw new Error("七牛未确认上传分片"); parts.push({ PartNumber: index + 1, ETag: part.ETag }); uploaded += body.length; const bytes = uploaded; progress = progress.then(() => params.onProgress?.(bytes)); await progress; } } catch (error) { controller.abort(); throw error; } })); const failure = workers.find(worker => worker.status === "rejected"); if (failure?.status === "rejected") throw failure.reason; await client.send(new CompleteMultipartUploadCommand({ ...upload, MultipartUpload: { Parts: parts.sort((a, b) => a.PartNumber - b.PartNumber) } }), { abortSignal: AbortSignal.timeout(30_000) }); } catch (error) { await client.send(new AbortMultipartUploadCommand(upload), { abortSignal: AbortSignal.timeout(10_000) }).catch(() => undefined); throw error; } finally { await handle.close(); } } /** Publish the complete ZIP before offering the accelerated download. */ export async function publishCanvasZip(params: ZipPublication) { const raw = await getDefaultStorageConfig(); const config = toDbS3UploadConfig(raw); if (config?.provider !== "qiniu-kodo" || !config.publicUrl?.startsWith("https://")) return null; const key = `canvas-downloads/${params.id}.zip`; const client = createS3Client(config); // Qiniu's S3 endpoint expects an ordinary known-length body, without AWS trailers. client.config.requestChecksumCalculation = async () => "WHEN_REQUIRED"; let phase: "check" | "upload" | "verify" = "check"; const startedAt = Date.now(); let uploadedBytes = 0; try { const existing = await client.send(new HeadObjectCommand({ Bucket: config.bucket, Key: key }), { abortSignal: AbortSignal.timeout(15_000) }).catch(error => { if (error.$metadata?.httpStatusCode === 404) return null; throw error; }); if (existing?.ContentLength !== params.bytes) { phase = "upload"; await uploadZip(client, config.bucket, key, { ...params, onProgress: async bytes => { uploadedBytes = bytes; await params.onProgress?.(bytes); } }); } phase = "verify"; const target = new URL(buildS3PublicUrl(config, key)); target.searchParams.set("v", String(params.bytes)); const url = target.toString(); const response = await fetch(url, { method: "HEAD", redirect: "error", signal: AbortSignal.timeout(10_000) }); const length = Number(response.headers.get("content-length")); if (!response.ok || length !== params.bytes) throw new Error("CDN object is not ready"); return { url, storageConfigId: raw!.id, key }; } catch (error) { const kind = (error as Error).name; const status = (error as { $metadata?: { httpStatusCode?: number } }).$metadata?.httpStatusCode; console.warn("[CanvasDownloadZipTask] CDN publication failed", { taskId: params.id, phase, kind, status, bytes: params.bytes, uploadedBytes, elapsedMs: Date.now() - startedAt }); const reason = status === 401 || status === 403 ? `七牛拒绝访问(HTTP ${status}),请管理员检查存储权限` : phase === "check" ? "暂时无法确认七牛文件状态" : phase === "verify" ? "七牛下载链接尚未可用" : /AbortError|TimeoutError/.test(kind) ? "传送至七牛时超时" : "传送至七牛失败"; const advice = status === 401 || status === 403 ? "存储权限恢复后再次点击下载即可继续传送原 ZIP" : "请再次点击下载以继续传送原 ZIP"; return { url: null, storageConfigId: raw!.id, key, error: `ZIP 已打包完成,但${reason}。${advice};不会重新生成图片。若持续失败,请联系管理员并提供下载编号:${params.id}` }; } finally { client.destroy(); } } export async function removeCanvasZipObject(object: { storageConfigId: string; key: string }) { if (!/^canvas-downloads\/[0-9a-f-]{36}\.zip$/i.test(object.key)) return; const config = toDbS3UploadConfig(await getStorageConfigById({ id: object.storageConfigId })); if (!config || config.provider !== "qiniu-kodo") return; const { DeleteObjectCommand } = await import("@aws-sdk/client-s3"); const client = createS3Client(config); try { await client.send(new DeleteObjectCommand({ Bucket: config.bucket, Key: object.key }), { abortSignal: AbortSignal.timeout(5000) }); } finally { client.destroy(); } }