boatramp 0.42.0

//! The `boatramp migrate` command: re-key a pre-0.2.0 (layout 1) control-plane store
//! to the project-scoped (0.2.0) layout, **offline** — the server refuses to serve an
//! unmigrated store (see the serve-startup guard), so this is the explicit,
//! operator-run step that unblocks it.
//!
//! The engine lives in [`boatramp_core::migrate`]; this wires it to the same KV store
//! `serve` opens (`--kv` / `--data-dir` must match). Supports a `--dry-run` audit, a
//! `--stage` copy-only pass (leaving the old keys for a soak/rollback window), and a
//! `--finalize` pass that deletes the staged old keys.

use std::path::PathBuf;

use boatramp_core::migrate::{self, MigrateOptions, MigrationReport, Status};

use crate::config::ServerConfig;
use crate::serve::build_control_plane_kv;
use boatramp_node::backends::KvBackend;

/// Errors from the `migrate` command.
#[derive(Debug, thiserror::Error)]
pub enum Error {
    /// Opening the control-plane KV store failed (delegated to the serve backend
    /// builder, so the same `--kv`/feature guidance applies).
    #[error(transparent)]
    Store(#[from] Box<crate::serve::Error>),
    /// The migration engine failed.
    #[error(transparent)]
    Migrate(#[from] boatramp_core::migrate::MigrateError),
    /// Flushing the migrated store to durable storage failed.
    #[error("flushing the migrated store failed: {0}")]
    Flush(String),
    /// Loading the `--node-config` for the `--finalize` substrate gate failed.
    #[error(transparent)]
    Config(#[from] boatramp_node::config::ConfigError),
    /// A control-plane KV read failed during the `--finalize` substrate gate.
    #[error(transparent)]
    Kv(#[from] boatramp_core::kv::KvError),
    /// `--finalize` is blocked (v0.31.0 Finding 8, HIGH-2c): clearing the KV dual soak would
    /// disengage the runtime blob read-fallback, so the blob + libsql substrates must be re-homed
    /// first. Reports which remains.
    #[error(
        "refusing `--finalize`: the default-scope re-home is not substrate-complete — {reason}. \
         Clearing the KV dual soak now would disengage the blob read-fallback and strand data. \
         Complete `boatramp blob rehome --finalize` + `boatramp sql rehome-default`, then re-run. \
         See docs/default-scope-migration.md."
    )]
    FinalizeNotSubstrateComplete {
        /// What is still outstanding (libsql bare DBs and/or the blob-retired marker).
        reason: String,
    },
    /// Reading the operator's input during an interactive migration failed.
    #[error("interactive migration: {0}")]
    Interactive(String),
}

/// `boatramp migrate` arguments.
#[derive(Debug, clap::Args)]
pub struct MigrateArgs {
    /// Data directory (flag/env > `[serve].data_dir` > `./data`). Must match `serve`.
    #[arg(long, env = "BOATRAMP_DATA_DIR")]
    data_dir: Option<PathBuf>,

    /// Metadata (KV) backend. Must match `serve`.
    #[arg(long, value_enum, default_value_t = KvBackend::Slatedb)]
    kv: KvBackend,

    /// Report what the migration would change, writing nothing.
    #[arg(long)]
    dry_run: bool,

    /// Copy + verify, keeping the old-layout keys for a soak/rollback window (the `dual` state).
    /// v0.31.0: this is now the DEFAULT behavior (the default-scope re-home MUST soak so the
    /// translating blob read-fallback engages), so the flag is accepted for clarity/compat but the
    /// forward pass always stages — a later `--finalize` reclaims the old keys.
    #[arg(long, conflicts_with_all = ["dry_run", "finalize"])]
    stage: bool,

    /// Delete the old-layout keys left by the staged forward pass, completing the migration. For the
    /// default-scope re-home (v0.31.0 Finding 8) this REFUSES until the blob + libsql substrates are
    /// re-homed (it reads `--node-config` to check), because clearing the dual soak disengages the
    /// blob read-fallback.
    #[arg(long, conflicts_with = "dry_run")]
    finalize: bool,

    /// Abort a staged (dual-soak) migration, rolling the store back to its exact pre-forward bytes —
    /// the reverse of a `stage`. OFFLINE (run with the node stopped). Fail-closed: refuses, deleting
    /// nothing, if any staged key received a write during the soak (that write would be lost); finalize
    /// forward instead, or reconcile manually. (libsql: abort removes the re-homed COPY; the retained
    /// bare DB is the rollback target — see docs/default-scope-migration.md.)
    #[arg(long, conflicts_with_all = ["dry_run", "stage", "finalize"])]
    abort: bool,

    /// Interactive (question-driven) migration: show the plan, PROMPT at each decision point (a
    /// name collision → rename / skip / abort), then stage honoring your answers. The guided CLI flow
    /// over the same `plan` + apply engine the console wizard uses. OFFLINE (node stopped).
    #[arg(long, short = 'i', conflicts_with_all = ["dry_run", "stage", "finalize", "abort"])]
    interactive: bool,

    /// Node config (`boatramp.cfg` shape) used by `--finalize` to locate the libsql dir (+ the blob
    /// retired marker) for the substrate-complete gate. Defaults to `boatramp.cfg`.
    #[arg(long, default_value = "boatramp.cfg")]
    node_config: PathBuf,
}

/// The result of a `boatramp migrate` run, rendered by [`crate::output`].
///
/// The `--interactive` flow prints its plan + decision prompts live (it is an
/// interactive stdin/stdout session); the terminal verdict it ends on is the typed
/// result here, rendered once by `output::emit`.
#[derive(Debug, serde::Serialize)]
#[serde(tag = "result", rename_all = "snake_case")]
pub enum MigrateOutput {
    /// `--abort` rolled the staged migration back; `status` is the store's resulting layout.
    Aborted { status: Status },
    /// A non-interactive pass (forward / `--dry-run` / `--stage` / `--finalize`): the report.
    Report {
        report: MigrationReport,
        dry_run: bool,
    },
    /// `-i`: the store is already at the current layout — nothing to do.
    InteractiveReady,
    /// `-i`: the store is already staged (dual soak).
    InteractiveAlreadyStaged,
    /// `-i`: the operator chose to abort at a decision prompt.
    InteractiveAbortedAtDecision,
    /// `-i`: out-of-band steps must be resolved first, then the plan re-run.
    InteractivePendingExternal { steps: Vec<String> },
    /// `-i`: the operator declined the final stage confirmation.
    InteractiveCancelled,
    /// `-i`: the apply engine aborted (a precheck returned `Aborted`).
    InteractiveAbortedDuringApply,
    /// `-i`: staged into the dual soak; `report` is the staged pass.
    InteractiveStaged { report: MigrationReport },
}

impl crate::output::Rendered for MigrateOutput {
    fn render_human(&self, w: &mut dyn std::io::Write) -> std::io::Result<()> {
        match self {
            Self::Aborted { status } => {
                writeln!(
                    w,
                    "aborted the staged migration — the store is rolled back to its pre-migration layout \
                     (old keys only); a prior binary can serve it, and `boatramp migrate` can re-stage it."
                )?;
                writeln!(w, "now: {}", describe(*status))
            }
            Self::Report { report, dry_run } => write_report(report, *dry_run, w),
            Self::InteractiveReady => {
                writeln!(
                    w,
                    "nothing to do: the store is already at the current layout."
                )
            }
            Self::InteractiveAlreadyStaged => writeln!(
                w,
                "the store is already staged (dual soak). Re-home the blob + libsql substrates, then \
                 `boatramp migrate --finalize` to complete it — or `boatramp migrate --abort` to roll \
                 it back."
            ),
            Self::InteractiveAbortedAtDecision => {
                writeln!(w, "\naborting the migration — nothing was changed.")
            }
            Self::InteractivePendingExternal { steps } => {
                writeln!(
                    w,
                    "\nResolve these first (out of band), then re-run `boatramp migrate --interactive` (the \
                     decision clears once the collision is gone):"
                )?;
                for step in steps {
                    writeln!(w, "  • {step}")?;
                }
                Ok(())
            }
            Self::InteractiveCancelled => writeln!(w, "cancelled."),
            Self::InteractiveAbortedDuringApply => {
                writeln!(w, "aborted — nothing was changed.")
            }
            Self::InteractiveStaged { report } => {
                write_report(report, false, w)?;
                writeln!(
                    w,
                    "\nStaged into the dual soak. NEXT:\n  \
                     1. RESTART the node so it adopts the new layout (a running daemon keeps serving the \
                     OLD layout until restarted).\n  \
                     2. Re-home the other substrates: `boatramp sql rehome-default` + `boatramp blob \
                     rehome`.\n  \
                     3. `boatramp migrate --finalize` to reclaim the old keys (gated until step 2).\n  \
                     Reversible until finalize: `boatramp migrate --abort`."
                )
            }
        }
    }
}

/// Run the `migrate` command.
pub async fn run(args: MigrateArgs, config: &ServerConfig) -> Result<MigrateOutput, Error> {
    let data_dir = args
        .data_dir
        .clone()
        .or_else(|| config.serve.as_ref().and_then(|s| s.data_dir.clone()))
        .unwrap_or_else(|| PathBuf::from("./data"));

    // The pre-0.2.0 → project-scoped re-key is a local-store migration (a fresh
    // remote-state deploy has no layout-1 store), so open the local SlateDB. Open STRICT here
    // (v0.9.0 KV-recovery): a migration is a re-key, not a crash recovery — a torn tail must fail
    // loud so the operator recovers the store FIRST (`boatramp kv recover`), never auto-quarantine
    // mid-migration.
    let kv = build_control_plane_kv(
        args.kv,
        &data_dir,
        None,
        boatramp_core::kv::KvOpenPolicy::Strict,
        // `migrate` is a SlateDB-store re-key, not a SQL target — no `[serve.kv.sql]` config.
        None,
    )
    .await
    .map_err(|e| Box::new(crate::serve::Error::from(e)))?;

    // Opening-status preamble: a diagnostic about the store being opened, not the
    // command's result — it rides the `tracing` sink (stderr).
    tracing::info!(
        "control-plane store ({:?} at {}): {}",
        args.kv,
        data_dir.display(),
        describe(migrate::status(kv.as_ref()).await?)
    );

    // ABORT: roll a staged (dual-soak) migration back to the pre-migration layout. Flushes, then
    // reports what it did plus the resulting status (both carried by the typed result).
    if args.abort {
        migrate::abort(kv.as_ref()).await?;
        kv.flush().await.map_err(|e| Error::Flush(e.to_string()))?;
        let status = migrate::status(kv.as_ref()).await?;
        return Ok(MigrateOutput::Aborted { status });
    }

    // INTERACTIVE: the question-driven flow (plan → prompt decisions → stage honoring the answers).
    if args.interactive {
        let out = interactive_migrate(kv.as_ref()).await?;
        kv.flush().await.map_err(|e| Error::Flush(e.to_string()))?;
        return Ok(out);
    }

    let report = if args.finalize {
        // Substrate-complete gate (Finding 8, HIGH-2c): clearing the KV dual soak disengages the
        // runtime blob read-fallback, so refuse until blob (the marker `blob rehome --finalize` sets)
        // AND libsql (a cheap on-disk scan) are re-homed.
        finalize_substrate_gate(kv.as_ref(), &args.node_config, &data_dir).await?;
        migrate::finalize(kv.as_ref()).await?
    } else {
        // FORCE STAGE (Finding 8, HIGH-2a): the default-scope re-home MUST enter the dual soak so the
        // translating blob read-fallback engages (a one-shot would strand cold default blobs). So the
        // forward pass never inline-finalizes; `--finalize` (gated above) reclaims the old keys.
        // Offline CLI: no object-store handle, so pass `None` — the data-aware collision gate then
        // treats every commingle as blocking (conservative; the relaxation is online/wizard-only).
        migrate::migrate(
            kv.as_ref(),
            MigrateOptions {
                dry_run: args.dry_run,
                finalize: false,
            },
            None,
        )
        .await?
    };

    // Force the migration durable (SlateDB flushes on a timer otherwise) before we
    // exit, so a `serve` started right after sees the completed layout.
    if !args.dry_run {
        kv.flush().await.map_err(|e| Error::Flush(e.to_string()))?;
    }
    Ok(MigrateOutput::Report {
        report,
        dry_run: args.dry_run,
    })
}

/// The interactive (question-driven) migration flow: show the plan, prompt the operator at each
/// decision (a name collision → rename / skip / abort), and stage honoring the answers. The CLI half
/// of the interactive migration — it drives the SAME `migrate::plan` + `migrate_with_answers` engine
/// the console wizard uses, so the two surfaces can never diverge. Thin stdin/stdout glue; the engine
/// logic (skip/abort/refuse) is unit-tested in `boatramp-core`.
async fn interactive_migrate(kv: &dyn boatramp_core::kv::KvStore) -> Result<MigrateOutput, Error> {
    let plan = migrate::plan(kv, None).await?; // offline CLI: no blob scanner (see the stage path above)
    match plan.status {
        Status::Ready => {
            return Ok(MigrateOutput::InteractiveReady);
        }
        Status::Dual => {
            return Ok(MigrateOutput::InteractiveAlreadyStaged);
        }
        Status::NeedsMigration => {}
    }

    // The plan + decision prompts below are printed live: this is an interactive
    // stdin/stdout session, so the operator must see them before answering. The
    // terminal verdict the flow ends on is the typed `MigrateOutput` returned here.
    println!("\nMigration plan:");
    for line in &plan.summary {
        println!("  • {line}");
    }

    // Collect a route per decision. GENERIC: the CLI acts on each route's `action` (the engine's
    // migration-agnostic contract), never on a route-`key` string — so a future migration's decisions
    // render + act with no CLI change. `External` routes (the operator resolves out-of-band, then
    // re-runs) are collected as guidance and never staged; `Stage` routes go into the answers; `Abort`
    // short-circuits.
    let mut answers = migrate::MigrationAnswers::default();
    let mut pending_external: Vec<String> = Vec::new();
    for d in &plan.decisions {
        println!("\n{}", d.question);
        for (i, r) in d.routes.iter().enumerate() {
            println!("  {}) {}", i + 1, r.label);
            if let Some(note) = &r.note {
                println!("     ({note})");
            }
        }
        let chosen = prompt_route(d)?;
        if chosen.action == migrate::RouteAction::Abort {
            return Ok(MigrateOutput::InteractiveAbortedAtDecision);
        }
        // Collect the route's free-text input (e.g. a new name), if it asks for one.
        let value = if let Some(input) = &chosen.input {
            Some(
                prompt_line(&format!("  {}: ", input.prompt))?
                    .trim()
                    .to_string(),
            )
        } else {
            None
        };
        match chosen.action {
            migrate::RouteAction::External => {
                // Must be performed out-of-band first; carry the migration's own guidance (its `note`)
                // verbatim so the CLI needs no per-migration knowledge of what the route entails.
                let resource = d.id.strip_prefix("collision:").unwrap_or(&d.id);
                let note = chosen
                    .note
                    .as_deref()
                    .unwrap_or("resolve this externally, then re-run the plan");
                let detail = value
                    .as_deref()
                    .map(|v| format!(" (you chose: `{v}`)"))
                    .unwrap_or_default();
                pending_external.push(format!("{resource}{detail}: {note}"));
            }
            migrate::RouteAction::Stage => {
                answers.choices.insert(
                    d.id.clone(),
                    migrate::DecisionChoice {
                        route: chosen.key.clone(),
                        value,
                    },
                );
            }
            migrate::RouteAction::Abort => unreachable!("handled above"),
        }
    }

    // Any out-of-band route must be done first (through the proven deploy path) BEFORE staging —
    // report the guidance and stop; the operator re-runs once the collisions are gone.
    if !pending_external.is_empty() {
        return Ok(MigrateOutput::InteractivePendingExternal {
            steps: pending_external,
        });
    }

    if !prompt_yes(
        "\nProceed to STAGE the re-home now (non-destructive; reversible with --abort)?",
    )? {
        return Ok(MigrateOutput::InteractiveCancelled);
    }

    match migrate::migrate_with_answers(kv, &answers, None).await? {
        migrate::InteractiveOutcome::Aborted => Ok(MigrateOutput::InteractiveAbortedDuringApply),
        migrate::InteractiveOutcome::Staged(report) => {
            Ok(MigrateOutput::InteractiveStaged { report })
        }
    }
}

/// Prompt the operator to choose one of a decision's routes (by number or by key); re-prompts on an
/// invalid entry. EOF/`q` is treated as a cancel (maps to the `abort` route if present, else errors).
fn prompt_route(d: &migrate::MigrationDecision) -> Result<&migrate::DecisionRoute, Error> {
    loop {
        let line = prompt_line("  choose [number or name]: ")?;
        let line = line.trim();
        if let Ok(n) = line.parse::<usize>()
            && n >= 1
            && n <= d.routes.len()
        {
            return Ok(&d.routes[n - 1]);
        }
        if let Some(r) = d.routes.iter().find(|r| r.key.eq_ignore_ascii_case(line)) {
            return Ok(r);
        }
        println!("  (please enter 1–{} or a route name)", d.routes.len());
    }
}

/// Read one line from stdin after printing `prompt`.
fn prompt_line(prompt: &str) -> Result<String, Error> {
    use std::io::Write;
    print!("{prompt}");
    std::io::stdout()
        .flush()
        .map_err(|e| Error::Interactive(e.to_string()))?;
    let mut line = String::new();
    let n = std::io::stdin()
        .read_line(&mut line)
        .map_err(|e| Error::Interactive(e.to_string()))?;
    if n == 0 {
        return Err(Error::Interactive("unexpected end of input".to_string()));
    }
    Ok(line)
}

/// A yes/no confirmation (default NO).
fn prompt_yes(question: &str) -> Result<bool, Error> {
    let line = prompt_line(&format!("{question} [y/N] "))?;
    Ok(matches!(line.trim(), "y" | "Y" | "yes" | "YES"))
}

/// The substrate-complete gate for `--finalize` (v0.31.0 Finding 8, HIGH-2c): clearing the KV dual
/// soak disengages the runtime blob read-fallback, so refuse until BOTH the blob substrate is retired
/// (the marker `boatramp blob rehome --finalize` sets) AND the default-project libsql is re-homed (no
/// bare `{dir}/*.db`). A remote sqld libsql binding (no local dir) is a documented operator step and
/// is not gated here. Reports every outstanding substrate at once.
async fn finalize_substrate_gate(
    kv: &dyn boatramp_core::kv::KvStore,
    node_config: &std::path::Path,
    data_dir: &std::path::Path,
) -> Result<(), Error> {
    let node_cfg = boatramp_node::config::ServerConfig::load(node_config, None)?;
    let mut outstanding: Vec<String> = Vec::new();

    // libsql: a cheap on-disk scan of the node's libsql dir (local bindings only). Use the SAME
    // run-level `data_dir` the serve guard + `assemble` use (honoring `--data-dir`), NOT
    // `node_cfg.serve.data_dir` — a `--data-dir`-overridden node keeps its DBs under the override.
    if let Some(dir) = boatramp_node::default_scope::libsql_dir_for(&node_cfg, data_dir) {
        let projects = boatramp_node::default_scope::known_projects(kv).await?;
        let bare = boatramp_node::default_scope::libsql_bare_default_idents(&dir, &projects);
        if !bare.is_empty() {
            outstanding.push(format!(
                "libsql not re-homed ({bare:?}) — run `boatramp sql rehome-default --site <site>`"
            ));
        }
    }
    // blob: require the marker `blob rehome --finalize` sets once the blob substrate is verified
    // retired (a node with no blob data still sets it by running that command once — a 0-object pass).
    if kv
        .get(boatramp_node::default_scope::BLOB_RETIRED_MARKER)
        .await?
        .is_none()
    {
        outstanding.push(
            "blob not verified-retired — run `boatramp blob rehome` (drain) then `boatramp blob \
             rehome --finalize` (retire)"
                .to_string(),
        );
    }

    if outstanding.is_empty() {
        Ok(())
    } else {
        Err(Error::FinalizeNotSubstrateComplete {
            reason: outstanding.join("; "),
        })
    }
}

/// A one-line description of a store's layout status.
fn describe(status: Status) -> &'static str {
    match status {
        Status::Ready => "ready (empty or already at the current schema version)",
        Status::NeedsMigration => "below the current schema version — needs migration",
        Status::Dual => "dual soak (migrated; old keys awaiting `--finalize`)",
    }
}

/// Write a human-readable summary of a migration pass.
fn write_report(
    report: &MigrationReport,
    dry_run: bool,
    w: &mut dyn std::io::Write,
) -> std::io::Result<()> {
    if report.already_migrated {
        return writeln!(w, "nothing to do: the store is already fully migrated.");
    }
    let verb = if dry_run { "would re-key" } else { "re-keyed" };
    if report.rekeyed.is_empty() && report.values_rewritten.is_empty() {
        writeln!(w, "no layout-1 records found.")?;
    }
    for (family, n) in &report.rekeyed {
        writeln!(w, "  {verb} {n} key(s) under {family}")?;
    }
    for (family, n) in &report.values_rewritten {
        let v = if dry_run { "would rewrite" } else { "rewrote" };
        writeln!(
            w,
            "  {v} {n} {family} index value(s) to the (project, site) form"
        )?;
    }
    if report.created_default_project {
        let v = if dry_run { "would create" } else { "created" };
        writeln!(w, "  {v} the `default` project")?;
    }
    if report.owner_entries > 0 {
        writeln!(
            w,
            "  built {} owner reverse-index entries",
            report.owner_entries
        )?;
    }
    if dry_run {
        writeln!(w, "(dry run — nothing was written)")?;
    } else if report.dual {
        writeln!(
            w,
            "staged to 2-dual — old keys kept; run `boatramp migrate --finalize` to reclaim them."
        )?;
    } else {
        writeln!(
            w,
            "migration complete: store is at the project-scoped layout."
        )?;
    }
    Ok(())
}