use super::{ DisruptionsBatch, PositionsBatch, StaticBatch, Store, TrainDeparturesBatch, TrainStationsBatch, VilloDiscoveryBatch, VilloStatusBatch, WaitingBatch, }; use crate::{ api::{collapse_dupes, fold}, model::{Bilingual, Disruption, LineSequence, RouteInfo, StopInfo, VehicleKind}, provider::{FeedStage, RawFeed}, shapes::Polyline, }; use anyhow::{ensure, Context}; use deadpool_sqlite::rusqlite::{ params, Connection, OptionalExtension, Transaction, TransactionBehavior, }; use sha2::{Digest, Sha256}; use std::{ collections::{HashMap, HashSet}, fmt::Write as _, }; struct PreparedRaw { provider: &'static str, feed: &'static str, scope: Box, captured_at: i64, source_at: Option, content_type: Box, payload_sha256: [u8; 32], payload_zstd: Vec, } struct PreparedSequenceStop { external_id: String, place_index: Option, } struct PreparedSequence { line: String, destination: Bilingual, source_order: i64, stops: Vec, } struct PreparedStatic { batch_captured_at: i64, raw: Vec, stops: Vec, aliases: Vec<(Box, usize, i64)>, routes: Vec, sequences: Vec, shapes: Vec<((String, String), Polyline)>, } struct PreparedWaiting { raw: PreparedRaw, passages: HashMap, Vec>, end_notices: HashMap, Vec>, } struct PreparedPositions { raw: PreparedRaw, positions: HashMap, Vec>, } struct PreparedDisruption { disruption: Disruption, payload_sha256: [u8; 32], payload_zstd: Vec, } struct PreparedDisruptions { raw: PreparedRaw, disruptions: Vec, } struct PreparedVilloDiscovery { batch_captured_at: i64, raw: Vec, stations: Vec<(Box, crate::villo::VilloStationMeta)>, } struct PreparedVilloStatus { raw: PreparedRaw, status: Vec<(Box, crate::villo::VilloStationCounts)>, } struct PreparedTrainStations { raw: PreparedRaw, stations: Vec, } struct PreparedTrainDeparture { label: String, destination: String, expected_at: i64, platform: Option, delay_seconds: Option, } struct PreparedTrainDepartures { raw: PreparedRaw, station_id: Box, departures: Vec, } impl PreparedStatic { fn new(batch: StaticBatch) -> anyhow::Result { batch.validate()?; let targets = batch.alias_targets()?; let mut aliases = Vec::with_capacity(batch.stops.len() * 2); for (index, stop) in batch.stops.iter().enumerate() { aliases.push((stop.id.clone().into_boxed_str(), index, 0)); } for (index, stop) in batch.stops.iter().enumerate() { if let Some(alias) = crate::bmc::strip_platform_suffix(&stop.id) { aliases.push((alias.into(), index, 1)); } } let mut sequence_groups: Vec<(&Box, &Vec)> = batch.sequences.iter().collect(); sequence_groups.sort_by(|(left, _), (right, _)| left.cmp(right)); let mut sequences = Vec::new(); for (line, group) in sequence_groups { for (source_order, sequence) in group.iter().enumerate() { let stops = sequence .stops .iter() .map(|stop| PreparedSequenceStop { external_id: stop.clone(), place_index: targets.get(stop.as_str()).copied(), }) .collect(); sequences.push(PreparedSequence { line: line.to_string(), destination: sequence.destination.clone(), source_order: source_order as i64, stops, }); } } drop(targets); let StaticBatch { raw, stops, routes, sequences: _, shapes, } = batch; let mut raw = raw .into_iter() .map(prepare_raw) .collect::>>()?; raw.sort_by_key(|feed| scope_rank(&feed.scope)); let batch_captured_at = raw .iter() .map(|feed| feed.captured_at) .max() .context("static raw feeds are empty")?; let mut routes: Vec = routes.into_values().collect(); routes.sort_by(|left, right| left.line.cmp(&right.line)); let mut shapes: Vec<((String, String), Polyline)> = shapes.into_iter().collect(); shapes.sort_by(|(left, _), (right, _)| left.cmp(right)); Ok(Self { batch_captured_at, raw, stops, aliases, routes, sequences, shapes, }) } } impl PreparedWaiting { fn new(batch: WaitingBatch) -> anyhow::Result { Ok(Self { raw: prepare_realtime_raw(batch.raw, "waiting")?, passages: batch.passages, end_notices: batch.end_notices, }) } } impl PreparedPositions { fn new(batch: PositionsBatch) -> anyhow::Result { Ok(Self { raw: prepare_realtime_raw(batch.raw, "positions")?, positions: batch.positions, }) } } impl PreparedDisruptions { fn new(batch: DisruptionsBatch) -> anyhow::Result { let raw = prepare_realtime_raw(batch.raw, "disruptions")?; let disruptions = batch .disruptions .into_iter() .map(|disruption| { let payload = serde_json::to_vec(&( "bmc", disruption.en.as_deref(), disruption.fr.as_deref(), disruption.nl.as_deref(), disruption.priority, &disruption.lines, &disruption.points, )) .context("failed to encode disruption fact")?; let payload_sha256 = Sha256::digest(&payload).into(); let payload_zstd = zstd::stream::encode_all(payload.as_slice(), 3) .context("failed to compress disruption fact")?; Ok(PreparedDisruption { disruption, payload_sha256, payload_zstd, }) }) .collect::>>()?; Ok(Self { raw, disruptions }) } } impl PreparedVilloDiscovery { fn new(batch: VilloDiscoveryBatch) -> anyhow::Result { ensure!( batch.raw.len() == 2 && batch .raw .iter() .all(|raw| raw.provider == "villo" && raw.feed == "discovery") && batch.raw.iter().any(|raw| raw.scope.as_ref() == "gbfs") && batch .raw .iter() .any(|raw| raw.scope.as_ref() == "station_information"), "villo discovery must contain exactly gbfs and station_information raw feeds" ); ensure!( !batch.status_url.trim().is_empty(), "villo status URL is empty" ); let discovery = batch .raw .iter() .find(|raw| raw.scope.as_ref() == "gbfs") .context("villo gbfs raw feed is missing")?; let body = std::str::from_utf8(&discovery.body).context("villo discovery encoding")?; let (_, status_url) = crate::villo::parse_discovery(body)?; ensure!( status_url == batch.status_url, "villo status URL does not match discovery raw feed" ); ensure!(!batch.stations.is_empty(), "villo stations are empty"); for (id, station) in &batch.stations { ensure!(!id.trim().is_empty(), "villo station ID is empty"); ensure!( station.lat.is_finite() && (-90.0..=90.0).contains(&station.lat) && station.lon.is_finite() && (-180.0..=180.0).contains(&station.lon), "villo station {id} has invalid coordinates" ); } let mut raw = batch .raw .into_iter() .map(prepare_raw) .collect::>>()?; raw.sort_by(|left, right| left.scope.cmp(&right.scope)); let batch_captured_at = raw .iter() .map(|raw| raw.captured_at) .max() .context("villo discovery raw feeds are empty")?; let mut stations: Vec<_> = batch.stations.into_iter().collect(); stations.sort_by(|(left, _), (right, _)| left.cmp(right)); Ok(Self { batch_captured_at, raw, stations, }) } } impl PreparedVilloStatus { fn new(batch: VilloStatusBatch) -> anyhow::Result { ensure!( batch.raw.provider == "villo" && batch.raw.feed == "status" && batch.raw.scope.is_empty(), "villo status raw feed must belong to villo/status with an empty scope" ); let raw = prepare_raw(batch.raw)?; let mut status: Vec<_> = batch.status.into_iter().collect(); status.sort_by(|(left, _), (right, _)| left.cmp(right)); Ok(Self { raw, status }) } } impl PreparedTrainStations { fn new(batch: TrainStationsBatch) -> anyhow::Result { ensure!( batch.raw.provider == "irail" && batch.raw.feed == "stations" && batch.raw.scope.is_empty(), "train stations raw feed must belong to irail/stations with an empty scope" ); ensure!(!batch.stations.is_empty(), "train stations are empty"); let mut ids = HashSet::with_capacity(batch.stations.len()); for station in &batch.stations { ensure!(!station.id.trim().is_empty(), "train station ID is empty"); ensure!( ids.insert(station.id.as_str()), "duplicate train station ID {}", station.id ); ensure!( station.lat.is_finite() && (-90.0..=90.0).contains(&station.lat) && station.lon.is_finite() && (-180.0..=180.0).contains(&station.lon), "train station {} has invalid coordinates", station.id ); } Ok(Self { raw: prepare_raw(batch.raw)?, stations: batch.stations, }) } } impl PreparedTrainDepartures { fn new(batch: TrainDeparturesBatch) -> anyhow::Result { ensure!(!batch.station_id.trim().is_empty(), "train station ID is empty"); ensure!( batch.raw.provider == "irail" && batch.raw.feed == "liveboard" && batch.raw.scope.as_ref() == batch.station_id.as_str(), "train liveboard raw feed scope must match its station ID" ); let raw = prepare_raw(batch.raw)?; let captured_second_ms = raw.captured_at.div_euclid(1000) * 1000; let mut departures = Vec::with_capacity(batch.departures.len()); for departure in batch.departures { ensure!( departure.delayed == (departure.delay_seconds.unwrap_or(0) > 0), "train departure delay fields disagree" ); let expected_at = departure .in_seconds .checked_mul(1000) .and_then(|offset| captured_second_ms.checked_add(offset)) .context("train departure expected time overflow")?; departures.push(PreparedTrainDeparture { label: departure.label, destination: departure.destination, expected_at, platform: departure.platform, delay_seconds: departure.delay_seconds, }); } departures.sort_by(|left, right| { ( left.expected_at, left.label.as_str(), left.destination.as_str(), ) .cmp(&( right.expected_at, right.label.as_str(), right.destination.as_str(), )) }); ensure!( departures.windows(2).all(|pair| { ( pair[0].expected_at, pair[0].label.as_str(), pair[0].destination.as_str(), ) != ( pair[1].expected_at, pair[1].label.as_str(), pair[1].destination.as_str(), ) }), "duplicate train departure" ); Ok(Self { raw, station_id: batch.station_id.into_boxed_str(), departures, }) } } fn prepare_realtime_raw(raw: RawFeed, feed: &'static str) -> anyhow::Result { ensure!( raw.provider == "bmc" && raw.feed == feed && raw.scope.is_empty(), "realtime raw feed must belong to bmc/{feed} with an empty scope" ); prepare_raw(raw) } fn prepare_raw(raw: RawFeed) -> anyhow::Result { let payload_sha256 = Sha256::digest(&raw.body).into(); let payload_zstd = zstd::stream::encode_all(raw.body.as_slice(), 3) .with_context(|| format!("failed to compress {} scope {}", raw.feed, raw.scope))?; Ok(PreparedRaw { provider: raw.provider, feed: raw.feed, scope: raw.scope, captured_at: raw.captured_at.timestamp_millis(), source_at: raw.source_at.map(|value| value.timestamp_millis()), content_type: raw.content_type, payload_sha256, payload_zstd, }) } fn scope_rank(scope: &str) -> u8 { match scope { "stop-details" => 0, "gtfs" => 1, "stops-by-line" => 2, _ => 3, } } impl Store { pub async fn apply_static(&self, batch: StaticBatch) -> anyhow::Result<()> { let batch = tokio::task::spawn_blocking(move || PreparedStatic::new(batch)) .await .context("static batch preparation task failed")??; let connection = self.pool.get().await.with_context(|| { format!("failed to get connection for {}", self.path.display()) })?; connection .interact(move |connection| apply_static_transaction(connection, batch)) .await .map_err(|error| anyhow::anyhow!("SQLite interaction failed: {error}"))? .with_context(|| format!("failed to apply static data to {}", self.path.display())) } pub async fn apply_waiting(&self, batch: WaitingBatch) -> anyhow::Result<()> { let batch = tokio::task::spawn_blocking(move || PreparedWaiting::new(batch)) .await .context("waiting batch preparation task failed")??; let connection = self.pool.get().await.with_context(|| { format!("failed to get connection for {}", self.path.display()) })?; connection .interact(move |connection| apply_waiting_transaction(connection, batch)) .await .map_err(|error| anyhow::anyhow!("SQLite interaction failed: {error}"))? .with_context(|| format!("failed to apply waiting data to {}", self.path.display())) } pub async fn apply_positions(&self, batch: PositionsBatch) -> anyhow::Result<()> { let batch = tokio::task::spawn_blocking(move || PreparedPositions::new(batch)) .await .context("positions batch preparation task failed")??; let connection = self.pool.get().await.with_context(|| { format!("failed to get connection for {}", self.path.display()) })?; connection .interact(move |connection| apply_positions_transaction(connection, batch)) .await .map_err(|error| anyhow::anyhow!("SQLite interaction failed: {error}"))? .with_context(|| format!("failed to apply positions data to {}", self.path.display())) } pub async fn apply_disruptions(&self, batch: DisruptionsBatch) -> anyhow::Result<()> { let batch = tokio::task::spawn_blocking(move || PreparedDisruptions::new(batch)) .await .context("disruptions batch preparation task failed")??; let connection = self.pool.get().await.with_context(|| { format!("failed to get connection for {}", self.path.display()) })?; connection .interact(move |connection| apply_disruptions_transaction(connection, batch)) .await .map_err(|error| anyhow::anyhow!("SQLite interaction failed: {error}"))? .with_context(|| format!("failed to apply disruptions data to {}", self.path.display())) } pub async fn apply_villo_discovery( &self, batch: VilloDiscoveryBatch, ) -> anyhow::Result<()> { let batch = tokio::task::spawn_blocking(move || PreparedVilloDiscovery::new(batch)) .await .context("villo discovery batch preparation task failed")??; let connection = self.pool.get().await.with_context(|| { format!("failed to get connection for {}", self.path.display()) })?; connection .interact(move |connection| apply_villo_discovery_transaction(connection, batch)) .await .map_err(|error| anyhow::anyhow!("SQLite interaction failed: {error}"))? .with_context(|| { format!( "failed to apply Villo discovery data to {}", self.path.display() ) }) } pub async fn apply_villo_status(&self, batch: VilloStatusBatch) -> anyhow::Result<()> { let batch = tokio::task::spawn_blocking(move || PreparedVilloStatus::new(batch)) .await .context("villo status batch preparation task failed")??; let connection = self.pool.get().await.with_context(|| { format!("failed to get connection for {}", self.path.display()) })?; connection .interact(move |connection| apply_villo_status_transaction(connection, batch)) .await .map_err(|error| anyhow::anyhow!("SQLite interaction failed: {error}"))? .with_context(|| { format!( "failed to apply Villo status data to {}", self.path.display() ) }) } pub async fn apply_train_stations( &self, batch: TrainStationsBatch, ) -> anyhow::Result<()> { let batch = tokio::task::spawn_blocking(move || PreparedTrainStations::new(batch)) .await .context("train stations batch preparation task failed")??; let connection = self.pool.get().await.with_context(|| { format!("failed to get connection for {}", self.path.display()) })?; connection .interact(move |connection| apply_train_stations_transaction(connection, batch)) .await .map_err(|error| anyhow::anyhow!("SQLite interaction failed: {error}"))? .with_context(|| { format!( "failed to apply train stations data to {}", self.path.display() ) }) } pub async fn apply_train_departures( &self, batch: TrainDeparturesBatch, ) -> anyhow::Result<()> { let batch = tokio::task::spawn_blocking(move || PreparedTrainDepartures::new(batch)) .await .context("train departures batch preparation task failed")??; let connection = self.pool.get().await.with_context(|| { format!("failed to get connection for {}", self.path.display()) })?; connection .interact(move |connection| apply_train_departures_transaction(connection, batch)) .await .map_err(|error| anyhow::anyhow!("SQLite interaction failed: {error}"))? .with_context(|| { format!( "failed to apply train departures data to {}", self.path.display() ) }) } pub(crate) async fn restore_feed_head_from_latest_snapshot( &self, provider: &'static str, feed: &'static str, ) -> anyhow::Result> { let connection = self.pool.get().await.with_context(|| { format!("failed to get connection for {}", self.path.display()) })?; connection .interact(move |connection| -> anyhow::Result> { let transaction = connection.transaction()?; let latest = transaction .query_row( "SELECT feed.id, snapshot.id FROM provider_feed AS feed JOIN provider_snapshot AS snapshot ON snapshot.feed_id = feed.id WHERE feed.provider = ?1 AND feed.feed = ?2 ORDER BY snapshot.captured_at DESC, snapshot.id DESC LIMIT 1", params![provider, feed], |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)), ) .optional()?; let restored = if let Some((feed_id, snapshot_id)) = latest { let updated = transaction.execute( "UPDATE provider_feed SET current_snapshot_id = ?2 WHERE id = ?1 AND current_snapshot_id IS NULL", params![feed_id, snapshot_id], )?; anyhow::ensure!(updated == 1, "{provider}/{feed} head changed"); Some(snapshot_id) } else { None }; transaction.commit()?; Ok(restored) }) .await .map_err(|error| anyhow::anyhow!("SQLite interaction failed: {error}"))? .with_context(|| { format!( "failed to restore {provider}/{feed} head in {}", self.path.display() ) }) } pub async fn record_feed_failure( &self, provider: &str, feed: &str, stage: FeedStage, message: &str, failed_at: chrono::DateTime, ) -> anyhow::Result<()> { let provider = provider.to_owned(); let feed = feed.to_owned(); let stage = stage.as_str(); let message = message.to_owned(); let failed_at = failed_at.timestamp_millis(); let connection = self.pool.get().await.with_context(|| { format!("failed to get connection for {}", self.path.display()) })?; connection .interact(move |connection| -> anyhow::Result<()> { let feed_id = connection.query_row( "SELECT id FROM provider_feed WHERE provider = ?1 AND feed = ?2", params![provider, feed], |row| row.get::<_, i64>(0), )?; connection.execute( "INSERT INTO feed_failure(feed_id, stage, message, failed_at) VALUES (?1, ?2, ?3, ?4) ON CONFLICT(feed_id) DO UPDATE SET stage = excluded.stage, message = excluded.message, failed_at = excluded.failed_at", params![feed_id, stage, message, failed_at], )?; Ok(()) }) .await .map_err(|error| anyhow::anyhow!("SQLite interaction failed: {error}"))? .with_context(|| format!("failed to record feed failure in {}", self.path.display())) } } fn apply_static_transaction( connection: &mut Connection, batch: PreparedStatic, ) -> anyhow::Result<()> { let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; let gtfs_snapshot_id = insert_static_snapshots(&transaction, &batch)?; replace_static_projection(&transaction, &batch)?; rebuild_transit_ranks(&transaction)?; let updated = transaction.execute( "UPDATE provider_feed SET current_snapshot_id = ?1 WHERE provider = 'bmc' AND feed = 'static'", [gtfs_snapshot_id], )?; ensure!(updated == 1, "bmc/static provider feed is missing"); transaction.execute( "DELETE FROM feed_failure WHERE feed_id = (SELECT id FROM provider_feed WHERE provider = 'bmc' AND feed = 'static')", [], )?; transaction.commit()?; Ok(()) } fn insert_static_snapshots( transaction: &Transaction<'_>, batch: &PreparedStatic, ) -> anyhow::Result { transaction.execute( "INSERT INTO provider_batch(provider, captured_at) VALUES ('bmc', ?1)", [batch.batch_captured_at], )?; let batch_id = transaction.last_insert_rowid(); let feed_id = transaction.query_row( "SELECT id FROM provider_feed WHERE provider = 'bmc' AND feed = 'static'", [], |row| row.get::<_, i64>(0), )?; let mut gtfs_snapshot_id = None; let mut statement = transaction.prepare( "INSERT INTO provider_snapshot(batch_id, feed_id, scope, captured_at, source_at, content_type, payload_sha256, payload_zstd) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", )?; for raw in &batch.raw { ensure!(raw.provider == "bmc" && raw.feed == "static", "invalid static raw feed"); statement.execute(params![ batch_id, feed_id, raw.scope, raw.captured_at, raw.source_at, raw.content_type, raw.payload_sha256.as_slice(), raw.payload_zstd, ])?; if raw.scope.as_ref() == "gtfs" { gtfs_snapshot_id = Some(transaction.last_insert_rowid()); } } gtfs_snapshot_id.context("gtfs snapshot is missing") } fn replace_static_projection( transaction: &Transaction<'_>, batch: &PreparedStatic, ) -> anyhow::Result<()> { transaction.execute_batch( "DELETE FROM transit_with_passages WHERE place_id IN (SELECT id FROM place WHERE provider = 'bmc' AND kind = 'transit'); DELETE FROM transit_with_notices WHERE place_id IN (SELECT id FROM place WHERE provider = 'bmc' AND kind = 'transit'); DELETE FROM transit_silent WHERE place_id IN (SELECT id FROM place WHERE provider = 'bmc' AND kind = 'transit'); DELETE FROM route_variant_stop WHERE variant_id IN ( SELECT route_variant.id FROM route_variant JOIN line ON line.id = route_variant.line_id WHERE line.provider = 'bmc' ); DELETE FROM route_variant WHERE line_id IN (SELECT id FROM line WHERE provider = 'bmc'); DELETE FROM route_shape WHERE line_id IN (SELECT id FROM line WHERE provider = 'bmc'); DELETE FROM place_alias WHERE provider = 'bmc'; DELETE FROM place_search WHERE place_id IN (SELECT id FROM place WHERE provider = 'bmc' AND kind = 'transit'); UPDATE line SET active = 0 WHERE provider = 'bmc'; UPDATE place SET active = 0 WHERE provider = 'bmc' AND kind = 'transit';", )?; let mut place_ids = Vec::with_capacity(batch.stops.len()); { let mut place_statement = transaction.prepare( "INSERT INTO place(provider, kind, external_id, name_fr, name_nl, folded_name, collapsed_name, normalized_id, source_order, active, geom) VALUES ('bmc', 'transit', ?1, ?2, ?3, ?4, ?5, ?6, ?7, 1, MakePoint(?8, ?9, 4326)) ON CONFLICT(provider, kind, external_id) DO UPDATE SET name_fr = excluded.name_fr, name_nl = excluded.name_nl, folded_name = excluded.folded_name, collapsed_name = excluded.collapsed_name, normalized_id = excluded.normalized_id, source_order = excluded.source_order, active = 1, geom = excluded.geom RETURNING id", )?; let mut search_statement = transaction.prepare( "INSERT INTO place_search(place_id, folded_name, collapsed_name, normalized_id) VALUES (?1, ?2, ?3, ?4)", )?; for (source_order, stop) in batch.stops.iter().enumerate() { let folded_name = format!("{} {}", fold(&stop.name.fr), fold(&stop.name.nl)); let collapsed_name = collapse_dupes(&folded_name); let normalized_id = crate::bmc::normalize_id(&stop.id); let place_id: i64 = place_statement.query_row( params![ stop.id, stop.name.fr, stop.name.nl, folded_name, collapsed_name, normalized_id, source_order as i64, stop.lon, stop.lat, ], |row| row.get(0), )?; search_statement.execute(params![ place_id, folded_name, collapsed_name, normalized_id, ])?; place_ids.push(place_id); } } { let mut statement = transaction.prepare( "INSERT INTO place_alias(provider, external_id, place_id, priority) VALUES ('bmc', ?1, ?2, ?3)", )?; for (external_id, stop_index, priority) in &batch.aliases { statement.execute(params![external_id, place_ids[*stop_index], priority])?; } } let mut line_ids = HashMap::with_capacity(batch.routes.len()); { let mut statement = transaction.prepare( "INSERT INTO line(provider, external_id, display_name, kind, color, text_color, active) VALUES ('bmc', ?1, ?2, ?3, ?4, ?5, 1) ON CONFLICT(provider, external_id) DO UPDATE SET display_name = excluded.display_name, kind = excluded.kind, color = excluded.color, text_color = excluded.text_color, active = 1 RETURNING id", )?; for route in &batch.routes { let line_id: i64 = statement.query_row( params![ route.line, route.line, vehicle_kind(route.kind), route.color, route.text_color, ], |row| row.get(0), )?; line_ids.insert(route.line.clone(), line_id); } } { let mut variant_statement = transaction.prepare( "INSERT INTO route_variant(line_id, destination_fr, destination_nl, source_order) VALUES (?1, ?2, ?3, ?4)", )?; let mut stop_statement = transaction.prepare( "INSERT INTO route_variant_stop(variant_id, ordinal, external_id, place_id) VALUES (?1, ?2, ?3, ?4)", )?; for sequence in &batch.sequences { let line_id = line_ids .get(&sequence.line) .with_context(|| format!("sequence line {} is unknown", sequence.line))?; variant_statement.execute(params![ line_id, sequence.destination.fr, sequence.destination.nl, sequence.source_order, ])?; let variant_id = transaction.last_insert_rowid(); for (ordinal, stop) in sequence.stops.iter().enumerate() { let place_id = stop.place_index.map(|index| place_ids[index]); stop_statement.execute(params![ variant_id, ordinal as i64, stop.external_id, place_id, ])?; } } } { let mut statement = transaction.prepare( "INSERT INTO route_shape(line_id, destination_key, geom) VALUES (?1, ?2, GeomFromText(?3, 4326))", )?; for ((line, destination), points) in &batch.shapes { let line_id = line_ids .get(line) .with_context(|| format!("shape line {line} is unknown"))?; statement.execute(params![line_id, destination, linestring(points)])?; } } clear_inactive_current_projections(transaction)?; Ok(()) } fn vehicle_kind(kind: VehicleKind) -> &'static str { match kind { VehicleKind::Tram => "tram", VehicleKind::Metro => "metro", VehicleKind::Bus => "bus", VehicleKind::Train => "train", VehicleKind::Bike => "bike", } } fn linestring(points: &[(f32, f32)]) -> String { let mut value = String::from("LINESTRING("); for (index, (lat, lon)) in points.iter().enumerate() { if index != 0 { value.push(','); } write!(&mut value, "{lon} {lat}").unwrap(); } value.push(')'); value } fn apply_villo_discovery_transaction( connection: &mut Connection, batch: PreparedVilloDiscovery, ) -> anyhow::Result<()> { let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; let snapshot_id = insert_villo_discovery_snapshots(&transaction, &batch)?; replace_villo_places(&transaction, &batch)?; let head = batch .raw .iter() .find(|raw| raw.scope.as_ref() == "station_information") .context("villo station_information snapshot is missing")?; finish_realtime_transaction(&transaction, head, snapshot_id)?; transaction.commit()?; Ok(()) } fn insert_villo_discovery_snapshots( transaction: &Transaction<'_>, batch: &PreparedVilloDiscovery, ) -> anyhow::Result { transaction.execute( "INSERT INTO provider_batch(provider, captured_at) VALUES ('villo', ?1)", [batch.batch_captured_at], )?; let batch_id = transaction.last_insert_rowid(); let feed_id = transaction .query_row( "SELECT id FROM provider_feed WHERE provider = 'villo' AND feed = 'discovery'", [], |row| row.get::<_, i64>(0), ) .context("villo/discovery provider feed is missing")?; let mut head = None; let mut statement = transaction.prepare( "INSERT INTO provider_snapshot(batch_id, feed_id, scope, captured_at, source_at, content_type, payload_sha256, payload_zstd) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", )?; for raw in &batch.raw { statement.execute(params![ batch_id, feed_id, raw.scope, raw.captured_at, raw.source_at, raw.content_type, raw.payload_sha256.as_slice(), raw.payload_zstd, ])?; if raw.scope.as_ref() == "station_information" { head = Some(transaction.last_insert_rowid()); } } head.context("villo station_information snapshot is missing") } fn replace_villo_places( transaction: &Transaction<'_>, batch: &PreparedVilloDiscovery, ) -> anyhow::Result<()> { transaction.execute( "UPDATE place SET active = 0 WHERE provider = 'villo' AND kind = 'bike'", [], )?; let mut statement = transaction.prepare( "INSERT INTO place(provider, kind, external_id, name_fr, name_nl, folded_name, collapsed_name, normalized_id, source_order, active, geom) VALUES ('villo', 'bike', ?1, ?2, ?2, ?3, ?4, ?5, ?6, 1, MakePoint(?7, ?8, 4326)) ON CONFLICT(provider, kind, external_id) DO UPDATE SET name_fr = excluded.name_fr, name_nl = excluded.name_nl, folded_name = excluded.folded_name, collapsed_name = excluded.collapsed_name, normalized_id = excluded.normalized_id, source_order = excluded.source_order, active = 1, geom = excluded.geom", )?; for (source_order, (external_id, station)) in batch.stations.iter().enumerate() { let folded_name = fold(&station.name); let collapsed_name = collapse_dupes(&folded_name); let normalized_id = crate::bmc::normalize_id(external_id); statement.execute(params![ external_id, station.name, folded_name, collapsed_name, normalized_id, source_order as i64, station.lon, station.lat, ])?; } drop(statement); clear_inactive_current_projections(transaction)?; Ok(()) } fn apply_villo_status_transaction( connection: &mut Connection, batch: PreparedVilloStatus, ) -> anyhow::Result<()> { let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; let snapshot_id = insert_realtime_snapshot(&transaction, &batch.raw)?; let mut place_ids = HashMap::new(); { let mut statement = transaction.prepare( "SELECT external_id, id FROM place WHERE provider = 'villo' AND kind = 'bike' AND active = 1", )?; let rows = statement.query_map([], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)) })?; for row in rows { let (external_id, place_id) = row?; place_ids.insert(external_id, place_id); } } transaction.execute( "DELETE FROM current_villo_status WHERE stand_id IN ( SELECT id FROM place WHERE provider = 'villo' AND kind = 'bike' )", [], )?; let bucket_at = batch.raw.captured_at.div_euclid(300_000) * 300_000; { let mut current = transaction.prepare( "INSERT INTO current_villo_status(snapshot_id, stand_id, bikes, docks, mechanical, electrical) VALUES (?1, ?2, ?3, ?4, ?5, ?6)", )?; let mut fact = transaction.prepare( "INSERT INTO villo_status_fact(stand_id, bucket_at, bikes, docks, mechanical, electrical, observed_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7) ON CONFLICT(stand_id, bucket_at) DO UPDATE SET bikes = excluded.bikes, docks = excluded.docks, mechanical = excluded.mechanical, electrical = excluded.electrical, observed_at = excluded.observed_at WHERE excluded.observed_at >= villo_status_fact.observed_at", )?; for (external_id, counts) in &batch.status { let Some(place_id) = place_ids.get(external_id.as_ref()) else { continue; }; let bikes = i64::from(counts.bikes); let docks = i64::from(counts.docks); let mechanical = counts.bikes_mechanical.map(i64::from); let electrical = counts.bikes_electrical.map(i64::from); current.execute(params![ snapshot_id, place_id, bikes, docks, mechanical, electrical, ])?; fact.execute(params![ place_id, bucket_at, bikes, docks, mechanical, electrical, batch.raw.captured_at, ])?; } } finish_realtime_transaction(&transaction, &batch.raw, snapshot_id)?; transaction.commit()?; Ok(()) } fn apply_train_stations_transaction( connection: &mut Connection, batch: PreparedTrainStations, ) -> anyhow::Result<()> { let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; let snapshot_id = insert_realtime_snapshot(&transaction, &batch.raw)?; transaction.execute( "UPDATE place SET active = 0 WHERE provider = 'irail' AND kind = 'train'", [], )?; { let mut statement = transaction.prepare( "INSERT INTO place(provider, kind, external_id, name_fr, name_nl, folded_name, collapsed_name, normalized_id, source_order, active, geom) VALUES ('irail', 'train', ?1, ?2, ?2, ?3, ?4, ?5, ?6, 1, MakePoint(?7, ?8, 4326)) ON CONFLICT(provider, kind, external_id) DO UPDATE SET name_fr = excluded.name_fr, name_nl = excluded.name_nl, folded_name = excluded.folded_name, collapsed_name = excluded.collapsed_name, normalized_id = excluded.normalized_id, source_order = excluded.source_order, active = 1, geom = excluded.geom", )?; for (source_order, station) in batch.stations.iter().enumerate() { let folded_name = fold(&station.name); let collapsed_name = collapse_dupes(&folded_name); let normalized_id = crate::bmc::normalize_id(&station.id); statement.execute(params![ station.id, station.name, folded_name, collapsed_name, normalized_id, source_order as i64, station.lon, station.lat, ])?; } } clear_inactive_current_projections(&transaction)?; transaction.execute( "UPDATE provider_feed SET current_snapshot_id = ( SELECT snapshot.id FROM provider_snapshot AS snapshot JOIN place ON place.provider = 'irail' AND place.kind = 'train' AND place.external_id = snapshot.scope AND place.active = 1 WHERE snapshot.feed_id = provider_feed.id ORDER BY snapshot.captured_at DESC, snapshot.id DESC LIMIT 1 ) WHERE provider = 'irail' AND feed = 'liveboard'", [], )?; finish_realtime_transaction(&transaction, &batch.raw, snapshot_id)?; transaction.commit()?; Ok(()) } fn apply_train_departures_transaction( connection: &mut Connection, batch: PreparedTrainDepartures, ) -> anyhow::Result<()> { let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; let snapshot_id = insert_realtime_snapshot(&transaction, &batch.raw)?; let station_id = transaction .query_row( "SELECT id FROM place WHERE provider = 'irail' AND kind = 'train' AND external_id = ?1 AND active = 1", [batch.station_id.as_ref()], |row| row.get::<_, i64>(0), ) .with_context(|| format!("unknown active train station {}", batch.station_id))?; transaction.execute( "DELETE FROM current_train_departure WHERE station_id = ?1", [station_id], )?; { let mut current = transaction.prepare( "INSERT INTO current_train_departure(snapshot_id, station_id, label, destination, expected_at, platform, delay_seconds) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", )?; let mut fact = transaction.prepare( "INSERT INTO train_departure_fact(station_id, label, destination, expected_at, first_seen_at, last_seen_at) VALUES (?1, ?2, ?3, ?4, ?5, ?5) ON CONFLICT(station_id, label, destination, expected_at) DO UPDATE SET first_seen_at = min(train_departure_fact.first_seen_at, excluded.first_seen_at), last_seen_at = max(train_departure_fact.last_seen_at, excluded.last_seen_at)", )?; for departure in &batch.departures { current.execute(params![ snapshot_id, station_id, departure.label, departure.destination, departure.expected_at, departure.platform, departure.delay_seconds, ])?; fact.execute(params![ station_id, departure.label, departure.destination, departure.expected_at, batch.raw.captured_at, ])?; } } finish_realtime_transaction(&transaction, &batch.raw, snapshot_id)?; transaction.commit()?; Ok(()) } fn apply_waiting_transaction( connection: &mut Connection, batch: PreparedWaiting, ) -> anyhow::Result<()> { let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; let snapshot_id = insert_realtime_snapshot(&transaction, &batch.raw)?; let aliases = bmc_place_aliases(&transaction)?; let lines = bmc_line_ids(&transaction)?; transaction.execute_batch( "DELETE FROM current_passage WHERE place_id IN ( SELECT id FROM place WHERE provider = 'bmc' AND kind = 'transit' ) OR line_id IN ( SELECT id FROM line WHERE provider = 'bmc' ); DELETE FROM current_end_notice WHERE place_id IN ( SELECT id FROM place WHERE provider = 'bmc' AND kind = 'transit' ) OR line_id IN ( SELECT id FROM line WHERE provider = 'bmc' );", )?; { let mut current = transaction.prepare( "INSERT INTO current_passage(snapshot_id, place_id, line_id, destination_fr, destination_nl, expected_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6)", )?; let mut fact = transaction.prepare( "INSERT INTO arrival_fact(place_id, line_id, destination_fr, expected_at, first_seen_at, last_seen_at) VALUES (?1, ?2, ?3, ?4, ?5, ?5) ON CONFLICT(place_id, line_id, destination_fr, expected_at) DO UPDATE SET first_seen_at = min(arrival_fact.first_seen_at, excluded.first_seen_at), last_seen_at = max(arrival_fact.last_seen_at, excluded.last_seen_at)", )?; for (external_id, passages) in &batch.passages { let Some(place_ids) = aliases.get(external_id.as_ref()) else { continue; }; for passage in passages { let Some(line_id) = lines.get(passage.line.as_str()).copied() else { continue; }; let expected_at = passage.expected.timestamp_millis(); for place_id in place_ids { current.execute(params![ snapshot_id, place_id, line_id, passage.destination.fr, passage.destination.nl, expected_at, ])?; fact.execute(params![ place_id, line_id, passage.destination.fr, expected_at, batch.raw.captured_at, ])?; } } } } { let mut current = transaction.prepare( "INSERT INTO current_end_notice(snapshot_id, place_id, line_id, message_fr, message_nl) VALUES (?1, ?2, ?3, ?4, ?5) ON CONFLICT(place_id, line_id) DO UPDATE SET snapshot_id = excluded.snapshot_id, message_fr = excluded.message_fr, message_nl = excluded.message_nl", )?; for (external_id, notices) in &batch.end_notices { let Some(place_ids) = aliases.get(external_id.as_ref()) else { continue; }; for notice in notices { let Some(line_id) = lines.get(notice.line.as_str()).copied() else { continue; }; for place_id in place_ids { current.execute(params![ snapshot_id, place_id, line_id, notice.message.fr, notice.message.nl, ])?; } } } } rebuild_transit_ranks(&transaction)?; finish_realtime_transaction(&transaction, &batch.raw, snapshot_id)?; transaction.commit()?; Ok(()) } fn apply_positions_transaction( connection: &mut Connection, batch: PreparedPositions, ) -> anyhow::Result<()> { let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; let snapshot_id = insert_realtime_snapshot(&transaction, &batch.raw)?; let aliases = bmc_place_aliases(&transaction)?; let lines = bmc_line_ids(&transaction)?; transaction.execute( "DELETE FROM current_vehicle_position WHERE line_id IN (SELECT id FROM line WHERE provider = 'bmc')", [], )?; let bucket_at = batch.raw.captured_at.div_euclid(60_000) * 60_000; { let mut current = transaction.prepare( "INSERT INTO current_vehicle_position( snapshot_id, line_id, terminus_external_id, point_external_id, terminus_place_id, point_place_id, distance_m ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", )?; let mut fact = transaction.prepare( "INSERT INTO vehicle_position_fact(line_id, terminus_place_id, point_place_id, distance_m, bucket_at, observed_at) VALUES (?1, ?2, ?3, ?4, ?5, ?6) ON CONFLICT(line_id, terminus_place_id, point_place_id, distance_m, bucket_at) DO UPDATE SET observed_at = max(vehicle_position_fact.observed_at, excluded.observed_at)", )?; for (external_line, positions) in &batch.positions { let Some(line_id) = lines.get(external_line.as_ref()).copied() else { continue; }; for position in positions { let terminus_place_ids = aliases.get(position.terminus.as_str()); let point_place_ids = aliases.get(position.point_id.as_str()); let mut insert_position = | terminus_place_id: Option, point_place_id: Option, | -> anyhow::Result<()> { current.execute(params![ snapshot_id, line_id, position.terminus, position.point_id, terminus_place_id, point_place_id, position.distance_m, ])?; if let (Some(terminus_place_id), Some(point_place_id)) = (terminus_place_id, point_place_id) { fact.execute(params![ line_id, terminus_place_id, point_place_id, position.distance_m, bucket_at, batch.raw.captured_at, ])?; } Ok(()) }; match (terminus_place_ids, point_place_ids) { (Some(terminus_place_ids), Some(point_place_ids)) => { for terminus_place_id in terminus_place_ids { for point_place_id in point_place_ids { insert_position( Some(*terminus_place_id), Some(*point_place_id), )?; } } } (Some(terminus_place_ids), None) => { for terminus_place_id in terminus_place_ids { insert_position(Some(*terminus_place_id), None)?; } } (None, Some(point_place_ids)) => { for point_place_id in point_place_ids { insert_position(None, Some(*point_place_id))?; } } (None, None) => insert_position(None, None)?, } } } } finish_realtime_transaction(&transaction, &batch.raw, snapshot_id)?; transaction.commit()?; Ok(()) } fn apply_disruptions_transaction( connection: &mut Connection, batch: PreparedDisruptions, ) -> anyhow::Result<()> { let transaction = connection.transaction_with_behavior(TransactionBehavior::Immediate)?; let snapshot_id = insert_realtime_snapshot(&transaction, &batch.raw)?; let aliases = bmc_place_aliases(&transaction)?; let lines = bmc_line_ids(&transaction)?; transaction.execute( "DELETE FROM current_disruption WHERE snapshot_id IN ( SELECT provider_snapshot.id FROM provider_snapshot JOIN provider_batch ON provider_batch.id = provider_snapshot.batch_id WHERE provider_batch.provider = 'bmc' )", [], )?; { let mut current = transaction.prepare( "INSERT INTO current_disruption(snapshot_id, priority, message_en, message_fr, message_nl) VALUES (?1, ?2, ?3, ?4, ?5)", )?; let mut current_line = transaction.prepare( "INSERT OR IGNORE INTO current_disruption_line(disruption_id, line_id) VALUES (?1, ?2)", )?; let mut current_place = transaction.prepare( "INSERT OR IGNORE INTO current_disruption_place(disruption_id, place_id) VALUES (?1, ?2)", )?; let mut fact = transaction.prepare( "INSERT INTO disruption_fact(payload_sha256, first_seen_at, last_seen_at, payload_zstd) VALUES (?1, ?2, ?2, ?3) ON CONFLICT(payload_sha256) DO UPDATE SET first_seen_at = min(disruption_fact.first_seen_at, excluded.first_seen_at), last_seen_at = max(disruption_fact.last_seen_at, excluded.last_seen_at)", )?; for prepared in &batch.disruptions { let mut line_ids = prepared .disruption .lines .iter() .filter_map(|external_line| lines.get(external_line.as_str()).copied()); let Some(first_line_id) = line_ids.next() else { continue; }; current.execute(params![ snapshot_id, prepared.disruption.priority, prepared.disruption.en, prepared.disruption.fr, prepared.disruption.nl, ])?; let disruption_id = transaction.last_insert_rowid(); current_line.execute(params![disruption_id, first_line_id])?; for line_id in line_ids { current_line.execute(params![disruption_id, line_id])?; } for external_place in &prepared.disruption.points { if let Some(place_ids) = aliases.get(external_place.as_str()) { for place_id in place_ids { current_place.execute(params![disruption_id, place_id])?; } } } fact.execute(params![ prepared.payload_sha256.as_slice(), batch.raw.captured_at, prepared.payload_zstd, ])?; } } finish_realtime_transaction(&transaction, &batch.raw, snapshot_id)?; transaction.commit()?; Ok(()) } fn insert_realtime_snapshot( transaction: &Transaction<'_>, raw: &PreparedRaw, ) -> anyhow::Result { transaction.execute( "INSERT INTO provider_batch(provider, captured_at) VALUES (?1, ?2)", params![raw.provider, raw.captured_at], )?; let batch_id = transaction.last_insert_rowid(); let feed_id = transaction .query_row( "SELECT id FROM provider_feed WHERE provider = ?1 AND feed = ?2", params![raw.provider, raw.feed], |row| row.get::<_, i64>(0), ) .with_context(|| format!("{}/{} provider feed is missing", raw.provider, raw.feed))?; transaction.execute( "INSERT INTO provider_snapshot(batch_id, feed_id, scope, captured_at, source_at, content_type, payload_sha256, payload_zstd) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)", params![ batch_id, feed_id, raw.scope, raw.captured_at, raw.source_at, raw.content_type, raw.payload_sha256.as_slice(), raw.payload_zstd, ], )?; Ok(transaction.last_insert_rowid()) } fn finish_realtime_transaction( transaction: &Transaction<'_>, raw: &PreparedRaw, snapshot_id: i64, ) -> anyhow::Result<()> { let updated = transaction.execute( "UPDATE provider_feed SET current_snapshot_id = ?1 WHERE provider = ?2 AND feed = ?3", params![snapshot_id, raw.provider, raw.feed], )?; ensure!( updated == 1, "{}/{} provider feed is missing", raw.provider, raw.feed ); transaction.execute( "DELETE FROM feed_failure WHERE feed_id = ( SELECT id FROM provider_feed WHERE provider = ?1 AND feed = ?2 )", params![raw.provider, raw.feed], )?; Ok(()) } fn bmc_place_aliases( transaction: &Transaction<'_>, ) -> anyhow::Result>> { let mut statement = transaction.prepare( "SELECT place_alias.external_id, place_alias.place_id, place_alias.priority FROM place_alias JOIN place ON place.id = place_alias.place_id WHERE place_alias.provider = 'bmc' AND place.provider = 'bmc' AND place.kind = 'transit' AND place.active = 1 ORDER BY place_alias.external_id, place_alias.priority, place.source_order, place.id", )?; let rows = statement.query_map([], |row| { Ok(( row.get::<_, String>(0)?, row.get::<_, i64>(1)?, row.get::<_, i64>(2)?, )) })?; let mut aliases: HashMap)> = HashMap::new(); for row in rows { let (external_id, place_id, priority) = row?; match aliases.get_mut(&external_id) { Some((preferred, place_ids)) if priority < *preferred => { *preferred = priority; place_ids.clear(); place_ids.push(place_id); } Some((preferred, place_ids)) if priority == *preferred => { place_ids.push(place_id); } Some(_) => {} None => { aliases.insert(external_id, (priority, vec![place_id])); } } } Ok(aliases .into_iter() .map(|(external_id, (_, place_ids))| (external_id, place_ids)) .collect()) } fn bmc_line_ids(transaction: &Transaction<'_>) -> anyhow::Result> { let mut statement = transaction.prepare( "SELECT external_id, id FROM line WHERE provider = 'bmc' AND active = 1", )?; let rows = statement.query_map([], |row| { Ok((row.get::<_, String>(0)?, row.get::<_, i64>(1)?)) })?; let mut lines = HashMap::new(); for row in rows { let (external_id, line_id) = row?; lines.insert(external_id, line_id); } Ok(lines) } fn clear_inactive_current_projections(transaction: &Transaction<'_>) -> anyhow::Result<()> { transaction.execute_batch( "DELETE FROM current_passage WHERE EXISTS ( SELECT 1 FROM place WHERE place.id = current_passage.place_id AND place.active = 0 ) OR EXISTS ( SELECT 1 FROM line WHERE line.id = current_passage.line_id AND line.active = 0 ); DELETE FROM current_end_notice WHERE EXISTS ( SELECT 1 FROM place WHERE place.id = current_end_notice.place_id AND place.active = 0 ) OR EXISTS ( SELECT 1 FROM line WHERE line.id = current_end_notice.line_id AND line.active = 0 ); DELETE FROM current_vehicle_position WHERE EXISTS ( SELECT 1 FROM line WHERE line.id = current_vehicle_position.line_id AND line.active = 0 ) OR EXISTS ( SELECT 1 FROM place WHERE place.id = current_vehicle_position.terminus_place_id AND place.active = 0 ) OR EXISTS ( SELECT 1 FROM place WHERE place.id = current_vehicle_position.point_place_id AND place.active = 0 ); DELETE FROM current_train_departure WHERE EXISTS ( SELECT 1 FROM place WHERE place.id = current_train_departure.station_id AND place.active = 0 ); DELETE FROM current_villo_status WHERE EXISTS ( SELECT 1 FROM place WHERE place.id = current_villo_status.stand_id AND place.active = 0 ); DELETE FROM current_disruption WHERE ( EXISTS ( SELECT 1 FROM current_disruption_line WHERE disruption_id = current_disruption.id ) OR EXISTS ( SELECT 1 FROM current_disruption_place WHERE disruption_id = current_disruption.id ) ) AND NOT EXISTS ( SELECT 1 FROM current_disruption_line JOIN line ON line.id = current_disruption_line.line_id WHERE current_disruption_line.disruption_id = current_disruption.id AND line.active = 1 ) AND NOT EXISTS ( SELECT 1 FROM current_disruption_place JOIN place ON place.id = current_disruption_place.place_id WHERE current_disruption_place.disruption_id = current_disruption.id AND place.active = 1 ); DELETE FROM current_disruption_line WHERE EXISTS ( SELECT 1 FROM line WHERE line.id = current_disruption_line.line_id AND line.active = 0 ); DELETE FROM current_disruption_place WHERE EXISTS ( SELECT 1 FROM place WHERE place.id = current_disruption_place.place_id AND place.active = 0 );", )?; Ok(()) } fn rebuild_transit_ranks(transaction: &Transaction<'_>) -> anyhow::Result<()> { transaction.execute_batch( "DELETE FROM transit_with_passages WHERE place_id IN ( SELECT id FROM place WHERE provider = 'bmc' AND kind = 'transit' ); DELETE FROM transit_with_notices WHERE place_id IN ( SELECT id FROM place WHERE provider = 'bmc' AND kind = 'transit' ); DELETE FROM transit_silent WHERE place_id IN ( SELECT id FROM place WHERE provider = 'bmc' AND kind = 'transit' );", )?; transaction.execute( "INSERT INTO transit_with_passages(place_id, source_order, geom) SELECT DISTINCT place.id, place.source_order, place.geom FROM current_passage JOIN line ON line.id = current_passage.line_id AND line.active = 1 JOIN place ON place.id = current_passage.place_id AND place.provider = 'bmc' AND place.kind = 'transit' AND place.active = 1", [], )?; transaction.execute( "INSERT INTO transit_with_notices(place_id, source_order, geom) SELECT DISTINCT place.id, place.source_order, place.geom FROM current_end_notice JOIN line ON line.id = current_end_notice.line_id AND line.active = 1 JOIN place ON place.id = current_end_notice.place_id AND place.provider = 'bmc' AND place.kind = 'transit' AND place.active = 1 AND NOT EXISTS ( SELECT 1 FROM transit_with_passages WHERE transit_with_passages.place_id = place.id )", [], )?; transaction.execute( "INSERT INTO transit_silent(place_id, source_order, geom) SELECT place.id, place.source_order, place.geom FROM place WHERE place.provider = 'bmc' AND place.kind = 'transit' AND place.active = 1 AND NOT EXISTS ( SELECT 1 FROM transit_with_passages WHERE transit_with_passages.place_id = place.id ) AND NOT EXISTS ( SELECT 1 FROM transit_with_notices WHERE transit_with_notices.place_id = place.id ) AND place.name_fr NOT IN ( SELECT place.name_fr FROM place JOIN transit_with_passages ON transit_with_passages.place_id = place.id UNION SELECT place.name_fr FROM place JOIN transit_with_notices ON transit_with_notices.place_id = place.id )", [], )?; Ok(()) }