411 lines
12 KiB
TypeScript
411 lines
12 KiB
TypeScript
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<Buffer> {
|
|
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<string, unknown> = {};
|
|
for (const [key, item] of Object.entries(value as Record<string, unknown>)) {
|
|
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<JobResult> {
|
|
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,
|
|
};
|
|
}
|
|
|