03b8a0deb0
Recover the existing 19 modified tracked files and 6 untracked source files from the production working tree without changing their contents. Production source HEAD: 439fd56fe120fdb9f855b50fa4df9ce2e71bb72a Production build ID: QhoygcZ4-x42PwCgCEj1d Captured: 2026-10-30; all 25 source file SHA256 values verified. This is a source recovery checkpoint; no deployment was performed.
1309 lines
45 KiB
TypeScript
1309 lines
45 KiB
TypeScript
import {
|
|
DeleteObjectCommand,
|
|
GetObjectCommand,
|
|
HeadObjectCommand,
|
|
ListObjectsV2Command,
|
|
PutObjectCommand,
|
|
S3Client,
|
|
} from "@aws-sdk/client-s3";
|
|
import { createHash } from "node:crypto";
|
|
import { Transform, type Readable } from "node:stream";
|
|
import postgres from "postgres";
|
|
|
|
type StorageConfig = {
|
|
id: string;
|
|
name: string;
|
|
type: string;
|
|
endpoint: string | null;
|
|
region: string | null;
|
|
bucket: string | null;
|
|
accessKeyId: string | null;
|
|
secretAccessKey: string | null;
|
|
publicUrl: string | null;
|
|
isEnabled: boolean;
|
|
};
|
|
|
|
type MigrationTask = {
|
|
id: string;
|
|
sourceUrl: string;
|
|
sourcePath: string;
|
|
attemptCount: number;
|
|
};
|
|
|
|
const SOURCE_PREFIX = "https://demo-sparkai.oss-cn-shanghai.aliyuncs.com";
|
|
const PROJECT_IMAGE_EXTENSION_PATTERN =
|
|
"(?:png|jpe?g|webp|jfif|gif|avif|tiff?|heic|heif|bmp|svg)";
|
|
const PROJECT_IMAGE_PATH_SQL_PATTERN =
|
|
`^uploads/[A-Za-z0-9._/-]+\\.${PROJECT_IMAGE_EXTENSION_PATTERN}$`;
|
|
const PROJECT_IMAGE_SOURCE_URL_PATTERN =
|
|
`(https://demo-sparkai\\.oss-cn-shanghai\\.aliyuncs\\.com/${PROJECT_IMAGE_PATH_SQL_PATTERN.slice(1, -1)})`;
|
|
const PROJECT_IMAGE_SOURCE_PATH_CAPTURE_PATTERN =
|
|
`https://demo-sparkai\\.oss-cn-shanghai\\.aliyuncs\\.com/(${PROJECT_IMAGE_PATH_SQL_PATTERN.slice(1, -1)})`;
|
|
const PROJECT_IMAGE_PATH_PATTERN = new RegExp(
|
|
PROJECT_IMAGE_PATH_SQL_PATTERN,
|
|
"i"
|
|
);
|
|
const TASK_STATUSES = ["pending", "failed", "copying", "copied"];
|
|
const args = process.argv.slice(2);
|
|
const command = args[0] || "audit";
|
|
|
|
function readArg(name: string, fallback: string) {
|
|
const prefix = `--${name}=`;
|
|
return args.find((item) => item.startsWith(prefix))?.slice(prefix.length) || fallback;
|
|
}
|
|
|
|
function readIntArg(name: string, fallback: number, min = 0, max = Number.MAX_SAFE_INTEGER) {
|
|
const parsed = Number(readArg(name, String(fallback)));
|
|
if (!Number.isFinite(parsed)) return fallback;
|
|
return Math.max(min, Math.min(max, Math.round(parsed)));
|
|
}
|
|
|
|
function cleanKey(value: string) {
|
|
return String(value || "").replace(/^\/+/, "");
|
|
}
|
|
|
|
function cleanUrl(value: string | null | undefined) {
|
|
return String(value || "").trim().replace(/\/+$/, "");
|
|
}
|
|
|
|
function formatBytes(value: number) {
|
|
const units = ["B", "KiB", "MiB", "GiB", "TiB"];
|
|
let amount = value;
|
|
let index = 0;
|
|
while (amount >= 1024 && index < units.length - 1) {
|
|
amount /= 1024;
|
|
index += 1;
|
|
}
|
|
return `${amount.toFixed(index === 0 ? 0 : 2)} ${units[index]}`;
|
|
}
|
|
|
|
function errorMessage(error: unknown) {
|
|
const record = (error && typeof error === "object" ? error : {}) as {
|
|
name?: string;
|
|
message?: string;
|
|
code?: string;
|
|
Code?: string;
|
|
cause?: unknown;
|
|
$metadata?: {
|
|
httpStatusCode?: number;
|
|
requestId?: string;
|
|
extendedRequestId?: string;
|
|
attempts?: number;
|
|
};
|
|
};
|
|
const cause =
|
|
record.cause && record.cause !== error
|
|
? String(
|
|
record.cause instanceof Error
|
|
? `${record.cause.name}: ${record.cause.message}`
|
|
: record.cause
|
|
)
|
|
: undefined;
|
|
return JSON.stringify({
|
|
name: record.name || (error instanceof Error ? error.name : undefined),
|
|
message: record.message || (error instanceof Error ? error.message : String(error || "Unknown error")),
|
|
code: record.code || record.Code,
|
|
status: record.$metadata?.httpStatusCode,
|
|
requestId: record.$metadata?.requestId,
|
|
extendedRequestId: record.$metadata?.extendedRequestId,
|
|
attempts: record.$metadata?.attempts,
|
|
cause,
|
|
}).slice(0, 1800);
|
|
}
|
|
|
|
function normalizeEtag(value: string | undefined) {
|
|
return String(value || "").replace(/^"|"$/g, "");
|
|
}
|
|
|
|
function isSimpleEtag(value: string | undefined) {
|
|
const normalized = normalizeEtag(value);
|
|
return /^[a-f0-9]{32}$/i.test(normalized);
|
|
}
|
|
|
|
function assertStorageConfig(config: StorageConfig | undefined, role: string) {
|
|
if (
|
|
!config ||
|
|
!config.endpoint ||
|
|
!config.bucket ||
|
|
!config.accessKeyId ||
|
|
!config.secretAccessKey
|
|
) {
|
|
throw new Error(`${role} storage configuration is incomplete`);
|
|
}
|
|
return config as StorageConfig & {
|
|
endpoint: string;
|
|
bucket: string;
|
|
accessKeyId: string;
|
|
secretAccessKey: string;
|
|
};
|
|
}
|
|
|
|
function createClient(config: ReturnType<typeof assertStorageConfig>) {
|
|
const endpoint = cleanUrl(config.endpoint);
|
|
const isCloud = /aliyuncs\.com|amazonaws\.com/i.test(endpoint);
|
|
return new S3Client({
|
|
endpoint,
|
|
region: config.region || "auto",
|
|
credentials: {
|
|
accessKeyId: config.accessKeyId,
|
|
secretAccessKey: config.secretAccessKey,
|
|
},
|
|
forcePathStyle: config.type === "qiniu-kodo" ? true : !isCloud,
|
|
maxAttempts: 4,
|
|
});
|
|
}
|
|
|
|
async function retry<T>(operation: () => Promise<T>, attempts = 3) {
|
|
let lastError: unknown;
|
|
for (let attempt = 1; attempt <= attempts; attempt += 1) {
|
|
try {
|
|
return await operation();
|
|
} catch (error) {
|
|
lastError = error;
|
|
if (attempt < attempts) {
|
|
await new Promise((resolve) => setTimeout(resolve, attempt * 1500));
|
|
}
|
|
}
|
|
}
|
|
throw lastError;
|
|
}
|
|
|
|
async function loadStorageConfigs(sql: ReturnType<typeof postgres>) {
|
|
const settings = await sql<{ value: Record<string, unknown> }[]>`
|
|
select "value"
|
|
from "SystemSettings"
|
|
where "key" = 'oss_image_migration_config'
|
|
limit 1
|
|
`;
|
|
const value = settings[0]?.value || {};
|
|
const sourceId = String(value.sourceStorageConfigId || "");
|
|
const targetId = String(value.targetStorageConfigId || "");
|
|
if (!sourceId || !targetId) {
|
|
throw new Error("Migration source or target storage ID is not configured");
|
|
}
|
|
|
|
const configs = await sql<StorageConfig[]>`
|
|
select
|
|
"id", "name", "type", "endpoint", "region", "bucket",
|
|
"accessKeyId", "secretAccessKey", "publicUrl", "isEnabled"
|
|
from "StorageConfig"
|
|
where "id" in (${sourceId}, ${targetId})
|
|
`;
|
|
const source = assertStorageConfig(configs.find((item) => item.id === sourceId), "Source");
|
|
const target = assertStorageConfig(configs.find((item) => item.id === targetId), "Target");
|
|
if (!/aliyuncs\.com/i.test(source.endpoint)) {
|
|
throw new Error(`Refusing to run: source is not Aliyun OSS (${source.name})`);
|
|
}
|
|
if (/aliyuncs\.com/i.test(target.endpoint)) {
|
|
throw new Error(`Refusing to run: target is also Aliyun OSS (${target.name})`);
|
|
}
|
|
if (!cleanUrl(target.publicUrl)) {
|
|
throw new Error("Target storage public URL is missing");
|
|
}
|
|
return { source, target };
|
|
}
|
|
|
|
function targetUrl(config: StorageConfig, key: string) {
|
|
return `${cleanUrl(config.publicUrl)}/${cleanKey(key)}`;
|
|
}
|
|
|
|
async function headPublicObject(url: string) {
|
|
const response = await fetch(url, {
|
|
method: "HEAD",
|
|
headers: {
|
|
"Accept-Encoding": "identity",
|
|
"User-Agent": "AI-Chatbot-Storage-Migrator/1.0",
|
|
},
|
|
signal: AbortSignal.timeout(20_000),
|
|
});
|
|
if (response.status === 404) return null;
|
|
if (!response.ok) {
|
|
throw new Error(`Target public HEAD failed: HTTP ${response.status} (${url})`);
|
|
}
|
|
let contentLength = Number(response.headers.get("content-length") || 0);
|
|
let etag = response.headers.get("etag") || undefined;
|
|
let contentType = response.headers.get("content-type") || undefined;
|
|
if (contentLength <= 0) {
|
|
const rangeResponse = await fetch(url, {
|
|
headers: {
|
|
"Accept-Encoding": "identity",
|
|
Range: "bytes=0-0",
|
|
"User-Agent": "AI-Chatbot-Storage-Migrator/1.0",
|
|
},
|
|
signal: AbortSignal.timeout(20_000),
|
|
});
|
|
if (!rangeResponse.ok) {
|
|
throw new Error(`Target public range read failed: HTTP ${rangeResponse.status} (${url})`);
|
|
}
|
|
const contentRange = rangeResponse.headers.get("content-range") || "";
|
|
contentLength = Number(
|
|
contentRange.match(/\/(\d+)$/)?.[1] ||
|
|
rangeResponse.headers.get("content-length") ||
|
|
0
|
|
);
|
|
etag = etag || rangeResponse.headers.get("etag") || undefined;
|
|
contentType = contentType || rangeResponse.headers.get("content-type") || undefined;
|
|
await rangeResponse.body?.cancel().catch(() => undefined);
|
|
}
|
|
return {
|
|
contentLength,
|
|
etag,
|
|
contentType,
|
|
};
|
|
}
|
|
|
|
async function copyObject(params: {
|
|
sourceClient: S3Client;
|
|
targetClient: S3Client;
|
|
source: ReturnType<typeof assertStorageConfig>;
|
|
target: ReturnType<typeof assertStorageConfig>;
|
|
key: string;
|
|
}) {
|
|
const key = cleanKey(params.key);
|
|
const sourceHead = await retry(() =>
|
|
params.sourceClient.send(new HeadObjectCommand({ Bucket: params.source.bucket, Key: key }))
|
|
);
|
|
const sourceBytes = Number(sourceHead.ContentLength || 0);
|
|
if (sourceBytes <= 0) throw new Error(`Source object is empty: ${key}`);
|
|
if (sourceBytes > 5 * 1024 * 1024 * 1024) {
|
|
throw new Error(`Object exceeds single-upload limit: ${key} (${sourceBytes})`);
|
|
}
|
|
|
|
const publicUrl = targetUrl(params.target, key);
|
|
let transferError: unknown;
|
|
for (let attempt = 1; attempt <= 3; attempt += 1) {
|
|
try {
|
|
const sourceObject = await params.sourceClient.send(
|
|
new GetObjectCommand({ Bucket: params.source.bucket, Key: key })
|
|
);
|
|
if (!sourceObject.Body) throw new Error(`Source object body is missing: ${key}`);
|
|
|
|
const body = sourceObject.Body as Readable;
|
|
if (typeof body.pipe !== "function") {
|
|
throw new Error(`Source object body is not a Node.js stream: ${key}`);
|
|
}
|
|
const sourceHash = createHash("md5");
|
|
const hashingStream = new Transform({
|
|
transform(chunk: Buffer, _encoding, callback) {
|
|
sourceHash.update(chunk);
|
|
callback(null, chunk);
|
|
},
|
|
});
|
|
const uploadResult = await params.targetClient.send(
|
|
new PutObjectCommand({
|
|
Bucket: params.target.bucket,
|
|
Key: key,
|
|
Body: body.pipe(hashingStream) as never,
|
|
ContentLength: sourceBytes,
|
|
ContentType: sourceHead.ContentType || "application/octet-stream",
|
|
CacheControl: sourceHead.CacheControl || "public, max-age=31536000, immutable",
|
|
ContentDisposition: sourceHead.ContentDisposition,
|
|
Metadata: sourceHead.Metadata,
|
|
ACL: "public-read",
|
|
})
|
|
);
|
|
const sourceMd5 = sourceHash.digest("hex");
|
|
|
|
const copiedHead = await retry(() => headPublicObject(publicUrl), 5);
|
|
if (!copiedHead || copiedHead.contentLength !== sourceBytes) {
|
|
throw new Error(
|
|
`Target size mismatch: ${key} source=${sourceBytes} target=${copiedHead?.contentLength}`
|
|
);
|
|
}
|
|
if (
|
|
isSimpleEtag(uploadResult.ETag || copiedHead.etag) &&
|
|
sourceMd5 !== normalizeEtag(uploadResult.ETag || copiedHead.etag)
|
|
) {
|
|
throw new Error(`Target checksum mismatch: ${key}`);
|
|
}
|
|
return { bytes: sourceBytes, reused: false };
|
|
} catch (error) {
|
|
transferError = error;
|
|
if (attempt < 3) {
|
|
await new Promise((resolve) => setTimeout(resolve, attempt * 2000));
|
|
}
|
|
}
|
|
}
|
|
throw transferError;
|
|
}
|
|
|
|
async function runPool<T>(items: T[], concurrency: number, worker: (item: T) => Promise<void>) {
|
|
let cursor = 0;
|
|
await Promise.all(
|
|
Array.from({ length: Math.min(concurrency, items.length) }, async () => {
|
|
while (cursor < items.length) {
|
|
const item = items[cursor];
|
|
cursor += 1;
|
|
await worker(item);
|
|
}
|
|
})
|
|
);
|
|
}
|
|
|
|
async function audit(sql: ReturnType<typeof postgres>, source: ReturnType<typeof assertStorageConfig>, target: ReturnType<typeof assertStorageConfig>) {
|
|
const statuses = await sql<{ status: string; count: number }[]>`
|
|
select "status", count(*)::int as "count"
|
|
from "StorageMigrationTask"
|
|
where "sourceStorageConfigId" = ${source.id}
|
|
and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`}
|
|
and "sourcePath" like 'uploads/%'
|
|
and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN}
|
|
group by "status"
|
|
order by "status"
|
|
`;
|
|
console.log("storage", {
|
|
source: { name: source.name, bucket: source.bucket, enabled: source.isEnabled },
|
|
target: { name: target.name, bucket: target.bucket, publicUrl: cleanUrl(target.publicUrl) },
|
|
});
|
|
console.log("task_statuses", statuses);
|
|
|
|
if (readArg("list", "false") !== "true") return;
|
|
|
|
const taskRows = await sql<{ sourcePath: string }[]>`
|
|
select "sourcePath"
|
|
from "StorageMigrationTask"
|
|
where "status" in ${sql(TASK_STATUSES)}
|
|
and "sourceStorageConfigId" = ${source.id}
|
|
and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`}
|
|
and "sourcePath" like 'uploads/%'
|
|
and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN}
|
|
`;
|
|
const taskKeys = new Set(taskRows.map((item) => cleanKey(item.sourcePath)));
|
|
const sourceClient = createClient(source);
|
|
const targetClient = createClient(target);
|
|
const sourceSizes = new Map<string, number>();
|
|
|
|
async function list(client: S3Client, bucket: string, label: string) {
|
|
let continuationToken: string | undefined;
|
|
let count = 0;
|
|
let bytes = 0;
|
|
let taskCount = 0;
|
|
let taskBytes = 0;
|
|
let maxBytes = 0;
|
|
do {
|
|
const page = await client.send(
|
|
new ListObjectsV2Command({
|
|
Bucket: bucket,
|
|
Prefix: "uploads/",
|
|
ContinuationToken: continuationToken,
|
|
})
|
|
);
|
|
for (const item of page.Contents || []) {
|
|
const key = cleanKey(item.Key || "");
|
|
const size = Number(item.Size || 0);
|
|
count += 1;
|
|
bytes += size;
|
|
maxBytes = Math.max(maxBytes, size);
|
|
if (taskKeys.has(key)) {
|
|
taskCount += 1;
|
|
taskBytes += size;
|
|
}
|
|
if (label === "source") sourceSizes.set(key, size);
|
|
}
|
|
continuationToken = page.IsTruncated ? page.NextContinuationToken : undefined;
|
|
if (count > 0 && count % 100_000 < 1000) {
|
|
console.log(`${label}_list_progress`, { count, bytes: formatBytes(bytes) });
|
|
}
|
|
} while (continuationToken);
|
|
console.log(`${label}_inventory`, {
|
|
count,
|
|
bytes: formatBytes(bytes),
|
|
taskCount,
|
|
taskBytes: formatBytes(taskBytes),
|
|
extraObjects: count - taskCount,
|
|
missingTaskObjects: taskKeys.size - taskCount,
|
|
maxObject: formatBytes(maxBytes),
|
|
});
|
|
}
|
|
|
|
try {
|
|
await list(sourceClient, source.bucket, "source");
|
|
await list(targetClient, target.bucket, "target");
|
|
} finally {
|
|
sourceClient.destroy();
|
|
targetClient.destroy();
|
|
sourceSizes.clear();
|
|
}
|
|
}
|
|
|
|
async function probe(sql: ReturnType<typeof postgres>, source: ReturnType<typeof assertStorageConfig>, target: ReturnType<typeof assertStorageConfig>) {
|
|
const rows = await sql<MigrationTask[]>`
|
|
select "id", "sourceUrl", "sourcePath", "attemptCount"
|
|
from "StorageMigrationTask"
|
|
where "status" in ('pending', 'failed')
|
|
and "sourceStorageConfigId" = ${source.id}
|
|
and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`}
|
|
and "sourcePath" like 'uploads/%'
|
|
and "sourcePath" not like '%..%'
|
|
and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN}
|
|
and "sourceUrl" = ${`${SOURCE_PREFIX}/`} || "sourcePath"
|
|
order by "createdAt" desc
|
|
limit 20
|
|
`;
|
|
if (rows.length === 0) throw new Error("No pending migration task is available for probe");
|
|
|
|
const sourceClient = createClient(source);
|
|
const targetClient = createClient(target);
|
|
const task = rows[0];
|
|
const originalKey = cleanKey(task.sourcePath);
|
|
const probeKey = `migration-probe/${Date.now()}-${originalKey.split("/").pop()}`;
|
|
try {
|
|
console.log("probe_stage", { stage: "source_head", key: originalKey });
|
|
const sourceHead = await sourceClient.send(
|
|
new HeadObjectCommand({ Bucket: source.bucket, Key: originalKey })
|
|
);
|
|
console.log("probe_stage", { stage: "source_get", bytes: sourceHead.ContentLength });
|
|
const sourceObject = await sourceClient.send(
|
|
new GetObjectCommand({ Bucket: source.bucket, Key: originalKey })
|
|
);
|
|
if (!sourceObject.Body) throw new Error("Probe source body is missing");
|
|
console.log("probe_stage", { stage: "target_put", key: probeKey });
|
|
await targetClient.send(
|
|
new PutObjectCommand({
|
|
Bucket: target.bucket,
|
|
Key: probeKey,
|
|
Body: sourceObject.Body as never,
|
|
ContentLength: sourceHead.ContentLength,
|
|
ContentType: sourceHead.ContentType || "application/octet-stream",
|
|
CacheControl: "no-store",
|
|
ACL: "public-read",
|
|
})
|
|
);
|
|
const publicUrl = targetUrl(target, probeKey);
|
|
console.log("probe_stage", { stage: "target_public_read" });
|
|
const response = await fetch(publicUrl, {
|
|
headers: { Range: "bytes=0-0", "User-Agent": "AI-Chatbot-Storage-Migration-Probe/1.0" },
|
|
signal: AbortSignal.timeout(20_000),
|
|
});
|
|
if (!response.ok) {
|
|
throw new Error(`Probe public read failed: HTTP ${response.status}`);
|
|
}
|
|
const contentRange = response.headers.get("content-range") || "";
|
|
const responseBytes = Number(contentRange.match(/\/(\d+)$/)?.[1] || 0);
|
|
if (responseBytes > 0 && responseBytes !== Number(sourceHead.ContentLength || 0)) {
|
|
throw new Error(
|
|
`Probe public size mismatch: source=${sourceHead.ContentLength} target=${responseBytes}`
|
|
);
|
|
}
|
|
console.log("probe_ok", {
|
|
sourceKey: originalKey,
|
|
bytes: sourceHead.ContentLength,
|
|
contentType: sourceHead.ContentType,
|
|
targetPublicReadStatus: response.status,
|
|
});
|
|
} finally {
|
|
await targetClient
|
|
.send(new DeleteObjectCommand({ Bucket: target.bucket, Key: probeKey }))
|
|
.catch(() => undefined);
|
|
sourceClient.destroy();
|
|
targetClient.destroy();
|
|
}
|
|
}
|
|
|
|
async function copyTasks(sql: ReturnType<typeof postgres>, source: ReturnType<typeof assertStorageConfig>, target: ReturnType<typeof assertStorageConfig>) {
|
|
const concurrency = readIntArg("concurrency", 4, 1, 16);
|
|
const batchSize = readIntArg("batch", concurrency * 5, concurrency, 500);
|
|
const limit = readIntArg("limit", 100, 0);
|
|
const maxAttempts = readIntArg("max-attempts", 12, 1, 100);
|
|
let stopping = false;
|
|
let processed = 0;
|
|
let succeeded = 0;
|
|
let failed = 0;
|
|
let bytes = 0;
|
|
let reused = 0;
|
|
const startedAt = Date.now();
|
|
process.once("SIGINT", () => {
|
|
stopping = true;
|
|
});
|
|
process.once("SIGTERM", () => {
|
|
stopping = true;
|
|
});
|
|
|
|
await sql`
|
|
update "StorageMigrationTask"
|
|
set "status" = 'failed',
|
|
"lastError" = 'Recovered stale copying task',
|
|
"updatedAt" = now()
|
|
where "status" = 'copying'
|
|
and "sourceStorageConfigId" = ${source.id}
|
|
and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`}
|
|
and "sourcePath" like 'uploads/%'
|
|
and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN}
|
|
and (
|
|
"targetStorageConfigId" is null
|
|
or "targetStorageConfigId" = ${target.id}
|
|
)
|
|
and "updatedAt" < now() - interval '30 minutes'
|
|
`;
|
|
|
|
const sourceClient = createClient(source);
|
|
const targetClient = createClient(target);
|
|
try {
|
|
while (!stopping && (limit === 0 || processed < limit)) {
|
|
const remaining = limit === 0 ? batchSize : Math.min(batchSize, limit - processed);
|
|
const tasks = await sql.begin(async (transaction) => {
|
|
return transaction<MigrationTask[]>`
|
|
with selected as (
|
|
select "id"
|
|
from "StorageMigrationTask"
|
|
where (
|
|
"status" = 'pending'
|
|
or (
|
|
"status" = 'failed'
|
|
and "updatedAt" < now() - interval '10 minutes'
|
|
)
|
|
)
|
|
and "sourceStorageConfigId" = ${source.id}
|
|
and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`}
|
|
and "sourcePath" like 'uploads/%'
|
|
and "sourcePath" not like '%..%'
|
|
and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN}
|
|
and (
|
|
"targetStorageConfigId" is null
|
|
or "targetStorageConfigId" = ${target.id}
|
|
)
|
|
and "attemptCount" < ${maxAttempts}
|
|
order by "createdAt", "id"
|
|
for update skip locked
|
|
limit ${remaining}
|
|
)
|
|
update "StorageMigrationTask" task
|
|
set "status" = 'copying',
|
|
"attemptCount" = task."attemptCount" + 1,
|
|
"lastError" = null,
|
|
"updatedAt" = now()
|
|
from selected
|
|
where task."id" = selected."id"
|
|
returning task."id", task."sourceUrl", task."sourcePath", task."attemptCount"
|
|
`;
|
|
});
|
|
if (tasks.length === 0) break;
|
|
|
|
await runPool(tasks, concurrency, async (task) => {
|
|
try {
|
|
if (!task.sourceUrl.startsWith(`${SOURCE_PREFIX}/uploads/`)) {
|
|
throw new Error(`Unexpected source URL: ${task.sourceUrl}`);
|
|
}
|
|
if (
|
|
!task.sourcePath.startsWith("uploads/") ||
|
|
task.sourcePath.includes("..") ||
|
|
!PROJECT_IMAGE_PATH_PATTERN.test(task.sourcePath) ||
|
|
task.sourceUrl !== `${SOURCE_PREFIX}/${task.sourcePath}`
|
|
) {
|
|
throw new Error(
|
|
`Project object scope mismatch: ${task.sourceUrl} (${task.sourcePath})`
|
|
);
|
|
}
|
|
const result = await copyObject({
|
|
sourceClient,
|
|
targetClient,
|
|
source,
|
|
target,
|
|
key: task.sourcePath,
|
|
});
|
|
const url = targetUrl(target, task.sourcePath);
|
|
await sql`
|
|
update "StorageMigrationTask"
|
|
set "status" = 'copied',
|
|
"targetUrl" = ${url},
|
|
"targetStorageConfigId" = ${target.id},
|
|
"lastError" = null,
|
|
"updatedAt" = now()
|
|
where "id" = ${task.id}
|
|
`;
|
|
succeeded += 1;
|
|
bytes += result.bytes;
|
|
if (result.reused) reused += 1;
|
|
} catch (error) {
|
|
failed += 1;
|
|
await sql`
|
|
update "StorageMigrationTask"
|
|
set "status" = 'failed',
|
|
"lastError" = ${errorMessage(error)},
|
|
"updatedAt" = now()
|
|
where "id" = ${task.id}
|
|
`;
|
|
} finally {
|
|
processed += 1;
|
|
if (processed % 25 === 0 || processed === tasks.length) {
|
|
const seconds = Math.max(1, (Date.now() - startedAt) / 1000);
|
|
console.log("copy_progress", {
|
|
processed,
|
|
succeeded,
|
|
failed,
|
|
reused,
|
|
bytes: formatBytes(bytes),
|
|
objectsPerSecond: Number((processed / seconds).toFixed(2)),
|
|
});
|
|
}
|
|
}
|
|
});
|
|
}
|
|
} finally {
|
|
sourceClient.destroy();
|
|
targetClient.destroy();
|
|
}
|
|
console.log("copy_finished", {
|
|
processed,
|
|
succeeded,
|
|
failed,
|
|
reused,
|
|
bytes: formatBytes(bytes),
|
|
stopped: stopping,
|
|
});
|
|
}
|
|
|
|
function quoteIdentifier(value: string) {
|
|
return `"${value.replaceAll('"', '""')}"`;
|
|
}
|
|
|
|
async function inventoryDatabaseReferences(sql: ReturnType<typeof postgres>) {
|
|
const sourcePrefix = `${SOURCE_PREFIX}/uploads/`;
|
|
const excludedTables = new Set([
|
|
"StorageConfig",
|
|
"StorageMigrationTask",
|
|
"SystemSettings",
|
|
"SystemSettingHistory",
|
|
"SystemSettingSnapshot",
|
|
"__drizzle_migrations",
|
|
]);
|
|
const columns = await sql<
|
|
{ tableName: string; columnName: string; dataType: string }[]
|
|
>`
|
|
select
|
|
table_name as "tableName",
|
|
column_name as "columnName",
|
|
data_type as "dataType"
|
|
from information_schema.columns
|
|
where table_schema = 'public'
|
|
and data_type in ('text', 'character varying', 'json', 'jsonb')
|
|
order by table_name, ordinal_position
|
|
`;
|
|
const candidates = columns.filter((item) => !excludedTables.has(item.tableName));
|
|
const matches: Array<{
|
|
tableName: string;
|
|
columnName: string;
|
|
dataType: string;
|
|
count: number;
|
|
}> = [];
|
|
const timeouts: Array<{ tableName: string; columnName: string; error: string }> = [];
|
|
|
|
for (let index = 0; index < candidates.length; index += 1) {
|
|
const item = candidates[index];
|
|
const table = quoteIdentifier(item.tableName);
|
|
const column = quoteIdentifier(item.columnName);
|
|
try {
|
|
const rows = await sql.unsafe<{ count: string }[]>(
|
|
`select count(*)::text as count from ${table} where ${column}::text ~* $1`,
|
|
[PROJECT_IMAGE_SOURCE_URL_PATTERN]
|
|
);
|
|
const count = Number(rows[0]?.count || 0);
|
|
if (count > 0) {
|
|
const match = { ...item, count };
|
|
matches.push(match);
|
|
console.log("reference_column", match);
|
|
}
|
|
} catch (error) {
|
|
const timeout = {
|
|
tableName: item.tableName,
|
|
columnName: item.columnName,
|
|
error: errorMessage(error),
|
|
};
|
|
timeouts.push(timeout);
|
|
console.log("reference_scan_failed", timeout);
|
|
}
|
|
if ((index + 1) % 25 === 0) {
|
|
console.log("reference_scan_progress", {
|
|
scanned: index + 1,
|
|
total: candidates.length,
|
|
matchedColumns: matches.length,
|
|
failedColumns: timeouts.length,
|
|
});
|
|
}
|
|
}
|
|
console.log("reference_inventory", {
|
|
sourcePrefix,
|
|
scannedColumns: candidates.length,
|
|
matchedColumns: matches.length,
|
|
failedColumns: timeouts.length,
|
|
totalMatchingRowsAcrossColumns: matches.reduce((sum, item) => sum + item.count, 0),
|
|
matches,
|
|
timeouts,
|
|
});
|
|
}
|
|
|
|
async function auditReferenceCoverage(
|
|
sql: ReturnType<typeof postgres>,
|
|
source: ReturnType<typeof assertStorageConfig>,
|
|
target: ReturnType<typeof assertStorageConfig>
|
|
) {
|
|
const sourcePrefix = `${SOURCE_PREFIX}/uploads/`;
|
|
const sourcePattern = PROJECT_IMAGE_SOURCE_URL_PATTERN;
|
|
const excludedTables = new Set([
|
|
"StorageConfig",
|
|
"StorageMigrationTask",
|
|
"SystemSettings",
|
|
"SystemSettingHistory",
|
|
"SystemSettingSnapshot",
|
|
"__drizzle_migrations",
|
|
]);
|
|
const columns = await sql<
|
|
{ tableName: string; columnName: string; dataType: string }[]
|
|
>`
|
|
select
|
|
table_name as "tableName",
|
|
column_name as "columnName",
|
|
data_type as "dataType"
|
|
from information_schema.columns
|
|
where table_schema = 'public'
|
|
and data_type in ('text', 'character varying', 'json', 'jsonb')
|
|
order by table_name, ordinal_position
|
|
`;
|
|
const candidates = columns.filter((item) => !excludedTables.has(item.tableName));
|
|
|
|
await sql.begin(async (transaction) => {
|
|
await transaction.unsafe(
|
|
"create temporary table migration_referenced_url (url text primary key) on commit drop"
|
|
);
|
|
for (let index = 0; index < candidates.length; index += 1) {
|
|
const item = candidates[index];
|
|
const table = quoteIdentifier(item.tableName);
|
|
const column = quoteIdentifier(item.columnName);
|
|
await transaction.unsafe(
|
|
`
|
|
insert into migration_referenced_url (url)
|
|
select distinct (matches_array)[1]
|
|
from ${table}
|
|
cross join lateral regexp_matches(${column}::text, $1, 'gi')
|
|
as extracted(matches_array)
|
|
where ${column}::text like $2
|
|
on conflict (url) do nothing
|
|
`,
|
|
[sourcePattern, `%${sourcePrefix}%`]
|
|
);
|
|
if ((index + 1) % 25 === 0) {
|
|
const progress = await transaction<{ count: string }[]>`
|
|
select count(*)::text as "count" from migration_referenced_url
|
|
`;
|
|
console.log("reference_coverage_progress", {
|
|
scanned: index + 1,
|
|
total: candidates.length,
|
|
distinctUrls: Number(progress[0]?.count || 0),
|
|
});
|
|
}
|
|
}
|
|
|
|
const coverage = await transaction<
|
|
{
|
|
distinctReferencedUrls: string;
|
|
coveredByTask: string;
|
|
missingTask: string;
|
|
copied: string;
|
|
unresolved: string;
|
|
}[]
|
|
>`
|
|
select
|
|
count(*)::text as "distinctReferencedUrls",
|
|
count(task."id")::text as "coveredByTask",
|
|
count(*) filter (where task."id" is null)::text as "missingTask",
|
|
count(*) filter (
|
|
where task."status" in ('copied', 'migrated', 'deleted')
|
|
and task."targetStorageConfigId" = ${target.id}
|
|
and task."targetUrl" = ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath"
|
|
)::text as "copied",
|
|
count(*) filter (
|
|
where task."id" is null
|
|
or task."status" not in ('copied', 'migrated', 'deleted')
|
|
or task."targetStorageConfigId" is distinct from ${target.id}
|
|
or task."targetUrl" is distinct from ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath"
|
|
)::text as "unresolved"
|
|
from migration_referenced_url referenced
|
|
left join "StorageMigrationTask" task
|
|
on task."sourceUrl" = referenced.url
|
|
and task."sourceStorageConfigId" = ${source.id}
|
|
`;
|
|
const missingSamples = await transaction<{ url: string }[]>`
|
|
select referenced.url
|
|
from migration_referenced_url referenced
|
|
left join "StorageMigrationTask" task
|
|
on task."sourceUrl" = referenced.url
|
|
and task."sourceStorageConfigId" = ${source.id}
|
|
where task."id" is null
|
|
order by referenced.url
|
|
limit 30
|
|
`;
|
|
const unreferencedTasks = await transaction<{ count: string }[]>`
|
|
select count(*)::text as "count"
|
|
from "StorageMigrationTask" task
|
|
left join migration_referenced_url referenced on referenced.url = task."sourceUrl"
|
|
where task."status" in ('pending', 'failed', 'copying', 'copied')
|
|
and task."sourceStorageConfigId" = ${source.id}
|
|
and task."sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`}
|
|
and task."sourcePath" like 'uploads/%'
|
|
and task."sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN}
|
|
and referenced.url is null
|
|
`;
|
|
console.log("reference_coverage", {
|
|
...coverage[0],
|
|
unreferencedActiveTasks: Number(unreferencedTasks[0]?.count || 0),
|
|
missingTaskSamples: missingSamples.map((item) => item.url),
|
|
});
|
|
});
|
|
}
|
|
|
|
async function enqueueMissingReferences(
|
|
sql: ReturnType<typeof postgres>,
|
|
source: ReturnType<typeof assertStorageConfig>
|
|
) {
|
|
const sourcePrefix = `${SOURCE_PREFIX}/uploads/`;
|
|
const sourcePattern = PROJECT_IMAGE_SOURCE_URL_PATTERN;
|
|
const excludedTables = new Set([
|
|
"StorageConfig",
|
|
"StorageMigrationTask",
|
|
"SystemSettings",
|
|
"SystemSettingHistory",
|
|
"SystemSettingSnapshot",
|
|
"__drizzle_migrations",
|
|
]);
|
|
const columns = await sql<
|
|
{ tableName: string; columnName: string; dataType: string }[]
|
|
>`
|
|
select
|
|
table_name as "tableName",
|
|
column_name as "columnName",
|
|
data_type as "dataType"
|
|
from information_schema.columns
|
|
where table_schema = 'public'
|
|
and data_type in ('text', 'character varying', 'json', 'jsonb')
|
|
order by table_name, ordinal_position
|
|
`;
|
|
const candidates = columns.filter((item) => !excludedTables.has(item.tableName));
|
|
|
|
await sql.begin(async (transaction) => {
|
|
await transaction.unsafe(
|
|
"create temporary table migration_referenced_url (url text primary key) on commit drop"
|
|
);
|
|
for (const item of candidates) {
|
|
const table = quoteIdentifier(item.tableName);
|
|
const column = quoteIdentifier(item.columnName);
|
|
await transaction.unsafe(
|
|
`
|
|
insert into migration_referenced_url (url)
|
|
select distinct (matches_array)[1]
|
|
from ${table}
|
|
cross join lateral regexp_matches(${column}::text, $1, 'gi')
|
|
as extracted(matches_array)
|
|
where ${column}::text like $2
|
|
on conflict (url) do nothing
|
|
`,
|
|
[sourcePattern, `%${sourcePrefix}%`]
|
|
);
|
|
}
|
|
const inserted = await transaction<{ id: string }[]>`
|
|
insert into "StorageMigrationTask" (
|
|
"sourceUrl",
|
|
"sourcePath",
|
|
"sourceStorageConfigId",
|
|
"status",
|
|
"migrateAfterAt",
|
|
"cleanupAfterAt",
|
|
"updatedAt"
|
|
)
|
|
select
|
|
referenced.url,
|
|
replace(referenced.url, ${`${SOURCE_PREFIX}/`}, ''),
|
|
${source.id},
|
|
'pending',
|
|
now(),
|
|
now() + interval '30 days',
|
|
now()
|
|
from migration_referenced_url referenced
|
|
left join "StorageMigrationTask" task on task."sourceUrl" = referenced.url
|
|
where task."id" is null
|
|
on conflict ("sourceUrl") do nothing
|
|
returning "id"
|
|
`;
|
|
console.log("missing_references_enqueued", { inserted: inserted.length });
|
|
});
|
|
}
|
|
|
|
async function replaceSampleReference(
|
|
sql: ReturnType<typeof postgres>,
|
|
source: ReturnType<typeof assertStorageConfig>,
|
|
target: ReturnType<typeof assertStorageConfig>
|
|
) {
|
|
const tasks = await sql<
|
|
{ id: string; sourceUrl: string; targetUrl: string; sourcePath: string }[]
|
|
>`
|
|
select task."id", task."sourceUrl", task."targetUrl", task."sourcePath"
|
|
from "StorageMigrationTask" task
|
|
where task."status" = 'copied'
|
|
and task."sourceStorageConfigId" = ${source.id}
|
|
and task."sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`}
|
|
and task."sourcePath" like 'uploads/%'
|
|
and task."sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN}
|
|
and task."targetStorageConfigId" = ${target.id}
|
|
and task."targetUrl" = ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath"
|
|
and (
|
|
exists (
|
|
select 1 from "GeneratedImage" image
|
|
where image."originalUrl" = task."sourceUrl"
|
|
or image."thumbnailUrl" = task."sourceUrl"
|
|
)
|
|
or exists (
|
|
select 1 from "MediaAsset" asset
|
|
where asset."originalUrl" = task."sourceUrl"
|
|
or asset."normalizedUrl" = lower(task."sourceUrl")
|
|
or asset."previewUrl" = task."sourceUrl"
|
|
or asset."thumbnailUrl" = task."sourceUrl"
|
|
)
|
|
)
|
|
order by task."updatedAt"
|
|
limit 1
|
|
`;
|
|
const task = tasks[0];
|
|
if (!task) throw new Error("No copied and referenced task is available for sample replacement");
|
|
|
|
const targetHead = await retry(() => headPublicObject(task.targetUrl), 5);
|
|
if (!targetHead || targetHead.contentLength <= 0) {
|
|
throw new Error(`Sample target is not publicly readable: ${task.targetUrl}`);
|
|
}
|
|
|
|
const excludedTables = new Set([
|
|
"StorageConfig",
|
|
"StorageMigrationTask",
|
|
"SystemSettings",
|
|
"SystemSettingHistory",
|
|
"SystemSettingSnapshot",
|
|
"__drizzle_migrations",
|
|
]);
|
|
const columns = await sql<
|
|
{ tableName: string; columnName: string; dataType: string }[]
|
|
>`
|
|
select
|
|
table_name as "tableName",
|
|
column_name as "columnName",
|
|
data_type as "dataType"
|
|
from information_schema.columns
|
|
where table_schema = 'public'
|
|
and data_type in ('text', 'character varying', 'json', 'jsonb')
|
|
order by table_name, ordinal_position
|
|
`;
|
|
const candidates = columns.filter((item) => !excludedTables.has(item.tableName));
|
|
|
|
const result = await sql.begin(async (transaction) => {
|
|
const changedColumns: Array<{
|
|
tableName: string;
|
|
columnName: string;
|
|
changedRows: number;
|
|
}> = [];
|
|
for (const item of candidates) {
|
|
const table = quoteIdentifier(item.tableName);
|
|
const column = quoteIdentifier(item.columnName);
|
|
const replacement =
|
|
item.dataType === "json" || item.dataType === "jsonb"
|
|
? `replace(${column}::text, $1, $2)::${item.dataType}`
|
|
: `replace(${column}, $1, $2)`;
|
|
const rows = await transaction.unsafe<{ count: string }[]>(
|
|
`
|
|
with updated as (
|
|
update ${table}
|
|
set ${column} = ${replacement}
|
|
where ${column}::text like $3
|
|
returning 1
|
|
)
|
|
select count(*)::text as count from updated
|
|
`,
|
|
[task.sourceUrl, task.targetUrl, `%${task.sourceUrl}%`]
|
|
);
|
|
const changedRows = Number(rows[0]?.count || 0);
|
|
if (changedRows > 0) {
|
|
changedColumns.push({
|
|
tableName: item.tableName,
|
|
columnName: item.columnName,
|
|
changedRows,
|
|
});
|
|
}
|
|
}
|
|
await transaction`
|
|
update "StorageMigrationTask"
|
|
set "status" = 'migrated',
|
|
"migratedAt" = now(),
|
|
"cleanupAfterAt" = now() + interval '30 days',
|
|
"lastError" = null,
|
|
"updatedAt" = now()
|
|
where "id" = ${task.id}
|
|
and "status" = 'copied'
|
|
`;
|
|
return changedColumns;
|
|
});
|
|
|
|
const remaining = await sql<{ count: string }[]>`
|
|
select (
|
|
select count(*) from "GeneratedImage"
|
|
where "originalUrl" = ${task.sourceUrl} or "thumbnailUrl" = ${task.sourceUrl}
|
|
) + (
|
|
select count(*) from "MediaAsset"
|
|
where "originalUrl" = ${task.sourceUrl}
|
|
or "previewUrl" = ${task.sourceUrl}
|
|
or "thumbnailUrl" = ${task.sourceUrl}
|
|
or "normalizedUrl" = ${task.sourceUrl.toLowerCase()}
|
|
) as "count"
|
|
`;
|
|
console.log("sample_reference_replaced", {
|
|
taskId: task.id,
|
|
sourcePath: task.sourcePath,
|
|
targetBytes: targetHead.contentLength,
|
|
changedColumns: result,
|
|
directSourceReferencesRemaining: Number(remaining[0]?.count || 0),
|
|
});
|
|
}
|
|
|
|
async function replaceReadyReferences(
|
|
sql: ReturnType<typeof postgres>,
|
|
source: ReturnType<typeof assertStorageConfig>,
|
|
target: ReturnType<typeof assertStorageConfig>
|
|
) {
|
|
const sourcePrefix = `${SOURCE_PREFIX}/uploads/`;
|
|
const sourcePattern = PROJECT_IMAGE_SOURCE_URL_PATTERN;
|
|
const targetReplacement = `${cleanUrl(target.publicUrl)}/\\1`;
|
|
const excludedTables = new Set([
|
|
"StorageConfig",
|
|
"StorageMigrationTask",
|
|
"SystemSettings",
|
|
"SystemSettingHistory",
|
|
"SystemSettingSnapshot",
|
|
"__drizzle_migrations",
|
|
]);
|
|
const columns = await sql<
|
|
{ tableName: string; columnName: string; dataType: string }[]
|
|
>`
|
|
select
|
|
column_info.table_name as "tableName",
|
|
column_info.column_name as "columnName",
|
|
column_info.data_type as "dataType"
|
|
from information_schema.columns column_info
|
|
join information_schema.tables table_info
|
|
on table_info.table_schema = column_info.table_schema
|
|
and table_info.table_name = column_info.table_name
|
|
where column_info.table_schema = 'public'
|
|
and table_info.table_type = 'BASE TABLE'
|
|
and column_info.is_generated = 'NEVER'
|
|
and column_info.is_updatable = 'YES'
|
|
and column_info.data_type in ('text', 'character varying', 'json', 'jsonb')
|
|
order by column_info.table_name, column_info.ordinal_position
|
|
`;
|
|
const candidates = columns.filter((item) => !excludedTables.has(item.tableName));
|
|
|
|
const readiness = await sql.begin(async (transaction) => {
|
|
await transaction.unsafe(
|
|
"create temporary table migration_referenced_url (url text primary key) on commit drop"
|
|
);
|
|
for (const item of candidates) {
|
|
const table = quoteIdentifier(item.tableName);
|
|
const column = quoteIdentifier(item.columnName);
|
|
await transaction.unsafe(
|
|
`
|
|
insert into migration_referenced_url (url)
|
|
select distinct (matches_array)[1]
|
|
from ${table}
|
|
cross join lateral regexp_matches(${column}::text, $1, 'gi')
|
|
as extracted(matches_array)
|
|
where ${column}::text like $2
|
|
on conflict (url) do nothing
|
|
`,
|
|
[sourcePattern, `%${sourcePrefix}%`]
|
|
);
|
|
}
|
|
|
|
const rows = await transaction<
|
|
{ referenced: string; ready: string; unresolved: string }[]
|
|
>`
|
|
select
|
|
count(*)::text as "referenced",
|
|
count(*) filter (
|
|
where task."status" in ('copied', 'migrated', 'deleted')
|
|
and task."targetStorageConfigId" = ${target.id}
|
|
and task."targetUrl" = ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath"
|
|
)::text as "ready",
|
|
count(*) filter (
|
|
where task."id" is null
|
|
or task."status" not in ('copied', 'migrated', 'deleted')
|
|
or task."targetStorageConfigId" is distinct from ${target.id}
|
|
or task."targetUrl" is distinct from ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath"
|
|
)::text as "unresolved"
|
|
from migration_referenced_url referenced
|
|
left join "StorageMigrationTask" task
|
|
on task."sourceUrl" = referenced.url
|
|
and task."sourceStorageConfigId" = ${source.id}
|
|
`;
|
|
const unresolvedSamples = await transaction<
|
|
{ url: string; status: string | null; lastError: string | null }[]
|
|
>`
|
|
select referenced.url, task."status", task."lastError"
|
|
from migration_referenced_url referenced
|
|
left join "StorageMigrationTask" task
|
|
on task."sourceUrl" = referenced.url
|
|
and task."sourceStorageConfigId" = ${source.id}
|
|
where task."id" is null
|
|
or task."status" not in ('copied', 'migrated', 'deleted')
|
|
or task."targetStorageConfigId" is distinct from ${target.id}
|
|
or task."targetUrl" is distinct from ${`${cleanUrl(target.publicUrl)}/`} || task."sourcePath"
|
|
order by referenced.url
|
|
limit 30
|
|
`;
|
|
return {
|
|
referenced: Number(rows[0]?.referenced || 0),
|
|
ready: Number(rows[0]?.ready || 0),
|
|
unresolved: Number(rows[0]?.unresolved || 0),
|
|
unresolvedSamples,
|
|
};
|
|
});
|
|
|
|
console.log("reference_replacement_readiness", readiness);
|
|
if (readiness.unresolved > 0) {
|
|
throw new Error(
|
|
`Refusing database URL replacement: ${readiness.unresolved} referenced objects are unresolved`
|
|
);
|
|
}
|
|
|
|
const changedColumns: Array<{
|
|
tableName: string;
|
|
columnName: string;
|
|
changedRows: number;
|
|
}> = [];
|
|
for (const item of candidates) {
|
|
const table = quoteIdentifier(item.tableName);
|
|
const column = quoteIdentifier(item.columnName);
|
|
const replacement =
|
|
item.dataType === "json" || item.dataType === "jsonb"
|
|
? `regexp_replace(${column}::text, $1, $2, 'gi')::${item.dataType}`
|
|
: `regexp_replace(${column}, $1, $2, 'gi')`;
|
|
const rows = await sql.begin(async (transaction) => {
|
|
await transaction.unsafe("set local lock_timeout = '3s'");
|
|
await transaction.unsafe("set local statement_timeout = '15min'");
|
|
return transaction.unsafe<{ count: string }[]>(
|
|
`
|
|
with updated as (
|
|
update ${table}
|
|
set ${column} = ${replacement}
|
|
where ${column}::text like $3
|
|
returning 1
|
|
)
|
|
select count(*)::text as count from updated
|
|
`,
|
|
[
|
|
PROJECT_IMAGE_SOURCE_PATH_CAPTURE_PATTERN,
|
|
targetReplacement,
|
|
`%${sourcePrefix}%`,
|
|
]
|
|
);
|
|
});
|
|
const changedRows = Number(rows[0]?.count || 0);
|
|
if (changedRows > 0) {
|
|
const changed = {
|
|
tableName: item.tableName,
|
|
columnName: item.columnName,
|
|
changedRows,
|
|
};
|
|
changedColumns.push(changed);
|
|
console.log("reference_column_replaced", changed);
|
|
}
|
|
}
|
|
|
|
const remainingColumns: Array<{
|
|
tableName: string;
|
|
columnName: string;
|
|
count: number;
|
|
}> = [];
|
|
for (const item of candidates) {
|
|
const table = quoteIdentifier(item.tableName);
|
|
const column = quoteIdentifier(item.columnName);
|
|
const rows = await sql.unsafe<{ count: string }[]>(
|
|
`select count(*)::text as count from ${table} where ${column}::text ~* $1`,
|
|
[PROJECT_IMAGE_SOURCE_URL_PATTERN]
|
|
);
|
|
const count = Number(rows[0]?.count || 0);
|
|
if (count > 0) {
|
|
remainingColumns.push({
|
|
tableName: item.tableName,
|
|
columnName: item.columnName,
|
|
count,
|
|
});
|
|
}
|
|
}
|
|
if (remainingColumns.length > 0) {
|
|
console.log("reference_replacement_remaining", remainingColumns);
|
|
throw new Error(
|
|
`Database URL replacement is incomplete in ${remainingColumns.length} columns`
|
|
);
|
|
}
|
|
|
|
const migrated = await sql<{ id: string }[]>`
|
|
update "StorageMigrationTask"
|
|
set "status" = 'migrated',
|
|
"migratedAt" = coalesce("migratedAt", now()),
|
|
"cleanupAfterAt" = greatest(
|
|
coalesce("cleanupAfterAt", now()),
|
|
now() + interval '30 days'
|
|
),
|
|
"lastError" = null,
|
|
"updatedAt" = now()
|
|
where "sourceStorageConfigId" = ${source.id}
|
|
and "targetStorageConfigId" = ${target.id}
|
|
and "status" = 'copied'
|
|
and "sourceUrl" like ${`${SOURCE_PREFIX}/uploads/%`}
|
|
and "sourcePath" like 'uploads/%'
|
|
and "sourcePath" ~* ${PROJECT_IMAGE_PATH_SQL_PATTERN}
|
|
and "targetUrl" = ${`${cleanUrl(target.publicUrl)}/`} || "sourcePath"
|
|
returning "id"
|
|
`;
|
|
console.log("reference_replacement_finished", {
|
|
referencedUrls: readiness.referenced,
|
|
changedColumns,
|
|
changedRows: changedColumns.reduce((sum, item) => sum + item.changedRows, 0),
|
|
tasksMarkedMigrated: migrated.length,
|
|
sourceRetentionDays: 30,
|
|
});
|
|
}
|
|
|
|
async function main() {
|
|
if (!process.env.POSTGRES_URL) throw new Error("POSTGRES_URL is required");
|
|
const concurrency = readIntArg("db-connections", 8, 2, 20);
|
|
const sql = postgres(process.env.POSTGRES_URL, {
|
|
max: concurrency,
|
|
idle_timeout: 30,
|
|
connect_timeout: 20,
|
|
prepare: false,
|
|
});
|
|
try {
|
|
const { source, target } = await loadStorageConfigs(sql);
|
|
if (command === "audit") {
|
|
await audit(sql, source, target);
|
|
} else if (command === "probe") {
|
|
await probe(sql, source, target);
|
|
} else if (command === "copy") {
|
|
await copyTasks(sql, source, target);
|
|
} else if (command === "inventory-db") {
|
|
await inventoryDatabaseReferences(sql);
|
|
} else if (command === "audit-reference-coverage") {
|
|
await auditReferenceCoverage(sql, source, target);
|
|
} else if (command === "enqueue-missing-references") {
|
|
await enqueueMissingReferences(sql, source);
|
|
} else if (command === "replace-sample") {
|
|
await replaceSampleReference(sql, source, target);
|
|
} else if (command === "replace-ready-references") {
|
|
await replaceReadyReferences(sql, source, target);
|
|
} else {
|
|
throw new Error(`Unknown command: ${command}`);
|
|
}
|
|
} finally {
|
|
await sql.end({ timeout: 10 });
|
|
}
|
|
}
|
|
|
|
main().catch((error) => {
|
|
console.error("migration_failed", errorMessage(error));
|
|
process.exitCode = 1;
|
|
});
|