use chrono::{DateTime, Duration, Utc};
use deadpool_sqlite::rusqlite::{params, Connection, OpenFlags};
use std::{
    collections::HashMap,
    net::TcpListener,
    path::{Path, PathBuf},
    process::{Child, Command, Stdio},
    time::{Duration as StdDuration, Instant},
};
use stibble_server::{
    api, dev_mock, irail,
    model::{
        Bilingual, Disruption, EndNotice, LineSequence, Passage, RouteInfo, StopInfo, VehicleKind,
        VehiclePosition,
    },
    provider::{FeedStage, RawFeed},
    shapes::ShapeIndex,
    state,
    storage::{
        DisruptionsBatch, PositionsBatch, StaticBatch, Store, TrainDeparturesBatch,
        TrainStationsBatch, VilloDiscoveryBatch, VilloStatusBatch, WaitingBatch,
    },
    villo,
};
use tempfile::TempDir;

struct FixtureStore {
    _dir: TempDir,
    path: PathBuf,
    extension: PathBuf,
    store: Store,
}

impl FixtureStore {
    async fn new() -> Self {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("stibble.sqlite");
        let extension = PathBuf::from(std::env::var("SPATIALITE_EXTENSION").unwrap());
        let store = Store::open(&path, &extension).await.unwrap();
        Self {
            _dir: dir,
            path,
            extension,
            store,
        }
    }

    async fn reopen(&self) -> Store {
        Store::open(&self.path, &self.extension).await.unwrap()
    }

    fn router_with_train_queue(
        &self,
        train_refresh_tx: tokio::sync::mpsc::Sender<Vec<String>>,
    ) -> axum::Router {
        api::router(state::AppState::new(
            self.store.clone(),
            train_refresh_tx,
        ))
    }
}


struct MockServerProcess {
    child: Child,
    base_url: String,
}

impl Drop for MockServerProcess {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
    }
}

async fn start_mock_server(path: &Path, extension: &Path) -> MockServerProcess {
    start_mock_server_with_refresh_interval(path, extension, None).await
}

async fn start_mock_server_with_refresh_interval(
    path: &Path,
    extension: &Path,
    refresh_interval_ms: Option<u64>,
) -> MockServerProcess {
    let socket = TcpListener::bind("127.0.0.1:0").unwrap();
    let address = socket.local_addr().unwrap();
    drop(socket);
    let mut command = Command::new(env!("CARGO_BIN_EXE_mock_server"));
    command
        .env("STIBBLE_DB", path)
        .env("SPATIALITE_EXTENSION", extension)
        .env("BIND_ADDR", address.to_string())
        .env("BMC_BASE", "http://127.0.0.1:1")
        .env_remove("BMC_PARTNER_KEY")
        .env("HTTP_PROXY", "http://127.0.0.1:1")
        .env("HTTPS_PROXY", "http://127.0.0.1:1")
        .env("NO_PROXY", "127.0.0.1,localhost")
        .env("RUST_LOG", "error")
        .env_remove("STIBBLE_MOCK_REFRESH_INTERVAL_MS")
        .stdout(Stdio::null());
    if let Some(refresh_interval_ms) = refresh_interval_ms {
        command.env(
            "STIBBLE_MOCK_REFRESH_INTERVAL_MS",
            refresh_interval_ms.to_string(),
        );
    }
    let mut child = command.spawn().unwrap();
    let base_url = format!("http://{address}");
    let client = reqwest::Client::new();
    let deadline = Instant::now() + StdDuration::from_secs(60);
    loop {
        if let Some(status) = child.try_wait().unwrap() {
            panic!("mock server exited before readiness with {status}");
        }
        if let Ok(response) = client.get(format!("{base_url}/readyz")).send().await {
            if response.status() == reqwest::StatusCode::OK {
                break;
            }
        }
        if Instant::now() >= deadline {
            let _ = child.kill();
            let _ = child.wait();
            panic!("mock server did not become ready");
        }
        tokio::time::sleep(StdDuration::from_millis(25)).await;
    }
    MockServerProcess { child, base_url }
}

async fn assert_mock_endpoint(server: &MockServerProcess, path: &str) {
    let response = reqwest::get(format!("{}{path}", server.base_url))
        .await
        .unwrap();
    let status = response.status();
    let body = response.text().await.unwrap();
    assert_eq!(status, reqwest::StatusCode::OK, "{path}: {body}");
}

struct UncommittedWriter {
    child: Child,
    marker: PathBuf,
}

impl Drop for UncommittedWriter {
    fn drop(&mut self) {
        let _ = self.child.kill();
        let _ = self.child.wait();
        let _ = std::fs::remove_file(&self.marker);
    }
}

fn start_uncommitted_writer(path: &Path) -> UncommittedWriter {
    let marker = path.with_extension("writer-ready");
    let _ = std::fs::remove_file(&marker);
    let mut child = Command::new(std::env::current_exe().unwrap())
        .arg("--ignored")
        .arg("--exact")
        .arg("uncommitted_writer_process")
        .env("STIBBLE_CRASH_DB", path)
        .env("STIBBLE_CRASH_READY", &marker)
        .stdout(Stdio::null())
        .spawn()
        .unwrap();
    let deadline = Instant::now() + StdDuration::from_secs(10);
    while !marker.exists() {
        if let Some(status) = child.try_wait().unwrap() {
            panic!("uncommitted writer exited before its transaction was ready with {status}");
        }
        if Instant::now() >= deadline {
            let _ = child.kill();
            let _ = child.wait();
            panic!("uncommitted writer did not become ready");
        }
        std::thread::sleep(StdDuration::from_millis(10));
    }
    UncommittedWriter { child, marker }
}

#[test]
#[ignore]
fn uncommitted_writer_process() {
    let (Some(path), Some(marker)) = (
        std::env::var_os("STIBBLE_CRASH_DB"),
        std::env::var_os("STIBBLE_CRASH_READY"),
    ) else {
        return;
    };
    let connection = Connection::open(path).unwrap();
    connection
        .execute_batch(
            "PRAGMA journal_mode=WAL;
             PRAGMA cache_size=1;
             PRAGMA cache_spill=ON;
             BEGIN IMMEDIATE;
             DELETE FROM current_passage;
             DELETE FROM current_villo_status;
             DELETE FROM current_train_departure;
             UPDATE provider_feed SET current_snapshot_id = NULL;",
        )
        .unwrap();
    std::fs::write(marker, b"ready").unwrap();
    loop {
        std::thread::sleep(StdDuration::from_secs(60));
    }
}

fn scalar(path: &Path, sql: &str) -> i64 {
    Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)
        .unwrap()
        .query_row(sql, [], |row| row.get(0))
        .unwrap()
}

fn blob(path: &Path, sql: &str) -> Vec<u8> {
    Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)
        .unwrap()
        .query_row(sql, [], |row| row.get(0))
        .unwrap()
}

fn feed_head(path: &Path, provider: &str, feed: &str) -> Option<i64> {
    Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)
        .unwrap()
        .query_row(
            "SELECT current_snapshot_id FROM provider_feed WHERE provider = ?1 AND feed = ?2",
            [provider, feed],
            |row| row.get(0),
        )
        .unwrap()
}

fn feed_scope_snapshot_count(path: &Path, provider: &str, feed: &str, scope: &str) -> i64 {
    Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY)
        .unwrap()
        .query_row(
            "SELECT count(*)
             FROM provider_snapshot AS snapshot
             JOIN provider_feed AS feed ON feed.id = snapshot.feed_id
             WHERE feed.provider = ?1 AND feed.feed = ?2 AND snapshot.scope = ?3",
            params![provider, feed, scope],
            |row| row.get(0),
        )
        .unwrap()
}

fn query_plan(path: &Path, extension: &Path, sql: &str) -> String {
    let connection = Connection::open_with_flags(path, OpenFlags::SQLITE_OPEN_READ_ONLY).unwrap();
    unsafe {
        connection.load_extension_enable().unwrap();
        connection
            .load_extension(extension, None::<&str>)
            .unwrap();
        connection.load_extension_disable().unwrap();
    }
    let mut statement = connection
        .prepare(&format!("EXPLAIN QUERY PLAN {sql}"))
        .unwrap();
    statement
        .query_map([], |row| row.get::<_, String>(3))
        .unwrap()
        .collect::<Result<Vec<_>, _>>()
        .unwrap()
        .join("\n")
}

async fn fixture_store() -> FixtureStore {
    FixtureStore::new().await
}

async fn seeded_store() -> FixtureStore {
    let fixture = fixture_store().await;
    dev_mock::seed(&fixture.store).await.unwrap();
    fixture
}

fn at(value: &str) -> DateTime<Utc> {
    value.parse().unwrap()
}

async fn seeded_store_at(now: &str) -> FixtureStore {
    let fixture = seeded_store().await;
    let now = at(now);
    let raw_cutoff = (now - Duration::days(7)).timestamp_millis();
    let fact_cutoff = (now - Duration::days(90)).timestamp_millis();
    let connection = Connection::open(&fixture.path).unwrap();
    let feed_id = connection
        .query_row(
            "SELECT id FROM provider_feed WHERE provider = 'bmc' AND feed = 'waiting'",
            [],
            |row| row.get::<_, i64>(0),
        )
        .unwrap();
    let (place_id, line_id) = connection
        .query_row(
            "SELECT
                 (SELECT id FROM place ORDER BY id LIMIT 1),
                 (SELECT id FROM line ORDER BY id LIMIT 1)",
            [],
            |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
        )
        .unwrap();

    for (index, captured_at) in [raw_cutoff - 1, raw_cutoff, raw_cutoff + 1]
        .into_iter()
        .enumerate()
    {
        connection
            .execute(
                "INSERT INTO provider_batch(provider, captured_at) VALUES ('bmc', ?1)",
                [captured_at],
            )
            .unwrap();
        let batch_id = connection.last_insert_rowid();
        let scope = format!("retention-raw-{index}");
        let hash = vec![0xD0 + index as u8; 32];
        connection
            .execute(
                "INSERT INTO provider_snapshot(
                     batch_id, feed_id, scope, captured_at, source_at,
                     content_type, payload_sha256, payload_zstd
                 ) VALUES (?1, ?2, ?3, ?4, NULL, 'application/json', ?5, X'')",
                params![batch_id, feed_id, scope, captured_at, hash],
            )
            .unwrap();
    }

    for (index, observed_at) in [fact_cutoff - 1, fact_cutoff, fact_cutoff + 1]
        .into_iter()
        .enumerate()
    {
        let marker = format!("RETENTION-{index}");
        let expected_at = 9_000_000_000_000_i64 + index as i64;
        connection
            .execute(
                "INSERT INTO arrival_fact(
                     place_id, line_id, destination_fr, expected_at, first_seen_at, last_seen_at
                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?5)",
                params![place_id, line_id, marker, expected_at, observed_at],
            )
            .unwrap();
        connection
            .execute(
                "INSERT INTO train_departure_fact(
                     station_id, label, destination, expected_at, first_seen_at, last_seen_at
                 ) VALUES (?1, ?2, 'RETENTION', ?3, ?4, ?4)",
                params![place_id, marker, expected_at, observed_at],
            )
            .unwrap();
        let hash = vec![0xE0 + index as u8; 32];
        connection
            .execute(
                "INSERT INTO disruption_fact(
                     payload_sha256, first_seen_at, last_seen_at, payload_zstd
                 ) VALUES (?1, ?2, ?2, 'retention')",
                params![hash, observed_at],
            )
            .unwrap();
        connection
            .execute(
                "INSERT INTO vehicle_position_fact(
                     line_id, terminus_place_id, point_place_id, distance_m,
                     bucket_at, observed_at
                 ) VALUES (?1, ?2, ?2, ?3, ?4, ?4)",
                params![line_id, place_id, 9_000 + index as i64, observed_at],
            )
            .unwrap();
        connection
            .execute(
                "INSERT INTO villo_status_fact(
                     stand_id, bucket_at, bikes, docks, mechanical, electrical, observed_at
                 ) VALUES (?1, ?2, ?3, 0, NULL, NULL, ?2)",
                params![place_id, observed_at, 9_000 + index as i64],
            )
            .unwrap();
    }

    fixture
}

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

fn static_batch() -> StaticBatch {
    let captured_at = "2026-08-25T00:00:00Z".parse::<chrono::DateTime<Utc>>().unwrap();
    let raw_specs: [(&str, &str, &[u8]); 3] = [
        ("stop-details", "application/json", br#"{"results":[]}"#),
        ("gtfs", "application/zip", b"mock-gtfs"),
        ("stops-by-line", "application/json", br#"{"results":[]}"#),
    ];
    let raw = raw_specs
        .into_iter()
        .map(|(scope, content_type, body)| RawFeed {
            provider: "bmc",
            feed: "static",
            scope: scope.into(),
            captured_at,
            source_at: None,
            content_type: content_type.into(),
            body: body.to_vec(),
        })
        .collect();
    let mut stops = Vec::with_capacity(3007);
    let mut routes: HashMap<Box<str>, RouteInfo> = HashMap::new();
    let mut sequences: HashMap<Box<str>, Vec<LineSequence>> = HashMap::new();
    let mut shapes: ShapeIndex = HashMap::new();

    for line in 0..96 {
        let id = format!("L{line}");
        routes.insert(
            id.clone().into_boxed_str(),
            RouteInfo {
                line: id.clone(),
                route_id: format!("R{line}"),
                kind: if line % 5 == 0 {
                    VehicleKind::Tram
                } else {
                    VehicleKind::Bus
                },
                color: if line % 5 == 0 {
                    "EA4F80".into()
                } else {
                    "FFB178".into()
                },
                text_color: "000000".into(),
            },
        );
    }

    for i in 0usize..3007 {
        stops.push(StopInfo {
            id: format!("S{i}"),
            name: bl(&format!("MOCK STOP {}", i % 600)),
            lat: 50.80 + (i % 100) as f64 * 0.0002,
            lon: 4.30 + (i / 100) as f64 * 0.0002,
        });
    }

    for i in 0usize..1500 {
        let id = format!("S{i}");
        let line_a = format!("L{}", i % 96);
        let line_b = format!("L{}", (i + 7) % 96);
        for (line, destination, offset) in [
            (line_a, "CENTER", 1usize),
            (line_b, "RING", 2usize),
        ] {
            let previous = format!("S{}", i.saturating_sub(offset));
            let next = format!("S{}", (i + offset).min(3006));
            sequences
                .entry(line.clone().into_boxed_str())
                .or_default()
                .push(LineSequence {
                    line: line.clone(),
                    destination: bl(destination),
                    stops: vec![previous, id.clone(), next.clone()],
                });
            shapes.insert(
                (line, next),
                vec![
                    (
                        50.80 + (i % 100) as f32 * 0.0002,
                        4.30 + (i / 100) as f32 * 0.0002,
                    ),
                    (
                        50.805 + (i % 100) as f32 * 0.0002,
                        4.305 + (i / 100) as f32 * 0.0002,
                    ),
                    (
                        50.810 + (i % 100) as f32 * 0.0002,
                        4.310 + (i / 100) as f32 * 0.0002,
                    ),
                ],
            );
        }
    }

    StaticBatch {
        raw,
        stops,
        routes,
        sequences,
        shapes,
    }
}

#[tokio::test]
async fn store_reopens_a_migrated_spatial_database() {
    let fixture = FixtureStore::new().await;
    assert_eq!(fixture.store.schema_version().await.unwrap(), 5);
    assert_eq!(scalar(&fixture.path, "SELECT version FROM schema_meta"), 5);
    assert_eq!(feed_head(&fixture.path, "bmc", "static"), None);
    assert!(query_plan(&fixture.path, &fixture.extension, "SELECT id FROM place").contains("place"));

    let second = fixture.reopen().await;
    assert_eq!(second.schema_version().await.unwrap(), 5);
    assert!(fixture.path.exists());
    assert!(
        fixture.path.with_extension("sqlite-wal").exists()
            || fixture.path.metadata().unwrap().len() > 0
    );
}

#[tokio::test]
async fn store_migrates_earlier_v1_and_restores_the_normalized_search_index() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("stibble.sqlite");
    let extension = PathBuf::from(std::env::var("SPATIALITE_EXTENSION").unwrap());
    let connection = Connection::open(&path).unwrap();
    unsafe {
        connection.load_extension_enable().unwrap();
        connection
            .load_extension(&extension, None::<&str>)
            .unwrap();
        connection.load_extension_disable().unwrap();
    }
    connection
        .execute_batch(include_str!("../migrations/0001_spatial_storage.sql"))
        .unwrap();
    connection
        .execute("DROP INDEX place_normalized_lookup", [])
        .unwrap();
    drop(connection);

    let reopened = Store::open(&path, &extension).await.unwrap();

    assert_eq!(reopened.schema_version().await.unwrap(), 5);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*) FROM sqlite_schema
             WHERE type = 'index' AND name = 'place_normalized_lookup'",
        ),
        1,
    );
    let plan = query_plan(
        &path,
        &extension,
        "SELECT id
         FROM place INDEXED BY place_normalized_lookup
         WHERE provider = 'bmc'
           AND kind = 'transit'
           AND normalized_id >= 'S1'
           AND normalized_id < 'S1\u{10ffff}'
           AND active = 1",
    );
    assert!(plan.contains("place_normalized_lookup"), "{plan}");
    assert!(!plan.contains("SCAN place"), "{plan}");
}

#[tokio::test]
async fn store_migrates_existing_v2_place_aliases_to_shared_schema() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("stibble.sqlite");
    let extension = PathBuf::from(std::env::var("SPATIALITE_EXTENSION").unwrap());
    let connection = Connection::open(&path).unwrap();
    unsafe {
        connection.load_extension_enable().unwrap();
        connection
            .load_extension(&extension, None::<&str>)
            .unwrap();
        connection.load_extension_disable().unwrap();
    }
    connection
        .execute_batch(include_str!("../migrations/0001_spatial_storage.sql"))
        .unwrap();
    connection
        .execute_batch(
            "DROP TABLE place_alias;
             CREATE TABLE place_alias (
                 provider TEXT NOT NULL,
                 external_id TEXT NOT NULL,
                 place_id INTEGER NOT NULL REFERENCES place(id),
                 priority INTEGER NOT NULL,
                 PRIMARY KEY (provider, external_id, priority)
             );",
        )
        .unwrap();
    connection
        .execute_batch(include_str!("../migrations/0002_maintenance_health.sql"))
        .unwrap();
    connection
        .execute_batch(
            "INSERT INTO place(
                 provider, kind, external_id, name_fr, name_nl, folded_name,
                 collapsed_name, normalized_id, source_order, active, geom
             ) VALUES
                 ('bmc', 'transit', '5009A', 'PLATFORM A', 'PLATFORM A',
                  'platform a', 'platform a', '5009A', 0, 1, MakePoint(0, 0, 4326)),
                 ('bmc', 'transit', '5009B', 'PLATFORM B', 'PLATFORM B',
                  'platform b', 'platform b', '5009B', 1, 1, MakePoint(0.0001, 0, 4326));
             INSERT INTO line(
                 provider, external_id, display_name, kind, color, text_color, active
             ) VALUES ('bmc', 'L0', 'L0', 'tram', '000000', 'FFFFFF', 1);
             INSERT INTO place_alias(provider, external_id, place_id, priority)
             SELECT 'bmc', '5009', id, 1
             FROM place
             WHERE provider = 'bmc' AND external_id = '5009A';
             INSERT INTO arrival_fact(
                 place_id, line_id, destination_fr, expected_at, first_seen_at, last_seen_at
             )
             SELECT place.id, line.id, 'CENTER', 1, 2, 3
             FROM place, line
             WHERE place.provider = 'bmc'
               AND place.external_id = '5009A'
               AND line.provider = 'bmc'
               AND line.external_id = 'L0';",
        )
        .unwrap();
    drop(connection);

    let reopened = Store::open(&path, &extension).await.unwrap();

    assert_eq!(reopened.schema_version().await.unwrap(), 5);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*) FROM place_alias
             WHERE provider = 'bmc' AND external_id = '5009' AND priority = 1",
        ),
        1,
    );
    assert_eq!(scalar(&path, "SELECT count(*) FROM arrival_fact"), 1);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*) FROM place
             WHERE provider = 'bmc' AND external_id IN ('5009A', '5009B')",
        ),
        2,
    );
    drop(reopened);

    let connection = Connection::open(&path).unwrap();
    connection
        .execute(
            "INSERT INTO place_alias(provider, external_id, place_id, priority)
             SELECT 'bmc', '5009', id, 1
             FROM place
             WHERE provider = 'bmc' AND external_id = '5009B'",
            [],
        )
        .unwrap();
    drop(connection);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*) FROM place_alias
             WHERE provider = 'bmc' AND external_id = '5009' AND priority = 1",
        ),
        2,
    );
}

#[tokio::test]
async fn store_migrates_existing_v3_route_stops_to_sequence_only_schema() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("stibble.sqlite");
    let extension = PathBuf::from(std::env::var("SPATIALITE_EXTENSION").unwrap());
    let connection = Connection::open(&path).unwrap();
    unsafe {
        connection.load_extension_enable().unwrap();
        connection
            .load_extension(&extension, None::<&str>)
            .unwrap();
        connection.load_extension_disable().unwrap();
    }
    connection
        .execute_batch(include_str!("../migrations/0001_spatial_storage.sql"))
        .unwrap();
    connection
        .execute_batch(
            "DROP INDEX route_variant_by_place;
             DROP TABLE route_variant_stop;
             CREATE TABLE route_variant_stop (
                 variant_id INTEGER NOT NULL REFERENCES route_variant(id),
                 ordinal INTEGER NOT NULL,
                 place_id INTEGER NOT NULL REFERENCES place(id),
                 PRIMARY KEY (variant_id, ordinal)
             );
             CREATE INDEX route_variant_by_place
             ON route_variant_stop(place_id, variant_id);",
        )
        .unwrap();
    connection
        .execute_batch(include_str!("../migrations/0002_maintenance_health.sql"))
        .unwrap();
    connection
        .execute_batch(include_str!("../migrations/0003_shared_place_alias.sql"))
        .unwrap();
    connection
        .execute_batch(
            "INSERT INTO place(
                 id, provider, kind, external_id, name_fr, name_nl, folded_name,
                 collapsed_name, normalized_id, source_order, active, geom
             ) VALUES (
                 17, 'bmc', 'transit', 'KNOWN', 'KNOWN', 'KNOWN', 'known',
                 'known', 'KNOWN', 0, 1, MakePoint(4.3, 50.8, 4326)
             );
             INSERT INTO line(
                 id, provider, external_id, display_name, kind, color, text_color, active
             ) VALUES (23, 'bmc', '62', '62', 'tram', 'EA4F80', '000000', 1);
             INSERT INTO route_variant(
                 id, line_id, destination_fr, destination_nl, source_order
             ) VALUES (42, 23, 'KNOWN', 'KNOWN', 0);
             INSERT INTO route_variant_stop(variant_id, ordinal, place_id)
             VALUES (42, 0, 17);",
        )
        .unwrap();
    drop(connection);

    let reopened = Store::open(&path, &extension).await.unwrap();
    assert_eq!(reopened.schema_version().await.unwrap(), 5);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*)
             FROM route_variant_stop
             WHERE variant_id = 42
               AND ordinal = 0
               AND external_id = 'KNOWN'
               AND place_id = 17",
        ),
        1,
    );
    assert_eq!(scalar(&path, "SELECT count(*) FROM route_variant WHERE id = 42"), 1);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*)
             FROM sqlite_schema
             WHERE type = 'index' AND name = 'route_variant_by_place'",
        ),
        1,
    );
    drop(reopened);

    let connection = Connection::open(&path).unwrap();
    connection
        .execute(
            "INSERT INTO route_variant_stop(variant_id, ordinal, external_id, place_id)
             VALUES (42, 1, '3107', NULL)",
            [],
        )
        .unwrap();
    drop(connection);

    let reopened = Store::open(&path, &extension).await.unwrap();
    assert_eq!(reopened.schema_version().await.unwrap(), 5);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*)
             FROM route_variant_stop
             WHERE variant_id = 42
               AND ordinal = 1
               AND external_id = '3107'
               AND place_id IS NULL",
        ),
        1,
    );
}

#[tokio::test]
async fn store_migrates_schema_four_vehicle_positions_with_external_ids() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("stibble.sqlite");
    let extension = PathBuf::from(std::env::var("SPATIALITE_EXTENSION").unwrap());
    let connection = Connection::open(&path).unwrap();
    unsafe {
        connection.load_extension_enable().unwrap();
        connection
            .load_extension(&extension, None::<&str>)
            .unwrap();
        connection.load_extension_disable().unwrap();
    }
    connection
        .execute_batch(include_str!("../migrations/0001_spatial_storage.sql"))
        .unwrap();
    connection
        .execute_batch(
            "DROP INDEX current_vehicle_by_line;
             DROP TABLE current_vehicle_position;
             CREATE TABLE current_vehicle_position (
                 snapshot_id INTEGER NOT NULL REFERENCES provider_snapshot(id),
                 line_id INTEGER NOT NULL REFERENCES line(id),
                 terminus_place_id INTEGER REFERENCES place(id),
                 point_place_id INTEGER REFERENCES place(id),
                 distance_m INTEGER NOT NULL
             );
             CREATE INDEX current_vehicle_by_line
             ON current_vehicle_position(line_id, terminus_place_id);",
        )
        .unwrap();
    for migration in [
        include_str!("../migrations/0002_maintenance_health.sql"),
        include_str!("../migrations/0003_shared_place_alias.sql"),
        include_str!("../migrations/0004_sequence_only_stops.sql"),
    ] {
        connection.execute_batch(migration).unwrap();
    }
    connection
        .execute_batch(
            "INSERT INTO provider_batch(id, provider, captured_at)
             VALUES (11, 'bmc', 1000);
             INSERT INTO place(
                 id, provider, kind, external_id, name_fr, name_nl, folded_name,
                 collapsed_name, normalized_id, source_order, active, geom
             ) VALUES
                 (17, 'bmc', 'transit', '2332', 'TERMINUS', 'TERMINUS',
                  'terminus', 'terminus', '2332', 0, 1, MakePoint(4.3, 50.8, 4326)),
                 (18, 'bmc', 'transit', '2888', 'POINT', 'POINT',
                  'point', 'point', '2888', 1, 1, MakePoint(4.31, 50.81, 4326));
             INSERT INTO line(
                 id, provider, external_id, display_name, kind, color, text_color, active
             ) VALUES (23, 'bmc', '21', '21', 'bus', '000000', 'FFFFFF', 1);
             INSERT INTO provider_snapshot(
                 id, batch_id, feed_id, scope, captured_at, source_at,
                 content_type, payload_sha256, payload_zstd
             ) SELECT
                 31, 11, id, '', 1000, NULL, 'application/json', zeroblob(32), X''
             FROM provider_feed
             WHERE provider = 'bmc' AND feed = 'positions';
             UPDATE provider_feed
             SET current_snapshot_id = 31
             WHERE provider = 'bmc' AND feed = 'positions';
             INSERT INTO current_vehicle_position(
                 snapshot_id, line_id, terminus_place_id, point_place_id, distance_m
             ) VALUES (31, 23, 17, 18, 321);",
        )
        .unwrap();
    drop(connection);

    let reopened = Store::open(&path, &extension).await.unwrap();
    assert_eq!(reopened.schema_version().await.unwrap(), 5);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*)
             FROM current_vehicle_position
             WHERE snapshot_id = 31
               AND line_id = 23
               AND terminus_external_id = '2332'
               AND point_external_id = '2888'
               AND terminus_place_id = 17
               AND point_place_id = 18
               AND distance_m = 321",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*)
             FROM pragma_table_info('current_vehicle_position')
             WHERE (name IN ('terminus_external_id', 'point_external_id') AND \"notnull\" = 1)
                OR (name IN ('terminus_place_id', 'point_place_id') AND \"notnull\" = 0)",
        ),
        4,
    );
    drop(reopened);

    let reopened = Store::open(&path, &extension).await.unwrap();
    assert_eq!(reopened.schema_version().await.unwrap(), 5);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*)
             FROM current_vehicle_position
             WHERE terminus_external_id = '2332'
               AND point_external_id = '2888'
               AND terminus_place_id = 17
               AND point_place_id = 18",
        ),
        1,
    );
}

#[tokio::test]
async fn maintenance_keeps_exact_retention_boundaries() {
    let now = at("2026-08-25T00:00:00Z");
    let fixture = seeded_store_at("2026-08-25T00:00:00Z").await;
    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")
        .unwrap();
    for _ in 0..20 {
        connection
            .execute(
                "UPDATE schema_meta SET applied_at = applied_at + 1",
                [],
            )
            .unwrap();
    }
    drop(connection);

    fixture.store.maintain(now).await.unwrap();

    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM provider_snapshot WHERE scope LIKE 'retention-raw-%'",
        ),
        2,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM provider_batch
             WHERE captured_at = 1787011199999",
        ),
        0,
    );
    for query in [
        "SELECT count(*) FROM arrival_fact WHERE destination_fr LIKE 'RETENTION-%'",
        "SELECT count(*) FROM train_departure_fact WHERE label LIKE 'RETENTION-%'",
        "SELECT count(*) FROM disruption_fact WHERE CAST(payload_zstd AS TEXT) = 'retention'",
        "SELECT count(*) FROM vehicle_position_fact WHERE distance_m >= 9000",
        "SELECT count(*) FROM villo_status_fact WHERE bikes >= 9000",
    ] {
        assert_eq!(scalar(&fixture.path, query), 2, "{query}");
    }
    let health = fixture.store.health().await.unwrap();
    assert_eq!(
        health.oldest_raw_at,
        Some(at("2026-08-18T00:00:00Z").timestamp_millis()),
    );
    assert_eq!(
        health.oldest_fact_at,
        Some(at("2026-05-27T00:00:00Z").timestamp_millis()),
    );
    assert_eq!(health.last_pruned_at, Some(now.timestamp_millis()));
    assert_eq!(health.last_checkpoint_at, Some(now.timestamp_millis()));
    let checkpoint = Connection::open(&fixture.path)
        .unwrap()
        .query_row("PRAGMA wal_checkpoint(PASSIVE)", [], |row| {
            Ok((
                row.get::<_, i64>(0)?,
                row.get::<_, i64>(1)?,
                row.get::<_, i64>(2)?,
            ))
        })
        .unwrap();
    assert_eq!(checkpoint.0, 0);
    assert!(checkpoint.1 <= 2, "{checkpoint:?}");
    assert_eq!(checkpoint.1, checkpoint.2);
}

#[tokio::test]
async fn maintenance_preserves_current_provider_snapshot_beyond_retention() {
    let fixture = fixture_store().await;
    fixture.store.apply_static(static_batch()).await.unwrap();
    let head = feed_head(&fixture.path, "bmc", "static").unwrap();

    fixture
        .store
        .maintain(at("2026-09-02T00:00:00Z"))
        .await
        .unwrap();

    assert_eq!(feed_head(&fixture.path, "bmc", "static"), Some(head));
    assert!(fixture.store.health().await.unwrap().ready);
    assert!(!fixture.store.search_stops("mock stop").await.unwrap().is_empty());
}

#[tokio::test]
async fn maintenance_preserves_every_snapshot_in_a_current_discovery_batch() {
    let fixture = fixture_store().await;
    let captured_at = at("2026-08-01T00:00:00Z");
    fixture
        .store
        .apply_villo_discovery(villo_discovery_batch(
            captured_at,
            VILLO_INFORMATION_BODY,
        ))
        .await
        .unwrap();
    let connection = Connection::open(&fixture.path).unwrap();
    let feed_id = connection
        .query_row(
            "SELECT id
             FROM provider_feed
             WHERE provider = 'villo' AND feed = 'discovery'",
            [],
            |row| row.get::<_, i64>(0),
        )
        .unwrap();
    connection
        .execute(
            "INSERT INTO provider_batch(provider, captured_at)
             VALUES ('villo', ?1)",
            [captured_at.timestamp_millis()],
        )
        .unwrap();
    let unrelated_batch = connection.last_insert_rowid();
    connection
        .execute(
            "INSERT INTO provider_snapshot(
                 batch_id, feed_id, scope, captured_at, source_at,
                 content_type, payload_sha256, payload_zstd
             ) VALUES (?1, ?2, 'maintenance-unrelated', ?3, NULL,
                       'application/json', ?4, X'')",
            params![
                unrelated_batch,
                feed_id,
                captured_at.timestamp_millis(),
                vec![0xA5_u8; 32],
            ],
        )
        .unwrap();
    drop(connection);

    fixture
        .store
        .maintain(at("2026-08-10T00:00:00Z"))
        .await
        .unwrap();

    assert_eq!(
        fixture.store.villo_status_url().await.unwrap().as_deref(),
        Some("https://villo.test/status"),
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM provider_snapshot
             WHERE scope = 'gbfs'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM provider_snapshot
             WHERE scope = 'maintenance-unrelated'",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            &format!(
                "SELECT count(*) FROM provider_batch WHERE id = {unrelated_batch}"
            ),
        ),
        0,
    );
}

#[tokio::test]
async fn health_reports_feed_failures_pool_files_and_schema() {
    let fixture = seeded_store().await;
    for (feed, stage) in [
        ("waiting", FeedStage::Provider),
        ("positions", FeedStage::Parse),
        ("disruptions", FeedStage::Commit),
    ] {
        fixture
            .store
            .record_feed_failure(
                "bmc",
                feed,
                stage,
                &format!("{feed} failed"),
                at("2026-08-25T00:00:00Z"),
            )
            .await
            .unwrap();
    }

    let health = fixture.store.health().await.unwrap();

    assert!(health.ready);
    assert_eq!(health.schema_version, 5);
    assert!(health.pool_size >= 1);
    assert!(health.pool_available <= health.pool_size);
    assert_eq!(health.pool_waiting, 0);
    assert!(health.database_bytes > 0);
    assert_eq!(health.feeds.len(), 8);
    let static_feed = health
        .feeds
        .iter()
        .find(|feed| feed.provider == "bmc" && feed.feed == "static")
        .unwrap();
    assert!(static_feed.current_snapshot_id.is_some());
    assert!(static_feed.age_seconds.is_some());
    for (feed, stage) in [
        ("waiting", "provider"),
        ("positions", "parse"),
        ("disruptions", "commit"),
    ] {
        let failure = health
            .feeds
            .iter()
            .find(|health| health.provider == "bmc" && health.feed == feed)
            .unwrap();
        assert_eq!(failure.failure_stage.as_deref(), Some(stage));
        assert_eq!(
            failure.failure_message.as_deref(),
            Some(format!("{feed} failed").as_str()),
        );
        assert_eq!(
            failure.failed_at,
            Some(at("2026-08-25T00:00:00Z").timestamp_millis()),
        );
    }
}

#[tokio::test]
async fn health_rebuilds_and_uses_time_leading_indexes_on_reopen() {
    let fixture = seeded_store().await;
    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute_batch(
            "DROP INDEX IF EXISTS provider_snapshot_retention;
             DROP INDEX IF EXISTS arrival_fact_retention;
             DROP INDEX IF EXISTS train_departure_fact_retention;
             DROP INDEX IF EXISTS disruption_fact_retention;
             DROP INDEX IF EXISTS vehicle_position_fact_retention;
             DROP INDEX IF EXISTS villo_status_fact_retention;",
        )
        .unwrap();
    drop(connection);

    let reopened = fixture.reopen().await;

    assert_eq!(reopened.schema_version().await.unwrap(), 5);
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM sqlite_schema
             WHERE type = 'index'
               AND name IN (
                   'provider_snapshot_retention',
                   'arrival_fact_retention',
                   'train_departure_fact_retention',
                   'disruption_fact_retention',
                   'vehicle_position_fact_retention',
                   'villo_status_fact_retention'
               )",
        ),
        6,
    );
    assert!(reopened.health().await.unwrap().oldest_fact_at.is_some());
    let raw_plan = query_plan(
        &fixture.path,
        &fixture.extension,
        "SELECT MIN(captured_at) FROM provider_snapshot",
    );
    assert!(
        raw_plan.contains("provider_snapshot_retention"),
        "{raw_plan}"
    );
    assert!(!raw_plan.contains("SCAN provider_snapshot"), "{raw_plan}");
    let fact_plan = query_plan(
        &fixture.path,
        &fixture.extension,
        "SELECT MIN(oldest_at)
         FROM (
             SELECT MIN(last_seen_at) AS oldest_at FROM arrival_fact
             UNION ALL
             SELECT MIN(last_seen_at) FROM train_departure_fact
             UNION ALL
             SELECT MIN(last_seen_at) FROM disruption_fact
             UNION ALL
             SELECT MIN(observed_at) FROM vehicle_position_fact
             UNION ALL
             SELECT MIN(observed_at) FROM villo_status_fact
         )",
    );
    for index in [
        "arrival_fact_retention",
        "train_departure_fact_retention",
        "disruption_fact_retention",
        "vehicle_position_fact_retention",
        "villo_status_fact_retention",
    ] {
        assert!(fact_plan.contains(index), "{index}: {fact_plan}");
    }
    for table in [
        "arrival_fact",
        "train_departure_fact",
        "disruption_fact",
        "vehicle_position_fact",
        "villo_status_fact",
    ] {
        assert!(!fact_plan.contains(&format!("SCAN {table}")), "{fact_plan}");
    }
}

#[tokio::test]
async fn readiness_survives_restart_without_providers() {
    use http_body_util::BodyExt;
    use tower::ServiceExt;

    let fixture = seeded_store().await;
    let reopened = fixture.reopen().await;
    assert!(reopened.health().await.unwrap().ready);
    let (train_refresh, _train_refresh_rx) = tokio::sync::mpsc::channel(1);
    let response = api::router(state::AppState::new(reopened, train_refresh))
        .oneshot(
            axum::http::Request::builder()
                .uri("/readyz")
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();

    assert_eq!(response.status(), axum::http::StatusCode::OK);
    let body = response.into_body().collect().await.unwrap().to_bytes();
    let health: serde_json::Value = serde_json::from_slice(&body).unwrap();
    assert_eq!(health["ready"], true);
}

#[tokio::test]
async fn waiting_commit_is_atomic_for_a_concurrent_reader() {
    let fixture = fixture_store().await;
    fixture.store.apply_static(static_batch()).await.unwrap();
    fixture
        .store
        .apply_waiting(waiting_for(&["S1"]))
        .await
        .unwrap();
    let reader = Connection::open(&fixture.path).unwrap();
    let current_stops = |connection: &Connection| {
        let mut statement = connection
            .prepare(
                "SELECT place.external_id
                 FROM current_passage
                 JOIN place ON place.id = current_passage.place_id
                 ORDER BY place.external_id",
            )
            .unwrap();
        statement
            .query_map([], |row| row.get::<_, String>(0))
            .unwrap()
            .collect::<Result<Vec<_>, _>>()
            .unwrap()
    };
    reader.execute_batch("BEGIN").unwrap();
    assert_eq!(current_stops(&reader), vec!["S1"]);

    fixture
        .store
        .apply_waiting(waiting_for(&["S2", "S3"]))
        .await
        .unwrap();

    assert_eq!(current_stops(&reader), vec!["S1"]);
    reader.execute_batch("COMMIT").unwrap();
    assert_eq!(current_stops(&reader), vec!["S2", "S3"]);
}

#[tokio::test]
async fn static_generation_reopens_with_places_lines_and_shapes() {
    let FixtureStore {
        store,
        path,
        extension,
        _dir,
    } = fixture_store().await;
    store.apply_static(static_batch()).await.unwrap();
    drop(store);

    let reopened = Store::open(&path, &extension).await.unwrap();
    assert_eq!(reopened.schema_version().await.unwrap(), 5);
    assert_eq!(scalar(&path, "SELECT count(*) FROM place"), 3007);
    assert_eq!(scalar(&path, "SELECT count(*) FROM line"), 96);
    assert!(scalar(&path, "SELECT count(*) FROM route_variant_stop") >= 9000);
    assert!(scalar(&path, "SELECT count(*) FROM route_shape") > 0);
    assert_eq!(scalar(&path, "SELECT count(*) FROM provider_batch"), 1);
    assert_eq!(scalar(&path, "SELECT count(*) FROM provider_snapshot"), 3);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*) FROM provider_feed f JOIN provider_snapshot s ON s.id = f.current_snapshot_id WHERE f.provider = 'bmc' AND f.feed = 'static' AND s.scope = 'gtfs'",
        ),
        1,
    );
    let gtfs = blob(
        &path,
        "SELECT payload_zstd FROM provider_snapshot WHERE scope = 'gtfs'",
    );
    assert_eq!(
        zstd::stream::decode_all(gtfs.as_slice()).unwrap(),
        b"mock-gtfs",
    );
    assert_eq!(
        scalar(
            &path,
            "SELECT length(payload_sha256) FROM provider_snapshot WHERE scope = 'gtfs'",
        ),
        32,
    );
    assert_eq!(scalar(&path, "SELECT count(*) FROM transit_silent"), 3007);
    assert_eq!(scalar(&path, "SELECT count(*) FROM transit_with_passages"), 0);
    assert_eq!(scalar(&path, "SELECT count(*) FROM transit_with_notices"), 0);
}

#[tokio::test]
async fn invalid_static_generation_preserves_the_previous_generation() {
    let fixture = fixture_store().await;
    fixture.store.apply_static(static_batch()).await.unwrap();
    let before = feed_head(&fixture.path, "bmc", "static");
    let places = scalar(&fixture.path, "SELECT count(*) FROM place");
    let batches = scalar(&fixture.path, "SELECT count(*) FROM provider_batch");
    let snapshots = scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot");

    let mut invalid = static_batch();
    invalid.raw.pop();
    assert!(fixture.store.apply_static(invalid).await.is_err());

    assert_eq!(feed_head(&fixture.path, "bmc", "static"), before);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM place"), places);
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM provider_batch"),
        batches,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot"),
        snapshots,
    );
}

#[tokio::test]
async fn static_generation_prefers_an_exact_alias() {
    let fixture = fixture_store().await;
    let mut exact = static_batch();
    exact.stops.extend([
        StopInfo {
            id: "2290".into(),
            name: bl("BARE"),
            lat: 50.9,
            lon: 4.4,
        },
        StopInfo {
            id: "2290B".into(),
            name: bl("PLATFORM"),
            lat: 50.91,
            lon: 4.41,
        },
    ]);
    exact
        .sequences
        .get_mut("L0")
        .unwrap()
        .push(LineSequence {
            line: "L0".into(),
            destination: bl("EXACT"),
            stops: vec!["2290".into()],
        });
    fixture.store.apply_static(exact).await.unwrap();
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place_alias a JOIN place p ON p.id = a.place_id WHERE a.provider = 'bmc' AND a.external_id = '2290' AND a.priority = 0 AND p.external_id = '2290'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place_alias a JOIN place p ON p.id = a.place_id WHERE a.provider = 'bmc' AND a.external_id = '2290' AND a.priority = 1 AND p.external_id = '2290B'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM route_variant v JOIN route_variant_stop s ON s.variant_id = v.id JOIN place p ON p.id = s.place_id WHERE v.destination_fr = 'EXACT' AND p.external_id = '2290'",
        ),
        1,
    );
}

#[tokio::test]
async fn shared_bare_aliases_fan_out_realtime_and_use_first_static_target() {
    let fixture = fixture_store().await;
    let mut batch = static_batch();
    batch.stops.extend([
        StopInfo {
            id: "5009A".into(),
            name: bl("PLATFORM A"),
            lat: 0.0,
            lon: 0.0,
        },
        StopInfo {
            id: "5009B".into(),
            name: bl("PLATFORM B"),
            lat: 0.0,
            lon: 0.0001,
        },
    ]);
    batch
        .sequences
        .get_mut("L0")
        .unwrap()
        .push(LineSequence {
            line: "L0".into(),
            destination: bl("SHARED ALIAS"),
            stops: vec!["5009".into()],
        });

    fixture.store.apply_static(batch).await.unwrap();

    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place_alias
             WHERE provider = 'bmc' AND external_id = '5009' AND priority = 1",
        ),
        2,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place_alias
             WHERE provider = 'bmc' AND external_id = '5009' AND priority = 0",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM route_variant
             JOIN route_variant_stop ON route_variant_stop.variant_id = route_variant.id
             JOIN place ON place.id = route_variant_stop.place_id
             WHERE route_variant.destination_fr = 'SHARED ALIAS'
               AND place.external_id = '5009A'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM route_variant
             JOIN route_variant_stop ON route_variant_stop.variant_id = route_variant.id
             JOIN place ON place.id = route_variant_stop.place_id
             WHERE route_variant.destination_fr = 'SHARED ALIAS'
               AND place.external_id = '5009B'",
        ),
        0,
    );

    let expected = at("2026-08-25T08:01:00Z").fixed_offset();
    fixture
        .store
        .apply_waiting(WaitingBatch {
            raw: realtime_raw(
                "waiting",
                "2026-08-25T08:00:00Z",
                br#"{"results":[]}"#,
            ),
            passages: HashMap::from([(
                "5009".into(),
                vec![Passage {
                    line: "L0".into(),
                    destination: bl("CENTER"),
                    expected,
                }],
            )]),
            end_notices: HashMap::from([(
                "5009".into(),
                vec![EndNotice {
                    line: "L0".into(),
                    message: bl("LAST SERVICE"),
                }],
            )]),
        })
        .await
        .unwrap();

    for table in [
        "current_passage",
        "current_end_notice",
        "transit_with_passages",
    ] {
        assert_eq!(
            scalar(
                &fixture.path,
                &format!(
                    "SELECT count(*)
                     FROM {table}
                     JOIN place ON place.id = {table}.place_id
                     WHERE place.external_id IN ('5009A', '5009B')"
                ),
            ),
            2,
            "{table}",
        );
    }
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM arrival_fact
             JOIN place ON place.id = arrival_fact.place_id
             WHERE place.external_id IN ('5009A', '5009B')",
        ),
        2,
    );

    let reopened = fixture.reopen().await;
    let board = reopened
        .transit_board(0.0, 0.00005, 2, at("2026-08-25T08:00:00Z"))
        .await
        .unwrap();
    let mut ids = board
        .stops
        .iter()
        .map(|stop| stop.id.as_str())
        .collect::<Vec<_>>();
    ids.sort_unstable();
    assert_eq!(ids, vec!["5009A", "5009B"]);
    assert!(board.stops.iter().all(|stop| stop.departures.len() == 1));
    drop(reopened);

    fixture
        .store
        .apply_waiting(waiting_for(&["5009A"]))
        .await
        .unwrap();
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_passage
             JOIN place ON place.id = current_passage.place_id
             WHERE place.external_id = '5009A'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_passage
             JOIN place ON place.id = current_passage.place_id
             WHERE place.external_id = '5009B'",
        ),
        0,
    );

    fixture
        .store
        .apply_positions(PositionsBatch {
            raw: realtime_raw(
                "positions",
                "2026-08-25T08:02:00Z",
                br#"{"results":[]}"#,
            ),
            positions: HashMap::from([(
                "L0".into(),
                vec![VehiclePosition {
                    terminus: "5009A".into(),
                    point_id: "5009".into(),
                    distance_m: 123,
                }],
            )]),
        })
        .await
        .unwrap();
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_vehicle_position
             JOIN place AS terminus ON terminus.id = current_vehicle_position.terminus_place_id
             JOIN place AS point ON point.id = current_vehicle_position.point_place_id
             WHERE terminus.external_id = '5009A'
               AND point.external_id IN ('5009A', '5009B')",
        ),
        2,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM vehicle_position_fact
             JOIN place AS terminus ON terminus.id = vehicle_position_fact.terminus_place_id
             JOIN place AS point ON point.id = vehicle_position_fact.point_place_id
             WHERE terminus.external_id = '5009A'
               AND point.external_id IN ('5009A', '5009B')",
        ),
        2,
    );

    fixture
        .store
        .apply_disruptions(DisruptionsBatch {
            raw: realtime_raw(
                "disruptions",
                "2026-08-25T08:03:00Z",
                br#"{"results":[]}"#,
            ),
            disruptions: vec![Disruption {
                en: Some("Delay".into()),
                fr: Some("Retard".into()),
                nl: Some("Vertraging".into()),
                priority: 4,
                lines: vec!["L0".into()],
                points: vec!["5009".into(), "5009A".into()],
            }],
        })
        .await
        .unwrap();
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_disruption_place
             JOIN place ON place.id = current_disruption_place.place_id
             WHERE place.external_id IN ('5009A', '5009B')",
        ),
        2,
    );
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM disruption_fact"), 1);
}

#[tokio::test]
async fn static_generation_clears_a_recorded_failure_on_commit() {
    let fixture = fixture_store().await;
    fixture
        .store
        .record_feed_failure(
            "bmc",
            "static",
            FeedStage::Parse,
            "invalid feed",
            "2026-08-25T00:00:00Z".parse().unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM feed_failure WHERE stage = 'parse' AND message = 'invalid feed'",
        ),
        1,
    );

    fixture.store.apply_static(static_batch()).await.unwrap();
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM feed_failure"), 0);
}

#[tokio::test]
async fn static_generation_preserves_shared_rows_and_stable_ids() {
    let fixture = fixture_store().await;
    let mut initial = static_batch();
    initial.stops.push(StopInfo {
        id: "99999Z".into(),
        name: bl("RETIRED STOP"),
        lat: 50.92,
        lon: 4.42,
    });
    initial.routes.insert(
        "OLD".into(),
        RouteInfo {
            line: "OLD".into(),
            route_id: "R-OLD".into(),
            kind: VehicleKind::Bus,
            color: "000000".into(),
            text_color: "FFFFFF".into(),
        },
    );
    fixture.store.apply_static(initial).await.unwrap();

    let bmc_place_id = scalar(
        &fixture.path,
        "SELECT id FROM place WHERE provider = 'bmc' AND external_id = 'S0'",
    );
    let bmc_line_id = scalar(
        &fixture.path,
        "SELECT id FROM line WHERE provider = 'bmc' AND external_id = 'L0'",
    );
    let retired_place_id = scalar(
        &fixture.path,
        "SELECT id FROM place WHERE provider = 'bmc' AND external_id = '99999Z'",
    );
    let retired_line_id = scalar(
        &fixture.path,
        "SELECT id FROM line WHERE provider = 'bmc' AND external_id = 'OLD'",
    );
    let snapshot_id = feed_head(&fixture.path, "bmc", "static").unwrap();

    let connection = Connection::open(&fixture.path).unwrap();
    unsafe {
        connection.load_extension_enable().unwrap();
        connection
            .load_extension(&fixture.extension, None::<&str>)
            .unwrap();
        connection.load_extension_disable().unwrap();
    }
    connection.execute_batch("PRAGMA foreign_keys=ON").unwrap();
    connection
        .execute(
            "INSERT INTO place(provider, kind, external_id, name_fr, name_nl, folded_name, collapsed_name, normalized_id, source_order, active, geom) SELECT 'irail', 'station', 'FOREIGN', 'FOREIGN', 'FOREIGN', 'FOREIGN', 'FOREIGN', 'FOREIGN', 0, 1, geom FROM place WHERE id = ?1",
            [bmc_place_id],
        )
        .unwrap();
    let foreign_place_id = connection.last_insert_rowid();
    connection
        .execute(
            "INSERT INTO place_search(place_id, folded_name, collapsed_name, normalized_id) VALUES (?1, 'FOREIGN', 'FOREIGN', 'FOREIGN')",
            [foreign_place_id],
        )
        .unwrap();
    connection
        .execute(
            "INSERT INTO place_alias(provider, external_id, place_id, priority) VALUES ('irail', 'FOREIGN', ?1, 0)",
            [foreign_place_id],
        )
        .unwrap();
    connection
        .execute(
            "INSERT INTO line(provider, external_id, display_name, kind, color, text_color) VALUES ('other', 'FOREIGN', 'FOREIGN', 'train', '000000', 'FFFFFF')",
            [],
        )
        .unwrap();
    let foreign_line_id = connection.last_insert_rowid();
    connection
        .execute(
            "INSERT INTO route_variant(line_id, destination_fr, destination_nl, source_order) VALUES (?1, 'FOREIGN', 'FOREIGN', 0)",
            [foreign_line_id],
        )
        .unwrap();
    let foreign_variant_id = connection.last_insert_rowid();
    connection
        .execute(
            "INSERT INTO route_variant_stop(variant_id, ordinal, external_id, place_id) VALUES (?1, 0, 'FOREIGN', ?2)",
            params![foreign_variant_id, foreign_place_id],
        )
        .unwrap();
    connection
        .execute(
            "INSERT INTO route_shape(line_id, destination_key, geom) SELECT ?1, 'FOREIGN', geom FROM route_shape LIMIT 1",
            [foreign_line_id],
        )
        .unwrap();
    let foreign_shape_id = connection.last_insert_rowid();
    connection
        .execute(
            "INSERT INTO current_passage(snapshot_id, place_id, line_id, destination_fr, destination_nl, expected_at) VALUES (?1, ?2, ?3, 'CENTER', 'CENTER', 1)",
            params![snapshot_id, bmc_place_id, bmc_line_id],
        )
        .unwrap();
    connection
        .execute(
            "INSERT INTO arrival_fact(place_id, line_id, destination_fr, expected_at, first_seen_at, last_seen_at) VALUES (?1, ?2, 'CENTER', 1, 1, 1)",
            params![bmc_place_id, bmc_line_id],
        )
        .unwrap();
    connection
        .execute(
            "INSERT INTO current_train_departure(snapshot_id, station_id, label, destination, expected_at) VALUES (?1, ?2, 'IC', 'OSTEND', 1)",
            params![snapshot_id, foreign_place_id],
        )
        .unwrap();
    connection
        .execute(
            "INSERT INTO train_departure_fact(station_id, label, destination, expected_at, first_seen_at, last_seen_at) VALUES (?1, 'IC', 'OSTEND', 1, 1, 1)",
            [foreign_place_id],
        )
        .unwrap();
    drop(connection);

    let mut refreshed = static_batch();
    refreshed
        .stops
        .iter_mut()
        .find(|stop| stop.id == "S0")
        .unwrap()
        .name = bl("UPDATED STOP");
    fixture.store.apply_static(refreshed).await.unwrap();

    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM place WHERE provider = 'bmc' AND external_id = 'S0'",
        ),
        bmc_place_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM line WHERE provider = 'bmc' AND external_id = 'L0'",
        ),
        bmc_line_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM place WHERE provider = 'irail' AND external_id = 'FOREIGN'",
        ),
        foreign_place_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM line WHERE provider = 'other' AND external_id = 'FOREIGN'",
        ),
        foreign_line_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT active FROM place WHERE provider = 'irail' AND external_id = 'FOREIGN'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT active FROM line WHERE provider = 'other' AND external_id = 'FOREIGN'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place_alias WHERE provider = 'irail' AND external_id = 'FOREIGN'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            &format!("SELECT count(*) FROM route_variant WHERE id = {foreign_variant_id}"),
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            &format!(
                "SELECT count(*) FROM route_variant_stop WHERE variant_id = {foreign_variant_id}"
            ),
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            &format!("SELECT count(*) FROM route_shape WHERE id = {foreign_shape_id}"),
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place_search WHERE place_id = (SELECT id FROM place WHERE provider = 'irail' AND external_id = 'FOREIGN')",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place WHERE provider = 'bmc' AND external_id = 'S0' AND name_fr = 'UPDATED STOP' AND active = 1",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM place WHERE provider = 'bmc' AND external_id = '99999Z'",
        ),
        retired_place_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT active FROM place WHERE provider = 'bmc' AND external_id = '99999Z'",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM line WHERE provider = 'bmc' AND external_id = 'OLD'",
        ),
        retired_line_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT active FROM line WHERE provider = 'bmc' AND external_id = 'OLD'",
        ),
        0,
    );
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_passage"), 1);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM arrival_fact"), 1);
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM current_train_departure"),
        1,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM train_departure_fact"),
        1,
    );
}

#[tokio::test]
async fn mock_static_generation_is_file_backed() {
    let fixture = seeded_store().await;
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM place"), 3009);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM line"), 97);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_passage"), 4500);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_end_notice"), 1000);
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM current_vehicle_position"),
        3000,
    );
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_disruption"), 0);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM arrival_fact"), 4500);
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM vehicle_position_fact"),
        3000,
    );
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM disruption_fact"), 0);
    assert!(feed_head(&fixture.path, "bmc", "waiting").is_some());
    assert!(feed_head(&fixture.path, "bmc", "positions").is_some());
    assert!(feed_head(&fixture.path, "bmc", "disruptions").is_some());
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_villo_status"), 5);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM villo_status_fact"), 5);
    assert!(feed_head(&fixture.path, "villo", "discovery").is_some());
    assert!(feed_head(&fixture.path, "villo", "status").is_some());
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_train_departure",
        ),
        6,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM train_departure_fact"),
        6,
    );
    assert!(feed_head(&fixture.path, "irail", "stations").is_some());
    assert!(feed_head(&fixture.path, "irail", "liveboard").is_some());
    drop(fixture.store);
    let reopened = Store::open(&fixture.path, &fixture.extension).await.unwrap();
    assert_eq!(reopened.schema_version().await.unwrap(), 5);
    let trains = reopened
        .nearby_trains(50.8317, 4.3856, Utc::now())
        .await
        .unwrap();
    assert_eq!(trains.stops.len(), 3);
    assert!(trains
        .stops
        .iter()
        .all(|station| station.departures.len() == 2));
}

#[tokio::test]
async fn mock_server_recovers_uncommitted_crash_without_reseeding_or_provider_access() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("stibble.sqlite");
    let extension = PathBuf::from(std::env::var("SPATIALITE_EXTENSION").unwrap());
    let endpoints = [
        "/readyz",
        "/healthz",
        "/v1/transit?lat=50.80&lon=4.30&n=6",
        "/v1/trains?lat=50.8317&lon=4.3856",
        "/v1/villo?lat=50.8317&lon=4.3856",
        "/v1/line?line=62&to=6419",
    ];
    {
        let server = start_mock_server(&path, &extension).await;
        for endpoint in endpoints {
            assert_mock_endpoint(&server, endpoint).await;
        }
    }
    Connection::open(&path)
        .unwrap()
        .execute_batch("PRAGMA wal_checkpoint(TRUNCATE)")
        .unwrap();
    let static_head = feed_head(&path, "bmc", "static");
    let batches = scalar(&path, "SELECT count(*) FROM provider_batch");
    let snapshots = scalar(&path, "SELECT count(*) FROM provider_snapshot");
    let arrival_facts = scalar(&path, "SELECT count(*) FROM arrival_fact");
    let current_passages = scalar(&path, "SELECT count(*) FROM current_passage");
    let current_villo = scalar(&path, "SELECT count(*) FROM current_villo_status");
    let current_trains = scalar(&path, "SELECT count(*) FROM current_train_departure");
    assert!(current_passages > 0);
    assert!(current_villo > 0);
    assert!(current_trains > 0);
    {
        let _writer = start_uncommitted_writer(&path);
        assert!(path.with_extension("sqlite-wal").metadata().unwrap().len() > 0);
    }
    {
        let server = start_mock_server(&path, &extension).await;
        for endpoint in endpoints {
            assert_mock_endpoint(&server, endpoint).await;
        }
    }
    assert_eq!(feed_head(&path, "bmc", "static"), static_head);
    assert_eq!(scalar(&path, "SELECT count(*) FROM provider_batch"), batches);
    assert_eq!(
        scalar(&path, "SELECT count(*) FROM provider_snapshot"),
        snapshots,
    );
    assert_eq!(
        scalar(&path, "SELECT count(*) FROM arrival_fact"),
        arrival_facts,
    );
    assert_eq!(
        scalar(&path, "SELECT count(*) FROM current_passage"),
        current_passages,
    );
    assert_eq!(
        scalar(&path, "SELECT count(*) FROM current_villo_status"),
        current_villo,
    );
    assert_eq!(
        scalar(&path, "SELECT count(*) FROM current_train_departure"),
        current_trains,
    );
}

#[test]
fn mock_server_seed_once_ignores_an_ambient_refresh_interval() {
    let parent_interval = std::env::var_os("STIBBLE_MOCK_REFRESH_INTERVAL_MS");
    let status = Command::new(std::env::current_exe().unwrap())
        .arg("--ignored")
        .arg("--exact")
        .arg("mock_server_seed_once_ambient_refresh_process")
        .env("STIBBLE_MOCK_REFRESH_INTERVAL_MS", "1")
        .status()
        .unwrap();
    assert!(status.success(), "ambient refresh child failed with {status}");
    assert_eq!(
        std::env::var_os("STIBBLE_MOCK_REFRESH_INTERVAL_MS"),
        parent_interval,
    );
}

#[tokio::test]
#[ignore]
async fn mock_server_seed_once_ambient_refresh_process() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("stibble.sqlite");
    let extension = PathBuf::from(std::env::var("SPATIALITE_EXTENSION").unwrap());
    let server = start_mock_server_with_refresh_interval(&path, &extension, None).await;
    let feeds = [
        ("bmc", "waiting"),
        ("bmc", "positions"),
        ("bmc", "disruptions"),
        ("villo", "status"),
        ("irail", "liveboard"),
    ];
    let initial_heads: Vec<_> = feeds
        .iter()
        .map(|(provider, feed)| feed_head(&path, provider, feed))
        .collect();
    let train_scopes = [
        "BE.NMBS.008812005",
        "BE.NMBS.008813003",
        "BE.NMBS.008814001",
    ];
    let initial_train_snapshots = train_scopes.map(|scope| {
        feed_scope_snapshot_count(&path, "irail", "liveboard", scope)
    });
    let initial_snapshots = scalar(&path, "SELECT count(*) FROM provider_snapshot");

    tokio::time::sleep(StdDuration::from_secs(1)).await;

    for ((provider, feed), initial) in feeds.into_iter().zip(initial_heads) {
        assert_eq!(feed_head(&path, provider, feed), initial);
    }
    for (scope, initial) in train_scopes.into_iter().zip(initial_train_snapshots) {
        assert_eq!(
            feed_scope_snapshot_count(&path, "irail", "liveboard", scope),
            initial,
        );
    }
    assert_eq!(
        scalar(&path, "SELECT count(*) FROM provider_snapshot"),
        initial_snapshots,
    );
    assert_mock_endpoint(&server, "/readyz").await;
}

#[tokio::test]
async fn mock_server_refreshes_repeatedly_while_every_endpoint_stays_available() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("stibble.sqlite");
    let extension = PathBuf::from(std::env::var("SPATIALITE_EXTENSION").unwrap());
    let server = start_mock_server_with_refresh_interval(&path, &extension, Some(25)).await;
    let endpoints = [
        "/readyz",
        "/healthz",
        "/v1/transit?lat=50.80&lon=4.30&n=6",
        "/v1/trains?lat=50.8317&lon=4.3856",
        "/v1/villo?lat=50.8317&lon=4.3856",
        "/v1/line?line=62&to=6419",
    ];
    let feeds = [
        ("bmc", "waiting"),
        ("bmc", "positions"),
        ("bmc", "disruptions"),
        ("villo", "status"),
        ("irail", "liveboard"),
    ];
    let initial_heads: Vec<_> = feeds
        .iter()
        .map(|(provider, feed)| feed_head(&path, provider, feed).unwrap())
        .collect();
    let train_scopes = [
        "BE.NMBS.008812005",
        "BE.NMBS.008813003",
        "BE.NMBS.008814001",
    ];
    let initial_train_snapshots: Vec<_> = train_scopes
        .iter()
        .map(|scope| feed_scope_snapshot_count(&path, "irail", "liveboard", scope))
        .collect();
    let initial_snapshots = scalar(&path, "SELECT count(*) FROM provider_snapshot");
    let deadline = Instant::now() + StdDuration::from_secs(60);
    let mut endpoint_rounds = 0;
    loop {
        for endpoint in &endpoints {
            assert_mock_endpoint(&server, endpoint).await;
        }
        endpoint_rounds += 1;
        let heads_advanced = feeds
            .iter()
            .zip(&initial_heads)
            .all(|((provider, feed), initial)| {
                feed_head(&path, provider, feed).unwrap() > *initial
            });
        let train_scopes_refreshed_twice = train_scopes
            .iter()
            .zip(&initial_train_snapshots)
            .all(|(scope, initial)| {
                feed_scope_snapshot_count(&path, "irail", "liveboard", scope) >= initial + 2
            });
        let snapshots = scalar(&path, "SELECT count(*) FROM provider_snapshot");
        if endpoint_rounds >= 2
            && heads_advanced
            && train_scopes_refreshed_twice
            && snapshots >= initial_snapshots + 14
        {
            break;
        }
        assert!(
            Instant::now() < deadline,
            "mock refresh loop did not commit two complete refreshes"
        );
        tokio::time::sleep(StdDuration::from_millis(10)).await;
    }
    assert_eq!(scalar(&path, "SELECT count(*) FROM current_passage"), 4500);
    assert_eq!(
        scalar(&path, "SELECT count(*) FROM current_vehicle_position"),
        3000
    );
    assert_eq!(scalar(&path, "SELECT count(*) FROM current_disruption"), 0);
    assert_eq!(scalar(&path, "SELECT count(*) FROM current_villo_status"), 5);
    assert_eq!(
        scalar(&path, "SELECT count(*) FROM current_train_departure"),
        6
    );
    for endpoint in endpoints {
        assert_mock_endpoint(&server, endpoint).await;
    }
}

#[tokio::test]
async fn mock_server_completes_a_partial_seed_without_replacing_static_data() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("stibble.sqlite");
    let extension = PathBuf::from(std::env::var("SPATIALITE_EXTENSION").unwrap());
    let store = Store::open(&path, &extension).await.unwrap();
    store.apply_static(static_batch()).await.unwrap();
    let static_head = feed_head(&path, "bmc", "static");
    let static_snapshots = scalar(
        &path,
        "SELECT count(*)
         FROM provider_snapshot AS snapshot
         JOIN provider_feed AS feed ON feed.id = snapshot.feed_id
         WHERE feed.provider = 'bmc' AND feed.feed = 'static'",
    );
    drop(store);

    {
        let server = start_mock_server(&path, &extension).await;
        for endpoint in [
            "/readyz",
            "/v1/transit?lat=50.80&lon=4.30&n=6",
            "/v1/trains?lat=50.8317&lon=4.3856",
            "/v1/villo?lat=50.8317&lon=4.3856",
            "/v1/line?line=L0&to=S0",
        ] {
            assert_mock_endpoint(&server, endpoint).await;
        }
    }

    assert_eq!(feed_head(&path, "bmc", "static"), static_head);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*)
             FROM provider_snapshot AS snapshot
             JOIN provider_feed AS feed ON feed.id = snapshot.feed_id
             WHERE feed.provider = 'bmc' AND feed.feed = 'static'",
        ),
        static_snapshots,
    );
    for (provider, feed) in [
        ("bmc", "static"),
        ("bmc", "waiting"),
        ("bmc", "positions"),
        ("bmc", "disruptions"),
        ("irail", "stations"),
        ("irail", "liveboard"),
        ("villo", "discovery"),
        ("villo", "status"),
    ] {
        assert!(
            feed_head(&path, provider, feed).is_some(),
            "{provider}/{feed}"
        );
    }
    assert!(scalar(&path, "SELECT count(*) FROM current_passage") > 0);
    assert!(scalar(&path, "SELECT count(*) FROM current_vehicle_position") > 0);
    assert!(scalar(&path, "SELECT count(*) FROM current_villo_status") > 0);
    assert!(scalar(&path, "SELECT count(*) FROM current_train_departure") > 0);
}

#[tokio::test]
async fn mock_server_completes_missing_liveboard_scopes_without_duplicating_existing_scope() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("stibble.sqlite");
    let extension = PathBuf::from(std::env::var("SPATIALITE_EXTENSION").unwrap());
    let required_scopes = [
        "BE.NMBS.008812005",
        "BE.NMBS.008813003",
        "BE.NMBS.008814001",
    ];
    let existing_scope = required_scopes[0];
    let store = Store::open(&path, &extension).await.unwrap();
    dev_mock::seed(&store).await.unwrap();
    let existing_snapshot = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_ONLY)
        .unwrap()
        .query_row(
            "SELECT snapshot.id
             FROM provider_snapshot AS snapshot
             JOIN provider_feed AS feed ON feed.id = snapshot.feed_id
             WHERE feed.provider = 'irail'
               AND feed.feed = 'liveboard'
               AND snapshot.scope = ?1
             ORDER BY snapshot.id DESC
             LIMIT 1",
            [existing_scope],
            |row| row.get::<_, i64>(0),
        )
        .unwrap();
    let existing_history =
        feed_scope_snapshot_count(&path, "irail", "liveboard", existing_scope);
    drop(store);
    let connection = Connection::open(&path).unwrap();
    connection
        .execute(
            "UPDATE provider_feed
             SET current_snapshot_id = ?1
             WHERE provider = 'irail' AND feed = 'liveboard'",
            [existing_snapshot],
        )
        .unwrap();
    connection
        .execute(
            "DELETE FROM provider_snapshot
             WHERE feed_id = (
                 SELECT id
                 FROM provider_feed
                 WHERE provider = 'irail' AND feed = 'liveboard'
             )
               AND scope <> ?1",
            [existing_scope],
        )
        .unwrap();
    connection
        .execute(
            "DELETE FROM current_train_departure
             WHERE station_id IN (
                 SELECT id
                 FROM place
                 WHERE provider = 'irail'
                   AND kind = 'train'
                   AND external_id <> ?1
             )",
            [existing_scope],
        )
        .unwrap();
    connection
        .execute(
            "DELETE FROM train_departure_fact
             WHERE station_id IN (
                 SELECT id
                 FROM place
                 WHERE provider = 'irail'
                   AND kind = 'train'
                   AND external_id <> ?1
             )",
            [existing_scope],
        )
        .unwrap();
    drop(connection);
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*)
             FROM provider_snapshot AS snapshot
             JOIN provider_feed AS feed ON feed.id = snapshot.feed_id
             WHERE feed.provider = 'irail' AND feed.feed = 'liveboard'",
        ),
        1,
    );

    {
        let server = start_mock_server(&path, &extension).await;
        assert_mock_endpoint(
            &server,
            "/v1/trains?lat=50.8317&lon=4.3856",
        )
        .await;
    }

    assert_eq!(
        feed_scope_snapshot_count(&path, "irail", "liveboard", existing_scope),
        existing_history,
    );
    for scope in required_scopes {
        assert_eq!(
            feed_scope_snapshot_count(&path, "irail", "liveboard", scope),
            1,
            "{scope}",
        );
    }
    let reopened = Store::open(&path, &extension).await.unwrap();
    let trains = reopened
        .nearby_trains(50.8317, 4.3856, Utc::now())
        .await
        .unwrap();
    assert_eq!(trains.stops.len(), 3);
    assert!(trains
        .stops
        .iter()
        .all(|station| station.departures.len() == 2));
}

#[tokio::test]
async fn mock_server_restores_a_missing_liveboard_head_without_replaying_scopes() {
    let directory = tempfile::tempdir().unwrap();
    let path = directory.path().join("stibble.sqlite");
    let extension = PathBuf::from(std::env::var("SPATIALITE_EXTENSION").unwrap());
    let required_scopes = [
        "BE.NMBS.008812005",
        "BE.NMBS.008813003",
        "BE.NMBS.008814001",
    ];
    let store = Store::open(&path, &extension).await.unwrap();
    dev_mock::seed(&store).await.unwrap();
    let expected_head = Connection::open_with_flags(&path, OpenFlags::SQLITE_OPEN_READ_ONLY)
        .unwrap()
        .query_row(
            "SELECT snapshot.id
             FROM provider_snapshot AS snapshot
             JOIN provider_feed AS feed ON feed.id = snapshot.feed_id
             WHERE feed.provider = 'irail' AND feed.feed = 'liveboard'
             ORDER BY snapshot.captured_at DESC, snapshot.id DESC
             LIMIT 1",
            [],
            |row| row.get::<_, i64>(0),
        )
        .unwrap();
    let scope_histories = required_scopes.map(|scope| {
        feed_scope_snapshot_count(&path, "irail", "liveboard", scope)
    });
    let snapshots = scalar(
        &path,
        "SELECT count(*)
         FROM provider_snapshot AS snapshot
         JOIN provider_feed AS feed ON feed.id = snapshot.feed_id
         WHERE feed.provider = 'irail' AND feed.feed = 'liveboard'",
    );
    let departures = scalar(&path, "SELECT count(*) FROM current_train_departure");
    let facts = scalar(&path, "SELECT count(*) FROM train_departure_fact");
    drop(store);
    Connection::open(&path)
        .unwrap()
        .execute(
            "UPDATE provider_feed
             SET current_snapshot_id = NULL
             WHERE provider = 'irail' AND feed = 'liveboard'",
            [],
        )
        .unwrap();

    {
        let server = start_mock_server(&path, &extension).await;
        assert_mock_endpoint(
            &server,
            "/v1/trains?lat=50.8317&lon=4.3856",
        )
        .await;
    }

    assert_eq!(
        feed_head(&path, "irail", "liveboard"),
        Some(expected_head),
    );
    for (scope, history) in required_scopes.into_iter().zip(scope_histories) {
        assert_eq!(
            feed_scope_snapshot_count(&path, "irail", "liveboard", scope),
            history,
            "{scope}",
        );
    }
    assert_eq!(
        scalar(
            &path,
            "SELECT count(*)
             FROM provider_snapshot AS snapshot
             JOIN provider_feed AS feed ON feed.id = snapshot.feed_id
             WHERE feed.provider = 'irail' AND feed.feed = 'liveboard'",
        ),
        snapshots,
    );
    assert_eq!(
        scalar(&path, "SELECT count(*) FROM current_train_departure"),
        departures,
    );
    assert_eq!(
        scalar(&path, "SELECT count(*) FROM train_departure_fact"),
        facts,
    );
    let reopened = Store::open(&path, &extension).await.unwrap();
    let trains = reopened
        .nearby_trains(50.8317, 4.3856, Utc::now())
        .await
        .unwrap();
    assert_eq!(trains.stops.len(), 3);
    assert!(trains
        .stops
        .iter()
        .all(|station| station.departures.len() == 2));
}

fn realtime_raw(
    feed: &'static str,
    captured_at: &str,
    body: &[u8],
) -> RawFeed {
    RawFeed {
        provider: "bmc",
        feed,
        scope: "".into(),
        captured_at: captured_at.parse().unwrap(),
        source_at: None,
        content_type: "application/json".into(),
        body: body.to_vec(),
    }
}

fn waiting_batch() -> WaitingBatch {
    let expected = "2026-08-25T08:00:00Z"
        .parse::<chrono::DateTime<Utc>>()
        .unwrap()
        .fixed_offset();
    let mut passages = HashMap::new();
    let mut end_notices = HashMap::new();
    for i in 0usize..1500 {
        let line_a = format!("L{}", i % 96);
        let line_b = format!("L{}", (i + 7) % 96);
        passages.insert(
            format!("S{i}").into_boxed_str(),
            vec![
                Passage {
                    line: line_a.clone(),
                    destination: bl("CENTER"),
                    expected: expected + Duration::seconds(60 + (i % 240) as i64),
                },
                Passage {
                    line: line_a,
                    destination: bl("CENTER"),
                    expected: expected + Duration::seconds(360 + (i % 240) as i64),
                },
                Passage {
                    line: line_b,
                    destination: bl("RING"),
                    expected: expected + Duration::seconds(120 + (i % 180) as i64),
                },
            ],
        );
        if i % 3 == 0 {
            end_notices.insert(
                format!("S{i}").into_boxed_str(),
                vec![EndNotice {
                    line: format!("L{}", (i + 13) % 96),
                    message: bl("LAST BUS HAS GONE"),
                }],
            );
        }
    }
    for i in 1500usize..2000 {
        end_notices.insert(
            format!("S{i}").into_boxed_str(),
            vec![EndNotice {
                line: format!("L{}", i % 96),
                message: bl("LAST BUS HAS GONE"),
            }],
        );
    }
    WaitingBatch {
        raw: realtime_raw(
            "waiting",
            "2026-08-25T08:00:00Z",
            br#"{"results":[]}"#,
        ),
        passages,
        end_notices,
    }
}


fn waiting_for(stops: &[&str]) -> WaitingBatch {
    let expected = "2026-08-25T08:01:00Z"
        .parse::<chrono::DateTime<Utc>>()
        .unwrap()
        .fixed_offset();
    let mut passages = HashMap::new();
    for stop in stops {
        passages.insert(
            (*stop).into(),
            vec![Passage {
                line: "L0".into(),
                destination: bl("CENTER"),
                expected,
            }],
        );
    }
    WaitingBatch {
        raw: realtime_raw(
            "waiting",
            "2026-08-25T08:01:00Z",
            br#"{"results":[]}"#,
        ),
        passages,
        end_notices: HashMap::new(),
    }
}

fn positions_batch() -> PositionsBatch {
    PositionsBatch {
        raw: realtime_raw(
            "positions",
            "2026-08-25T08:00:05Z",
            br#"{"results":[]}"#,
        ),
        positions: HashMap::from([(
            "L0".into(),
            vec![
                VehiclePosition {
                    terminus: "S2".into(),
                    point_id: "S0".into(),
                    distance_m: 123,
                },
                VehiclePosition {
                    terminus: "S3".into(),
                    point_id: "S1".into(),
                    distance_m: 456,
                },
            ],
        )]),
    }
}

fn disruptions_batch() -> DisruptionsBatch {
    DisruptionsBatch {
        raw: realtime_raw(
            "disruptions",
            "2026-08-25T08:00:00Z",
            br#"{"results":[]}"#,
        ),
        disruptions: vec![
            Disruption {
                en: Some("Delay".into()),
                fr: Some("Retard".into()),
                nl: Some("Vertraging".into()),
                priority: 4,
                lines: vec!["L0".into(), "L1".into()],
                points: vec!["S0".into()],
            },
            Disruption {
                en: None,
                fr: Some("Travaux".into()),
                nl: Some("Werken".into()),
                priority: 2,
                lines: vec!["L2".into()],
                points: vec!["S2".into(), "S3".into()],
            },
        ],
    }
}

fn line_now() -> chrono::DateTime<Utc> {
    "2026-08-25T12:00:00Z".parse().unwrap()
}

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

fn line_static_batch(shape: Vec<(f32, f32)>) -> StaticBatch {
    let captured_at = line_now();
    let raw = [
        ("stop-details", "application/json", br#"{"results":[]}"#.as_slice()),
        ("gtfs", "application/zip", b"mock-gtfs".as_slice()),
        ("stops-by-line", "application/json", br#"{"results":[]}"#.as_slice()),
    ]
    .into_iter()
    .map(|(scope, content_type, body)| RawFeed {
        provider: "bmc",
        feed: "static",
        scope: scope.into(),
        captured_at,
        source_at: None,
        content_type: content_type.into(),
        body: body.to_vec(),
    })
    .collect();
    let stops = 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("BOTANIQUE"),
            lat: 50.82,
            lon: 4.32,
        },
        StopInfo {
            id: "s4".into(),
            name: bl("D"),
            lat: 50.83,
            lon: 4.33,
        },
        StopInfo {
            id: "6155F".into(),
            name: bl("STOP"),
            lat: 50.84,
            lon: 4.34,
        },
        StopInfo {
            id: "2290B".into(),
            name: bl("HANKAR"),
            lat: 50.85,
            lon: 4.35,
        },
        StopInfo {
            id: "A".into(),
            name: bl("DUPLICATE A"),
            lat: 50.86,
            lon: 4.36,
        },
        StopInfo {
            id: "B".into(),
            name: bl("DUPLICATE B"),
            lat: 50.87,
            lon: 4.37,
        },
        StopInfo {
            id: "C".into(),
            name: bl("DUPLICATE C"),
            lat: 50.88,
            lon: 4.38,
        },
    ];
    let routes = [
        (
            "62",
            VehicleKind::Tram,
            "EA4F80",
            "000000",
        ),
        (
            "81",
            VehicleKind::Bus,
            "555555",
            "FFFFFF",
        ),
        (
            "34",
            VehicleKind::Bus,
            "000000",
            "FFFFFF",
        ),
        (
            "10",
            VehicleKind::Bus,
            "000000",
            "FFFFFF",
        ),
        (
            "55",
            VehicleKind::Bus,
            "123456",
            "FFFFFF",
        ),
    ]
    .into_iter()
    .map(|(line, kind, color, text_color)| {
        (
            line.into(),
            RouteInfo {
                line: line.into(),
                route_id: format!("R{line}"),
                kind,
                color: color.into(),
                text_color: text_color.into(),
            },
        )
    })
    .collect();
    let sequences = HashMap::from([
        (
            "62".into(),
            vec![LineSequence {
                line: "62".into(),
                destination: bilingual("SAINT-LAMBERT", "SINT-LAMBRECHTS"),
                stops: vec!["s1".into(), "s2".into(), "6419".into(), "s4".into()],
            }],
        ),
        (
            "81".into(),
            vec![LineSequence {
                line: "81".into(),
                destination: bl("MARIUS RENARD"),
                stops: vec!["6155F".into()],
            }],
        ),
        (
            "34".into(),
            vec![LineSequence {
                line: "34".into(),
                destination: bl("DELTA"),
                stops: vec!["2290B".into()],
            }],
        ),
        (
            "10".into(),
            vec![LineSequence {
                line: "10".into(),
                destination: bl("TERVUREN"),
                stops: vec!["6419".into()],
            }],
        ),
        (
            "55".into(),
            vec![LineSequence {
                line: "55".into(),
                destination: bl("DUPLICATE"),
                stops: vec!["A".into(), "B".into(), "A".into(), "C".into()],
            }],
        ),
    ]);
    StaticBatch {
        raw,
        stops,
        routes,
        sequences,
        shapes: HashMap::from([(("62".into(), "s4".into()), shape)]),
    }
}

fn line_waiting_batch() -> WaitingBatch {
    let expected = line_now().fixed_offset();
    WaitingBatch {
        raw: realtime_raw(
            "waiting",
            "2026-08-25T12:00:00Z",
            br#"{"results":[]}"#,
        ),
        passages: HashMap::from([
            (
                "6419".into(),
                vec![
                    Passage {
                        line: "62".into(),
                        destination: bilingual("SAINT-LAMBERT", "SINT-LAMBRECHTS"),
                        expected: expected + Duration::seconds(300),
                    },
                    Passage {
                        line: "62".into(),
                        destination: bilingual("SAINT-LAMBERT", "SINT-LAMBRECHTS"),
                        expected: expected + Duration::seconds(90),
                    },
                    Passage {
                        line: "62".into(),
                        destination: bilingual("SAINTE-ANNE", "SINT-ANNA"),
                        expected: expected + Duration::seconds(40),
                    },
                ],
            ),
            (
                "6155F".into(),
                vec![
                    Passage {
                        line: "81".into(),
                        destination: bl("TRINITE"),
                        expected: expected + Duration::seconds(120),
                    },
                    Passage {
                        line: "81".into(),
                        destination: bl("FLAGEY"),
                        expected: expected + Duration::seconds(300),
                    },
                ],
            ),
            (
                "2290".into(),
                vec![Passage {
                    line: "34".into(),
                    destination: bl("DELTA"),
                    expected: expected + Duration::seconds(180),
                }],
            ),
        ]),
        end_notices: HashMap::new(),
    }
}

fn line_positions_batch() -> PositionsBatch {
    PositionsBatch {
        raw: realtime_raw(
            "positions",
            "2026-08-25T12:00:00Z",
            br#"{"results":[]}"#,
        ),
        positions: HashMap::from([
            (
                "62".into(),
                vec![
                    VehiclePosition {
                        terminus: "s4".into(),
                        point_id: "s1".into(),
                        distance_m: 500,
                    },
                    VehiclePosition {
                        terminus: "s4".into(),
                        point_id: "s1".into(),
                        distance_m: 300,
                    },
                ],
            ),
            (
                "34".into(),
                vec![VehiclePosition {
                    terminus: "2290".into(),
                    point_id: "2290".into(),
                    distance_m: 75,
                }],
            ),
            (
                "55".into(),
                vec![VehiclePosition {
                    terminus: "C".into(),
                    point_id: "A".into(),
                    distance_m: 50,
                }],
            ),
        ]),
    }
}

fn line_disruptions_batch() -> DisruptionsBatch {
    DisruptionsBatch {
        raw: realtime_raw(
            "disruptions",
            "2026-08-25T12:00:00Z",
            br#"{"results":[]}"#,
        ),
        disruptions: vec![
            Disruption {
                en: Some("LINE 62 EN".into()),
                fr: Some("LINE 62".into()),
                nl: Some("LIJN 62".into()),
                priority: 3,
                lines: vec!["62".into()],
                points: Vec::new(),
            },
            Disruption {
                en: None,
                fr: Some("UNRELATED".into()),
                nl: None,
                priority: 1,
                lines: vec!["34".into()],
                points: vec!["6419".into()],
            },
            Disruption {
                en: None,
                fr: Some("HANKAR".into()),
                nl: None,
                priority: 5,
                lines: vec!["34".into()],
                points: vec!["2290".into()],
            },
        ],
    }
}

async fn line_fixture_with_shape(shape: Vec<(f32, f32)>) -> FixtureStore {
    let fixture = fixture_store().await;
    fixture.store.apply_static(line_static_batch(shape)).await.unwrap();
    fixture.store.apply_waiting(line_waiting_batch()).await.unwrap();
    fixture.store.apply_positions(line_positions_batch()).await.unwrap();
    fixture
        .store
        .apply_disruptions(line_disruptions_batch())
        .await
        .unwrap();
    fixture
}

async fn line_fixture() -> FixtureStore {
    line_fixture_with_shape(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),
    ])
    .await
}

fn line_static_without_62() -> StaticBatch {
    let mut batch = line_static_batch(Vec::new());
    batch.routes.remove("62");
    batch.sequences.remove("62");
    batch.shapes.retain(|(line, _), _| line != "62");
    batch
}

#[tokio::test]
async fn sequence_only_stop_is_retained_without_becoming_a_spatial_place() {
    use http_body_util::BodyExt;
    use tower::ServiceExt;

    let fixture = fixture_store().await;
    let mut batch = line_static_batch(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),
    ]);
    batch.sequences.get_mut("62").unwrap()[0]
        .stops
        .insert(2, "3107".into());
    fixture.store.apply_static(batch).await.unwrap();

    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM route_variant_stop
             WHERE external_id = '3107' AND place_id IS NULL",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place WHERE provider = 'bmc' AND external_id = '3107'",
        ),
        0,
    );

    let detail = fixture
        .store
        .line_detail("62", "6419", None, line_now())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        detail
            .strip
            .iter()
            .map(|stop| stop.id.as_str())
            .collect::<Vec<_>>(),
        vec!["s1", "s2", "3107", "6419"],
    );
    assert_eq!(detail.strip[2].name, bl("3107"));
    assert_eq!(detail.shape.unwrap().len(), 3);

    let board = fixture
        .store
        .transit_board(50.81, 4.31, 20, line_now())
        .await
        .unwrap();
    assert!(!board.stops.is_empty());
    assert!(board.stops.iter().all(|stop| stop.id != "3107"));
    assert!(fixture.store.search_stops("3107").await.unwrap().is_empty());

    let (tx, _rx) = tokio::sync::mpsc::channel(1);
    let response = fixture
        .router_with_train_queue(tx)
        .oneshot(
            axum::http::Request::builder()
                .uri("/v1/line?line=62&to=6419")
                .body(axum::body::Body::empty())
                .unwrap(),
        )
        .await
        .unwrap();
    assert_eq!(response.status(), axum::http::StatusCode::OK);
    let body = response.into_body().collect().await.unwrap().to_bytes();
    let value: serde_json::Value = serde_json::from_slice(&body).unwrap();
    assert_eq!(value["strip"][2]["id"], "3107");
    assert_eq!(value["strip"][2]["name"]["fr"], "3107");
    assert_eq!(value["strip"][2]["name"]["nl"], "3107");
}

#[tokio::test]
async fn line_detail_matches_sequence_only_vehicle_tokens_once() {
    let fixture = fixture_store().await;
    let mut batch = line_static_batch(Vec::new());
    let stops = &mut batch.sequences.get_mut("62").unwrap()[0].stops;
    stops.insert(2, "3107F".into());
    stops.insert(3, "3107F".into());
    stops.push("2332F".into());
    batch.shapes.clear();
    fixture.store.apply_static(batch).await.unwrap();
    fixture
        .store
        .apply_positions(PositionsBatch {
            raw: realtime_raw(
                "positions",
                "2026-08-25T12:01:00Z",
                br#"{"results":[]}"#,
            ),
            positions: HashMap::from([(
                "62".into(),
                vec![
                    VehiclePosition {
                        terminus: "2332F".into(),
                        point_id: "3107F".into(),
                        distance_m: 90,
                    },
                    VehiclePosition {
                        terminus: "2332".into(),
                        point_id: "3107".into(),
                        distance_m: 80,
                    },
                ],
            )]),
        })
        .await
        .unwrap();

    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_vehicle_position
             WHERE terminus_place_id IS NULL
               AND point_place_id IS NULL
               AND terminus_external_id IN ('2332', '2332F')
               AND point_external_id IN ('3107', '3107F')",
        ),
        2,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM place
             WHERE provider = 'bmc'
               AND external_id IN ('2332F', '3107F')",
        ),
        0,
    );

    let detail = fixture
        .store
        .line_detail("62", "6419", None, line_now())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        detail
            .strip
            .iter()
            .map(|stop| (stop.id.as_str(), stop.vehicle_after_m))
            .collect::<Vec<_>>(),
        vec![
            ("s1", None),
            ("s2", None),
            ("3107F", Some(80)),
            ("3107F", None),
            ("6419", None),
        ],
    );
}

#[tokio::test]
async fn static_removal_clears_current_bmc_projections_for_inactive_line() {
    let fixture = line_fixture().await;
    let mut waiting = line_waiting_batch();
    waiting.end_notices.insert(
        "s1".into(),
        vec![EndNotice {
            line: "62".into(),
            message: bl("ENDED"),
        }],
    );
    fixture.store.apply_waiting(waiting).await.unwrap();

    fixture
        .store
        .apply_static(line_static_without_62())
        .await
        .unwrap();

    for table in ["current_passage", "current_end_notice"] {
        assert_eq!(
            scalar(
                &fixture.path,
                &format!(
                    "SELECT count(*) FROM {table}
                     JOIN line ON line.id = {table}.line_id
                     WHERE line.external_id = '62'"
                ),
            ),
            0,
        );
    }
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_vehicle_position
             JOIN line ON line.id = current_vehicle_position.line_id
             WHERE line.external_id = '62'",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_disruption_line
             JOIN line ON line.id = current_disruption_line.line_id
             WHERE line.external_id = '62'",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_disruption
             WHERE message_fr = 'LINE 62'",
        ),
        0,
    );
    fixture
        .store
        .maintain(line_now() + Duration::days(91))
        .await
        .unwrap();
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM line
             WHERE provider = 'bmc' AND external_id = '62'",
        ),
        0,
    );
}

#[tokio::test]
async fn inactive_line_is_hidden_when_waiting_rows_survive_static_refresh() {
    let fixture = line_fixture().await;
    let mut waiting = line_waiting_batch();
    waiting.passages.insert(
        "s1".into(),
        vec![Passage {
            line: "62".into(),
            destination: bl("REMOVED"),
            expected: line_now().fixed_offset() + Duration::minutes(2),
        }],
    );
    fixture.store.apply_waiting(waiting).await.unwrap();
    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute_batch(
            "CREATE TRIGGER preserve_removed_waiting
             BEFORE DELETE ON current_passage
             WHEN OLD.line_id = (
                 SELECT id FROM line WHERE provider = 'bmc' AND external_id = '62'
             )
             BEGIN
                 SELECT RAISE(IGNORE);
             END;",
        )
        .unwrap();

    fixture
        .store
        .apply_static(line_static_without_62())
        .await
        .unwrap();
    connection
        .execute_batch("DROP TRIGGER preserve_removed_waiting")
        .unwrap();
    let line_id = scalar(
        &fixture.path,
        "SELECT id FROM line WHERE provider = 'bmc' AND external_id = '62'",
    );
    let place_id = scalar(
        &fixture.path,
        "SELECT id FROM place WHERE provider = 'bmc' AND external_id = 's1'",
    );
    connection
        .execute(
            "INSERT INTO route_variant(line_id, destination_fr, destination_nl, source_order)
             VALUES (?1, 'REMOVED', 'REMOVED', 0)",
            [line_id],
        )
        .unwrap();
    let variant_id = connection.last_insert_rowid();
    connection
        .execute(
            "INSERT INTO route_variant_stop(variant_id, ordinal, external_id, place_id)
             VALUES (?1, 0, 's1', ?2)",
            params![variant_id, place_id],
        )
        .unwrap();

    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_passage WHERE place_id = (
                 SELECT id FROM place WHERE provider = 'bmc' AND external_id = 's1'
             )",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM transit_with_passages WHERE place_id = (
                 SELECT id FROM place WHERE provider = 'bmc' AND external_id = 's1'
             )",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM transit_silent WHERE place_id = (
                 SELECT id FROM place WHERE provider = 'bmc' AND external_id = 's1'
             )",
        ),
        1,
    );
    let board = fixture
        .store
        .transit_board(50.80, 4.30, 10, line_now())
        .await
        .unwrap();
    let stop = board.stops.iter().find(|stop| stop.id == "s1").unwrap();
    assert!(stop.departures.is_empty());
    let search = fixture.store.search_stops("s1").await.unwrap();
    let hit = search.iter().find(|hit| hit.id == "s1").unwrap();
    assert_eq!(hit.line_count, 0);
    assert_eq!(hit.direction, None);
    assert!(fixture
        .store
        .line_detail("62", "s1", None, line_now())
        .await
        .unwrap()
        .is_none());
}

#[tokio::test]
async fn line_detail_uses_relational_joins_and_preserves_wire_order() {
    let fixture = line_fixture().await;
    let detail = fixture
        .store
        .line_detail("62", "6419", None, line_now())
        .await
        .unwrap()
        .unwrap();

    assert_eq!(detail.line, "62");
    assert_eq!(detail.kind, VehicleKind::Tram);
    assert_eq!(detail.color, "EA4F80");
    assert_eq!(detail.text_color, "000000");
    assert_eq!(detail.destination, bilingual("SAINTE-ANNE", "SINT-ANNA"));
    assert_eq!(
        detail
            .strip
            .iter()
            .map(|stop| stop.id.as_str())
            .collect::<Vec<_>>(),
        vec!["s1", "s2", "6419"]
    );
    assert_eq!(detail.strip[0].vehicle_after_m, Some(300));
    assert_eq!(detail.strip[1].vehicle_after_m, None);
    assert!(!detail.strip[0].you);
    assert!(detail.strip.last().unwrap().you);
    assert_eq!(detail.in_seconds, vec![40, 90, 300]);
    let disruption = detail.disruption.as_ref().unwrap();
    assert_eq!(disruption.en.as_deref(), Some("LINE 62 EN"));
    assert_eq!(disruption.fr.as_deref(), Some("LINE 62"));
    assert_eq!(disruption.nl.as_deref(), Some("LIJN 62"));
    let shape = detail.shape.unwrap();
    assert_eq!(shape.len(), 3);
    for (actual, expected) in shape.iter().zip([(50.80, 4.30), (50.81, 4.31), (50.82, 4.32)]) {
        assert!((actual.0 - expected.0).abs() < 1e-5);
        assert!((actual.1 - expected.1).abs() < 1e-5);
    }
}

#[tokio::test]
async fn line_detail_preserves_destination_label_selection() {
    let fixture = line_fixture().await;

    let asked = fixture
        .store
        .line_detail("62", "6419", Some("SINT-ANNA"), line_now())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(asked.destination, bilingual("SAINTE-ANNE", "SINT-ANNA"));
    assert_eq!(asked.in_seconds, vec![40, 90, 300]);

    let fallback = fixture
        .store
        .line_detail("62", "6419", Some("NOPE"), line_now())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        fallback.destination,
        bilingual("SAINT-LAMBERT", "SINT-LAMBRECHTS")
    );
    assert_eq!(fallback.in_seconds, vec![40, 90, 300]);
}

#[tokio::test]
async fn line_detail_falls_back_to_all_line_passages_when_diverted() {
    let fixture = line_fixture().await;
    let detail = fixture
        .store
        .line_detail("81", "6155F", None, line_now())
        .await
        .unwrap()
        .unwrap();

    assert_eq!(detail.destination, bl("TRINITE"));
    assert_eq!(detail.in_seconds, vec![120, 300]);
}

#[tokio::test]
async fn line_detail_prefers_target_disruption_and_resolves_bare_aliases() {
    let fixture = line_fixture().await;
    let detail = fixture
        .store
        .line_detail("34", "2290", None, line_now())
        .await
        .unwrap()
        .unwrap();

    assert_eq!(detail.strip[0].id, "2290B");
    assert_eq!(detail.strip[0].vehicle_after_m, Some(75));
    assert_eq!(detail.in_seconds, vec![180]);
    assert_eq!(detail.disruption.unwrap().fr.as_deref(), Some("HANKAR"));
}

#[tokio::test]
async fn line_detail_exact_alias_wins_over_stripped_alias() {
    let fixture = fixture_store().await;
    let mut batch = line_static_batch(vec![(50.79, 4.29), (50.84, 4.34)]);
    batch.stops.push(StopInfo {
        id: "2290".into(),
        name: bl("BARE"),
        lat: 50.851,
        lon: 4.351,
    });
    batch.sequences.get_mut("34").unwrap().push(LineSequence {
        line: "34".into(),
        destination: bl("EXACT"),
        stops: vec!["2290".into()],
    });
    fixture.store.apply_static(batch).await.unwrap();
    fixture.store.apply_waiting(line_waiting_batch()).await.unwrap();

    let detail = fixture
        .store
        .line_detail("34", "2290", None, line_now())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(detail.strip[0].id, "2290");
    assert_eq!(detail.strip[0].name, bl("BARE"));
    assert_eq!(detail.destination, bl("DELTA"));
}

#[tokio::test]
async fn line_detail_marks_only_the_first_duplicate_stop_occurrence() {
    let fixture = line_fixture().await;
    let detail = fixture
        .store
        .line_detail("55", "C", None, line_now())
        .await
        .unwrap()
        .unwrap();

    assert_eq!(
        detail
            .strip
            .iter()
            .map(|stop| (stop.id.as_str(), stop.vehicle_after_m))
            .collect::<Vec<_>>(),
        vec![
            ("A", Some(50)),
            ("B", None),
            ("A", None),
            ("C", None)
        ]
    );
}

#[tokio::test]
async fn line_detail_shape_retains_slice_endpoints_within_wire_budget() {
    let shape = (0..=100)
        .map(|index| {
            let fraction = index as f32 / 100.0;
            (50.79 + 0.05 * fraction, 4.29 + 0.05 * fraction)
        })
        .collect();
    let fixture = line_fixture_with_shape(shape).await;
    let detail = fixture
        .store
        .line_detail("62", "6419", None, line_now())
        .await
        .unwrap()
        .unwrap();
    let shape = detail.shape.unwrap();

    assert!(shape.len() <= 24);
    assert!((shape.first().unwrap().0 - 50.80).abs() < 1e-4);
    assert!((shape.last().unwrap().0 - 50.82).abs() < 1e-4);
}

#[tokio::test]
async fn line_detail_omits_an_absent_shape() {
    let fixture = line_fixture().await;
    let detail = fixture
        .store
        .line_detail("10", "6419", None, line_now())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(detail.shape, None);
}

#[tokio::test]
async fn waiting_filters_unknown_references_and_commits_empty_replacements() {
    let fixture = fixture_store().await;
    fixture.store.apply_static(static_batch()).await.unwrap();
    fixture.store.apply_waiting(waiting_for(&["S2"])).await.unwrap();
    fixture
        .store
        .record_feed_failure(
            "bmc",
            "waiting",
            FeedStage::Commit,
            "old failure",
            at("2026-08-25T08:01:30Z"),
        )
        .await
        .unwrap();
    let previous_head = feed_head(&fixture.path, "bmc", "waiting");
    let expected = at("2026-08-25T08:03:00Z").fixed_offset();

    fixture
        .store
        .apply_waiting(WaitingBatch {
            raw: realtime_raw(
                "waiting",
                "2026-08-25T08:02:00Z",
                br#"{"results":["mixed"]}"#,
            ),
            passages: HashMap::from([
                (
                    "S0".into(),
                    vec![
                        Passage {
                            line: "L0".into(),
                            destination: bl("MIXED"),
                            expected,
                        },
                        Passage {
                            line: "21".into(),
                            destination: bl("UNKNOWN LINE"),
                            expected,
                        },
                    ],
                ),
                (
                    "2888".into(),
                    vec![Passage {
                        line: "L0".into(),
                        destination: bl("UNKNOWN STOP"),
                        expected,
                    }],
                ),
            ]),
            end_notices: HashMap::from([
                (
                    "S1".into(),
                    vec![
                        EndNotice {
                            line: "L1".into(),
                            message: bl("KNOWN NOTICE"),
                        },
                        EndNotice {
                            line: "21".into(),
                            message: bl("UNKNOWN LINE"),
                        },
                    ],
                ),
                (
                    "2888".into(),
                    vec![EndNotice {
                        line: "L1".into(),
                        message: bl("UNKNOWN STOP"),
                    }],
                ),
            ]),
        })
        .await
        .unwrap();

    let mixed_head = feed_head(&fixture.path, "bmc", "waiting");
    assert_ne!(mixed_head, previous_head);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_passage"), 1);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_end_notice"), 1);
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM arrival_fact
             JOIN place ON place.id = arrival_fact.place_id
             JOIN line ON line.id = arrival_fact.line_id
             WHERE place.external_id = 'S0'
               AND line.external_id = 'L0'
               AND destination_fr = 'MIXED'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM feed_failure
             JOIN provider_feed ON provider_feed.id = feed_failure.feed_id
             WHERE provider_feed.provider = 'bmc'
               AND provider_feed.feed = 'waiting'",
        ),
        0,
    );
    let facts = scalar(&fixture.path, "SELECT count(*) FROM arrival_fact");

    fixture
        .store
        .apply_waiting(WaitingBatch {
            raw: realtime_raw(
                "waiting",
                "2026-08-25T08:04:00Z",
                br#"{"results":["unknown-only"]}"#,
            ),
            passages: HashMap::from([
                (
                    "2888".into(),
                    vec![Passage {
                        line: "L0".into(),
                        destination: bl("UNKNOWN STOP"),
                        expected,
                    }],
                ),
                (
                    "S0".into(),
                    vec![Passage {
                        line: "21".into(),
                        destination: bl("UNKNOWN LINE"),
                        expected,
                    }],
                ),
            ]),
            end_notices: HashMap::from([(
                "2888".into(),
                vec![EndNotice {
                    line: "21".into(),
                    message: bl("UNKNOWN"),
                }],
            )]),
        })
        .await
        .unwrap();

    assert_ne!(feed_head(&fixture.path, "bmc", "waiting"), mixed_head);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_passage"), 0);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_end_notice"), 0);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM arrival_fact"), facts);
}

#[tokio::test]
async fn positions_filter_unknown_lines_and_retain_unknown_place_tokens() {
    let fixture = fixture_store().await;
    fixture.store.apply_static(static_batch()).await.unwrap();
    fixture.store.apply_positions(positions_batch()).await.unwrap();
    fixture
        .store
        .record_feed_failure(
            "bmc",
            "positions",
            FeedStage::Commit,
            "old failure",
            at("2026-08-25T08:01:30Z"),
        )
        .await
        .unwrap();
    let previous_head = feed_head(&fixture.path, "bmc", "positions");

    fixture
        .store
        .apply_positions(PositionsBatch {
            raw: realtime_raw(
                "positions",
                "2026-08-25T08:02:00Z",
                br#"{"results":["mixed"]}"#,
            ),
            positions: HashMap::from([
                (
                    "L0".into(),
                    vec![
                        VehiclePosition {
                            terminus: "S2".into(),
                            point_id: "S0".into(),
                            distance_m: 700,
                        },
                        VehiclePosition {
                            terminus: "2332".into(),
                            point_id: "S1".into(),
                            distance_m: 701,
                        },
                        VehiclePosition {
                            terminus: "S3".into(),
                            point_id: "2888".into(),
                            distance_m: 702,
                        },
                        VehiclePosition {
                            terminus: "2332".into(),
                            point_id: "2888".into(),
                            distance_m: 703,
                        },
                    ],
                ),
                (
                    "21".into(),
                    vec![VehiclePosition {
                        terminus: "S2".into(),
                        point_id: "S0".into(),
                        distance_m: 704,
                    }],
                ),
            ]),
        })
        .await
        .unwrap();

    let mixed_head = feed_head(&fixture.path, "bmc", "positions");
    assert_ne!(mixed_head, previous_head);
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_vehicle_position",
        ),
        4,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_vehicle_position
             WHERE terminus_external_id = '2332'
               AND point_external_id = 'S1'
               AND terminus_place_id IS NULL
               AND point_place_id IS NOT NULL
               AND distance_m = 701",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_vehicle_position
             WHERE terminus_external_id = 'S3'
               AND point_external_id = '2888'
               AND terminus_place_id IS NOT NULL
               AND point_place_id IS NULL
               AND distance_m = 702",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_vehicle_position
             WHERE terminus_external_id = '2332'
               AND point_external_id = '2888'
               AND terminus_place_id IS NULL
               AND point_place_id IS NULL
               AND distance_m = 703",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM vehicle_position_fact WHERE distance_m = 700",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM vehicle_position_fact
             WHERE distance_m BETWEEN 701 AND 704",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM feed_failure
             JOIN provider_feed ON provider_feed.id = feed_failure.feed_id
             WHERE provider_feed.provider = 'bmc'
               AND provider_feed.feed = 'positions'",
        ),
        0,
    );
    let facts = scalar(&fixture.path, "SELECT count(*) FROM vehicle_position_fact");

    fixture
        .store
        .apply_positions(PositionsBatch {
            raw: realtime_raw(
                "positions",
                "2026-08-25T08:03:00Z",
                br#"{"results":["unknown-only"]}"#,
            ),
            positions: HashMap::from([(
                "21".into(),
                vec![VehiclePosition {
                    terminus: "2332".into(),
                    point_id: "2888".into(),
                    distance_m: 705,
                }],
            )]),
        })
        .await
        .unwrap();

    assert_ne!(feed_head(&fixture.path, "bmc", "positions"), mixed_head);
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_vehicle_position",
        ),
        0,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM vehicle_position_fact"),
        facts,
    );
}

#[tokio::test]
async fn disruptions_filter_unknown_links_and_require_a_known_line() {
    let fixture = fixture_store().await;
    fixture.store.apply_static(static_batch()).await.unwrap();
    fixture
        .store
        .apply_disruptions(disruptions_batch())
        .await
        .unwrap();
    fixture
        .store
        .record_feed_failure(
            "bmc",
            "disruptions",
            FeedStage::Commit,
            "old failure",
            at("2026-08-25T08:01:30Z"),
        )
        .await
        .unwrap();
    let previous_head = feed_head(&fixture.path, "bmc", "disruptions");
    let previous_facts = scalar(&fixture.path, "SELECT count(*) FROM disruption_fact");

    fixture
        .store
        .apply_disruptions(DisruptionsBatch {
            raw: realtime_raw(
                "disruptions",
                "2026-08-25T08:02:00Z",
                br#"{"results":["mixed"]}"#,
            ),
            disruptions: vec![
                Disruption {
                    en: Some("MIXED".into()),
                    fr: None,
                    nl: None,
                    priority: 1,
                    lines: vec!["L0".into(), "21".into()],
                    points: vec!["S0".into(), "2888".into()],
                },
                Disruption {
                    en: Some("UNKNOWN LINE".into()),
                    fr: None,
                    nl: None,
                    priority: 2,
                    lines: vec!["21".into()],
                    points: vec!["S1".into()],
                },
                Disruption {
                    en: Some("KNOWN LINE UNKNOWN POINT".into()),
                    fr: None,
                    nl: None,
                    priority: 3,
                    lines: vec!["L1".into()],
                    points: vec!["2888".into()],
                },
            ],
        })
        .await
        .unwrap();

    let mixed_head = feed_head(&fixture.path, "bmc", "disruptions");
    assert_ne!(mixed_head, previous_head);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_disruption"), 2);
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_disruption_line",
        ),
        2,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_disruption_place",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_disruption
             JOIN current_disruption_line
               ON current_disruption_line.disruption_id = current_disruption.id
             JOIN line ON line.id = current_disruption_line.line_id
             WHERE current_disruption.message_en = 'MIXED'
               AND line.external_id = 'L0'",
        ),
        1,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM disruption_fact"),
        previous_facts + 2,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM feed_failure
             JOIN provider_feed ON provider_feed.id = feed_failure.feed_id
             WHERE provider_feed.provider = 'bmc'
               AND provider_feed.feed = 'disruptions'",
        ),
        0,
    );
    let facts = scalar(&fixture.path, "SELECT count(*) FROM disruption_fact");

    fixture
        .store
        .apply_disruptions(DisruptionsBatch {
            raw: realtime_raw(
                "disruptions",
                "2026-08-25T08:03:00Z",
                br#"{"results":["unknown-only"]}"#,
            ),
            disruptions: vec![Disruption {
                en: Some("UNKNOWN".into()),
                fr: None,
                nl: None,
                priority: 1,
                lines: vec!["21".into()],
                points: vec!["2888".into()],
            }],
        })
        .await
        .unwrap();

    assert_ne!(
        feed_head(&fixture.path, "bmc", "disruptions"),
        mixed_head,
    );
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_disruption"), 0);
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM disruption_fact"),
        facts,
    );
}

#[tokio::test]
async fn waiting_commit_rebuilds_exclusive_transit_ranks() {
    let fixture = seeded_store().await;
    fixture.store.apply_waiting(waiting_batch()).await.unwrap();

    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_passage"), 4500);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_end_notice"), 1000);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM transit_with_passages"), 1500);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM transit_with_notices"), 500);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM transit_silent"), 1001);
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT (SELECT count(*) FROM transit_with_passages) + (SELECT count(*) FROM transit_with_notices) + (SELECT count(*) FROM transit_silent)",
        ),
        3001,
    );
    assert!(feed_head(&fixture.path, "bmc", "waiting").is_some());
}

#[tokio::test]
async fn a_failed_waiting_commit_preserves_current_rows_and_facts() {
    let fixture = seeded_store().await;
    fixture.store.apply_waiting(waiting_batch()).await.unwrap();
    let snapshot = feed_head(&fixture.path, "bmc", "waiting");
    let snapshots = scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot");
    let current = scalar(&fixture.path, "SELECT count(*) FROM current_passage");
    let facts = scalar(&fixture.path, "SELECT count(*) FROM arrival_fact");

    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute_batch(
            "CREATE TRIGGER reject_waiting_insert
             BEFORE INSERT ON current_passage
             BEGIN
                 SELECT RAISE(FAIL, 'forced waiting insert failure');
             END;",
        )
        .unwrap();
    drop(connection);
    assert!(fixture.store.apply_waiting(waiting_batch()).await.is_err());
    assert_eq!(feed_head(&fixture.path, "bmc", "waiting"), snapshot);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot"), snapshots);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_passage"), current);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM arrival_fact"), facts);
}

#[tokio::test]
async fn realtime_alias_resolution_prefers_exact_aliases() {
    let fixture = seeded_store().await;
    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute(
            "INSERT INTO place_alias(provider, external_id, place_id, priority)
             SELECT 'bmc', 'S0', id, 1 FROM place
             WHERE provider = 'bmc' AND kind = 'transit' AND external_id = 'S1'",
            [],
        )
        .unwrap();
    drop(connection);

    fixture.store.apply_waiting(waiting_for(&["S0"])).await.unwrap();

    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_passage
             JOIN place ON place.id = current_passage.place_id
             WHERE place.external_id = 'S0'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_passage
             JOIN place ON place.id = current_passage.place_id
             WHERE place.external_id = 'S1'",
        ),
        0,
    );
}

#[tokio::test]
async fn silent_rank_excludes_names_present_in_active_ranks() {
    let fixture = seeded_store().await;
    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute(
            "UPDATE place
             SET name_fr = 'MOCK ACTIVE 0'
             WHERE provider = 'bmc' AND kind = 'transit' AND external_id = 'S2000'",
            [],
        )
        .unwrap();
    drop(connection);

    fixture.store.apply_waiting(waiting_for(&["S0"])).await.unwrap();

    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM transit_silent
             JOIN place ON place.id = transit_silent.place_id
             WHERE place.name_fr = 'MOCK ACTIVE 0'",
        ),
        0,
    );
}

#[tokio::test]
async fn active_rank_retains_distinct_stops_with_the_same_name() {
    let fixture = seeded_store().await;
    fixture
        .store
        .apply_waiting(waiting_for(&["S0", "S300"]))
        .await
        .unwrap();

    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM transit_with_passages
             JOIN place ON place.id = transit_with_passages.place_id
             WHERE place.name_fr = 'MOCK ACTIVE 0'",
        ),
        2,
    );
}

#[tokio::test]
async fn position_facts_dedupe_equal_observations_within_each_minute() {
    let fixture = seeded_store().await;
    let before = scalar(&fixture.path, "SELECT count(*) FROM vehicle_position_fact");
    fixture.store.apply_positions(positions_batch()).await.unwrap();

    let mut repeated = positions_batch();
    repeated.raw.captured_at = "2026-08-25T08:00:55Z".parse().unwrap();
    fixture.store.apply_positions(repeated).await.unwrap();
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM vehicle_position_fact"),
        before + 2,
    );

    let mut changed = positions_batch();
    changed.raw.captured_at = "2026-08-25T08:00:56Z".parse().unwrap();
    changed.positions.get_mut("L0").unwrap()[0].distance_m = 124;
    fixture.store.apply_positions(changed).await.unwrap();
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM vehicle_position_fact"),
        before + 3,
    );

    let mut next_minute = positions_batch();
    next_minute.raw.captured_at = "2026-08-25T08:01:05Z".parse().unwrap();
    fixture.store.apply_positions(next_minute).await.unwrap();
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM vehicle_position_fact"),
        before + 5,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM current_vehicle_position"),
        2,
    );
}

#[tokio::test]
async fn disruption_facts_dedupe_unchanged_payloads_and_retain_changes() {
    let fixture = seeded_store().await;
    fixture.store.apply_disruptions(disruptions_batch()).await.unwrap();
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM disruption_fact"), 2);

    let mut repeated = disruptions_batch();
    repeated.raw.captured_at = "2026-08-25T08:05:00Z".parse().unwrap();
    fixture.store.apply_disruptions(repeated).await.unwrap();
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM disruption_fact"), 2);

    let mut changed = disruptions_batch();
    changed.raw.captured_at = "2026-08-25T08:10:00Z".parse().unwrap();
    changed.disruptions[0].fr = Some("Retard prolongé".into());
    fixture.store.apply_disruptions(changed).await.unwrap();
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM disruption_fact"), 3);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_disruption"), 2);
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM current_disruption_line"),
        3,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM current_disruption_place"),
        3,
    );
}

#[tokio::test]
async fn failed_position_and_disruption_commits_preserve_heads_and_facts() {
    let fixture = seeded_store().await;
    fixture.store.apply_positions(positions_batch()).await.unwrap();
    fixture.store.apply_disruptions(disruptions_batch()).await.unwrap();
    let positions_head = feed_head(&fixture.path, "bmc", "positions");
    let positions_current = scalar(&fixture.path, "SELECT count(*) FROM current_vehicle_position");
    let position_facts = scalar(&fixture.path, "SELECT count(*) FROM vehicle_position_fact");
    let disruptions_head = feed_head(&fixture.path, "bmc", "disruptions");
    let disruptions_current = scalar(&fixture.path, "SELECT count(*) FROM current_disruption");
    let disruption_facts = scalar(&fixture.path, "SELECT count(*) FROM disruption_fact");
    let snapshots = scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot");
    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute_batch(
            "CREATE TRIGGER reject_position_insert
             BEFORE INSERT ON current_vehicle_position
             BEGIN
                 SELECT RAISE(FAIL, 'forced position insert failure');
             END;
             CREATE TRIGGER reject_disruption_insert
             BEFORE INSERT ON current_disruption
             BEGIN
                 SELECT RAISE(FAIL, 'forced disruption insert failure');
             END;",
        )
        .unwrap();
    drop(connection);

    assert!(fixture.store.apply_positions(positions_batch()).await.is_err());
    assert_eq!(feed_head(&fixture.path, "bmc", "positions"), positions_head);
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM current_vehicle_position"),
        positions_current,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM vehicle_position_fact"),
        position_facts,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot"),
        snapshots,
    );

    assert!(fixture
        .store
        .apply_disruptions(disruptions_batch())
        .await
        .is_err());
    assert_eq!(
        feed_head(&fixture.path, "bmc", "disruptions"),
        disruptions_head,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM current_disruption"),
        disruptions_current,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM disruption_fact"),
        disruption_facts,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot"),
        snapshots,
    );
}

#[tokio::test]
async fn transit_uses_live_rank_then_exact_spatial_distance() {
    let fixture = seeded_store().await;
    let board = fixture
        .store
        .transit_board(50.8535, 4.3661, 6, Utc::now())
        .await
        .unwrap();

    assert_eq!(board.stops.len(), 6);
    assert!(board.stops[0]
        .departures
        .iter()
        .any(|row| matches!(row, stibble_server::departure::Departure::Transit { .. })));
    assert!(board.stops.windows(2).all(|pair| pair[0].rank >= pair[1].rank));
    assert!(board
        .stops
        .windows(2)
        .all(|pair| pair[0].rank != pair[1].rank || pair[0].distance_m <= pair[1].distance_m));
}

#[tokio::test]
async fn search_uses_exact_then_collapsed_candidates_and_short_ids() {
    let fixture = seeded_store().await;

    let exact = fixture.store.search_stops("MOCK ACTIVE 1").await.unwrap();
    assert_eq!(exact[0].id, "S1");
    assert_eq!(exact[0].line_count, 2);
    assert!(exact[0].direction.is_some());

    let collapsed = fixture.store.search_stops("MOCCK ACTIVE 1").await.unwrap();
    assert_eq!(collapsed[0].id, "S1");

    let short_id = fixture.store.search_stops("s1").await.unwrap();
    assert_eq!(short_id[0].id, "S1");
    assert!(short_id.len() <= 20);
}

#[tokio::test]
async fn static_refresh_rebuilds_live_ranks_and_failed_waiting_preserves_them() {
    let fixture = fixture_store().await;
    fixture.store.apply_static(static_batch()).await.unwrap();
    fixture.store.apply_waiting(waiting_batch()).await.unwrap();
    let waiting_head = feed_head(&fixture.path, "bmc", "waiting");

    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM transit_with_passages"), 1500);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM transit_with_notices"), 500);

    fixture.store.apply_static(static_batch()).await.unwrap();

    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM transit_with_passages"), 1500);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM transit_with_notices"), 500);
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT EXISTS(
                SELECT 1
                FROM transit_with_passages
                JOIN place ON place.id = transit_with_passages.place_id
                WHERE place.external_id = 'S0'
            )",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT EXISTS(
                SELECT 1
                FROM transit_with_notices
                JOIN place ON place.id = transit_with_notices.place_id
                WHERE place.external_id = 'S1500'
            )",
        ),
        1,
    );
    assert_eq!(feed_head(&fixture.path, "bmc", "waiting"), waiting_head);

    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute_batch(
            "CREATE TRIGGER reject_waiting_insert
             BEFORE INSERT ON current_passage
             BEGIN
                 SELECT RAISE(FAIL, 'forced waiting insert failure');
             END;",
        )
        .unwrap();
    drop(connection);
    assert!(fixture.store.apply_waiting(waiting_batch()).await.is_err());
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM transit_with_passages"), 1500);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM transit_with_notices"), 500);
    assert_eq!(feed_head(&fixture.path, "bmc", "waiting"), waiting_head);
}

const VILLO_DISCOVERY_BODY: &str = r#"{"data":{"feeds":[
  {"name":"station_information","url":"https://villo.test/information"},
  {"name":"station_status","url":"https://villo.test/status"}
]}}"#;

const VILLO_INFORMATION_BODY: &str = r#"{"data":{"stations":[
  {"station_id":"1","name":"ARTS-LOI","lat":50.8467,"lon":4.3676},
  {"station_id":"2","name":"MAELBEEK","lat":50.8470,"lon":4.3676},
  {"station_id":"3","name":"TOO FAR","lat":50.8600,"lon":4.3676}
]}}"#;

const VILLO_INFORMATION_REFRESH_BODY: &str = r#"{"data":{"stations":[
  {"station_id":"1","name":"ARTS-LOI UPDATED","lat":50.8468,"lon":4.3677},
  {"station_id":"3","name":"TOO FAR","lat":50.8600,"lon":4.3676},
  {"station_id":"4","name":"SCHUMAN","lat":50.8472,"lon":4.3690}
]}}"#;

const VILLO_STATUS_BODY: &str = r#"{"data":{"stations":[
  {"station_id":"1","num_vehicles_available":6,"num_docks_available":9,"vehicle_types_available":[{"vehicle_type_id":"mechanical","count":4},{"vehicle_type_id":"electrical","count":2}]},
  {"station_id":"2","num_vehicles_available":3,"num_docks_available":7},
  {"station_id":"missing","num_vehicles_available":1,"num_docks_available":1}
]}}"#;

const VILLO_STATUS_CHANGED_BODY: &str = r#"{"data":{"stations":[
  {"station_id":"1","num_vehicles_available":8,"num_docks_available":7,"vehicle_types_available":[{"vehicle_type_id":"mechanical","count":5},{"vehicle_type_id":"electrical","count":3}]},
  {"station_id":"2","num_vehicles_available":2,"num_docks_available":8}
]}}"#;

const VILLO_STATUS_REJECTED_BODY: &str = r#"{"data":{"stations":[
  {"station_id":"1","num_vehicles_available":999,"num_docks_available":0},
  {"station_id":"2","num_vehicles_available":2,"num_docks_available":8}
]}}"#;

fn villo_raw(
    feed: &'static str,
    scope: &str,
    captured_at: chrono::DateTime<Utc>,
    body: &str,
) -> RawFeed {
    RawFeed {
        provider: "villo",
        feed,
        scope: scope.into(),
        captured_at,
        source_at: None,
        content_type: "application/json".into(),
        body: body.as_bytes().to_vec(),
    }
}

fn villo_discovery_batch(
    captured_at: chrono::DateTime<Utc>,
    information: &str,
) -> VilloDiscoveryBatch {
    VilloDiscoveryBatch {
        raw: vec![
            villo_raw("discovery", "gbfs", captured_at, VILLO_DISCOVERY_BODY),
            villo_raw(
                "discovery",
                "station_information",
                captured_at,
                information,
            ),
        ],
        status_url: "https://villo.test/status".into(),
        stations: villo::parse_information(information).unwrap(),
    }
}

fn villo_status_batch(captured_at: chrono::DateTime<Utc>, body: &str) -> VilloStatusBatch {
    VilloStatusBatch {
        raw: villo_raw("status", "", captured_at, body),
        status: villo::parse_status(body).unwrap(),
    }
}

#[tokio::test]
async fn villo_discovery_persists_raw_and_reconciles_stable_provider_rows() {
    let fixture = fixture_store().await;
    fixture.store.apply_static(static_batch()).await.unwrap();
    let bmc_id = scalar(
        &fixture.path,
        "SELECT id FROM place WHERE provider = 'bmc' AND external_id = 'S0'",
    );
    let first_at = "2026-08-25T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_villo_discovery(villo_discovery_batch(first_at, VILLO_INFORMATION_BODY))
        .await
        .unwrap();
    fixture
        .store
        .apply_villo_status(villo_status_batch(
            first_at + Duration::minutes(1),
            VILLO_STATUS_BODY,
        ))
        .await
        .unwrap();
    let villo_id = scalar(
        &fixture.path,
        "SELECT id FROM place WHERE provider = 'villo' AND kind = 'bike' AND external_id = '1'",
    );

    assert_eq!(
        fixture.store.villo_status_url().await.unwrap().as_deref(),
        Some("https://villo.test/status"),
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM provider_snapshot
             JOIN provider_feed ON provider_feed.id = provider_snapshot.feed_id
             WHERE provider_feed.provider = 'villo' AND provider_feed.feed = 'discovery'",
        ),
        2,
    );
    let discovery = blob(
        &fixture.path,
        "SELECT payload_zstd FROM provider_snapshot WHERE scope = 'gbfs'",
    );
    assert_eq!(
        zstd::stream::decode_all(discovery.as_slice()).unwrap(),
        VILLO_DISCOVERY_BODY.as_bytes(),
    );

    fixture
        .store
        .record_feed_failure(
            "villo",
            "discovery",
            FeedStage::Provider,
            "temporary",
            Utc::now(),
        )
        .await
        .unwrap();
    let second_at = "2026-08-26T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_villo_discovery(villo_discovery_batch(
            second_at,
            VILLO_INFORMATION_REFRESH_BODY,
        ))
        .await
        .unwrap();

    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM place WHERE provider = 'villo' AND kind = 'bike' AND external_id = '1'",
        ),
        villo_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT active FROM place WHERE provider = 'villo' AND kind = 'bike' AND external_id = '2'",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_villo_status
             JOIN place ON place.id = current_villo_status.stand_id
             WHERE place.external_id = '2'",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM villo_status_fact
             JOIN place ON place.id = villo_status_fact.stand_id
             WHERE place.external_id = '2'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT active FROM place WHERE provider = 'villo' AND kind = 'bike' AND external_id = '4'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM place WHERE provider = 'bmc' AND external_id = 'S0'",
        ),
        bmc_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM feed_failure
             JOIN provider_feed ON provider_feed.id = feed_failure.feed_id
             WHERE provider_feed.provider = 'villo' AND provider_feed.feed = 'discovery'",
        ),
        0,
    );
}

#[tokio::test]
async fn failed_villo_discovery_commit_preserves_head_and_places() {
    let fixture = fixture_store().await;
    let first_at = "2026-08-25T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_villo_discovery(villo_discovery_batch(first_at, VILLO_INFORMATION_BODY))
        .await
        .unwrap();
    let head = feed_head(&fixture.path, "villo", "discovery");
    let snapshots = scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot");
    let stable_id = scalar(
        &fixture.path,
        "SELECT id FROM place WHERE provider = 'villo' AND external_id = '1'",
    );

    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute_batch(
            "CREATE TRIGGER reject_villo_discovery
             BEFORE INSERT ON place
             WHEN NEW.provider = 'villo' AND NEW.external_id = '4'
             BEGIN
                 SELECT RAISE(ABORT, 'rejected villo discovery');
             END;",
        )
        .unwrap();
    let rejected_at = "2026-08-26T00:00:00Z".parse().unwrap();
    assert!(
        fixture
            .store
            .apply_villo_discovery(villo_discovery_batch(
                rejected_at,
                VILLO_INFORMATION_REFRESH_BODY,
            ))
            .await
            .is_err()
    );
    connection
        .execute_batch("DROP TRIGGER reject_villo_discovery")
        .unwrap();

    assert_eq!(feed_head(&fixture.path, "villo", "discovery"), head);
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot"),
        snapshots,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM place WHERE provider = 'villo' AND external_id = '1'",
        ),
        stable_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT active FROM place WHERE provider = 'villo' AND external_id = '2'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place WHERE provider = 'villo' AND external_id = '4'",
        ),
        0,
    );
}

#[tokio::test]
async fn villo_status_replaces_current_counts_and_buckets_five_minute_facts() {
    let fixture = fixture_store().await;
    let discovery_at = "2026-08-25T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_villo_discovery(villo_discovery_batch(
            discovery_at,
            VILLO_INFORMATION_BODY,
        ))
        .await
        .unwrap();
    let first_at = "2026-08-25T00:01:00Z".parse().unwrap();
    fixture
        .store
        .apply_villo_status(villo_status_batch(first_at, VILLO_STATUS_BODY))
        .await
        .unwrap();

    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM current_villo_status"), 2);
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM villo_status_fact"), 2);
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT bikes FROM current_villo_status
             JOIN place ON place.id = current_villo_status.stand_id
             WHERE place.external_id = '1'",
        ),
        6,
    );
    let status = blob(
        &fixture.path,
        "SELECT payload_zstd FROM provider_snapshot
         JOIN provider_feed ON provider_feed.id = provider_snapshot.feed_id
         WHERE provider_feed.provider = 'villo' AND provider_feed.feed = 'status'",
    );
    assert_eq!(
        zstd::stream::decode_all(status.as_slice()).unwrap(),
        VILLO_STATUS_BODY.as_bytes(),
    );
    fixture
        .store
        .record_feed_failure(
            "villo",
            "status",
            FeedStage::Parse,
            "temporary",
            Utc::now(),
        )
        .await
        .unwrap();

    let changed_at = "2026-08-25T00:04:00Z".parse().unwrap();
    fixture
        .store
        .apply_villo_status(villo_status_batch(changed_at, VILLO_STATUS_CHANGED_BODY))
        .await
        .unwrap();
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM feed_failure
             JOIN provider_feed ON provider_feed.id = feed_failure.feed_id
             WHERE provider_feed.provider = 'villo' AND provider_feed.feed = 'status'",
        ),
        0,
    );
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM villo_status_fact"), 2);
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT bikes FROM villo_status_fact
             JOIN place ON place.id = villo_status_fact.stand_id
             WHERE place.external_id = '1'",
        ),
        8,
    );

    let next_bucket_at = "2026-08-25T00:06:00Z".parse().unwrap();
    fixture
        .store
        .apply_villo_status(villo_status_batch(
            next_bucket_at,
            VILLO_STATUS_CHANGED_BODY,
        ))
        .await
        .unwrap();
    assert_eq!(scalar(&fixture.path, "SELECT count(*) FROM villo_status_fact"), 4);
}

#[tokio::test]
async fn failed_villo_status_commit_preserves_head_current_and_facts() {
    let fixture = fixture_store().await;
    let discovery_at = "2026-08-25T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_villo_discovery(villo_discovery_batch(
            discovery_at,
            VILLO_INFORMATION_BODY,
        ))
        .await
        .unwrap();
    let first_at = "2026-08-25T00:01:00Z".parse().unwrap();
    fixture
        .store
        .apply_villo_status(villo_status_batch(first_at, VILLO_STATUS_BODY))
        .await
        .unwrap();
    let head = feed_head(&fixture.path, "villo", "status");
    let snapshots = scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot");
    let current = scalar(&fixture.path, "SELECT sum(bikes) FROM current_villo_status");
    let facts = scalar(&fixture.path, "SELECT count(*) FROM villo_status_fact");

    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute_batch(
            "CREATE TRIGGER reject_villo_status
             BEFORE INSERT ON current_villo_status
             WHEN NEW.bikes = 999
             BEGIN
                 SELECT RAISE(ABORT, 'rejected villo status');
             END;",
        )
        .unwrap();
    let rejected_at = "2026-08-25T00:06:00Z".parse().unwrap();
    assert!(
        fixture
            .store
            .apply_villo_status(villo_status_batch(
                rejected_at,
                VILLO_STATUS_REJECTED_BODY,
            ))
            .await
            .is_err()
    );
    connection
        .execute_batch("DROP TRIGGER reject_villo_status")
        .unwrap();

    assert_eq!(feed_head(&fixture.path, "villo", "status"), head);
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot"),
        snapshots,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT sum(bikes) FROM current_villo_status"),
        current,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM villo_status_fact"),
        facts,
    );
}

#[tokio::test]
async fn villo_radius_is_inclusive_and_spatially_indexed() {
    let fixture = seeded_store().await;
    let rows = fixture
        .store
        .nearby_villo(50.8467, 4.3676)
        .await
        .unwrap();
    assert!(!rows.is_empty());
    assert!(
        rows.windows(2)
            .all(|pair| pair[0].distance_m <= pair[1].distance_m)
    );
    assert!(rows.iter().all(|row| row.distance_m <= 500.0));
}

#[tokio::test]
async fn villo_radius_includes_the_exact_floating_point_boundary() {
    let fixture = fixture_store().await;
    let captured_at = "2026-08-25T00:00:00Z".parse().unwrap();
    let information = r#"{"data":{"stations":[
      {"station_id":"boundary","name":"BOUNDARY","lat":0.004521847385157641,"lon":0.0}
    ]}}"#;
    let status = r#"{"data":{"stations":[
      {"station_id":"boundary","num_vehicles_available":1,"num_docks_available":2}
    ]}}"#;
    fixture
        .store
        .apply_villo_discovery(villo_discovery_batch(captured_at, information))
        .await
        .unwrap();
    fixture
        .store
        .apply_villo_status(villo_status_batch(captured_at, status))
        .await
        .unwrap();

    let rows = fixture.store.nearby_villo(0.0, 0.0).await.unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, "boundary");
    assert!(rows[0].distance_m > 499.999_999);
    assert!(rows[0].distance_m <= 500.0);
}

#[tokio::test]
async fn villo_radius_keeps_near_antimeridian_point_and_excludes_far_point() {
    let fixture = fixture_store().await;
    let captured_at = "2026-08-25T00:00:00Z".parse().unwrap();
    let information = r#"{"data":{"stations":[
      {"station_id":"near","name":"NEAR","lat":0.0,"lon":-179.999},
      {"station_id":"far","name":"FAR","lat":0.0,"lon":-179.99}
    ]}}"#;
    let status = r#"{"data":{"stations":[
      {"station_id":"near","num_vehicles_available":1,"num_docks_available":2},
      {"station_id":"far","num_vehicles_available":3,"num_docks_available":4}
    ]}}"#;
    fixture
        .store
        .apply_villo_discovery(villo_discovery_batch(captured_at, information))
        .await
        .unwrap();
    fixture
        .store
        .apply_villo_status(villo_status_batch(captured_at, status))
        .await
        .unwrap();

    let rows = fixture
        .store
        .nearby_villo(0.0, 179.999)
        .await
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, "near");
    assert!(rows[0].distance_m <= 500.0);
}

#[tokio::test]
async fn villo_radius_keeps_near_pole_point_and_excludes_far_point() {
    let fixture = fixture_store().await;
    let captured_at = "2026-08-25T00:00:00Z".parse().unwrap();
    let information = r#"{"data":{"stations":[
      {"station_id":"near","name":"NEAR","lat":89.998,"lon":179.999},
      {"station_id":"far","name":"FAR","lat":89.99,"lon":0.0}
    ]}}"#;
    let status = r#"{"data":{"stations":[
      {"station_id":"near","num_vehicles_available":1,"num_docks_available":2},
      {"station_id":"far","num_vehicles_available":3,"num_docks_available":4}
    ]}}"#;
    fixture
        .store
        .apply_villo_discovery(villo_discovery_batch(captured_at, information))
        .await
        .unwrap();
    fixture
        .store
        .apply_villo_status(villo_status_batch(captured_at, status))
        .await
        .unwrap();

    let rows = fixture.store.nearby_villo(89.998, 0.0).await.unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, "near");
    assert!(rows[0].distance_m <= 500.0);
}

#[tokio::test]
async fn villo_restart_serves_the_last_committed_counts() {
    let fixture = seeded_store().await;
    let before = fixture
        .store
        .nearby_villo(50.8467, 4.3676)
        .await
        .unwrap();
    let reopened = fixture.reopen().await;
    assert_eq!(
        reopened
            .nearby_villo(50.8467, 4.3676)
            .await
            .unwrap(),
        before,
    );
    assert_eq!(
        reopened.villo_status_url().await.unwrap().as_deref(),
        Some("https://villo.test/status"),
    );
}

const TRAIN_STATIONS_BODY: &str = r#"{"station":[
  {"id":"BE.NMBS.008812005","name":"Brussels-North","standardname":"BRUSSELS NORTH","locationX":"4.3661","locationY":"50.8535"},
  {"id":"BE.NMBS.008813003","name":"Brussels-Central","standardname":"BRUSSELS CENTRAL","locationX":"4.3700","locationY":"50.8535"}
]}"#;

const TRAIN_STATIONS_REFRESH_BODY: &str = r#"{"station":[
  {"id":"BE.NMBS.008812005","name":"Brussels-North","standardname":"BRUSSELS NORTH UPDATED","locationX":"4.3662","locationY":"50.8536"},
  {"id":"BE.NMBS.NEW","name":"New","standardname":"NEW STATION","locationX":"4.3720","locationY":"50.8535"}
]}"#;

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

fn train_stations_batch(
    captured_at: chrono::DateTime<Utc>,
    body: &str,
) -> TrainStationsBatch {
    TrainStationsBatch {
        raw: train_raw("stations", "", captured_at, body),
        stations: irail::parse_stations(body).unwrap(),
    }
}

fn train_departure(
    label: &str,
    destination: &str,
    in_seconds: i64,
    delay_seconds: Option<i64>,
) -> irail::TrainDeparture {
    irail::TrainDeparture {
        label: label.into(),
        destination: destination.into(),
        in_seconds,
        delayed: delay_seconds.unwrap_or(0) > 0,
        platform: Some("3".into()),
        delay_seconds,
    }
}

fn train_departures_batch(
    captured_at: chrono::DateTime<Utc>,
    station_id: &str,
    body: &str,
    departures: Vec<irail::TrainDeparture>,
) -> TrainDeparturesBatch {
    TrainDeparturesBatch {
        raw: train_raw("liveboard", station_id, captured_at, body),
        station_id: station_id.into(),
        departures,
    }
}

#[tokio::test]
async fn train_stations_persist_raw_and_reconcile_stable_provider_rows() {
    let fixture = fixture_store().await;
    fixture.store.apply_static(static_batch()).await.unwrap();
    let bmc_id = scalar(
        &fixture.path,
        "SELECT id FROM place WHERE provider = 'bmc' AND external_id = 'S0'",
    );
    let first_at = "2026-08-25T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_train_stations(train_stations_batch(first_at, TRAIN_STATIONS_BODY))
        .await
        .unwrap();
    let stable_id = scalar(
        &fixture.path,
        "SELECT id FROM place WHERE provider = 'irail' AND kind = 'train' AND external_id = 'BE.NMBS.008812005'",
    );
    let snapshot = blob(
        &fixture.path,
        "SELECT payload_zstd
         FROM provider_snapshot
         JOIN provider_feed ON provider_feed.id = provider_snapshot.feed_id
         WHERE provider_feed.provider = 'irail' AND provider_feed.feed = 'stations'",
    );
    assert_eq!(
        zstd::stream::decode_all(snapshot.as_slice()).unwrap(),
        TRAIN_STATIONS_BODY.as_bytes(),
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM provider_snapshot
             JOIN provider_feed ON provider_feed.id = provider_snapshot.feed_id
             WHERE provider_feed.provider = 'irail'
               AND provider_feed.feed = 'stations'
               AND provider_snapshot.scope = ''",
        ),
        1,
    );

    fixture
        .store
        .record_feed_failure(
            "irail",
            "stations",
            FeedStage::Provider,
            "temporary",
            Utc::now(),
        )
        .await
        .unwrap();
    let second_at = "2026-08-26T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_train_stations(train_stations_batch(
            second_at,
            TRAIN_STATIONS_REFRESH_BODY,
        ))
        .await
        .unwrap();

    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM place WHERE provider = 'irail' AND kind = 'train' AND external_id = 'BE.NMBS.008812005'",
        ),
        stable_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT active FROM place WHERE provider = 'irail' AND kind = 'train' AND external_id = 'BE.NMBS.008813003'",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT active FROM place WHERE provider = 'irail' AND kind = 'train' AND external_id = 'BE.NMBS.NEW'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM place WHERE provider = 'bmc' AND external_id = 'S0'",
        ),
        bmc_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM feed_failure
             JOIN provider_feed ON provider_feed.id = feed_failure.feed_id
             WHERE provider_feed.provider = 'irail' AND provider_feed.feed = 'stations'",
        ),
        0,
    );
}

#[tokio::test]
async fn failed_train_station_commit_preserves_head_snapshots_and_places() {
    let fixture = fixture_store().await;
    let first_at = "2026-08-25T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_train_stations(train_stations_batch(first_at, TRAIN_STATIONS_BODY))
        .await
        .unwrap();
    let head = feed_head(&fixture.path, "irail", "stations");
    let snapshots = scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot");
    let stable_id = scalar(
        &fixture.path,
        "SELECT id FROM place WHERE provider = 'irail' AND external_id = 'BE.NMBS.008812005'",
    );
    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute_batch(
            "CREATE TRIGGER reject_train_station
             BEFORE INSERT ON place
             WHEN NEW.provider = 'irail' AND NEW.external_id = 'BE.NMBS.NEW'
             BEGIN
                 SELECT RAISE(ABORT, 'rejected train station');
             END;",
        )
        .unwrap();

    let rejected_at = "2026-08-26T00:00:00Z".parse().unwrap();
    assert!(
        fixture
            .store
            .apply_train_stations(train_stations_batch(
                rejected_at,
                TRAIN_STATIONS_REFRESH_BODY,
            ))
            .await
            .is_err()
    );
    connection
        .execute_batch("DROP TRIGGER reject_train_station")
        .unwrap();

    assert_eq!(feed_head(&fixture.path, "irail", "stations"), head);
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot"),
        snapshots,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT id FROM place WHERE provider = 'irail' AND external_id = 'BE.NMBS.008812005'",
        ),
        stable_id,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT active FROM place WHERE provider = 'irail' AND external_id = 'BE.NMBS.008813003'",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place WHERE provider = 'irail' AND external_id = 'BE.NMBS.NEW'",
        ),
        0,
    );
}

#[tokio::test]
async fn train_departures_replace_current_rows_retain_facts_and_report_stale_ids() {
    let fixture = fixture_store().await;
    let station_at = "2026-08-25T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_train_stations(train_stations_batch(station_at, TRAIN_STATIONS_BODY))
        .await
        .unwrap();
    let first_at = "2026-08-25T00:01:00Z".parse().unwrap();
    fixture
        .store
        .apply_train_departures(train_departures_batch(
            first_at,
            "BE.NMBS.008812005",
            r#"{"departures":"first"}"#,
            vec![
                train_departure("IC", "LEUVEN", 120, Some(0)),
                train_departure("S1", "GENT", 60, Some(120)),
            ],
        ))
        .await
        .unwrap();
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_train_departure",
        ),
        2,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM train_departure_fact"),
        2,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM provider_snapshot
             JOIN provider_feed ON provider_feed.id = provider_snapshot.feed_id
             WHERE provider_feed.provider = 'irail'
               AND provider_feed.feed = 'liveboard'
               AND provider_snapshot.scope = 'BE.NMBS.008812005'",
        ),
        1,
    );

    fixture
        .store
        .record_feed_failure(
            "irail",
            "liveboard",
            FeedStage::Parse,
            "temporary",
            Utc::now(),
        )
        .await
        .unwrap();
    let second_at = first_at + Duration::seconds(10);
    fixture
        .store
        .apply_train_departures(train_departures_batch(
            second_at,
            "BE.NMBS.008812005",
            r#"{"departures":"second"}"#,
            vec![train_departure("IC", "LEUVEN", 110, Some(0))],
        ))
        .await
        .unwrap();
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_train_departure",
        ),
        1,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM train_departure_fact"),
        2,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT last_seen_at
             FROM train_departure_fact
             JOIN place ON place.id = train_departure_fact.station_id
             WHERE place.external_id = 'BE.NMBS.008812005'
               AND train_departure_fact.label = 'IC'",
        ),
        second_at.timestamp_millis(),
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM feed_failure
             JOIN provider_feed ON provider_feed.id = feed_failure.feed_id
             WHERE provider_feed.provider = 'irail' AND provider_feed.feed = 'liveboard'",
        ),
        0,
    );

    let fresh = fixture
        .store
        .nearby_trains(50.8535, 4.3661, second_at)
        .await
        .unwrap();
    assert_eq!(fresh.stops[0].id, "BE.NMBS.008812005");
    assert_eq!(fresh.stops[0].departures.len(), 1);
    assert_eq!(fresh.stops[0].departures[0].in_seconds, 110);
    assert!(!fresh.stops[0].departures[0].delayed);
    assert_eq!(
        fresh.refresh_station_ids,
        vec!["BE.NMBS.008813003".to_string()],
    );

    let stale = fixture
        .store
        .nearby_trains(
            50.8535,
            4.3661,
            second_at + Duration::seconds(30),
        )
        .await
        .unwrap();
    assert_eq!(
        stale.refresh_station_ids,
        vec![
            "BE.NMBS.008812005".to_string(),
            "BE.NMBS.008813003".to_string(),
        ],
    );
}

#[tokio::test]
async fn removed_train_station_releases_snapshot_and_prunes_after_fact_retention() {
    let fixture = fixture_store().await;
    let first_at = at("2026-01-01T00:00:00Z");
    fixture
        .store
        .apply_train_stations(train_stations_batch(first_at, TRAIN_STATIONS_BODY))
        .await
        .unwrap();
    fixture
        .store
        .apply_train_departures(train_departures_batch(
            first_at + Duration::minutes(1),
            "BE.NMBS.008813003",
            r#"{"departures":"removed"}"#,
            vec![train_departure("IC", "ANTWERP", 120, Some(0))],
        ))
        .await
        .unwrap();
    fixture
        .store
        .apply_train_departures(train_departures_batch(
            first_at + Duration::minutes(2),
            "BE.NMBS.008812005",
            r#"{"departures":"retained"}"#,
            vec![train_departure("S1", "LEUVEN", 180, Some(0))],
        ))
        .await
        .unwrap();
    assert!(fixture
        .store
        .nearby_trains(50.8535, 4.3700, first_at + Duration::minutes(2))
        .await
        .unwrap()
        .stops
        .iter()
        .any(|station| station.id == "BE.NMBS.008813003"
            && !station.departures.is_empty()));

    fixture
        .store
        .apply_train_stations(train_stations_batch(
            first_at + Duration::minutes(3),
            TRAIN_STATIONS_REFRESH_BODY,
        ))
        .await
        .unwrap();

    assert!(!fixture
        .store
        .nearby_trains(50.8535, 4.3700, first_at + Duration::minutes(3))
        .await
        .unwrap()
        .stops
        .iter()
        .any(|station| station.id == "BE.NMBS.008813003"));
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM current_train_departure
             JOIN place ON place.id = current_train_departure.station_id
             WHERE place.external_id = 'BE.NMBS.008813003'",
        ),
        0,
    );
    assert_eq!(
        feed_scope_snapshot_count(
            &fixture.path,
            "irail",
            "liveboard",
            "BE.NMBS.008813003",
        ),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*)
             FROM train_departure_fact
             JOIN place ON place.id = train_departure_fact.station_id
             WHERE place.external_id = 'BE.NMBS.008813003'",
        ),
        1,
    );

    fixture
        .store
        .maintain(first_at + Duration::days(8))
        .await
        .unwrap();
    assert_eq!(
        feed_scope_snapshot_count(
            &fixture.path,
            "irail",
            "liveboard",
            "BE.NMBS.008813003",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place
             WHERE provider = 'irail' AND external_id = 'BE.NMBS.008813003'",
        ),
        1,
    );

    fixture
        .store
        .maintain(first_at + Duration::days(91))
        .await
        .unwrap();
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM train_departure_fact
             JOIN place ON place.id = train_departure_fact.station_id
             WHERE place.external_id = 'BE.NMBS.008813003'",
        ),
        0,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM place
             WHERE provider = 'irail' AND external_id = 'BE.NMBS.008813003'",
        ),
        0,
    );
}

#[tokio::test]
async fn repeated_train_polls_share_identity_across_subsecond_capture_times() {
    let fixture = fixture_store().await;
    let station_at = "2026-08-25T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_train_stations(train_stations_batch(station_at, TRAIN_STATIONS_BODY))
        .await
        .unwrap();
    let first_at = "2026-08-25T00:01:00.100Z".parse().unwrap();
    fixture
        .store
        .apply_train_departures(train_departures_batch(
            first_at,
            "BE.NMBS.008812005",
            r#"{"departures":"first-subsecond"}"#,
            vec![train_departure("IC", "LEUVEN", 120, Some(0))],
        ))
        .await
        .unwrap();
    let second_at = "2026-08-25T00:01:00.900Z".parse().unwrap();
    fixture
        .store
        .apply_train_departures(train_departures_batch(
            second_at,
            "BE.NMBS.008812005",
            r#"{"departures":"second-subsecond"}"#,
            vec![train_departure("IC", "LEUVEN", 120, Some(0))],
        ))
        .await
        .unwrap();

    let expected_at = "2026-08-25T00:03:00Z"
        .parse::<chrono::DateTime<Utc>>()
        .unwrap()
        .timestamp_millis();
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM current_train_departure"),
        1,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM train_departure_fact"),
        1,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT expected_at FROM current_train_departure",
        ),
        expected_at,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT expected_at FROM train_departure_fact"),
        expected_at,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT last_seen_at FROM train_departure_fact",
        ),
        second_at.timestamp_millis(),
    );
}

#[tokio::test]
async fn failed_train_departure_commit_preserves_head_current_and_facts() {
    let fixture = fixture_store().await;
    let station_at = "2026-08-25T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_train_stations(train_stations_batch(station_at, TRAIN_STATIONS_BODY))
        .await
        .unwrap();
    let first_at = "2026-08-25T00:01:00Z".parse().unwrap();
    fixture
        .store
        .apply_train_departures(train_departures_batch(
            first_at,
            "BE.NMBS.008812005",
            r#"{"departures":"first"}"#,
            vec![train_departure("IC", "LEUVEN", 120, Some(0))],
        ))
        .await
        .unwrap();
    let head = feed_head(&fixture.path, "irail", "liveboard");
    let snapshots = scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot");
    let current = scalar(
        &fixture.path,
        "SELECT count(*) FROM current_train_departure",
    );
    let facts = scalar(&fixture.path, "SELECT count(*) FROM train_departure_fact");
    let connection = Connection::open(&fixture.path).unwrap();
    connection
        .execute_batch(
            "CREATE TRIGGER reject_train_departure
             BEFORE INSERT ON current_train_departure
             WHEN NEW.label = 'FAIL'
             BEGIN
                 SELECT RAISE(ABORT, 'rejected train departure');
             END;",
        )
        .unwrap();

    let rejected_at = first_at + Duration::minutes(1);
    assert!(
        fixture
            .store
            .apply_train_departures(train_departures_batch(
                rejected_at,
                "BE.NMBS.008812005",
                r#"{"departures":"rejected"}"#,
                vec![train_departure("FAIL", "NOWHERE", 60, None)],
            ))
            .await
            .is_err()
    );
    connection
        .execute_batch("DROP TRIGGER reject_train_departure")
        .unwrap();

    assert_eq!(feed_head(&fixture.path, "irail", "liveboard"), head);
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM provider_snapshot"),
        snapshots,
    );
    assert_eq!(
        scalar(
            &fixture.path,
            "SELECT count(*) FROM current_train_departure",
        ),
        current,
    );
    assert_eq!(
        scalar(&fixture.path, "SELECT count(*) FROM train_departure_fact"),
        facts,
    );
}

#[tokio::test]
async fn train_radius_is_inclusive_and_uses_exact_distance_order() {
    let fixture = fixture_store().await;
    let captured_at = "2026-08-25T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_train_stations(train_stations_batch(captured_at, TRAIN_STATIONS_BODY))
        .await
        .unwrap();
    let board = fixture
        .store
        .nearby_trains(50.8535, 4.3661, captured_at)
        .await
        .unwrap();
    assert!(!board.stops.is_empty());
    assert!(
        board
            .stops
            .windows(2)
            .all(|pair| pair[0].distance_m <= pair[1].distance_m),
    );
    assert!(board.stops.iter().all(|row| row.distance_m <= 1000.0));
}

#[tokio::test]
async fn train_radius_includes_the_exact_floating_point_boundary() {
    let fixture = fixture_store().await;
    let captured_at = "2026-08-25T00:00:00Z".parse().unwrap();
    let body = r#"{"station":[
      {"id":"boundary","name":"Boundary","standardname":"BOUNDARY","locationX":"0.0","locationY":"0.009043694769315282"}
    ]}"#;
    fixture
        .store
        .apply_train_stations(train_stations_batch(captured_at, body))
        .await
        .unwrap();

    let board = fixture
        .store
        .nearby_trains(0.0, 0.0, captured_at)
        .await
        .unwrap();
    assert_eq!(board.stops.len(), 1);
    assert_eq!(board.stops[0].id, "boundary");
    assert!(board.stops[0].distance_m > 999.999_999);
    assert!(board.stops[0].distance_m <= 1000.0);
}

#[tokio::test]
async fn train_radius_keeps_global_edge_points_and_excludes_far_points() {
    let fixture = fixture_store().await;
    let captured_at = "2026-08-25T00:00:00Z".parse().unwrap();
    let body = r#"{"station":[
      {"id":"antimeridian-near","name":"Near","standardname":"NEAR","locationX":"-179.999","locationY":"0.0"},
      {"id":"antimeridian-far","name":"Far","standardname":"FAR","locationX":"-179.98","locationY":"0.0"},
      {"id":"pole-near","name":"Near","standardname":"NEAR","locationX":"179.999","locationY":"89.998"},
      {"id":"pole-far","name":"Far","standardname":"FAR","locationX":"0.0","locationY":"89.98"}
    ]}"#;
    fixture
        .store
        .apply_train_stations(train_stations_batch(captured_at, body))
        .await
        .unwrap();

    let antimeridian = fixture
        .store
        .nearby_trains(0.0, 179.999, captured_at)
        .await
        .unwrap();
    assert_eq!(antimeridian.stops.len(), 1);
    assert_eq!(antimeridian.stops[0].id, "antimeridian-near");
    let pole = fixture
        .store
        .nearby_trains(89.998, 0.0, captured_at)
        .await
        .unwrap();
    assert_eq!(pole.stops.len(), 1);
    assert_eq!(pole.stops[0].id, "pole-near");
}

#[tokio::test]
async fn train_radius_keeps_near_polar_longitude_and_rejects_far_peer() {
    let fixture = fixture_store().await;
    let captured_at = "2026-08-25T00:00:00Z".parse().unwrap();
    let body = r#"{"station":[
      {"id":"polar-near","name":"Near","standardname":"NEAR","locationX":"53.0","locationY":"89.99"},
      {"id":"polar-far","name":"Far","standardname":"FAR","locationX":"54.0","locationY":"89.99"}
    ]}"#;
    fixture
        .store
        .apply_train_stations(train_stations_batch(captured_at, body))
        .await
        .unwrap();

    let board = fixture
        .store
        .nearby_trains(89.99, 0.0, captured_at)
        .await
        .unwrap();
    assert_eq!(board.stops.len(), 1);
    assert_eq!(board.stops[0].id, "polar-near");
    assert!(board.stops[0].distance_m <= 1000.0);
}

#[tokio::test]
async fn train_restart_serves_last_committed_departures() {
    let fixture = fixture_store().await;
    let captured_at = "2026-08-25T00:00:00Z".parse().unwrap();
    fixture
        .store
        .apply_train_stations(train_stations_batch(captured_at, TRAIN_STATIONS_BODY))
        .await
        .unwrap();
    fixture
        .store
        .apply_train_departures(train_departures_batch(
            captured_at,
            "BE.NMBS.008812005",
            r#"{"departures":"committed"}"#,
            vec![train_departure("IC", "LEUVEN", 120, Some(0))],
        ))
        .await
        .unwrap();
    let before = fixture
        .store
        .nearby_trains(50.8535, 4.3661, captured_at)
        .await
        .unwrap();
    let reopened = fixture.reopen().await;
    assert_eq!(
        reopened
            .nearby_trains(50.8535, 4.3661, captured_at)
            .await
            .unwrap(),
        before,
    );
}

#[tokio::test]
async fn train_request_reads_committed_data_and_only_schedules_refresh() {
    use http_body_util::BodyExt;
    use tower::ServiceExt;

    let fixture = fixture_store().await;
    let captured_at = Utc::now() - Duration::seconds(31);
    fixture
        .store
        .apply_train_stations(train_stations_batch(captured_at, TRAIN_STATIONS_BODY))
        .await
        .unwrap();
    fixture
        .store
        .apply_train_departures(train_departures_batch(
            captured_at,
            "BE.NMBS.008812005",
            r#"{"departures":"committed"}"#,
            vec![train_departure("IC", "LEUVEN", 180, Some(0))],
        ))
        .await
        .unwrap();
    let (tx, mut rx) = tokio::sync::mpsc::channel(8);
    let app = fixture.router_with_train_queue(tx);
    let request = axum::http::Request::builder()
        .uri("/v1/trains?lat=50.8535&lon=4.3661")
        .body(axum::body::Body::empty())
        .unwrap();

    let response = tokio::time::timeout(
        std::time::Duration::from_secs(1),
        app.oneshot(request),
    )
    .await
    .unwrap()
    .unwrap();
    assert_eq!(response.status(), axum::http::StatusCode::OK);
    let body = response.into_body().collect().await.unwrap().to_bytes();
    let value: serde_json::Value = serde_json::from_slice(&body).unwrap();
    assert_eq!(value["stops"][0]["rows"][0]["line"], "IC");
    assert_eq!(value["stops"][0]["rows"][0]["destination"]["fr"], "LEUVEN");
    let refresh = tokio::time::timeout(std::time::Duration::from_secs(1), rx.recv())
        .await
        .unwrap()
        .unwrap();
    assert!(refresh.contains(&"BE.NMBS.008812005".to_string()));
}