|
|
@@ -0,0 +1,128 @@
|
|
|
+#!/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;
|
|
|
+}
|
彭峰