From d161c74eeedb47ad425592c16533b7227d613b01 Mon Sep 17 00:00:00 2001
From: lambda <lambda@bjornmossa.net>
Date: Mon, 26 Oct 2026 12:42:31 +0300
Subject: [PATCH] feat: reconnect telemetry stream on disconnect

---
 src-tauri/src/lib.rs | 106 +++++++++++++++++++++++++++++++++++++------
 1 file changed, 91 insertions(+), 15 deletions(-)

diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs
index 44dfa03..66364a1 100644
--- a/src-tauri/src/lib.rs
+++ b/src-tauri/src/lib.rs
@@ -103,21 +103,12 @@ async fn connect(app: AppHandle, device_id: String, category: String) -> Result<
     }
 
     let source = state.source.clone();
-    let stream = source.stream(&device).await.map_err(|e| e.to_string())?;
-
+    let source_for_spawn = source.clone();
     let app_t = app.clone();
-    let device_id_t = device.id.clone();
+    let device_t = device.clone();
+    let category_t = category.clone();
     let handle = async_runtime::spawn(async move {
-        let mut stream = stream;
-        while let Some(telemetry) = stream.next().await {
-            let _ = app_t.emit(
-                "telemetry",
-                TelemetryEvent {
-                    device_id: device_id_t.clone(),
-                    telemetry,
-                },
-            );
-        }
+        run_watchdog(app_t, source_for_spawn, device_t, category_t).await;
     });
 
     state
@@ -154,7 +145,9 @@ async fn set_target_power(app: AppHandle, watts: i16) -> Result<(), String> {
         .source
         .command(&device, Command::FtmsSetTargetPower(watts))
         .await
-        .map_err(|e| e.to_string())
+        .map_err(|e| e.to_string())?;
+    state.trainer_targets.lock().unwrap().power = Some(watts);
+    Ok(())
 }
 
 /// Задать целевой каденс станку (rpm; внутри переводится в 0.5-rpm единицы FTMS).
@@ -166,7 +159,9 @@ async fn set_target_cadence(app: AppHandle, rpm: u16) -> Result<(), String> {
         .source
         .command(&device, Command::FtmsSetTargetCadence(rpm * 2))
         .await
-        .map_err(|e| e.to_string())
+        .map_err(|e| e.to_string())?;
+    state.trainer_targets.lock().unwrap().cadence = Some(rpm);
+    Ok(())
 }
 
 /// Запустить станок (FTMS Start or Resume, opcode 0x07).
@@ -234,6 +229,87 @@ where
     }
 }
 
+async fn run_watchdog(
+    app: AppHandle,
+    source: Arc<dyn SensorSource>,
+    device: Device,
+    category: String,
+) {
+    let device_id = device.id.clone();
+    let mut attempts: u32 = 0;
+
+    loop {
+        // stream() does ensure_connected() internally, so a fresh call = reconnect.
+        let mut stream = match source.stream(&device).await {
+            Ok(s) => {
+                attempts = 0;
+                emit_connection(&app, &device_id, &category, "reconnected");
+                s
+            }
+            Err(_) => {
+                attempts += 1;
+                emit_connection(&app, &device_id, &category, "reconnecting");
+                if attempts >= RECONNECT_MAX_ATTEMPTS {
+                    emit_connection(&app, &device_id, &category, "gave_up");
+                    return;
+                }
+                time::sleep(RECONNECT_BACKOFF * attempts).await;
+                continue;
+            }
+        };
+
+        // Trainer: re-request control and re-apply last targets after reconnect.
+        if category == "trainer" {
+            let _ = source.command(&device, Command::FtmsRequestControl).await;
+            reapply_trainer_targets(&app, &source, &device).await;
+        }
+
+        let app_emit = app.clone();
+        let dev_id = device_id.clone();
+        let outcome = consume_with_watchdog(
+            &mut stream,
+            WATCHDOG_STALE_TIMEOUT,
+            move |t| {
+                let _ = app_emit.emit("telemetry", TelemetryEvent {
+                    device_id: dev_id.clone(),
+                    telemetry: t,
+                });
+            },
+        ).await;
+
+        // Stream broke (Ended or Stale). Loop back and reconnect.
+        let _ = outcome;
+        emit_connection(&app, &device_id, &category, "reconnecting");
+        time::sleep(RECONNECT_BACKOFF).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_targets(
+    app: &AppHandle,
+    source: &Arc<dyn SensorSource>,
+    device: &Device,
+) {
+    let state = app.state::<State>();
+    let (power, cadence) = {
+        let targets = state.trainer_targets.lock().unwrap();
+        (targets.power, targets.cadence)
+    };
+    if let Some(watts) = power {
+        let _ = source.command(device, Command::FtmsSetTargetPower(watts)).await;
+    }
+    if let Some(rpm) = cadence {
+        let _ = source.command(device, Command::FtmsSetTargetCadence(rpm * 2)).await;
+    }
+}
+
 #[cfg_attr(mobile, tauri::mobile_entry_point)]
 pub fn run() {
     tauri::Builder::default()