ai-chatbot/lib/storage/service.ts
2026-05-27 21:37:21 +08:00

645 lines
19 KiB
TypeScript
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

import "server-only";
import { DeleteObjectCommand, PutObjectCommand, S3Client } from "@aws-sdk/client-s3";
import { put } from "@vercel/blob";
import {
getDefaultStorageConfig,
getStorageConfigById,
getSystemSetting,
} from "@/lib/db/queries";
import {
enqueueStorageMigrationTask,
getPreferredAliyunStorageConfig,
isAliyunStorageConfig,
} from "@/lib/storage/migration-db";
import {
defaultOssImageMigrationConfig,
normalizeOssImageMigrationConfig,
OSS_MIGRATION_CONFIG_KEY,
type OssImageMigrationConfig,
} from "@/lib/storage/migration-config";
import type { StorageConfig } from "@/lib/db/schema";
import { createLogger } from "@/lib/observability/logger";
const logger = createLogger("storage");
export type S3UploadConfig = {
endpoint: string;
region: string;
bucket: string;
accessKeyId: string;
secretAccessKey: string;
publicUrl?: string;
provider?: "s3" | "qiniu-kodo";
};
export type UploadedFileResult = {
url: string;
pathname: string;
contentType: string;
};
export type StorageUploadFunction = (
filename: string,
buffer: ArrayBuffer
) => Promise<UploadedFileResult>;
let lastKnownGoodDbS3Config: S3UploadConfig | null = null;
const STORAGE_FAILOVER_CONFIG_KEY = "storage_failover_config";
function normalizeEndpoint(endpoint: string): string {
const trimmed = endpoint.trim().replace(/\/+$/, "");
if (!trimmed) return "";
// 阿里云 OSS 统一强制 HTTPS,避免误填 http 或带空格导致连接失败
if (/aliyuncs\.com/i.test(trimmed)) {
const withoutProtocol = trimmed.replace(/^https?:\/\//i, "");
return `https://${withoutProtocol}`;
}
return trimmed;
}
function normalizePublicUrl(publicUrl?: string): string | undefined {
if (!publicUrl) return undefined;
const trimmed = publicUrl.trim().replace(/\/+$/, "");
return trimmed || undefined;
}
// 获取文件的 Content-Type
export function getContentType(filename: string): string {
const ext = filename.toLowerCase().split(".").pop();
const types: Record<string, string> = {
jpg: "image/jpeg",
jpeg: "image/jpeg",
png: "image/png",
gif: "image/gif",
webp: "image/webp",
svg: "image/svg+xml",
psd: "image/vnd.adobe.photoshop",
psb: "application/octet-stream",
pdf: "application/pdf",
json: "application/json",
txt: "text/plain",
html: "text/html",
css: "text/css",
js: "application/javascript",
ttf: "font/ttf",
otf: "font/otf",
woff: "font/woff",
woff2: "font/woff2",
glb: "model/gltf-binary",
gltf: "model/gltf+json",
usdz: "model/vnd.usdz+zip",
fbx: "application/octet-stream",
obj: "text/plain",
stl: "model/stl",
"3mf": "model/3mf",
ply: "application/octet-stream",
spz: "application/octet-stream",
splat: "application/octet-stream",
ksplat: "application/octet-stream",
zip: "application/zip",
};
return types[ext ?? ""] ?? "application/octet-stream";
}
export function createS3Client(config: S3UploadConfig): S3Client {
const normalizedEndpoint = normalizeEndpoint(config.endpoint);
const lowerEndpoint = normalizedEndpoint.toLowerCase();
const isCloudS3 =
lowerEndpoint.includes("aliyuncs.com") ||
lowerEndpoint.includes("amazonaws.com");
return new S3Client({
endpoint: normalizedEndpoint,
region: config.region || "auto",
credentials: {
accessKeyId: config.accessKeyId,
secretAccessKey: config.secretAccessKey,
},
forcePathStyle: config.provider === "qiniu-kodo" ? true : !isCloudS3,
});
}
export function buildStorageObjectKey(filename: string, prefix = "uploads"): string {
const timestamp = Date.now();
const randomString = Math.random().toString(36).slice(2, 8);
const cleanExt = String(filename.split(".").pop() || "bin")
.trim()
.toLowerCase()
.replace(/[^a-z0-9]+/g, "")
.slice(0, 16) || "bin";
const cleanPrefix = prefix.replace(/^\/+|\/+$/g, "") || "uploads";
return `${cleanPrefix}/${timestamp}-${randomString}.${cleanExt}`;
}
export function buildS3PublicUrl(config: S3UploadConfig, key: string): string {
const normalizedEndpoint = normalizeEndpoint(config.endpoint);
const normalizedPublicUrl = normalizePublicUrl(config.publicUrl);
const cleanKey = key.replace(/^\/+/, "");
if (normalizedPublicUrl) {
return `${normalizedPublicUrl}/${cleanKey}`;
}
return `${normalizedEndpoint}/${config.bucket}/${cleanKey}`;
}
export async function resolveS3UploadTarget(): Promise<
| {
config: S3UploadConfig;
storageConfig?: StorageConfig;
migrationConfig: OssImageMigrationConfig;
}
| null
> {
let migrationConfig = defaultOssImageMigrationConfig;
const migrationSetting = await getSystemSetting({
key: OSS_MIGRATION_CONFIG_KEY,
}).catch(() => undefined);
migrationConfig = normalizeOssImageMigrationConfig(migrationSetting?.value);
const selectedStorageConfig = await resolvePreferredUploadStorageConfig(
migrationConfig
);
const dbS3Config = toDbS3UploadConfig(selectedStorageConfig);
if (dbS3Config) {
return {
config: dbS3Config,
storageConfig: selectedStorageConfig,
migrationConfig,
};
}
const envS3Config = buildEnvS3Config();
if (envS3Config) {
return {
config: envS3Config,
migrationConfig,
};
}
return null;
}
export async function maybeEnqueueStorageMigrationTask(params: {
uploaded: { url: string; pathname: string };
storageConfig?: StorageConfig;
migrationConfig: OssImageMigrationConfig;
}) {
await maybeEnqueueAliyunMigrationTask(params);
}
// S3/MinIO 上传函数(使用 AWS SDK)
async function uploadToS3(
filename: string,
buffer: ArrayBuffer,
config: S3UploadConfig,
s3Client = createS3Client(config)
) {
const {
endpoint,
bucket,
publicUrl,
} = config;
const normalizedEndpoint = normalizeEndpoint(endpoint);
const normalizedPublicUrl = normalizePublicUrl(publicUrl);
// 生成唯一的文件名
const key = buildStorageObjectKey(filename);
const contentType = getContentType(filename);
try {
logger.debug(`[S3 Upload] Uploading to: ${normalizedEndpoint}/${bucket}/${key}`);
// 上传文件
const command = new PutObjectCommand({
Bucket: bucket,
Key: key,
Body: Buffer.from(buffer),
ContentType: contentType,
// 设置为公开可读(如果您的 MinIO 桶支持)
ACL: "public-read",
});
await s3Client.send(command);
logger.debug(`[S3 Upload] Upload successful: ${key}`);
// 返回文件的公开 URL
// MinIO 的公开 URL 格式通常是: http://endpoint:port/bucket/key
const fileUrl = buildS3PublicUrl(
{ ...config, endpoint: normalizedEndpoint, publicUrl: normalizedPublicUrl },
key
);
logger.debug(`[S3 Upload] File URL: ${fileUrl}`);
return {
url: fileUrl,
pathname: key,
contentType,
};
} catch (error) {
logger.error("[S3 Upload] Error:", error);
throw error;
}
}
export async function uploadFileWithS3Config(
filename: string,
buffer: ArrayBuffer,
config: S3UploadConfig
) {
return uploadToS3(filename, buffer, config);
}
export async function deleteS3ObjectByPath(pathname: string, config: S3UploadConfig) {
const key = String(pathname || "").replace(/^\/+/, "");
if (!key) return;
const s3Client = createS3Client(config);
await s3Client.send(
new DeleteObjectCommand({
Bucket: config.bucket,
Key: key,
})
);
}
export function toDbS3UploadConfig(raw: unknown): S3UploadConfig | null {
const value = raw as Record<string, unknown> | null | undefined;
const provider =
value?.type === "qiniu-kodo" ? "qiniu-kodo" : value?.type === "s3" ? "s3" : null;
const endpoint = value?.endpoint ? normalizeEndpoint(String(value.endpoint)) : "";
const bucket = value?.bucket ? String(value.bucket).trim() : "";
const accessKeyId = value?.accessKeyId ? String(value.accessKeyId).trim() : "";
const secretAccessKey = value?.secretAccessKey
? String(value.secretAccessKey).trim()
: "";
if (
!value ||
!provider ||
!value.isEnabled ||
!endpoint ||
!bucket ||
!accessKeyId ||
!secretAccessKey
) {
return null;
}
return {
endpoint,
region: String(value.region || "auto").trim() || "auto",
bucket,
accessKeyId,
secretAccessKey,
publicUrl: normalizePublicUrl(value.publicUrl ? String(value.publicUrl) : undefined),
provider,
};
}
function buildEnvS3Config(): S3UploadConfig | null {
const endpoint = process.env.S3_ENDPOINT
? normalizeEndpoint(process.env.S3_ENDPOINT)
: "";
const bucket = process.env.S3_BUCKET?.trim() ?? "";
const accessKeyId = process.env.S3_ACCESS_KEY_ID?.trim() ?? "";
const secretAccessKey = process.env.S3_SECRET_ACCESS_KEY?.trim() ?? "";
if (endpoint && bucket && accessKeyId && secretAccessKey) {
return {
endpoint,
region: process.env.S3_REGION?.trim() || "auto",
bucket,
accessKeyId,
secretAccessKey,
publicUrl: normalizePublicUrl(process.env.S3_PUBLIC_URL),
};
}
return null;
}
function normalizeStorageFailoverConfig(value: unknown): {
enabled: boolean;
backupStorageConfigId: string;
} {
const raw =
value && typeof value === "object" && !Array.isArray(value)
? (value as Record<string, unknown>)
: {};
return {
enabled: raw.enabled === true,
backupStorageConfigId: String(raw.backupStorageConfigId || "").trim(),
};
}
async function resolvePreferredUploadStorageConfig(
migrationConfig: OssImageMigrationConfig
): Promise<StorageConfig | undefined> {
if (!migrationConfig.enabled || !migrationConfig.preferAliyunForUploads) {
return await getDefaultStorageConfig();
}
const preferredAliyunConfig = await getPreferredAliyunStorageConfig({
preferredId: migrationConfig.sourceStorageConfigId || undefined,
});
if (preferredAliyunConfig) {
return preferredAliyunConfig;
}
return await getDefaultStorageConfig();
}
async function maybeEnqueueAliyunMigrationTask(params: {
uploaded: { url: string; pathname: string };
storageConfig?: StorageConfig;
migrationConfig: OssImageMigrationConfig;
}) {
const { uploaded, storageConfig, migrationConfig } = params;
if (!migrationConfig.enabled) return;
if (!storageConfig || !isAliyunStorageConfig(storageConfig)) return;
if (!uploaded.url || !uploaded.pathname) return;
const targetStorageConfig =
(migrationConfig.targetStorageConfigId
? await getStorageConfigById({ id: migrationConfig.targetStorageConfigId }).catch(() => undefined)
: undefined) || (await getDefaultStorageConfig().catch(() => undefined));
if (!targetStorageConfig) return;
if (targetStorageConfig.id === storageConfig.id) return;
if (isAliyunStorageConfig(targetStorageConfig)) return;
if (!toDbS3UploadConfig(targetStorageConfig)) return;
const now = Date.now();
const migrateAfterAt = new Date(
now + migrationConfig.migrateAfterHours * 60 * 60 * 1000
);
const cleanupAfterAt = new Date(
now + migrationConfig.cleanupAfterHours * 60 * 60 * 1000
);
await enqueueStorageMigrationTask({
sourceUrl: uploaded.url,
sourcePath: uploaded.pathname,
sourceStorageConfigId: storageConfig.id,
migrateAfterAt,
cleanupAfterAt,
});
}
async function uploadWithBackupStorageConfig(
filename: string,
buffer: ArrayBuffer,
primaryStorageConfigId?: string
): Promise<
| {
uploaded: { url: string; pathname: string; contentType: string };
storageConfig?: StorageConfig;
}
| null
> {
try {
const setting = await getSystemSetting({ key: STORAGE_FAILOVER_CONFIG_KEY });
const failover = normalizeStorageFailoverConfig(setting?.value);
if (!failover.enabled || !failover.backupStorageConfigId) {
return null;
}
if (
primaryStorageConfigId &&
failover.backupStorageConfigId === primaryStorageConfigId
) {
logger.warn(
"[Upload] Backup storage config equals default storage config, skipping failover retry"
);
return null;
}
const backupStorageConfig = await getStorageConfigById({
id: failover.backupStorageConfigId,
});
const backupS3Config = toDbS3UploadConfig(backupStorageConfig);
if (!backupS3Config) {
logger.warn(
"[Upload] Backup storage config is invalid, disabled, or not S3 compatible"
);
return null;
}
logger.debug(
`[Upload] Retrying with backup S3 config: ${backupStorageConfig?.name ?? failover.backupStorageConfigId}`
);
const uploaded = await uploadToS3(filename, buffer, backupS3Config);
lastKnownGoodDbS3Config = backupS3Config;
return {
uploaded,
storageConfig: backupStorageConfig,
};
} catch (error) {
logger.error("[Upload] Backup storage config retry failed:", error);
return null;
}
}
// 上传文件到存储服务
export async function uploadFile(
filename: string,
buffer: ArrayBuffer
): Promise<UploadedFileResult> {
logger.debug(`[Upload] Starting upload for file: ${filename}`);
let dbS3Config: S3UploadConfig | null = null;
let primaryStorageConfigId: string | undefined;
let selectedStorageConfig: StorageConfig | undefined;
let migrationConfig = defaultOssImageMigrationConfig;
try {
const migrationSetting = await getSystemSetting({
key: OSS_MIGRATION_CONFIG_KEY,
}).catch(() => undefined);
migrationConfig = normalizeOssImageMigrationConfig(migrationSetting?.value);
selectedStorageConfig = await resolvePreferredUploadStorageConfig(migrationConfig);
primaryStorageConfigId = selectedStorageConfig?.id;
dbS3Config = toDbS3UploadConfig(selectedStorageConfig);
if (dbS3Config) {
logger.debug(`[Upload] Using database S3 config: ${selectedStorageConfig?.name}`);
logger.debug(`[Upload] Endpoint: ${dbS3Config.endpoint}`);
logger.debug(`[Upload] Bucket: ${dbS3Config.bucket}`);
const uploaded = await uploadToS3(filename, buffer, dbS3Config);
lastKnownGoodDbS3Config = dbS3Config;
await maybeEnqueueAliyunMigrationTask({
uploaded,
storageConfig: selectedStorageConfig,
migrationConfig,
}).catch((error) => {
logger.error("[Upload] Failed to enqueue OSS migration task:", error);
});
return uploaded;
}
logger.warn("[Upload] No valid S3 config in database");
} catch (error) {
logger.error("[Upload] Database storage path failed:", error);
const uploadedFromBackup = await uploadWithBackupStorageConfig(
filename,
buffer,
primaryStorageConfigId
);
if (uploadedFromBackup) {
await maybeEnqueueAliyunMigrationTask({
uploaded: uploadedFromBackup.uploaded,
storageConfig: uploadedFromBackup.storageConfig,
migrationConfig,
}).catch((enqueueError) => {
logger.error("[Upload] Failed to enqueue backup OSS migration task:", enqueueError);
});
return uploadedFromBackup.uploaded;
}
if (lastKnownGoodDbS3Config) {
try {
logger.debug("[Upload] Retrying with last known good database S3 config");
return await uploadToS3(filename, buffer, lastKnownGoodDbS3Config);
} catch (retryError) {
logger.error("[Upload] Last known S3 config upload failed:", retryError);
}
}
}
if (!dbS3Config) {
const uploadedFromBackup = await uploadWithBackupStorageConfig(
filename,
buffer,
primaryStorageConfigId
);
if (uploadedFromBackup) {
await maybeEnqueueAliyunMigrationTask({
uploaded: uploadedFromBackup.uploaded,
storageConfig: uploadedFromBackup.storageConfig,
migrationConfig,
}).catch((enqueueError) => {
logger.error("[Upload] Failed to enqueue backup OSS migration task:", enqueueError);
});
return uploadedFromBackup.uploaded;
}
}
const envS3Config = buildEnvS3Config();
if (envS3Config) {
logger.debug("[Upload] Using environment variable S3 config");
return await uploadToS3(filename, buffer, envS3Config);
}
if (!process.env.BLOB_READ_WRITE_TOKEN) {
throw new Error(
"No available storage backend: database S3 failed, env S3 missing, and BLOB_READ_WRITE_TOKEN is not configured."
);
}
// 最终回退到 Vercel Blob
logger.debug("[Upload] Falling back to Vercel Blob");
const data = await put(filename, buffer, {
access: "public",
});
return {
url: data.url,
pathname: data.pathname,
contentType: getContentType(filename),
};
}
export async function createStorageUploadSession(): Promise<StorageUploadFunction> {
try {
let migrationConfig = defaultOssImageMigrationConfig;
const migrationSetting = await getSystemSetting({
key: OSS_MIGRATION_CONFIG_KEY,
}).catch(() => undefined);
migrationConfig = normalizeOssImageMigrationConfig(migrationSetting?.value);
const selectedStorageConfig = await resolvePreferredUploadStorageConfig(migrationConfig);
const primaryStorageConfigId = selectedStorageConfig?.id;
const dbS3Config = toDbS3UploadConfig(selectedStorageConfig);
if (!dbS3Config) {
return uploadFile;
}
const s3Client = createS3Client(dbS3Config);
logger.debug(`[Upload] Created reusable upload session: ${selectedStorageConfig?.name}`);
logger.debug(`[Upload] Endpoint: ${dbS3Config.endpoint}`);
logger.debug(`[Upload] Bucket: ${dbS3Config.bucket}`);
return async (filename: string, buffer: ArrayBuffer) => {
logger.debug(`[Upload] Starting session upload for file: ${filename}`);
try {
const uploaded = await uploadToS3(filename, buffer, dbS3Config, s3Client);
lastKnownGoodDbS3Config = dbS3Config;
await maybeEnqueueAliyunMigrationTask({
uploaded,
storageConfig: selectedStorageConfig,
migrationConfig,
}).catch((error) => {
logger.error("[Upload] Failed to enqueue OSS migration task:", error);
});
return uploaded;
} catch (error) {
logger.error("[Upload] Reusable storage session failed:", error);
const uploadedFromBackup = await uploadWithBackupStorageConfig(
filename,
buffer,
primaryStorageConfigId
);
if (uploadedFromBackup) {
await maybeEnqueueAliyunMigrationTask({
uploaded: uploadedFromBackup.uploaded,
storageConfig: uploadedFromBackup.storageConfig,
migrationConfig,
}).catch((migrationError) => {
logger.error("[Upload] Failed to enqueue backup OSS migration task:", migrationError);
});
return uploadedFromBackup.uploaded;
}
throw error;
}
};
} catch (error) {
logger.error("[Upload] Failed to create reusable upload session:", error);
return uploadFile;
}
}
// 获取当前使用的存储类型
export async function getStorageType(): Promise<
"database-s3" | "env-s3" | "vercel-blob"
> {
try {
const storageConfig = await getDefaultStorageConfig();
if (
storageConfig &&
(storageConfig.type === "s3" || storageConfig.type === "qiniu-kodo") &&
storageConfig.endpoint &&
storageConfig.bucket
) {
return "database-s3";
}
} catch (error) {
// ignore
}
if (process.env.S3_ENDPOINT && process.env.S3_BUCKET) {
return "env-s3";
}
return "vercel-blob";
}