lesavka/server/src/main.rs

//! lesavka-server — receive HidReport and write to /dev/hidg0
// sever/src/main.rs
#![forbid(unsafe_code)]

use futures_util::{Stream, StreamExt};
use std::{pin::Pin, sync::Arc, time::Duration};
use std::sync::atomic::{AtomicBool, Ordering};
use tokio::{fs::{OpenOptions}, io::AsyncWriteExt, sync::Mutex};
use tokio_stream::{wrappers::ReceiverStream};
use tonic::{transport::Server, Request, Response, Status};
use tracing::{info, trace, warn};
use tracing_subscriber::{fmt, EnvFilter};
use udev::{MonitorBuilder};

use lesavka_server::{usb_gadget::UsbGadget, video};

use lesavka_common::lesavka::{
    relay_server::{Relay, RelayServer},
    KeyboardReport, MouseReport,
    MonitorRequest, VideoPacket,
};

/*─────────────────── tonic service ─────────────────────*/
struct Handler {
    kb: Arc<Mutex<tokio::fs::File>>,
    ms: Arc<Mutex<tokio::fs::File>>,
    gadget: UsbGadget,
    did_cycle: AtomicBool,
}

impl Handler {
    async fn make(gadget: UsbGadget) -> anyhow::Result<Self> {
        let kb = OpenOptions::new().write(true).open("/dev/hidg0").await?;
        let ms = OpenOptions::new().write(true)
                                   .custom_flags(libc::O_NONBLOCK)
                                   .open("/dev/hidg1").await?;
        Ok(Self { kb: Arc::new(Mutex::new(kb)),
                  ms: Arc::new(Mutex::new(ms)),
                  gadget,
                  did_cycle: AtomicBool::new(false),
        })
    }
}

#[tonic::async_trait]
impl Relay for Handler {
    type StreamKeyboardStream = ReceiverStream<Result<KeyboardReport, Status>>;
    type StreamMouseStream    = ReceiverStream<Result<MouseReport, Status>>;
    type CaptureVideoStream   = Pin<Box<dyn Stream<Item = Result<VideoPacket, Status>> + Send + Sync + 'static>>;

    async fn stream_keyboard(
        &self,
        req: Request<tonic::Streaming<KeyboardReport>>,
    ) -> Result<Response<Self::StreamKeyboardStream>, Status> {
        // self.gadget.cycle().map_err(|e| Status::internal(e.to_string()))?;
        if !self.did_cycle.swap(true, Ordering::SeqCst) {
            self.gadget
                .cycle()
                .map_err(|e| Status::internal(e.to_string()))?;
            tokio::time::sleep(Duration::from_millis(500)).await;
        }
        let (tx, rx) = tokio::sync::mpsc::channel(32);
        let kb = self.kb.clone();

        tokio::spawn(async move {
            let mut s = req.into_inner();
            while let Some(pkt) = s.next().await.transpose()? {
                // kb.lock().await.write_all(&pkt.data).await?;
                for attempt in 0..50 {
                    match kb.lock().await.write_all(&pkt.data).await {
                        Ok(()) => {
                            trace!("⌨️  wrote {}", pkt.data.iter()
                                   .map(|b| format!("{b:02X}")).collect::<Vec<_>>().join(" "));
                            break;
                        },
                        Err(e) if e.raw_os_error() == Some(libc::EBUSY) => {
                            tokio::time::sleep(Duration::from_millis(20)).await;
                        }
                        Err(e) => return Err(Status::internal(e.to_string())),
                    }
                    if attempt == 49 {
                        return Err(Status::internal("hidg0 stayed BUSY"));
                    }
                }
                tx.send(Ok(pkt)).await;//.ok();       // best-effort echo
            }
            Ok::<(), Status>(())
        });

        Ok(Response::new(ReceiverStream::new(rx)))
    }

    async fn stream_mouse(
        &self,
        req: Request<tonic::Streaming<MouseReport>>,
    ) -> Result<Response<Self::StreamMouseStream>, Status> {
        let (tx, rx) = tokio::sync::mpsc::channel(4096);   // higher burst
        let ms = self.ms.clone();

        tokio::spawn(async move {
            let mut s = req.into_inner();
            let mut boot_mode = true;
            while let Some(pkt) = s.next().await.transpose()? {
                loop {
                    match ms.lock().await.write_all(&pkt.data).await {
                        Ok(()) => {
                            trace!("🖱️  wrote {}", pkt.data.iter()
                                   .map(|b| format!("{b:02X}")).collect::<Vec<_>>().join(" "));
                            break;
                        }
                        Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                            tokio::time::sleep(Duration::from_micros(500)).await;
                        }
                        Err(e) => return Err(Status::internal(format!("hidg1: {e}"))),
                    }
                }
                let _ = tx.send(Ok(pkt)).await;
            }
            Ok::<(), Status>(())
        });

        Ok(Response::new(ReceiverStream::new(rx)))
    }

    async fn capture_video(
        &self,
        req: Request<MonitorRequest>,
    ) -> Result<Response<Self::CaptureVideoStream>, Status> {
        let r = req.into_inner();

        // let devs = loop {
        //     let list = list_gc311_devices()
        //         .map_err(|e| Status::internal(format!("enum v4l2: {e}")))?;
        //     if !list.is_empty() { break list; }
        //     tokio::time::sleep(Duration::from_secs(1)).await;
        // };

        // let dev = devs
        //     .get(r.id as usize)
        //     .ok_or_else(|| Status::invalid_argument(format!("monitor id {} absent", r.id)))?
        //     .to_owned();
    
        let dev = match r.id {
            0 => "/dev/lesavka_l_eye",
            1 => "/dev/lesavka_r_eye",
            _ => return Err(Status::invalid_argument("monitor id must be 0 or 1")),
        }
        .to_string();

        info!("🎥 streaming {dev} at ≤{} kb/s", r.max_bitrate);

        let s = video::spawn_camera(&dev, r.id, r.max_bitrate)
            .await
            .map_err(|e| Status::internal(format!("{e:#?}")))?;

        Ok(Response::new(Box::pin(s) as _))
    }
}

/*─────────────────── main ──────────────────────────────*/
#[tokio::main(flavor = "multi_thread", worker_threads = 4)]
async fn main() -> anyhow::Result<()> {
    /* logging */
    fmt().with_env_filter(
        EnvFilter::try_from_default_env()
            .unwrap_or_else(|_| EnvFilter::new("lesavka_server=info")),
    )
    .init();

    /* auto‑cycle task */
    // tokio::spawn(async { monitor_gc311_disconnect().await.ok(); });

    let gadget = UsbGadget::new("lesavka");
    let handler = Handler::make(gadget.clone()).await?;

    // tokio::spawn({
    //     let gadget = gadget.clone();
    //     async move {
    //         loop {
    //             tokio::time::sleep(Duration::from_secs(4)).await;
    //             if LAST_HID_WRITE.elapsed().as_secs() > 3 {
    //                 warn!("no HID traffic in 3 s – cycling UDC");
    //                 let _ = gadget.cycle();
    //             }
    //         }
    //     }
    // });

    println!("🌐 lesavka-server listening on 0.0.0.0:50051");
    Server::builder()
        .add_service(RelayServer::new(handler))
        .serve(([0, 0, 0, 0], 50051).into())
        .await?;
    Ok(())
}