diff --git a/src/handlers/routes.rs b/src/handlers/routes.rs index 3c513d8..0f3255c 100644 --- a/src/handlers/routes.rs +++ b/src/handlers/routes.rs @@ -20,9 +20,10 @@ use tracing::{error, info}; use crate::environment::AppState; use crate::graphql::TripQueryParams; use crate::models::{ - BusScheduleDetail, BusScheduleDetails, GTFSStop, MemoryUsageStats, MinimalEmployee, - NandiRoutesRes, RouteStopMapping, StopCodeFromProviderStopCodeResponse, TripDetails, - VehicleData, VehicleMetadataResponse, VehicleOperationData, VehicleServiceTypeResponse, + BusScheduleDetail, BusScheduleDetails, GTFSStop, MemoryUsageStats, + MinimalEmployee, NandiRoutesRes, RouteStopMapping, StopCodeFromProviderStopCodeResponse, + TripDetails, VehicleData, VehicleMetadataResponse, VehicleOperationData, + VehicleServiceTypeResponse, }; use crate::services::db_vehicle_reader::{chalo_gtfs_ids, is_chalo_gtfs_id}; // alias for query param map (string->string) @@ -176,6 +177,10 @@ pub fn create_routes(cfg: &mut actix_web::web::ServiceConfig) { "/station-children/{gtfs_id}/{stop_code}", actix_web::web::get().to(get_station_children), ) + .route( + "/cluster/{gtfs_id}/destinations/{stop_code}", + actix_web::web::get().to(get_cluster_destinations), + ) .route("/ready", actix_web::web::get().to(readiness_probe)) .route("/version/{gtfs_id}", actix_web::web::get().to(get_version)) .route( @@ -667,6 +672,35 @@ pub async fn get_route_stop_mapping_by_stop( Ok(HttpResponse::Ok().json(mappings)) } +#[utoipa::path( + get, + path = "/cluster/{gtfs_id}/destinations/{stop_code}", + tag = "Cluster", + params( + ("gtfs_id" = String, Path, description = "GTFS feed identifier"), + ("stop_code" = String, Path, description = "Source stop code"), + ), + responses(( + status = 200, + description = "Destination stop_codes reachable downstream of the source on the same trip pattern, \ + deduplicated by H3 cluster (one representative stop_code per reachable cluster). \ + Does NOT include destinations reachable via a transfer. \ + Falls back to a single-stop walk when the source stop has no cluster_id (logged at warn). \ + Returns 404 on unknown gtfs_id; returns [] for an unknown stop_code or a stop with no outgoing routes.", + body = Vec, + )) +)] +pub async fn get_cluster_destinations( + app_state: Data, + path: Path<(String, String)>, +) -> AppResult { + let (gtfs_id, stop_code) = path.into_inner(); + let destinations = app_state + .gtfs_service + .get_cluster_destinations_for_stop(>fs_id, &stop_code)?; + Ok(HttpResponse::Ok().json(destinations)) +} + #[utoipa::path( get, path = "/routes/{gtfs_id}/fuzzy/{query}", diff --git a/src/models.rs b/src/models.rs index 516180e..2097d43 100644 --- a/src/models.rs +++ b/src/models.rs @@ -424,6 +424,14 @@ pub struct GTFSStop { pub hindi_name: Option, #[serde(rename = "regionalName")] pub regional_name: Option, + #[serde(default, skip_serializing)] + pub desc: Option, + #[serde( + rename = "clusterId", + default, + skip_serializing_if = "Option::is_none" + )] + pub cluster_id: Option, } #[derive(Debug, Clone, Serialize, Deserialize)] @@ -450,6 +458,8 @@ pub struct GTFSRouteData { #[derive(Debug, Clone, Default, Serialize)] pub struct GTFSStopData { pub stops: HashMap, + #[serde(default)] + pub by_cluster_id: HashMap>, } #[derive(Debug, Clone, Default, Serialize)] diff --git a/src/services/gtfs_service.rs b/src/services/gtfs_service.rs index cd9ffa8..1934e43 100644 --- a/src/services/gtfs_service.rs +++ b/src/services/gtfs_service.rs @@ -1,10 +1,11 @@ use crate::environment::AppConfig; use crate::models::{ - cast_vehicle_type, clean_identifier, CachedDataResponse, GTFSData, GTFSRouteData, GTFSStop, - GTFSStopData, LatLong, NandiPattern, NandiPatternDetails, NandiRoutesRes, PlatformInfo, - ProviderStopCodeRecord, RouteServiceTierRecord, RouteStopMapping, SeatLayoutMappingRecord, - ServiceTierType, StaticFleetInfo, StaticFleetInfoRecord, StopGeojson, StopGeojsonRecord, - StopRegionalNameRecord, SuburbanStopInfo, SuburbanStopInfoRecord, + cast_vehicle_type, clean_identifier, CachedDataResponse, GTFSData, + GTFSRouteData, GTFSStop, GTFSStopData, LatLong, NandiPattern, NandiPatternDetails, + NandiRoutesRes, PlatformInfo, ProviderStopCodeRecord, RouteServiceTierRecord, + RouteStopMapping, SeatLayoutMappingRecord, ServiceTierType, StaticFleetInfo, + StaticFleetInfoRecord, StopGeojson, StopGeojsonRecord, StopRegionalNameRecord, + SuburbanStopInfo, SuburbanStopInfoRecord, }; use crate::models::{GTFSAlternateStopData, TripDetails}; use crate::tools::error::{AppError, AppResult}; @@ -36,6 +37,21 @@ fn get_sha256_hash(val: &T) -> String { format!("{:x}", hasher.finalize()) } +const SENTINEL_CLUSTER_ID: &str = "INVALID_SENTINEL"; + +fn parse_cluster_id_from_desc(desc: Option<&str>) -> Option { + let raw = desc?.trim(); + if raw.is_empty() || raw == "{}" { + return None; + } + let parsed: serde_json::Value = serde_json::from_str(raw).ok()?; + let cid = parsed.get("clusterId")?.as_str()?; + if cid == SENTINEL_CLUSTER_ID { + return None; + } + Some(cid.to_string()) +} + fn normalize_stop_name(name: &str) -> String { name.to_lowercase() .chars() @@ -760,6 +776,8 @@ impl GTFSService { .get(gtfs_id) .and_then(|m| m.get(stop_code)); + let cluster_id = parse_cluster_id_from_desc(stop.desc.as_deref()); + // Create a new GTFSStop with the clean stop code let stop_res = GTFSStop { id: stop.id.clone(), @@ -771,6 +789,8 @@ impl GTFSService { cluster: stop.cluster.clone(), hindi_name: regional_name.map(|r| r.hindi_name.clone()), regional_name: regional_name.map(|r| r.regional_name.clone()), + desc: None, + cluster_id: cluster_id.clone(), }; if stop.cluster.is_some() { let cluster_stop_res = GTFSStop { @@ -783,15 +803,49 @@ impl GTFSService { cluster: stop.cluster.clone(), hindi_name: regional_name.map(|r| r.hindi_name.clone()), regional_name: regional_name.map(|r| r.regional_name.clone()), + desc: None, + cluster_id: cluster_id.clone(), }; stop_data .stops .insert(stop.cluster.clone().unwrap(), cluster_stop_res); } + if let Some(cid) = &cluster_id { + stop_data + .by_cluster_id + .entry(cid.clone()) + .or_default() + .push(stop_code.to_string()); + } + stop_data.stops.insert(stop_code.to_string(), stop_res); } + for (gtfs_id, data) in &stops_by_gtfs { + let total = data.stops.len(); + let clustered = data + .stops + .values() + .filter(|s| s.cluster_id.is_some()) + .count(); + if clustered == 0 { + warn!( + gtfs_id = %gtfs_id, + total_stops = total, + "No stops have a cluster_id; cluster destinations endpoint will fall back to single-stop walks for every request", + ); + } else { + info!( + gtfs_id = %gtfs_id, + total_stops = total, + clustered_stops = clustered, + distinct_clusters = data.by_cluster_id.len(), + "cluster_id coverage after build", + ); + } + } + stops_by_gtfs } @@ -1442,6 +1496,136 @@ impl GTFSService { Ok(stops) } + pub fn get_cluster_destinations_for_stop( + &self, + gtfs_id: &str, + stop_code: &str, + ) -> AppResult> { + let data = self.data.load_full(); + let gtfs_id = clean_identifier(gtfs_id); + let stop_code = clean_identifier(stop_code); + + // Unknown gtfs_id is a configuration error → 404. Unknown stop_code + // inside a known feed is semantically "no destinations" → 200 []. + let stops_data = data.stops_by_gtfs.get(>fs_id).ok_or_else(|| { + AppError::NotFound(format!("Stops data not found for gtfs_id: {}", gtfs_id)) + })?; + + let src_stop = match stops_data.stops.get(&stop_code) { + Some(s) => s, + None => { + info!( + gtfs_id = %gtfs_id, + stop_code = %stop_code, + "destinations: stop not found in feed, returning empty list", + ); + return Ok(Vec::new()); + } + }; + + let sibling_codes: Vec = match src_stop.cluster_id.as_ref() { + Some(cid) => { + let siblings = stops_data + .by_cluster_id + .get(cid) + .cloned() + .unwrap_or_default(); + info!( + gtfs_id = %gtfs_id, + stop_code = %stop_code, + cluster_id = %cid, + siblings = siblings.len(), + "destinations: cluster walk", + ); + siblings + } + None => { + warn!( + gtfs_id = %gtfs_id, + stop_code = %stop_code, + "destinations: no cluster_id, falling back to single-stop walk", + ); + vec![stop_code.clone()] + } + }; + + let route_data = match data.route_data_by_gtfs.get(>fs_id) { + Some(r) => r, + None => return Ok(Vec::new()), + }; + + // For each sibling source stop, collect (route_code, src_seq) pairs. + // Multiple siblings may serve the same route — keep the min src_seq so + // we look as far downstream as possible on that route. + let mut src_seq_by_route: HashMap, i32> = HashMap::new(); + for sib in &sibling_codes { + if let Some(idxs) = route_data.by_stop.get(sib) { + for &i in idxs { + if let Some(m) = route_data.mappings.get(i) { + src_seq_by_route + .entry(m.route_code.clone()) + .and_modify(|existing| { + if m.sequence_num < *existing { + *existing = m.sequence_num; + } + }) + .or_insert(m.sequence_num); + } + } + } + } + + let src_cluster = src_stop.cluster_id.as_deref(); + let mut rep_by_key: HashMap = HashMap::new(); + for (route_code, src_seq) in &src_seq_by_route { + let idxs = match route_data.by_route.get(route_code.as_ref()) { + Some(v) => v, + None => continue, + }; + for &i in idxs { + let m = match route_data.mappings.get(i) { + Some(m) => m, + None => continue, + }; + if m.sequence_num <= *src_seq { + continue; + } + let dst_code = m.stop_code.as_ref(); + let dst_stop = match stops_data.stops.get(dst_code) { + Some(s) => s, + None => continue, + }; + match (src_cluster, dst_stop.cluster_id.as_deref()) { + (Some(s), Some(d)) if s == d => continue, + (None, _) if dst_code == stop_code.as_str() => continue, + _ => {} + } + let dedup_key = dst_stop + .cluster_id + .clone() + .unwrap_or_else(|| dst_code.to_string()); + rep_by_key + .entry(dedup_key) + .and_modify(|existing| { + if dst_code < existing.as_str() { + *existing = dst_code.to_string(); + } + }) + .or_insert_with(|| dst_code.to_string()); + } + } + + let mut out: Vec = rep_by_key.into_values().collect(); + out.sort(); + info!( + gtfs_id = %gtfs_id, + stop_code = %stop_code, + destinations = out.len(), + "destinations: result", + ); + Ok(out) + } + pub async fn get_stop( &self, gtfs_id: &str, diff --git a/src/swagger.rs b/src/swagger.rs index d89df90..f9e0629 100644 --- a/src/swagger.rs +++ b/src/swagger.rs @@ -47,6 +47,7 @@ use crate::services::operator::QueryBody; routes::get_all_stops_by_ids, routes::get_all_route_stop_mappings_by_route_codes, routes::get_all_route_stop_mappings_by_stop_codes, + routes::get_cluster_destinations, routes::get_all_vehicles_by_ids, routes::get_routes_by_ids, // Trip @@ -145,6 +146,7 @@ use crate::services::operator::QueryBody; (name = "Vehicle", description = "Vehicle and service type endpoints"), (name = "Schedule", description = "Bus schedule endpoints"), (name = "Bulk", description = "Bulk data retrieval endpoints"), + (name = "Cluster", description = "H3 cluster-based stop endpoints"), (name = "Trip", description = "Trip information endpoints"), (name = "System", description = "Health and system info endpoints"), (name = "Internal Operator", description = "Internal operator CRUD and management endpoints"),