motorcortex-rust 0.5.0

//! Driver thread for the async-first `Request`.
//!
//! The driver owns the `ConnectionManager` (sole owner, no `Mutex`
//! needed) and serves one `Cmd` at a time from its command channel.
//! It runs on a plain `std::thread::spawn` — no tokio runtime on the
//! driver side, because NNG calls are blocking sync. Users `.await`
//! on the `oneshot::Receiver` that came with the command; that works
//! from any tokio runtime context.
//!
//! Shutdown: the driver exits when every `Request` handle is dropped,
//! which closes the mpsc channel and makes `blocking_recv()` return
//! `None`. `ConnectionManager::drop()` then closes the socket.

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, RwLock};
use std::thread;
use std::time::Duration;

use tokio::sync::{mpsc::UnboundedReceiver, watch};

use crate::client::{ParameterTree, receive_message};
use crate::connection::{ConnectionManager, PipeEvent};
use crate::core::proto::{decode_message, encode_with_hash};
use crate::core::request::Cmd;
use crate::core::state::ConnectionState;
use crate::core::subscribe::SubCmd;
use crate::core::subscription::Subscription;
use crate::error::{MotorcortexError, Result};
use crate::msg::{
    CreateGroupMsg, GetParameterListMsg, GetParameterMsg, GetParameterTreeHashMsg,
    GetParameterTreeMsg, GetSessionTokenMsg, GroupStatusMsg, LoginMsg, LogoutMsg, ParameterListMsg,
    ParameterMsg, ParameterTreeHashMsg, ParameterTreeMsg, RemoveGroupMsg, RestoreSessionMsg,
    SessionTokenMsg, SetParameterListMsg, SetParameterMsg, StatusCode, StatusMsg,
};

pub(crate) fn run_request_driver(
    self_tx: tokio::sync::mpsc::UnboundedSender<Cmd>,
    mut rx: UnboundedReceiver<Cmd>,
    state_tx: watch::Sender<ConnectionState>,
    tree: Arc<RwLock<ParameterTree>>,
    last_token: Arc<RwLock<Option<String>>>,
    refresh_count: Arc<AtomicU64>,
) {
    let mut conn = ConnectionManager::new();
    // `user_wants_connected` tracks the most recent user intent so
    // spontaneous pipe events (server died, transport dropped) can
    // be distinguished from a planned `disconnect()`.
    let mut user_wants_connected = false;
    // Flips true the moment we see a pipe `Removed` event while the
    // user wanted to stay connected. The next `Added` event is then
    // treated as a reconnect and drives the RestoreSession flow.
    let mut pending_reconnect = false;
    // Mirrors `ConnectionOptions::reconnect` for the currently-live
    // connection. When `false`, REM_POST publishes `Disconnected`
    // instead of `ConnectionLost` — the dialer isn't configured to
    // redial, so there's no recovery to wait for.
    let mut reconnect_enabled = true;
    // Cap on consecutive `ADD_POST → RestoreSession failure` cycles
    // before the driver pulls the plug on auto-reconnect. `None`
    // means retry forever (NNG's default).
    let mut max_reconnect_attempts: Option<u32> = None;
    // Running count of consecutive RestoreSession failures — reset
    // on any successful restore or on a reconnect that skipped
    // restore (no cached token).
    let mut consecutive_restore_failures: u32 = 0;
    let on_pipe = pipe_handler_for_request(self_tx.clone());
    let refresh_stop = Arc::new(AtomicBool::new(false));
    let mut refresh_thread: Option<thread::JoinHandle<()>> = None;
    while let Some(cmd) = rx.blocking_recv() {
        match cmd {
            Cmd::Connect { url, opts, reply } => {
                let refresh_interval = opts.token_refresh_interval;
                let opt_reconnect = opts.reconnect;
                let opt_max_attempts = opts.max_reconnect_attempts;
                let result = conn.connect(
                    &url,
                    opts,
                    nng_c_sys::nng_req0_open,
                    Arc::clone(&on_pipe),
                );
                if result.is_ok() {
                    user_wants_connected = true;
                    pending_reconnect = false;
                    reconnect_enabled = opt_reconnect;
                    max_reconnect_attempts = opt_max_attempts;
                    consecutive_restore_failures = 0;
                    // Clear any stale token from a previous
                    // connection attempt — user is starting fresh.
                    if let Ok(mut g) = last_token.write() {
                        *g = None;
                    }
                    let _ = state_tx.send(ConnectionState::Connected);
                    // Spawn the token-refresh helper thread when the
                    // interval is > 0 and isn't already running.
                    if refresh_thread.is_none() && !refresh_interval.is_zero() {
                        refresh_stop.store(false, Ordering::Relaxed);
                        let tx_clone = self_tx.clone();
                        let stop_clone = Arc::clone(&refresh_stop);
                        refresh_thread = Some(
                            thread::Builder::new()
                                .name("mcx-token-refresh".into())
                                .spawn(move || {
                                    run_token_refresh(tx_clone, stop_clone, refresh_interval);
                                })
                                .expect("spawning the token-refresh thread must succeed"),
                        );
                    }
                } else {
                    let _ = state_tx.send(ConnectionState::Disconnected);
                }
                let _ = reply.send(result);
            }
            Cmd::Disconnect { reply } => {
                user_wants_connected = false;
                pending_reconnect = false;
                refresh_stop.store(true, Ordering::Relaxed);
                let result = conn.disconnect();
                if let Some(handle) = refresh_thread.take() {
                    let _ = handle.join();
                }
                if let Ok(mut g) = last_token.write() {
                    *g = None;
                }
                // Even if disconnect returns an Err from NNG, the
                // handle is no longer usable — publish Disconnected.
                let _ = state_tx.send(ConnectionState::Disconnected);
                let _ = reply.send(result);
            }
            Cmd::Login { user, pass, reply } => {
                let result = do_login(&conn, user, pass);
                let _ = reply.send(result);
            }
            Cmd::Logout { reply } => {
                let result = do_logout(&conn);
                let _ = reply.send(result);
            }
            Cmd::RequestParameterTree { reply } => {
                let result = do_request_parameter_tree(&conn, &tree);
                let _ = reply.send(result);
            }
            Cmd::GetParameter { path, reply } => {
                let result = do_get_parameter(&conn, path);
                let _ = reply.send(result);
            }
            Cmd::SetParameter { path, value, reply } => {
                let result = do_set_parameter(&conn, path, value);
                let _ = reply.send(result);
            }
            Cmd::GetParameters { msg, reply } => {
                let result = do_get_parameters(&conn, msg);
                let _ = reply.send(result);
            }
            Cmd::SetParameters { msg, reply } => {
                let result = do_set_parameters(&conn, msg);
                let _ = reply.send(result);
            }
            Cmd::CreateGroup { msg, reply } => {
                let result = do_create_group(&conn, msg);
                let _ = reply.send(result);
            }
            Cmd::RemoveGroup { alias, reply } => {
                let result = do_remove_group(&conn, alias);
                let _ = reply.send(result);
            }
            Cmd::GetParameterTreeHash { reply } => {
                let result = do_get_parameter_tree_hash(&conn);
                let _ = reply.send(result);
            }
            Cmd::GetSessionToken { reply } => {
                let result = do_get_session_token(&conn, &last_token);
                let _ = reply.send(result);
            }
            Cmd::RestoreSession { token, reply } => {
                let result = do_restore_session(&conn, token);
                let _ = reply.send(result);
            }
            Cmd::RefreshTokenTick => {
                // Skip the RPC entirely while the pipe is dead —
                // there's no point spending the driver's time
                // (and the NNG recv timeout) on a request that
                // can't succeed. The refresh resumes on the next
                // ADD_POST → Connected transition.
                if *state_tx.borrow() == ConnectionState::Connected {
                    // Fire-and-forget: update the cache, discard
                    // the result. Errors here are silent on
                    // purpose — they usually mean the connection
                    // just flipped flaky and the next PipeEvent
                    // will surface the real state transition.
                    let _ = do_get_session_token(&conn, &last_token);
                    refresh_count.fetch_add(1, Ordering::Relaxed);
                }
            }
            Cmd::Pipe(event) => {
                apply_pipe_event_request(
                    event,
                    &mut pending_reconnect,
                    user_wants_connected,
                    &mut reconnect_enabled,
                    max_reconnect_attempts,
                    &mut consecutive_restore_failures,
                    &state_tx,
                    &conn,
                    &last_token,
                );
            }
        }
    }
    refresh_stop.store(true, Ordering::Relaxed);
    if let Some(handle) = refresh_thread.take() {
        let _ = handle.join();
    }
    // All senders dropped → exit; ConnectionManager::drop() runs next.
}

/// Build the closure the NNG pipe-notify callback invokes on every
/// ADD_POST / REM_POST event for a Request connection. The closure
/// pushes events back into the driver's own command channel so
/// they're processed in order alongside user RPCs.
fn pipe_handler_for_request(
    tx: tokio::sync::mpsc::UnboundedSender<Cmd>,
) -> crate::connection::PipeEventHandler {
    Arc::new(move |event: PipeEvent| {
        // Ignore the error: if the driver has already shut down,
        // nobody's listening and there's nothing to be done.
        let _ = tx.send(Cmd::Pipe(event));
    })
}

fn pipe_handler_for_subscribe(
    tx: tokio::sync::mpsc::UnboundedSender<SubCmd>,
) -> crate::connection::PipeEventHandler {
    Arc::new(move |event: PipeEvent| {
        let _ = tx.send(SubCmd::Pipe(event));
    })
}

/// Apply a PipeEvent to the connection state, respecting the user's
/// most recent connect/disconnect intent so we don't clobber a
/// planned disconnect with a `ConnectionLost` transition.
fn apply_pipe_event(
    event: PipeEvent,
    user_wants_connected: bool,
    state_tx: &watch::Sender<ConnectionState>,
) {
    if !user_wants_connected {
        // The user has asked to be disconnected; pipe events from the
        // teardown or a stale dialer are noise.
        return;
    }
    let new_state = match event {
        PipeEvent::Added => ConnectionState::Connected,
        PipeEvent::Removed => ConnectionState::ConnectionLost,
    };
    let _ = state_tx.send(new_state);
}

/// Outcome of the `RestoreSession` attempt the reconnect handler
/// runs — kept as a tiny enum so the pure decision function
/// [`decide_state_after_pipe_event`] stays free of I/O and is
/// unit-testable.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum RestoreOutcome {
    /// Skipped — no cached token, treated as a bare reconnect.
    NotAttempted,
    /// Server accepted the token (`Ok` or `ReadOnlyMode`).
    Accepted,
    /// Server rejected the token or the RPC errored.
    Rejected,
}

/// Pure helper that computes the next `ConnectionState` + updated
/// reconnect bookkeeping. Split out of [`apply_pipe_event_request`]
/// so the state-machine can be exercised without a real
/// [`ConnectionManager`] or server.
///
/// Returns `(new_state, new_reconnect_enabled, new_consecutive_failures,
/// disable_dialer)`. The caller is responsible for actually
/// publishing the state, flipping its local flags, and killing the
/// dialer if requested.
pub(crate) fn decide_state_after_pipe_event(
    event: PipeEvent,
    pending_reconnect: bool,
    user_wants_connected: bool,
    reconnect_enabled: bool,
    max_reconnect_attempts: Option<u32>,
    consecutive_restore_failures: u32,
    restore_outcome: RestoreOutcome,
) -> Option<PipeEventDecision> {
    if !user_wants_connected {
        return None;
    }
    match event {
        PipeEvent::Removed => {
            if reconnect_enabled {
                Some(PipeEventDecision {
                    new_state: ConnectionState::ConnectionLost,
                    new_pending_reconnect: true,
                    new_reconnect_enabled: reconnect_enabled,
                    new_consecutive_failures: consecutive_restore_failures,
                    disable_dialer: false,
                })
            } else {
                Some(PipeEventDecision {
                    new_state: ConnectionState::Disconnected,
                    new_pending_reconnect: false,
                    new_reconnect_enabled: reconnect_enabled,
                    new_consecutive_failures: consecutive_restore_failures,
                    disable_dialer: false,
                })
            }
        }
        PipeEvent::Added if pending_reconnect => match restore_outcome {
            RestoreOutcome::Accepted | RestoreOutcome::NotAttempted => {
                Some(PipeEventDecision {
                    new_state: ConnectionState::Connected,
                    new_pending_reconnect: false,
                    new_reconnect_enabled: reconnect_enabled,
                    new_consecutive_failures: 0,
                    disable_dialer: false,
                })
            }
            RestoreOutcome::Rejected => {
                let new_failures = consecutive_restore_failures.saturating_add(1);
                let cap_hit = matches!(max_reconnect_attempts, Some(cap) if new_failures >= cap);
                if cap_hit {
                    Some(PipeEventDecision {
                        new_state: ConnectionState::Disconnected,
                        new_pending_reconnect: false,
                        new_reconnect_enabled: false,
                        new_consecutive_failures: new_failures,
                        disable_dialer: true,
                    })
                } else {
                    Some(PipeEventDecision {
                        new_state: ConnectionState::SessionExpired,
                        new_pending_reconnect: false,
                        new_reconnect_enabled: reconnect_enabled,
                        new_consecutive_failures: new_failures,
                        disable_dialer: false,
                    })
                }
            }
        },
        PipeEvent::Added => Some(PipeEventDecision {
            new_state: ConnectionState::Connected,
            new_pending_reconnect: false,
            new_reconnect_enabled: reconnect_enabled,
            new_consecutive_failures: consecutive_restore_failures,
            disable_dialer: false,
        }),
    }
}

/// Output of [`decide_state_after_pipe_event`].
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct PipeEventDecision {
    pub new_state: ConnectionState,
    pub new_pending_reconnect: bool,
    pub new_reconnect_enabled: bool,
    pub new_consecutive_failures: u32,
    pub disable_dialer: bool,
}

/// Request-side pipe handler that also drives the session-restore
/// flow on reconnect.
///
/// - `PipeEvent::Removed` while connected → `ConnectionLost`, mark
///   `pending_reconnect = true`.
/// - `PipeEvent::Added` while `pending_reconnect` is set → try
///   `RestoreSession(cached_token)`. On `Ok` / `ReadOnlyMode` →
///   `Connected`. On any other status → `SessionExpired` (the user
///   must re-login or give up). If no token is cached (user never
///   logged in / never fetched one) → just `Connected`. If
///   `max_reconnect_attempts` caps out, disable the dialer and
///   publish `Disconnected`.
// Every parameter is a distinct piece of driver state the handler
// must both read and mutate; bundling them into a struct would
// trade a linter warning for a call-site shuffle with no benefit.
#[allow(clippy::too_many_arguments)]
fn apply_pipe_event_request(
    event: PipeEvent,
    pending_reconnect: &mut bool,
    user_wants_connected: bool,
    reconnect_enabled: &mut bool,
    max_reconnect_attempts: Option<u32>,
    consecutive_restore_failures: &mut u32,
    state_tx: &watch::Sender<ConnectionState>,
    conn: &ConnectionManager,
    last_token: &Arc<RwLock<Option<String>>>,
) {
    // Run the I/O up-front — we need the restore outcome before we
    // can ask the pure state-machine what to do next.
    let restore_outcome = if matches!(event, PipeEvent::Added) && *pending_reconnect {
        match last_token.read().ok().and_then(|g| g.clone()) {
            Some(tok) => match do_restore_session(conn, tok) {
                Ok(StatusCode::Ok) | Ok(StatusCode::ReadOnlyMode) => RestoreOutcome::Accepted,
                _ => RestoreOutcome::Rejected,
            },
            None => RestoreOutcome::NotAttempted,
        }
    } else {
        RestoreOutcome::NotAttempted
    };

    let decision = match decide_state_after_pipe_event(
        event,
        *pending_reconnect,
        user_wants_connected,
        *reconnect_enabled,
        max_reconnect_attempts,
        *consecutive_restore_failures,
        restore_outcome,
    ) {
        Some(d) => d,
        None => return,
    };

    *pending_reconnect = decision.new_pending_reconnect;
    *reconnect_enabled = decision.new_reconnect_enabled;
    *consecutive_restore_failures = decision.new_consecutive_failures;
    if decision.disable_dialer {
        conn.disable_dialer_reconnect();
    }
    let _ = state_tx.send(decision.new_state);
}

/// Helper loop for the background token-refresh thread. Sleeps
/// `interval`, sends `Cmd::RefreshTokenTick`, repeats until
/// `stop` fires or the command channel closes.
fn run_token_refresh(
    tx: tokio::sync::mpsc::UnboundedSender<Cmd>,
    stop: Arc<AtomicBool>,
    interval: Duration,
) {
    // Use a short sleep tick so the stop signal gets observed
    // quickly on disconnect, even when the refresh interval is
    // long (default 30 s).
    let tick = Duration::from_millis(200);
    let mut elapsed = Duration::ZERO;
    loop {
        if stop.load(Ordering::Relaxed) {
            break;
        }
        thread::sleep(tick);
        elapsed += tick;
        if elapsed >= interval {
            elapsed = Duration::ZERO;
            if tx.send(Cmd::RefreshTokenTick).is_err() {
                // Channel closed → driver has shut down.
                break;
            }
        }
    }
}

/// Length of the group-id prefix NNG matches subscriptions against.
/// The id is written little-endian; we only subscribe with 3 bytes
/// because the 4th would collide with unrelated message framing.
const ID_BYTE_SIZE: usize = 3;

/// Driver for the pub/sub side. Same actor pattern as the Request
/// driver: blocking recv on the SubCmd channel, owns the SUB
/// `ConnectionManager`, publishes to a `watch` channel.
///
/// Connect spawns a second thread (the receive loop) that consumes
/// packets off the SUB socket and dispatches them into the active
/// subscriptions. Disconnect stops the receive thread and joins it
/// before returning.
pub(crate) fn run_subscribe_driver(
    self_tx: tokio::sync::mpsc::UnboundedSender<SubCmd>,
    mut rx: UnboundedReceiver<SubCmd>,
    state_tx: watch::Sender<ConnectionState>,
    subscriptions: Arc<RwLock<HashMap<u32, Subscription>>>,
) {
    let mut conn = ConnectionManager::new();
    let stop = Arc::new(AtomicBool::new(false));
    let mut receive_thread: Option<thread::JoinHandle<()>> = None;
    let mut user_wants_connected = false;
    let on_pipe = pipe_handler_for_subscribe(self_tx);

    while let Some(cmd) = rx.blocking_recv() {
        match cmd {
            SubCmd::Connect { url, opts, reply } => {
                let result = conn.connect(
                    &url,
                    opts,
                    nng_c_sys::nng_sub0_open,
                    Arc::clone(&on_pipe),
                );
                if result.is_ok() {
                    user_wants_connected = true;
                    stop.store(false, Ordering::Relaxed);
                    let sock = conn.sock.expect("connect succeeded → sock is Some");
                    let subs = Arc::clone(&subscriptions);
                    let stop_signal = Arc::clone(&stop);
                    receive_thread = Some(
                        thread::Builder::new()
                            .name("mcx-subscribe-receive".into())
                            .spawn(move || run_subscribe_receive(sock, subs, stop_signal))
                            .expect("spawning the subscribe receive thread must succeed"),
                    );
                    let _ = state_tx.send(ConnectionState::Connected);
                } else {
                    let _ = state_tx.send(ConnectionState::Disconnected);
                }
                let _ = reply.send(result);
            }
            SubCmd::Disconnect { reply } => {
                user_wants_connected = false;
                stop.store(true, Ordering::Relaxed);
                let result = conn.disconnect();
                if let Some(handle) = receive_thread.take() {
                    let _ = handle.join();
                }
                subscriptions.write().unwrap().clear();
                let _ = state_tx.send(ConnectionState::Disconnected);
                let _ = reply.send(result);
            }
            SubCmd::Pipe(event) => {
                apply_pipe_event(event, user_wants_connected, &state_tx);
            }
            SubCmd::Subscribe { group_msg, fdiv, reply } => {
                let result = do_subscribe(&conn, &subscriptions, group_msg, fdiv);
                let _ = reply.send(result);
            }
            SubCmd::Unsubscribe { id, reply } => {
                let result = do_unsubscribe(&conn, &subscriptions, id);
                let _ = reply.send(result);
            }
            SubCmd::ApplyResubscribe { results, reply } => {
                let result = do_apply_resubscribe(&conn, &subscriptions, results);
                let _ = reply.send(result);
            }
        }
    }

    // All senders dropped → clean shutdown. The receive thread is
    // likely blocked in `receive_message(&sock)`; closing the socket
    // here wakes it with `Err(Closed)` so the stop-flag check
    // breaks the loop. Without this close, `join()` below would
    // hang forever when the user drops `Subscribe` without calling
    // `disconnect()` first.
    stop.store(true, Ordering::Relaxed);
    let _ = conn.disconnect();
    if let Some(handle) = receive_thread.take() {
        let _ = handle.join();
    }
}

/// Second thread owned by the subscribe driver: blocks on `nng_recv`,
/// parses the 3-byte group id prefix, and pushes the raw buffer into
/// the matching `Subscription::update`.
fn run_subscribe_receive(
    sock: nng_c_sys::nng_socket,
    subscriptions: Arc<RwLock<HashMap<u32, Subscription>>>,
    stop: Arc<AtomicBool>,
) {
    const HEADER_LEN: usize = 4;
    loop {
        if stop.load(Ordering::Relaxed) {
            break;
        }
        match receive_message(&sock) {
            Ok(buffer) => {
                if buffer.len() > HEADER_LEN {
                    let id = (buffer[0] as u32)
                        | ((buffer[1] as u32) << 8)
                        | ((buffer[2] as u32) << 16);
                    let sub_opt = subscriptions.read().unwrap().get(&id).cloned();
                    if let Some(sub) = sub_opt {
                        sub.update(buffer);
                    }
                }
            }
            Err(_) => {
                // Receive errors typically mean the socket was closed
                // during shutdown — re-check stop and back off slightly
                // so we don't spin if the socket stays up for some
                // other reason.
                if stop.load(Ordering::Relaxed) {
                    break;
                }
                thread::sleep(Duration::from_millis(50));
            }
        }
    }
}

fn do_subscribe(
    conn: &ConnectionManager,
    subscriptions: &Arc<RwLock<HashMap<u32, Subscription>>>,
    group_msg: crate::msg::GroupStatusMsg,
    fdiv: u32,
) -> Result<Subscription> {
    let sock = conn.sock.as_ref().ok_or_else(|| {
        MotorcortexError::Connection("Socket is not available. Connect first.".into())
    })?;
    let id = group_msg.id;
    nng_sub(*sock, id)?;

    let sub = Subscription::new(group_msg, fdiv);
    let returned = sub.clone();
    subscriptions.write().unwrap().insert(id, sub);
    Ok(returned)
}

/// For each `(old_id, new_group_msg)`:
///
/// 1. Look up the existing `Subscription` under `old_id`; skip if gone
///    (concurrent unsubscribe).
/// 2. Unsubscribe the old NNG filter — noisy error is logged but
///    non-fatal, since the new filter replaces it on the wire anyway.
/// 3. Subscribe the new NNG filter for the server-assigned id.
/// 4. [`rebind`](Subscription::rebind) the handle in place and rekey
///    the map.
///
/// The map is held under a write lock across the whole batch so the
/// receive thread sees a single consistent "old ids gone, new ids
/// live" transition rather than the partial states in between.
fn do_apply_resubscribe(
    conn: &ConnectionManager,
    subscriptions: &Arc<RwLock<HashMap<u32, Subscription>>>,
    results: Vec<(u32, crate::msg::GroupStatusMsg)>,
) -> Result<()> {
    let sock = conn.sock.as_ref().ok_or_else(|| {
        MotorcortexError::Connection("Socket is not available. Connect first.".into())
    })?;
    let mut guard = subscriptions.write().unwrap();
    for (old_id, new_group) in results {
        let sub = match guard.remove(&old_id) {
            Some(s) => s,
            // Concurrent unsubscribe dropped the entry between the
            // snapshot and now — nothing to re-register.
            None => continue,
        };
        // Best-effort: an NNG filter may already be gone after a
        // socket-level redial; tolerate the error.
        let _ = nng_unsub(*sock, old_id);
        let new_id = new_group.id;
        nng_sub(*sock, new_id)?;
        sub.rebind(new_group);
        guard.insert(new_id, sub);
    }
    Ok(())
}

fn nng_sub(sock: nng_c_sys::nng_socket, id: u32) -> Result<()> {
    let bytes = id.to_le_bytes();
    // SAFETY: `bytes` outlives the call (stack-owned in this
    // function); `ID_BYTE_SIZE` is ≤ `bytes.len()`. `nng_setopt`
    // copies the value internally — no lifetime hazard.
    let rv = unsafe {
        nng_c_sys::nng_setopt(
            sock,
            nng_c_sys::NNG_OPT_SUB_SUBSCRIBE.as_ptr() as *const core::ffi::c_char,
            bytes.as_ptr() as *const std::ffi::c_void,
            ID_BYTE_SIZE,
        )
    };
    if rv != 0 {
        return Err(MotorcortexError::Subscription(format!(
            "Failed to subscribe via NNG. Error code: {rv}"
        )));
    }
    Ok(())
}

fn nng_unsub(sock: nng_c_sys::nng_socket, id: u32) -> Result<()> {
    let bytes = id.to_le_bytes();
    // SAFETY: mirrors `nng_sub` — the buffer outlives the call and
    // NNG copies the option value internally.
    let rv = unsafe {
        nng_c_sys::nng_setopt(
            sock,
            nng_c_sys::NNG_OPT_SUB_UNSUBSCRIBE.as_ptr() as *const core::ffi::c_char,
            bytes.as_ptr() as *const std::ffi::c_void,
            ID_BYTE_SIZE,
        )
    };
    if rv != 0 {
        return Err(MotorcortexError::Subscription(format!(
            "Failed to unsubscribe via NNG. Error code: {rv}"
        )));
    }
    Ok(())
}

fn do_unsubscribe(
    conn: &ConnectionManager,
    subscriptions: &Arc<RwLock<HashMap<u32, Subscription>>>,
    id: u32,
) -> Result<()> {
    let removed = subscriptions.write().unwrap().remove(&id);
    if removed.is_none() {
        // Stale handle — local state already clean; server-side cleanup
        // happens in the caller via `req.remove_group(alias)`.
        return Ok(());
    }
    let sock = conn.sock.as_ref().ok_or_else(|| {
        MotorcortexError::Connection("Socket is not available.".into())
    })?;
    nng_unsub(*sock, id)
}

/// `send_message` + `receive_message` + the sock-missing guard.
/// Every hash-tagged RPC goes through here.
fn rpc_round_trip<M, T>(conn: &ConnectionManager, msg: &M) -> Result<T>
where
    M: prost::Message + crate::msg::Hash,
    T: prost::Message + Default + crate::msg::Hash,
{
    let sock = conn.sock.as_ref().ok_or_else(|| {
        MotorcortexError::Connection("Socket is not available. Connect first.".into())
    })?;
    let buffer = encode_with_hash(msg)?;

    let data_ptr = buffer.as_ptr() as *mut std::ffi::c_void;
    // SAFETY: `buffer` lives on this function's stack across the
    // call; `nng_send` with flag `0` copies the payload into NNG's
    // own queue before returning, so the pointer doesn't need to
    // outlive the call.
    let rv = unsafe { nng_c_sys::nng_send(*sock, data_ptr, buffer.len(), 0) };
    if rv != 0 {
        return Err(MotorcortexError::Io(format!(
            "nng_send failed with code: {rv}"
        )));
    }

    let reply = receive_message(sock)?;
    decode_message::<T>(&reply)
}

fn do_login(conn: &ConnectionManager, user: String, pass: String) -> Result<StatusCode> {
    let msg = LoginMsg {
        header: None,
        login: user,
        password: pass,
    };
    let status: StatusMsg = rpc_round_trip(conn, &msg)?;
    Ok(StatusCode::try_from(status.status)
        .unwrap_or(StatusCode::Failed))
}

fn do_logout(conn: &ConnectionManager) -> Result<StatusCode> {
    let msg = LogoutMsg { header: None };
    let status: StatusMsg = rpc_round_trip(conn, &msg)?;
    Ok(StatusCode::try_from(status.status)
        .unwrap_or(StatusCode::Failed))
}

/// Fetch a single parameter's value bytes. The caller (holding the
/// dtype from the local tree) decodes them into the desired Rust type.
fn do_get_parameter(conn: &ConnectionManager, path: String) -> Result<Vec<u8>> {
    let msg = GetParameterMsg { header: None, path };
    let reply: ParameterMsg = rpc_round_trip(conn, &msg)?;
    Ok(reply.value)
}

/// Write a single parameter. The caller encoded the value against the
/// correct dtype before sending this command.
fn do_set_parameter(
    conn: &ConnectionManager,
    path: String,
    value: Vec<u8>,
) -> Result<StatusCode> {
    let msg = SetParameterMsg {
        header: None,
        offset: None,
        path,
        value,
    };
    let status: StatusMsg = rpc_round_trip(conn, &msg)?;
    Ok(StatusCode::try_from(status.status).unwrap_or(StatusCode::Failed))
}

/// Batch read: caller supplies the fully-built `GetParameterListMsg`;
/// driver ships it and returns the `ParameterListMsg` for the
/// client-side tuple decode.
fn do_get_parameters(conn: &ConnectionManager, msg: GetParameterListMsg) -> Result<ParameterListMsg> {
    rpc_round_trip(conn, &msg)
}

/// Batch write: caller has already encoded every tuple element into
/// the `SetParameterMsg::value` buffers.
fn do_set_parameters(conn: &ConnectionManager, msg: SetParameterListMsg) -> Result<StatusCode> {
    let status: StatusMsg = rpc_round_trip(conn, &msg)?;
    Ok(StatusCode::try_from(status.status).unwrap_or(StatusCode::Failed))
}

fn do_create_group(conn: &ConnectionManager, msg: CreateGroupMsg) -> Result<GroupStatusMsg> {
    rpc_round_trip(conn, &msg)
}

/// Fetch a fresh session token from the server. On `StatusCode::Ok`,
/// store it in the driver's `last_token` cache and return it to the
/// caller. Non-OK replies surface as `MotorcortexError::Status`.
fn do_get_session_token(
    conn: &ConnectionManager,
    last_token: &Arc<RwLock<Option<String>>>,
) -> Result<String> {
    let msg = GetSessionTokenMsg { header: None };
    let reply: SessionTokenMsg = rpc_round_trip(conn, &msg)?;
    let status = StatusCode::try_from(reply.status).unwrap_or(StatusCode::Failed);
    if status != StatusCode::Ok {
        return Err(MotorcortexError::Status(status));
    }
    if let Ok(mut g) = last_token.write() {
        *g = Some(reply.token.clone());
    }
    Ok(reply.token)
}

/// Restore a session by handing the server a previously-issued
/// token. The server replies with a `StatusMsg`; we pass the status
/// back unchanged so the caller can distinguish `Ok` /
/// `ReadOnlyMode` / `PermissionDenied` / `Failed`.
fn do_restore_session(conn: &ConnectionManager, token: String) -> Result<StatusCode> {
    let msg = RestoreSessionMsg {
        header: None,
        token,
    };
    let status: StatusMsg = rpc_round_trip(conn, &msg)?;
    Ok(StatusCode::try_from(status.status).unwrap_or(StatusCode::Failed))
}

/// Decodes the reply as `ParameterTreeHashMsg` (fix: the legacy
/// implementation decoded it as the wrong type and every call 404'd
/// with "Invalid message hash" — see `1f87dd5` on `development`).
fn do_get_parameter_tree_hash(conn: &ConnectionManager) -> Result<u32> {
    let msg = GetParameterTreeHashMsg { header: None };
    let reply: ParameterTreeHashMsg = rpc_round_trip(conn, &msg)?;
    Ok(reply.hash)
}

fn do_remove_group(conn: &ConnectionManager, alias: String) -> Result<StatusCode> {
    // Same rule as `set_parameter`, `login`, `request_parameter_tree`,
    // etc.: an RPC returning `Result<StatusCode>` never converts a
    // non-OK server reply into `Err`. Err means transport / decode
    // trouble; the server's own judgement is the caller's
    // responsibility to branch on.
    let msg = RemoveGroupMsg { header: None, alias };
    let status: StatusMsg = rpc_round_trip(conn, &msg)?;
    Ok(StatusCode::try_from(status.status).unwrap_or(StatusCode::Failed))
}

fn do_request_parameter_tree(
    conn: &ConnectionManager,
    tree: &Arc<RwLock<ParameterTree>>,
) -> Result<StatusCode> {
    let msg = GetParameterTreeMsg { header: None };
    let reply: ParameterTreeMsg = rpc_round_trip(conn, &msg)?;
    let status = StatusCode::try_from(reply.status).unwrap_or(StatusCode::Failed);
    // Only refresh the cache on a healthy reply. A non-OK reply from
    // the server leaves the previous tree intact (helpful if the
    // caller retries).
    if status == StatusCode::Ok {
        if let Some(new_tree) = ParameterTree::from_message(reply) {
            let mut guard = tree
                .write()
                .map_err(|_| MotorcortexError::Decode("parameter tree lock poisoned".into()))?;
            *guard = new_tree;
        }
    }
    Ok(status)
}

#[cfg(test)]
mod state_machine_tests {
    use super::*;

    /// Helper to reduce call-site boilerplate when exercising the
    /// pure decision function.
    fn decide(
        event: PipeEvent,
        pending: bool,
        outcome: RestoreOutcome,
        failures: u32,
        cap: Option<u32>,
        reconnect_enabled: bool,
    ) -> PipeEventDecision {
        decide_state_after_pipe_event(
            event,
            pending,
            true, // user_wants_connected
            reconnect_enabled,
            cap,
            failures,
            outcome,
        )
        .expect("user_wants_connected = true always returns Some")
    }

    #[test]
    fn user_disconnect_swallows_events() {
        // Any event on a disconnected handle is a no-op.
        for event in [PipeEvent::Added, PipeEvent::Removed] {
            let d = decide_state_after_pipe_event(
                event,
                false,
                false, // user_wants_connected = false
                true,
                None,
                0,
                RestoreOutcome::NotAttempted,
            );
            assert!(d.is_none(), "events while disconnected must be ignored");
        }
    }

    #[test]
    fn pipe_removed_with_reconnect_enabled_goes_to_connection_lost() {
        let d = decide(
            PipeEvent::Removed,
            false,
            RestoreOutcome::NotAttempted,
            0,
            None,
            true,
        );
        assert_eq!(d.new_state, ConnectionState::ConnectionLost);
        assert!(d.new_pending_reconnect);
        assert!(d.new_reconnect_enabled);
        assert!(!d.disable_dialer);
    }

    #[test]
    fn pipe_removed_with_reconnect_disabled_goes_to_disconnected() {
        let d = decide(
            PipeEvent::Removed,
            false,
            RestoreOutcome::NotAttempted,
            0,
            None,
            false,
        );
        assert_eq!(d.new_state, ConnectionState::Disconnected);
        assert!(!d.new_pending_reconnect);
    }

    #[test]
    fn pipe_added_without_pending_reconnect_is_plain_connected() {
        let d = decide(
            PipeEvent::Added,
            false,
            RestoreOutcome::NotAttempted,
            0,
            None,
            true,
        );
        assert_eq!(d.new_state, ConnectionState::Connected);
    }

    #[test]
    fn pipe_added_with_no_token_treats_reconnect_as_success() {
        let d = decide(
            PipeEvent::Added,
            true, // pending_reconnect
            RestoreOutcome::NotAttempted,
            3, // stale failure count
            Some(5),
            true,
        );
        assert_eq!(d.new_state, ConnectionState::Connected);
        assert_eq!(d.new_consecutive_failures, 0, "counter resets on success");
    }

    #[test]
    fn pipe_added_with_accepted_restore_resets_counter() {
        let d = decide(
            PipeEvent::Added,
            true,
            RestoreOutcome::Accepted,
            2,
            Some(3),
            true,
        );
        assert_eq!(d.new_state, ConnectionState::Connected);
        assert_eq!(d.new_consecutive_failures, 0);
    }

    #[test]
    fn pipe_added_with_rejected_restore_emits_session_expired_under_cap() {
        let d = decide(
            PipeEvent::Added,
            true,
            RestoreOutcome::Rejected,
            0,
            Some(3),
            true,
        );
        assert_eq!(d.new_state, ConnectionState::SessionExpired);
        assert_eq!(d.new_consecutive_failures, 1);
        assert!(!d.disable_dialer);
    }

    #[test]
    fn pipe_added_with_rejected_restore_trips_cap() {
        // Cap of 2, already 1 failure — this is the 2nd failure and
        // should trip the guard.
        let d = decide(
            PipeEvent::Added,
            true,
            RestoreOutcome::Rejected,
            1,
            Some(2),
            true,
        );
        assert_eq!(d.new_state, ConnectionState::Disconnected);
        assert_eq!(d.new_consecutive_failures, 2);
        assert!(d.disable_dialer, "cap-exceeded must request dialer disable");
        assert!(!d.new_reconnect_enabled, "reconnect flag must flip off");
    }

    #[test]
    fn pipe_added_with_rejected_restore_and_no_cap_keeps_retrying() {
        // No cap set: every rejection surfaces as SessionExpired, dialer
        // stays alive, counter climbs forever.
        let d = decide(
            PipeEvent::Added,
            true,
            RestoreOutcome::Rejected,
            1_000_000,
            None,
            true,
        );
        assert_eq!(d.new_state, ConnectionState::SessionExpired);
        assert_eq!(d.new_consecutive_failures, 1_000_001);
        assert!(!d.disable_dialer);
        assert!(d.new_reconnect_enabled);
    }

    #[test]
    fn cap_exactly_1_trips_on_first_rejection() {
        let d = decide(
            PipeEvent::Added,
            true,
            RestoreOutcome::Rejected,
            0,
            Some(1),
            true,
        );
        assert_eq!(d.new_state, ConnectionState::Disconnected);
        assert!(d.disable_dialer);
    }

    #[test]
    fn consecutive_restore_failures_saturates_rather_than_overflowing() {
        // Absurd but cheap: confirm saturating_add isn't surprising at
        // u32::MAX so the state machine can't panic on pathological
        // reconnect loops.
        let d = decide(
            PipeEvent::Added,
            true,
            RestoreOutcome::Rejected,
            u32::MAX,
            None,
            true,
        );
        assert_eq!(d.new_consecutive_failures, u32::MAX);
    }
}