use super::{
    LineDetailData, LineDetailUnavailable, LineDisruptionData, LineStripStopData, SearchHitData,
    Store, TrainBoardData, TrainStopData, TransitBoardData, TransitStopData, VilloBoardData,
    VilloStopData,
};
use crate::{
    api::{collapse_dupes, fold},
    departure::{Badge, Departure},
    model::{Bilingual, VehicleKind},
};
use anyhow::Context;
use deadpool_sqlite::rusqlite::{params, OptionalExtension};
use std::collections::{BTreeMap, HashMap, HashSet};

const KNN_RADIUS_DEGREES: f64 = 0.01;
const TRANSIT_SELECT_SQL: &str = r#"
WITH
passage_candidates AS (
    SELECT ranked.place_id, ranked.source_order, 2 AS rank, knn2.distance_m
    FROM knn2
    JOIN transit_with_passages AS ranked ON ranked.rowid = knn2.fid
    WHERE knn2.f_table_name = 'transit_with_passages'
      AND knn2.f_geometry_column = 'geom'
      AND knn2.ref_geometry = MakePoint(?1, ?2, 4326)
      AND knn2.radius = ?3
      AND knn2.max_items = ?4
      AND knn2.expand = 1
),
notice_candidates AS (
    SELECT ranked.place_id, ranked.source_order, 1 AS rank, knn2.distance_m
    FROM knn2
    JOIN transit_with_notices AS ranked ON ranked.rowid = knn2.fid
    WHERE knn2.f_table_name = 'transit_with_notices'
      AND knn2.f_geometry_column = 'geom'
      AND knn2.ref_geometry = MakePoint(?1, ?2, 4326)
      AND knn2.radius = ?3
      AND knn2.max_items = ?4
      AND knn2.expand = 1
),
silent_candidates AS (
    SELECT ranked.place_id, ranked.source_order, 0 AS rank, knn2.distance_m
    FROM knn2
    JOIN transit_silent AS ranked ON ranked.rowid = knn2.fid
    WHERE knn2.f_table_name = 'transit_silent'
      AND knn2.f_geometry_column = 'geom'
      AND knn2.ref_geometry = MakePoint(?1, ?2, 4326)
      AND knn2.radius = ?3
      AND knn2.max_items = ?4
      AND knn2.expand = 1
),
all_candidates AS (
    SELECT * FROM passage_candidates
    UNION ALL
    SELECT * FROM notice_candidates
    UNION ALL
    SELECT * FROM silent_candidates
),
selected AS MATERIALIZED (
    SELECT place_id, source_order, rank, distance_m
    FROM all_candidates
    ORDER BY rank DESC, distance_m, source_order
    LIMIT ?4
),
"#;

const STOPS_SELECT_SQL: &str = r#"
WITH
selected AS MATERIALIZED (
    SELECT place.id AS place_id, place.source_order, 0 AS rank, 0.0 AS distance_m
    FROM place
    WHERE place.external_id IN (SELECT value FROM json_each(?1))
      AND place.provider = 'bmc'
      AND place.kind = 'transit'
      AND place.active = 1
),
"#;

const BOARD_DETAILS_SQL: &str = r#"
details AS (
    SELECT
        selected.place_id,
        selected.source_order,
        selected.rank,
        selected.distance_m,
        place.external_id,
        place.name_fr,
        place.name_nl,
        0 AS detail_kind,
        NULL AS line_id,
        NULL AS line_external,
        NULL AS line_name,
        NULL AS vehicle_kind,
        NULL AS color,
        NULL AS text_color,
        NULL AS destination_fr,
        NULL AS destination_nl,
        NULL AS expected_at,
        NULL AS message_fr,
        NULL AS message_nl,
        0 AS disrupted
    FROM selected
    JOIN place ON place.id = selected.place_id

    UNION ALL

    SELECT
        selected.place_id,
        selected.source_order,
        selected.rank,
        selected.distance_m,
        place.external_id,
        place.name_fr,
        place.name_nl,
        1 AS detail_kind,
        line.id AS line_id,
        line.external_id AS line_external,
        line.display_name AS line_name,
        line.kind AS vehicle_kind,
        line.color,
        line.text_color,
        passage.destination_fr,
        passage.destination_nl,
        passage.expected_at,
        NULL AS message_fr,
        NULL AS message_nl,
        EXISTS(
            SELECT 1
            FROM current_disruption_line AS disruption_line
            WHERE disruption_line.line_id = line.id
        ) AS disrupted
    FROM selected
    JOIN place ON place.id = selected.place_id
    CROSS JOIN current_passage AS passage ON passage.place_id = selected.place_id
    JOIN line ON line.id = passage.line_id AND line.active = 1
    WHERE passage.expected_at >= $CUTOFF

    UNION ALL

    SELECT
        selected.place_id,
        selected.source_order,
        selected.rank,
        selected.distance_m,
        place.external_id,
        place.name_fr,
        place.name_nl,
        2 AS detail_kind,
        line.id AS line_id,
        line.external_id AS line_external,
        line.display_name AS line_name,
        line.kind AS vehicle_kind,
        line.color,
        line.text_color,
        NULL AS destination_fr,
        NULL AS destination_nl,
        NULL AS expected_at,
        notice.message_fr,
        notice.message_nl,
        EXISTS(
            SELECT 1
            FROM current_disruption_line AS disruption_line
            WHERE disruption_line.line_id = line.id
        ) AS disrupted
    FROM selected
    JOIN place ON place.id = selected.place_id
    CROSS JOIN current_end_notice AS notice ON notice.place_id = selected.place_id
    JOIN line ON line.id = notice.line_id AND line.active = 1
)
SELECT *
FROM details
ORDER BY rank DESC, distance_m, source_order, detail_kind, line_external, expected_at
"#;

static TRANSIT_SQL: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
    format!("{}{}", TRANSIT_SELECT_SQL, BOARD_DETAILS_SQL.replace("$CUTOFF", "?5"))
});

static STOPS_SQL: std::sync::LazyLock<String> = std::sync::LazyLock::new(|| {
    format!("{}{}", STOPS_SELECT_SQL, BOARD_DETAILS_SQL.replace("$CUTOFF", "?2"))
});

const VILLO_RADIUS_M: f64 = 500.0;
const MIN_METERS_PER_DEGREE: f64 = 110_000.0;
const VILLO_SQL: &str = r#"
WITH spatial_candidates(rowid) AS MATERIALIZED (
    SELECT rowid
    FROM SpatialIndex AS VirtualSpatialIndex
    WHERE f_table_name = 'place'
      AND f_geometry_column = 'geom'
      AND search_frame = BuildMbr(?3, ?4, ?5, ?6, 4326)
),
measured AS MATERIALIZED (
    SELECT
        place.external_id,
        place.name_fr,
        place.name_nl,
        Distance(place.geom, MakePoint(?1, ?2, 4326), 1) AS distance_m,
        status.bikes,
        status.docks,
        status.mechanical,
        status.electrical,
        place.source_order
    FROM spatial_candidates
    JOIN place ON place.rowid = spatial_candidates.rowid
    JOIN current_villo_status AS status ON status.stand_id = place.id
    WHERE place.provider = 'villo'
      AND place.kind = 'bike'
      AND place.active = 1
)
SELECT external_id, name_fr, name_nl, distance_m, bikes, docks, mechanical, electrical
FROM measured
WHERE distance_m <= 500.0
ORDER BY distance_m, source_order
"#;

const TRAIN_RADIUS_M: f64 = 1000.0;
const TRAIN_SQL: &str = r#"
WITH spatial_candidates(rowid) AS MATERIALIZED (
    SELECT rowid
    FROM SpatialIndex AS VirtualSpatialIndex
    WHERE f_table_name = 'place'
      AND f_geometry_column = 'geom'
      AND search_frame = BuildMbr(?3, ?4, ?5, ?6, 4326)
),
measured AS MATERIALIZED (
    SELECT
        place.id,
        place.external_id,
        place.name_fr,
        Distance(place.geom, MakePoint(?1, ?2, 4326), 1) AS distance_m,
        (
            SELECT snapshot.captured_at
            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 = place.external_id
            ORDER BY snapshot.captured_at DESC, snapshot.id DESC
            LIMIT 1
        ) AS liveboard_captured_at
    FROM spatial_candidates
    JOIN place ON place.rowid = spatial_candidates.rowid
    WHERE place.provider = 'irail'
      AND place.kind = 'train'
      AND place.active = 1
),
selected AS MATERIALIZED (
    SELECT *
    FROM measured
    WHERE distance_m <= 1000.0
)
SELECT
    selected.id,
    selected.external_id,
    selected.name_fr,
    selected.distance_m,
    selected.liveboard_captured_at,
    departure.label,
    departure.destination,
    CASE
        WHEN departure.expected_at IS NULL THEN NULL
        ELSE (departure.expected_at - ?7) / 1000
    END AS in_seconds,
    departure.platform,
    departure.delay_seconds
FROM selected
LEFT JOIN current_train_departure AS departure ON departure.station_id = selected.id
ORDER BY
    selected.distance_m,
    selected.external_id,
    departure.expected_at,
    departure.label,
    departure.destination
"#;

const SEARCH_SQL: &str = r#"
WITH
exact_hits(place_id) AS MATERIALIZED (
    SELECT place_id
    FROM place_search
    WHERE place_search MATCH ?1
      AND instr(folded_name, ?2) > 0
    UNION
    SELECT id
    FROM place INDEXED BY place_normalized_lookup
    WHERE provider = 'bmc'
      AND kind = 'transit'
      AND normalized_id >= ?3
      AND normalized_id < ?4
      AND active = 1
),
collapsed_hits(place_id) AS (
    SELECT place_id
    FROM place_search
    WHERE place_search MATCH ?5
      AND instr(collapsed_name, ?6) > 0
),
candidates AS (
    SELECT place_id, 0 AS tier FROM exact_hits
    UNION ALL
    SELECT collapsed_hits.place_id, 1 AS tier
    FROM collapsed_hits
    WHERE NOT EXISTS (
        SELECT 1 FROM exact_hits WHERE exact_hits.place_id = collapsed_hits.place_id
    )
),
selected AS MATERIALIZED (
    SELECT place.id, place.external_id, place.name_fr, place.name_nl, candidates.tier, place.source_order
    FROM candidates
    JOIN place ON place.id = candidates.place_id
    WHERE place.provider = 'bmc'
      AND place.kind = 'transit'
      AND place.active = 1
    ORDER BY candidates.tier, place.source_order
    LIMIT 20
)
SELECT
    selected.external_id,
    selected.name_fr,
    selected.name_nl,
    (
        SELECT count(DISTINCT passage.line_id)
        FROM current_passage AS passage
        JOIN line ON line.id = passage.line_id AND line.active = 1
        WHERE passage.place_id = selected.id
    ) AS line_count,
    (
        SELECT variant.destination_fr
        FROM route_variant_stop AS variant_stop
        JOIN route_variant AS variant ON variant.id = variant_stop.variant_id
        JOIN line ON line.id = variant.line_id AND line.active = 1
        WHERE variant_stop.place_id = selected.id
        ORDER BY line.external_id, variant.source_order, variant.id
        LIMIT 1
    ) AS direction_fr,
    (
        SELECT variant.destination_nl
        FROM route_variant_stop AS variant_stop
        JOIN route_variant AS variant ON variant.id = variant_stop.variant_id
        JOIN line ON line.id = variant.line_id AND line.active = 1
        WHERE variant_stop.place_id = selected.id
        ORDER BY line.external_id, variant.source_order, variant.id
        LIMIT 1
    ) AS direction_nl
FROM selected
ORDER BY selected.tier, selected.source_order
"#;

const SEARCH_SHORT_ID_SQL: &str = r#"
WITH selected AS MATERIALIZED (
    SELECT id, external_id, name_fr, name_nl, source_order
    FROM place INDEXED BY place_normalized_lookup
    WHERE provider = 'bmc'
      AND kind = 'transit'
      AND normalized_id >= ?1
      AND normalized_id < ?2
      AND active = 1
    ORDER BY source_order
    LIMIT 20
)
SELECT
    selected.external_id,
    selected.name_fr,
    selected.name_nl,
    (
        SELECT count(DISTINCT passage.line_id)
        FROM current_passage AS passage
        JOIN line ON line.id = passage.line_id AND line.active = 1
        WHERE passage.place_id = selected.id
    ) AS line_count,
    (
        SELECT variant.destination_fr
        FROM route_variant_stop AS variant_stop
        JOIN route_variant AS variant ON variant.id = variant_stop.variant_id
        JOIN line ON line.id = variant.line_id AND line.active = 1
        WHERE variant_stop.place_id = selected.id
        ORDER BY line.external_id, variant.source_order, variant.id
        LIMIT 1
    ) AS direction_fr,
    (
        SELECT variant.destination_nl
        FROM route_variant_stop AS variant_stop
        JOIN route_variant AS variant ON variant.id = variant_stop.variant_id
        JOIN line ON line.id = variant.line_id AND line.active = 1
        WHERE variant_stop.place_id = selected.id
        ORDER BY line.external_id, variant.source_order, variant.id
        LIMIT 1
    ) AS direction_nl
FROM selected
ORDER BY selected.source_order
"#;

const LINE_VARIANT_SQL: &str = r#"
WITH target AS MATERIALIZED (
    SELECT alias.place_id
    FROM place_alias AS alias
    JOIN place ON place.id = alias.place_id
    WHERE alias.provider = 'bmc'
      AND alias.external_id = ?2
      AND place.provider = 'bmc'
      AND place.kind = 'transit'
      AND place.active = 1
    ORDER BY alias.priority, place.source_order, place.id
    LIMIT 1
),
selected AS MATERIALIZED (
    SELECT
        line.id AS line_id,
        variant.id AS variant_id,
        variant.destination_fr,
        variant.destination_nl,
        MIN(target_stop.ordinal) AS target_ordinal,
        target.place_id AS target_place_id
    FROM line
    JOIN route_variant AS variant ON variant.line_id = line.id
    JOIN target ON 1 = 1
    JOIN route_variant_stop AS target_stop
      ON target_stop.variant_id = variant.id
     AND target_stop.place_id = target.place_id
    WHERE line.provider = 'bmc'
      AND line.external_id = ?1
      AND line.active = 1
    GROUP BY
        line.id,
        variant.id,
        variant.destination_fr,
        variant.destination_nl,
        variant.source_order,
        target.place_id
    ORDER BY variant.source_order, variant.id
    LIMIT 1
)
SELECT
    line.id,
    line.display_name,
    line.kind,
    line.color,
    line.text_color,
    selected.variant_id,
    selected.destination_fr,
    selected.destination_nl,
    selected.target_ordinal,
    selected.target_place_id,
    (
        SELECT stop.place_id
        FROM route_variant_stop AS stop
        WHERE stop.variant_id = selected.variant_id
        ORDER BY stop.ordinal DESC
        LIMIT 1
    ) AS terminus_place_id
FROM selected
JOIN line ON line.id = selected.line_id
"#;

const LINE_STOPS_SQL: &str = r#"
SELECT
    variant_stop.ordinal,
    COALESCE(place.external_id, variant_stop.external_id),
    COALESCE(place.name_fr, variant_stop.external_id),
    COALESCE(place.name_nl, variant_stop.external_id),
    Y(place.geom),
    X(place.geom)
FROM route_variant_stop AS variant_stop
LEFT JOIN place ON place.id = variant_stop.place_id
WHERE variant_stop.variant_id = ?1
  AND variant_stop.ordinal BETWEEN max(?2 - 5, 0) AND ?2
ORDER BY variant_stop.ordinal
"#;

const LINE_VEHICLES_SQL: &str = r#"
WITH terminus AS MATERIALIZED (
    SELECT place_id, external_id
    FROM route_variant_stop
    WHERE variant_id = ?1
    ORDER BY ordinal DESC
    LIMIT 1
),
matched AS MATERIALIZED (
    SELECT
        MIN(stop.ordinal) AS ordinal,
        position.distance_m
    FROM current_vehicle_position AS position
    JOIN terminus ON
        (
            terminus.place_id IS NOT NULL
            AND terminus.place_id = position.terminus_place_id
        )
        OR (
            terminus.place_id IS NULL
            AND bmc_ids_match(terminus.external_id, position.terminus_external_id)
        )
    JOIN route_variant_stop AS stop ON
        stop.variant_id = ?1
        AND (
            (
                stop.place_id IS NOT NULL
                AND stop.place_id = position.point_place_id
            )
            OR (
                stop.place_id IS NULL
                AND bmc_ids_match(stop.external_id, position.point_external_id)
            )
        )
    WHERE position.line_id = ?2
    GROUP BY position.rowid, position.distance_m
)
SELECT ordinal, MIN(distance_m)
FROM matched
WHERE ordinal BETWEEN max(?3 - 5, 0) AND ?3
GROUP BY ordinal
ORDER BY ordinal
"#;

const LINE_PASSAGES_SQL: &str = r#"
SELECT destination_fr, destination_nl, expected_at
FROM current_passage
WHERE place_id = ?1
  AND line_id = ?2
ORDER BY rowid
"#;

const LINE_DISRUPTIONS_SQL: &str = r#"
SELECT
    disruption.priority,
    disruption.message_en,
    disruption.message_fr,
    disruption.message_nl,
    EXISTS (
        SELECT 1
        FROM current_disruption_place AS disruption_place
        WHERE disruption_place.disruption_id = disruption.id
          AND disruption_place.place_id = ?2
    )
FROM current_disruption_line AS disruption_line
JOIN current_disruption AS disruption ON disruption.id = disruption_line.disruption_id
WHERE disruption_line.line_id = ?1
ORDER BY disruption.id
"#;

const LINE_SHAPE_SQL: &str = r#"
SELECT AsGeoJSON(route_shape.geom, 15, 0)
FROM route_shape
JOIN place_alias AS alias
  ON alias.provider = 'bmc'
 AND alias.external_id = route_shape.destination_key
WHERE route_shape.line_id = ?1
  AND alias.place_id = ?2
ORDER BY alias.priority, route_shape.id
LIMIT 1
"#;

fn spatial_bounds(lat: f64, lon: f64, radius_m: f64) -> (f64, f64, f64, f64) {
    let latitude_delta = radius_m / MIN_METERS_PER_DEGREE;
    let min_lat = lat - latitude_delta;
    let max_lat = lat + latitude_delta;
    if min_lat < -90.0 || max_lat > 90.0 {
        return (-180.0, -90.0, 180.0, 90.0);
    }
    let edge_latitude = min_lat.abs().max(max_lat.abs());
    let longitude_scale = edge_latitude.to_radians().cos().abs();
    let longitude_delta = if longitude_scale <= f64::EPSILON {
        180.0
    } else {
        (radius_m / (MIN_METERS_PER_DEGREE * longitude_scale)).min(180.0)
    };
    let min_lon = lon - longitude_delta;
    let max_lon = lon + longitude_delta;
    if min_lon < -180.0 || max_lon > 180.0 {
        (-180.0, -90.0, 180.0, 90.0)
    } else {
        (min_lon, min_lat, max_lon, max_lat)
    }
}

struct LineSelection {
    line_id: i64,
    line: String,
    kind: String,
    color: String,
    text_color: String,
    variant_id: i64,
    destination_fr: String,
    destination_nl: String,
    target_ordinal: i64,
    target_place_id: i64,
    terminus_place_id: Option<i64>,
}

struct LineStopRow {
    ordinal: i64,
    id: String,
    name: Bilingual,
    lat: Option<f64>,
    lon: Option<f64>,
}

struct LinePassageRow {
    destination_fr: String,
    destination_nl: String,
    expected_at: i64,
}

struct LineDisruptionRow {
    priority: i64,
    en: Option<String>,
    fr: Option<String>,
    nl: Option<String>,
    target: bool,
}

#[derive(serde::Deserialize)]
struct LineGeometry {
    coordinates: Vec<(f64, f64)>,
}

struct TimedDeparture {
    badge: Badge,
    destination: Bilingual,
    in_seconds: Vec<i64>,
    disrupted: bool,
}

struct NoticeDeparture {
    line_id: i64,
    badge: Badge,
    message: Bilingual,
    disrupted: bool,
}

struct StopBuilder {
    stop: TransitStopData,
    timed: BTreeMap<i64, TimedDeparture>,
    notices: Vec<NoticeDeparture>,
}

fn waiting_captured_at(
    transaction: &deadpool_sqlite::rusqlite::Transaction<'_>,
) -> anyhow::Result<Option<i64>> {
    Ok(transaction
        .query_row(
            "SELECT snapshot.captured_at
             FROM provider_feed AS feed
             JOIN provider_snapshot AS snapshot ON snapshot.id = feed.current_snapshot_id
             WHERE feed.provider = 'bmc' AND feed.feed = 'waiting'",
            [],
            |row| row.get::<_, i64>(0),
        )
        .optional()?)
}

fn assemble_transit_stops(
    rows: &mut deadpool_sqlite::rusqlite::Rows<'_>,
    now_ms: i64,
) -> anyhow::Result<Vec<TransitStopData>> {
    let mut stops = Vec::<StopBuilder>::new();
    let mut stop_indexes = HashMap::<i64, usize>::new();
    while let Some(row) = rows.next()? {
        let place_id = row.get::<_, i64>(0)?;
        let detail_kind = row.get::<_, i64>(7)?;
        let index = if let Some(index) = stop_indexes.get(&place_id).copied() {
            index
        } else {
            let index = stops.len();
            stop_indexes.insert(place_id, index);
            stops.push(StopBuilder {
                stop: TransitStopData {
                    id: row.get(4)?,
                    name: Bilingual { fr: row.get(5)?, nl: row.get(6)? },
                    distance_m: row.get::<_, f64>(3)? as u32,
                    rank: row.get::<_, u8>(2)?,
                    departures: Vec::new(),
                },
                timed: BTreeMap::new(),
                notices: Vec::new(),
            });
            index
        };
        if detail_kind == 0 {
            continue;
        }
        let line_id = row.get::<_, i64>(8)?;
        let badge = Badge {
            line: row.get(10)?,
            kind: parse_vehicle_kind(&row.get::<_, String>(11)?)?,
            color: row.get(12)?,
            text_color: row.get(13)?,
        };
        let disrupted = row.get(19)?;
        if detail_kind == 1 {
            let expected_at = row.get::<_, i64>(16)?;
            if let Some(group) = stops[index].timed.get_mut(&line_id) {
                group.in_seconds.push(((expected_at - now_ms) / 1000).max(0));
            } else {
                stops[index].timed.insert(
                    line_id,
                    TimedDeparture {
                        badge,
                        destination: Bilingual { fr: row.get(14)?, nl: row.get(15)? },
                        in_seconds: vec![((expected_at - now_ms) / 1000).max(0)],
                        disrupted,
                    },
                );
            }
        } else {
            stops[index].notices.push(NoticeDeparture {
                line_id,
                badge,
                message: Bilingual {
                    fr: row.get::<_, Option<String>>(17)?.unwrap_or_default(),
                    nl: row.get::<_, Option<String>>(18)?.unwrap_or_default(),
                },
                disrupted,
            });
        }
    }
    Ok(stops
        .into_iter()
        .map(|mut builder| {
            let timed_lines: HashSet<i64> = builder.timed.keys().copied().collect();
            let mut timed = builder
                .timed
                .into_values()
                .map(|mut departure| {
                    departure.in_seconds.sort_unstable();
                    let first = departure.in_seconds.first().copied().unwrap_or(i64::MAX);
                    let line = departure.badge.line.clone();
                    (
                        first,
                        line,
                        Departure::Transit {
                            badge: departure.badge,
                            destination: departure.destination,
                            in_seconds: departure.in_seconds,
                            disrupted: departure.disrupted,
                        },
                    )
                })
                .collect::<Vec<_>>();
            timed.sort_by(|left, right| (left.0, &left.1).cmp(&(right.0, &right.1)));
            builder.stop.departures.extend(timed.into_iter().map(|(_, _, row)| row));
            builder.stop.departures.extend(
                builder
                    .notices
                    .into_iter()
                    .filter(|notice| !timed_lines.contains(&notice.line_id))
                    .map(|notice| Departure::EndOfService {
                        badge: notice.badge,
                        message: notice.message,
                        disrupted: notice.disrupted,
                    }),
            );
            builder.stop
        })
        .collect())
}

impl Store {
    pub async fn line_detail(
        &self,
        line: &str,
        to: &str,
        destination: Option<&str>,
        now: chrono::DateTime<chrono::Utc>,
    ) -> anyhow::Result<Option<LineDetailData>> {
        let line = crate::bmc::normalize_id(line);
        let to = crate::bmc::normalize_id(to);
        let destination = destination.unwrap_or_default().to_owned();
        let now_ms = now.timestamp_millis();
        let (ready, detail) = self
            .read(
                "line_detail",
                move |connection| -> anyhow::Result<_> {
                    let transaction = connection.transaction()?;
                    let ready = transaction.query_row(
                        "SELECT EXISTS (
                            SELECT 1
                            FROM place
                            WHERE provider = 'bmc'
                              AND kind = 'transit'
                              AND active = 1
                        )",
                        [],
                        |row| Ok(row.get::<_, i64>(0)? != 0),
                    )?;
                    if !ready {
                        transaction.commit()?;
                        return Ok((false, None));
                    }
                    let selection = {
                        let mut statement = transaction.prepare(LINE_VARIANT_SQL)?;
                        statement
                            .query_row(params![line, to], |row| {
                                Ok(LineSelection {
                                    line_id: row.get(0)?,
                                    line: row.get(1)?,
                                    kind: row.get(2)?,
                                    color: row.get(3)?,
                                    text_color: row.get(4)?,
                                    variant_id: row.get(5)?,
                                    destination_fr: row.get(6)?,
                                    destination_nl: row.get(7)?,
                                    target_ordinal: row.get(8)?,
                                    target_place_id: row.get(9)?,
                                    terminus_place_id: row.get(10)?,
                                })
                            })
                            .optional()?
                    };
                    let Some(selection) = selection else {
                        transaction.commit()?;
                        return Ok((true, None));
                    };
                    let stops = {
                        let mut statement = transaction.prepare(LINE_STOPS_SQL)?;
                        let rows = statement.query_map(
                            params![selection.variant_id, selection.target_ordinal],
                            |row| {
                                Ok(LineStopRow {
                                    ordinal: row.get(0)?,
                                    id: row.get(1)?,
                                    name: Bilingual {
                                        fr: row.get(2)?,
                                        nl: row.get(3)?,
                                    },
                                    lat: row.get(4)?,
                                    lon: row.get(5)?,
                                })
                            },
                        )?;
                        rows.collect::<Result<Vec<_>, _>>()?
                    };
                    let markers = {
                        let mut statement = transaction.prepare(LINE_VEHICLES_SQL)?;
                        let rows = statement.query_map(
                            params![
                                selection.variant_id,
                                selection.line_id,
                                selection.target_ordinal,
                            ],
                            |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)),
                        )?;
                        let mut markers = HashMap::new();
                        for row in rows {
                            let (ordinal, distance_m) = row?;
                            markers.insert(
                                ordinal,
                                u32::try_from(distance_m)
                                    .context("line vehicle distance is out of range")?,
                            );
                        }
                        markers
                    };
                    let passages = {
                        let mut statement = transaction.prepare(LINE_PASSAGES_SQL)?;
                        let rows = statement.query_map(
                            params![selection.target_place_id, selection.line_id],
                            |row| {
                                Ok(LinePassageRow {
                                    destination_fr: row.get(0)?,
                                    destination_nl: row.get(1)?,
                                    expected_at: row.get(2)?,
                                })
                            },
                        )?;
                        rows.collect::<Result<Vec<_>, _>>()?
                    };
                    let disruptions = {
                        let mut statement = transaction.prepare(LINE_DISRUPTIONS_SQL)?;
                        let rows = statement.query_map(
                            params![selection.line_id, selection.target_place_id],
                            |row| {
                                Ok(LineDisruptionRow {
                                    priority: row.get(0)?,
                                    en: row.get(1)?,
                                    fr: row.get(2)?,
                                    nl: row.get(3)?,
                                    target: row.get::<_, i64>(4)? != 0,
                                })
                            },
                        )?;
                        rows.collect::<Result<Vec<_>, _>>()?
                    };
                    let shape_json = {
                        let mut statement = transaction.prepare(LINE_SHAPE_SQL)?;
                        statement
                            .query_row(
                                params![selection.line_id, selection.terminus_place_id],
                                |row| row.get::<_, String>(0),
                            )
                            .optional()?
                    };
                    transaction.commit()?;

                    let response_destination = if destination.is_empty() {
                        passages
                            .iter()
                            .min_by_key(|passage| passage.expected_at)
                            .map_or_else(
                                || Bilingual {
                                    fr: selection.destination_fr.clone(),
                                    nl: selection.destination_nl.clone(),
                                },
                                |passage| Bilingual {
                                    fr: passage.destination_fr.clone(),
                                    nl: passage.destination_nl.clone(),
                                },
                            )
                    } else {
                        passages
                            .iter()
                            .find(|passage| {
                                passage.destination_fr == destination
                                    || passage.destination_nl == destination
                            })
                            .map_or_else(
                                || Bilingual {
                                    fr: selection.destination_fr.clone(),
                                    nl: selection.destination_nl.clone(),
                                },
                                |passage| Bilingual {
                                    fr: passage.destination_fr.clone(),
                                    nl: passage.destination_nl.clone(),
                                },
                            )
                    };
                    let mut in_seconds = passages
                        .iter()
                        .filter_map(|passage| {
                            let seconds = passage.expected_at.saturating_sub(now_ms) / 1000;
                            (seconds >= -30).then_some(seconds.max(0))
                        })
                        .collect::<Vec<_>>();
                    in_seconds.sort_unstable();

                    let target_disruption =
                        disruptions.iter().any(|disruption| disruption.target);
                    let disruption = disruptions
                        .iter()
                        .filter(|disruption| !target_disruption || disruption.target)
                        .min_by_key(|disruption| disruption.priority)
                        .map(|disruption| LineDisruptionData {
                            en: disruption.en.clone(),
                            fr: disruption.fr.clone(),
                            nl: disruption.nl.clone(),
                        });

                    let shape = match shape_json {
                        Some(shape_json) => {
                            let geometry: LineGeometry = serde_json::from_str(&shape_json)
                                .context("line shape is not valid GeoJSON")?;
                            let points = geometry
                                .coordinates
                                .into_iter()
                                .map(|(lon, lat)| (lat as f32, lon as f32))
                                .collect::<Vec<_>>();
                            let first = stops.iter().find_map(|stop| stop.lat.zip(stop.lon));
                            let last = stops
                                .iter()
                                .rev()
                                .find_map(|stop| stop.lat.zip(stop.lon));
                            match (first, last) {
                                (Some(first), Some(last)) => {
                                    crate::shapes::section(&points, first, last)
                                        .map(|section| crate::shapes::simplify(&section, 24))
                                }
                                _ => None,
                            }
                        }
                        None => None,
                    };
                    let strip = stops
                        .into_iter()
                        .map(|stop| LineStripStopData {
                            id: stop.id,
                            name: stop.name,
                            you: stop.ordinal == selection.target_ordinal,
                            vehicle_after_m: markers.get(&stop.ordinal).copied(),
                        })
                        .collect();
                    Ok((
                        true,
                        Some(LineDetailData {
                            line: selection.line,
                            kind: parse_vehicle_kind(&selection.kind)?,
                            color: selection.color,
                            text_color: selection.text_color,
                            destination: response_destination,
                            strip,
                            in_seconds,
                            disruption,
                            shape,
                        }),
                    ))
                },
            )
            .await?;
        if ready {
            Ok(detail)
        } else {
            Err(LineDetailUnavailable.into())
        }
    }

    pub async fn transit_board(
        &self,
        lat: f64,
        lon: f64,
        n: usize,
        now: chrono::DateTime<chrono::Utc>,
    ) -> anyhow::Result<TransitBoardData> {
        let now_ms = now.timestamp_millis();
        let max_items = i64::try_from(n.min(1024))?;
        self.read(
            "transit_board",
            move |connection| {
                let transaction = connection.transaction()?;
                let captured_at = waiting_captured_at(&transaction)?;
                let data_age_seconds = captured_at.map(|captured_at| (now_ms - captured_at) / 1000);
                let stale = data_age_seconds.map(|age| age > 90).unwrap_or(true);
                if max_items == 0 {
                    transaction.commit()?;
                    return Ok(TransitBoardData { data_age_seconds, stale, stops: Vec::new() });
                }

                let mut statement = transaction.prepare(TRANSIT_SQL.as_str())?;
                let mut rows = statement.query(params![
                    lon,
                    lat,
                    KNN_RADIUS_DEGREES,
                    max_items,
                    now_ms - 30_000,
                ])?;
                let stops = assemble_transit_stops(&mut rows, now_ms)?;
                drop(rows);
                drop(statement);
                transaction.commit()?;
                Ok(TransitBoardData { data_age_seconds, stale, stops })
            },
        )
            .await
    }

    pub async fn stops_board(
        &self,
        ids: Vec<String>,
        now: chrono::DateTime<chrono::Utc>,
    ) -> anyhow::Result<TransitBoardData> {
        let now_ms = now.timestamp_millis();
        let ids_json = serde_json::to_string(&ids)?;
        self.read(
            "stops_board",
            move |connection| {
                let transaction = connection.transaction()?;
                let captured_at = waiting_captured_at(&transaction)?;
                let data_age_seconds = captured_at.map(|captured_at| (now_ms - captured_at) / 1000);
                let stale = data_age_seconds.map(|age| age > 90).unwrap_or(true);
                if ids_json == "[]" {
                    transaction.commit()?;
                    return Ok(TransitBoardData { data_age_seconds, stale, stops: Vec::new() });
                }
                let mut statement = transaction.prepare(STOPS_SQL.as_str())?;
                let mut rows = statement.query(params![ids_json, now_ms - 30_000])?;
                let stops = assemble_transit_stops(&mut rows, now_ms)?;
                drop(rows);
                drop(statement);
                transaction.commit()?;
                Ok(TransitBoardData { data_age_seconds, stale, stops })
            },
        )
        .await
    }

    pub async fn search_stops(&self, query: &str) -> anyhow::Result<Vec<SearchHitData>> {
        let needle = fold(query.trim());
        if needle.is_empty() {
            return Ok(Vec::new());
        }
        let normalized_id = crate::bmc::normalize_id(&needle);
        let normalized_upper = format!("{normalized_id}\u{10ffff}");
        let short_id = needle.chars().count() < 3;
        let collapsed = collapse_dupes(&needle);
        let exact_match = fts_phrase("folded_name", &needle);
        let collapsed_match = fts_phrase("collapsed_name", &collapsed);
        self.read(
            "search_stops",
            move |connection| {
                if short_id {
                    let mut statement = connection.prepare(SEARCH_SHORT_ID_SQL)?;
                    let rows = statement.query_map(
                        params![normalized_id, normalized_upper],
                        search_hit_row,
                    )?;
                    return rows.collect::<Result<Vec<_>, _>>().map_err(Into::into);
                }
                let mut statement = connection.prepare(SEARCH_SQL)?;
                let rows = statement.query_map(
                    params![
                        exact_match,
                        needle,
                        normalized_id,
                        normalized_upper,
                        collapsed_match,
                        collapsed,
                    ],
                    search_hit_row,
                )?;
                rows.collect::<Result<Vec<_>, _>>().map_err(Into::into)
            },
        )
            .await
    }

    pub async fn nearby_trains(
        &self,
        lat: f64,
        lon: f64,
        now: chrono::DateTime<chrono::Utc>,
    ) -> anyhow::Result<TrainBoardData> {
        let (min_lon, min_lat, max_lon, max_lat) =
            spatial_bounds(lat, lon, TRAIN_RADIUS_M);
        let now_ms = now.timestamp_millis();
        self.read(
            "nearby_trains",
            move |connection| {
                let mut statement = connection.prepare(TRAIN_SQL)?;
                let mut rows = statement.query(params![
                    lon, lat, min_lon, min_lat, max_lon, max_lat, now_ms,
                ])?;
                let mut stops = Vec::<TrainStopData>::new();
                let mut refresh_station_ids = Vec::new();
                let mut current_station_id = None;
                while let Some(row) = rows.next()? {
                    let station_id = row.get::<_, i64>(0)?;
                    if current_station_id != Some(station_id) {
                        current_station_id = Some(station_id);
                        let external_id = row.get::<_, String>(1)?;
                        let captured_at = row.get::<_, Option<i64>>(4)?;
                        if captured_at
                            .map(|captured_at| now_ms - captured_at >= 30_000)
                            .unwrap_or(true)
                        {
                            refresh_station_ids.push(external_id.clone());
                        }
                        stops.push(TrainStopData {
                            id: external_id,
                            name: row.get(2)?,
                            distance_m: row.get(3)?,
                            departures: Vec::new(),
                        });
                    }
                    let Some(label) = row.get::<_, Option<String>>(5)? else {
                        continue;
                    };
                    let delay_seconds = row.get::<_, Option<i64>>(9)?;
                    stops
                        .last_mut()
                        .context("train departure has no station")?
                        .departures
                        .push(crate::irail::TrainDeparture {
                            label,
                            destination: row.get(6)?,
                            in_seconds: row.get(7)?,
                            delayed: delay_seconds.unwrap_or(0) > 0,
                            platform: row.get(8)?,
                            delay_seconds,
                        });
                }
                Ok(TrainBoardData {
                    stops,
                    refresh_station_ids,
                })
            },
        )
            .await
    }

    pub async fn villo_board(
        &self,
        location: Option<(f64, f64)>,
        now: chrono::DateTime<chrono::Utc>,
    ) -> anyhow::Result<VilloBoardData> {
        let bounds = location.map(|(lat, lon)| {
            let (min_lon, min_lat, max_lon, max_lat) =
                spatial_bounds(lat, lon, VILLO_RADIUS_M);
            (lat, lon, min_lon, min_lat, max_lon, max_lat)
        });
        let now = now.timestamp_millis();
        self.read(
            "villo_board",
            move |connection| {
                let transaction = connection.transaction()?;
                let discovery_ready = transaction.query_row(
                    "SELECT EXISTS(
                         SELECT 1 FROM provider_feed
                         WHERE provider = 'villo'
                           AND feed = 'discovery'
                           AND current_snapshot_id IS NOT NULL
                     )",
                    [],
                    |row| row.get::<_, bool>(0),
                )?;
                let captured_at = transaction
                    .query_row(
                        "SELECT snapshot.captured_at
                         FROM provider_feed AS feed
                         JOIN provider_snapshot AS snapshot
                           ON snapshot.id = feed.current_snapshot_id
                         WHERE feed.provider = 'villo' AND feed.feed = 'status'",
                        [],
                        |row| row.get::<_, i64>(0),
                    )
                    .optional()?;
                let data_age_seconds =
                    captured_at.map(|captured_at| (now - captured_at) / 1000);
                let stops = if let Some((lat, lon, min_lon, min_lat, max_lon, max_lat)) = bounds {
                    let mut statement = transaction.prepare(VILLO_SQL)?;
                    let rows = statement.query_map(
                        params![lon, lat, min_lon, min_lat, max_lon, max_lat],
                        |row| {
                            Ok(VilloStopData {
                                id: row.get(0)?,
                                name: Bilingual {
                                    fr: row.get(1)?,
                                    nl: row.get(2)?,
                                },
                                distance_m: row.get(3)?,
                                bikes: row.get::<_, i64>(4)? as u16,
                                docks: row.get::<_, i64>(5)? as u16,
                                mechanical: row
                                    .get::<_, Option<i64>>(6)?
                                    .map(|value| value as u16),
                                electrical: row
                                    .get::<_, Option<i64>>(7)?
                                    .map(|value| value as u16),
                            })
                        },
                    )?;
                    rows.collect::<Result<Vec<_>, _>>()?
                } else {
                    Vec::new()
                };
                transaction.commit()?;
                Ok(VilloBoardData {
                    discovery_ready,
                    data_age_seconds,
                    stops,
                })
            },
        )
            .await
    }

    pub async fn nearby_villo(&self, lat: f64, lon: f64) -> anyhow::Result<Vec<VilloStopData>> {
        self.villo_board(Some((lat, lon)), chrono::Utc::now())
            .await
            .map(|board| board.stops)
    }

    pub async fn villo_status_url(&self) -> anyhow::Result<Option<String>> {
        self.read(
            "villo_status_url",
            |connection| {
                let payload = connection
                    .query_row(
                        "SELECT discovery.payload_zstd
                         FROM provider_feed AS feed
                         JOIN provider_snapshot AS head ON head.id = feed.current_snapshot_id
                         JOIN provider_snapshot AS discovery
                           ON discovery.batch_id = head.batch_id
                          AND discovery.feed_id = feed.id
                          AND discovery.scope = 'gbfs'
                         WHERE feed.provider = 'villo' AND feed.feed = 'discovery'",
                        [],
                        |row| row.get::<_, Vec<u8>>(0),
                    )
                    .optional()?;
                let Some(payload) = payload else {
                    return Ok(None);
                };
                let body = zstd::stream::decode_all(payload.as_slice())
                    .context("failed to decompress Villo discovery snapshot")?;
                let body =
                    std::str::from_utf8(&body).context("stored Villo discovery encoding")?;
                let (_, status_url) = crate::villo::parse_discovery(body)?;
                Ok(Some(status_url))
            },
        )
            .await
    }


    pub(crate) async fn has_feed_head(
        &self,
        provider: &'static str,
        feed: &'static str,
    ) -> anyhow::Result<bool> {
        self.read("has_feed_head", move |connection| {
            Ok(connection.query_row(
                "SELECT EXISTS(
                     SELECT 1
                     FROM provider_feed AS feed
                     JOIN provider_snapshot AS snapshot
                       ON snapshot.id = feed.current_snapshot_id
                     WHERE feed.provider = ?1 AND feed.feed = ?2
                 )",
                params![provider, feed],
                |row| row.get(0),
            )?)
        })
        .await
    }

    pub(crate) async fn has_feed_snapshot(
        &self,
        provider: &'static str,
        feed: &'static str,
        scope: &'static str,
    ) -> anyhow::Result<bool> {
        self.read("has_feed_snapshot", move |connection| {
            Ok(connection.query_row(
                "SELECT EXISTS(
                     SELECT 1
                     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),
            )?)
        })
        .await
    }

}

fn parse_vehicle_kind(value: &str) -> anyhow::Result<VehicleKind> {
    match value {
        "tram" => Ok(VehicleKind::Tram),
        "metro" => Ok(VehicleKind::Metro),
        "bus" => Ok(VehicleKind::Bus),
        "train" => Ok(VehicleKind::Train),
        "bike" => Ok(VehicleKind::Bike),
        _ => anyhow::bail!("unknown vehicle kind {value}"),
    }
}

fn fts_phrase(column: &str, value: &str) -> String {
    format!("{column}: \"{}\"", value.replace('"', "\"\""))
}

fn search_hit_row(
    row: &deadpool_sqlite::rusqlite::Row<'_>,
) -> deadpool_sqlite::rusqlite::Result<SearchHitData> {
    let direction_fr = row.get::<_, Option<String>>(4)?;
    let direction_nl = row.get::<_, Option<String>>(5)?;
    Ok(SearchHitData {
        id: row.get(0)?,
        name: Bilingual { fr: row.get(1)?, nl: row.get(2)? },
        line_count: row.get::<_, i64>(3)? as usize,
        direction: direction_fr.zip(direction_nl).map(|(fr, nl)| Bilingual { fr, nl }),
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use http_body_util::BodyExt as _;
    use tower::ServiceExt as _;

    fn register_commit_gate(
        connection: &deadpool_sqlite::rusqlite::Connection,
        name: &str,
        started: std::sync::mpsc::SyncSender<()>,
        committed: std::sync::mpsc::Receiver<()>,
    ) {
        let fired = std::sync::atomic::AtomicBool::new(false);
        connection
            .create_scalar_function(
                name,
                1,
                deadpool_sqlite::rusqlite::functions::FunctionFlags::SQLITE_UTF8,
                move |context| {
                    let value = context.get::<i64>(0)?;
                    if !fired.swap(true, std::sync::atomic::Ordering::SeqCst) {
                        started.send(()).unwrap();
                        committed.recv().unwrap();
                    }
                    Ok(value)
                },
            )
            .unwrap();
    }

    #[tokio::test]
    async fn transit_plan_uses_knn2_without_scanning_place() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("stibble.sqlite");
        let extension = std::env::var("SPATIALITE_EXTENSION").unwrap();
        let store = Store::open(path, extension).await.unwrap();
        crate::dev_mock::seed(&store).await.unwrap();
        let connection = store.pool.get().await.unwrap();
        let plan = connection
            .interact(|connection| {
                let mut statement = connection
                    .prepare(&format!("EXPLAIN QUERY PLAN {}", TRANSIT_SQL.as_str()))
                    .unwrap();
                statement
                    .query_map(
                        params![4.3661, 50.8535, KNN_RADIUS_DEGREES, 6, 0],
                        |row| row.get::<_, String>(3),
                    )
                    .unwrap()
                    .collect::<Result<Vec<_>, _>>()
                    .unwrap()
                    .join("\n")
            })
            .await
            .unwrap();
        assert!(plan.to_ascii_lowercase().contains("knn2"), "{plan}");
        assert!(!plan.contains("SCAN place"), "{plan}");
    }

    #[tokio::test]
    async fn search_plans_use_indexes_for_names_and_short_ids() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("stibble.sqlite");
        let extension = std::env::var("SPATIALITE_EXTENSION").unwrap();
        let store = Store::open(path, extension).await.unwrap();
        let connection = store.pool.get().await.unwrap();
        let (normal, short) = connection
            .interact(|connection| {
                let mut normal_statement = connection
                    .prepare(&format!("EXPLAIN QUERY PLAN {SEARCH_SQL}"))
                    .unwrap();
                let normal = normal_statement
                    .query_map(
                        params![
                            "folded_name: \"HANKAR\"",
                            "HANKAR",
                            "HANKAR",
                            "HANKAR\u{10ffff}",
                            "collapsed_name: \"HANKAR\"",
                            "HANKAR",
                        ],
                        |row| row.get::<_, String>(3),
                    )
                    .unwrap()
                    .collect::<Result<Vec<_>, _>>()
                    .unwrap()
                    .join("\n");
                let mut short_statement = connection
                    .prepare(&format!("EXPLAIN QUERY PLAN {SEARCH_SHORT_ID_SQL}"))
                    .unwrap();
                let short = short_statement
                    .query_map(params!["64", "64\u{10ffff}"], |row| row.get::<_, String>(3))
                    .unwrap()
                    .collect::<Result<Vec<_>, _>>()
                    .unwrap()
                    .join("\n");
                (normal, short)
            })
            .await
            .unwrap();
        assert!(normal.contains("VIRTUAL TABLE INDEX 0:M"), "{normal}");
        assert!(!normal.lines().any(is_full_search_scan), "{normal}");
        assert!(!short.contains("place_search"), "{short}");
        assert!(short.contains("place_normalized_lookup"), "{short}");
        assert!(!short.lines().any(is_full_search_scan), "{short}");
    }

    #[tokio::test]
    async fn villo_plan_uses_virtual_spatial_index() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("stibble.sqlite");
        let extension = std::env::var("SPATIALITE_EXTENSION").unwrap();
        let store = Store::open(path, extension).await.unwrap();
        crate::dev_mock::seed(&store).await.unwrap();
        let connection = store.pool.get().await.unwrap();
        let plan = connection
            .interact(|connection| {
                let mut statement = connection
                    .prepare(&format!("EXPLAIN QUERY PLAN {VILLO_SQL}"))
                    .unwrap();
                statement
                    .query_map(
                        params![4.3676, 50.8467, 4.35, 50.84, 4.38, 50.86],
                        |row| row.get::<_, String>(3),
                    )
                    .unwrap()
                    .collect::<Result<Vec<_>, _>>()
                    .unwrap()
                    .join("\n")
            })
            .await
            .unwrap();
        assert!(plan.contains("VirtualSpatialIndex"), "{plan}");
        assert!(!plan.contains("SCAN place"), "{plan}");
    }

    #[tokio::test]
    async fn train_plan_uses_virtual_spatial_index() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("stibble.sqlite");
        let extension = std::env::var("SPATIALITE_EXTENSION").unwrap();
        let store = Store::open(path, extension).await.unwrap();
        crate::dev_mock::seed(&store).await.unwrap();
        let connection = store.pool.get().await.unwrap();
        let plan = connection
            .interact(|connection| {
                let mut statement = connection
                    .prepare(&format!("EXPLAIN QUERY PLAN {TRAIN_SQL}"))
                    .unwrap();
                statement
                    .query_map(
                        params![4.3661, 50.8535, 4.35, 50.84, 4.38, 50.86, 0],
                        |row| row.get::<_, String>(3),
                    )
                    .unwrap()
                    .collect::<Result<Vec<_>, _>>()
                    .unwrap()
                    .join("\n")
            })
            .await
            .unwrap();
        assert!(plan.contains("VirtualSpatialIndex"), "{plan}");
        assert!(!plan.contains("SCAN place"), "{plan}");
        let brussels = spatial_bounds(50.8535, 4.3661, TRAIN_RADIUS_M);
        assert!(brussels.2 - brussels.0 < 0.1, "{brussels:?}");
        assert!(brussels.3 - brussels.1 < 0.1, "{brussels:?}");
        let polar = spatial_bounds(89.99, 0.0, TRAIN_RADIUS_M);
        assert_eq!((polar.0, polar.2), (-180.0, 180.0));
    }

    #[tokio::test]
    async fn transit_head_and_rows_share_one_deferred_snapshot() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("stibble.sqlite");
        let extension = std::env::var("SPATIALITE_EXTENSION").unwrap();
        let store = Store::open(&path, extension).await.unwrap();
        crate::dev_mock::seed(&store).await.unwrap();
        let now = chrono::Utc::now();
        let old_at = now - chrono::Duration::seconds(120);
        let new_at = now;
        let (started_tx, started_rx) = std::sync::mpsc::sync_channel(0);
        let (committed_tx, committed_rx) = std::sync::mpsc::sync_channel(0);
        let connection = store.pool.get().await.unwrap();
        connection
            .interact(move |connection| {
                connection
                    .execute(
                        "UPDATE provider_snapshot
                         SET captured_at = ?1
                         WHERE id = (
                             SELECT current_snapshot_id FROM provider_feed
                             WHERE provider = 'bmc' AND feed = 'waiting'
                         )",
                        [old_at.timestamp_millis()],
                    )
                    .unwrap();
                connection
                    .execute(
                        "UPDATE current_passage
                         SET destination_fr = 'BEFORE',
                             destination_nl = 'BEFORE',
                             expected_at = ?1",
                        [(now + chrono::Duration::minutes(2)).timestamp_millis()],
                    )
                    .unwrap();
                register_commit_gate(
                    connection,
                    "sync_transit_head",
                    started_tx,
                    committed_rx,
                );
                connection
                    .execute_batch(
                        "ALTER TABLE provider_snapshot RENAME TO provider_snapshot_data;
                         CREATE VIEW provider_snapshot AS
                         SELECT
                             id,
                             batch_id,
                             feed_id,
                             scope,
                             sync_transit_head(captured_at) AS captured_at,
                             source_at,
                             content_type,
                             payload_sha256,
                             payload_zstd
                         FROM provider_snapshot_data;",
                    )
                    .unwrap();
            })
            .await
            .unwrap();
        drop(connection);
        let writer_path = path.clone();
        let writer = std::thread::spawn(move || {
            let mut connection = deadpool_sqlite::rusqlite::Connection::open(writer_path).unwrap();
            started_rx.recv().unwrap();
            let transaction = connection.transaction().unwrap();
            transaction
                .execute(
                    "UPDATE provider_snapshot_data
                     SET captured_at = ?1
                     WHERE id = (
                         SELECT current_snapshot_id FROM provider_feed
                         WHERE provider = 'bmc' AND feed = 'waiting'
                     )",
                    [new_at.timestamp_millis()],
                )
                .unwrap();
            transaction
                .execute(
                    "UPDATE current_passage
                     SET destination_fr = 'AFTER', destination_nl = 'AFTER'",
                    [],
                )
                .unwrap();
            transaction.commit().unwrap();
            committed_tx.send(()).unwrap();
        });

        let board = store
            .transit_board(50.80, 4.30, 1, now)
            .await
            .unwrap();
        writer.join().unwrap();

        assert_eq!(board.data_age_seconds, Some(120));
        assert!(board.stops.iter().any(|stop| !stop.departures.is_empty()));
        assert!(board
            .stops
            .iter()
            .flat_map(|stop| &stop.departures)
            .all(|departure| match departure {
                Departure::Transit { destination, .. } => destination.fr == "BEFORE",
                Departure::EndOfService { .. } => true,
                _ => false,
            }));
    }

    #[tokio::test]
    async fn villo_endpoint_reads_status_age_and_rows_from_one_snapshot() {
        let directory = tempfile::tempdir().unwrap();
        let path = directory.path().join("stibble.sqlite");
        let extension = std::env::var("SPATIALITE_EXTENSION").unwrap();
        let store = Store::open(&path, extension).await.unwrap();
        crate::dev_mock::seed(&store).await.unwrap();
        let now = chrono::Utc::now();
        let old_at = now - chrono::Duration::seconds(120);
        let new_at = now;
        let (started_tx, started_rx) = std::sync::mpsc::sync_channel(0);
        let (committed_tx, committed_rx) = std::sync::mpsc::sync_channel(0);
        let connection = store.pool.get().await.unwrap();
        connection
            .interact(move |connection| {
                connection
                    .execute(
                        "UPDATE provider_snapshot
                         SET captured_at = ?1
                         WHERE id = (
                             SELECT current_snapshot_id FROM provider_feed
                             WHERE provider = 'villo' AND feed = 'status'
                         )",
                        [old_at.timestamp_millis()],
                    )
                    .unwrap();
                connection
                    .execute("UPDATE current_villo_status SET bikes = 1", [])
                    .unwrap();
                register_commit_gate(
                    connection,
                    "sync_villo_status",
                    started_tx,
                    committed_rx,
                );
                connection
                    .execute_batch(
                        "ALTER TABLE current_villo_status RENAME TO current_villo_status_data;
                         CREATE VIEW current_villo_status AS
                         SELECT
                             snapshot_id,
                             stand_id,
                             sync_villo_status(bikes) AS bikes,
                             docks,
                             mechanical,
                             electrical
                         FROM current_villo_status_data;",
                    )
                    .unwrap();
            })
            .await
            .unwrap();
        drop(connection);
        let writer_path = path.clone();
        let writer = std::thread::spawn(move || {
            let mut connection = deadpool_sqlite::rusqlite::Connection::open(writer_path).unwrap();
            started_rx.recv().unwrap();
            let transaction = connection.transaction().unwrap();
            transaction
                .execute(
                    "UPDATE provider_snapshot
                     SET captured_at = ?1
                     WHERE id = (
                         SELECT current_snapshot_id FROM provider_feed
                         WHERE provider = 'villo' AND feed = 'status'
                     )",
                    [new_at.timestamp_millis()],
                )
                .unwrap();
            transaction
                .execute("UPDATE current_villo_status_data SET bikes = 99", [])
                .unwrap();
            transaction.commit().unwrap();
            committed_tx.send(()).unwrap();
        });
        let (refresh_tx, _refresh_rx) = tokio::sync::mpsc::channel(1);
        let app = crate::api::router(crate::state::AppState::new(store, refresh_tx));
        let response = app
            .oneshot(
                axum::http::Request::builder()
                    .uri("/v1/villo?lat=50.8317&lon=4.3856")
                    .body(axum::body::Body::empty())
                    .unwrap(),
            )
            .await
            .unwrap();
        writer.join().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["stale"], true);
        assert!(value["data_age_seconds"].as_i64().unwrap() >= 119);
        let station = value["stops"]
            .as_array()
            .unwrap()
            .iter()
            .find(|station| station["id"] == "VILLO:1")
            .unwrap();
        assert_eq!(
            station["rows"][0]["destination"]["fr"],
            serde_json::json!("1 velos \u{b7} 2 places")
        );
    }

    fn is_full_search_scan(detail: &str) -> bool {
        detail == "SCAN place" || detail == "SCAN place_search VIRTUAL TABLE INDEX 0:"
    }
}