//! Build the head of an agent's config repo that moved without a build.
//!
//! A merged config PR is built and deployed from its webhook delivery
//! (`crate::webhook`). A delivery that never arrives leaves `main` ahead of
//! what the agent runs, and nothing else asks for a build. Every
//! `config_pr::POLL_INTERVAL` the config scan calls [`run`], which compares each
//! placed agent's `agent-configs/<agent>` `main` with the config commit its
//! latest green build used, and queues the build and deploy the webhook would
//! have queued ([`crate::jobq::insert_merged_deploy`]) for each that differs.
//!
//! The config commit a build used is the pin of `agent-<agent>` in the
//! `flake.nix` of the `internal/meta` commit its [`AgentClosure`] record names.
//! The pin in `internal/meta`'s head is not the record: a later meta update for
//! another agent moves every pin, built or not.
//!
//! Left alone:
//! - An agent no hive places, or more than one does.
//! - An agent with no record, or whose recorded flake pins no config commit for
//!   it: nothing was built from its config repo to differ from.
//! - A head whose deploy the graph holds in any state, or an agent with a deploy
//!   in flight. A head that was tried and failed is not queued again here: the
//!   meta timer builds it again every round (`jobq::queue_meta_update`). The
//!   graph is in memory, so a restart tries each such head once more.

use std::collections::{BTreeSet, HashMap};
use std::future::Future;
use std::sync::{Arc, Mutex};

use anyhow::{Context as _, Result};
use async_nats::jetstream::kv::Store;
use swarm_wire::agent_closures::AgentClosure;

use crate::forge::{self, objects::SHARED_ORG};
use crate::jobq::{SwarmNodeKind, SwarmResourceKind};
use crate::meta_flake::{CONFIG_BRANCH, META_REPO};
use crate::wanted::WantedWriter;

type Jobq = Mutex<hive_jobq::scheduler::Scheduler<SwarmNodeKind, SwarmResourceKind>>;

/// What a pass reads and does outside the process, so a test can stand in for it.
trait World {
    /// Every `(hive, agent)` a hive places.
    fn placements(&self) -> impl Future<Output = Result<BTreeSet<(String, String)>>> + Send;
    /// Every config repo's `(agent, head commit)`.
    fn heads(&self) -> impl Future<Output = Result<Vec<(String, String)>>> + Send;
    /// The config commit `agent`'s latest green build used; `None` when there is
    /// no such build or it pinned none.
    fn built(&self, agent: &str) -> impl Future<Output = Result<Option<String>>> + Send;
    /// Whether the graph holds a deploy of `agent` at `head`, or one still
    /// running.
    fn attempted(&self, agent: &str, head: &str) -> bool;
    /// Queue the build and deploy of `agent` at `head` on `hive`. Whether it was
    /// queued.
    fn queue(&self, hive: &str, agent: &str, head: &str) -> bool;
}

/// Whether `graph` holds a `TriggerDeploy` of `agent` that is not finished, or
/// one at `head` in any state.
fn attempted(
    graph: &hive_jobq::Graph<SwarmNodeKind, SwarmResourceKind>,
    agent: &str,
    head: &str,
) -> bool {
    graph.nodes().any(|n| match &n.payload {
        SwarmNodeKind::TriggerDeploy { agent: a, rev, .. } if a == agent => {
            !n.state.is_terminal() || rev.as_deref() == Some(head)
        }
        _ => false,
    })
}

/// One pass: each head that differs from what was built is queued once. The
/// `(agent, head)` it queued, in order.
///
/// # Errors
/// A read that failed before any agent was looked at (the placements, the
/// heads). A failure on one agent is logged and the pass goes on.
async fn pass(world: &impl World) -> Result<Vec<(String, String)>> {
    let placed = world
        .placements()
        .await
        .context("reading where the hives place agents")?;
    let heads = world.heads().await.context("reading the config heads")?;
    let mut queued = Vec::new();
    for (agent, head) in heads {
        let mut hives = placed.iter().filter(|(_, a)| *a == agent);
        let (Some((hive, _)), None) = (hives.next(), hives.next()) else {
            continue;
        };
        if world.attempted(&agent, &head) {
            continue;
        }
        let built = match world.built(&agent).await {
            Ok(Some(built)) => built,
            Ok(None) => continue,
            Err(e) => {
                tracing::warn!(%agent, error = %format!("{e:#}"), "config drift: reading the built config commit failed");
                continue;
            }
        };
        if built == head {
            continue;
        }
        tracing::info!(
            %agent, %head, %built, %hive,
            "config drift: agent-configs main moved to a head no build used, and no webhook build was seen; queueing it"
        );
        if world.queue(hive, &agent, &head) {
            queued.push((agent, head));
        }
    }
    Ok(queued)
}

/// What [`run`] needs beyond the forge client.
pub(crate) struct Deps {
    pub(crate) jobq: Arc<Jobq>,
    pub(crate) wanted: Arc<WantedWriter>,
    pub(crate) hives: Vec<String>,
    pub(crate) queue: async_nats::Client,
}

/// One pass against the live swarm. A pass that fails is logged and the next
/// scan tries again.
pub(crate) async fn run(deps: &Deps, forge: &Arc<forge::Client>) {
    let closures = match swarm_queue_client::agent_closures::open_or_create(&deps.queue).await {
        Ok(closures) => closures,
        Err(e) => {
            tracing::warn!(error = %e, "config drift: opening the agent-closures bucket failed");
            return;
        }
    };
    let world = Live {
        deps,
        forge: Arc::clone(forge),
        closures,
        flakes: Mutex::default(),
    };
    match pass(&world).await {
        Ok(queued) if !queued.is_empty() => {
            tracing::info!("config drift: pass queued {} builds", queued.len());
        }
        Ok(_) => tracing::debug!("config drift: pass done, no drift"),
        Err(e) => tracing::warn!(error = %format!("{e:#}"), "config drift: pass failed"),
    }
}

/// The real world of one pass.
struct Live<'a> {
    deps: &'a Deps,
    forge: Arc<forge::Client>,
    closures: Store,
    /// `internal/meta`'s `flake.nix` by commit, for the commits read this pass.
    flakes: Mutex<HashMap<String, Option<String>>>,
}

impl Live<'_> {
    /// `flake.nix` of `internal/meta` at `meta_rev`; `None` when it has none.
    async fn flake_at(&self, meta_rev: &str) -> Result<Option<String>> {
        if let Some(read) = self.flakes.lock().unwrap().get(meta_rev) {
            return Ok(read.clone());
        }
        let flake = self
            .forge
            .file_at_commit(SHARED_ORG, META_REPO, meta_rev, "flake.nix")
            .await
            .with_context(|| {
                format!("reading flake.nix of {SHARED_ORG}/{META_REPO} at {meta_rev}")
            })?
            .map(|bytes| String::from_utf8_lossy(&bytes).into_owned());
        self.flakes
            .lock()
            .unwrap()
            .insert(meta_rev.to_owned(), flake.clone());
        Ok(flake)
    }
}

impl World for Live<'_> {
    async fn placements(&self) -> Result<BTreeSet<(String, String)>> {
        crate::jobq::placements(&self.deps.wanted, &self.deps.hives).await
    }

    async fn heads(&self) -> Result<Vec<(String, String)>> {
        self.forge.list_config_heads(CONFIG_BRANCH).await
    }

    async fn built(&self, agent: &str) -> Result<Option<String>> {
        let (record, _) = crate::schedules::read_with(&self.closures, agent, |bytes| {
            Ok(Some(serde_json::from_slice::<AgentClosure>(bytes)?))
        })
        .await?;
        let Some(record) = record else {
            return Ok(None);
        };
        let Some(flake) = self.flake_at(&record.meta_rev).await? else {
            return Ok(None);
        };
        Ok(hive_meta_flake::pinned_rev(&flake, &format!("agent-{agent}")).map(str::to_owned))
    }

    fn attempted(&self, agent: &str, head: &str) -> bool {
        attempted(
            self.deps
                .jobq
                .lock()
                .unwrap_or_else(std::sync::PoisonError::into_inner)
                .graph(),
            agent,
            head,
        )
    }

    fn queue(&self, hive: &str, agent: &str, head: &str) -> bool {
        crate::jobq::insert_merged_deploy(&self.deps.jobq, hive, agent, head, None)
    }
}

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

    use hive_jobq::State;
    use hive_jobq::scheduler::{Outcome, Scheduler};

    use super::*;

    const OLD: &str = "1111111111111111111111111111111111111111";
    const HEAD: &str = "2222222222222222222222222222222222222222";
    const NEWER: &str = "3333333333333333333333333333333333333333";

    /// The swarm as a pass sees it. The graph is a real scheduler: what the pass
    /// queues, and what it finds queued, go through the production code.
    struct Fake {
        placed: BTreeSet<(String, String)>,
        heads: Vec<(String, String)>,
        built: BTreeMap<String, Result<Option<String>, String>>,
        jobq: Arc<Jobq>,
    }

    impl Fake {
        /// `atlas` is placed on `pr1ma`, its `main` is at `HEAD` and its last
        /// build used `OLD`.
        fn drifted() -> Self {
            Self {
                placed: [("pr1ma".to_owned(), "atlas".to_owned())].into(),
                heads: vec![("atlas".to_owned(), HEAD.to_owned())],
                built: BTreeMap::from([("atlas".to_owned(), Ok(Some(OLD.to_owned())))]),
                jobq: Arc::new(Mutex::new(Scheduler::new(
                    hive_jobq::Graph::new(),
                    hive_jobq::resources::ResourceTable::new(),
                ))),
            }
        }

        fn built_at(mut self, agent: &str, rev: &str) -> Self {
            self.built
                .insert(agent.to_owned(), Ok(Some(rev.to_owned())));
            self
        }

        /// The `rev` of every deploy of `agent` in the graph, oldest first.
        fn deploys(&self, agent: &str) -> Vec<(Option<String>, State)> {
            let guard = self.jobq.lock().unwrap();
            let mut nodes: Vec<_> = guard
                .graph()
                .nodes()
                .filter_map(|n| match &n.payload {
                    SwarmNodeKind::TriggerDeploy { agent: a, rev, .. } if a == agent => {
                        Some((n.id, rev.clone(), n.state))
                    }
                    _ => None,
                })
                .collect();
            nodes.sort_by_key(|(id, _, _)| *id);
            nodes
                .into_iter()
                .map(|(_, rev, state)| (rev, state))
                .collect()
        }

        /// Run every runnable node; the builds fail when `builds_fail`.
        async fn drain(&self, builds_fail: bool) {
            while let Some(runner) =
                Scheduler::claim_next(&self.jobq, move |_id, kind, builder| async move {
                    let outcome = match kind {
                        SwarmNodeKind::BuildMeta { .. } if builds_fail => {
                            Outcome::Failed("error: attribute missing".to_owned())
                        }
                        _ => Outcome::Done,
                    };
                    (builder, outcome)
                })
            {
                runner.await.1.expect("no growth declared");
            }
        }
    }

    impl World for Fake {
        async fn placements(&self) -> Result<BTreeSet<(String, String)>> {
            Ok(self.placed.clone())
        }

        async fn heads(&self) -> Result<Vec<(String, String)>> {
            Ok(self.heads.clone())
        }

        async fn built(&self, agent: &str) -> Result<Option<String>> {
            match self.built.get(agent) {
                Some(Ok(rev)) => Ok(rev.clone()),
                Some(Err(e)) => anyhow::bail!("{e}"),
                None => Ok(None),
            }
        }

        fn attempted(&self, agent: &str, head: &str) -> bool {
            attempted(self.jobq.lock().unwrap().graph(), agent, head)
        }

        fn queue(&self, hive: &str, agent: &str, head: &str) -> bool {
            crate::jobq::insert_merged_deploy(&self.jobq, hive, agent, head, None)
        }
    }

    fn queued(agent: &str, head: &str) -> Vec<(String, String)> {
        vec![(agent.to_owned(), head.to_owned())]
    }

    #[tokio::test]
    async fn a_head_that_moved_without_a_build_queues_one_build() {
        let world = Fake::drifted();

        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));

        assert_eq!(
            world.deploys("atlas"),
            [(Some(HEAD.to_owned()), State::Pending)]
        );
        let guard = world.jobq.lock().unwrap();
        let hives: Vec<_> = guard
            .graph()
            .nodes()
            .filter_map(|n| match &n.payload {
                SwarmNodeKind::TriggerDeploy { hive, .. } => Some(hive.clone()),
                _ => None,
            })
            .collect();
        assert_eq!(hives, ["pr1ma"]);
    }

    #[tokio::test]
    async fn a_head_that_was_built_queues_nothing() {
        let world = Fake::drifted().built_at("atlas", HEAD);

        assert_eq!(pass(&world).await.unwrap(), []);
        assert_eq!(world.deploys("atlas"), []);
    }

    #[tokio::test]
    async fn a_head_already_queued_is_not_queued_again() {
        let world = Fake::drifted();
        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));

        assert_eq!(pass(&world).await.unwrap(), []);

        assert_eq!(world.deploys("atlas").len(), 1);
    }

    #[tokio::test]
    async fn a_head_a_webhook_queued_is_not_queued_again() {
        let world = Fake::drifted();
        assert!(crate::jobq::insert_merged_deploy(
            &world.jobq,
            "pr1ma",
            "atlas",
            HEAD,
            Some(7)
        ));

        assert_eq!(pass(&world).await.unwrap(), []);

        assert_eq!(world.deploys("atlas").len(), 1);
    }

    #[tokio::test]
    async fn a_head_whose_build_failed_is_not_queued_every_pass() {
        let world = Fake::drifted();
        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
        world.drain(true).await;
        assert_eq!(
            world.deploys("atlas"),
            [(Some(HEAD.to_owned()), State::Skipped)]
        );

        for _ in 0..3 {
            assert_eq!(pass(&world).await.unwrap(), []);
        }

        assert_eq!(world.deploys("atlas").len(), 1);
    }

    #[tokio::test]
    async fn a_newer_head_is_queued_after_a_failed_one() {
        let mut world = Fake::drifted();
        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
        world.drain(true).await;
        world.heads = vec![("atlas".to_owned(), NEWER.to_owned())];

        assert_eq!(pass(&world).await.unwrap(), queued("atlas", NEWER));
    }

    #[tokio::test]
    async fn a_deploy_in_flight_holds_back_a_newer_head_until_it_ends() {
        let mut world = Fake::drifted();
        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
        world.heads = vec![("atlas".to_owned(), NEWER.to_owned())];
        assert_eq!(pass(&world).await.unwrap(), []);

        world.drain(false).await;
        // The build recorded the closure from `HEAD`.
        world
            .built
            .insert("atlas".to_owned(), Ok(Some(HEAD.to_owned())));

        assert_eq!(pass(&world).await.unwrap(), queued("atlas", NEWER));
    }

    #[tokio::test]
    async fn an_agent_nobody_places_or_two_hives_place_is_left_alone() {
        let mut world = Fake::drifted();
        world.heads = vec![
            ("atlas".to_owned(), HEAD.to_owned()),
            ("ghost".to_owned(), HEAD.to_owned()),
            ("twin".to_owned(), HEAD.to_owned()),
        ];
        world.placed.insert(("pr1ma".to_owned(), "twin".to_owned()));
        world.placed.insert(("hive2".to_owned(), "twin".to_owned()));
        let world = world.built_at("ghost", OLD).built_at("twin", OLD);

        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
    }

    #[tokio::test]
    async fn an_agent_with_no_build_to_compare_is_left_alone() {
        let mut world = Fake::drifted();
        world.heads = vec![
            ("atlas".to_owned(), HEAD.to_owned()),
            ("fresh".to_owned(), HEAD.to_owned()),
        ];
        world
            .placed
            .insert(("pr1ma".to_owned(), "fresh".to_owned()));
        world.built.insert("fresh".to_owned(), Ok(None));

        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
    }

    #[tokio::test]
    async fn an_agent_whose_build_cannot_be_read_does_not_stop_the_others() {
        let mut world = Fake::drifted();
        world.heads = vec![
            ("broken".to_owned(), HEAD.to_owned()),
            ("atlas".to_owned(), HEAD.to_owned()),
        ];
        world
            .placed
            .insert(("pr1ma".to_owned(), "broken".to_owned()));
        world
            .built
            .insert("broken".to_owned(), Err("kv unreachable".to_owned()));

        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
        assert_eq!(world.deploys("broken"), []);
    }

    #[tokio::test]
    async fn another_agents_deploy_does_not_hold_this_one_back() {
        let mut world = Fake::drifted();
        world.heads = vec![
            ("atlas".to_owned(), HEAD.to_owned()),
            ("iris".to_owned(), HEAD.to_owned()),
        ];
        world.placed.insert(("pr1ma".to_owned(), "iris".to_owned()));
        let world = world.built_at("iris", OLD);
        assert!(crate::jobq::insert_merged_deploy(
            &world.jobq,
            "pr1ma",
            "iris",
            HEAD,
            None
        ));

        assert_eq!(pass(&world).await.unwrap(), queued("atlas", HEAD));
    }
}