//! Converging this hive onto the agent set the swarm controller declares.
//!
//! The controller gives each hive its own `hive-wanted-<hive>` bucket (see
//! [`swarm_queue_client::wanted`]); this reads its own and acts on it. It is
//! the **repair** path, not the fast one: a deploy event
//! ([`crate::swarm_status`]) is core NATS, so a hive that was down never hears
//! it, and until this existed that left the hive on its old shape with nothing
//! looking wrong.
//!
//! # Two things a declaration does not mean
//!
//! **Absence is not a deletion order.** An agent missing from the value — or
//! no value, or no bucket — is one the controller has said nothing about.
//! Swarm-side lifecycle does not yet cover agents that predate it, so
//! converging "to exactly this set" would tear down every agent the swarm has
//! not adopted.
//!
//! **An unknown state is not a partial instruction.** [`AgentState`] is closed,
//! so a value this build cannot read fails the whole decode: the hive converges
//! nothing rather than obeying the agents it happened to understand.
//!
//! # What "diverged" is measured against
//!
//! The hive's **stored power intent**, not the container's observed running
//! state. An agent down while its intent says `Up` is already the boot
//! reconcile's work; a loop reading `is_running` would insert a start DAG
//! behind that reconcile's back on every boot.
//!
//! One consequence, deliberate: a declaration outranks a local stop — an agent
//! declared `Up` returns at the next boot, because the controller owns that.

use std::collections::{BTreeMap, BTreeSet};
use std::sync::Arc;

use anyhow::{Context, Result};
use swarm_queue_client::wanted::{AgentState, HiveWanted};

use crate::coordinator::Coordinator;
use crate::power::Wanted;

/// Read this hive's declaration and converge the agents it names.
///
/// `Ok(())` covers three quiet cases that are **not** an empty declaration: no
/// hive name, no bucket, no key. Each logs its own line, because a loop that
/// silently does nothing and one that is working look identical otherwise.
/// Anything else — an unreachable queue, an undecodable value, an agent list
/// this hive cannot enumerate — is an error, so the node fails visibly on the
/// dashboard instead of reading as "nothing was declared".
///
/// # Errors
/// Propagates a failed queue connect, KV read, decode, or agent enumeration.
pub async fn pull(coord: &Arc<Coordinator>) -> Result<()> {
    let Some(hive) = crate::container_view::hive_swarm_names().0 else {
        tracing::debug!("wanted state: HYPERHIVE_HIVE_NAME unset; no key to read");
        return Ok(());
    };
    let Some(client) = crate::swarm_queue::client().await else {
        // Absent or failed — either way already reported by the shared
        // connector (an `info` log or a `swarm_queue_config` banner).
        return Ok(());
    };
    // An unconnected client does not fail a JetStream request, it hangs on it.
    // A hung node is worse than a failed one: the dashboard shows it running.
    swarm_queue_client::ensure_connected(&client)?;

    let Some(store) = swarm_queue_client::wanted::open_read_only(&client, &hive).await else {
        tracing::info!(%hive, "wanted state: no bucket yet; nothing has been declared");
        return Ok(());
    };
    let Some(raw) = store
        .get(&hive)
        .await
        .context("reading this hive's wanted state")?
    else {
        tracing::info!(%hive, "wanted state: no key for this hive; nothing has been declared");
        return Ok(());
    };
    let declared: HiveWanted =
        serde_json::from_slice(&raw).context("decoding this hive's wanted state")?;

    converge(coord, &declared).await
}

/// Converge again each time the controller republishes this hive's declaration.
///
/// The **fast** path, and it does not replace [`pull`]: this hears only what is
/// published while it is listening, so a hive that was down still learns the
/// current declaration from the boot read. Both, not either.
///
/// ⚠️ A refused watch and a quiet one are told apart here, unlike the core-NATS
/// subscriptions in [`crate::swarm_status`]: a watch is a `JetStream` consumer,
/// so the request is answered, and a hive lacking the grant gets `None` rather
/// than silence. That is why this warns instead of returning quietly.
pub async fn watch_declarations(
    client: async_nats::Client,
    coord: Arc<Coordinator>,
    hive: String,
    mut shutdown: tokio::sync::watch::Receiver<bool>,
) {
    use futures_util::StreamExt as _;

    let Some(mut updates) = swarm_queue_client::wanted::watch(&client, &hive).await else {
        tracing::warn!(
            %hive,
            "wanted state: cannot watch this hive's bucket; changes will be \
             picked up at the next boot instead"
        );
        return;
    };
    tracing::info!(%hive, "wanted state: watching for declarations");

    loop {
        tokio::select! {
            entry = updates.next() => {
                let Some(entry) = entry else {
                    // Same reading as the sibling subscriptions: `async-nats`
                    // reconnects underneath a live watch, so an ended stream is
                    // the connection going away for good rather than a blip to
                    // spin on.
                    tracing::warn!(%hive, "wanted state: watch closed");
                    return;
                };
                match entry {
                    Ok(entry) => apply_entry(&coord, &hive, &entry).await,
                    // The watch survives one bad entry; the stream ending is
                    // the case above.
                    Err(e) => tracing::warn!(%hive, error = %e, "wanted state: watch error"),
                }
            }
            _ = shutdown.changed() => {
                tracing::info!("wanted state: shutdown signal received");
                return;
            }
        }
    }
}

/// Converge one watched update, or decline to.
///
/// A delete is **not** a deletion order — the module docs' rule, and the reason
/// this is not simply "decode and converge": the controller removing the key
/// says nothing about the agents this hive runs, so acting on it would tear
/// down the very set that absence is defined not to touch.
async fn apply_entry(
    coord: &Arc<Coordinator>,
    hive: &str,
    entry: &async_nats::jetstream::kv::Entry,
) {
    if !carries_a_declaration(entry.operation) {
        tracing::info!(%hive, "wanted state: declaration withdrawn; nothing to converge");
        return;
    }
    let declared: HiveWanted = match serde_json::from_slice(&entry.value) {
        Ok(declared) => declared,
        // Warned rather than propagated: this end and the controller share one
        // type, so a decode failure means they disagree about it — and the
        // watch must keep running to pick up the next, possibly good, value.
        Err(e) => {
            tracing::warn!(%hive, error = %e, "wanted state: undecodable declaration");
            return;
        }
    };
    if let Err(e) = converge(coord, &declared).await {
        tracing::warn!(%hive, error = ?e, "wanted state: converging a watched update failed");
    }
}

/// Whether a watched operation carries a declaration to converge to.
///
/// Pure, and separate from [`apply_entry`], so the module's "absence is not a
/// deletion order" rule is enforced by a test rather than only asserted in
/// prose — converging on a removed key is the one mistake here that would tear
/// down agents nobody asked to stop.
fn carries_a_declaration(operation: async_nats::jetstream::kv::Operation) -> bool {
    use async_nats::jetstream::kv::Operation;

    match operation {
        Operation::Put => true,
        Operation::Delete | Operation::Purge => false,
    }
}

/// Queue whatever the declaration asks for and this hive is not already doing.
async fn converge(coord: &Arc<Coordinator>, declared: &HiveWanted) -> Result<()> {
    // Fail closed, for the reason the deploy event's own arm gives: without
    // the list, "this hive does not have that agent" cannot be told from
    // "this hive cannot see its agents", and acting on the guess creates a
    // container over one that already exists.
    let present: BTreeSet<String> = crate::lifecycle::agents_for_meta_listing()
        .await
        .context("enumerating this hive's agents")?
        .into_iter()
        .map(|spec| spec.name)
        .collect();

    let mut intents = BTreeMap::new();
    for agent in declared.agents.keys() {
        match coord.power.get(agent) {
            Ok(intent) => {
                intents.insert(agent.clone(), intent);
            }
            // Left out of the map, which is what makes `plan` skip it.
            Err(e) => {
                tracing::warn!(%agent, error = ?e, "wanted state: cannot read power intent; skipping");
            }
        }
    }

    let plan = plan(declared, &present, &intents);
    tracing::info!(
        declared = declared.agents.len(),
        deploy = plan.deploy.len(),
        start = plan.start.len(),
        stop = plan.stop.len(),
        "wanted state: converging"
    );

    let mut queued = false;
    for agent in &plan.deploy {
        match crate::swarm_status::queue_first_deploy(coord, agent) {
            Ok(_) => queued = true,
            // Per agent rather than the whole sweep: one agent that cannot be
            // queued is not a reason to leave the others undeclared.
            Err(e) => tracing::warn!(%agent, error = %e, "wanted state: queueing a deploy failed"),
        }
    }
    if !plan.start.is_empty() {
        crate::job_queue::power::start_many(coord, &plan.start)
            .await
            .context("queueing the declared starts")?;
    }
    if !plan.stop.is_empty() {
        crate::job_queue::power::stop_many(coord, &plan.stop, true)
            .await
            .context("queueing the declared stops")?;
    }
    if queued {
        // The power ops emit their own; the first deploys do not.
        coord.emit_rebuild_queue_snapshot();
    }
    Ok(())
}

/// Everything one declaration asks this hive to queue.
#[derive(Debug, Default, PartialEq, Eq)]
struct Plan {
    deploy: Vec<String>,
    start: Vec<String>,
    stop: Vec<String>,
}

/// Turn one declaration into that plan — pure, so what the loop does with a
/// given declaration is testable without a queue, a container or a store.
///
/// `intents` is the hive's stored power intent per agent, and an agent
/// **missing from the map** is one whose intent could not be read: it is
/// skipped. That is not `Some(None)` — "no row yet" is a reading and gets
/// converged, "the read failed" is not and would be a guess.
///
/// An agent the declaration does not name reaches none of the three lists.
/// That is the whole of "absence is not a deletion order": the agents this
/// hive runs that the swarm has not adopted are never even inspected.
fn plan(
    declared: &HiveWanted,
    present: &BTreeSet<String>,
    intents: &BTreeMap<String, Option<Wanted>>,
) -> Plan {
    let mut plan = Plan::default();
    for (agent, decl) in &declared.agents {
        let Some(intent) = intents.get(agent) else {
            continue;
        };
        match decide(decl.state, present.contains(agent), *intent) {
            Converge::Deploy => plan.deploy.push(agent.clone()),
            Converge::Start => plan.start.push(agent.clone()),
            Converge::Stop => plan.stop.push(agent.clone()),
            Converge::Nothing => {}
        }
    }
    plan
}

/// What the loop will do about one declared agent.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Converge {
    /// Declared, and this hive has no container for it — the missed-deploy
    /// repair.
    Deploy,
    Start,
    Stop,
    /// The hive already agrees with the declaration.
    Nothing,
}

/// The whole decision, pure: no queue, no store, no container. The three
/// inputs are exactly what [`converge`] reads per agent, so the table below is
/// the behaviour rather than a model of it.
fn decide(state: AgentState, present: bool, intent: Option<Wanted>) -> Converge {
    match state {
        AgentState::Up if !present => Converge::Deploy,
        AgentState::Up => {
            if intent == Some(Wanted::Up) {
                Converge::Nothing
            } else {
                Converge::Start
            }
        }
        // Declared offline and not here: creating a container in order to
        // leave it stopped is not what the declaration asks for.
        AgentState::Offline if !present => Converge::Nothing,
        AgentState::Offline => {
            if intent == Some(Wanted::Offline) {
                Converge::Nothing
            } else {
                Converge::Stop
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use std::collections::{BTreeMap, BTreeSet};

    use super::{Converge, Plan, carries_a_declaration, decide, plan};
    use crate::power::Wanted;
    use swarm_queue_client::wanted::{AgentState, AgentWanted, HiveWanted};

    fn declaration(agents: &[(&str, AgentState)]) -> HiveWanted {
        HiveWanted {
            agents: agents
                .iter()
                .map(|(name, state)| ((*name).to_owned(), AgentWanted { state: *state }))
                .collect(),
        }
    }

    fn set(names: &[&str]) -> BTreeSet<String> {
        names.iter().map(|n| (*n).to_owned()).collect()
    }

    fn intents(entries: &[(&str, Option<Wanted>)]) -> BTreeMap<String, Option<Wanted>> {
        entries
            .iter()
            .map(|(name, intent)| ((*name).to_owned(), *intent))
            .collect()
    }

    /// The property the whole migration rides on: this hive's own agents are
    /// not converged, they are not looked at. `adopted` is the control — the
    /// same call must still act on what the declaration *does* name, or this
    /// passes on a loop that plans nothing at all.
    #[test]
    fn an_agent_the_declaration_does_not_name_is_never_touched() {
        let declared = declaration(&[("adopted", AgentState::Up)]);
        let planned = plan(
            &declared,
            &set(&["adopted", "legacy"]),
            &intents(&[("adopted", Some(Wanted::Offline)), ("legacy", None)]),
        );
        assert_eq!(
            planned,
            Plan {
                deploy: vec![],
                start: vec!["adopted".to_owned()],
                stop: vec![],
            }
        );
    }

    /// An intent that could not be read is skipped rather than treated as "no
    /// row". `readable` is the control for the same call.
    #[test]
    fn an_unreadable_intent_skips_only_that_agent() {
        let declared = declaration(&[("unreadable", AgentState::Up), ("readable", AgentState::Up)]);
        let planned = plan(
            &declared,
            &set(&["unreadable", "readable"]),
            &intents(&[("readable", None)]),
        );
        assert_eq!(planned.start, vec!["readable".to_owned()]);
        assert!(planned.deploy.is_empty() && planned.stop.is_empty());
    }

    #[test]
    fn a_declared_agent_this_hive_does_not_have_is_deployed() {
        assert_eq!(decide(AgentState::Up, false, None), Converge::Deploy);
        // Even holding an intent: a row can outlive its container.
        assert_eq!(
            decide(AgentState::Up, false, Some(Wanted::Up)),
            Converge::Deploy
        );
    }

    #[test]
    fn a_hive_that_already_agrees_queues_nothing() {
        assert_eq!(
            decide(AgentState::Up, true, Some(Wanted::Up)),
            Converge::Nothing
        );
        assert_eq!(
            decide(AgentState::Offline, true, Some(Wanted::Offline)),
            Converge::Nothing
        );
    }

    #[test]
    fn a_disagreeing_intent_is_converged_both_ways() {
        assert_eq!(
            decide(AgentState::Up, true, Some(Wanted::Offline)),
            Converge::Start
        );
        assert_eq!(
            decide(AgentState::Offline, true, Some(Wanted::Up)),
            Converge::Stop
        );
    }

    /// No row is not agreement. An agent that predates the power store is
    /// converged to what it was declared as, not left on whatever it happens
    /// to be doing.
    #[test]
    fn an_agent_with_no_intent_row_is_converged() {
        assert_eq!(decide(AgentState::Up, true, None), Converge::Start);
        assert_eq!(decide(AgentState::Offline, true, None), Converge::Stop);
    }

    /// Declared offline, no container: the one absent case that must not
    /// deploy. Its control is the `Up` case above, which must.
    #[test]
    fn an_absent_agent_declared_offline_is_left_absent() {
        assert_eq!(decide(AgentState::Offline, false, None), Converge::Nothing);
    }

    /// The watch's half of "absence is not a deletion order". `Put` is the
    /// control: without it this would pass on a function that refused
    /// everything, which would silently stop the fast path converging at all.
    #[test]
    fn only_a_put_carries_a_declaration_to_converge_to() {
        use async_nats::jetstream::kv::Operation;

        assert!(carries_a_declaration(Operation::Put));
        for withdrawn in [Operation::Delete, Operation::Purge] {
            assert!(
                !carries_a_declaration(withdrawn),
                "{withdrawn:?} must not converge"
            );
        }
    }

    /// Version skew is handled one layer up, at the decode: `AgentState` is
    /// closed, so an unknown value never reaches [`decide`] — it fails the
    /// whole declaration in `swarm-queue-client`, which owns that test
    /// (`an_unknown_state_fails_the_whole_declaration`). There is deliberately
    /// no case for it here: a test asserting something unrepresentable would
    /// pass forever without measuring anything.
    #[test]
    fn every_state_this_build_knows_is_covered_above() {
        for state in [AgentState::Up, AgentState::Offline] {
            let seen = [true, false].iter().any(|present| {
                decide(state, *present, None) != Converge::Nothing
                    || decide(state, *present, Some(Wanted::Up)) != Converge::Nothing
            });
            assert!(seen, "{state:?} produces no action in any combination");
        }
    }
}