use crate::departure::{self, Badge, Departure};
use crate::model::{Bilingual, VehicleKind};
use crate::state::AppState;
use axum::extract::{Query, State};
use axum::http::StatusCode;
use axum::response::Json;
use axum::routing::get;
use axum::Router;
use chrono::Utc;
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use tower_http::trace::{DefaultOnFailure, TraceLayer};

type BoardResponse = ([(axum::http::HeaderName, &'static str); 1], String);

fn board_response(body: String) -> BoardResponse {
    ([(axum::http::header::CONTENT_TYPE, "application/json")], body)
}

type ApiError = (StatusCode, Json<serde_json::Value>);

fn storage_unavailable(error: anyhow::Error) -> ApiError {
    tracing::error!(%error, "storage read failed");
    (
        StatusCode::SERVICE_UNAVAILABLE,
        Json(serde_json::json!({"error": "storage"})),
    )
}

fn serialization_failure(error: serde_json::Error) -> ApiError {
    tracing::error!(%error, "response serialization failed");
    (
        StatusCode::INTERNAL_SERVER_ERROR,
        Json(serde_json::json!({"error": "serialize"})),
    )
}

#[derive(Deserialize)]
pub struct BoardQuery {
    lat: Option<f64>,
    lon: Option<f64>,
    n: Option<usize>,
}

#[derive(Serialize)]
struct BoardRow {
    line: String,
    kind: VehicleKind,
    color: String,
    text_color: String,
    destination: Bilingual,
    in_seconds: Vec<i64>,
    disrupted: bool,
    end_of_service: bool,
    #[serde(skip_serializing_if = "Option::is_none")]
    platform: Option<String>,
    #[serde(skip_serializing_if = "Option::is_none")]
    delay_seconds: Option<i64>,
    #[serde(skip_serializing_if = "Option::is_none")]
    bikes_mechanical: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    bikes_electrical: Option<u32>,
    #[serde(skip_serializing_if = "Option::is_none")]
    docks: Option<u32>,
}

impl From<Departure> for BoardRow {
    fn from(departure: Departure) -> Self {
        let blank = BoardRow {
            line: String::new(),
            kind: VehicleKind::Bus,
            color: String::new(),
            text_color: String::new(),
            destination: Bilingual { fr: String::new(), nl: String::new() },
            in_seconds: Vec::new(),
            disrupted: false,
            end_of_service: false,
            platform: None,
            delay_seconds: None,
            bikes_mechanical: None,
            bikes_electrical: None,
            docks: None,
        };
        match departure {
            Departure::Transit {
                badge,
                destination,
                in_seconds,
                disrupted,
            } => BoardRow {
                line: badge.line,
                kind: badge.kind,
                color: badge.color,
                text_color: badge.text_color,
                destination,
                in_seconds,
                disrupted,
                ..blank
            },
            Departure::EndOfService { badge, message, disrupted } => BoardRow {
                line: badge.line,
                kind: badge.kind,
                color: badge.color,
                text_color: badge.text_color,
                destination: message,
                disrupted,
                end_of_service: true,
                ..blank
            },
            Departure::Train {
                badge,
                destination,
                in_seconds,
                platform,
                delay_seconds,
                delayed,
            } => BoardRow {
                line: badge.line,
                kind: badge.kind,
                color: badge.color,
                text_color: badge.text_color,
                destination: Bilingual { fr: destination.clone(), nl: destination },
                in_seconds: vec![in_seconds],
                disrupted: delayed,
                platform,
                delay_seconds,
                ..blank
            },
            Departure::Stand { bikes, bikes_mechanical, bikes_electrical, docks } => BoardRow {
                line: departure::STAND_LINE.into(),
                kind: VehicleKind::Bike,
                color: departure::STAND_COLOR.into(),
                text_color: departure::STAND_TEXT_COLOR.into(),
                destination: departure::stand_summary(bikes, docks),
                bikes_mechanical: bikes_mechanical.map(u32::from),
                bikes_electrical: bikes_electrical.map(u32::from),
                docks: Some(u32::from(docks)),
                ..blank
            },
        }
    }
}

#[derive(Serialize)]
struct BoardStop {
    id: String,
    name: Bilingual,
    distance_m: Option<u32>,
    rows: Vec<BoardRow>,
}

#[derive(Serialize)]
struct Board {
    data_age_seconds: Option<i64>,
    stale: bool,
    stops: Vec<BoardStop>,
}

#[derive(Deserialize)]
struct LineQuery {
    line: Option<String>,
    to: Option<String>,
    dest: Option<String>,
}

#[derive(Serialize)]
struct StripStopOut {
    id: String,
    name: Bilingual,
    you: bool,
    vehicle_after_m: Option<u32>,
}

#[derive(Serialize)]
struct DisruptionOut {
    en: Option<String>,
    fr: Option<String>,
    nl: Option<String>,
}

#[derive(Serialize)]
struct LineDetail {
    line: String,
    kind: VehicleKind,
    color: String,
    text_color: String,
    destination: Bilingual,
    strip: Vec<StripStopOut>,
    in_seconds: Vec<i64>,
    disruption: Option<DisruptionOut>,
    #[serde(skip_serializing_if = "Option::is_none")]
    shape: Option<Vec<(f32, f32)>>,
}

pub fn fold(s: &str) -> String {
    s.chars()
        .map(|c| match c {
            'à' | 'â' | 'ä' | 'À' | 'Â' | 'Ä' => 'A',
            'ç' | 'Ç' => 'C',
            'è' | 'é' | 'ê' | 'ë' | 'È' | 'É' | 'Ê' | 'Ë' => 'E',
            'î' | 'ï' | 'Î' | 'Ï' => 'I',
            'ô' | 'ö' | 'Ô' | 'Ö' => 'O',
            'ù' | 'û' | 'ü' | 'Ù' | 'Û' | 'Ü' => 'U',
            other => other.to_ascii_uppercase(),
        })
        .collect()
}

pub fn collapse_dupes(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    let mut prev: Option<char> = None;
    for c in s.chars() {
        if prev != Some(c) {
            out.push(c);
        }
        prev = Some(c);
    }
    out
}

#[derive(Deserialize)]
struct SearchQuery {
    q: Option<String>,
}

#[derive(Serialize)]
struct SearchHit {
    id: String,
    name: Bilingual,
    lines: usize,
    #[serde(skip_serializing_if = "Option::is_none")]
    direction: Option<Bilingual>,
}

async fn search(
    State(state): State<Arc<AppState>>,
    Query(q): Query<SearchQuery>,
) -> Result<Json<Vec<SearchHit>>, (StatusCode, Json<serde_json::Value>)> {
    let hits = state
        .store
        .search_stops(q.q.as_deref().unwrap_or(""))
        .await
        .map_err(storage_unavailable)?
        .into_iter()
        .map(|hit| SearchHit {
            id: hit.id,
            name: hit.name,
            lines: hit.line_count,
            direction: hit.direction,
        })
        .collect();
    Ok(Json(hits))
}

async fn healthz(
    State(state): State<Arc<AppState>>,
) -> Result<Json<crate::storage::StorageHealth>, (StatusCode, Json<serde_json::Value>)> {
    state
        .store
        .health()
        .await
        .map(Json)
        .map_err(storage_unavailable)
}

async fn readyz(
    State(state): State<Arc<AppState>>,
) -> Result<
    (StatusCode, Json<crate::storage::StorageHealth>),
    (StatusCode, Json<serde_json::Value>),
> {
    let health = state.store.health().await.map_err(storage_unavailable)?;
    let status = if health.ready {
        StatusCode::OK
    } else {
        StatusCode::SERVICE_UNAVAILABLE
    };
    Ok((status, Json(health)))
}

async fn config() -> impl axum::response::IntoResponse {
    (
        [(axum::http::header::CONTENT_TYPE, "text/html; charset=utf-8")],
        include_str!("../assets/config.html"),
    )
}

pub fn router(state: Arc<AppState>) -> Router {
    Router::new()
        .route("/v1/transit", get(transit))
        .route("/v1/trains", get(trains))
        .route("/v1/villo", get(villo))
        .route("/v1/line", get(line_detail))
        .route("/v1/stops/search", get(search))
        .route("/v1/stops/board", get(stops_board))
        .route("/healthz", get(healthz))
        .route("/readyz", get(readyz))
        .route("/config", get(config))
        .layer(
            TraceLayer::new_for_http()
                .make_span_with(|req: &axum::http::Request<axum::body::Body>| {
                    tracing::info_span!(
                        "request",
                        method = %req.method(),
                        path = %req.uri().path(),
                        query = req.uri().query().unwrap_or("")
                    )
                })
                .on_request(())
                .on_response(
                    |res: &axum::http::Response<axum::body::Body>,
                     latency: std::time::Duration,
                     _span: &tracing::Span| {
                        tracing::info!(
                            status = res.status().as_u16(),
                            latency_ms = latency.as_millis() as u64,
                            "handled"
                        );
                    },
                )
                .on_failure(DefaultOnFailure::new().level(tracing::Level::WARN)),
        )
        .with_state(state)
}

async fn transit(
    State(state): State<Arc<AppState>>,
    Query(q): Query<BoardQuery>,
) -> Result<BoardResponse, (StatusCode, Json<serde_json::Value>)> {
    let now = Utc::now();
    let requested = q.n.unwrap_or(6).min(10);
    let board = match (q.lat, q.lon) {
        (Some(lat), Some(lon)) => state.store.transit_board(lat, lon, requested, now).await,
        _ => state.store.transit_board(0.0, 0.0, 0, now).await,
    }
    .map_err(storage_unavailable)?;
    if requested > 0 && q.lat.is_some() && q.lon.is_some() && board.stops.is_empty() {
        return Err((
            StatusCode::SERVICE_UNAVAILABLE,
            Json(serde_json::json!({"error": "warming up"})),
        ));
    }
    let stops = board
        .stops
        .into_iter()
        .map(|stop| BoardStop {
            id: stop.id,
            name: stop.name,
            distance_m: Some(stop.distance_m),
            rows: stop.departures.into_iter().map(Into::into).collect(),
        })
        .collect();
    let body = serde_json::to_string(&Board {
        data_age_seconds: board.data_age_seconds,
        stale: board.stale,
        stops,
    })
    .map_err(serialization_failure)?;
    Ok(board_response(body))
}

#[derive(Deserialize)]
struct StopsBoardQuery {
    ids: Option<String>,
}

async fn stops_board(
    State(state): State<Arc<AppState>>,
    Query(q): Query<StopsBoardQuery>,
) -> Result<BoardResponse, (StatusCode, Json<serde_json::Value>)> {
    let now = Utc::now();
    let ids: Vec<String> = q
        .ids
        .unwrap_or_default()
        .split(',')
        .map(str::trim)
        .filter(|s| !s.is_empty())
        .map(crate::bmc::normalize_id)
        .take(24)
        .collect();
    let board = state.store.stops_board(ids, now).await.map_err(storage_unavailable)?;
    if board.data_age_seconds.is_none() {
        return Err((
            StatusCode::SERVICE_UNAVAILABLE,
            Json(serde_json::json!({"error": "warming up"})),
        ));
    }
    let mut stops: Vec<BoardStop> = board
        .stops
        .into_iter()
        .map(|stop| BoardStop {
            id: stop.id,
            name: stop.name,
            distance_m: Some(0),
            rows: stop.departures.into_iter().map(Into::into).collect(),
        })
        .collect();
    stops.sort_by_key(|stop| {
        stop.rows
            .iter()
            .filter_map(|row| row.in_seconds.first().copied())
            .next()
            .unwrap_or(i64::MAX)
    });
    let body = serde_json::to_string(&Board {
        data_age_seconds: board.data_age_seconds,
        stale: board.stale,
        stops,
    })
    .map_err(serialization_failure)?;
    Ok(board_response(body))
}

async fn trains(
    State(state): State<Arc<AppState>>,
    Query(q): Query<BoardQuery>,
) -> Result<BoardResponse, (StatusCode, Json<serde_json::Value>)> {
    let stations_ready = state
        .store
        .has_feed_head("irail", "stations")
        .await
        .map_err(storage_unavailable)?;
    if !stations_ready {
        return Err((
            StatusCode::SERVICE_UNAVAILABLE,
            Json(serde_json::json!({"error": "warming up"})),
        ));
    }
    let stops = match (q.lat, q.lon) {
        (Some(lat), Some(lon)) => {
            let board = state
                .store
                .nearby_trains(lat, lon, Utc::now())
                .await
                .map_err(storage_unavailable)?;
            if !board.refresh_station_ids.is_empty() {
                let _ = state
                    .train_refresh
                    .try_send(board.refresh_station_ids);
            }
            train_pseudo_stop(board.stops)
        }
        _ => Vec::new(),
    };
    let body = serde_json::to_string(&Board {
        data_age_seconds: None,
        stale: false,
        stops,
    })
    .map_err(serialization_failure)?;
    Ok(board_response(body))
}

async fn villo(
    State(state): State<Arc<AppState>>,
    Query(q): Query<BoardQuery>,
) -> Result<BoardResponse, (StatusCode, Json<serde_json::Value>)> {
    let board = state
        .store
        .villo_board(q.lat.zip(q.lon), Utc::now())
        .await
        .map_err(storage_unavailable)?;
    if !board.discovery_ready {
        return Err((
            StatusCode::SERVICE_UNAVAILABLE,
            Json(serde_json::json!({"error": "warming up"})),
        ));
    }
    let stale = board.data_age_seconds.map(|age| age > 90).unwrap_or(true);
    let stops = board
        .stops
        .into_iter()
        .map(|station| BoardStop {
            id: format!("VILLO:{}", station.id),
            name: station.name,
            distance_m: Some(station.distance_m as u32),
            rows: vec![Departure::Stand {
                bikes: station.bikes,
                bikes_mechanical: station.mechanical,
                bikes_electrical: station.electrical,
                docks: station.docks,
            }
            .into()],
        })
        .collect();
    let body = serde_json::to_string(&Board {
        data_age_seconds: board.data_age_seconds,
        stale,
        stops,
    })
    .map_err(serialization_failure)?;
    Ok(board_response(body))
}

fn short_station_id(id: &str) -> &str {
    let tail = id.rsplit('.').next().unwrap_or(id);
    let stripped = tail.trim_start_matches('0');
    if stripped.is_empty() { tail } else { stripped }
}

fn train_pseudo_stop(stations: Vec<crate::storage::TrainStopData>) -> Vec<BoardStop> {
    stations
        .into_iter()
        .map(|station| {
            let mut rows: Vec<BoardRow> = station
                .departures
                .into_iter()
                .filter(|departure| !departure.destination.trim().is_empty())
                .map(|departure| {
                    Departure::Train {
                        badge: Badge {
                            line: departure.label,
                            kind: VehicleKind::Train,
                            color: departure::TRAIN_COLOR.into(),
                            text_color: departure::TRAIN_TEXT_COLOR.into(),
                        },
                        destination: departure.destination,
                        in_seconds: departure.in_seconds,
                        platform: departure.platform,
                        delay_seconds: departure.delay_seconds,
                        delayed: departure.delayed,
                    }
                    .into()
                })
                .collect();
            rows.sort_by_key(|row| row.in_seconds.first().copied().unwrap_or(i64::MAX));
            BoardStop {
                id: format!("SNCB:{}", short_station_id(&station.id)),
                name: Bilingual {
                    fr: station.name.clone(),
                    nl: station.name,
                },
                distance_m: Some(station.distance_m as u32),
                rows,
            }
        })
        .collect()
}

async fn line_detail(
    State(state): State<Arc<AppState>>,
    Query(q): Query<LineQuery>,
) -> Result<Json<LineDetail>, (StatusCode, Json<serde_json::Value>)> {
    let line = q.line.as_deref().unwrap_or_default();
    let to = q.to.as_deref().unwrap_or_default();
    let detail = match state
        .store
        .line_detail(line, to, q.dest.as_deref(), Utc::now())
        .await
    {
        Ok(Some(detail)) => detail,
        Ok(None) => {
            tracing::warn!(line, stop = to, "line detail 404");
            return Err((
                StatusCode::NOT_FOUND,
                Json(serde_json::json!({"error": "unknown"})),
            ));
        }
        Err(error)
            if error
                .downcast_ref::<crate::storage::LineDetailUnavailable>()
                .is_some() =>
        {
            return Err((
                StatusCode::SERVICE_UNAVAILABLE,
                Json(serde_json::json!({"error": "warming up"})),
            ));
        }
        Err(error) => return Err(storage_unavailable(error)),
    };
    Ok(Json(LineDetail {
        line: detail.line,
        kind: detail.kind,
        color: detail.color,
        text_color: detail.text_color,
        destination: detail.destination,
        strip: detail
            .strip
            .into_iter()
            .map(|stop| StripStopOut {
                id: stop.id,
                name: stop.name,
                you: stop.you,
                vehicle_after_m: stop.vehicle_after_m,
            })
            .collect(),
        in_seconds: detail.in_seconds,
        disruption: detail.disruption.map(|disruption| DisruptionOut {
            en: disruption.en,
            fr: disruption.fr,
            nl: disruption.nl,
        }),
        shape: detail.shape,
    }))
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::model::{
        Bilingual, Disruption, EndNotice, LineSequence, Passage, RouteInfo, StopInfo,
        VehicleKind, VehiclePosition,
    };
    use axum::body::Body;
    use axum::http::{Request, StatusCode};
    use http_body_util::BodyExt;
    use tower::util::ServiceExt;
    use parking_lot::RwLock;

    #[derive(Clone)]
    struct ApiFixture {
        app: Arc<crate::state::AppState>,
        _directory: Arc<tempfile::TempDir>,
        stops: Arc<RwLock<Arc<Vec<StopInfo>>>>,
        routes: Arc<RwLock<Arc<std::collections::HashMap<Box<str>, RouteInfo>>>>,
        passages: Arc<RwLock<Arc<std::collections::HashMap<Box<str>, Vec<Passage>>>>>,
        positions: Arc<RwLock<Arc<std::collections::HashMap<Box<str>, Vec<VehiclePosition>>>>>,
        sequences: Arc<RwLock<Arc<std::collections::HashMap<Box<str>, Vec<LineSequence>>>>>,
        shapes: Arc<RwLock<Arc<crate::shapes::ShapeIndex>>>,
        disruptions: Arc<RwLock<Arc<Vec<Disruption>>>>,
        end_notices: Arc<RwLock<Arc<std::collections::HashMap<Box<str>, Vec<EndNotice>>>>>,
        captured_at: Arc<RwLock<Option<chrono::DateTime<Utc>>>>,
    }

    impl ApiFixture {
        fn new(
            directory: Arc<tempfile::TempDir>,
            store: crate::storage::Store,
            train_refresh: tokio::sync::mpsc::Sender<Vec<String>>,
        ) -> Self {
            Self {
                app: crate::state::AppState::new(store, train_refresh),
                _directory: directory,
                stops: Arc::new(RwLock::new(Arc::new(Vec::new()))),
                routes: Arc::new(RwLock::new(Arc::new(std::collections::HashMap::new()))),
                passages: Arc::new(RwLock::new(Arc::new(std::collections::HashMap::new()))),
                positions: Arc::new(RwLock::new(Arc::new(std::collections::HashMap::new()))),
                sequences: Arc::new(RwLock::new(Arc::new(std::collections::HashMap::new()))),
                shapes: Arc::new(RwLock::new(Arc::new(std::collections::HashMap::new()))),
                disruptions: Arc::new(RwLock::new(Arc::new(Vec::new()))),
                end_notices: Arc::new(RwLock::new(Arc::new(std::collections::HashMap::new()))),
                captured_at: Arc::new(RwLock::new(None)),
            }
        }

        fn app(&self) -> Arc<crate::state::AppState> {
            self.app.clone()
        }
    }

    impl std::ops::Deref for ApiFixture {
        type Target = crate::state::AppState;

        fn deref(&self) -> &Self::Target {
            self.app.as_ref()
        }
    }

    trait TestAppState {
        fn app_state(&self) -> Arc<crate::state::AppState>;
    }

    impl TestAppState for ApiFixture {
        fn app_state(&self) -> Arc<crate::state::AppState> {
            self.app()
        }
    }

    impl TestAppState for Arc<crate::state::AppState> {
        fn app_state(&self) -> Arc<crate::state::AppState> {
            self.clone()
        }
    }

    fn frozen_board() -> Board {
        Board {
            data_age_seconds: Some(12),
            stale: false,
            stops: vec![
                BoardStop {
                    id: "6419".into(),
                    name: bl("BOTANIQUE"),
                    distance_m: Some(120),
                    rows: vec![
                        BoardRow {
                            line: "62".into(),
                            kind: VehicleKind::Tram,
                            color: "EA4F80".into(),
                            text_color: "000000".into(),
                            destination: bl("SAINT-LAMBERT"),
                            in_seconds: vec![90, 300],
                            disrupted: true,
                            end_of_service: false,
                            platform: None,
                            delay_seconds: None,
                            bikes_mechanical: None,
                            bikes_electrical: None,
                            docks: None,
                        },
                        BoardRow {
                            line: "10".into(),
                            kind: VehicleKind::Bus,
                            color: "000000".into(),
                            text_color: "FFFFFF".into(),
                            destination: bl("LAST BUS HAS GONE"),
                            in_seconds: Vec::new(),
                            disrupted: false,
                            end_of_service: true,
                            platform: None,
                            delay_seconds: None,
                            bikes_mechanical: None,
                            bikes_electrical: None,
                            docks: None,
                        },
                    ],
                },
                BoardStop {
                    id: "SNCB:8813003".into(),
                    name: bl("BRUXELLES-CENTRAL"),
                    distance_m: Some(320),
                    rows: vec![BoardRow {
                        line: "IC".into(),
                        kind: VehicleKind::Train,
                        color: "1A5FB4".into(),
                        text_color: "FFFFFF".into(),
                        destination: bl("OOSTENDE"),
                        in_seconds: vec![420],
                        disrupted: true,
                        end_of_service: false,
                        platform: Some("3".into()),
                        delay_seconds: Some(120),
                        bikes_mechanical: None,
                        bikes_electrical: None,
                        docks: None,
                    }],
                },
                BoardStop {
                    id: "VILLO:046".into(),
                    name: bl("VILLO FLAGEY"),
                    distance_m: Some(90),
                    rows: vec![BoardRow {
                        line: "V".into(),
                        kind: VehicleKind::Bike,
                        color: "8FF0A4".into(),
                        text_color: "000000".into(),
                        destination: Bilingual {
                            fr: "10 velos \u{b7} 5 places".into(),
                            nl: "10 fietsen \u{b7} 5 plaatsen".into(),
                        },
                        in_seconds: Vec::new(),
                        disrupted: false,
                        end_of_service: false,
                        platform: None,
                        delay_seconds: None,
                        bikes_mechanical: Some(7),
                        bikes_electrical: Some(3),
                        docks: Some(5),
                    }],
                },
            ],
        }
    }

    #[test]
    fn wire_shape_is_frozen() {
        let json = serde_json::to_string(&frozen_board()).unwrap();
        let expected = std::fs::read_to_string("tests/wire-golden.json").unwrap();
        assert_eq!(json.trim(), expected.trim());
    }

    fn bl(s: &str) -> Bilingual {
        Bilingual { fr: s.into(), nl: s.into() }
    }

    async fn empty_state_with_train_queue() -> (
        ApiFixture,
        tokio::sync::mpsc::Receiver<Vec<String>>,
    ) {
        let directory = Arc::new(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 (train_refresh, train_refresh_rx) = tokio::sync::mpsc::channel(64);
        (
            ApiFixture::new(directory, store, train_refresh),
            train_refresh_rx,
        )
    }

    async fn empty_state() -> ApiFixture {
        empty_state_with_train_queue().await.0
    }

    async fn spatial_state() -> (Arc<tempfile::TempDir>, ApiFixture) {
        let directory = Arc::new(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();
        crate::dev_mock::seed(&store).await.unwrap();
        let (train_refresh, _train_refresh_rx) = tokio::sync::mpsc::channel(64);
        let state = ApiFixture::new(directory.clone(), store, train_refresh);
        (directory, state)
    }

    async fn persist_transit_state(state: &ApiFixture) {
        let captured_at = state.captured_at.read().unwrap_or_else(Utc::now);
        let mut stops = state.stops.read().as_ref().clone();
        let passages = state.passages.read().as_ref().clone();
        let end_notices = state.end_notices.read().as_ref().clone();
        let disruptions = state.disruptions.read().as_ref().clone();
        let source_sequences = state.sequences.read().as_ref().clone();
        let positions = state.positions.read().as_ref().clone();
        let shapes = state.shapes.read().as_ref().clone();
        let mut routes = state.routes.read().as_ref().clone();
        let mut referenced_stops = Vec::new();
        for sequences in source_sequences.values() {
            for sequence in sequences {
                referenced_stops.extend(sequence.stops.iter().cloned());
            }
        }
        for rows in positions.values() {
            for position in rows {
                referenced_stops.push(position.terminus.clone());
                referenced_stops.push(position.point_id.clone());
            }
        }
        let mut known_stops =
            stops.iter().map(|stop| stop.id.clone()).collect::<std::collections::HashSet<_>>();
        for id in referenced_stops {
            if known_stops.insert(id.clone()) {
                stops.push(StopInfo {
                    name: bl(&id),
                    id,
                    lat: 0.0,
                    lon: 0.0,
                });
            }
        }
        let mut line_ids = std::collections::BTreeSet::new();
        for rows in passages.values() {
            line_ids.extend(rows.iter().map(|row| row.line.clone()));
        }
        for rows in end_notices.values() {
            line_ids.extend(rows.iter().map(|row| row.line.clone()));
        }
        for disruption in &disruptions {
            line_ids.extend(disruption.lines.iter().cloned());
        }
        line_ids.extend(source_sequences.keys().map(|line| line.to_string()));
        line_ids.extend(positions.keys().map(|line| line.to_string()));
        line_ids.extend(shapes.keys().map(|(line, _)| line.clone()));
        if routes.is_empty() && line_ids.is_empty() {
            line_ids.insert("TEST".into());
        }
        for line in line_ids {
            routes.entry(line.clone().into_boxed_str()).or_insert(RouteInfo {
                line: line.clone(),
                route_id: format!("R{line}"),
                kind: VehicleKind::Bus,
                color: "555555".into(),
                text_color: "FFFFFF".into(),
            });
        }
        let stop_ids =
            stops.iter().map(|stop| stop.id.as_str()).collect::<std::collections::HashSet<_>>();
        let mut sequences = std::collections::HashMap::new();
        for (line, group) in &source_sequences {
            let valid = group
                .iter()
                .filter(|sequence| {
                    routes.contains_key(line.as_ref())
                        && sequence.stops.iter().all(|stop| stop_ids.contains(stop.as_str()))
                })
                .cloned()
                .collect::<Vec<_>>();
            if !valid.is_empty() {
                sequences.insert(line.clone(), valid);
            }
        }
        if sequences.is_empty() {
            let line = routes.keys().min().unwrap().to_string();
            sequences.insert(
                line.clone().into_boxed_str(),
                vec![LineSequence {
                    line,
                    destination: bl("TEST"),
                    stops: vec![stops[0].id.clone()],
                }],
            );
        }
        let raw = ["stop-details", "gtfs", "stops-by-line"]
            .into_iter()
            .map(|scope| crate::provider::RawFeed {
                provider: "bmc",
                feed: "static",
                scope: scope.into(),
                captured_at,
                source_at: None,
                content_type: "application/octet-stream".into(),
                body: scope.as_bytes().to_vec(),
            })
            .collect();
        state
            .store
            .apply_static(crate::storage::StaticBatch {
                raw,
                stops,
                routes,
                sequences,
                shapes,
            })
            .await
            .unwrap();
        state
            .store
            .apply_waiting(crate::storage::WaitingBatch {
                raw: crate::provider::RawFeed {
                    provider: "bmc",
                    feed: "waiting",
                    scope: "".into(),
                    captured_at,
                    source_at: None,
                    content_type: "application/json".into(),
                    body: br#"{"results":[]}"#.to_vec(),
                },
                passages,
                end_notices,
            })
            .await
            .unwrap();
        state
            .store
            .apply_positions(crate::storage::PositionsBatch {
                raw: crate::provider::RawFeed {
                    provider: "bmc",
                    feed: "positions",
                    scope: "".into(),
                    captured_at,
                    source_at: None,
                    content_type: "application/json".into(),
                    body: br#"{"results":[]}"#.to_vec(),
                },
                positions,
            })
            .await
            .unwrap();
        state
            .store
            .apply_disruptions(crate::storage::DisruptionsBatch {
                raw: crate::provider::RawFeed {
                    provider: "bmc",
                    feed: "disruptions",
                    scope: "".into(),
                    captured_at,
                    source_at: None,
                    content_type: "application/json".into(),
                    body: br#"{"results":[]}"#.to_vec(),
                },
                disruptions,
            })
            .await
            .unwrap();
    }

    async fn transit_ready_state() -> ApiFixture {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![StopInfo {
            id: "TRANSIT".into(),
            name: bl("TRANSIT"),
            lat: 0.0,
            lon: 0.0,
        }]);
        persist_transit_state(&state).await;
        state
    }

    #[derive(Clone, Copy)]
    struct VilloFixture {
        id: &'static str,
        name: &'static str,
        lat: f64,
        lon: f64,
        bikes: u16,
        docks: u16,
        mechanical: Option<u16>,
        electrical: Option<u16>,
    }

    async fn persist_villo_state(
        state: &ApiFixture,
        captured_at: chrono::DateTime<Utc>,
        stations: &[VilloFixture],
    ) {
        let information_rows: Vec<_> = stations
            .iter()
            .map(|station| {
                serde_json::json!({
                    "station_id": station.id,
                    "name": station.name,
                    "lat": station.lat,
                    "lon": station.lon,
                })
            })
            .collect();
        let status_rows: Vec<_> = stations
            .iter()
            .map(|station| {
                serde_json::json!({
                    "station_id": station.id,
                    "num_vehicles_available": station.bikes,
                    "num_docks_available": station.docks,
                    "vehicle_types_available": [
                        {"vehicle_type_id": "mechanical", "count": station.mechanical},
                        {"vehicle_type_id": "electrical", "count": station.electrical},
                    ],
                })
            })
            .collect();
        let information_body =
            serde_json::to_vec(&serde_json::json!({"data": {"stations": information_rows}}))
                .unwrap();
        let status_body =
            serde_json::to_vec(&serde_json::json!({"data": {"stations": status_rows}})).unwrap();
        let discovery_body = br#"{"data":{"feeds":[
          {"name":"station_information","url":"https://villo.test/information"},
          {"name":"station_status","url":"https://villo.test/status"}
        ]}}"#;
        let raw = |feed: &'static str, scope: &str, body: Vec<u8>| crate::provider::RawFeed {
            provider: "villo",
            feed,
            scope: scope.into(),
            captured_at,
            source_at: None,
            content_type: "application/json".into(),
            body,
        };
        state
            .store
            .apply_villo_discovery(crate::storage::VilloDiscoveryBatch {
                raw: vec![
                    raw("discovery", "gbfs", discovery_body.to_vec()),
                    raw(
                        "discovery",
                        "station_information",
                        information_body,
                    ),
                ],
                status_url: "https://villo.test/status".into(),
                stations: stations
                    .iter()
                    .map(|station| {
                        (
                            station.id.into(),
                            crate::villo::VilloStationMeta {
                                name: station.name.into(),
                                lat: station.lat,
                                lon: station.lon,
                            },
                        )
                    })
                    .collect(),
            })
            .await
            .unwrap();
        state
            .store
            .apply_villo_status(crate::storage::VilloStatusBatch {
                raw: raw("status", "", status_body),
                status: stations
                    .iter()
                    .map(|station| {
                        (
                            station.id.into(),
                            crate::villo::VilloStationCounts {
                                bikes: station.bikes,
                                docks: station.docks,
                                bikes_mechanical: station.mechanical,
                                bikes_electrical: station.electrical,
                            },
                        )
                    })
                    .collect(),
            })
            .await
            .unwrap();
    }

    async fn persist_train_state(
        state: &ApiFixture,
        captured_at: chrono::DateTime<Utc>,
        destination: &str,
    ) {
        state
            .store
            .apply_train_stations(crate::storage::TrainStationsBatch {
                raw: crate::provider::RawFeed {
                    provider: "irail",
                    feed: "stations",
                    scope: "".into(),
                    captured_at,
                    source_at: None,
                    content_type: "application/json".into(),
                    body: br#"{"station":[]}"#.to_vec(),
                },
                stations: vec![crate::irail::TrainStation {
                    id: "BE.NMBS.008812005".into(),
                    name: "Brussel-Noord/Bruxelles-Nord".into(),
                    lat: 50.8535,
                    lon: 4.3661,
                }],
            })
            .await
            .unwrap();
        state
            .store
            .apply_train_departures(crate::storage::TrainDeparturesBatch {
                raw: crate::provider::RawFeed {
                    provider: "irail",
                    feed: "liveboard",
                    scope: "BE.NMBS.008812005".into(),
                    captured_at,
                    source_at: None,
                    content_type: "application/json".into(),
                    body: br#"{"departures":[]}"#.to_vec(),
                },
                station_id: "BE.NMBS.008812005".into(),
                departures: vec![crate::irail::TrainDeparture {
                    label: "IC".into(),
                    destination: destination.into(),
                    in_seconds: 300,
                    delayed: true,
                    platform: Some("3".into()),
                    delay_seconds: Some(120),
                }],
            })
            .await
            .unwrap();
    }

    async fn seeded() -> ApiFixture {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "8021".into(), name: bl("MERODE"), lat: 50.84, lon: 4.40 },
            StopInfo { id: "6419".into(), name: bl("BOTANIQUE"), lat: 50.8535, lon: 4.3661 },
            StopInfo { id: "1057".into(), name: bl("ARTS-LOI"), lat: 50.8467, lon: 4.3676 },
        ]);
        Arc::make_mut(&mut *state.routes.write()).insert(
            "62".into(),
            RouteInfo { line: "62".into(), route_id: "R62".into(), kind: VehicleKind::Tram, color: "EA4F80".into(), text_color: "000000".into() },
        );
        Arc::make_mut(&mut *state.passages.write()).insert(
            "6419".into(),
            vec![
                Passage { line: "62".into(), destination: bl("SAINT-LAMBERT"), expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(300) },
                Passage { line: "62".into(), destination: bl("SAINT-LAMBERT"), expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(90) },
                Passage { line: "99".into(), destination: bl("NOWHERE"), expected: chrono::Utc::now().fixed_offset() - chrono::Duration::seconds(120) },
            ],
        );
        Arc::make_mut(&mut *state.passages.write()).insert(
            "8021".into(),
            vec![
                Passage { line: "77".into(), destination: bl("UNKNOWN"), expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(300) },
                Passage { line: "62".into(), destination: bl("MONTGOMERY"), expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(50) },
                Passage { line: "88".into(), destination: bl("CLAMP"), expected: chrono::Utc::now().fixed_offset() - chrono::Duration::seconds(10) },
            ],
        );
        Arc::make_mut(&mut *state.sequences.write()).insert(
            "62".into(),
            vec![LineSequence {
                line: "62".into(),
                destination: bl("SAINT-LAMBERT"),
                stops: vec!["s1".into(), "s2".into(), "6419".into(), "s4".into()],
            }],
        );
        Arc::make_mut(&mut *state.positions.write()).insert(
            "62".into(),
            vec![VehiclePosition { terminus: "s4".into(), point_id: "s1".into(), distance_m: 300 }],
        );
        Arc::make_mut(&mut *state.routes.write()).insert(
            "10".into(),
            RouteInfo { line: "10".into(), route_id: "R10".into(), kind: VehicleKind::Bus, color: "000000".into(), text_color: "FFFFFF".into() },
        );
        Arc::make_mut(&mut *state.sequences.write()).insert(
            "10".into(),
            vec![LineSequence {
                line: "10".into(),
                destination: bl("TERVUREN"),
                stops: vec!["6419".into()],
            }],
        );
        *state.disruptions.write() = Arc::new(vec![Disruption {
            en: None,
            fr: None,
            nl: None,
            priority: 1,
            lines: vec!["62".into()],
            points: vec!["6419".into()],
        }]);
        Arc::make_mut(&mut *state.end_notices.write()).insert(
            "8021".into(),
            vec![EndNotice { line: "62".into(), message: bl("SUPPRESSED") }],
        );
        Arc::make_mut(&mut *state.end_notices.write()).insert(
            "1057".into(),
            vec![EndNotice { line: "99".into(), message: bl("LAST TRAM DEPARTED") }],
        );
        persist_villo_state(
            &state,
            Utc::now(),
            &[VilloFixture {
                id: "1",
                name: "LEOPOLD II",
                lat: 50.7000,
                lon: 4.2000,
                bikes: 6,
                docks: 9,
                mechanical: Some(4),
                electrical: Some(2),
            }],
        )
        .await;
        *state.captured_at.write() = Some(chrono::Utc::now());
        persist_transit_state(&state).await;
        persist_train_state(&state, Utc::now(), "LEUVEN").await;
        state
    }

    async fn get_raw(
        state: impl TestAppState,
        uri: &str,
    ) -> (StatusCode, String, String) {
        let resp = router(state.app_state())
            .oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
            .await
            .unwrap();
        let status = resp.status();
        let content_type = resp
            .headers()
            .get(axum::http::header::CONTENT_TYPE)
            .map(|v| v.to_str().unwrap().to_string())
            .unwrap_or_default();
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        (status, content_type, String::from_utf8(bytes.to_vec()).unwrap())
    }

    async fn get_json(state: impl TestAppState, uri: &str) -> (StatusCode, serde_json::Value) {
        let resp = router(state.app_state())
            .oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap())
            .await
            .unwrap();
        let status = resp.status();
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        (status, serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null))
    }

    #[tokio::test]
    async fn transit_endpoint_reads_the_file_backed_store() {
        let (_directory, state) = spatial_state().await;
        *state.stops.write() = Arc::new(Vec::new());
        *state.passages.write() = Arc::new(std::collections::HashMap::new());
        *state.end_notices.write() = Arc::new(std::collections::HashMap::new());
        let (status, value) =
            get_json(state, "/v1/transit?lat=50.80&lon=4.30&n=1").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(value["stops"][0]["id"], "S0");
        assert_eq!(value["stops"][0]["rows"].as_array().unwrap().len(), 3);
    }

    #[tokio::test]
    async fn search_endpoint_reads_the_file_backed_store() {
        let (_directory, state) = spatial_state().await;
        *state.stops.write() = Arc::new(Vec::new());
        *state.passages.write() = Arc::new(std::collections::HashMap::new());
        *state.sequences.write() = Arc::new(std::collections::HashMap::new());
        let (status, value) = get_json(state, "/v1/stops/search?q=mock%20active%200").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(value[0]["id"], "S0");
        assert_eq!(value[0]["lines"], 2);
        assert!(value[0]["direction"].is_object());
    }

    #[tokio::test]
    async fn line_endpoint_reads_the_file_backed_store() {
        let seeded = seeded().await;
        let (train_refresh_tx, _train_refresh_rx) = tokio::sync::mpsc::channel(64);
        let state = crate::state::AppState::new(seeded.store.clone(), train_refresh_tx);
        let (status, value) = get_json(state, "/v1/line?line=62&to=6419").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(value["line"], "62");
        assert_eq!(value["strip"][2]["id"], "6419");
    }

    #[tokio::test]
    async fn villo_endpoint_reads_the_file_backed_store() {
        let (_directory, state) = spatial_state().await;
        let (status, value) =
            get_json(state, "/v1/villo?lat=50.8317&lon=4.3856").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(value["stops"][0]["id"], "VILLO:1");
        assert_eq!(value["stops"][0]["rows"][0]["bikes_mechanical"], 31);
    }

    #[tokio::test]
    async fn transit_endpoint_serves_only_transit_stops() {
        let (status, v) = get_json(seeded().await, "/v1/transit?lat=50.8535&lon=4.3661&n=6").await;
        assert_eq!(status, StatusCode::OK);
        for stop in v["stops"].as_array().unwrap() {
            let id = stop["id"].as_str().unwrap();
            assert!(!id.starts_with("SNCB:"));
            assert!(!id.starts_with("VILLO:"));
        }
    }

    #[tokio::test]
    async fn villo_endpoint_serves_only_villo_stops() {
        let (status, v) = get_json(seeded().await, "/v1/villo?lat=50.7000&lon=4.2000").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert!(!stops.is_empty());
        for stop in stops {
            assert!(stop["id"].as_str().unwrap().starts_with("VILLO:"));
        }
    }

    #[tokio::test]
    async fn board_endpoint_is_gone() {
        let (status, _) = get_json(seeded().await, "/v1/board?lat=50.8535&lon=4.3661").await;
        assert_eq!(status, StatusCode::NOT_FOUND);
    }


    #[tokio::test]
    async fn transit_endpoint_returns_nearest_with_grouped_rows() {
        let (status, v) = get_json(seeded().await, "/v1/transit?lat=50.8535&lon=4.3661&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let stop = &v["stops"][0];
        assert_eq!(stop["id"], "6419");
        let rows = stop["rows"].as_array().unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0]["line"], "62");
        assert_eq!(rows[0]["kind"], "tram");
        assert_eq!(rows[0]["color"], "EA4F80");
        let secs = rows[0]["in_seconds"].as_array().unwrap();
        assert_eq!(secs.len(), 2);
        assert!(secs[0].as_i64().unwrap() < secs[1].as_i64().unwrap());
        assert_eq!(v["stale"], false);
    }

    #[tokio::test]
    async fn transit_endpoint_uses_the_earliest_destination_per_line() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "STOP".into(), name: bl("STOP"), lat: 0.0, lon: 0.0 }]);
        Arc::make_mut(&mut *state.passages.write()).insert(
            "STOP".into(),
            vec![
                Passage {
                    line: "62".into(),
                    destination: bl("LATER"),
                    expected: chrono::Utc::now().fixed_offset()
                        + chrono::Duration::seconds(300),
                },
                Passage {
                    line: "62".into(),
                    destination: bl("EARLIER"),
                    expected: chrono::Utc::now().fixed_offset()
                        + chrono::Duration::seconds(60),
                },
            ],
        );
        persist_transit_state(&state).await;
        let (status, value) = get_json(state, "/v1/transit?lat=0&lon=0&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = value["stops"][0]["rows"].as_array().unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0]["destination"]["fr"], "EARLIER");
        assert_eq!(rows[0]["in_seconds"].as_array().unwrap().len(), 2);
    }

    #[tokio::test]
    async fn transit_endpoint_orders_rows_clamps_negative_and_defaults_unknown_lines() {
        let (status, v) = get_json(seeded().await, "/v1/transit?lat=50.84&lon=4.40&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = v["stops"][0]["rows"].as_array().unwrap();
        assert_eq!(rows.len(), 3);
        assert_eq!(rows[0]["line"], "88");
        assert_eq!(rows[0]["in_seconds"], serde_json::json!([0]));
        assert_eq!(rows[1]["line"], "62");
        assert_eq!(rows[1]["kind"], "tram");
        assert_eq!(rows[1]["color"], "EA4F80");
        assert_eq!(rows[2]["line"], "77");
        assert_eq!(rows[2]["kind"], "bus");
        assert_eq!(rows[2]["color"], "555555");
        assert_eq!(rows[2]["text_color"], "FFFFFF");
    }

    #[tokio::test]
    async fn transit_endpoint_warms_up_as_503() {
        let state = empty_state().await;
        let (status, v) = get_json(state, "/v1/transit?lat=1&lon=1").await;
        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(v["error"], "warming up");
    }

    fn push_passage(state: &ApiFixture, stop: &str, line: &str, seconds: i64) {
        Arc::make_mut(&mut *state.passages.write())
            .entry(stop.into())
            .or_default()
            .push(Passage {
                line: line.into(),
                destination: bl("DEST"),
                expected: chrono::Utc::now().fixed_offset()
                    + chrono::Duration::seconds(seconds),
            });
    }

    #[tokio::test]
    async fn stops_board_returns_rows_for_selected_ids() {
        let (status, v) = get_json(seeded().await, "/v1/stops/board?ids=6419").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert_eq!(stops.len(), 1);
        assert_eq!(stops[0]["id"], "6419");
        assert_eq!(stops[0]["distance_m"], 0);
        assert!(!stops[0]["rows"].as_array().unwrap().is_empty());
    }

    #[tokio::test]
    async fn stops_board_skips_unknown_ids_and_normalizes() {
        let (status, v) = get_json(seeded().await, "/v1/stops/board?ids=06419,%20NOPE,").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert_eq!(stops.len(), 1);
        assert_eq!(stops[0]["id"], "6419");
    }

    #[tokio::test]
    async fn stops_board_ignores_villo_train_and_inactive_places() {
        let state = seeded().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "8021".into(), name: bl("MERODE"), lat: 50.84, lon: 4.40 },
            StopInfo { id: "6419".into(), name: bl("BOTANIQUE"), lat: 50.8535, lon: 4.3661 },
        ]);
        persist_transit_state(&state).await;
        let (status, v) =
            get_json(state, "/v1/stops/board?ids=1,BE.NMBS.008812005,1057,6419").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert_eq!(stops.len(), 1);
        assert_eq!(stops[0]["id"], "6419");
    }

    #[tokio::test]
    async fn stops_board_orders_by_next_departure() {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "LATE".into(), name: bl("LATE"), lat: 0.0, lon: 0.0 },
            StopInfo { id: "SOON".into(), name: bl("SOON"), lat: 0.0, lon: 0.0 },
            StopInfo { id: "SILENT".into(), name: bl("SILENT"), lat: 0.0, lon: 0.0 },
        ]);
        push_passage(&state, "LATE", "62", 900);
        push_passage(&state, "SOON", "62", 60);
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/stops/board?ids=LATE,SOON,SILENT").await;
        assert_eq!(status, StatusCode::OK);
        let ids: Vec<&str> = v["stops"].as_array().unwrap().iter()
            .map(|s| s["id"].as_str().unwrap()).collect();
        assert_eq!(ids, vec!["SOON", "LATE", "SILENT"]);
    }

    #[tokio::test]
    async fn stops_board_caps_ids_at_twenty_four() {
        let mut ids: Vec<String> = (0..30).map(|i| format!("X{i}")).collect();
        ids.push("6419".into());
        let url = format!("/v1/stops/board?ids={}", ids.join(","));
        let (status, v) = get_json(seeded().await, &url).await;
        assert_eq!(status, StatusCode::OK);
        assert!(v["stops"].as_array().unwrap().is_empty());
    }

    #[tokio::test]
    async fn stops_board_warms_up_as_503() {
        let (status, v) = get_json(empty_state().await, "/v1/stops/board?ids=6419").await;
        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(v["error"], "warming up");
    }

    #[tokio::test]
    async fn stops_board_with_no_ids_serves_an_empty_board() {
        let (status, v) = get_json(seeded().await, "/v1/stops/board?ids=").await;
        assert_eq!(status, StatusCode::OK);
        assert!(v["stops"].as_array().unwrap().is_empty());
    }

    #[tokio::test]
    async fn trains_endpoint_warms_up_without_a_station_head() {
        let state = transit_ready_state().await;
        let (ready_status, _) = get_json(state.clone(), "/readyz").await;
        assert_eq!(ready_status, StatusCode::OK);

        let (status, value) =
            get_json(state, "/v1/trains?lat=50.8535&lon=4.3661").await;

        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(value, serde_json::json!({"error": "warming up"}));
    }

    #[tokio::test]
    async fn villo_endpoint_warms_up_without_a_discovery_head() {
        let state = transit_ready_state().await;
        let (ready_status, _) = get_json(state.clone(), "/readyz").await;
        assert_eq!(ready_status, StatusCode::OK);

        let (status, value) =
            get_json(state, "/v1/villo?lat=50.8535&lon=4.3661").await;

        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(value, serde_json::json!({"error": "warming up"}));
    }

    #[test]
    fn fold_strips_diacritics_and_case() {
        assert_eq!(fold("Mérode"), "MERODE");
        assert_eq!(fold("Groot-Bijgaarden"), "GROOT-BIJGAARDEN");
        assert_eq!(fold("Père Éternel àâäçèêëîïôöùûü"), "PERE ETERNEL AAACEEEIIOOUUU");
    }

    #[test]
    fn collapse_dupes_folds_consecutive_repeats() {
        assert_eq!(collapse_dupes("HERRMANN"), "HERMAN");
        assert_eq!(collapse_dupes("HERMANN"), "HERMAN");
        assert_eq!(collapse_dupes("HERMMAN"), "HERMAN");
        assert_eq!(collapse_dupes("ANNEESSENS"), "ANESENS");
    }

    #[tokio::test]
    async fn search_forgives_double_letter_misspellings() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "HD".into(), name: bl("HERRMANN-DEBROUX"), lat: 0.0, lon: 0.0 }]);
        persist_transit_state(&state).await;
        let (_, v) = get_json(state.clone(), "/v1/stops/search?q=hermann").await;
        assert_eq!(v[0]["id"], "HD");
        let (_, v) = get_json(state, "/v1/stops/search?q=hermman").await;
        assert_eq!(v[0]["id"], "HD");
    }

    #[tokio::test]
    async fn search_ranks_exact_match_before_collapsed_match() {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "HD".into(), name: bl("HERRMANN-DEBROUX"), lat: 0.0, lon: 0.0 },
            StopInfo { id: "HM".into(), name: bl("HERMAN"), lat: 0.0, lon: 0.0 },
        ]);
        persist_transit_state(&state).await;
        let (_, v) = get_json(state, "/v1/stops/search?q=herman").await;
        let ids: Vec<&str> = v.as_array().unwrap().iter().map(|h| h["id"].as_str().unwrap()).collect();
        assert_eq!(ids, vec!["HM", "HD"]);
    }

    #[tokio::test]
    async fn search_folds_accents_in_query_and_in_name() {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "A".into(), name: bl("MERODE"), lat: 0.0, lon: 0.0 },
            StopInfo { id: "B".into(), name: bl("Général Anneessens"), lat: 0.0, lon: 0.0 },
        ]);
        persist_transit_state(&state).await;
        let (_, v) = get_json(state.clone(), "/v1/stops/search?q=m%C3%A9rode").await;
        assert_eq!(v[0]["id"], "A");
        let (_, v) = get_json(state, "/v1/stops/search?q=general").await;
        assert_eq!(v[0]["id"], "B");
    }

    #[tokio::test]
    async fn search_collapse_does_not_merge_distinct_names() {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "ANN".into(), name: bl("ANNEESSENS"), lat: 0.0, lon: 0.0 },
            StopInfo { id: "AND".into(), name: bl("ANDERLECHT"), lat: 0.0, lon: 0.0 },
        ]);
        persist_transit_state(&state).await;
        let (_, v) = get_json(state, "/v1/stops/search?q=anesens").await;
        let ids: Vec<&str> = v.as_array().unwrap().iter().map(|h| h["id"].as_str().unwrap()).collect();
        assert_eq!(ids, vec!["ANN"]);
    }

    #[tokio::test]
    async fn search_result_cap_stays_twenty_with_collapsed_fill() {
        let state = empty_state().await;
        let stops: Vec<StopInfo> = (0..25)
            .map(|i| StopInfo {
                id: format!("S{i}"),
                name: bl(&format!("HERRMANN-DEBROUX {i}")),
                lat: 0.0,
                lon: 0.0,
            })
            .collect();
        *state.stops.write() = Arc::new(stops);
        persist_transit_state(&state).await;
        let (_, v) = get_json(state, "/v1/stops/search?q=hermann").await;
        assert_eq!(v.as_array().unwrap().len(), 20);
    }

    #[tokio::test]
    async fn search_matches_folded_names_and_ids() {
        let state = seeded().await;
        let (status, v) = get_json(state.clone(), "/v1/stops/search?q=m%C3%A9ro").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v[0]["id"], "8021");
        let (_, v) = get_json(state.clone(), "/v1/stops/search?q=64").await;
        assert_eq!(v[0]["id"], "6419");
        let (_, v) = get_json(state, "/v1/stops/search?q=").await;
        assert_eq!(v.as_array().unwrap().len(), 0);
    }

    #[tokio::test]
    async fn search_reports_distinct_line_count() {
        let state = seeded().await;
        let (status, v) = get_json(state.clone(), "/v1/stops/search?q=m%C3%A9ro").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v[0]["id"], "8021");
        assert_eq!(v[0]["lines"], 3);
        let (_, v) = get_json(state.clone(), "/v1/stops/search?q=64").await;
        assert_eq!(v[0]["id"], "6419");
        assert_eq!(v[0]["lines"], 2);
        let (_, v) = get_json(state, "/v1/stops/search?q=arts").await;
        assert_eq!(v[0]["id"], "1057");
        assert_eq!(v[0]["lines"], 0);
    }

    #[tokio::test]
    async fn search_hit_carries_direction_when_stop_is_in_a_sequence() {
        let state = seeded().await;
        let (status, v) = get_json(state, "/v1/stops/search?q=64").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v[0]["id"], "6419");
        assert!(v[0]["direction"]["fr"].is_string());
    }

    #[tokio::test]
    async fn search_hit_omits_direction_when_stop_is_in_no_sequence() {
        let state = seeded().await;
        let (status, v) = get_json(state, "/v1/stops/search?q=m%C3%A9ro").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v[0]["id"], "8021");
        assert!(v[0].get("direction").is_none());
    }

    #[tokio::test]
    async fn healthz_reports_storage_status_before_readiness() {
        let (status, value) = get_json(empty_state().await, "/healthz").await;

        assert_eq!(status, StatusCode::OK);
        assert_eq!(value["ready"], false);
        assert_eq!(value["schema_version"], 5);
        assert_eq!(value["feeds"].as_array().unwrap().len(), 8);
        assert!(
            value["database_bytes"].as_u64().unwrap()
                + value["wal_bytes"].as_u64().unwrap()
                > 0
        );
        assert!(value["last_pruned_at"].is_null());
        assert!(value["last_checkpoint_at"].is_null());
    }

    #[tokio::test]
    async fn readyz_rejects_an_empty_store() {
        let (status, value) = get_json(empty_state().await, "/readyz").await;

        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(value["ready"], false);
        assert_eq!(value["schema_version"], 5);
    }

    #[tokio::test]
    async fn readyz_accepts_a_committed_static_projection() {
        let (status, value) = get_json(seeded().await, "/readyz").await;

        assert_eq!(status, StatusCode::OK);
        assert_eq!(value["ready"], true);
        assert!(value["feeds"]
            .as_array()
            .unwrap()
            .iter()
            .any(|feed| feed["provider"] == "bmc"
                && feed["feed"] == "static"
                && feed["current_snapshot_id"].is_number()));
    }

    #[tokio::test]
    async fn readyz_rejects_a_noncurrent_schema() {
        let state = seeded().await;
        let connection = deadpool_sqlite::rusqlite::Connection::open(
            state._directory.path().join("stibble.sqlite"),
        )
        .unwrap();
        connection
            .execute("UPDATE schema_meta SET version = 1", [])
            .unwrap();
        drop(connection);

        let (status, value) = get_json(state, "/readyz").await;

        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(value["ready"], false);
        assert_eq!(value["schema_version"], 1);
    }

    #[tokio::test]
    async fn readyz_rejects_a_storage_query_failure() {
        let state = seeded().await;
        let connection = deadpool_sqlite::rusqlite::Connection::open(
            state._directory.path().join("stibble.sqlite"),
        )
        .unwrap();
        connection.execute("DROP TABLE schema_meta", []).unwrap();
        drop(connection);

        let (status, value) = get_json(state, "/readyz").await;

        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(value, serde_json::json!({"error": "storage"}));
    }

    #[tokio::test]
    async fn storage_query_failure_is_503_and_reported_by_health() {
        let state = seeded().await;
        let connection = deadpool_sqlite::rusqlite::Connection::open(
            state._directory.path().join("stibble.sqlite"),
        )
        .unwrap();
        connection.execute("DROP TABLE place_search", []).unwrap();
        drop(connection);

        let (status, value) =
            get_json(state.clone(), "/v1/stops/search?q=botanique").await;
        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(value, serde_json::json!({"error": "storage"}));

        let (status, health) = get_json(state, "/healthz").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(health["last_read_failure"]["stage"], "query");
        assert_eq!(
            health["last_read_failure"]["operation"],
            "search_stops"
        );
        assert!(health["last_read_failure"]["message"]
            .as_str()
            .unwrap()
            .contains("place_search"));
        assert!(health["last_read_failure"]["failed_at"].is_number());
    }

    #[tokio::test]
    async fn storage_pool_timeout_is_503_and_reported_by_health() {
        let state = empty_state().await;
        let (ready_tx, mut ready_rx) = tokio::sync::mpsc::channel(8);
        let release = Arc::new(std::sync::Barrier::new(9));
        let mut holders = Vec::new();
        for _ in 0..8 {
            let store = state.store.clone();
            let ready = ready_tx.clone();
            let release = Arc::clone(&release);
            holders.push(tokio::spawn(async move {
                store
                    .hold_connection_for_test(ready, release)
                    .await
                    .unwrap();
            }));
        }
        drop(ready_tx);
        for _ in 0..8 {
            ready_rx.recv().await.unwrap();
        }

        let (status, value) =
            get_json(state.clone(), "/v1/stops/search?q=botanique").await;
        let release_wait = Arc::clone(&release);
        tokio::task::spawn_blocking(move || release_wait.wait())
            .await
            .unwrap();
        for holder in holders {
            holder.await.unwrap();
        }
        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(value, serde_json::json!({"error": "storage"}));

        let (status, health) = get_json(state, "/healthz").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(health["last_read_failure"]["stage"], "acquire");
        assert_eq!(
            health["last_read_failure"]["operation"],
            "search_stops"
        );
        assert!(health["last_read_failure"]["failed_at"].is_number());
    }

    #[tokio::test]
    async fn config_serves_html() {
        let state = seeded().await;
        let resp = router(state.app())
            .oneshot(Request::builder().uri("/config").body(Body::empty()).unwrap())
            .await
            .unwrap();
        assert_eq!(resp.status(), StatusCode::OK);
        let ct = resp.headers()["content-type"].to_str().unwrap().to_string();
        assert!(ct.starts_with("text/html"));
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let page = String::from_utf8(bytes.to_vec()).unwrap();
        assert!(page.contains("pebblejs://close#"));
        assert!(page.contains("/v1/stops/search"));
    }

    #[tokio::test]
    async fn config_page_groups_search_results() {
        let state = seeded().await;
        let resp = router(state.app())
            .oneshot(Request::builder().uri("/config").body(Body::empty()).unwrap())
            .await
            .unwrap();
        let bytes = resp.into_body().collect().await.unwrap().to_bytes();
        let page = String::from_utf8(bytes.to_vec()).unwrap();
        assert!(page.contains("function groupHits"));
        assert!(page.contains("dirCount"));
    }

    #[tokio::test]
    async fn search_matches_zero_padded_ids() {
        let (_, v) = get_json(seeded().await, "/v1/stops/search?q=06419").await;
        assert_eq!(v[0]["id"], "6419");
    }

    #[tokio::test]
    async fn transit_endpoint_flags_disrupted_row() {
        let (status, v) = get_json(seeded().await, "/v1/transit?lat=50.8535&lon=4.3661&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = v["stops"][0]["rows"].as_array().unwrap();
        let row = rows.iter().find(|r| r["line"] == "62").unwrap();
        assert_eq!(row["disrupted"], true);
    }

    #[tokio::test]
    async fn transit_endpoint_disruption_with_bare_point_attaches_to_suffixed_stop_row() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "2290B".into(), name: bl("HANKAR"), lat: 0.0, lon: 0.0 }]);
        Arc::make_mut(&mut *state.passages.write()).insert(
            "2290".into(),
            vec![Passage {
                line: "34".into(),
                destination: bl("DELTA"),
                expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(300),
            }],
        );
        *state.disruptions.write() = Arc::new(vec![Disruption {
            en: None,
            fr: None,
            nl: None,
            priority: 1,
            lines: vec!["34".into()],
            points: vec!["2290".into()],
        }]);
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=0&lon=0&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = v["stops"][0]["rows"].as_array().unwrap();
        let row = rows.iter().find(|r| r["line"] == "34").unwrap();
        assert_eq!(row["disrupted"], true);
    }

    #[tokio::test]
    async fn line_detail_bare_point_disruption_wins_over_unrelated_stop_disruption() {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "2290B".into(), name: bl("HANKAR"), lat: 0.0, lon: 0.0 },
            StopInfo { id: "9999".into(), name: bl("OTHER"), lat: 1.0, lon: 1.0 },
        ]);
        Arc::make_mut(&mut *state.routes.write()).insert(
            "34".into(),
            RouteInfo {
                line: "34".into(),
                route_id: "R34".into(),
                kind: VehicleKind::Bus,
                color: "000000".into(),
                text_color: "FFFFFF".into(),
            },
        );
        Arc::make_mut(&mut *state.sequences.write()).insert(
            "34".into(),
            vec![LineSequence { line: "34".into(), destination: bl("DELTA"), stops: vec!["2290B".into()] }],
        );
        *state.disruptions.write() = Arc::new(vec![
            Disruption {
                en: None,
                fr: Some("UNRELATED".into()),
                nl: None,
                priority: 1,
                lines: vec!["34".into()],
                points: vec!["9999".into()],
            },
            Disruption {
                en: Some("HANKAR EN".into()),
                fr: Some("HANKAR FR".into()),
                nl: Some("HANKAR NL".into()),
                priority: 5,
                lines: vec!["34".into()],
                points: vec!["2290".into()],
            },
        ]);
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/line?line=34&to=2290B").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v["disruption"]["en"], "HANKAR EN");
        assert_eq!(v["disruption"]["fr"], "HANKAR FR");
        assert_eq!(v["disruption"]["nl"], "HANKAR NL");
    }

    #[tokio::test]
    async fn transit_endpoint_appends_end_of_service_row() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "1057".into(), name: bl("ARTS-LOI"), lat: 50.8467, lon: 4.3676 }]);
        Arc::make_mut(&mut *state.end_notices.write()).insert(
            "1057".into(),
            vec![EndNotice { line: "99".into(), message: bl("LAST TRAM DEPARTED") }],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=50.8467&lon=4.3676&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = v["stops"][0]["rows"].as_array().unwrap();
        assert_eq!(rows.len(), 1);
        let row = &rows[0];
        assert_eq!(row["line"], "99");
        assert_eq!(row["destination"]["fr"], "LAST TRAM DEPARTED");
        assert_eq!(row["in_seconds"], serde_json::json!([]));
        assert_eq!(row["end_of_service"], true);
    }

    #[tokio::test]
    async fn transit_endpoint_suppresses_end_notice_when_timed_row_exists() {
        let (status, v) = get_json(seeded().await, "/v1/transit?lat=50.84&lon=4.40&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = v["stops"][0]["rows"].as_array().unwrap();
        assert_eq!(rows.len(), 3);
        let count_62 = rows.iter().filter(|r| r["line"] == "62").count();
        assert_eq!(count_62, 1);
        let row = rows.iter().find(|r| r["line"] == "62").unwrap();
        assert_eq!(row["end_of_service"], false);
    }

    #[tokio::test]
    async fn transit_endpoint_defaults_live_fields_without_data() {
        let (status, v) = get_json(seeded().await, "/v1/transit?lat=50.84&lon=4.40&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = v["stops"][0]["rows"].as_array().unwrap();
        let row = rows.iter().find(|r| r["line"] == "88").unwrap();
        assert_eq!(row["disrupted"], false);
        assert_eq!(row["end_of_service"], false);
    }

    #[tokio::test]
    async fn line_returns_strip_you_flag_vehicle_marker_and_in_seconds() {
        let (status, v) = get_json(seeded().await, "/v1/line?line=62&to=6419").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v["line"], "62");
        assert_eq!(v["kind"], "tram");
        assert_eq!(v["color"], "EA4F80");
        assert_eq!(v["text_color"], "000000");
        assert_eq!(v["destination"]["fr"], "SAINT-LAMBERT");
        let strip = v["strip"].as_array().unwrap();
        assert_eq!(strip.len(), 3);
        assert_eq!(strip[0]["id"], "s1");
        assert_eq!(strip[1]["id"], "s2");
        assert_eq!(strip[2]["id"], "6419");
        assert_eq!(strip[2]["name"]["fr"], "BOTANIQUE");
        assert_eq!(strip[0]["you"], false);
        assert_eq!(strip[1]["you"], false);
        assert_eq!(strip[2]["you"], true);
        assert_eq!(strip[0]["vehicle_after_m"], 300);
        assert!(strip[1]["vehicle_after_m"].is_null());
        assert!(strip[2]["vehicle_after_m"].is_null());
        let secs = v["in_seconds"].as_array().unwrap();
        assert_eq!(secs.len(), 2);
        assert!(secs[0].as_i64().unwrap() < secs[1].as_i64().unwrap());
    }

    #[tokio::test]
    async fn line_dest_labels_the_row_but_keeps_all_line_passages() {
        let state = seeded().await;
        Arc::make_mut(&mut *state.passages.write())
            .get_mut("6419")
            .unwrap()
            .push(Passage {
                line: "62".into(),
                destination: Bilingual { fr: "SAINTE-ANNE".into(), nl: "SINT-ANNA".into() },
                expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(40),
            });
        persist_transit_state(&state).await;
        let (status, v) = get_json(state.clone(), "/v1/line?line=62&to=6419&dest=SAINTE-ANNE").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v["destination"]["fr"], "SAINTE-ANNE");
        assert_eq!(v["destination"]["nl"], "SINT-ANNA");
        let secs = v["in_seconds"].as_array().unwrap();
        assert_eq!(secs.len(), 3);
        assert!(secs[0].as_i64().unwrap() <= 40);
        let (status, v) = get_json(state, "/v1/line?line=62&to=6419&dest=SINT-ANNA").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v["destination"]["fr"], "SAINTE-ANNE");
        assert_eq!(v["in_seconds"].as_array().unwrap().len(), 3);
    }

    #[tokio::test]
    async fn line_dest_matching_nothing_falls_back_to_static() {
        let (status, v) = get_json(seeded().await, "/v1/line?line=62&to=6419&dest=NOPE").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v["destination"]["fr"], "SAINT-LAMBERT");
        assert_eq!(v["in_seconds"].as_array().unwrap().len(), 2);
    }

    #[tokio::test]
    async fn line_ignores_empty_dest() {
        let (status, v) = get_json(seeded().await, "/v1/line?line=62&to=6419&dest=").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v["destination"]["fr"], "SAINT-LAMBERT");
        assert_eq!(v["in_seconds"].as_array().unwrap().len(), 2);
    }

    #[tokio::test]
    async fn line_returns_404_on_unknown_line() {
        let (status, v) = get_json(seeded().await, "/v1/line?line=999&to=6419").await;
        assert_eq!(status, StatusCode::NOT_FOUND);
        assert_eq!(v["error"], "unknown");
    }

    #[tokio::test]
    async fn line_returns_404_on_stop_absent_from_every_sequence() {
        let (status, v) = get_json(seeded().await, "/v1/line?line=62&to=9999").await;
        assert_eq!(status, StatusCode::NOT_FOUND);
        assert_eq!(v["error"], "unknown");
    }

    #[tokio::test]
    async fn line_warms_up_as_503() {
        let state = empty_state().await;
        let (status, v) = get_json(state, "/v1/line?line=62&to=6419").await;
        assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE);
        assert_eq!(v["error"], "warming up");
    }

    #[tokio::test]
    async fn line_reports_matching_disruption() {
        let (status, v) = get_json(seeded().await, "/v1/line?line=62&to=6419").await;
        assert_eq!(status, StatusCode::OK);
        assert!(!v["disruption"].is_null());
        assert!(v["disruption"]["fr"].is_null());
    }

    #[tokio::test]
    async fn line_disruption_is_null_when_none_apply() {
        let (status, v) = get_json(seeded().await, "/v1/line?line=10&to=6419").await;
        assert_eq!(status, StatusCode::OK);
        assert!(v["disruption"].is_null());
    }

    #[tokio::test]
    async fn line_carries_shape_sliced_to_the_strip_span() {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "s1".into(), name: bl("A"), lat: 50.80, lon: 4.30 },
            StopInfo { id: "s2".into(), name: bl("B"), lat: 50.81, lon: 4.31 },
            StopInfo { id: "6419".into(), name: bl("C"), lat: 50.82, lon: 4.32 },
            StopInfo { id: "s4".into(), name: bl("D"), lat: 50.83, lon: 4.33 },
        ]);
        Arc::make_mut(&mut *state.routes.write()).insert(
            "62".into(),
            RouteInfo { line: "62".into(), route_id: "R62".into(), kind: VehicleKind::Tram, color: "EA4F80".into(), text_color: "000000".into() },
        );
        Arc::make_mut(&mut *state.sequences.write()).insert(
            "62".into(),
            vec![LineSequence {
                line: "62".into(),
                destination: bl("SAINT-LAMBERT"),
                stops: vec!["s1".into(), "s2".into(), "6419".into(), "s4".into()],
            }],
        );
        Arc::make_mut(&mut *state.shapes.write()).insert(
            ("62".into(), "s4".into()),
            vec![
                (50.79, 4.29),
                (50.80, 4.30),
                (50.81, 4.31),
                (50.82, 4.32),
                (50.83, 4.33),
                (50.84, 4.34),
            ],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/line?line=62&to=6419").await;
        assert_eq!(status, StatusCode::OK);
        let shape = v["shape"].as_array().unwrap();
        assert_eq!(shape.len(), 3);
        assert!((shape[0][0].as_f64().unwrap() - 50.80).abs() < 1e-4);
        assert!((shape.last().unwrap()[0].as_f64().unwrap() - 50.82).abs() < 1e-4);
    }

    #[tokio::test]
    async fn line_serves_full_shape_when_slice_collapses_to_a_point() {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "s1".into(), name: bl("A"), lat: 50.80, lon: 4.30 },
            StopInfo { id: "s2".into(), name: bl("B"), lat: 50.81, lon: 4.31 },
            StopInfo { id: "6419".into(), name: bl("C"), lat: 50.82, lon: 4.32 },
            StopInfo { id: "s4".into(), name: bl("D"), lat: 50.83, lon: 4.33 },
        ]);
        Arc::make_mut(&mut *state.sequences.write()).insert(
            "62".into(),
            vec![LineSequence {
                line: "62".into(),
                destination: bl("SAINT-LAMBERT"),
                stops: vec!["s1".into(), "s2".into(), "6419".into(), "s4".into()],
            }],
        );
        Arc::make_mut(&mut *state.shapes.write()).insert(
            ("62".into(), "s4".into()),
            vec![(50.829, 4.329), (50.831, 4.331)],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/line?line=62&to=6419").await;
        assert_eq!(status, StatusCode::OK);
        let shape = v["shape"].as_array().unwrap();
        assert!(shape.len() >= 2, "degenerate 1-point shape served: {shape:?}");
    }

    #[tokio::test]
    async fn line_omits_shape_when_none_is_stored() {
        let (status, v) = get_json(seeded().await, "/v1/line?line=10&to=6419").await;
        assert_eq!(status, StatusCode::OK);
        assert!(v.get("shape").is_none());
    }

    #[test]
    fn train_pseudo_stop_preserves_wire_rows_and_disrupted_delay() {
        let stops = train_pseudo_stop(vec![crate::storage::TrainStopData {
            id: "BE.NMBS.008812005".into(),
            name: "Brussel-Noord/Bruxelles-Nord".into(),
            distance_m: 42.9,
            departures: vec![
                crate::irail::TrainDeparture {
                    label: "IC".into(),
                    destination: "Leuven".into(),
                    in_seconds: 600,
                    delayed: false,
                    platform: None,
                    delay_seconds: Some(0),
                },
                crate::irail::TrainDeparture {
                    label: "S1".into(),
                    destination: "Gent".into(),
                    in_seconds: 120,
                    delayed: true,
                    platform: Some("3".into()),
                    delay_seconds: Some(120),
                },
                crate::irail::TrainDeparture {
                    label: "P".into(),
                    destination: "".into(),
                    in_seconds: 60,
                    delayed: false,
                    platform: None,
                    delay_seconds: None,
                },
            ],
        }]);
        assert_eq!(stops.len(), 1);
        let stop = &stops[0];
        assert_eq!(stop.id, "SNCB:8812005");
        assert_eq!(stop.name.fr, "Brussel-Noord/Bruxelles-Nord");
        assert_eq!(stop.name.nl, "Brussel-Noord/Bruxelles-Nord");
        assert_eq!(stop.distance_m, Some(42));
        assert_eq!(stop.rows.len(), 2);
        assert_eq!(stop.rows[0].line, "S1");
        assert_eq!(stop.rows[0].kind, VehicleKind::Train);
        assert_eq!(stop.rows[0].color, "1A5FB4");
        assert_eq!(stop.rows[0].text_color, "FFFFFF");
        assert_eq!(stop.rows[0].destination.fr, "Gent");
        assert_eq!(stop.rows[0].in_seconds, vec![120]);
        assert!(stop.rows[0].disrupted);
        assert!(!stop.rows[0].end_of_service);
        assert_eq!(stop.rows[0].platform.as_deref(), Some("3"));
        assert_eq!(stop.rows[0].delay_seconds, Some(120));
        assert_eq!(stop.rows[1].line, "IC");
        assert!(!stop.rows[1].disrupted);
        assert_eq!(stop.rows[1].platform, None);
        assert_eq!(stop.rows[1].delay_seconds, Some(0));
    }

    #[test]
    fn train_pseudo_stop_ids_are_short_and_unique_across_stations() {
        let stops = train_pseudo_stop(vec![
            crate::storage::TrainStopData {
                id: "BE.NMBS.008812005".into(),
                name: "North".into(),
                distance_m: 10.0,
                departures: Vec::new(),
            },
            crate::storage::TrainStopData {
                id: "BE.NMBS.008813003".into(),
                name: "Central".into(),
                distance_m: 20.0,
                departures: Vec::new(),
            },
        ]);
        assert_eq!(stops.len(), 2);
        assert_eq!(stops[0].id, "SNCB:8812005");
        assert_eq!(stops[1].id, "SNCB:8813003");
        assert_ne!(stops[0].id, stops[1].id);
        assert!(stops.iter().all(|stop| stop.id.len() <= 15));
    }

    #[tokio::test]
    async fn villo_endpoint_appends_pseudo_stop_with_counts() {
        let (status, v) = get_json(seeded().await, "/v1/villo?lat=50.70005&lon=4.20005&n=0").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        let stop = stops.iter().find(|s| s["id"] == "VILLO:1").unwrap();
        let rows = stop["rows"].as_array().unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0]["line"], "V");
        assert_eq!(rows[0]["kind"], "bike");
        assert_eq!(rows[0]["color"], "8FF0A4");
        assert_eq!(rows[0]["text_color"], "000000");
        assert_eq!(rows[0]["destination"]["fr"], "6 velos \u{b7} 9 places");
        assert_eq!(rows[0]["destination"]["nl"], "6 fietsen \u{b7} 9 plaatsen");
        assert_eq!(rows[0]["in_seconds"], serde_json::json!([]));
        assert_eq!(rows[0]["disrupted"], false);
        assert_eq!(rows[0]["end_of_service"], false);
        assert_eq!(rows[0]["bikes_mechanical"], 4);
        assert_eq!(rows[0]["bikes_electrical"], 2);
        assert_eq!(rows[0]["docks"], 9);
    }

    #[tokio::test]
    async fn villo_endpoint_row_omits_bike_kinds_when_feed_lacks_the_breakdown() {
        let state = seeded().await;
        persist_villo_state(
            &state,
            Utc::now(),
            &[VilloFixture {
                id: "1",
                name: "LEOPOLD II",
                lat: 50.7000,
                lon: 4.2000,
                bikes: 6,
                docks: 9,
                mechanical: None,
                electrical: None,
            }],
        )
        .await;
        let (status, v) = get_json(state, "/v1/villo?lat=50.70005&lon=4.20005&n=0").await;
        assert_eq!(status, StatusCode::OK);
        let stop = v["stops"].as_array().unwrap().iter().find(|s| s["id"] == "VILLO:1").unwrap();
        let row = &stop["rows"][0];
        assert!(row.get("bikes_mechanical").is_none());
        assert!(row.get("bikes_electrical").is_none());
        assert_eq!(row["docks"], 9);
        assert_eq!(row["destination"]["fr"], "6 velos \u{b7} 9 places");
    }

    #[tokio::test]
    async fn transit_endpoint_stib_rows_omit_the_new_optional_fields() {
        let (status, v) = get_json(seeded().await, "/v1/transit?lat=50.8535&lon=4.3661&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = v["stops"][0]["rows"].as_array().unwrap();
        for row in rows {
            assert!(row.get("platform").is_none());
            assert!(row.get("delay_seconds").is_none());
            assert!(row.get("bikes_mechanical").is_none());
            assert!(row.get("bikes_electrical").is_none());
            assert!(row.get("docks").is_none());
        }
    }

    #[tokio::test]
    async fn transit_endpoint_row_shape_is_additive_over_v1() {
        let (status, v) = get_json(seeded().await, "/v1/transit?lat=50.8535&lon=4.3661&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let row = &v["stops"][0]["rows"][0];
        for key in [
            "line",
            "kind",
            "color",
            "text_color",
            "destination",
            "in_seconds",
            "disrupted",
            "end_of_service",
        ] {
            assert!(row.get(key).is_some(), "missing {key}");
        }
    }

    #[tokio::test]
    async fn villo_endpoint_present_just_inside_the_radius() {
        let (status, v) = get_json(seeded().await, "/v1/villo?lat=50.70405&lon=4.2000&n=0").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert!(stops.iter().any(|s| s["id"] == "VILLO:1"));
    }

    #[tokio::test]
    async fn villo_endpoint_absent_just_outside_the_radius() {
        let (status, v) = get_json(seeded().await, "/v1/villo?lat=50.70495&lon=4.2000&n=0").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert!(stops.iter().all(|s| s["id"] != "VILLO:1"));
    }

    #[tokio::test]
    async fn villo_endpoint_includes_every_stand_in_range() {
        let state = empty_state().await;
        persist_villo_state(
            &state,
            Utc::now(),
            &[
                VilloFixture {
                    id: "1",
                    name: "LEOPOLD II",
                    lat: 50.7020,
                    lon: 4.2000,
                    bikes: 6,
                    docks: 9,
                    mechanical: None,
                    electrical: None,
                },
                VilloFixture {
                    id: "2",
                    name: "PARVIS",
                    lat: 50.7045,
                    lon: 4.2000,
                    bikes: 3,
                    docks: 4,
                    mechanical: None,
                    electrical: None,
                },
            ],
        )
        .await;
        let (status, v) = get_json(state, "/v1/villo?lat=50.7045&lon=4.2000&n=0").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert!(stops.iter().any(|s| s["id"] == "VILLO:1"));
        assert!(stops.iter().any(|s| s["id"] == "VILLO:2"));
    }

    #[tokio::test]
    async fn villo_endpoint_ignores_n_and_returns_every_stand_in_range() {
        let state = seeded().await;
        let (s1, _, b1) =
            get_raw(state.clone(), "/v1/villo?lat=50.7035&lon=4.2000&n=1").await;
        assert_eq!(s1, StatusCode::OK);
        let (s2, _, b2) =
            get_raw(state, "/v1/villo?lat=50.7035&lon=4.2000&n=6").await;
        assert_eq!(s2, StatusCode::OK);
        assert_eq!(b2, b1);
    }

    #[tokio::test]
    async fn transit_endpoint_distinguishes_n_values() {
        let state = seeded().await;
        let (s1, _, b1) = get_raw(state.clone(), "/v1/transit?lat=50.8535&lon=4.3661&n=1").await;
        assert_eq!(s1, StatusCode::OK);
        let (s2, _, b2) = get_raw(state.clone(), "/v1/transit?lat=50.8535&lon=4.3661&n=3").await;
        assert_eq!(s2, StatusCode::OK);
        assert_ne!(b2, b1);
    }

    #[tokio::test]
    async fn transit_serves_independently_of_the_train_projection() {
        let state = seeded().await;
        let (status, value) =
            get_json(state, "/v1/transit?lat=50.8535&lon=4.3661&n=6").await;
        assert_eq!(status, StatusCode::OK);
        assert!(!value["stops"].as_array().unwrap().is_empty());
    }

    #[tokio::test]
    async fn trains_serves_committed_rows_when_the_refresh_queue_is_closed() {
        let state = seeded().await;
        persist_train_state(
            &state,
            Utc::now() - chrono::Duration::seconds(31),
            "LEUVEN",
        )
        .await;
        let (status, value) =
            get_json(state, "/v1/trains?lat=50.8535&lon=4.3661").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(value["stops"][0]["rows"][0]["destination"]["fr"], "LEUVEN");
    }

    #[tokio::test]
    async fn trains_reads_each_commit_and_only_queues_stale_station_ids() {
        let (state, mut refreshes) = empty_state_with_train_queue().await;
        persist_train_state(
            &state,
            Utc::now() - chrono::Duration::seconds(31),
            "LEUVEN",
        )
        .await;
        let (first_status, first_type, first_body) =
            get_raw(state.clone(), "/v1/trains?lat=50.8535&lon=4.3661").await;
        assert_eq!(first_status, StatusCode::OK);
        assert!(first_type.starts_with("application/json"));
        let first: serde_json::Value = serde_json::from_str(&first_body).unwrap();
        assert_eq!(first["stops"][0]["rows"][0]["destination"]["fr"], "LEUVEN");
        let refresh = tokio::time::timeout(
            std::time::Duration::from_secs(1),
            refreshes.recv(),
        )
        .await
        .unwrap()
        .unwrap();
        assert_eq!(refresh, vec!["BE.NMBS.008812005".to_string()]);

        persist_train_state(&state, Utc::now(), "GENT").await;
        let (second_status, second_type, second_body) =
            get_raw(state, "/v1/trains?lat=50.8535&lon=4.3661").await;
        assert_eq!(second_status, StatusCode::OK);
        assert_eq!(second_type, first_type);
        assert_ne!(second_body, first_body);
        let second: serde_json::Value = serde_json::from_str(&second_body).unwrap();
        assert_eq!(second["stops"][0]["rows"][0]["destination"]["fr"], "GENT");
        assert!(refreshes.try_recv().is_err());
    }

    #[tokio::test]
    async fn transit_endpoint_stale_tracks_waiting_snapshot() {
        let state = seeded().await;
        *state.captured_at.write() = Some(Utc::now() - chrono::Duration::hours(2));
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=50.8535&lon=4.3661&n=1").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v["stale"], true);
    }

    #[tokio::test]
    async fn trains_endpoint_never_reports_bmc_staleness() {
        let state = seeded().await;
        let (status, v) = get_json(state, "/v1/trains?lat=50.8535&lon=4.3661").await;
        assert_eq!(status, StatusCode::OK);
        assert!(v["data_age_seconds"].is_null());
        assert_eq!(v["stale"], false);
    }

    #[tokio::test]
    async fn villo_endpoint_stale_tracks_status_snapshot_not_waiting_state() {
        let state = seeded().await;
        let (status, v) =
            get_json(state, "/v1/villo?lat=50.7000&lon=4.2000&n=0").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v["stale"], false);
        assert!(v["data_age_seconds"].as_i64().unwrap() < 5);
    }

    #[tokio::test]
    async fn villo_endpoint_marks_old_status_snapshot_stale() {
        let state = seeded().await;
        persist_villo_state(
            &state,
            Utc::now() - chrono::Duration::seconds(91),
            &[VilloFixture {
                id: "1",
                name: "LEOPOLD II",
                lat: 50.7000,
                lon: 4.2000,
                bikes: 6,
                docks: 9,
                mechanical: Some(4),
                electrical: Some(2),
            }],
        )
        .await;
        let (status, value) =
            get_json(state, "/v1/villo?lat=50.7000&lon=4.2000").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(value["stale"], true);
        assert!(value["data_age_seconds"].as_i64().unwrap() >= 91);
    }

    #[tokio::test]
    async fn villo_endpoint_orders_every_stand_in_range_by_distance() {
        let state = empty_state().await;
        persist_villo_state(
            &state,
            Utc::now(),
            &[
                VilloFixture {
                    id: "2",
                    name: "NEAR TWO",
                    lat: 50.8520,
                    lon: 4.3500,
                    bikes: 2,
                    docks: 8,
                    mechanical: Some(2),
                    electrical: Some(0),
                },
                VilloFixture {
                    id: "1",
                    name: "NEAR ONE",
                    lat: 50.8500,
                    lon: 4.3500,
                    bikes: 4,
                    docks: 6,
                    mechanical: Some(3),
                    electrical: Some(1),
                },
                VilloFixture {
                    id: "3",
                    name: "TOO FAR",
                    lat: 50.8600,
                    lon: 4.3500,
                    bikes: 9,
                    docks: 1,
                    mechanical: Some(9),
                    electrical: Some(0),
                },
            ],
        )
        .await;

        let (status, value) = get_json(state, "/v1/villo?lat=50.8500&lon=4.3500").await;
        assert_eq!(status, StatusCode::OK);
        let stops = value["stops"].as_array().unwrap();
        assert_eq!(stops.len(), 2);
        assert_eq!(stops[0]["name"]["fr"], "NEAR ONE");
        assert_eq!(stops[1]["name"]["fr"], "NEAR TWO");
    }

    #[tokio::test]
    async fn transit_endpoint_nearest_skips_silent_stop_for_farther_stop_with_passages() {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "SILENT".into(), name: bl("SILENT"), lat: 0.0, lon: 0.0 },
            StopInfo { id: "LOUD".into(), name: bl("LOUD"), lat: 0.0, lon: 0.01 },
        ]);
        Arc::make_mut(&mut *state.passages.write()).insert(
            "LOUD".into(),
            vec![Passage {
                line: "62".into(),
                destination: bl("SAINT-LAMBERT"),
                expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(300),
            }],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=0&lon=0&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert_eq!(stops.len(), 1);
        assert_eq!(stops[0]["id"], "LOUD");
    }

    #[tokio::test]
    async fn transit_endpoint_nearest_keeps_stop_with_only_end_notice() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "NOTICE_ONLY".into(), name: bl("NOTICE_ONLY"), lat: 0.0, lon: 0.0 }]);
        Arc::make_mut(&mut *state.end_notices.write()).insert(
            "NOTICE_ONLY".into(),
            vec![EndNotice { line: "62".into(), message: bl("LAST TRAM DEPARTED") }],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=0&lon=0&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert_eq!(stops.len(), 1);
        assert_eq!(stops[0]["id"], "NOTICE_ONLY");
    }

    #[tokio::test]
    async fn transit_endpoint_selects_silent_stop_without_waiting_data() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "SILENT".into(), name: bl("SILENT"), lat: 0.0, lon: 0.0 }]);
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=0&lon=0&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert_eq!(stops.len(), 1);
        assert_eq!(stops[0]["id"], "SILENT");
    }

    #[tokio::test]
    async fn transit_endpoint_nearest_drops_silent_duplicate_of_selected_data_twin() {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "6463".into(), name: bl("SQUARE"), lat: 0.0, lon: 0.0 },
            StopInfo { id: "9931".into(), name: bl("SQUARE"), lat: 0.0, lon: 0.0001 },
        ]);
        Arc::make_mut(&mut *state.passages.write()).insert(
            "6463".into(),
            vec![Passage {
                line: "62".into(),
                destination: bl("SAINT-LAMBERT"),
                expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(300),
            }],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=0&lon=0&n=2").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert_eq!(stops.len(), 1);
        assert_eq!(stops[0]["id"], "6463");
    }

    #[tokio::test]
    async fn transit_endpoint_nearest_keeps_both_same_named_stops_with_data() {
        let state = empty_state().await;
        *state.stops.write() = Arc::new(vec![
            StopInfo { id: "6463A".into(), name: bl("SQUARE"), lat: 0.0, lon: 0.0 },
            StopInfo { id: "9931".into(), name: bl("SQUARE"), lat: 0.0, lon: 0.0001 },
        ]);
        Arc::make_mut(&mut *state.passages.write()).insert(
            "6463A".into(),
            vec![Passage {
                line: "62".into(),
                destination: bl("SAINT-LAMBERT"),
                expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(300),
            }],
        );
        Arc::make_mut(&mut *state.passages.write()).insert(
            "9931".into(),
            vec![Passage {
                line: "71".into(),
                destination: bl("DELTA"),
                expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(400),
            }],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=0&lon=0&n=2").await;
        assert_eq!(status, StatusCode::OK);
        let stops = v["stops"].as_array().unwrap();
        assert_eq!(stops.len(), 2);
        let ids: Vec<&str> = stops.iter().map(|s| s["id"].as_str().unwrap()).collect();
        assert!(ids.contains(&"6463A"));
        assert!(ids.contains(&"9931"));
    }

    #[tokio::test]
    async fn line_returns_404_on_pseudo_id() {
        let (status, v) = get_json(
            seeded().await,
            "/v1/line?line=SNCB:8812005&to=SNCB:8812005",
        )
        .await;
        assert_eq!(status, StatusCode::NOT_FOUND);
        assert_eq!(v["error"], "unknown");
    }

    #[tokio::test]
    async fn transit_endpoint_suffixed_stop_matches_bare_realtime_passages() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "2290B".into(), name: bl("HANKAR"), lat: 0.0, lon: 0.0 }]);
        Arc::make_mut(&mut *state.passages.write()).insert(
            "2290".into(),
            vec![Passage {
                line: "34".into(),
                destination: bl("DELTA"),
                expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(300),
            }],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=0&lon=0&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = v["stops"][0]["rows"].as_array().unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0]["line"], "34");
    }

    #[tokio::test]
    async fn transit_endpoint_suffixed_stop_matches_bare_end_notice() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "2291B".into(), name: bl("HANKAR"), lat: 0.0, lon: 0.0 }]);
        Arc::make_mut(&mut *state.end_notices.write()).insert(
            "2291".into(),
            vec![EndNotice { line: "34".into(), message: bl("LAST BUS DEPARTED") }],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=0&lon=0&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = v["stops"][0]["rows"].as_array().unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0]["end_of_service"], true);
        assert_eq!(rows[0]["destination"]["fr"], "LAST BUS DEPARTED");
    }

    #[tokio::test]
    async fn transit_endpoint_exact_match_stop_unaffected_by_fallback() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "2290".into(), name: bl("BARE"), lat: 0.0, lon: 0.0 }]);
        Arc::make_mut(&mut *state.passages.write()).insert(
            "2290".into(),
            vec![Passage {
                line: "34".into(),
                destination: bl("DELTA"),
                expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(300),
            }],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=0&lon=0&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = v["stops"][0]["rows"].as_array().unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0]["line"], "34");
    }

    #[tokio::test]
    async fn transit_endpoint_suffixed_stop_with_dead_bare_sibling_stays_silent() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "7968B".into(), name: bl("PETILLON"), lat: 0.0, lon: 0.0 }]);
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/transit?lat=0&lon=0&n=1").await;
        assert_eq!(status, StatusCode::OK);
        let rows = v["stops"][0]["rows"].as_array().unwrap();
        assert!(rows.is_empty());
    }

    #[tokio::test]
    async fn search_reports_lines_via_bare_fallback_for_suffixed_stop() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "2290B".into(), name: bl("HANKAR"), lat: 0.0, lon: 0.0 }]);
        Arc::make_mut(&mut *state.passages.write()).insert(
            "2290".into(),
            vec![Passage {
                line: "34".into(),
                destination: bl("DELTA"),
                expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(300),
            }],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/stops/search?q=hankar").await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(v[0]["id"], "2290B");
        assert_eq!(v[0]["lines"], 1);
    }

    #[tokio::test]
    async fn line_detail_in_seconds_matches_bare_realtime_passages_for_suffixed_stop() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "2290B".into(), name: bl("HANKAR"), lat: 0.0, lon: 0.0 }]);
        Arc::make_mut(&mut *state.routes.write()).insert(
            "34".into(),
            RouteInfo {
                line: "34".into(),
                route_id: "R34".into(),
                kind: VehicleKind::Bus,
                color: "000000".into(),
                text_color: "FFFFFF".into(),
            },
        );
        Arc::make_mut(&mut *state.sequences.write()).insert(
            "34".into(),
            vec![LineSequence {
                line: "34".into(),
                destination: bl("DELTA"),
                stops: vec!["2290B".into()],
            }],
        );
        Arc::make_mut(&mut *state.passages.write()).insert(
            "2290".into(),
            vec![Passage {
                line: "34".into(),
                destination: bl("DELTA"),
                expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(300),
            }],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/line?line=34&to=2290B").await;
        assert_eq!(status, StatusCode::OK);
        let secs = v["in_seconds"].as_array().unwrap();
        assert_eq!(secs.len(), 1);
    }

    #[tokio::test]
    async fn line_detail_falls_back_to_all_line_passages_when_diverted() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "6155F".into(), name: bl("STOP"), lat: 0.0, lon: 0.0 }]);
        Arc::make_mut(&mut *state.sequences.write()).insert(
            "81".into(),
            vec![LineSequence {
                line: "81".into(),
                destination: bl("MARIUS RENARD"),
                stops: vec!["6155F".into()],
            }],
        );
        Arc::make_mut(&mut *state.passages.write()).insert(
            "6155F".into(),
            vec![
                Passage { line: "81".into(), destination: bl("TRINITE"), expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(120) },
                Passage { line: "81".into(), destination: bl("FLAGEY"), expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(300) },
            ],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/line?line=81&to=6155F").await;
        assert_eq!(status, StatusCode::OK);
        let secs = v["in_seconds"].as_array().unwrap();
        assert_eq!(secs.len(), 2);
        assert!(secs[0].as_i64().unwrap() < secs[1].as_i64().unwrap());
    }

    #[tokio::test]
    async fn line_detail_keeps_all_same_line_passages_when_destinations_vary() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "6155F".into(), name: bl("STOP"), lat: 0.0, lon: 0.0 }]);
        Arc::make_mut(&mut *state.sequences.write()).insert(
            "81".into(),
            vec![LineSequence {
                line: "81".into(),
                destination: bl("MARIUS RENARD"),
                stops: vec!["6155F".into()],
            }],
        );
        Arc::make_mut(&mut *state.passages.write()).insert(
            "6155F".into(),
            vec![
                Passage { line: "81".into(), destination: bl("MARIUS RENARD"), expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(120) },
                Passage { line: "81".into(), destination: bl("TRINITE"), expected: chrono::Utc::now().fixed_offset() + chrono::Duration::seconds(60) },
            ],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/line?line=81&to=6155F").await;
        assert_eq!(status, StatusCode::OK);
        let secs = v["in_seconds"].as_array().unwrap();
        assert_eq!(secs.len(), 2);
        assert!(secs[0].as_i64().unwrap() <= 60);
    }

    #[tokio::test]
    async fn line_detail_in_seconds_empty_when_no_passages_for_line() {
        let state = empty_state().await;
        *state.stops.write() =
            Arc::new(vec![StopInfo { id: "6155F".into(), name: bl("STOP"), lat: 0.0, lon: 0.0 }]);
        Arc::make_mut(&mut *state.sequences.write()).insert(
            "81".into(),
            vec![LineSequence {
                line: "81".into(),
                destination: bl("MARIUS RENARD"),
                stops: vec!["6155F".into()],
            }],
        );
        persist_transit_state(&state).await;
        let (status, v) = get_json(state, "/v1/line?line=81&to=6155F").await;
        assert_eq!(status, StatusCode::OK);
        let secs = v["in_seconds"].as_array().unwrap();
        assert!(secs.is_empty());
    }
}