[feat]: unpacked cluster id and returning destination using the clust…

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<String>,
+    ))
+)]
+pub async fn get_cluster_destinations(
+    app_state: Data<AppState>,
+    path: Path<(String, String)>,
+) -> AppResult<HttpResponse> {
+    let (gtfs_id, stop_code) = path.into_inner();
+    let destinations = app_state
+        .gtfs_service
+        .get_cluster_destinations_for_stop(&gtfs_id, &stop_code)?;
+    Ok(HttpResponse::Ok().json(destinations))
+}
+
 #[utoipa::path(
     get,
     path = "/routes/{gtfs_id}/fuzzy/{query}",

src/models.rs

@@ -424,6 +424,14 @@ pub struct GTFSStop {
     pub hindi_name: Option<String>,
     #[serde(rename = "regionalName")]
     pub regional_name: Option<String>,
+    #[serde(default, skip_serializing)]
+    pub desc: Option<String>,
+    #[serde(
+        rename = "clusterId",
+        default,
+        skip_serializing_if = "Option::is_none"
+    )]
+    pub cluster_id: Option<String>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
@@ -450,6 +458,8 @@ pub struct GTFSRouteData {
 #[derive(Debug, Clone, Default, Serialize)]
 pub struct GTFSStopData {
     pub stops: HashMap<String, GTFSStop>,
+    #[serde(default)]
+    pub by_cluster_id: HashMap<String, Vec<String>>,
 }
 
 #[derive(Debug, Clone, Default, Serialize)]

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<T: Serialize>(val: &T) -> String {
     format!("{:x}", hasher.finalize())
 }
 
+const SENTINEL_CLUSTER_ID: &str = "INVALID_SENTINEL";
+
+fn parse_cluster_id_from_desc(desc: Option<&str>) -> Option<String> {
+    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<Vec<String>> {
+        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(&gtfs_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<String> = 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(&gtfs_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<Arc<str>, 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<String, String> = 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<String> = 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,

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"),