hyperhive/swarm-controller/src/forge/operator_profiles.rs

//! Each operator's forge-profile pronouns, published as one list under the
//! `operator-profiles` KV bucket's one key
//! (`swarm_queue_client::operator_profiles`) at start and every
//! [`RECONCILE_INTERVAL`] after. Every agent reads that key into its system
//! prompt.
//!
//! An operator is a member of authelia's operator group, read the way
//! `crate::matrix_account::operator_admin` reads it. A pass writes the key
//! only when the list changed, so a name no longer in the group drops out of
//! it. An operator whose forge profile could not be read keeps the entry the
//! stored list holds (name only when it holds none); one with no forge account
//! yet is listed with no pronouns.

use std::sync::Arc;

use anyhow::{Context, Result};
use async_nats::jetstream::kv::Store;
use swarm_queue_client::operator_profiles::{self, KEY, OperatorProfile};

use super::Client;
use super::agent_token::is_not_found;
use crate::auth::AuthBridge;

/// How often [`spawn`] re-reads the operator group and their profiles.
const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5);

/// Where an operator's pronouns come from.
pub(crate) trait Profiles {
    /// The pronouns on `name`'s forge profile: `None` when unset or when the
    /// forge has no such account.
    fn pronouns(&self, name: &str) -> impl Future<Output = Result<Option<String>>> + Send;
}

impl Profiles for Client {
    async fn pronouns(&self, name: &str) -> Result<Option<String>> {
        match self.api.call(|api| api.user_get(name).into_future()).await {
            Ok(user) => Ok(user
                .pronouns
                .map(|p| p.trim().to_owned())
                .filter(|p| !p.is_empty())),
            Err(e) if is_not_found(&e) => Ok(None),
            Err(e) => Err(e).with_context(|| format!("read forge user {name}")),
        }
    }
}

/// What one pass did.
#[derive(Debug, Default, PartialEq, Eq)]
pub(crate) struct Outcome {
    /// The list differed from the stored one and was written.
    written: bool,
    /// Operators whose profile could not be read, left as stored.
    kept: usize,
}

/// Bring [`KEY`] in `store` to the list of `operators`, each with its
/// pronouns as `forge` reports them, sorted by name.
///
/// # Errors
/// When the key cannot be read or written. A forge read that fails is logged
/// and keeps that operator's stored entry. A stored value this build cannot
/// decode reads as an empty list, so the pass replaces it.
pub(crate) async fn sync(
    operators: &[String],
    forge: &impl Profiles,
    store: &Store,
) -> Result<Outcome> {
    let stored: Vec<OperatorProfile> = store
        .get(KEY)
        .await
        .context("reading the operator list")?
        .and_then(|value| serde_json::from_slice(&value).ok())
        .unwrap_or_default();
    let mut outcome = Outcome::default();
    let mut list = Vec::with_capacity(operators.len());
    for name in operators {
        let pronouns = match forge.pronouns(name).await {
            Ok(pronouns) => pronouns,
            Err(e) => {
                tracing::warn!(
                    operator = %name,
                    error = %format!("{e:#}"),
                    "operator profiles: forge profile unreadable; keeping the stored entry"
                );
                outcome.kept += 1;
                stored
                    .iter()
                    .find(|o| o.name == *name)
                    .and_then(|o| o.pronouns.clone())
            }
        };
        list.push(OperatorProfile {
            name: name.clone(),
            pronouns,
        });
    }
    list.sort_by(|a, b| a.name.cmp(&b.name));
    list.dedup_by(|a, b| a.name == b.name);
    if list != stored {
        let value = serde_json::to_vec(&list).context("encoding the operator list")?;
        store
            .put(KEY, value.into())
            .await
            .context("writing the operator list")?;
        outcome.written = true;
    }
    Ok(outcome)
}

/// One pass: read the operator group, then [`sync`] the key to it. Fails as a
/// whole, changing nothing, when the group cannot be read.
async fn pass(forge: &Client, bridge: &AuthBridge, queue: &async_nats::Client) -> Result<Outcome> {
    let operators = bridge
        .list_operators()
        .await
        .context("reading authelia's operator group")?;
    let store = operator_profiles::open_or_create(queue)
        .await
        .context("opening the operator-profiles bucket")?;
    sync(&operators, forge, &store).await
}

/// Publish the operators' profiles now and every [`RECONCILE_INTERVAL`]
/// after. A pass that fails is logged and retried on the next tick; it never
/// stops the daemon.
pub fn spawn(forge: Arc<Client>, bridge: Arc<AuthBridge>, queue: async_nats::Client) {
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
        loop {
            ticker.tick().await;
            let Some(outcome) =
                super::within_forge_deadline("operator profiles", pass(&forge, &bridge, &queue))
                    .await
            else {
                continue;
            };
            match outcome {
                Ok(outcome) if outcome.written => tracing::info!(
                    kept = outcome.kept,
                    "operator profiles: operator list updated"
                ),
                Ok(outcome) => tracing::debug!(kept = outcome.kept, "operator profiles: unchanged"),
                Err(e) => tracing::warn!(
                    error = %format!("{e:#}"),
                    retry_in_s = RECONCILE_INTERVAL.as_secs(),
                    "operator profiles: pass failed; retrying next tick"
                ),
            }
        }
    });
}

#[cfg(test)]
mod tests {
    use std::collections::BTreeMap;

    use super::*;

    /// A forge answering from a table: `Err` for a name whose read fails.
    struct Fake(BTreeMap<&'static str, Result<Option<&'static str>, ()>>);

    impl Profiles for Fake {
        async fn pronouns(&self, name: &str) -> Result<Option<String>> {
            match self.0.get(name) {
                Some(Ok(p)) => Ok(p.map(str::to_owned)),
                Some(Err(())) => anyhow::bail!("the forge refused the read"),
                None => Ok(None),
            }
        }
    }

    fn names(names: &[&str]) -> Vec<String> {
        names.iter().map(|&n| n.to_owned()).collect()
    }

    async fn stored(store: &Store) -> Vec<OperatorProfile> {
        let value = store.get(KEY).await.unwrap().expect("the list is written");
        serde_json::from_slice(&value).unwrap()
    }

    fn profile(name: &str, pronouns: Option<&str>) -> OperatorProfile {
        OperatorProfile {
            name: name.to_owned(),
            pronouns: pronouns.map(str::to_owned),
        }
    }

    #[tokio::test]
    async fn the_list_is_written_once_and_a_departed_operator_drops_out() {
        let nats = crate::test_nats::TestNats::start();
        let client = nats.client().await;
        let store = operator_profiles::open_or_create(&client).await.unwrap();
        let forge = Fake([("mara", Ok(Some("she/her"))), ("sol", Ok(None))].into());

        let outcome = sync(&names(&["sol", "mara"]), &forge, &store)
            .await
            .unwrap();
        assert!(outcome.written);
        assert_eq!(
            stored(&store).await,
            [profile("mara", Some("she/her")), profile("sol", None)]
        );
        let revision = store.entry(KEY).await.unwrap().unwrap().revision;

        // An unchanged list is not rewritten.
        let outcome = sync(&names(&["mara", "sol"]), &forge, &store)
            .await
            .unwrap();
        assert_eq!(outcome, Outcome::default());
        assert_eq!(store.entry(KEY).await.unwrap().unwrap().revision, revision);

        let outcome = sync(&names(&["mara"]), &forge, &store).await.unwrap();
        assert!(outcome.written);
        assert_eq!(stored(&store).await, [profile("mara", Some("she/her"))]);
    }

    #[tokio::test]
    async fn a_failed_forge_read_keeps_the_stored_entry() {
        let nats = crate::test_nats::TestNats::start();
        let client = nats.client().await;
        let store = operator_profiles::open_or_create(&client).await.unwrap();
        let reachable = Fake(
            [
                ("mara", Ok(Some("she/her"))),
                ("sol", Ok(Some("they/them"))),
            ]
            .into(),
        );
        sync(&names(&["mara", "sol"]), &reachable, &store)
            .await
            .unwrap();

        // Control: the same sync does change a value the forge reads anew.
        // `ren` is new and unreadable, so it is listed by name only.
        let failing = Fake(
            [
                ("mara", Err(())),
                ("sol", Ok(Some("he/him"))),
                ("ren", Err(())),
            ]
            .into(),
        );
        let outcome = sync(&names(&["mara", "sol", "ren"]), &failing, &store)
            .await
            .unwrap();
        assert_eq!(outcome.kept, 2);
        assert_eq!(
            stored(&store).await,
            [
                profile("mara", Some("she/her")),
                profile("ren", None),
                profile("sol", Some("he/him"))
            ]
        );
    }

    /// The agent's read (`operator_profiles::read`, one direct get) finds
    /// what a pass wrote, an empty list for a bucket with no key yet, and
    /// `NoBucket` before the bucket exists.
    #[tokio::test]
    async fn the_agent_read_finds_the_list_and_falls_back_without_one() {
        use swarm_queue_client::operator_profiles::{Read, read};

        let nats = crate::test_nats::TestNats::start();
        let client = nats.client().await;
        assert!(matches!(read(&client).await.unwrap(), Read::NoBucket));

        let store = operator_profiles::open_or_create(&client).await.unwrap();
        let Read::Current(empty) = read(&client).await.unwrap() else {
            panic!("the bucket exists");
        };
        assert_eq!(empty, []);

        let forge = Fake([("mara", Ok(Some("she/her")))].into());
        sync(&names(&["mara"]), &forge, &store).await.unwrap();
        let Read::Current(list) = read(&client).await.unwrap() else {
            panic!("the bucket exists");
        };
        assert_eq!(list, [profile("mara", Some("she/her"))]);

        store.delete(KEY).await.unwrap();
        let Read::Current(deleted) = read(&client).await.unwrap() else {
            panic!("the bucket exists");
        };
        assert_eq!(deleted, []);
    }
}