zlayer-overlay 0.14.2

//! Edge client supervisor: the integration point that wires the whole edge
//! data plane together.
//!
//! [`EdgeClient::start`] takes a minted [`EdgeConfig`] plus operator
//! [`EdgeOptions`] and stands up one of two transports:
//!
//! - **netstack mode** (unprivileged): a stable smoltcp [`Netstack`] +
//!   [`NetstackChannel`] is built once, the `--forward` listeners are started
//!   once against a [`NetstackDialer`], and a reconnect [`NetstackSupervisor`]
//!   loop rebuilds *only* the `WireGuard` tunnel (a fresh UDP socket + `Tunn`
//!   peer + [`TunnDriver`]) with exponential jittered backoff. The netstack and
//!   its forward listeners survive tunnel rebuilds untouched.
//! - **device mode** (privileged): a real `zl-edge0` TUN interface is stood up
//!   via [`OverlayTransport`] and the OS routes overlay traffic; the
//!   `--forward` mappings degrade to plain TCP proxies ([`DeviceDialer`]) that
//!   reuse the same [`serve_forward`] accept/pump loop.
//!
//! [`EdgeMode::Auto`] probes the host ([`probe_device_mode`]) and prefers a
//! real device when it looks grantable, transparently falling back to the
//! userspace netstack on any device-setup error.
//!
//! The names below (`EdgeClient`, `EdgeHandle`, `EdgeOptions`, `EdgeStatus`)
//! intentionally repeat the `edge`/`client` module context — they are the
//! stable public surface Wave-3 wires the CLI + API route onto.
#![allow(clippy::module_name_repetitions)]

use std::ffi::OsString;
use std::net::{IpAddr, Ipv4Addr, SocketAddr};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

use async_trait::async_trait;
use dashmap::DashMap;
use parking_lot::RwLock;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{lookup_host, TcpStream, UdpSocket};
use tokio::sync::{watch, Mutex as AsyncMutex, Notify};
use tokio::task::JoinHandle;

use zlayer_types::overlayd::EdgeConfig;

use crate::config::{OverlayConfig, PeerInfo};
use crate::edge::artifact::parse_allowed_ips;
use crate::edge::forward::{serve_forward, ForwardSpec, TunnelDialer};
use crate::edge::netstack::{Netstack, NetstackChannel, NetstackConfig};
use crate::edge::probe::{probe_device_mode, DeviceVerdict, EdgeMode};
use crate::edge::{
    EdgeError, TcpConn, EDGE_BACKOFF_MAX, EDGE_HANDSHAKE_GRACE, EDGE_HANDSHAKE_STALE,
    EDGE_TCP_BUFFER_BYTES, EDGE_TCP_TIMEOUT,
};
use crate::transport::OverlayTransport;
use crate::tunn_loop::{build_tunn, decode_key_b64, PeerState, TunnDriver};

/// Advisory MTU for the edge overlay tunnel. [`EdgeConfig`] carries no MTU
/// field, so the standard `WireGuard` tunnel MTU (1500-byte underlay minus 80
/// bytes of encapsulation) is used for both the netstack and the device
/// interface, matching [`crate::config::OverlayConfig`]'s own default.
const EDGE_DEFAULT_MTU: u32 = 1420;

/// Interface name for the device-mode TUN adapter (≤15 chars for Linux
/// `IFNAMSIZ`; ignored on macOS where the kernel assigns `utunN`).
const DEVICE_INTERFACE_NAME: &str = "zl-edge0";

/// How often the reconnect supervisor samples the peer's last-handshake clock.
const WATCHDOG_INTERVAL: Duration = Duration::from_secs(10);

/// Default persistent-keepalive interval (seconds) when the minted peer spec
/// carries none.
const DEFAULT_KEEPALIVE_SECS: u16 = 25;

/// Read/copy chunk size for the device-mode plain-TCP proxy adapter.
const DEVICE_ADAPT_CHUNK: usize = 64 * 1024;

/// Per-direction channel depth for the device-mode plain-TCP proxy adapter.
const DEVICE_ADAPT_CHANNEL_BOUND: usize = 128;

// ---------------------------------------------------------------------------
// Public surface
// ---------------------------------------------------------------------------

/// Operator-supplied knobs layered on top of the minted [`EdgeConfig`].
#[derive(Debug, Clone)]
pub struct EdgeOptions {
    /// `--forward LOCAL=OVERLAY` mappings to expose once the tunnel is up.
    pub forwards: Vec<ForwardSpec>,
    /// Which transport to use ([`EdgeMode::Auto`] probes and chooses).
    pub mode: EdgeMode,
    /// Override for the on-disk state directory (device mode's UAPI socket
    /// dir). Defaults per [`resolve_state_dir`]; netstack mode writes nothing.
    pub state_dir: Option<PathBuf>,
}

impl Default for EdgeOptions {
    /// No forwards, [`EdgeMode::Auto`], default state dir. (`EdgeMode` has no
    /// `Default` impl, so this cannot be derived.)
    fn default() -> Self {
        Self {
            forwards: Vec::new(),
            mode: EdgeMode::Auto,
            state_dir: None,
        }
    }
}

/// Coarse connection state published by a running edge client.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum EdgeStatus {
    /// The first tunnel is being established (no handshake yet).
    Connecting,
    /// A `WireGuard` handshake has completed; the tunnel is live.
    Connected {
        /// When the current live session was observed to come up.
        since: SystemTime,
    },
    /// The session dropped and the supervisor is rebuilding the tunnel.
    Reconnecting {
        /// 1-based reconnect attempt counter.
        attempt: u32,
    },
    /// The client is shutting down or has torn its tunnel down.
    Down,
}

/// The zero-sized entry point to the edge data plane.
pub struct EdgeClient;

impl EdgeClient {
    /// Resolve the transport mode and stand up the edge client, returning a
    /// live [`EdgeHandle`].
    ///
    /// - [`EdgeMode::Device`] forces device mode (no fallback).
    /// - [`EdgeMode::Netstack`] forces the userspace netstack.
    /// - [`EdgeMode::Auto`] probes the host: a grantable verdict tries device
    ///   mode and falls back to netstack on any setup error; a denied verdict
    ///   goes straight to netstack.
    ///
    /// # Errors
    ///
    /// Returns [`EdgeError`] if key material fails to decode, the peer spec is
    /// missing/malformed, a socket cannot be bound, or (device mode) the TUN
    /// interface cannot be created or configured.
    pub async fn start(cfg: EdgeConfig, opts: EdgeOptions) -> Result<EdgeHandle, EdgeError> {
        let plan = match opts.mode {
            EdgeMode::Netstack => ModePlan::Netstack,
            EdgeMode::Device => ModePlan::Device {
                allow_fallback: false,
            },
            EdgeMode::Auto => {
                let verdict = probe_device_mode();
                if let DeviceVerdict::Denied(reason) = &verdict {
                    tracing::debug!(%reason, "edge auto: device mode unavailable, using netstack");
                }
                plan_mode(EdgeMode::Auto, &verdict)
            }
        };

        match plan {
            ModePlan::Netstack => netstack_mode(cfg, opts),
            ModePlan::Device {
                allow_fallback: false,
            } => device_mode(cfg, opts).await,
            ModePlan::Device {
                allow_fallback: true,
            } => match device_mode(cfg.clone(), opts.clone()).await {
                Ok(handle) => Ok(handle),
                Err(e) => {
                    tracing::warn!(error = %e, "edge auto: device mode failed, falling back to netstack");
                    netstack_mode(cfg, opts)
                }
            },
        }
    }
}

/// Live handle to a running edge client: status stream, netstack (netstack
/// mode only), and a clean shutdown path.
///
/// `netstack` is [`Some`] in netstack mode (shared as an [`Arc`] so both the
/// forward dialers and Wave-3 callers can `connect`/`listen` through the one
/// poll task) and [`None`] in device mode, where the OS routing table carries
/// overlay traffic instead.
pub struct EdgeHandle {
    /// The netstack control handle in netstack mode; [`None`] in device mode.
    pub netstack: Option<Arc<Netstack>>,
    status: watch::Receiver<EdgeStatus>,
    cancel: Cancel,
    join: JoinHandle<()>,
}

impl EdgeHandle {
    /// A fresh receiver on the client's [`EdgeStatus`] stream.
    #[must_use]
    pub fn status(&self) -> watch::Receiver<EdgeStatus> {
        self.status.clone()
    }

    /// Signal the supervisor to tear the tunnel down and await it.
    ///
    /// Drops all driver/forward tasks, releases the UDP socket (or destroys the
    /// TUN interface, device mode), and publishes [`EdgeStatus::Down`].
    pub async fn shutdown(self) {
        self.cancel.cancel();
        // `EdgeHandle` has no `Drop`, so moving `join` out is a plain partial
        // move; the remaining fields drop when this scope ends.
        let _ = self.join.await;
    }
}

// ---------------------------------------------------------------------------
// Mode resolution (pure, independently testable)
// ---------------------------------------------------------------------------

/// The concrete transport decision produced from an [`EdgeMode`] + probe
/// verdict.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum ModePlan {
    Netstack,
    Device { allow_fallback: bool },
}

/// Map an [`EdgeMode`] and a device probe verdict to a concrete [`ModePlan`].
///
/// Forced modes ignore the verdict; `Auto` uses it, and only `Auto` permits a
/// device→netstack fallback.
fn plan_mode(mode: EdgeMode, verdict: &DeviceVerdict) -> ModePlan {
    match mode {
        EdgeMode::Netstack => ModePlan::Netstack,
        EdgeMode::Device => ModePlan::Device {
            allow_fallback: false,
        },
        EdgeMode::Auto => match verdict {
            DeviceVerdict::Grantable => ModePlan::Device {
                allow_fallback: true,
            },
            DeviceVerdict::Denied(_) => ModePlan::Netstack,
        },
    }
}

// ---------------------------------------------------------------------------
// Cancellation (tokio-util is not a dependency of this crate)
// ---------------------------------------------------------------------------

/// A minimal broadcast cancellation token built on [`Notify`] + an atomic
/// flag.
///
/// `tokio-util`'s `CancellationToken` is not a dependency of `zlayer-overlay`,
/// so the supervisor uses this instead: [`Cancel::cancel`] is idempotent and
/// wakes every current waiter, and [`Cancel::cancelled`] is safe to call from
/// inside `select!` (a late waiter that registers after `cancel` simply
/// observes the flag and returns immediately).
#[derive(Clone)]
struct Cancel {
    flag: Arc<AtomicBool>,
    notify: Arc<Notify>,
}

impl Cancel {
    fn new() -> Self {
        Self {
            flag: Arc::new(AtomicBool::new(false)),
            notify: Arc::new(Notify::new()),
        }
    }

    /// Trip the token and wake all current waiters. Idempotent.
    fn cancel(&self) {
        self.flag.store(true, Ordering::SeqCst);
        self.notify.notify_waiters();
    }

    fn is_cancelled(&self) -> bool {
        self.flag.load(Ordering::SeqCst)
    }

    /// Resolve once the token is cancelled (immediately if already tripped).
    async fn cancelled(&self) {
        if self.is_cancelled() {
            return;
        }
        // Register the waiter *before* the second flag check so a concurrent
        // `cancel` cannot slip its `notify_waiters` in between and be missed.
        let notified = self.notify.notified();
        tokio::pin!(notified);
        notified.as_mut().enable();
        if self.is_cancelled() {
            return;
        }
        notified.await;
    }
}

// ---------------------------------------------------------------------------
// Backoff (pure, independently testable)
// ---------------------------------------------------------------------------

/// The un-jittered backoff for a reconnect `attempt` (1-based): 1s doubling to
/// [`EDGE_BACKOFF_MAX`]. Saturates without overflow for large `attempt`.
fn backoff_base(attempt: u32) -> Duration {
    let shift = attempt.saturating_sub(1).min(32);
    let secs = 1u64 << shift;
    Duration::from_secs(secs).min(EDGE_BACKOFF_MAX)
}

/// Apply ±20% uniform jitter to a backoff duration.
fn apply_jitter(base: Duration) -> Duration {
    // `rand::random::<f64>()` yields `[0, 1)`, so the factor is `[0.8, 1.2)`.
    let factor = 0.8 + rand::random::<f64>() * 0.4;
    base.mul_f64(factor)
}

/// The jittered backoff for a reconnect `attempt`: [`backoff_base`] with ±20%
/// jitter, capped (before jitter) at [`EDGE_BACKOFF_MAX`].
fn next_backoff(attempt: u32) -> Duration {
    apply_jitter(backoff_base(attempt))
}

// ---------------------------------------------------------------------------
// Shared helpers
// ---------------------------------------------------------------------------

/// Current unix time in whole seconds (monotonic-ish; parity with the tunnel
/// loops' `last_handshake_sec` clock).
fn now_unix() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap_or_default()
        .as_secs()
}

/// Clamp a minted peer's `persistent_keepalive_secs` (`u64` on the wire) into
/// the `u16` `WireGuard` keepalive field, defaulting a zero/absent value to
/// [`DEFAULT_KEEPALIVE_SECS`].
fn keepalive_secs(secs: u64) -> u16 {
    let clamped = u16::try_from(secs).unwrap_or(u16::MAX);
    if clamped == 0 {
        DEFAULT_KEEPALIVE_SECS
    } else {
        clamped
    }
}

/// Resolve a `host:port` endpoint to a concrete [`SocketAddr`], surviving DNS
/// hostnames and NAT-driven IP changes (called each reconnect in netstack
/// mode, once at setup in device mode).
async fn resolve_endpoint(host_port: &str) -> Result<SocketAddr, EdgeError> {
    lookup_host(host_port)
        .await
        .map_err(|e| EdgeError::Connect(format!("resolve endpoint {host_port:?}: {e}")))?
        .next()
        .ok_or_else(|| EdgeError::Connect(format!("endpoint {host_port:?} resolved to no address")))
}

/// The edge client's on-disk state directory.
///
/// An explicit override wins; otherwise [`default_state_dir`] resolves
/// `$XDG_STATE_HOME/zlayer/edge`, else `~/.local/state/zlayer/edge`. Only
/// device mode materializes anything beneath it (its UAPI socket dir); netstack
/// mode never touches disk. Never `/var/lib`.
fn resolve_state_dir(explicit: Option<&Path>) -> PathBuf {
    explicit.map_or_else(
        || default_state_dir(std::env::var_os("XDG_STATE_HOME"), std::env::var_os("HOME")),
        Path::to_path_buf,
    )
}

/// Pure `$XDG_STATE_HOME` / `$HOME` → state-dir resolution (env values passed
/// in so the branch logic is testable without touching process env).
fn default_state_dir(xdg_state_home: Option<OsString>, home: Option<OsString>) -> PathBuf {
    if let Some(xdg) = xdg_state_home.filter(|s| !s.is_empty()) {
        return PathBuf::from(xdg).join("zlayer").join("edge");
    }
    let base = home
        .filter(|s| !s.is_empty())
        .map_or_else(std::env::temp_dir, |h| {
            PathBuf::from(h).join(".local").join("state")
        });
    base.join("zlayer").join("edge")
}

// ---------------------------------------------------------------------------
// Netstack mode
// ---------------------------------------------------------------------------

/// A [`TunnelDialer`] backed by the userspace [`Netstack`] connect path.
struct NetstackDialer {
    netstack: Arc<Netstack>,
}

#[async_trait]
impl TunnelDialer for NetstackDialer {
    async fn dial(&self, dst: SocketAddr) -> Result<TcpConn, EdgeError> {
        self.netstack.connect(dst).await
    }
}

/// The minting node peer, resolved from `cfg.peers[0]` into the raw material
/// the reconnect loop needs each iteration.
struct NodePeer {
    public_key: [u8; 32],
    /// Kept as a `host:port` string and re-resolved every reconnect.
    endpoint: String,
    allowed_ips: Arc<Vec<ipnet::IpNet>>,
    keepalive: u16,
}

/// Owns the netstack reconnect loop and every task torn down at shutdown.
struct NetstackSupervisor {
    edge_priv: [u8; 32],
    node: NodePeer,
    /// Stable across reconnects — cloned into each fresh [`TunnDriver`].
    channel: Arc<NetstackChannel>,
    status_tx: watch::Sender<EdgeStatus>,
    cancel: Cancel,
    /// The netstack poll task (never restarted; aborted at teardown).
    poll_task: JoinHandle<()>,
    /// The forward listeners (started once; aborted at teardown).
    forward_tasks: Vec<JoinHandle<()>>,
}

/// Outcome of watching one tunnel session.
enum SessionOutcome {
    /// Shutdown was requested.
    Cancelled,
    /// The session failed/stalled; rebuild the tunnel.
    Reconnect,
}

impl NetstackSupervisor {
    /// The reconnect loop: rebuild only the tunnel (UDP + `Tunn` + driver),
    /// watch it, back off, repeat — until cancelled.
    async fn run(self) {
        let mut attempt: u32 = 0;
        loop {
            if attempt > 0 {
                let _ = self.status_tx.send(EdgeStatus::Reconnecting { attempt });
            }

            let endpoint = match self.resolve().await {
                Ok(addr) => addr,
                Err(e) => {
                    tracing::warn!(error = %e, endpoint = %self.node.endpoint, "edge: endpoint resolution failed");
                    attempt += 1;
                    if self.sleep_backoff(attempt).await {
                        break;
                    }
                    continue;
                }
            };

            let udp = match self.bind_udp().await {
                Ok(sock) => sock,
                Err(e) => {
                    tracing::warn!(error = %e, "edge: udp bind failed");
                    attempt += 1;
                    if self.sleep_backoff(attempt).await {
                        break;
                    }
                    continue;
                }
            };

            let (peers, last_hs) = self.build_peers(endpoint);
            let driver = TunnDriver::spawn(Arc::clone(&udp), Arc::clone(&self.channel), peers);

            match self.watch_session(&last_hs).await {
                SessionOutcome::Cancelled => {
                    driver.abort_all();
                    break;
                }
                SessionOutcome::Reconnect => {
                    driver.abort_all();
                    drop(udp);
                    attempt += 1;
                    if self.sleep_backoff(attempt).await {
                        break;
                    }
                }
            }
        }
        self.teardown();
    }

    async fn resolve(&self) -> Result<SocketAddr, EdgeError> {
        resolve_endpoint(&self.node.endpoint).await
    }

    async fn bind_udp(&self) -> Result<Arc<UdpSocket>, EdgeError> {
        let bind: SocketAddr = (Ipv4Addr::UNSPECIFIED, 0).into();
        let sock = UdpSocket::bind(bind)
            .await
            .map_err(|e| EdgeError::Connect(format!("edge udp bind failed: {e}")))?;
        Ok(Arc::new(sock))
    }

    /// Build a one-entry peer map for the node peer + expose its shared
    /// last-handshake clock for the watchdog.
    fn build_peers(
        &self,
        endpoint: SocketAddr,
    ) -> (Arc<DashMap<[u8; 32], PeerState>>, Arc<AtomicU64>) {
        let tunn = build_tunn(
            &self.edge_priv,
            &self.node.public_key,
            None,
            Some(self.node.keepalive),
        );
        let last_hs = Arc::new(AtomicU64::new(0));
        let state = PeerState {
            tunn: Arc::new(AsyncMutex::new(tunn)),
            endpoint: Arc::new(RwLock::new(Some(endpoint))),
            last_handshake_sec: Arc::clone(&last_hs),
            allowed_ips: Arc::clone(&self.node.allowed_ips),
            persistent_keepalive: Some(self.node.keepalive),
        };
        let peers = DashMap::new();
        peers.insert(self.node.public_key, state);
        (Arc::new(peers), last_hs)
    }

    /// Sample the peer's last-handshake clock every [`WATCHDOG_INTERVAL`] and
    /// drive the [`EdgeStatus`] transitions:
    /// `Connecting`→`Connected` once a handshake lands within
    /// [`EDGE_HANDSHAKE_GRACE`], `Connected`→reconnect once it goes older than
    /// [`EDGE_HANDSHAKE_STALE`], and reconnect if no handshake lands within the
    /// grace period.
    async fn watch_session(&self, last_hs: &Arc<AtomicU64>) -> SessionOutcome {
        let session_start = Instant::now();
        let mut established = false;
        let mut ticker = tokio::time::interval(WATCHDOG_INTERVAL);
        ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        loop {
            tokio::select! {
                () = self.cancel.cancelled() => return SessionOutcome::Cancelled,
                _ = ticker.tick() => {
                    let last = last_hs.load(Ordering::Relaxed);
                    if last == 0 {
                        if !established && session_start.elapsed() > EDGE_HANDSHAKE_GRACE {
                            tracing::warn!("edge: no handshake within grace period; reconnecting");
                            let _ = self.status_tx.send(EdgeStatus::Down);
                            return SessionOutcome::Reconnect;
                        }
                    } else if established {
                        if now_unix().saturating_sub(last) > EDGE_HANDSHAKE_STALE.as_secs() {
                            tracing::warn!(last_handshake = last, "edge: handshake went stale; reconnecting");
                            let _ = self.status_tx.send(EdgeStatus::Down);
                            return SessionOutcome::Reconnect;
                        }
                    } else {
                        established = true;
                        tracing::info!("edge: overlay handshake established");
                        let _ = self.status_tx.send(EdgeStatus::Connected { since: SystemTime::now() });
                    }
                }
            }
        }
    }

    /// Sleep the jittered backoff, returning `true` if cancelled during it.
    async fn sleep_backoff(&self, attempt: u32) -> bool {
        let delay = next_backoff(attempt);
        tracing::debug!(?delay, attempt, "edge: backing off before reconnect");
        tokio::select! {
            () = self.cancel.cancelled() => true,
            () = tokio::time::sleep(delay) => false,
        }
    }

    /// Abort the persistent tasks and publish [`EdgeStatus::Down`].
    fn teardown(&self) {
        self.poll_task.abort();
        for task in &self.forward_tasks {
            task.abort();
        }
        let _ = self.status_tx.send(EdgeStatus::Down);
    }
}

/// Stand up netstack mode: build the netstack + forward listeners once, then
/// spawn the reconnect supervisor.
///
/// Synchronous (it only *spawns* tasks — the netstack poll task, the forward
/// listeners, and the reconnect supervisor — and never awaits), so it must be
/// called from within a Tokio runtime.
fn netstack_mode(cfg: EdgeConfig, opts: EdgeOptions) -> Result<EdgeHandle, EdgeError> {
    let edge_priv = decode_key_b64(&cfg.private_key)
        .map_err(|e| EdgeError::Connect(format!("edge private key decode failed: {e}")))?;
    let local_ip = cfg.overlay_ip;
    let prefix_len = cfg.prefix_len;
    let node = node_peer(cfg)?;

    let ns_cfg = NetstackConfig {
        local_ip,
        prefix_len,
        mtu: EDGE_DEFAULT_MTU,
        tcp_buffer_bytes: EDGE_TCP_BUFFER_BYTES,
        tcp_timeout: EDGE_TCP_TIMEOUT,
    };
    let (netstack, channel, poll_task) = Netstack::spawn(ns_cfg);
    let netstack = Arc::new(netstack);
    let channel = Arc::new(channel);

    let mut forward_tasks = Vec::with_capacity(opts.forwards.len());
    for spec in opts.forwards {
        let dialer = NetstackDialer {
            netstack: Arc::clone(&netstack),
        };
        forward_tasks.push(tokio::spawn(async move {
            if let Err(e) = serve_forward(spec, dialer).await {
                tracing::warn!(local = %spec.local, error = %e, "edge forward listener exited");
            }
        }));
    }

    let (status_tx, status_rx) = watch::channel(EdgeStatus::Connecting);
    let cancel = Cancel::new();
    let supervisor = NetstackSupervisor {
        edge_priv,
        node,
        channel,
        status_tx,
        cancel: cancel.clone(),
        poll_task,
        forward_tasks,
    };
    let join = tokio::spawn(supervisor.run());

    Ok(EdgeHandle {
        netstack: Some(netstack),
        status: status_rx,
        cancel,
        join,
    })
}

/// Resolve `cfg.peers[0]` into a [`NodePeer`] (edge configs carry exactly one
/// peer — the minting node — which [`crate::edge::artifact`] already validated),
/// consuming `cfg` so the peer's owned fields move rather than clone.
fn node_peer(mut cfg: EdgeConfig) -> Result<NodePeer, EdgeError> {
    if cfg.peers.is_empty() {
        return Err(EdgeError::Connect(
            "edge config carries no peer".to_string(),
        ));
    }
    let peer = cfg.peers.swap_remove(0);
    let public_key = decode_key_b64(&peer.public_key)
        .map_err(|e| EdgeError::Connect(format!("node public key decode failed: {e}")))?;
    let allowed_ips = Arc::new(parse_allowed_ips(&peer.allowed_ips)?);
    Ok(NodePeer {
        public_key,
        endpoint: peer.endpoint,
        allowed_ips,
        keepalive: keepalive_secs(peer.persistent_keepalive_secs),
    })
}

// ---------------------------------------------------------------------------
// Device mode
// ---------------------------------------------------------------------------

/// A [`TunnelDialer`] that "degrades to a plain proxy": it dials a real
/// [`TcpStream`] to the overlay address (the OS routes it through the TUN
/// device) and adapts it to a [`TcpConn`] byte pipe.
struct DeviceDialer;

#[async_trait]
impl TunnelDialer for DeviceDialer {
    async fn dial(&self, dst: SocketAddr) -> Result<TcpConn, EdgeError> {
        let stream = TcpStream::connect(dst)
            .await
            .map_err(|e| EdgeError::Forward(format!("device-mode dial {dst} failed: {e}")))?;
        Ok(tcp_stream_to_conn(stream))
    }
}

/// Adapt a live [`TcpStream`] into the [`TcpConn`] byte-pipe the shared pump
/// expects: one task copies host→overlay bytes (`conn.tx`) into the stream, a
/// second copies stream reads back onto `conn.rx`.
fn tcp_stream_to_conn(stream: TcpStream) -> TcpConn {
    let (host_tx, mut host_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(DEVICE_ADAPT_CHANNEL_BOUND);
    let (back_tx, back_rx) = tokio::sync::mpsc::channel::<Vec<u8>>(DEVICE_ADAPT_CHANNEL_BOUND);
    let (mut read_half, mut write_half) = stream.into_split();

    // host → overlay: pump-written chunks into the socket; EOF/drop half-closes.
    tokio::spawn(async move {
        while let Some(chunk) = host_rx.recv().await {
            if write_half.write_all(&chunk).await.is_err() {
                break;
            }
        }
        let _ = write_half.shutdown().await;
    });

    // overlay → host: socket reads onto the return channel; close drops `back_tx`.
    tokio::spawn(async move {
        let mut buf = vec![0u8; DEVICE_ADAPT_CHUNK];
        loop {
            match read_half.read(&mut buf).await {
                Ok(0) | Err(_) => break,
                Ok(n) => {
                    if back_tx.send(buf[..n].to_vec()).await.is_err() {
                        break;
                    }
                }
            }
        }
    });

    TcpConn {
        tx: host_tx,
        rx: back_rx,
    }
}

/// Stand up device mode: create + configure a real `zl-edge0` TUN interface,
/// start the plain-TCP forward proxies, and hold the interface alive until
/// shutdown.
async fn device_mode(mut cfg: EdgeConfig, opts: EdgeOptions) -> Result<EdgeHandle, EdgeError> {
    if cfg.peers.is_empty() {
        return Err(EdgeError::Connect(
            "edge config carries no peer".to_string(),
        ));
    }
    let peer = cfg.peers.swap_remove(0);
    let endpoint = resolve_endpoint(&peer.endpoint).await?;
    let uapi_sock_dir = resolve_state_dir(opts.state_dir.as_deref()).join("wireguard");

    let overlay_cidr = format!("{}/{}", cfg.overlay_ip, cfg.prefix_len);
    let keepalive = keepalive_secs(peer.persistent_keepalive_secs);
    let overlay_cfg = OverlayConfig {
        local_endpoint: SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 0),
        private_key: cfg.private_key,
        public_key: cfg.public_key,
        overlay_cidr,
        cluster_cidr: Some(peer.allowed_ips.clone()),
        uapi_sock_dir,
        mtu: EDGE_DEFAULT_MTU,
        ..OverlayConfig::default()
    };

    let mut transport = OverlayTransport::new(overlay_cfg, DEVICE_INTERFACE_NAME.to_string());
    transport
        .create_interface()
        .await
        .map_err(|e| EdgeError::Connect(format!("edge device create_interface failed: {e}")))?;

    let peer_info = PeerInfo::new(
        peer.public_key,
        endpoint,
        &peer.allowed_ips,
        Duration::from_secs(u64::from(keepalive)),
    );
    transport
        .configure(&[peer_info])
        .await
        .map_err(|e| EdgeError::Connect(format!("edge device configure failed: {e}")))?;

    // Device mode has no in-process handshake signal we own; interface-up is
    // treated as Connected.
    let (status_tx, status_rx) = watch::channel(EdgeStatus::Connecting);
    let _ = status_tx.send(EdgeStatus::Connected {
        since: SystemTime::now(),
    });

    let cancel = Cancel::new();
    let mut forward_tasks = Vec::with_capacity(opts.forwards.len());
    for spec in opts.forwards {
        forward_tasks.push(tokio::spawn(async move {
            if let Err(e) = serve_forward(spec, DeviceDialer).await {
                tracing::warn!(local = %spec.local, error = %e, "edge device forward listener exited");
            }
        }));
    }

    let join = tokio::spawn(device_teardown(
        transport,
        forward_tasks,
        status_tx,
        cancel.clone(),
    ));

    Ok(EdgeHandle {
        netstack: None,
        status: status_rx,
        cancel,
        join,
    })
}

/// Device-mode lifetime task: hold the TUN interface up until cancelled, then
/// abort the forward proxies, drop the transport (destroying the interface),
/// and publish [`EdgeStatus::Down`].
async fn device_teardown(
    transport: OverlayTransport,
    forward_tasks: Vec<JoinHandle<()>>,
    status_tx: watch::Sender<EdgeStatus>,
    cancel: Cancel,
) {
    cancel.cancelled().await;
    for task in &forward_tasks {
        task.abort();
    }
    drop(transport);
    let _ = status_tx.send(EdgeStatus::Down);
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[cfg(test)]
mod tests {
    use super::*;
    use std::str::FromStr;

    // --- mode resolution ---

    #[test]
    fn plan_netstack_is_forced() {
        assert_eq!(
            plan_mode(EdgeMode::Netstack, &DeviceVerdict::Grantable),
            ModePlan::Netstack
        );
        assert_eq!(
            plan_mode(EdgeMode::Netstack, &DeviceVerdict::Denied("x".into())),
            ModePlan::Netstack
        );
    }

    #[test]
    fn plan_device_is_forced_without_fallback() {
        assert_eq!(
            plan_mode(EdgeMode::Device, &DeviceVerdict::Grantable),
            ModePlan::Device {
                allow_fallback: false
            }
        );
        assert_eq!(
            plan_mode(EdgeMode::Device, &DeviceVerdict::Denied("nope".into())),
            ModePlan::Device {
                allow_fallback: false
            }
        );
    }

    #[test]
    fn plan_auto_grantable_prefers_device_with_fallback() {
        assert_eq!(
            plan_mode(EdgeMode::Auto, &DeviceVerdict::Grantable),
            ModePlan::Device {
                allow_fallback: true
            }
        );
    }

    #[test]
    fn plan_auto_denied_uses_netstack() {
        assert_eq!(
            plan_mode(
                EdgeMode::Auto,
                &DeviceVerdict::Denied("no CAP_NET_ADMIN".into())
            ),
            ModePlan::Netstack
        );
    }

    // --- backoff ---

    #[test]
    fn backoff_base_doubles_and_caps() {
        assert_eq!(backoff_base(1), Duration::from_secs(1));
        assert_eq!(backoff_base(2), Duration::from_secs(2));
        assert_eq!(backoff_base(3), Duration::from_secs(4));
        assert_eq!(backoff_base(4), Duration::from_secs(8));
        // 2^6 = 64 > 60 → capped.
        assert_eq!(backoff_base(7), EDGE_BACKOFF_MAX);
        // Large attempt: saturates, no shift overflow, still capped.
        assert_eq!(backoff_base(1000), EDGE_BACKOFF_MAX);
        assert_eq!(backoff_base(u32::MAX), EDGE_BACKOFF_MAX);
    }

    #[test]
    fn next_backoff_stays_within_jitter_band() {
        for attempt in 1..=10u32 {
            let base = backoff_base(attempt);
            let lower = base.mul_f64(0.8);
            let upper = base.mul_f64(1.2);
            for _ in 0..200 {
                let d = next_backoff(attempt);
                assert!(d >= lower, "attempt {attempt}: {d:?} < {lower:?}");
                assert!(d <= upper, "attempt {attempt}: {d:?} > {upper:?}");
            }
        }
    }

    #[test]
    fn next_backoff_hits_the_cap_band() {
        // A large attempt is capped at EDGE_BACKOFF_MAX (60s) before jitter, so
        // the jittered result stays within [48s, 72s].
        for _ in 0..200 {
            let d = next_backoff(50);
            assert!(d >= Duration::from_secs(48), "{d:?} below cap band");
            assert!(d <= Duration::from_secs(72), "{d:?} above cap band");
        }
    }

    // --- keepalive clamping ---

    #[test]
    fn keepalive_clamps_and_defaults() {
        assert_eq!(keepalive_secs(25), 25);
        assert_eq!(keepalive_secs(0), DEFAULT_KEEPALIVE_SECS);
        assert_eq!(keepalive_secs(u64::from(u16::MAX) + 100), u16::MAX);
    }

    // --- state dir defaulting ---

    #[test]
    fn state_dir_explicit_override_wins() {
        assert_eq!(
            resolve_state_dir(Some(Path::new("/custom/edge/state"))),
            PathBuf::from("/custom/edge/state")
        );
    }

    #[test]
    fn state_dir_prefers_xdg_state_home() {
        let d = default_state_dir(
            Some(OsString::from("/x/state")),
            Some(OsString::from("/home/u")),
        );
        assert_eq!(d, PathBuf::from("/x/state/zlayer/edge"));
    }

    #[test]
    fn state_dir_falls_back_to_home_local_state() {
        let d = default_state_dir(None, Some(OsString::from("/home/u")));
        assert_eq!(d, PathBuf::from("/home/u/.local/state/zlayer/edge"));
    }

    #[test]
    fn state_dir_ignores_empty_env_values() {
        let d = default_state_dir(Some(OsString::new()), Some(OsString::from("/home/u")));
        assert_eq!(d, PathBuf::from("/home/u/.local/state/zlayer/edge"));
    }

    // --- dialer construction (compile-time trait proof; no tunnel spun up) ---

    #[test]
    fn dialers_implement_tunnel_dialer() {
        fn assert_dialer<D: TunnelDialer>() {}
        assert_dialer::<DeviceDialer>();
        assert_dialer::<NetstackDialer>();
    }

    // --- options / status plumbing ---

    #[test]
    fn edge_options_carry_and_clone_fields() {
        let spec = ForwardSpec::from_str("127.0.0.1:8080=10.42.0.5:80").unwrap();
        let opts = EdgeOptions {
            forwards: vec![spec],
            mode: EdgeMode::Netstack,
            state_dir: Some(PathBuf::from("/s")),
        };
        assert_eq!(opts.forwards.len(), 1);
        assert_eq!(opts.mode, EdgeMode::Netstack);
        assert_eq!(opts.state_dir.as_deref(), Some(Path::new("/s")));

        let cloned = opts.clone();
        assert_eq!(cloned.forwards[0], spec);
        assert_eq!(cloned.mode, EdgeMode::Netstack);
    }

    #[test]
    fn edge_options_default_is_auto_and_empty() {
        let opts = EdgeOptions::default();
        assert!(opts.forwards.is_empty());
        assert_eq!(opts.mode, EdgeMode::Auto);
        assert!(opts.state_dir.is_none());
    }

    #[test]
    fn edge_status_equality() {
        assert_eq!(EdgeStatus::Down, EdgeStatus::Down);
        assert_ne!(EdgeStatus::Connecting, EdgeStatus::Down);

        let t = SystemTime::now();
        assert_eq!(
            EdgeStatus::Connected { since: t },
            EdgeStatus::Connected { since: t }
        );

        assert_eq!(
            EdgeStatus::Reconnecting { attempt: 3 },
            EdgeStatus::Reconnecting { attempt: 3 }
        );
        assert_ne!(
            EdgeStatus::Reconnecting { attempt: 3 },
            EdgeStatus::Reconnecting { attempt: 4 }
        );
    }

    // --- endpoint resolution (IP literals only; no DNS / no network) ---

    #[tokio::test]
    async fn resolve_endpoint_parses_ip_literal() {
        let addr = resolve_endpoint("203.0.113.7:51820").await.unwrap();
        assert_eq!(addr, "203.0.113.7:51820".parse::<SocketAddr>().unwrap());
    }

    #[tokio::test]
    async fn resolve_endpoint_rejects_portless_literal() {
        // A portless literal is an invalid socket address; `lookup_host` rejects
        // it synchronously without a network round-trip.
        assert!(resolve_endpoint("203.0.113.7").await.is_err());
    }

    // --- cancellation token ---

    #[tokio::test]
    async fn cancel_resolves_after_trip() {
        let cancel = Cancel::new();
        assert!(!cancel.is_cancelled());
        let waiter = cancel.clone();
        let h = tokio::spawn(async move { waiter.cancelled().await });
        cancel.cancel();
        assert!(cancel.is_cancelled());
        tokio::time::timeout(Duration::from_secs(5), h)
            .await
            .expect("cancelled() must resolve after cancel()")
            .expect("waiter task joins");
    }

    #[tokio::test]
    async fn cancel_is_immediate_when_already_tripped() {
        let cancel = Cancel::new();
        cancel.cancel();
        // Already tripped: resolves immediately.
        tokio::time::timeout(Duration::from_secs(5), cancel.cancelled())
            .await
            .expect("already-cancelled token resolves immediately");
    }
}