nlink 0.13.0

//! High-level netlink connection with request/response handling.

use std::{os::unix::io::RawFd, path::Path, time::Duration};

use tracing::{instrument, warn};

use super::{
    builder::MessageBuilder,
    error::{Error, Result},
    interface_ref::InterfaceRef,
    message::{
        MessageIter, NLM_F_ACK, NLM_F_DUMP, NLM_F_REQUEST, NLMSG_HDRLEN, NlMsgError, NlMsgHdr,
        NlMsgType,
    },
    parse::FromNetlink,
    protocol::{ProtocolState, Route},
    socket::NetlinkSocket,
    tc_handle::TcHandle,
};

/// High-level netlink connection parameterized by protocol state.
///
/// The type parameter `P` determines which protocol this connection uses
/// and which methods are available:
///
/// - [`Connection<Route>`]: RTNetlink for interfaces, addresses, routes, TC
/// - [`Connection<Generic>`]: Generic netlink for WireGuard, MACsec, etc.
///
/// # Construction
///
/// **Sync protocols** — use [`Connection::new()`]:
/// `Route`, `SockDiag`, `Generic`, `Nftables`
///
/// **GENL protocols** — use `Connection::new_async().await`:
/// `Wireguard`, `Macsec`, `Mptcp`, `Ethtool`, `Nl80211`, `Devlink`
///
/// These require async construction to resolve their Generic Netlink family ID.
/// While `Connection::new()` compiles for them (they implement `Default`), the
/// resulting connection will have an unresolved family ID and will not work.
///
/// **Other protocols** have their own constructors (e.g., `Connector`, `KobjectUevent`).
///
/// # Example
///
/// ```ignore
/// use nlink::netlink::{Connection, Route, Generic, Wireguard};
///
/// // Sync construction (Route, Generic, Nftables, SockDiag)
/// let route = Connection::<Route>::new()?;
/// let genl = Connection::<Generic>::new()?;
///
/// // Async construction (Wireguard, Macsec, Mptcp, Ethtool, Nl80211, Devlink)
/// let wg = Connection::<Wireguard>::new_async().await?;
/// ```
pub struct Connection<P: ProtocolState> {
    socket: NetlinkSocket,
    state: P,
    timeout: Option<Duration>,
}

// ============================================================================
// Shared methods for protocol types that implement Default
// ============================================================================

impl<P: ProtocolState + Default> Connection<P> {
    /// Create a new connection for this protocol type.
    ///
    /// This is available for protocols that implement `Default`, but should
    /// only be used for sync protocol types (`Route`, `SockDiag`, `Generic`,
    /// `Nftables`).
    ///
    /// # Important
    ///
    /// **Do not use this for GENL protocol types** (`Wireguard`, `Macsec`,
    /// `Mptcp`, `Ethtool`, `Nl80211`, `Devlink`). While it compiles, the
    /// connection will have an unresolved family ID (0) and operations will
    /// fail. Use `Connection::<Wireguard>::new_async().await?` instead.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::netlink::{Connection, Route, Generic};
    ///
    /// let route = Connection::<Route>::new()?;
    /// let genl = Connection::<Generic>::new()?;
    /// ```
    #[instrument(level = "info", skip_all, fields(protocol = std::any::type_name::<P>()))]
    pub fn new() -> Result<Self> {
        Ok(Self {
            socket: NetlinkSocket::new(P::PROTOCOL)?,
            state: P::default(),
            timeout: None,
        })
    }

    /// Create a connection that operates in a specific network namespace.
    ///
    /// The namespace is specified by an open file descriptor to a namespace file
    /// (e.g., `/proc/<pid>/ns/net` or `/var/run/netns/<name>`).
    ///
    /// # Example
    ///
    /// ```ignore
    /// use std::fs::File;
    /// use std::os::unix::io::AsRawFd;
    /// use nlink::netlink::{Connection, Route};
    ///
    /// let ns_file = File::open("/var/run/netns/myns")?;
    /// let conn = Connection::<Route>::new_in_namespace(ns_file.as_raw_fd())?;
    ///
    /// // All operations now occur in the "myns" namespace
    /// let links = conn.get_links().await?;
    /// ```
    #[instrument(level = "info", skip_all, fields(protocol = std::any::type_name::<P>(), ns_fd))]
    pub fn new_in_namespace(ns_fd: RawFd) -> Result<Self> {
        Ok(Self {
            socket: NetlinkSocket::new_in_namespace(P::PROTOCOL, ns_fd)?,
            state: P::default(),
            timeout: None,
        })
    }

    /// Create a connection that operates in a network namespace specified by path.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::netlink::{Connection, Route};
    ///
    /// // For a named namespace (created via `ip netns add myns`)
    /// let conn = Connection::<Route>::new_in_namespace_path("/var/run/netns/myns")?;
    ///
    /// // For a container's namespace
    /// let conn = Connection::<Route>::new_in_namespace_path("/proc/1234/ns/net")?;
    ///
    /// // Query interfaces in that namespace
    /// let links = conn.get_links().await?;
    /// ```
    #[instrument(level = "info", skip_all, fields(protocol = std::any::type_name::<P>(), ns_path = %ns_path.as_ref().display()))]
    pub fn new_in_namespace_path<T: AsRef<Path>>(ns_path: T) -> Result<Self> {
        Ok(Self {
            socket: NetlinkSocket::new_in_namespace_path(P::PROTOCOL, ns_path)?,
            state: P::default(),
            timeout: None,
        })
    }
}

// ============================================================================
// Shared methods for all protocol types
// ============================================================================

impl<P: ProtocolState> Connection<P> {
    /// Get the underlying socket.
    pub fn socket(&self) -> &NetlinkSocket {
        &self.socket
    }

    /// Get a mutable reference to the underlying socket.
    pub(crate) fn socket_mut(&mut self) -> &mut NetlinkSocket {
        &mut self.socket
    }

    /// Get the protocol state.
    pub fn state(&self) -> &P {
        &self.state
    }

    /// Set a default timeout for all netlink operations.
    ///
    /// Operations that exceed the timeout return [`Error::Timeout`].
    /// By default, no timeout is set and operations wait indefinitely.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::{Connection, Route};
    /// use std::time::Duration;
    ///
    /// let conn = Connection::<Route>::new()?
    ///     .timeout(Duration::from_secs(5));
    /// ```
    pub fn timeout(mut self, timeout: Duration) -> Self {
        self.timeout = Some(timeout);
        self
    }

    /// Clear the timeout (operations will wait indefinitely).
    pub fn no_timeout(mut self) -> Self {
        self.timeout = None;
        self
    }

    /// Get the configured timeout.
    pub fn get_timeout(&self) -> Option<Duration> {
        self.timeout
    }

    /// Wrap a future with the configured timeout.
    ///
    /// If no timeout is set, the future runs without time limit.
    /// On timeout, returns [`Error::Timeout`].
    pub(crate) async fn with_timeout<F, T>(&self, fut: F) -> Result<T>
    where
        F: std::future::Future<Output = Result<T>>,
    {
        match self.timeout {
            Some(dur) => tokio::time::timeout(dur, fut)
                .await
                .map_err(|_| Error::Timeout)?,
            None => fut.await,
        }
    }

    /// Create a connection from its parts.
    ///
    /// This is primarily used internally for protocols that require
    /// async initialization (like WireGuard which needs family ID resolution).
    pub(crate) fn from_parts(socket: NetlinkSocket, state: P) -> Self {
        Self {
            socket,
            state,
            timeout: None,
        }
    }

    // ========================================================================
    // Internal request methods (pub(crate) - not part of public API)
    // ========================================================================

    /// Send a request and wait for a response.
    ///
    /// Respects the configured timeout. This is a low-level method.
    /// Prefer using typed methods like `get_links()`, `add_route()`, etc.
    pub(crate) async fn send_request(&self, builder: MessageBuilder) -> Result<Vec<u8>> {
        self.with_timeout(self.send_request_inner(builder)).await
    }

    /// Send a request that expects an ACK only (no data response).
    ///
    /// Respects the configured timeout. This is a low-level method.
    /// Prefer using typed methods like `add_link()`, `del_route()`, etc.
    pub(crate) async fn send_ack(&self, builder: MessageBuilder) -> Result<()> {
        self.with_timeout(self.send_ack_inner(builder)).await
    }

    /// Send a dump request and collect all responses.
    ///
    /// Respects the configured timeout. This is a low-level method.
    /// Prefer using typed methods like `get_links()`, `get_routes()`, etc.
    pub(crate) async fn send_dump(&self, builder: MessageBuilder) -> Result<Vec<Vec<u8>>> {
        self.with_timeout(self.send_dump_inner(builder)).await
    }

    #[instrument(level = "trace", skip_all, fields(seq))]
    async fn send_request_inner(&self, mut builder: MessageBuilder) -> Result<Vec<u8>> {
        let seq = self.socket.next_seq();
        builder.set_seq(seq);
        builder.set_pid(self.socket.pid());
        tracing::Span::current().record("seq", seq);

        let msg = builder.finish();
        self.socket.send(&msg).await?;

        let response = self.socket.recv_msg().await?;
        self.process_response(&response, seq).inspect_err(|e| {
            warn!(errno = ?e.errno(), "kernel returned error for request");
        })?;

        Ok(response)
    }

    #[instrument(level = "trace", skip_all, fields(seq))]
    async fn send_ack_inner(&self, mut builder: MessageBuilder) -> Result<()> {
        let seq = self.socket.next_seq();
        builder.set_seq(seq);
        builder.set_pid(self.socket.pid());
        tracing::Span::current().record("seq", seq);

        let msg = builder.finish();
        self.socket.send(&msg).await?;

        let response = self.socket.recv_msg().await?;
        self.process_ack(&response, seq).inspect_err(|e| {
            warn!(errno = ?e.errno(), "kernel returned error for ack");
        })?;

        Ok(())
    }

    #[instrument(level = "trace", skip_all, fields(seq, responses))]
    async fn send_dump_inner(&self, mut builder: MessageBuilder) -> Result<Vec<Vec<u8>>> {
        let seq = self.socket.next_seq();
        builder.set_seq(seq);
        builder.set_pid(self.socket.pid());
        tracing::Span::current().record("seq", seq);

        let msg = builder.finish();
        self.socket.send(&msg).await?;

        let mut responses = Vec::new();

        loop {
            let data = self.socket.recv_msg().await?;
            let mut done = false;

            for result in MessageIter::new(&data) {
                let (header, payload) = result?;

                // Check sequence number
                if header.nlmsg_seq != seq {
                    continue;
                }

                if header.is_error() {
                    let err = NlMsgError::from_bytes(payload)?;
                    if !err.is_ack() {
                        return Err(Error::from_errno(err.error));
                    }
                }

                if header.is_done() {
                    done = true;
                    break;
                }

                // Collect the full message (header + payload)
                let msg_len = header.nlmsg_len as usize;
                let msg_start = payload.as_ptr() as usize
                    - data.as_ptr() as usize
                    - std::mem::size_of::<NlMsgHdr>();
                if msg_start + msg_len <= data.len() {
                    responses.push(data[msg_start..msg_start + msg_len].to_vec());
                }
            }

            if done {
                break;
            }
        }

        tracing::Span::current().record("responses", responses.len());
        Ok(responses)
    }

    /// Process a response and check for errors.
    fn process_response(&self, data: &[u8], expected_seq: u32) -> Result<()> {
        for result in MessageIter::new(data) {
            let (header, payload) = result?;

            if header.nlmsg_seq != expected_seq {
                continue;
            }

            if header.is_error() {
                let err = NlMsgError::from_bytes(payload)?;
                if !err.is_ack() {
                    return Err(Error::from_errno(err.error));
                }
            }
        }

        Ok(())
    }

    /// Process an ACK response.
    fn process_ack(&self, data: &[u8], expected_seq: u32) -> Result<()> {
        for result in MessageIter::new(data) {
            let (header, payload) = result?;

            if header.nlmsg_seq != expected_seq {
                continue;
            }

            if header.is_error() {
                let err = NlMsgError::from_bytes(payload)?;
                if !err.is_ack() {
                    return Err(Error::from_errno(err.error));
                }
                return Ok(());
            }
        }

        Err(Error::InvalidMessage("expected ACK message".into()))
    }
}

// ============================================================================
// Route protocol multicast groups
// ============================================================================

/// Multicast groups for Route protocol event notifications.
///
/// Use with [`Connection<Route>::subscribe`] to receive specific event types.
///
/// # Example
///
/// ```ignore
/// use nlink::netlink::{Connection, Route, RtnetlinkGroup};
///
/// let mut conn = Connection::<Route>::new()?;
/// conn.subscribe(&[RtnetlinkGroup::Link, RtnetlinkGroup::Tc])?;
/// let mut events = conn.events();
/// ```
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[non_exhaustive]
pub enum RtnetlinkGroup {
    /// Link (interface) state changes (RTM_NEWLINK, RTM_DELLINK).
    Link,
    /// IPv4 address changes (RTM_NEWADDR, RTM_DELADDR).
    Ipv4Addr,
    /// IPv6 address changes (RTM_NEWADDR, RTM_DELADDR).
    Ipv6Addr,
    /// IPv4 route changes (RTM_NEWROUTE, RTM_DELROUTE).
    Ipv4Route,
    /// IPv6 route changes (RTM_NEWROUTE, RTM_DELROUTE).
    Ipv6Route,
    /// Neighbor (ARP/NDP) cache changes (RTM_NEWNEIGH, RTM_DELNEIGH).
    Neigh,
    /// Traffic control changes (qdiscs, classes, filters, actions).
    Tc,
    /// Namespace ID changes (RTM_NEWNSID, RTM_DELNSID).
    NsId,
    /// IPv4 routing rule changes (RTM_NEWRULE, RTM_DELRULE).
    Ipv4Rule,
    /// IPv6 routing rule changes (RTM_NEWRULE, RTM_DELRULE).
    Ipv6Rule,
}

impl RtnetlinkGroup {
    /// Convert to the raw netlink multicast group number.
    fn to_group(self) -> u32 {
        use super::socket::rtnetlink_groups::*;
        match self {
            Self::Link => RTNLGRP_LINK,
            Self::Ipv4Addr => RTNLGRP_IPV4_IFADDR,
            Self::Ipv6Addr => RTNLGRP_IPV6_IFADDR,
            Self::Ipv4Route => RTNLGRP_IPV4_ROUTE,
            Self::Ipv6Route => RTNLGRP_IPV6_ROUTE,
            Self::Neigh => RTNLGRP_NEIGH,
            Self::Tc => RTNLGRP_TC,
            Self::NsId => RTNLGRP_NSID,
            Self::Ipv4Rule => RTNLGRP_IPV4_RULE,
            Self::Ipv6Rule => RTNLGRP_IPV6_RULE,
        }
    }
}

// ============================================================================
// Route protocol specific methods
// ============================================================================

impl Connection<Route> {
    /// Create a connection for the specified namespace.
    ///
    /// This is a convenience method that creates a Route protocol connection
    /// for any namespace specification.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::netlink::{Connection, Route};
    /// use nlink::netlink::namespace::NamespaceSpec;
    ///
    /// // For a named namespace
    /// let conn = Connection::<Route>::for_namespace(NamespaceSpec::Named("myns"))?;
    ///
    /// // For a container by PID
    /// let conn = Connection::<Route>::for_namespace(NamespaceSpec::Pid(1234))?;
    ///
    /// // For the default namespace
    /// let conn = Connection::<Route>::for_namespace(NamespaceSpec::Default)?;
    /// ```
    pub fn for_namespace(spec: super::namespace::NamespaceSpec<'_>) -> Result<Self> {
        spec.connection()
    }

    /// Subscribe to multicast groups for event notifications.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::netlink::{Connection, Route, RtnetlinkGroup};
    ///
    /// let mut conn = Connection::<Route>::new()?;
    /// conn.subscribe(&[RtnetlinkGroup::Link, RtnetlinkGroup::Tc])?;
    /// ```
    #[instrument(level = "info", skip(self), fields(groups = ?groups))]
    pub fn subscribe(&mut self, groups: &[RtnetlinkGroup]) -> Result<()> {
        for group in groups {
            self.socket.add_membership(group.to_group())?;
        }
        Ok(())
    }

    /// Subscribe to all commonly-used event groups.
    ///
    /// Subscribes to: Link, Ipv4Addr, Ipv6Addr, Ipv4Route, Ipv6Route, Neigh, Tc.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::netlink::{Connection, Route};
    ///
    /// let mut conn = Connection::<Route>::new()?;
    /// conn.subscribe_all()?;
    /// let mut events = conn.events();
    /// ```
    pub fn subscribe_all(&mut self) -> Result<()> {
        self.subscribe(&[
            RtnetlinkGroup::Link,
            RtnetlinkGroup::Ipv4Addr,
            RtnetlinkGroup::Ipv6Addr,
            RtnetlinkGroup::Ipv4Route,
            RtnetlinkGroup::Ipv6Route,
            RtnetlinkGroup::Neigh,
            RtnetlinkGroup::Tc,
        ])
    }

    // ========================================================================
    // Strongly-typed API for Route protocol
    // ========================================================================

    /// Send a dump request and parse all responses into typed messages.
    ///
    /// This is a convenience method that combines `dump()` with parsing.
    /// The type T must implement `FromNetlink::write_dump_header` to provide
    /// the required message header (e.g., IfInfoMsg for links, IfAddrMsg for addresses).
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::netlink::messages::AddressMessage;
    /// use nlink::netlink::message::NlMsgType;
    ///
    /// let addresses: Vec<AddressMessage> = conn.dump_typed(NlMsgType::RTM_GETADDR).await?;
    /// for addr in addresses {
    ///     println!("{}: {:?}", addr.ifindex(), addr.address);
    /// }
    /// ```
    pub async fn dump_typed<T: FromNetlink>(&self, msg_type: u16) -> Result<Vec<T>> {
        let mut builder = dump_request(msg_type);

        // Get the header from the type and append it to the request
        let mut header_buf = Vec::new();
        T::write_dump_header(&mut header_buf);
        builder.append_bytes(&header_buf);

        let responses = self.send_dump(builder).await?;

        let mut parsed = Vec::with_capacity(responses.len());
        for response in responses {
            if response.len() < NLMSG_HDRLEN {
                continue;
            }
            let payload = &response[NLMSG_HDRLEN..];
            if let Ok(msg) = T::from_bytes(payload) {
                parsed.push(msg);
            }
        }

        Ok(parsed)
    }

    /// Parse a single response into a typed message.
    pub fn parse_response<T: FromNetlink>(&self, response: &[u8]) -> Result<T> {
        if response.len() < NLMSG_HDRLEN {
            return Err(Error::Truncated {
                expected: NLMSG_HDRLEN,
                actual: response.len(),
            });
        }
        let payload = &response[NLMSG_HDRLEN..];
        T::from_bytes(payload)
    }
}

/// Helper to build a dump request.
pub(crate) fn dump_request(msg_type: u16) -> MessageBuilder {
    MessageBuilder::new(msg_type, NLM_F_REQUEST | NLM_F_DUMP)
}

/// Helper to build a request expecting ACK.
pub(crate) fn ack_request(msg_type: u16) -> MessageBuilder {
    MessageBuilder::new(msg_type, NLM_F_REQUEST | NLM_F_ACK)
}

/// Helper to build a create request.
pub(crate) fn create_request(msg_type: u16) -> MessageBuilder {
    MessageBuilder::new(msg_type, NLM_F_REQUEST | NLM_F_ACK | 0x400) // NLM_F_CREATE
}

/// Helper to build a create-or-replace request.
pub(crate) fn replace_request(msg_type: u16) -> MessageBuilder {
    MessageBuilder::new(msg_type, NLM_F_REQUEST | NLM_F_ACK | 0x400 | 0x100) // NLM_F_CREATE | NLM_F_REPLACE
}

// ============================================================================
// Batch Operations
// ============================================================================

impl Connection<Route> {
    /// Create a batch for executing multiple operations in minimal syscalls.
    ///
    /// Operations are buffered and sent as concatenated messages in a single
    /// `sendmsg()`. The kernel processes them sequentially and returns one
    /// ACK per message.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::netlink::{Connection, Route};
    /// use nlink::netlink::route::Ipv4Route;
    ///
    /// let conn = Connection::<Route>::new()?;
    /// let results = conn.batch()
    ///     .add_route(Ipv4Route::new("10.0.0.0", 8).dev_index(5))
    ///     .add_route(Ipv4Route::new("10.1.0.0", 16).dev_index(5))
    ///     .execute()
    ///     .await?;
    /// ```
    pub fn batch(&self) -> super::batch::Batch<'_> {
        super::batch::Batch::new(self)
    }
}

// ============================================================================
// Convenience Query Methods
// ============================================================================

use super::messages::{
    AddressMessage, LinkMessage, NeighborMessage, RouteMessage, RuleMessage, TcMessage,
};

impl Connection<Route> {
    /// Get all network interfaces.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let links = conn.get_links().await?;
    /// for link in links {
    ///     println!("{}: {}", link.ifindex(), link.name.as_deref().unwrap_or("?"));
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_links"))]
    pub async fn get_links(&self) -> Result<Vec<LinkMessage>> {
        self.dump_typed(NlMsgType::RTM_GETLINK).await
    }

    /// Get a network interface by name.
    ///
    /// Returns `None` if the interface doesn't exist.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_link_by_name"))]
    pub async fn get_link_by_name(
        &self,
        name: impl Into<InterfaceRef>,
    ) -> Result<Option<LinkMessage>> {
        let iface = name.into();
        match iface {
            InterfaceRef::Name(ref name_str) => {
                let links = self.get_links().await?;
                Ok(links
                    .into_iter()
                    .find(|l| l.name.as_deref() == Some(name_str)))
            }
            InterfaceRef::Index(idx) => self.get_link_by_index(idx).await,
        }
    }

    /// Get a network interface by index.
    ///
    /// Returns `None` if the interface doesn't exist.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_link_by_index"))]
    pub async fn get_link_by_index(&self, index: u32) -> Result<Option<LinkMessage>> {
        let links = self.get_links().await?;
        Ok(links.into_iter().find(|l| l.ifindex() == index))
    }

    /// Resolve an interface reference to an index.
    ///
    /// This method is namespace-safe: it uses netlink to resolve interface names,
    /// which queries the namespace that this connection is bound to.
    ///
    /// - If the reference is already an index, returns it directly.
    /// - If the reference is a name, queries the kernel via netlink.
    ///
    /// # Errors
    ///
    /// Returns [`Error::InterfaceNotFound`] if the interface name doesn't exist.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::netlink::{Connection, Route, InterfaceRef};
    ///
    /// let conn = Connection::<Route>::new()?;
    ///
    /// // Resolve a name
    /// let ifindex = conn.resolve_interface(&InterfaceRef::name("eth0")).await?;
    ///
    /// // Pass-through an index
    /// let ifindex = conn.resolve_interface(&InterfaceRef::index(2)).await?;
    /// assert_eq!(ifindex, 2);
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "resolve_interface"))]
    pub async fn resolve_interface(&self, iface: &InterfaceRef) -> Result<u32> {
        match iface {
            InterfaceRef::Index(idx) => Ok(*idx),
            InterfaceRef::Name(name) => {
                let link = self
                    .get_link_by_name(name)
                    .await?
                    .ok_or_else(|| Error::interface_not_found(name))?;
                Ok(link.ifindex())
            }
        }
    }

    /// Resolve an optional interface reference.
    ///
    /// Returns `None` if the input is `None`, otherwise resolves the reference.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "resolve_interface_opt"))]
    pub async fn resolve_interface_opt(&self, iface: Option<&InterfaceRef>) -> Result<Option<u32>> {
        match iface {
            Some(iface) => Ok(Some(self.resolve_interface(iface).await?)),
            None => Ok(None),
        }
    }

    /// Build a map of interface index to name.
    ///
    /// This is a convenience method for code that needs to look up interface
    /// names by index (e.g., when displaying addresses, routes, or TC objects).
    ///
    /// # Example
    ///
    /// ```ignore
    /// let names = conn.get_interface_names().await?;
    /// let addresses = conn.get_addresses().await?;
    /// for addr in addresses {
    ///     let name = names.get(&addr.ifindex()).map(|s| s.as_str()).unwrap_or("?");
    ///     println!("{}: {:?}", name, addr.address);
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_interface_names"))]
    pub async fn get_interface_names(&self) -> Result<std::collections::HashMap<u32, String>> {
        let links = self.get_links().await?;
        Ok(links
            .into_iter()
            .filter_map(|l| l.name.clone().map(|n| (l.ifindex(), n)))
            .collect())
    }

    /// Get interface name by index.
    ///
    /// This is a convenience method for getting a single interface name.
    /// For looking up multiple names, prefer [`Connection::get_interface_names()`]
    /// to build a lookup map.
    ///
    /// Returns `None` if no interface with that index exists.
    ///
    /// # Example
    ///
    /// ```ignore
    /// if let Some(name) = conn.interface_name(route.oif.unwrap()).await? {
    ///     println!("Route via {}", name);
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "interface_name"))]
    pub async fn interface_name(&self, ifindex: u32) -> Result<Option<String>> {
        let link = self.get_link_by_index(ifindex).await?;
        Ok(link.and_then(|l| l.name))
    }

    /// Get interface name by index, or return a default value.
    ///
    /// This is a convenience method for display purposes when you want
    /// a fallback value like "-" or "?" for unknown interfaces.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let dev = conn.interface_name_or(route.oif.unwrap_or(0), "-").await?;
    /// println!("Route via {}", dev);
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "interface_name_or"))]
    pub async fn interface_name_or(&self, ifindex: u32, default: &str) -> Result<String> {
        Ok(self
            .interface_name(ifindex)
            .await?
            .unwrap_or_else(|| default.to_string()))
    }

    /// Get bond information for a bond interface.
    ///
    /// Returns the bond configuration as reported by the kernel.
    ///
    /// # Errors
    ///
    /// Returns an error if the interface doesn't exist or is not a bond.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let info = conn.get_bond_info("bond0").await?;
    /// println!("Mode: {:?}, miimon: {}ms", info.bond_mode(), info.miimon);
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_bond_info"))]
    pub async fn get_bond_info(
        &self,
        iface: impl Into<InterfaceRef>,
    ) -> Result<crate::netlink::messages::BondInfo> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        let link = self
            .get_link_by_index(ifindex)
            .await?
            .ok_or_else(|| Error::InvalidMessage("interface not found".into()))?;
        link.bond_info()
            .ok_or_else(|| Error::InvalidMessage("not a bond interface".into()))
    }

    /// List all slaves of a bond interface with their status.
    ///
    /// Returns a list of `(LinkMessage, BondSlaveInfo)` pairs for each slave.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let slaves = conn.get_bond_slaves("bond0").await?;
    /// for (link, info) in &slaves {
    ///     println!("{}: state={:?}, mii={:?}",
    ///         link.name_or("?"), info.state, info.mii_status);
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_bond_slaves"))]
    pub async fn get_bond_slaves(
        &self,
        bond: impl Into<InterfaceRef>,
    ) -> Result<Vec<(LinkMessage, crate::netlink::messages::BondSlaveInfo)>> {
        let bond_ifindex = self.resolve_interface(&bond.into()).await?;
        let all_links = self.get_links().await?;
        let mut slaves = Vec::new();

        for link in all_links {
            if link.master() == Some(bond_ifindex)
                && let Some(info) = link.bond_slave_info()
            {
                slaves.push((link, info));
            }
        }

        Ok(slaves)
    }

    /// Get all IP addresses.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let addresses = conn.get_addresses().await?;
    /// for addr in addresses {
    ///     println!("{:?}/{} on idx {}", addr.address, addr.prefix_len(), addr.ifindex());
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_addresses"))]
    pub async fn get_addresses(&self) -> Result<Vec<AddressMessage>> {
        self.dump_typed(NlMsgType::RTM_GETADDR).await
    }

    /// Get IP addresses for a specific interface.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_addresses_by_name"))]
    pub async fn get_addresses_by_name(
        &self,
        iface: impl Into<InterfaceRef>,
    ) -> Result<Vec<AddressMessage>> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.get_addresses_by_index(ifindex).await
    }

    /// Get IP addresses for a specific interface by index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_addresses_by_index"))]
    pub async fn get_addresses_by_index(&self, ifindex: u32) -> Result<Vec<AddressMessage>> {
        let addresses = self.get_addresses().await?;
        Ok(addresses
            .into_iter()
            .filter(|a| a.ifindex() == ifindex)
            .collect())
    }

    /// Get an address entry by IP address.
    ///
    /// Returns `None` if no address entry matches the given IP.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use std::net::IpAddr;
    ///
    /// let ip: IpAddr = "192.168.1.100".parse()?;
    /// if let Some(addr) = conn.get_address_by_ip(ip).await? {
    ///     println!("Found on interface index {}", addr.ifindex());
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_address_by_ip"))]
    pub async fn get_address_by_ip(
        &self,
        addr: std::net::IpAddr,
    ) -> Result<Option<AddressMessage>> {
        let addresses = self.get_addresses().await?;
        Ok(addresses.into_iter().find(|a| a.address == Some(addr)))
    }

    /// Get all routes.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let routes = conn.get_routes().await?;
    /// for route in routes {
    ///     println!("{:?}/{}", route.destination(), route.dst_len());
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_routes"))]
    pub async fn get_routes(&self) -> Result<Vec<RouteMessage>> {
        self.dump_typed(NlMsgType::RTM_GETROUTE).await
    }

    /// Get routes for a specific table.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_routes_for_table"))]
    pub async fn get_routes_for_table(&self, table_id: u32) -> Result<Vec<RouteMessage>> {
        let routes = self.get_routes().await?;
        Ok(routes
            .into_iter()
            .filter(|r| r.table_id() == table_id)
            .collect())
    }

    /// Get a specific IPv4 route by destination and prefix length.
    ///
    /// Uses RTM_GETROUTE without NLM_F_DUMP to query the kernel directly,
    /// which is more efficient than dumping all routes for large routing tables.
    ///
    /// Returns `None` if no matching route is found.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use std::net::Ipv4Addr;
    ///
    /// // Look up route to 10.0.0.0/8
    /// if let Some(route) = conn.get_route_v4(Ipv4Addr::new(10, 0, 0, 0), 8).await? {
    ///     println!("Gateway: {:?}", route.gateway);
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_route_v4"))]
    pub async fn get_route_v4(
        &self,
        destination: std::net::Ipv4Addr,
        prefix_len: u8,
    ) -> Result<Option<RouteMessage>> {
        use crate::netlink::types::route::{RtMsg, RtaAttr};

        // Build RTM_GETROUTE request WITHOUT NLM_F_DUMP
        let mut builder = MessageBuilder::new(NlMsgType::RTM_GETROUTE, NLM_F_REQUEST);

        let rtmsg = RtMsg::new()
            .with_family(libc::AF_INET as u8)
            .with_dst_len(prefix_len);
        builder.append(&rtmsg);
        builder.append_attr(RtaAttr::Dst as u16, &destination.octets());

        // Send single request (not dump)
        match self.send_request(builder).await {
            Ok(response) => {
                // Parse the response - skip netlink header
                if response.len() >= NLMSG_HDRLEN {
                    let payload = &response[NLMSG_HDRLEN..];
                    if let Ok(msg) = RouteMessage::from_bytes(payload) {
                        return Ok(Some(msg));
                    }
                }
                Ok(None)
            }
            Err(e) if e.is_not_found() => Ok(None),
            Err(e) => Err(e),
        }
    }

    /// Get a specific IPv6 route by destination and prefix length.
    ///
    /// Uses RTM_GETROUTE without NLM_F_DUMP to query the kernel directly,
    /// which is more efficient than dumping all routes for large routing tables.
    ///
    /// Returns `None` if no matching route is found.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use std::net::Ipv6Addr;
    ///
    /// // Look up route to 2001:db8::/32
    /// let dest: Ipv6Addr = "2001:db8::".parse()?;
    /// if let Some(route) = conn.get_route_v6(dest, 32).await? {
    ///     println!("Gateway: {:?}", route.gateway);
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_route_v6"))]
    pub async fn get_route_v6(
        &self,
        destination: std::net::Ipv6Addr,
        prefix_len: u8,
    ) -> Result<Option<RouteMessage>> {
        use crate::netlink::types::route::{RtMsg, RtaAttr};

        let mut builder = MessageBuilder::new(NlMsgType::RTM_GETROUTE, NLM_F_REQUEST);

        let rtmsg = RtMsg::new()
            .with_family(libc::AF_INET6 as u8)
            .with_dst_len(prefix_len);
        builder.append(&rtmsg);
        builder.append_attr(RtaAttr::Dst as u16, &destination.octets());

        match self.send_request(builder).await {
            Ok(response) => {
                if response.len() >= NLMSG_HDRLEN {
                    let payload = &response[NLMSG_HDRLEN..];
                    if let Ok(msg) = RouteMessage::from_bytes(payload) {
                        return Ok(Some(msg));
                    }
                }
                Ok(None)
            }
            Err(e) if e.is_not_found() => Ok(None),
            Err(e) => Err(e),
        }
    }

    /// Get all neighbor entries.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let neighbors = conn.get_neighbors().await?;
    /// for neigh in neighbors {
    ///     println!("{:?} -> {:?}", neigh.destination, neigh.lladdr);
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_neighbors"))]
    pub async fn get_neighbors(&self) -> Result<Vec<NeighborMessage>> {
        self.dump_typed(NlMsgType::RTM_GETNEIGH).await
    }

    /// Get neighbor entries for a specific interface.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    ///
    /// See also [`get_neighbors_by_index`](Self::get_neighbors_by_index) in the neighbor module.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_neighbors_by_name"))]
    pub async fn get_neighbors_by_name(
        &self,
        iface: impl Into<InterfaceRef>,
    ) -> Result<Vec<NeighborMessage>> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.get_neighbors_by_index(ifindex).await
    }

    /// Get all routing rules.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let rules = conn.get_rules().await?;
    /// for rule in rules {
    ///     println!("{}: {:?} -> table {}", rule.priority, rule.source, rule.table);
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_rules"))]
    pub async fn get_rules(&self) -> Result<Vec<RuleMessage>> {
        self.dump_typed(NlMsgType::RTM_GETRULE).await
    }

    /// Get routing rules for a specific address family.
    ///
    /// # Arguments
    ///
    /// * `family` - Address family: `libc::AF_INET` for IPv4, `libc::AF_INET6` for IPv6
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_rules_for_family"))]
    pub async fn get_rules_for_family(&self, family: u8) -> Result<Vec<RuleMessage>> {
        let rules = self.get_rules().await?;
        Ok(rules.into_iter().filter(|r| r.family() == family).collect())
    }

    /// Get IPv4 routing rules.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_rules_v4"))]
    pub async fn get_rules_v4(&self) -> Result<Vec<RuleMessage>> {
        self.get_rules_for_family(libc::AF_INET as u8).await
    }

    /// Get IPv6 routing rules.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_rules_v6"))]
    pub async fn get_rules_v6(&self) -> Result<Vec<RuleMessage>> {
        self.get_rules_for_family(libc::AF_INET6 as u8).await
    }

    /// Add a routing rule.
    ///
    /// Use the [`super::rule::RuleBuilder`] to construct the rule.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::netlink::rule::RuleBuilder;
    ///
    /// // Add a rule to lookup table 100 for traffic from 10.0.0.0/8
    /// conn.add_rule(
    ///     RuleBuilder::v4()
    ///         .priority(100)
    ///         .from("10.0.0.0", 8)
    ///         .table(100)
    /// ).await?;
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "add_rule"))]
    pub async fn add_rule(&self, rule: super::rule::RuleBuilder) -> Result<()> {
        let builder = rule.build()?;
        self.send_ack(builder)
            .await
            .map_err(|e| e.with_context("add_rule"))
    }

    /// Delete a routing rule.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::netlink::rule::RuleBuilder;
    ///
    /// conn.del_rule(
    ///     RuleBuilder::v4()
    ///         .priority(100)
    /// ).await?;
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "del_rule"))]
    pub async fn del_rule(&self, rule: super::rule::RuleBuilder) -> Result<()> {
        let builder = rule.build_delete()?;
        self.send_ack(builder)
            .await
            .map_err(|e| e.with_context("del_rule"))
    }

    /// Delete a rule by priority.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "del_rule_by_priority"))]
    pub async fn del_rule_by_priority(&self, family: u8, priority: u32) -> Result<()> {
        let rule = super::rule::RuleBuilder::new(family).priority(priority);
        self.del_rule(rule).await
    }

    /// Flush all non-default routing rules for a family.
    ///
    /// This deletes all rules except the default ones (priority 0, 32766, 32767).
    #[tracing::instrument(level = "debug", skip_all, fields(method = "flush_rules"))]
    pub async fn flush_rules(&self, family: u8) -> Result<()> {
        let rules = self.get_rules_for_family(family).await?;

        for rule in rules {
            // Skip default rules
            if rule.priority == 0 || rule.priority == 32766 || rule.priority == 32767 {
                continue;
            }

            // Delete by priority
            let _ = self.del_rule_by_priority(family, rule.priority).await;
        }

        Ok(())
    }

    /// Get all qdiscs.
    ///
    /// # Example
    ///
    /// ```ignore
    /// let qdiscs = conn.get_qdiscs().await?;
    /// for qdisc in qdiscs {
    ///     println!("{}: {}", qdisc.ifindex(), qdisc.kind().unwrap_or("?"));
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_qdiscs"))]
    pub async fn get_qdiscs(&self) -> Result<Vec<TcMessage>> {
        self.dump_typed(NlMsgType::RTM_GETQDISC).await
    }

    /// Get qdiscs for a specific interface.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_qdiscs_by_name"))]
    pub async fn get_qdiscs_by_name(
        &self,
        iface: impl Into<InterfaceRef>,
    ) -> Result<Vec<TcMessage>> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.get_qdiscs_by_index(ifindex).await
    }

    /// Get qdiscs for a specific interface by index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_qdiscs_by_index"))]
    pub async fn get_qdiscs_by_index(&self, ifindex: u32) -> Result<Vec<TcMessage>> {
        let qdiscs = self.get_qdiscs().await?;
        Ok(qdiscs
            .into_iter()
            .filter(|q| q.ifindex() == ifindex)
            .collect())
    }

    /// Get all TC classes.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_classes"))]
    pub async fn get_classes(&self) -> Result<Vec<TcMessage>> {
        self.dump_typed(NlMsgType::RTM_GETTCLASS).await
    }

    /// Get TC classes for a specific interface.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_classes_by_name"))]
    pub async fn get_classes_by_name(
        &self,
        iface: impl Into<InterfaceRef>,
    ) -> Result<Vec<TcMessage>> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.get_classes_by_index(ifindex).await
    }

    /// Get TC classes for a specific interface by index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_classes_by_index"))]
    pub async fn get_classes_by_index(&self, ifindex: u32) -> Result<Vec<TcMessage>> {
        let classes = self.get_classes().await?;
        Ok(classes
            .into_iter()
            .filter(|c| c.ifindex() == ifindex)
            .collect())
    }

    /// Get all TC filters.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_filters"))]
    pub async fn get_filters(&self) -> Result<Vec<TcMessage>> {
        self.dump_typed(NlMsgType::RTM_GETTFILTER).await
    }

    /// Get TC filters for a specific interface.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_filters_by_name"))]
    pub async fn get_filters_by_name(
        &self,
        iface: impl Into<InterfaceRef>,
    ) -> Result<Vec<TcMessage>> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.get_filters_by_index(ifindex).await
    }

    /// Get TC filters for a specific interface by index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_filters_by_index"))]
    pub async fn get_filters_by_index(&self, ifindex: u32) -> Result<Vec<TcMessage>> {
        let filters = self.get_filters().await?;
        Ok(filters
            .into_iter()
            .filter(|f| f.ifindex() == ifindex)
            .collect())
    }

    /// Get TC filters for a specific interface, filtered by parent handle.
    ///
    /// Equivalent to `get_filters_by_name(...).await?` followed by a
    /// client-side `.filter(|f| f.parent() == parsed_parent)`. Useful for
    /// reconcile-style consumers that need targeted teardown without
    /// scanning every filter on the interface.
    ///
    /// `parent` accepts the standard `tc` handle syntax: `"1:"`, `"1:5"`,
    /// `"root"`, `"ingress"`, `"clsact"`. Returns `Error::InvalidMessage`
    /// for unparseable handles.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_filters_by_parent"))]
    pub async fn get_filters_by_parent(
        &self,
        iface: impl Into<InterfaceRef>,
        parent: TcHandle,
    ) -> Result<Vec<TcMessage>> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.get_filters_by_parent_index(ifindex, parent).await
    }

    /// Get TC filters by interface index, filtered by parent handle.
    ///
    /// Namespace-safe variant of [`Connection::get_filters_by_parent`].
    #[tracing::instrument(
        level = "debug",
        skip_all,
        fields(method = "get_filters_by_parent_index")
    )]
    pub async fn get_filters_by_parent_index(
        &self,
        ifindex: u32,
        parent: TcHandle,
    ) -> Result<Vec<TcMessage>> {
        let filters = self.get_filters_by_index(ifindex).await?;
        Ok(filters
            .into_iter()
            .filter(|f| f.parent() == parent)
            .collect())
    }

    /// Get all TC filter chains for an interface.
    ///
    /// Filter chains provide logical grouping of filters for better
    /// performance and organization (Linux 4.1+).
    ///
    /// # Example
    ///
    /// ```ignore
    /// let chains = conn.get_tc_chains("eth0", "ingress").await?;
    /// for chain in chains {
    ///     println!("Chain: {}", chain);
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_tc_chains"))]
    pub async fn get_tc_chains(
        &self,
        ifname: impl Into<InterfaceRef>,
        parent: TcHandle,
    ) -> Result<Vec<u32>> {
        let ifindex = self.resolve_interface(&ifname.into()).await?;
        self.get_tc_chains_by_index(ifindex, parent).await
    }

    /// Get all TC filter chains for an interface by index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_tc_chains_by_index"))]
    pub async fn get_tc_chains_by_index(&self, ifindex: u32, parent: TcHandle) -> Result<Vec<u32>> {
        use super::types::tc::TcMsg;

        let tcmsg = TcMsg::new()
            .with_ifindex(ifindex as i32)
            .with_parent(parent.as_raw());

        let mut builder = dump_request(NlMsgType::RTM_GETCHAIN);
        builder.append(&tcmsg);

        let responses = self.send_dump(builder).await?;
        let mut chains = Vec::new();

        for response in responses {
            if response.len() < NLMSG_HDRLEN {
                continue;
            }
            let payload = &response[NLMSG_HDRLEN..];
            if let Ok(tc) = TcMessage::from_bytes(payload)
                && let Some(chain) = tc.chain()
            {
                chains.push(chain);
            }
        }

        Ok(chains)
    }

    /// Add a TC filter chain.
    ///
    /// Chains provide logical grouping of filters (Linux 4.1+).
    /// Chain 0 is created automatically when adding filters.
    ///
    /// # Example
    ///
    /// ```ignore
    /// // Create chain 100 on ingress qdisc
    /// conn.add_tc_chain("eth0", "ingress", 100).await?;
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "add_tc_chain"))]
    pub async fn add_tc_chain(
        &self,
        ifname: impl Into<InterfaceRef>,
        parent: TcHandle,
        chain: u32,
    ) -> Result<()> {
        let ifindex = self.resolve_interface(&ifname.into()).await?;
        self.add_tc_chain_by_index(ifindex, parent, chain).await
    }

    /// Add a TC filter chain by interface index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "add_tc_chain_by_index"))]
    pub async fn add_tc_chain_by_index(
        &self,
        ifindex: u32,
        parent: TcHandle,
        chain: u32,
    ) -> Result<()> {
        use super::types::tc::{TcMsg, TcaAttr};

        let tcmsg = TcMsg::new()
            .with_ifindex(ifindex as i32)
            .with_parent(parent.as_raw());

        let mut builder = create_request(NlMsgType::RTM_NEWCHAIN);
        builder.append(&tcmsg);
        builder.append_attr_u32(TcaAttr::Chain as u16, chain);

        self.send_ack(builder)
            .await
            .map_err(|e| e.with_context("add_tc_chain"))
    }

    /// Delete a TC filter chain.
    ///
    /// All filters in the chain must be deleted before the chain can be removed.
    ///
    /// # Example
    ///
    /// ```ignore
    /// conn.del_tc_chain("eth0", "ingress", 100).await?;
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "del_tc_chain"))]
    pub async fn del_tc_chain(
        &self,
        ifname: impl Into<InterfaceRef>,
        parent: TcHandle,
        chain: u32,
    ) -> Result<()> {
        let ifindex = self.resolve_interface(&ifname.into()).await?;
        self.del_tc_chain_by_index(ifindex, parent, chain).await
    }

    /// Delete a TC filter chain by interface index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "del_tc_chain_by_index"))]
    pub async fn del_tc_chain_by_index(
        &self,
        ifindex: u32,
        parent: TcHandle,
        chain: u32,
    ) -> Result<()> {
        use super::types::tc::{TcMsg, TcaAttr};

        let tcmsg = TcMsg::new()
            .with_ifindex(ifindex as i32)
            .with_parent(parent.as_raw());

        let mut builder = create_request(NlMsgType::RTM_DELCHAIN);
        builder.append(&tcmsg);
        builder.append_attr_u32(TcaAttr::Chain as u16, chain);

        self.send_ack(builder)
            .await
            .map_err(|e| e.with_context("del_tc_chain"))
    }

    /// Get the root qdisc for an interface (parent == ROOT).
    ///
    /// Returns `None` if no root qdisc is configured.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    ///
    /// # Example
    ///
    /// ```ignore
    /// if let Some(root) = conn.get_root_qdisc_by_name("eth0").await? {
    ///     println!("Root qdisc: {}", root.kind().unwrap_or("?"));
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_root_qdisc_by_name"))]
    pub async fn get_root_qdisc_by_name(
        &self,
        iface: impl Into<InterfaceRef>,
    ) -> Result<Option<TcMessage>> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.get_root_qdisc_by_index(ifindex).await
    }

    /// Get the root qdisc for an interface by index.
    ///
    /// Returns `None` if no root qdisc is configured.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_root_qdisc_by_index"))]
    pub async fn get_root_qdisc_by_index(&self, ifindex: u32) -> Result<Option<TcMessage>> {
        let qdiscs = self.get_qdiscs().await?;
        Ok(qdiscs
            .into_iter()
            .find(|q| q.ifindex() == ifindex && q.is_root()))
    }

    /// Get a qdisc by interface name and handle.
    ///
    /// # Example
    ///
    /// ```ignore
    /// // Get the qdisc with handle 1:0 on eth0
    /// if let Some(qdisc) = conn.get_qdisc_by_handle("eth0", "1:").await? {
    ///     println!("Found qdisc: {}", qdisc.kind().unwrap_or("?"));
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_qdisc_by_handle"))]
    pub async fn get_qdisc_by_handle(
        &self,
        ifname: &str,
        handle: TcHandle,
    ) -> Result<Option<TcMessage>> {
        let ifindex = self
            .resolve_interface(&InterfaceRef::Name(ifname.to_string()))
            .await?;
        self.get_qdisc_by_handle_index(ifindex, handle).await
    }

    /// Get a qdisc by interface index and handle.
    ///
    /// # Example
    ///
    /// ```ignore
    /// // Get the qdisc with handle 1:0 on interface index 2
    /// if let Some(qdisc) = conn.get_qdisc_by_handle_index(2, TcHandle::major_only(1)).await? {
    ///     println!("Found qdisc: {}", qdisc.kind().unwrap_or("?"));
    /// }
    /// ```
    #[tracing::instrument(
        level = "debug",
        skip_all,
        fields(method = "get_qdisc_by_handle_index")
    )]
    pub async fn get_qdisc_by_handle_index(
        &self,
        ifindex: u32,
        handle: TcHandle,
    ) -> Result<Option<TcMessage>> {
        let qdiscs = self.get_qdiscs().await?;
        Ok(qdiscs
            .into_iter()
            .find(|q| q.ifindex() == ifindex && q.handle() == handle))
    }

    /// Get netem options for an interface, if a netem qdisc is configured at root.
    ///
    /// This is a convenience method that returns `Some` only if a netem qdisc
    /// is the root qdisc and its options can be parsed.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    ///
    /// # Example
    ///
    /// ```ignore
    /// if let Some(netem) = conn.get_netem_by_name("eth0").await? {
    ///     if let Some(delay) = netem.delay() {
    ///         println!("Delay: {:?}", delay);
    ///     }
    ///     if let Some(loss) = netem.loss() {
    ///         println!("Loss: {:.2}%", loss);
    ///     }
    ///     if let Some(rate) = netem.rate_bps() {
    ///         println!("Rate limit: {} bytes/sec", rate);
    ///     }
    /// }
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_netem_by_name"))]
    pub async fn get_netem_by_name(
        &self,
        iface: impl Into<InterfaceRef>,
    ) -> Result<Option<super::tc_options::NetemOptions>> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.get_netem_by_index(ifindex).await
    }

    /// Get netem options for an interface by index.
    ///
    /// Returns `None` if no netem qdisc is configured at root.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_netem_by_index"))]
    pub async fn get_netem_by_index(
        &self,
        ifindex: u32,
    ) -> Result<Option<super::tc_options::NetemOptions>> {
        use super::tc_options::QdiscOptions;
        let root = self.get_root_qdisc_by_index(ifindex).await?;
        Ok(match root.and_then(|q| q.options()) {
            Some(QdiscOptions::Netem(opts)) => Some(opts),
            _ => None,
        })
    }
}

// ============================================================================
// Link State Management
// ============================================================================

use super::types::link::{IfInfoMsg, iff};

impl Connection<Route> {
    /// Bring a network interface up.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    ///
    /// # Example
    ///
    /// ```ignore
    /// conn.set_link_up("eth0").await?;
    /// conn.set_link_up(5u32).await?;  // by index
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "set_link_up"))]
    pub async fn set_link_up(&self, iface: impl Into<InterfaceRef>) -> Result<()> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.set_link_up_by_index(ifindex).await
    }

    /// Bring a network interface up by index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "set_link_up_by_index"))]
    pub async fn set_link_up_by_index(&self, ifindex: u32) -> Result<()> {
        self.set_link_state_by_index(ifindex, true).await
    }

    /// Bring a network interface down.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    ///
    /// # Example
    ///
    /// ```ignore
    /// conn.set_link_down("eth0").await?;
    /// conn.set_link_down(5u32).await?;  // by index
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "set_link_down"))]
    pub async fn set_link_down(&self, iface: impl Into<InterfaceRef>) -> Result<()> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.set_link_down_by_index(ifindex).await
    }

    /// Bring a network interface down by index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "set_link_down_by_index"))]
    pub async fn set_link_down_by_index(&self, ifindex: u32) -> Result<()> {
        self.set_link_state_by_index(ifindex, false).await
    }

    /// Set the state of a network interface (up or down).
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    ///
    /// # Arguments
    ///
    /// * `iface` - The interface name or index
    /// * `up` - `true` to bring the interface up, `false` to bring it down
    ///
    /// # Example
    ///
    /// ```ignore
    /// // Bring interface up
    /// conn.set_link_state("eth0", true).await?;
    ///
    /// // Bring interface down by index
    /// conn.set_link_state(5u32, false).await?;
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "set_link_state"))]
    pub async fn set_link_state(&self, iface: impl Into<InterfaceRef>, up: bool) -> Result<()> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.set_link_state_by_index(ifindex, up).await
    }

    /// Set the state of a network interface by index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "set_link_state_by_index"))]
    pub async fn set_link_state_by_index(&self, ifindex: u32, up: bool) -> Result<()> {
        let mut ifinfo = IfInfoMsg::new().with_index(ifindex as i32);

        if up {
            ifinfo.ifi_flags = iff::UP;
            ifinfo.ifi_change = iff::UP;
        } else {
            ifinfo.ifi_flags = 0;
            ifinfo.ifi_change = iff::UP;
        }

        let mut builder = ack_request(NlMsgType::RTM_SETLINK);
        builder.append(&ifinfo);

        let state = if up { "up" } else { "down" };
        self.send_ack(builder).await.map_err(|e| {
            if e.is_not_found() {
                Error::InterfaceNotFound {
                    name: format!("ifindex {ifindex}"),
                }
            } else {
                e.with_context(format!("set_link_{state}(ifindex {ifindex})"))
            }
        })
    }

    /// Set the MTU of a network interface.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    ///
    /// # Example
    ///
    /// ```ignore
    /// conn.set_link_mtu("eth0", 9000).await?;
    /// conn.set_link_mtu(5u32, 9000).await?;  // by index
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "set_link_mtu"))]
    pub async fn set_link_mtu(&self, iface: impl Into<InterfaceRef>, mtu: u32) -> Result<()> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.set_link_mtu_by_index(ifindex, mtu).await
    }

    /// Set the MTU of a network interface by index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "set_link_mtu_by_index"))]
    pub async fn set_link_mtu_by_index(&self, ifindex: u32, mtu: u32) -> Result<()> {
        use super::types::link::IflaAttr;

        let ifinfo = IfInfoMsg::new().with_index(ifindex as i32);

        let mut builder = ack_request(NlMsgType::RTM_SETLINK);
        builder.append(&ifinfo);
        builder.append_attr_u32(IflaAttr::Mtu as u16, mtu);

        self.send_ack(builder)
            .await
            .map_err(|e| e.with_context("set_link_mtu"))
    }

    /// Delete a network interface.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    ///
    /// # Example
    ///
    /// ```ignore
    /// conn.del_link("veth0").await?;
    /// conn.del_link(5u32).await?;  // by index
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "del_link"))]
    pub async fn del_link(&self, iface: impl Into<InterfaceRef>) -> Result<()> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.del_link_by_index(ifindex).await
    }

    /// Delete a network interface by index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "del_link_by_index"))]
    pub async fn del_link_by_index(&self, ifindex: u32) -> Result<()> {
        let ifinfo = IfInfoMsg::new().with_index(ifindex as i32);

        let mut builder = ack_request(NlMsgType::RTM_DELLINK);
        builder.append(&ifinfo);

        self.send_ack(builder).await.map_err(|e| {
            if e.is_not_found() {
                Error::InterfaceNotFound {
                    name: format!("ifindex {ifindex}"),
                }
            } else {
                e.with_context(format!("del_link(ifindex {ifindex})"))
            }
        })
    }

    /// Set the TX queue length of a network interface.
    ///
    /// Accepts either an interface name or index via [`InterfaceRef`].
    ///
    /// # Example
    ///
    /// ```ignore
    /// conn.set_link_txqlen("eth0", 1000).await?;
    /// conn.set_link_txqlen(5u32, 1000).await?;  // by index
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "set_link_txqlen"))]
    pub async fn set_link_txqlen(&self, iface: impl Into<InterfaceRef>, txqlen: u32) -> Result<()> {
        let ifindex = self.resolve_interface(&iface.into()).await?;
        self.set_link_txqlen_by_index(ifindex, txqlen).await
    }

    /// Set the TX queue length of a network interface by index.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "set_link_txqlen_by_index"))]
    pub async fn set_link_txqlen_by_index(&self, ifindex: u32, txqlen: u32) -> Result<()> {
        use super::types::link::IflaAttr;

        let ifinfo = IfInfoMsg::new().with_index(ifindex as i32);

        let mut builder = ack_request(NlMsgType::RTM_SETLINK);
        builder.append(&ifinfo);
        builder.append_attr_u32(IflaAttr::TxqLen as u16, txqlen);

        self.send_ack(builder)
            .await
            .map_err(|e| e.with_context("set_link_txqlen"))
    }
}

// ============================================================================
// Namespace ID Queries
// ============================================================================

use super::{
    messages::NsIdMessage,
    types::nsid::{RTM_GETNSID, RtGenMsg, netnsa},
};

impl Connection<Route> {
    /// Get the namespace ID for a given file descriptor.
    ///
    /// The file descriptor should be an open reference to a network namespace
    /// (e.g., from opening `/proc/<pid>/ns/net`).
    ///
    /// # Errors
    ///
    /// Returns an error if the namespace ID cannot be determined.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use std::fs::File;
    /// use std::os::unix::io::AsRawFd;
    ///
    /// let ns_file = File::open("/var/run/netns/myns")?;
    /// let nsid = conn.get_nsid(ns_file.as_raw_fd()).await?;
    /// println!("Namespace ID: {}", nsid);
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_nsid"))]
    pub async fn get_nsid(&self, ns_fd: RawFd) -> Result<u32> {
        let mut builder = ack_request(RTM_GETNSID);

        // Append rtgenmsg header (1 byte + 3 padding)
        builder.append(&RtGenMsg::new());
        builder.append_bytes(&[0u8; 3]); // Padding to 4 bytes

        // Add NETNSA_FD attribute
        builder.append_attr_u32(netnsa::FD, ns_fd as u32);

        let response = self.send_request(builder).await?;

        // Parse the response
        if response.len() >= super::message::NLMSG_HDRLEN {
            let payload = &response[super::message::NLMSG_HDRLEN..];
            if let Some(nsid_msg) = NsIdMessage::parse(payload)
                && let Some(nsid) = nsid_msg.nsid
            {
                return Ok(nsid);
            }
        }

        Err(Error::InvalidMessage(
            "namespace ID not found in response".into(),
        ))
    }

    /// Get the namespace ID for a given process's network namespace.
    ///
    /// # Errors
    ///
    /// Returns an error if the namespace ID cannot be determined.
    ///
    /// # Example
    ///
    /// ```ignore
    /// // Get the namespace ID for process 1234
    /// let nsid = conn.get_nsid_for_pid(1234).await?;
    /// println!("Namespace ID for PID 1234: {}", nsid);
    /// ```
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_nsid_for_pid"))]
    pub async fn get_nsid_for_pid(&self, pid: u32) -> Result<u32> {
        let mut builder = ack_request(RTM_GETNSID);

        // Append rtgenmsg header (1 byte + 3 padding)
        builder.append(&RtGenMsg::new());
        builder.append_bytes(&[0u8; 3]); // Padding to 4 bytes

        // Add NETNSA_PID attribute
        builder.append_attr_u32(netnsa::PID, pid);

        let response = self.send_request(builder).await?;

        // Parse the response
        if response.len() >= super::message::NLMSG_HDRLEN {
            let payload = &response[super::message::NLMSG_HDRLEN..];
            if let Some(nsid_msg) = NsIdMessage::parse(payload)
                && let Some(nsid) = nsid_msg.nsid
            {
                return Ok(nsid);
            }
        }

        Err(Error::InvalidMessage(
            "namespace ID not found in response".into(),
        ))
    }
}

// ============================================================================
// Generic Netlink protocol methods
// ============================================================================

use std::collections::HashMap;

use super::{
    genl::{
        CtrlAttr, CtrlAttrMcastGrp, CtrlCmd, FamilyInfo, GENL_HDRLEN, GENL_ID_CTRL, GenlMsgHdr,
    },
    protocol::Generic,
};

impl Connection<Generic> {
    /// Get information about a Generic Netlink family.
    ///
    /// The result is cached, so subsequent calls for the same family
    /// do not require kernel communication.
    ///
    /// # Example
    ///
    /// ```ignore
    /// use nlink::netlink::{Connection, Generic};
    ///
    /// let conn = Connection::<Generic>::new()?;
    /// let wg = conn.get_family("wireguard").await?;
    /// println!("WireGuard family ID: {}", wg.id);
    /// ```
    #[instrument(level = "info", skip(self), fields(family = %name, id, cached))]
    pub async fn get_family(&self, name: &str) -> Result<FamilyInfo> {
        // Check cache first
        {
            let cache = self.state.cache.read().unwrap();
            if let Some(info) = cache.get(name) {
                let span = tracing::Span::current();
                span.record("id", info.id);
                span.record("cached", true);
                return Ok(info.clone());
            }
        }

        // Query kernel for family info
        let info = self.query_family(name).await?;
        let span = tracing::Span::current();
        span.record("id", info.id);
        span.record("cached", false);

        // Cache the result
        {
            let mut cache = self.state.cache.write().unwrap();
            cache.insert(name.to_string(), info.clone());
        }

        Ok(info)
    }

    /// Get the family ID for a given family name.
    ///
    /// This is a convenience method that returns just the ID.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "get_family_id"))]
    pub async fn get_family_id(&self, name: &str) -> Result<u16> {
        Ok(self.get_family(name).await?.id)
    }

    /// Clear the family cache.
    ///
    /// This is rarely needed, but may be useful if families are
    /// dynamically loaded/unloaded.
    pub fn clear_cache(&self) {
        let mut cache = self.state.cache.write().unwrap();
        cache.clear();
    }

    /// Query the kernel for family information.
    async fn query_family(&self, name: &str) -> Result<FamilyInfo> {
        // Build CTRL_CMD_GETFAMILY request
        let mut builder = MessageBuilder::new(GENL_ID_CTRL, NLM_F_REQUEST | NLM_F_ACK);

        // Append GENL header
        let genl_hdr = GenlMsgHdr::new(CtrlCmd::GetFamily as u8, 1);
        builder.append(&genl_hdr);

        // Append family name attribute
        builder.append_attr_str(CtrlAttr::FamilyName as u16, name);

        // Send request
        let seq = self.socket.next_seq();
        builder.set_seq(seq);
        builder.set_pid(self.socket.pid());

        let msg = builder.finish();
        self.socket.send(&msg).await?;

        // Receive response
        let response = self.socket.recv_msg().await?;

        // Parse response
        self.parse_family_response(&response, seq, name)
    }

    /// Parse a CTRL_CMD_GETFAMILY response.
    fn parse_family_response(&self, data: &[u8], seq: u32, name: &str) -> Result<FamilyInfo> {
        for result in MessageIter::new(data) {
            let (header, payload) = result?;

            // Check sequence number
            if header.nlmsg_seq != seq {
                continue;
            }

            // Check for error
            if header.is_error() {
                let err = NlMsgError::from_bytes(payload)?;
                if !err.is_ack() {
                    // ENOENT means family not found
                    if err.error == -libc::ENOENT {
                        return Err(Error::FamilyNotFound {
                            name: name.to_string(),
                        });
                    }
                    return Err(Error::from_errno(err.error));
                }
                continue;
            }

            // Skip DONE message
            if header.is_done() {
                continue;
            }

            // Parse GENL header
            if payload.len() < GENL_HDRLEN {
                return Err(Error::InvalidMessage("GENL header too short".into()));
            }

            // Parse attributes after GENL header
            let attrs_data = &payload[GENL_HDRLEN..];
            return self.parse_family_attrs(attrs_data);
        }

        Err(Error::FamilyNotFound {
            name: name.to_string(),
        })
    }

    /// Parse family attributes from a CTRL_CMD_GETFAMILY response.
    fn parse_family_attrs(&self, data: &[u8]) -> Result<FamilyInfo> {
        use super::attr::{AttrIter, get};

        let mut id: Option<u16> = None;
        let mut version: u8 = 0;
        let mut hdr_size: u32 = 0;
        let mut max_attr: u32 = 0;
        let mut mcast_groups = HashMap::new();

        for (attr_type, payload) in AttrIter::new(data) {
            match attr_type {
                t if t == CtrlAttr::FamilyId as u16 => {
                    id = Some(get::u16_ne(payload)?);
                }
                t if t == CtrlAttr::Version as u16 => {
                    version = get::u32_ne(payload)? as u8;
                }
                t if t == CtrlAttr::HdrSize as u16 => {
                    hdr_size = get::u32_ne(payload)?;
                }
                t if t == CtrlAttr::MaxAttr as u16 => {
                    max_attr = get::u32_ne(payload)?;
                }
                t if t == CtrlAttr::McastGroups as u16 => {
                    mcast_groups = self.parse_mcast_groups(payload)?;
                }
                _ => {}
            }
        }

        let id = id.ok_or_else(|| Error::InvalidMessage("missing family ID".into()))?;

        Ok(FamilyInfo {
            id,
            version,
            hdr_size,
            max_attr,
            mcast_groups,
        })
    }

    /// Parse multicast groups from CTRL_ATTR_MCAST_GROUPS.
    fn parse_mcast_groups(&self, data: &[u8]) -> Result<HashMap<String, u32>> {
        use super::attr::{AttrIter, get};

        let mut groups = HashMap::new();

        // The mcast_groups attribute contains nested arrays
        for (_group_idx, group_payload) in AttrIter::new(data) {
            let mut name: Option<String> = None;
            let mut grp_id: Option<u32> = None;

            // Parse the nested group attributes
            for (attr_type, payload) in AttrIter::new(group_payload) {
                match attr_type {
                    t if t == CtrlAttrMcastGrp::Name as u16 => {
                        name = Some(get::string(payload)?.to_string());
                    }
                    t if t == CtrlAttrMcastGrp::Id as u16 => {
                        grp_id = Some(get::u32_ne(payload)?);
                    }
                    _ => {}
                }
            }

            if let (Some(name), Some(id)) = (name, grp_id) {
                groups.insert(name, id);
            }
        }

        Ok(groups)
    }

    /// Send a GENL command and wait for a response.
    ///
    /// This is a low-level method for sending arbitrary GENL commands.
    /// Family-specific wrappers (like `Connection<Wireguard>`) should use this.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "command"))]
    pub async fn command(
        &self,
        family_id: u16,
        cmd: u8,
        version: u8,
        build_attrs: impl FnOnce(&mut MessageBuilder),
    ) -> Result<Vec<u8>> {
        let mut builder = MessageBuilder::new(family_id, NLM_F_REQUEST | NLM_F_ACK);

        // Append GENL header
        let genl_hdr = GenlMsgHdr::new(cmd, version);
        builder.append(&genl_hdr);

        // Let caller append attributes
        build_attrs(&mut builder);

        // Send request
        let seq = self.socket.next_seq();
        builder.set_seq(seq);
        builder.set_pid(self.socket.pid());

        let msg = builder.finish();
        self.socket.send(&msg).await?;

        // Receive response
        let response = self.socket.recv_msg().await?;
        self.process_genl_response(&response, seq)?;

        Ok(response)
    }

    /// Send a GENL dump command and collect all responses.
    #[tracing::instrument(level = "debug", skip_all, fields(method = "dump_command"))]
    pub async fn dump_command(
        &self,
        family_id: u16,
        cmd: u8,
        version: u8,
        build_attrs: impl FnOnce(&mut MessageBuilder),
    ) -> Result<Vec<Vec<u8>>> {
        let mut builder = MessageBuilder::new(family_id, NLM_F_REQUEST | NLM_F_DUMP);

        // Append GENL header
        let genl_hdr = GenlMsgHdr::new(cmd, version);
        builder.append(&genl_hdr);

        // Let caller append attributes
        build_attrs(&mut builder);

        // Send request
        let seq = self.socket.next_seq();
        builder.set_seq(seq);
        builder.set_pid(self.socket.pid());

        let msg = builder.finish();
        self.socket.send(&msg).await?;

        let mut responses = Vec::new();

        loop {
            let data = self.socket.recv_msg().await?;
            let mut done = false;

            for result in MessageIter::new(&data) {
                let (header, payload) = result?;

                if header.nlmsg_seq != seq {
                    continue;
                }

                if header.is_error() {
                    let err = NlMsgError::from_bytes(payload)?;
                    if !err.is_ack() {
                        return Err(Error::from_errno(err.error));
                    }
                    continue;
                }

                if header.is_done() {
                    done = true;
                    break;
                }

                // Include the payload (with GENL header)
                responses.push(payload.to_vec());
            }

            if done {
                break;
            }
        }

        Ok(responses)
    }

    /// Process a GENL response, checking for errors.
    fn process_genl_response(&self, data: &[u8], seq: u32) -> Result<()> {
        for result in MessageIter::new(data) {
            let (header, payload) = result?;

            if header.nlmsg_seq != seq {
                continue;
            }

            if header.is_error() {
                let err = NlMsgError::from_bytes(payload)?;
                if !err.is_ack() {
                    return Err(Error::from_errno(err.error));
                }
            }
        }
        Ok(())
    }
}

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

    fn assert_send<T: Send>() {}
    fn assert_sync<T: Sync>() {}

    #[test]
    fn connection_is_send_sync() {
        assert_send::<Connection<Route>>();
        assert_sync::<Connection<Route>>();
    }
}