//! The permissions the swarm grants this agent.
//!
//! swarm-controller keeps the agent's set under its key in the
//! `agent-permissions` KV bucket. This task reads that key every
//! [`POLL_INTERVAL`] with the one subject the agent's own queue credential is
//! granted there, and [`current`] answers with the last value read. The turn
//! loop asks [`current`] for every turn's `--tools`, `--allowedTools` and MCP
//! config, so a change reaches the next turn.
//!
//! Until the first successful read after a start the agent holds nothing
//! ([`BEFORE_FIRST_READ`]), so a queue that is down at start never leaves it
//! wider than the swarm grants. After one, a set key wins, `[]` holds none and
//! an unset key is the default. A failed read keeps the last value.

use std::sync::{PoisonError, RwLock};
use std::time::Duration;

use hive_types::Permission;
use swarm_queue_client::agent_permissions::{BEFORE_FIRST_READ, Read, held, read};

/// How often the key is read. Also the longest a change can take to reach
/// the agent's next turn.
const POLL_INTERVAL: Duration = Duration::from_secs(10);

/// The last value read; `None` until the first read succeeds.
static CURRENT: RwLock<Option<Vec<Permission>>> = RwLock::new(None);

/// The permissions this agent holds now.
#[must_use]
pub fn current() -> Vec<Permission> {
    let last = CURRENT
        .read()
        .unwrap_or_else(PoisonError::into_inner)
        .clone();
    held_now(last)
}

/// What the agent holds given the last value read: that value, or nothing
/// before the first read has succeeded.
fn held_now(last: Option<Vec<Permission>>) -> Vec<Permission> {
    last.unwrap_or_else(|| BEFORE_FIRST_READ.to_vec())
}

/// What one read of the key leaves the agent holding; `None` (keep the last
/// value) when the read failed or the bucket did not answer.
fn after(read: Option<Read>) -> Option<Vec<Permission>> {
    match read? {
        Read::Current(record) => Some(held(&record)),
        Read::NoBucket => None,
    }
}

/// The last value read once `read` has been applied to `last`.
fn settle(last: Option<Vec<Permission>>, read: Option<Read>) -> Option<Vec<Permission>> {
    after(read).or(last)
}

/// Run the poller for the harness's lifetime, if this agent has a label to
/// name its key with.
pub async fn run() {
    let agent = crate::identity::label();
    if agent.is_empty() {
        tracing::warn!("this agent has no label; it holds no swarm permissions");
        return;
    }
    let client = crate::swarm_queue::client().await;
    tracing::warn!(
        %agent,
        "permissions: holding none (no bash) until the first read of this agent's key succeeds"
    );
    loop {
        let read = read_once(&client, &agent).await;
        {
            let mut current = CURRENT.write().unwrap_or_else(PoisonError::into_inner);
            let next = settle(current.clone(), read);
            if *current != next {
                tracing::info!(perms = ?next, "permissions: the swarm's set changed");
                *current = next;
            }
        }
        tokio::time::sleep(POLL_INTERVAL).await;
    }
}

/// The key's current value, or `None` when it could not be read.
async fn read_once(client: &async_nats::Client, agent: &str) -> Option<Read> {
    // An unconnected client does not fail a JetStream request, it hangs on it.
    swarm_queue_client::ensure_connected(client).ok()?;
    match read(client, agent).await {
        Ok(read) => Some(read),
        Err(e) => {
            tracing::warn!(error = %e, "permissions: reading this agent's key failed");
            None
        }
    }
}

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

    fn record(json: &str) -> Read {
        Read::Current(serde_json::from_str(json).expect("decodes"))
    }

    /// What the agent holds after each read in turn, from a fresh start.
    fn held_after(reads: Vec<Option<Read>>) -> Vec<Permission> {
        held_now(reads.into_iter().fold(None, settle))
    }

    #[test]
    fn before_the_first_read_execution_is_not_held() {
        let held = held_after(Vec::new());
        assert!(held.is_empty(), "held {held:?} before any read");
        assert!(!held.contains(&Permission::Execution));
        assert!(
            Permission::DEFAULT.contains(&Permission::Execution),
            "control: the default does hold it"
        );
        for failed in [None, Some(Read::NoBucket)] {
            assert!(
                held_after(vec![failed]).is_empty(),
                "a read that fails leaves the agent closed"
            );
        }
    }

    #[test]
    fn a_read_with_the_key_unset_holds_the_default() {
        let held = held_after(vec![Some(record("{}"))]);
        assert_eq!(held, Permission::DEFAULT.to_vec());
        assert!(held.contains(&Permission::Execution));
    }

    #[test]
    fn a_read_with_a_restricted_set_holds_exactly_that_set() {
        assert_eq!(
            held_after(vec![Some(record(r#"{"permissions":["web_tools"]}"#))]),
            vec![Permission::WebTools]
        );
        assert_eq!(
            held_after(vec![Some(record(r#"{"permissions":[]}"#))]),
            Vec::new()
        );
    }

    #[test]
    fn a_failed_read_after_a_success_keeps_the_last_set() {
        let web = Some(record(r#"{"permissions":["web_tools"]}"#));
        assert_eq!(
            held_after(vec![web, None, Some(Read::NoBucket)]),
            vec![Permission::WebTools]
        );
    }
}