#!/usr/bin/env node
import {spawnSync} from 'node:child_process';
import {createHash, createHmac} from 'node:crypto';
import {
  chmodSync,
  closeSync,
  createReadStream,
  createWriteStream,
  existsSync,
  lstatSync,
  mkdirSync,
  openSync,
  readdirSync,
  readFileSync,
  readlinkSync,
  realpathSync,
  renameSync,
  rmSync,
  statfsSync,
  statSync,
  symlinkSync,
  writeFileSync,
} from 'node:fs';
import {basename, dirname, isAbsolute, join, relative, resolve, sep} from 'node:path';
import {Readable} from 'node:stream';
import {pipeline} from 'node:stream/promises';
import {setTimeout as sleep} from 'node:timers/promises';
import {fileURLToPath} from 'node:url';

const __filename = fileURLToPath(import.meta.url);
const __dirname = dirname(__filename);
const REPO_ROOT = resolve(__dirname, '..');
const DEFAULT_LOCAL_DATA_DIR = resolve(REPO_ROOT, 'server-rs/.spacetimedb/local/data');
const DEFAULT_LOCAL_WORK_DIR = resolve(REPO_ROOT, 'server-rs/.data/database-backups');
const DEFAULT_PRODUCTION_DATA_DIR = '/stdb';
const DEFAULT_PRODUCTION_WORK_DIR = '/var/lib/genarrative/database-backups';
const DEFAULT_SPACE_SAFETY_RATIO = 1.1;
const DEFAULT_EXTRA_FREE_BYTES = 512 * 1024 * 1024;
const OSS_ALGORITHM = 'OSS4-HMAC-SHA256';
const OSS_SERVICE = 'oss';
const OSS_REQUEST = 'aliyun_v4_request';
const UNSIGNED_PAYLOAD = 'UNSIGNED-PAYLOAD';
const DEFAULT_OSS_MULTIPART_PART_SIZE_BYTES = 128 * 1024 * 1024;
const OSS_MIN_MULTIPART_PART_SIZE_BYTES = 100 * 1024;
const OSS_MAX_MULTIPART_PART_SIZE_BYTES = 5 * 1024 * 1024 * 1024;
const OSS_MAX_MULTIPART_PARTS = 10_000;
const DEFAULT_OSS_REQUEST_MAX_ATTEMPTS = 5;
const DEFAULT_OSS_RETRY_BASE_DELAY_MS = 1_000;
const DEFAULT_OSS_RETRY_MAX_DELAY_MS = 30_000;
const DEFAULT_DIRECT_FILES_CONCURRENCY = 16;
const MAX_DIRECT_FILES_CONCURRENCY = 64;
const DIRECT_FILES_SINGLE_PUT_MAX_BYTES = 16 * 1024 * 1024;
const RETRYABLE_OSS_HTTP_STATUSES = new Set([408, 429, 500, 502, 503, 504]);
const HISTORY_STATE_SCHEMA_VERSION = 1;
const HISTORY_MANIFEST_SCHEMA_VERSION = 1;
const DIRECT_FILES_STATE_SCHEMA_VERSION = 1;
const DIRECT_FILES_CATALOG_SCHEMA_VERSION = 1;
const DIRECT_FILES_LATEST_SCHEMA_VERSION = 1;

function usage() {
  console.log(`用法:
  npm run database:backup:oss -- [--mode full|history] [--storage-format archive|files] [--data-dir <path>] [--work-dir <path>] [--bucket <bucket>] [--object-prefix <prefix>] [--keep-local]
  node -- scripts/database-backup-to-oss.mjs [--stop-service spacetimedb.service] [--restart-service-after genarrative-api.service] [--defer-upload]
  node -- scripts/database-backup-to-oss.mjs --upload-archive <path>
  node -- scripts/database-backup-to-oss.mjs --publish-manifest <path>
  node -- scripts/database-backup-to-oss.mjs --restore-files-state <path> --restore-dir <path>
  node -- scripts/database-backup-to-oss.mjs --restore-files-latest --restore-dir <path> [--dry-run]

说明:
  将 SpacetimeDB 数据目录以 .tar.gz 或逐文件 catalog 形式上传到阿里云 OSS 指定 bucket。
  默认 full 模式保持原有全量冷备行为；history 模式只归档已被最新 snapshot 覆盖的历史 commitlog 与旧 snapshot。
  --storage-format files 不打包：按原相对路径建立 catalog，文件内容以 SHA-256 不可变对象上传；重复运行只上传新增或变化内容。
  archive history 必须有已验真的 full baseline state；files history 必须复用同一 work-dir 中已发布的 full catalog state。
  --defer-upload 只生成本地冷备份和 manifest，不上传；后续用 --upload-archive 异步上传。
  默认读取 .env / .env.local / .env.secrets.local；生产服务可传 --env-file /etc/genarrative/api-server.env。
  shell 环境变量优先级最高，不会被 env 文件覆盖。

常用环境变量:
  GENARRATIVE_DATABASE_BACKUP_DATA_DIR        数据目录；生产建议 /stdb
  GENARRATIVE_DATABASE_BACKUP_WORK_DIR        本地临时备份目录；生产建议 /var/lib/genarrative/database-backups
  GENARRATIVE_DATABASE_BACKUP_OSS_BUCKET      备份 bucket；未设置时回退 ALIYUN_OSS_BUCKET
  GENARRATIVE_DATABASE_BACKUP_OSS_PREFIX      对象前缀，默认 database-backups
  GENARRATIVE_DATABASE_BACKUP_OSS_ENDPOINT    OSS endpoint；未设置时回退 ALIYUN_OSS_ENDPOINT
  GENARRATIVE_DATABASE_BACKUP_STORAGE_FORMAT  archive（默认）或 files
  GENARRATIVE_DATABASE_BACKUP_KEEP_LOCAL      true 时保留本地 tar.gz
  GENARRATIVE_DATABASE_BACKUP_MIN_FREE_BYTES  备份前要求 work dir 所在文件系统至少有这些可用字节；未设置时按数据目录大小估算
  GENARRATIVE_DATABASE_BACKUP_BASELINE_STATE  history 使用的 full baseline 与追加批次状态文件
  GENARRATIVE_DATABASE_BACKUP_BASELINE_MANIFEST 首次初始化 history state 的 uploaded full manifest
  ALIYUN_OSS_ACCESS_KEY_ID / ALIYUN_OSS_ACCESS_KEY_SECRET
`);
}

function loadEnvFile(filePath, target, protectedKeys) {
  if (!existsSync(filePath)) {
    return;
  }
  const rawText = readFileSync(filePath, 'utf8');
  for (const rawLine of rawText.split(/\r?\n/u)) {
    const line = rawLine.trim();
    if (!line || line.startsWith('#')) {
      continue;
    }
    const match = line.match(/^([A-Za-z_][A-Za-z0-9_]*)=(.*)$/u);
    if (!match) {
      continue;
    }
    const [, key, rawValue] = match;
    if (protectedKeys.has(key)) {
      continue;
    }
    target[key] = rawValue.replace(/^['"]|['"]$/gu, '');
  }
}

function loadRepoEnv() {
  const env = {...process.env};
  const protectedKeys = new Set(
    Object.entries(process.env)
      .filter(([, value]) => String(value ?? '').trim())
      .map(([key]) => key),
  );
  for (const fileName of ['.env', '.env.local', '.env.secrets.local']) {
    loadEnvFile(resolve(REPO_ROOT, fileName), env, protectedKeys);
  }
  return env;
}

function loadEffectiveEnv(envFiles) {
  const env = loadRepoEnv();
  const protectedKeys = new Set(
    Object.entries(process.env)
      .filter(([, value]) => String(value ?? '').trim())
      .map(([key]) => key),
  );
  for (const filePath of envFiles) {
    loadEnvFile(resolvePath(filePath), env, protectedKeys);
  }
  return env;
}

function parseArgs(argv) {
  const options = {
    dataDir: '',
    workDir: '',
    bucket: '',
    endpoint: '',
    objectPrefix: '',
    accessKeyId: '',
    accessKeySecret: '',
    envFiles: [],
    keepLocal: false,
    stopService: '',
    restartServicesAfter: [],
    database: '',
    dryRun: false,
    deferUpload: false,
    uploadArchive: '',
    manifestFile: '',
    objectKey: '',
    resultFile: '',
    minFreeBytes: '',
    mode: 'full',
    baselineState: '',
    baselineManifest: '',
    publishManifest: '',
    storageFormat: '',
    restoreFilesState: '',
    restoreFilesLatest: false,
    restoreDir: '',
  };

  for (let index = 0; index < argv.length; index += 1) {
    const arg = argv[index];
    const readValue = () => {
      const value = argv[index + 1];
      if (!value || value.startsWith('--')) {
        throw new Error(`${arg} 缺少参数值`);
      }
      index += 1;
      return value;
    };

    switch (arg) {
      case '--help':
      case '-h':
        usage();
        process.exit(0);
        break;
      case '--data-dir':
        options.dataDir = readValue();
        break;
      case '--work-dir':
        options.workDir = readValue();
        break;
      case '--bucket':
        options.bucket = readValue();
        break;
      case '--endpoint':
        options.endpoint = readValue();
        break;
      case '--object-prefix':
        options.objectPrefix = readValue();
        break;
      case '--object-key':
        options.objectKey = readValue();
        break;
      case '--access-key-id':
        options.accessKeyId = readValue();
        break;
      case '--access-key-secret':
        options.accessKeySecret = readValue();
        break;
      case '--env-file':
        options.envFiles.push(readValue());
        break;
      case '--database':
        options.database = readValue();
        break;
      case '--stop-service':
        options.stopService = readValue();
        break;
      case '--restart-service-after':
        options.restartServicesAfter.push(readValue());
        break;
      case '--keep-local':
        options.keepLocal = true;
        break;
      case '--dry-run':
        options.dryRun = true;
        break;
      case '--defer-upload':
        options.deferUpload = true;
        options.keepLocal = true;
        break;
      case '--upload-archive':
        options.uploadArchive = readValue();
        break;
      case '--manifest-file':
        options.manifestFile = readValue();
        break;
      case '--result-file':
        options.resultFile = readValue();
        break;
      case '--min-free-bytes':
        options.minFreeBytes = readValue();
        break;
      case '--mode':
        options.mode = readValue();
        break;
      case '--baseline-state':
        options.baselineState = readValue();
        break;
      case '--baseline-manifest':
        options.baselineManifest = readValue();
        break;
      case '--publish-manifest':
        options.publishManifest = readValue();
        break;
      case '--storage-format':
        options.storageFormat = readValue();
        break;
      case '--restore-files-state':
        options.restoreFilesState = readValue();
        break;
      case '--restore-files-latest':
        options.restoreFilesLatest = true;
        break;
      case '--restore-dir':
        options.restoreDir = readValue();
        break;
      default:
        throw new Error(`未知参数: ${arg}`);
    }
  }

  return options;
}

function firstNonEmpty(...values) {
  return values.map((value) => String(value ?? '').trim()).find(Boolean) ?? '';
}

function parseDirectFilesConcurrency(rawValue) {
  const value = Number(String(rawValue ?? DEFAULT_DIRECT_FILES_CONCURRENCY).trim());
  if (!Number.isSafeInteger(value) || value < 1 || value > MAX_DIRECT_FILES_CONCURRENCY) {
    throw new Error(`GENARRATIVE_DATABASE_BACKUP_FILES_CONCURRENCY 必须是 1-${MAX_DIRECT_FILES_CONCURRENCY} 的整数，实际: ${rawValue}`);
  }
  return value;
}

export function createUploadBandwidthLimiter(rawValue, {nowFn = Date.now, sleepImpl = sleep} = {}) {
  const normalized = String(rawValue ?? '').trim();
  if (!normalized || normalized === '0') {
    return null;
  }
  const maxBytesPerSecond = Number(normalized);
  if (!Number.isSafeInteger(maxBytesPerSecond) || maxBytesPerSecond < 1024) {
    throw new Error(`GENARRATIVE_DATABASE_BACKUP_UPLOAD_MAX_BYTES_PER_SECOND 必须为空、0 或 >= 1024 的整数，实际: ${rawValue}`);
  }
  let nextAvailableAtMs = 0;
  const waitForChunk = async (sizeBytes) => {
    const now = nowFn();
    const startAt = Math.max(now, nextAvailableAtMs);
    const finishAt = startAt + (sizeBytes / maxBytesPerSecond) * 1000;
    nextAvailableAtMs = finishAt;
    const delayMs = Math.max(0, finishAt - now);
    if (delayMs > 0) {
      await sleepImpl(delayMs);
    }
  };
  return {
    maxBytesPerSecond,
    wrap(readable) {
      return Readable.from((async function* throttleUpload() {
        for await (const chunk of readable) {
          await waitForChunk(chunk.length);
          yield chunk;
        }
      })(), {objectMode: false});
    },
  };
}

function createBufferReadStream(buffer, chunkSizeBytes = 64 * 1024) {
  return Readable.from((function* readChunks() {
    for (let offset = 0; offset < buffer.length; offset += chunkSizeBytes) {
      yield buffer.subarray(offset, Math.min(offset + chunkSizeBytes, buffer.length));
    }
  })(), {objectMode: false});
}

function resolvePath(value) {
  return isAbsolute(value) ? value : resolve(REPO_ROOT, value);
}

function normalizeEndpoint(raw) {
  return String(raw ?? '')
    .trim()
    .replace(/^https?:\/\//u, '')
    .replace(/\/+$/u, '');
}

function sanitizeObjectPart(value, fallback) {
  const sanitized = String(value ?? '')
    .trim()
    .toLowerCase()
    .replace(/[^a-z0-9._-]+/gu, '-')
    .replace(/-+/gu, '-')
    .replace(/^-|-$/gu, '');
  return sanitized || fallback;
}

function timestampForFile(date = new Date()) {
  const pad = (value) => String(value).padStart(2, '0');
  return `${date.getUTCFullYear()}${pad(date.getUTCMonth() + 1)}${pad(date.getUTCDate())}T${pad(date.getUTCHours())}${pad(date.getUTCMinutes())}${pad(date.getUTCSeconds())}Z`;
}

function buildBackupNames({database, dataDir, objectPrefix}) {
  const timestamp = timestampForFile();
  const databasePart = sanitizeObjectPart(database || basename(dataDir), 'spacetimedb');
  const fileName = `${databasePart}-${timestamp}.tar.gz`;
  const prefix = String(objectPrefix || 'database-backups')
    .trim()
    .replace(/^\/+|\/+$/gu, '')
    .split('/')
    .filter(Boolean)
    .map((part) => sanitizeObjectPart(part, 'backup'))
    .join('/');
  const objectKey = [prefix, databasePart, fileName].filter(Boolean).join('/');
  return {fileName, objectKey};
}

function atomicWriteJson(filePath, payload) {
  mkdirSync(dirname(filePath), {recursive: true});
  const tempPath = `${filePath}.${process.pid}.${Date.now()}.tmp`;
  writeFileSync(tempPath, `${JSON.stringify(payload, null, 2)}\n`, {encoding: 'utf8', mode: 0o600});
  chmodSync(tempPath, 0o600);
  renameSync(tempPath, filePath);
}

function processIsAlive(pid) {
  try {
    process.kill(pid, 0);
    return true;
  } catch (error) {
    return error?.code === 'EPERM';
  }
}

function acquireBackupLock({workDir, database}) {
  mkdirSync(workDir, {recursive: true});
  const lockPath = join(workDir, `${sanitizeObjectPart(database, 'spacetimedb')}.backup.lock`);
  try {
    const fd = openSync(lockPath, 'wx', 0o600);
    writeFileSync(fd, `${process.pid}\n`, 'utf8');
    closeSync(fd);
    const release = () => {
      try {
        const ownerPid = Number(String(readFileSync(lockPath, 'utf8')).trim());
        if (ownerPid === process.pid) {
          rmSync(lockPath, {force: true});
        }
      } catch {
        // The lock may already have been removed by the normal exit path.
      }
    };
    process.once('exit', release);
    for (const signal of ['SIGINT', 'SIGTERM']) {
      process.once(signal, () => {
        release();
        process.exit(signal === 'SIGINT' ? 130 : 143);
      });
    }
    return lockPath;
  } catch (error) {
    if (error?.code !== 'EEXIST') {
      throw error;
    }
  }
  const ownerPid = Number(String(readFileSync(lockPath, 'utf8')).trim());
  if (Number.isSafeInteger(ownerPid) && ownerPid > 0 && processIsAlive(ownerPid)) {
    throw new Error(`已有数据库备份进程持有锁: ${lockPath} pid=${ownerPid}`);
  }
  throw new Error(`发现失效数据库备份锁，拒绝自动抢锁；请核对 OSS multipart 与进程后手工删除: ${lockPath} pid=${ownerPid || '<invalid>'}`);
}

function historyStatePath({args, env, workDir, database}) {
  return resolvePath(firstNonEmpty(
    args.baselineState,
    env.GENARRATIVE_DATABASE_BACKUP_BASELINE_STATE,
    join(workDir, `${sanitizeObjectPart(database, 'spacetimedb')}-history-state.json`),
  ));
}

function baselineIdFor(baseline) {
  return sha256Hex([
    baseline.bucket,
    baseline.objectKey,
    baseline.verifiedAt,
    baseline.contentLength,
    baseline.archiveSha256,
  ].join('\0')).slice(0, 24);
}

function normalizeUploadedBaselineManifest(manifest, {database, dataDir}) {
  if (manifest.uploadStatus !== 'uploaded') {
    throw new Error(`baseline manifest 必须是 uploaded，实际: ${manifest.uploadStatus ?? '<missing>'}`);
  }
  if (manifest.backupKind !== 'spacetimedb-data-dir') {
    throw new Error(`baseline manifest backupKind 必须是 spacetimedb-data-dir，实际: ${manifest.backupKind ?? '<missing>'}`);
  }
  const baseline = {
    backupKind: 'spacetimedb-data-dir',
    database: firstNonEmpty(manifest.database, database),
    dataDir: resolvePath(firstNonEmpty(manifest.dataDir, dataDir)),
    bucket: String(manifest.bucket ?? '').trim(),
    objectKey: String(manifest.objectKey ?? '').trim(),
    verifiedAt: String(manifest.verifiedAt ?? '').trim(),
    uploadedAt: String(manifest.uploadedAt ?? '').trim(),
    contentLength: Number(manifest.contentLength),
    archiveSha256: String(manifest.archiveSha256 ?? '').trim().toLowerCase(),
    manifestObjectKey: String(manifest.manifestObjectKey ?? '').trim(),
    manifestContentLength: Number(manifest.manifestContentLength),
    manifestArchiveSha256: String(manifest.manifestArchiveSha256 ?? '').trim().toLowerCase(),
    manifestVerifiedAt: String(manifest.manifestVerifiedAt ?? '').trim(),
  };
  if (
    !baseline.bucket
    || !baseline.objectKey
    || !baseline.verifiedAt
    || !Number.isSafeInteger(baseline.contentLength)
    || baseline.contentLength <= 0
    || !/^[a-f0-9]{64}$/u.test(baseline.archiveSha256)
    || !baseline.manifestObjectKey
    || !Number.isSafeInteger(baseline.manifestContentLength)
    || baseline.manifestContentLength <= 0
    || !/^[a-f0-9]{64}$/u.test(baseline.manifestArchiveSha256)
    || !baseline.manifestVerifiedAt
  ) {
    throw new Error('baseline manifest 缺少已验真 OSS 归档或 sidecar 信息。');
  }
  baseline.id = baselineIdFor(baseline);
  return baseline;
}

function validateHistoryState(state, {database, dataDir}) {
  if (state.schemaVersion !== HISTORY_STATE_SCHEMA_VERSION || !state.baseline) {
    throw new Error('history state schemaVersion 或 baseline 无效。');
  }
  const baseline = normalizeUploadedBaselineManifest(
    {...state.baseline, uploadStatus: 'uploaded'},
    {database, dataDir},
  );
  if (baseline.database !== database) {
    throw new Error(`history state database 不匹配: expected=${database}, actual=${baseline.database}`);
  }
  if (resolvePath(baseline.dataDir) !== resolvePath(dataDir)) {
    throw new Error(`history state dataDir 不匹配: expected=${resolvePath(dataDir)}, actual=${resolvePath(baseline.dataDir)}`);
  }
  return {
    ...state,
    baseline,
    batches: Array.isArray(state.batches) ? state.batches : [],
  };
}

function writeBaselineState({statePath, baseline, previousState = null}) {
  const state = {
    schemaVersion: HISTORY_STATE_SCHEMA_VERSION,
    updatedAt: new Date().toISOString(),
    baseline,
    batches: previousState?.baseline?.id === baseline.id && Array.isArray(previousState.batches)
      ? previousState.batches
      : [],
  };
  atomicWriteJson(statePath, state);
  return state;
}

function loadOrImportHistoryState({args, env, statePath, database, dataDir}) {
  if (existsSync(statePath)) {
    return validateHistoryState(readManifest(statePath), {database, dataDir});
  }
  const importPath = firstNonEmpty(args.baselineManifest, env.GENARRATIVE_DATABASE_BACKUP_BASELINE_MANIFEST);
  if (!importPath) {
    throw new Error(`history 模式缺少已验真 baseline state: ${statePath}；可用 --baseline-manifest 导入已有 uploaded baseline manifest。`);
  }
  const baseline = normalizeUploadedBaselineManifest(readManifest(resolvePath(importPath)), {database, dataDir});
  return writeBaselineState({statePath, baseline});
}

function assertSafeRelativePath(dataDir, absolutePath) {
  const relativePath = relative(resolvePath(dataDir), resolvePath(absolutePath));
  if (!relativePath || relativePath === '..' || relativePath.startsWith(`..${sep}`) || isAbsolute(relativePath)) {
    throw new Error(`history 候选路径越界或等于数据目录: ${absolutePath}`);
  }
  return relativePath.split(sep).join('/');
}

function statFingerprint(absolutePath, rootPath = absolutePath) {
  const entries = [];
  let totalSize = 0n;
  const visit = (currentPath) => {
    const stat = lstatSync(currentPath, {bigint: true});
    if (stat.isSymbolicLink()) {
      throw new Error(`history 候选不得包含符号链接: ${currentPath}`);
    }
    const entryPath = currentPath === rootPath ? '.' : relative(rootPath, currentPath).split(sep).join('/');
    const kind = stat.isDirectory() ? 'directory' : stat.isFile() ? 'file' : 'other';
    if (kind === 'other') {
      throw new Error(`history 候选只允许普通文件或目录: ${currentPath}`);
    }
    entries.push([
      entryPath,
      kind,
      stat.dev.toString(),
      stat.ino.toString(),
      stat.mode.toString(),
      stat.size.toString(),
      stat.mtimeNs.toString(),
    ].join('\0'));
    if (stat.isFile()) {
      totalSize += stat.size;
    } else {
      for (const name of readdirSync(currentPath).sort()) {
        visit(join(currentPath, name));
      }
    }
  };
  visit(rootPath);
  return {
    fingerprint: sha256Hex(entries.join('\n')),
    sizeBytes: totalSize.toString(),
    entryCount: entries.length,
  };
}

function findReplicasDir(dataDir) {
  const candidates = [resolve(dataDir, 'replicas'), resolve(dataDir, 'data', 'replicas')]
    .filter((candidate) => existsSync(candidate) && lstatSync(candidate).isDirectory());
  if (candidates.length !== 1) {
    throw new Error(`无法唯一确定 replicas 目录: ${candidates.length === 0 ? '<none>' : candidates.join(', ')}`);
  }
  return candidates[0];
}

function historyCandidate({dataDir, absolutePath, kind, replicaId, transaction}) {
  const stat = statFingerprint(absolutePath);
  return {
    path: assertSafeRelativePath(dataDir, absolutePath),
    kind,
    replicaId,
    transaction: transaction.toString(),
    ...stat,
  };
}

export function discoverHistoryPlan({dataDir}) {
  const resolvedDataDir = resolvePath(dataDir);
  const replicasDir = findReplicasDir(resolvedDataDir);
  const replicaEntries = readdirSync(replicasDir, {withFileTypes: true});
  const replicas = [];
  const candidates = [];

  for (const replicaEntry of replicaEntries.sort((left, right) => left.name.localeCompare(right.name))) {
    if (!replicaEntry.isDirectory()) {
      continue;
    }
    if (!/^\d+$/u.test(replicaEntry.name)) {
      throw new Error(`replica 目录名不符合预期: ${replicaEntry.name}`);
    }
    const replicaId = replicaEntry.name;
    const replicaDir = join(replicasDir, replicaId);
    const snapshotsDir = join(replicaDir, 'snapshots');
    const clogDir = join(replicaDir, 'clog');
    if (!existsSync(snapshotsDir) || !lstatSync(snapshotsDir).isDirectory()) {
      replicas.push({replicaId, status: 'skipped', reason: 'no-snapshots-directory'});
      continue;
    }
    const snapshotEntries = readdirSync(snapshotsDir, {withFileTypes: true});
    const snapshots = snapshotEntries.flatMap((entry) => {
      const match = /^(\d{20})\.snapshot_dir$/u.exec(entry.name);
      if (!match) {
        return [];
      }
      const transaction = BigInt(match[1]);
      if (transaction > 0xffff_ffff_ffff_ffffn) {
        throw new Error(`snapshot transaction 超出 u64: ${entry.name}`);
      }
      if (!entry.isDirectory()) {
        throw new Error(`snapshot 候选必须是目录: ${join(snapshotsDir, entry.name)}`);
      }
      const snapshotDir = join(snapshotsDir, entry.name);
      const lockPath = join(snapshotsDir, `${match[1]}.lock`);
      const snapshotFile = join(snapshotDir, `${match[1]}.snapshot_bsatn`);
      if (existsSync(lockPath) || !existsSync(snapshotFile) || !lstatSync(snapshotFile).isFile()) {
        return [];
      }
      return [{name: entry.name, transaction}];
    }).sort((left, right) => left.transaction < right.transaction ? -1 : left.transaction > right.transaction ? 1 : 0);
    if (snapshots.length === 0) {
      replicas.push({replicaId, status: 'skipped', reason: 'no-snapshot'});
      continue;
    }
    if (!existsSync(clogDir) || !lstatSync(clogDir).isDirectory()) {
      throw new Error(`replica ${replicaId} 缺少 clog 目录。`);
    }
    const segmentFiles = new Map();
    for (const entry of readdirSync(clogDir, {withFileTypes: true})) {
      const match = /^(\d{20})\.stdb\.(log|ofs)$/u.exec(entry.name);
      if (!match) {
        throw new Error(`commitlog 文件名不符合预期: ${entry.name}`);
      }
      if (!entry.isFile()) {
        throw new Error(`commitlog 候选必须是普通文件: ${join(clogDir, entry.name)}`);
      }
      const transaction = BigInt(match[1]);
      if (transaction > 0xffff_ffff_ffff_ffffn) {
        throw new Error(`commitlog transaction 超出 u64: ${entry.name}`);
      }
      const key = transaction.toString();
      const group = segmentFiles.get(key) ?? {transaction};
      group[match[2]] = entry.name;
      segmentFiles.set(key, group);
    }
    for (const group of segmentFiles.values()) {
      if (group.ofs && !group.log) {
        throw new Error(`commitlog offset 缺少对应 log: replica=${replicaId}, transaction=${group.transaction}`);
      }
    }
    const segments = [...segmentFiles.values()]
      .filter((group) => group.log)
      .sort((left, right) => left.transaction < right.transaction ? -1 : left.transaction > right.transaction ? 1 : 0);
    const latestSnapshot = snapshots.at(-1).transaction;
    const boundarySegment = segments.filter((segment) => segment.transaction <= latestSnapshot).at(-1);
    if (!boundarySegment) {
      throw new Error(`replica ${replicaId} 无法找到覆盖 latest snapshot ${latestSnapshot} 的 commitlog 边界。`);
    }
    for (const snapshot of snapshots.slice(0, -1)) {
      candidates.push(historyCandidate({
        dataDir: resolvedDataDir,
        absolutePath: join(snapshotsDir, snapshot.name),
        kind: 'snapshot',
        replicaId,
        transaction: snapshot.transaction,
      }));
    }
    for (const segment of segments.filter((item) => item.transaction < boundarySegment.transaction)) {
      candidates.push(historyCandidate({
        dataDir: resolvedDataDir,
        absolutePath: join(clogDir, segment.log),
        kind: 'commitlog',
        replicaId,
        transaction: segment.transaction,
      }));
      if (segment.ofs) {
        candidates.push(historyCandidate({
          dataDir: resolvedDataDir,
          absolutePath: join(clogDir, segment.ofs),
          kind: 'commitlog-offset',
          replicaId,
          transaction: segment.transaction,
        }));
      }
    }
    replicas.push({
      replicaId,
      status: 'ready',
      latestSnapshot: latestSnapshot.toString(),
      boundarySegment: boundarySegment.transaction.toString(),
    });
  }
  candidates.sort((left, right) => left.path.localeCompare(right.path));
  return {
    dataDir: resolvedDataDir,
    replicasDir: assertSafeRelativePath(resolvedDataDir, replicasDir),
    replicas,
    candidates,
    totalSizeBytes: candidates.reduce((sum, item) => sum + BigInt(item.sizeBytes), 0n).toString(),
  };
}

function runCommand(command, args, options = {}) {
  const result = spawnSync(command, args, {
    cwd: options.cwd ?? REPO_ROOT,
    env: options.env ?? process.env,
    encoding: 'utf8',
    stdio: options.stdio ?? 'pipe',
    shell: process.platform === 'win32',
  });
  if (result.error) {
    throw new Error(`${command} 启动失败: ${result.error.message}`);
  }
  if (result.status !== 0) {
    const output = `${result.stdout ?? ''}\n${result.stderr ?? ''}`.trim();
    throw new Error(`${command} 退出码 ${result.status}: ${output}`);
  }
  return result;
}

function parseByteSize(rawValue, label) {
  const value = String(rawValue ?? '').trim();
  if (!value) {
    return null;
  }
  const match = /^(\d+)(?:\s*([KMGTPE]?)(?:I?B?)?)?$/iu.exec(value);
  if (!match) {
    throw new Error(`${label} 必须是字节数或 K/M/G/T/P/E 后缀大小，实际: ${rawValue}`);
  }
  const [, amountText, unitText = ''] = match;
  const multipliers = {
    '': 1n,
    K: 1024n,
    M: 1024n ** 2n,
    G: 1024n ** 3n,
    T: 1024n ** 4n,
    P: 1024n ** 5n,
    E: 1024n ** 6n,
  };
  return BigInt(amountText) * multipliers[unitText.toUpperCase()];
}

function formatBytes(bytes) {
  const value = BigInt(bytes);
  const gib = Number(value) / (1024 ** 3);
  if (gib >= 1) {
    return `${gib.toFixed(1)}GiB`;
  }
  const mib = Number(value) / (1024 ** 2);
  if (mib >= 1) {
    return `${mib.toFixed(1)}MiB`;
  }
  return `${value}B`;
}

function getDirectorySizeBytes(dataDir) {
  const result = runCommand('du', ['-sk', dataDir]);
  const [sizeKbText] = String(result.stdout ?? '').trim().split(/\s+/u);
  if (!sizeKbText || !/^\d+$/u.test(sizeKbText)) {
    throw new Error(`无法解析数据目录大小: ${result.stdout}`);
  }
  return BigInt(sizeKbText) * 1024n;
}

function getAvailableBytes(fileSystemPath) {
  const stat = statfsSync(fileSystemPath, {bigint: true});
  return stat.bavail * stat.bsize;
}

function parseSafetyRatio(rawValue) {
  const value = String(rawValue ?? '').trim();
  if (!value) {
    return DEFAULT_SPACE_SAFETY_RATIO;
  }
  const ratio = Number(value);
  if (!Number.isFinite(ratio) || ratio < 1) {
    throw new Error(`GENARRATIVE_DATABASE_BACKUP_SPACE_SAFETY_RATIO 必须是 >= 1 的数字，实际: ${rawValue}`);
  }
  return ratio;
}

function calculateRequiredFreeBytes({dataSizeBytes, args, env}) {
  const explicitMinFreeBytes = parseByteSize(
    firstNonEmpty(args.minFreeBytes, env.GENARRATIVE_DATABASE_BACKUP_MIN_FREE_BYTES),
    'GENARRATIVE_DATABASE_BACKUP_MIN_FREE_BYTES',
  );
  if (explicitMinFreeBytes !== null) {
    return explicitMinFreeBytes;
  }
  const ratio = parseSafetyRatio(env.GENARRATIVE_DATABASE_BACKUP_SPACE_SAFETY_RATIO);
  const ratioBasisPoints = BigInt(Math.ceil(ratio * 10000));
  const ratioRequirement = (dataSizeBytes * ratioBasisPoints + 9999n) / 10000n;
  const extraFreeBytes = parseByteSize(
    firstNonEmpty(env.GENARRATIVE_DATABASE_BACKUP_EXTRA_FREE_BYTES, String(DEFAULT_EXTRA_FREE_BYTES)),
    'GENARRATIVE_DATABASE_BACKUP_EXTRA_FREE_BYTES',
  );
  const extraRequirement = dataSizeBytes + extraFreeBytes;
  return ratioRequirement > extraRequirement ? ratioRequirement : extraRequirement;
}

function assertSufficientWorkDirSpace({dataDir, workDir, args, env}) {
  mkdirSync(workDir, {recursive: true});
  const dataSizeBytes = getDirectorySizeBytes(dataDir);
  const availableBytes = getAvailableBytes(workDir);
  const requiredFreeBytes = calculateRequiredFreeBytes({dataSizeBytes, args, env});
  console.log(
    `[database-backup] 备份空间预检: data=${formatBytes(dataSizeBytes)}, available=${formatBytes(availableBytes)}, required=${formatBytes(requiredFreeBytes)}`,
  );
  if (availableBytes < requiredFreeBytes) {
    throw new Error(
      [
        `备份工作目录所在文件系统剩余空间不足: ${workDir}`,
        `available=${formatBytes(availableBytes)}`,
        `required=${formatBytes(requiredFreeBytes)}`,
        `dataDir=${dataDir}`,
        `dataSize=${formatBytes(dataSizeBytes)}`,
        '为避免停库后写满磁盘，本次备份已在停止服务前中止。',
      ].join('；'),
    );
  }
}

function assertSufficientHistoryWorkDirSpace({historySizeBytes, workDir, args, env}) {
  mkdirSync(workDir, {recursive: true});
  const availableBytes = getAvailableBytes(workDir);
  const requiredFreeBytes = calculateRequiredFreeBytes({dataSizeBytes: BigInt(historySizeBytes), args, env});
  console.log(
    `[database-backup] history 空间预检: candidates=${formatBytes(historySizeBytes)}, available=${formatBytes(availableBytes)}, required=${formatBytes(requiredFreeBytes)}`,
  );
  if (availableBytes < requiredFreeBytes) {
    throw new Error(`history 工作目录剩余空间不足: available=${formatBytes(availableBytes)}；required=${formatBytes(requiredFreeBytes)}`);
  }
}

function collectRestartServicesAfterBackup({args, env}) {
  const serviceNames = [
    ...String(env.GENARRATIVE_DATABASE_BACKUP_RESTART_SERVICE_AFTER ?? '')
      .split(',')
      .map((value) => value.trim())
      .filter(Boolean),
    ...args.restartServicesAfter,
  ];
  return [...new Set(serviceNames.filter(Boolean))];
}

function stopServiceIfNeeded(serviceName) {
  if (!serviceName) {
    return false;
  }
  console.log(`[database-backup] 停止服务以获取冷备份: ${serviceName}`);
  runCommand('systemctl', ['stop', serviceName], {stdio: 'inherit'});
  return true;
}

function startServiceIfNeeded(serviceName, wasStopped) {
  if (!serviceName || !wasStopped) {
    return;
  }
  console.log(`[database-backup] 恢复服务: ${serviceName}`);
  runCommand('systemctl', ['start', serviceName], {stdio: 'inherit'});
}

function restartServicesAfterBackup(serviceNames) {
  const errors = [];
  for (const serviceName of serviceNames) {
    if (!serviceName) {
      continue;
    }
    console.log(`[database-backup] 冷备份后重启依赖服务: ${serviceName}`);
    try {
      runCommand('systemctl', ['restart', serviceName], {stdio: 'inherit'});
    } catch (error) {
      errors.push(error);
    }
  }
  if (errors.length > 0) {
    throw new AggregateError(errors, `冷备份后重启依赖服务失败: ${errors.map((error) => error.message).join('; ')}`);
  }
}

function restoreServicesAfterBackup({stopService, serviceStopped, restartServicesAfter}) {
  const errors = [];
  try {
    startServiceIfNeeded(stopService, serviceStopped);
  } catch (error) {
    errors.push(error);
  }
  try {
    restartServicesAfterBackup(restartServicesAfter);
  } catch (error) {
    errors.push(error);
  }
  if (errors.length > 0) {
    throw new AggregateError(errors, `恢复冷备份相关服务失败: ${errors.map((error) => error.message).join('; ')}`);
  }
}

function createArchive({dataDir, workDir, fileName}) {
  if (!existsSync(dataDir)) {
    throw new Error(`数据库数据目录不存在: ${dataDir}`);
  }
  const stat = statSync(dataDir);
  if (!stat.isDirectory()) {
    throw new Error(`数据库数据路径不是目录: ${dataDir}`);
  }
  mkdirSync(workDir, {recursive: true});
  const archivePath = resolve(workDir, fileName);
  const parentDir = dirname(dataDir);
  const entryName = basename(dataDir);
  console.log(`[database-backup] 打包: ${dataDir} -> ${archivePath}`);
  runCommand('tar', ['-czf', archivePath, '-C', parentDir, entryName], {stdio: 'inherit'});
  verifyArchive(archivePath);
  return archivePath;
}

function verifyArchive(archivePath) {
  console.log(`[database-backup] 校验归档: ${archivePath}`);
  runCommand('tar', ['-tzf', archivePath], {stdio: 'ignore'});
}

function historyBatchId({baselineId, plan}) {
  const identity = plan.candidates.map((candidate) => [
    candidate.path,
    candidate.kind,
    candidate.transaction,
    candidate.fingerprint,
  ].join('\0')).join('\n');
  return sha256Hex(`${baselineId}\0${identity}`).slice(0, 32);
}

function buildHistoryNames({database, objectPrefix, baselineId, batchId}) {
  const databasePart = sanitizeObjectPart(database, 'spacetimedb');
  const prefix = String(objectPrefix || 'database-backups')
    .trim()
    .replace(/^\/+|\/+$/gu, '')
    .split('/')
    .filter(Boolean)
    .map((part) => sanitizeObjectPart(part, 'backup'))
    .join('/');
  const fileName = `${databasePart}-history-${batchId}.tar.gz`;
  return {
    fileName,
    objectKey: [prefix, databasePart, 'history', baselineId, fileName].filter(Boolean).join('/'),
  };
}

function createHistoryArchive({dataDir, workDir, fileName, manifestPath, candidates}) {
  mkdirSync(workDir, {recursive: true});
  const archivePath = resolve(workDir, fileName);
  const candidatePaths = candidates.map((candidate) => candidate.path);
  console.log(`[database-backup] 打包 history: ${candidatePaths.length} 个候选 -> ${archivePath}`);
  runCommand('tar', [
    '-czf',
    archivePath,
    '-C',
    dataDir,
    ...candidatePaths,
    '-C',
    dirname(manifestPath),
    basename(manifestPath),
  ], {stdio: 'inherit'});
  verifyArchive(archivePath);
  return archivePath;
}

function recordHistoryBatch({statePath, state, manifest, uploadResult, manifestUpload, status, cleanedAt = ''}) {
  const batch = {
    batchId: manifest.batchId,
    objectKey: uploadResult.objectKey,
    contentLength: uploadResult.contentLength,
    archiveSha256: uploadResult.archiveSha256,
    verifiedAt: uploadResult.verifiedAt,
    manifestObjectKey: manifestUpload.objectKey,
    manifestContentLength: manifestUpload.contentLength,
    manifestArchiveSha256: manifestUpload.archiveSha256,
    manifestVerifiedAt: manifestUpload.verifiedAt,
    uploadedAt: manifest.uploadedAt,
    status,
    cleanedAt,
    candidates: manifest.candidates,
  };
  const batches = state.batches.filter((item) => item.batchId !== batch.batchId);
  batches.push(batch);
  const nextState = {...state, updatedAt: new Date().toISOString(), batches};
  atomicWriteJson(statePath, nextState);
  return nextState;
}

function candidateKey(candidate) {
  return `${candidate.kind}\0${candidate.path}`;
}

export function cleanupHistoryCandidates({dataDir, candidates}) {
  const currentPlan = discoverHistoryPlan({dataDir});
  const eligible = new Map(currentPlan.candidates.map((candidate) => [candidateKey(candidate), candidate]));
  const existing = [];
  for (const candidate of candidates) {
    const absolutePath = resolve(dataDir, candidate.path);
    assertSafeRelativePath(dataDir, absolutePath);
    if (!existsSync(absolutePath)) {
      continue;
    }
    const current = eligible.get(candidateKey(candidate));
    if (!current) {
      throw new Error(`history 候选已不在当前安全边界内，拒绝删除: ${candidate.path}`);
    }
    const currentStat = statFingerprint(absolutePath);
    if (currentStat.fingerprint !== candidate.fingerprint || currentStat.sizeBytes !== candidate.sizeBytes) {
      throw new Error(`history 候选 stat 漂移，拒绝删除: ${candidate.path}`);
    }
    existing.push({candidate, absolutePath});
  }
  existing.sort((left, right) => {
    const priority = {'commitlog-offset': 0, commitlog: 1, snapshot: 2};
    return (priority[left.candidate.kind] ?? 3) - (priority[right.candidate.kind] ?? 3)
      || left.candidate.path.localeCompare(right.candidate.path);
  });
  for (const {candidate, absolutePath} of existing) {
    rmSync(absolutePath, {recursive: candidate.kind === 'snapshot', force: false});
    console.log(`[database-backup] 已清理 history 源文件: ${candidate.path}`);
  }
  return {deletedCount: existing.length, alreadyMissingCount: candidates.length - existing.length};
}

function writeManifest({manifestPath, payload}) {
  writeFileSync(manifestPath, `${JSON.stringify(payload, null, 2)}\n`, 'utf8');
}

function readManifest(manifestPath) {
  if (!existsSync(manifestPath)) {
    throw new Error(`备份清单不存在: ${manifestPath}`);
  }
  return JSON.parse(readFileSync(manifestPath, 'utf8'));
}

function hmac(key, content, encoding) {
  return createHmac('sha256', key).update(content).digest(encoding);
}

function sha256Hex(content) {
  return createHash('sha256').update(content).digest('hex');
}

async function sha256FileHex(filePath) {
  const hash = createHash('sha256');
  for await (const chunk of createReadStream(filePath)) {
    hash.update(chunk);
  }
  return hash.digest('hex');
}

function directFilesStatePath({workDir, database}) {
  return join(workDir, `${sanitizeObjectPart(database, 'spacetimedb')}-files-state.json`);
}

function normalizeObjectPrefix(objectPrefix, database) {
  const prefix = String(objectPrefix || 'database-backups')
    .trim()
    .replace(/^\/+|\/+$/gu, '')
    .split('/')
    .filter(Boolean)
    .map((part) => sanitizeObjectPart(part, 'backup'))
    .join('/');
  return [prefix, sanitizeObjectPart(database, 'spacetimedb')].filter(Boolean).join('/');
}

function directFileIdentity(filePath) {
  const stat = lstatSync(filePath, {bigint: true});
  if (!stat.isFile() || stat.isSymbolicLink()) {
    throw new Error(`files 模式只允许普通文件: ${filePath}`);
  }
  return {
    dev: stat.dev.toString(),
    ino: stat.ino.toString(),
    size: stat.size.toString(),
    mtimeNs: stat.mtimeNs.toString(),
    mode: Number(stat.mode & 0o7777n),
  };
}

function sameDirectFileIdentity(left, right) {
  return left.dev === right.dev
    && left.ino === right.ino
    && left.size === right.size
    && left.mtimeNs === right.mtimeNs
    && left.mode === right.mode;
}

export async function collectDirectFileEntries({dataDir, candidates = null, objectPrefix, database}) {
  const resolvedDataDir = resolvePath(dataDir);
  if (!existsSync(resolvedDataDir) || !lstatSync(resolvedDataDir).isDirectory()) {
    throw new Error(`files 数据目录不存在或不是目录: ${resolvedDataDir}`);
  }
  const files = new Map();
  const symlinks = new Map();
  const directories = new Set(['.']);
  const roots = candidates === null
    ? [{absolutePath: resolvedDataDir, relativePath: '.'}]
    : candidates.map((candidate) => ({
      absolutePath: resolve(resolvedDataDir, candidate.path),
      relativePath: assertSafeRelativePath(resolvedDataDir, resolve(resolvedDataDir, candidate.path)),
    }));

  const visit = async (absolutePath, relativePath) => {
    const stat = lstatSync(absolutePath);
    if (stat.isSymbolicLink()) {
      const target = readlinkSync(absolutePath, 'utf8');
      if (!target || isAbsolute(target)) {
        throw new Error(`files 模式只允许 data-dir 内部的相对符号链接: ${absolutePath} -> ${target}`);
      }
      assertSafeRelativePath(resolvedDataDir, resolve(dirname(absolutePath), target));
      symlinks.set(relativePath, {path: relativePath, target});
      return;
    }
    if (stat.isDirectory()) {
      directories.add(relativePath);
      for (const name of readdirSync(absolutePath).sort()) {
        const childRelative = relativePath === '.' ? name : `${relativePath}/${name}`;
        await visit(join(absolutePath, name), childRelative);
      }
      return;
    }
    if (!stat.isFile()) {
      throw new Error(`files 模式只允许普通文件或目录: ${absolutePath}`);
    }
    const before = directFileIdentity(absolutePath);
    const sha256 = await sha256FileHex(absolutePath);
    const after = directFileIdentity(absolutePath);
    if (!sameDirectFileIdentity(before, after)) {
      throw new Error(`files 扫描期间源文件发生变化: ${relativePath}`);
    }
    const basePrefix = normalizeObjectPrefix(objectPrefix, database);
    files.set(relativePath, {
      path: relativePath,
      sizeBytes: Number(after.size),
      sha256,
      mode: after.mode,
      objectKey: `${basePrefix}/files/sha256/${sha256.slice(0, 2)}/${sha256}`,
      sourceStat: after,
    });
  };

  for (const root of roots.sort((left, right) => left.relativePath.localeCompare(right.relativePath))) {
    if (!existsSync(root.absolutePath)) {
      throw new Error(`files 候选在扫描前消失: ${root.relativePath}`);
    }
    await visit(root.absolutePath, root.relativePath);
  }
  return {
    directories: [...directories].sort(),
    files: [...files.values()].sort((left, right) => left.path.localeCompare(right.path)),
    symlinks: [...symlinks.values()].sort((left, right) => left.path.localeCompare(right.path)),
  };
}

function directCatalogIdentity({mode, baselineCatalogId, rootName, directories, files, symlinks}) {
  return sha256Hex(JSON.stringify({
    mode,
    baselineCatalogId: baselineCatalogId || '',
    rootName,
    directories,
    files: files.map(({path, sizeBytes, sha256, mode, objectKey}) => ({path, sizeBytes, sha256, mode, objectKey})),
    symlinks,
  }));
}

function readDirectFilesState(statePath, {database, bucket}) {
  if (!existsSync(statePath)) {
    return null;
  }
  const state = readManifest(statePath);
  if (
    state.schemaVersion !== DIRECT_FILES_STATE_SCHEMA_VERSION
    || state.backupKind !== 'spacetimedb-direct-files-state'
    || state.database !== database
    || state.bucket !== bucket
  ) {
    throw new Error(`files state 与本次数据源或 bucket 不匹配: ${statePath}`);
  }
  return state;
}

async function ensureDirectObject({
  file,
  dataDir,
  uploadOptions,
  previousFile,
  verifyCatalogReuse = false,
  uploadFn,
  verifyFn,
}) {
  const absolutePath = resolve(dataDir, file.path);
  assertSafeRelativePath(dataDir, absolutePath);
  if (previousFile?.sha256 === file.sha256
    && previousFile?.sizeBytes === file.sizeBytes
    && previousFile?.objectKey === file.objectKey) {
    if (verifyCatalogReuse) {
      await verifyFn({
        ...uploadOptions,
        objectKey: file.objectKey,
        contentLength: file.sizeBytes,
        archiveSha256: file.sha256,
      });
    }
    return {
      status: verifyCatalogReuse ? 'catalog-reused-verified' : 'catalog-reused',
      objectKey: file.objectKey,
    };
  }
  try {
    await verifyFn({
      ...uploadOptions,
      objectKey: file.objectKey,
      contentLength: file.sizeBytes,
      archiveSha256: file.sha256,
    });
    return {status: 'oss-reused', objectKey: file.objectKey};
  } catch (error) {
    if (error?.status !== 404) {
      throw error;
    }
  }
  const beforeUpload = directFileIdentity(absolutePath);
  if (!sameDirectFileIdentity(beforeUpload, file.sourceStat)) {
    throw new Error(`files 上传前源文件 stat 漂移: ${file.path}`);
  }
  await uploadFn({
    archivePath: absolutePath,
    ...uploadOptions,
    objectKey: file.objectKey,
    archiveSha256: file.sha256,
    backupKind: 'spacetimedb-direct-file',
    contentType: 'application/octet-stream',
    allowEmpty: true,
  });
  const afterUpload = directFileIdentity(absolutePath);
  if (!sameDirectFileIdentity(afterUpload, file.sourceStat)) {
    throw new Error(`files 上传期间源文件 stat 漂移: ${file.path}`);
  }
  return {status: 'uploaded', objectKey: file.objectKey};
}

async function ensureDirectManifest({manifestPath, objectKey, uploadOptions, uploadManifestFn, verifyFn}) {
  const body = readFileSync(manifestPath);
  const archiveSha256 = sha256Hex(body);
  try {
    const verification = await verifyFn({
      ...uploadOptions,
      objectKey,
      contentLength: body.length,
      archiveSha256,
    });
    return {objectKey, contentLength: body.length, archiveSha256, verifiedAt: verification.verifiedAt, reused: true};
  } catch (error) {
    if (error?.status !== 404) {
      throw error;
    }
  }
  return uploadManifestFn({manifestPath, ...uploadOptions, objectKey});
}

function directCatalogRef(catalog) {
  return {
    mode: catalog.mode,
    catalogId: catalog.catalogId,
    objectKey: catalog.objectKey,
    contentLength: catalog.contentLength,
    sha256: catalog.sha256,
    verifiedAt: catalog.verifiedAt,
  };
}

function assertDirectCatalogRef(catalog, expectedMode, label) {
  if (
    !catalog
    || catalog.mode !== expectedMode
    || !/^[a-f0-9]{64}$/u.test(catalog.catalogId)
    || typeof catalog.objectKey !== 'string'
    || !catalog.objectKey
    || !Number.isSafeInteger(catalog.contentLength)
    || catalog.contentLength <= 0
    || !/^[a-f0-9]{64}$/u.test(catalog.sha256)
    || typeof catalog.verifiedAt !== 'string'
    || !catalog.verifiedAt
  ) {
    throw new Error(`files ${label} catalog ref 无效。`);
  }
  return directCatalogRef(catalog);
}

function buildDirectFilesLatest({database, bucket, state}) {
  const latestFullCatalog = assertDirectCatalogRef(state?.latestCatalog, 'full', 'latest full');
  const historyCatalogs = (state?.historyCatalogs ?? []).map((catalog) => (
    assertDirectCatalogRef(catalog, 'history', 'history')
  ));
  return {
    schemaVersion: DIRECT_FILES_LATEST_SCHEMA_VERSION,
    backupKind: 'spacetimedb-direct-files-latest',
    database,
    bucket,
    updatedAt: new Date().toISOString(),
    latestFullCatalog,
    historyCatalogs,
  };
}

function validateDirectFilesLatest(latest, {database, bucket}) {
  if (
    latest?.schemaVersion !== DIRECT_FILES_LATEST_SCHEMA_VERSION
    || latest.backupKind !== 'spacetimedb-direct-files-latest'
    || latest.database !== database
    || latest.bucket !== bucket
    || !Array.isArray(latest.historyCatalogs)
  ) {
    throw new Error('files latest pointer 契约无效。');
  }
  return {
    ...latest,
    latestFullCatalog: assertDirectCatalogRef(latest.latestFullCatalog, 'full', 'latest full'),
    historyCatalogs: latest.historyCatalogs.map((catalog) => assertDirectCatalogRef(catalog, 'history', 'history')),
  };
}

async function publishDirectFilesLatest({
  workDir,
  database,
  bucket,
  objectPrefix,
  state,
  uploadOptions,
  uploadManifestFn,
  verifyFn,
}) {
  const latest = buildDirectFilesLatest({database, bucket, state});
  for (const catalogRef of [latest.latestFullCatalog, ...latest.historyCatalogs]) {
    await verifyFn({
      ...uploadOptions,
      objectKey: catalogRef.objectKey,
      contentLength: catalogRef.contentLength,
      archiveSha256: catalogRef.sha256,
    });
  }
  const latestPath = join(workDir, `${sanitizeObjectPart(database, 'spacetimedb')}-latest.json`);
  const latestObjectKey = `${normalizeObjectPrefix(objectPrefix, database)}/latest.json`;
  writeManifest({manifestPath: latestPath, payload: latest});
  const uploaded = await uploadManifestFn({manifestPath: latestPath, ...uploadOptions, objectKey: latestObjectKey});
  const verification = await verifyFn({
    ...uploadOptions,
    objectKey: latestObjectKey,
    contentLength: uploaded.contentLength,
    archiveSha256: uploaded.archiveSha256,
  });
  return {
    latest,
    latestPath,
    latestObjectKey,
    contentLength: uploaded.contentLength,
    sha256: uploaded.archiveSha256,
    verifiedAt: verification.verifiedAt,
  };
}

export async function runDirectFilesBackup({
  mode,
  dataDir,
  workDir,
  database,
  bucket,
  objectPrefix,
  dryRun = false,
  resultFile = '',
  uploadOptions,
  uploadFn = uploadDirectFile,
  uploadManifestFn = uploadManifestFile,
  verifyFn = verifyOssObject,
  concurrency = 1,
}) {
  mkdirSync(workDir, {recursive: true});
  const statePath = directFilesStatePath({workDir, database});
  const state = readDirectFilesState(statePath, {database, bucket});
  if (mode === 'history' && (!state?.baselineCatalog || state?.latestCatalog?.mode !== 'full')) {
    throw new Error(`files history 模式缺少已发布 full baseline catalog: ${statePath}`);
  }
  const plan = mode === 'history' ? discoverHistoryPlan({dataDir}) : null;
  const collected = await collectDirectFileEntries({
    dataDir,
    candidates: plan?.candidates ?? null,
    objectPrefix,
    database,
  });
  const baselineCatalogId = mode === 'history' ? (state?.baselineCatalog?.catalogId ?? '') : '';
  const rootName = basename(dataDir);
  const catalogId = directCatalogIdentity({mode, baselineCatalogId, rootName, ...collected});
  const basePrefix = normalizeObjectPrefix(objectPrefix, database);
  const catalogObjectKey = `${basePrefix}/catalogs/${mode}/${catalogId}.json`;
  const catalogPath = join(workDir, `${sanitizeObjectPart(database, 'spacetimedb')}-${mode}-${catalogId}.catalog.json`);
  const catalog = {
    schemaVersion: DIRECT_FILES_CATALOG_SCHEMA_VERSION,
    backupKind: mode === 'full' ? 'spacetimedb-data-dir-files' : 'spacetimedb-history-files',
    database,
    bucket,
    mode,
    catalogId,
    catalogObjectKey,
    baselineCatalogId,
    rootName,
    directories: collected.directories,
    files: collected.files.map(({sourceStat: _sourceStat, ...file}) => file),
    symlinks: collected.symlinks,
  };
  writeManifest({manifestPath: catalogPath, payload: catalog});
  const summary = {
    statePath,
    catalogPath,
    catalogObjectKey,
    catalogId,
    fileCount: collected.files.length,
    symlinkCount: collected.symlinks.length,
    totalSizeBytes: collected.files.reduce((sum, file) => sum + BigInt(file.sizeBytes), 0n).toString(),
    candidateCount: plan?.candidates.length ?? 0,
  };
  if (resultFile) {
    atomicWriteJson(resolvePath(resultFile), {...summary, dryRun});
  }
  console.log(`[database-backup] files ${mode}: files=${summary.fileCount}, symlinks=${summary.symlinkCount}, size=${formatBytes(summary.totalSizeBytes)}, catalog=${catalogId}`);
  if (dryRun) {
    console.log('[database-backup] files dry-run，仅扫描并生成本地 catalog，不上传或删除。');
    return {...summary, catalog, uploadedCount: 0, reusedCount: 0};
  }
  if (mode === 'history' && plan.candidates.length === 0) {
    await verifyFn({
      ...uploadOptions,
      objectKey: state.latestCatalog.objectKey,
      contentLength: state.latestCatalog.contentLength,
      archiveSha256: state.latestCatalog.sha256,
    });
    const latestPointer = await publishDirectFilesLatest({
      workDir,
      database,
      bucket,
      objectPrefix,
      state,
      uploadOptions,
      uploadManifestFn,
      verifyFn,
    });
    console.log('[database-backup] files history 没有可归档候选。');
    const emptyResult = {...summary, catalog, latestPointer, uploadedCount: 0, reusedCount: 0, cleanup: null};
    if (resultFile) {
      atomicWriteJson(resolvePath(resultFile), emptyResult);
    }
    return emptyResult;
  }

  if (state?.latestCatalog?.catalogId === catalogId && state.latestCatalog.mode === mode) {
    await verifyFn({...uploadOptions, objectKey: state.latestCatalog.objectKey, contentLength: state.latestCatalog.contentLength, archiveSha256: state.latestCatalog.sha256});
    if (mode === 'full') {
      const latestPointer = await publishDirectFilesLatest({
        workDir,
        database,
        bucket,
        objectPrefix,
        state,
        uploadOptions,
        uploadManifestFn,
        verifyFn,
      });
      console.log('[database-backup] files catalog 未变化，无文件需要上传。');
      return {...summary, catalog, latestPointer, uploadedCount: 0, reusedCount: collected.files.length, unchanged: true};
    }
  }

  if (state?.latestCatalog) {
    await verifyFn({
      ...uploadOptions,
      objectKey: state.latestCatalog.objectKey,
      contentLength: state.latestCatalog.contentLength,
      archiveSha256: state.latestCatalog.sha256,
    });
  }

  const previousFiles = new Map((state?.latestCatalog?.files ?? []).map((file) => [file.path, file]));
  let uploadedCount = 0;
  let reusedCount = 0;
  let nextIndex = 0;
  let completedCount = 0;
  const workerCount = Math.min(concurrency, collected.files.length);
  const workers = Array.from({length: workerCount}, async () => {
    while (nextIndex < collected.files.length) {
      const index = nextIndex;
      nextIndex += 1;
      const file = collected.files[index];
      const result = await ensureDirectObject({
        file,
        dataDir,
        uploadOptions,
        previousFile: previousFiles.get(file.path),
        verifyCatalogReuse: mode === 'history',
        uploadFn,
        verifyFn,
      });
      if (result.status === 'uploaded') {
        uploadedCount += 1;
      } else {
        reusedCount += 1;
      }
      completedCount += 1;
      if (collected.files.length <= 100 || completedCount % 1000 === 0 || completedCount === collected.files.length) {
        console.log(`[database-backup] files 进度: ${completedCount}/${collected.files.length} (${result.status}) ${file.path}`);
      }
    }
  });
  await Promise.all(workers);
  const catalogUpload = await ensureDirectManifest({
    manifestPath: catalogPath,
    objectKey: catalogObjectKey,
    uploadOptions,
    uploadManifestFn,
    verifyFn,
  });
  await verifyFn({...uploadOptions, objectKey: catalogObjectKey, contentLength: catalogUpload.contentLength, archiveSha256: catalogUpload.archiveSha256});

  if (mode === 'history') {
    await verifyFn({
      ...uploadOptions,
      objectKey: state.baselineCatalog.objectKey,
      contentLength: state.baselineCatalog.contentLength,
      archiveSha256: state.baselineCatalog.sha256,
    });
  }
  const catalogRef = {
    mode,
    catalogId,
    objectKey: catalogObjectKey,
    contentLength: catalogUpload.contentLength,
    sha256: catalogUpload.archiveSha256,
    verifiedAt: catalogUpload.verifiedAt,
    files: catalog.files,
    symlinks: catalog.symlinks,
  };
  const nextState = {
    schemaVersion: DIRECT_FILES_STATE_SCHEMA_VERSION,
    backupKind: 'spacetimedb-direct-files-state',
    database,
    dataDir,
    bucket,
    updatedAt: new Date().toISOString(),
    baselineCatalog: state?.baselineCatalog ?? catalogRef,
    latestCatalog: mode === 'full' ? catalogRef : state.latestCatalog,
    historyCatalogs: mode === 'history'
      ? [...(state.historyCatalogs ?? []).filter((item) => item.catalogId !== catalogId), catalogRef]
      : (state?.historyCatalogs ?? []),
  };
  const latestPointer = await publishDirectFilesLatest({
    workDir,
    database,
    bucket,
    objectPrefix,
    state: nextState,
    uploadOptions,
    uploadManifestFn,
    verifyFn,
  });
  atomicWriteJson(statePath, nextState);
  let cleanup = null;
  if (mode === 'history') {
    cleanup = cleanupHistoryCandidates({dataDir, candidates: plan.candidates});
  }
  const finalResult = {...summary, catalog, latestPointer, uploadedCount, reusedCount, cleanup};
  if (resultFile) {
    atomicWriteJson(resolvePath(resultFile), finalResult);
  }
  return finalResult;
}

async function downloadOssBuffer({objectKey, uploadOptions}) {
  const response = await signedOssRequest({
    ...ossRequestDefaults(uploadOptions),
    method: 'GET',
    objectKey,
    operation: '下载对象',
  });
  return Buffer.from(await response.arrayBuffer());
}

async function downloadOssFile({objectKey, destinationPath, uploadOptions}) {
  const response = await signedOssRequest({
    ...ossRequestDefaults(uploadOptions),
    method: 'GET',
    objectKey,
    operation: '下载对象',
  });
  const tempPath = `${destinationPath}.partial-${process.pid}`;
  rmSync(tempPath, {force: true});
  try {
    if (response.body) {
      await pipeline(Readable.fromWeb(response.body), createWriteStream(tempPath, {mode: 0o600}));
    } else {
      writeFileSync(tempPath, Buffer.alloc(0), {mode: 0o600});
    }
    renameSync(tempPath, destinationPath);
  } catch (error) {
    rmSync(tempPath, {force: true});
    throw error;
  }
}

async function loadDirectFilesCatalog({catalogRef, database, bucket, uploadOptions, downloadBufferFn}) {
  const catalogBody = await downloadBufferFn({objectKey: catalogRef.objectKey, uploadOptions});
  if (catalogBody.length !== catalogRef.contentLength || sha256Hex(catalogBody) !== catalogRef.sha256) {
    throw new Error(`files restore catalog 长度或 SHA-256 不一致: ${catalogRef.objectKey}`);
  }
  const catalog = JSON.parse(catalogBody.toString('utf8'));
  if (
    catalog.schemaVersion !== DIRECT_FILES_CATALOG_SCHEMA_VERSION
    || catalog.backupKind !== 'spacetimedb-data-dir-files'
    || catalog.database !== database
    || catalog.bucket !== bucket
    || catalog.catalogId !== catalogRef.catalogId
    || !Array.isArray(catalog.directories)
    || !Array.isArray(catalog.files)
  ) {
    throw new Error(`files restore catalog 契约无效: ${catalogRef.objectKey}`);
  }
  return {...catalog, symlinks: catalog.symlinks ?? []};
}

function assertDirectCatalogFile(file, index) {
  if (
    !file
    || typeof file.path !== 'string'
    || !Number.isSafeInteger(file.sizeBytes)
    || file.sizeBytes < 0
    || !/^[a-f0-9]{64}$/u.test(file.sha256)
    || typeof file.objectKey !== 'string'
    || !file.objectKey
    || !Number.isSafeInteger(file.mode)
  ) {
    throw new Error(`files restore catalog 文件项无效: index=${index}`);
  }
}

function assertDirectCatalogSymlink(symlink, index, restoreDir) {
  if (
    !symlink
    || typeof symlink.path !== 'string'
    || !symlink.path
    || typeof symlink.target !== 'string'
    || !symlink.target
    || isAbsolute(symlink.target)
  ) {
    throw new Error(`files restore catalog 符号链接项无效: index=${index}`);
  }
  const destinationPath = resolve(restoreDir, symlink.path);
  assertSafeRelativePath(restoreDir, destinationPath);
  assertSafeRelativePath(restoreDir, resolve(dirname(destinationPath), symlink.target));
}

async function restoreDirectFilesCatalog({
  catalog,
  restoreDir,
  uploadOptions,
  resultFile = '',
  dryRun = false,
  downloadFileFn = downloadOssFile,
}) {
  const resolvedRestoreDir = resolvePath(restoreDir);
  catalog.files.forEach(assertDirectCatalogFile);
  catalog.symlinks.forEach((symlink, index) => assertDirectCatalogSymlink(symlink, index, resolvedRestoreDir));
  const totalSizeBytes = catalog.files.reduce((sum, file) => sum + BigInt(file.sizeBytes), 0n).toString();
  if (dryRun) {
    const result = {
      restoreDir: resolvedRestoreDir,
      catalogId: catalog.catalogId,
      fileCount: catalog.files.length,
      symlinkCount: catalog.symlinks.length,
      totalSizeBytes,
      downloadedCount: 0,
      reusedCount: 0,
      dryRun: true,
    };
    if (resultFile) {
      atomicWriteJson(resolvePath(resultFile), result);
    }
    return result;
  }
  mkdirSync(resolvedRestoreDir, {recursive: true, mode: 0o700});
  for (const directoryPath of catalog.directories) {
    if (directoryPath === '.') {
      continue;
    }
    const absolutePath = resolve(resolvedRestoreDir, directoryPath);
    assertSafeRelativePath(resolvedRestoreDir, absolutePath);
    mkdirSync(absolutePath, {recursive: true});
  }

  let downloadedCount = 0;
  let reusedCount = 0;
  for (const [index, file] of catalog.files.entries()) {
    const destinationPath = resolve(resolvedRestoreDir, file.path);
    assertSafeRelativePath(resolvedRestoreDir, destinationPath);
    mkdirSync(dirname(destinationPath), {recursive: true});
    let reusable = false;
    if (existsSync(destinationPath) && lstatSync(destinationPath).isFile()) {
      const stat = statSync(destinationPath);
      reusable = stat.size === file.sizeBytes && await sha256FileHex(destinationPath) === file.sha256;
    }
    if (reusable) {
      reusedCount += 1;
    } else {
      rmSync(destinationPath, {force: true});
      await downloadFileFn({objectKey: file.objectKey, destinationPath, uploadOptions});
      const stat = statSync(destinationPath);
      const sha256 = await sha256FileHex(destinationPath);
      if (stat.size !== file.sizeBytes || sha256 !== file.sha256) {
        rmSync(destinationPath, {force: true});
        throw new Error(`files restore 对象长度或 SHA-256 不一致: ${file.path}`);
      }
      downloadedCount += 1;
    }
    chmodSync(destinationPath, file.mode & 0o7777);
    console.log(`[database-backup] files restore: ${index + 1}/${catalog.files.length} (${reusable ? 'reused' : 'downloaded'}) ${file.path}`);
  }
  for (const symlink of catalog.symlinks) {
    const destinationPath = resolve(resolvedRestoreDir, symlink.path);
    assertSafeRelativePath(resolvedRestoreDir, destinationPath);
    mkdirSync(dirname(destinationPath), {recursive: true});
    rmSync(destinationPath, {recursive: true, force: true});
    symlinkSync(symlink.target, destinationPath);
    console.log(`[database-backup] files restore: symlink ${symlink.path} -> ${symlink.target}`);
  }
  const result = {
    restoreDir: resolvedRestoreDir,
    catalogId: catalog.catalogId,
    fileCount: catalog.files.length,
    symlinkCount: catalog.symlinks.length,
    totalSizeBytes,
    downloadedCount,
    reusedCount,
  };
  if (resultFile) {
    atomicWriteJson(resolvePath(resultFile), result);
  }
  return result;
}

export async function restoreDirectFilesBackup({
  statePath,
  restoreDir,
  database,
  bucket,
  uploadOptions,
  resultFile = '',
  dryRun = false,
  downloadBufferFn = downloadOssBuffer,
  downloadFileFn = downloadOssFile,
}) {
  const state = readDirectFilesState(resolvePath(statePath), {database, bucket});
  if (!state?.latestCatalog || state.latestCatalog.mode !== 'full') {
    throw new Error(`files restore 缺少 full baseline catalog: ${statePath}`);
  }
  const catalogRef = assertDirectCatalogRef(state.latestCatalog, 'full', 'latest full');
  const catalog = await loadDirectFilesCatalog({catalogRef, database, bucket, uploadOptions, downloadBufferFn});
  return restoreDirectFilesCatalog({
    catalog,
    restoreDir,
    uploadOptions,
    resultFile,
    dryRun,
    downloadFileFn,
  });
}

export async function restoreDirectFilesLatest({
  restoreDir,
  database,
  bucket,
  objectPrefix,
  uploadOptions,
  resultFile = '',
  dryRun = false,
  downloadBufferFn = downloadOssBuffer,
  downloadFileFn = downloadOssFile,
  verifyFn = verifyOssObject,
}) {
  const latestObjectKey = `${normalizeObjectPrefix(objectPrefix, database)}/latest.json`;
  const latestBody = await downloadBufferFn({objectKey: latestObjectKey, uploadOptions});
  const latestSha256 = sha256Hex(latestBody);
  await verifyFn({
    ...uploadOptions,
    objectKey: latestObjectKey,
    contentLength: latestBody.length,
    archiveSha256: latestSha256,
  });
  const latest = validateDirectFilesLatest(JSON.parse(latestBody.toString('utf8')), {database, bucket});
  const catalogRef = latest.latestFullCatalog;
  await verifyFn({
    ...uploadOptions,
    objectKey: catalogRef.objectKey,
    contentLength: catalogRef.contentLength,
    archiveSha256: catalogRef.sha256,
  });
  const catalog = await loadDirectFilesCatalog({catalogRef, database, bucket, uploadOptions, downloadBufferFn});
  return restoreDirectFilesCatalog({
    catalog,
    restoreDir,
    uploadOptions,
    resultFile,
    dryRun,
    downloadFileFn,
  });
}

function regionFromEndpoint(endpoint) {
  const match = /^oss-([a-z0-9-]+)\./u.exec(endpoint);
  if (!match) {
    throw new Error(`无法从 OSS endpoint 推断 region: ${endpoint}`);
  }
  return match[1];
}

function formatScopeDate(date) {
  return timestampForFile(date).slice(0, 8);
}

function formatOssDate(date) {
  return timestampForFile(date).replace(/[-:]/gu, '');
}

function encodePath(path) {
  return path
    .split('/')
    .map((segment) => encodeURIComponent(segment).replace(/[!'()*]/gu, (char) => `%${char.charCodeAt(0).toString(16).toUpperCase()}`))
    .join('/');
}

function encodeQueryComponent(value) {
  return encodeURIComponent(String(value)).replace(
    /[!'()*]/gu,
    (char) => `%${char.charCodeAt(0).toString(16).toUpperCase()}`,
  );
}

export function buildCanonicalQuery(queries = {}) {
  return Object.entries(queries)
    .map(([key, value]) => [encodeQueryComponent(key), value === null ? null : encodeQueryComponent(value)])
    .sort(([leftKey, leftValue], [rightKey, rightValue]) => {
      if (leftKey !== rightKey) {
        return leftKey < rightKey ? -1 : 1;
      }
      const left = leftValue ?? '';
      const right = rightValue ?? '';
      return left === right ? 0 : left < right ? -1 : 1;
    })
    .map(([key, value]) => value === null ? key : `${key}=${value}`)
    .join('&');
}

function canonicalHeaderValue(value) {
  return String(value).trim().replace(/\s+/gu, ' ');
}

export function buildAuthorization({method, bucket, endpoint, objectKey, accessKeyId, accessKeySecret, headers, date, queries = {}}) {
  const region = regionFromEndpoint(endpoint);
  const scopeDate = formatScopeDate(date);
  const scope = `${scopeDate}/${region}/${OSS_SERVICE}/${OSS_REQUEST}`;
  const canonicalUri = `/${encodeURIComponent(bucket)}/${encodePath(objectKey)}`;
  const signedHeaders = Object.fromEntries(
    Object.entries(headers).map(([key, value]) => [key.toLowerCase(), canonicalHeaderValue(value)]),
  );
  const canonicalHeaders = Object.entries(signedHeaders)
    .sort(([left], [right]) => left.localeCompare(right))
    .map(([key, value]) => `${key}:${value}\n`)
    .join('');
  const additionalHeaders = 'host';
  const canonicalRequest = [
    method,
    canonicalUri,
    buildCanonicalQuery(queries),
    canonicalHeaders,
    additionalHeaders,
    UNSIGNED_PAYLOAD,
  ].join('\n');
  const stringToSign = [OSS_ALGORITHM, headers['x-oss-date'], scope, sha256Hex(canonicalRequest)].join('\n');
  const signature = hmac(Buffer.from(`aliyun_v4${accessKeySecret}`, 'utf8'), scopeDate);
  const regionKey = hmac(signature, region);
  const serviceKey = hmac(regionKey, OSS_SERVICE);
  const signingKey = hmac(serviceKey, OSS_REQUEST);
  const finalSignature = hmac(signingKey, stringToSign, 'hex');
  return `${OSS_ALGORITHM} Credential=${accessKeyId}/${scope},AdditionalHeaders=${additionalHeaders},Signature=${finalSignature}`;
}

function buildOssUrl({bucket, endpoint, objectKey, queries = {}}) {
  const canonicalQuery = buildCanonicalQuery(queries);
  return `https://${bucket}.${endpoint}/${encodePath(objectKey)}${canonicalQuery ? `?${canonicalQuery}` : ''}`;
}

function isRetryableOssStatus(status) {
  return RETRYABLE_OSS_HTTP_STATUSES.has(status);
}

function retryDelayMs({attempt, baseDelayMs, maxDelayMs, randomFn}) {
  const ceiling = Math.min(maxDelayMs, baseDelayMs * (2 ** Math.max(0, attempt - 1)));
  return Math.floor(randomFn() * ceiling);
}

function ossRequestDefaults({
  bucket,
  endpoint,
  accessKeyId,
  accessKeySecret,
  fetchImpl = globalThis.fetch,
  nowFn = () => new Date(),
  sleepImpl = sleep,
  randomFn = Math.random,
  maxAttempts = DEFAULT_OSS_REQUEST_MAX_ATTEMPTS,
  retryBaseDelayMs = DEFAULT_OSS_RETRY_BASE_DELAY_MS,
  retryMaxDelayMs = DEFAULT_OSS_RETRY_MAX_DELAY_MS,
}) {
  return {
    bucket,
    endpoint,
    accessKeyId,
    accessKeySecret,
    fetchImpl,
    nowFn,
    sleepImpl,
    randomFn,
    maxAttempts,
    retryBaseDelayMs,
    retryMaxDelayMs,
  };
}

async function signedOssRequest({
  method,
  bucket,
  endpoint,
  objectKey,
  accessKeyId,
  accessKeySecret,
  queries = {},
  headers = {},
  bodyFactory,
  contentLength,
  operation,
  fetchImpl,
  nowFn,
  sleepImpl,
  randomFn,
  maxAttempts,
  retryBaseDelayMs,
  retryMaxDelayMs,
}) {
  const targetUrl = buildOssUrl({bucket, endpoint, objectKey, queries});
  let lastError = null;

  for (let attempt = 1; attempt <= maxAttempts; attempt += 1) {
    const now = nowFn();
    const signedHeaders = {
      host: `${bucket}.${endpoint}`,
      ...headers,
      'x-oss-content-sha256': UNSIGNED_PAYLOAD,
      'x-oss-date': formatOssDate(now),
    };
    const authorization = buildAuthorization({
      method,
      bucket,
      endpoint,
      objectKey,
      accessKeyId,
      accessKeySecret,
      headers: signedHeaders,
      date: now,
      queries,
    });
    const requestHeaders = {...signedHeaders, authorization};
    if (contentLength !== undefined) {
      requestHeaders['content-length'] = String(contentLength);
    }
    const body = bodyFactory ? bodyFactory() : undefined;
    const requestOptions = {method, headers: requestHeaders};
    if (body !== undefined) {
      requestOptions.body = body;
      requestOptions.duplex = 'half';
    }

    let response;
    try {
      response = await fetchImpl(targetUrl, requestOptions);
    } catch (error) {
      lastError = new Error(`OSS ${operation}请求失败: oss://${bucket}/${objectKey}`, {cause: error});
    }

    if (response?.ok) {
      return response;
    }
    if (response) {
      const responseText = await response.text();
      const requestId = response.headers.get('x-oss-request-id');
      lastError = new Error(
        `OSS ${operation}失败 HTTP ${response.status}${requestId ? ` requestId=${requestId}` : ''}: ${responseText.slice(0, 500)}`,
      );
      lastError.status = response.status;
    }

    const retryable = !response || isRetryableOssStatus(response.status);
    if (!retryable || attempt >= maxAttempts) {
      throw lastError;
    }
    const delayMs = retryDelayMs({attempt, baseDelayMs: retryBaseDelayMs, maxDelayMs: retryMaxDelayMs, randomFn});
    console.warn(`[database-backup] OSS ${operation}失败，${delayMs}ms 后重试 (${attempt}/${maxAttempts})`);
    await sleepImpl(delayMs);
  }

  throw lastError ?? new Error(`OSS ${operation}失败: oss://${bucket}/${objectKey}`);
}

function readXmlTag(xml, tagName) {
  const match = new RegExp(`<${tagName}>([\\s\\S]*?)</${tagName}>`, 'u').exec(xml);
  if (!match) {
    return '';
  }
  return match[1]
    .replace(/&lt;/gu, '<')
    .replace(/&gt;/gu, '>')
    .replace(/&quot;/gu, '"')
    .replace(/&apos;/gu, "'")
    .replace(/&amp;/gu, '&')
    .trim();
}

function escapeXml(value) {
  return String(value)
    .replace(/&/gu, '&amp;')
    .replace(/</gu, '&lt;')
    .replace(/>/gu, '&gt;')
    .replace(/"/gu, '&quot;')
    .replace(/'/gu, '&apos;');
}

function buildCompleteMultipartBody(parts) {
  const partXml = parts
    .map(({partNumber, etag}) => [
      '<Part>',
      `<PartNumber>${partNumber}</PartNumber>`,
      `<ETag>${escapeXml(etag)}</ETag>`,
      '</Part>',
    ].join(''))
    .join('');
  return `<?xml version="1.0" encoding="UTF-8"?><CompleteMultipartUpload>${partXml}</CompleteMultipartUpload>`;
}

function resolveMultipartPartSize(fileSize, configuredPartSize) {
  if (!Number.isSafeInteger(configuredPartSize) || configuredPartSize < OSS_MIN_MULTIPART_PART_SIZE_BYTES) {
    throw new Error(`OSS multipart part size 必须是 >= ${OSS_MIN_MULTIPART_PART_SIZE_BYTES} 的安全整数`);
  }
  const minimumForPartLimit = Math.ceil(fileSize / OSS_MAX_MULTIPART_PARTS);
  const partSize = Math.max(configuredPartSize, minimumForPartLimit);
  if (partSize > OSS_MAX_MULTIPART_PART_SIZE_BYTES) {
    throw new Error(`OSS multipart part size 超过 5GiB: ${partSize}`);
  }
  return partSize;
}

async function verifyUploadedObject({requestOptions, expectedContentLength, expectedArchiveSha256}) {
  const response = await signedOssRequest({
    ...requestOptions,
    method: 'HEAD',
    operation: 'HEAD 验证',
  });
  const contentLengthHeader = response.headers.get('content-length');
  const metadataLengthHeader = response.headers.get('x-oss-meta-file-size');
  const effectiveLengthHeader = /^\d+$/u.test(contentLengthHeader ?? '')
    ? contentLengthHeader
    : metadataLengthHeader;
  if (!effectiveLengthHeader || !/^\d+$/u.test(effectiveLengthHeader)) {
    throw new Error(
      `OSS HEAD 验证缺少有效 content-length/file-size: content-length=${contentLengthHeader ?? '<missing>'}, file-size=${metadataLengthHeader ?? '<missing>'}`,
    );
  }
  const remoteContentLength = Number(effectiveLengthHeader);
  if (remoteContentLength !== expectedContentLength) {
    throw new Error(`OSS HEAD 验证长度不一致: local=${expectedContentLength}, remote=${remoteContentLength}`);
  }
  const remoteArchiveSha256 = String(
    response.headers.get('x-oss-meta-file-sha256')
    ?? response.headers.get('x-oss-meta-archive-sha256')
    ?? '',
  ).trim().toLowerCase();
  if (remoteArchiveSha256 !== expectedArchiveSha256) {
    throw new Error(`OSS HEAD 验证 SHA-256 不一致: local=${expectedArchiveSha256}, remote=${remoteArchiveSha256 || '<missing>'}`);
  }
  return {verifiedAt: new Date().toISOString(), remoteContentLength, remoteArchiveSha256};
}

export async function verifyOssObject({
  bucket,
  endpoint,
  objectKey,
  accessKeyId,
  accessKeySecret,
  contentLength,
  archiveSha256,
  fetchImpl = globalThis.fetch,
  nowFn = () => new Date(),
  sleepImpl = sleep,
  randomFn = Math.random,
  maxAttempts = DEFAULT_OSS_REQUEST_MAX_ATTEMPTS,
}) {
  const requestOptions = {
    bucket,
    endpoint,
    objectKey,
    accessKeyId,
    accessKeySecret,
    fetchImpl,
    nowFn,
    sleepImpl,
    randomFn,
    maxAttempts,
    retryBaseDelayMs: DEFAULT_OSS_RETRY_BASE_DELAY_MS,
    retryMaxDelayMs: DEFAULT_OSS_RETRY_MAX_DELAY_MS,
  };
  return verifyUploadedObject({
    requestOptions,
    expectedContentLength: Number(contentLength),
    expectedArchiveSha256: String(archiveSha256 ?? '').trim().toLowerCase(),
  });
}

async function abortMultipartUpload({requestOptions, uploadId}) {
  try {
    await signedOssRequest({
      ...requestOptions,
      method: 'DELETE',
      queries: {uploadId},
      operation: 'AbortMultipartUpload',
      maxAttempts: Math.min(2, requestOptions.maxAttempts),
    });
    console.warn(`[database-backup] 已清理失败的 multipart upload: ${uploadId}`);
  } catch (error) {
    console.warn(`[database-backup] 清理 multipart upload 失败: ${error.message}`);
  }
}

export async function uploadArchive({
  archivePath,
  bucket,
  endpoint,
  objectKey,
  accessKeyId,
  accessKeySecret,
  partSizeBytes = DEFAULT_OSS_MULTIPART_PART_SIZE_BYTES,
  maxAttempts = DEFAULT_OSS_REQUEST_MAX_ATTEMPTS,
  retryBaseDelayMs = DEFAULT_OSS_RETRY_BASE_DELAY_MS,
  retryMaxDelayMs = DEFAULT_OSS_RETRY_MAX_DELAY_MS,
  fetchImpl = globalThis.fetch,
  nowFn = () => new Date(),
  sleepImpl = sleep,
  randomFn = Math.random,
  backupKind = 'spacetimedb-data-dir',
  archiveSha256 = '',
  contentType = 'application/gzip',
  allowEmpty = false,
  bandwidthLimiter = null,
}) {
  const fileStat = statSync(archivePath);
  if (!fileStat.isFile() || (!allowEmpty && fileStat.size <= 0)) {
    throw new Error(`待上传备份必须是${allowEmpty ? '' : '非空'}普通文件: ${archivePath}`);
  }
  const verifiedArchiveSha256 = archiveSha256 || await sha256FileHex(archivePath);
  if (!/^[a-f0-9]{64}$/u.test(verifiedArchiveSha256)) {
    throw new Error(`归档 SHA-256 无效: ${verifiedArchiveSha256}`);
  }
  const requestOptions = {
    bucket,
    endpoint,
    objectKey,
    accessKeyId,
    accessKeySecret,
    fetchImpl,
    nowFn,
    sleepImpl,
    randomFn,
    maxAttempts,
    retryBaseDelayMs,
    retryMaxDelayMs,
  };
  if (fileStat.size === 0) {
    await signedOssRequest({
      ...requestOptions,
      method: 'PUT',
      headers: {
        'content-type': contentType,
        'x-oss-meta-archive-sha256': verifiedArchiveSha256,
        'x-oss-meta-file-sha256': verifiedArchiveSha256,
        'x-oss-meta-file-size': '0',
        'x-oss-meta-backup-kind': backupKind,
      },
      contentLength: 0,
      bodyFactory: () => Buffer.alloc(0),
      operation: '上传空文件',
    });
    const verification = await verifyUploadedObject({requestOptions, expectedContentLength: 0, expectedArchiveSha256: verifiedArchiveSha256});
    return {bucket, objectKey, contentLength: 0, archiveSha256: verifiedArchiveSha256, etag: '', uploadMode: 'single', partCount: 1, partSizeBytes: 0, verifiedAt: verification.verifiedAt};
  }
  const partSize = resolveMultipartPartSize(fileStat.size, partSizeBytes);
  const partCount = Math.ceil(fileStat.size / partSize);
  let uploadId = '';
  let uploadCompleted = false;

  console.log(`[database-backup] multipart 上传 OSS: oss://${bucket}/${objectKey} (${partCount} parts)`);
  try {
    const initiateResponse = await signedOssRequest({
      ...requestOptions,
      method: 'POST',
      queries: {uploads: null},
      headers: {
        'content-type': contentType,
        'x-oss-meta-archive-sha256': verifiedArchiveSha256,
        'x-oss-meta-file-sha256': verifiedArchiveSha256,
        'x-oss-meta-file-size': String(fileStat.size),
        'x-oss-meta-backup-kind': backupKind,
      },
      operation: 'InitiateMultipartUpload',
    });
    uploadId = readXmlTag(await initiateResponse.text(), 'UploadId');
    if (!uploadId) {
      throw new Error('OSS InitiateMultipartUpload 响应缺少 UploadId');
    }

    const parts = [];
    for (let partNumber = 1; partNumber <= partCount; partNumber += 1) {
      const start = (partNumber - 1) * partSize;
      const end = Math.min(fileStat.size, start + partSize) - 1;
      const contentLength = end - start + 1;
      const response = await signedOssRequest({
        ...requestOptions,
        method: 'PUT',
        queries: {partNumber, uploadId},
        headers: {'content-type': 'application/octet-stream'},
        contentLength,
        bodyFactory: () => {
          const stream = createReadStream(archivePath, {start, end});
          return bandwidthLimiter ? bandwidthLimiter.wrap(stream) : stream;
        },
        operation: `UploadPart ${partNumber}/${partCount}`,
      });
      const etag = response.headers.get('etag');
      if (!etag) {
        throw new Error(`OSS UploadPart ${partNumber}/${partCount} 响应缺少 ETag`);
      }
      parts.push({partNumber, etag});
      console.log(`[database-backup] multipart 进度: ${partNumber}/${partCount}`);
    }

    const completeBody = buildCompleteMultipartBody(parts);
    let completeResponse;
    try {
      completeResponse = await signedOssRequest({
        ...requestOptions,
        method: 'POST',
        queries: {uploadId},
        headers: {'content-type': 'application/xml'},
        contentLength: Buffer.byteLength(completeBody),
        bodyFactory: () => completeBody,
        operation: 'CompleteMultipartUpload',
      });
      const completeResponseText = await completeResponse.text();
      if (/<Error(?:\s|>)/u.test(completeResponseText)) {
        throw new Error(`OSS CompleteMultipartUpload 返回错误: ${completeResponseText.slice(0, 500)}`);
      }
    } catch (completeError) {
      try {
        await verifyUploadedObject({requestOptions, expectedContentLength: fileStat.size, expectedArchiveSha256: verifiedArchiveSha256});
        completeResponse = null;
      } catch {
        throw completeError;
      }
    }

    const verification = await verifyUploadedObject({
      requestOptions,
      expectedContentLength: fileStat.size,
      expectedArchiveSha256: verifiedArchiveSha256,
    });
    uploadCompleted = true;
    return {
      bucket,
      objectKey,
      contentLength: fileStat.size,
      archiveSha256: verifiedArchiveSha256,
      etag: completeResponse?.headers.get('etag')?.replace(/^"|"$/gu, '') ?? '',
      uploadMode: 'multipart',
      partCount,
      partSizeBytes: partSize,
      verifiedAt: verification.verifiedAt,
    };
  } catch (error) {
    if (uploadId && !uploadCompleted) {
      await abortMultipartUpload({requestOptions, uploadId});
    }
    throw error;
  }
}

export async function uploadDirectFile({
  archivePath,
  bucket,
  endpoint,
  objectKey,
  accessKeyId,
  accessKeySecret,
  fetchImpl = globalThis.fetch,
  nowFn = () => new Date(),
  sleepImpl = sleep,
  randomFn = Math.random,
  maxAttempts = DEFAULT_OSS_REQUEST_MAX_ATTEMPTS,
  retryBaseDelayMs = DEFAULT_OSS_RETRY_BASE_DELAY_MS,
  retryMaxDelayMs = DEFAULT_OSS_RETRY_MAX_DELAY_MS,
  backupKind = 'spacetimedb-direct-file',
  archiveSha256 = '',
  contentType = 'application/octet-stream',
  allowEmpty = true,
  bandwidthLimiter = null,
}) {
  const fileStat = statSync(archivePath);
  if (!fileStat.isFile() || (!allowEmpty && fileStat.size <= 0)) {
    throw new Error(`待上传备份必须是${allowEmpty ? '' : '非空'}普通文件: ${archivePath}`);
  }
  if (fileStat.size > DIRECT_FILES_SINGLE_PUT_MAX_BYTES) {
    return uploadArchive({
      archivePath,
      bucket,
      endpoint,
      objectKey,
      accessKeyId,
      accessKeySecret,
      fetchImpl,
      nowFn,
      sleepImpl,
      randomFn,
      maxAttempts,
      retryBaseDelayMs,
      retryMaxDelayMs,
      backupKind,
      archiveSha256,
      contentType,
      allowEmpty,
      bandwidthLimiter,
    });
  }
  const verifiedArchiveSha256 = archiveSha256 || await sha256FileHex(archivePath);
  if (!/^[a-f0-9]{64}$/u.test(verifiedArchiveSha256)) {
    throw new Error(`归档 SHA-256 无效: ${verifiedArchiveSha256}`);
  }
  const requestOptions = {
    bucket,
    endpoint,
    objectKey,
    accessKeyId,
    accessKeySecret,
    fetchImpl,
    nowFn,
    sleepImpl,
    randomFn,
    maxAttempts,
    retryBaseDelayMs,
    retryMaxDelayMs,
  };
  await signedOssRequest({
    ...requestOptions,
    method: 'PUT',
    headers: {
      'content-type': contentType,
      'x-oss-meta-archive-sha256': verifiedArchiveSha256,
      'x-oss-meta-file-sha256': verifiedArchiveSha256,
      'x-oss-meta-file-size': String(fileStat.size),
      'x-oss-meta-backup-kind': backupKind,
    },
    contentLength: fileStat.size,
    bodyFactory: () => {
      if (fileStat.size === 0) {
        return Buffer.alloc(0);
      }
      const stream = createReadStream(archivePath);
      return bandwidthLimiter ? bandwidthLimiter.wrap(stream) : stream;
    },
    operation: '上传逐文件对象',
  });
  const verification = await verifyUploadedObject({
    requestOptions,
    expectedContentLength: fileStat.size,
    expectedArchiveSha256: verifiedArchiveSha256,
  });
  return {
    bucket,
    objectKey,
    contentLength: fileStat.size,
    archiveSha256: verifiedArchiveSha256,
    etag: '',
    uploadMode: 'single',
    partCount: 1,
    partSizeBytes: fileStat.size,
    verifiedAt: verification.verifiedAt,
  };
}

export async function uploadManifestFile({
  manifestPath,
  bucket,
  endpoint,
  objectKey,
  accessKeyId,
  accessKeySecret,
  fetchImpl = globalThis.fetch,
  nowFn = () => new Date(),
  sleepImpl = sleep,
  randomFn = Math.random,
  maxAttempts = DEFAULT_OSS_REQUEST_MAX_ATTEMPTS,
  bandwidthLimiter = null,
}) {
  const body = readFileSync(manifestPath);
  if (body.length === 0) {
    throw new Error(`待上传 manifest 不能为空: ${manifestPath}`);
  }
  const archiveSha256 = sha256Hex(body);
  const requestOptions = {
    bucket,
    endpoint,
    objectKey,
    accessKeyId,
    accessKeySecret,
    fetchImpl,
    nowFn,
    sleepImpl,
    randomFn,
    maxAttempts,
    retryBaseDelayMs: DEFAULT_OSS_RETRY_BASE_DELAY_MS,
    retryMaxDelayMs: DEFAULT_OSS_RETRY_MAX_DELAY_MS,
  };
  await signedOssRequest({
    ...requestOptions,
    method: 'PUT',
    headers: {
      'content-type': 'application/json',
      'x-oss-meta-archive-sha256': archiveSha256,
      'x-oss-meta-file-size': String(body.length),
      'x-oss-meta-backup-kind': 'spacetimedb-backup-manifest',
    },
    contentLength: body.length,
    bodyFactory: () => {
      const stream = createBufferReadStream(body);
      return bandwidthLimiter ? bandwidthLimiter.wrap(stream) : stream;
    },
    operation: '上传 manifest',
  });
  const verification = await verifyUploadedObject({
    requestOptions,
    expectedContentLength: body.length,
    expectedArchiveSha256: archiveSha256,
  });
  return {objectKey, contentLength: body.length, archiveSha256, verifiedAt: verification.verifiedAt};
}

function uploadedManifestPayload({manifest, database, result}) {
  return {
    ...manifest,
    database,
    bucket: result.bucket,
    objectKey: result.objectKey,
    manifestObjectKey: `${result.objectKey}.manifest.json`,
    contentLength: result.contentLength,
    archiveSha256: result.archiveSha256,
    etag: result.etag,
    uploadMode: result.uploadMode,
    partCount: result.partCount,
    partSizeBytes: result.partSizeBytes,
    verifiedAt: result.verifiedAt,
    uploadedAt: new Date().toISOString(),
    uploadStatus: 'uploaded',
  };
}

export async function uploadHistoryArchiveWithCleanup({
  archivePath,
  manifestPath,
  manifest,
  statePath,
  uploadOptions,
  uploadFn = uploadArchive,
  manifestUploadFn = uploadManifestFile,
  verifyFn = verifyOssObject,
}) {
  const result = await uploadFn({
    archivePath,
    ...uploadOptions,
    backupKind: 'spacetimedb-history',
  });
  const uploadedManifest = uploadedManifestPayload({manifest, database: manifest.database, result});
  writeManifest({manifestPath, payload: uploadedManifest});
  const manifestUpload = await manifestUploadFn({
    manifestPath,
    ...uploadOptions,
    objectKey: uploadedManifest.manifestObjectKey,
  });
  uploadedManifest.manifestVerifiedAt = manifestUpload.verifiedAt;
  uploadedManifest.manifestContentLength = manifestUpload.contentLength;
  uploadedManifest.manifestArchiveSha256 = manifestUpload.archiveSha256;
  writeManifest({manifestPath, payload: uploadedManifest});
  let state = validateHistoryState(readManifest(statePath), {
    database: uploadedManifest.database,
    dataDir: uploadedManifest.dataDir,
  });
  if (state.baseline.id !== uploadedManifest.baselineId) {
    throw new Error(`history manifest baselineId 与 state 不匹配: manifest=${uploadedManifest.baselineId}, state=${state.baseline.id}`);
  }
  await verifyFn({
    ...uploadOptions,
    bucket: state.baseline.bucket,
    objectKey: state.baseline.objectKey,
    contentLength: state.baseline.contentLength,
    archiveSha256: state.baseline.archiveSha256,
  });
  await verifyFn({
    ...uploadOptions,
    bucket: state.baseline.bucket,
    objectKey: state.baseline.manifestObjectKey,
    contentLength: state.baseline.manifestContentLength,
    archiveSha256: state.baseline.manifestArchiveSha256,
  });
  state = recordHistoryBatch({
    statePath,
    state,
    manifest: uploadedManifest,
    uploadResult: result,
    manifestUpload,
    status: 'uploaded',
  });
  const cleanup = cleanupHistoryCandidates({
    dataDir: uploadedManifest.dataDir,
    candidates: uploadedManifest.candidates,
  });
  state = recordHistoryBatch({
    statePath,
    state,
    manifest: uploadedManifest,
    uploadResult: result,
    manifestUpload,
    status: 'cleaned',
    cleanedAt: new Date().toISOString(),
  });
  return {result, uploadedManifest, cleanup, state};
}

async function uploadExistingArchive({args, env, bucket, endpoint, accessKeyId, accessKeySecret, objectPrefix, bandwidthLimiter}) {
  const archivePath = resolvePath(args.uploadArchive);
  if (!existsSync(archivePath)) {
    throw new Error(`待上传备份文件不存在: ${archivePath}`);
  }

  const manifestPath = resolvePath(args.manifestFile || `${archivePath}.manifest.json`);
  const manifest = existsSync(manifestPath) ? readManifest(manifestPath) : {};
  const dataDir = firstNonEmpty(manifest.dataDir, env.GENARRATIVE_DATABASE_BACKUP_DATA_DIR, DEFAULT_PRODUCTION_DATA_DIR);
  const database = firstNonEmpty(args.database, manifest.database, env.GENARRATIVE_SPACETIME_DATABASE, basename(dataDir));
  const objectKey = firstNonEmpty(args.objectKey, manifest.objectKey, buildBackupNames({database, dataDir, objectPrefix}).objectKey);
  if (manifest.backupKind !== 'spacetimedb-history') {
    manifest.backupKind = 'spacetimedb-data-dir';
    manifest.baselineStatePath = firstNonEmpty(
      manifest.baselineStatePath,
      historyStatePath({args, env, workDir: dirname(archivePath), database}),
    );
  }

  console.log(`[database-backup] 上传已有备份: ${archivePath}`);
  console.log(`[database-backup] 目标对象: oss://${bucket}/${objectKey}`);

  if (args.dryRun) {
    console.log('[database-backup] dry-run，仅校验上传配置。');
    return;
  }

  const statePath = resolvePath(firstNonEmpty(
    manifest.baselineStatePath,
    historyStatePath({args, env, workDir: dirname(archivePath), database}),
  ));
  let result;
  let uploadedAt;
  if (manifest.backupKind === 'spacetimedb-history') {
    const historyResult = await uploadHistoryArchiveWithCleanup({
      archivePath,
      manifestPath,
      manifest,
      statePath,
      uploadOptions: {bucket, endpoint, objectKey, accessKeyId, accessKeySecret, bandwidthLimiter},
    });
    result = historyResult.result;
    uploadedAt = historyResult.uploadedManifest.uploadedAt;
    console.log(`[database-backup] history 上传并清理完成: ${JSON.stringify(historyResult.cleanup)}`);
  } else {
    result = await uploadArchive({archivePath, bucket, endpoint, objectKey, accessKeyId, accessKeySecret, bandwidthLimiter});
    const uploadedManifest = uploadedManifestPayload({manifest, database, result});
    uploadedAt = uploadedManifest.uploadedAt;
    writeManifest({manifestPath, payload: uploadedManifest});
    const manifestUpload = await uploadManifestFile({
      manifestPath,
      bucket,
      endpoint,
      objectKey: uploadedManifest.manifestObjectKey,
      accessKeyId,
      accessKeySecret,
      bandwidthLimiter,
    });
    uploadedManifest.manifestVerifiedAt = manifestUpload.verifiedAt;
    uploadedManifest.manifestContentLength = manifestUpload.contentLength;
    uploadedManifest.manifestArchiveSha256 = manifestUpload.archiveSha256;
    writeManifest({manifestPath, payload: uploadedManifest});
    const previousState = existsSync(statePath)
      ? validateHistoryState(readManifest(statePath), {database, dataDir})
      : null;
    const baseline = normalizeUploadedBaselineManifest(uploadedManifest, {database, dataDir});
    writeBaselineState({statePath, baseline, previousState});
    console.log(`[database-backup] 已写入 baseline state: ${statePath}`);
  }
  console.log(`[database-backup] 上传完成: ${JSON.stringify(result)}`);

  if (args.resultFile) {
    writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({archivePath, manifestPath, statePath, ...result, uploadedAt}, null, 2)}\n`, 'utf8');
  }

  const keepLocal = args.keepLocal || String(env.GENARRATIVE_DATABASE_BACKUP_KEEP_LOCAL ?? '').trim().toLowerCase() === 'true';
  if (!keepLocal) {
    rmSync(archivePath, {force: true});
    rmSync(manifestPath, {force: true});
    console.log('[database-backup] 已删除本地临时备份文件；如需保留请设置 --keep-local。');
  } else {
    console.log(`[database-backup] 已保留本地备份: ${archivePath}`);
    console.log(`[database-backup] 已保留备份清单: ${manifestPath}`);
  }
}

async function publishExistingManifest({args, bucket, endpoint, accessKeyId, accessKeySecret, bandwidthLimiter}) {
  const manifestPath = resolvePath(args.publishManifest);
  const manifest = readManifest(manifestPath);
  if (manifest.uploadStatus !== 'uploaded' || !manifest.objectKey) {
    throw new Error('只允许发布 uploadStatus=uploaded 且包含 objectKey 的备份 manifest。');
  }
  manifest.manifestObjectKey = manifest.manifestObjectKey || `${manifest.objectKey}.manifest.json`;
  writeManifest({manifestPath, payload: manifest});
  const result = await uploadManifestFile({
    manifestPath,
    bucket,
    endpoint,
    objectKey: manifest.manifestObjectKey,
    accessKeyId,
    accessKeySecret,
    bandwidthLimiter,
  });
  manifest.manifestVerifiedAt = result.verifiedAt;
  manifest.manifestContentLength = result.contentLength;
  manifest.manifestArchiveSha256 = result.archiveSha256;
  writeManifest({manifestPath, payload: manifest});
  console.log(`[database-backup] manifest 上传并验真完成: ${JSON.stringify(result)}`);
}

export async function resumeUploadedHistoryBatch({statePath, state, dataDir, verificationOptions, verifyFn = verifyOssObject}) {
  const pendingBatch = state.batches.find((batch) => batch.status === 'uploaded');
  if (!pendingBatch) {
    return state;
  }
  console.log(`[database-backup] 重试已上传 history 批次的本地清理: ${pendingBatch.batchId}`);
  await verifyFn({
    ...verificationOptions,
    objectKey: pendingBatch.objectKey,
    contentLength: pendingBatch.contentLength,
    archiveSha256: pendingBatch.archiveSha256,
  });
  await verifyFn({
    ...verificationOptions,
    objectKey: pendingBatch.manifestObjectKey,
    contentLength: pendingBatch.manifestContentLength,
    archiveSha256: pendingBatch.manifestArchiveSha256,
  });
  const cleanup = cleanupHistoryCandidates({dataDir, candidates: pendingBatch.candidates});
  const manifest = {
    batchId: pendingBatch.batchId,
    uploadedAt: pendingBatch.uploadedAt,
    candidates: pendingBatch.candidates,
  };
  const uploadResult = {
    objectKey: pendingBatch.objectKey,
    contentLength: pendingBatch.contentLength,
    archiveSha256: pendingBatch.archiveSha256,
    verifiedAt: pendingBatch.verifiedAt,
  };
  const manifestUpload = {
    objectKey: pendingBatch.manifestObjectKey,
    contentLength: pendingBatch.manifestContentLength,
    archiveSha256: pendingBatch.manifestArchiveSha256,
    verifiedAt: pendingBatch.manifestVerifiedAt,
  };
  const nextState = recordHistoryBatch({
    statePath,
    state,
    manifest,
    uploadResult,
    manifestUpload,
    status: 'cleaned',
    cleanedAt: new Date().toISOString(),
  });
  console.log(`[database-backup] 已完成 history 清理重试: ${JSON.stringify(cleanup)}`);
  return nextState;
}

async function runHistoryBackup({
  args,
  env,
  dataDir,
  workDir,
  database,
  bucket,
  endpoint,
  accessKeyId,
  accessKeySecret,
  objectPrefix,
  keepLocal,
  bandwidthLimiter,
}) {
  const statePath = historyStatePath({args, env, workDir, database});
  let state = loadOrImportHistoryState({args, env, statePath, database, dataDir});
  if (!args.dryRun && !args.deferUpload) {
    console.log(`[database-backup] 重新验真 full baseline: oss://${state.baseline.bucket}/${state.baseline.objectKey}`);
    await verifyOssObject({
      bucket: state.baseline.bucket,
      endpoint,
      objectKey: state.baseline.objectKey,
      accessKeyId,
      accessKeySecret,
      contentLength: state.baseline.contentLength,
      archiveSha256: state.baseline.archiveSha256,
    });
    await verifyOssObject({
      bucket: state.baseline.bucket,
      endpoint,
      objectKey: state.baseline.manifestObjectKey,
      accessKeyId,
      accessKeySecret,
      contentLength: state.baseline.manifestContentLength,
      archiveSha256: state.baseline.manifestArchiveSha256,
    });
    state = await resumeUploadedHistoryBatch({
      statePath,
      state,
      dataDir,
      verificationOptions: {bucket, endpoint, accessKeyId, accessKeySecret},
    });
  }
  const plan = discoverHistoryPlan({dataDir});
  console.log(`[database-backup] history replicas: ${JSON.stringify(plan.replicas)}`);
  console.log(`[database-backup] history 候选: count=${plan.candidates.length}, size=${formatBytes(plan.totalSizeBytes)}`);
  if (args.resultFile) {
    writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({statePath, baseline: state.baseline, ...plan}, null, 2)}\n`, 'utf8');
  }
  if (args.dryRun) {
    console.log('[database-backup] history dry-run，仅输出安全候选，不打包、上传或删除。');
    return;
  }
  if (plan.candidates.length === 0) {
    console.log('[database-backup] 没有可归档的 history 候选。');
    return;
  }

  assertSufficientHistoryWorkDirSpace({historySizeBytes: plan.totalSizeBytes, workDir, args, env});
  const batchId = historyBatchId({baselineId: state.baseline.id, plan});
  const {fileName, objectKey} = buildHistoryNames({
    database,
    objectPrefix,
    baselineId: state.baseline.id,
    batchId,
  });
  const archivePath = resolve(workDir, fileName);
  const manifestPath = `${archivePath}.manifest.json`;
  const manifest = {
    schemaVersion: HISTORY_MANIFEST_SCHEMA_VERSION,
    backupKind: 'spacetimedb-history',
    createdAt: new Date().toISOString(),
    database,
    dataDir,
    bucket,
    objectKey,
    archivePath,
    baselineId: state.baseline.id,
    baselineStatePath: statePath,
    batchId,
    replicas: plan.replicas,
    candidates: plan.candidates,
    totalSizeBytes: plan.totalSizeBytes,
    uploadStatus: args.deferUpload ? 'deferred' : 'pending',
  };
  writeManifest({manifestPath, payload: manifest});
  createHistoryArchive({
    dataDir,
    workDir,
    fileName,
    manifestPath,
    candidates: plan.candidates,
  });

  if (args.deferUpload) {
    console.log(`[database-backup] 已生成 history 归档，延后上传且未清理源文件: ${archivePath}`);
    if (args.resultFile) {
      writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({archivePath, manifestPath, statePath, bucket, objectKey, batchId}, null, 2)}\n`, 'utf8');
    }
    return;
  }

  const historyResult = await uploadHistoryArchiveWithCleanup({
    archivePath,
    manifestPath,
    manifest,
    statePath,
    uploadOptions: {bucket, endpoint, objectKey, accessKeyId, accessKeySecret, bandwidthLimiter},
  });
  console.log(`[database-backup] history 上传并清理完成: ${JSON.stringify(historyResult.cleanup)}`);
  if (args.resultFile) {
    writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({
      archivePath,
      manifestPath,
      statePath,
      batchId,
      ...historyResult.result,
      uploadedAt: historyResult.uploadedManifest.uploadedAt,
    }, null, 2)}\n`, 'utf8');
  }
  if (!keepLocal) {
    rmSync(archivePath, {force: true});
    rmSync(manifestPath, {force: true});
    console.log('[database-backup] 已删除本地 history 临时归档和清单。');
  }
}

async function main() {
  const args = parseArgs(process.argv.slice(2));
  const env = loadEffectiveEnv(args.envFiles);
  const isProductionLike = existsSync(DEFAULT_PRODUCTION_DATA_DIR) && process.platform !== 'win32';
  const dataDir = resolvePath(firstNonEmpty(
    args.dataDir,
    env.GENARRATIVE_DATABASE_BACKUP_DATA_DIR,
    isProductionLike ? DEFAULT_PRODUCTION_DATA_DIR : DEFAULT_LOCAL_DATA_DIR,
  ));
  const workDir = resolvePath(firstNonEmpty(
    args.workDir,
    env.GENARRATIVE_DATABASE_BACKUP_WORK_DIR,
    isProductionLike ? DEFAULT_PRODUCTION_WORK_DIR : DEFAULT_LOCAL_WORK_DIR,
  ));
  const bucket = firstNonEmpty(args.bucket, env.GENARRATIVE_DATABASE_BACKUP_OSS_BUCKET, env.ALIYUN_OSS_BUCKET);
  const endpoint = normalizeEndpoint(firstNonEmpty(args.endpoint, env.GENARRATIVE_DATABASE_BACKUP_OSS_ENDPOINT, env.ALIYUN_OSS_ENDPOINT));
  const accessKeyId = firstNonEmpty(args.accessKeyId, env.GENARRATIVE_DATABASE_BACKUP_OSS_ACCESS_KEY_ID, env.ALIYUN_OSS_ACCESS_KEY_ID);
  const accessKeySecret = firstNonEmpty(args.accessKeySecret, env.GENARRATIVE_DATABASE_BACKUP_OSS_ACCESS_KEY_SECRET, env.ALIYUN_OSS_ACCESS_KEY_SECRET);
  const objectPrefix = firstNonEmpty(args.objectPrefix, env.GENARRATIVE_DATABASE_BACKUP_OSS_PREFIX, 'database-backups');
  const database = firstNonEmpty(args.database, env.GENARRATIVE_SPACETIME_DATABASE, basename(dataDir));
  const keepLocal = args.keepLocal || String(env.GENARRATIVE_DATABASE_BACKUP_KEEP_LOCAL ?? '').trim().toLowerCase() === 'true';
  const storageFormat = firstNonEmpty(args.storageFormat, env.GENARRATIVE_DATABASE_BACKUP_STORAGE_FORMAT, 'archive');
  const directFilesConcurrency = parseDirectFilesConcurrency(env.GENARRATIVE_DATABASE_BACKUP_FILES_CONCURRENCY);
  const uploadBandwidthLimiter = createUploadBandwidthLimiter(env.GENARRATIVE_DATABASE_BACKUP_UPLOAD_MAX_BYTES_PER_SECOND);

  if (!['full', 'history'].includes(args.mode)) {
    throw new Error(`--mode 只能是 full 或 history，实际: ${args.mode}`);
  }
  if (!['archive', 'files'].includes(storageFormat)) {
    throw new Error(`--storage-format 只能是 archive 或 files，实际: ${storageFormat}`);
  }

  for (const [label, value] of Object.entries({bucket, endpoint, accessKeyId, accessKeySecret})) {
    if (!value) {
      throw new Error(`缺少 ${label} 配置`);
    }
  }

  if (args.restoreFilesState && args.restoreFilesLatest) {
    throw new Error('--restore-files-state 与 --restore-files-latest 不能同时使用。');
  }
  if (args.restoreFilesState) {
    if (!args.restoreDir) {
      throw new Error('--restore-files-state 必须同时传 --restore-dir。');
    }
    await restoreDirectFilesBackup({
      statePath: args.restoreFilesState,
      restoreDir: args.restoreDir,
      database,
      bucket,
      uploadOptions: {bucket, endpoint, accessKeyId, accessKeySecret},
      resultFile: args.resultFile,
      dryRun: args.dryRun,
    });
    return;
  }
  if (args.restoreFilesLatest) {
    if (!args.restoreDir) {
      throw new Error('--restore-files-latest 必须同时传 --restore-dir。');
    }
    await restoreDirectFilesLatest({
      restoreDir: args.restoreDir,
      database,
      bucket,
      objectPrefix,
      uploadOptions: {bucket, endpoint, accessKeyId, accessKeySecret},
      resultFile: args.resultFile,
      dryRun: args.dryRun,
    });
    return;
  }
  if (args.restoreDir) {
    throw new Error('--restore-dir 只能与 --restore-files-state 或 --restore-files-latest 一起使用。');
  }

  if (!args.dryRun) {
    const lockPath = acquireBackupLock({workDir, database});
    console.log(`[database-backup] 已获取进程锁: ${lockPath}`);
  }

  if (args.publishManifest) {
    await publishExistingManifest({args, bucket, endpoint, accessKeyId, accessKeySecret, bandwidthLimiter: uploadBandwidthLimiter});
    return;
  }

  if (args.uploadArchive) {
    await uploadExistingArchive({
      args,
      env,
      bucket,
      endpoint,
      accessKeyId,
      accessKeySecret,
      objectPrefix,
      bandwidthLimiter: uploadBandwidthLimiter,
    });
    return;
  }

  if (storageFormat === 'files') {
    if (args.deferUpload) {
      throw new Error('files 模式无需本地归档且不支持 --defer-upload；失败后使用同一 work-dir 重跑即可续传。');
    }
    const stopService = args.stopService || firstNonEmpty(env.GENARRATIVE_DATABASE_BACKUP_STOP_SERVICE);
    const restartServicesAfter = collectRestartServicesAfterBackup({args, env});
    let serviceStopped = false;
    let backupError = null;
    let restoreError = null;
    try {
      if (args.mode === 'full' && !args.dryRun) {
        serviceStopped = stopServiceIfNeeded(stopService);
      }
      await runDirectFilesBackup({
        mode: args.mode,
        dataDir,
        workDir,
        database,
        bucket,
        objectPrefix,
        dryRun: args.dryRun,
        resultFile: args.resultFile,
        uploadOptions: {bucket, endpoint, accessKeyId, accessKeySecret, bandwidthLimiter: uploadBandwidthLimiter},
        concurrency: directFilesConcurrency,
      });
    } catch (error) {
      backupError = error;
    } finally {
      try {
        if (serviceStopped) {
          restoreServicesAfterBackup({stopService, serviceStopped, restartServicesAfter});
        } else if (!backupError && args.mode === 'full' && !args.dryRun) {
          restartServicesAfterBackup(restartServicesAfter);
        }
      } catch (error) {
        restoreError = error;
      }
    }
    if (backupError && restoreError) {
      throw new AggregateError([backupError, restoreError], `files 备份失败，且恢复依赖服务时也失败: ${backupError.message}; ${restoreError.message}`);
    }
    if (backupError) {
      throw backupError;
    }
    if (restoreError) {
      throw restoreError;
    }
    return;
  }

  if (args.mode === 'history') {
    await runHistoryBackup({
      args,
      env,
      dataDir,
      workDir,
      database,
      bucket,
      endpoint,
      accessKeyId,
      accessKeySecret,
      objectPrefix,
      keepLocal,
      bandwidthLimiter: uploadBandwidthLimiter,
    });
    return;
  }

  const {fileName, objectKey} = buildBackupNames({database, dataDir, objectPrefix});
  console.log(`[database-backup] 数据目录: ${dataDir}`);
  console.log(`[database-backup] 本地临时目录: ${workDir}`);
  console.log(`[database-backup] 目标对象: oss://${bucket}/${objectKey}`);

  if (args.dryRun) {
    console.log('[database-backup] dry-run，仅校验配置，不打包上传。');
    return;
  }

  let archivePath = '';
  let serviceStopped = false;
  let backupError = null;
  let restoreError = null;
  const stopService = args.stopService || firstNonEmpty(env.GENARRATIVE_DATABASE_BACKUP_STOP_SERVICE);
  const restartServicesAfter = collectRestartServicesAfterBackup({args, env});
  try {
    assertSufficientWorkDirSpace({dataDir, workDir, args, env});
    serviceStopped = stopServiceIfNeeded(stopService);
    archivePath = createArchive({dataDir, workDir, fileName});
  } catch (error) {
    backupError = error;
  } finally {
    try {
      if (serviceStopped) {
        restoreServicesAfterBackup({stopService, serviceStopped, restartServicesAfter});
      } else if (!backupError) {
        restartServicesAfterBackup(restartServicesAfter);
      }
    } catch (error) {
      restoreError = error;
    }
  }
  if (backupError) {
    if (restoreError) {
      throw new AggregateError([backupError, restoreError], `数据库备份失败，且恢复依赖服务时也失败: ${backupError.message}; ${restoreError.message}`);
    }
    throw backupError;
  }
  if (restoreError) {
    throw restoreError;
  }

  const manifestPath = `${archivePath}.manifest.json`;
  const baselineStatePath = historyStatePath({args, env, workDir, database});
  const fullManifest = {
    backupKind: 'spacetimedb-data-dir',
    createdAt: new Date().toISOString(),
    database,
    dataDir,
    bucket,
    objectKey,
    archivePath,
    baselineStatePath,
    uploadStatus: args.deferUpload ? 'deferred' : 'pending',
  };
  writeManifest({
    manifestPath,
    payload: fullManifest,
  });

  if (args.deferUpload) {
    console.log(`[database-backup] 已生成本地冷备份，延后上传: ${archivePath}`);
    console.log(`[database-backup] 已写入备份清单: ${manifestPath}`);
    if (args.resultFile) {
      writeFileSync(resolvePath(args.resultFile), `${JSON.stringify({archivePath, manifestPath, baselineStatePath, bucket, objectKey}, null, 2)}\n`, 'utf8');
    }
    return;
  }

  const result = await uploadArchive({
    archivePath,
    bucket,
    endpoint,
    objectKey,
    accessKeyId,
    accessKeySecret,
    bandwidthLimiter: uploadBandwidthLimiter,
  });
  console.log(`[database-backup] 上传完成: ${JSON.stringify(result)}`);
  const uploadedManifest = uploadedManifestPayload({manifest: fullManifest, database, result});
  writeManifest({manifestPath, payload: uploadedManifest});
  const manifestUpload = await uploadManifestFile({
    manifestPath,
    bucket,
    endpoint,
    objectKey: uploadedManifest.manifestObjectKey,
    accessKeyId,
    accessKeySecret,
    bandwidthLimiter: uploadBandwidthLimiter,
  });
  uploadedManifest.manifestVerifiedAt = manifestUpload.verifiedAt;
  uploadedManifest.manifestContentLength = manifestUpload.contentLength;
  uploadedManifest.manifestArchiveSha256 = manifestUpload.archiveSha256;
  writeManifest({manifestPath, payload: uploadedManifest});
  const previousState = existsSync(baselineStatePath)
    ? validateHistoryState(readManifest(baselineStatePath), {database, dataDir})
    : null;
  const baseline = normalizeUploadedBaselineManifest(uploadedManifest, {database, dataDir});
  writeBaselineState({statePath: baselineStatePath, baseline, previousState});
  console.log(`[database-backup] 已写入 baseline state: ${baselineStatePath}`);

  if (!keepLocal) {
    rmSync(archivePath, {force: true});
    rmSync(manifestPath, {force: true});
    console.log('[database-backup] 已删除本地临时备份文件；如需保留请设置 --keep-local。');
  } else {
    console.log(`[database-backup] 已保留本地备份: ${archivePath}`);
    console.log(`[database-backup] 已保留备份清单: ${manifestPath}`);
  }
}

function formatErrorDetails(error) {
  if (!error || typeof error !== 'object') {
    return '';
  }
  return ['code', 'errno', 'syscall', 'hostname', 'host', 'port', 'address']
    .map((field) => {
      const value = error[field];
      return value === undefined || value === null || value === '' ? '' : `${field}=${String(value)}`;
    })
    .filter(Boolean)
    .join(' ');
}

function describeError(error) {
  const lines = [];
  let current = error;
  for (let depth = 0; current && depth < 5; depth += 1) {
    const label = depth === 0 ? 'error' : `cause[${depth}]`;
    if (!(current instanceof Error)) {
      lines.push(`${label}: ${String(current)}`);
      break;
    }

    lines.push(`${label}: ${current.name}: ${current.message}`);
    const details = formatErrorDetails(current);
    if (details) {
      lines.push(`${label} details: ${details}`);
    }
    if (current instanceof AggregateError) {
      current.errors.slice(0, 3).forEach((item, index) => {
        const itemText = item instanceof Error ? `${item.name}: ${item.message}` : String(item);
        const itemDetails = formatErrorDetails(item);
        lines.push(`${label}.errors[${index}]: ${itemText}${itemDetails ? ` (${itemDetails})` : ''}`);
      });
    }
    current = current.cause;
  }
  return lines;
}

if (process.argv[1] && realpathSync(resolve(process.argv[1])) === realpathSync(__filename)) {
  main().catch((error) => {
    for (const line of describeError(error)) {
      console.error(`[database-backup] ${line}`);
    }
    process.exit(1);
  });
}