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

use std::{io::ErrorKind, pin::Pin, sync::Arc, panic::AssertUnwindSafe};
use std::time::Duration;
use tokio::{fs::{File, OpenOptions}, io::AsyncWriteExt, sync::Mutex};
use tokio_stream::{wrappers::ReceiverStream, Stream, StreamExt};
use tonic::{transport::Server, Request, Response, Status};
use tracing::{error, info, trace, warn, debug};
use tracing_subscriber::{fmt, EnvFilter};
use futures_util::FutureExt;

use navka_common::navka::{
    relay_server::{Relay, RelayServer},
    KeyboardReport, MouseReport,
};

struct Handler {
    kb: Arc<Mutex<tokio::fs::File>>,
    ms: Arc<Mutex<tokio::fs::File>>,
}

#[tonic::async_trait]
impl Relay for Handler {
    type StreamKeyboardStream = ReceiverStream<Result<KeyboardReport, Status>>;
    type StreamMouseStream    = ReceiverStream<Result<MouseReport, Status>>;

    async fn stream_keyboard(
        &self,
        req: Request<tonic::Streaming<KeyboardReport>>,
    ) -> Result<Response<Self::StreamKeyboardStream>, Status> {
        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?;
                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)))
    }
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    fmt().with_env_filter(
        // honour RUST_LOG but fall back to very chatty defaults
        EnvFilter::try_from_default_env().unwrap_or_else(|_| //{
            EnvFilter::new(
                // "navka_client=trace,\
                //  navka_server=trace,\
                //  tonic=debug,\
                //  h2=debug,\
                //  tower=debug",
                "navka_server=info"
            )
        //}
        ),
    )
    // .with_target(true)
    // .with_thread_ids(true)
    // .with_file(true)
    .init();

    let kb = OpenOptions::new()
        .write(true)
        // .read(true)
        // .custom_flags(libc::O_NONBLOCK)
        .open("/dev/hidg0")
        .await?;

    let ms = OpenOptions::new()
        .write(true)
        // .read(true)
        .custom_flags(libc::O_NONBLOCK)
        .open("/dev/hidg1")
        .await?;

    let handler = Handler {
        kb: Arc::new(Mutex::new(kb)),
        ms: Arc::new(Mutex::new(ms)),
    };

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