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

//! Each agent's read access to `internal/docs`: a collaborator grant at
//! **read**, made here with the admin API so it does not depend on which hive
//! the agent runs on or whether that hive has a forge container.
//!
//! Two callers insert the same `GrantDocsRead` job node: agent creation, and
//! [`spawn`]'s pass at start and every [`RECONCILE_INTERVAL`] over every
//! agent that holds a store identity. The pass reaches agents created before
//! this node existed and agents whose creation-time grant failed. It only ever
//! adds: nobody is removed, whatever the roster or the forge says. The
//! decision is pure ([`plan`]), so the tests pin it.

use std::sync::Arc;

use anyhow::{Context, Result};
use forgejo_api::structs::{AddCollaboratorOptionPermission, RepoCollaboratorPermission};
use swarm_secret_client::{client::DEFAULT_CERT_MOUNT, policy};

use super::Client;
use super::agent_token::is_not_found;
use super::objects::{SHARED_DOCS_REPO, SHARED_ORG};

/// How often [`spawn`] re-checks every agent.
const RECONCILE_INTERVAL: std::time::Duration = std::time::Duration::from_mins(5);

/// What one pass found for one agent.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Observed {
    /// The agent is already a collaborator on `internal/docs`, at any level.
    Granted,
    /// The forge says the agent is not a collaborator, or has no such user —
    /// the queued job creates the user before it grants.
    Missing,
    /// The read failed. Nothing is known, so nothing is done.
    Unknown,
}

/// The agents a pass queues a grant for: those the forge says are not
/// collaborators. [`Observed::Unknown`] is never queued.
pub fn plan(observed: &[(String, Observed)]) -> Vec<String> {
    observed
        .iter()
        .filter(|(_, o)| *o == Observed::Missing)
        .map(|(agent, _)| agent.clone())
        .collect()
}

/// Whether a permission read says the user already has read access or more.
fn holds_read(p: &RepoCollaboratorPermission) -> bool {
    matches!(
        p.permission.as_deref(),
        Some("read" | "write" | "admin" | "owner")
    )
}

impl Client {
    /// Make `agent` a **read** collaborator on `internal/docs` — the whole job
    /// of the `GrantDocsRead` node. Idempotent, and never lowers: an agent that
    /// already holds read or more (write, admin, owner) is left as it is, since
    /// the add endpoint would overwrite the permission. Needs the forge user to
    /// exist.
    pub async fn grant_docs_read(&self, agent: &str) -> Result<()> {
        match self
            .api
            .repo_get_repo_permissions(SHARED_ORG, SHARED_DOCS_REPO, agent)
            .await
        {
            Ok(p) if holds_read(&p) => return Ok(()),
            Ok(_) => {}
            Err(e) if is_not_found(&e) => {}
            Err(e) => {
                return Err(e).with_context(|| {
                    format!("read {agent}'s permission on {SHARED_ORG}/{SHARED_DOCS_REPO}")
                });
            }
        }
        self.add_collaborator_to(
            SHARED_ORG,
            SHARED_DOCS_REPO,
            agent,
            AddCollaboratorOptionPermission::Read,
        )
        .await
    }

    /// Whether `agent` is a collaborator on `internal/docs` now.
    async fn observe_grant(&self, agent: &str) -> Observed {
        match self
            .api
            .repo_check_collaborator(SHARED_ORG, SHARED_DOCS_REPO, agent)
            .await
        {
            Ok(()) => Observed::Granted,
            Err(e) if is_not_found(&e) => Observed::Missing,
            Err(e) => {
                tracing::warn!(agent, error = %e, "docs grant: forge read failed");
                Observed::Unknown
            }
        }
    }

    /// Whether `internal/docs` exists now: `None` when the forge could not
    /// say. A missing repo makes every collaborator check answer 404 too, so
    /// it has to be told apart from "not a collaborator" before they run.
    async fn docs_repo_exists(&self) -> Option<bool> {
        match self.api.repo_get(SHARED_ORG, SHARED_DOCS_REPO).await {
            Ok(_) => Some(true),
            Err(e) if is_not_found(&e) => Some(false),
            Err(e) => {
                tracing::warn!(error = %e, "docs grants: forge read of the repo failed");
                None
            }
        }
    }

    /// One pass over `roster`.
    async fn observe_grants(&self, roster: &[String]) -> Vec<(String, Observed)> {
        let mut observed = Vec::with_capacity(roster.len());
        for agent in roster {
            observed.push((agent.clone(), self.observe_grant(agent).await));
        }
        observed
    }
}

/// The store's `hive-agent-*` cert-auth roles: exactly the agents that hold a
/// store identity.
async fn read_roster() -> Result<Vec<String>> {
    let store = crate::store::connect()
        .await
        .context("logging in to the swarm secret store")?;
    let roles = store
        .list_cert_roles(DEFAULT_CERT_MOUNT)
        .await
        .context("listing the store's cert-auth roles")?;
    Ok(policy::agents_from_role_names(&roles))
}

/// One pass: observe every agent of `roster` and return the ones without a
/// grant. A roster that could not be read yields none and touches nothing; so
/// does a pass whose `internal/docs` is missing or unreadable, which is checked
/// once before any agent.
async fn run_pass(client: &Client, roster: Result<Vec<String>>) -> Vec<String> {
    let roster = match roster {
        Ok(roster) => roster,
        Err(e) => {
            tracing::warn!(
                error = %format!("{e:#}"),
                retry_in_s = RECONCILE_INTERVAL.as_secs(),
                "docs grants: no roster; retrying next tick"
            );
            return Vec::new();
        }
    };
    match client.docs_repo_exists().await {
        Some(true) => {}
        Some(false) => {
            tracing::warn!(
                repo = format!("{SHARED_ORG}/{SHARED_DOCS_REPO}"),
                "docs grants: repo missing; no grants queued"
            );
            return Vec::new();
        }
        None => return Vec::new(),
    }
    let observed = client.observe_grants(&roster).await;
    let agents = plan(&observed);
    if agents.is_empty() {
        tracing::debug!(checked = observed.len(), "docs grants: all current");
    } else {
        tracing::info!(
            checked = observed.len(),
            queueing = agents.len(),
            "docs grants: queueing grants"
        );
    }
    agents
}

/// Check every agent now and every [`RECONCILE_INTERVAL`] after, and hand the
/// agents without a grant to `enqueue`, which inserts a `GrantDocsRead` node
/// for each. A pass that fails is logged and retried on the next tick; it
/// never stops the daemon.
pub fn spawn(client: Arc<Client>, enqueue: impl Fn(Vec<String>) + Send + 'static) {
    tokio::spawn(async move {
        let mut ticker = tokio::time::interval(RECONCILE_INTERVAL);
        loop {
            ticker.tick().await;
            let Some(agents) = super::within_forge_deadline("docs grants", async {
                run_pass(&client, read_roster().await).await
            })
            .await
            else {
                continue;
            };
            if !agents.is_empty() {
                enqueue(agents);
            }
        }
    });
}

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

    use axum::extract::{Path, State};
    use axum::http::{StatusCode, header::CONTENT_TYPE};
    use forgejo_api::{Auth, Forgejo};

    use super::*;

    /// What a stand-in `internal/docs` holds: who is a collaborator, and the
    /// request log.
    #[derive(Default)]
    struct Docs {
        /// The repo does not exist: its own lookup and every collaborator
        /// check answer 404.
        repo_missing: bool,
        collaborators: BTreeMap<String, String>,
        requests: Vec<(String, String)>,
        /// Answer every collaborator check with a 500.
        broken: bool,
    }

    type Shared = Arc<Mutex<Docs>>;

    /// A stand-in forge on a loopback port — never the real one — serving
    /// the collaborator check (`GET`) and add (`PUT`) on `internal/docs`.
    async fn stub_forge(collaborators: &[&str], broken: bool) -> (Client, Shared) {
        serve(Docs {
            collaborators: collaborators
                .iter()
                .map(|c| ((*c).to_owned(), "read".to_owned()))
                .collect(),
            broken,
            ..Docs::default()
        })
        .await
    }

    async fn serve(docs: Docs) -> (Client, Shared) {
        let docs: Shared = Arc::new(Mutex::new(docs));
        let app = axum::Router::new()
            .route("/api/v1/repos/internal/docs", axum::routing::get(repo))
            .route(
                "/api/v1/repos/internal/docs/collaborators/{user}",
                axum::routing::get(check).put(add),
            )
            .route(
                "/api/v1/repos/internal/docs/collaborators/{user}/permission",
                axum::routing::get(permission),
            )
            .with_state(Arc::clone(&docs));
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = url::Url::parse(&format!("http://{}/", listener.local_addr().unwrap())).unwrap();
        tokio::spawn(async move { axum::serve(listener, app).await });
        let api = Forgejo::new(Auth::Token("stub-token"), url.clone()).unwrap();
        (Client { api, url }, docs)
    }

    async fn check(
        State(docs): State<Shared>,
        Path(user): Path<String>,
    ) -> (
        StatusCode,
        [(axum::http::HeaderName, &'static str); 1],
        &'static str,
    ) {
        let mut docs = docs.lock().unwrap();
        docs.requests.push(("GET".to_owned(), user.clone()));
        let status = if docs.repo_missing {
            StatusCode::NOT_FOUND
        } else if docs.broken {
            StatusCode::INTERNAL_SERVER_ERROR
        } else if docs.collaborators.contains_key(&user) {
            StatusCode::NO_CONTENT
        } else {
            StatusCode::NOT_FOUND
        };
        (status, [(CONTENT_TYPE, "application/json")], "{}")
    }

    async fn repo(
        State(docs): State<Shared>,
    ) -> (
        StatusCode,
        [(axum::http::HeaderName, &'static str); 1],
        &'static str,
    ) {
        let mut docs = docs.lock().unwrap();
        docs.requests.push(("GET".to_owned(), "repo".to_owned()));
        let status = if docs.repo_missing {
            StatusCode::NOT_FOUND
        } else {
            StatusCode::OK
        };
        (status, [(CONTENT_TYPE, "application/json")], "{}")
    }

    async fn permission(State(docs): State<Shared>, Path(user): Path<String>) -> String {
        let mut docs = docs.lock().unwrap();
        docs.requests
            .push(("GET".to_owned(), format!("{user} permission")));
        let level = docs.collaborators.get(&user).map_or("none", String::as_str);
        format!(r#"{{"permission":"{level}"}}"#)
    }

    async fn add(State(docs): State<Shared>, Path(user): Path<String>, body: String) -> StatusCode {
        let mut docs = docs.lock().unwrap();
        docs.requests
            .push(("PUT".to_owned(), format!("{user} {body}")));
        docs.collaborators.insert(user, "read".to_owned());
        StatusCode::NO_CONTENT
    }

    fn roster(agents: &[&str]) -> Vec<String> {
        agents.iter().map(|a| (*a).to_owned()).collect()
    }

    fn puts(docs: &Shared) -> Vec<String> {
        docs.lock()
            .unwrap()
            .requests
            .iter()
            .filter(|(m, _)| m == "PUT")
            .map(|(_, what)| what.clone())
            .collect()
    }

    #[test]
    fn only_a_missing_grant_is_planned() {
        let observed = [
            ("a".to_owned(), Observed::Granted),
            ("b".to_owned(), Observed::Missing),
            ("c".to_owned(), Observed::Unknown),
        ];
        assert_eq!(plan(&observed), ["b"]);
    }

    /// The `GrantDocsRead` node's work: a **read** collaborator on
    /// `internal/docs`, not write.
    #[tokio::test]
    async fn the_grant_adds_the_agent_as_a_read_collaborator() {
        let (client, docs) = stub_forge(&[], false).await;

        client.grant_docs_read("alice").await.unwrap();

        let sent = puts(&docs);
        assert_eq!(sent.len(), 1, "{sent:?}");
        let (user, body) = sent[0].split_once(' ').unwrap();
        assert_eq!(user, "alice");
        let body: serde_json::Value = serde_json::from_str(body).unwrap();
        assert_eq!(body["permission"], "read");
    }

    /// The backfill queues the agents that lack the grant, and only those.
    #[tokio::test]
    async fn the_pass_queues_the_agents_missing_the_grant() {
        let (client, docs) = stub_forge(&["bob"], false).await;
        let queued = run_pass(&client, Ok(roster(&["alice", "bob", "carol"]))).await;

        assert_eq!(queued, ["alice", "carol"]);
        assert!(puts(&docs).is_empty(), "the pass itself only reads");
    }

    /// Once every agent holds the grant a pass queues nothing, and writes
    /// nothing — in particular it does not re-add a collaborator that holds a
    /// higher permission.
    #[tokio::test]
    async fn the_pass_is_a_no_op_for_agents_already_granted() {
        let (client, docs) = stub_forge(&["alice", "bob"], false).await;
        let queued = run_pass(&client, Ok(roster(&["alice", "bob"]))).await;

        assert!(queued.is_empty());
        assert!(puts(&docs).is_empty());
    }

    /// A roster that could not be read changes nothing: no grant queued, no
    /// request to the forge at all.
    #[tokio::test]
    async fn an_unreadable_roster_changes_nothing() {
        let (client, docs) = stub_forge(&["alice"], false).await;
        let queued = run_pass(&client, Err(anyhow::anyhow!("store is down"))).await;

        assert!(queued.is_empty());
        assert!(docs.lock().unwrap().requests.is_empty());
    }

    /// A forge that cannot answer the check is not a forge that says "not a
    /// collaborator": nothing is queued.
    #[tokio::test]
    async fn an_unreadable_forge_queues_nothing() {
        let (client, docs) = stub_forge(&[], true).await;
        let queued = run_pass(&client, Ok(roster(&["alice"]))).await;

        assert!(queued.is_empty());
        assert!(puts(&docs).is_empty());
    }

    /// A missing `internal/docs` answers 404 to every collaborator check, which
    /// would read as "not a collaborator" for every agent: nothing is queued,
    /// and the agents are not asked about one by one.
    #[tokio::test]
    async fn a_missing_repo_queues_nothing() {
        let (client, docs) = serve(Docs {
            repo_missing: true,
            ..Docs::default()
        })
        .await;
        let queued = run_pass(&client, Ok(roster(&["alice", "bob"]))).await;

        assert!(queued.is_empty());
        let requests = docs.lock().unwrap().requests.clone();
        assert_eq!(requests, [("GET".to_owned(), "repo".to_owned())]);
    }

    /// An agent an operator already raised above read keeps it: the node reads
    /// the permission first and sends no write.
    #[tokio::test]
    async fn the_grant_leaves_a_higher_permission_alone() {
        let (client, docs) = serve(Docs {
            collaborators: BTreeMap::from([("alice".to_owned(), "write".to_owned())]),
            ..Docs::default()
        })
        .await;

        client.grant_docs_read("alice").await.unwrap();

        assert!(puts(&docs).is_empty(), "{:?}", puts(&docs));
        assert_eq!(docs.lock().unwrap().collaborators["alice"], "write");
    }
}