diff --git a/app/(admin)/settings/admin-settings-client.tsx b/app/(admin)/settings/admin-settings-client.tsx index b642ca574..58d6e7bcd 100644 --- a/app/(admin)/settings/admin-settings-client.tsx +++ b/app/(admin)/settings/admin-settings-client.tsx @@ -235,6 +235,7 @@ type SiteSettings = { lovartPromptTagVisibility: LovartPromptTagVisibility; contactConfig: ContactConfig; storageFailoverConfig: StorageFailoverConfig; + ossImageMigrationConfig: OssImageMigrationConfig; pointsSystemConfig: PointsSystemConfig; pointsPackages: PointsPackage[]; dulupayConfig: DulupayConfig; @@ -267,6 +268,15 @@ type StorageFailoverConfig = { backupStorageConfigId: string; }; +type OssImageMigrationConfig = { + enabled: boolean; + sourceStorageConfigId: string; + targetStorageConfigId: string; + migrateAfterHours: number; + cleanupAfterHours: number; + batchSize: number; +}; + const defaultLovartPromptTagVisibility: LovartPromptTagVisibility = { nanoBananaPro: true, design: true, @@ -286,6 +296,15 @@ const defaultStorageFailoverConfig: StorageFailoverConfig = { backupStorageConfigId: "", }; +const defaultOssImageMigrationConfig: OssImageMigrationConfig = { + enabled: true, + sourceStorageConfigId: "", + targetStorageConfigId: "", + migrateAfterHours: 48, + cleanupAfterHours: 72, + batchSize: 20, +}; + const lovartPromptTagOptions: Array<{ key: keyof LovartPromptTagVisibility; label: string; @@ -582,6 +601,7 @@ export function AdminSettingsClient() { lovartPromptTagVisibility: defaultLovartPromptTagVisibility, contactConfig: defaultContactConfig, storageFailoverConfig: defaultStorageFailoverConfig, + ossImageMigrationConfig: defaultOssImageMigrationConfig, pointsSystemConfig: defaultPointsSystemConfig, pointsPackages: defaultPointsPackages, dulupayConfig: defaultDulupayConfig, @@ -675,6 +695,39 @@ export function AdminSettingsClient() { (siteSettingsData as any).storageFailoverConfig?.backupStorageConfigId || "" ).trim(), }, + ossImageMigrationConfig: { + enabled: (siteSettingsData as any).ossImageMigrationConfig?.enabled !== false, + sourceStorageConfigId: String( + (siteSettingsData as any).ossImageMigrationConfig?.sourceStorageConfigId || "" + ).trim(), + targetStorageConfigId: String( + (siteSettingsData as any).ossImageMigrationConfig?.targetStorageConfigId || "" + ).trim(), + migrateAfterHours: Math.max( + 1, + Math.min( + 720, + Number((siteSettingsData as any).ossImageMigrationConfig?.migrateAfterHours) || + defaultOssImageMigrationConfig.migrateAfterHours + ) + ), + cleanupAfterHours: Math.max( + 2, + Math.min( + 1440, + Number((siteSettingsData as any).ossImageMigrationConfig?.cleanupAfterHours) || + defaultOssImageMigrationConfig.cleanupAfterHours + ) + ), + batchSize: Math.max( + 1, + Math.min( + 100, + Number((siteSettingsData as any).ossImageMigrationConfig?.batchSize) || + defaultOssImageMigrationConfig.batchSize + ) + ), + }, pointsSystemConfig: normalizedPointsSystemConfig, pointsPackages: normalizedPointsPackages, dulupayConfig: normalizedDulupayConfig, @@ -1524,6 +1577,38 @@ export function AdminSettingsClient() { return; } + if (siteSettings.ossImageMigrationConfig.enabled) { + if ( + siteSettings.ossImageMigrationConfig.cleanupAfterHours <= + siteSettings.ossImageMigrationConfig.migrateAfterHours + ) { + toast.error("OSS 清理时间必须晚于转存时间"); + return; + } + + if ( + siteSettings.ossImageMigrationConfig.sourceStorageConfigId && + siteSettings.ossImageMigrationConfig.targetStorageConfigId && + siteSettings.ossImageMigrationConfig.sourceStorageConfigId === + siteSettings.ossImageMigrationConfig.targetStorageConfigId + ) { + toast.error("OSS 源桶和目标桶不能相同"); + return; + } + + const targetConfig = (storageData?.configs || []).find( + (config) => config.id === siteSettings.ossImageMigrationConfig.targetStorageConfigId + ); + if ( + targetConfig && + targetConfig.type === "s3" && + targetConfig.endpoint?.includes("aliyuncs.com") + ) { + toast.error("OSS 转存目标桶不能是阿里云 OSS"); + return; + } + } + const parsedMembershipConfig = normalizeMembershipConfig(siteSettings.membershipConfig); const parsedMembershipOverrides = normalizeMembershipUserOverrides( siteSettings.membershipUserOverrides @@ -7307,6 +7392,185 @@ export function AdminSettingsClient() { + + + + 阿里云 OSS 自动转存 + + 若已配置阿里云 OSS，上传优先进入 OSS；到期后自动转存到目标默认桶并替换数据库 URL。 + + + +
+
+ +

+ 开启后执行“先 OSS、后迁移、再清理”的生命周期策略（不依赖 CDN）。 +

+
+ + setSiteSettings((prev) => ({ + ...prev, + ossImageMigrationConfig: { + ...prev.ossImageMigrationConfig, + enabled: checked, + }, + })) + } + /> +
+ +
+
+ + +
+ +
+ + +
+
+ +
+
+ + + setSiteSettings((prev) => ({ + ...prev, + ossImageMigrationConfig: { + ...prev.ossImageMigrationConfig, + migrateAfterHours: Math.max( + 1, + Math.min(720, Number(e.target.value) || 48) + ), + }, + })) + } + /> +
+
+ + + setSiteSettings((prev) => ({ + ...prev, + ossImageMigrationConfig: { + ...prev.ossImageMigrationConfig, + cleanupAfterHours: Math.max( + 2, + Math.min(1440, Number(e.target.value) || 72) + ), + }, + })) + } + /> +
+
+ + + setSiteSettings((prev) => ({ + ...prev, + ossImageMigrationConfig: { + ...prev.ossImageMigrationConfig, + batchSize: Math.max( + 1, + Math.min(100, Number(e.target.value) || 20) + ), + }, + })) + } + /> +
+
+ +
+ 推荐值：转存 48 小时，清理 72 小时。并配置定时调用 + /api/jobs/migrate-oss-images + 执行迁移任务。 +
+ +
+ +
+ + {/* 本地AI设置 */} diff --git a/app/api/jobs/migrate-oss-images/route.ts b/app/api/jobs/migrate-oss-images/route.ts new file mode 100644 index 000000000..11e11de7a --- /dev/null +++ b/app/api/jobs/migrate-oss-images/route.ts @@ -0,0 +1,71 @@ +import { NextResponse } from "next/server"; +import { + getOssImageMigrationJobStats, + runOssImageMigrationJob, +} from "@/lib/jobs/oss-image-migration-job"; + +function verifyAuthToken(request: Request): boolean { + const authHeader = request.headers.get("authorization"); + const token = authHeader?.replace("Bearer ", ""); + const cronSecret = process.env.CRON_SECRET; + + if (!cronSecret) { + console.warn("[OSSMigrationAPI] ⚠️ CRON_SECRET 未配置，允许所有请求"); + return true; + } + + return token === cronSecret; +} + +export async function POST(request: Request) { + try { + if (!verifyAuthToken(request)) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + const result = await runOssImageMigrationJob(); + return NextResponse.json({ + success: result.success, + message: result.message, + migrate: result.migrate, + cleanup: result.cleanup, + skippedReason: result.skippedReason || null, + timestamp: new Date().toISOString(), + }); + } catch (error: any) { + console.error("[OSSMigrationAPI] ❌ Job execution failed:", error); + return NextResponse.json( + { + success: false, + error: error?.message || "Job failed", + timestamp: new Date().toISOString(), + }, + { status: 500 } + ); + } +} + +export async function GET(request: Request) { + try { + if (!verifyAuthToken(request)) { + return NextResponse.json({ error: "Unauthorized" }, { status: 401 }); + } + + const data = await getOssImageMigrationJobStats(); + return NextResponse.json({ + ...data, + timestamp: new Date().toISOString(), + }); + } catch (error: any) { + console.error("[OSSMigrationAPI] ❌ Query failed:", error); + return NextResponse.json( + { + success: false, + error: error?.message || "Query failed", + timestamp: new Date().toISOString(), + }, + { status: 500 } + ); + } +} + diff --git a/app/api/site-settings/route.ts b/app/api/site-settings/route.ts index 6ba0e8023..4684813e6 100644 --- a/app/api/site-settings/route.ts +++ b/app/api/site-settings/route.ts @@ -24,6 +24,11 @@ import { normalizeDulupayConfig, normalizeDulupayForPublic, } from "@/lib/payment/dulupay-config"; +import { + defaultOssImageMigrationConfig, + normalizeOssImageMigrationConfig, + OSS_MIGRATION_CONFIG_KEY, +} from "@/lib/storage/migration-config"; const HOME_VARIANT_KEY = "home_variant"; // Historically stored as "home_surprise_categories" but used for the featured gallery categories. @@ -173,6 +178,7 @@ export async function GET() { dulupayConfigSetting, realtimeCanvasConfigSetting, storageFailoverConfigSetting, + ossImageMigrationConfigSetting, ] = await Promise.all([ db.select().from(siteSettings).limit(1).then(rows => rows[0]), db @@ -265,6 +271,12 @@ export async function GET() { .where(eq(systemSettings.key, STORAGE_FAILOVER_CONFIG_KEY)) .limit(1) .then(rows => rows[0]), + db + .select() + .from(systemSettings) + .where(eq(systemSettings.key, OSS_MIGRATION_CONFIG_KEY)) + .limit(1) + .then(rows => rows[0]), ]); const notification = notificationSetting?.value || { enabled: false, title: "", content: "" }; @@ -297,6 +309,9 @@ export async function GET() { const storageFailoverConfig = normalizeStorageFailoverConfig( storageFailoverConfigSetting?.value ); + const ossImageMigrationConfig = normalizeOssImageMigrationConfig( + ossImageMigrationConfigSetting?.value + ); // 如果没有设置记录，返回默认值 const baseSettings = settings || { @@ -324,6 +339,7 @@ export async function GET() { realtimeCanvasEnabled: true, realtimeCanvasServiceEnabled: true, storageFailoverConfig, + ossImageMigrationConfig, canvasDefaultModelId: null, canvasAutoModeDefaultModelId: null, canvasDetailDefaultModelId: null, @@ -368,6 +384,7 @@ export async function GET() { dulupayConfig, realtimeCanvasServiceEnabled, storageFailoverConfig, + ossImageMigrationConfig, }); // 禁用缓存，确保总是获取最新设置 @@ -447,6 +464,16 @@ const updateSchema = z.object({ backupStorageConfigId: z.string().max(128).optional(), }) .optional(), + ossImageMigrationConfig: z + .object({ + enabled: z.boolean().optional(), + sourceStorageConfigId: z.string().max(128).optional(), + targetStorageConfigId: z.string().max(128).optional(), + migrateAfterHours: z.number().min(1).max(720).optional(), + cleanupAfterHours: z.number().min(1).max(1440).optional(), + batchSize: z.number().min(1).max(100).optional(), + }) + .optional(), }); // PUT /api/site-settings - 更新网站设置（仅管理员） @@ -478,6 +505,7 @@ export async function PUT(request: NextRequest) { pointsPackages, dulupayConfig, storageFailoverConfig, + ossImageMigrationConfig, ...siteSettingsPayload } = data; @@ -754,6 +782,26 @@ export async function PUT(request: NextRequest) { }); } + if (ossImageMigrationConfig !== undefined) { + const normalized = normalizeOssImageMigrationConfig(ossImageMigrationConfig); + await db + .insert(systemSettings) + .values({ + key: OSS_MIGRATION_CONFIG_KEY, + value: normalized, + updatedAt: new Date(), + updatedBy: session.user.id, + }) + .onConflictDoUpdate({ + target: systemSettings.key, + set: { + value: normalized, + updatedAt: new Date(), + updatedBy: session.user.id, + }, + }); + } + return NextResponse.json({ ...updated, homeVariant: homeVariant ?? normalizeHomeVariant(undefined), @@ -802,6 +850,10 @@ export async function PUT(request: NextRequest) { storageFailoverConfig !== undefined ? normalizeStorageFailoverConfig(storageFailoverConfig) : normalizeStorageFailoverConfig(undefined), + ossImageMigrationConfig: + ossImageMigrationConfig !== undefined + ? normalizeOssImageMigrationConfig(ossImageMigrationConfig) + : defaultOssImageMigrationConfig, }); } else { // 创建新设置 @@ -1101,6 +1153,26 @@ export async function PUT(request: NextRequest) { }); } + if (ossImageMigrationConfig !== undefined) { + const normalized = normalizeOssImageMigrationConfig(ossImageMigrationConfig); + await db + .insert(systemSettings) + .values({ + key: OSS_MIGRATION_CONFIG_KEY, + value: normalized, + updatedAt: new Date(), + updatedBy: session.user.id, + }) + .onConflictDoUpdate({ + target: systemSettings.key, + set: { + value: normalized, + updatedAt: new Date(), + updatedBy: session.user.id, + }, + }); + } + return NextResponse.json({ ...created, homeVariant: homeVariant ?? normalizeHomeVariant(undefined), @@ -1149,6 +1221,10 @@ export async function PUT(request: NextRequest) { storageFailoverConfig !== undefined ? normalizeStorageFailoverConfig(storageFailoverConfig) : normalizeStorageFailoverConfig(undefined), + ossImageMigrationConfig: + ossImageMigrationConfig !== undefined + ? normalizeOssImageMigrationConfig(ossImageMigrationConfig) + : defaultOssImageMigrationConfig, }); } } catch (error) { diff --git a/lib/db/migrations/0047_add_storage_migration_tasks.sql b/lib/db/migrations/0047_add_storage_migration_tasks.sql new file mode 100644 index 000000000..c450a6e18 --- /dev/null +++ b/lib/db/migrations/0047_add_storage_migration_tasks.sql @@ -0,0 +1,23 @@ +CREATE TABLE IF NOT EXISTS "StorageMigrationTask" ( + "id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL, + "sourceUrl" text NOT NULL UNIQUE, + "sourcePath" text NOT NULL, + "sourceStorageConfigId" uuid REFERENCES "StorageConfig"("id"), + "targetStorageConfigId" uuid REFERENCES "StorageConfig"("id"), + "targetUrl" text, + "status" varchar(32) NOT NULL DEFAULT 'pending', + "attemptCount" integer NOT NULL DEFAULT 0, + "lastError" text, + "migrateAfterAt" timestamp NOT NULL, + "cleanupAfterAt" timestamp NOT NULL, + "migratedAt" timestamp, + "cleanedAt" timestamp, + "createdAt" timestamp NOT NULL DEFAULT now(), + "updatedAt" timestamp NOT NULL DEFAULT now() +); + +CREATE INDEX IF NOT EXISTS "StorageMigrationTask_status_migrate_idx" + ON "StorageMigrationTask"("status", "migrateAfterAt"); + +CREATE INDEX IF NOT EXISTS "StorageMigrationTask_status_cleanup_idx" + ON "StorageMigrationTask"("status", "cleanupAfterAt"); diff --git a/lib/db/migrations/meta/_journal.json b/lib/db/migrations/meta/_journal.json index 9153f1a0e..45d3e72a5 100644 --- a/lib/db/migrations/meta/_journal.json +++ b/lib/db/migrations/meta/_journal.json @@ -232,6 +232,13 @@ "when": 1772517600000, "tag": "0046_add_canvas_detail_generation_mode", "breakpoints": true + }, + { + "idx": 33, + "version": "7", + "when": 1772604000000, + "tag": "0047_add_storage_migration_tasks", + "breakpoints": true } ] } diff --git a/lib/db/schema.ts b/lib/db/schema.ts index 3bb144449..eb58361c7 100644 --- a/lib/db/schema.ts +++ b/lib/db/schema.ts @@ -324,6 +324,27 @@ export const storageConfig = pgTable("StorageConfig", { export type StorageConfig = InferSelectModel; +// 阿里云 OSS 图片迁移任务表 +export const storageMigrationTask = pgTable("StorageMigrationTask", { + id: uuid("id").primaryKey().notNull().defaultRandom(), + sourceUrl: text("sourceUrl").notNull().unique(), + sourcePath: text("sourcePath").notNull(), + sourceStorageConfigId: uuid("sourceStorageConfigId").references(() => storageConfig.id), + targetStorageConfigId: uuid("targetStorageConfigId").references(() => storageConfig.id), + targetUrl: text("targetUrl"), + status: varchar("status", { length: 32 }).notNull().default("pending"), + attemptCount: integer("attemptCount").notNull().default(0), + lastError: text("lastError"), + migrateAfterAt: timestamp("migrateAfterAt").notNull(), + cleanupAfterAt: timestamp("cleanupAfterAt").notNull(), + migratedAt: timestamp("migratedAt"), + cleanedAt: timestamp("cleanedAt"), + createdAt: timestamp("createdAt").notNull().defaultNow(), + updatedAt: timestamp("updatedAt").notNull().defaultNow(), +}); + +export type StorageMigrationTask = InferSelectModel; + // 网站设置表 export const siteSettings = pgTable("SiteSettings", { id: uuid("id").primaryKey().notNull().defaultRandom(), diff --git a/lib/jobs/oss-image-migration-job.ts b/lib/jobs/oss-image-migration-job.ts new file mode 100644 index 000000000..95d992284 --- /dev/null +++ b/lib/jobs/oss-image-migration-job.ts @@ -0,0 +1,410 @@ +import "server-only"; + +import { and, eq, like, or, sql } from "drizzle-orm"; +import { db } from "@/lib/db"; +import { + canvasDetailArchive, + canvasProject, + generatedImage, + message, + messageDeprecated, +} from "@/lib/db/schema"; +import { getDefaultStorageConfig, getSystemSetting } from "@/lib/db/queries"; +import { + getStorageConfigByIdForMigration, + getStorageMigrationTaskStats, + getStorageTasksToCleanup, + getStorageTasksToMigrate, + isAliyunStorageConfig, + markStorageTaskCleaned, + markStorageTaskCleanupFailed, + markStorageTaskMigrated, + markStorageTaskMigrating, + markStorageTaskMigrationFailed, +} from "@/lib/storage/migration-db"; +import { + normalizeOssImageMigrationConfig, + OSS_MIGRATION_CONFIG_KEY, +} from "@/lib/storage/migration-config"; +import { + deleteS3ObjectByPath, + toDbS3UploadConfig, + uploadFileWithS3Config, +} from "@/lib/storage/service"; + +type JobResult = { + success: boolean; + message: string; + migrate: { total: number; succeeded: number; failed: number }; + cleanup: { total: number; succeeded: number; failed: number }; + skippedReason?: string; +}; + +function toArrayBuffer(buffer: Buffer): ArrayBuffer { + return buffer.buffer.slice( + buffer.byteOffset, + buffer.byteOffset + buffer.byteLength + ) as ArrayBuffer; +} + +function deriveFilenameFromPath(pathname: string): string { + const cleaned = String(pathname || "").split("?")[0].replace(/^\/+/, ""); + const tail = cleaned.split("/").pop() || ""; + return tail || `image-${Date.now()}.png`; +} + +async function downloadImageAsBuffer(url: string, retries = 2): Promise { + for (let attempt = 0; attempt <= retries; attempt += 1) { + try { + const response = await fetch(url, { + signal: AbortSignal.timeout(30_000), + headers: { + "User-Agent": "AI-Chatbot-OSS-Migrator/1.0", + }, + }); + + if (!response.ok) { + throw new Error(`HTTP ${response.status} ${response.statusText}`); + } + + const arrayBuffer = await response.arrayBuffer(); + if (!arrayBuffer || arrayBuffer.byteLength === 0) { + throw new Error("Downloaded file is empty"); + } + return Buffer.from(arrayBuffer); + } catch (error) { + if (attempt >= retries) { + throw error; + } + await new Promise((resolve) => setTimeout(resolve, 1000 * (attempt + 1))); + } + } + + throw new Error("Download failed"); +} + +function replaceDeepUrls( + value: unknown, + sourceUrl: string, + targetUrl: string +): { value: unknown; changed: boolean } { + if (typeof value === "string") { + if (!value.includes(sourceUrl)) { + return { value, changed: false }; + } + return { value: value.split(sourceUrl).join(targetUrl), changed: true }; + } + + if (Array.isArray(value)) { + let changed = false; + const next = value.map((item) => { + const result = replaceDeepUrls(item, sourceUrl, targetUrl); + if (result.changed) changed = true; + return result.value; + }); + return { value: changed ? next : value, changed }; + } + + if (value && typeof value === "object") { + let changed = false; + const next: Record = {}; + for (const [key, item] of Object.entries(value as Record)) { + const result = replaceDeepUrls(item, sourceUrl, targetUrl); + next[key] = result.value; + if (result.changed) changed = true; + } + return { + value: changed ? next : value, + changed, + }; + } + + return { value, changed: false }; +} + +async function replaceImageUrlInDatabase(sourceUrl: string, targetUrl: string) { + if (!sourceUrl || !targetUrl || sourceUrl === targetUrl) return; + + await db + .update(generatedImage) + .set({ thumbnailUrl: targetUrl }) + .where(eq(generatedImage.thumbnailUrl, sourceUrl)); + + await db + .update(generatedImage) + .set({ originalUrl: targetUrl }) + .where(eq(generatedImage.originalUrl, sourceUrl)); + + const messages = await db + .select({ + id: message.id, + parts: message.parts, + attachments: message.attachments, + }) + .from(message) + .where( + or( + sql`${message.parts}::text like ${`%${sourceUrl}%`}`, + sql`${message.attachments}::text like ${`%${sourceUrl}%`}` + ) + ) + .limit(500); + + for (const item of messages) { + const partsResult = replaceDeepUrls(item.parts, sourceUrl, targetUrl); + const attachmentsResult = replaceDeepUrls(item.attachments, sourceUrl, targetUrl); + + if (!partsResult.changed && !attachmentsResult.changed) continue; + + await db + .update(message) + .set({ + parts: (partsResult.value ?? item.parts) as any, + attachments: (attachmentsResult.value ?? item.attachments) as any, + }) + .where(eq(message.id, item.id)); + } + + const legacyMessages = await db + .select({ + id: messageDeprecated.id, + content: messageDeprecated.content, + }) + .from(messageDeprecated) + .where(sql`${messageDeprecated.content}::text like ${`%${sourceUrl}%`}`) + .limit(500); + + for (const item of legacyMessages) { + const replaced = replaceDeepUrls(item.content, sourceUrl, targetUrl); + if (!replaced.changed) continue; + await db + .update(messageDeprecated) + .set({ + content: replaced.value as any, + }) + .where(eq(messageDeprecated.id, item.id)); + } + + const projects = await db + .select({ + id: canvasProject.id, + canvasData: canvasProject.canvasData, + }) + .from(canvasProject) + .where(sql`${canvasProject.canvasData}::text like ${`%${sourceUrl}%`}`) + .limit(200); + + for (const item of projects) { + const replaced = replaceDeepUrls(item.canvasData, sourceUrl, targetUrl); + if (!replaced.changed) continue; + await db + .update(canvasProject) + .set({ + canvasData: replaced.value as any, + updatedAt: new Date(), + }) + .where(eq(canvasProject.id, item.id)); + } + + const detailArchives = await db + .select({ + id: canvasDetailArchive.id, + coverUrl: canvasDetailArchive.coverUrl, + brief: canvasDetailArchive.brief, + slots: canvasDetailArchive.slots, + }) + .from(canvasDetailArchive) + .where( + or( + like(canvasDetailArchive.coverUrl, `%${sourceUrl}%`), + sql`${canvasDetailArchive.brief}::text like ${`%${sourceUrl}%`}`, + sql`${canvasDetailArchive.slots}::text like ${`%${sourceUrl}%`}` + ) + ) + .limit(200); + + for (const item of detailArchives) { + const cover = replaceDeepUrls(item.coverUrl, sourceUrl, targetUrl); + const brief = replaceDeepUrls(item.brief, sourceUrl, targetUrl); + const slots = replaceDeepUrls(item.slots, sourceUrl, targetUrl); + + if (!cover.changed && !brief.changed && !slots.changed) continue; + + await db + .update(canvasDetailArchive) + .set({ + coverUrl: (cover.value as string | null | undefined) ?? item.coverUrl, + brief: (brief.value ?? item.brief) as any, + slots: (slots.value ?? item.slots) as any, + updatedAt: new Date(), + }) + .where(eq(canvasDetailArchive.id, item.id)); + } +} + +async function resolveMigrationConfigs() { + const setting = await getSystemSetting({ + key: OSS_MIGRATION_CONFIG_KEY, + }).catch(() => undefined); + const config = normalizeOssImageMigrationConfig(setting?.value); + + if (!config.enabled) { + return { + config, + targetConfig: undefined, + skipReason: "OSS migration is disabled", + }; + } + + const targetConfig = + (await getStorageConfigByIdForMigration({ + id: config.targetStorageConfigId || undefined, + })) || (await getDefaultStorageConfig()); + + if (!targetConfig) { + return { + config, + targetConfig: undefined, + skipReason: "No target storage config found", + }; + } + + if (isAliyunStorageConfig(targetConfig)) { + return { + config, + targetConfig: undefined, + skipReason: "Target storage config cannot be Aliyun OSS", + }; + } + + return { config, targetConfig, skipReason: undefined }; +} + +export async function runOssImageMigrationJob(): Promise { + console.log("[OSSMigrationJob] ⏰ Start running OSS migration job..."); + + const { config, targetConfig, skipReason } = await resolveMigrationConfigs(); + if (!targetConfig) { + return { + success: true, + message: "OSS migration skipped", + migrate: { total: 0, succeeded: 0, failed: 0 }, + cleanup: { total: 0, succeeded: 0, failed: 0 }, + skippedReason: skipReason, + }; + } + + const targetS3Config = toDbS3UploadConfig(targetConfig); + if (!targetS3Config) { + return { + success: false, + message: "Target storage config is invalid", + migrate: { total: 0, succeeded: 0, failed: 0 }, + cleanup: { total: 0, succeeded: 0, failed: 0 }, + }; + } + + const tasks = await getStorageTasksToMigrate({ limit: config.batchSize }); + let migrateSucceeded = 0; + let migrateFailed = 0; + + for (const task of tasks) { + await markStorageTaskMigrating(task.id); + try { + const sourceConfig = await getStorageConfigByIdForMigration({ + id: task.sourceStorageConfigId, + }); + if (!sourceConfig || !isAliyunStorageConfig(sourceConfig)) { + throw new Error("Source Aliyun OSS config is missing or disabled"); + } + if (sourceConfig.id === targetConfig.id) { + throw new Error("Source and target storage configs are identical"); + } + + const filename = deriveFilenameFromPath(task.sourcePath || task.sourceUrl); + const downloaded = await downloadImageAsBuffer(task.sourceUrl); + const uploaded = await uploadFileWithS3Config( + filename, + toArrayBuffer(downloaded), + targetS3Config + ); + + await replaceImageUrlInDatabase(task.sourceUrl, uploaded.url); + + await markStorageTaskMigrated({ + id: task.id, + targetUrl: uploaded.url, + targetStorageConfigId: targetConfig.id, + }); + + migrateSucceeded += 1; + } catch (error) { + console.error(`[OSSMigrationJob] migrate failed (${task.id}):`, error); + await markStorageTaskMigrationFailed({ id: task.id, error }); + migrateFailed += 1; + } + } + + const cleanupTasks = await getStorageTasksToCleanup({ limit: config.batchSize }); + let cleanupSucceeded = 0; + let cleanupFailed = 0; + + for (const task of cleanupTasks) { + try { + const sourceConfig = await getStorageConfigByIdForMigration({ + id: task.sourceStorageConfigId, + }); + if (!sourceConfig || !isAliyunStorageConfig(sourceConfig)) { + throw new Error("Source Aliyun OSS config unavailable for cleanup"); + } + + const sourceS3Config = toDbS3UploadConfig(sourceConfig); + if (!sourceS3Config) { + throw new Error("Source Aliyun OSS config is invalid"); + } + + await deleteS3ObjectByPath(task.sourcePath, sourceS3Config); + await markStorageTaskCleaned(task.id); + cleanupSucceeded += 1; + } catch (error) { + console.error(`[OSSMigrationJob] cleanup failed (${task.id}):`, error); + await markStorageTaskCleanupFailed({ id: task.id, error }); + cleanupFailed += 1; + } + } + + return { + success: true, + message: "OSS migration job completed", + migrate: { + total: tasks.length, + succeeded: migrateSucceeded, + failed: migrateFailed, + }, + cleanup: { + total: cleanupTasks.length, + succeeded: cleanupSucceeded, + failed: cleanupFailed, + }, + }; +} + +export async function getOssImageMigrationJobStats() { + const { config, targetConfig, skipReason } = await resolveMigrationConfigs(); + const stats = await getStorageMigrationTaskStats(); + + return { + config, + targetStorageConfig: targetConfig + ? { + id: targetConfig.id, + name: targetConfig.name, + bucket: targetConfig.bucket, + endpoint: targetConfig.endpoint, + } + : null, + skipReason: skipReason || null, + stats, + }; +} + diff --git a/lib/storage/migration-config.ts b/lib/storage/migration-config.ts new file mode 100644 index 000000000..2aa5da807 --- /dev/null +++ b/lib/storage/migration-config.ts @@ -0,0 +1,61 @@ +export const OSS_MIGRATION_CONFIG_KEY = "oss_image_migration_config"; + +export type OssImageMigrationConfig = { + enabled: boolean; + sourceStorageConfigId: string; + targetStorageConfigId: string; + migrateAfterHours: number; + cleanupAfterHours: number; + batchSize: number; +}; + +export const defaultOssImageMigrationConfig: OssImageMigrationConfig = { + enabled: true, + sourceStorageConfigId: "", + targetStorageConfigId: "", + migrateAfterHours: 48, + cleanupAfterHours: 72, + batchSize: 20, +}; + +export function normalizeOssImageMigrationConfig( + value: unknown +): OssImageMigrationConfig { + const raw = + value && typeof value === "object" && !Array.isArray(value) + ? (value as Record) + : {}; + + const migrateAfterHours = Number(raw.migrateAfterHours); + const cleanupAfterHours = Number(raw.cleanupAfterHours); + const batchSize = Number(raw.batchSize); + + const normalizedMigrateAfterHours = + Number.isFinite(migrateAfterHours) && migrateAfterHours >= 1 + ? Math.min(720, Math.round(migrateAfterHours)) + : defaultOssImageMigrationConfig.migrateAfterHours; + + const normalizedCleanupAfterHours = + Number.isFinite(cleanupAfterHours) && + cleanupAfterHours > normalizedMigrateAfterHours + ? Math.min(1440, Math.round(cleanupAfterHours)) + : Math.max( + defaultOssImageMigrationConfig.cleanupAfterHours, + normalizedMigrateAfterHours + 1 + ); + + const normalizedBatchSize = + Number.isFinite(batchSize) && batchSize >= 1 + ? Math.min(100, Math.round(batchSize)) + : defaultOssImageMigrationConfig.batchSize; + + return { + enabled: raw.enabled !== false, + sourceStorageConfigId: String(raw.sourceStorageConfigId || "").trim(), + targetStorageConfigId: String(raw.targetStorageConfigId || "").trim(), + migrateAfterHours: normalizedMigrateAfterHours, + cleanupAfterHours: normalizedCleanupAfterHours, + batchSize: normalizedBatchSize, + }; +} + diff --git a/lib/storage/migration-db.ts b/lib/storage/migration-db.ts new file mode 100644 index 000000000..f67599fd3 --- /dev/null +++ b/lib/storage/migration-db.ts @@ -0,0 +1,250 @@ +import "server-only"; + +import { and, asc, desc, eq, inArray, lte, sql } from "drizzle-orm"; +import { db } from "@/lib/db"; +import { storageConfig, storageMigrationTask, type StorageConfig } from "@/lib/db/schema"; + +export type StorageMigrationTaskRecord = typeof storageMigrationTask.$inferSelect; + +export function isAliyunStorageConfig(config: Pick | null | undefined) { + if (!config) return false; + return config.type === "s3" && /aliyuncs\.com/i.test(String(config.endpoint || "")); +} + +function isUsableS3Config(config: StorageConfig | null | undefined): config is StorageConfig { + return Boolean( + config && + config.type === "s3" && + config.isEnabled && + String(config.endpoint || "").trim() && + String(config.bucket || "").trim() && + String(config.accessKeyId || "").trim() && + String(config.secretAccessKey || "").trim() + ); +} + +export async function getEnabledS3StorageConfigs(): Promise { + const configs = await db + .select() + .from(storageConfig) + .where(and(eq(storageConfig.type, "s3"), eq(storageConfig.isEnabled, true))) + .orderBy(desc(storageConfig.isDefault), asc(storageConfig.createdAt)); + + return configs.filter(isUsableS3Config); +} + +export async function getPreferredAliyunStorageConfig({ + preferredId, +}: { + preferredId?: string; +} = {}): Promise { + const configs = await getEnabledS3StorageConfigs(); + const aliyunConfigs = configs.filter((config) => isAliyunStorageConfig(config)); + + if (aliyunConfigs.length === 0) return undefined; + + if (preferredId) { + const preferred = aliyunConfigs.find((item) => item.id === preferredId); + if (preferred) return preferred; + } + + const defaultAliyun = aliyunConfigs.find((item) => item.isDefault); + return defaultAliyun || aliyunConfigs[0]; +} + +export async function getStorageConfigByIdForMigration({ + id, +}: { + id?: string | null; +}): Promise { + if (!id) return undefined; + const [config] = await db + .select() + .from(storageConfig) + .where(eq(storageConfig.id, id)) + .limit(1); + + return isUsableS3Config(config) ? config : undefined; +} + +export async function enqueueStorageMigrationTask({ + sourceUrl, + sourcePath, + sourceStorageConfigId, + migrateAfterAt, + cleanupAfterAt, +}: { + sourceUrl: string; + sourcePath: string; + sourceStorageConfigId: string; + migrateAfterAt: Date; + cleanupAfterAt: Date; +}) { + if (!sourceUrl || !sourcePath || !sourceStorageConfigId) return; + + await db + .insert(storageMigrationTask) + .values({ + sourceUrl, + sourcePath, + sourceStorageConfigId, + migrateAfterAt, + cleanupAfterAt, + status: "pending", + updatedAt: new Date(), + }) + .onConflictDoNothing({ + target: storageMigrationTask.sourceUrl, + }); +} + +export async function getStorageTasksToMigrate({ + limit, +}: { + limit: number; +}) { + const now = new Date(); + return await db + .select() + .from(storageMigrationTask) + .where( + and( + inArray(storageMigrationTask.status, ["pending", "failed"]), + lte(storageMigrationTask.migrateAfterAt, now) + ) + ) + .orderBy(asc(storageMigrationTask.migrateAfterAt)) + .limit(Math.max(1, Math.min(100, limit))); +} + +export async function markStorageTaskMigrating(id: string) { + await db + .update(storageMigrationTask) + .set({ + status: "migrating", + attemptCount: sql`${storageMigrationTask.attemptCount} + 1`, + updatedAt: new Date(), + }) + .where(eq(storageMigrationTask.id, id)); +} + +export async function markStorageTaskMigrated({ + id, + targetUrl, + targetStorageConfigId, +}: { + id: string; + targetUrl: string; + targetStorageConfigId: string; +}) { + await db + .update(storageMigrationTask) + .set({ + status: "migrated", + targetUrl, + targetStorageConfigId, + migratedAt: new Date(), + lastError: null, + updatedAt: new Date(), + }) + .where(eq(storageMigrationTask.id, id)); +} + +export async function markStorageTaskMigrationFailed({ + id, + error, +}: { + id: string; + error: unknown; +}) { + const message = String(error instanceof Error ? error.message : error || "Unknown error").slice( + 0, + 2000 + ); + + await db + .update(storageMigrationTask) + .set({ + status: "failed", + lastError: message, + updatedAt: new Date(), + }) + .where(eq(storageMigrationTask.id, id)); +} + +export async function getStorageTasksToCleanup({ + limit, +}: { + limit: number; +}) { + const now = new Date(); + return await db + .select() + .from(storageMigrationTask) + .where( + and( + eq(storageMigrationTask.status, "migrated"), + lte(storageMigrationTask.cleanupAfterAt, now) + ) + ) + .orderBy(asc(storageMigrationTask.cleanupAfterAt)) + .limit(Math.max(1, Math.min(100, limit))); +} + +export async function markStorageTaskCleaned(id: string) { + await db + .update(storageMigrationTask) + .set({ + status: "deleted", + cleanedAt: new Date(), + lastError: null, + updatedAt: new Date(), + }) + .where(eq(storageMigrationTask.id, id)); +} + +export async function markStorageTaskCleanupFailed({ + id, + error, +}: { + id: string; + error: unknown; +}) { + const message = String(error instanceof Error ? error.message : error || "Unknown error").slice( + 0, + 2000 + ); + + await db + .update(storageMigrationTask) + .set({ + lastError: message, + updatedAt: new Date(), + }) + .where(eq(storageMigrationTask.id, id)); +} + +export async function getStorageMigrationTaskStats() { + const grouped = await db + .select({ + status: storageMigrationTask.status, + count: sql`count(*)::int`, + }) + .from(storageMigrationTask) + .groupBy(storageMigrationTask.status); + + const pending = grouped + .filter((item) => item.status === "pending" || item.status === "failed") + .reduce((sum, item) => sum + (item.count || 0), 0); + + const cleanupPending = grouped + .filter((item) => item.status === "migrated") + .reduce((sum, item) => sum + (item.count || 0), 0); + + return { + grouped, + pending, + cleanupPending, + }; +} + diff --git a/lib/storage/service.ts b/lib/storage/service.ts index 5dcbea391..51cbd5cfe 100644 --- a/lib/storage/service.ts +++ b/lib/storage/service.ts @@ -1,14 +1,26 @@ import "server-only"; -import { S3Client, PutObjectCommand } from "@aws-sdk/client-s3"; +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"; -type S3UploadConfig = { +export type S3UploadConfig = { endpoint: string; region: string; bucket: string; @@ -59,6 +71,24 @@ function getContentType(filename: string): string { return types[ext ?? ""] ?? "application/octet-stream"; } +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: !isCloudS3, + }); +} + // S3/MinIO 上传函数（使用 AWS SDK） async function uploadToS3( filename: string, @@ -85,23 +115,13 @@ async function uploadToS3( const contentType = getContentType(filename); try { - // 创建 S3 客户端（支持 MinIO/Aliyun OSS/AWS S3） - // 自动检测是否为云厂商S3（阿里云/AWS），如果是则使用 Virtual Hosted Style，否则默认为 Path Style (MinIO) - const lowerEndpoint = normalizedEndpoint.toLowerCase(); - const isCloudS3 = - lowerEndpoint.includes("aliyuncs.com") || - lowerEndpoint.includes("amazonaws.com"); - - const s3Client = new S3Client({ + const s3Client = createS3Client({ endpoint: normalizedEndpoint, - region: region || "auto", - credentials: { - accessKeyId: accessKeyId, - secretAccessKey: secretAccessKey, - }, - // MinIO 需要使用 path-style URL (true) - // 阿里云 OSS/AWS S3 推荐使用 virtual-hosted-style URL (false) - forcePathStyle: !isCloudS3, + region, + bucket, + accessKeyId, + secretAccessKey, + publicUrl: normalizedPublicUrl, }); console.log(`[S3 Upload] Uploading to: ${normalizedEndpoint}/${bucket}/${key}`); @@ -144,7 +164,28 @@ async function uploadToS3( } } -function toDbS3UploadConfig(raw: unknown): S3UploadConfig | null { +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 | null | undefined; const endpoint = value?.endpoint ? normalizeEndpoint(String(value.endpoint)) : ""; const bucket = value?.bucket ? String(value.bucket).trim() : ""; @@ -210,11 +251,68 @@ function normalizeStorageFailoverConfig(value: unknown): { }; } +async function resolvePreferredUploadStorageConfig( + migrationConfig: OssImageMigrationConfig +): Promise { + 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<{ url: string; pathname: string; contentType: string } | null> { +): 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); @@ -251,7 +349,10 @@ async function uploadWithBackupStorageConfig( const uploaded = await uploadToS3(filename, buffer, backupS3Config); lastKnownGoodDbS3Config = backupS3Config; - return uploaded; + return { + uploaded, + storageConfig: backupStorageConfig, + }; } catch (error) { console.error("[Upload] Backup storage config retry failed:", error); return null; @@ -267,16 +368,30 @@ export async function uploadFile( let dbS3Config: S3UploadConfig | null = null; let primaryStorageConfigId: string | undefined; + let selectedStorageConfig: StorageConfig | undefined; + let migrationConfig = defaultOssImageMigrationConfig; try { - const storageConfig = await getDefaultStorageConfig(); - primaryStorageConfigId = storageConfig?.id; - dbS3Config = toDbS3UploadConfig(storageConfig); + 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) { - console.log(`[Upload] Using database S3 config: ${storageConfig?.name}`); + console.log(`[Upload] Using database S3 config: ${selectedStorageConfig?.name}`); console.log(`[Upload] Endpoint: ${dbS3Config.endpoint}`); console.log(`[Upload] Bucket: ${dbS3Config.bucket}`); const uploaded = await uploadToS3(filename, buffer, dbS3Config); lastKnownGoodDbS3Config = dbS3Config; + await maybeEnqueueAliyunMigrationTask({ + uploaded, + storageConfig: selectedStorageConfig, + migrationConfig, + }).catch((error) => { + console.error("[Upload] Failed to enqueue OSS migration task:", error); + }); return uploaded; } console.warn("[Upload] No valid S3 config in database"); @@ -289,7 +404,14 @@ export async function uploadFile( primaryStorageConfigId ); if (uploadedFromBackup) { - return uploadedFromBackup; + await maybeEnqueueAliyunMigrationTask({ + uploaded: uploadedFromBackup.uploaded, + storageConfig: uploadedFromBackup.storageConfig, + migrationConfig, + }).catch((enqueueError) => { + console.error("[Upload] Failed to enqueue backup OSS migration task:", enqueueError); + }); + return uploadedFromBackup.uploaded; } if (lastKnownGoodDbS3Config) { @@ -309,7 +431,14 @@ export async function uploadFile( primaryStorageConfigId ); if (uploadedFromBackup) { - return uploadedFromBackup; + await maybeEnqueueAliyunMigrationTask({ + uploaded: uploadedFromBackup.uploaded, + storageConfig: uploadedFromBackup.storageConfig, + migrationConfig, + }).catch((enqueueError) => { + console.error("[Upload] Failed to enqueue backup OSS migration task:", enqueueError); + }); + return uploadedFromBackup.uploaded; } }