balls 0.5.10

//! §16 `bl import` — the write inverse of the bedrock read (`show --json`).
//!
//! `show --json` reads the lossless record OUT; `import` writes the same shape
//! BACK IN, through the real store and the real `task.rs` serializer: id and
//! timestamps verbatim, NO mint / NO stamp / NO gate. It is not
//! migration-specific — "reproduce a task that already has an identity" is the
//! shared primitive behind federation join, restore-from-backup, and
//! clone-seeding — and it is deliberately not `create --id`: `create` mints a
//! NEW identity (and rightly refuses a foreign one); `import`'s ids are
//! authoritative at the source.
//!
//! **Collision semantics — refuse, don't guess (§16):** an id that already
//! lives in the store (or repeats within the stream) aborts the WHOLE stream
//! before anything is written, exit nonzero naming the id(s). Skip would lie
//! (a "restored" ball silently isn't), replace would destroy local state on no
//! explicit signal — both guesses; the caller composes the remedy (`bl close`
//! the holder, or filter the stream) with the verbs that already exist, so no
//! flag is added.
//!
//! `--legacy[=REF]` is the cutover button: pure orchestration over the §16
//! read shim + this primitive — exactly `bl list --legacy --json | bl import`
//! in-process, plus the epic reciprocal-edge pass wired through the ordinary
//! `update --needs` machinery. It carries no policy the pipe can't express.

use std::collections::{BTreeMap, HashSet};
use std::io::{self, Read};
use std::path::Path;

use crate::clock;
use crate::edge::Edge;
use crate::lifecycle::BaseChange;
use crate::message::Message;
use crate::mutate::{self, Op};
use crate::reads::legacy;
use crate::reads::scope;
use crate::task::Task;
use crate::taskfile::{exists, write_task};
use crate::verb::Verb;
use crate::wire::Command;

#[path = "import_stream.rs"]
mod stream;

/// Run `bl import`: parse flags, source the records — the verbatim bedrock
/// JSON on `input` (one object, an array, or a concatenated stream: whatever
/// `show --json`/`list --json` emit), or the §16 legacy projection under
/// `--legacy` — and seal them in. `input` is the caller's stdin, injected so
/// the verb is testable and never read on the `--legacy` path.
pub fn run(edge: &Edge, input: &mut dyn Read, args: &[String]) -> io::Result<()> {
    let flags = parse(args, &edge.default_actor)?;
    if let Some(spec) = &flags.legacy {
        return button(edge, &flags, spec);
    }
    let mut text = String::new();
    input.read_to_string(&mut text)?;
    ingest(edge, &flags, stream::records(&text)?)
}

/// `import`'s flags: identity, narration, the §12 `--remote` per-op override, and
/// the `--legacy[=REF]` button. No positionals — records ride stdin.
#[derive(Debug)]
struct Flags {
    actor: String,
    message: Option<String>,
    remote: Option<String>,
    legacy: Option<String>,
}

/// Parse `import`'s argv. The flag vocabulary is deliberately the mutating
/// verbs' (`--as`/`-m`/`--remote`) plus the §16 `--legacy` spelling shared with
/// the read verbs ([`legacy::flag`]). `--center` is prime-only (enrollment,
/// bl-35e5), so on `import` it falls through to the unexpected-argument error.
fn parse(args: &[String], default_actor: &str) -> io::Result<Flags> {
    let mut f = Flags { actor: default_actor.to_string(), message: None, remote: None, legacy: None };
    let mut args = args.iter();
    while let Some(arg) = args.next() {
        match arg.as_str() {
            "--as" => f.actor = value(&mut args, "--as")?,
            "-m" | "--message" => f.message = Some(value(&mut args, "-m")?),
            "--remote" => f.remote = Some(value(&mut args, "--remote")?),
            a if legacy::flag(a).is_some() => f.legacy = legacy::flag(a),
            other => {
                return Err(crate::usage(format!(
                    "import: unexpected argument '{other}' — records ride stdin"
                )))
            }
        }
    }
    Ok(f)
}

/// The value following a value-taking flag, or a "needs a value" error.
fn value(args: &mut std::slice::Iter<'_, String>, flag: &str) -> io::Result<String> {
    args.next().cloned().ok_or_else(|| crate::usage(format!("import: {flag} needs a value")))
}

/// Seal `balls` into the store as ONE op through the shared engine wiring
/// ([`mutate::seal_op`]) — one commit, all-or-nothing, so the refuse-the-stream
/// collision check and the §14 rollback give the same guarantee: a failed
/// import imports nothing. Confirmation on stderr; stdout stays silent (§9 —
/// the caller supplied every id, so the verb has no product to print).
fn ingest(edge: &Edge, flags: &Flags, balls: Vec<(String, Task)>) -> io::Result<()> {
    let n = balls.len();
    let base = Ingest { balls, actor: flags.actor.clone(), message: flags.message.clone() };
    let op = Op {
        actor: flags.actor.clone(),
        remote: flags.remote.clone(),
        // A bulk import authors no delivery and no single ball — no id, no
        // worktree, no target (§16).
        command: Command {
            op: Verb::Import.token().to_string(),
            id: None,
            body_change: None,
            message: None,
            target: None,
        },
    };
    // Records carry their OWN timestamps verbatim (§16 — no restamp), so the op
    // instant dates only the seal commit, keeping the import's store commit on the
    // same one-clock-per-op discipline as every other seal (§8, bl-8b98).
    let instant = clock::for_op(edge)?;
    mutate::seal_op(edge, Verb::Import, &op, &base, None, &instant)?;
    eprintln!("import {n} ball{}", if n == 1 { "" } else { "s" });
    if let Some(hint) = foreign_root_hint(&edge.invocation_path, &base.balls) {
        eprintln!("{hint}");
    }
    Ok(())
}

/// The foreign-root hint (bl-d3fa): ONE stderr line decorating a SUCCESSFUL
/// import whose records carry a `root_commit` this checkout does not answer to.
///
/// Nothing is wrong — the balls are real, `show` resolves them — but the
/// root-aware default `list` scope (bl-0161) rightly hides them, so the
/// operator's very next command reads as a contradiction of the confirmation
/// just printed: "import 1 ball", then an empty `bl list`. Correct behavior can
/// still be silent in the wrong place. So the success is decorated in the
/// `[source]`-hint voice — name the fact, name the lifted-scope read — and nothing
/// else changes: no flag, no scope change, no refusal. The hint is the fix.
///
/// The root walk is LAZY on [`scope::checkout_roots`]'s own rule (skipped when no
/// record carries a root — a rootless stream stays git-free) and runs only AFTER
/// the seal, so a refused stream pays nothing for a line it will never print.
fn foreign_root_hint(invocation_path: &Path, balls: &[(String, Task)]) -> Option<String> {
    let carried = balls.iter().any(|(_, t)| t.root_commit.is_some());
    let roots = scope::checkout_roots(invocation_path, carried);
    let foreign = balls.iter().filter(|(_, t)| scope::is_foreign(t.root_commit.as_deref(), &roots)).count();
    (foreign > 0).then(|| {
        format!(
            "import: {foreign} of {} rooted in another project, hidden from this checkout's default `bl list` scope — read with `bl list --everywhere`",
            balls.len()
        )
    })
}

/// The bulk [`BaseChange`]: write every record verbatim through the §3
/// serializer ([`write_task`]). No enforce gate runs — the records are
/// reproductions, not transitions — and nothing is restamped.
struct Ingest {
    balls: Vec<(String, Task)>,
    actor: String,
    message: Option<String>,
}

impl BaseChange for Ingest {
    fn stage(&self, dir: &Path) -> io::Result<()> {
        // Collision = an id the store already holds OR one repeated in the
        // stream: refuse the WHOLE stream before writing anything.
        let mut seen = HashSet::new();
        let held: Vec<&str> = self
            .balls
            .iter()
            .map(|(id, _)| id.as_str())
            .filter(|id| !seen.insert(*id) || exists(dir, id))
            .collect();
        if !held.is_empty() {
            return Err(io::Error::new(
                io::ErrorKind::AlreadyExists,
                format!(
                    "import: id(s) already held: {} — refusing the whole stream, nothing imported (retire the holder or strip the record, then re-run)",
                    held.join(", ")
                ),
            ));
        }
        for (id, task) in &self.balls {
            write_task(dir, id, task)?;
        }
        Ok(())
    }

    fn finalize(&self, _dir: &Path) -> io::Result<String> {
        // A bulk op names no single ball, so it takes §5's id-less form (like
        // a checkout-scoped op): the subject counts, the file diff is the record.
        let n = self.balls.len();
        Message {
            verb: Verb::Import,
            actor: self.actor.clone(),
            id: None,
            subject: format!("import {n} ball{}", if n == 1 { "" } else { "s" }),
            body: self.message.clone(),
        }
        .render()
    }

    fn narrated(&self) -> bool {
        // bl-cf93's loud-loss rule: an empty stream with `-m` would converge the
        // seal and silently drop the note — abort instead (≥1 record never
        // converges, the collision check refuses every existing id first).
        self.message.is_some()
    }
}

/// `bl import --legacy[=REF]` — the §16 cutover button: read the legacy store
/// through the SAME projection `bl list --legacy` renders (the pipe,
/// in-process), import every live ball, then mint the epic reciprocal edges —
/// each live child claim-blocks its live parent — as ordinary `update --needs`
/// ops, one per parent. Legacy derived "an epic waits on its children" from
/// status; greenfield `parent:` is containment-only (§10), so the edge is a
/// reconstruction, and it rides the real edge machinery (real ops, real
/// `updated` stamps) rather than a second transform path.
///
/// The edge stays a CLAIM edge, deliberately — not the close-gate `--subtask-of`
/// now mints (bl-e844). It reproduces what the legacy store meant, and a
/// migrated epic therefore keeps FLAT delivery (a claim-gate is not the §11
/// nesting declaration, [`crate::target`]). Nesting is opted into by spelling a
/// close edge afterwards, never by conversion in the migration path.
fn button(edge: &Edge, flags: &Flags, spec: &str) -> io::Result<()> {
    let balls = legacy::balls(&edge.invocation_path, spec)?;
    // The projection already nulled dangling parents (§16), so every surviving
    // `parent` is in the imported live set.
    let mut edges: BTreeMap<String, Vec<String>> = BTreeMap::new();
    for (id, task) in &balls {
        if let Some(parent) = &task.parent {
            edges.entry(parent.clone()).or_default().push(id.clone());
        }
    }
    ingest(edge, flags, balls)?;
    for (parent, children) in edges {
        let mut args = vec![parent];
        for child in children {
            args.push("--needs".into());
            args.push(child);
        }
        args.extend(["--as".into(), flags.actor.clone()]);
        if let Some(remote) = &flags.remote {
            args.extend(["--remote".into(), remote.clone()]);
        }
        mutate::run(edge, Verb::Update, &args)?;
    }
    Ok(())
}

#[cfg(test)]
#[path = "import_tests.rs"]
mod tests;