feat(storage): add aliyun oss auto migration and cleanup pipeline
This commit is contained in:
Your Name
parent
82c9173467
commit
bd418a3fdb
@ -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() {
|
||||
</div>
|
||||
</CardContent>
|
||||
</Card>
|
||||
|
||||
<Card>
|
||||
<CardHeader>
|
||||
<CardTitle>阿里云 OSS 自动转存</CardTitle>
|
||||
<CardDescription>
|
||||
若已配置阿里云 OSS，上传优先进入 OSS；到期后自动转存到目标默认桶并替换数据库 URL。
|
||||
</CardDescription>
|
||||
</CardHeader>
|
||||
<CardContent className="space-y-4">
|
||||
<div className="flex items-center justify-between rounded-lg border p-4">
|
||||
<div className="space-y-1">
|
||||
<Label className="text-sm font-medium">启用 OSS 自动转存</Label>
|
||||
<p className="text-xs text-muted-foreground">
|
||||
开启后执行“先 OSS、后迁移、再清理”的生命周期策略（不依赖 CDN）。
|
||||
</p>
|
||||
</div>
|
||||
<Switch
|
||||
checked={siteSettings.ossImageMigrationConfig.enabled}
|
||||
onCheckedChange={(checked) =>
|
||||
setSiteSettings((prev) => ({
|
||||
...prev,
|
||||
ossImageMigrationConfig: {
|
||||
...prev.ossImageMigrationConfig,
|
||||
enabled: checked,
|
||||
},
|
||||
}))
|
||||
}
|
||||
/>
|
||||
</div>
|
||||
|
||||
<div className="grid gap-4 md:grid-cols-2">
|
||||
<div className="space-y-2">
|
||||
<Label>OSS 源桶（可选指定）</Label>
|
||||
<Select
|
||||
value={siteSettings.ossImageMigrationConfig.sourceStorageConfigId || "auto"}
|
||||
onValueChange={(value) =>
|
||||
setSiteSettings((prev) => ({
|
||||
...prev,
|
||||
ossImageMigrationConfig: {
|
||||
...prev.ossImageMigrationConfig,
|
||||
sourceStorageConfigId: value === "auto" ? "" : value,
|
||||
},
|
||||
}))
|
||||
}
|
||||
>
|
||||
<SelectTrigger>
|
||||
<SelectValue placeholder="自动选择阿里云 OSS（默认优先）" />
|
||||
</SelectTrigger>
|
||||
<SelectContent>
|
||||
<SelectItem value="auto">自动选择阿里云 OSS</SelectItem>
|
||||
{(storageData?.configs || [])
|
||||
.filter(
|
||||
(config) =>
|
||||
config.type === "s3" &&
|
||||
config.isEnabled &&
|
||||
config.endpoint?.includes("aliyuncs.com")
|
||||
)
|
||||
.map((config) => (
|
||||
<SelectItem key={config.id} value={config.id}>
|
||||
{config.name}
|
||||
{config.isDefault ? "（默认）" : ""}
|
||||
</SelectItem>
|
||||
))}
|
||||
</SelectContent>
|
||||
</Select>
|
||||
</div>
|
||||
|
||||
<div className="space-y-2">
|
||||
<Label>目标转存桶（可选指定）</Label>
|
||||
<Select
|
||||
value={siteSettings.ossImageMigrationConfig.targetStorageConfigId || "default"}
|
||||
onValueChange={(value) =>
|
||||
setSiteSettings((prev) => ({
|
||||
...prev,
|
||||
ossImageMigrationConfig: {
|
||||
...prev.ossImageMigrationConfig,
|
||||
targetStorageConfigId: value === "default" ? "" : value,
|
||||
},
|
||||
}))
|
||||
}
|
||||
>
|
||||
<SelectTrigger>
|
||||
<SelectValue placeholder="默认存储桶（建议非阿里云）" />
|
||||
</SelectTrigger>
|
||||
<SelectContent>
|
||||
<SelectItem value="default">使用当前默认存储桶</SelectItem>
|
||||
{(storageData?.configs || [])
|
||||
.filter((config) => config.type === "s3" && config.isEnabled)
|
||||
.map((config) => (
|
||||
<SelectItem key={config.id} value={config.id}>
|
||||
{config.name}
|
||||
{config.endpoint?.includes("aliyuncs.com")
|
||||
? "（阿里云 OSS，不建议）"
|
||||
: ""}
|
||||
{config.isDefault ? "（默认）" : ""}
|
||||
</SelectItem>
|
||||
))}
|
||||
</SelectContent>
|
||||
</Select>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="grid gap-4 md:grid-cols-3">
|
||||
<div className="space-y-2">
|
||||
<Label>转存时间（小时）</Label>
|
||||
<Input
|
||||
type="number"
|
||||
min={1}
|
||||
max={720}
|
||||
value={siteSettings.ossImageMigrationConfig.migrateAfterHours}
|
||||
onChange={(e) =>
|
||||
setSiteSettings((prev) => ({
|
||||
...prev,
|
||||
ossImageMigrationConfig: {
|
||||
...prev.ossImageMigrationConfig,
|
||||
migrateAfterHours: Math.max(
|
||||
1,
|
||||
Math.min(720, Number(e.target.value) || 48)
|
||||
),
|
||||
},
|
||||
}))
|
||||
}
|
||||
/>
|
||||
</div>
|
||||
<div className="space-y-2">
|
||||
<Label>清理时间（小时）</Label>
|
||||
<Input
|
||||
type="number"
|
||||
min={2}
|
||||
max={1440}
|
||||
value={siteSettings.ossImageMigrationConfig.cleanupAfterHours}
|
||||
onChange={(e) =>
|
||||
setSiteSettings((prev) => ({
|
||||
...prev,
|
||||
ossImageMigrationConfig: {
|
||||
...prev.ossImageMigrationConfig,
|
||||
cleanupAfterHours: Math.max(
|
||||
2,
|
||||
Math.min(1440, Number(e.target.value) || 72)
|
||||
),
|
||||
},
|
||||
}))
|
||||
}
|
||||
/>
|
||||
</div>
|
||||
<div className="space-y-2">
|
||||
<Label>单次批量任务</Label>
|
||||
<Input
|
||||
type="number"
|
||||
min={1}
|
||||
max={100}
|
||||
value={siteSettings.ossImageMigrationConfig.batchSize}
|
||||
onChange={(e) =>
|
||||
setSiteSettings((prev) => ({
|
||||
...prev,
|
||||
ossImageMigrationConfig: {
|
||||
...prev.ossImageMigrationConfig,
|
||||
batchSize: Math.max(
|
||||
1,
|
||||
Math.min(100, Number(e.target.value) || 20)
|
||||
),
|
||||
},
|
||||
}))
|
||||
}
|
||||
/>
|
||||
</div>
|
||||
</div>
|
||||
|
||||
<div className="rounded-lg border border-blue-200 bg-blue-50 px-3 py-2 text-xs text-blue-700">
|
||||
推荐值：转存 48 小时，清理 72 小时。并配置定时调用
|
||||
<code className="mx-1">/api/jobs/migrate-oss-images</code>
|
||||
执行迁移任务。
|
||||
</div>
|
||||
|
||||
<div className="flex justify-end">
|
||||
<Button onClick={handleSaveSiteSettings}>保存 OSS 自动转存设置</Button>
|
||||
</div>
|
||||
</CardContent>
|
||||
</Card>
|
||||
</TabsContent>
|
||||
|
||||
{/* 本地AI设置 */}
|
||||
|
||||
71
app/api/jobs/migrate-oss-images/route.ts
Normal file
71
app/api/jobs/migrate-oss-images/route.ts
Normal file
@ -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 }
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@ -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) {
|
||||
|
||||
23
lib/db/migrations/0047_add_storage_migration_tasks.sql
Normal file
23
lib/db/migrations/0047_add_storage_migration_tasks.sql
Normal file
@ -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");
|
||||
@ -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
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
@ -324,6 +324,27 @@ export const storageConfig = pgTable("StorageConfig", {
|
||||
|
||||
export type StorageConfig = InferSelectModel<typeof storageConfig>;
|
||||
|
||||
// 阿里云 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<typeof storageMigrationTask>;
|
||||
|
||||
// 网站设置表
|
||||
export const siteSettings = pgTable("SiteSettings", {
|
||||
id: uuid("id").primaryKey().notNull().defaultRandom(),
|
||||
|
||||
410
lib/jobs/oss-image-migration-job.ts
Normal file
410
lib/jobs/oss-image-migration-job.ts
Normal file
@ -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<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,
|
||||
};
|
||||
}
|
||||
|
||||
61
lib/storage/migration-config.ts
Normal file
61
lib/storage/migration-config.ts
Normal file
@ -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<string, unknown>)
|
||||
: {};
|
||||
|
||||
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,
|
||||
};
|
||||
}
|
||||
|
||||
250
lib/storage/migration-db.ts
Normal file
250
lib/storage/migration-db.ts
Normal file
@ -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<StorageConfig, "type" | "endpoint"> | 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<StorageConfig[]> {
|
||||
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<StorageConfig | undefined> {
|
||||
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<StorageConfig | undefined> {
|
||||
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<number>`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,
|
||||
};
|
||||
}
|
||||
|
||||
@ -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<string, unknown> | 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<StorageConfig | undefined> {
|
||||
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;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user