/** # 脚本说明
 * 压测发射器(在 37 服务器以 root 运行):
 *   从 pool.json 逐个(并发 workers)执行:
 *     节点导入(幂等)→ POST lease(ttlMs=2h,规避 10 分钟空闲自动关闭)
 *     → POST browser/start → 按标签数经 CDP /json/new 开真实页面。
 *   直到「资源耗尽」触发:MemAvailable<1.5GB / swap已用≥7.2GB /
 *   连续启动失败≥10 / 已开浏览器被 OOM 杀≥10 / 达到 --max 封顶。
 *   触发后停止发射(已开浏览器保持常驻),结论写 data/<label>-stop.json。
 *   断点续跑:进度在 data/<label>-launched.jsonl,重启脚本会跳过已开指纹。
 *
 * 用法:bun run-stress.ts --pool data/pool.json --tabs 3 --label test1 [--max 500] [--concurrency 6]
 */
/** # 成员变量 辅助:进程内同步读文本(供 /proc 扫描) **/
const fs = require("fs") as typeof import("fs");

/** # 成员变量 服务基址(同机) **/
const ADMIN = "http://127.0.0.1:5310";
const NODE = "http://127.0.0.1:3106";

/** # 成员变量 基目录与数据目录 **/
const BASE = new URL(".", import.meta.url).pathname;
const DATA = `${BASE}data`;

/** # 成员变量 3 标签站点;1 标签测试只取第一个(google) **/
const URLS3 = ["https://www.google.com/", "https://www.yahoo.com/", "https://x.com/"];

/** # 成员变量 Lease TTL:2 小时(远大于 10 分钟空闲关闭窗口) **/
const LEASE_TTL_MS = 7_200_000;

/** # 成员变量 资源耗尽触发阈值(连续失败放宽,配合冷却防瞬时抖动误停) **/
const MEM_FLOOR_MB = 1500;
const SWAP_CEIL_MB = 7200;
const MAX_CONSEC_FAIL = 20;
const MAX_OOM_KILLED = 10;
/** # 成员变量 发射节流:load1 超过核数×3(37 为 8 核 → 24)暂停发射 **/
const LOAD_GATE = 24;

// ---- 参数解析 ----
const args = process.argv.slice(2);
function argOf(k: string, d: string): string { const i = args.indexOf(k); return i >= 0 ? args[i + 1] : d; }
const TABS = Number(argOf("--tabs", "3"));
const LABEL = argOf("--label", "test1");
const POOL_FILE = argOf("--pool", `${DATA}/pool.json`);
const MAX = Number(argOf("--max", "0"));
const CONCURRENCY = Number(argOf("--concurrency", "6"));
/** # 成员变量 本轮要开的真实站点列表 **/
const OPEN_URLS = TABS === 1 ? URLS3.slice(0, 1) : URLS3;

type Entry = {
  fp: string; name: string; startedAt: string;
  cdpPort: number | null; mainPid: number | null; tabs: string[];
  leaseId?: string; userDataDir?: string;
};

// ---- 密钥(仅内存,不落盘不回显) ----
/** # 函数:通过 docker exec 查询 admin DB 取单值文本 **/
function sqlOne(sql: string): string {
  const p = Bun.spawnSync(["docker", "exec", "serpmax-pg", "psql", "-U", "postgres", "-d", "serpmax_admin", "-tAc", sql], { stdout: "pipe", stderr: "pipe" });
  const out = p.stdout.toString().trim();
  if (p.exitCode !== 0 || !out) throw new Error(`SQL 失败: ${p.stderr.toString().slice(0, 300)}`);
  return out;
}
const TOKEN = sqlOne("SELECT api_token FROM users WHERE id=2");
const envText = fs.readFileSync("/opt/serpmax/node/.env", "utf8");
const NODE_KEY = /^NODE_KEY=(.*)$/m.exec(envText)?.[1]?.trim() ?? "";
/** # 成员变量 ipmart.io-US 代理四元组(DB 直取) **/
const PROXY_ID = "19e3ab5c-c61e-4936-8a7f-d78f6635f57c";
const prow = sqlOne(`SELECT type,host,port,coalesce(username,''),coalesce(password,'') FROM proxies WHERE id='${PROXY_ID}'`).split("|");

/** # 成员变量 节点公共头(与 admin relayHeaders 一致:Token+节点密钥双证) **/
function nodeHeaders(extra: Record<string, string> = {}): Record<string, string> {
  return { "content-type": "application/json", authorization: `Bearer ${TOKEN}`, "x-node-key": NODE_KEY, ...extra };
}

// ---- 状态 ----
const pool = (await Bun.file(POOL_FILE).json()) as { fingerprint: string; name: string }[];
const LAUNCHED_FILE = `${DATA}/${LABEL}-launched.json`;
/** # 成员变量 已启动记录(含续跑读取) **/
let launched: Entry[] = [];
try { launched = JSON.parse(await Bun.file(LAUNCHED_FILE).text() || "[]"); } catch { launched = []; }
const doneSet = new Set(launched.map((e) => e.fp));
console.log(`[run] pool=${pool.length} tabs=${TABS} 续跑已开=${launched.length}`);
const persist = () => Bun.write(LAUNCHED_FILE, JSON.stringify(launched));

// ---- 统计/控制 ----
let consecutiveFail = 0;
let totalFail = 0;
let oomKilled = 0;
let stopReason: string | null = null;
const t0 = Date.now();

/** # 函数:读内存水位(MB) **/
function meminfo(): { availMB: number; swapUsedMB: number } {
  const txt = fs.readFileSync("/proc/meminfo", "utf8");
  const get = (k: string) => Number(new RegExp(`${k}:\\s+(\\d+)`).exec(txt)?.[1] ?? 0);
  return { availMB: get("MemAvailable") / 1024, swapUsedMB: (get("SwapTotal") - get("SwapFree")) / 1024 };
}

/** # 函数:统计已开浏览器中被杀死的数量。
 * 判据:该 Profile 的 user-data-dir 下已无任何 chrome 进程,且记录时长超过 60s
 * (startBrowser 返回的 processId 是包装进程,会自然退出,不能逐 PID 判)。
 **/
function countKilled(): number {
  const alive = new Set<string>();
  for (const pid of fs.readdirSync("/proc").filter((d) => /^\d+$/.test(d))) {
    try {
      const cmd = fs.readFileSync(`/proc/${pid}/cmdline`, "utf8");
      const m = /--user-data-dir=\S*\/profiles\/([\w.]+)\/browser/.exec(cmd);
      if (m) alive.add(m[1]);
    } catch { /* 进程已退出 */ }
  }
  const now = Date.now();
  let dead = 0;
  for (const e of launched) {
    if (now - Date.parse(e.startedAt) < 60_000) continue;
    if (!alive.has(e.fp)) dead++;
  }
  return dead;
}

/** # 函数:能否继续发射;返回触发原因或 null,顺带刷新 oomKilled **/
function budget(): string | null {
  if (stopReason) return stopReason;
  if (consecutiveFail >= MAX_CONSEC_FAIL) return `连续启动失败 ${consecutiveFail} 次(共失败 ${totalFail})`;
  oomKilled = countKilled();
  if (oomKilled >= MAX_OOM_KILLED) return `浏览器主进程消失(疑似 OOM)${oomKilled} 个`;
  if (MAX > 0 && launched.length >= MAX) return `达到封顶数 ${MAX}`;
  const m = meminfo();
  if (m.availMB < MEM_FLOOR_MB) return `MemAvailable ${m.availMB | 0}MB < ${MEM_FLOOR_MB}MB`;
  if (m.swapUsedMB >= SWAP_CEIL_MB) return `swap 已用 ${m.swapUsedMB | 0}MB ≥ ${SWAP_CEIL_MB}MB`;
  return null;
}

async function nodeFetch(path: string, init: RequestInit = {}, timeout = 120000): Promise<Response> {
  return fetch(`${NODE}${path}`, { ...init, headers: nodeHeaders(init.headers as Record<string, string> | undefined), signal: AbortSignal.timeout(timeout) });
}

/** # 函数:解开 {success, data} 包装,拿 payload **/
function unwrap<T>(j: T | { data: T }): T {
  return (j as any)?.data ?? (j as any);
}

/** # 函数:节点导入(幂等;载荷与 admin relay 的 importProfileToNode 等价) **/
async function ensureImported(fp: string, name: string): Promise<void> {
  const chk = await nodeFetch(`/profiles/${fp}`, {}, 10000);
  if (chk.ok) return;
  if (chk.status !== 404) throw new Error(`节点查询 ${fp} 状态 ${chk.status}`);
  const res = await nodeFetch("/profiles", {
    method: "POST",
    headers: { ...nodeHeaders(), "idempotency-key": `stress-import-${fp}` },
    body: JSON.stringify({
      id: fp, name,
      browser: { kind: "cloak", theme: "light" },
      network: {},
      proxy: { id: PROXY_ID, protocol: prow[0], host: prow[1], port: Number(prow[2]), username: prow[3], password: prow[4] },
      metadata: { importedBy: "stress-run" },
    }),
  }, 15000);
  if (!res.ok && res.status !== 409) throw new Error(`导入 ${fp} 失败 ${res.status}`);
}

/** # 函数
 * 通过 WS 对指定标签发送 Page.navigate(内核 /json/new?url= 不生效,需逐标签导航)。
 * 参数 port:CDP 端口;targetId:标签 id;url:目标网址
 * 返回值:是否发出成功(10s 超时视为失败,不阻塞)
 */
async function navigateTab(port: number, targetId: string, url: string): Promise<boolean> {
  return new Promise((resolve) => {
    let ws: WebSocket | null = null;
    const done = (ok: boolean) => { try { ws?.close(); } catch { /* ignore */ } resolve(ok); };
    const timer = setTimeout(() => done(false), 10000);
    try {
      ws = new WebSocket(`ws://127.0.0.1:${port}/devtools/page/${targetId}`);
      ws.onopen = () => ws!.send(JSON.stringify({ id: 1, method: "Page.navigate", params: { url } }));
      ws.onmessage = (ev) => {
        try {
          const m = JSON.parse(String(ev.data));
          if (m.id === 1) { clearTimeout(timer); done(!m.error); }
        } catch { /* ignore */ }
      };
      ws.onerror = () => { clearTimeout(timer); done(false); };
    } catch { clearTimeout(timer); done(false); }
  });
}

/** # 函数:lease(2h)→ start → 开标签;返回每 Profile 记录 **/
async function launchOne(fp: string, name: string): Promise<Entry> {
  await ensureImported(fp, name);
  // 已有未过期 lease(如脚本重启续跑)则复用其 leaseId,否则新拿一个
  let leaseId = "";
  const detail = await nodeFetch(`/profiles/${fp}`, {}, 10000);
  if (detail.ok) {
    const d = unwrap<{ lease?: { leaseId: string; expiresAt: string } | null }>(await detail.json());
    if (d.lease && new Date(d.lease.expiresAt).getTime() > Date.now()) leaseId = d.lease.leaseId;
  }
  if (!leaseId) {
    const r = await nodeFetch(`/profiles/${fp}/lease`, { method: "POST", body: JSON.stringify({ consumerId: "stress", ttlMs: LEASE_TTL_MS }) }, 15000);
    if (!r.ok) throw new Error(`lease ${r.status} ${(await r.text()).slice(0, 100)}`);
    leaseId = unwrap<{ leaseId: string }>(await r.json()).leaseId;
  }
  const start = await nodeFetch(`/profiles/${fp}/browser/start`, { method: "POST", headers: { "x-lease-id": leaseId }, body: "{}" }, 150000);
  if (!start.ok) throw new Error(`start ${start.status} ${(await start.text()).slice(0, 130)}`);
  const s = unwrap<{ cdp?: { endpoint?: string; port?: number }; processId?: number; userDataDir?: string }>(await start.json());
  const entry: Entry = { fp, name, startedAt: new Date().toISOString(), cdpPort: s.cdp?.port ?? null, mainPid: s.processId ?? null, tabs: [], leaseId, userDataDir: s.userDataDir };
  // OOM 角色反转:浏览器主进程调「易杀」,系统其它进程由采样器置「受保护」
  if (entry.mainPid) { try { fs.writeFileSync(`/proc/${entry.mainPid}/oom_score_adj`, "200"); } catch { /* 进程已退出 */ } }
  // 开标签:CDP /json/new 建空页(Chrome 111+ 要求 PUT;405 回退 POST)→ WS Page.navigate 逐个导航
  if (entry.cdpPort) {
    for (const u of OPEN_URLS) {
      const url = `http://127.0.0.1:${entry.cdpPort}/json/new`;
      try {
        let t = await fetch(url, { method: "PUT", signal: AbortSignal.timeout(15000) });
        if (t.status === 405) t = await fetch(url, { method: "POST", signal: AbortSignal.timeout(15000) });
        if (t.ok) {
          const j: any = unwrap(await t.json().catch(() => null) ?? {});
          const tid = j?.id ?? "";
          entry.tabs.push(tid);
          if (tid && !j?.url?.startsWith(u)) void navigateTab(entry.cdpPort, tid, u);
        }
      } catch { /* 单个标签失败不阻塞 */ }
    }
  }
  return entry;
}

// ---- 主 workers ----
let cursor = 0;
/** # 成员变量 每 Profile 启动失败计数(≥2 判代理/数据坏,节点强删后跳过) **/
const failByFp = new Map<string, number>();
/** # 函数:读 1 分钟负载(发射节流用) **/
function load1(): number {
  try { return Number(fs.readFileSync("/proc/loadavg", "utf8").split(/\s+/)[0]) || 0; } catch { return 0; }
}
async function worker(): Promise<void> {
  while (!stopReason) {
    if (cursor >= pool.length) return;
    // 发射节流:系统负载超过门限则等待(防 CPU 打满导致的启动超时雪崩)
    while (load1() > LOAD_GATE && !stopReason) await new Promise((r) => setTimeout(r, 10_000));
    const i = cursor++;
    const item = pool[i];
    if (!item || doneSet.has(item.fingerprint)) continue;
    const b = budget();
    if (b) { stopReason = stopReason ?? b; return; }
    try {
      const entry = await launchOne(item.fingerprint, item.name);
      doneSet.add(item.fingerprint);
      launched.push(entry);
      consecutiveFail = 0;
      if (launched.length % 10 === 0) {
        const m = meminfo();
        console.log(`[run] launched=${launched.length} avail=${m.availMB | 0}MB swapUsed=${m.swapUsedMB | 0}MB t=${((Date.now() - t0) / 60000).toFixed(1)}min`);
        persist();
      }
    } catch (e) {
      consecutiveFail++;
      totalFail++;
      console.error(`[run] fail(${consecutiveFail}) ${item.fingerprint}: ${String(e).slice(0, 160)}`);
      // 坏 Profile(嫌疑代理不通/数据脏,非超时类错误)失败两次 → 节点强删后跳过
      // 启动超时(Timeout)不删:属系统瞬时拥塞,续跑时还会重试
      const msg = String(e);
      if (!msg.includes("Timeout")) {
        const n = (failByFp.get(item.fingerprint) ?? 0) + 1;
        failByFp.set(item.fingerprint, n);
        if (n >= 2) {
          try { await nodeFetch(`/profiles/${item.fingerprint}?force=true`, { method: "DELETE" }, 30000); } catch { /* best-effort */ }
          doneSet.add(item.fingerprint);
          console.error(`[run] drop ${item.fingerprint}(失败 ${n} 次,节点已强删)`);
        }
      }
      // 连续失败冷却:6 次后等 45s 再继续(躲开发射抖动期)
      if (consecutiveFail === 6) { console.error(`[run] 连续失败,冷却 45s`); await new Promise((r) => setTimeout(r, 45_000)); }
    }
    const b2 = budget();
    if (b2 && !stopReason) stopReason = b2;
  }
}
await persist();
await Promise.all(Array.from({ length: CONCURRENCY }, () => worker()));
await persist();
const m = meminfo();
const final = { stopReason, launchedCount: launched.length, totalFail, oomKilled, tabs: TABS, minutes: +((Date.now() - t0) / 60000).toFixed(1), availMB: m.availMB | 0, swapUsedMB: m.swapUsedMB | 0, stoppedAt: new Date().toISOString() };
await Bun.write(`${DATA}/${LABEL}-stop.json`, JSON.stringify(final, null, 2));
console.log(`[run] STOP: ${JSON.stringify(final)}`);