//! The connection side of the per-agent closure KV bucket. The bucket's names
//! and stored type are `swarm_wire::agent_closures`.

/// Open the closures bucket for writing, creating it if nothing has yet.
///
/// **Controller-side only.** `history: 1`: a reader wants the latest build.
#[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_closures::BUCKET,
        "agent-closures",
        "The latest system closure the swarm builder built for each agent".to_owned(),
    )
    .await
}

/// What reading one agent's key found.
#[cfg(feature = "agent-closures")]
#[derive(Debug)]
pub enum Read {
    /// The bucket answered: the agent's latest build, or `None` for an absent
    /// or deleted key.
    Current(Option<swarm_wire::agent_closures::AgentClosure>),
    /// Nothing answers for the bucket's stream: no build has been recorded
    /// yet, or `JetStream` is not serving it right now.
    NoBucket,
}

/// Read `agent`'s latest closure record with one direct get.
///
/// # 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-closures")]
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;
    use swarm_wire::agent_closures::{STREAM, key_subject};

    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(None));
        }
        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 record: a KV delete or purge is stored as an empty
/// marker carrying the `KV-Operation` header, and reads as no record.
#[cfg(feature = "agent-closures")]
fn decode(
    operation: Option<&str>,
    payload: &[u8],
) -> serde_json::Result<Option<swarm_wire::agent_closures::AgentClosure>> {
    if matches!(operation, Some("DEL" | "PURGE")) {
        return Ok(None);
    }
    serde_json::from_slice(payload).map(Some)
}

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

    #[test]
    fn a_stored_record_reads_and_a_delete_is_no_record() {
        let payload = br#"{"meta_rev":"abc","store_path":"/nix/store/x","system":"x86_64-linux","built_unix":7,"cached":true}"#;
        let record = decode(None, payload).expect("decodes").expect("a record");
        assert_eq!(record.meta_rev, "abc");
        assert!(record.cached);
        assert!(decode(Some("DEL"), b"").expect("decodes").is_none());
        assert!(decode(Some("PURGE"), b"").expect("decodes").is_none());
    }
}