ridein/src-tauri/src/lib.rs

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

use std::pin::Pin;
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};
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"
}

#[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())
}

/// 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()),
                });
                Ok::<(), Box<dyn std::error::Error>>(())
            })
        })
        .invoke_handler(tauri::generate_handler![
            init,
            connect,
            set_target_power,
            set_target_cadence,
            trainer_start,
            trainer_stop
        ])
        .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);
    }
}