119 lines
7.0 KiB
TypeScript
119 lines
7.0 KiB
TypeScript
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<void> };
|
||
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(); }
|
||
}
|