//! An [`opentelemetry_http::HttpClient`] that authenticates every request
//! with a fresh, audience-scoped bearer token minted from this daemon's own
//! [`swarm_queue_client::QueueConfig`] identity — "one identity per
//! principal" applied to a third destination (the queue connection and
//! `auth.rs`'s bridge client, `swarm_queue_client::mint_token_for`, are the
//! other two).
//!
//! Lives here, not in `swarm-queue-client` (mara's own call during review:
//! it does not belong in the queue-connect crate) — this daemon is the only
//! caller, and the module only *reuses* that crate's
//! `QueueConfig`/`mint_token_for_blocking`, it doesn't extend the
//! queue-connect contract those exist for.
//!
//! Exists for exactly one shape of caller: an OTLP metric exporter whose
//! `PeriodicReader` drives export from a background thread with **no tokio
//! reactor** (`opentelemetry-otlp`'s `reqwest-blocking-client` feature is
//! chosen everywhere in this workspace for that reason — see
//! `vcs_metrics`'s module doc). That constraint is why this mints via
//! [`swarm_queue_client::mint_token_for_blocking`] rather than the async
//! `swarm_queue_client::mint_token_for`: the latter needs a runtime to
//! `.await` on, which the calling thread does not have.
//!
//! Per-request, not per-tick or cached: `opentelemetry-otlp` 0.32 exposes
//! `WithHttpConfig::with_http_client`, a seam for a caller-supplied
//! [`opentelemetry_http::HttpClient`] — strictly better than minting once
//! and setting a static header (`WithHttpConfig::with_headers`), because a
//! tick that runs long, retries, or fires late can outlive a token minted
//! before it started, and that failure is intermittent — the worst kind to
//! debug through a metrics pipeline that is itself the thing being fixed.

use std::fmt;
use std::sync::{Mutex, PoisonError};
use std::time::{Duration, Instant};

use async_trait::async_trait;
use bytes::Bytes;
use http::{Request, Response};
use opentelemetry_http::{HttpClient, HttpError};
use swarm_queue_client::{QueueConfig, chain, mint_token_for_blocking};

/// How long a failure of the same kind stays quiet after it was logged.
/// Every push in between is counted and the count rides on the next line.
const REPEAT_INTERVAL: Duration = Duration::from_mins(10);

/// The most of a refusal's response body a log line carries. The collector
/// and authelia both answer with a one-line JSON error, well under this.
const BODY_SNIPPET_CHARS: usize = 200;

/// See the module doc. Wraps a plain `reqwest::blocking::Client` — deliberately
/// not `opentelemetry_http`'s own blanket impl for that type (behind its
/// `reqwest-blocking` cargo feature, unused here): that impl has no seam to
/// mint a token per call, which is this type's entire reason to exist.
#[derive(Debug)]
pub struct AuthenticatedHttpClient {
    inner: reqwest::blocking::Client,
    cfg: QueueConfig,
    /// The audience requested at the token endpoint — see
    /// [`swarm_queue_client::mint_token_for_blocking`]'s own doc (via
    /// `swarm_queue_client::mint_token_for`'s) for why this has to be asked
    /// for explicitly rather than left to whatever the identity's default is.
    audience: String,
    failures: Mutex<FailureLog>,
}

impl AuthenticatedHttpClient {
    /// `inner` is the caller's own client (already configured with
    /// whatever timeout/TLS trust the destination needs) — this type only
    /// adds the auth header, it does not build a client of its own, so a
    /// caller reusing an existing `reqwest::blocking::Client` doesn't pay
    /// for a second connection pool.
    #[must_use]
    pub fn new(
        inner: reqwest::blocking::Client,
        cfg: QueueConfig,
        audience: impl Into<String>,
    ) -> Self {
        Self {
            inner,
            cfg,
            audience: audience.into(),
            failures: Mutex::default(),
        }
    }

    /// One push: mint, then post. A failure carries what to log beside the
    /// error the SDK gets back.
    fn send(&self, request: Request<Bytes>) -> Result<Response<Bytes>, Failed> {
        let token = mint_token_for_blocking(&self.cfg, Some(&self.audience), None)
            .map_err(|e| Failed::new(Failure::Token, chain(&e), e))?;
        self.post(&token, request)
    }

    /// The HTTP half of [`Self::send`], with the token already minted.
    fn post(&self, token: &str, mut request: Request<Bytes>) -> Result<Response<Bytes>, Failed> {
        let header = http::HeaderValue::from_str(&format!("Bearer {token}"))
            .map_err(|e| Failed::new(Failure::Token, e.to_string(), e))?;
        request
            .headers_mut()
            .insert(http::header::AUTHORIZATION, header);

        let req = request
            .try_into()
            .map_err(|e: reqwest::Error| Failed::new(Failure::Transport, chain(&e), e))?;
        let mut response = self
            .inner
            .execute(req)
            .map_err(|e| Failed::new(Failure::Transport, chain(&e), e))?;
        let status = response.status();
        // Same header/body reassembly `opentelemetry_http`'s own blocking
        // impl does — `reqwest::blocking::Response` and `http::Response`
        // are different types, so this is the required conversion, not an
        // extra step invented here.
        let headers = std::mem::take(response.headers_mut());
        let body = response
            .bytes()
            .map_err(|e| Failed::new(Failure::Transport, chain(&e), e))?;
        if !status.is_success() {
            let detail = snippet(&body);
            let error = format!("the collector answered {status}: {detail}");
            return Err(Failed::new(Failure::Status(status.as_u16()), detail, error));
        }
        let mut http_response = Response::builder()
            .status(status)
            .body(body)
            .map_err(|e| Failed::new(Failure::Transport, e.to_string(), e))?;
        *http_response.headers_mut() = headers;
        Ok(http_response)
    }

    /// The SDK reports a failed export only at debug level, so this is the
    /// one place a refused push shows up: a warning per kind of failure,
    /// repeated every [`REPEAT_INTERVAL`] while it persists, and one info
    /// line when a push succeeds after failing.
    fn log(&self, outcome: Result<(), &Failed>) {
        let mut failures = self.failures.lock().unwrap_or_else(PoisonError::into_inner);
        match outcome {
            Ok(()) => {
                if failures.succeeded() == Verdict::Recovered {
                    tracing::info!(
                        client_id = %self.cfg.client_id,
                        "otel export: push to the collector succeeded after failing"
                    );
                }
            }
            Err(failed) => {
                if let Verdict::Warn { suppressed } = failures.failed(failed.kind, Instant::now()) {
                    tracing::warn!(
                        failure = %failed.kind,
                        detail = %failed.detail,
                        suppressed,
                        client_id = %self.cfg.client_id,
                        audience = %self.audience,
                        "otel export: push to the collector failed"
                    );
                }
            }
        }
    }
}

#[async_trait]
impl HttpClient for AuthenticatedHttpClient {
    /// `async fn` in signature, synchronous in body — matching
    /// `opentelemetry_http`'s own `reqwest::blocking::Client` impl exactly
    /// (see that impl's source): the trait is async because some callers
    /// need that, not because this one does, and the module doc explains
    /// why this specific caller deliberately runs on a thread with nothing
    /// to yield to.
    async fn send_bytes(&self, request: Request<Bytes>) -> Result<Response<Bytes>, HttpError> {
        let outcome = self.send(request);
        self.log(outcome.as_ref().map(|_| ()));
        outcome.map_err(|failed| failed.error)
    }
}

/// What stopped one push. Coarse on purpose: a collector down for an hour
/// is one kind of failure, logged once, not sixty.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
enum Failure {
    /// No token: authelia refused the request or could not be reached.
    Token,
    /// No answer from the collector: DNS, connect, TLS, or a cut-off body.
    Transport,
    /// The collector, or the gateway in front of it, answered non-2xx.
    Status(u16),
}

impl fmt::Display for Failure {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            Self::Token => f.write_str("token"),
            Self::Transport => f.write_str("transport"),
            Self::Status(code) => write!(f, "status {code}"),
        }
    }
}

struct Failed {
    kind: Failure,
    detail: String,
    error: HttpError,
}

impl Failed {
    fn new(kind: Failure, detail: String, error: impl Into<HttpError>) -> Self {
        Self {
            kind,
            detail,
            error: error.into(),
        }
    }
}

/// Whether a push outcome is worth a log line.
#[derive(Debug, PartialEq, Eq)]
enum Verdict {
    Quiet,
    /// Log this failure; `suppressed` pushes failed the same way unlogged
    /// since the last line about it.
    Warn {
        suppressed: u64,
    },
    /// The first success after a logged failure.
    Recovered,
}

/// The failure streak in progress, if any.
#[derive(Debug, Default)]
struct FailureLog {
    streak: Option<Streak>,
}

#[derive(Debug)]
struct Streak {
    kind: Failure,
    logged_at: Instant,
    suppressed: u64,
}

impl FailureLog {
    fn failed(&mut self, kind: Failure, now: Instant) -> Verdict {
        match &mut self.streak {
            Some(streak) if streak.kind == kind => {
                if now.duration_since(streak.logged_at) < REPEAT_INTERVAL {
                    streak.suppressed += 1;
                    return Verdict::Quiet;
                }
                streak.logged_at = now;
                Verdict::Warn {
                    suppressed: std::mem::take(&mut streak.suppressed),
                }
            }
            _ => {
                self.streak = Some(Streak {
                    kind,
                    logged_at: now,
                    suppressed: 0,
                });
                Verdict::Warn { suppressed: 0 }
            }
        }
    }

    fn succeeded(&mut self) -> Verdict {
        if self.streak.take().is_some() {
            Verdict::Recovered
        } else {
            Verdict::Quiet
        }
    }
}

/// The start of a response body as one log-safe line.
fn snippet(body: &[u8]) -> String {
    let text = String::from_utf8_lossy(body);
    let text = text.trim();
    let mut out: String = text
        .chars()
        .take(BODY_SNIPPET_CHARS)
        .map(|c| if c.is_control() { ' ' } else { c })
        .collect();
    if text.chars().count() > BODY_SNIPPET_CHARS {
        out.push('…');
    }
    out
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::path::PathBuf;
    use swarm_queue_client::ClientSecret;

    fn client() -> AuthenticatedHttpClient {
        let inner = reqwest::blocking::Client::builder()
            .no_proxy()
            .build()
            .expect("client");
        let cfg = QueueConfig {
            url: "nats://127.0.0.1:4222".to_owned(),
            token_endpoint: "http://127.0.0.1:9/token".to_owned(),
            client_id: "swarm-controller".to_owned(),
            client_secret: ClientSecret::File(PathBuf::from("/nonexistent")),
            ca_file: None,
        };
        AuthenticatedHttpClient::new(inner, cfg, "swarm-controller")
    }

    fn request(url: &str) -> Request<Bytes> {
        Request::post(url)
            .body(Bytes::from_static(b"{}"))
            .expect("request")
    }

    /// A stub collector answering every request with `status` and `body`.
    async fn collector(status: http::StatusCode, body: &'static str) -> String {
        let app = axum::Router::new().fallback(move || async move { (status, body) });
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .expect("bind");
        let url = format!("http://{}/v1/metrics", listener.local_addr().expect("addr"));
        tokio::spawn(async move { axum::serve(listener, app).await });
        url
    }

    async fn post(url: String) -> Result<Response<Bytes>, (Failure, String)> {
        tokio::task::spawn_blocking(move || {
            client()
                .post("a-token", request(&url))
                .map_err(|f| (f.kind, f.detail))
        })
        .await
        .expect("joined")
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_refusal_is_classified_by_status_with_the_body() {
        let url = collector(
            http::StatusCode::UNAUTHORIZED,
            r#"{"code":16,"message":"authentication didn't succeed"}"#,
        )
        .await;
        let (kind, detail) = post(url).await.expect_err("refused");
        assert_eq!(kind, Failure::Status(401));
        assert_eq!(
            detail,
            r#"{"code":16,"message":"authentication didn't succeed"}"#
        );
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn an_accepted_push_is_returned_as_is() {
        let url = collector(http::StatusCode::OK, "{}").await;
        let response = post(url).await.expect("accepted");
        assert_eq!(response.status(), http::StatusCode::OK);
        assert_eq!(response.body().as_ref(), b"{}");
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn an_unreachable_collector_is_a_transport_failure() {
        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
        let url = format!("http://{}/v1/metrics", listener.local_addr().expect("addr"));
        drop(listener);
        let (kind, _) = post(url).await.expect_err("unreachable");
        assert_eq!(kind, Failure::Transport);
    }

    /// The mint runs first: no token, no request, and the failure says so.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_failed_mint_is_a_token_failure() {
        let url = collector(http::StatusCode::OK, "{}").await;
        let kind = tokio::task::spawn_blocking(move || client().send(request(&url)).err())
            .await
            .expect("joined")
            .expect("no token")
            .kind;
        assert_eq!(kind, Failure::Token);
    }

    #[test]
    fn a_repeated_failure_is_logged_once_per_interval_with_the_count() {
        let mut log = FailureLog::default();
        let t0 = Instant::now();
        let kind = Failure::Status(401);
        assert_eq!(log.failed(kind, t0), Verdict::Warn { suppressed: 0 });
        for minute in 1..10 {
            assert_eq!(
                log.failed(kind, t0 + Duration::from_mins(minute)),
                Verdict::Quiet
            );
        }
        assert_eq!(
            log.failed(kind, t0 + REPEAT_INTERVAL),
            Verdict::Warn { suppressed: 9 }
        );
        assert_eq!(
            log.failed(kind, t0 + REPEAT_INTERVAL + Duration::from_mins(1)),
            Verdict::Quiet
        );
    }

    #[test]
    fn a_different_failure_is_logged_at_once() {
        let mut log = FailureLog::default();
        let t0 = Instant::now();
        assert_eq!(
            log.failed(Failure::Token, t0),
            Verdict::Warn { suppressed: 0 }
        );
        assert_eq!(
            log.failed(Failure::Status(401), t0),
            Verdict::Warn { suppressed: 0 }
        );
        assert_eq!(
            log.failed(Failure::Status(404), t0),
            Verdict::Warn { suppressed: 0 }
        );
    }

    #[test]
    fn recovery_is_logged_once_and_only_after_a_failure() {
        let mut log = FailureLog::default();
        let t0 = Instant::now();
        assert_eq!(log.succeeded(), Verdict::Quiet);
        log.failed(Failure::Transport, t0);
        assert_eq!(log.succeeded(), Verdict::Recovered);
        assert_eq!(log.succeeded(), Verdict::Quiet);
        assert_eq!(
            log.failed(Failure::Transport, t0),
            Verdict::Warn { suppressed: 0 }
        );
    }

    #[test]
    fn a_long_body_is_cut_on_a_character_boundary() {
        let body = "é".repeat(BODY_SNIPPET_CHARS + 5);
        let cut = snippet(body.as_bytes());
        assert_eq!(cut.chars().count(), BODY_SNIPPET_CHARS + 1);
        assert!(cut.ends_with('…'));
        assert_eq!(snippet(b"  short\nbody \n"), "short body");
    }
}