lesavka/client/src/app.rs

// client/src/app.rs
#![forbid(unsafe_code)]
use anyhow::Result;
use std::time::Duration;
use tokio::{sync::broadcast, task::JoinHandle};
use tokio_stream::{wrappers::BroadcastStream, StreamExt};
use tonic::Request;
use tracing::{info, warn, error, debug};

use navka_common::navka::{relay_client::RelayClient, KeyboardReport, MouseReport,};

use input::inputs::InputAggregator;


pub struct NavkaClientApp {
    aggregator: Option<InputAggregator>,
    server_addr: String,
    dev_mode: bool,
    kbd_tx: broadcast::Sender<KeyboardReport>,
    mou_tx: broadcast::Sender<MouseReport>,
}

impl NavkaClientApp {
    pub fn new() -> Result<Self> {
        info!("Creating navka-client app!");
        let dev_mode    = std::env::var("NAVKA_DEV_MODE").is_ok();
        let server      = std::env::args()
            .nth(1)
            .or_else(|| std::env::var("NAVKA_SERVER_ADDR").ok())
            .unwrap_or_else(|| "http://127.0.0.1:50051".to_owned());

        let (kbd_tx, _) = broadcast::channel::<KeyboardReport>(1024);
        let (mou_tx, _) = broadcast::channel::<MouseReport>(4096);
        let mut agg = InputAggregator::new(dev_mode, kbd_tx.clone(), mou_tx.clone());
        agg.init()?;

        Ok(Self {
            keyboard: Some(kbd_aggregator),
            mouse: Some(mou_aggregator),
            server_addr: addr,
            dev_mode,
            tx,
        })
    }

    pub async fn run(&mut self) -> Result<()> {
        info!("Running navka-client app!");
        /* ─ spawn the input thread ─ */
        let agg_task: JoinHandle<Result<()>> =
            tokio::spawn(async move { self.aggregator.take().unwrap().run().await });

        /* ─ stream loops ─ */
        let kbd_loop = self.stream_loop_keyboard();
        let mou_loop = self.stream_loop_mouse();

        /* ─ suicide timer for DEV mode ─ */
        let suicide_future = async {
            if self.dev_mode {
                info!("DEV-mode: will kill itself in 30s");
                tokio::time::sleep(Duration::from_secs(30)).await;
                Err::<(), _>(anyhow::anyhow!("dev-mode timer expired - 💀goodbye💀 cruel world..."))
            } else {
                futures::future::pending().await
            }
        };

        // Combine aggregator + dev + reconnect logic
        tokio::select! {
            // aggregator finishes
            agg_res = aggregator_task => {
                match agg_res {
                    Ok(Ok(())) => {
                        error!("Aggregator ended normally, exiting.");
                        std::process::exit(0);
                    }
                    Ok(Err(e)) => {
                        error!("Aggregator ended with error: {e}");
                        std::process::exit(1);
                    }
                    Err(join_err) => {
                        error!("Aggregator task panicked or was cancelled: {join_err}");
                        std::process::exit(1);
                    }
                }
            },
            // dev-mode
            res = suicide_future => {
                warn!("Dev-mode: {res:?}");
                std::process::exit(0);
            },
            // reconnect loop
            _ = self.reconnect_loop() => {
                warn!("Reconnect loop ended??");
                std::process::exit(0);
            }
        }
    }

    #[inline(always)]
    async fn delay() { tokio::time::sleep(Duration::from_secs(1)).await; }

    /*──────────────────────────────── keyboard stream ─────────────────────────*/
    async fn stream_loop_keyboard(&self) {
        loop {
            info!("⌨️  dial {}", self.server_addr);
            let mut cli = match RelayClient::connect(self.server_addr.clone()).await {
                Ok(c) => c, Err(e) => { error!("connect: {e}"); delay().await; continue }
            };
            let outbound = BroadcastStream::new(self.kbd_tx.subscribe())
                .filter_map(|r| r.ok());

            info!("⌨️  start stream_keyboard()");
            let resp = match cli.stream_keyboard(Request::new(outbound)).await {
                Ok(r) => r, Err(e) => { error!("stream_keyboard: {e}"); delay().await; continue }
            };

            let mut inbound = resp.into_inner();
            while let Some(m) = inbound.message().await.transpose() {
                match m {
                    Ok(r) => debug!("↩️  kbd echo {} B", r.data.len()),
                    Err(e) => { error!("kbd inbound: {e}"); break }
                }
            }
            warn!("⌨️  disconnected, retry");
            delay().await;
        }
    }

    /*──────────────────────────────── mouse stream ────────────────────────────*/
    async fn stream_loop_mouse(&self) {
        loop {
            info!("🖱️  dial {}", self.server_addr);
            let mut cli = match RelayClient::connect(self.server_addr.clone()).await {
                Ok(c) => c, Err(e) => { error!("connect: {e}"); delay().await; continue }
            };
            let outbound = BroadcastStream::new(self.mou_tx.subscribe())
                .filter_map(|r| r.ok());

            info!("🖱️  start stream_mouse()");
            let resp = match cli.stream_mouse(Request::new(outbound)).await {
                Ok(r) => r, Err(e) => { error!("stream_mouse: {e}"); delay().await; continue }
            };

            let mut inbound = resp.into_inner();
            while let Some(m) = inbound.message().await.transpose() {
                match m {
                    Ok(r) => debug!("↩️  mouse echo {} B", r.data.len()),
                    Err(e) => { error!("mouse inbound: {e}"); break }
                }
            }
            warn!("🖱️  disconnected, retry");
            delay().await;
        }
    }
}