hyperhive/swarm-queue-client/src/wanted.rs

//! The hive-wanted KV buckets: the agent set the controller declares for each
//! hive, **one bucket per hive** (see [`crate::wanted::bucket`] for why the
//! split is a grant boundary rather than a data-modelling choice), keyed inside
//! it by `hiveName`.
//!
//! Sibling of [`crate::status`] and deliberately **not** a mirror of it: that
//! one is **observed** — each hive republishes what it is, so its store losing
//! everything "degrades to honesty" (`swarm-nats.nix`). This one is
//! **declared**, and nothing regenerates it; lose it and a reconcile loop has
//! nothing to converge to, which is the failure such a loop exists to remove.
//!
//! # Absence is not a deletion order
//!
//! A hive that finds no key for itself — or no bucket at all — has learned
//! nothing about the agents it is running, not that it should have none.
//! Swarm-side lifecycle does not yet cover agents that predate it, so callers
//! converge the agents a value NAMES and leave the rest alone.
//!
//! # Only the controller creates it
//!
//! Unlike [`crate::status::open_or_create`], where either end may legitimately
//! arrive first, the writer here is single and known. A hive opens read-only
//! and treats absence as the case above, so "not published yet" stays quiet
//! rather than looking like an error a hive could fix.

#[cfg(feature = "kv")]
use crate::Error;

/// The prefix every hive's wanted-state bucket name starts with.
///
/// Derived from, and not configurable, for the reason [`crate::status::BUCKET`]
/// gives: writer and reader must name the same bucket, and an option is a way
/// for two deployments to disagree about which one that is.
pub const BUCKET_PREFIX: &str = "hive-wanted-";

/// The KV bucket one hive's declaration lives in — **one bucket per hive**, not
/// one bucket keyed by hive.
///
/// The split is a grant-scoping decision, not a data-modelling one. A KV read
/// scopes per key, because `DIRECT.GET` carries the key in the subject; a
/// *watch* does not, because a consumer's filter travels in the request payload
/// and `$JS.API.CONSUMER.CREATE.<stream>` therefore grants the whole stream. So
/// with every hive in one bucket, letting a hive watch its own declaration
/// means letting it read every other hive's. One stream per hive makes the
/// grant a hive can hold exactly as wide as what it is allowed to see.
///
/// A bucket name may contain `[a-zA-Z0-9_-]`, and a hive name is an `Ident`
/// (`[a-z0-9-]`), so this composition is always a legal bucket name — plain
/// code span rather than an intra-doc link because this crate does not depend
/// on `hive_types`.
#[must_use]
pub fn bucket(hive: &str) -> String {
    format!("{BUCKET_PREFIX}{hive}")
}

/// One hive's whole declaration: the agents the controller names, and what it
/// wants of each. The value under `BUCKET`/`<hiveName>`.
///
/// The agents it does **not** name are not part of the declaration at all —
/// see the module docs. A hive converges the ones listed here and leaves the
/// rest alone.
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct HiveWanted {
    /// Agent name → what the controller wants of it.
    #[serde(default)]
    pub agents: std::collections::BTreeMap<String, AgentWanted>,
}

/// One agent's entry in a [`HiveWanted`].
///
/// A struct around one field rather than the bare state, because this is where
/// a later declaration grows — a config revision, a schedule — and a hive
/// built before that field existed keeps deserialising the whole document.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AgentWanted {
    /// The state to converge this agent to.
    pub state: AgentState,
}

/// What the controller wants an agent to be.
///
/// **Closed on purpose.** A value this build does not know fails the whole
/// document's decode, so a hive running older code converges **nothing** rather
/// than part of a declaration it only half understands. Adding a state means
/// adding a variant here and shipping it to both ends — which is the intended
/// workflow, not an obstacle to route around with a catch-all variant.
#[derive(
    Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize, strum::IntoStaticStr,
)]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
pub enum AgentState {
    /// Exists on the hive and is running.
    Up,
    /// Exists on the hive and is not running.
    Offline,
    /// Exists on the hive, is running, and its turn loop is parked — the
    /// swarm-declared counterpart of the hive-local `hivectl agent pause`
    /// marker. Orthogonal to the power axis `Up`/`Offline` express: a
    /// paused agent still has a running container (the web UI and MCP
    /// daemons stay reachable), it just drives no turns. See
    /// `hive-c0re::workers::wanted::decide_pause` for the convergence side
    /// of that split.
    Paused,
    /// Torn down entirely — the hive runs the destroy template once, then
    /// treats absence as agreement rather than re-running it. The key
    /// stays in the declared set with this state forever, on purpose: it's
    /// what lets a redeclare of an already-destroyed agent be rejected as
    /// a terminal-state transition instead of silently reviving it.
    /// Irreversible: there is no state that brings a destroyed agent back
    /// short of a fresh deploy.
    Destroyed,
}

/// Open the wanted-state bucket for writing, creating it if nothing has yet.
///
/// **Controller-side only.** `history: 1` because a hive converges to the
/// current declaration and never asks what the previous one was. The
/// controller keeps no second copy either: the value published here is the
/// record, and it reads it back from this bucket when it needs it.
#[cfg(feature = "kv")]
pub async fn open_or_create(
    client: &async_nats::Client,
    hive: &str,
) -> Result<async_nats::jetstream::kv::Store, Error> {
    let js = async_nats::jetstream::new(client.clone());
    let bucket = bucket(hive);
    let description = format!("Agent set the swarm controller declares for hive {hive}");
    crate::kv::open_or_create(&js, &bucket, "wanted-state", description).await
}

/// Open the wanted-state bucket for reading, or `None` when it does not exist.
///
/// **Hive-side.** `None` is the ordinary pre-publication state, not a failure:
/// a hive holds no grant to create this bucket and must not treat its absence
/// as a reason to converge to an empty agent set — see the module docs.
///
/// # Errors
///
/// [`Error::OpenBucket`] for every failure other than the bucket's stream not
/// existing: a refused grant, a timeout, `JetStream` unavailable.
#[cfg(feature = "kv")]
pub async fn open_read_only(
    client: &async_nats::Client,
    hive: &str,
) -> Result<Option<async_nats::jetstream::kv::Store>, Error> {
    let js = async_nats::jetstream::new(client.clone());
    let bucket = bucket(hive);
    match js.get_key_value(&bucket).await {
        Ok(store) => Ok(Some(store)),
        Err(e) if is_bucket_not_found(&e) => Ok(None),
        Err(source) => Err(Error::OpenBucket { bucket, source }),
    }
}

/// `get_key_value` reports a missing bucket as `GetBucket` wrapping the
/// stream lookup's `STREAM_NOT_FOUND`; every other kind is a real failure.
#[cfg(feature = "kv")]
fn is_bucket_not_found(e: &async_nats::jetstream::context::KeyValueError) -> bool {
    use async_nats::jetstream::context::{GetStreamError, GetStreamErrorKind, KeyValueErrorKind};

    e.kind() == KeyValueErrorKind::GetBucket
        && std::error::Error::source(e)
            .and_then(|s| s.downcast_ref::<GetStreamError>())
            .is_some_and(|s| {
                matches!(
                    s.kind(),
                    GetStreamErrorKind::JetStream(js)
                        if js.error_code() == async_nats::jetstream::ErrorCode::STREAM_NOT_FOUND
                )
            })
}

/// Watch this hive's own declaration for changes.
///
/// **Hive-side.** `None` on the same terms as [`open_read_only`]: no bucket
/// yet. A watch is a `JetStream` *consumer*, so it needs a grant a plain `get`
/// does not; a hive missing `CONSUMER.CREATE` on its own stream gets
/// [`Error::WatchKey`] here while `get` keeps working.
///
/// Updates only, deliberately: the boot-time read already has the current
/// value, and a watch that replayed history would re-converge the whole
/// declaration on every reconnect for nothing.
///
/// # Errors
///
/// [`Error::OpenBucket`] as for [`open_read_only`], and [`Error::WatchKey`]
/// when the watch itself is refused or fails.
#[cfg(feature = "kv")]
pub async fn watch(
    client: &async_nats::Client,
    hive: &str,
) -> Result<Option<async_nats::jetstream::kv::Watch>, Error> {
    let Some(store) = open_read_only(client, hive).await? else {
        return Ok(None);
    };
    // The bucket holds exactly one key, named for the hive — same key
    // `open_read_only`'s caller reads, so both paths address one declaration.
    store
        .watch(hive)
        .await
        .map(Some)
        .map_err(|source| Error::WatchKey {
            key: hive.to_owned(),
            source,
        })
}

#[cfg(test)]
mod tests {
    use super::{AgentState, BUCKET_PREFIX, HiveWanted, bucket};

    #[test]
    fn each_hive_gets_its_own_bucket_name() {
        assert_eq!(bucket("alpha"), "hive-wanted-alpha");
        assert_ne!(bucket("alpha"), bucket("beta"));
    }

    #[test]
    fn a_bucket_name_is_legal_for_every_legal_hive_name() {
        // The client rejects a bucket name outside `[a-zA-Z0-9_-]`, and a hive
        // name is an `Ident` — lowercase, digits, hyphen. Pinned here because
        // the grant scoping in `swarm-nats-auth` names this string, so a bucket
        // the client refuses to open would surface as a permissions problem
        // rather than as the naming problem it is.
        let legal = |s: &str| {
            s.chars()
                .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-')
        };
        assert!(legal(&bucket("alpha")));
        assert!(legal(&bucket("a-hive-with-hyphens")));
        assert!(legal(&bucket("h9")));
        // Control: the predicate can fail, so the assertions above are not
        // vacuously true of any string.
        assert!(!legal("hive.wanted"));
    }

    #[test]
    fn the_prefix_is_what_every_bucket_starts_with() {
        assert!(bucket("alpha").starts_with(BUCKET_PREFIX));
    }

    #[test]
    fn the_known_states_decode_and_round_trip() {
        let doc = r#"{"agents":{"a":{"state":"up"},"b":{"state":"offline"},"c":{"state":"destroyed"},"d":{"state":"paused"}}}"#;
        let decoded: HiveWanted = serde_json::from_str(doc).expect("decodes");
        assert_eq!(decoded.agents["a"].state, AgentState::Up);
        assert_eq!(decoded.agents["b"].state, AgentState::Offline);
        assert_eq!(decoded.agents["c"].state, AgentState::Destroyed);
        assert_eq!(decoded.agents["d"].state, AgentState::Paused);
        assert_eq!(serde_json::to_string(&decoded).expect("serialises"), doc);
    }

    /// The enum is closed, so a state this build does not know takes the
    /// **whole document** down rather than one agent. That is the point: a hive
    /// converges nothing rather than part of a declaration it half understands.
    /// The `up` beside it is the control — a valid entry in the same document
    /// does not rescue it, and this test would pass on a broken decoder without
    /// the round-trip test above proving the happy path still works.
    #[test]
    fn an_unknown_state_fails_the_whole_declaration() {
        // `"paused"` used to be this test's unknown example; it is a real
        // variant now, so `"sleeping"` takes its place as one this build
        // still does not know.
        let doc = r#"{"agents":{"a":{"state":"up"},"c":{"state":"sleeping"}}}"#;
        assert!(serde_json::from_str::<HiveWanted>(doc).is_err());
    }

    /// Closed *values*, tolerant *fields*: a field a newer controller adds must
    /// not cost an older hive the document, since adding one is not a semantic
    /// a hive has to understand to obey the rest.
    #[test]
    fn unknown_fields_are_ignored() {
        let doc = r#"{"agents":{"a":{"state":"up","config_rev":"deadbeef"}},"epoch":3}"#;
        let decoded: HiveWanted = serde_json::from_str(doc).expect("decodes");
        assert_eq!(decoded.agents["a"].state, AgentState::Up);
    }

    /// A declaration missing the one required field is an error, so a malformed
    /// document cannot read as "no agents".
    #[test]
    fn a_missing_state_is_an_error() {
        assert!(serde_json::from_str::<HiveWanted>(r#"{"agents":{"a":{}}}"#).is_err());
    }

    /// `as_str` and the `rename_all` are two spellings of one fact, and a
    /// renderer that disagrees with the wire is worse than one that does not
    /// exist. The `match` is what makes this exhaustive: a new variant fails
    /// to compile here rather than quietly going untested.
    #[test]
    fn as_str_matches_the_serde_spelling() {
        let every = [
            AgentState::Up,
            AgentState::Offline,
            AgentState::Paused,
            AgentState::Destroyed,
        ];
        for state in every {
            match state {
                AgentState::Up
                | AgentState::Offline
                | AgentState::Paused
                | AgentState::Destroyed => {}
            }
            assert_eq!(
                serde_json::to_string(&state).expect("serialises"),
                format!("\"{}\"", <&str>::from(state))
            );
        }
    }
}