//! Provisioning a hive's access to the swarm's secret store: the policy its
//! certificate carries, and the cert-auth role that hands it that policy.
//!
//! The controller writes a credential; the hive fetches it with its own
//! certificate. Without the role the certificate authenticates to nothing, and
//! without the policy the token it gets may read nothing — either way the read
//! half of a delivery answers 403.
//!
//! A pass runs at startup, and repeats until it reaches the store. The two
//! start together, so the store being the slower one is the ordinary case
//! rather than an edge one, and a pass that gave up there provisions nobody
//! until something restarts this daemon. Per-hive failures are different: they
//! are reported and skipped — one hive that cannot be provisioned should not
//! take down a daemon that serves everything else, and a store that answers is
//! answering the same way next time, so the next start is the retry for those.
//!
//! The document names the hive it is written for: one stanza is the same
//! everywhere, the other is scoped to the reader's own name
//! ([`swarm_secret_client::policy`] explains the asymmetry). There is no shared
//! document to hoist this render up to — doing that hands every hive the stanza
//! naming one of them. The hive list is loaded once because a config change
//! means a redeploy.

use std::{future::Future, time::Duration};

use anyhow::{Context, Result};
use swarm_secret_client::{Error, SecretStore, client::DEFAULT_CERT_MOUNT, policy};

/// File holding the authority hives are issued from, as
/// `swarm-controller.nix` names it.
pub const ENV_HIVE_CLIENT_CA: &str = "SWARM_CONTROLLER_HIVE_CLIENT_CA_FILE";

/// How long to wait before re-attempting a pass that never reached the store.
///
/// `nix/host-modules/swarm-bao.nix`'s bootstrap unit retries the other side of
/// this same race on the same cadence: the two halves of one race should not
/// disagree about how long a wait is worth.
const RETRY_INTERVAL: Duration = Duration::from_secs(30);

/// How many times [`RETRY_INTERVAL`] is waited before giving up. A store that
/// has not appeared within [`RETRY_WINDOW`] is absent rather than slow, and a
/// daemon logging into the void forever tells an operator nothing.
const RETRY_ATTEMPTS: u32 = 2_880;

/// What [`RETRY_ATTEMPTS`] at [`RETRY_INTERVAL`] come to. A named third
/// constant because it is the claim the other two make together, and either
/// one of them can be edited alone.
const RETRY_WINDOW: Duration = Duration::from_hours(24);

/// A pass that never got as far as the store, carrying what to say about it.
///
/// Returned rather than logged where it happens: the first attempt deserves a
/// warning and the two-hundredth is noise, and only the caller knows which one
/// this is.
struct Unreachable(String);

/// Give every hive in `hives` the access its certificate needs.
///
/// Returns once the first pass is done, so a store that is already up leaves
/// nothing deferred. When that pass cannot reach the store, it repeats in the
/// background for up to [`RETRY_WINDOW`].
pub async fn ensure_hive_access(hives: Vec<String>) {
    if let Err(Unreachable(reason)) = provision(&hives).await {
        tracing::warn!(
            reason,
            retry_in = ?RETRY_INTERVAL,
            "hive provisioning could not reach the swarm secret store; retrying in the background"
        );
        spawn_retry(hives);
    }
}

/// Repeat [`provision`] until it reaches the store, or [`RETRY_WINDOW`] passes.
fn spawn_retry(hives: Vec<String>) {
    tokio::spawn(async move {
        let reached = retry_until_ok(RETRY_INTERVAL, RETRY_ATTEMPTS, move || {
            let hives = hives.clone();
            async move {
                let outcome = provision(&hives).await;
                if let Err(Unreachable(reason)) = &outcome {
                    tracing::debug!(reason, "hive provisioning: store still unreachable");
                }
                outcome
            }
        })
        .await;
        if let Some(attempt) = reached {
            tracing::info!(attempt, "hive provisioning reached the store");
        } else {
            tracing::warn!(
                attempts = RETRY_ATTEMPTS,
                window = ?RETRY_WINDOW,
                "giving up on hive provisioning; no hive can log in to the store until this daemon is restarted"
            );
        }
    });
}

/// Run `attempt` every `interval` until it succeeds, at most `attempts` times.
/// Answers with the attempt that succeeded, or `None` if none did.
///
/// The cadence and the bound are parameters rather than the constants above so
/// that a test can drive this loop without waiting out a real day.
async fn retry_until_ok<F, Fut>(interval: Duration, attempts: u32, mut attempt: F) -> Option<u32>
where
    F: FnMut() -> Fut,
    Fut: Future<Output = Result<(), Unreachable>>,
{
    let mut ticker = tokio::time::interval(interval);
    // `interval`'s first tick is immediate, and that pass is the one the caller
    // already ran before deciding to retry at all.
    ticker.tick().await;
    for n in 1..=attempts {
        ticker.tick().await;
        if attempt().await.is_ok() {
            return Some(n);
        }
    }
    None
}

/// One pass: log in once, write every hive's policy, then every hive's role.
///
/// Policy first, because the role names the policy — the other order leaves a
/// window in which the role points at nothing.
///
/// # Errors
/// [`Unreachable`] for the states a later attempt can resolve: a store that is
/// not up, and an authority file that is not placed yet. A deployment that is
/// configured for neither is `Ok` — there is nothing to wait for.
async fn provision(hives: &[String]) -> Result<(), Unreachable> {
    if hives.is_empty() {
        return Ok(());
    }
    let store = match crate::store::connect().await {
        Ok(store) => store,
        // An unset `BAO_*` variable is a deployment that runs no store, not
        // one whose store is late: asking again changes nothing.
        Err(Error::MissingEnv(var)) => {
            tracing::info!(
                var,
                "no secret store configured; hive access is not managed here"
            );
            return Ok(());
        }
        Err(e) => return Err(Unreachable(e.to_string())),
    };
    for hive in hives {
        if let Err(e) = write_policy_for(&store, hive).await {
            tracing::warn!(hive, error = %format!("{e:#}"), "writing this hive's read policy failed");
        }
    }

    let Some(path) = std::env::var_os(ENV_HIVE_CLIENT_CA) else {
        tracing::info!("no hive certificate authority configured; hive roles are not managed here");
        return Ok(());
    };
    let path = path.to_string_lossy().into_owned();
    // Retryable rather than fatal: the unit that places this file and this
    // daemon start together, so "not there yet" is a moment, not a verdict.
    let ca = std::fs::read_to_string(&path)
        .map_err(|e| Unreachable(format!("reading the hive authority {path}: {e}")))?;
    for hive in hives {
        if let Err(e) = write_role(&store, hive, &ca).await {
            tracing::warn!(hive, error = %format!("{e:#}"), "creating this hive's store role failed");
        }
    }
    Ok(())
}

/// One hive's policy: its own name, its own document.
async fn write_policy_for(store: &SecretStore, hive: &str) -> Result<()> {
    let name = policy::hive_object_name(hive)?;
    store
        .write_policy(&name, &policy::render(hive)?)
        .await
        .with_context(|| format!("writing the read policy {name}"))?;
    tracing::info!(hive, policy = %name, "hive read policy in place");
    Ok(())
}

/// The role for one hive: its own certificate's subject, its own policy.
async fn write_role(store: &SecretStore, hive: &str, ca: &str) -> Result<()> {
    let name = policy::hive_object_name(hive)?;
    store
        // The subject is the hive's name: `glue-bao-tls.nix` mints a hive's
        // client leaf with its name as the CN, and cert auth matches on that.
        .write_cert_role(DEFAULT_CERT_MOUNT, &name, ca, hive, &name)
        .await
        .with_context(|| format!("writing the cert-auth role {name}"))?;
    tracing::info!(hive, role = %name, "hive store role in place");
    Ok(())
}

#[cfg(test)]
mod tests {
    use std::{cell::Cell, time::Duration};

    use super::{RETRY_ATTEMPTS, RETRY_INTERVAL, RETRY_WINDOW, Unreachable, retry_until_ok};

    #[test]
    fn the_retry_bound_is_the_window_it_claims() {
        assert_eq!(RETRY_INTERVAL * RETRY_ATTEMPTS, RETRY_WINDOW);
    }

    /// The property this issue is about: a pass that fails because the store is
    /// not up yet happens again.
    #[tokio::test]
    async fn a_pass_is_repeated_until_it_reaches_the_store() {
        let calls = Cell::new(0_u32);
        let reached = retry_until_ok(Duration::from_millis(1), 10, || {
            calls.set(calls.get() + 1);
            let still_down = calls.get() < 3;
            async move {
                if still_down {
                    Err(Unreachable("store not up".to_owned()))
                } else {
                    Ok(())
                }
            }
        })
        .await;
        assert_eq!(reached, Some(3));
        assert_eq!(calls.get(), 3, "the loop stops at the first success");
    }

    #[tokio::test]
    async fn the_loop_is_bounded_rather_than_endless() {
        let calls = Cell::new(0_u32);
        let reached = retry_until_ok(Duration::from_millis(1), 5, || {
            calls.set(calls.get() + 1);
            async { Err(Unreachable("store not up".to_owned())) }
        })
        .await;
        assert_eq!(reached, None, "a store that never appears is given up on");
        assert_eq!(calls.get(), 5, "one attempt per interval, and no more");
    }
}