hyperhive/swarm-queue-client/src/operator_profiles.rs

//! The operator-profile KV bucket: one key,
//! [`crate::operator_profiles::KEY`], holding every operator's name and the
//! pronouns from that operator's own forge profile, as one JSON array.
//!
//! **swarm-controller is the only writer**, and creates the bucket on its
//! first write. An operator is a member of authelia's operator group; one who
//! leaves it drops out of the list on the controller's next pass.
//!
//! **Every agent reads the key with one `DIRECT.GET`**, read-only, when it
//! renders its system prompt
//! ([`crate::operator_profiles::direct_get_subject`]). One key rather than one
//! per operator so that the read is a single request on one literal subject,
//! the shape of [`crate::agent_inference`]'s read: no `STREAM.INFO`, no
//! consumer.

/// The KV bucket operator profiles are published into.
pub const BUCKET: &str = "operator-profiles";

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

/// The one key in [`BUCKET`]: the whole operator list.
pub const KEY: &str = "operators";

/// The subject [`KEY`] is stored under.
#[must_use]
pub fn key_subject() -> String {
    format!("$KV.{BUCKET}.{KEY}")
}

/// The request subject that reads [`KEY`]'s current entry, and so the one
/// subject the queue grants every agent in this bucket.
#[must_use]
pub fn direct_get_subject() -> String {
    format!("$JS.API.DIRECT.GET.{STREAM}.{}", key_subject())
}

/// One operator's profile: an element of the array under [`KEY`].
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
pub struct OperatorProfile {
    /// The operator's authelia username.
    pub name: String,
    /// The pronouns set on the operator's forge profile. `None` when unset or
    /// when the operator has no forge account yet.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub pronouns: Option<String>,
}

/// Open the profile bucket for writing, creating it if nothing has yet.
///
/// **Controller-side only.** `history: 1`: an agent 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,
        "operator-profiles",
        "Each operator's name and forge-profile pronouns, set by the swarm controller".to_owned(),
    )
    .await
}

/// What reading [`KEY`] found.
#[cfg(feature = "operator-profiles")]
#[derive(Debug)]
pub enum Read {
    /// The bucket answered; an absent or deleted key is no operators.
    Current(Vec<OperatorProfile>),
    /// Nothing answers for the bucket's stream: the controller has not written
    /// it yet, or `JetStream` is not serving it right now.
    NoBucket,
}

/// Read the operator list with the one subject the agent grant 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 = "operator-profiles")]
pub async fn read(
    client: &async_nats::Client,
) -> 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()).await {
        Ok(message) => message,
        Err(e) if matches!(e.kind(), DirectGetErrorKind::NotFound) => {
            return Ok(Read::Current(Vec::new()));
        }
        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()),
    };
    // A KV delete or purge is stored as an empty marker carrying this header.
    let removed = message
        .headers
        .get("KV-Operation")
        .is_some_and(|op| matches!(op.as_str(), "DEL" | "PURGE"));
    if removed {
        return Ok(Read::Current(Vec::new()));
    }
    Ok(Read::Current(serde_json::from_slice(&message.payload)?))
}

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

    /// The names the auth responder grants agents from.
    #[test]
    fn names_match_the_queue_grant() {
        assert_eq!(STREAM, format!("KV_{BUCKET}"));
        assert_eq!(key_subject(), "$KV.operator-profiles.operators");
        assert_eq!(
            direct_get_subject(),
            "$JS.API.DIRECT.GET.KV_operator-profiles.$KV.operator-profiles.operators"
        );
    }

    #[test]
    fn unset_pronouns_are_left_out_of_the_value() {
        let list = vec![
            OperatorProfile {
                name: "mara".to_owned(),
                pronouns: Some("she/her".to_owned()),
            },
            OperatorProfile {
                name: "sol".to_owned(),
                pronouns: None,
            },
        ];
        let json = serde_json::to_string(&list).unwrap();
        assert_eq!(
            json,
            r#"[{"name":"mara","pronouns":"she/her"},{"name":"sol"}]"#
        );
        assert_eq!(
            serde_json::from_str::<Vec<OperatorProfile>>(&json).unwrap(),
            list
        );
    }
}