ridein/src-tauri/src/lib.rs

mod device;
mod ble;
mod device_repo;
mod state;
mod telemetry;
mod commands;
mod workout;

use std::pin::Pin;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use std::time::Duration;

use futures::{Stream, StreamExt};
use serde::Serialize;
use tauri::{AppHandle, Emitter, Manager, async_runtime};
use tokio::time;

use crate::ble::{BleSource, MockSource, SensorSource};
use crate::commands::Command;
use crate::device::Device;
use crate::device_repo::{repo_from_devices, Repo};
use crate::state::{State, TrainerState, WorkoutHandle};
use crate::telemetry::Telemetry;
use btleplug::api::Manager as ApiManager;
use btleplug::platform::Manager as BleManager;

/// Сколько секунд сканируем BLE на старте.
const SCAN_SECONDS: u64 = 5;

#[derive(Clone, Serialize)]
struct TelemetryEvent {
    device_id: String,
    telemetry: Telemetry,
}

#[derive(Clone, Serialize)]
struct BatteryEvent {
    device_id: String,
    percent: u8,
}

// Watchdog: how long without telemetry before the connection is considered dead.
const WATCHDOG_STALE_TIMEOUT: Duration = Duration::from_secs(5);
// How many consecutive failed/silent cycles before giving up on reconnect.
const RECONNECT_MAX_ATTEMPTS: u32 = 5;
// Base backoff between reconnect attempts.
const RECONNECT_BACKOFF: Duration = Duration::from_secs(2);

#[derive(Clone, Serialize)]
struct ConnectionEvent {
    device_id: String,
    category: String,
    status: &'static str, // "reconnecting" | "reconnected" | "gave_up"
}

#[derive(Clone, Serialize)]
struct WorkoutProgressEvent {
    step_index: usize,
    step_name: String,
    target_watts: Option<i16>,
    target_cadence: Option<u16>,
    elapsed_secs: u64,
    total_secs: u64,
}

#[derive(Clone, Serialize)]
struct WorkoutFinishedEvent {
    ok: bool,
}

#[tauri::command]
async fn init(app: AppHandle) -> Result<(), String> {
    let repo = scan_and_classify(&app).await.map_err(|e| e.to_string())?;
    present_repo(&app, repo).map_err(|e| e.to_string())?;
    Ok(())
}

/// Этап 1: скан BLE-устройств и раскладка по типам датчиков.
async fn scan_and_classify(app: &AppHandle) -> Result<Repo, ble::BleError> {
    let state = app.state::<State>();
    let source = state.source.clone();
    let devices = source.scan(SCAN_SECONDS).await?;
    *state.devices.write().unwrap() = devices.clone();
    Ok(repo_from_devices(devices))
}

/// Этап 2: закрываем сплэш, показываем главное окно, отдаём список датчиков на фронт.
fn present_repo(app: &AppHandle, repo: Repo) -> Result<(), String> {
    let splash = app
        .get_webview_window("splashscreen")
        .ok_or("окно splashscreen не найдено")?;
    let main = app
        .get_webview_window("main")
        .ok_or("окно main не найдено")?;

    splash.close().map_err(|e| format!("не удалось закрыть splashscreen: {e}"))?;
    main.show().map_err(|e| format!("не удалось показать main: {e}"))?;
    app.emit("devices-loaded", repo)
        .map_err(|e| format!("не удалось отправить событие devices-loaded: {e}"))?;

    Ok(())
}

#[tauri::command]
async fn connect(app: AppHandle, device_id: String, category: String) -> Result<(), String> {
    let state = app.state::<State>();

    let device = {
        let devices = state.devices.read().unwrap();
        devices
            .iter()
            .find(|d| d.id == device_id)
            .cloned()
            .ok_or_else(|| "устройство не найдено".to_string())?
    };

    if let Some(old) = state.streams.lock().unwrap().remove(&category) {
        old.abort();
    }

    let source = state.source.clone();

    // Battery first, before the watchdog task touches the peripheral (avoids a
    // connect/discover race on the same peripheral).
    if let Ok(percent) = source.battery(&device).await {
        let _ = app.emit(
            "battery",
            BatteryEvent {
                device_id: device.id.clone(),
                percent,
            },
        );
    }

    let app_t = app.clone();
    let device_t = device.clone();
    let category_t = category.clone();
    let handle = async_runtime::spawn(async move {
        run_watchdog(app_t, source, device_t, category_t).await;
    });

    state
        .streams
        .lock()
        .unwrap()
        .insert(category.clone(), handle);

    // Trainer: remember the device id and reset the previous trainer's targets.
    // Control is requested (with confirmation) inside run_watchdog, after the
    // notification subscription is live.
    if category == "trainer" {
        *state.trainer.lock().unwrap() = Some(device.id.clone());
        *state.trainer_state.lock().unwrap() = TrainerState::default();
    }

    Ok(())
}

/// Задать целевую мощность станку (ERG).
#[tauri::command]
async fn set_target_power(app: AppHandle, watts: i16) -> Result<(), String> {
    let state = app.state::<State>();
    if watts < 0 {
        return Err("мощность не может быть отрицательной".to_string());
    }
    let device = trainer_device(&state)?;
    state
        .source
        .command(&device, Command::FtmsSetTargetPower(watts))
        .await
        .map_err(|e| e.to_string())?;
    state.trainer_state.lock().unwrap().power = Some(watts);
    Ok(())
}

/// Задать целевой каденс станку (rpm; внутри переводится в 0.5-rpm единицы FTMS).
#[tauri::command]
async fn set_target_cadence(app: AppHandle, rpm: u16) -> Result<(), String> {
    let state = app.state::<State>();
    let half_rpm = rpm.checked_mul(2).ok_or("каденс слишком большой")?;
    let device = trainer_device(&state)?;
    state
        .source
        .command(&device, Command::FtmsSetTargetCadence(half_rpm))
        .await
        .map_err(|e| e.to_string())?;
    state.trainer_state.lock().unwrap().cadence = Some(rpm);
    Ok(())
}

/// Запустить станок (FTMS Start or Resume, opcode 0x07).
#[tauri::command]
async fn trainer_start(app: AppHandle) -> Result<(), String> {
    let state = app.state::<State>();
    let device = trainer_device(&state)?;
    state
        .source
        .command(&device, Command::FtmsStart)
        .await
        .map_err(|e| e.to_string())?;
    state.trainer_state.lock().unwrap().running = true;
    Ok(())
}

/// Остановить станок (FTMS Stop or Pause, opcode 0x08).
#[tauri::command]
async fn trainer_stop(app: AppHandle) -> Result<(), String> {
    let state = app.state::<State>();
    let device = trainer_device(&state)?;
    state
        .source
        .command(&device, Command::FtmsStop)
        .await
        .map_err(|e| e.to_string())?;
    state.trainer_state.lock().unwrap().running = false;
    Ok(())
}

/// Возвращает Device подключённого станка.
fn trainer_device(state: &State) -> Result<Device, String> {
    let trainer_id = state
        .trainer
        .lock()
        .unwrap()
        .clone()
        .ok_or("станок не подключен")?;
    let devices = state.devices.read().unwrap();
    devices
        .iter()
        .find(|d| d.id == trainer_id)
        .cloned()
        .ok_or_else(|| "станок не найден".to_string())
}

/// Список доступных Ride-воркаутов (сводки).
#[tauri::command]
fn list_workouts() -> Result<Vec<workout::WorkoutSummary>, String> {
    workout::list_workouts(&workout::workouts_dir()).map_err(|e| e.to_string())
}

/// Полный воркаут по имени файла (из каталога воркаутов).
#[tauri::command]
fn get_workout(path: String) -> Result<workout::Workout, String> {
    workout::read_workout(&path).map_err(|e| e.to_string())
}

/// Задать FTP гонщика (Вт). Используется для перевода %FTP → ватты.
#[tauri::command]
fn set_ftp(app: AppHandle, ftp: u16) -> Result<(), String> {
    if ftp == 0 {
        return Err("FTP должен быть больше нуля".to_string());
    }
    *app.state::<State>().ftp.lock().unwrap() = ftp;
    Ok(())
}

/// Текущий FTP гонщика (Вт).
#[tauri::command]
fn get_ftp(app: AppHandle) -> u16 {
    *app.state::<State>().ftp.lock().unwrap()
}

/// Вычисляет и отправляет целевую мощность сегмента, если она изменилась.
async fn apply_power(
    source: &Arc<dyn SensorSource>,
    device: &Device,
    seg: &workout::Segment,
    elapsed_secs: f64,
    ftp: u16,
    last_watts: Option<i16>,
) -> Option<i16> {
    let watts = seg.power.as_ref().map(|p| {
        workout::percent_to_watts(workout::target_percent(p, elapsed_secs, seg.duration_secs), ftp)
    });
    if let Some(w) = watts.filter(|w| Some(*w) != last_watts) {
        let _ = source.command(device, Command::FtmsSetTargetPower(w)).await;
    }
    watts
}

/// Собирает событие прогресса воркаута.
fn progress_event(
    idx: usize,
    seg: &workout::Segment,
    watts: Option<i16>,
    elapsed_total: u64,
    total: u64,
) -> WorkoutProgressEvent {
    WorkoutProgressEvent {
        step_index: idx,
        step_name: seg.text.clone().unwrap_or_default(),
        target_watts: watts,
        target_cadence: seg.cadence_rpm,
        elapsed_secs: elapsed_total,
        total_secs: total,
    }
}

/// Ядро исполнения воркаута (без AppHandle) — тестируемо против MockSource.
async fn run_workout_engine(
    source: &Arc<dyn SensorSource>,
    device: &Device,
    segments: &[workout::Segment],
    ftp: u16,
    paused: &AtomicBool,
    tick_duration: Duration,
    mut on_progress: impl FnMut(WorkoutProgressEvent) + Send,
) -> bool {
    if source.request_control(device).await.is_err() {
        return false;
    }
    let _ = source.command(device, Command::FtmsStart).await;

    let total = workout::total_duration_secs(segments);
    let mut elapsed_total: u64 = 0;

    for (idx, seg) in segments.iter().enumerate() {
        let dur = seg.duration_secs as u64;

        // каденс — константа сегмента, шлём один раз (в 0.5-rpm единицах FTMS).
        if let Some(half) = seg.cadence_rpm.and_then(|rpm| rpm.checked_mul(2)) {
            let _ = source.command(device, Command::FtmsSetTargetCadence(half)).await;
        }

        let mut last_watts: Option<i16> = None;
        let mut tick = time::interval(tick_duration);
        tick.set_missed_tick_behavior(time::MissedTickBehavior::Delay);

        // t = 0
        let w = apply_power(source, device, seg, 0.0, ftp, last_watts).await;
        last_watts = w;
        on_progress(progress_event(idx, seg, w, elapsed_total, total));

        let mut elapsed_in_seg: u64 = 0;
        while elapsed_in_seg < dur {
            tick.tick().await;
            if paused.load(Ordering::SeqCst) {
                continue;
            }
            elapsed_in_seg += 1;
            let w = apply_power(source, device, seg, elapsed_in_seg as f64, ftp, last_watts).await;
            last_watts = w;
            elapsed_total += 1;
            on_progress(progress_event(idx, seg, w, elapsed_total, total));
        }
    }

    let _ = source.command(device, Command::FtmsStop).await;
    true
}

/// Исполняет воркаут: обёртка run_workout_engine с эмитом событий в приложение.
async fn run_workout(
    app: AppHandle,
    source: Arc<dyn SensorSource>,
    device: Device,
    segments: Vec<workout::Segment>,
    ftp: u16,
    paused: Arc<AtomicBool>,
) {
    let app_progress = app.clone();
    let ok = run_workout_engine(
        &source,
        &device,
        &segments,
        ftp,
        paused.as_ref(),
        Duration::from_secs(1),
        move |p| {
            let _ = app_progress.emit("workout-progress", p);
        },
    )
    .await;
    let _ = app.emit("workout-finished", WorkoutFinishedEvent { ok });
}

/// Запустить воркаут (по имени файла из каталога воркаутов).
#[tauri::command]
fn start_workout(app: AppHandle, path: String) -> Result<(), String> {
    let workout = workout::read_workout(&path).map_err(|e| e.to_string())?;
    let segments = workout::flatten(&workout.steps);
    if segments.is_empty() {
        return Err("воркаут пустой".to_string());
    }

    let state = app.state::<State>();
    let ftp = *state.ftp.lock().unwrap();
    let device = trainer_device(&state)?;
    let source = state.source.clone();

    // остановить предыдущий воркаут, если был
    if let Some(old) = state.workout.lock().unwrap().take() {
        old.task.abort();
    }

    let paused = Arc::new(AtomicBool::new(false));
    let app_t = app.clone();
    let source_t = source.clone();
    let device_t = device.clone();
    let paused_t = paused.clone();
    let handle = async_runtime::spawn(async move {
        run_workout(app_t, source_t, device_t, segments, ftp, paused_t).await;
    });

    *state.workout.lock().unwrap() = Some(WorkoutHandle { task: handle, paused });
    Ok(())
}

/// Остановить текущий воркаут.
#[tauri::command]
async fn stop_workout(app: AppHandle) -> Result<(), String> {
    let state = app.state::<State>();
    // взять ручку (guard отпускается сразу, до await)
    let handle = {
        let mut guard = state.workout.lock().unwrap();
        guard.take()
    };
    if let Some(h) = handle {
        h.task.abort();
        let device = trainer_device(&state)?;
        let _ = state.source.command(&device, Command::FtmsStop).await;
    }
    Ok(())
}

/// Пауза / возобновление текущего воркаута.
#[tauri::command]
fn pause_workout(app: AppHandle, paused: bool) -> Result<(), String> {
    let state = app.state::<State>();
    let guard = state.workout.lock().unwrap();
    let handle = guard.as_ref().ok_or("воркаут не запущен")?;
    handle.paused.store(paused, Ordering::SeqCst);
    Ok(())
}

/// Why the watchdog loop stopped consuming a stream.
enum StreamOutcome {
    /// The stream ended (peripheral disconnected / notifications closed).
    Ended,
    /// No item arrived within `timeout` (stale connection).
    Stale,
}

/// Consumes a telemetry stream, invoking `on_item` per item, until it ends or goes
/// silent for `timeout`. Returns the outcome and whether at least one item was seen.
async fn consume_with_watchdog<F>(
    stream: &mut Pin<Box<dyn Stream<Item = Telemetry> + Send>>,
    timeout: Duration,
    mut on_item: F,
) -> (StreamOutcome, bool)
where
    F: FnMut(Telemetry),
{
    let mut saw = false;
    loop {
        match time::timeout(timeout, stream.next()).await {
            Ok(Some(t)) => {
                saw = true;
                on_item(t);
            }
            Ok(None) => return (StreamOutcome::Ended, saw),
            Err(_) => return (StreamOutcome::Stale, saw),
        }
    }
}

async fn run_watchdog(
    app: AppHandle,
    source: Arc<dyn SensorSource>,
    device: Device,
    category: String,
) {
    let device_id = device.id.clone();
    let mut silent_cycles: u32 = 0;

    loop {
        // stream() does ensure_connected() internally, so a fresh call = reconnect.
        let mut stream = match source.stream(&device).await {
            Ok(s) => {
                emit_connection(&app, &device_id, &category, "reconnected");
                s
            }
            Err(_) => {
                silent_cycles += 1;
                emit_connection(&app, &device_id, &category, "reconnecting");
                if silent_cycles >= RECONNECT_MAX_ATTEMPTS {
                    emit_connection(&app, &device_id, &category, "gave_up");
                    app.state::<State>().streams.lock().unwrap().remove(&category);
                    return;
                }
                time::sleep(RECONNECT_BACKOFF * silent_cycles).await;
                continue;
            }
        };

        // Trainer: request control (with confirmation) and restore targets/start.
        if category == "trainer" {
            match source.request_control(&device).await {
                Ok(()) => reapply_trainer_state(&app, &source, &device).await,
                Err(e) => eprintln!("request_control failed: {e}"),
            }
        }

        let app_emit = app.clone();
        let dev_id = device_id.clone();
        let (outcome, saw_data) = consume_with_watchdog(
            &mut stream,
            WATCHDOG_STALE_TIMEOUT,
            move |t| {
                let _ = app_emit.emit(
                    "telemetry",
                    TelemetryEvent {
                        device_id: dev_id.clone(),
                        telemetry: t,
                    },
                );
            },
        )
        .await;

        let _ = outcome;
        // A healthy cycle (received data) resets the counter; a silent one does not.
        if saw_data {
            silent_cycles = 0;
        } else {
            silent_cycles += 1;
        }
        emit_connection(&app, &device_id, &category, "reconnecting");
        if silent_cycles >= RECONNECT_MAX_ATTEMPTS {
            emit_connection(&app, &device_id, &category, "gave_up");
            app.state::<State>().streams.lock().unwrap().remove(&category);
            return;
        }
        time::sleep(RECONNECT_BACKOFF * silent_cycles.max(1)).await;
    }
}

fn emit_connection(app: &AppHandle, device_id: &str, category: &str, status: &'static str) {
    let _ = app.emit(
        "connection",
        ConnectionEvent {
            device_id: device_id.to_string(),
            category: category.to_string(),
            status,
        },
    );
}

async fn reapply_trainer_state(
    app: &AppHandle,
    source: &Arc<dyn SensorSource>,
    device: &Device,
) {
    let state = app.state::<State>();
    let (power, cadence, running) = {
        let t = state.trainer_state.lock().unwrap();
        (t.power, t.cadence, t.running)
    };
    if let Some(watts) = power {
        let _ = source.command(device, Command::FtmsSetTargetPower(watts)).await;
    }
    if let Some(rpm) = cadence {
        if let Some(half_rpm) = rpm.checked_mul(2) {
            let _ = source.command(device, Command::FtmsSetTargetCadence(half_rpm)).await;
        }
    }
    if running {
        let _ = source.command(device, Command::FtmsStart).await;
    }
}

#[cfg_attr(mobile, tauri::mobile_entry_point)]
pub fn run() {
    tauri::Builder::default()
        .setup(|app| {
            async_runtime::block_on(async move {
                let source: Arc<dyn SensorSource> = if std::env::var("RIDEIN_MOCK").is_ok() {
                    Arc::new(MockSource::new())
                } else {
                    let manager = BleManager::new().await?;
                    let adapters = manager.adapters().await?;
                    let adapter = adapters.into_iter().next()
                        .ok_or_else(|| Box::<dyn std::error::Error>::from("No BLE adapter found"))?;
                    Arc::new(BleSource::new(adapter))
                };

                app.manage(State {
                    source,
                    devices: RwLock::new(Vec::new()),
                    trainer: Mutex::new(None),
                    streams: Mutex::new(std::collections::HashMap::new()),
                    trainer_state: Mutex::new(TrainerState::default()),
                    ftp: Mutex::new(state::DEFAULT_FTP),
                    workout: Mutex::new(None),
                });
                Ok::<(), Box<dyn std::error::Error>>(())
            })
        })
        .invoke_handler(tauri::generate_handler![
            init,
            connect,
            set_target_power,
            set_target_cadence,
            trainer_start,
            trainer_stop,
            list_workouts,
            get_workout,
            set_ftp,
            get_ftp,
            start_workout,
            stop_workout,
            pause_workout
        ])
        .run(tauri::generate_context!())
        .expect("error while running tauri application");
}

#[cfg(test)]
mod tests {
    use super::*;
    use futures::stream;
    use std::time::Duration;

    fn pin(s: impl Stream<Item = Telemetry> + Send + 'static) -> Pin<Box<dyn Stream<Item = Telemetry> + Send>> {
        Box::pin(s)
    }

    #[tokio::test]
    async fn ends_when_stream_is_done() {
        let items = vec![Telemetry::HeartRate { bpm: 120 }];
        let mut s = pin(stream::iter(items));
        let mut seen = Vec::new();
        let (outcome, saw) = consume_with_watchdog(&mut s, Duration::from_secs(5), |t| seen.push(t)).await;
        assert!(matches!(outcome, StreamOutcome::Ended));
        assert!(saw);
        assert_eq!(seen, vec![Telemetry::HeartRate { bpm: 120 }]);
    }

    #[tokio::test]
    async fn stale_when_silent_past_timeout() {
        let mut s = pin(stream::pending::<Telemetry>());
        let mut seen = Vec::new();
        let (outcome, saw) = consume_with_watchdog(&mut s, Duration::from_millis(50), |t| seen.push(t)).await;
        assert!(matches!(outcome, StreamOutcome::Stale));
        assert!(!saw);
        assert!(seen.is_empty());
    }

    #[tokio::test]
    async fn forwards_every_item_via_callback() {
        let items = vec![
            Telemetry::Power { watts: 200 },
            Telemetry::Cadence { rpm: 90.0 },
        ];
        let mut s = pin(stream::iter(items));
        let mut seen = Vec::new();
        let (_outcome, saw) = consume_with_watchdog(&mut s, Duration::from_secs(5), |t| seen.push(t)).await;
        assert!(saw);
        assert_eq!(seen.len(), 2);
    }

    #[tokio::test]
    async fn workout_engine_sends_expected_commands() {
        let mock = Arc::new(MockSource::new());
        let source: Arc<dyn SensorSource> = mock.clone();
        let devices = source.scan(0).await.unwrap();
        let trainer = devices.iter().find(|d| d.id == "mock-trainer").unwrap().clone();

        // короткий воркаут: фикс-сегмент + ramp-сегмент
        let segments = vec![
            workout::Segment {
                text: Some("Warmup".into()),
                duration_secs: 1,
                power: Some(workout::PowerTarget::Fixed { percent: 50.0 }),
                cadence_rpm: Some(85),
            },
            workout::Segment {
                text: Some("Ramp".into()),
                duration_secs: 2,
                power: Some(workout::PowerTarget::Ramp { from: 100.0, to: 200.0 }),
                cadence_rpm: None,
            },
        ];

        let paused = Arc::new(AtomicBool::new(false));
        let ok = run_workout_engine(
            &source,
            &trainer,
            &segments,
            250,
            paused.as_ref(),
            Duration::from_millis(20),
            |_| {},
        )
        .await;
        assert!(ok);

        assert_eq!(
            mock.commands(),
            vec![
                Command::FtmsStart,
                Command::FtmsSetTargetCadence(170), // 85 rpm * 2
                Command::FtmsSetTargetPower(125),   // 50% of 250
                Command::FtmsSetTargetPower(250),   // 100% (ramp start)
                Command::FtmsSetTargetPower(375),   // 150%
                Command::FtmsSetTargetPower(500),   // 200% (ramp end)
                Command::FtmsStop,
            ]
        );
    }
}