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/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 f27a071..8ffcc57 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,29 @@ 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 +347,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?), @@ -379,10 +410,7 @@ impl AppState { } } - info!( - "Loaded fleet tag list for {} GTFS feeds", - fleet_map.len() - ); + info!("Loaded fleet tag list for {} GTFS feeds", fleet_map.len()); Ok(fleet_map) } diff --git a/src/handlers/routes.rs b/src/handlers/routes.rs index 8942195..be74d66 100644 --- a/src/handlers/routes.rs +++ b/src/handlers/routes.rs @@ -23,6 +23,7 @@ use crate::models::{ StopCodeFromProviderStopCodeResponse, VehicleMetadataResponse, 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::{ @@ -578,6 +579,18 @@ async fn get_routes_fuzzy( 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)) } @@ -620,6 +633,17 @@ 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) @@ -2901,7 +2925,9 @@ async fn fleet_operator_trip_action( let trip_number = match action { TripAction::Reset => 0, _ => req.trip_number.ok_or_else(|| { - AppError::BadRequest("trip_number is required for 'start' and 'end' actions.".to_string()) + AppError::BadRequest( + "trip_number is required for 'start' and 'end' actions.".to_string(), + ) })?, }; diff --git a/src/main.rs b/src/main.rs index f2cb6f7..7524c8d 100644 --- a/src/main.rs +++ b/src/main.rs @@ -41,6 +41,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(); // Create and run the web server with performance optimizations diff --git a/src/services/chalo_vehicle_cache.rs b/src/services/chalo_vehicle_cache.rs index e6b0e44..5cc5500 100644 --- a/src/services/chalo_vehicle_cache.rs +++ b/src/services/chalo_vehicle_cache.rs @@ -2,8 +2,8 @@ use crate::services::gtfs_service::GTFSService; use crate::tools::error::{AppError, AppResult}; use chrono::{DateTime, Utc}; use csv::ReaderBuilder; -use reqwest::Client; use once_cell::sync::Lazy; +use reqwest::Client; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; use std::sync::Arc; @@ -95,9 +95,8 @@ pub const CHALO_CITY_DEFS: [ChaloCityDef; 2] = [ /// Membership for [`CHALO_CITY_DEFS`], built once per process. The set only changes with a deploy /// (same as edits to [`CHALO_CITY_DEFS`]), so no runtime invalidation. -static CHALO_GTFS_ID_SET: Lazy> = Lazy::new(|| { - CHALO_CITY_DEFS.iter().map(|c| c.gtfs_id).collect() -}); +static CHALO_GTFS_ID_SET: Lazy> = + Lazy::new(|| CHALO_CITY_DEFS.iter().map(|c| c.gtfs_id).collect()); /// Same order as [`CHALO_CITY_DEFS`]; built once per process (see [`CHALO_GTFS_ID_SET`]). static CHALO_GTFS_IDS_SLICE: Lazy> = Lazy::new(|| { diff --git a/src/services/fleet_operator.rs b/src/services/fleet_operator.rs index 5b44b0b..24c2740 100644 --- a/src/services/fleet_operator.rs +++ b/src/services/fleet_operator.rs @@ -843,7 +843,8 @@ impl FleetOperatorService for DBFleetOperatorService { if action != TripAction::Reset { self.validate_trip_exists(&waybill, trip_number).await?; } - self.apply_trip_action(&waybill, &action, trip_number, timestamp).await?; + self.apply_trip_action(&waybill, &action, trip_number, timestamp) + .await?; Ok(TripActionResponse { success: true }) } 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..8a4fc0f --- /dev/null +++ b/src/services/osrtc_station_cache.rs @@ -0,0 +1,295 @@ +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, + } +}