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 = std::sync::LazyLock::new(|| { format!("{}{}", TRANSIT_SELECT_SQL, BOARD_DETAILS_SQL.replace("$CUTOFF", "?5")) }); static STOPS_SQL: std::sync::LazyLock = 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, } struct LineStopRow { ordinal: i64, id: String, name: Bilingual, lat: Option, lon: Option, } struct LinePassageRow { destination_fr: String, destination_nl: String, expected_at: i64, } struct LineDisruptionRow { priority: i64, en: Option, fr: Option, nl: Option, target: bool, } #[derive(serde::Deserialize)] struct LineGeometry { coordinates: Vec<(f64, f64)>, } struct TimedDeparture { badge: Badge, destination: Bilingual, in_seconds: Vec, disrupted: bool, } struct NoticeDeparture { line_id: i64, badge: Badge, message: Bilingual, disrupted: bool, } struct StopBuilder { stop: TransitStopData, timed: BTreeMap, notices: Vec, } fn waiting_captured_at( transaction: &deadpool_sqlite::rusqlite::Transaction<'_>, ) -> anyhow::Result> { 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> { let mut stops = Vec::::new(); let mut stop_indexes = HashMap::::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>(17)?.unwrap_or_default(), nl: row.get::<_, Option>(18)?.unwrap_or_default(), }, disrupted, }); } } Ok(stops .into_iter() .map(|mut builder| { let timed_lines: HashSet = 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::>(); 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(¬ice.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, ) -> anyhow::Result> { 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::, _>>()? }; 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::, _>>()? }; 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::, _>>()? }; 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::>(); 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::>(); 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(§ion, 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, ) -> anyhow::Result { 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, now: chrono::DateTime, ) -> anyhow::Result { 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> { 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::, _>>().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::, _>>().map_err(Into::into) }, ) .await } pub async fn nearby_trains( &self, lat: f64, lon: f64, now: chrono::DateTime, ) -> anyhow::Result { 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::::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>(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>(5)? else { continue; }; let delay_seconds = row.get::<_, Option>(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, ) -> anyhow::Result { 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>(6)? .map(|value| value as u16), electrical: row .get::<_, Option>(7)? .map(|value| value as u16), }) }, )?; rows.collect::, _>>()? } 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> { self.villo_board(Some((lat, lon)), chrono::Utc::now()) .await .map(|board| board.stops) } pub async fn villo_status_url(&self) -> anyhow::Result> { 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>(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 { 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 { 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 { 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 { let direction_fr = row.get::<_, Option>(4)?; let direction_nl = row.get::<_, Option>(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::(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::, _>>() .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::, _>>() .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::, _>>() .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::, _>>() .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::, _>>() .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:" } }