AgentHub/server/hub/remote-ingest.js

/**
 * @file remote-ingest.js
 * @description The hub's side of a remote client's hourly upload (server mode
 * only). A client sends, for each company-account session it registered, the
 * NEW part of each transcript file — the session's JSONL and its sub-agents'
 * — already stripped of secret-shaped strings on the client. The hub appends
 * it to its own copy and imports that copy with the SAME importer it uses for
 * local history (`importFromDirectory`), so the session's tokens and cost are
 * computed exactly like a local one's; the session is then marked as the
 * device's (`source = device:<id>`) and its user's (`owner_id`).
 *
 * Every path is built from ids the hub checked — user id, device id, a
 * registered session UUID, and one of two fixed file-name shapes — never from
 * request text. A chunk is accepted only at the offset the hub already holds
 * (`from` = consumed): a client that lost its place is told where to resume
 * (409 with `expected`), so nothing is written twice or skipped.
 *
 * Importing parses the whole copy, so a long session sent as many chunks
 * would be parsed once per chunk. A client marks every chunk but the last of
 * a round `more: true`; the hub imports once when the round's last request
 * arrives (or FLUSH_MS after a round that stopped half-way).
 *
 * Conversations are kept `RETAIN_DAYS` (30) after a session's last upload:
 * `purgeOld` deletes the copy and the session's events, and keeps its token
 * rows, so spend and cost history stay whole.
 * @author Nguyễn Ngọc Trí Vĩ <vinnt@smartgift.vn>
 */

const fs = require("node:fs");
const path = require("node:path");

const { getDataDir } = require("../lib/claude-home");
const sessions = require("./remote-sessions");

const err = (code, message, status, extra = {}) =>
  Object.assign(new Error(message), { code, status, ...extra });

const RETAIN_DAYS = Number(process.env.AGENTHUB_REMOTE_RETAIN_DAYS) || 30;
/** One file's copy may not grow past this. */
const MAX_FILE_BYTES = 512 * 1024 * 1024;
const MAX_FILES_PER_UPLOAD = 200;
/** A round that stopped before its last request is imported this long after. */
const FLUSH_MS = () => Number(process.env.AGENTHUB_REMOTE_FLUSH_MS) || 2 * 60 * 1000;
const SUBAGENT_RE =
  /^subagents\/(?:workflows\/[A-Za-z0-9_-]{1,100}\/)?agent-[A-Za-z0-9_-]{1,100}\.jsonl$/;

/** `<data dir>/remote/u<user>/d<device>/<session>` — the session's import root. */
function sessionRoot(userId, deviceId, sessionId) {
  return path.join(
    getDataDir(),
    "remote",
    `u${Number(userId)}`,
    `d${Number(deviceId)}`,
    String(sessionId)
  );
}

/** The folder a session's files sit in, named like Claude Code names a project. */
function projectDir(row) {
  const slug = (row.cwd || "unknown").replace(/[^A-Za-z0-9]/g, "-").slice(0, 200) || "unknown";
  return path.join(sessionRoot(row.user_id, row.device_id, row.session_id), slug);
}

/**
 * Where `rel` of session `row` is kept: `<sid>.jsonl` (the session itself) or
 * `<sid>/subagents/agent-*.jsonl` (a sub-agent's) — anything else is refused.
 */
function fileFor(row, rel) {
  const sid = row.session_id;
  if (rel === `${sid}.jsonl`) return path.join(projectDir(row), rel);
  const prefix = `${sid}/`;
  if (
    typeof rel === "string" &&
    rel.startsWith(prefix) &&
    SUBAGENT_RE.test(rel.slice(prefix.length))
  ) {
    return path.join(projectDir(row), ...rel.split("/"));
  }
  throw err("EBADPATH", `not a transcript file of this session: ${String(rel).slice(0, 120)}`, 400);
}

const sql = (q) => require("../db").db.prepare(q);

function consumedOf(sessionId, rel) {
  const r = sql("SELECT consumed FROM remote_files WHERE session_id = ? AND rel_path = ?").get(
    sessionId,
    rel
  );
  return r ? r.consumed : 0;
}

/**
 * Take one upload of `device`: `{sessionId, files: [{path, from, to, data}], ended}`.
 * Returns `{files: [{path, consumed}], imported}`; throws a coded error
 * (409 EOFFSET carries `expected`).
 */
async function takeUpload(device, body = {}) {
  const row = sessions.sessionOfDevice(body.sessionId, device.id);
  if (!row) throw err("ENOSESSION", "this device registered no such session", 404);
  if (row.purged_at) throw err("EPURGED", "this session's conversation was already removed", 410);
  const files = Array.isArray(body.files) ? body.files : [];
  if (files.length > MAX_FILES_PER_UPLOAD)
    throw err("ETOOMANY", "too many files in one upload", 400);

  // Check every chunk before writing any, so one bad entry writes nothing.
  const plan = files.map((f) => {
    const rel = String(f?.path || "");
    const file = fileFor(row, rel);
    const from = Number(f.from);
    const to = Number(f.to);
    const data = typeof f.data === "string" ? f.data : null;
    if (!Number.isSafeInteger(from) || !Number.isSafeInteger(to) || from < 0 || to < from) {
      throw err("EBADRANGE", "from/to must be offsets with from ≤ to", 400);
    }
    if (data == null || (data && !data.endsWith("\n"))) {
      throw err("EBADDATA", "data must be whole lines", 400);
    }
    const consumed = consumedOf(row.session_id, rel);
    if (from !== consumed) {
      throw err("EOFFSET", "upload does not continue where the hub stopped", 409, {
        expected: { path: rel, consumed },
      });
    }
    let size = 0;
    try {
      size = fs.statSync(file).size;
    } catch {
      /* new file */
    }
    if (size + Buffer.byteLength(data) > MAX_FILE_BYTES) {
      throw err("ETOOBIG", "this transcript is over the hub's size limit", 413);
    }
    return { rel, file, to, data };
  });

  for (const p of plan) {
    fs.mkdirSync(path.dirname(p.file), { recursive: true, mode: 0o700 });
    if (p.data) fs.appendFileSync(p.file, p.data, { mode: 0o600 });
    sql(
      `INSERT INTO remote_files (session_id, rel_path, consumed, updated_at)
         VALUES (?, ?, ?, strftime('%Y-%m-%dT%H:%M:%fZ','now'))
       ON CONFLICT(session_id, rel_path) DO UPDATE SET consumed = excluded.consumed,
         updated_at = excluded.updated_at`
    ).run(row.session_id, p.rel, p.to);
  }

  let imported = false;
  const wrote = plan.some((p) => p.data);
  if (wrote && body.more === true) {
    pending.add(row.session_id);
    scheduleFlush(row.session_id);
  } else if (wrote || pending.has(row.session_id)) {
    imported = await importSession(row.session_id);
  }
  sql(
    "UPDATE remote_sessions SET last_upload_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE session_id = ?"
  ).run(row.session_id);
  if (body.ended === true) sessions.markEnded(row.session_id, device.id);
  return {
    files: plan.map((p) => ({ path: p.rel, consumed: p.to })),
    imported,
  };
}

// Sessions with chunks written but not imported yet, and their fallback timers.
const pending = new Set();
const flushTimers = new Map();
// One import at a time per session: two would parse the same copy twice.
const importing = new Map();

function scheduleFlush(sessionId) {
  clearTimeout(flushTimers.get(sessionId));
  const t = setTimeout(() => {
    flushTimers.delete(sessionId);
    if (pending.has(sessionId)) {
      importSession(sessionId).catch((e) =>
        console.warn(`[remote] import of ${sessionId} failed: ${e.message}`)
      );
    }
  }, FLUSH_MS());
  t.unref();
  flushTimers.set(sessionId, t);
}

/** Import the hub's copy of `sessionId` and mark the row its user's. */
function importSession(sessionId) {
  const prev = importing.get(sessionId) || Promise.resolve();
  const run = prev
    .catch(() => {})
    .then(async () => {
      pending.delete(sessionId);
      clearTimeout(flushTimers.get(sessionId));
      flushTimers.delete(sessionId);
      // Re-read: the session may have moved device, or been purged, meanwhile.
      const row = sessions.getSession(sessionId);
      if (!row || row.purged_at) return false;
      const dbModule = require("../db");
      const { importFromDirectory } = require("../../scripts/import-history");
      const counters = await importFromDirectory(
        dbModule,
        sessionRoot(row.user_id, row.device_id, row.session_id)
      );
      // The import made (or grew) the session row: it is this device's user's.
      // Never over a session someone else owns (defence in depth: the
      // registration already refuses a session id the hub knew).
      sql(
        "UPDATE sessions SET source = ?, owner_id = ? WHERE id = ? AND (owner_id IS NULL OR owner_id = ?) AND lane_id IS NULL"
      ).run(`device:${row.device_id}`, row.user_id, row.session_id, row.user_id);
      return counters.imported > 0 || counters.backfilled > 0;
    });
  importing.set(sessionId, run);
  run
    .finally(() => {
      if (importing.get(sessionId) === run) importing.delete(sessionId);
    })
    .catch(() => {});
  return run;
}

/**
 * A session moves to another device of the SAME user (registerSession): its
 * copy moves with it, so the next upload continues the same files.
 */
function moveSession(sessionId, fromDevice, toDevice, userId) {
  const from = sessionRoot(userId, fromDevice, sessionId);
  const to = sessionRoot(userId, toDevice, sessionId);
  if (fs.existsSync(from)) {
    fs.mkdirSync(path.dirname(to), { recursive: true, mode: 0o700 });
    fs.renameSync(from, to);
  }
  const { db } = require("../db");
  db.transaction(() => {
    db.prepare("UPDATE remote_sessions SET device_id = ? WHERE session_id = ?").run(
      toDevice,
      sessionId
    );
    db.prepare(
      "UPDATE sessions SET source = ?, transcript_path = replace(transcript_path, ?, ?) WHERE id = ?"
    ).run(`device:${toDevice}`, from, to, sessionId);
  })();
}

/** Where the hub stands on every file of a session (a client re-syncs from it). */
function fileState(device, sessionId) {
  const row = sessions.sessionOfDevice(sessionId, device.id);
  if (!row) throw err("ENOSESSION", "this device registered no such session", 404);
  return sql("SELECT rel_path AS path, consumed FROM remote_files WHERE session_id = ?").all(
    row.session_id
  );
}

/**
 * Remove the conversation of every remote session whose last upload (or
 * registration) is older than `days`: the copied transcripts and the
 * session's events go; the session row and its token rows stay.
 */
function purgeOld({ days = RETAIN_DAYS, now = Date.now() } = {}) {
  const cutoff = new Date(now - days * 864e5).toISOString();
  const { db } = require("../db");
  const old = db
    .prepare(
      `SELECT * FROM remote_sessions
        WHERE purged_at IS NULL AND COALESCE(last_upload_at, registered_at) < ?`
    )
    .all(cutoff);
  for (const row of old) {
    try {
      fs.rmSync(sessionRoot(row.user_id, row.device_id, row.session_id), {
        recursive: true,
        force: true,
      });
    } catch (e) {
      console.warn(`[remote] could not remove ${row.session_id}: ${e.message}`);
      continue;
    }
    pending.delete(row.session_id);
    clearTimeout(flushTimers.get(row.session_id));
    flushTimers.delete(row.session_id);
    db.transaction(() => {
      db.prepare("DELETE FROM events WHERE session_id = ?").run(row.session_id);
      db.prepare("UPDATE sessions SET transcript_path = NULL WHERE id = ?").run(row.session_id);
      db.prepare("DELETE FROM remote_files WHERE session_id = ?").run(row.session_id);
      db.prepare(
        "UPDATE remote_sessions SET purged_at = strftime('%Y-%m-%dT%H:%M:%fZ','now') WHERE session_id = ?"
      ).run(row.session_id);
    })();
  }
  return old.length;
}

module.exports = {
  RETAIN_DAYS,
  takeUpload,
  fileState,
  moveSession,
  purgeOld,
  sessionRoot,
  __pending: () => [...pending],
  __fileFor: fileFor,
};