xoq 0.3.6

//! socketcan-compatible interface to remote CAN sockets over P2P.
//!
//! This module provides a `socketcan` crate compatible API for connecting
//! to remote CAN interfaces over iroh P2P or MoQ relay.
//!
//! # Example
//!
//! ```no_run
//! use xoq::socketcan;
//!
//! let mut socket = socketcan::new("server-endpoint-id").open()?;
//! let frame = socketcan::CanFrame::new(0x123, &[1, 2, 3, 4])?;
//! socket.write_frame(&frame)?;
//! let received = socket.read_frame()?;
//! println!("Received: ID={:x} Data={:?}", received.id(), received.data());
//! # Ok::<(), anyhow::Error>(())
//! ```

use anyhow::Result;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;

use crate::can_types::wire;
use crate::iroh::{IrohClientBuilder, IrohConnection};

// Re-export frame types for convenience
pub use crate::can_types::{CanFdFlags, CanInterfaceInfo};

/// A client that connects to a remote CAN interface over iroh P2P
pub struct CanClient {
    send: Arc<Mutex<iroh::endpoint::SendStream>>,
    recv: Arc<Mutex<iroh::endpoint::RecvStream>>,
    _conn: IrohConnection,
}

impl CanClient {
    /// Connect to a remote CAN bridge server
    pub async fn connect(server_id: &str) -> Result<Self> {
        tracing::debug!("Connecting to CAN server: {}", server_id);
        let conn = IrohClientBuilder::new()
            .connect_str(server_id)
            .await
            .map_err(|e| anyhow::anyhow!("Failed to connect to server: {}", e))?;
        tracing::debug!("Connected to server, opening stream...");
        let stream = conn
            .open_stream()
            .await
            .map_err(|e| anyhow::anyhow!("Failed to open stream: {}", e))?;
        tracing::debug!("Stream opened successfully");

        // Split stream so reads and writes don't block each other
        let (send, recv) = stream.split();

        Ok(Self {
            send: Arc::new(Mutex::new(send)),
            recv: Arc::new(Mutex::new(recv)),
            _conn: conn,
        })
    }

    /// Write a CAN frame to the remote interface
    pub async fn write_frame(&self, frame: &CanFrame) -> Result<()> {
        let encoded = wire::encode(&AnyCanFrame::Can(frame.clone()));
        let mut send = self.send.lock().await;
        send.write_all(&encoded).await?;
        drop(send);
        tokio::task::yield_now().await;
        Ok(())
    }

    /// Write a CAN FD frame to the remote interface
    pub async fn write_fd_frame(&self, frame: &CanFdFrame) -> Result<()> {
        let encoded = wire::encode(&AnyCanFrame::CanFd(frame.clone()));
        let mut send = self.send.lock().await;
        send.write_all(&encoded).await?;
        drop(send);
        tokio::task::yield_now().await;
        Ok(())
    }

    /// Read a CAN frame from the remote interface
    pub async fn read_frame(&self) -> Result<Option<AnyCanFrame>> {
        let mut buf = [0u8; wire::FRAME_SIZE];
        let mut recv = self.recv.lock().await;

        let mut read = 0;
        while read < wire::FRAME_SIZE {
            match recv.read(&mut buf[read..]).await? {
                Some(n) if n > 0 => read += n,
                Some(_) => continue,
                None => return Ok(None),
            }
        }

        let (frame, _) = wire::decode(&buf)?;
        Ok(Some(frame))
    }
}

/// Transport type for the CAN socket connection.
#[derive(Clone)]
pub enum Transport {
    /// Iroh P2P connection (default)
    Iroh {
        /// Custom ALPN protocol
        alpn: Option<Vec<u8>>,
        /// Custom relay URL (None = use default iroh relays)
        relay_url: Option<String>,
    },
    /// MoQ relay connection
    Moq {
        /// Relay URL
        relay: String,
        /// Disable TLS verification (for self-signed certs)
        insecure: bool,
    },
}

impl Default for Transport {
    fn default() -> Self {
        Transport::Iroh {
            alpn: None,
            relay_url: None,
        }
    }
}

/// Builder for creating a remote CAN socket connection.
///
/// Mimics a builder pattern similar to `socketcan` for drop-in compatibility.
///
/// # Example
///
/// ```no_run
/// use xoq::socketcan;
/// use std::time::Duration;
///
/// // Simple iroh P2P connection (default)
/// let socket = socketcan::new("server-endpoint-id").open()?;
///
/// // With timeout
/// let socket = socketcan::new("server-endpoint-id")
///     .timeout(Duration::from_millis(500))
///     .open()?;
///
/// // With CAN FD enabled
/// let socket = socketcan::new("server-endpoint-id")
///     .enable_fd(true)
///     .open()?;
/// # Ok::<(), anyhow::Error>(())
/// ```
pub struct CanSocketBuilder {
    server_id: String,
    timeout: Duration,
    transport: Transport,
    fd_enabled: bool,
}

impl CanSocketBuilder {
    /// Create a new CAN socket builder.
    pub fn new(server_id: &str) -> Self {
        Self {
            server_id: server_id.to_string(),
            timeout: Duration::from_secs(1),
            transport: Transport::default(),
            fd_enabled: false,
        }
    }

    /// Set the read/write timeout.
    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }

    /// Enable CAN FD support.
    pub fn enable_fd(mut self, enabled: bool) -> Self {
        self.fd_enabled = enabled;
        self
    }

    /// Use iroh P2P transport (default).
    pub fn with_iroh(mut self) -> Self {
        self.transport = Transport::Iroh {
            alpn: None,
            relay_url: None,
        };
        self
    }

    /// Set custom ALPN for iroh connection.
    pub fn alpn(mut self, alpn: &[u8]) -> Self {
        if let Transport::Iroh {
            alpn: ref mut a, ..
        } = self.transport
        {
            *a = Some(alpn.to_vec());
        }
        self
    }

    /// Set custom iroh relay URL (for self-hosted iroh relay).
    pub fn iroh_relay(mut self, url: &str) -> Self {
        if let Transport::Iroh {
            relay_url: ref mut r,
            ..
        } = self.transport
        {
            *r = Some(url.to_string());
        }
        self
    }

    /// Use MoQ relay transport.
    pub fn with_moq(mut self, relay: &str) -> Self {
        self.transport = Transport::Moq {
            relay: relay.to_string(),
            insecure: false,
        };
        self
    }

    /// Disable TLS verification (for self-signed certs on MoQ relay).
    pub fn insecure(mut self, val: bool) -> Self {
        if let Transport::Moq {
            insecure: ref mut i,
            ..
        } = self.transport
        {
            *i = val;
        }
        self
    }

    /// Open the connection to the remote CAN interface.
    pub fn open(self) -> Result<RemoteCanSocket> {
        let runtime = tokio::runtime::Runtime::new()?;

        let client = match self.transport {
            Transport::Iroh { alpn, relay_url } => runtime.block_on(async {
                let mut builder = IrohClientBuilder::new();
                if let Some(alpn) = alpn {
                    builder = builder.alpn(&alpn);
                }
                if let Some(url) = relay_url {
                    builder = builder.relay_url(url);
                }
                let conn = builder.connect_str(&self.server_id).await?;
                let stream = conn.open_stream().await?;
                // Split for true full-duplex: reads and writes don't block each other
                let (send, recv) = stream.split();
                Ok::<_, anyhow::Error>(ClientInner::Iroh {
                    send: Arc::new(Mutex::new(send)),
                    recv: Arc::new(Mutex::new(recv)),
                    _conn: Box::new(conn),
                })
            })?,
            Transport::Moq { relay, insecure } => runtime.block_on(async {
                let mut builder = crate::moq::MoqBuilder::new().relay(&relay);
                if insecure {
                    builder = builder.disable_tls_verify();
                }

                // Client publishes commands on {path}/commands, subscribes to state on {path}/state
                let cmd_builder = builder
                    .clone()
                    .path(&format!("{}/commands", self.server_id));
                let state_builder = builder.path(&format!("{}/state", self.server_id));

                let (pub_result, sub_result) = tokio::join!(
                    cmd_builder.connect_publisher_with_track("can"),
                    Self::connect_and_subscribe_retry(state_builder, "can")
                );

                let (publisher, cmd_writer) = pub_result?;
                let (subscriber, state_reader) = sub_result?;

                Ok::<_, anyhow::Error>(ClientInner::Moq {
                    cmd_writer: Arc::new(Mutex::new(cmd_writer)),
                    state_reader: Arc::new(Mutex::new(state_reader)),
                    _publisher: Box::new(publisher),
                    _subscriber: Box::new(subscriber),
                })
            })?,
        };

        Ok(RemoteCanSocket {
            client,
            runtime,
            server_id: self.server_id,
            timeout: self.timeout,
            fd_enabled: self.fd_enabled,
            read_buffer: std::sync::Mutex::new(Vec::new()),
        })
    }
}

/// Internal client representation supporting multiple transports.
enum ClientInner {
    Iroh {
        // Split stream for true full-duplex: reads and writes don't block each other
        send: Arc<Mutex<iroh::endpoint::SendStream>>,
        recv: Arc<Mutex<iroh::endpoint::RecvStream>>,
        _conn: Box<IrohConnection>,
    },
    Moq {
        cmd_writer: Arc<Mutex<crate::moq::MoqTrackWriter>>,
        state_reader: Arc<Mutex<crate::moq::MoqTrackReader>>,
        _publisher: Box<crate::moq::MoqPublisher>,
        _subscriber: Box<crate::moq::MoqSubscriber>,
    },
}

impl CanSocketBuilder {
    /// Connect subscriber with retry loop — keeps reconnecting until broadcast is found.
    async fn connect_and_subscribe_retry(
        builder: crate::moq::MoqBuilder,
        track_name: &str,
    ) -> anyhow::Result<(crate::moq::MoqSubscriber, crate::moq::MoqTrackReader)> {
        let track = track_name.to_string();
        loop {
            let mut subscriber = builder.clone().connect_subscriber().await?;

            match tokio::time::timeout(Duration::from_secs(2), subscriber.subscribe_track(&track))
                .await
            {
                Ok(Ok(Some(reader))) => return Ok((subscriber, reader)),
                Ok(Ok(None)) => {
                    tracing::debug!("Broadcast ended, reconnecting...");
                }
                Ok(Err(e)) => {
                    tracing::debug!("Subscribe error: {}, reconnecting...", e);
                }
                Err(_) => {
                    tracing::debug!("No broadcast yet, reconnecting...");
                }
            }

            tokio::time::sleep(Duration::from_millis(500)).await;
        }
    }
}

/// Create a new remote CAN socket builder.
///
/// This function provides a familiar API for CAN socket creation.
///
/// # Example
///
/// ```no_run
/// use xoq::socketcan;
///
/// // Connect to remote CAN interface
/// let socket = socketcan::new("server-endpoint-id").open()?;
/// # Ok::<(), anyhow::Error>(())
/// ```
pub fn new(server_id: &str) -> CanSocketBuilder {
    CanSocketBuilder::new(server_id)
}

/// A remote CAN socket that provides a blocking API.
///
/// This struct mimics the `socketcan` crate's interface for drop-in compatibility.
///
/// # Example
///
/// ```no_run
/// use xoq::socketcan;
///
/// // Iroh P2P (default)
/// let mut socket = socketcan::new("server-endpoint-id").open()?;
///
/// // Write a frame
/// let frame = socketcan::CanFrame::new(0x123, &[1, 2, 3])?;
/// socket.write_frame(&frame)?;
///
/// // Read a frame
/// if let Some(frame) = socket.read_frame()? {
///     println!("ID: {:x}, Data: {:?}", frame.id(), frame.data());
/// }
/// # Ok::<(), anyhow::Error>(())
/// ```
pub struct RemoteCanSocket {
    client: ClientInner,
    runtime: tokio::runtime::Runtime,
    server_id: String,
    timeout: Duration,
    fd_enabled: bool,
    // Use Mutex for interior mutability so trait methods with &self can update buffer
    read_buffer: std::sync::Mutex<Vec<u8>>,
}

impl RemoteCanSocket {
    /// Open a connection to a remote CAN interface via iroh P2P.
    ///
    /// Prefer using `xoq::socketcan::new(server_id).open()` for more options.
    pub fn open(server_id: &str) -> Result<Self> {
        new(server_id).open()
    }

    /// Get the server ID.
    pub fn server_id(&self) -> &str {
        &self.server_id
    }

    /// Get the current timeout.
    pub fn timeout(&self) -> Duration {
        self.timeout
    }

    /// Set the read/write timeout.
    pub fn set_timeout(&mut self, timeout: Duration) -> Result<()> {
        self.timeout = timeout;
        Ok(())
    }

    /// Check if CAN FD mode is enabled.
    pub fn is_fd_enabled(&self) -> bool {
        self.fd_enabled
    }

    /// Write a standard CAN frame to the remote interface.
    pub fn write_frame(&mut self, frame: &CanFrame) -> Result<()> {
        let encoded = wire::encode(&AnyCanFrame::Can(frame.clone()));
        self.write_raw(&encoded)
    }

    /// Write a CAN FD frame to the remote interface.
    pub fn write_fd_frame(&mut self, frame: &CanFdFrame) -> Result<()> {
        let encoded = wire::encode(&AnyCanFrame::CanFd(frame.clone()));
        self.write_raw(&encoded)
    }

    /// Write any CAN frame to the remote interface.
    pub fn write_any_frame(&mut self, frame: &AnyCanFrame) -> Result<()> {
        let encoded = wire::encode(frame);
        self.write_raw(&encoded)
    }

    fn write_raw(&mut self, data: &[u8]) -> Result<()> {
        self.runtime.block_on(async {
            match &self.client {
                ClientInner::Iroh { send, .. } => {
                    let mut s = send.lock().await;
                    s.write_all(data).await?;
                    drop(s);
                    // quinn's flush() is a no-op — yield to let the connection
                    // task pack the data into a UDP packet before we return
                    tokio::time::sleep(std::time::Duration::from_micros(100)).await;
                }
                ClientInner::Moq { cmd_writer, .. } => {
                    let mut w = cmd_writer.lock().await;
                    w.write_stream(data.to_vec());
                }
            }
            Ok::<_, anyhow::Error>(())
        })
    }

    /// Read a CAN frame from the remote interface.
    ///
    /// Returns `None` if the connection is closed or timeout occurs.
    pub fn read_frame(&mut self) -> Result<Option<AnyCanFrame>> {
        // First check if we have a complete frame in the buffer
        if let Some(frame) = self.try_decode_buffered()? {
            return Ok(Some(frame));
        }

        // Read more data with timeout
        let data = self.read_raw_with_timeout()?;
        if let Some(data) = data {
            {
                let mut buffer = self.read_buffer.lock().unwrap();
                buffer.extend_from_slice(&data);
            }
            self.try_decode_buffered()
        } else {
            Ok(None)
        }
    }

    fn read_raw_with_timeout(&mut self) -> Result<Option<Vec<u8>>> {
        let timeout = self.timeout;
        let result = self.runtime.block_on(async {
            // No flush needed — quinn's flush() is a no-op.
            // Writes are sent by the connection task asynchronously.
            tokio::time::timeout(timeout, async {
                let mut temp_buf = vec![0u8; 128];
                match &self.client {
                    ClientInner::Iroh { recv, .. } => {
                        let mut r = recv.lock().await;
                        match r.read(&mut temp_buf).await? {
                            Some(n) => {
                                temp_buf.truncate(n);
                                Ok(Some(temp_buf))
                            }
                            None => Ok(None),
                        }
                    }
                    ClientInner::Moq { state_reader, .. } => {
                        let mut r = state_reader.lock().await;
                        if let Some(data) = r.read().await? {
                            Ok(Some(data.to_vec()))
                        } else {
                            Ok(None)
                        }
                    }
                }
            })
            .await
        });

        match result {
            Ok(Ok(data)) => Ok(data),
            Ok(Err(e)) => Err(e),
            Err(_) => Ok(None), // Timeout
        }
    }

    fn try_decode_buffered(&self) -> Result<Option<AnyCanFrame>> {
        let mut buffer = self.read_buffer.lock().unwrap();

        if buffer.len() < wire::FRAME_SIZE {
            return Ok(None);
        }

        match wire::decode(&buffer) {
            Ok((frame, consumed)) => {
                buffer.drain(..consumed);
                Ok(Some(frame))
            }
            Err(e) => {
                tracing::warn!("CAN frame decode error: {}, skipping frame to resync", e);
                buffer.drain(..wire::FRAME_SIZE);
                Ok(None)
            }
        }
    }

    /// Read frames continuously, calling the provided callback for each frame.
    pub fn read_frames<F>(&mut self, mut callback: F) -> Result<()>
    where
        F: FnMut(AnyCanFrame) -> bool,
    {
        loop {
            match self.read_frame()? {
                Some(frame) => {
                    if !callback(frame) {
                        break;
                    }
                }
                None => continue, // Timeout, keep trying
            }
        }
        Ok(())
    }
}

// Re-export for public API
pub use crate::can_types::AnyCanFrame;
pub use crate::can_types::CanBusSocket;
pub use crate::can_types::CanFdFrame;
pub use crate::can_types::CanFrame;

impl CanBusSocket for RemoteCanSocket {
    fn is_open(&self) -> bool {
        // Remote socket is always considered open once connected
        true
    }

    fn write_raw(&self, can_id: u32, data: &[u8]) -> anyhow::Result<()> {
        let frame = CanFrame::new(can_id, data)?;
        // Need mutable access, but trait requires &self for Send+Sync compatibility.
        // We use interior mutability via the runtime's block_on.
        let encoded = wire::encode(&AnyCanFrame::Can(frame));
        self.runtime.block_on(async {
            match &self.client {
                ClientInner::Iroh { send, .. } => {
                    let mut s = send.lock().await;
                    s.write_all(&encoded).await?;
                    drop(s);
                    // quinn's flush() is a no-op — yield to let connection task send
                    tokio::time::sleep(std::time::Duration::from_micros(100)).await;
                }
                ClientInner::Moq { cmd_writer, .. } => {
                    let mut w = cmd_writer.lock().await;
                    w.write_stream(encoded.to_vec());
                }
            }
            Ok::<_, anyhow::Error>(())
        })
    }

    fn read_raw(&self) -> anyhow::Result<Option<(u32, Vec<u8>)>> {
        // First check if we already have a complete frame in the buffer
        if let Some(frame) = self.try_decode_buffered()? {
            return Ok(Some((frame.id(), frame.data().to_vec())));
        }

        // Read more data with timeout
        let timeout = self.timeout;
        let result = self.runtime.block_on(async {
            tokio::time::timeout(timeout, async {
                let mut temp_buf = vec![0u8; 128];
                match &self.client {
                    ClientInner::Iroh { recv, .. } => {
                        let mut r = recv.lock().await;
                        match r.read(&mut temp_buf).await? {
                            Some(n) => {
                                temp_buf.truncate(n);
                                Ok(Some(temp_buf))
                            }
                            None => Ok(None),
                        }
                    }
                    ClientInner::Moq { state_reader, .. } => {
                        let mut r = state_reader.lock().await;
                        if let Some(data) = r.read().await? {
                            Ok(Some(data.to_vec()))
                        } else {
                            Ok(None)
                        }
                    }
                }
            })
            .await
        });

        match result {
            Ok(Ok(Some(data))) => {
                // Add to buffer and try to decode
                {
                    let mut buffer = self.read_buffer.lock().unwrap();
                    buffer.extend_from_slice(&data);
                }
                // Try to decode a complete frame
                if let Some(frame) = self.try_decode_buffered()? {
                    Ok(Some((frame.id(), frame.data().to_vec())))
                } else {
                    Ok(None) // Partial frame, need more data
                }
            }
            Ok(Ok(None)) => Ok(None), // Stream closed
            Ok(Err(e)) => Err(e),
            Err(_) => Ok(None), // Timeout
        }
    }

    fn is_data_available(&self, _timeout_us: u64) -> anyhow::Result<bool> {
        let buffer = self.read_buffer.lock().unwrap();
        Ok(buffer.len() >= wire::FRAME_SIZE)
    }

    fn set_recv_timeout(&mut self, timeout_us: u64) -> anyhow::Result<()> {
        self.timeout = std::time::Duration::from_micros(timeout_us);
        Ok(())
    }
}