stack/packages/cli/src/thread.mjs

// Row 128 (#1622, REQ-CLI-1, REQ-CLI-2, REQ-THREAD-2, REQ-USER-2): the human's side of the
// bus's project and thread ops (row 121). `mosaic threads`, `mosaic thread create|open|read`,
// and the thread resolver `mosaic talk --thread` and `mosaic follow --thread` (src/cli.mjs) use.
// The broker owns every rule: ownership, uniqueness, the one threaded instance. The CLI adds
// no check of those; it resolves a reference to an id and prints what the broker returns.
// Row 184 (#1682, Q173): each verb takes --tester and runs as the tester through tester-cli.mjs,
// on the tester's own threads (owner `tester`, row 177). The broker's ownership check is the only one.

import { CliError } from "./errors.mjs";
import { shortId } from "./format.mjs";
import { configDir, loadBusiness } from "../../business/src/index.mjs";
import { rolesDir } from "./config.mjs";

export const THREADS_USAGE = "mosaic threads [--project <project>] [--business <id>] [--json] [--tester]";
export const THREAD_USAGE = [
  "mosaic thread create <project> <name> [--business <id>] [--tester]",
  "mosaic thread open <thread> [--business <id>] [--tester]",
  "mosaic thread read <thread> [--business <id>] [--json] [--tester]",
].join("\n  ");
export const THREAD_TALK_USAGE = [
  "mosaic talk --thread <thread> <text> [--wait <seconds> | --follow] [--business <id>] [--tester]",
  "mosaic follow --thread <thread> [--business <id>] [--tester]",
].join("\n  ");

// The parse shape callFor reads (flags, values, positional). Errors carry the verb's own usage.
export function parseArgs(args, usage, { flags = [], values = [] } = {}) {
  const out = { positional: [], flags: new Set(), values: {} };
  const refuse = (why) => new CliError(`${why}\n${usage}`, 4);
  for (let i = 0; i < args.length; i++) {
    const a = args[i];
    if (flags.includes(a)) out.flags.add(a);
    else if (values.includes(a)) {
      if (i + 1 >= args.length || Object.hasOwn(out.values, a)) throw refuse(`${a} needs one value`);
      out.values[a] = args[++i];
    } else if (a.startsWith("--")) throw refuse(`unknown option ${a}`);
    else out.positional.push(a);
  }
  return out;
}

const UUID = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
const HEX = /^[0-9a-f-]+$/i;

// A project is its name or its id.
export async function resolveProject(call, ref, { business, env = process.env } = {}) {
  const projects = await call("project.list");
  const hit = projects.find((p) => p.id === ref) ?? projects.find((p) => p.name === ref);
  if (hit) return hit;
  let repository = false;
  try {
    const b = loadBusiness(business, { dir: configDir(env), rolesDir: rolesDir(env) });
    repository = Object.hasOwn(b.projects, ref);
  } catch { /* A diagnostic read cannot replace the original refusal if config is unavailable. */ }
  if (repository) throw new CliError(`${JSON.stringify(ref)} is a repository in the business file, not a thread project yet; mosaic project create ${ref} --repo ${ref} makes one`, 2);
  throw new CliError(`no project is named or numbered ${JSON.stringify(ref)}; see mosaic project list`, 2);
}

const labelOf = (projects, t) => `${projects.find((p) => p.id === t.project)?.name ?? shortId(t.project)}/${t.name}`;

// A thread is its id, an id prefix of at least 8 characters, or <project>/<name>. thread.list
// holds only the caller's threads, so a full id it doesn't list is passed on: the broker says
// thread-not-owned for another user's thread and thread-not-found for none (REQ-USER-2).
export async function resolveThread(call, ref) {
  const threads = await call("thread.list");
  const projects = await call("project.list");
  const exact = threads.find((t) => t.id === ref) ?? threads.find((t) => labelOf(projects, t) === ref);
  if (exact) return { id: exact.id, label: labelOf(projects, exact) };
  if (UUID.test(ref)) return { id: ref, label: ref };
  if (HEX.test(ref)) {
    if (ref.length < 8) throw new CliError(`thread reference ${JSON.stringify(ref)} is too short; use at least 8 characters of the id, or <project>/<name>`, 2);
    const hits = threads.filter((t) => t.id.startsWith(ref));
    if (hits.length === 1) return { id: hits[0].id, label: labelOf(projects, hits[0]) };
    if (hits.length > 1) throw new CliError(`${ref} matches ${hits.length} threads; use more of the id, or <project>/<name>`, 2);
  }
  throw new CliError(`no thread of yours matches ${JSON.stringify(ref)}; mosaic threads lists them as <project>/<name>`, 2);
}

const PAGE = 200;

// The thread's messages after `after`, oldest first, from every page of thread.messages.
// Reading writes no delivery or read row.
export async function threadMessages(call, thread, after) {
  const out = [];
  for (;;) {
    const page = await call("thread.messages", { thread, limit: PAGE, ...(after ? { after } : {}) });
    out.push(...page);
    if (page.length < PAGE) return out;
    after = page.at(-1).id;
  }
}

// A thread's message ids with their sequence numbers, kept current by refresh(). message.receive
// does not say which thread a message is in, so `--follow` receives first and asks the thread
// second; the broker records a message's thread when it stores the message, so a message it
// received is already listed.
export function threadMembers(call, thread) {
  const seqs = new Map();
  let last;
  return {
    seq: (id) => seqs.get(id),
    async refresh() {
      for (const m of await threadMessages(call, thread, last)) {
        seqs.set(m.id, m.seq);
        last = m.id;
      }
    },
  };
}

const json = (io, value) => io.stdout.write(`${JSON.stringify(value, null, 2)}\n`);

// Row 175 (#1673, Q173): with --tester the tester lists threads. Since row 177 (#1675) those are
// its own threads (owner tester).
export async function threadsCommand(args, io, callFor) {
  const parsed = parseArgs(args, THREADS_USAGE, { flags: ["--json", "--tester"], values: ["--business", "--project"] });
  if (parsed.positional.length !== 0) throw new CliError(`threads takes no arguments\n${THREADS_USAGE}`, 4);
  const call = callFor(parsed);
  const project = parsed.values["--project"] === undefined ? null : await resolveProject(call, parsed.values["--project"], { business: call.business, env: io.env });
  const threads = await call("thread.list", project ? { project: project.id } : {});
  if (parsed.flags.has("--json")) return json(io, threads);
  if (threads.length === 0) return io.stdout.write("no threads\n");
  const projects = await call("project.list");
  io.stdout.write(threads.map((t) => `${labelOf(projects, t)}  ${t.instance}  ${shortId(t.id)}\n`).join(""));
}

// `format` is cli.mjs's formatMessage, the one line talk and inbox print a message with.
export async function threadCommand(args, io, callFor, format) {
  const [sub, ...rest] = args;
  if (!["create", "open", "read"].includes(sub)) throw new CliError(`usage:\n  ${THREAD_USAGE}`, 4);
  const parsed = parseArgs(rest, THREAD_USAGE, { flags: sub === "read" ? ["--json", "--tester"] : ["--tester"], values: ["--business"] });
  if (parsed.positional.length !== (sub === "create" ? 2 : 1)) {
    throw new CliError(`thread ${sub} takes ${sub === "create" ? "a project and a name" : "one thread"}\n${THREAD_USAGE}`, 4);
  }
  const call = callFor(parsed);
  if (sub === "create") {
    const [ref, name] = parsed.positional;
    const project = await resolveProject(call, ref, { business: call.business, env: io.env });
    const made = await call("thread.create", { project: project.id, name });
    return io.stdout.write(`created ${project.name}/${name}  id ${made.id}  instance ${made.instance}\n`);
  }
  const thread = await resolveThread(call, parsed.positional[0]);
  if (sub === "open") {
    await call("thread.open", { thread: thread.id });
    return io.stdout.write(`opened ${thread.label}\n`);
  }
  const messages = await threadMessages(call, thread.id);
  if (parsed.flags.has("--json")) return json(io, messages);
  io.stdout.write(messages.length === 0 ? `${thread.label}: no messages\n` : messages.map(format).join(""));
}