codex-market-app/electron/lib/provider_sync.js

// provider_sync.js —— 一键把历史会话同步到「当前账号」的 provider，使其在 Codex 侧栏重新显示。
// 复刻 CodexPlusPlus 的 provider-sync：改 rollout 首行 provider + state DB(provider/has_user_event/cwd)。
// 必须在 Codex 已退出时运行。数据库用纯 JS 的 sql.js（免原生编译，跨平台）。
const fs = require("fs");
const path = require("path");
const os = require("os");
const { execFileSync, spawnSync } = require("child_process");
const initSqlJs = require("sql.js");

const CODEX_HOME = path.join(os.homedir(), ".codex");
const SESSION_INDEX = path.join(CODEX_HOME, "session_index.jsonl");
const GLOBAL_STATE = path.join(CODEX_HOME, ".codex-global-state.json");

// 检测 Codex 是否在运行（尽力而为，跨平台）
function codexRunning() {
  try {
    if (process.platform === "darwin") {
      return execFileSync("pgrep", ["-f", "Codex.app/Contents/MacOS/Codex"], { encoding: "utf-8" }).trim().length > 0;
    }
    if (process.platform === "win32") {
      return /Codex\.exe/i.test(execFileSync("tasklist", [], { encoding: "utf-8" }));
    }
  } catch {
    /* pgrep 无匹配时退出码非 0，视为未运行 */
  }
  return false;
}

// 目标 provider = config.toml 根级 model_provider，缺省 openai
function targetProvider() {
  try {
    const head = fs.readFileSync(path.join(CODEX_HOME, "config.toml"), "utf-8").split("\n[")[0];
    const m = head.match(/^\s*model_provider\s*=\s*["']([^"']+)/m);
    if (m) return m[1];
  } catch {}
  return "openai";
}

function findRollouts() {
  const out = [];
  for (const root of [path.join(CODEX_HOME, "sessions"), path.join(CODEX_HOME, "archived_sessions")]) {
    const walk = (dir) => {
      let ents;
      try {
        ents = fs.readdirSync(dir, { withFileTypes: true });
      } catch {
        return;
      }
      for (const e of ents) {
        const full = path.join(dir, e.name);
        if (e.isDirectory()) walk(full);
        else if (e.isFile() && e.name.startsWith("rollout-") && e.name.endsWith(".jsonl")) out.push(full);
      }
    };
    walk(root);
  }
  return out;
}

function isArchivedRollout(file) {
  return file.startsWith(path.join(CODEX_HOME, "archived_sessions") + path.sep);
}

function loadIndexTitles() {
  const map = {};
  try {
    for (const line of fs.readFileSync(SESSION_INDEX, "utf-8").split("\n")) {
      if (!line.trim()) continue;
      try {
        const d = JSON.parse(line);
        if (d.id && d.thread_name) map[d.id] = d.thread_name;
      } catch {}
    }
  } catch {}
  return map;
}

function parseTimeMs(tsText, fallbackMs = Date.now()) {
  const n = Date.parse(tsText || "");
  return Number.isFinite(n) ? n : fallbackMs;
}

function firstText(value) {
  if (typeof value === "string") return value;
  if (Array.isArray(value)) {
    return value
      .map((x) => (typeof x === "string" ? x : x && typeof x.text === "string" ? x.text : ""))
      .filter(Boolean)
      .join("\n");
  }
  return "";
}

function sql(v) {
  if (v === null || v === undefined) return "NULL";
  if (typeof v === "number") return Number.isFinite(v) ? String(v) : "NULL";
  return "'" + String(v).replace(/'/g, "''") + "'";
}

function sqlite3Available() {
  const r = spawnSync("sqlite3", ["--version"], { encoding: "utf-8" });
  return !r.error && r.status === 0;
}

function runSqlite3(dbPath, input) {
  const r = spawnSync("sqlite3", [dbPath], {
    input,
    encoding: "utf-8",
    maxBuffer: 20 * 1024 * 1024,
  });
  if (r.error || r.status !== 0) {
    throw new Error((r.stderr || r.error && r.error.message || "sqlite3 执行失败").trim());
  }
  return r.stdout || "";
}

function sqliteCols(dbPath) {
  const out = runSqlite3(dbPath, "PRAGMA table_info(threads);\n");
  return new Set(out.split("\n").map((line) => line.split("|")[1]).filter(Boolean));
}

function readRolloutRecord(file, titles) {
  let content;
  try {
    content = fs.readFileSync(file, "utf-8");
  } catch {
    return null;
  }
  const stat = fs.statSync(file);
  const rec = {
    id: "",
    rollout_path: file,
    cwd: "",
    source: "vscode",
    model_provider: "",
    thread_source: "user",
    title: "",
    first_user_message: "",
    preview: "",
    has_user_event: content.includes('"user_message"') || content.includes('"user_input"') ? 1 : 0,
    archived: isArchivedRollout(file) ? 1 : 0,
    archived_at: null,
    created_at_ms: stat.birthtimeMs || stat.mtimeMs || Date.now(),
    updated_at_ms: stat.mtimeMs || Date.now(),
    sandbox_policy: JSON.stringify({ type: "disabled" }),
    approval_mode: "never",
    tokens_used: 0,
    cli_version: "",
    git_sha: null,
    git_branch: null,
    git_origin_url: null,
    model: null,
    reasoning_effort: null,
    history_mode: "legacy",
  };
  for (const line of content.split("\n")) {
    if (!line.trim()) continue;
    let d;
    try {
      d = JSON.parse(line);
    } catch {
      continue;
    }
    const p = d.payload || {};
    if (d.type === "session_meta" && !rec.id) {
      rec.id = p.id || p.session_id || rec.id;
      rec.cwd = p.cwd || rec.cwd;
      rec.source = p.source || rec.source;
      rec.model_provider = p.model_provider || rec.model_provider;
      rec.thread_source = p.thread_source || rec.thread_source;
      rec.cli_version = p.cli_version || rec.cli_version;
      rec.history_mode = p.history_mode || rec.history_mode;
      rec.created_at_ms = parseTimeMs(p.timestamp || d.timestamp, rec.created_at_ms);
      if (p.git && typeof p.git === "object") {
        rec.git_sha = p.git.commit_hash || null;
        rec.git_branch = p.git.branch || null;
        rec.git_origin_url = p.git.repository_url || null;
      }
    } else if (d.type === "event_msg" && (p.type === "user_message" || p.type === "user_input")) {
      rec.has_user_event = 1;
      if (!rec.first_user_message) rec.first_user_message = firstText(p.message || p.input).trim();
    }
  }
  rec.updated_at_ms = Math.max(rec.updated_at_ms, rec.created_at_ms);
  rec.title = titles[rec.id] || rec.first_user_message.replace(/\s+/g, " ").slice(0, 80) || "(无标题对话)";
  rec.preview = rec.first_user_message.replace(/\s+/g, " ").slice(0, 200);
  return rec.id ? rec : null;
}

function buildThreadUpsertSql(records, target, cols) {
  const desired = [
    "id", "rollout_path", "created_at", "updated_at", "source", "model_provider", "cwd", "title",
    "sandbox_policy", "approval_mode", "tokens_used", "has_user_event", "archived", "archived_at",
    "git_sha", "git_branch", "git_origin_url", "cli_version", "first_user_message", "memory_mode",
    "model", "reasoning_effort", "created_at_ms", "updated_at_ms", "thread_source", "preview",
    "recency_at", "recency_at_ms", "history_mode",
  ].filter((c) => cols.has(c));
  const valueFor = (r, c) => {
    const created = Math.floor(r.created_at_ms / 1000);
    const updated = Math.floor(r.updated_at_ms / 1000);
    const values = {
      id: r.id,
      rollout_path: r.rollout_path,
      created_at: created,
      updated_at: updated,
      source: r.source,
      model_provider: target,
      cwd: r.cwd,
      title: r.title,
      sandbox_policy: r.sandbox_policy,
      approval_mode: r.approval_mode,
      tokens_used: r.tokens_used,
      has_user_event: r.has_user_event,
      archived: r.archived,
      archived_at: r.archived ? updated : null,
      git_sha: r.git_sha,
      git_branch: r.git_branch,
      git_origin_url: r.git_origin_url,
      cli_version: r.cli_version,
      first_user_message: r.first_user_message,
      memory_mode: "enabled",
      model: r.model,
      reasoning_effort: r.reasoning_effort,
      created_at_ms: Math.round(r.created_at_ms),
      updated_at_ms: Math.round(r.updated_at_ms),
      thread_source: r.thread_source,
      preview: r.preview,
      recency_at: updated,
      recency_at_ms: Math.round(r.updated_at_ms),
      history_mode: r.history_mode,
    };
    return values[c];
  };
  const updateSet = [];
  for (const c of desired) {
    if (c === "id") continue;
    if (c === "has_user_event") updateSet.push(`${c}=CASE WHEN excluded.${c}=1 THEN 1 ELSE threads.${c} END`);
    else if (["cwd", "first_user_message", "preview", "thread_source"].includes(c)) {
      updateSet.push(`${c}=CASE WHEN COALESCE(excluded.${c},'')<>'' THEN excluded.${c} ELSE threads.${c} END`);
    } else if (c === "title") {
      updateSet.push(`${c}=CASE WHEN COALESCE(threads.${c},'')='' OR threads.${c}='(无标题对话)' THEN excluded.${c} ELSE threads.${c} END`);
    } else if (["updated_at", "updated_at_ms", "recency_at", "recency_at_ms"].includes(c)) {
      updateSet.push(`${c}=MAX(threads.${c}, excluded.${c})`);
    } else {
      updateSet.push(`${c}=excluded.${c}`);
    }
  }
  const lines = ["BEGIN;"];
  for (const r of records) {
    lines.push(
      `INSERT INTO threads (${desired.join(",")}) VALUES (${desired.map((c) => sql(valueFor(r, c))).join(",")}) ` +
      `ON CONFLICT(id) DO UPDATE SET ${updateSet.join(",")};`
    );
  }
  lines.push(`UPDATE threads SET model_provider=${sql(target)} WHERE COALESCE(model_provider,'')<>${sql(target)};`);
  lines.push("COMMIT;");
  return lines.join("\n") + "\n";
}

function updateProjectState(records, log = () => {}) {
  if (!fs.existsSync(GLOBAL_STATE)) return { addedProjects: 0, removedProjectless: 0 };
  let state;
  try {
    state = JSON.parse(fs.readFileSync(GLOBAL_STATE, "utf-8"));
  } catch (e) {
    log(`全局项目索引跳过：${e.message || e}`);
    return { addedProjects: 0, removedProjectless: 0 };
  }
  const projects = [...new Set(records.map((r) => r.cwd).filter(Boolean))];
  const idsWithProject = new Set(records.filter((r) => r.cwd).map((r) => r.id));
  const before = Array.isArray(state["project-order"]) ? state["project-order"] : [];
  const seen = new Set(before);
  state["project-order"] = before.concat(projects.filter((p) => !seen.has(p)));
  const oldProjectless = Array.isArray(state["projectless-thread-ids"]) ? state["projectless-thread-ids"] : [];
  state["projectless-thread-ids"] = oldProjectless.filter((id) => !idsWithProject.has(id));
  const addedProjects = state["project-order"].length - before.length;
  const removedProjectless = oldProjectless.length - state["projectless-thread-ids"].length;
  if (addedProjects || removedProjectless) {
    fs.writeFileSync(GLOBAL_STATE, JSON.stringify(state, null, 2) + "\n", "utf-8");
    log(`项目索引更新：新增项目 ${addedProjects} 个，移出无项目会话 ${removedProjectless} 条`);
  }
  return { addedProjects, removedProjectless };
}

function ts() {
  const d = new Date();
  const p = (n) => String(n).padStart(2, "0");
  return `${d.getFullYear()}${p(d.getMonth() + 1)}${p(d.getDate())}-${p(d.getHours())}${p(d.getMinutes())}${p(d.getSeconds())}`;
}

async function syncProviderToCurrent(log = () => {}, onProgress = () => {}) {
  if (codexRunning()) throw new Error("检测到 Codex 正在运行，请先完全退出 Codex 再同步。");
  const target = targetProvider();
  log(`目标 provider：${target}`);
  onProgress({ phase: "备份中", done: 0, total: 0 });

  // 1) 备份
  const backup = path.join(CODEX_HOME, "backups_state", "provider-sync", ts());
  fs.mkdirSync(backup, { recursive: true });
  for (const name of ["config.toml", ".codex-global-state.json", ".codex-global-state.json.bak"]) {
    const src = path.join(CODEX_HOME, name);
    if (fs.existsSync(src)) fs.copyFileSync(src, path.join(backup, name));
  }
  for (const db of fs.readdirSync(CODEX_HOME).filter((f) => /^state_.*\.sqlite(?:-(?:wal|shm))?$/.test(f))) {
    fs.copyFileSync(path.join(CODEX_HOME, db), path.join(backup, db));
  }
  log(`已备份到 ${backup}`);

  // 2) 遍历 rollout：改首行 provider，收集完整线程元信息
  const userEventIds = new Set();
  const cwdById = {};
  const titles = loadIndexTitles();
  const records = [];
  let changedFiles = 0;
  const rolloutFiles = findRollouts();
  const rtotal = rolloutFiles.length;
  let ri = 0;
  for (const file of rolloutFiles) {
    ri++;
    if (ri % 5 === 0 || ri === rtotal) onProgress({ phase: "改写会话文件", done: ri, total: rtotal });
    let lines;
    try {
      lines = fs.readFileSync(file, "utf-8").split(/(?<=\n)/); // 保留换行
    } catch {
      continue;
    }
    if (!lines.length) continue;
    const blob = lines.join("");
    const hasUE = blob.includes('"user_message"') || blob.includes('"user_input"');
    let tid = null;
    let rewrote = false;
    for (let i = 0; i < lines.length; i++) {
      const line = lines[i];
      if (!line.trim()) continue;
      let rec;
      try {
        rec = JSON.parse(line);
      } catch {
        continue;
      }
      if (rec.type === "session_meta" && rec.payload && typeof rec.payload === "object") {
        tid = rec.payload.id || rec.payload.session_id || null;
        const cwd = rec.payload.cwd;
        if (tid && cwd) cwdById[tid] = cwd;
        if (rec.payload.model_provider !== target) {
          rec.payload.model_provider = target;
          lines[i] = JSON.stringify(rec) + "\n";
          rewrote = true;
        }
        break;
      }
    }
    if (rewrote) {
      fs.writeFileSync(file, lines.join(""), "utf-8");
      changedFiles++;
    }
    if (tid && hasUE) userEventIds.add(tid);
    const rec = readRolloutRecord(file, titles);
    if (rec) {
      rec.model_provider = target;
      records.push(rec);
    }
  }
  log(`改写 rollout 文件：${changedFiles} 个；含用户消息的会话：${userEventIds.size} 条；可回填线程：${records.length} 条`);

  // 2.5) Codex 项目分组来自全局 state，历史 cwd 需要合并到 project-order。
  const projectState = updateProjectState(records, log);

  // 3) 更新所有 state_*.sqlite
  onProgress({ phase: "更新数据库", done: 0, total: 0 });
  if (sqlite3Available()) {
    let provRows = 0;
    let ueRows = 0;
    let cwdRows = 0;
    let insertedRows = 0;
    for (const name of fs.readdirSync(CODEX_HOME).filter((f) => /^state_.*\.sqlite$/.test(f))) {
      const dbPath = path.join(CODEX_HOME, name);
      try {
        const cols = sqliteCols(dbPath);
        if (!cols.has("model_provider") || !cols.has("has_user_event")) continue;
        const beforeProvider = Number(runSqlite3(dbPath, `SELECT COUNT(*) FROM threads WHERE COALESCE(model_provider,'')<>${sql(target)};\n`).trim() || 0);
        const beforeUserEvent = Number(runSqlite3(dbPath, "SELECT COUNT(*) FROM threads WHERE COALESCE(has_user_event,0)<>1;\n").trim() || 0);
        const beforeRows = Number(runSqlite3(dbPath, "SELECT COUNT(*) FROM threads;\n").trim() || 0);
        runSqlite3(dbPath, buildThreadUpsertSql(records, target, cols));
        runSqlite3(dbPath, "PRAGMA wal_checkpoint(TRUNCATE);\n");
        const afterUserEvent = Number(runSqlite3(dbPath, "SELECT COUNT(*) FROM threads WHERE COALESCE(has_user_event,0)<>1;\n").trim() || 0);
        const afterRows = Number(runSqlite3(dbPath, "SELECT COUNT(*) FROM threads;\n").trim() || 0);
        provRows += beforeProvider;
        ueRows += Math.max(0, beforeUserEvent - afterUserEvent);
        insertedRows += Math.max(0, afterRows - beforeRows);
        cwdRows += records.filter((r) => r.cwd).length;
      } catch (e) {
        log(`  sqlite3 数据库跳过 ${name}：${e.message || e}`);
      }
    }
    log(`数据库更新：provider ${provRows} 行、has_user_event ${ueRows} 行、新增线程 ${insertedRows} 行、项目索引 +${projectState.addedProjects}`);
    onProgress({ phase: "完成", done: 0, total: 0 });

    return {
      target,
      totalConversations: rtotal,
      changedFiles,
      userEvents: userEventIds.size,
      provRows,
      ueRows,
      cwdRows,
      insertedRows,
      projectRows: projectState.addedProjects,
      projectlessRemoved: projectState.removedProjectless,
      backup,
    };
  }

  const SQL = await initSqlJs({
    // wasm 恒在【包内】node_modules（热更包不含 node_modules），故用 BUNDLE_ROOT 而非 __dirname
    locateFile: (f) =>
      path.join(global.__BUNDLE_ROOT__ || path.join(__dirname, "..", ".."),
        "node_modules", "sql.js", "dist", f),
  });
  let provRows = 0;
  let ueRows = 0;
  let cwdRows = 0;
  for (const name of fs.readdirSync(CODEX_HOME).filter((f) => /^state_.*\.sqlite$/.test(f))) {
    const dbPath = path.join(CODEX_HOME, name);
    let db;
    try {
      db = new SQL.Database(fs.readFileSync(dbPath));
      const tbl = db.exec("SELECT name FROM sqlite_master WHERE type='table' AND name='threads'");
      if (!tbl.length) {
        db.close();
        continue;
      }
      const cols = new Set(db.exec("PRAGMA table_info(threads)")[0].values.map((r) => r[1]));
      if (!cols.has("model_provider")) {
        db.close();
        continue;
      }
      db.run("UPDATE threads SET model_provider=? WHERE COALESCE(model_provider,'')<>?", [target, target]);
      provRows += db.getRowsModified();
      if (cols.has("has_user_event")) {
        for (const tid of userEventIds) {
          db.run("UPDATE threads SET has_user_event=1 WHERE id=? AND COALESCE(has_user_event,0)<>1", [tid]);
          ueRows += db.getRowsModified();
        }
      }
      if (cols.has("cwd")) {
        for (const tid of Object.keys(cwdById)) {
          db.run("UPDATE threads SET cwd=? WHERE id=? AND COALESCE(cwd,'')<>?", [cwdById[tid], tid, cwdById[tid]]);
          cwdRows += db.getRowsModified();
        }
      }
      fs.writeFileSync(dbPath, Buffer.from(db.export()));
      db.close();
    } catch (e) {
      if (db) try { db.close(); } catch {}
      log(`  数据库跳过 ${name}：${e.message || e}`);
    }
  }
  log(`数据库更新：provider ${provRows} 行、has_user_event ${ueRows} 行、cwd ${cwdRows} 行`);
  onProgress({ phase: "完成", done: 0, total: 0 });

  return {
    target,
    totalConversations: rtotal,
    changedFiles,
    userEvents: userEventIds.size,
    provRows,
    ueRows,
    cwdRows,
    insertedRows: 0,
    projectRows: projectState.addedProjects,
    projectlessRemoved: projectState.removedProjectless,
    backup,
  };
}

module.exports = { syncProviderToCurrent, codexRunning, targetProvider };