chajia/scripts/prove-rpa-chain.ts

/**
 * RPA 链路存在性证明（单次 HTTP 全路径，不改业务逻辑）
 *
 * 证明目标：候选 → 入队 → Worker 消费 → axel/quote 网络抓价 → DB sourceType=rpa
 * 前置：npm run dev:start && RPA_MOCK_MODE=false
 * mock 下游验证：RPA_ADDRESS_MODE=mock（自动跳过联想 API，用内置地址）
 * PowerShell：$env:RPA_ADDRESS_MODE="mock"; npm run prove:rpa-chain
 * 强制跳过联想：$env:PROOF_SKIP_CANDIDATES="true"
 * 轮询上限（默认 60s）：$env:PROVE_POLL_TIMEOUT_MS="60000"
 * 用法：npm run prove:rpa-chain
 */
import { randomUUID } from "node:crypto";
import { existsSync, mkdirSync, readFileSync, readdirSync, writeFileSync } from "node:fs";
import { join } from "node:path";
import { loadDevEnv } from "@/lib/dev-env";
import { getRpaAddressMode } from "@/lib/rpa/address-mode";
import { getRpaQuoteMode } from "@/lib/rpa/env";
import { buildSyntheticCandidates } from "./lib/proof-synthetic-candidates";
import { ensureNativeInfraIfNeeded } from "@/lib/infra/ensure-native-infra";
import {
  hostCreateQuote,
  hostFetchMothershipCandidates,
  hostGetQuote,
  hostPreheatQuoteSession,
} from "@/lib/frontend/api-client";
import { applyMothershipCandidate } from "@/lib/frontend/mothership-address";
import type { QuoteRequestBody, MothershipAddressCandidate } from "@/lib/frontend/types";
import { pollQuoteUntilDone } from "@/hooks/use-quote-polling";
import { isAnyRpaWorkerHealthy } from "@/lib/rpa/worker-health";
import { prisma } from "@/lib/prisma";
import { ensureRedisReady, resetRedisClient } from "@/lib/redis";
import { recoverOrphanActiveJobs } from "@/workers/rpa/queue-recovery";

/** prove 脚本轮询上限（与生产 QUOTE_TIMEOUT_MS 解耦，默认 60s） */
const PROVE_POLL_TIMEOUT_MS = Number(
  process.env.PROVE_POLL_TIMEOUT_MS ?? 60_000,
);

loadDevEnv();

const BASE = process.env.PROVE_BASE_URL ?? "http://localhost:3000";
const TOKEN = "demo-host-token";
const CUSTOMER_ID = "CUST_001";
const WORKER_LOG_DIR = join(process.cwd(), ".dev", "logs");

function decodeLogBuffer(buf: Buffer): string {
  if (buf.length >= 2 && buf[0] === 0xff && buf[1] === 0xfe) {
    return buf.subarray(2).toString("utf16le");
  }
  if (buf.length >= 2 && buf[0] === 0xfe && buf[1] === 0xff) {
    return buf.subarray(2).toString("utf16le");
  }
  // PowerShell 重定向常见无 BOM UTF-16LE
  const sample = buf.subarray(0, Math.min(buf.length, 200));
  let nullCount = 0;
  for (let i = 1; i < sample.length; i += 2) {
    if (sample[i] === 0) {
      nullCount++;
    }
  }
  if (nullCount > sample.length / 4) {
    return buf.toString("utf16le");
  }
  return buf.toString("utf8");
}

function readLogFile(path: string, sinceByte = 0): string {
  const buf = readFileSync(path);
  const slice = buf.subarray(Math.min(sinceByte, buf.length));
  return decodeLogBuffer(slice);
}

function snapshotWorkerLogOffsets(): Map<string, number> {
  const snap = new Map<string, number>();
  if (!existsSync(WORKER_LOG_DIR)) {
    return snap;
  }
  const files = readdirSync(WORKER_LOG_DIR)
    .filter((name) => name.startsWith("worker-rpa") && name.endsWith(".log"))
    .map((name) => join(WORKER_LOG_DIR, name));
  if (files.length === 0) {
    const legacy = join(WORKER_LOG_DIR, "worker-rpa.log");
    if (existsSync(legacy)) {
      snap.set(legacy, readFileSync(legacy).length);
    }
    return snap;
  }
  for (const path of files) {
    snap.set(path, readFileSync(path).length);
  }
  return snap;
}

function readWorkerLogSince(snapshot: Map<string, number>): string {
  if (snapshot.size === 0) {
    return readWorkerLogTail(0);
  }
  return [...snapshot.entries()]
    .map(([path, offset]) => readLogFile(path, offset))
    .join("\n");
}

function readWorkerLogTail(sinceByte: number): string {
  if (!existsSync(WORKER_LOG_DIR)) {
    return "";
  }
  const files = readdirSync(WORKER_LOG_DIR)
    .filter((name) => name.startsWith("worker-rpa") && name.endsWith(".log"))
    .map((name) => join(WORKER_LOG_DIR, name));
  if (files.length === 0) {
    const legacy = join(WORKER_LOG_DIR, "worker-rpa.log");
    if (!existsSync(legacy)) {
      return "";
    }
    return readLogFile(legacy, sinceByte);
  }
  return files.map((path) => readLogFile(path, sinceByte)).join("\n");
}

type ProofAddress = { street: string; city: string; state: string; zip: string };
type ProofCargo = {
  pallet_count: number;
  weight_lb: number;
  length_in: number;
  width_in: number;
  height_in: number;
  cargo_type: QuoteRequestBody["cargo_type"];
};

/** 可选测试组（PROOF_PAIR_ID=01~10）；默认 01 基线 */
const PROOF_PAIRS: Record<
  string,
  { label: string; pickup: Omit<ProofAddress, "zip">; delivery: Omit<ProofAddress, "zip">; cargo: ProofCargo }
> = {
  "01": {
    label: "洛杉矶 CA → Farmers Branch TX（基线）",
    pickup: { street: "1234 Warehouse Street", city: "Los Angeles", state: "CA" },
    delivery: { street: "5678 Distribution Way", city: "Farmers Branch", state: "TX" },
    cargo: { pallet_count: 2, weight_lb: 500, length_in: 48, width_in: 40, height_in: 48, cargo_type: "general_freight" },
  },
  "02": {
    label: "西雅图 WA → 迈阿密 FL",
    pickup: { street: "3131 Western Ave", city: "Seattle", state: "WA" },
    delivery: { street: "7000 NW 52nd St", city: "Miami", state: "FL" },
    cargo: { pallet_count: 3, weight_lb: 550, length_in: 48, width_in: 40, height_in: 45, cargo_type: "general_freight" },
  },
  "06": {
    label: "丹佛 CO → 夏洛特 NC",
    pickup: { street: "4800 Race St", city: "Denver", state: "CO" },
    delivery: { street: "9300 South Blvd", city: "Charlotte", state: "NC" },
    cargo: { pallet_count: 3, weight_lb: 450, length_in: 48, width_in: 40, height_in: 44, cargo_type: "general_freight" },
  },
  "99": {
    label: "Key West FL → Los Angeles CA（Duval / Hollywood 通断测试）",
    pickup: { street: "Duval Street", city: "Key West", state: "FL" },
    delivery: { street: "Hollywood Boulevard", city: "Los Angeles", state: "CA" },
    cargo: { pallet_count: 2, weight_lb: 503, length_in: 48, width_in: 40, height_in: 48, cargo_type: "general_freight" },
  },
  "10": {
    label: "西雅图 Convention Center → 迈阿密 Beach Ocean Drive（widget）",
    pickup: {
      street: "Washington State Convention Center Main Garage, Pike Street",
      city: "Seattle",
      state: "WA",
    },
    delivery: {
      street: "Ocean Drive",
      city: "Miami Beach",
      state: "FL",
    },
    cargo: {
      pallet_count: 2,
      weight_lb: 500,
      length_in: 48,
      width_in: 40,
      height_in: 48,
      cargo_type: "general_freight",
    },
  },
  "11": {
    label: "真人录证 5519e0ea：293 Pike St Seattle → 826 Ocean Dr Miami Beach",
    pickup: { street: "293 Pike St", city: "Seattle", state: "WA" },
    delivery: { street: "826 Ocean Dr", city: "Miami Beach", state: "FL" },
    cargo: {
      pallet_count: 2,
      weight_lb: 500,
      length_in: 48,
      width_in: 40,
      height_in: 48,
      cargo_type: "general_freight",
    },
  },
};

const PROOF_PAIR_ID = process.env.PROOF_PAIR_ID ?? "01";
const PROOF_PAIR = PROOF_PAIRS[PROOF_PAIR_ID] ?? PROOF_PAIRS["01"]!;

/** 单托重量 + 宽域 nonce，避免 L2 与历史 prove 撞 cargo_hash */
const PROOF_WEIGHT_NONCE = Number(
  process.env.PROOF_WEIGHT_NONCE ??
    (parseInt(randomUUID().replace(/-/g, ""), 16) % 9500) + 1,
);

type Check = { id: string; ok: boolean; detail: string };

const checks: Check[] = [];

function record(id: string, ok: boolean, detail: string): void {
  checks.push({ id, ok, detail });
  console.log(`${ok ? "[PASS]" : "[FAIL]"} ${id}: ${detail}`);
}

const PROOF_DIR = join(process.cwd(), ".rpa");

async function ensureStackReady(): Promise<void> {
  ensureNativeInfraIfNeeded();
  resetRedisClient();
  await ensureRedisReady(12, 1_000);
  try {
    await prisma.$queryRaw`SELECT 1`;
  } catch {
    throw new Error("MySQL 未响应（127.0.0.1:3307），请先 npm run dev:infra:start");
  }
  const recovered = await recoverOrphanActiveJobs();
  if (recovered > 0) {
    console.warn(`[prove] 已回收 ${recovered} 条 orphan 队列 job`);
  }
  const res = await fetch(BASE, { method: "GET", redirect: "manual" });
  if (res.status <= 0) {
    throw new Error("Next 未响应，请先 npm run dev:start");
  }
  if (!(await isAnyRpaWorkerHealthy())) {
    throw new Error("RPA Worker 无心跳，请先 npm run dev:start");
  }
}

function isProofSkipCandidates(): boolean {
  return process.env.PROOF_SKIP_CANDIDATES === "true";
}

async function main(): Promise<void> {
  console.log("=== RPA 链路存在性证明（单次 HTTP）===\n");

  if (process.env.RPA_MOCK_MODE === "true") {
    record("mock-off", false, "RPA_MOCK_MODE=true，无法证明真实 RPA");
    process.exit(1);
  }
  record("mock-off", true, "RPA_MOCK_MODE≠true");

  const addressMode = getRpaAddressMode();
  const quoteMode = getRpaQuoteMode();
  const skipCandidates = isProofSkipCandidates();
  if (addressMode === "mock") {
    record(
      "address-mode",
      false,
      "RPA_ADDRESS_MODE=mock 不产生 MotherShip 真实报价，请改为 real",
    );
    process.exit(1);
  }
  record("address-mode", true, "RPA_ADDRESS_MODE=real（MotherShip 真实 axel/quote）");
  record("quote-mode", true, `RPA_QUOTE_MODE=${quoteMode}`);
  if (skipCandidates) {
    record(
      "candidates-skip",
      true,
      "跳过 MotherShip 联想 API（仅 PROOF_SKIP_CANDIDATES=true 时）",
    );
  }

  await ensureStackReady();
  record("infra", true, "Next + Redis + Worker 心跳就绪");

  const logSnapshot = snapshotWorkerLogOffsets();
  const runId = randomUUID().slice(0, 8);
  const startedAt = new Date().toISOString();

  console.log(`[prove] 测试组 ${PROOF_PAIR_ID}：${PROOF_PAIR.label}`);

  const userPickup: ProofAddress = { ...PROOF_PAIR.pickup, zip: "" };
  const userDelivery: ProofAddress = { ...PROOF_PAIR.delivery, zip: "" };
  const proofCargo = PROOF_PAIR.cargo;

  function draftAddress(draft: typeof userPickup): QuoteRequestBody["pickup_address"] {
    const street = draft.street.trim();
    const city = draft.city.trim();
    const state = draft.state.trim();
    return {
      street,
      city,
      state,
      zip: "",
      place_id: `draft_${street}_${city}_${state}`.replace(/\s+/g, "_").slice(0, 64),
      formatted_address: `${street}, ${city}, ${state}`,
      selected_from_suggestions: true,
    };
  }

  let selectedPickup: MothershipAddressCandidate;
  let selectedDelivery: MothershipAddressCandidate;
  let quote_session_id: string | undefined;

  if (skipCandidates) {
    console.log("\n[1/4] 跳过联想 API，使用内置合成地址 …");
    const syn = buildSyntheticCandidates(
      PROOF_PAIR_ID,
      userPickup,
      userDelivery,
    );
    selectedPickup = syn.pickup;
    selectedDelivery = syn.delivery;
    record(
      "candidates-rpa",
      true,
      `synthetic pickup=${selectedPickup.display_label} delivery=${selectedDelivery.display_label}`,
    );
  } else {
    console.log("\n[1/4] POST /api/addresses/mothership-candidates …");
    const candRes = await hostFetchMothershipCandidates(
      BASE,
      TOKEN,
      CUSTOMER_ID,
      userPickup,
      userDelivery,
    );
    if (candRes.code !== 0 || !candRes.data) {
      record("candidates-http", false, candRes.message ?? "联想 API 失败");
      writeReport(runId, startedAt, null, checks);
      process.exit(1);
    }

    const data = candRes.data;
    quote_session_id = data.quote_session_id;
    const pickupOk = data.pickup_candidates.length > 0;
    const deliveryOk = data.delivery_candidates.length > 0;
    record(
      "candidates-rpa",
      pickupOk && deliveryOk,
      `pickup=${data.pickup_candidates.length} delivery=${data.delivery_candidates.length}`,
    );

    selectedPickup = data.pickup_candidates[0] ?? null;
    selectedDelivery = data.delivery_candidates[0] ?? null;
    if (!selectedPickup || !selectedDelivery) {
      writeReport(runId, startedAt, null, checks);
      process.exit(1);
    }

    const userStreetNorm = `${userPickup.street}|${userDelivery.street}`;
    const confirmedNorm = `${selectedPickup.street ?? ""}|${selectedDelivery.street ?? ""}`;
    record(
      "candidates-not-raw-input",
      userStreetNorm !== confirmedNorm ||
        selectedPickup.display_label !== userPickup.street,
      `用户=${userPickup.street} → 确认=${selectedPickup.display_label}`,
    );
  }

  if (!selectedPickup || !selectedDelivery) {
    writeReport(runId, startedAt, null, checks);
    process.exit(1);
  }

  if (quote_session_id) {
    await hostPreheatQuoteSession(BASE, TOKEN, CUSTOMER_ID, quote_session_id);
  }

  console.log("\n[2/4] POST /api/quotes …");
  let created: Awaited<ReturnType<typeof hostCreateQuote>> | null = null;
  for (let attempt = 0; attempt < 6; attempt++) {
    const weightLb =
      PROOF_PAIR.cargo.weight_lb + PROOF_WEIGHT_NONCE + attempt * 17;
    const body: QuoteRequestBody = {
      request_id: randomUUID(),
      customer_id: CUSTOMER_ID,
      quote_session_id,
      pickup_address: applyMothershipCandidate(
        draftAddress(userPickup),
        selectedPickup,
      ),
      delivery_address: applyMothershipCandidate(
        draftAddress(userDelivery),
        selectedDelivery,
      ),
      weight: { value: weightLb, unit: "lb" },
      dimensions: {
        length: proofCargo.length_in,
        width: proofCargo.width_in,
        height: proofCargo.height_in,
        unit: "in",
      },
      pallet_count: proofCargo.pallet_count,
      cargo_type: proofCargo.cargo_type,
    };
    created = await hostCreateQuote(BASE, TOKEN, body);
    if (created.code !== 0 || !created.data?.quote_id) {
      break;
    }
    if (
      created.data.status === "done" &&
      created.data.source_type === "cache"
    ) {
      console.warn(
        `[prove] L2 命中，换 weight=${weightLb} 重试 (${attempt + 1}/6)`,
      );
      continue;
    }
    break;
  }
  if (!created || created.code !== 0 || !created.data?.quote_id) {
    record("quote-create", false, created?.message ?? "创建询价失败");
    writeReport(runId, startedAt, null, checks);
    process.exit(1);
  }

  const quoteId = created.data.quote_id;
  const createStatus = created.data.status;
  const createSource = created.data.source_type;

  if (createStatus === "done" && createSource === "cache") {
    record(
      "quote-not-cache",
      false,
      `创建即 done + source_type=cache（L2 假阳性，非本次 RPA）`,
    );
    writeReport(runId, startedAt, quoteId, checks);
    process.exit(1);
  }
  record(
    "quote-not-cache",
    createStatus === "processing",
    `quote_id=${quoteId} status=${createStatus} source=${createSource ?? "—"}`,
  );

  console.log(
    `\n[3/4] 轮询 GET /api/quotes/${quoteId}（上限 ${PROVE_POLL_TIMEOUT_MS / 1000}s）…`,
  );
  let finalDetail;
  if (createStatus === "done") {
    const detail = await hostGetQuote(BASE, TOKEN, CUSTOMER_ID, quoteId);
    finalDetail = detail.data;
  } else {
    const poll = await pollQuoteUntilDone(
      async () => {
        const detail = await hostGetQuote(BASE, TOKEN, CUSTOMER_ID, quoteId);
        if (detail.code !== 0) {
          return { ok: false, errorMessage: detail.message };
        }
        return { ok: true, data: detail.data };
      },
      { timeoutMs: PROVE_POLL_TIMEOUT_MS },
    );
    if (poll.type !== "done") {
      record(
        "quote-final",
        false,
        poll.type === "timeout" ? "轮询超时" : poll.message,
      );
      writeReport(runId, startedAt, quoteId, checks);
      process.exit(1);
    }
    finalDetail = poll.quote;
  }

  const tierCount = finalDetail?.quotes?.length ?? 0;
  const finalSource = finalDetail?.source_type;
  record(
    "quote-final-rpa",
    finalDetail?.status === "done" &&
      finalSource === "rpa" &&
      tierCount >= 4,
    `status=${finalDetail?.status} source=${finalSource} tiers=${tierCount}`,
  );

  console.log("\n[4/4] 校验 Worker 日志 + DB …");
  const newLog = readWorkerLogSince(logSnapshot);
  const quoteScopedLog = newLog
    .split(/\r?\n/)
    .filter((line) => line.includes(quoteId))
    .join("\n");
  const logChecks: Array<[string, RegExp]> = [
    ["log-quote-consume", new RegExp(`quote_id=${quoteId} attempt=`)],
    ["log-axel-quote", /services\.mothership\.com\/axel\/quote|\[rpa\] direct axel quote:/],
    ["log-quote-done", new RegExp(`\\[rpa-worker\\] job .*quote_id=${quoteId}`)],
  ];
  if (!skipCandidates) {
    if (quoteMode === "widget") {
      logChecks.push([
        "log-place-commit",
        /axel\/location\/place\//,
      ]);
    } else {
      logChecks.push([
        "log-direct-quote",
        /\[rpa\] direct axel quote:/,
      ]);
    }
    record(
      "log-candidates-job",
      true,
      "direct/widget 候选走 axel API（不经 worker-rpa）",
    );
  }
  record(
    "log-no-pickup-lost",
    !/pickupLostAfterDelivery=true/.test(newLog),
    "worker 日志不得出现 pickupLostAfterDelivery=true",
  );
  record(
    "log-no-fixture",
    !/mock-chain-fixture/.test(quoteScopedLog || newLog),
    "禁止 mock-chain-fixture 占位价（仅本次 quote 日志）",
  );
  for (const [id, pattern] of logChecks) {
    record(id, pattern.test(newLog), pattern.source);
  }

  const dbRec = await prisma.quoteRecord.findFirst({
    where: { quoteId },
    select: {
      status: true,
      sourceType: true,
      quotesJson: true,
      createdAt: true,
    },
  });
  const dbTierCount = Array.isArray(dbRec?.quotesJson)
    ? dbRec.quotesJson.length
    : 0;
  record(
    "db-rpa",
    dbRec?.status === "done" &&
      dbRec?.sourceType === "rpa" &&
      dbTierCount >= 4,
    dbRec
      ? `status=${dbRec.status} source=${dbRec.sourceType} tiers=${dbTierCount}`
      : "无记录",
  );

  const reportPath = writeReport(runId, startedAt, quoteId, checks, {
    proofPairId: PROOF_PAIR_ID,
    proofPairLabel: PROOF_PAIR.label,
    rpaAddressMode: addressMode,
    rpaQuoteMode: quoteMode,
    proofSkipCandidates: skipCandidates,
    userPickup,
    userDelivery,
    confirmedPickup: selectedPickup.display_label,
    confirmedDelivery: selectedDelivery.display_label,
    quote_session_id,
    tiers: finalDetail?.quotes?.map((q) => ({
      tier: `${q.service_level}/${q.rate_option}`,
      final_total: q.final_total,
    })),
    workerLogSnippet: newLog.slice(-4000),
  });

  const allPass = checks.every((c) => c.ok);
  console.log(`\n证据文件：${reportPath}`);
  if (allPass) {
    console.log("\n=== VERDICT: PASS（HTTP 全链路 + Worker 日志 + DB 一致）===");
    console.log(
      "下一步：人工打开 MotherShip 同候选地址核对档位/金额（本脚本不替代人工验收）",
    );
    process.exit(0);
  }

  console.log("\n=== VERDICT: FAIL（链路未完整证明，勿改业务逻辑）===");
  process.exit(1);
}

function writeReport(
  runId: string,
  startedAt: string,
  quoteId: string | null,
  checks: Check[],
  extra?: Record<string, unknown>,
): string {
  mkdirSync(PROOF_DIR, { recursive: true });
  const path = join(PROOF_DIR, `chain-proof-${runId}.json`);
  writeFileSync(
    path,
    JSON.stringify(
      {
        runId,
        startedAt,
        finishedAt: new Date().toISOString(),
        quoteId,
        verdict: checks.every((c) => c.ok) ? "PASS" : "FAIL",
        checks,
        ...extra,
      },
      null,
      2,
    ),
    "utf8",
  );
  return path;
}

main()
  .catch((error) => {
    console.error("[FAIL] unexpected:", error);
    process.exit(1);
  })
  .finally(async () => {
    await prisma.$disconnect().catch(() => undefined);
  });