Genarrative/scripts/spacetime-maintain-external-generation-jobs.mjs

#!/usr/bin/env node

import {
  callSpacetimeProcedureViaCli,
  encodeSpacetimeCliOption,
  ensureProcedureOk,
  parsePositiveInteger,
} from './spacetime-migration-common.mjs';

const MAX_BATCH_SIZE = 25;

function usage() {
  return `用法：
  node scripts/spacetime-maintain-external-generation-jobs.mjs --database <name> [选项]

默认只 dry-run 一批历史终态任务 payload 压缩，不修改数据库。

公共选项：
  --database <name>              目标数据库（必填，也可用 GENARRATIVE_SPACETIME_DATABASE）
  --server <name-or-url>         spacetime CLI server 名或 URL
  --server-url <url>             显式 server URL
  --limit <1-${MAX_BATCH_SIZE}>               单批任务数，默认 10
  --cursor-job-id <jobId>        从上一批 next_cursor_job_id 继续
  --apply                        执行写入；省略时始终 dry-run
  --backfill-summaries           改为回填轻量摘要投影
  --owner-user-id <userId>       仅摘要回填可选，限定 owner
  --completed-before-micros <n>  仅 payload 压缩可选，限定终态完成时间
  --help                         显示帮助

必须使用已授权 migration operator 的 spacetime CLI 登录态。脚本每次只处理一批；
根据返回的 next_cursor_job_id 与 has_more 手工继续，避免在生产一次长事务扫完整历史。`;
}

function parseOptions(argv) {
  const options = {
    apply: false,
    backfillSummaries: false,
    completedBeforeMicros: null,
    cursorJobId: '',
    database: process.env.GENARRATIVE_SPACETIME_DATABASE || '',
    limit: 10,
    ownerUserId: '',
    passthrough: [],
    server: process.env.GENARRATIVE_SPACETIME_SERVER || '',
    serverUrl: process.env.GENARRATIVE_SPACETIME_SERVER_URL || '',
  };

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

    if (arg === '--database') {
      options.database = readValue(arg);
    } else if (arg === '--server') {
      options.server = readValue(arg);
    } else if (arg === '--server-url') {
      options.serverUrl = readValue(arg);
    } else if (arg === '--limit') {
      options.limit = parsePositiveInteger(readValue(arg), arg);
    } else if (arg === '--cursor-job-id') {
      options.cursorJobId = readValue(arg).trim();
    } else if (arg === '--completed-before-micros') {
      const value = readValue(arg);
      if (!/^-?[0-9]+$/u.test(value)) {
        throw new Error(`${arg} 必须是整数。`);
      }
      const parsed = Number.parseInt(value, 10);
      if (!Number.isSafeInteger(parsed)) {
        throw new Error(`${arg} 超出 JavaScript 安全整数范围。`);
      }
      options.completedBeforeMicros = parsed;
    } else if (arg === '--owner-user-id') {
      options.ownerUserId = readValue(arg).trim();
    } else if (arg === '--apply') {
      options.apply = true;
    } else if (arg === '--backfill-summaries') {
      options.backfillSummaries = true;
    } else if (arg === '--help' || arg === '-h') {
      options.help = true;
    } else {
      throw new Error(`未知参数: ${arg}`);
    }
  }

  if (options.limit > MAX_BATCH_SIZE) {
    throw new Error(`--limit 不能超过 ${MAX_BATCH_SIZE}。`);
  }
  if (options.ownerUserId && !options.backfillSummaries) {
    throw new Error('--owner-user-id 只能与 --backfill-summaries 一起使用。');
  }
  if (options.completedBeforeMicros !== null && options.backfillSummaries) {
    throw new Error('--completed-before-micros 不能用于摘要回填。');
  }
  return options;
}

try {
  const options = parseOptions(process.argv.slice(2));
  if (options.help) {
    console.log(usage());
    process.exit(0);
  }
  if (!options.database) {
    throw new Error(
      '必须传入 --database，或设置 GENARRATIVE_SPACETIME_DATABASE。',
    );
  }

  const procedureName = options.backfillSummaries
    ? 'backfill_external_generation_job_summaries_and_return'
    : 'compact_external_generation_job_payloads_and_return';
  const input = options.backfillSummaries
    ? {
        owner_user_id: encodeSpacetimeCliOption(options.ownerUserId || null),
        limit: options.limit,
        cursor_job_id: encodeSpacetimeCliOption(options.cursorJobId || null),
        dry_run: !options.apply,
      }
    : {
        dry_run: !options.apply,
        limit: options.limit,
        cursor_job_id: encodeSpacetimeCliOption(options.cursorJobId || null),
        completed_before_micros: encodeSpacetimeCliOption(
          options.completedBeforeMicros,
        ),
      };
  const result = await callSpacetimeProcedureViaCli(
    options,
    procedureName,
    input,
  );
  ensureProcedureOk(result);

  console.log(JSON.stringify({ procedure: procedureName, ...result }, null, 2));
  const pendingApplyCount = options.backfillSummaries
    ? Number(result.selected_count ?? 0)
    : Number(result.matched_count ?? 0);
  if (result.has_more && options.apply) {
    console.log(
      `仍有后续批次；下一次追加 --cursor-job-id ${result.next_cursor_job_id ?? '<missing>'}。`,
    );
  } else if (!options.apply && (result.has_more || pendingApplyCount > 0)) {
    const currentCursor = options.cursorJobId
      ? `保留 --cursor-job-id ${options.cursorJobId}`
      : '仍从首批开始';
    console.log(
      `当前仅 dry-run；请${currentCursor}并追加 --apply 重跑同一批。apply 成功后再使用其 next_cursor_job_id 进入下一批。`,
    );
  }
} catch (error) {
  console.error(
    `[spacetime:external-generation:maintenance] ${error instanceof Error ? error.message : String(error)}`,
  );
  process.exit(1);
}