rzmq 0.5.21

use std::time::Duration;

use crate::throttle::types::AdaptiveThrottleConfig;
use crate::{Blob, CoreState, ZmqError};

pub const DEFAULT_SNDBATCH_COUNT: usize = 128;
pub const DEFAULT_SNDBATCH_BYTES: usize = 256 * 1024; // 256 KB
pub const DEFAULT_RCVBATCH_COUNT: usize = 128;
pub const DEFAULT_RCVBATCH_BYTES: usize = 256 * 1024; // 256 KB;

// Use values consistent with libzmq where possible
pub const SNDBUF: i32 = 11;
pub const RCVBUF: i32 = 12;
pub const SNDHWM: i32 = 23;
pub const RCVHWM: i32 = 24;
pub const LINGER: i32 = 17;
pub const SUBSCRIBE: i32 = 6;
pub const UNSUBSCRIBE: i32 = 7;
pub const ROUTING_ID: i32 = 5; // Often called ZMQ_IDENTITY
pub const RECONNECT_IVL: i32 = 18; // ZMQ_RECONNECT_IVL
pub const RECONNECT_IVL_MAX: i32 = 21; // ZMQ_RECONNECT_IVL_MAX
pub const RCVTIMEO: i32 = 27;
pub const SNDTIMEO: i32 = 28;
pub const LAST_ENDPOINT: i32 = 32;
pub const TCP_KEEPALIVE: i32 = 34;
pub const TCP_KEEPALIVE_IDLE: i32 = 35;
pub const TCP_KEEPALIVE_CNT: i32 = 36;
pub const TCP_KEEPALIVE_INTVL: i32 = 37;
pub const HEARTBEAT_IVL: i32 = 38; // ZMQ_HEARTBEAT_IVL
pub const HEARTBEAT_TIMEOUT: i32 = 39; // ZMQ_HEARTBEAT_TIMEOUT
pub const HEARTBEAT_TTL: i32 = 40; // ZMQ_HEARTBEAT_TTL (Often derived from TIMEOUT)
pub const HANDSHAKE_IVL: i32 = 41; // ZMQ_HANDSHAKE_IVL

pub const ROUTER_MANDATORY: i32 = 33;
pub const AUTO_DELIMITER: i32 = 42; // Router/Dealer Auto Delimiter insertion/stripping handling. Enabled by default.

// Security Options
/// Not used currently
pub const ZAP_DOMAIN: i32 = 55; //TODO  will remove if 100% sure ZAP won't be impl
pub const PLAIN_SERVER: i32 = 44;
pub const PLAIN_USERNAME: i32 = 45;
pub const PLAIN_PASSWORD: i32 = 46;

// Security/Noise XX
pub const NOISE_XX_ENABLED: i32 = 1202; // Boolean (0 or 1)
pub const NOISE_XX_STATIC_SECRET_KEY: i32 = 1200; // Expects 32-byte secret key
pub const NOISE_XX_REMOTE_STATIC_PUBLIC_KEY: i32 = 1201; // Client uses this for server's PK, expects 32-byte public key
                                                         // Optional: For server, a list of allowed client public keys (if not using ZAP for this)
                                                         // pub const NOISE_XX_ALLOWED_PEERS: i32 = 1203; // Would take a list of PKs

// Security/CURVE
pub const CURVE_SERVER: i32 = 47; // Matches libzmq's ZMQ_CURVE_SERVER
pub const CURVE_SECRET_KEY: i32 = 49; // Matches libzmq's ZMQ_CURVE_SECRETKEY
pub const CURVE_SERVER_KEY: i32 = 48; // Matches libzmq's ZMQ_CURVE_SERVERKEY

pub const MAXMSGSIZE: i32 = 22;
pub const MAX_CONNECTIONS: i32 = 1000;

// IO Uring Options
#[cfg(feature = "io-uring")]
pub const IO_URING_SNDZEROCOPY: i32 = 1170;
#[cfg(feature = "io-uring")]
pub const IO_URING_RCVMULTISHOT: i32 = 1171;

/// Socket option: Enable/disable TCP_CORK (Linux only).
/// Value is i32 (0 or 1).
pub const TCP_CORK: i32 = 1172;

pub const IO_URING_SESSION_ENABLED: i32 = 1175;

/// Minimum payload size in bytes at which a send is issued as io_uring zero-copy (SEND_ZC)
/// rather than a regular copy-based send. Tuning this trades kernel memory-lock overhead
/// against the cost of a memcpy; the default 16384 bytes is a reasonable crossover point.
#[cfg(feature = "io-uring")]
pub const IO_URING_ZC_SEND_THRESHOLD: i32 = 1176;

pub const ADAPTIVE_THROTTLE: i32 = 1210;

/// Allow downgrading to the ZMTP/2.0 wire protocol when a peer announces it.
/// Boolean (0 or 1). Enabled by default (matching libzmq, which accepts v2 peers).
pub const ALLOW_ZMTP2: i32 = 1220;

pub const SNDBATCH_COUNT: i32 = 1215; // Max logical messages to coalesce per outbound write
pub const SNDBATCH_BYTES: i32 = 1216; // Max payload bytes to coalesce per outbound write
pub const RCVBATCH_COUNT: i32 = 1217; // Max logical messages to extract per inbound wakeup
pub const RCVBATCH_BYTES: i32 = 1218; // Max payload bytes to extract per inbound wakeup


pub const DEFAULT_RECONNECT_IVL_MS: u64 = 1000;

/// Holds parsed and validated socket options.
#[derive(Debug, Clone)]
pub(crate) struct SocketOptions {
  // High Water Marks (applied to Pipes / internal queues)
  pub rcvhwm: usize,
  pub sndhwm: usize,
  // Timeouts: None = -1 (infinite), Some(ZERO) = 0 (immediate), Some(>0) = timeout
  pub rcvtimeo: Option<Duration>,
  pub sndtimeo: Option<Duration>,
  // Connection Behavior
  pub linger: Option<Duration>, // ZMQ uses -1 for infinite, 0 for immediate, >0 for ms -> map to Duration
  pub reconnect_ivl: Option<Duration>, // Initial reconnect interval (None = ZMQ default, often 0 = no reconnect)
  pub reconnect_ivl_max: Option<Duration>, // Max reconnect interval (for exponential backoff)
  pub backlog: Option<u32>,
  // Identity
  pub routing_id: Option<Blob>,
  pub socket_type_name: String, // e.g., "REQ", "REP" - needed for READY cmd
  // TCP Specific (mirrored for setting on stream)
  pub tcp_keepalive_enabled: i32, // ZMQ standard: -1 off, 0 system, 1 on
  pub tcp_keepalive_idle: Option<Duration>,
  pub tcp_keepalive_count: Option<u32>,
  pub tcp_keepalive_interval: Option<Duration>,
  pub tcp_nodelay: bool, // Usually enabled by default
  pub max_connections: Option<usize>,
  /// Maximum inbound frame size in bytes. -1 means unlimited.
  pub maxmsgsize: i64,

  /// Interval between sending ZMTP PING probes if no traffic received.
  /// `None` disables PINGs.
  pub heartbeat_ivl: Option<Duration>,
  /// Time to wait for PONG reply before considering connection dead.
  /// `None` uses a default derived from `heartbeat_ivl`.
  pub heartbeat_timeout: Option<Duration>,
  pub handshake_ivl: Option<Duration>,

  /// ROUTER behavior when routing ID is unknown.
  /// Default false (drop message). True = return EHOSTUNREACH.
  pub router_mandatory: bool,
  /// Allow downgrading the handshake to ZMTP/2.0 when the peer announces it.
  /// Default true (libzmq accepts v2 peers).
  pub allow_zmtp2: bool,
  // Add other commonly used options as needed
  // pub heartbeat_ttl: Option<Duration>, // TTL often derived from timeout
  pub tcp_cork: bool,
  pub sndbuf: Option<usize>,
  pub rcvbuf: Option<usize>,
  pub io_uring: IOURingSocketOptions,
  pub zap_domain: Option<String>, // ZAP Domain
  #[cfg(feature = "plain")]
  pub plain_options: PlainMechanismSocketOptions,
  #[cfg(feature = "curve")]
  pub curve_options: CurveMechanismSocketOptions,
  #[cfg(feature = "noise_xx")]
  pub noise_xx_options: NoiseXxSocketOptions,
  pub throttle_config: AdaptiveThrottleConfig,
  pub sndbatch_count: usize,
  pub sndbatch_bytes: usize,
  pub rcvbatch_count: usize,
  pub rcvbatch_bytes: usize,
}

impl Default for SocketOptions {
  fn default() -> Self {
    Self {
      // ZMQ Defaults:
      rcvhwm: 256,
      sndhwm: 256,
      rcvtimeo: None,               // -1 in ZMQ
      sndtimeo: None,               // -1 in ZMQ
      linger: Some(Duration::ZERO), // 0 in ZMQ (different from socket default!)
      reconnect_ivl: Some(Duration::from_millis(DEFAULT_RECONNECT_IVL_MS)),
      reconnect_ivl_max: Some(Duration::ZERO), // ZMQ default is 0 (disable max/backoff)
      backlog: None,
      routing_id: None,
      socket_type_name: "UNKNOWN".to_string(), // Default, should be set on creation
      tcp_keepalive_enabled: 0,                // 0 (use system default) in ZMQ
      tcp_keepalive_idle: None,
      tcp_keepalive_count: None,
      tcp_keepalive_interval: None,
      tcp_nodelay: true, // Common default for messaging
      max_connections: Some(1024),
      maxmsgsize: -1,      // -1 = unlimited
      heartbeat_ivl: None, // Disabled by default
      heartbeat_timeout: None,
      handshake_ivl: None,
      router_mandatory: false, // Default ZMQ behavior is to drop silently
      allow_zmtp2: true,       // Accept ZMTP/2.0 peers by default (matches libzmq)
      tcp_cork: false,
      sndbuf: None,
      rcvbuf: None,
      io_uring: Default::default(),
      zap_domain: None,
      #[cfg(feature = "plain")]
      plain_options: Default::default(),
      #[cfg(feature = "noise_xx")]
      noise_xx_options: NoiseXxSocketOptions::default(),
      #[cfg(feature = "curve")]
      curve_options: CurveMechanismSocketOptions::default(),
      throttle_config: {
        let mut c = AdaptiveThrottleConfig::default();
        c.credit_per_message = 5;
        c.healthy_balance_width = 1024000;
        c.max_imbalance = 6553600;
        c.yield_after_n_consecutive = 256;
        c.priority_boost_factor = 5.0;
        c
      },
      sndbatch_count: DEFAULT_SNDBATCH_COUNT,
      sndbatch_bytes: DEFAULT_SNDBATCH_BYTES,
      rcvbatch_count: DEFAULT_RCVBATCH_COUNT,
      rcvbatch_bytes: DEFAULT_RCVBATCH_BYTES,
    }
  }
}

#[derive(Debug, Clone)]
pub struct IOURingSocketOptions {
  pub send_zerocopy: bool,
  pub recv_multishot: bool,
  pub session_enabled: bool,
  /// Minimum payload bytes to use zero-copy send (SEND_ZC) instead of a copy-based send.
  /// Default 16384 bytes — below this the memcpy is cheaper than the kernel memory-lock cost.
  pub zc_send_threshold: usize,
}

impl Default for IOURingSocketOptions {
  fn default() -> Self {
    Self {
      session_enabled: false,
      send_zerocopy: false,
      recv_multishot: false,
      zc_send_threshold: 16384,
    }
  }
}

#[cfg(feature = "curve")]
#[derive(Debug, Clone, Default)]
pub struct CurveMechanismSocketOptions {
  pub enabled: bool,
  pub server_role: bool, // True if the socket is acting as a CURVE server
  pub secret_key: Option<[u8; 32]>,
  pub server_public_key: Option<[u8; 32]>, // For clients connecting to a server
}

#[cfg(feature = "noise_xx")]
#[derive(Debug, Clone, Default)]
pub struct NoiseXxSocketOptions {
  pub enabled: bool,
  pub static_secret_key_bytes: Option<[u8; 32]>,
  pub remote_static_public_key_bytes: Option<[u8; 32]>,
  // pub allowed_peers: Option<Vec<[u8; 32]>>, // If you add this later
}

/// Configuration passed to TCP Listener/Connecter for initial socket setup.
#[derive(Debug, Clone, Default)]
pub(crate) struct TcpTransportConfig {
  pub tcp_nodelay: bool,
  pub keepalive_time: Option<Duration>,
  pub keepalive_interval: Option<Duration>,
  pub keepalive_count: Option<u32>,
  pub sndbuf: Option<usize>,
  pub rcvbuf: Option<usize>,
}

#[cfg(feature = "plain")]
#[derive(Debug, Clone, Default)]
pub struct PlainMechanismSocketOptions {
  pub enabled: bool,
  pub server_role: Option<bool>,
  pub username: Option<String>,
  pub password: Option<String>,
}

// Config specific to TCP transport, potentially influenced by socket options
#[derive(Debug, Clone)]
pub(crate) struct ZmtpEngineConfig {
  /// Identity to present in READY command (for Client role)
  pub routing_id: Option<Blob>,
  /// Socket type name to include in READY command
  pub socket_type_name: String,
  pub security_enabled: bool,
  /// Allow downgrading the handshake to ZMTP/2.0 when the peer announces it.
  pub allow_zmtp2: bool,
  pub heartbeat_ivl: Option<Duration>,
  pub heartbeat_timeout: Option<Duration>,
  pub handshake_timeout: Option<Duration>,
  pub rcvtimeo: Option<Duration>,
  pub sndtimeo: Option<Duration>,
  // io-uring specific options
  pub use_send_zerocopy: bool,
  pub use_recv_multishot: bool,
  // TCP Corking
  pub use_cork: bool,
  #[cfg(feature = "noise_xx")]
  pub use_noise_xx: bool,
  #[cfg(feature = "noise_xx")]
  pub noise_xx_local_sk_bytes_for_engine: Option<[u8; 32]>,
  #[cfg(feature = "noise_xx")]
  pub noise_xx_remote_pk_bytes_for_engine: Option<[u8; 32]>,
  #[cfg(feature = "curve")]
  pub use_curve: bool,
  #[cfg(feature = "curve")]
  pub curve_local_secret_key: Option<[u8; 32]>,
  #[cfg(feature = "curve")]
  pub curve_remote_public_key: Option<[u8; 32]>,
  #[cfg(feature = "plain")]
  pub use_plain: bool,
  #[cfg(feature = "plain")]
  pub plain_username_for_engine: Option<String>,
  #[cfg(feature = "plain")]
  pub plain_password_for_engine: Option<String>,
  /// Maximum inbound frame size in bytes. -1 means unlimited.
  pub max_msg_size: i64,
  pub throttle_config: AdaptiveThrottleConfig,
  pub sndhwm: usize,
  pub rcvhwm: usize,
  pub sndbatch_count: usize,
  pub sndbatch_bytes: usize,
  pub sndbatch_bytes_physical: usize,
  pub rcvbatch_count: usize,
  pub rcvbatch_bytes: usize,
  /// Receive buffer size hint in bytes, derived from the RCVBUF socket option.
  /// Used to size read buffers and greedy-read limits. `None` means use the compile-time default.
  pub rcvbuf: Option<usize>,
  /// Minimum payload bytes before a send uses SEND_ZC instead of a copy-based send.
  /// Derived from IO_URING_ZC_SEND_THRESHOLD socket option; default 16384 bytes.
  #[cfg(feature = "io-uring")]
  pub zc_send_threshold: usize,
}

impl Default for ZmtpEngineConfig {
  fn default() -> Self {
    Self {
      routing_id: None,
      socket_type_name: String::new(),
      security_enabled: false,
      allow_zmtp2: true,
      heartbeat_ivl: None,
      heartbeat_timeout: None,
      handshake_timeout: None,
      rcvtimeo: None,
      sndtimeo: None,
      use_send_zerocopy: false,
      use_recv_multishot: false,
      use_cork: false,
      #[cfg(feature = "noise_xx")]
      use_noise_xx: false,
      #[cfg(feature = "noise_xx")]
      noise_xx_local_sk_bytes_for_engine: None,
      #[cfg(feature = "noise_xx")]
      noise_xx_remote_pk_bytes_for_engine: None,
      #[cfg(feature = "curve")]
      use_curve: false,
      #[cfg(feature = "curve")]
      curve_local_secret_key: None,
      #[cfg(feature = "curve")]
      curve_remote_public_key: None,
      #[cfg(feature = "plain")]
      use_plain: false,
      #[cfg(feature = "plain")]
      plain_username_for_engine: None,
      #[cfg(feature = "plain")]
      plain_password_for_engine: None,
      max_msg_size: -1, // -1 = unlimited, mirrors SocketOptions::maxmsgsize default
      throttle_config: AdaptiveThrottleConfig::default(),
      sndhwm: 256,
      rcvhwm: 256,
      sndbatch_count: DEFAULT_SNDBATCH_COUNT,
      sndbatch_bytes: DEFAULT_SNDBATCH_BYTES,
      sndbatch_bytes_physical: DEFAULT_SNDBATCH_BYTES + (DEFAULT_SNDBATCH_COUNT * 9),
      rcvbatch_count: DEFAULT_RCVBATCH_COUNT,
      rcvbatch_bytes: DEFAULT_RCVBATCH_BYTES,
      rcvbuf: None,
      #[cfg(feature = "io-uring")]
      zc_send_threshold: 16384,
    }
  }
}

impl From<&SocketOptions> for ZmtpEngineConfig {
  fn from(options: &SocketOptions) -> Self {
    // Determine if any security mechanism is active.
    let security_enabled = {
      #[cfg(feature = "plain")]
      {
        options.plain_options.enabled
      }
      #[cfg(not(feature = "plain"))]
      {
        false
      }
    } || {
      #[cfg(feature = "noise_xx")]
      {
        options.noise_xx_options.enabled
      }
      #[cfg(not(feature = "noise_xx"))]
      {
        false
      }
    } || {
      #[cfg(feature = "curve")]
      {
        options.curve_options.enabled
      }
      #[cfg(not(feature = "curve"))]
      {
        false
      }
    };

    // When zero-copy sends are enabled, clamp sndbatch_bytes to DEFAULT_IO_URING_SND_BUFFER_SIZE.
    // The physical pool slot is sized by calculate_required_slot_size to absorb framing overhead
    // on top of this logical payload ceiling, so no further per-message headroom deduction is
    // needed here.
    #[cfg(feature = "io-uring")]
    let sndbatch_bytes = if options.io_uring.send_zerocopy {
      let ceiling = crate::uring::DEFAULT_IO_URING_SND_BUFFER_SIZE;
      if options.sndbatch_bytes > ceiling {
        tracing::warn!(
          sndbatch_bytes = options.sndbatch_bytes,
          ceiling,
          "sndbatch_bytes clamped to io_uring logical payload ceiling to prevent ZC slot overflow"
        );
        ceiling
      } else {
        options.sndbatch_bytes
      }
    } else {
      options.sndbatch_bytes
    };
    #[cfg(not(feature = "io-uring"))]
    let sndbatch_bytes = options.sndbatch_bytes;

    // Pre-calculate physical size including worst-case framing overhead on-demand
    let sndbatch_bytes_physical =
      calculate_required_slot_size(sndbatch_bytes, options.sndbatch_count);

    ZmtpEngineConfig {
      routing_id: options.routing_id.clone(),
      socket_type_name: options.socket_type_name.clone(),
      security_enabled,
      allow_zmtp2: options.allow_zmtp2,
      heartbeat_ivl: options.heartbeat_ivl,
      heartbeat_timeout: options.heartbeat_timeout,
      handshake_timeout: options.handshake_ivl,
      rcvtimeo: options.rcvtimeo,
      sndtimeo: options.sndtimeo,
      use_send_zerocopy: options.io_uring.send_zerocopy,
      use_recv_multishot: options.io_uring.recv_multishot,
      use_cork: options.tcp_cork,
      #[cfg(feature = "noise_xx")]
      use_noise_xx: options.noise_xx_options.enabled,
      #[cfg(feature = "noise_xx")]
      noise_xx_local_sk_bytes_for_engine: options.noise_xx_options.static_secret_key_bytes,
      #[cfg(feature = "noise_xx")]
      noise_xx_remote_pk_bytes_for_engine: options.noise_xx_options.remote_static_public_key_bytes,
      #[cfg(feature = "curve")]
      use_curve: options.curve_options.enabled,
      #[cfg(feature = "curve")]
      curve_local_secret_key: options.curve_options.secret_key,
      #[cfg(feature = "curve")]
      curve_remote_public_key: options.curve_options.server_public_key,
      #[cfg(feature = "plain")]
      use_plain: options.plain_options.enabled,
      #[cfg(feature = "plain")]
      plain_username_for_engine: options.plain_options.username.clone(),
      #[cfg(feature = "plain")]
      plain_password_for_engine: options.plain_options.password.clone(),
      max_msg_size: options.maxmsgsize,
      throttle_config: options.throttle_config.clone(),
      sndhwm: options.sndhwm,
      rcvhwm: options.rcvhwm,
      sndbatch_count: options.sndbatch_count,
      sndbatch_bytes,
      sndbatch_bytes_physical,
      rcvbatch_count: options.rcvbatch_count,
      rcvbatch_bytes: options.rcvbatch_bytes,
      rcvbuf: options.rcvbuf,
      #[cfg(feature = "io-uring")]
      zc_send_threshold: options.io_uring.zc_send_threshold,
    }
  }
}

// --- Helper functions for parsing option values ---
/// Parses a byte slice representing an integer option (like HWM, linger).
pub(crate) fn parse_i32_option(value: &[u8]) -> Result<i32, ZmqError> {
  let arr: [u8; 4] = value
    .try_into()
    .map_err(|_| ZmqError::InvalidOptionValue(0))?; // Use generic error for now

  Ok(i32::from_ne_bytes(arr)) // Assuming native endianness for socket options based on ZMQ C API usage
}

/// Parses a byte slice representing a boolean option (0 or 1).
pub(crate) fn parse_bool_option(value: &[u8]) -> Result<bool, ZmqError> {
  Ok(parse_i32_option(value)? == 1)
}

/// Parses a byte slice representing a timeout or linger value in milliseconds.
/// ZMQ uses -1 for infinite, 0 for immediate (no linger), >0 for duration.
pub(crate) fn parse_duration_ms_option(value: &[u8]) -> Result<Option<Duration>, ZmqError> {
  let val = parse_i32_option(value)?;
  match val {
    -1 => Ok(None),                                     // Infinite timeout / linger
    0.. => Ok(Some(Duration::from_millis(val as u64))), // Non-negative -> Duration
    // Negative values other than -1 are invalid for timeouts/linger
    _ => Err(ZmqError::InvalidOptionValue(0)), // Use generic error
  }
}

/// Parses a byte slice representing a duration in seconds for TCP Keepalive options.
/// ZMQ uses integers for seconds. 0 might mean "use system default".
pub(crate) fn parse_secs_duration_option(value: &[u8]) -> Result<Option<Duration>, ZmqError> {
  let val = parse_i32_option(value)?;
  match val {
    0..=i32::MAX => Ok(Some(Duration::from_secs(val as u64))),
    // Negative values invalid? Or does -1 mean system default? Check ZMQ spec/impl. Assume invalid for now.
    _ => Err(ZmqError::InvalidOptionValue(0)),
  }
}

/// Parses a byte slice representing a timeout or linger value in milliseconds.
/// ZMQ uses -1 for infinite, 0 for immediate/no-wait, >0 for duration.
/// Maps to Option<Duration>: None=-1, Some(ZERO)=0, Some(>0)=millis.
pub(crate) fn parse_timeout_option(
  value: &[u8],
  option_id: i32,
) -> Result<Option<Duration>, ZmqError> {
  let val = parse_i32_option(value).map_err(|_| ZmqError::InvalidOptionValue(option_id))?;
  match val {
    -1 => Ok(None),                                     // Infinite timeout
    0 => Ok(Some(Duration::ZERO)),                      // Zero timeout (non-blocking indication)
    1.. => Ok(Some(Duration::from_millis(val as u64))), // Positive timeout
    _ => Err(ZmqError::InvalidOptionValue(option_id)),  // Other negative values invalid
  }
}

pub(crate) fn parse_linger_option(value: &[u8]) -> Result<Option<Duration>, ZmqError> {
  let val = parse_i32_option(value)?;
  match val {
    -1 => Ok(None),                                     // None represents infinite linger
    0.. => Ok(Some(Duration::from_millis(val as u64))), // Non-negative -> Duration
    _ => Err(ZmqError::InvalidOptionValue(LINGER)),     // Other negative values invalid
  }
}

/// Parses a byte slice representing a count for TCP Keepalive.
pub(crate) fn parse_u32_option(value: &[u8]) -> Result<Option<u32>, ZmqError> {
  let val = parse_i32_option(value)?; // ZMQ uses int
  match val {
    0..=i32::MAX => Ok(Some(val as u32)),
    _ => Err(ZmqError::InvalidOptionValue(0)),
  }
}

/// Parses the ZMQ_TCP_KEEPALIVE option (-1, 0, 1).
pub(crate) fn parse_keepalive_mode_option(value: &[u8]) -> Result<i32, ZmqError> {
  let val = parse_i32_option(value)?;
  if val >= -1 && val <= 1 {
    Ok(val)
  } else {
    Err(ZmqError::InvalidOptionValue(TCP_KEEPALIVE))
  }
}

/// Parses a byte slice into a Blob (for ROUTING_ID).
pub(crate) fn parse_blob_option(value: &[u8]) -> Result<Blob, ZmqError> {
  // ZMQ identities have length limits (max 255 bytes)
  if value.len() > 255 {
    Err(ZmqError::InvalidOptionValue(ROUTING_ID)) // Or specific error
  } else {
    Ok(Blob::from(value.to_vec())) // Clone into Blob
  }
}

/// Parses heartbeat interval/timeout values in milliseconds.
/// ZMQ uses 0 to disable. Negative is invalid.
pub(crate) fn parse_heartbeat_option(
  value: &[u8],
  option_id: i32,
) -> Result<Option<Duration>, ZmqError> {
  let val = parse_i32_option(value).map_err(|_| ZmqError::InvalidOptionValue(option_id))?;
  match val {
    0 => Ok(None),                                      // 0 disables heartbeat
    1.. => Ok(Some(Duration::from_millis(val as u64))), // Positive timeout
    _ => Err(ZmqError::InvalidOptionValue(option_id)),  // Negative values invalid
  }
}

pub(crate) fn parse_handshake_option(
  value: &[u8],
  option_id: i32,
) -> Result<Option<Duration>, ZmqError> {
  let val = parse_i32_option(value).map_err(|_| ZmqError::InvalidOptionValue(option_id))?;
  match val {
    0 => Ok(None),
    1.. => Ok(Some(Duration::from_millis(val as u64))), // Positive timeout
    _ => Err(ZmqError::InvalidOptionValue(option_id)),  // Negative values invalid
  }
}

pub(crate) fn parse_reconnect_ivl_option(value: &[u8]) -> Result<Option<Duration>, ZmqError> {
  let val = parse_i32_option(value)?;
  match val {
    -1 => Ok(None), // Treat -1 as disable? ZMQ uses 0. Let's use 0.
    0 => Ok(None),  // 0 disables reconnect according to ZMQ docs for IVL
    1.. => Ok(Some(Duration::from_millis(val as u64))),
    _ => Err(ZmqError::InvalidOptionValue(RECONNECT_IVL)),
  }
}

pub(crate) fn parse_reconnect_ivl_max_option(value: &[u8]) -> Result<Option<Duration>, ZmqError> {
  let val = parse_i32_option(value)?;
  match val {
    0 => Ok(Some(Duration::ZERO)), // 0 disables max/backoff according to ZMQ docs
    1.. => Ok(Some(Duration::from_millis(val as u64))),
    _ => Err(ZmqError::InvalidOptionValue(RECONNECT_IVL_MAX)),
  }
}

/// Parses ZMQ_MAXMSGSIZE: accepts -1 (unlimited) or any non-negative i64.
pub(crate) fn parse_maxmsgsize_option(value: &[u8]) -> Result<i64, ZmqError> {
  let arr: [u8; 8] = value
    .try_into()
    .map_err(|_| ZmqError::InvalidOptionValue(MAXMSGSIZE))?;
  let v = i64::from_ne_bytes(arr);
  if v < -1 {
    return Err(ZmqError::InvalidOptionValue(MAXMSGSIZE));
  }
  Ok(v)
}

pub(crate) fn parse_max_connections_option(
  value: &[u8],
  option_id: i32,
) -> Result<Option<usize>, ZmqError> {
  let val = parse_i32_option(value).map_err(|_| ZmqError::InvalidOptionValue(option_id))?;
  match val {
    -1 => Ok(None), // ZMQ often uses -1 for "no limit" or "system default"
    0 => Err(ZmqError::InvalidOptionValue(option_id)), // 0 is invalid for max connections
    1.. => Ok(Some(val as usize)),
    _ => Err(ZmqError::InvalidOptionValue(option_id)),
  }
}

/// Parses a fixed-length binary key option from a byte slice.
///
/// # Arguments
/// * `value`: The byte slice containing the key data.
/// * `option_id`: The integer ID of the socket option being parsed (for error reporting).
/// * `N`: A const generic representing the expected length of the key in bytes.
///
/// # Returns
/// `Ok([u8; N])` if the value has the correct length.
/// `Err(ZmqError::InvalidOptionValue)` if the value's length does not match `N`.
pub(crate) fn parse_key_option<const N: usize>(
  value: &[u8],
  option_id: i32,
) -> Result<[u8; N], ZmqError> {
  value.try_into().map_err(|_e| {
    // The error from try_into (TryFromSliceError) doesn't carry much info itself,
    // so we create our own ZmqError.
    tracing::error!(
      option_id = option_id,
      expected_len = N,
      actual_len = value.len(),
      "Invalid key length provided for socket option."
    );
    ZmqError::InvalidOptionValue(option_id) // Indicate which option had the invalid value
  })
}

pub(crate) fn parse_string_option(value: &[u8], option_id: i32) -> Result<String, ZmqError> {
  String::from_utf8(value.to_vec()).map_err(|_| ZmqError::InvalidOptionValue(option_id))
}

// --- New Helper Functions for Applying/Retrieving Core Options ---

/// Applies a core-level socket option value to the `SocketOptions` struct.
/// This function centralizes the logic for parsing and setting options that
/// are managed by `SocketCore` or affect its underlying configuration.
/// Pattern-specific options (like SUBSCRIBE for SUB) are handled by `ISocket::set_pattern_option`.
pub(crate) fn apply_core_option_value(
  options: &mut SocketOptions, // Mutable reference to update
  option_id: i32,
  value: &[u8],
) -> Result<(), ZmqError> {
  tracing::debug!(
    option_id,
    value_len = value.len(),
    "Applying core socket option"
  );
  match option_id {
        SNDBUF => options.sndbuf = Some(parse_i32_option(value)?.max(0) as usize),
        RCVBUF => options.rcvbuf = Some(parse_i32_option(value)?.max(0) as usize),
        SNDHWM => options.sndhwm = parse_i32_option(value)?.max(0) as usize,
        RCVHWM => options.rcvhwm = parse_i32_option(value)?.max(0) as usize,
        LINGER => options.linger = parse_linger_option(value)?,
        ROUTING_ID => options.routing_id = Some(parse_blob_option(value)?),
        RECONNECT_IVL => options.reconnect_ivl = parse_reconnect_ivl_option(value)?,
        RECONNECT_IVL_MAX => options.reconnect_ivl_max = parse_reconnect_ivl_max_option(value)?,
        RCVTIMEO => options.rcvtimeo = parse_timeout_option(value, option_id)?,
        SNDTIMEO => options.sndtimeo = parse_timeout_option(value, option_id)?,
        TCP_KEEPALIVE => options.tcp_keepalive_enabled = parse_keepalive_mode_option(value)?,
        TCP_KEEPALIVE_IDLE => options.tcp_keepalive_idle = parse_secs_duration_option(value)?,
        TCP_KEEPALIVE_CNT => options.tcp_keepalive_count = parse_u32_option(value)?,
        TCP_KEEPALIVE_INTVL => options.tcp_keepalive_interval = parse_secs_duration_option(value)?,
        HEARTBEAT_IVL => options.heartbeat_ivl = parse_heartbeat_option(value, option_id)?,
        HEARTBEAT_TIMEOUT => options.heartbeat_timeout = parse_heartbeat_option(value, option_id)?,
        HANDSHAKE_IVL => options.handshake_ivl = parse_handshake_option(value, option_id)?,
        MAXMSGSIZE => options.maxmsgsize = parse_maxmsgsize_option(value)?,
        MAX_CONNECTIONS => options.max_connections = parse_max_connections_option(value, option_id)?,
        TCP_CORK => options.tcp_cork = parse_bool_option(value)?,
        ALLOW_ZMTP2 => options.allow_zmtp2 = parse_bool_option(value)?,
        ZAP_DOMAIN => options.zap_domain = Some(parse_string_option(value, option_id)?),
        #[cfg(feature = "plain")]
        PLAIN_SERVER => {
            options.plain_options.server_role = Some(parse_bool_option(value)?);
            options.plain_options.enabled = true;
        }
        #[cfg(feature = "plain")]
        PLAIN_USERNAME => {
            options.plain_options.username = Some(parse_string_option(value, option_id)?);
            options.plain_options.enabled = true;
        }
        #[cfg(feature = "plain")]
        PLAIN_PASSWORD => {
            options.plain_options.password = Some(parse_string_option(value, option_id)?);
            options.plain_options.enabled = true;
        }
        
        #[cfg(feature = "curve")]
        CURVE_SERVER => {
          options.curve_options.server_role = parse_bool_option(value)?;
          options.curve_options.enabled = true; // Setting any CURVE option enables it
        }
        #[cfg(feature = "curve")]
        CURVE_SECRET_KEY => {
          options.curve_options.secret_key = Some(parse_key_option::<32>(value, option_id)?);
          options.curve_options.enabled = true;
        }
        #[cfg(feature = "curve")]
        CURVE_SERVER_KEY => {
          options.curve_options.server_public_key = Some(parse_key_option::<32>(value, option_id)?);
          options.curve_options.enabled = true;
        }

        #[cfg(feature = "noise_xx")]
        NOISE_XX_ENABLED => options.noise_xx_options.enabled = parse_bool_option(value)?,
        #[cfg(feature = "noise_xx")]
        NOISE_XX_STATIC_SECRET_KEY => options.noise_xx_options.static_secret_key_bytes = Some(parse_key_option::<32>(value, option_id)?),
        #[cfg(feature = "noise_xx")]
        NOISE_XX_REMOTE_STATIC_PUBLIC_KEY => options.noise_xx_options.remote_static_public_key_bytes = Some(parse_key_option::<32>(value, option_id)?),

        #[cfg(feature = "io-uring")]
        IO_URING_SESSION_ENABLED => options.io_uring.session_enabled = parse_bool_option(value)?,

        #[cfg(feature = "io-uring")]
        IO_URING_SNDZEROCOPY => options.io_uring.send_zerocopy = parse_bool_option(value)?,

        #[cfg(feature = "io-uring")]
        IO_URING_RCVMULTISHOT => options.io_uring.recv_multishot = parse_bool_option(value)?,

        #[cfg(feature = "io-uring")]
        IO_URING_ZC_SEND_THRESHOLD => options.io_uring.zc_send_threshold = parse_i32_option(value)?.max(1) as usize,

        ADAPTIVE_THROTTLE => options.throttle_config.enabled = parse_bool_option(value)?,
        SNDBATCH_COUNT => options.sndbatch_count = parse_i32_option(value)?.max(1) as usize,
        SNDBATCH_BYTES => options.sndbatch_bytes = parse_i32_option(value)?.max(1) as usize,
        RCVBATCH_COUNT => options.rcvbatch_count = parse_i32_option(value)?.max(1) as usize,
        RCVBATCH_BYTES => options.rcvbatch_bytes = parse_i32_option(value)?.max(1) as usize,

        // Options handled by pattern logic (ISocket) or read-only, or not applicable for set_option
        SUBSCRIBE | UNSUBSCRIBE | LAST_ENDPOINT  /* Pattern specific */ | ROUTER_MANDATORY |
        AUTO_DELIMITER | 16 /* ZMQ_TYPE (read-only) */ => return Err(ZmqError::UnsupportedOption(option_id)),

        _ => return Err(ZmqError::InvalidOption(option_id)), // Unknown option ID
    }
  Ok(())
}

/// Retrieves a core-level socket option value from the `SocketOptions` and `CoreState` structs.
pub(crate) fn retrieve_core_option_value(
  options: &SocketOptions,   // Read reference
  core_s_reader: &CoreState, // Read reference to CoreState for things like LAST_ENDPOINT
  option_id: i32,
) -> Result<Vec<u8>, ZmqError> {
  match option_id {
        SNDBUF => Ok(options.sndbuf.map_or(0, |v| v as i32).to_ne_bytes().to_vec()),
        RCVBUF => Ok(options.rcvbuf.map_or(0, |v| v as i32).to_ne_bytes().to_vec()),
        SNDHWM => Ok((options.sndhwm as i32).to_ne_bytes().to_vec()),
        RCVHWM => Ok((options.rcvhwm as i32).to_ne_bytes().to_vec()),
        LINGER => Ok(options.linger.map_or(-1, |d| d.as_millis().try_into().unwrap_or(i32::MAX)).to_ne_bytes().to_vec()),
        ROUTING_ID => options.routing_id.as_ref().map(|b| b.to_vec()).ok_or(ZmqError::Internal("Option ROUTING_ID not set".into())),
        RECONNECT_IVL => Ok(options.reconnect_ivl.map_or(0, |d| d.as_millis() as i32).to_ne_bytes().to_vec()), // 0 if None
        RECONNECT_IVL_MAX => Ok(options.reconnect_ivl_max.map_or(0, |d| d.as_millis() as i32).to_ne_bytes().to_vec()), // 0 if None
        RCVTIMEO => Ok(options.rcvtimeo.map_or(-1, |d| d.as_millis().try_into().unwrap_or(i32::MAX)).to_ne_bytes().to_vec()),
        SNDTIMEO => Ok(options.sndtimeo.map_or(-1, |d| d.as_millis().try_into().unwrap_or(i32::MAX)).to_ne_bytes().to_vec()),
        LAST_ENDPOINT => Ok(core_s_reader.last_bound_endpoint.as_deref().unwrap_or("").as_bytes().to_vec()),
        TCP_KEEPALIVE => Ok(options.tcp_keepalive_enabled.to_ne_bytes().to_vec()),
        TCP_KEEPALIVE_IDLE => Ok(options.tcp_keepalive_idle.map_or(0, |d| d.as_secs() as i32).to_ne_bytes().to_vec()),
        TCP_KEEPALIVE_CNT => Ok(options.tcp_keepalive_count.map_or(0, |c| c as i32).to_ne_bytes().to_vec()),
        TCP_KEEPALIVE_INTVL => Ok(options.tcp_keepalive_interval.map_or(0, |d| d.as_secs() as i32).to_ne_bytes().to_vec()),
        HEARTBEAT_IVL => Ok(options.heartbeat_ivl.map_or(0, |d| d.as_millis() as i32).to_ne_bytes().to_vec()),
        HEARTBEAT_TIMEOUT => Ok(options.heartbeat_timeout.map_or(0, |d| d.as_millis() as i32).to_ne_bytes().to_vec()),
        HANDSHAKE_IVL => Ok(options.handshake_ivl.map_or(0, |d| d.as_millis() as i32).to_ne_bytes().to_vec()),
        MAXMSGSIZE => Ok(options.maxmsgsize.to_ne_bytes().to_vec()),
        MAX_CONNECTIONS => Ok(options.max_connections.map_or(-1, |v| v as i32).to_ne_bytes().to_vec()),
        TCP_CORK => Ok((options.tcp_cork as i32).to_ne_bytes().to_vec()),
        ALLOW_ZMTP2 => Ok((options.allow_zmtp2 as i32).to_ne_bytes().to_vec()),
        ZAP_DOMAIN => options.zap_domain.as_ref().map(|s| s.as_bytes().to_vec()).ok_or(ZmqError::Internal("Option ZAP_DOMAIN not set".into())),
        #[cfg(feature = "plain")]
        PLAIN_SERVER => options.plain_options.server_role.map(|b| (b as i32).to_ne_bytes().to_vec()).ok_or(ZmqError::Internal("Option PLAIN_SERVER not set".into())),
        #[cfg(feature = "plain")]
        PLAIN_USERNAME => options.plain_options.username.as_ref().map(|s| s.as_bytes().to_vec()).ok_or(ZmqError::Internal("Option PLAIN_USERNAME not set".into())),
        #[cfg(feature = "plain")]
        PLAIN_PASSWORD => Err(ZmqError::PermissionDenied("PLAIN_PASSWORD is write-only".into())),


        #[cfg(feature = "noise_xx")]
        NOISE_XX_ENABLED => Ok((options.noise_xx_options.enabled as i32).to_ne_bytes().to_vec()),
        #[cfg(feature = "noise_xx")]
        NOISE_XX_STATIC_SECRET_KEY => Err(ZmqError::PermissionDenied("NOISE_XX_STATIC_SECRET_KEY is write-only".into())),
        #[cfg(feature = "noise_xx")]
        NOISE_XX_REMOTE_STATIC_PUBLIC_KEY => options.noise_xx_options.remote_static_public_key_bytes.map(|k| k.to_vec()).ok_or(ZmqError::Internal("Option NOISE_XX_REMOTE_STATIC_PUBLIC_KEY not set".into())),

        #[cfg(feature = "io-uring")]
        IO_URING_SESSION_ENABLED => Ok((options.io_uring.session_enabled as i32).to_ne_bytes().to_vec()),

        #[cfg(feature = "io-uring")]
        IO_URING_SNDZEROCOPY => Ok((options.io_uring.send_zerocopy as i32).to_ne_bytes().to_vec()),

        #[cfg(feature = "io-uring")]
        IO_URING_RCVMULTISHOT => Ok((options.io_uring.recv_multishot as i32).to_ne_bytes().to_vec()),

        #[cfg(feature = "io-uring")]
        IO_URING_ZC_SEND_THRESHOLD => Ok((options.io_uring.zc_send_threshold as i32).to_ne_bytes().to_vec()),

        ADAPTIVE_THROTTLE => Ok((options.throttle_config.enabled as i32).to_ne_bytes().to_vec()),
        SNDBATCH_COUNT => Ok((options.sndbatch_count as i32).to_ne_bytes().to_vec()),
        SNDBATCH_BYTES => Ok((options.sndbatch_bytes as i32).to_ne_bytes().to_vec()),
        RCVBATCH_COUNT => Ok((options.rcvbatch_count as i32).to_ne_bytes().to_vec()),
        RCVBATCH_BYTES => Ok((options.rcvbatch_bytes as i32).to_ne_bytes().to_vec()),

        // Options handled by pattern logic or read-only by nature
        16 /* ZMQ_TYPE */ => Ok((core_s_reader.socket_type as i32).to_ne_bytes().to_vec()),
        SUBSCRIBE | UNSUBSCRIBE | ROUTER_MANDATORY | AUTO_DELIMITER => Err(ZmqError::UnsupportedOption(option_id)), // Pattern specific

        _ => Err(ZmqError::InvalidOption(option_id)),
    }
}

/// Calculates the minimum physical send-buffer slot size needed to hold a fully-framed
/// ZMTP batch without allocation fallback.
///
/// Each ZMTP long frame (payload ≥ 256 bytes) costs 9 bytes of overhead; each short frame
/// costs 2 bytes. This function computes the worst-case total and rounds up to a 4 KB page
/// boundary for kernel/MMU efficiency.
///
/// Example: `calculate_required_slot_size(65_536, 128)` → 69_632 bytes (68 KB).
pub fn calculate_required_slot_size(target_payload_bytes: usize, max_batch_count: usize) -> usize {
  let max_long_frames = std::cmp::min(max_batch_count, target_payload_bytes / 256);
  let long_frame_overhead = max_long_frames * 9;
  let short_frame_overhead = max_batch_count.saturating_sub(max_long_frames) * 2;
  let raw_physical_size = target_payload_bytes + long_frame_overhead + short_frame_overhead;
  let page_size = unsafe { libc::sysconf(libc::_SC_PAGESIZE) as usize };
  ((raw_physical_size + page_size - 1) / page_size) * page_size
}

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

  #[test]
  fn test_allow_zmtp2_default_true() {
    assert!(SocketOptions::default().allow_zmtp2);
    assert!(ZmtpEngineConfig::default().allow_zmtp2);
  }

  #[test]
  fn test_allow_zmtp2_apply_and_map_through() {
    let mut opts = SocketOptions::default();
    apply_core_option_value(&mut opts, ALLOW_ZMTP2, &0i32.to_ne_bytes()).unwrap();
    assert!(!opts.allow_zmtp2);

    let cfg = ZmtpEngineConfig::from(&opts);
    assert!(!cfg.allow_zmtp2, "From<&SocketOptions> must carry allow_zmtp2 through");

    apply_core_option_value(&mut opts, ALLOW_ZMTP2, &1i32.to_ne_bytes()).unwrap();
    assert!(opts.allow_zmtp2);
    assert!(ZmtpEngineConfig::from(&opts).allow_zmtp2);
  }
}