//! This responder's own OIDC client secret, the credential it introspects
//! with. Read from the secret store as the responder starts and held in
//! memory, never in a file, `argv` or the environment.
//!
//! The secret can change under a running responder: authelia's host
//! re-publishes it when it is re-minted. So when the `IdP` refuses an
//! introspection (a non-2xx), the secret is read from the store again and,
//! when it changed, the introspection is tried once more. Both fit in the one
//! request's [`INTROSPECTION_TIMEOUT`](crate::introspect::INTROSPECTION_TIMEOUT).
//! A re-read that does not is left running: that request is denied, and the
//! next one introspects with what the re-read landed. One re-read runs at a
//! time; a refusal while one is running waits for it rather than starting
//! another.

use std::future::Future;
use std::sync::{Arc, Mutex, RwLock};
use std::time::Duration;

use anyhow::{Context, Result};
use futures_util::FutureExt;
use futures_util::future::{BoxFuture, Shared};
use swarm_secret_client::client::{DEFAULT_CERT_MOUNT, Settings};
use swarm_secret_client::path::{Kind, principal_prefix};
use tokio::task::AbortHandle;
use tokio::time::Instant;

use crate::introspect::Answer;
use crate::keystore::KeyStore;

/// The path under the store's KV mount holding `client_id`'s secret.
/// `swarm-secret-publisher.nix` writes it, and `natsAuthReaders` in
/// `swarm-bao.nix` grants this responder's role the read.
///
/// # Errors
/// When `client_id` is not a single path segment.
pub fn path(client_id: &str) -> Result<String> {
    let prefix = principal_prefix(Kind::Service, client_id)?;
    Ok(format!("{prefix}/oidc/client"))
}

/// The secret at `path`, with surrounding whitespace trimmed: the publisher
/// stores the minted file as is, and a trailing newline is not part of the
/// secret. Errors name the path, never the value.
///
/// # Errors
/// When the store cannot be read, or holds nothing or only whitespace there.
pub async fn read(store: &impl KeyStore, path: &str) -> Result<String> {
    let stored = store.read(path).await?.with_context(|| {
        format!("nothing is stored at {path}; swarm-secret-publish on authelia's host writes it")
    })?;
    let trimmed = stored.trim();
    if trimmed.is_empty() {
        anyhow::bail!("{path} holds an empty secret");
    }
    Ok(trimmed.to_owned())
}

/// Where a re-read of the secret comes from.
pub trait Source: Send + Sync + 'static {
    /// The secret as the store holds it now.
    fn read(&self) -> impl Future<Output = Result<String>> + Send;
}

/// The secret store, logged in to per read: a re-read can come long after
/// the start's token has expired.
pub struct Store {
    settings: Settings,
    cert_role: String,
    path: String,
}

impl Store {
    /// The secret at `path`, read as `cert_role` from the store at `settings`.
    pub fn new(settings: Settings, cert_role: String, path: String) -> Self {
        Self {
            settings,
            cert_role,
            path,
        }
    }
}

impl Source for Store {
    async fn read(&self) -> Result<String> {
        let store = swarm_secret_client::SecretStore::connect(
            &self.settings,
            &self.cert_role,
            DEFAULT_CERT_MOUNT,
        )
        .await
        .context("logging in to the secret store")?;
        read(&store, &self.path).await
    }
}

/// The longest one re-read runs. The store client sets no timeout of its
/// own, and while a re-read runs every refusal waits on it instead of
/// starting another.
const REREAD_TIMEOUT: Duration = Duration::from_secs(10);

/// A re-read in progress, shared by every request waiting on it. It yields
/// whether the store was read; a failed or timed-out read logs for itself.
type Reread = Shared<BoxFuture<'static, bool>>;

/// The secret in use, and the re-read that replaces it. No `Debug`: it holds
/// the secret.
pub struct OidcSecret<S> {
    source: Arc<S>,
    current: Arc<RwLock<String>>,
    reread: Mutex<Option<(AbortHandle, Reread)>>,
}

impl<S: Source> OidcSecret<S> {
    /// `initial`, re-read from `source` when the `IdP` refuses it.
    pub fn new(source: S, initial: String) -> Self {
        Self {
            source: Arc::new(source),
            current: Arc::new(RwLock::new(initial)),
            reread: Mutex::new(None),
        }
    }

    fn current(&self) -> String {
        self.current.read().expect("unpoisoned").clone()
    }

    /// The caller `attempt` names, introspecting with the current secret.
    /// When the `IdP` refuses that, the secret is re-read and, if it changed
    /// and `deadline` has not passed, `attempt` runs once more with the new
    /// one. `attempt` bounds itself by the same `deadline`.
    ///
    /// # Errors
    /// When an attempt could not be made at all; the caller denies.
    pub async fn introspect<F, Fut>(&self, deadline: Instant, attempt: F) -> Result<Option<String>>
    where
        F: Fn(String) -> Fut,
        Fut: Future<Output = Result<Answer>>,
    {
        let used = self.current();
        if let Answer::Caller(caller) = attempt(used.clone()).await? {
            return Ok(caller);
        }
        match tokio::time::timeout_at(deadline, self.reread()).await {
            Err(_) => {
                tracing::warn!(
                    "re-reading the OIDC client secret did not finish inside the introspection budget; \
                     denying, and the next request introspects with what it reads"
                );
                return Ok(None);
            }
            Ok(false) => return Ok(None),
            Ok(true) => {}
        }
        let fresh = self.current();
        if fresh == used {
            tracing::warn!("the stored OIDC client secret is the one the IdP refused; denying");
            return Ok(None);
        }
        tracing::info!("re-read a changed OIDC client secret; introspecting again");
        match attempt(fresh).await? {
            Answer::Caller(caller) => Ok(caller),
            Answer::Refused => Ok(None),
        }
    }

    /// The running re-read, or a new one when none is running. Spawned, so it
    /// finishes even when every request waiting on it has given up.
    fn reread(&self) -> Reread {
        let mut slot = self.reread.lock().expect("unpoisoned");
        if let Some((handle, running)) = slot.as_ref()
            && !handle.is_finished()
        {
            return running.clone();
        }
        let source = Arc::clone(&self.source);
        let current = Arc::clone(&self.current);
        let task = tokio::spawn(async move {
            match tokio::time::timeout(REREAD_TIMEOUT, source.read()).await {
                Ok(Ok(secret)) => {
                    *current.write().expect("unpoisoned") = secret;
                    true
                }
                Ok(Err(e)) => {
                    tracing::warn!(
                        error = format!("{e:#}"),
                        "could not re-read the OIDC client secret; keeping the one in use"
                    );
                    false
                }
                Err(_) => {
                    tracing::warn!(
                        timeout = ?REREAD_TIMEOUT,
                        "re-reading the OIDC client secret timed out; keeping the one in use"
                    );
                    false
                }
            }
        });
        let handle = task.abort_handle();
        let shared = task
            .map(|read| {
                read.unwrap_or_else(|e| {
                    tracing::warn!(
                        error = %e,
                        "the OIDC client secret re-read task died; keeping the one in use"
                    );
                    false
                })
            })
            .boxed()
            .shared();
        *slot = Some((handle, shared.clone()));
        shared
    }
}

#[cfg(test)]
mod tests {
    use std::sync::atomic::{AtomicUsize, Ordering};
    use std::time::Duration;

    use super::*;
    use crate::introspect::INTROSPECTION_TIMEOUT;
    use crate::keystore::tests::Fake as FakeKeys;

    /// A store holding `secret`, or failing, that takes `delay` to answer.
    struct Fake {
        secret: Option<&'static str>,
        delay: Duration,
        reads: Arc<AtomicUsize>,
    }

    impl Fake {
        fn holding(secret: &'static str) -> Self {
            Self {
                secret: Some(secret),
                delay: Duration::ZERO,
                reads: Arc::default(),
            }
        }
    }

    impl Source for Fake {
        async fn read(&self) -> Result<String> {
            self.reads.fetch_add(1, Ordering::SeqCst);
            tokio::time::sleep(self.delay).await;
            self.secret.map(str::to_owned).context("store unreachable")
        }
    }

    /// An `IdP` that admits as `hive-alpha` only when introspected with
    /// `registered`, and refuses every other secret, counting its calls.
    fn idp<'a>(
        registered: &'static str,
        calls: &'a AtomicUsize,
    ) -> impl Fn(String) -> std::future::Ready<Result<Answer>> + 'a {
        move |secret| {
            calls.fetch_add(1, Ordering::SeqCst);
            std::future::ready(Ok(if secret == registered {
                Answer::Caller(Some("hive-alpha".to_owned()))
            } else {
                Answer::Refused
            }))
        }
    }

    fn budget() -> Instant {
        Instant::now() + INTROSPECTION_TIMEOUT
    }

    #[test]
    fn the_path_is_the_one_the_publisher_writes() {
        assert_eq!(
            path("swarm-nats").expect("path"),
            "swarm/services/swarm-nats/oidc/client"
        );
        assert!(path("../swarm-grafana").is_err());
    }

    #[tokio::test]
    async fn the_stored_secret_is_trimmed_and_an_empty_or_absent_one_refused() {
        let at = "swarm/services/swarm-nats/oidc/client";
        let keys = FakeKeys::holding(&[(at, "s3cret\n")]);
        assert_eq!(read(&keys, at).await.expect("stored"), "s3cret");
        keys.put(at, " \n");
        assert!(read(&keys, at).await.is_err());
        assert!(read(&FakeKeys::default(), at).await.is_err());
    }

    #[tokio::test]
    async fn an_answered_introspection_reads_nothing() {
        let store = Fake::holding("new");
        let reads = Arc::clone(&store.reads);
        let secret = OidcSecret::new(store, "old".to_owned());
        let calls = AtomicUsize::new(0);
        let caller = secret
            .introspect(budget(), idp("old", &calls))
            .await
            .expect("answered");
        assert_eq!(caller.as_deref(), Some("hive-alpha"));
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        assert_eq!(reads.load(Ordering::SeqCst), 0);
    }

    /// The rotation the re-read exists for: the `IdP` refuses the secret held
    /// since start, the store holds the new one, and the same request is
    /// admitted with it.
    #[tokio::test]
    async fn a_refused_secret_is_re_read_and_the_request_retried_once() {
        let store = Fake::holding("new");
        let reads = Arc::clone(&store.reads);
        let secret = OidcSecret::new(store, "old".to_owned());
        let calls = AtomicUsize::new(0);
        let caller = secret
            .introspect(budget(), idp("new", &calls))
            .await
            .expect("answered");
        assert_eq!(caller.as_deref(), Some("hive-alpha"));
        assert_eq!(calls.load(Ordering::SeqCst), 2);
        assert_eq!(reads.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn a_refusal_after_the_re_read_is_a_denial_with_no_third_attempt() {
        let store = Fake::holding("newer");
        let reads = Arc::clone(&store.reads);
        let secret = OidcSecret::new(store, "old".to_owned());
        let calls = AtomicUsize::new(0);
        let caller = secret
            .introspect(budget(), idp("unknown", &calls))
            .await
            .expect("answered");
        assert_eq!(caller, None);
        assert_eq!(calls.load(Ordering::SeqCst), 2);
        assert_eq!(reads.load(Ordering::SeqCst), 1);
    }

    #[tokio::test]
    async fn an_unchanged_secret_is_not_retried() {
        let secret = OidcSecret::new(Fake::holding("old"), "old".to_owned());
        let calls = AtomicUsize::new(0);
        let caller = secret
            .introspect(budget(), idp("unknown", &calls))
            .await
            .expect("answered");
        assert_eq!(caller, None);
        assert_eq!(calls.load(Ordering::SeqCst), 1);
    }

    thread_local! {
        static LOG: std::cell::RefCell<Vec<u8>> = const { std::cell::RefCell::new(Vec::new()) };
    }

    /// What `tracing` records on the calling thread. The subscriber is
    /// process-wide, because a scoped one loses to the callsite cache when
    /// another test reaches the same event first; the buffer is per thread,
    /// which is where a current-thread test runtime runs the spawned re-read.
    struct Logs;

    impl std::io::Write for Logs {
        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
            LOG.with(|log| log.borrow_mut().extend_from_slice(buf));
            Ok(buf.len())
        }

        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }

    impl Logs {
        /// Starts this thread's capture empty.
        fn capture() {
            static INSTALLED: std::sync::Once = std::sync::Once::new();
            INSTALLED.call_once(|| {
                tracing_subscriber::fmt()
                    .with_writer(|| Logs)
                    .with_ansi(false)
                    .init();
            });
            LOG.with(|log| log.borrow_mut().clear());
        }

        fn text() -> String {
            LOG.with(|log| String::from_utf8(log.borrow().clone()).expect("utf-8"))
        }
    }

    #[tokio::test]
    async fn a_failed_re_read_denies_and_keeps_the_secret_in_use() {
        Logs::capture();
        let store = Fake {
            secret: None,
            delay: Duration::ZERO,
            reads: Arc::default(),
        };
        let secret = OidcSecret::new(store, "old".to_owned());
        let calls = AtomicUsize::new(0);
        let caller = secret
            .introspect(budget(), idp("new", &calls))
            .await
            .expect("answered");
        assert_eq!(caller, None);
        assert_eq!(calls.load(Ordering::SeqCst), 1);
        assert_eq!(secret.current(), "old");

        let text = Logs::text();
        assert!(text.contains("could not re-read the OIDC client secret"));
        assert!(!text.contains("is the one the IdP refused"), "{text}");
        assert!(!text.contains("old"), "no secret in the log: {text}");
    }

    /// The store was read and holds what the `IdP` already refused: the one
    /// case that says so.
    #[tokio::test]
    async fn the_unchanged_secret_is_logged_as_the_one_the_idp_refused() {
        Logs::capture();
        let secret = OidcSecret::new(Fake::holding("old"), "old".to_owned());
        let calls = AtomicUsize::new(0);
        let caller = secret
            .introspect(budget(), idp("new", &calls))
            .await
            .expect("answered");
        assert_eq!(caller, None);

        let text = Logs::text();
        assert!(text.contains("is the one the IdP refused"), "{text}");
        assert!(!text.contains("could not re-read"), "{text}");
    }

    /// A store slower than the budget: the request is denied at its
    /// deadline, the re-read finishes behind it, and the next request is
    /// admitted on its first attempt without reading again.
    #[tokio::test(start_paused = true)]
    async fn a_re_read_past_the_budget_serves_the_next_request() {
        let store = Fake {
            secret: Some("new"),
            delay: Duration::from_secs(3),
            reads: Arc::default(),
        };
        let reads = Arc::clone(&store.reads);
        let secret = OidcSecret::new(store, "old".to_owned());
        let calls = AtomicUsize::new(0);

        let started = Instant::now();
        let deadline = started + INTROSPECTION_TIMEOUT;
        let first = secret
            .introspect(deadline, idp("new", &calls))
            .await
            .expect("answered");
        assert_eq!(first, None);
        assert_eq!(Instant::now(), deadline);
        assert_eq!(calls.load(Ordering::SeqCst), 1);

        tokio::time::sleep(Duration::from_secs(2)).await;
        let next = secret
            .introspect(budget(), idp("new", &calls))
            .await
            .expect("answered");
        assert_eq!(next.as_deref(), Some("hive-alpha"));
        assert_eq!(calls.load(Ordering::SeqCst), 2);
        assert_eq!(reads.load(Ordering::SeqCst), 1);
    }

    /// A refusal while a re-read is still running waits on that one, and is
    /// admitted when it lands inside this request's budget.
    #[tokio::test(start_paused = true)]
    async fn a_refusal_during_a_running_re_read_joins_it() {
        let store = Fake {
            secret: Some("new"),
            delay: Duration::from_secs(2),
            reads: Arc::default(),
        };
        let reads = Arc::clone(&store.reads);
        let secret = OidcSecret::new(store, "old".to_owned());
        let calls = AtomicUsize::new(0);

        let first = secret
            .introspect(budget(), idp("new", &calls))
            .await
            .expect("answered");
        assert_eq!(first, None);

        let second = secret
            .introspect(budget(), idp("new", &calls))
            .await
            .expect("answered");
        assert_eq!(second.as_deref(), Some("hive-alpha"));
        assert_eq!(calls.load(Ordering::SeqCst), 3);
        assert_eq!(reads.load(Ordering::SeqCst), 1);
    }

    /// A re-read that stalls is given up at [`REREAD_TIMEOUT`]: the secret in
    /// use stays, and the next refusal starts a fresh re-read rather than
    /// waiting on the stalled one.
    #[tokio::test(start_paused = true)]
    async fn a_stalled_re_read_is_dropped_at_its_timeout() {
        let store = Fake {
            secret: Some("new"),
            delay: REREAD_TIMEOUT * 6,
            reads: Arc::default(),
        };
        let reads = Arc::clone(&store.reads);
        let secret = OidcSecret::new(store, "old".to_owned());
        let calls = AtomicUsize::new(0);

        let first = secret
            .introspect(budget(), idp("new", &calls))
            .await
            .expect("answered");
        assert_eq!(first, None);

        tokio::time::sleep(REREAD_TIMEOUT).await;
        assert_eq!(secret.current(), "old");
        let second = secret
            .introspect(budget(), idp("new", &calls))
            .await
            .expect("answered");
        assert_eq!(second, None);
        assert_eq!(reads.load(Ordering::SeqCst), 2);
    }

    /// An `IdP` on a loopback port that admits as `hive-alpha` a request
    /// whose basic auth is `swarm-nats:{registered}`, and answers any other
    /// with a 401.
    async fn stub_idp(registered: &str) -> String {
        use axum::http::{HeaderMap, StatusCode, header::AUTHORIZATION};

        let expected = format!(
            "Basic {}",
            data_encoding::BASE64.encode(format!("swarm-nats:{registered}").as_bytes())
        );
        let app = axum::Router::new().fallback(move |headers: HeaderMap| async move {
            if headers.get(AUTHORIZATION).and_then(|v| v.to_str().ok()) == Some(&*expected) {
                (
                    StatusCode::OK,
                    r#"{"active":true,"client_id":"hive-alpha"}"#,
                )
            } else {
                (StatusCode::UNAUTHORIZED, "")
            }
        });
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let url = format!("http://{}/introspect", listener.local_addr().expect("addr"));
        tokio::spawn(async move { axum::serve(listener, app).await });
        url
    }

    /// The `IdP`'s 401 to the secret held since start is what starts the
    /// re-read: driven through the real HTTP introspection, not an [`Answer`]
    /// handed in.
    #[tokio::test]
    async fn an_idp_401_re_reads_the_secret_and_admits_with_the_new_one() {
        let url = stub_idp("new").await;
        let http = reqwest::Client::builder()
            .no_proxy()
            .build()
            .expect("client");
        let store = Fake::holding("new");
        let reads = Arc::clone(&store.reads);
        let secret = OidcSecret::new(store, "old".to_owned());
        let deadline = budget();
        let caller = secret
            .introspect(deadline, |own| {
                let (http, url) = (&http, &url);
                async move {
                    crate::introspect::identify_caller(
                        http,
                        url,
                        "swarm-nats",
                        &own,
                        "a-token",
                        deadline,
                    )
                    .await
                }
            })
            .await
            .expect("answered");
        assert_eq!(caller.as_deref(), Some("hive-alpha"));
        assert_eq!(reads.load(Ordering::SeqCst), 1);
    }
}