//! REST API server for recording management and live telemetry access. //! //! Runs as a parallel tokio task alongside the UDP listeners. use axum::{ Json, Router, extract::{Path, State}, http::StatusCode, response::{IntoResponse, Response}, routing::{get, post}, }; use serde::{Deserialize, Serialize}; use std::sync::{Arc, Mutex, MutexGuard}; use tower_http::cors::CorsLayer; use crate::analysis; use crate::feed_state::FeedState; use crate::recording::{Game, RecordingManager, RecordingStatus, StartRecordingRequest, UseCase}; use crate::session::SessionManager; use crate::ui::FeedDisplayInfo; pub struct AppState { pub recording_manager: Arc, pub feed_displays: Vec<(FeedDisplayInfo, Arc>)>, pub session_manager: Arc>, } #[derive(Serialize)] struct ApiError { error: String, } fn ok_json(data: T) -> Response { (StatusCode::OK, Json(data)).into_response() } fn err(status: StatusCode, msg: &str) -> Response { ( status, Json(ApiError { error: msg.to_string(), }), ) .into_response() } fn lock_sessions(state: &Arc) -> MutexGuard<'_, SessionManager> { state.session_manager.lock().unwrap_or_else(|e| { eprintln!("[WARN] SessionManager mutex poisoned, recovering: {}", e); e.into_inner() }) } #[derive(Serialize)] struct LiveFeedSnapshot { label: String, port: u16, health: String, frequency_hz: f64, total_packets: u64, data_rate_kb_s: f64, speed_kmh: Option, gear: Option, engine_rpm: Option, throttle: Option, brake: Option, lap_time: Option, spline_position: Option, } #[derive(Serialize)] struct LiveTelemetryResponse { feeds: Vec, active_recording: Option, } #[derive(Deserialize)] struct CreateSessionRequest { use_case: UseCase, label: Option, } #[derive(Deserialize)] struct StartRunRequest { feed_name: String, label: Option, } async fn create_session( State(state): State>, Json(req): Json, ) -> Response { let mut mgr = lock_sessions(&state); match mgr.create_session(req.use_case, req.label) { Ok(session) => (StatusCode::CREATED, Json(session)).into_response(), Err(msg) => err(StatusCode::BAD_REQUEST, &msg), } } async fn list_sessions(State(state): State>) -> Response { let mut mgr = lock_sessions(&state); ok_json(mgr.list_sessions()) } async fn get_session( State(state): State>, Path(id): Path, ) -> Response { let mgr = lock_sessions(&state); match mgr.get_session(&id) { Some(session) => ok_json(session), None => err(StatusCode::NOT_FOUND, "Session not found"), } } async fn delete_session( State(state): State>, Path(id): Path, ) -> Response { let mut mgr = lock_sessions(&state); match mgr.delete_session(&id) { Ok(()) => StatusCode::NO_CONTENT.into_response(), Err(msg) => err(StatusCode::NOT_FOUND, &msg), } } async fn start_run( State(state): State>, Path(id): Path, Json(req): Json, ) -> Response { let mut mgr = lock_sessions(&state); match mgr.start_run(&id, &req.feed_name, req.label) { Ok(run) => (StatusCode::CREATED, Json(run)).into_response(), Err(msg) => err(StatusCode::BAD_REQUEST, &msg), } } async fn stop_run( State(state): State>, Path((id, run_id)): Path<(String, String)>, ) -> Response { let mut mgr = lock_sessions(&state); match mgr.stop_run(&id, &run_id) { Ok(run) => ok_json(run), Err(msg) => err(StatusCode::BAD_REQUEST, &msg), } } async fn complete_session( State(state): State>, Path(id): Path, ) -> Response { let (use_case, frames_by_run, run_labels) = { let mut mgr = lock_sessions(&state); match mgr.complete_session(&id) { Ok(_session) => {} Err(msg) => return err(StatusCode::BAD_REQUEST, &msg), } match mgr.collect_run_frames(&id) { Some((uc, frames, labels)) => (uc, frames, labels), None => return err(StatusCode::NOT_FOUND, "Session not found after completion"), } }; let result = analysis::analyze(&use_case, &frames_by_run, &run_labels); { let mut mgr = lock_sessions(&state); let _ = mgr.set_analysis(&id, result); } let mgr = lock_sessions(&state); match mgr.get_session(&id) { Some(session) => ok_json(session), None => err(StatusCode::INTERNAL_SERVER_ERROR, "Session lost after analysis"), } } async fn get_analysis( State(state): State>, Path(id): Path, ) -> Response { let mgr = lock_sessions(&state); match mgr.get_session(&id) { Some(session) => { match session.analysis { Some(ref analysis) => ok_json(analysis), None => err(StatusCode::NOT_FOUND, "Analysis not available yet. Complete the session first."), } } None => err(StatusCode::NOT_FOUND, "Session not found"), } } async fn run_analysis( State(state): State>, Path(id): Path, ) -> Response { let (use_case, frames_by_run, run_labels) = { let mgr = lock_sessions(&state); match mgr.get_session(&id) { Some(session) => { if session.status != crate::session::SessionStatus::Completed { return err(StatusCode::BAD_REQUEST, "Session must be completed before analysis"); } } None => return err(StatusCode::NOT_FOUND, "Session not found"), } match mgr.collect_run_frames(&id) { Some(data) => data, None => return err(StatusCode::NOT_FOUND, "Session not found"), } }; let result = analysis::analyze(&use_case, &frames_by_run, &run_labels); { let mut mgr = lock_sessions(&state); let _ = mgr.set_analysis(&id, result.clone()); } ok_json(result) } async fn list_recordings(State(state): State>) -> Response { match state.recording_manager.list_recordings().await { Ok(recordings) => ok_json(recordings), Err(e) => err( StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to list recordings: {}", e), ), } } async fn get_recording( State(state): State>, Path(id): Path, ) -> Response { match state.recording_manager.get_recording(&id).await { Ok(manifest) => ok_json(manifest), Err(e) if e.kind() == std::io::ErrorKind::NotFound => { err(StatusCode::NOT_FOUND, "Recording not found") } Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => { err(StatusCode::BAD_REQUEST, &e.to_string()) } Err(e) => err( StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to get recording: {}", e), ), } } async fn start_recording( State(state): State>, Json(req): Json, ) -> Response { match state.recording_manager.start_recording(req).await { Ok(manifest) => (StatusCode::CREATED, Json(manifest)).into_response(), Err(e) => err( StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to start recording: {}", e), ), } } async fn stop_recording(State(state): State>) -> Response { match state.recording_manager.stop_recording().await { Ok(Some(manifest)) => ok_json(manifest), Ok(None) => err(StatusCode::BAD_REQUEST, "No active recording to stop"), Err(e) => err( StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to stop recording: {}", e), ), } } async fn stop_specific_recording( State(state): State>, Path(id): Path, ) -> Response { let manifest = match state.recording_manager.get_recording(&id).await { Ok(m) => m, Err(e) if e.kind() == std::io::ErrorKind::NotFound => { return err(StatusCode::NOT_FOUND, "Recording not found") } Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => { return err(StatusCode::BAD_REQUEST, &e.to_string()) } Err(e) => { return err( StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to get recording: {}", e), ) } }; if manifest.status != RecordingStatus::Recording { return err(StatusCode::CONFLICT, "Recording is not active"); } match state .recording_manager .stop_recording_internal(&id) .await { Ok(stopped) => { state .recording_manager .clear_active_if_matches(&id) .await; ok_json(stopped) } Err(e) => err( StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to stop recording: {}", e), ), } } async fn delete_recording( State(state): State>, Path(id): Path, ) -> Response { match state.recording_manager.delete_recording(&id).await { Ok(()) => StatusCode::NO_CONTENT.into_response(), Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => { err(StatusCode::CONFLICT, &e.to_string()) } Err(e) if e.kind() == std::io::ErrorKind::NotFound => { err(StatusCode::NOT_FOUND, "Recording not found") } Err(e) => err( StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to delete recording: {}", e), ), } } async fn get_feed_data( State(state): State>, Path((id, feed_name)): Path<(String, String)>, ) -> Response { let active_id = state.recording_manager.active_recording_id().await; if active_id.as_deref() == Some(id.as_str()) { match state.recording_manager.read_frames(&feed_name) { Some(frames) => return ok_json(frames), None => { return err( StatusCode::NOT_FOUND, &format!("Feed '{}' not found in active recording", feed_name), ) } } } let manifest = match state.recording_manager.get_recording(&id).await { Ok(m) => m, Err(e) if e.kind() == std::io::ErrorKind::NotFound => { return err(StatusCode::NOT_FOUND, "Recording not found") } Err(e) if e.kind() == std::io::ErrorKind::InvalidInput => { return err(StatusCode::BAD_REQUEST, &e.to_string()) } Err(e) => { return err( StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to get recording: {}", e), ) } }; let feed_stats = match manifest.feeds.iter().find(|f| f.name == feed_name) { Some(f) => f, None => { return err( StatusCode::NOT_FOUND, &format!("Feed '{}' not found in recording", feed_name), ) } }; let file_path = manifest.dir_path(state.recording_manager.data_dir()).join(&feed_stats.file); if !file_path.exists() { return err(StatusCode::NOT_FOUND, "Feed data file not found on disk (recording may still be in progress)"); } match tokio::fs::read_to_string(&file_path).await { Ok(content) => { let frames: Vec = content .lines() .filter(|l| !l.is_empty()) .filter_map(|l| serde_json::from_str(l).ok()) .collect(); ok_json(frames) } Err(e) => err( StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to read feed data: {}", e), ), } } async fn get_live(State(state): State>) -> Response { let mut feeds = Vec::new(); for (info, state_arc) in &state.feed_displays { let snapshot = state_arc.lock().ok().map(|mut s| { s.tick(); s.clone() }); let snap = match snapshot { Some(s) => s, None => continue, }; let (speed_kmh, gear, engine_rpm, throttle, brake, lap_time, spline_position) = if let Some(ref frame) = snap.last_frame { ( Some(frame.speed_kmh), Some(frame.gear_display()), Some(frame.engine_rpm), Some(frame.throttle), Some(frame.brake), Some( crate::telemetry_frame::TelemetryFrame::format_lap_time(frame.lap_time_ms), ), Some(frame.spline_position), ) } else { (None, None, None, None, None, None, None) }; feeds.push(LiveFeedSnapshot { label: info.label.to_string(), port: info.port, health: format!("{:?}", snap.health).to_lowercase(), frequency_hz: snap.frequency_hz, total_packets: snap.total_packets, data_rate_kb_s: snap.data_rate_kb_s, speed_kmh, gear, engine_rpm, throttle, brake, lap_time, spline_position, }); } let active_recording = state.recording_manager.active_recording_id().await; ok_json(LiveTelemetryResponse { feeds, active_recording, }) } async fn list_use_cases() -> Response { ok_json(UseCase::all()) } async fn list_games() -> Response { ok_json(Game::valid_choices()) } pub fn build_router(state: Arc) -> Router { Router::new() .route("/api/sessions", get(list_sessions).post(create_session)) .route("/api/sessions/{id}", get(get_session).delete(delete_session)) .route("/api/sessions/{id}/runs", post(start_run)) .route("/api/sessions/{id}/runs/{run_id}/stop", post(stop_run)) .route("/api/sessions/{id}/complete", post(complete_session)) .route("/api/sessions/{id}/analysis", get(get_analysis)) .route("/api/sessions/{id}/analyze", post(run_analysis)) .route("/api/recordings", get(list_recordings).post(start_recording)) .route("/api/recordings/stop", post(stop_recording)) .route( "/api/recordings/{id}", get(get_recording).delete(delete_recording), ) .route("/api/recordings/{id}/stop", post(stop_specific_recording)) .route( "/api/recordings/{id}/data/{feed_name}", get(get_feed_data), ) .route( "/api/recordings/{id}/frames/{feed_name}", get(get_feed_data), ) .route("/api/live", get(get_live)) .route("/api/use-cases", get(list_use_cases)) .route("/api/games", get(list_games)) .layer(CorsLayer::permissive()) .with_state(state) } pub async fn run_server(state: Arc, addr: String, mut shutdown_rx: tokio::sync::watch::Receiver) { let router = build_router(state); let listener = match tokio::net::TcpListener::bind(&addr).await { Ok(l) => l, Err(e) => { eprintln!("[ERROR] Failed to bind API server on {}: {}", addr, e); return; } }; eprintln!("[INFO] REST API server listening on {}", addr); let shutdown_signal = async move { while shutdown_rx.changed().await.is_ok() { if *shutdown_rx.borrow() { break; } } }; if let Err(e) = axum::serve(listener, router) .with_graceful_shutdown(shutdown_signal) .await { eprintln!("[ERROR] API server error: {}", e); } eprintln!("[INFO] REST API server shut down."); }