gunnar-server 2.0.0

//! **Replication's configuration, over the wire and back.**
//!
//! One file, one direction each way, and it is deliberately not part of
//! [`crate::mirror`]. `mirror.rs` is the replicator: a subscriber, a queue and a
//! send-pack, and it knows nothing about gRPC. `control_plane.rs` is the
//! control plane and knows nothing about replication's types. This is the seam
//! between them, so neither has to grow a dependency on the other's vocabulary.
//!
//! # Why the operator surface exists at all
//!
//! The 2026-08-30 drill measured replication working end to end and then said
//! the thing that mattered: it was not in the appliance binary. Compiling it in
//! is half the job. The other half is that the reference appliance is an **OS-less
//! image running as PID 1** — no shell, no editor, no `scp` — so
//! "edit `<data-root>/replication.json` and restart" is not a procedure anyone
//! can carry out on the box that ships. A feature that ships and cannot be
//! turned on has not shipped.
//!
//! So `Mirror.SetReplication` / `RemoveReplication` / `ReplicationStatus`, and
//! this file is the translation under them.
//!
//! # What is NOT translated, and why
//!
//! [`ReplicationConfig`]'s tuning — `queue_capacity`, `retry_base_ms`,
//! `retry_max_ms`, `subscriber_disabled` — has no field on the wire. Those are
//! a deployment's shape, not a standby's identity, and the one that is a test
//! seam (`subscriber_disabled`) must never be reachable from a network call:
//! `mirror.rs` says so by name. `set_target` on a config read back from the
//! slot therefore preserves whatever they are, rather than resetting them to
//! defaults on every `SetReplication` — which is exactly what building a fresh
//! `ReplicationConfig` here would have done.

use std::path::{Path, PathBuf};

use gunnar_grpc::{ControlError, ControlResult};

use crate::mirror::{ReplicationScope, ReplicationTarget, Replicator, VisibilityScope};

/// **The deploy key the appliance pushes to its standby with**, under the data
/// root beside `replication.json`.
///
/// It is minted ON THE BOX and the private half never leaves it. That is not a
/// nicety, it is the only design that works here: an appliance with no shell
/// cannot be handed a file, and a private key travelling through a control-plane
/// RPC would be a private key in a caller's shell history, in a terminal
/// scrollback and in whatever logs that call passed through.
///
/// So the exchange runs the other way. `SetReplication` mints this key if it is
/// not there, and answers with its PUBLIC half; the operator authorises that one
/// line on the standby, exactly as they would any other deploy key. `0600`,
/// because it is a private key and the appliance is the only thing that has ever
/// held it.
pub const REPLICATION_KEY: &str = "replication_key";

/// Mint the replication deploy key if it does not exist, and return its path
/// and its public OpenSSH line.
///
/// Idempotent and stable: a second `SetReplication` reuses the same key, so an
/// operator who has already authorised it on the standby does not have to do it
/// again for every target they add.
pub fn ensure_identity(root: &Path) -> ControlResult<(PathBuf, String)> {
    let path = root.join(REPLICATION_KEY);
    let bytes = match std::fs::read(&path) {
        Ok(bytes) => bytes,
        Err(err) if err.kind() == std::io::ErrorKind::NotFound => {
            let minted = gunnar_core::mint_ed25519().map_err(|err| {
                ControlError::Internal(format!("minting the replication deploy key: {err}"))
            })?;
            // Write-then-rename, and 0600 BEFORE the bytes are visible under the
            // final name: a private key that is briefly world-readable is a
            // private key that was world-readable.
            let tmp = root.join(format!("{REPLICATION_KEY}.tmp"));
            std::fs::write(&tmp, &minted).map_err(|err| {
                ControlError::Internal(format!("writing {}: {err}", tmp.display()))
            })?;
            #[cfg(unix)]
            {
                use std::os::unix::fs::PermissionsExt;
                std::fs::set_permissions(&tmp, std::fs::Permissions::from_mode(0o600)).map_err(
                    |err| ControlError::Internal(format!("chmod 600 {}: {err}", tmp.display())),
                )?;
            }
            std::fs::rename(&tmp, &path).map_err(|err| {
                ControlError::Internal(format!("renaming {} into place: {err}", tmp.display()))
            })?;
            tracing::info!(
                target: crate::mirror::TARGET_REPLICATION,
                event = "replication.identity_minted",
                path = %path.display(),
                "minted this server's replication deploy key; its public half must be authorised \
                 on the standby"
            );
            minted
        }
        Err(err) => {
            return Err(ControlError::Internal(format!(
                "reading {}: {err}",
                path.display()
            )))
        }
    };
    let key = gunnar_core::HostKey::from_openssh(bytes).map_err(|err| {
        ControlError::Internal(format!("{} is not an OpenSSH key: {err}", path.display()))
    })?;
    let public = key.public_openssh().map_err(|err| {
        ControlError::Internal(format!("encoding the replication key's public half: {err}"))
    })?;
    Ok((path, public))
}

/// An empty string is "not set", never a path of length zero.
///
/// proto3 has no `null` for a scalar, so the wire cannot tell "the operator did
/// not name a key" from "the operator named the empty path". Treating `""` as
/// `None` is the only reading that is ever useful: an empty `identity_path`
/// falls back to the client key ladder, which is what a target with no key does
/// today, and an empty `known_hosts` means the server's own.
fn opt_path(s: String) -> Option<PathBuf> {
    let s = s.trim();
    (!s.is_empty()).then(|| PathBuf::from(s))
}

fn path_str(p: &Option<PathBuf>) -> String {
    p.as_ref()
        .map(|p| p.display().to_string())
        .unwrap_or_default()
}

/// Wire → [`ReplicationTarget`], refusing everything a standby cannot be.
///
/// Every refusal here is an `InvalidArgument` naming the field, because the
/// caller is an operator at a terminal and the whole value of the verb is that
/// it tells them what is wrong while they can still fix it. The alternative —
/// accepting a target that can never connect — is a standby that reports
/// `failed` forever into a log nobody reads, which is the failure this feature
/// was rebuilt out of.
pub fn target_from_proto(t: gunnar_proto::ReplicationTarget) -> ControlResult<ReplicationTarget> {
    let name = t.name.trim().to_owned();
    if name.is_empty() {
        return Err(ControlError::InvalidArgument(
            "name is required: every event, log line and status row identifies a standby by \
             its name rather than by its URL"
                .into(),
        ));
    }
    let remote = t.remote.trim().to_owned();
    if remote.is_empty() {
        return Err(ControlError::InvalidArgument("remote is required".into()));
    }
    // SSH ONLY, refused by name. The push half is `gunnar-client`'s send-pack
    // over its russh transport; it has no HTTP transport and the appliance has
    // no `http` feature to give it one. An `https://` here would be accepted,
    // stored, and then fail on every attempt forever.
    if !(remote.starts_with("ssh://") || remote.starts_with("git+ssh://")) {
        return Err(ControlError::InvalidArgument(format!(
            "remote {remote:?} is not an ssh url. Replication is a send-pack over SSH; \
             the server's client half has no HTTP transport to dial anything else with"
        )));
    }
    let scope = match t.scope.and_then(|s| s.of) {
        None | Some(gunnar_proto::replication_scope::Of::Server(_)) => ReplicationScope::Server,
        Some(gunnar_proto::replication_scope::Of::Namespace(ns)) => {
            let ns = ns.trim().to_owned();
            if ns.is_empty() {
                return Err(ControlError::InvalidArgument(
                    "a namespace-scoped target with an empty namespace would take nothing".into(),
                ));
            }
            ReplicationScope::Namespace { namespace: ns }
        }
        Some(gunnar_proto::replication_scope::Of::Repo(path)) => {
            path.validate()
                .map_err(|err| ControlError::InvalidArgument(err.to_string()))?;
            ReplicationScope::Repo {
                namespace: path.namespace,
                repo: path.repo,
            }
        }
    };
    // **PublicOnly needs a control address** (the browse replica): the Public → Private
    // delete has nowhere to go without one, and a replica that could not be
    // told to forget would keep private data — refused at add time, by name,
    // rather than discovered at the first flip.
    let control = Some(t.control.trim().to_owned()).filter(|c| !c.is_empty());
    let visibility = if t.public_only {
        if control.is_none() {
            return Err(ControlError::InvalidArgument(
                "a public-only target needs `control` (the replica's control-plane host:port): \
                 a repository that goes Private is deleted there over that address, and \
                 without it the replica would keep a copy it must not"
                    .into(),
            ));
        }
        // The identity requirement is NOT checked here — see
        // [`require_public_only_identity`], which runs after the mint.
        VisibilityScope::PublicOnly
    } else {
        VisibilityScope::All
    };
    Ok(ReplicationTarget {
        name,
        scope,
        remote,
        identity: opt_path(t.identity_path),
        known_hosts: opt_path(t.known_hosts_path),
        strict_host_key: t.strict_host_key,
        visibility,
        control,
    })
}

/// **A public-only target must end up with a deploy key — checked AFTER the
/// mint, never before it.**
///
/// The requirement is right: a repository that goes Private is DELETED on the
/// replica over `control`, and that call is signed with the same deploy key the
/// send-pack authenticates with. A browse replica that could not be told to
/// forget would keep private data.
///
/// It used to be enforced inside [`target_from_proto`], which is one statement
/// too early. MEASURED 2026-09-10 against a running appliance:
///
/// ```text
/// $ gunnar-cli control replication add public ssh://git@…:2222 \
///       --scope public-only --control …:50051
/// gunnar: Client specified an invalid argument: invalid argument: a public-only
/// target needs `identity_path`: the delete on a Private flip is signed with the
/// same deploy key the send-pack authenticates with
/// ```
///
/// — refused for the absence of a key the server MINTS on the very next line
/// (`control_plane::replication_set` calls [`ensure_identity`] the moment
/// `identity` is `None`, for the reason recorded on [`REPLICATION_KEY`]: an
/// appliance with no shell cannot be handed a file). So the one target shape
/// that most needs the minted key was the one shape the mint could not reach,
/// and the documented one-line recipe for a browse replica could not be run at
/// all on the box it was written for.
///
/// Moving it here keeps the guarantee — nothing is stored without a key — and
/// lets the mint satisfy it. What is left is a genuine failure: the box could
/// not produce a key at all.
pub fn require_public_only_identity(t: &ReplicationTarget) -> ControlResult<()> {
    if t.visibility == VisibilityScope::PublicOnly && t.identity.is_none() {
        return Err(ControlError::InvalidArgument(
            "a public-only target needs a deploy key and this box has none: the delete on a \
             Private flip is signed with the same key the send-pack authenticates with. The \
             server mints one itself unless `identity_path` names another, so this means the \
             mint failed"
                .into(),
        ));
    }
    Ok(())
}

/// [`ReplicationTarget`] → wire.
pub fn target_to_proto(t: &ReplicationTarget) -> gunnar_proto::ReplicationTarget {
    use gunnar_proto::replication_scope::Of;
    let of = match &t.scope {
        ReplicationScope::Server => Of::Server(true),
        ReplicationScope::Namespace { namespace } => Of::Namespace(namespace.clone()),
        ReplicationScope::Repo { namespace, repo } => Of::Repo(gunnar_proto::RepoPath {
            namespace: namespace.clone(),
            repo: repo.clone(),
        }),
    };
    gunnar_proto::ReplicationTarget {
        name: t.name.clone(),
        scope: Some(gunnar_proto::ReplicationScope { of: Some(of) }),
        remote: t.remote.clone(),
        identity_path: path_str(&t.identity),
        known_hosts_path: path_str(&t.known_hosts),
        strict_host_key: t.strict_host_key,
        public_only: t.visibility == VisibilityScope::PublicOnly,
        control: t.control.clone().unwrap_or_default(),
    }
}

/// The RUNNING state, straight off [`crate::replication_view::report`].
///
/// **Through the one reader and not around it.** That function is what the
/// console band on the box's own screen and `Admin.Stats` already draw from,
/// including the staleness verdict and the "never synced" zero. Building a
/// second view here out of `Replicator::status` would have been a twin, and two
/// readers of the same counters are two chances to disagree about whether a
/// standby has ever worked — which is the one question this whole surface
/// exists to answer.
pub fn running_to_proto(
    replicator: &Replicator,
    include_refs: bool,
    out: &mut gunnar_proto::ReplicationStatusResponse,
) {
    let report = crate::replication_view::report(replicator);
    out.configured = report.configured;
    out.observed_since_unix_ms = replicator.observed_since_unix_ms();
    // **The ref-coverage walk is opt-in, and it is taken ONCE here.**
    //
    // `include_refs` is false for a plain `replication list` and for every
    // status a mutating verb prints back: those want the verdict, the pending
    // clock and the per-repository copy state, none of which opens a store. It
    // is true only for `replication list --refs`, when the operator has asked
    // what a mirror does not carry — and then this is the single place the
    // reference databases are opened, one FRESH walk whose per-repository rows
    // and server-wide sum both come off it (one implementation), rather than the old code's
    // walk on every call that made a cold box unresponsive.
    let coverage =
        include_refs.then(|| replicator.ref_coverage_by_repo(std::time::Duration::ZERO));
    // **Per repository, and driven from the catalogue.** A failure to read the
    // catalogue leaves this empty rather than failing the whole status call:
    // the aggregates are still worth answering, and an empty list is
    // distinguishable from a list of zeroes.
    out.repos = match replicator.repo_copies(coverage.as_deref()) {
        Ok(copies) => copies
            .into_iter()
            .map(|c| gunnar_proto::RepoReplication {
                repo: Some(gunnar_proto::RepoPath::new(&c.namespace, &c.repo)),
                // `None` stays `None`: "we could not read this repository's
                // refs" and "none of them are unmirrored" are opposite answers.
                refs_total: c.refs.map(|r| r.total),
                refs_unmirrored: c.refs.map(|r| r.unmirrored),
                standby: c.standby,
                last_success_unix_ms: c.last_success_unix_ms,
                refs_at_last_success: c.refs_at_last_success,
                outstanding: c.outstanding,
                last_error: c.last_error,
                stopped: c.stopped,
                sha1: c.sha1,
            })
            .collect(),
        Err(err) => {
            tracing::warn!(
                target: crate::mirror::TARGET_REPLICATION,
                %err,
                "could not list repositories for the per-repository replication report; the \
                 aggregates below are still answered"
            );
            Vec::new()
        }
    };
    // The server-wide summary is the sum of the same walk the rows came off
    // (one implementation), or left at zero when refs were not asked for — which every
    // printer already renders as "no ref figure", not as "nothing unmirrored".
    if let Some(coverage) = coverage.as_deref() {
        let mut total = crate::replication_view::RefCoverage::default();
        for c in coverage.values() {
            total.add(*c);
        }
        out.refs_total = total.total;
        out.refs_unmirrored = total.unmirrored;
    }
    out.enqueued = report.enqueued;
    out.synced = report.synced;
    out.failed = report.failed;
    out.bus_gaps = report.bus_gaps;
    out.standbys = report
        .standbys
        .into_iter()
        .map(|s| gunnar_proto::ReplicationStandby {
            verdict: s.verdict().word().to_owned(),
            name: s.name,
            remote: s.remote,
            pending: s.pending,
            consecutive_failures: s.consecutive_failures,
            last_error: s.last_error,
            last_success_unix_ms: s.last_success_unix_ms,
            needs_full_resync: s.needs_full_resync,
            stale: s.stale,
            in_flight: s.in_flight,
            outstanding_for_ms: s.outstanding_for_ms,
            behind: s.behind,
        })
        .collect();
}

#[cfg(test)]
mod tests {
    use super::*;

    fn public_only_wire(identity: &str) -> gunnar_proto::ReplicationTarget {
        gunnar_proto::ReplicationTarget {
            name: "public".into(),
            scope: Some(gunnar_proto::ReplicationScope {
                of: Some(gunnar_proto::replication_scope::Of::Server(true)),
            }),
            remote: "ssh://git@replica:2222".into(),
            identity_path: identity.into(),
            known_hosts_path: String::new(),
            strict_host_key: false,
            public_only: true,
            control: "replica:50051".into(),
        }
    }

    /// **A public-only target with no `identity_path` must PARSE**, because the
    /// server mints the key immediately afterwards.
    ///
    /// Red-first, and the red is what shipped. MEASURED 2026-09-10 against a
    /// running server, typing the one-line recipe the `replication browse`
    /// help text itself prints:
    ///
    /// ```text
    /// invalid argument: a public-only target needs `identity_path`: the delete
    /// on a Private flip is signed with the same deploy key the send-pack
    /// authenticates with
    /// ```
    ///
    /// `control_plane::replication_set` calls [`ensure_identity`] the moment
    /// `identity` is `None` — one statement after this parse — precisely
    /// because an appliance has no shell and cannot be handed a key file. So
    /// the target shape that needs the minted key most was the only shape the
    /// mint could not reach, and a browse replica could not be configured on
    /// the box this control plane exists for.
    #[test]
    fn a_public_only_target_parses_without_an_identity_because_the_mint_follows() {
        let parsed = target_from_proto(public_only_wire("")).expect(
            "a public-only target with no identity must parse: the server mints one on the \
             next line, and refusing here makes that mint unreachable",
        );
        assert_eq!(parsed.visibility, VisibilityScope::PublicOnly);
        assert!(parsed.identity.is_none(), "nothing invented an identity");
    }

    /// And the guarantee still holds: nothing is stored public-only without a
    /// key. The check simply moved to where a key exists.
    #[test]
    fn the_identity_requirement_still_bites_after_the_mint() {
        let mut minted = target_from_proto(public_only_wire("")).expect("parses");
        let err = require_public_only_identity(&minted)
            .expect_err("a public-only target with no key must be refused somewhere");
        let text = format!("{err:?}");
        assert!(
            text.contains("mint failed"),
            "the refusal has to say what actually went wrong now — the operator did not \
             forget a flag, the box could not make a key:\n{text}"
        );

        minted.identity = Some(std::path::PathBuf::from("/data/replication_key"));
        require_public_only_identity(&minted).expect("a minted key satisfies it");
    }

    /// The OTHER public-only requirement is unmoved: `control` is a thing the
    /// operator types and nothing can mint, so it is still refused at parse.
    #[test]
    fn a_public_only_target_without_a_control_address_is_still_refused_at_parse() {
        let mut wire = public_only_wire("/data/replication_key");
        wire.control = String::new();
        let err = target_from_proto(wire).expect_err("no control address");
        assert!(
            format!("{err:?}").contains("needs `control`"),
            "a replica that cannot be told to forget a repository that went Private would \
             keep private data"
        );
    }
}