diff --git a/.gitignore b/.gitignore index dc772e6..5d3d95b 100644 --- a/.gitignore +++ b/.gitignore @@ -48,6 +48,8 @@ dhall-configs/secrets/gtfs_in_memory_server_rust.dhall # Temporary files *.tmp *.temp +CLAUDE.md +.claude/ # Unused files ignored/ \ No newline at end of file diff --git a/assets/fleet_tag_list.csv b/assets/fleet_tag_list.csv index 702a297..4e87134 100644 --- a/assets/fleet_tag_list.csv +++ b/assets/fleet_tag_list.csv @@ -8,4 +8,9 @@ chennai_bus,K5659,P6 chennai_bus,K5660,P7 chennai_bus,K5661,P8 chennai_bus,K5662,P10 -chennai_bus,K5663,P9 \ No newline at end of file +chennai_bus,K5663,P9 +kolkata_bus,K1001,P1 +kolkata_bus,K1002,P2 +kolkata_bus,K1003,P3 +kolkata_bus,K1004,P4 +kolkata_bus,K1005,P5 \ No newline at end of file diff --git a/assets/route_service_tiers.csv b/assets/route_service_tiers.csv index 395a6ce..be5cbe9 100644 --- a/assets/route_service_tiers.csv +++ b/assets/route_service_tiers.csv @@ -7,7 +7,11 @@ kolkata_bus,sh1,PREMIUM kolkata_bus,sh2,PREMIUM kolkata_bus,sh3,PREMIUM kolkata_bus,sh4,PREMIUM -kolkata_bus,1001,PREMIUM -kolkata_bus,1002,PREMIUM -kolkata_bus,1003,PREMIUM -kolkata_bus,1004,PREMIUM \ No newline at end of file +kolkata_bus,1001,SHUTTLE +kolkata_bus,1002,SHUTTLE +kolkata_bus,1003,SHUTTLE +kolkata_bus,1004,SHUTTLE +kolkata_bus,1001033,SHUTTLE +kolkata_bus,1002033,SHUTTLE +kolkata_bus,1003033,SHUTTLE +kolkata_bus,1004033,SHUTTLE \ No newline at end of file diff --git a/dhall-configs/dev/gtfs_in_memory_server_rust.dhall b/dhall-configs/dev/gtfs_in_memory_server_rust.dhall index c142448..baacb09 100644 --- a/dhall-configs/dev/gtfs_in_memory_server_rust.dhall +++ b/dhall-configs/dev/gtfs_in_memory_server_rust.dhall @@ -5,6 +5,8 @@ let logger_cfg = { log_to_file = False } +let secrets = ../secrets/gtfs_in_memory_server_rust.example.dhall + in { -- Logger configuration logger_cfg = logger_cfg, @@ -55,5 +57,13 @@ in { -- Bhubaneswar vehicle cache configuration bhubaneswar_cache_update_interval = 10, - phone_number_hash_key = "HASH_KEY" + phone_number_hash_key = "HASH_KEY", + + enable_schedule_reconciliation=True, + -- OSRTC station cache configuration + osrtc_base_url = Some "OSRTC_BASE_URL", + osrtc_username = secrets.osrtc_username, + osrtc_secret_key = secrets.osrtc_secret_key, + osrtc_station_refresh_interval_hours = 1, + osrtc_feed_key = Some "odisha_osrtc" } diff --git a/dhall-configs/secrets/gtfs_in_memory_server_rust.example.dhall b/dhall-configs/secrets/gtfs_in_memory_server_rust.example.dhall index e63ecd9..dbfd8d2 100644 --- a/dhall-configs/secrets/gtfs_in_memory_server_rust.example.dhall +++ b/dhall-configs/secrets/gtfs_in_memory_server_rust.example.dhall @@ -4,5 +4,7 @@ otp_url = "OTP_URL", bhubaneswar_external_auth = Some "TEXT", phone_number_hash_key = Some "TEXT", - enable_schedule_reconciliation = False + enable_schedule_reconciliation = False, + osrtc_username = Some "OSRTC_USERNAME", + osrtc_secret_key = Some "OSRTC_PASSWORD" } diff --git a/src/environment.rs b/src/environment.rs index 030ed88..80415c4 100644 --- a/src/environment.rs +++ b/src/environment.rs @@ -20,6 +20,7 @@ use crate::services::{ fleet_operator::{DBFleetOperatorService, FleetOperatorService, MockFleetOperatorService}, gtfs_service::GTFSService, operator::{DBOperatorService, MockOperatorService, OperatorService}, + osrtc_station_cache::OsrtcStationCache, trip_service::TripService, }; use crate::tools::dhall::read_dhall_config as dhall_read_config; @@ -72,6 +73,11 @@ pub struct AppConfig { pub phone_number_hash_key: String, /// Enable schedule-based active trip reconciliation (default: false) pub enable_schedule_reconciliation: bool, + pub osrtc_base_url: Option, + pub osrtc_username: Option, + pub osrtc_secret_key: Option, + pub osrtc_station_refresh_interval_hours: u64, + pub osrtc_feed_key: Option, } impl OtpConfig { @@ -137,6 +143,7 @@ pub struct AppState { pub fleet_operator_service: Arc, pub trip_service: Arc, pub chalo_vehicle_cache: Arc, + pub osrtc_cache: Option>, pub config: AppConfig, pub bus_registration_mapping: Arc>>, pub fleet_list: Arc>>>, @@ -299,6 +306,32 @@ impl AppState { chalo_vehicle_cache.set_update_interval(app_config.bhubaneswar_cache_update_interval); let chalo_vehicle_cache = Arc::new(chalo_vehicle_cache); + let osrtc_cache = match ( + &app_config.osrtc_base_url, + &app_config.osrtc_username, + &app_config.osrtc_secret_key, + ) { + (Some(base_url), Some(username), Some(secret_key)) => { + let cache = OsrtcStationCache::new( + base_url.clone(), + username.clone(), + secret_key.clone(), + app_config.osrtc_station_refresh_interval_hours * 3600, + )?; + if let Err(e) = cache.initialize().await { + error!( + "OSRTC station cache initial load failed (will retry in background): {}", + e + ); + } + Some(Arc::new(cache)) + } + _ => { + info!("OSRTC credentials not configured; OSRTC station cache disabled"); + None + } + }; + // Load bus registration mapping from CSV let bus_registration_mapping = Arc::new(Self::load_bus_registration_mapping().await?); @@ -317,6 +350,7 @@ impl AppState { fleet_operator_service, trip_service, chalo_vehicle_cache, + osrtc_cache, config: app_config, bus_registration_mapping, fleet_list: Arc::new(Self::load_fleet_list().await?), diff --git a/src/handlers/routes.rs b/src/handlers/routes.rs index 3c513d8..676000f 100644 --- a/src/handlers/routes.rs +++ b/src/handlers/routes.rs @@ -1,4 +1,7 @@ -use crate::services::fleet_operator::{TripAction, WaybillAnchor}; +use crate::services::fleet_operator::{ + EmployeeLoginRequest, EmployeeLoginResponse, EmployeeRegisterRequest, EmployeeRegisterResponse, + TripAction, WaybillAnchor, +}; use crate::services::operator::{ break_types, day_types, shift_types, trip_types, waybill_statuses, QueryBody, SUPPORTED_OPERATOR_GTFS_IDS, @@ -25,6 +28,7 @@ use crate::models::{ VehicleData, VehicleMetadataResponse, VehicleOperationData, VehicleServiceTypeResponse, }; use crate::services::db_vehicle_reader::{chalo_gtfs_ids, is_chalo_gtfs_id}; +use crate::services::osrtc_station_cache::osrtc_station_to_route_stop_mapping; // alias for query param map (string->string) type MapStringString = std::collections::HashMap; use crate::{ @@ -132,7 +136,15 @@ pub fn create_routes(cfg: &mut actix_web::web::ServiceConfig) { "/currentTripDetails", web::post().to(fleet_operator_current_trip_details), ) - .route("/verify", web::post().to(fleet_operator_verify)), + .route("/verify", web::post().to(fleet_operator_verify)) + .route( + "/employee/login", + web::post().to(fleet_operator_employee_login), + ) + .route( + "/employee/register", + web::post().to(fleet_operator_employee_register), + ), ) .route( "/bus-route-schedule/{gtfs_id}/{route_id}", @@ -176,6 +188,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( @@ -206,6 +222,10 @@ pub fn create_routes(cfg: &mut actix_web::web::ServiceConfig) { actix_web::web::get().to(get_connection_stats), ) .route("/trip/{trip_id}", actix_web::web::get().to(get_trip_data)) + .route( + "/waybill/{gtfs_id}/metadata/{waybill_no}", + actix_web::web::get().to(get_waybill_metadata), + ) .route( "/trip-cache/stats", actix_web::web::get().to(get_trip_cache_stats), @@ -667,6 +687,39 @@ 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, computed against the \ + server's precomputed representative pattern per route (the longest pattern observed \ + for each route at build time — see build_route_data). Deduplicated by H3 cluster \ + so the response carries one representative stop_code per reachable cluster. \ + Does NOT include destinations reachable via a transfer, and does NOT enumerate \ + every trip pattern — patterns shorter than the longest for the same route are not \ + walked. Falls back to a single-stop walk when the source stop has no cluster_id \ + (logged at debug). Returns 404 on unknown gtfs_id; returns [] for an unknown \ + stop_code or a stop with no outgoing routes on the representative pattern.", + 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}", @@ -724,6 +777,18 @@ pub async fn get_routes_fuzzy( )] pub async fn get_stops(app_state: Data, path: Path) -> AppResult { let gtfs_id = path.into_inner(); + if app_state.config.osrtc_feed_key.as_deref() == Some(gtfs_id.as_str()) { + let cache = app_state + .osrtc_cache + .as_ref() + .ok_or_else(|| AppError::Internal("OSRTC cache not configured".to_string()))?; + let stations = cache.get_all_stations().await; + let mappings: Vec = stations + .iter() + .map(osrtc_station_to_route_stop_mapping) + .collect(); + return Ok(HttpResponse::Ok().json(mappings)); + } let stops = app_state.gtfs_service.get_stops(>fs_id).await?; Ok(HttpResponse::Ok().json(stops)) } @@ -776,6 +841,17 @@ pub async fn get_stop( path: Path<(String, String)>, ) -> AppResult { let (gtfs_id, stop_code) = path.into_inner(); + if app_state.config.osrtc_feed_key.as_deref() == Some(gtfs_id.as_str()) { + let cache = app_state + .osrtc_cache + .as_ref() + .ok_or_else(|| AppError::Internal("OSRTC cache not configured".to_string()))?; + let station = cache + .get_station_by_id(&stop_code) + .await + .ok_or_else(|| AppError::NotFound(format!("OSRTC station not found: {stop_code}")))?; + return Ok(HttpResponse::Ok().json(osrtc_station_to_route_stop_mapping(&station))); + } let (stop, maybe_mapping) = app_state .gtfs_service .get_stop(>fs_id, &stop_code) @@ -900,7 +976,7 @@ pub async fn get_version(app_state: Data, path: Path) -> AppRe } #[derive(Deserialize)] -struct TripQuery { +pub struct TripQuery { trip_number: Option, #[serde(rename = "passVerifyReq")] pass_verify_req: Option, @@ -997,7 +1073,7 @@ pub async fn get_vehicle_metadata_by_gtfs_id( .and_then(|by_vehicle| by_vehicle.get(&vehicle_no)) .cloned(); - let service_type = get_vehicle_service_type_with_optional_chennai_cache( + let service_type = get_vehicle_service_type_with_optional_internal_cache( app_state.as_ref(), >fs_id, &vehicle_no, @@ -1025,19 +1101,20 @@ pub async fn get_vehicle_metadata_by_gtfs_id( })) } -async fn get_vehicle_service_type_with_optional_chennai_cache( +async fn get_vehicle_service_type_with_optional_internal_cache( app_state: &AppState, gtfs_id: &str, vehicle_no: &str, ) -> AppResult> { - if gtfs_id != "chennai_bus" { + if !SUPPORTED_OPERATOR_GTFS_IDS.contains(>fs_id) { return fetch_vehicle_service_type(app_state, gtfs_id, vehicle_no).await; } + let cache_key = format!("{}:{}", gtfs_id, vehicle_no); let now = Instant::now(); { let cache = app_state.chennai_service_type_cache.read().await; - if let Some((expires_at, cached)) = cache.get(vehicle_no) { + if let Some((expires_at, cached)) = cache.get(&cache_key) { if *expires_at > now { return Ok(cached.clone()); } @@ -1045,9 +1122,9 @@ async fn get_vehicle_service_type_with_optional_chennai_cache( } { let mut cache = app_state.chennai_service_type_cache.write().await; - if let Some((expires_at, _)) = cache.get(vehicle_no) { + if let Some((expires_at, _)) = cache.get(&cache_key) { if *expires_at <= now { - cache.remove(vehicle_no); + cache.remove(&cache_key); } } } @@ -1055,7 +1132,7 @@ async fn get_vehicle_service_type_with_optional_chennai_cache( if computed.is_some() { let mut cache = app_state.chennai_service_type_cache.write().await; cache.insert( - vehicle_no.to_string(), + cache_key, (now + Duration::from_secs(60 * 60), computed.clone()), ); } @@ -1718,7 +1795,9 @@ async fn get_service_type_by_vehicle_impl( .map(|details| details.iter().any(|t| t.is_active_trip.unwrap_or(false))) .unwrap_or(false); - if !response.is_active_trip && !has_active_in_remaining && response.trip_number != Some(1) + if !response.is_active_trip + && !has_active_in_remaining + && response.trip_number != Some(1) { // Try to find trip 1 in schedule_details and promote it if let Some(ref schedule_map) = vehicle_data.schedule_details { @@ -2183,6 +2262,31 @@ pub async fn get_trip_data( Ok(HttpResponse::Ok().json(trip_data)) } + +#[utoipa::path( + get, + path = "/waybill/{gtfs_id}/metadata/{waybill_no}", + tag = "Waybill", + params( + ("gtfs_id" = String, Path, description = "GTFS feed identifier"), + ("waybill_no" = String, Path, description = "Waybill number"), + ), + responses((status = 200, description = "Waybill metadata with driver details", body = crate::models::WaybillMetadataResponse)) +)] +pub async fn get_waybill_metadata( + app_state: Data, + path: Path<(String, String)>, +) -> AppResult { + let (gtfs_id, waybill_no) = path.into_inner(); + + let waybill_metadata = app_state + .db_vehicle_reader_internal + .get_waybill_metadata(>fs_id, &waybill_no) + .await?; + + Ok(HttpResponse::Ok().json(waybill_metadata)) +} + const SPEED_KM_PER_HOUR: f64 = 25.0; const EARTH_RADIUS_KM: f64 = 6371.0; // Earth radius in kilometers @@ -2372,18 +2476,29 @@ pub async fn get_bus_trip_schedule( .await .unwrap_or_default(); - // Fetch from external tables - let external_rows = app_state - .db_vehicle_reader - .get_chennai_waybill_by_waybill_and_trip(&waybill_no, trip_number) - .await?; - - // Fetch from internal tables - let internal_rows = app_state - .db_vehicle_reader_internal - .get_chennai_waybill_by_waybill_and_trip(&waybill_no, trip_number, >fs_id) - .await - .unwrap_or_default(); + // kolkata_bus: internal only; chennai_bus: both external + internal + let (external_rows, internal_rows) = if gtfs_id == "kolkata_bus" { + ( + vec![], + app_state + .db_vehicle_reader_internal + .get_chennai_waybill_by_waybill_and_trip(&waybill_no, trip_number, >fs_id) + .await + .unwrap_or_default(), + ) + } else { + ( + app_state + .db_vehicle_reader + .get_chennai_waybill_by_waybill_and_trip(&waybill_no, trip_number) + .await?, + app_state + .db_vehicle_reader_internal + .get_chennai_waybill_by_waybill_and_trip(&waybill_no, trip_number, >fs_id) + .await + .unwrap_or_default(), + ) + }; let all: Vec = external_rows .into_iter() @@ -2463,10 +2578,12 @@ pub async fn get_bus_route_schedule( ) -> AppResult { let (gtfs_id, route_id) = path.into_inner(); - // chennai_bus guy + // chennai_bus / kolkata_bus - internal DB flow // Single join query returns waybills + trip (bstd & bstf) times // No per-vehicle get_vehicle_data call req - if gtfs_id == "chennai_bus" { + if SUPPORTED_OPERATOR_GTFS_IDS.contains(>fs_id.as_str()) { + // kolkata_bus: internal reader only; chennai_bus: both internal + external + let is_kolkata = gtfs_id == "kolkata_bus"; let just_internal = query.just_internal.unwrap_or(false); let just_external = query.just_external.unwrap_or(false); let vehicle_number = query.vehicle_number.as_deref(); @@ -2480,7 +2597,8 @@ pub async fn get_bus_route_schedule( let mut all_rows = Vec::new(); // 1. Fetch from external (existing) tables unless justInternal is strictly true - if !just_internal { + // Skip for kolkata_bus (internal only) + if !just_internal && !is_kolkata { let mut ext_rows = app_state .db_vehicle_reader .get_chennai_waybills_by_route_id(&route_id, vehicle_number) @@ -3733,3 +3851,51 @@ pub async fn fleet_operator_verify( .await?; Ok(HttpResponse::Ok().json(response)) } + +#[utoipa::path( + post, + path = "/internal/fleet-operator/{gtfs_id}/employee/login", + tag = "Fleet Operator", + params(("gtfs_id" = String, Path, description = "GTFS dataset identifier")), + request_body = EmployeeLoginRequest, + responses((status = 200, description = "Login response", body = EmployeeLoginResponse)) +)] +pub async fn fleet_operator_employee_login( + app_state: Data, + path: Path, + body: Json, +) -> AppResult { + let gtfs_id = path.into_inner(); + let req = body.into_inner(); + + let response = app_state + .fleet_operator_service + .login(>fs_id, &req) + .await?; + + Ok(HttpResponse::Ok().json(response)) +} + +#[utoipa::path( + post, + path = "/internal/fleet-operator/{gtfs_id}/employee/register", + tag = "Fleet Operator", + params(("gtfs_id" = String, Path, description = "GTFS dataset identifier")), + request_body = EmployeeRegisterRequest, + responses((status = 200, description = "Registration response", body = EmployeeRegisterResponse)) +)] +pub async fn fleet_operator_employee_register( + app_state: Data, + path: Path, + body: Json, +) -> AppResult { + let gtfs_id = path.into_inner(); + let req = body.into_inner(); + + let response = app_state + .fleet_operator_service + .register(>fs_id, &req) + .await?; + + Ok(HttpResponse::Ok().json(response)) +} diff --git a/src/main.rs b/src/main.rs index 348bfee..d5503b3 100644 --- a/src/main.rs +++ b/src/main.rs @@ -44,6 +44,13 @@ async fn main() -> anyhow::Result<()> { chalo_cache_clone.start_background_update_task().await; }); + // Start background task to periodically refresh OSRTC station list + if let Some(osrtc_cache) = app_state.osrtc_cache.clone() { + tokio::spawn(async move { + osrtc_cache.start_background_refresh_task().await; + }); + } + let prometheus = prometheus_metrics(); let openapi = ApiDoc::openapi(); diff --git a/src/models.rs b/src/models.rs index 516180e..9329433 100644 --- a/src/models.rs +++ b/src/models.rs @@ -267,6 +267,30 @@ pub struct VehicleMetadataResponse { pub is_actually_valid: Option, } +#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)] +pub struct WaybillMetadataResponse { + pub waybill_no: String, + pub vehicle_no: String, + #[serde(rename = "serviceType")] + pub service_type: String, + pub driver_id: Option, + #[serde(rename = "driverName")] + pub driver_name: Option, + #[serde(rename = "driverMobileNumber")] + pub driver_mobile_number: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, sqlx::FromRow)] +pub struct WaybillTripInfo { + pub waybill_no: String, + pub vehicle_no: String, + pub service_type: String, + pub driver_token_no: Option, + pub driver_first_name: Option, + pub driver_last_name: Option, + pub driver_mobile_number: Option, +} + #[derive(Debug, Clone, PartialEq, Serialize, Deserialize, ToSchema)] #[serde(rename_all = "lowercase")] pub enum WaybillStatus { @@ -424,6 +448,10 @@ 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 +478,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/db_vehicle_reader.rs b/src/services/db_vehicle_reader.rs index f0d5c88..cb682a0 100644 --- a/src/services/db_vehicle_reader.rs +++ b/src/services/db_vehicle_reader.rs @@ -527,6 +527,7 @@ impl DBVehicleReader { w.conductor_token_no::text AS conductor_code, w.deleted AS deleted, w.status AS status, + w.is_flexi as is_flexi FROM waybills w LEFT JOIN entities e ON e.entity_id = w.entity_id diff --git a/src/services/db_vehicle_reader_internal.rs b/src/services/db_vehicle_reader_internal.rs index 7942f4d..c6aa763 100644 --- a/src/services/db_vehicle_reader_internal.rs +++ b/src/services/db_vehicle_reader_internal.rs @@ -7,7 +7,10 @@ use tokio::sync::RwLock; use sqlx::PgPool; use tracing::{debug, error, info}; -use crate::models::{BusSchedule, VehicleData, VehicleDataWithRouteId, WaybillStatus}; +use crate::models::{ + BusSchedule, VehicleData, VehicleDataWithRouteId, WaybillMetadataResponse, WaybillStatus, + WaybillTripInfo, +}; use crate::tools::error::{AppError, AppResult}; #[async_trait] @@ -41,6 +44,12 @@ pub trait VehicleDataReaderInternal: Send + Sync { destination_station_code: &str, eta_in_seconds: i32, ) -> AppResult<()>; + + async fn get_waybill_metadata( + &self, + gtfs_id: &str, + waybill_no: &str, + ) -> AppResult; } // Mock implementation for local testing without a database @@ -106,6 +115,16 @@ impl VehicleDataReaderInternal for MockDBVehicleReaderInternal { ) -> AppResult<()> { Ok(()) } + + async fn get_waybill_metadata( + &self, + _gtfs_id: &str, + _waybill_no: &str, + ) -> AppResult { + Err(AppError::NotFound( + "Database is not connected in local testing mode.".to_string(), + )) + } } #[allow(clippy::type_complexity)] @@ -1287,4 +1306,78 @@ impl VehicleDataReaderInternal for DBVehicleReaderInternal { ) .await } + + async fn get_waybill_metadata( + &self, + gtfs_id: &str, + waybill_no: &str, + ) -> AppResult { + self.get_waybill_metadata_impl(gtfs_id, waybill_no).await + } +} + +impl DBVehicleReaderInternal { + async fn get_waybill_metadata_impl( + &self, + gtfs_id: &str, + waybill_no: &str, + ) -> AppResult { + let pool = self.pool()?; + + let waybill_query = r#" + SELECT + w.waybill_no::text, + w.vehicle_no, + w.service_type, + w.driver_token_no::text AS driver_token_no, + d.first_name AS driver_first_name, + d.last_name AS driver_last_name, + d.mobile_no AS driver_mobile_number + FROM waybills_internal w + LEFT JOIN employees_internal d + ON d.token_no = w.driver_token_no::text AND d.gtfs_id = $2 AND d.deleted = false + WHERE w.waybill_no = $1 + AND w.gtfs_id = $2 + AND w.deleted = false + ORDER BY w.updated_at DESC + LIMIT 1 + "#; + + let waybill_row = sqlx::query_as::<_, WaybillTripInfo>(waybill_query) + .bind(waybill_no) + .bind(gtfs_id) + .fetch_optional(pool) + .await + .map_err(|e| AppError::DbError(format!("Waybill query failed: {}", e)))? + .ok_or_else(|| AppError::NotFound(format!("Waybill not found: {}", waybill_no)))?; + + // Build driver name + let driver_name = match (waybill_row.driver_first_name, waybill_row.driver_last_name) { + (Some(f), Some(l)) => { + let combined = format!("{} {}", f, l).trim().to_string(); + if combined.is_empty() { + None + } else { + Some(combined) + } + } + (Some(f), None) if !f.trim().is_empty() => Some(f), + (None, Some(l)) if !l.trim().is_empty() => Some(l), + _ => None, + }; + let driver_mobile_number = waybill_row + .driver_mobile_number + .filter(|m| !m.trim().is_empty()); + + let response = WaybillMetadataResponse { + waybill_no: waybill_row.waybill_no, + vehicle_no: waybill_row.vehicle_no, + service_type: waybill_row.service_type, + driver_id: waybill_row.driver_token_no, + driver_name, + driver_mobile_number, + }; + + Ok(response) + } } diff --git a/src/services/fleet_operator.rs b/src/services/fleet_operator.rs index 24c2740..76012d5 100644 --- a/src/services/fleet_operator.rs +++ b/src/services/fleet_operator.rs @@ -1,5 +1,5 @@ use async_trait::async_trait; -use serde::Serialize; +use serde::{Deserialize, Serialize}; use sqlx::PgPool; use std::collections::HashMap; use std::sync::Arc; @@ -51,6 +51,8 @@ pub struct CurrentOperationResponse { pub conductor_token: Option, pub driver_token: Option, pub number_of_trips: i64, + pub driver_person_id: Option, + pub conductor_person_id: Option, } #[derive(Debug, Serialize)] @@ -86,6 +88,47 @@ pub struct CurrentTripDetailsResponse { pub upcoming: Vec, } +#[derive(Debug, Clone, Deserialize, Serialize, utoipa::ToSchema)] +pub enum AuthType { + Email, +} + +#[derive(Debug, Clone, Copy, Deserialize, Serialize, utoipa::ToSchema)] +#[serde(rename_all = "lowercase")] +pub enum Role { + Driver, + Conductor, +} + +#[derive(Debug, Clone, Deserialize, Serialize, utoipa::ToSchema)] +pub struct EmployeeLoginRequest { + pub auth_type: Option, + pub email_hash: Option, + pub password_hash: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, utoipa::ToSchema)] +pub struct EmployeeLoginResponse { + pub verified: bool, + pub token: Option, + pub role: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, utoipa::ToSchema)] +pub struct EmployeeRegisterRequest { + pub token_no: String, + pub email_hash: String, + pub password_hash: String, + pub first_name: String, + pub role: Option, +} + +#[derive(Debug, Clone, Deserialize, Serialize, utoipa::ToSchema)] +pub struct EmployeeRegisterResponse { + pub success: bool, + pub token_no: String, +} + // ─── Service trait ───────────────────────────────────────────────────────────── #[async_trait] @@ -125,6 +168,18 @@ pub trait FleetOperatorService: Send + Sync { token: &str, device_serial_no: &str, ) -> AppResult; + + async fn login( + &self, + gtfs_id: &str, + req: &EmployeeLoginRequest, + ) -> AppResult; + + async fn register( + &self, + gtfs_id: &str, + req: &EmployeeRegisterRequest, + ) -> AppResult; } // ─── Mock implementation ─────────────────────────────────────────────────────── @@ -200,6 +255,26 @@ impl FleetOperatorService for MockFleetOperatorService { "Database is not connected in local testing mode.".to_string(), )) } + + async fn login( + &self, + _gtfs_id: &str, + _req: &EmployeeLoginRequest, + ) -> AppResult { + Err(AppError::NotFound( + "Database is not connected in local testing mode.".to_string(), + )) + } + + async fn register( + &self, + _gtfs_id: &str, + _req: &EmployeeRegisterRequest, + ) -> AppResult { + Err(AppError::NotFound( + "Database is not connected in local testing mode.".to_string(), + )) + } } // ─── DB implementation ───────────────────────────────────────────────────────── @@ -219,6 +294,30 @@ impl DBFleetOperatorService { } } + async fn emp_id_for_token(&self, gtfs_id: &str, token: &str) -> AppResult> { + sqlx::query_scalar::<_, i64>( + r#" + SELECT emp_id + FROM employees_internal + WHERE gtfs_id = $1 + AND token_no = $2 + AND deleted = false + LIMIT 1 + "#, + ) + .bind(gtfs_id) + .bind(token) + .fetch_optional(&self.pool) + .await + .map_err(|e| { + error!( + "emp_id lookup failed for gtfs_id={}, token_no={}: {}", + gtfs_id, token, e + ); + AppError::Internal(e.to_string()) + }) + } + // ── Waybill resolution ─────────────────────────────────────────────────── async fn resolve_waybill( @@ -822,12 +921,23 @@ impl FleetOperatorService for DBFleetOperatorService { let waybill = self.resolve_waybill(gtfs_id, &anchor).await?; let number_of_trips = self.get_trip_count(&waybill).await?; + let driver_person_id = match waybill.driver_token_no.as_deref() { + Some(t) if !t.is_empty() => self.emp_id_for_token(gtfs_id, t).await?, + _ => None, + }; + let conductor_person_id = match waybill.conductor_token_no.as_deref() { + Some(t) if !t.is_empty() => self.emp_id_for_token(gtfs_id, t).await?, + _ => None, + }; + Ok(CurrentOperationResponse { waybill_no: waybill.waybill_no, vehicle_number: waybill.vehicle_no, conductor_token: waybill.conductor_token_no, driver_token: waybill.driver_token_no, number_of_trips, + driver_person_id, + conductor_person_id, }) } @@ -1029,4 +1139,167 @@ impl FleetOperatorService for DBFleetOperatorService { Ok(VerifyResponse { verified: etm_ok }) } + + async fn login( + &self, + gtfs_id: &str, + req: &EmployeeLoginRequest, + ) -> AppResult { + match req.auth_type { + Some(AuthType::Email) => { + let email_hash = req.email_hash.as_ref().ok_or_else(|| { + AppError::BadRequest("email_hash is required for email auth".into()) + })?; + let password_hash = req.password_hash.as_ref().ok_or_else(|| { + AppError::BadRequest("password_hash is required for email auth".into()) + })?; + + let row: Option<(String, Option)> = sqlx::query_as( + r#" + SELECT e.token_no, LOWER(d.designation_name) + FROM employees_internal e + LEFT JOIN designations_internal d + ON d.designation_id = e.designation_id + AND d.gtfs_id = e.gtfs_id + AND d.deleted = false + WHERE e.email_hash = $1 + AND e.password_hash = $2 + AND e.gtfs_id = $3 + AND e.deleted = false + LIMIT 1 + "#, + ) + .bind(email_hash) + .bind(password_hash) + .bind(gtfs_id) + .fetch_optional(&self.pool) + .await + .map_err(|e| { + error!("login failed for gtfs_id={}: {}", gtfs_id, e); + AppError::Internal(e.to_string()) + })?; + + match row { + Some((token, designation_name)) => { + let role = designation_name.as_deref().and_then(|n| match n { + "driver" => Some(Role::Driver), + "conductor" => Some(Role::Conductor), + _ => None, + }); + Ok(EmployeeLoginResponse { + verified: true, + token: Some(token), + role, + }) + } + None => Ok(EmployeeLoginResponse { + verified: false, + token: None, + role: None, + }), + } + } + _ => Ok(EmployeeLoginResponse { + verified: false, + token: None, + role: None, + }), + } + } + + async fn register( + &self, + gtfs_id: &str, + req: &EmployeeRegisterRequest, + ) -> AppResult { + let designation_id: Option = match req.role { + Some(role) => { + let name = match role { + Role::Driver => "driver", + Role::Conductor => "conductor", + }; + let id: Option = sqlx::query_scalar( + r#" + SELECT designation_id + FROM designations_internal + WHERE LOWER(designation_name) = $1 + AND gtfs_id = $2 + AND deleted = false + LIMIT 1 + "#, + ) + .bind(name) + .bind(gtfs_id) + .fetch_optional(&self.pool) + .await + .map_err(|e| { + error!( + "register designation lookup failed for gtfs_id={}, role={}: {}", + gtfs_id, name, e + ); + AppError::Internal(e.to_string()) + })?; + Some(id.ok_or_else(|| { + AppError::NotFound(format!( + "designation '{}' not found for gtfs_id={}", + name, gtfs_id + )) + })?) + } + None => None, + }; + + if let Some(did) = designation_id { + sqlx::query( + r#" + INSERT INTO employees_internal (token_no, email_hash, password_hash, gtfs_id, first_name, designation_id) + VALUES ($1, $2, $3, $4, $5, $6) + ON CONFLICT (gtfs_id, token_no) DO UPDATE SET + email_hash = EXCLUDED.email_hash, + password_hash = EXCLUDED.password_hash, + designation_id = EXCLUDED.designation_id, + updated_at = NOW() + "#, + ) + .bind(&req.token_no) + .bind(&req.email_hash) + .bind(&req.password_hash) + .bind(gtfs_id) + .bind(&req.first_name) + .bind(did) + .execute(&self.pool) + .await + .map_err(|e| { + error!("register upsert failed for gtfs_id={}, token_no={}: {}", gtfs_id, req.token_no, e); + AppError::Internal(e.to_string()) + })?; + } else { + sqlx::query( + r#" + INSERT INTO employees_internal (token_no, email_hash, password_hash, gtfs_id, first_name) + VALUES ($1, $2, $3, $4, $5) + ON CONFLICT (gtfs_id, token_no) DO UPDATE SET + email_hash = EXCLUDED.email_hash, + password_hash = EXCLUDED.password_hash, + updated_at = NOW() + "#, + ) + .bind(&req.token_no) + .bind(&req.email_hash) + .bind(&req.password_hash) + .bind(gtfs_id) + .bind(&req.first_name) + .execute(&self.pool) + .await + .map_err(|e| { + error!("register upsert failed for gtfs_id={}, token_no={}: {}", gtfs_id, req.token_no, e); + AppError::Internal(e.to_string()) + })?; + } + + Ok(EmployeeRegisterResponse { + success: true, + token_no: req.token_no.clone(), + }) + } } diff --git a/src/services/gtfs_service.rs b/src/services/gtfs_service.rs index cd9ffa8..ad42700 100644 --- a/src/services/gtfs_service.rs +++ b/src/services/gtfs_service.rs @@ -36,6 +36,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() @@ -746,6 +761,7 @@ impl GTFSService { stop_regional_names_by_gtfs: &HashMap>, ) -> HashMap { let mut stops_by_gtfs: HashMap = HashMap::new(); + let mut by_cluster_set: HashMap>> = HashMap::new(); for stop in stops { let parts: Vec<&str> = stop.id.split(':').collect(); @@ -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,60 @@ 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: None, }; stop_data .stops .insert(stop.cluster.clone().unwrap(), cluster_stop_res); } + if let Some(cid) = &cluster_id { + by_cluster_set + .entry(gtfs_id.to_string()) + .or_default() + .entry(cid.clone()) + .or_default() + .insert(stop_code.to_string()); + } + stop_data.stops.insert(stop_code.to_string(), stop_res); } + for (gtfs_id, per_cluster) in by_cluster_set { + if let Some(stop_data) = stops_by_gtfs.get_mut(>fs_id) { + for (cid, codes) in per_cluster { + let mut v: Vec = codes.into_iter().collect(); + v.sort(); + stop_data.by_cluster_id.insert(cid, v); + } + } + } + + 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 +1507,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 => { + debug!( + 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/services/mod.rs b/src/services/mod.rs index 77e0b4c..3fa855e 100644 --- a/src/services/mod.rs +++ b/src/services/mod.rs @@ -6,4 +6,5 @@ pub mod field_generator; pub mod fleet_operator; pub mod gtfs_service; pub mod operator; +pub mod osrtc_station_cache; pub mod trip_service; diff --git a/src/services/osrtc_station_cache.rs b/src/services/osrtc_station_cache.rs new file mode 100644 index 0000000..8799cd5 --- /dev/null +++ b/src/services/osrtc_station_cache.rs @@ -0,0 +1,294 @@ +use crate::models::{LatLong, RouteStopMapping}; +use crate::tools::error::{AppError, AppResult}; +use chrono::{DateTime, Duration, Utc}; +use reqwest::Client; +use serde::Deserialize; +use std::collections::HashMap; +use std::sync::Arc; +use tokio::sync::{Mutex, RwLock}; +use tracing::{error, info}; + +// --- API response types (private) --- + +#[derive(Deserialize)] +struct GenerateAccessResponse { + data: Vec, + issuccess: bool, +} + +#[derive(Deserialize)] +struct AccessTokenData { + #[serde(rename = "strAccessToken")] + access_token: String, + #[serde(rename = "dteAccessTokenExpirationTime")] + expires_at: DateTime, +} + +#[derive(Deserialize)] +struct GetStationListResponse { + data: Vec, + issuccess: bool, +} + +#[derive(Deserialize)] +struct OsrtcApiStation { + #[serde(rename = "intStationID")] + station_id: i64, + #[serde(rename = "strStationName")] + station_name: String, + #[serde(rename = "strStationCode")] + station_code: String, +} + +// --- Internal token cache --- + +struct OsrtcToken { + access_token: String, + expires_at: DateTime, +} + +// --- Public types --- + +#[derive(Clone)] +pub struct OsrtcStation { + pub station_id: i64, + pub station_name: String, + pub station_code: String, // strStationCode, kept for observability. Not actually used by the frontend. +} + +pub struct OsrtcStationCache { + http_client: Client, + base_url: String, + username: String, + secret_key: String, + // Mutex so check-and-refresh is atomic — prevents concurrent callers from all hitting the auth endpoint on expiry + token: Arc>>, + // Keyed by intStationID.to_string() — the identifier used by frontend/backend + stations: Arc>>, + // Pre-built snapshot; get_all_stations is a single Arc::clone instead of a full HashMap copy + station_list: Arc>>>, + station_refresh_interval_secs: u64, +} + +impl OsrtcStationCache { + pub fn new( + base_url: String, + username: String, + secret_key: String, + station_refresh_interval_secs: u64, + ) -> AppResult { + let http_client = Client::builder() + .timeout(std::time::Duration::from_secs(30)) + .build() + .map_err(|e| AppError::Internal(format!("Failed to create HTTP client: {}", e)))?; + + Ok(Self { + http_client, + base_url, + username, + secret_key, + token: Arc::new(Mutex::new(None)), + stations: Arc::new(RwLock::new(HashMap::new())), + station_list: Arc::new(RwLock::new(Arc::new(Vec::new()))), + station_refresh_interval_secs, + }) + } + + async fn fetch_fresh_token(&self) -> AppResult { + info!("Refreshing OSRTC auth token..."); + let url = format!("{}/api/Auth/GenerateAccess", self.base_url); + let body = serde_json::json!({ + "UserName": self.username, + "SecretKey": self.secret_key, + }); + + let response = self + .http_client + .post(&url) + .header("content-type", "application/json") + .json(&body) + .send() + .await + .map_err(|e| { + AppError::Internal(format!("OSRTC GenerateAccess request failed: {}", e)) + })?; + + if !response.status().is_success() { + return Err(AppError::Internal(format!( + "OSRTC GenerateAccess returned status: {}", + response.status() + ))); + } + + let parsed: GenerateAccessResponse = response.json().await.map_err(|e| { + AppError::Internal(format!( + "Failed to parse OSRTC GenerateAccess response: {}", + e + )) + })?; + + if !parsed.issuccess { + return Err(AppError::Internal( + "OSRTC GenerateAccess returned issuccess=false".to_string(), + )); + } + + parsed + .data + .into_iter() + .next() + .map(|td| OsrtcToken { + access_token: td.access_token, + expires_at: td.expires_at, + }) + .ok_or_else(|| { + AppError::Internal("OSRTC GenerateAccess returned empty data array".to_string()) + }) + } + + // Lazy refresh: returns cached token if it has >5 min left, otherwise re-fetches. + // Holds the mutex across the HTTP call so only one caller refreshes at a time. + async fn get_valid_token(&self) -> AppResult { + let mut token_guard = self.token.lock().await; + if let Some(t) = &*token_guard { + if t.expires_at > Utc::now() + Duration::minutes(5) { + return Ok(t.access_token.clone()); + } + } + let new_token = self.fetch_fresh_token().await?; + let access_token = new_token.access_token.clone(); + *token_guard = Some(new_token); + info!("OSRTC auth token refreshed successfully"); + info!("OSRTC access token: {}", access_token); + Ok(access_token) + } + + async fn refresh_stations(&self) -> AppResult<()> { + info!("Refreshing OSRTC station list..."); + let token = self.get_valid_token().await?; + let url = format!("{}/api/List/GetStationList", self.base_url); + + let response = self + .http_client + .post(&url) + .header("Authorization", format!("Bearer {}", token)) + .json(&serde_json::json!({})) + .send() + .await + .map_err(|e| { + AppError::Internal(format!("OSRTC GetStationList request failed: {}", e)) + })?; + + if !response.status().is_success() { + let status = response.status(); + let body = response.text().await.unwrap_or_default(); + return Err(AppError::Internal(format!( + "OSRTC GetStationList returned status: {} body: {}", + status, body + ))); + } + + let parsed: GetStationListResponse = response.json().await.map_err(|e| { + AppError::Internal(format!( + "Failed to parse OSRTC GetStationList response: {}", + e + )) + })?; + + if !parsed.issuccess { + return Err(AppError::Internal( + "OSRTC GetStationList returned issuccess=false".to_string(), + )); + } + + let mut new_stations: HashMap = + HashMap::with_capacity(parsed.data.len()); + for api_station in parsed.data { + new_stations.insert( + api_station.station_id.to_string(), + OsrtcStation { + station_id: api_station.station_id, + station_name: api_station.station_name, + station_code: api_station.station_code, + }, + ); + } + + let count = new_stations.len(); + let list: Arc> = Arc::new(new_stations.values().cloned().collect()); + + let mut stations_lock = self.stations.write().await; + *stations_lock = new_stations; + drop(stations_lock); + + let mut list_lock = self.station_list.write().await; + *list_lock = list; + + info!("OSRTC station list refreshed: {} stations loaded", count); + Ok(()) + } + + pub async fn initialize(&self) -> AppResult<()> { + info!("Initializing OSRTC station cache..."); + self.refresh_stations().await?; + info!("OSRTC station cache initialized successfully"); + Ok(()) + } + + pub async fn get_station_by_id(&self, id: &str) -> Option { + if self.stations.read().await.is_empty() { + if let Err(e) = self.refresh_stations().await { + error!("OSRTC on-demand station refresh failed: {}", e); + } + } + self.stations.read().await.get(id).cloned() + } + + pub async fn get_all_stations(&self) -> Arc> { + if self.station_list.read().await.is_empty() { + if let Err(e) = self.refresh_stations().await { + error!("OSRTC on-demand station refresh failed: {}", e); + } + } + self.station_list.read().await.clone() + } + + pub async fn start_background_refresh_task(self: Arc) { + let interval_secs = self.station_refresh_interval_secs; + info!( + "Starting OSRTC station cache background refresh task with interval: {}s", + interval_secs + ); + let duration = std::time::Duration::from_secs(interval_secs); + let mut interval = tokio::time::interval(duration); + interval.tick().await; // discard the immediate first tick + loop { + interval.tick().await; + if let Err(e) = self.refresh_stations().await { + error!("Failed to refresh OSRTC station cache: {}", e); + } + } + } +} + +// --- Conversion helper --- + +pub fn osrtc_station_to_route_stop_mapping(station: &OsrtcStation) -> RouteStopMapping { + let id_str: Arc = station.station_id.to_string().into(); + RouteStopMapping { + stop_code: id_str.clone(), + provider_code: id_str, + stop_name: station.station_name.as_str().into(), + stop_point: LatLong { lat: 0.0, lon: 0.0 }, // Doesn't matter, and won't be consumed. Better to do this than object structure + route_code: Arc::from(""), + sequence_num: 0, + vehicle_type: Arc::from("BUS"), + estimated_travel_time_from_previous_stop: None, + geo_json: None, + gates: None, + hindi_name: None, + regional_name: None, + platform: None, + parent_stop_code: None, + } +} 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"),