xiaoshu-app/scripts/run-nightly-legacy-sync.mjs

#!/usr/bin/env node

import { mkdir, writeFile, rename, rm } from 'node:fs/promises';
import { homedir } from 'node:os';
import { resolve, join } from 'node:path';
import { randomUUID } from 'node:crypto';

const defaultDatasets = ['nodes', 'vocabulary', 'special-training', 'course-bindings', 'learning-records', 'practice-records', 'memory-records', 'assessments', 'appointments', 'lessons'];
const reportDir = resolve(process.env.XIAOSHU_SYNC_REPORT_DIR || join(homedir(), '.local/state/xiaoshu/legacy-sync'));
const token = String(process.env.XIAOSHU_SYNC_OPERATOR_TOKEN || '').trim();
const companyId = process.env.XIAOSHU_SYNC_COMPANY_ID || '7pIbDBJmKx';
const functionUrl = (process.env.XIAOSHU_FUNCTION_URL || 'https://server.xiaoshu.pro/api/functions').replace(/\/$/, '');
const bridgeUrl = (process.env.XIAOSHU_LEGACY_SYNC_BASE_URL || 'https://a018.2018.z01.com/xiaoshu-sync').replace(/\/$/, '');
const appId = process.env.XIAOSHU_PARSE_APP_ID || '7pIbDBJmKx_main';
const datasets = String(process.env.XIAOSHU_SYNC_DATASETS || defaultDatasets.join(',')).split(',').map(value => value.trim()).filter(Boolean);
const maxRounds = Number(process.env.XIAOSHU_SYNC_MAX_ROUNDS || 200);
const maxDurationMs = Number(process.env.XIAOSHU_SYNC_MAX_DURATION_MS || 2_700_000);
const deadline = Date.now() + maxDurationMs;
const runId = randomUUID();
const businessDate = new Intl.DateTimeFormat('en-CA', { timeZone: 'Asia/Shanghai', year: 'numeric', month: '2-digit', day: '2-digit' }).format(new Date());
const report = { runId, businessDate, timeZone: 'Asia/Shanghai', startedAt: new Date().toISOString(), status: 'running', reason: '', datasets: {}, retries: 0 };
const lock = join(reportDir, 'nightly.lock');
let locked = false;

class RunError extends Error {
  constructor(reason, status = 'failed') { super(reason); this.runStatus = status; }
}
function abort(reason, status = 'blocked') { throw new RunError(reason, status); }
async function request(url, options = {}, allowRetry = true) {
  for (let attempt = 0; ; attempt++) {
    if (Date.now() >= deadline) abort('run_time_limit', 'incomplete');
    try {
      const response = await fetch(url, { ...options, signal: AbortSignal.timeout(Math.min(90_000, deadline - Date.now())) });
      if (allowRetry && [429, 502, 503, 504].includes(response.status) && attempt < 2) {
        report.retries++;
        await new Promise(done => setTimeout(done, 1000 * (attempt + 1)));
        continue;
      }
      return response;
    } catch (error) {
      if (error instanceof RunError) throw error;
      if (!allowRetry || attempt >= 2) abort('network_request_failed', 'failed');
      report.retries++;
      await new Promise(done => setTimeout(done, 1000 * (attempt + 1)));
    }
  }
}
async function gateway(params) {
  const response = await request(`${functionUrl}/xiaoshu/ops/gateway-v3`, {
    method: 'POST',
    headers: { 'Content-Type': 'application/json', 'X-Parse-Application-Id': appId },
    body: JSON.stringify({ token, params: { ...params, companyId } }),
  }, params.operation === 'ops/sync/status');
  if (!response.ok) abort(`gateway_http_${response.status}`);
  const payload = await response.json().catch(() => null);
  if (payload?.success !== true || !payload.data) abort('invalid_gateway_response');
  return payload.data;
}
function assertProtection(status) {
  // Do not infer protection from a caller-supplied flag. The deployed gateway
  // must advertise this only after its implementation and acceptance checks pass.
  if (status.syncProtection?.conflictPolicy !== 'preserve-new' || status.syncProtection?.atomicProjection !== true) {
    abort('preserve_new_conflict_protection_not_verified');
  }
}

try {
  await mkdir(reportDir, { recursive: true, mode: 0o700 });
  try { await mkdir(lock, { mode: 0o700 }); locked = true; }
  catch (error) { if (error.code === 'EEXIST') abort('another_run_or_stale_lock', 'skipped'); throw error; }
  await writeFile(join(lock, 'owner.json'), JSON.stringify({ runId, pid: process.pid, startedAt: report.startedAt }), { mode: 0o600 });
  if (!Number.isSafeInteger(maxRounds) || maxRounds < 1 || !Number.isSafeInteger(maxDurationMs) || maxDurationMs < 1) abort('invalid_run_limits');
  if (!datasets.length || new Set(datasets).size !== datasets.length || datasets.some(dataset => !defaultDatasets.includes(dataset))) abort('invalid_datasets');
  if (datasets.includes('memory-records') && !datasets.includes('learning-records')) abort('missing_learning_dependency');
  if (datasets.includes('lessons') && !datasets.includes('appointments')) abort('missing_appointment_dependency');
  const healthResponse = await request(`${bridgeUrl}/v1/health`);
  if (!healthResponse.ok) abort(`bridge_http_${healthResponse.status}`);
  const health = await healthResponse.json().catch(() => null);
  if (health?.status !== 'ok' || health.service !== 'xiaoshu-legacy-sync' || health.version !== 'v1') abort('wrong_bridge_service');
  if (!token) abort('missing_super_admin_token');
  if (process.env.XIAOSHU_SYNC_OPERATOR_EXPIRES_AT && !(Date.parse(process.env.XIAOSHU_SYNC_OPERATOR_EXPIRES_AT) > Date.now())) abort('expired_super_admin_token');
  const initial = await gateway({ operation: 'ops/sync/status' });
  if (!initial.configured || !initial.bridgeReachable) abort('bridge_not_configured_or_signed_manifest_unavailable');
  assertProtection(initial);
  if (process.argv.includes('--check')) {
    report.status = 'ready';
  } else {
    // Use a fixed topological order, regardless of environment list order.
    for (const dataset of defaultDatasets.filter(value => datasets.includes(value))) {
      const totals = report.datasets[dataset] = { rounds: 0, processed: 0, created: 0, updated: 0, deleted: 0, conflicts: 0, failures: 0, caughtUp: false };
      for (let round = 0; round < maxRounds; round++) {
        // A POST timeout is ambiguous. Stop and retain the server checkpoint;
        // the next scheduled run resumes it rather than replaying blindly.
        const result = await gateway({ operation: 'ops/sync/run', dataset, maxPages: 10, pageSize: 500, conflictPolicy: 'preserve-new' });
        if (result.dataset !== dataset || typeof result.hasMore !== 'boolean' || !Number.isFinite(result.failures)) abort('invalid_sync_result', 'failed');
        totals.rounds++;
        for (const key of ['processed', 'created', 'updated', 'deleted', 'conflicts', 'failures']) totals[key] += Number(result[key] || 0);
        if (result.failures) abort(`dataset_failed:${dataset}`, 'failed');
        if (!result.hasMore) { totals.caughtUp = true; break; }
      }
      if (!totals.caughtUp) abort(`round_limit:${dataset}`, 'incomplete');
    }
    const final = await gateway({ operation: 'ops/sync/status' });
    assertProtection(final);
    report.pendingFailures = Number(final.totalFailures || 0);
    report.pendingConflicts = Number(final.totalConflicts || 0);
    if (report.pendingFailures) abort('pending_failures', 'failed');
    if (report.pendingConflicts) abort('pending_conflicts', 'completed_with_conflicts');
    if (!final.configured || !final.bridgeReachable || !datasets.every(key => final.datasets?.some(row => row.key === key && row.reconciled === true))) {
      abort('business_id_checksum_reconciliation_not_verified', 'awaiting_reconciliation');
    }
    report.status = 'success';
  }
} catch (error) {
  report.status = error.runStatus || 'failed';
  // Never persist a raw server error, token, payload, or business record.
  report.reason = error instanceof RunError ? error.message : 'unexpected_runner_error';
} finally {
  report.finishedAt = new Date().toISOString();
  await mkdir(reportDir, { recursive: true, mode: 0o700 });
  const reportPath = join(reportDir, `${businessDate}-${runId}.json`);
  const temporary = `${reportPath}.tmp`;
  await writeFile(temporary, JSON.stringify(report, null, 2), { mode: 0o600 });
  await rename(temporary, reportPath);
  if (locked) await rm(lock, { recursive: true });
  console.log(JSON.stringify({ status: report.status, reason: report.reason, reportPath }));
  process.exitCode = ['success', 'ready'].includes(report.status) ? 0 : 1;
}