use crate::{
    provider::{FeedFailure, FeedStage, RawFeed},
    quota::Quota,
    storage::{Store, TrainDeparturesBatch, TrainStationsBatch},
};
use anyhow::{Context, Result};
use chrono::Utc;
use serde::Deserialize;
use std::{collections::HashSet, future::Future, sync::Arc};

#[derive(Clone, Debug, PartialEq)]
pub struct TrainStation {
    pub id: String,
    pub name: String,
    pub lat: f64,
    pub lon: f64,
}

#[derive(Clone, Debug, PartialEq)]
pub struct TrainDeparture {
    pub label: String,
    pub destination: String,
    pub in_seconds: i64,
    pub delayed: bool,
    pub platform: Option<String>,
    pub delay_seconds: Option<i64>,
}

#[derive(Deserialize)]
struct StationsDoc {
    station: Vec<RawStation>,
}

#[derive(Deserialize)]
struct RawStation {
    id: String,
    name: String,
    standardname: String,
    #[serde(rename = "locationX")]
    location_x: String,
    #[serde(rename = "locationY")]
    location_y: String,
}

pub fn parse_stations(body: &str) -> Result<Vec<TrainStation>> {
    let doc: StationsDoc = serde_json::from_str(body).context("stations envelope")?;
    Ok(doc
        .station
        .into_iter()
        .filter_map(|s| {
            let lon = s.location_x.parse::<f64>().ok()?;
            let lat = s.location_y.parse::<f64>().ok()?;
            let name = if s.standardname.trim().is_empty() { s.name } else { s.standardname };
            Some(TrainStation { id: s.id, name, lat, lon })
        })
        .collect())
}

#[derive(Deserialize)]
struct LiveboardDoc {
    departures: DeparturesDoc,
}

#[derive(Deserialize)]
struct DeparturesDoc {
    departure: Vec<RawDeparture>,
}

#[derive(Deserialize)]
struct RawDeparture {
    station: Option<String>,
    time: Option<String>,
    delay: Option<String>,
    platform: Option<String>,
    vehicleinfo: Option<RawVehicleInfo>,
}

#[derive(Deserialize)]
struct RawVehicleInfo {
    shortname: Option<String>,
    #[serde(rename = "type")]
    kind: Option<String>,
}

fn label_for(info: &Option<RawVehicleInfo>) -> String {
    if let Some(info) = info {
        if let Some(kind) = &info.kind {
            if !kind.trim().is_empty() {
                return kind.clone();
            }
        }
        if let Some(shortname) = &info.shortname {
            if let Some(first) = shortname.split_whitespace().next() {
                return first.to_string();
            }
        }
    }
    "T".to_string()
}

pub fn parse_liveboard(body: &str, now_unix: i64) -> Result<Vec<TrainDeparture>> {
    let doc: LiveboardDoc = serde_json::from_str(body).context("liveboard envelope")?;
    let mut departures: Vec<TrainDeparture> = doc
        .departures
        .departure
        .into_iter()
        .filter_map(|d| {
            let time: i64 = d.time.as_deref()?.parse().ok()?;
            let delay_seconds: Option<i64> = d.delay.as_deref().and_then(|s| s.parse().ok());
            let delay = delay_seconds.unwrap_or(0);
            let in_seconds = time + delay - now_unix;
            if in_seconds < -30 {
                return None;
            }
            Some(TrainDeparture {
                label: label_for(&d.vehicleinfo),
                destination: d.station.unwrap_or_default(),
                in_seconds,
                delayed: delay > 0,
                platform: d
                    .platform
                    .as_deref()
                    .map(str::trim)
                    .filter(|p| !p.is_empty() && *p != "?")
                    .map(String::from),
                delay_seconds,
            })
        })
        .collect();
    departures.sort_by_key(|d| d.in_seconds);
    departures.truncate(4);
    Ok(departures)
}

const STATIONS_URL: &str = "https://api.irail.be/stations/?format=json";
const LIVEBOARD_URL: &str = "https://api.irail.be/liveboard/";

async fn fetch_raw(
    http: &reqwest::Client,
    url: &str,
    feed: &'static str,
    scope: &str,
) -> std::result::Result<RawFeed, FeedFailure> {
    let response = http
        .get(url)
        .send()
        .await
        .with_context(|| format!("failed to fetch {url}"))
        .and_then(|response| {
            response
                .error_for_status()
                .with_context(|| format!("failed to fetch {url}"))
        })
        .map_err(|source| FeedFailure {
            stage: FeedStage::Provider,
            source,
        })?;
    let content_type = response
        .headers()
        .get(reqwest::header::CONTENT_TYPE)
        .with_context(|| format!("missing content type for {url}"))
        .and_then(|value| {
            value
                .to_str()
                .with_context(|| format!("invalid content type for {url}"))
        })
        .map_err(|source| FeedFailure {
            stage: FeedStage::Provider,
            source,
        })?
        .to_owned()
        .into_boxed_str();
    let body = response
        .bytes()
        .await
        .with_context(|| format!("failed to read {url}"))
        .map_err(|source| FeedFailure {
            stage: FeedStage::Provider,
            source,
        })?
        .to_vec();
    Ok(RawFeed {
        provider: "irail",
        feed,
        scope: scope.into(),
        captured_at: Utc::now(),
        source_at: None,
        content_type,
        body,
    })
}

async fn fetch_stations_at(
    http: &reqwest::Client,
    quota: &Quota,
    url: &str,
) -> std::result::Result<TrainStationsBatch, FeedFailure> {
    if !quota.try_take(Utc::now()) {
        return Err(FeedFailure {
            stage: FeedStage::Provider,
            source: anyhow::anyhow!("irail quota exhausted"),
        });
    }
    let raw = fetch_raw(http, url, "stations", "").await?;
    let stations = std::str::from_utf8(&raw.body)
        .context("irail stations encoding")
        .and_then(parse_stations)
        .and_then(|stations| {
            anyhow::ensure!(!stations.is_empty(), "irail stations empty");
            Ok(stations)
        })
        .map_err(|source| FeedFailure {
            stage: FeedStage::Parse,
            source,
        })?;
    Ok(TrainStationsBatch { raw, stations })
}

pub async fn refresh_stations(
    http: &reqwest::Client,
    quota: &Quota,
) -> std::result::Result<TrainStationsBatch, FeedFailure> {
    fetch_stations_at(http, quota, STATIONS_URL).await
}

async fn fetch_departures_at(
    http: &reqwest::Client,
    url: &str,
    station_id: &str,
) -> std::result::Result<TrainDeparturesBatch, FeedFailure> {
    let raw = fetch_raw(http, url, "liveboard", station_id).await?;
    let departures = std::str::from_utf8(&raw.body)
        .context("irail liveboard encoding")
        .and_then(|body| parse_liveboard(body, raw.captured_at.timestamp()))
        .map_err(|source| FeedFailure {
            stage: FeedStage::Parse,
            source,
        })?;
    Ok(TrainDeparturesBatch {
        raw,
        station_id: station_id.into(),
        departures,
    })
}

async fn fetch_departures_admitted(
    http: &reqwest::Client,
    station_id: &str,
) -> std::result::Result<TrainDeparturesBatch, FeedFailure> {
    let url = format!("{LIVEBOARD_URL}?id={station_id}&format=json&arrdep=departure");
    fetch_departures_at(http, &url, station_id).await
}

pub async fn fetch_departures(
    http: &reqwest::Client,
    quota: &Quota,
    station_id: &str,
) -> std::result::Result<TrainDeparturesBatch, FeedFailure> {
    if !quota.try_take(Utc::now()) {
        return Err(FeedFailure {
            stage: FeedStage::Provider,
            source: anyhow::anyhow!("irail quota exhausted"),
        });
    }
    fetch_departures_admitted(http, station_id).await
}

async fn run_liveboard_worker_with<F, Fut>(
    store: Store,
    quota: Arc<Quota>,
    mut requests: tokio::sync::mpsc::Receiver<Vec<String>>,
    fetch: F,
) -> anyhow::Result<()>
where
    F: Fn(String) -> Fut,
    Fut: Future<Output = std::result::Result<TrainDeparturesBatch, FeedFailure>>,
{
    let mut next = Vec::new();
    loop {
        let initial = if next.is_empty() {
            match requests.recv().await {
                Some(station_ids) => station_ids,
                None => return Ok(()),
            }
        } else {
            std::mem::take(&mut next)
        };
        let mut seen = HashSet::new();
        let mut pending = Vec::new();
        for station_id in initial {
            if seen.insert(station_id.clone()) {
                pending.push(station_id);
            }
        }
        while let Ok(station_ids) = requests.try_recv() {
            for station_id in station_ids {
                if seen.insert(station_id.clone()) {
                    pending.push(station_id);
                }
            }
        }
        let admitted = pending
            .into_iter()
            .filter(|_| quota.try_take(Utc::now()));
        let results = futures::future::join_all(admitted.map(&fetch)).await;
        let mut failures = Vec::new();
        for result in results {
            let applied = match result {
                Ok(batch) => store
                    .apply_train_departures(batch)
                    .await
                    .map_err(|source| FeedFailure {
                        stage: FeedStage::Commit,
                        source,
                    }),
                Err(failure) => Err(failure),
            };
            if let Err(failure) = applied {
                tracing::warn!("irail liveboard refresh failed: {failure}");
                failures.push(failure);
            }
        }
        if !failures.is_empty() {
            let stage = if failures
                .iter()
                .any(|failure| failure.stage == FeedStage::Commit)
            {
                FeedStage::Commit
            } else if failures
                .iter()
                .any(|failure| failure.stage == FeedStage::Parse)
            {
                FeedStage::Parse
            } else {
                FeedStage::Provider
            };
            let mut messages = failures
                .iter()
                .map(|failure| {
                    format!("{}: {:#}", failure.stage.as_str(), failure.source)
                })
                .collect::<Vec<_>>();
            messages.sort_unstable();
            messages.dedup();
            if let Err(error) = store
                .record_feed_failure(
                    "irail",
                    "liveboard",
                    stage,
                    &messages.join("; "),
                    Utc::now(),
                )
                .await
            {
                tracing::warn!("irail liveboard failure recording failed: {error:#}");
            }
        }
        let mut queued = HashSet::new();
        while let Ok(station_ids) = requests.try_recv() {
            for station_id in station_ids {
                if !seen.contains(&station_id) && queued.insert(station_id.clone()) {
                    next.push(station_id);
                }
            }
        }
        if next.is_empty() && requests.is_closed() {
            return Ok(());
        }
    }
}

pub async fn run_liveboard_worker(
    store: Store,
    http: reqwest::Client,
    quota: Arc<Quota>,
    requests: tokio::sync::mpsc::Receiver<Vec<String>>,
) -> anyhow::Result<()> {
    run_liveboard_worker_with(store, quota, requests, move |station_id| {
        let http = http.clone();
        async move { fetch_departures_admitted(&http, &station_id).await }
    })
    .await
}

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

    const STATIONS: &str = r#"{"station":[
      {"id":"BE.NMBS.008812005","name":"Brussels-North","standardname":"Brussel-Noord/Bruxelles-Nord","locationX":"4.36","locationY":"50.85"},
      {"id":"BE.NMBS.008813003","name":"Brussels-Central","standardname":"","locationX":"4.356801","locationY":"50.845658"},
      {"id":"BE.NMBS.999","name":"Bad","standardname":"Bad","locationX":"bogus","locationY":"50.0"}
    ]}"#;

    #[test]
    fn parses_stations_and_prefers_standardname() {
        let stations = parse_stations(STATIONS).unwrap();
        assert_eq!(stations.len(), 2);
        assert_eq!(stations[0].id, "BE.NMBS.008812005");
        assert_eq!(stations[0].name, "Brussel-Noord/Bruxelles-Nord");
        assert!((stations[0].lon - 4.36).abs() < 1e-9);
        assert!((stations[0].lat - 50.85).abs() < 1e-9);
    }

    #[test]
    fn falls_back_to_name_when_standardname_empty() {
        let stations = parse_stations(STATIONS).unwrap();
        assert_eq!(stations[1].name, "Brussels-Central");
    }

    #[test]
    fn skips_station_with_unparsable_coordinates() {
        let stations = parse_stations(STATIONS).unwrap();
        assert!(!stations.iter().any(|s| s.id == "BE.NMBS.999"));
    }

    fn liveboard(departures: &str) -> String {
        format!(r#"{{"departures":{{"number":"0","departure":[{departures}]}}}}"#)
    }

    #[test]
    fn parses_liveboard_with_documented_shape() {
        let body = liveboard(
            r#"{"station":"Leuven","time":"1755733200","delay":"60","platform":"3","vehicleinfo":{"shortname":"IC 1234","type":"IC"}}"#,
        );
        let departures = parse_liveboard(&body, 1755733000).unwrap();
        assert_eq!(departures.len(), 1);
        assert_eq!(departures[0].label, "IC");
        assert_eq!(departures[0].destination, "Leuven");
        assert_eq!(departures[0].in_seconds, 260);
        assert!(departures[0].delayed);
        assert_eq!(departures[0].platform.as_deref(), Some("3"));
        assert_eq!(departures[0].delay_seconds, Some(60));
    }

    #[test]
    fn platform_and_delay_absent_from_the_feed_stay_none() {
        let body = liveboard(r#"{"station":"A","time":"100"}"#);
        let departures = parse_liveboard(&body, 0).unwrap();
        assert_eq!(departures[0].platform, None);
        assert_eq!(departures[0].delay_seconds, None);
    }

    #[test]
    fn unknown_or_blank_platform_marker_becomes_none() {
        let body = liveboard(
            r#"{"station":"A","time":"100","delay":"0","platform":"?"},
               {"station":"B","time":"100","delay":"0","platform":" "},
               {"station":"C","time":"100","delay":"0","platform":" ? "},
               {"station":"D","time":"100","delay":"0","platform":" 3 "}"#,
        );
        let departures = parse_liveboard(&body, 0).unwrap();
        assert_eq!(departures[0].platform, None);
        assert_eq!(departures[1].platform, None);
        assert_eq!(departures[2].platform, None);
        assert_eq!(departures[3].platform.as_deref(), Some("3"));
        assert_eq!(departures[0].delay_seconds, Some(0));
    }

    #[test]
    fn label_falls_back_to_first_token_of_shortname_then_t() {
        let body = liveboard(
            r#"{"station":"A","time":"100","delay":"0","vehicleinfo":{"shortname":"P 7008"}},
               {"station":"B","time":"100","delay":"0"}"#,
        );
        let departures = parse_liveboard(&body, 0).unwrap();
        assert_eq!(departures[0].label, "P");
        assert_eq!(departures[1].label, "T");
    }

    #[test]
    fn drops_unparsable_time_and_past_departures() {
        let body = liveboard(
            r#"{"station":"A","time":"not-a-number","delay":"0"},
               {"station":"B","time":"50","delay":"0"}"#,
        );
        let departures = parse_liveboard(&body, 100).unwrap();
        assert!(departures.is_empty());
    }

    #[test]
    fn keeps_departures_within_the_grace_window() {
        let body = liveboard(r#"{"station":"A","time":"71","delay":"0"}"#);
        let departures = parse_liveboard(&body, 100).unwrap();
        assert_eq!(departures.len(), 1);
        assert_eq!(departures[0].in_seconds, -29);
    }

    #[test]
    fn sorts_ascending_and_truncates_to_four() {
        let entries: Vec<String> = (0..6)
            .map(|i| format!(r#"{{"station":"S{i}","time":"{}","delay":"0"}}"#, 1000 - i * 10))
            .collect();
        let body = liveboard(&entries.join(","));
        let departures = parse_liveboard(&body, 0).unwrap();
        assert_eq!(departures.len(), 4);
        assert_eq!(departures[0].destination, "S5");
        assert_eq!(departures[3].destination, "S2");
        assert!(departures.windows(2).all(|w| w[0].in_seconds <= w[1].in_seconds));
    }

    fn departure(destination: &str) -> TrainDeparture {
        TrainDeparture {
            label: "IC".into(),
            destination: destination.into(),
            in_seconds: 300,
            delayed: false,
            platform: None,
            delay_seconds: None,
        }
    }

    fn raw(
        feed: &'static str,
        scope: &str,
        captured_at: chrono::DateTime<Utc>,
        body: &[u8],
    ) -> crate::provider::RawFeed {
        crate::provider::RawFeed {
            provider: "irail",
            feed,
            scope: scope.into(),
            captured_at,
            source_at: None,
            content_type: "application/json".into(),
            body: body.to_vec(),
        }
    }

    async fn worker_store() -> (tempfile::TempDir, crate::storage::Store) {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("stibble.sqlite");
        let extension = std::env::var("SPATIALITE_EXTENSION").unwrap();
        let store = crate::storage::Store::open(path, extension).await.unwrap();
        let captured_at = Utc::now();
        store
            .apply_train_stations(crate::storage::TrainStationsBatch {
                raw: raw("stations", "", captured_at, STATIONS.as_bytes()),
                stations: vec![
                    TrainStation {
                        id: "first".into(),
                        name: "FIRST".into(),
                        lat: 50.84665,
                        lon: 4.35243,
                    },
                    TrainStation {
                        id: "second".into(),
                        name: "SECOND".into(),
                        lat: 50.85,
                        lon: 4.35243,
                    },
                ],
            })
            .await
            .unwrap();
        (directory, store)
    }

    fn worker_batch(station_id: String) -> crate::storage::TrainDeparturesBatch {
        let captured_at = Utc::now();
        crate::storage::TrainDeparturesBatch {
            raw: raw("liveboard", &station_id, captured_at, b"{}"),
            station_id: station_id.clone(),
            departures: vec![departure(&station_id)],
        }
    }

    #[tokio::test]
    async fn liveboard_worker_deduplicates_pending_station_ids_and_commits_each() {
        let (_directory, store) = worker_store().await;
        let (tx, rx) = tokio::sync::mpsc::channel(8);
        tx.send(vec!["first".into(), "first".into()])
            .await
            .unwrap();
        tx.send(vec!["first".into(), "second".into()])
            .await
            .unwrap();
        drop(tx);
        let calls = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let fetched = calls.clone();

        run_liveboard_worker_with(
            store.clone(),
            Arc::new(Quota::new(10, 100)),
            rx,
            move |station_id| {
                let fetched = fetched.clone();
                async move {
                    fetched.lock().await.push(station_id.clone());
                    Ok(worker_batch(station_id))
                }
            },
        )
        .await
        .unwrap();

        assert_eq!(
            calls.lock().await.as_slice(),
            ["first".to_string(), "second".to_string()],
        );
        let board = store
            .nearby_trains(50.84665, 4.35243, Utc::now())
            .await
            .unwrap();
        assert_eq!(board.stops[0].departures[0].destination, "first");
        assert_eq!(board.stops[1].departures[0].destination, "second");
    }

    #[tokio::test]
    async fn liveboard_worker_admits_only_the_available_quota_in_request_order() {
        let (_directory, store) = worker_store().await;
        let (tx, rx) = tokio::sync::mpsc::channel(8);
        tx.send(vec!["first".into(), "second".into()])
            .await
            .unwrap();
        drop(tx);
        let calls = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let fetched = calls.clone();

        run_liveboard_worker_with(
            store,
            Arc::new(Quota::new(1, 100)),
            rx,
            move |station_id| {
                let fetched = fetched.clone();
                async move {
                    fetched.lock().await.push(station_id.clone());
                    Ok(worker_batch(station_id))
                }
            },
        )
        .await
        .unwrap();

        assert_eq!(calls.lock().await.as_slice(), ["first".to_string()]);
    }

    #[tokio::test]
    async fn liveboard_worker_fetches_admitted_stations_concurrently() {
        use std::sync::atomic::{AtomicUsize, Ordering};

        let (_directory, store) = worker_store().await;
        let (tx, rx) = tokio::sync::mpsc::channel(8);
        tx.send(vec!["first".into(), "second".into()])
            .await
            .unwrap();
        drop(tx);
        let in_flight = Arc::new(AtomicUsize::new(0));
        let peak = Arc::new(AtomicUsize::new(0));
        let observed_in_flight = in_flight.clone();
        let observed_peak = peak.clone();

        run_liveboard_worker_with(
            store,
            Arc::new(Quota::new(10, 100)),
            rx,
            move |station_id| {
                let in_flight = observed_in_flight.clone();
                let peak = observed_peak.clone();
                async move {
                    let current = in_flight.fetch_add(1, Ordering::SeqCst) + 1;
                    peak.fetch_max(current, Ordering::SeqCst);
                    tokio::time::sleep(std::time::Duration::from_millis(50)).await;
                    in_flight.fetch_sub(1, Ordering::SeqCst);
                    Ok(worker_batch(station_id))
                }
            },
        )
        .await
        .unwrap();

        assert_eq!(peak.load(Ordering::SeqCst), 2);
    }

    async fn assert_mixed_liveboard_batch_retains_failure(station_ids: [&str; 2]) {
        let (directory, store) = worker_store().await;
        let (tx, rx) = tokio::sync::mpsc::channel(8);
        tx.send(
            station_ids
                .into_iter()
                .map(String::from)
                .collect(),
        )
        .await
        .unwrap();
        drop(tx);

        run_liveboard_worker_with(
            store.clone(),
            Arc::new(Quota::new(10, 100)),
            rx,
            |station_id| async move {
                if station_id == "first" {
                    Err(crate::provider::FeedFailure {
                        stage: crate::provider::FeedStage::Provider,
                        source: anyhow::anyhow!("failed first"),
                    })
                } else {
                    Ok(worker_batch(station_id))
                }
            },
        )
        .await
        .unwrap();

        let connection = deadpool_sqlite::rusqlite::Connection::open(
            directory.path().join("stibble.sqlite"),
        )
        .unwrap();
        let failure_count = connection
            .query_row(
                "SELECT count(*)
                 FROM feed_failure
                 JOIN provider_feed ON provider_feed.id = feed_failure.feed_id
                 WHERE provider_feed.provider = 'irail'
                   AND provider_feed.feed = 'liveboard'",
                [],
                |row| row.get::<_, i64>(0),
            )
            .unwrap();
        assert_eq!(failure_count, 1);
        let stage = connection
            .query_row(
                "SELECT stage
                 FROM feed_failure
                 JOIN provider_feed ON provider_feed.id = feed_failure.feed_id
                 WHERE provider_feed.provider = 'irail'
                   AND provider_feed.feed = 'liveboard'",
                [],
                |row| row.get::<_, String>(0),
            )
            .unwrap();
        assert_eq!(stage, "provider");
        let board = store
            .nearby_trains(50.84665, 4.35243, Utc::now())
            .await
            .unwrap();
        assert!(board.stops[0].departures.is_empty());
        assert_eq!(board.stops[1].departures[0].destination, "second");
    }

    #[tokio::test]
    async fn liveboard_worker_retains_failure_before_success() {
        assert_mixed_liveboard_batch_retains_failure(["first", "second"]).await;
    }

    #[tokio::test]
    async fn liveboard_worker_retains_failure_after_success() {
        assert_mixed_liveboard_batch_retains_failure(["second", "first"]).await;
    }

    async fn irail_server() -> (String, tokio::task::JoinHandle<()>) {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .unwrap();
        let address = listener.local_addr().unwrap();
        let departure_at = Utc::now().timestamp() + 300;
        let liveboard = liveboard(&format!(
            r#"{{"station":"LEUVEN","time":"{departure_at}","delay":"0","platform":"3","vehicleinfo":{{"shortname":"IC 1","type":"IC"}}}}"#,
        ));
        let app = axum::Router::new()
            .route(
                "/stations",
                axum::routing::get(|| async {
                    (
                        [(axum::http::header::CONTENT_TYPE, "application/json")],
                        STATIONS,
                    )
                }),
            )
            .route(
                "/liveboard",
                axum::routing::get(move || {
                    let liveboard = liveboard.clone();
                    async move {
                        (
                            [(axum::http::header::CONTENT_TYPE, "application/json")],
                            liveboard,
                        )
                    }
                }),
            );
        let server = tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });
        (format!("http://{address}"), server)
    }

    #[tokio::test]
    async fn fetches_station_and_liveboard_raw_parsed_batches_with_exact_scopes() {
        let (base, server) = irail_server().await;
        let http = reqwest::Client::new();
        let quota = Quota::new(10, 100);
        let stations = fetch_stations_at(&http, &quota, &format!("{base}/stations"))
            .await
            .unwrap();
        assert_eq!(stations.raw.provider, "irail");
        assert_eq!(stations.raw.feed, "stations");
        assert!(stations.raw.scope.is_empty());
        assert_eq!(stations.raw.body, STATIONS.as_bytes());
        assert_eq!(stations.stations.len(), 2);

        let departures =
            fetch_departures_at(&http, &format!("{base}/liveboard"), "first")
                .await
                .unwrap();
        server.abort();
        assert_eq!(departures.raw.provider, "irail");
        assert_eq!(departures.raw.feed, "liveboard");
        assert_eq!(departures.raw.scope.as_ref(), "first");
        assert_eq!(departures.station_id, "first");
        assert_eq!(departures.departures[0].destination, "LEUVEN");
    }

    #[tokio::test]
    async fn station_fetch_maps_http_and_parser_failures_to_their_stages() {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
            .await
            .unwrap();
        let address = listener.local_addr().unwrap();
        let app = axum::Router::new().route(
            "/invalid",
            axum::routing::get(|| async {
                (
                    [(axum::http::header::CONTENT_TYPE, "application/json")],
                    "invalid",
                )
            }),
        );
        let server = tokio::spawn(async move {
            axum::serve(listener, app).await.unwrap();
        });
        let http = reqwest::Client::new();
        let quota = Quota::new(10, 100);
        let provider =
            match fetch_stations_at(&http, &quota, &format!("http://{address}/missing")).await {
                Ok(_) => panic!("expected provider failure"),
                Err(failure) => failure,
            };
        let parsed =
            match fetch_stations_at(&http, &quota, &format!("http://{address}/invalid")).await {
                Ok(_) => panic!("expected parse failure"),
                Err(failure) => failure,
            };
        server.abort();
        assert_eq!(provider.stage, crate::provider::FeedStage::Provider);
        assert_eq!(parsed.stage, crate::provider::FeedStage::Parse);
    }
}