//! The per-agent permissions KV bucket: one key per agent, holding the
//! [`hive_sh4re::permissions::Permission`] set the swarm grants it.
//!
//! **swarm-controller is the only writer**, and creates the bucket when it
//! starts. A missing key or an unset `permissions` field leaves the agent on
//! the compiled-in default (`Permission::default_for`); an empty list grants
//! nothing.
//!
//! **Two readers, each enforcing its own part.** The hive running the agent
//! applies `manage_root_agent` as bind mounts at container start; its grant
//! covers every key of this bucket, read-only, for the reason
//! [`crate::agent_limits`] gives. The agent itself (its harness and subagent
//! daemon) applies the tool permissions; its grant covers only its own key.
//! Both read with one `DIRECT.GET` on the agent's key
//! ([`crate::agent_permissions::key_subject`]).
//!
//! Keyed by the agent alone, like [`crate::agent_limits`]: an agent can move
//! between hives, and its permissions go with it.
//!
//! Also carries the one-shot import a hive sends of the grants its meta repo
//! held ([`crate::agent_permissions::import_subject`]), answered by the
//! controller.

/// The KV bucket permissions are published into, keyed by agent name.
pub const BUCKET: &str = "agent-permissions";

/// The stream backing [`BUCKET`]. `create_key_value` prefixes `KV_`.
pub const STREAM: &str = "KV_agent-permissions";

/// The subject `agent`'s entry is stored under.
#[must_use]
pub fn key_subject(agent: &str) -> String {
    format!("$KV.{BUCKET}.{agent}")
}

const IMPORT_SUBJECT_PREFIX: &str = "$SWARM.permissions-import";

/// The subject `hive` sends its import on. One per hive, so the hive's grant
/// names only its own, and the controller takes the hive from the subject
/// rather than from the payload.
#[must_use]
pub fn import_subject(hive: &str) -> String {
    format!("{IMPORT_SUBJECT_PREFIX}.{hive}")
}

/// Every [`import_subject`] at once, for the controller's subscription.
#[must_use]
pub fn import_subject_wildcard() -> String {
    import_subject("*")
}

/// The hive an [`import_subject`] names, or `None` for any other subject.
#[must_use]
pub fn hive_from_import_subject(subject: &str) -> Option<&str> {
    let hive = subject
        .strip_prefix(IMPORT_SUBJECT_PREFIX)?
        .strip_prefix('.')?;
    (!hive.is_empty() && !hive.contains('.')).then_some(hive)
}

/// How long a hive waits for the controller's answer to an import, and the
/// queue lets the controller answer.
pub const IMPORT_REPLY_DEADLINE: std::time::Duration = std::time::Duration::from_secs(30);

/// One agent's permissions: the value under `BUCKET`/`<agent>`. `None` is
/// "not set by the swarm", which a hive reads as its default.
///
/// No `deny_unknown_fields`: a field a newer controller adds must not cost an
/// older hive the fields it knows.
#[cfg(feature = "agent-permissions")]
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct AgentPermissions {
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub permissions: Option<Vec<hive_sh4re::permissions::Permission>>,
}

/// What a hive sends once: the permissions its meta repo held, by agent.
#[cfg(feature = "agent-permissions")]
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct Import {
    pub permissions: std::collections::BTreeMap<String, Vec<hive_sh4re::permissions::Permission>>,
}

/// The controller's answer to an [`Import`].
#[cfg(feature = "agent-permissions")]
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(tag = "state", rename_all = "kebab-case")]
pub enum ImportReply {
    /// Every agent was handled. `written` had no key and got the hive's value;
    /// `kept` already had a key, which was left alone; `skipped` is not an
    /// agent the sending hive's wanted state declares.
    Imported {
        written: Vec<String>,
        kept: Vec<String>,
        skipped: Vec<String>,
    },
    /// Nothing is known to be written; the hive sends again later.
    Failed { error: String },
}

/// 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,
        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,
}

/// The file hive-c0re writes an agent's [`HiveCopy`] to in that agent's
/// runtime dir.
#[cfg(feature = "agent-permissions")]
pub const HIVE_COPY_FILE: &str = "swarm-permissions.json";

/// Where the agent's own processes read [`HIVE_COPY_FILE`]: the runtime dir
/// is mounted at `/run/hive` in the container.
#[cfg(feature = "agent-permissions")]
pub const HIVE_COPY_PATH: &str = "/run/hive/swarm-permissions.json";

/// What hive-c0re holds for one agent: its record as the hive last stored it
/// (read from this bucket, or taken from the meta repo before the import),
/// and whether the meta repo's import is still pending. The agent's harness
/// and subagent daemon hold [`HiveCopy::before_read`] until their own first
/// read of the key succeeds, and judge every read with
/// [`HiveCopy::after_read`].
#[cfg(feature = "agent-permissions")]
#[derive(Debug, Clone, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct HiveCopy {
    #[serde(flatten)]
    pub stored: AgentPermissions,
    #[serde(default)]
    pub import_pending: bool,
}

#[cfg(feature = "agent-permissions")]
impl HiveCopy {
    /// The copy at `path`. A missing or unreadable file is the empty copy,
    /// which holds the default and has no import pending.
    #[must_use]
    pub fn read_at(path: &std::path::Path) -> Self {
        let bytes = match std::fs::read(path) {
            Ok(bytes) => bytes,
            Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Self::default(),
            Err(e) => {
                tracing::warn!(path = %path.display(), error = %e, "permissions: the hive's copy is unreadable");
                return Self::default();
            }
        };
        serde_json::from_slice(&bytes).unwrap_or_else(|e| {
            tracing::warn!(path = %path.display(), error = %e, "permissions: the hive's copy does not decode");
            Self::default()
        })
    }

    /// What `agent` holds before any read of its key has succeeded: the
    /// stored list, or the default when the hive stored none.
    #[must_use]
    pub fn before_read(&self, agent: &str) -> Vec<hive_sh4re::permissions::Permission> {
        self.stored
            .permissions
            .clone()
            .unwrap_or_else(|| hive_sh4re::permissions::Permission::default_for(agent))
    }

    /// What `agent` holds after reading `read` from its key. An unset key
    /// while the import is pending may only mean the grant is not imported
    /// yet, so the stored list stands; otherwise unset is the default.
    #[must_use]
    pub fn after_read(
        &self,
        agent: &str,
        read: &AgentPermissions,
    ) -> Vec<hive_sh4re::permissions::Permission> {
        match &read.permissions {
            Some(list) => list.clone(),
            None if self.import_pending => self.before_read(agent),
            None => hive_sh4re::permissions::Permission::default_for(agent),
        }
    }
}

/// 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(STREAM).await?;
    let message = match stream.direct_get_last_for_subject(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(test)]
mod tests {
    use super::*;

    /// The names the grants in `swarm-nats-auth` are built from.
    #[test]
    fn names_match_the_queue_grant() {
        assert_eq!(STREAM, format!("KV_{BUCKET}"));
        assert_eq!(key_subject("iris"), "$KV.agent-permissions.iris");
        assert_eq!(import_subject("pr1ma"), "$SWARM.permissions-import.pr1ma");
        assert_eq!(import_subject_wildcard(), "$SWARM.permissions-import.*");
    }

    #[test]
    fn the_hive_is_read_back_from_its_import_subject_only() {
        assert_eq!(
            hive_from_import_subject(&import_subject("pr1ma")),
            Some("pr1ma")
        );
        for other in [
            "$SWARM.permissions-import",
            "$SWARM.permissions-import.",
            "$SWARM.permissions-import.a.b",
            "$SWARM.permissions-importx.pr1ma",
            "$SWARM.deploy.pr1ma",
        ] {
            assert_eq!(hive_from_import_subject(other), None, "{other}");
        }
    }

    /// Unset is left out on the wire, `{}` is unset, and an empty list stays
    /// an empty list: "grants nothing" is not "not set".
    #[cfg(feature = "agent-permissions")]
    #[test]
    fn unset_and_empty_are_different_values() {
        use hive_sh4re::permissions::Permission;
        assert_eq!(
            serde_json::to_string(&AgentPermissions::default()).unwrap(),
            "{}"
        );
        let none: AgentPermissions = serde_json::from_str(r#"{"permissions":[]}"#).unwrap();
        assert_eq!(none.permissions, Some(Vec::new()));
        let ruth: AgentPermissions =
            serde_json::from_str(r#"{"permissions":["manage_root_agent"]}"#).unwrap();
        assert_eq!(ruth.permissions, Some(vec![Permission::ManageRootAgent]));
    }

    /// Before a read the stored list stands; after one, a set key wins and an
    /// unset key is the default unless the import is pending.
    #[cfg(feature = "agent-permissions")]
    #[test]
    fn the_hive_copy_stands_until_read_and_while_the_import_is_pending() {
        use hive_sh4re::permissions::Permission;
        let record = |json: &str| -> AgentPermissions { serde_json::from_str(json).unwrap() };
        let copy = |json: &str| -> HiveCopy { serde_json::from_str(json).unwrap() };
        let web = vec![Permission::WebTools];
        let default = Permission::DEFAULT.to_vec();

        assert_eq!(
            copy(r#"{"permissions":["web_tools"]}"#).before_read("atlas"),
            web
        );
        assert_eq!(copy("{}").before_read("atlas"), default);

        let pending = copy(r#"{"permissions":["web_tools"],"import_pending":true}"#);
        assert_eq!(pending.after_read("atlas", &record("{}")), web);
        assert_eq!(
            pending.after_read("atlas", &record(r#"{"permissions":[]}"#)),
            Vec::new()
        );
        let settled = copy(r#"{"permissions":["web_tools"]}"#);
        assert_eq!(settled.after_read("atlas", &record("{}")), default);
    }

    /// 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()
        );
    }
}