//! Fetching each source on its own interval.
//!
//! One task per source, each reporting into a channel the UI drains. Sources
//! are independent on purpose: a daemon that hangs must not stop the others
//! from updating, because the tab that stops updating is exactly the one you
//! need to see.

use std::time::Duration;

use chrono::Utc;
use ops_status::Payload;
use tokio::sync::mpsc;

use crate::config::Source;

/// A poll result, tagged with which source it came from.
pub(crate) struct Update {
    pub index: usize,
    pub at: chrono::DateTime<Utc>,
    pub result: Result<Payload, String>,
}

/// How long a single request may take before it counts as unreachable.
///
/// Bounded well under any sane poll interval: a request that outlives its own
/// interval would stack tasks up behind it, and a source that is merely slow
/// should read as a problem rather than silently delaying every later poll.
const REQUEST_TIMEOUT: Duration = Duration::from_secs(10);

/// Spawn one polling task per source. Returns the receiving end.
pub(crate) fn spawn_all(sources: &[Source]) -> mpsc::Receiver<Update> {
    // Capacity comfortably exceeds one in-flight update per source, so a
    // momentarily busy UI loop cannot make a poller block or drop a result.
    let (tx, rx) = mpsc::channel(sources.len().max(1) * 4);
    for (index, source) in sources.iter().enumerate() {
        tokio::spawn(poll_forever(index, source.clone(), tx.clone()));
    }
    rx
}

async fn poll_forever(index: usize, source: Source, tx: mpsc::Sender<Update>) {
    let client = match reqwest::Client::builder().timeout(REQUEST_TIMEOUT).build() {
        Ok(client) => client,
        Err(e) => {
            let _ = tx
                .send(Update {
                    index,
                    at: Utc::now(),
                    result: Err(format!("client: {e}")),
                })
                .await;
            return;
        }
    };

    let url = source.status_url();
    let mut ticker = tokio::time::interval(source.poll_interval());
    // A poll that overruns its interval must not cause a burst of catch-up
    // requests at the next tick.
    ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);

    loop {
        ticker.tick().await;
        let result = fetch(&client, &url, source.token().as_deref()).await;
        // The token is read per request rather than captured once, so rotating
        // it does not require restarting the viewer.
        if tx
            .send(Update {
                index,
                at: Utc::now(),
                result,
            })
            .await
            .is_err()
        {
            return; // UI is gone
        }
    }
}

async fn fetch(
    client: &reqwest::Client,
    url: &str,
    token: Option<&str>,
) -> Result<Payload, String> {
    let mut request = client.get(url);
    if let Some(token) = token {
        request = request.bearer_auth(token);
    }

    let response = request.send().await.map_err(short_error)?;
    let status = response.status();
    if !status.is_success() {
        // 401 is worth its own words: it is the most likely misconfiguration
        // and the least self-evident from a bare status code.
        return Err(match status.as_u16() {
            401 | 403 => "unauthorized (check the source's token_env)".into(),
            code => format!("HTTP {code}"),
        });
    }

    let body = response.bytes().await.map_err(short_error)?;
    serde_json::from_slice(&body).map_err(|e| format!("bad payload: {e}"))
}

/// reqwest errors stringify into a paragraph with a full URL chain. The rollup
/// has one column for this, so keep the part that says what went wrong.
// e is consumed into the short string.
#[allow(clippy::needless_pass_by_value)]
fn short_error(e: reqwest::Error) -> String {
    if e.is_timeout() {
        return "timed out".into();
    }
    if e.is_connect() {
        return "connection refused".into();
    }
    if e.is_decode() {
        return "bad response body".into();
    }
    let text = e.to_string();
    text.split(':')
        .next_back()
        .unwrap_or(&text)
        .trim()
        .to_string()
}

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

    #[test]
    fn a_status_url_is_built_from_the_configured_base() {
        let source: Source = toml::from_str(
            r#"
name = "sando"
url = "http://fw13:8080"
"#,
        )
        .unwrap();
        assert_eq!(source.status_url(), "http://fw13:8080/status.json");
    }

    #[tokio::test]
    async fn an_unreachable_host_reports_a_short_reason() {
        let client = reqwest::Client::builder()
            .timeout(Duration::from_millis(300))
            .build()
            .unwrap();
        // Port 1 on loopback: nothing listens, so this refuses fast.
        let err = fetch(&client, "http://127.0.0.1:1/status.json", None)
            .await
            .unwrap_err();
        assert!(
            err.len() < 60,
            "the rollup has one column for this: {err:?}"
        );
        assert!(
            !err.contains("http://"),
            "no URL chain in the summary: {err:?}"
        );
    }

    #[tokio::test]
    async fn a_body_that_is_not_a_payload_is_an_error_not_a_panic() {
        let client = reqwest::Client::new();
        let bad: Result<Payload, String> =
            serde_json::from_slice(b"not json").map_err(|e| format!("bad payload: {e}"));
        assert!(bad.unwrap_err().starts_with("bad payload"));
        drop(client);
    }
}