//! Client for `swarm-authelia-bridge` — this daemon's only access to the
//! swarm's authelia users database, in either direction (see that crate's
//! README for why the file cannot be touched from here).
//!
//! Authenticated with THIS daemon's own queue OIDC identity
//! ([`crate::queue_identity`], the same one `swarm-queue-client` mints for
//! the queue connection) — "one identity per principal" means a second
//! op that needs to prove who this process is reuses the identity it
//! already has rather than provisioning a new one. A fresh token is
//! minted per call for the same reason the queue client mints one per
//! connection attempt: no window in which this process holds a token
//! that outlives its intended use.
//!
//! `None` when this deployment set no bridge address; the default is the
//! `/bridge/` location on the authelia vhost, from every host. Without one,
//! this daemon can neither create an identity nor read the roster.
//!
//! Each caller answers that absence in its own terms rather than this
//! module inventing a shared one: `SwarmNodeKind::CreateIdentity` fails
//! the job explicitly, and the roster endpoint refuses. Neither substitutes
//! an empty answer, which is the failure mode a default here would create.

use anyhow::{Context, Result};
use swarm_authelia_bridge_sock::{BridgeRequest, BridgeResponse};

/// Bound on reaching the bridge.
const HTTP_CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
/// Bound on one whole bridge call, body included. Ensuring an identity has
/// the bridge shell out to `authelia` for an argon2 hash before it answers,
/// so this is looser than a plain API call needs.
const HTTP_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);

/// A configured connection to `swarm-authelia-bridge`.
#[derive(Clone)]
pub struct AuthBridge {
    http: reqwest::Client,
    base_url: String,
    queue_cfg: swarm_queue_client::QueueConfig,
}

impl AuthBridge {
    /// Read `SWARM_CONTROLLER_AUTH_BRIDGE_URL`; `Ok(None)` when unset.
    ///
    /// The queue identity ([`crate::queue_identity`]) is not optional once
    /// the bridge URL is set: the nix module sets `queueEnv` unconditionally
    /// for every controller (the queue is required, not just co-located
    /// service), so a bridge URL with no queue identity to authenticate
    /// with is a deployment bug, not an absent-feature case — hence the
    /// hard error rather than a second `None`.
    pub fn from_env() -> Result<Option<Self>> {
        let Ok(base_url) = std::env::var("SWARM_CONTROLLER_AUTH_BRIDGE_URL") else {
            return Ok(None);
        };
        let queue_cfg = crate::queue_identity::get().cloned().ok_or_else(|| {
            anyhow::anyhow!(
                "SWARM_CONTROLLER_AUTH_BRIDGE_URL is set but SWARM_CONTROLLER_OIDC_* is \
                     not — the bridge is authenticated with this daemon's queue identity, so \
                     that identity must exist first"
            )
        })?;
        let http = reqwest::Client::builder()
            .connect_timeout(HTTP_CONNECT_TIMEOUT)
            .timeout(HTTP_TIMEOUT)
            .build()
            .context("building the swarm-authelia-bridge HTTP client")?;
        Ok(Some(Self {
            http,
            base_url,
            queue_cfg,
        }))
    }

    /// Idempotently ensure `name` exists as an authelia subject.
    pub async fn ensure_agent_identity(&self, name: &str) -> Result<BridgeResponse> {
        self.request(&BridgeRequest::EnsureAgentIdentity {
            name: name.to_owned(),
        })
        .await
    }

    /// The swarm's agent roster, straight from the identity store.
    ///
    /// Returns the names rather than the whole [`BridgeResponse`]: every
    /// other variant is a protocol error for this request, and a caller that
    /// had to match them would be free to treat one as an empty roster. An
    /// empty roster and a bridge that answered something else are different
    /// facts, and only one of them is a valid render.
    pub async fn list_agent_identities(&self) -> Result<Vec<String>> {
        match self.request(&BridgeRequest::ListAgentIdentities).await? {
            BridgeResponse::Agents { agents } => Ok(agents
                .into_iter()
                .map(|entry| entry.name.into_string())
                .collect()),
            other => {
                anyhow::bail!("swarm-authelia-bridge answered a roster request with {other:?}")
            }
        }
    }

    /// The usernames in authelia's operator group. The bridge answers this
    /// to this daemon's client only.
    ///
    /// Names only, for [`Self::list_agent_identities`]'s reason.
    pub async fn list_operators(&self) -> Result<Vec<String>> {
        match self.request(&BridgeRequest::ListOperators).await? {
            BridgeResponse::Operators { operators } => Ok(operators),
            other => {
                anyhow::bail!(
                    "swarm-authelia-bridge answered an operator-list request with {other:?}"
                )
            }
        }
    }

    /// One authenticated round-trip to the bridge.
    ///
    /// A fresh token per call, deliberately — see the module doc. Shared by
    /// every operation so the auth and the error handling cannot drift
    /// between them; a second copy is how one path ends up treating a 500 as
    /// an answer.
    async fn request(&self, req: &BridgeRequest) -> Result<BridgeResponse> {
        // `mint_token_for` builds its own HTTP client (trusting the queue's
        // configured CA, if any) — deliberately not `self.http`, which is
        // the bridge's own client and has nothing to do with authelia's
        // token endpoint's trust anchors.
        let token = swarm_queue_client::mint_token_for(&self.queue_cfg, None, None)
            .await
            .context("minting a bearer token for swarm-authelia-bridge")?;

        post_with_retry(&self.http, &self.base_url, &token, req, RETRY_BACKOFF).await
    }
}

/// Waits between attempts at one bridge call: three tries in all.
const RETRY_BACKOFF: &[std::time::Duration] = &[
    std::time::Duration::from_secs(1),
    std::time::Duration::from_secs(4),
];

/// POST `req` to the bridge, retrying a `503` or a failure to connect after
/// each wait in `backoff`.
///
/// `503` is the bridge saying it could not get an answer out of authelia, so
/// the token was never evaluated; the same call usually succeeds once
/// authelia recovers. `401` and `403` are verdicts on the token and are
/// final, as is any other status. A timeout after the connection was made
/// is not retried: the request may have been acted on, and a whole attempt
/// already took [`HTTP_TIMEOUT`]. Every bridge request is safe to repeat
/// regardless (idempotent ensure, or a read).
async fn post_with_retry<T: serde::de::DeserializeOwned>(
    http: &reqwest::Client,
    base_url: &str,
    token: &str,
    req: &BridgeRequest,
    backoff: &[std::time::Duration],
) -> Result<T> {
    let mut waits = backoff.iter();
    loop {
        let sent = http
            .post(format!("{base_url}/requests"))
            .bearer_auth(token)
            .json(req)
            .send()
            .await;
        let (failure, retryable) = match sent {
            Ok(response) => {
                let status = response.status();
                let body = response.text().await.unwrap_or_default();
                if status.is_success() {
                    return serde_json::from_str(&body)
                        .context("parsing swarm-authelia-bridge's response");
                }
                (
                    anyhow::anyhow!("swarm-authelia-bridge refused the request ({status}): {body}"),
                    status == reqwest::StatusCode::SERVICE_UNAVAILABLE,
                )
            }
            Err(e) => {
                let retryable = e.is_connect();
                let what = if e.is_connect() && e.is_timeout() {
                    format!(
                        "calling swarm-authelia-bridge: connect timed out after {HTTP_CONNECT_TIMEOUT:?}"
                    )
                } else if e.is_timeout() {
                    format!("calling swarm-authelia-bridge: timed out after {HTTP_TIMEOUT:?}")
                } else {
                    "calling swarm-authelia-bridge".to_owned()
                };
                (anyhow::Error::new(e).context(what), retryable)
            }
        };
        match waits.next() {
            Some(wait) if retryable => {
                tracing::warn!(error = %failure, ?wait, "swarm-authelia-bridge call failed; retrying");
                tokio::time::sleep(*wait).await;
            }
            _ => return Err(failure),
        }
    }
}

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

    use std::sync::Arc;
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::time::Duration;

    const NO_WAIT: &[Duration] = &[Duration::ZERO, Duration::ZERO];

    /// A bridge on loopback answering each request with the next entry of
    /// `statuses` (the last one repeats), and counting the requests.
    async fn bridge(statuses: &'static [u16]) -> (String, Arc<AtomicUsize>) {
        use axum::http::StatusCode;
        let hits = Arc::new(AtomicUsize::new(0));
        let counter = hits.clone();
        let app = axum::Router::new().route(
            "/requests",
            axum::routing::post(move || {
                let n = counter.fetch_add(1, Ordering::SeqCst);
                let code = statuses[n.min(statuses.len() - 1)];
                async move {
                    let body = if code == 200 {
                        r#"{"status":"operators","operators":["mara"]}"#
                    } else {
                        r#"{"status":"error","message":"refused"}"#
                    };
                    (StatusCode::from_u16(code).unwrap(), body)
                }
            }),
        );
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let url = format!("http://{}", listener.local_addr().unwrap());
        tokio::spawn(async move { axum::serve(listener, app).await.unwrap() });
        (url, hits)
    }

    async fn call(url: &str, backoff: &[Duration]) -> Result<BridgeResponse> {
        post_with_retry(
            &reqwest::Client::new(),
            url,
            "tok",
            &BridgeRequest::ListOperators,
            backoff,
        )
        .await
    }

    #[tokio::test]
    async fn a_503_is_retried_until_the_bridge_answers() {
        let (url, hits) = bridge(&[503, 503, 200]).await;
        let got = call(&url, NO_WAIT).await.expect("third try succeeds");
        assert!(matches!(got, BridgeResponse::Operators { .. }), "{got:?}");
        assert_eq!(hits.load(Ordering::SeqCst), 3);
    }

    #[tokio::test]
    async fn a_persistent_503_fails_after_three_tries() {
        let (url, hits) = bridge(&[503]).await;
        let err = call(&url, NO_WAIT).await.expect_err("never recovers");
        assert!(format!("{err:#}").contains("503"), "{err:#}");
        assert_eq!(hits.load(Ordering::SeqCst), 3);
    }

    /// 401 and 403 are verdicts on the token: one try each.
    #[tokio::test]
    async fn a_401_or_403_is_final() {
        for code in [401, 403] {
            let (url, hits) = bridge(match code {
                401 => &[401],
                _ => &[403],
            })
            .await;
            let err = call(&url, NO_WAIT).await.expect_err("refused");
            assert!(format!("{err:#}").contains(&code.to_string()), "{err:#}");
            assert_eq!(hits.load(Ordering::SeqCst), 1, "status {code}");
        }
    }

    #[tokio::test]
    async fn a_connection_refusal_is_retried_then_fails() {
        // Bind then drop, so the port refuses connections.
        let port = {
            let l = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
            l.local_addr().unwrap().port()
        };
        let started = tokio::time::Instant::now();
        let err = call(
            &format!("http://127.0.0.1:{port}"),
            &[Duration::from_millis(150); 2],
        )
        .await
        .expect_err("nothing listens");
        assert!(
            format!("{err:#}").contains("calling swarm-authelia-bridge"),
            "{err:#}"
        );
        assert!(
            started.elapsed() >= Duration::from_millis(300),
            "both waits were taken: {:?}",
            started.elapsed()
        );
    }

    /// The common case: no bridge wired up for this deployment.
    #[test]
    fn absent_url_is_not_an_error() {
        assert!(std::env::var("SWARM_CONTROLLER_AUTH_BRIDGE_URL").is_err());
        assert!(
            AuthBridge::from_env()
                .expect("absent is not an error")
                .is_none()
        );
    }
}