//! Reading 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.
//!
//! Two origins, one loop. An HTTP source is fetched and a file source is read,
//! and past that nothing downstream can tell them apart: both produce a
//! `Payload` or a short reason, on the same tick, into the same channel. A batch
//! producer is legible here because the UI already keeps the last payload across
//! a failed poll, so a sweep that finished at 03:20 still reads at 15:00.

use std::path::Path;
use std::time::Duration;

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

use crate::config::{Origin, 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 origin = source.origin();
    // Built once per task, and only for an HTTP source: a file source has no
    // transport to fail, so a broken TLS setup must not take its tab down with
    // the others.
    let client = if matches!(origin, Origin::Http(_)) {
        match crate::tls::client(REQUEST_TIMEOUT) {
            Ok(client) => Some(client),
            Err(e) => {
                let _ = tx
                    .send(Update {
                        index,
                        at: Utc::now(),
                        result: Err(format!("client: {e}")),
                    })
                    .await;
                return;
            }
        }
    } else {
        None
    };

    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 = match &origin {
            // The token is read per request rather than captured once, so
            // rotating it does not require restarting magicmirror.
            Origin::Http(url) => match &client {
                Some(client) => fetch(client, url, source.token().as_deref()).await,
                None => Err("client: not built".into()),
            },
            Origin::File(path) => read(path).await,
        };
        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}"))
}

/// Read a payload a batch producer left on disk.
///
/// Every way this fails is the same condition an unreachable daemon is in --
/// nothing current to show -- so it reports the same shape, and the UI keeps the
/// last good payload either way. A missing file is the ordinary case on a fresh
/// machine: the producer has not run yet.
async fn read(path: &Path) -> Result<Payload, String> {
    let bytes = tokio::fs::read(path).await.map_err(short_io_error)?;
    serde_json::from_slice(&bytes).map_err(|e| format!("bad payload: {e}"))
}

/// The rollup has one column for this, and an io error stringifies with the
/// whole path in it, which is already the tab's own configuration.
#[allow(clippy::needless_pass_by_value)]
fn short_io_error(e: std::io::Error) -> String {
    match e.kind() {
        std::io::ErrorKind::NotFound => "no payload written yet".into(),
        std::io::ErrorKind::PermissionDenied => "permission denied".into(),
        std::io::ErrorKind::IsADirectory => "path is a directory".into(),
        kind => kind.to_string(),
    }
}

/// 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.origin(),
            Origin::Http("http://fw13:8080/status.json".into())
        );
    }

    /// The batch producer's ordinary state on a fresh machine: configured,
    /// pointed at the right place, and nothing has run yet. It has to read as a
    /// source with nothing current rather than as a broken config.
    #[tokio::test]
    async fn a_payload_that_has_not_been_written_yet_is_a_short_reason() {
        let dir = tempfile::tempdir().unwrap();
        let err = read(&dir.path().join("status.json")).await.unwrap_err();
        assert_eq!(err, "no payload written yet");
    }

    #[tokio::test]
    async fn a_payload_on_disk_reads_the_same_as_one_off_the_wire() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("status.json");
        let written = Payload::new("witchbroom", Utc::now());
        std::fs::write(&path, serde_json::to_string(&written).unwrap()).unwrap();

        let got = read(&path).await.unwrap();
        assert_eq!(got.source, "witchbroom");
        assert_eq!(got, written);
    }

    #[tokio::test]
    async fn a_file_that_is_not_a_payload_says_so_rather_than_panicking() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("status.json");
        std::fs::write(&path, b"half a write").unwrap();
        assert!(read(&path).await.unwrap_err().starts_with("bad payload"));
    }

    // Opens a socket, which miri has no shim for. See the note in `exec`.
    #[tokio::test]
    #[cfg_attr(miri, ignore)]
    async fn an_unreachable_host_reports_a_short_reason() {
        let client = crate::tls::client(Duration::from_millis(300)).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:?}"
        );
    }

    // Builds a reqwest client, which reaches for the same socket machinery.
    #[tokio::test]
    #[cfg_attr(miri, ignore)]
    async fn a_body_that_is_not_a_payload_is_an_error_not_a_panic() {
        let client = crate::tls::client(Duration::from_millis(300)).unwrap();
        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);
    }
}