nwep/

server.rs

#![allow(unsafe_op_in_unsafe_fn)]

use crate::anchorserver::AnchorServer;
use crate::error::{Error, check};
use crate::ffi;
use crate::sock_compat::{AF_INET, AF_INET6, sa_family_t, sockaddr_in, sockaddr_in6};
use crate::keypair::Keypair;
use crate::logserver::LogServer;
use crate::msg::CRequest;
use crate::role::ServerRole;
use crate::types::{
    DEFAULT_MAX_MESSAGE_SIZE, DEFAULT_MAX_STREAMS, DEFAULT_TIMEOUT, Header, Identity, NodeId,
    SECONDS,
};
use std::collections::{HashMap, HashSet};
use std::ffi::CString;
use std::net::{SocketAddr, UdpSocket};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, mpsc};
use std::time::Duration;

const UDP_BUF_SIZE: usize = 65536;

/// `Settings` controls transport-level parameters for a NWEP server.
///
/// All fields have sensible defaults via [`Settings::default`]; only override what you need.
pub struct Settings {
    /// Maximum number of concurrent QUIC streams per connection.
    pub max_streams: u32,
    /// Maximum allowed size of a single NWEP message body, in bytes.
    pub max_message_size: u32,
    /// QUIC idle timeout in milliseconds.
    ///
    /// After this period of inactivity the connection is dropped by the transport layer.
    pub timeout_ms: u32,
    /// Optional compression algorithm name (e.g. `"zstd"`).  Empty string means no compression.
    pub compression: String,
    /// Role advertised to connecting peers (e.g. `ServerRole::Regular` or `ServerRole::Bootstrap`).
    pub role: ServerRole,
}

impl Default for Settings {
    fn default() -> Self {
        Settings {
            max_streams: DEFAULT_MAX_STREAMS,
            max_message_size: DEFAULT_MAX_MESSAGE_SIZE as u32,
            timeout_ms: (DEFAULT_TIMEOUT / (SECONDS / 1000)) as u32,
            compression: String::new(),
            role: ServerRole::Regular,
        }
    }
}

/// `NotifyOptions` carries optional metadata for a server-initiated push notification.
///
/// Pass this to [`Server::notify_with_options`] when you need to attach custom headers or
/// a stable notification identifier to the push message.
pub struct NotifyOptions {
    /// Custom headers to include in the notification message.
    pub headers: Vec<Header>,
    /// Optional 16-byte notification identifier.
    ///
    /// When `Some`, the C library sets `has_notify_id = 1` in the wire message so the
    /// peer can deduplicate retransmitted notifications.
    pub notify_id: Option<[u8; 16]>,
}

/// `ConnInfo` describes a peer that has connected to (or disconnected from) the server.
///
/// It is delivered to the `on_connect` and `on_disconnect` callbacks registered on
/// [`ServerBuilder`].
#[derive(Clone, Debug)]
pub struct ConnInfo {
    /// The 32-byte node identifier of the peer, derived from its Ed25519 public key.
    pub node_id: NodeId,
    /// The full Ed25519 identity of the peer (public key + node ID).
    pub peer_identity: Identity,
    /// The role the peer advertised during the handshake.
    pub role: ServerRole,
}

enum ServerEvent {
    Packet {
        data: Vec<u8>,
        local: SocketAddr,
        remote: SocketAddr,
    },
    TimerExpiry,
    Notify {
        peer_node_id: NodeId,
        event: String,
        path: String,
        body: Vec<u8>,
        options: Option<NotifyOptions>,
    },
    NotifyAll {
        event: String,
        path: String,
        body: Vec<u8>,
    },
    CloseConn {
        peer_node_id: NodeId,
        error: i32,
    },
    Shutdown,
}

/// `Request` represents an inbound NWEP request delivered to a [`Handler`].
///
/// The request is constructed by the event loop from the C library's `nwep_request` struct and
/// passed (together with a [`ResponseWriter`]) to [`Handler::serve`].
pub struct Request {
    /// NWEP method string (e.g. `"read"` or `"write"`).
    pub method: String,
    /// Request path (e.g. `"/hello"`).
    pub path: String,
    /// Request body bytes.
    pub body: Vec<u8>,
    /// 16-byte opaque request identifier generated by the client.
    pub request_id: [u8; 16],
    /// 16-byte opaque trace identifier for distributed tracing.
    pub trace_id: [u8; 16],
    /// Custom headers sent by the client.
    pub headers: Vec<Header>,
    /// Identity and role of the peer that sent this request.
    pub conn: ConnInfo,
}

impl Request {
    /// `header` looks up a request header by name, returning its value if present.
    pub fn header(&self, name: &str) -> Option<&str> {
        self.headers
            .iter()
            .find(|h| h.name == name)
            .map(|h| h.value.as_str())
    }
}

/// `Handler` is implemented by any type that can process an inbound NWEP request.
///
/// The event loop invokes [`Handler::serve`] once per request, on the event loop thread.
/// Closures of type `Fn(&mut ResponseWriter, &Request) + Send + 'static` implement this
/// trait automatically, so you can pass a closure wherever a `Handler` is expected.
pub trait Handler: Send + 'static {
    /// `serve` processes a single request and writes the response through `w`.
    ///
    /// If `serve` returns without having called any write method on `w`, the event loop
    /// automatically sends an empty `"ok"` response so the stream is always closed cleanly.
    fn serve(&self, w: &mut ResponseWriter, r: &Request);
}

impl<F: Fn(&mut ResponseWriter, &Request) + Send + 'static> Handler for F {
    fn serve(&self, w: &mut ResponseWriter, r: &Request) {
        self(w, r)
    }
}

/// `ResponseWriter` is the handle through which a [`Handler`] writes its response.
///
/// Two response modes are supported:
///
/// - **Buffered**: call [`set_status`](ResponseWriter::set_status) / [`set_header`](ResponseWriter::set_header)
///   then [`write`](ResponseWriter::write) (or the shorthand [`respond`](ResponseWriter::respond)).
///   This sends headers + body atomically and closes the stream.
/// - **Streaming**: call [`stream_write`](ResponseWriter::stream_write) one or more times, then
///   [`stream_end`](ResponseWriter::stream_end) (or [`stream_close`](ResponseWriter::stream_close)
///   on error).  Response headers must have been sent by a prior buffered call in this case, or
///   the stream is left headerless.
pub struct ResponseWriter {
    pub(crate) stream: *mut ffi::nwep_stream,
    pub(crate) status: String,
    pub(crate) headers: Vec<Header>,
    pub(crate) sent: bool,
}

impl ResponseWriter {
    /// `set_status` sets the NWEP status string for the pending buffered response.
    ///
    /// The default status is `"ok"`.  Must be called before [`write`](ResponseWriter::write).
    pub fn set_status(&mut self, status: &str) {
        self.status = status.to_string();
    }

    /// `set_header` appends a response header to the pending buffered response.
    pub fn set_header(&mut self, name: &str, value: &str) {
        self.headers.push(Header::new(name, value));
    }

    /// `write` sends the buffered response headers and body, then closes the stream.
    ///
    /// If called more than once the subsequent calls are no-ops; only the first call
    /// transmits data.  Use [`stream_write`](ResponseWriter::stream_write) /
    /// [`stream_end`](ResponseWriter::stream_end) for incremental streaming.
    ///
    /// # Errors
    ///
    /// Returns an [`Error`] if the underlying `nwep_stream_respond` or `nwep_stream_end`
    /// C call fails.
    pub fn write(&mut self, body: &[u8]) -> Result<(), Error> {
        if self.sent {
            return Ok(());
        }
        self.sent = true;
        let status = CString::new(self.status.as_str()).unwrap_or_default();
        let c_headers: Vec<ffi::nwep_header> = self
            .headers
            .iter()
            .map(|h| ffi::nwep_header {
                name: h.name.as_ptr(),
                name_len: h.name.len(),
                value: h.value.as_ptr(),
                value_len: h.value.len(),
            })
            .collect();
        let resp = ffi::nwep_response {
            status: status.as_ptr(),
            status_len: self.status.len(),
            status_details: std::ptr::null(),
            status_details_len: 0,
            headers: if c_headers.is_empty() {
                std::ptr::null()
            } else {
                c_headers.as_ptr()
            },
            header_count: c_headers.len(),
            body: if body.is_empty() {
                std::ptr::null()
            } else {
                body.as_ptr()
            },
            body_len: body.len(),
        };
        check(unsafe { ffi::nwep_stream_respond(self.stream, &resp) })?;
        check(unsafe { ffi::nwep_stream_end(self.stream) })?;
        Ok(())
    }

    /// `respond` sets `status` and writes `body` in a single call.
    ///
    /// Equivalent to calling [`set_status`](ResponseWriter::set_status) followed by
    /// [`write`](ResponseWriter::write).
    ///
    /// # Errors
    ///
    /// Returns an [`Error`] if the underlying send fails.
    pub fn respond(&mut self, status: &str, body: &[u8]) -> Result<(), Error> {
        self.set_status(status);
        self.write(body)
    }

    /// `stream_write` writes a chunk of body data directly to the open QUIC stream.
    ///
    /// This is the low-level building block for streaming responses.  Call
    /// [`stream_end`](ResponseWriter::stream_end) when all chunks have been written.
    /// Returns the number of bytes written.
    ///
    /// # Errors
    ///
    /// Returns an [`Error`] if `nwep_stream_write` returns a negative result code.
    pub fn stream_write(&mut self, data: &[u8]) -> Result<isize, Error> {
        let n = unsafe { ffi::nwep_stream_write(self.stream, data.as_ptr(), data.len()) };
        if n < 0 {
            Err(Error::from_code(n as i32))
        } else {
            Ok(n)
        }
    }

    /// `stream_end` signals the end of the response body and closes the QUIC stream cleanly.
    ///
    /// # Errors
    ///
    /// Returns an [`Error`] if `nwep_stream_end` fails.
    pub fn stream_end(&mut self) -> Result<(), Error> {
        check(unsafe { ffi::nwep_stream_end(self.stream) })
    }

    /// `stream_close` aborts the stream with an application error code.
    ///
    /// Use this to signal an error mid-stream without sending a complete response.
    /// Unlike [`stream_end`](ResponseWriter::stream_end), this does not return a result
    /// because the close is best-effort.
    pub fn stream_close(&mut self, err: i32) {
        unsafe { ffi::nwep_stream_close(self.stream, err) }
    }

    /// `stream_id` returns the QUIC stream identifier for this response stream.
    ///
    /// Useful for correlating log output or for advanced stream management.
    pub fn stream_id(&self) -> i64 {
        unsafe { ffi::nwep_stream_get_id(self.stream) }
    }

    /// `is_server_initiated` returns `true` if this stream was opened by the server.
    ///
    /// Server-initiated streams correspond to notify messages; client-initiated streams
    /// correspond to ordinary requests.
    pub fn is_server_initiated(&self) -> bool {
        unsafe { ffi::nwep_stream_is_server_initiated(self.stream) != 0 }
    }
}

// All fields are owned directly; no double-indirection via raw pointers.
// This is exclusively accessed from the event loop thread (callbacks fire there too).
struct CallbackData {
    handler: Box<dyn Handler>,
    on_connect: Option<Box<dyn Fn(ConnInfo) + Send>>,
    on_disconnect: Option<Box<dyn Fn(ConnInfo, i32) + Send>>,
    // Keyed by NodeId; conn pointers are only valid while the peer is connected.
    // Maintained by server_on_connect / server_on_disconnect.
    conns: HashMap<NodeId, *mut ffi::nwep_conn>,
    // Reverse map: conn pointer (as usize) → NodeId.  Used in server_on_disconnect to
    // look up the correct NodeId even when nwep_conn_get_peer_identity returns zeroed data.
    conn_to_node_id: HashMap<usize, NodeId>,
    // Connections closed via close_connection() that haven't yet been freed by the C library.
    // server_on_request aborts streams on these connections so no application response is sent
    // before CONNECTION_CLOSE reaches the client.
    closing_conns: HashSet<usize>,
    // Shared with Server so connection_count() / connected_peers() work from any thread.
    peers: Arc<Mutex<Vec<NodeId>>>,
    // Optional sub-servers; if set, intercept /log and /checkpoint requests respectively.
    // Shared with Server so callers can access them after build().
    log_server: Option<Arc<Mutex<LogServer>>>,
    anchor_server: Option<Arc<Mutex<AnchorServer>>>,
}

unsafe extern "C" fn server_on_connect(
    conn: *mut ffi::nwep_conn,
    peer: *const ffi::nwep_identity,
    user_data: *mut std::ffi::c_void,
) -> std::ffi::c_int {
    let cb = &mut *(user_data as *mut CallbackData);
    // The C library populates peer.pubkey but may leave peer.nodeid zeroed.
    // Derive the node_id from the pubkey when that happens.
    let mut identity = if peer.is_null() {
        Identity::default()
    } else {
        Identity::from(*peer)
    };
    if identity.node_id.0 == [0u8; 32] && identity.pubkey != [0u8; 32] {
        let mut nid = ffi::nwep_nodeid { data: [0u8; 32] };
        if ffi::nwep_nodeid_from_pubkey(&mut nid, identity.pubkey.as_ptr()) == 0 {
            identity.node_id = NodeId(nid.data);
        }
    }
    let node_id = NodeId(identity.node_id.0);
    let role = if conn.is_null() {
        ServerRole::Regular
    } else {
        let role_ptr = ffi::nwep_conn_get_role(conn);
        if role_ptr.is_null() {
            ServerRole::Regular
        } else {
            ServerRole::from_str(
                std::ffi::CStr::from_ptr(role_ptr)
                    .to_str()
                    .unwrap_or("regular"),
            )
        }
    };

    // Register conn for Notify and update the shared peer list.
    if !conn.is_null() {
        cb.conns.insert(node_id, conn);
        cb.conn_to_node_id.insert(conn as usize, node_id);
        if let Ok(mut peers) = cb.peers.lock() {
            peers.push(node_id);
        }
    }

    if let Some(cb_fn) = &cb.on_connect {
        cb_fn(ConnInfo {
            node_id,
            peer_identity: identity,
            role,
        });
    }
    0
}

unsafe extern "C" fn server_on_disconnect(
    conn: *mut ffi::nwep_conn,
    error: std::ffi::c_int,
    user_data: *mut std::ffi::c_void,
) {
    let cb = &mut *(user_data as *mut CallbackData);
    // Use the reverse map for a reliable node_id lookup: nwep_conn_get_peer_identity(conn)
    // may return null or zeroed data when called from a CLOSING/DRAINING connection.
    let node_id = match cb.conn_to_node_id.remove(&(conn as usize)) {
        Some(nid) => nid,
        None => {
            // Already handled by close_connection(); clean up closing_conns and skip.
            cb.closing_conns.remove(&(conn as usize));
            return;
        }
    };
    cb.closing_conns.remove(&(conn as usize));
    cb.conns.remove(&node_id);
    if let Ok(mut peers) = cb.peers.lock() {
        peers.retain(|&n| n != node_id);
    }

    if let Some(cb_fn) = &cb.on_disconnect {
        // Build the best available identity, falling back to node_id-only if necessary.
        let peer = if conn.is_null() {
            std::ptr::null()
        } else {
            ffi::nwep_conn_get_peer_identity(conn)
        };
        let identity = if peer.is_null() {
            Identity {
                pubkey: [0u8; 32],
                node_id,
            }
        } else {
            let mut id = Identity::from(*peer);
            if id.node_id.0 == [0u8; 32] {
                id.node_id = node_id;
            }
            id
        };
        cb_fn(
            ConnInfo {
                node_id,
                peer_identity: identity,
                role: ServerRole::Regular,
            },
            error,
        );
    }
}

unsafe extern "C" fn server_on_request(
    conn: *mut ffi::nwep_conn,
    stream: *mut ffi::nwep_stream,
    req: *const ffi::nwep_request,
    user_data: *mut std::ffi::c_void,
) -> std::ffi::c_int {
    let cb = &mut *(user_data as *mut CallbackData);
    if req.is_null() || stream.is_null() {
        return 0;
    }
    // If this connection is being closed, abort the stream without responding.
    // The QUIC CLOSING state will send CONNECTION_CLOSE to the client.
    if !conn.is_null() && cb.closing_conns.contains(&(conn as usize)) {
        ffi::nwep_stream_close(stream, 0);
        return 0;
    }
    let c_req = CRequest::from_ffi(&*req);

    // Intercept requests for registered sub-servers (matching Go routing).
    // The C handle_request functions call nwep_stream_respond but NOT nwep_stream_end.
    if cb.log_server.is_some() || cb.anchor_server.is_some() {
        if let Some(ref ls_arc) = cb.log_server {
            if c_req.path == "/log" || c_req.path.starts_with("/log/") {
                if let Ok(mut ls) = ls_arc.lock() {
                    if ls.handle_request(stream, req).is_err() {
                        sub_server_error_respond(stream, crate::protocol::STATUS_INTERNAL_ERROR);
                    }
                }
                ffi::nwep_stream_end(stream);
                return 0;
            }
        }
        if let Some(ref as_arc) = cb.anchor_server {
            if c_req.path == "/checkpoint" || c_req.path.starts_with("/checkpoint/") {
                if let Ok(mut as_) = as_arc.lock() {
                    if as_.handle_request(stream, req).is_err() {
                        sub_server_error_respond(stream, crate::protocol::STATUS_INTERNAL_ERROR);
                    }
                }
                ffi::nwep_stream_end(stream);
                return 0;
            }
        }
    }

    let peer = if conn.is_null() {
        std::ptr::null()
    } else {
        ffi::nwep_conn_get_peer_identity(conn)
    };
    let mut identity = if peer.is_null() {
        Identity::default()
    } else {
        Identity::from(*peer)
    };
    if identity.node_id.0 == [0u8; 32] && identity.pubkey != [0u8; 32] {
        let mut nid = ffi::nwep_nodeid { data: [0u8; 32] };
        if ffi::nwep_nodeid_from_pubkey(&mut nid, identity.pubkey.as_ptr()) == 0 {
            identity.node_id = NodeId(nid.data);
        }
    }
    let role = if conn.is_null() {
        ServerRole::Regular
    } else {
        let role_ptr = ffi::nwep_conn_get_role(conn);
        if role_ptr.is_null() {
            ServerRole::Regular
        } else {
            ServerRole::from_str(
                std::ffi::CStr::from_ptr(role_ptr)
                    .to_str()
                    .unwrap_or("regular"),
            )
        }
    };
    let conn_info = ConnInfo {
        node_id: NodeId(identity.node_id.0),
        peer_identity: identity,
        role,
    };
    let request = Request {
        method: c_req.method,
        path: c_req.path,
        body: c_req.body,
        request_id: c_req.request_id,
        trace_id: c_req.trace_id,
        headers: c_req.headers,
        conn: conn_info,
    };
    let mut writer = ResponseWriter {
        stream,
        status: crate::protocol::STATUS_OK.to_string(),
        headers: Vec::new(),
        sent: false,
    };
    cb.handler.serve(&mut writer, &request);
    if !writer.sent {
        let _ = writer.write(b"");
    }
    0
}

unsafe extern "C" fn server_rand(
    dest: *mut u8,
    len: usize,
    _user_data: *mut std::ffi::c_void,
) -> std::ffi::c_int {
    let slice = std::slice::from_raw_parts_mut(dest, len);
    match crate::crypto::random_bytes(slice) {
        Ok(()) => 0,
        Err(_) => -1,
    }
}

/// Sends an error response on `stream` for a failed sub-server request.
unsafe fn sub_server_error_respond(stream: *mut ffi::nwep_stream, status: &str) {
    let status_c = CString::new(status).unwrap_or_default();
    let resp = ffi::nwep_response {
        status: status_c.as_ptr(),
        status_len: status.len(),
        status_details: std::ptr::null(),
        status_details_len: 0,
        headers: std::ptr::null(),
        header_count: 0,
        body: std::ptr::null(),
        body_len: 0,
    };
    let _ = ffi::nwep_stream_respond(stream, &resp);
}

/// Calls `nwep_conn_notify` for the given conn and drains writes.
/// All pointers must remain valid for the duration of this call.
unsafe fn do_conn_notify(
    c_server: *mut ffi::nwep_server,
    conn: *mut ffi::nwep_conn,
    notify: &ffi::nwep_notify,
    socket: &UdpSocket,
    write_buf: &mut Vec<u8>,
    ts: u64,
) {
    let mut stream: *mut ffi::nwep_stream = std::ptr::null_mut();
    let rc = ffi::nwep_conn_notify(conn, notify, &mut stream);
    if rc == 0 && !stream.is_null() {
        ffi::nwep_stream_end(stream);
    }
    drain_writes(c_server, socket, write_buf, ts);
}

/// `Server` is the thread-safe handle to a running NWEP server.
///
/// `Server` is obtained from [`ServerBuilder::build`] alongside an [`EventLoop`].  It can be
/// cloned and sent across threads freely; all mutating operations are communicated to the event
/// loop via an internal channel.
pub struct Server {
    event_tx: mpsc::SyncSender<ServerEvent>,
    node_id: NodeId,
    addr: SocketAddr,
    shutdown_flag: Arc<AtomicBool>,
    /// Shared peer list; updated by the event loop's connect/disconnect callbacks.
    peers: Arc<Mutex<Vec<NodeId>>>,
    /// Shared with CallbackData so callers can access sub-servers after build().
    log_server: Option<Arc<Mutex<LogServer>>>,
    anchor_server: Option<Arc<Mutex<AnchorServer>>>,
}

impl Server {
    /// `node_id` returns the server's own 32-byte NWEP node identifier.
    pub fn node_id(&self) -> NodeId {
        self.node_id
    }

    /// `addr` returns the local UDP socket address the server is listening on.
    pub fn addr(&self) -> SocketAddr {
        self.addr
    }

    /// `url` formats a `web://` URL for the given path on this server.
    ///
    /// If the server was bound to an unspecified address (`0.0.0.0` or `::`), the URL uses
    /// the loopback address instead, matching Go's `server.URL()` behaviour.
    ///
    /// # Example
    ///
    /// ```no_run
    /// # let server: nwep::server::Server = unimplemented!();
    /// println!("{}", server.url("/hello")); // e.g. "web://127.0.0.1:<node_id>:8080/hello"
    /// ```
    pub fn url(&self, path: &str) -> String {
        use std::net::{IpAddr, Ipv4Addr};
        // Replace unspecified (0.0.0.0 / ::) with loopback, matching Go's server.URL() behaviour.
        let ip = match self.addr.ip() {
            ip if ip.is_unspecified() => IpAddr::V4(Ipv4Addr::LOCALHOST),
            ip => ip,
        };
        let nwep_addr = match ip {
            IpAddr::V4(v4) => crate::addr::Addr::new_ipv4(v4, self.node_id, self.addr.port()),
            IpAddr::V6(v6) => crate::addr::Addr::new_ipv6(v6, self.node_id, self.addr.port()),
        };
        let url = crate::addr::Url {
            addr: nwep_addr,
            path: path.to_string(),
        };
        url.format()
            .unwrap_or_else(|_| format!("web://[{}]:{}{}", ip, self.addr.port(), path))
    }

    /// `connection_count` returns the number of currently authenticated, connected peers.
    ///
    /// The count is updated synchronously on the event loop thread when a peer connects or
    /// disconnects, so it may lag slightly behind the network state.
    pub fn connection_count(&self) -> usize {
        self.peers.lock().map(|p| p.len()).unwrap_or(0)
    }

    /// `connected_peers` returns a snapshot of the node IDs of all currently connected peers.
    pub fn connected_peers(&self) -> Vec<NodeId> {
        self.peers.lock().map(|p| p.clone()).unwrap_or_default()
    }

    /// `shutdown` signals the event loop to stop accepting new connections and exit.
    ///
    /// This call returns immediately; wait for [`EventLoop::run`] to return to confirm the
    /// server has fully stopped.
    pub fn shutdown(&self) {
        self.shutdown_flag.store(true, Ordering::SeqCst);
        let _ = self.event_tx.send(ServerEvent::Shutdown);
    }

    /// `notify` sends a server-initiated push message to a single connected peer.
    ///
    /// The peer receives the notification in its `on_notify` callback.  `event` is a
    /// free-form event type string; `path` is the resource path; `body` is the payload.
    ///
    /// # Errors
    ///
    /// Returns an error if the internal channel has been closed (i.e. the event loop has
    /// already stopped).
    pub fn notify(
        &self,
        peer_node_id: NodeId,
        event: &str,
        path: &str,
        body: Vec<u8>,
    ) -> Result<(), Error> {
        self.event_tx
            .send(ServerEvent::Notify {
                peer_node_id,
                event: event.to_string(),
                path: path.to_string(),
                body,
                options: None,
            })
            .map_err(|_| Error::from_code(crate::error::ERR_INTERNAL_INVALID_STATE))
    }

    /// `notify_with_options` sends a server-initiated push message with additional metadata.
    ///
    /// Like [`notify`](Server::notify) but also accepts a [`NotifyOptions`] to attach custom
    /// headers or a stable notification identifier to the wire message.
    ///
    /// # Errors
    ///
    /// Returns an error if the internal channel has been closed.
    pub fn notify_with_options(
        &self,
        peer_node_id: NodeId,
        event: &str,
        path: &str,
        body: Vec<u8>,
        options: NotifyOptions,
    ) -> Result<(), Error> {
        self.event_tx
            .send(ServerEvent::Notify {
                peer_node_id,
                event: event.to_string(),
                path: path.to_string(),
                body,
                options: Some(options),
            })
            .map_err(|_| Error::from_code(crate::error::ERR_INTERNAL_INVALID_STATE))
    }

    /// `notify_all` broadcasts a push notification to every currently connected peer.
    ///
    /// The message is sent to the snapshot of connections held by the event loop at the
    /// time it processes the event; peers that disconnect between the call and processing
    /// are silently skipped.
    ///
    /// # Errors
    ///
    /// Returns an error if the internal channel has been closed.
    pub fn notify_all(&self, event: &str, path: &str, body: Vec<u8>) -> Result<(), Error> {
        self.event_tx
            .send(ServerEvent::NotifyAll {
                event: event.to_string(),
                path: path.to_string(),
                body,
            })
            .map_err(|_| Error::from_code(crate::error::ERR_INTERNAL_INVALID_STATE))
    }

    /// `close_connection` forcefully closes the connection to the identified peer.
    ///
    /// The peer is removed from internal state immediately and the `on_disconnect` callback
    /// fires before the QUIC `CONNECTION_CLOSE` is sent.  This is intentional: the C
    /// library's QUIC CLOSING state is asynchronous and the disconnect callback would
    /// otherwise be delayed until the peer sends a packet.
    ///
    /// # Errors
    ///
    /// Returns an error if the internal channel has been closed.
    pub fn close_connection(&self, peer_node_id: NodeId, error: i32) -> Result<(), Error> {
        self.event_tx
            .send(ServerEvent::CloseConn {
                peer_node_id,
                error,
            })
            .map_err(|_| Error::from_code(crate::error::ERR_INTERNAL_INVALID_STATE))
    }

    /// `log_server` returns the shared [`LogServer`] registered during build, if any.
    ///
    /// The returned `Arc<Mutex<LogServer>>` is the same instance intercepting `/log/*`
    /// requests, so callers can inspect or modify log state without additional coordination.
    pub fn log_server(&self) -> Option<Arc<Mutex<LogServer>>> {
        self.log_server.clone()
    }

    /// `anchor_server` returns the shared [`AnchorServer`] registered during build, if any.
    ///
    /// The returned `Arc<Mutex<AnchorServer>>` is the same instance intercepting
    /// `/checkpoint/*` requests.
    pub fn anchor_server(&self) -> Option<Arc<Mutex<AnchorServer>>> {
        self.anchor_server.clone()
    }
}

/// `ServerBuilder` constructs and configures a NWEP server before binding.
///
/// Use the builder pattern to attach callbacks, override [`Settings`], and register optional
/// sub-servers.  Call [`build`](ServerBuilder::build) when configuration is complete.
pub struct ServerBuilder {
    addr: String,
    keypair: Keypair,
    settings: Settings,
    on_connect: Option<Box<dyn Fn(ConnInfo) + Send>>,
    on_disconnect: Option<Box<dyn Fn(ConnInfo, i32) + Send>>,
    log_server: Option<Arc<Mutex<LogServer>>>,
    anchor_server: Option<Arc<Mutex<AnchorServer>>>,
}

impl ServerBuilder {
    /// `new` creates a builder that will bind the server to `addr` and authenticate with `keypair`.
    ///
    /// `addr` is any string accepted by [`std::net::UdpSocket::bind`] (e.g. `"0.0.0.0:4433"`).
    /// The `keypair` provides the Ed25519 identity used in the mutual-authentication handshake.
    pub fn new(addr: impl Into<String>, keypair: Keypair) -> Self {
        ServerBuilder {
            addr: addr.into(),
            keypair,
            settings: Settings::default(),
            on_connect: None,
            on_disconnect: None,
            log_server: None,
            anchor_server: None,
        }
    }

    /// `settings` overrides the default transport settings for this server.
    pub fn settings(mut self, s: Settings) -> Self {
        self.settings = s;
        self
    }

    /// `on_connect` registers a callback invoked each time a peer completes the handshake.
    ///
    /// The callback receives a [`ConnInfo`] describing the new peer.  It runs on the event
    /// loop thread, so it should be fast and non-blocking.
    pub fn on_connect<F: Fn(ConnInfo) + Send + 'static>(mut self, f: F) -> Self {
        self.on_connect = Some(Box::new(f));
        self
    }

    /// `on_disconnect` registers a callback invoked each time a peer disconnects.
    ///
    /// The second argument is an error code: `0` means a clean shutdown, non-zero indicates
    /// a transport error.  The callback runs on the event loop thread.
    pub fn on_disconnect<F: Fn(ConnInfo, i32) + Send + 'static>(mut self, f: F) -> Self {
        self.on_disconnect = Some(Box::new(f));
        self
    }

    /// Register a `LogServer` to automatically handle requests to `/log` and `/log/*`.
    /// The `Arc<Mutex<LogServer>>` is shared with the returned `Server` so callers
    /// can still access it via `Server::log_server()` after starting.
    pub fn log_server(mut self, ls: Arc<Mutex<LogServer>>) -> Self {
        self.log_server = Some(ls);
        self
    }

    /// Register an `AnchorServer` to automatically handle requests to `/checkpoint` and `/checkpoint/*`.
    pub fn anchor_server(mut self, as_: Arc<Mutex<AnchorServer>>) -> Self {
        self.anchor_server = Some(as_);
        self
    }

    pub fn build<H: Handler>(mut self, handler: H) -> Result<(Server, EventLoop), Error> {
        let node_id = self.keypair.node_id()?;
        // Safety: `nwep_server_new` stores the keypair pointer for use during
        // TLS signing. We must keep the Keypair alive for the server's lifetime.
        // The keypair will be moved into the EventLoop below.

        let socket = UdpSocket::bind(&self.addr)
            .map_err(|_| Error::from_code(crate::error::ERR_NETWORK_SOCKET))?;
        let local_addr = socket
            .local_addr()
            .map_err(|_| Error::from_code(crate::error::ERR_NETWORK_SOCKET))?;

        let compression = CString::new(self.settings.compression.as_str()).unwrap_or_default();
        let role_str = CString::new(self.settings.role.as_str()).unwrap_or_default();
        let ffi_settings = ffi::nwep_settings {
            max_streams: self.settings.max_streams,
            max_message_size: self.settings.max_message_size,
            timeout_ms: self.settings.timeout_ms,
            compression: compression.as_ptr(),
            role: role_str.as_ptr(),
        };

        // Shared peer list between CallbackData (event loop thread) and Server (any thread).
        let peers = Arc::new(Mutex::new(Vec::<NodeId>::new()));

        let cb_data = Box::new(CallbackData {
            handler: Box::new(handler),
            on_connect: self.on_connect,
            on_disconnect: self.on_disconnect,
            conns: HashMap::new(),
            conn_to_node_id: HashMap::new(),
            closing_conns: HashSet::new(),
            peers: Arc::clone(&peers),
            log_server: self.log_server.clone(),
            anchor_server: self.anchor_server.clone(),
        });
        let cb_data_ptr = Box::into_raw(cb_data) as *mut std::ffi::c_void;

        let callbacks = ffi::nwep_callbacks {
            on_connect: Some(server_on_connect),
            on_disconnect: Some(server_on_disconnect),
            on_request: Some(server_on_request),
            on_response: None,
            on_notify: None,
            on_stream_data: None,
            on_stream_end: None,
            rand: Some(server_rand),
            log: None,
        };

        let mut c_server: *mut ffi::nwep_server = std::ptr::null_mut();
        check(unsafe {
            ffi::nwep_server_new(
                &mut c_server,
                &ffi_settings,
                &callbacks,
                self.keypair.as_ffi_mut(),
                cb_data_ptr,
            )
        })?;

        let (event_tx, event_rx) = mpsc::sync_channel::<ServerEvent>(1024);
        let shutdown_flag = Arc::new(AtomicBool::new(false));

        let server = Server {
            event_tx: event_tx.clone(),
            node_id,
            addr: local_addr,
            shutdown_flag: shutdown_flag.clone(),
            peers,
            log_server: self.log_server,
            anchor_server: self.anchor_server,
        };

        let event_loop = EventLoop {
            c_server,
            socket,
            local_addr,
            event_rx,
            event_tx,
            shutdown_flag,
            cb_data_ptr,
            _keypair: self.keypair,
        };

        Ok((server, event_loop))
    }
}

pub struct EventLoop {
    c_server: *mut ffi::nwep_server,
    socket: UdpSocket,
    local_addr: SocketAddr,
    event_rx: mpsc::Receiver<ServerEvent>,
    event_tx: mpsc::SyncSender<ServerEvent>,
    shutdown_flag: Arc<AtomicBool>,
    cb_data_ptr: *mut std::ffi::c_void,
    // Keep keypair alive so the C library's stored pointer remains valid
    // for signing during TLS handshakes.
    _keypair: crate::keypair::Keypair,
}

unsafe impl Send for EventLoop {}

impl EventLoop {
    pub fn run(self) -> Result<(), Error> {
        // Spawn UDP reader thread
        let socket_recv = self
            .socket
            .try_clone()
            .map_err(|_| Error::from_code(crate::error::ERR_NETWORK_SOCKET))?;
        let event_tx_recv = self.event_tx.clone();
        let local_addr = self.local_addr;
        let shutdown_flag_recv = self.shutdown_flag.clone();
        std::thread::spawn(move || {
            let mut buf = vec![0u8; UDP_BUF_SIZE];
            loop {
                if shutdown_flag_recv.load(Ordering::SeqCst) {
                    break;
                }
                socket_recv
                    .set_read_timeout(Some(Duration::from_millis(100)))
                    .ok();
                match socket_recv.recv_from(&mut buf) {
                    Ok((n, remote)) => {
                        let _ = event_tx_recv.send(ServerEvent::Packet {
                            data: buf[..n].to_vec(),
                            local: local_addr,
                            remote,
                        });
                    }
                    Err(e)
                        if e.kind() == std::io::ErrorKind::WouldBlock
                            || e.kind() == std::io::ErrorKind::TimedOut => {}
                    Err(_) => {
                        if shutdown_flag_recv.load(Ordering::SeqCst) {
                            break;
                        }
                    }
                }
            }
        });

        // Timer thread
        let event_tx_timer = self.event_tx.clone();
        let shutdown_flag_timer = self.shutdown_flag.clone();
        std::thread::spawn(move || {
            loop {
                std::thread::sleep(Duration::from_millis(10));
                if shutdown_flag_timer.load(Ordering::SeqCst) {
                    break;
                }
                let _ = event_tx_timer.send(ServerEvent::TimerExpiry);
            }
        });

        let mut write_buf = vec![0u8; UDP_BUF_SIZE];
        let socket_send = self
            .socket
            .try_clone()
            .map_err(|_| Error::from_code(crate::error::ERR_NETWORK_SOCKET))?;

        loop {
            let event = match self.event_rx.recv_timeout(Duration::from_millis(100)) {
                Ok(e) => e,
                Err(mpsc::RecvTimeoutError::Timeout) => {
                    if self.shutdown_flag.load(Ordering::SeqCst) {
                        break;
                    }
                    continue;
                }
                Err(mpsc::RecvTimeoutError::Disconnected) => break,
            };

            let ts = now_ns();

            match event {
                ServerEvent::Shutdown => break,

                ServerEvent::Packet {
                    data,
                    local,
                    remote,
                } => {
                    let path = make_path(&local, &remote);
                    unsafe {
                        ffi::nwep_server_read(self.c_server, &path, data.as_ptr(), data.len(), ts);
                    }
                    drain_writes(self.c_server, &socket_send, &mut write_buf, ts);
                }

                ServerEvent::TimerExpiry => {
                    let expiry = unsafe { ffi::nwep_server_get_expiry(self.c_server) };
                    if expiry != u64::MAX && ts >= expiry {
                        unsafe {
                            ffi::nwep_server_handle_expiry(self.c_server, ts);
                        }
                        drain_writes(self.c_server, &socket_send, &mut write_buf, ts);
                    }
                }

                ServerEvent::Notify {
                    peer_node_id,
                    event,
                    path,
                    body,
                    options,
                } => {
                    let cb = unsafe { &*(self.cb_data_ptr as *const CallbackData) };
                    if let Some(&conn) = cb.conns.get(&peer_node_id) {
                        let ev = CString::new(event.as_str()).unwrap_or_default();
                        let pth = CString::new(path.as_str()).unwrap_or_default();
                        // Keep header strings alive across the FFI call.
                        let c_headers: Vec<ffi::nwep_header> = options
                            .as_ref()
                            .map(|o| {
                                o.headers
                                    .iter()
                                    .map(|h| ffi::nwep_header {
                                        name: h.name.as_ptr(),
                                        name_len: h.name.len(),
                                        value: h.value.as_ptr(),
                                        value_len: h.value.len(),
                                    })
                                    .collect()
                            })
                            .unwrap_or_default();
                        let notify = ffi::nwep_notify {
                            event: ev.as_ptr(),
                            event_len: event.len(),
                            path: pth.as_ptr(),
                            path_len: path.len(),
                            notify_id: options
                                .as_ref()
                                .and_then(|o| o.notify_id)
                                .unwrap_or([0u8; 16]),
                            has_notify_id: options.as_ref().and_then(|o| o.notify_id).is_some()
                                as i32,
                            headers: if c_headers.is_empty() {
                                std::ptr::null()
                            } else {
                                c_headers.as_ptr()
                            },
                            header_count: c_headers.len(),
                            body: if body.is_empty() {
                                std::ptr::null()
                            } else {
                                body.as_ptr()
                            },
                            body_len: body.len(),
                        };
                        unsafe {
                            do_conn_notify(
                                self.c_server,
                                conn,
                                &notify,
                                &socket_send,
                                &mut write_buf,
                                ts,
                            );
                        }
                    }
                }

                ServerEvent::NotifyAll { event, path, body } => {
                    let cb = unsafe { &*(self.cb_data_ptr as *const CallbackData) };
                    // Snapshot conn list so we don't hold cb borrow across async ops.
                    let conns: Vec<*mut ffi::nwep_conn> = cb.conns.values().copied().collect();
                    if !conns.is_empty() {
                        let ev = CString::new(event.as_str()).unwrap_or_default();
                        let pth = CString::new(path.as_str()).unwrap_or_default();
                        let notify = ffi::nwep_notify {
                            event: ev.as_ptr(),
                            event_len: event.len(),
                            path: pth.as_ptr(),
                            path_len: path.len(),
                            notify_id: [0u8; 16],
                            has_notify_id: 0,
                            headers: std::ptr::null(),
                            header_count: 0,
                            body: if body.is_empty() {
                                std::ptr::null()
                            } else {
                                body.as_ptr()
                            },
                            body_len: body.len(),
                        };
                        for conn in conns {
                            unsafe {
                                do_conn_notify(
                                    self.c_server,
                                    conn,
                                    &notify,
                                    &socket_send,
                                    &mut write_buf,
                                    ts,
                                );
                            }
                        }
                    }
                }

                ServerEvent::CloseConn {
                    peer_node_id,
                    error,
                } => {
                    let cb = unsafe { &mut *(self.cb_data_ptr as *mut CallbackData) };
                    if let Some(&conn) = cb.conns.get(&peer_node_id) {
                        let conn_key = conn as usize;
                        // Remove from active maps immediately so connection_count() and
                        // connected_peers() reflect the close without waiting for the C library.
                        let node_id = cb.conn_to_node_id.remove(&conn_key).unwrap_or(peer_node_id);
                        cb.conns.remove(&peer_node_id);
                        if let Ok(mut peers) = cb.peers.lock() {
                            peers.retain(|&n| n != peer_node_id);
                        }
                        // Track as closing so server_on_request aborts any incoming streams.
                        cb.closing_conns.insert(conn_key);
                        // Manually fire the Rust disconnect callback.  The C library won't do
                        // it promptly because nwep_conn_close only enters QUIC CLOSING state;
                        // it sends CONNECTION_CLOSE only in response to incoming client packets.
                        if let Some(cb_fn) = &cb.on_disconnect {
                            let peer = unsafe {
                                if conn.is_null() {
                                    std::ptr::null()
                                } else {
                                    ffi::nwep_conn_get_peer_identity(conn)
                                }
                            };
                            let identity = if peer.is_null() {
                                Identity {
                                    pubkey: [0u8; 32],
                                    node_id,
                                }
                            } else {
                                let mut id = unsafe { Identity::from(*peer) };
                                if id.node_id.0 == [0u8; 32] {
                                    id.node_id = node_id;
                                }
                                id
                            };
                            cb_fn(
                                ConnInfo {
                                    node_id,
                                    peer_identity: identity,
                                    role: ServerRole::Regular,
                                },
                                error,
                            );
                        }
                        unsafe {
                            ffi::nwep_conn_close(conn, error);
                        }
                        drain_writes(self.c_server, &socket_send, &mut write_buf, ts);
                    }
                }
            }
        }

        unsafe {
            ffi::nwep_server_close(self.c_server);
            // Pump timer ticks so CONNECTION_CLOSE packets are sent to all connected clients.
            // nwep_server_close schedules the close; handle_expiry calls generate the packets.
            let deadline = std::time::Instant::now() + std::time::Duration::from_millis(500);
            loop {
                let ts = now_ns();
                ffi::nwep_server_handle_expiry(self.c_server, ts);
                drain_writes(self.c_server, &socket_send, &mut write_buf, ts);
                let expiry = ffi::nwep_server_get_expiry(self.c_server);
                if expiry == u64::MAX || std::time::Instant::now() >= deadline {
                    break;
                }
                std::thread::sleep(std::time::Duration::from_millis(10));
            }
            ffi::nwep_server_free(self.c_server);
        }

        Ok(())
    }
}

impl Drop for EventLoop {
    fn drop(&mut self) {
        if !self.cb_data_ptr.is_null() {
            unsafe {
                drop(Box::from_raw(self.cb_data_ptr as *mut CallbackData));
            }
            self.cb_data_ptr = std::ptr::null_mut();
        }
    }
}

fn now_ns() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_nanos() as u64
}

fn drain_writes(server: *mut ffi::nwep_server, socket: &UdpSocket, buf: &mut Vec<u8>, ts: u64) {
    loop {
        let mut path = unsafe { std::mem::zeroed::<ffi::nwep_path>() };
        let n =
            unsafe { ffi::nwep_server_write(server, &mut path, buf.as_mut_ptr(), buf.len(), ts) };
        if n <= 0 {
            break;
        }
        if let Some(addr) = sockaddr_to_socketaddr(&path.remote_addr, path.remote_addrlen) {
            socket.send_to(&buf[..n as usize], addr).ok();
        }
    }
}

fn make_path(local: &SocketAddr, remote: &SocketAddr) -> ffi::nwep_path {
    let mut path = unsafe { std::mem::zeroed::<ffi::nwep_path>() };
    let (local_sa, local_len) = socketaddr_to_sockaddr(local);
    let (remote_sa, remote_len) = socketaddr_to_sockaddr(remote);
    path.local_addr = local_sa;
    path.local_addrlen = local_len;
    path.remote_addr = remote_sa;
    path.remote_addrlen = remote_len;
    path
}

fn socketaddr_to_sockaddr(addr: &SocketAddr) -> (ffi::sockaddr_storage, usize) {
    let mut storage = unsafe { std::mem::zeroed::<ffi::sockaddr_storage>() };
    // Matching Go's fillServerSockaddr: use AF_INET for IPv4 addresses (including
    // IPv4-mapped IPv6 ::ffff:x.x.x.x received from a dual-stack socket), and
    // AF_INET6 for pure IPv6.
    let (is_v4, v4_octets, port) = match addr {
        SocketAddr::V4(v4) => (true, Some(v4.ip().octets()), v4.port()),
        SocketAddr::V6(v6) => {
            let octets = v6.ip().octets();
            let is_mapped =
                octets[..10].iter().all(|&b| b == 0) && octets[10] == 0xff && octets[11] == 0xff;
            if is_mapped {
                (
                    true,
                    Some([octets[12], octets[13], octets[14], octets[15]]),
                    v6.port(),
                )
            } else {
                (false, None, v6.port())
            }
        }
    };
    if is_v4 {
        let sin: &mut sockaddr_in = unsafe { &mut *((&mut storage) as *mut _ as *mut _) };
        sin.sin_family = AF_INET as sa_family_t;
        sin.sin_port = port.to_be();
        sin.sin_addr.s_addr = u32::from_ne_bytes(v4_octets.unwrap());
        (storage, std::mem::size_of::<sockaddr_in>())
    } else {
        let sin6: &mut sockaddr_in6 = unsafe { &mut *((&mut storage) as *mut _ as *mut _) };
        sin6.sin6_family = AF_INET6 as sa_family_t;
        sin6.sin6_port = port.to_be();
        sin6.sin6_addr.s6_addr = match addr {
            SocketAddr::V6(v6) => v6.ip().octets(),
            SocketAddr::V4(_) => unreachable!(),
        };
        (storage, std::mem::size_of::<sockaddr_in6>())
    }
}

fn sockaddr_to_socketaddr(storage: &ffi::sockaddr_storage, _len: usize) -> Option<SocketAddr> {
    let family = storage.ss_family as i32;
    match family {
        AF_INET => {
            // Promote to IPv4-mapped IPv6 so send_to works on the AF_INET6 server socket.
            let sin: &sockaddr_in = unsafe { &*(storage as *const _ as *const _) };
            let v4 = std::net::Ipv4Addr::from(u32::from_be(sin.sin_addr.s_addr));
            let port = u16::from_be(sin.sin_port);
            let mapped = std::net::Ipv6Addr::from(v4.to_ipv6_mapped().octets());
            Some(SocketAddr::new(mapped.into(), port))
        }
        AF_INET6 => {
            let sin6: &sockaddr_in6 = unsafe { &*(storage as *const _ as *const _) };
            let ip = std::net::Ipv6Addr::from(sin6.sin6_addr.s6_addr);
            let port = u16::from_be(sin6.sin6_port);
            Some(SocketAddr::new(ip.into(), port))
        }
        _ => None,
    }
}

pub struct Router {
    routes: Vec<(String, Box<dyn Handler>)>,
    prefix_routes: Vec<(String, Box<dyn Handler>)>,
}

impl Router {
    pub fn new() -> Self {
        Router {
            routes: Vec::new(),
            prefix_routes: Vec::new(),
        }
    }

    pub fn handle(&mut self, path: &str, h: impl Handler) {
        self.routes.push((path.to_string(), Box::new(h)));
    }

    pub fn handle_func<F: Fn(&mut ResponseWriter, &Request) + Send + 'static>(
        &mut self,
        path: &str,
        f: F,
    ) {
        self.routes.push((path.to_string(), Box::new(f)));
    }

    pub fn handle_prefix(&mut self, prefix: &str, h: impl Handler) {
        self.prefix_routes.push((prefix.to_string(), Box::new(h)));
    }
}

impl Handler for Router {
    fn serve(&self, w: &mut ResponseWriter, r: &Request) {
        for (path, h) in &self.routes {
            if path == &r.path {
                h.serve(w, r);
                return;
            }
        }
        for (prefix, h) in &self.prefix_routes {
            if r.path.starts_with(prefix.as_str()) {
                h.serve(w, r);
                return;
            }
        }
        let _ = w.respond(crate::protocol::STATUS_NOT_FOUND, b"");
    }
}

impl Default for Router {
    fn default() -> Self {
        Self::new()
    }
}