//! Opening and reading the per-agent permissions KV bucket. The names, the
//! stored type and the import messages are `swarm_wire::agent_permissions`;
//! reading needs the `agent-permissions` feature and writing needs `kv`.

#[cfg(feature = "agent-permissions")]
use swarm_wire::agent_permissions::AgentPermissions;

/// Open the permissions bucket, creating it if nothing has yet.
///
/// **Controller-side only.** `history: 1`: a hive wants the current value.
#[cfg(feature = "kv")]
pub async fn open_or_create(
    client: &async_nats::Client,
) -> Result<async_nats::jetstream::kv::Store, crate::Error> {
    let js = async_nats::jetstream::new(client.clone());
    crate::kv::open_or_create(
        &js,
        swarm_wire::agent_permissions::BUCKET,
        "agent-permissions",
        "Permissions the swarm controller sets for each agent".to_owned(),
    )
    .await
}

/// What reading one agent's key found.
#[cfg(feature = "agent-permissions")]
#[derive(Debug)]
pub enum Read {
    /// The bucket answered; an absent or deleted key has every field unset.
    Current(AgentPermissions),
    /// Nothing answers for the bucket's stream. Not an unset value: it says
    /// nothing about the agent's permissions.
    NoBucket,
}

/// What an agent holds until its first read of its key succeeds: nothing, the
/// narrowest set. Not the default, which includes `Execution`; a read of an
/// unset key is what makes an agent hold the default.
#[cfg(feature = "agent-permissions")]
pub const BEFORE_FIRST_READ: &[hive_types::Permission] = &[];

/// The permissions held under `record`: its list when set (`[]` holds none),
/// the default when unset.
#[cfg(feature = "agent-permissions")]
#[must_use]
pub fn held(record: &AgentPermissions) -> Vec<hive_types::Permission> {
    record
        .permissions
        .clone()
        .unwrap_or_else(|| hive_types::Permission::DEFAULT.to_vec())
}

/// Whether `record` lets the agent's traffic leave unfiltered: its flag when
/// set, **on** when absent. A record written before the flag existed, or a
/// missing key, is an agent that has always had open access.
#[cfg(feature = "agent-permissions")]
#[must_use]
pub fn unrestricted(record: &AgentPermissions) -> bool {
    record.unrestricted_internet.unwrap_or(true)
}

/// How squid treats an agent's requests to names its lists do not allow.
#[cfg(feature = "agent-permissions")]
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum InternetMode {
    /// Any destination the safety denies leave.
    Unrestricted,
    /// As `Restricted`, except that a request to an unlisted name goes through
    /// and squid logs it as `would_deny`.
    Log,
    /// Only the listed names.
    Restricted,
}

/// The mode `record` puts the agent in. `Log` needs an explicit `false` on
/// `unrestricted_internet`: an absent flag is [`InternetMode::Unrestricted`]
/// whatever `internet_log` says.
#[cfg(feature = "agent-permissions")]
#[must_use]
pub fn internet_mode(record: &AgentPermissions) -> InternetMode {
    if unrestricted(record) {
        InternetMode::Unrestricted
    } else if record.internet_log == Some(true) {
        InternetMode::Log
    } else {
        InternetMode::Restricted
    }
}

/// Read `agent`'s current permissions with one direct get: the one subject
/// an agent's own grant covers, and one of those a hive's covers.
///
/// # Errors
///
/// A failed request or an undecodable value. A request the queue refused for
/// permissions reaches the client as a timeout, not as a refusal.
#[cfg(feature = "agent-permissions")]
pub async fn read(
    client: &async_nats::Client,
    agent: &str,
) -> Result<Read, Box<dyn std::error::Error + Send + Sync>> {
    use async_nats::jetstream::stream::DirectGetErrorKind;

    let js = async_nats::jetstream::new(client.clone());
    let stream = js
        .get_stream_no_info(swarm_wire::agent_permissions::STREAM)
        .await?;
    let message = match stream
        .direct_get_last_for_subject(swarm_wire::agent_permissions::key_subject(agent))
        .await
    {
        Ok(message) => message,
        Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => {
            return Ok(Read::Current(AgentPermissions::default()));
        }
        Err(e)
            if matches!(
                e.kind(),
                DirectGetErrorKind::ErrorResponse(status, _)
                    if status == async_nats::StatusCode::NO_RESPONDERS
            ) =>
        {
            return Ok(Read::NoBucket);
        }
        Err(e) => return Err(e.into()),
    };
    Ok(Read::Current(decode(
        message
            .headers
            .get("KV-Operation")
            .map(async_nats::HeaderValue::as_str),
        &message.payload,
    )?))
}

/// One stored message as a value: a KV delete or purge is stored as an empty
/// marker carrying the `KV-Operation` header, and reads as every field unset.
#[cfg(feature = "agent-permissions")]
fn decode(operation: Option<&str>, payload: &[u8]) -> serde_json::Result<AgentPermissions> {
    if matches!(operation, Some("DEL" | "PURGE")) {
        return Ok(AgentPermissions::default());
    }
    serde_json::from_slice(payload)
}

#[cfg(all(test, feature = "agent-permissions"))]
mod tests {
    use super::*;

    /// A set key wins, `[]` holds none, and an unset key is the default.
    #[test]
    fn an_unset_key_holds_the_default() {
        use hive_types::Permission;
        let record = |json: &str| -> AgentPermissions { serde_json::from_str(json).unwrap() };

        assert_eq!(
            held(&record(r#"{"permissions":["web_tools"]}"#)),
            vec![Permission::WebTools]
        );
        assert_eq!(held(&record(r#"{"permissions":[]}"#)), Vec::new());
        assert_eq!(held(&record("{}")), Permission::DEFAULT.to_vec());
        assert_ne!(
            Permission::DEFAULT.to_vec(),
            Vec::new(),
            "control: the default is not the empty list"
        );
    }

    /// An absent flag reads on, and so does a missing or deleted key; only an
    /// explicit `false` reads off.
    #[test]
    fn an_absent_internet_flag_reads_on() {
        let record = |json: &str| -> AgentPermissions { serde_json::from_str(json).unwrap() };
        assert!(unrestricted(&record("{}")));
        assert!(unrestricted(&record(r#"{"permissions":[]}"#)));
        assert!(unrestricted(&record(r#"{"unrestricted_internet":true}"#)));
        assert!(!unrestricted(&record(r#"{"unrestricted_internet":false}"#)));
        assert!(unrestricted(&decode(Some("DEL"), b"").unwrap()));
    }

    /// `Log` takes `unrestricted_internet: false` and `internet_log: true`
    /// together; either alone, or the pair on an unrestricted record, is not it.
    #[test]
    fn log_needs_the_old_flag_off_and_the_new_one_on() {
        let mode = |json: &str| internet_mode(&serde_json::from_str(json).unwrap());
        assert_eq!(
            mode(r#"{"unrestricted_internet":false,"internet_log":true}"#),
            InternetMode::Log
        );
        assert_eq!(
            mode(r#"{"unrestricted_internet":false}"#),
            InternetMode::Restricted
        );
        assert_eq!(
            mode(r#"{"unrestricted_internet":false,"internet_log":false}"#),
            InternetMode::Restricted
        );
        assert_eq!(mode(r#"{"internet_log":true}"#), InternetMode::Unrestricted);
        assert_eq!(
            mode(r#"{"unrestricted_internet":true,"internet_log":true}"#),
            InternetMode::Unrestricted
        );
        assert_eq!(mode("{}"), InternetMode::Unrestricted);
    }

    /// A decoder built before the flag existed has no such field and ignores
    /// unknown ones; it reads a record carrying the flag.
    #[test]
    fn a_decoder_without_the_flag_tolerates_it() {
        #[derive(serde::Deserialize)]
        struct Old {
            permissions: Option<Vec<hive_types::Permission>>,
        }
        let old: Old =
            serde_json::from_str(r#"{"permissions":["web_tools"],"unrestricted_internet":false}"#)
                .unwrap();
        assert_eq!(
            old.permissions,
            Some(vec![hive_types::Permission::WebTools])
        );
    }

    /// Fields a newer controller adds must not cost an older hive the ones it
    /// knows; a deleted key is unset.
    #[cfg(feature = "agent-permissions")]
    #[test]
    fn decoding_tolerates_unknown_fields_and_reads_a_delete_as_unset() {
        let decoded = decode(
            None,
            br#"{"permissions":["manage_root_agent"],"later_field":["meta"]}"#,
        )
        .unwrap();
        assert_eq!(decoded.permissions.map(|c| c.len()), Some(1));
        assert_eq!(
            decode(Some("DEL"), b"").unwrap(),
            AgentPermissions::default()
        );
        assert_eq!(
            decode(Some("PURGE"), b"").unwrap(),
            AgentPermissions::default()
        );
    }
}