// rusty-telemetry — Sim racing telemetry analysis platform // Copyright (C) 2026 Jan-Peter v. Hunnius // // This program is free software: you can redistribute it and/or modify // it under the terms of the GNU Affero General Public License as published by // the Free Software Foundation, either version 3 of the License, or // (at your option) any later version. // // This program is distributed in the hope that it will be useful, // but WITHOUT ANY WARRANTY; without even the implied warranty of // MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the // GNU Affero General Public License for more details. // // You should have received a copy of the GNU Affero General Public License // along with this program. If not, see . //! 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::{delete, get, post}, }; use serde::{Deserialize, Serialize}; use std::sync::{Arc, Mutex, MutexGuard}; use tower_http::cors::CorsLayer; use crate::analysis; use crate::characteristics; use crate::feed_state::FeedState; use crate::recording::{Game, RecordingManager, RecordingStatus, StartRecordingRequest, UseCase}; use crate::session::SessionManager; use crate::telemetry_frame::TelemetryFrame; use crate::ui::FeedDisplayInfo; pub struct AppState { pub recording_manager: Arc, pub feed_displays: Vec<(FeedDisplayInfo, Arc>)>, pub session_manager: Arc>, pub track_model_store: 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, #[serde(default)] merge_feeds: Vec, } #[derive(Deserialize)] struct BuildTrackRequest { session_id: String, game: Option, track_name: Option, } async fn create_session( State(state): State>, Json(req): Json, ) -> Response { let recording_id = state.recording_manager.active_recording_id().await; let mut mgr = lock_sessions(&state); match mgr.create_session(req.use_case, req.label, recording_id) { 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, req.merge_feeds) { 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 delete_run( State(state): State>, Path((id, run_id)): Path<(String, String)>, ) -> Response { let mut mgr = lock_sessions(&state); match mgr.delete_run(&id, &run_id) { Ok(()) => StatusCode::NO_CONTENT.into_response(), Err(msg) => err(StatusCode::BAD_REQUEST, &msg), } } async fn reopen_session( State(state): State>, Path(id): Path, ) -> Response { let mut mgr = lock_sessions(&state); match mgr.reopen_session(&id) { Ok(session) => ok_json(session), Err(msg) => err(StatusCode::BAD_REQUEST, &msg), } } async fn enrich_with_recording_meta( state: &Arc, result: &mut serde_json::Value, recording_id: Option<&str>, ) { let rec_id = match recording_id { Some(id) => id, None => return, }; if let Ok(manifest) = state.recording_manager.get_recording(rec_id).await { if let Some(obj) = result.as_object_mut() { obj.insert("car".to_string(), serde_json::json!(manifest.car)); obj.insert("game".to_string(), serde_json::json!(manifest.game)); obj.insert("track".to_string(), serde_json::json!(manifest.track)); } } } async fn execute_and_store_analysis( state: &Arc, session_id: &str, use_case: UseCase, frames_by_run: Vec>, run_labels: Vec, recording_id: Option, ) -> serde_json::Value { let mut result = analysis::analyze(&use_case, &frames_by_run, &run_labels); enrich_with_recording_meta(state, &mut result, recording_id.as_deref()).await; let mut mgr = lock_sessions(state); let _ = mgr.set_analysis(session_id, result.clone()); result } async fn complete_session( State(state): State>, Path(id): Path, ) -> Response { let (use_case, frames_by_run, run_labels, recording_id) = { 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)) => { let rec_id = mgr.get_session(&id).and_then(|s| s.recording_id.clone()); (uc, frames, labels, rec_id) } None => return err(StatusCode::NOT_FOUND, "Session not found after completion"), } }; execute_and_store_analysis(&state, &id, use_case, frames_by_run, run_labels, recording_id).await; 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, recording_id) = { let mut 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"), } let rec_id = mgr.get_session(&id).unwrap().recording_id.clone(); match mgr.collect_run_frames(&id) { Some((uc, frames, labels)) => (uc, frames, labels, rec_id), None => return err(StatusCode::NOT_FOUND, "Session not found"), } }; let result = execute_and_store_analysis(&state, &id, use_case, frames_by_run, run_labels, recording_id).await; 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, }) } #[derive(Serialize)] struct UseCaseInfo { id: String, name: String, guidance: String, } async fn list_tracks(State(state): State>) -> Response { match state.track_model_store.list().await { Ok(summaries) => ok_json(summaries), Err(e) => err( StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to list tracks: {}", e), ), } } async fn get_track( State(state): State>, Path((game, track_name)): Path<(String, String)>, ) -> Response { match state.track_model_store.load(&game, &track_name).await { Ok(model) => ok_json(model), Err(e) if e.kind() == std::io::ErrorKind::NotFound => { err(StatusCode::NOT_FOUND, "Track model 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 load track model: {}", e), ), } } async fn delete_track( State(state): State>, Path((game, track_name)): Path<(String, String)>, ) -> Response { match state.track_model_store.delete(&game, &track_name).await { Ok(()) => StatusCode::NO_CONTENT.into_response(), Err(e) if e.kind() == std::io::ErrorKind::NotFound => { err(StatusCode::NOT_FOUND, "Track model 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 delete track model: {}", e), ), } } async fn get_track_corners( State(state): State>, Path((game, track_name)): Path<(String, String)>, ) -> Response { match state.track_model_store.load(&game, &track_name).await { Ok(model) => ok_json(&model.corners), Err(e) if e.kind() == std::io::ErrorKind::NotFound => { err(StatusCode::NOT_FOUND, "Track model 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 load track model: {}", e), ), } } async fn build_track( State(state): State>, Json(req): Json, ) -> Response { let rec_id = { let mgr = lock_sessions(&state); let session = match mgr.get_session(&req.session_id) { Some(s) => s, None => return err(StatusCode::NOT_FOUND, "Session not found"), }; match session.recording_id.clone() { Some(id) => id, None => return err(StatusCode::BAD_REQUEST, "Session has no associated recording"), } }; let manifest = match state.recording_manager.get_recording(&rec_id).await { Ok(m) => m, Err(_) => return err(StatusCode::NOT_FOUND, "Recording not found"), }; let game = req.game.unwrap_or_else(|| manifest.game.to_track_dir_name().to_string()); let track_name = req.track_name.unwrap_or(manifest.track); if game.is_empty() || track_name.is_empty() { return err(StatusCode::BAD_REQUEST, "Game and track name are required. Specify them in the request or set them when starting the recording."); } let boundaries = { let mgr = lock_sessions(&state); let session = match mgr.get_session(&req.session_id) { Some(s) => s, None => return err(StatusCode::NOT_FOUND, "Session not found"), }; if session.use_case != UseCase::TrackMapping { return err(StatusCode::BAD_REQUEST, "Session is not a track_mapping session"); } if session.status != crate::session::SessionStatus::Completed { return err(StatusCode::BAD_REQUEST, "Session must be completed first"); } let analysis = match &session.analysis { Some(a) => a.clone(), None => return err(StatusCode::BAD_REQUEST, "Session has no analysis results"), }; let boundaries_val = match analysis.get("boundaries") { Some(v) => v.clone(), None => return err(StatusCode::BAD_REQUEST, "No boundaries in analysis"), }; match serde_json::from_value::>(boundaries_val) { Ok(b) => b, Err(e) => return err(StatusCode::BAD_REQUEST, &format!("Failed to parse boundaries: {}", e)), } }; let source = crate::track_model::TrackModelSource::BoundaryRecording; let track_model = match analysis::build_track_model(&boundaries, game.clone(), track_name.clone(), source) { Ok(m) => m, Err(e) => return err(StatusCode::BAD_REQUEST, &format!("Failed to build track model: {}", e)), }; match state.track_model_store.save(&track_model).await { Ok(()) => (StatusCode::CREATED, Json(track_model)).into_response(), Err(e) => err(StatusCode::INTERNAL_SERVER_ERROR, &format!("Failed to save track model: {}", e)), } } async fn list_use_cases() -> Response { let use_cases: Vec = UseCase::all() .into_iter() .map(|uc| UseCaseInfo { id: serde_json::to_value(&uc) .ok() .and_then(|v| v.as_str().map(String::from)) .unwrap_or_default(), name: format!("{:?}", uc), guidance: uc.guidance().to_string(), }) .collect(); ok_json(use_cases) } async fn list_games() -> Response { ok_json(Game::valid_choices()) } async fn list_car_profiles( State(state): State>, ) -> Response { let mgr = lock_sessions(&state); ok_json(characteristics::car_profiles(&mgr)) } async fn get_car_profile( State(state): State>, Path(car): Path, ) -> Response { let mgr = lock_sessions(&state); match characteristics::car_profile(&mgr, &car) { Some(profile) => ok_json(profile), None => err(StatusCode::NOT_FOUND, &format!("No data for car '{}'", car)), } } 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}/runs/{run_id}", delete(delete_run)) .route("/api/sessions/{id}/reopen", post(reopen_session)) .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/tracks", get(list_tracks)) .route("/api/tracks/{game}/{track_name}", get(get_track).delete(delete_track)) .route("/api/tracks/{game}/{track_name}/corners", get(get_track_corners)) .route("/api/tracks/build", post(build_track)) .route("/api/use-cases", get(list_use_cases)) .route("/api/games", get(list_games)) .route("/api/characteristics/cars", get(list_car_profiles)) .route("/api/characteristics/cars/{car}", get(get_car_profile)) .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."); }