//! Socket handle and `zmq_socket`/close/bind/connect/unbind/disconnect.

use std::collections::VecDeque;
use std::ffi::{CStr, c_int, c_void};
use std::str::FromStr;
use std::sync::atomic::{AtomicBool, AtomicI64, Ordering};
use std::sync::{Arc, Mutex};

use std::net::{IpAddr, Ipv6Addr};

use bytes::Bytes;
use omq_tokio::SocketType;
use omq_tokio::endpoint::{Endpoint, Host};

use std::os::raw::c_char;

use crate::context::{OmqContext, next_socket_id};
use crate::error::{ETERM, fail, map_omq_err, set_errno};
use crate::opts::SocketOverlay;

/// Rewrite `Host::Wildcard` to IPv6 unspecified (::) for dual-stack bind.
fn ipv6_rewrite_wildcard(ep: Endpoint) -> Endpoint {
    match ep {
        Endpoint::Tcp {
            host: Host::Wildcard,
            port,
        } => Endpoint::Tcp {
            host: Host::Ip(IpAddr::V6(Ipv6Addr::UNSPECIFIED)),
            port,
        },
        other => other,
    }
}

// Default channel capacity (matches default HWM).
pub(crate) const DEFAULT_HWM: usize = 1000;

/// Eventfd-based notification pair on Linux; pipe pair on other platforms.
#[cfg(target_os = "linux")]
pub(crate) struct NotifyFd {
    /// eventfd signaled (+1) for each message delivered to the recv ring.
    pub recv_fd: std::os::unix::io::RawFd,
    /// eventfd signaled (+1) for each send slot freed.
    pub send_fd: std::os::unix::io::RawFd,
}

#[cfg(not(target_os = "linux"))]
pub(crate) struct NotifyFd {
    pub recv_read: std::os::unix::io::RawFd,
    pub recv_write: std::os::unix::io::RawFd,
    pub send_read: std::os::unix::io::RawFd,
    pub send_write: std::os::unix::io::RawFd,
}

impl std::fmt::Debug for NotifyFd {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str("NotifyFd { .. }")
    }
}

#[cfg(target_os = "linux")]
impl NotifyFd {
    fn new() -> Option<Self> {
        // Non-semaphore: a single read returns the accumulated count
        // and resets to 0. This allows O(1) drain instead of O(N).
        // SAFETY: eventfd returns a valid fd on success, -1 on failure.
        let recv_fd = unsafe { libc::eventfd(0, libc::EFD_NONBLOCK) };
        if recv_fd < 0 {
            return None;
        }
        // SAFETY: same as above.
        let send_fd = unsafe { libc::eventfd(0, libc::EFD_NONBLOCK) };
        if send_fd < 0 {
            // SAFETY: recv_fd is a valid open fd from the successful eventfd call above.
            unsafe { libc::close(recv_fd) };
            return None;
        }
        Some(Self { recv_fd, send_fd })
    }

    fn close(&self) {
        // SAFETY: recv_fd and send_fd are valid fds opened by new().
        unsafe {
            libc::close(self.recv_fd);
            libc::close(self.send_fd);
        }
    }

    pub(crate) fn signal_recv(fd: std::os::unix::io::RawFd) {
        let val: u64 = 1;
        // SAFETY: fd is a valid eventfd; writing 8 bytes atomically increments the counter.
        unsafe {
            libc::write(fd, (&raw const val).cast::<libc::c_void>(), 8);
        }
    }
}

#[cfg(not(target_os = "linux"))]
impl NotifyFd {
    fn new() -> Option<Self> {
        let mut fds = [-1i32; 2];
        // SAFETY: fds is a valid 2-element array; pipe() writes two fds into it on success.
        let rc = unsafe { libc::pipe(fds.as_mut_ptr()) };
        if rc != 0 {
            return None;
        }
        // SAFETY: fds[0] and fds[1] are valid fds from the successful pipe() call.
        unsafe {
            libc::fcntl(fds[0], libc::F_SETFL, libc::O_NONBLOCK);
            libc::fcntl(fds[1], libc::F_SETFL, libc::O_NONBLOCK);
        }
        let (recv_read, recv_write) = (fds[0], fds[1]);
        // SAFETY: same as above.
        let rc = unsafe { libc::pipe(fds.as_mut_ptr()) };
        if rc != 0 {
            // SAFETY: recv_read and recv_write are valid open fds.
            unsafe {
                libc::close(recv_read);
                libc::close(recv_write);
            }
            return None;
        }
        // SAFETY: fds[0] and fds[1] are valid fds from the successful pipe() call.
        unsafe {
            libc::fcntl(fds[0], libc::F_SETFL, libc::O_NONBLOCK);
            libc::fcntl(fds[1], libc::F_SETFL, libc::O_NONBLOCK);
        }
        let (send_read, send_write) = (fds[0], fds[1]);
        Some(Self {
            recv_read,
            recv_write,
            send_read,
            send_write,
        })
    }

    fn close(&self) {
        // SAFETY: all fds are valid, opened by new().
        unsafe {
            libc::close(self.recv_read);
            libc::close(self.recv_write);
            libc::close(self.send_read);
            libc::close(self.send_write);
        }
    }

    pub(crate) fn signal_recv(fd: std::os::unix::io::RawFd) {
        let b: u8 = 1;
        // SAFETY: fd is a valid pipe write end; writing 1 byte signals readiness.
        unsafe {
            libc::write(fd, (&raw const b).cast::<libc::c_void>(), 1);
        }
    }
}

/// Dual yring consumers for the recv path.
#[derive(Debug)]
pub(crate) struct RecvConsumers {
    /// Filled directly by the first peer's `ConnectionDriver`.
    pub fast: yring::Consumer<omq_tokio::Message>,
    /// Filled by the recv pump task (fallback for second+ peers).
    pub pump: yring::Consumer<omq_tokio::Message>,
}

#[expect(dead_code)]
#[derive(Debug)]
pub(crate) struct OmqSocket {
    pub id: u64,
    pub thread_idx: usize,
    pub ctx: Arc<OmqContext>,
    pub socket_type: SocketType,
    pub overlay: Mutex<SocketOverlay>,
    pub sndtimeo_ms: AtomicI64,
    pub rcvtimeo_ms: AtomicI64,
    /// Accumulator for SNDMORE multipart assembly.
    /// `UnsafeCell` because zmq guarantees single-threaded socket access.
    pub send_accum: std::cell::UnsafeCell<Vec<Bytes>>,
    /// Lock-free inproc bypass (sender half). Set once during connect;
    /// accessed only from the `zmq_send` caller thread (ZMQ's single-thread
    /// contract per socket).
    pub bypass_send: std::cell::UnsafeCell<Option<crate::inproc_bypass::BypassSend>>,
    /// Lock-free inproc bypass (receiver half).
    pub bypass_recv: std::cell::UnsafeCell<Option<crate::inproc_bypass::BypassRecv>>,
    /// Leftover frames from a multipart recv (RCVMORE).
    pub recv_drain: Mutex<VecDeque<Bytes>>,
    /// True when `recv_drain` is non-empty. Checked without the lock so the
    /// mutex is skipped entirely on the common single-frame recv path.
    pub drain_nonempty: AtomicBool,
    /// Dual yring consumers. `fast` is filled directly by the first
    /// peer's `ConnectionDriver` (bypasses `async_channel` + recv pump).
    /// `pump` is filled by the recv pump task for second+ peers.
    /// Accessed only from the `zmq_recv` caller thread.
    pub recv_cons: std::cell::UnsafeCell<Option<RecvConsumers>>,
    /// The inner omq-tokio socket. Send+Sync, stored directly.
    pub inner: std::sync::OnceLock<Arc<omq_tokio::Socket>>,
    /// Backpressure: recv pump waits on this when the recv ring is full.
    pub recv_space: std::sync::OnceLock<Arc<tokio::sync::Notify>>,
    /// Shared config for recycling the recv fast yring on peer churn.
    pub recv_sink_config: std::sync::OnceLock<Arc<omq_tokio::engine::RecvSinkConfig>>,
    pub last_endpoint: Mutex<Option<String>>,
    pub notify: NotifyFd,
    pub bound_or_connected: AtomicBool,
    pub recv_pump: std::sync::OnceLock<tokio::task::JoinHandle<()>>,
    /// Send yield state: (message count, bytes queued). Accessed only
    /// from the `zmq_send` caller thread (ZMQ single-thread contract).
    pub send_yield: std::cell::UnsafeCell<(u32, usize)>,
}

/// Map ZMQ socket-type integer to `SocketType`.
fn map_socket_type(t: c_int) -> Option<SocketType> {
    match t {
        0 => Some(SocketType::Pair),
        1 => Some(SocketType::Pub),
        2 => Some(SocketType::Sub),
        3 => Some(SocketType::Req),
        4 => Some(SocketType::Rep),
        5 => Some(SocketType::Dealer),
        6 => Some(SocketType::Router),
        7 => Some(SocketType::Pull),
        8 => Some(SocketType::Push),
        9 => Some(SocketType::XPub),
        10 => Some(SocketType::XSub),
        11 => Some(SocketType::Stream),
        12 => Some(SocketType::Server),
        13 => Some(SocketType::Client),
        14 => Some(SocketType::Radio),
        15 => Some(SocketType::Dish),
        16 => Some(SocketType::Gather),
        17 => Some(SocketType::Scatter),
        19 => Some(SocketType::Peer),
        20 => Some(SocketType::Channel),
        _ => None,
    }
}

/// Run an async op against the socket's inner Arc and return the result.
/// Spawns the future on the io thread's tokio runtime and blocks the
/// calling thread until completion.
pub(crate) fn with_socket<F, Fut, T>(
    ctx: &Arc<OmqContext>,
    thread_idx: usize,
    inner: &Arc<omq_tokio::Socket>,
    op: F,
) -> Result<T, ()>
where
    F: FnOnce(Arc<omq_tokio::Socket>) -> Fut + Send + 'static,
    Fut: std::future::Future<Output = T> + Send + 'static,
    T: Send + 'static,
{
    let (otx, orx) = flume::bounded::<T>(1);
    let s = inner.clone();
    ctx.handle(thread_idx).spawn(async move {
        let out = op(s).await;
        let _ = otx.send(out);
    });
    orx.recv().map_err(|_| ())
}

fn is_bypass_eligible(a: SocketType, b: SocketType) -> bool {
    matches!(
        (a, b),
        (SocketType::Push, SocketType::Pull) | (SocketType::Pull, SocketType::Push)
    )
}

fn try_install_bypass(sender: &Arc<OmqSocket>, receiver: &Arc<OmqSocket>) {
    let capacity = {
        let Ok(s_ov) = sender.overlay.lock() else {
            return;
        };
        let Ok(r_ov) = receiver.overlay.lock() else {
            return;
        };
        let shwm = s_ov.send_hwm.unwrap_or(DEFAULT_HWM as u32) as usize;
        let rhwm = r_ov.recv_hwm.unwrap_or(DEFAULT_HWM as u32) as usize;
        shwm.min(rhwm).max(16)
    };

    #[cfg(target_os = "linux")]
    let recv_fd = receiver.notify.recv_fd;
    #[cfg(not(target_os = "linux"))]
    let recv_fd = receiver.notify.recv_write;

    // Byte ring capacity: enough for `capacity` messages at a generous
    // average size. Rounded up to a power of two internally.
    let byte_ring_cap = capacity * 1024;
    let (bsend, brecv) = crate::inproc_bypass::create_bypass(byte_ring_cap, recv_fd);
    // SAFETY: called from zmq_bind/zmq_connect before any send/recv,
    // so no concurrent access to the UnsafeCells.
    unsafe { *sender.bypass_send.get() = Some(bsend) };
    unsafe { *receiver.bypass_recv.get() = Some(brecv) };
}

/// Register an inproc bind. If there are pending connectors, install
/// bypass pipes for eligible pairs.
fn register_inproc_bind(sock: &Arc<OmqSocket>, name: &str) {
    let ctx = &sock.ctx;
    let Ok(mut binds) = ctx.inproc_binds.lock() else {
        return;
    };
    binds.insert(name.to_owned(), Arc::downgrade(sock));
    drop(binds);

    let Ok(mut waiting) = ctx.inproc_waiting.lock() else {
        return;
    };
    let waiters = waiting.remove(name).unwrap_or_default();
    drop(waiting);
    for w in waiters {
        if let Some(connector) = w.upgrade()
            && is_bypass_eligible(connector.socket_type, sock.socket_type)
        {
            let (sender, receiver) = if connector.socket_type == SocketType::Push {
                (&connector, sock)
            } else {
                (sock, &connector)
            };
            try_install_bypass(sender, receiver);
        }
    }
}

/// Register an inproc connect. If the binder exists, install bypass.
fn register_inproc_connect(sock: &Arc<OmqSocket>, name: &str) {
    let ctx = &sock.ctx;
    let binder = ctx
        .inproc_binds
        .lock()
        .ok()
        .and_then(|g| g.get(name).and_then(std::sync::Weak::upgrade));

    if let Some(binder) = binder {
        if is_bypass_eligible(sock.socket_type, binder.socket_type) {
            let (sender, receiver) = if sock.socket_type == SocketType::Push {
                (sock, &binder)
            } else {
                (&binder, sock)
            };
            try_install_bypass(sender, receiver);
        }
    } else if let Ok(mut waiting) = ctx.inproc_waiting.lock() {
        waiting
            .entry(name.to_owned())
            .or_default()
            .push(Arc::downgrade(sock));
    }
}

#[unsafe(no_mangle)]
#[expect(clippy::arc_with_non_send_sync)]
pub extern "C" fn zmq_socket(ctx_ptr: *mut c_void, type_int: c_int) -> *mut c_void {
    if ctx_ptr.is_null() {
        set_errno(libc::EFAULT);
        return std::ptr::null_mut();
    }
    // SAFETY: caller must pass a valid context pointer from zmq_ctx_new.
    let ctx = unsafe { &*(ctx_ptr.cast::<Arc<OmqContext>>()) };
    if ctx.terminated.load(Ordering::Acquire) {
        set_errno(ETERM);
        return std::ptr::null_mut();
    }
    let Some(socket_type) = map_socket_type(type_int) else {
        set_errno(libc::EINVAL);
        return std::ptr::null_mut();
    };
    if ctx.socket_count.load(Ordering::Relaxed) >= ctx.max_sockets.load(Ordering::Relaxed) {
        set_errno(libc::EMFILE);
        return std::ptr::null_mut();
    }

    let Some(notify) = NotifyFd::new() else {
        set_errno(libc::EMFILE);
        return std::ptr::null_mut();
    };
    let thread_idx = ctx.assign_thread();
    let id = next_socket_id();

    let ctx_arc = ctx.clone();
    ctx_arc.socket_opened();

    let sock = Arc::new(OmqSocket {
        id,
        thread_idx,
        ctx: ctx_arc,
        socket_type,
        overlay: Mutex::new(SocketOverlay::default()),
        sndtimeo_ms: AtomicI64::new(-1),
        rcvtimeo_ms: AtomicI64::new(-1),
        send_accum: std::cell::UnsafeCell::new(Vec::new()),
        bypass_send: std::cell::UnsafeCell::new(None),
        bypass_recv: std::cell::UnsafeCell::new(None),
        recv_drain: Mutex::new(VecDeque::new()),
        drain_nonempty: AtomicBool::new(false),
        recv_cons: std::cell::UnsafeCell::new(None),
        inner: std::sync::OnceLock::new(),
        recv_space: std::sync::OnceLock::new(),
        recv_sink_config: std::sync::OnceLock::new(),
        last_endpoint: Mutex::new(None),
        notify,
        bound_or_connected: AtomicBool::new(false),
        recv_pump: std::sync::OnceLock::new(),
        send_yield: std::cell::UnsafeCell::new((0, 0)),
    });

    Box::into_raw(Box::new(sock)).cast()
}

/// Materialize the omq-tokio socket on the io thread with current overlay
/// options, then start the recv pump. Called once on first bind/connect
/// so that options set between `zmq_socket` and first bind/connect take
/// effect.
#[expect(clippy::too_many_lines)]
pub(crate) fn ensure_materialized(sock: &Arc<OmqSocket>) {
    // CAS guarantees exactly one thread wins the materialization race.
    if sock
        .bound_or_connected
        .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
        .is_err()
    {
        // Another thread already materialized. Spin until inner is set.
        while sock.inner.get().is_none() {
            std::hint::spin_loop();
        }
        return;
    }
    let Ok(overlay) = sock.overlay.lock() else {
        return;
    };
    let opts = overlay.to_options();
    let recv_hwm = overlay.recv_hwm.unwrap_or(DEFAULT_HWM as u32) as usize;
    drop(overlay);

    let socket_type = sock.socket_type;

    #[cfg(target_os = "linux")]
    let recv_signal_fd = sock.notify.recv_fd;
    #[cfg(not(target_os = "linux"))]
    let recv_signal_fd = sock.notify.recv_write;

    let cap = recv_hwm.max(16);
    let (fast_prod, fast_cons) = yring::spsc(cap);
    let (mut pump_prod, pump_cons) = yring::spsc(cap);
    // SAFETY: called once during materialization before any recv.
    unsafe {
        *sock.recv_cons.get() = Some(RecvConsumers {
            fast: fast_cons,
            pump: pump_cons,
        });
    };

    let recv_space = Arc::new(tokio::sync::Notify::new());
    let _ = sock.recv_space.set(recv_space.clone());

    // Build the RecvSink::Yring for the driver's direct fast path.
    let signal_fd = recv_signal_fd;
    let signal_cb: Arc<dyn Fn() + Send + Sync> = Arc::new(move || NotifyFd::signal_recv(signal_fd));
    let recv_sink = omq_tokio::engine::RecvSink::Yring(omq_tokio::engine::YringSink {
        producer: fast_prod,
        signal: Box::new({
            let f = signal_cb.clone();
            move || f()
        }),
        space: recv_space.clone(),
    });
    let recv_sink_cfg = Arc::new(omq_tokio::engine::RecvSinkConfig::new(
        recv_sink,
        signal_cb,
        recv_space.clone(),
        cap,
    ));
    let _ = sock.recv_sink_config.set(recv_sink_cfg.clone());

    // Build the inner socket ON the tokio io thread.
    // Socket::new() calls tokio::spawn internally (spawn_driver), so it
    // must run within a tokio runtime context.
    let (otx, orx) = flume::bounded(1);
    sock.ctx.submit(
        sock.thread_idx,
        Box::new(move || {
            let inner = Arc::new(omq_tokio::Socket::new_with_recv_sink_config(
                socket_type,
                opts,
                recv_sink_cfg,
            ));

            // Recv pump: relay from async_channel into the pump yring.
            // Handles second+ peers whose drivers push to async_channel.
            // For the single-peer case, the async_channel stays empty
            // and this task idles.
            let s_recv = inner.clone();
            let recv_pump = tokio::spawn(async move {
                while let Ok(msg) = s_recv.recv().await {
                    let mut m = msg;
                    loop {
                        match pump_prod.push(m) {
                            Ok(()) => {
                                if let yring::FlushResult::Flushed {
                                    was_empty: true, ..
                                } = pump_prod.flush_and_check()
                                {
                                    NotifyFd::signal_recv(recv_signal_fd);
                                }
                                break;
                            }
                            Err(returned) => {
                                m = returned;
                                let notified = recv_space.notified();
                                tokio::pin!(notified);
                                notified.as_mut().enable();
                                match pump_prod.push(m) {
                                    Ok(()) => {
                                        if let yring::FlushResult::Flushed {
                                            was_empty: true, ..
                                        } = pump_prod.flush_and_check()
                                        {
                                            NotifyFd::signal_recv(recv_signal_fd);
                                        }
                                        break;
                                    }
                                    Err(returned2) => {
                                        m = returned2;
                                        notified.await;
                                    }
                                }
                            }
                        }
                    }
                }
            });

            let _ = otx.send((inner, recv_pump));
        }),
    );

    let Ok((inner, recv_pump)) = orx.recv() else {
        return;
    };
    let _ = sock.inner.set(inner);
    let _ = sock.recv_pump.set(recv_pump);
}

#[unsafe(no_mangle)]
pub extern "C" fn zmq_close(sock_ptr: *mut c_void) -> c_int {
    if sock_ptr.is_null() {
        return fail(libc::EFAULT);
    }
    // SAFETY: sock_ptr came from Box::into_raw in zmq_socket; reclaiming ownership.
    let arc = unsafe { *Box::from_raw(sock_ptr.cast::<Arc<OmqSocket>>()) };

    if let Some(h) = arc.recv_pump.get() {
        h.abort();
    }

    // Enter the tokio runtime context so that dropping OmqSocket (and
    // the Arc<omq_tokio::Socket> it holds) doesn't panic from missing
    // reactor.
    let _guard = arc.ctx.handle(arc.thread_idx).enter();

    arc.notify.close();
    arc.ctx.socket_closed();
    drop(arc);
    0
}

/// Validate `sock_ptr` and `addr`, parse `addr` as a `CStr` and then as an
/// `Endpoint`. Returns the socket reference, owned address string, and
/// parsed endpoint on success. Also checks the context terminated flag.
///
/// # Safety
///
/// `sock_ptr` must be a valid `*mut Arc<OmqSocket>` or null (returns EFAULT).
/// `addr` must be a valid C string pointer or null (returns EFAULT).
unsafe fn parse_endpoint_args<'a>(
    sock_ptr: *mut c_void,
    addr: *const c_char,
) -> Result<(&'a Arc<OmqSocket>, String, Endpoint), c_int> {
    if sock_ptr.is_null() || addr.is_null() {
        return Err(libc::EFAULT);
    }
    // SAFETY: caller guarantees sock_ptr is a valid socket from zmq_socket.
    let sock = unsafe { &*(sock_ptr.cast::<Arc<OmqSocket>>()) };
    if sock.ctx.terminated.load(Ordering::Acquire) {
        return Err(ETERM);
    }
    // SAFETY: addr is non-null (checked above); caller guarantees a valid C string.
    let Ok(addr_str) = unsafe { CStr::from_ptr(addr) }.to_str() else {
        return Err(libc::EINVAL);
    };
    let addr_str = addr_str.to_owned();
    let Ok(endpoint) = omq_tokio::Endpoint::from_str(&addr_str) else {
        return Err(libc::EINVAL);
    };
    Ok((sock, addr_str, endpoint))
}

/// Validate `sock_ptr` and `cstr`, parse `cstr` as a UTF-8 `CStr`.
/// Does **not** parse an endpoint or check the terminated flag (join/leave
/// operate on group names, not transport addresses).
///
/// # Safety
///
/// `sock_ptr` must be a valid `*mut Arc<OmqSocket>` or null.
/// `cstr` must be a valid C string pointer or null.
unsafe fn parse_group_args<'a>(
    sock_ptr: *mut c_void,
    cstr: *const c_char,
) -> Result<(&'a Arc<OmqSocket>, Bytes), c_int> {
    if sock_ptr.is_null() || cstr.is_null() {
        return Err(libc::EFAULT);
    }
    // SAFETY: caller guarantees sock_ptr is a valid socket from zmq_socket.
    let sock = unsafe { &*(sock_ptr.cast::<Arc<OmqSocket>>()) };
    // SAFETY: cstr is non-null (checked above); caller guarantees a valid C string.
    let Ok(g) = unsafe { CStr::from_ptr(cstr) }.to_str() else {
        return Err(libc::EINVAL);
    };
    Ok((sock, Bytes::copy_from_slice(g.as_bytes())))
}

/// Map the two-level `Result<Result<T, omq Error>, ()>` returned by
/// `with_socket` to a C return code (0 on success).
fn result_to_rc<T>(result: &Result<Result<T, omq_tokio::error::Error>, ()>) -> c_int {
    match result {
        Ok(Ok(_)) => 0,
        Ok(Err(e)) => fail(map_omq_err(e)),
        Err(()) => fail(ETERM),
    }
}

#[unsafe(no_mangle)]
pub extern "C" fn zmq_bind(sock_ptr: *mut c_void, addr: *const libc::c_char) -> c_int {
    let (sock, addr_str, mut endpoint) = match unsafe { parse_endpoint_args(sock_ptr, addr) } {
        Ok(t) => t,
        Err(e) => return fail(e),
    };

    // ZMQ_IPV6: rewrite wildcard bind to IPv6 unspecified (dual-stack).
    let Ok(ov) = sock.overlay.lock() else {
        return fail(ETERM);
    };
    if ov.ipv6 {
        endpoint = ipv6_rewrite_wildcard(endpoint);
    }
    drop(ov);

    ensure_materialized(sock);

    let Some(inner) = sock.inner.get() else {
        return fail(ETERM);
    };

    let result = with_socket(&sock.ctx, sock.thread_idx, inner, move |s| async move {
        s.bind(endpoint.clone()).await?;
        let resolved = s.last_bound_endpoint().map(|ep| ep.to_string());
        Ok::<_, omq_tokio::error::Error>(resolved)
    });

    match result {
        Ok(Ok(resolved)) => {
            if let Ok(mut ep) = sock.last_endpoint.lock() {
                *ep = resolved.or(Some(addr_str.clone()));
            }
            if addr_str.starts_with("inproc://") {
                register_inproc_bind(sock, &addr_str);
            }
            0
        }
        Ok(Err(ref e)) => fail(map_omq_err(e)),
        Err(()) => fail(ETERM),
    }
}

#[unsafe(no_mangle)]
pub extern "C" fn zmq_connect(sock_ptr: *mut c_void, addr: *const libc::c_char) -> c_int {
    let (sock, addr_str, endpoint) = match unsafe { parse_endpoint_args(sock_ptr, addr) } {
        Ok(t) => t,
        Err(e) => return fail(e),
    };

    ensure_materialized(sock);

    let Some(inner) = sock.inner.get() else {
        return fail(ETERM);
    };

    let result = with_socket(&sock.ctx, sock.thread_idx, inner, move |s| async move {
        s.connect(endpoint).await
    });

    match result {
        Ok(Ok(())) => {
            if let Ok(mut ep) = sock.last_endpoint.lock() {
                *ep = Some(addr_str.clone());
            }
            if addr_str.starts_with("inproc://") {
                register_inproc_connect(sock, &addr_str);
            }
            0
        }
        Ok(Err(ref e)) => fail(map_omq_err(e)),
        Err(()) => fail(ETERM),
    }
}

#[unsafe(no_mangle)]
pub extern "C" fn zmq_unbind(sock_ptr: *mut c_void, addr: *const libc::c_char) -> c_int {
    let (sock, _addr_str, endpoint) = match unsafe { parse_endpoint_args(sock_ptr, addr) } {
        Ok(t) => t,
        Err(e) => return fail(e),
    };

    let Some(inner) = sock.inner.get() else {
        return fail(ETERM);
    };

    let result = with_socket(&sock.ctx, sock.thread_idx, inner, move |s| async move {
        s.unbind(endpoint).await
    });

    result_to_rc(&result)
}

#[unsafe(no_mangle)]
pub extern "C" fn zmq_disconnect(sock_ptr: *mut c_void, addr: *const libc::c_char) -> c_int {
    let (sock, _addr_str, endpoint) = match unsafe { parse_endpoint_args(sock_ptr, addr) } {
        Ok(t) => t,
        Err(e) => return fail(e),
    };

    let Some(inner) = sock.inner.get() else {
        return fail(ETERM);
    };

    let result = with_socket(&sock.ctx, sock.thread_idx, inner, move |s| async move {
        s.disconnect(endpoint).await
    });

    result_to_rc(&result)
}

#[unsafe(no_mangle)]
pub extern "C" fn zmq_join(sock_ptr: *mut c_void, group: *const libc::c_char) -> c_int {
    let (sock, g) = match unsafe { parse_group_args(sock_ptr, group) } {
        Ok(t) => t,
        Err(e) => return fail(e),
    };
    ensure_materialized(sock);
    let Some(inner) = sock.inner.get() else {
        return fail(ETERM);
    };
    let result = with_socket(&sock.ctx, sock.thread_idx, inner, move |s| async move {
        s.join(g).await
    });
    result_to_rc(&result)
}

#[unsafe(no_mangle)]
pub extern "C" fn zmq_leave(sock_ptr: *mut c_void, group: *const libc::c_char) -> c_int {
    let (sock, g) = match unsafe { parse_group_args(sock_ptr, group) } {
        Ok(t) => t,
        Err(e) => return fail(e),
    };
    ensure_materialized(sock);
    let Some(inner) = sock.inner.get() else {
        return fail(ETERM);
    };
    let result = with_socket(&sock.ctx, sock.thread_idx, inner, move |s| async move {
        s.leave(g).await
    });
    result_to_rc(&result)
}

/// Start monitoring events on `sock`, publishing them as two-frame messages
/// on an inproc PAIR socket bound to `addr`.
///
/// Frame layout (libzmq v1 monitor protocol):
///   frame 1: `event_id` (`u16` LE) + value (`i32` LE) = 6 bytes
///   frame 2: endpoint string (UTF-8)
#[unsafe(no_mangle)]
pub extern "C" fn zmq_socket_monitor(
    sock_ptr: *mut c_void,
    addr: *const libc::c_char,
    events: c_int,
) -> c_int {
    if sock_ptr.is_null() {
        return fail(libc::EFAULT);
    }
    // addr == NULL means stop monitoring.
    if addr.is_null() {
        return 0;
    }
    // SAFETY: sock_ptr is non-null (checked above); caller guarantees a valid socket.
    let sock = unsafe { &*(sock_ptr.cast::<Arc<OmqSocket>>()) };
    // SAFETY: addr is non-null (checked above); caller guarantees a valid C string.
    let Ok(addr_str) = unsafe { CStr::from_ptr(addr) }.to_str() else {
        return fail(libc::EINVAL);
    };
    let addr_str = addr_str.to_owned();
    let events_mask = events as u16;

    ensure_materialized(sock);

    let Some(inner) = sock.inner.get() else {
        return fail(ETERM);
    };

    let result = with_socket(&sock.ctx, sock.thread_idx, inner, move |s| async move {
        let mut stream = s.monitor();

        // Create a PAIR socket for publishing events, bind it to addr.
        let pair = std::sync::Arc::new(omq_tokio::Socket::new(
            omq_tokio::SocketType::Pair,
            omq_tokio::Options::default(),
        ));
        let ep = omq_tokio::Endpoint::from_str(&addr_str)
            .map_err(|e| omq_tokio::error::Error::InvalidEndpoint(e.to_string()))?;
        pair.bind(ep).await?;

        tokio::spawn(async move {
            use omq_tokio::MonitorEvent;
            use omq_tokio::message::Message;

            while let Ok(ev) = stream.recv().await {
                let (event_id, value, endpoint): (u16, i32, String) = match &ev {
                    MonitorEvent::Listening { endpoint } => (0x0008, 0, endpoint.to_string()),
                    MonitorEvent::Accepted { endpoint, .. } => (0x0020, 0, endpoint.to_string()),
                    MonitorEvent::Connected { endpoint, .. } => (0x0001, 0, endpoint.to_string()),
                    MonitorEvent::ConnectDelayed { endpoint, .. } => {
                        (0x0002, 0, endpoint.to_string())
                    }
                    MonitorEvent::HandshakeSucceeded { endpoint, .. } => {
                        (0x1000, 0, endpoint.to_string())
                    }
                    MonitorEvent::HandshakeFailed { endpoint, .. } => {
                        (0x2000, 0, endpoint.to_string())
                    }
                    MonitorEvent::Disconnected { endpoint, .. } => {
                        (0x0200, 0, endpoint.to_string())
                    }
                    MonitorEvent::Closed => (0x0400, 0, String::new()),
                    _ => continue,
                };

                if events_mask != 0xFFFF && (events_mask & event_id) == 0 {
                    continue;
                }

                let mut header = [0u8; 6];
                header[0..2].copy_from_slice(&event_id.to_le_bytes());
                header[2..6].copy_from_slice(&value.to_le_bytes());

                let msg = Message::multipart([
                    bytes::Bytes::copy_from_slice(&header),
                    bytes::Bytes::copy_from_slice(endpoint.as_bytes()),
                ]);
                if pair.send(msg).await.is_err() {
                    break;
                }
            }
        });

        Ok::<_, omq_tokio::error::Error>(())
    });

    match result {
        Ok(Ok(())) => 0,
        Ok(Err(ref e)) => fail(map_omq_err(e)),
        Err(()) => fail(ETERM),
    }
}
