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;
});