//! A small HTTP client with per-host pacing.
//!
//! Tangled's mirror and knots answer a burst with a bare `403` and no
//! `Retry-After`. The crawler is a long, polite background job, so the client
//! spaces requests per host and retries the statuses that mean "later" rather
//! than treating a single 403 as a dead repository.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::Duration;

use anyhow::{Context, Result, bail};
use reqwest::header::{HeaderMap, HeaderValue, USER_AGENT};
use reqwest::{Client, StatusCode};
use tokio::sync::Mutex;
use tokio::time::{Instant, sleep};

#[derive(Debug, Clone)]
pub struct HttpClient {
    client: Client,
    gates: Arc<Mutex<HashMap<String, Arc<HostGate>>>>,
    interval: Duration,
    attempts: u32,
    circuit_failures: u32,
    circuit_cooldown: Duration,
}

#[derive(Debug)]
struct HostGate {
    state: Mutex<GateState>,
}

#[derive(Debug)]
struct GateState {
    next: Instant,
    failures: u32,
    open_until: Option<Instant>,
    /// The failure that opened the circuit, so a request it refuses says
    /// what is wrong with the host rather than only that it was skipped.
    opened_by: Option<String>,
}

#[derive(Debug)]
pub struct HttpResponse {
    pub headers: HeaderMap,
    pub body: Vec<u8>,
}

impl HttpClient {
    pub fn new(timeout: Duration, per_host_rps: f64) -> Result<Self> {
        let rps = if per_host_rps > 0.0 {
            per_host_rps
        } else {
            1.0
        };
        let client = Client::builder()
            .timeout(timeout)
            .user_agent(format!(
                "tally/{} (+https://dev.permadeath.com)",
                env!("CARGO_PKG_VERSION")
            ))
            .build()
            .context("building HTTP client")?;
        Ok(Self {
            client,
            gates: Arc::new(Mutex::new(HashMap::new())),
            interval: Duration::from_secs_f64(1.0 / rps),
            attempts: 5,
            circuit_failures: 5,
            circuit_cooldown: Duration::from_secs(60),
        })
    }

    /// The global crawl can use fewer attempts and a shorter circuit than an
    /// interactive run, because a broken mirror is better skipped than retried
    /// for every repository.
    pub fn with_policy(
        timeout: Duration,
        per_host_rps: f64,
        attempts: u32,
        circuit_failures: u32,
        circuit_cooldown: Duration,
    ) -> Result<Self> {
        let mut client = Self::new(timeout, per_host_rps)?;
        client.attempts = attempts.max(1);
        client.circuit_failures = circuit_failures.max(1);
        client.circuit_cooldown = circuit_cooldown;
        Ok(client)
    }

    /// GET a URL, pacing by host and retrying transient failures.
    pub async fn get(&self, url: &str, bearer: Option<&str>) -> Result<HttpResponse> {
        let host = url
            .split("://")
            .nth(1)
            .and_then(|rest| rest.split('/').next())
            .unwrap_or(url)
            .to_ascii_lowercase();

        let gate = {
            let mut gates = self.gates.lock().await;
            gates
                .entry(host.clone())
                .or_insert_with(|| {
                    Arc::new(HostGate {
                        state: Mutex::new(GateState {
                            next: Instant::now(),
                            failures: 0,
                            open_until: None,
                            opened_by: None,
                        }),
                    })
                })
                .clone()
        };

        let attempts = self.attempts;
        for attempt in 0..attempts {
            self.wait_for_turn(&host, &gate).await?;

            let mut request = self.client.get(url).header(
                USER_AGENT,
                HeaderValue::from_static(concat!(
                    "tally/",
                    env!("CARGO_PKG_VERSION"),
                    " (+https://dev.permadeath.com)"
                )),
            );
            if let Some(token) = bearer {
                request = request.bearer_auth(token);
            }

            let response = match request.send().await {
                Ok(response) => response,
                Err(err) => {
                    let opened = self
                        .record_failure(&host, &gate, || {
                            format!("GET {url}: {}", error_chain(&err))
                        })
                        .await;
                    if opened || attempt + 1 >= attempts {
                        return Err(err).with_context(|| format!("GET {url}"));
                    }
                    tracing::warn!(url, attempt, error = %err, "HTTP request failed; retrying");
                    sleep(backoff(attempt)).await;
                    continue;
                }
            };

            let status = response.status();
            let headers = response.headers().clone();

            if status.is_success() {
                self.record_success(&gate).await;
                let body = response.bytes().await?.to_vec();
                return Ok(HttpResponse { headers, body });
            }

            // GitHub tells us exactly when the window resets. Waiting here
            // would stall a scheduled run; returning lets the caller record
            // the partial scan and resume on the next tick.
            if status == StatusCode::FORBIDDEN
                && headers
                    .get("x-ratelimit-remaining")
                    .and_then(|value| value.to_str().ok())
                    == Some("0")
            {
                let reset = headers
                    .get("x-ratelimit-reset")
                    .and_then(|value| value.to_str().ok())
                    .unwrap_or("unknown");
                bail!("rate limited by {host}; reset at {reset}");
            }

            let transient = status == StatusCode::TOO_MANY_REQUESTS
                || status == StatusCode::FORBIDDEN
                || status == StatusCode::BAD_GATEWAY
                || status == StatusCode::SERVICE_UNAVAILABLE
                || status == StatusCode::GATEWAY_TIMEOUT
                || status.is_server_error();

            if transient {
                let opened = self
                    .record_failure(&host, &gate, || {
                        format!("GET {url} returned {}", status.as_u16())
                    })
                    .await;
                if opened || attempt + 1 >= attempts {
                    let body = response.text().await.unwrap_or_default();
                    bail!(
                        "GET {url} returned {}: {}",
                        status.as_u16(),
                        body.chars().take(300).collect::<String>()
                    );
                }
                let delay = backoff(attempt);
                tracing::warn!(
                    url,
                    attempt,
                    status = status.as_u16(),
                    delay_ms = delay.as_millis() as u64,
                    "transient HTTP response; backing off"
                );
                sleep(delay).await;
                continue;
            }

            let body = response.text().await.unwrap_or_default();
            bail!(
                "GET {url} returned {}: {}",
                status.as_u16(),
                body.chars().take(300).collect::<String>()
            );
        }

        unreachable!("the retry loop always returns")
    }

    pub async fn get_json<T: serde::de::DeserializeOwned>(
        &self,
        url: &str,
        bearer: Option<&str>,
    ) -> Result<T> {
        let response = self.get(url, bearer).await?;
        serde_json::from_slice(&response.body).with_context(|| format!("decoding JSON from {url}"))
    }

    async fn wait_for_turn(&self, host: &str, gate: &HostGate) -> Result<()> {
        let mut state = gate.state.lock().await;
        let now = Instant::now();
        if let Some(open_until) = state.open_until {
            if open_until > now {
                bail!(
                    "circuit open for {host} for {}s after: {}",
                    (open_until - now).as_secs().max(1),
                    state.opened_by.as_deref().unwrap_or("repeated failures")
                );
            }
            state.open_until = None;
            state.opened_by = None;
        }
        let wait = if state.next > now {
            state.next - now
        } else {
            Duration::ZERO
        };
        state.next = state.next.max(now) + self.interval;
        drop(state);
        if !wait.is_zero() {
            sleep(wait).await;
        }
        Ok(())
    }

    /// Returns true when this failure opened the circuit. `failure` says
    /// what failed, for the requests the open circuit refuses.
    async fn record_failure(
        &self,
        host: &str,
        gate: &HostGate,
        failure: impl FnOnce() -> String,
    ) -> bool {
        let mut state = gate.state.lock().await;
        state.failures += 1;
        if state.failures < self.circuit_failures {
            return false;
        }
        state.open_until = Some(Instant::now() + self.circuit_cooldown);
        state.opened_by = Some(failure());
        state.failures = 0;
        tracing::warn!(
            host,
            cooldown_secs = self.circuit_cooldown.as_secs(),
            "HTTP circuit opened after repeated failures"
        );
        true
    }

    async fn record_success(&self, gate: &HostGate) {
        let mut state = gate.state.lock().await;
        state.failures = 0;
        state.open_until = None;
        state.opened_by = None;
    }
}

/// An error and its causes on one line, as `{:#}` prints an `anyhow` error:
/// a connection failure's cause ("dns error", "connection refused") is in
/// the chain, not the top-level message.
fn error_chain(err: &dyn std::error::Error) -> String {
    let mut text = err.to_string();
    let mut source = err.source();
    while let Some(cause) = source {
        text.push_str(": ");
        text.push_str(&cause.to_string());
        source = cause.source();
    }
    text
}

fn backoff(attempt: u32) -> Duration {
    // 300ms, 600ms, 1.2s, 2.4s ... with a small deterministic stagger so two
    // concurrent requests do not wake together.
    let base = 300u64.saturating_mul(1u64 << attempt.min(6));
    Duration::from_millis(base + (u64::from(attempt) * 73))
}

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

    #[test]
    fn backoff_grows() {
        assert!(backoff(3) > backoff(1));
    }
}