use clap::Parser;
use futures::future;
use rust_bfd::config::{Config, ConfigError, ResolvedConfig, ResolvedPeer};
use rust_bfd::events::{EventMessageSink, ScriptHookSink, StateChangeEvent};
use rust_bfd::{bfd::*, events::EventMessage};
use rust_bfd::{set_ttl_or_hop_limit, AtomicInstant, AtomicIntervalStats, BfdSessionStats};
use std::{
    collections::{HashMap, HashSet},
    error::Error,
    fmt::Display,
    io::ErrorKind,
    mem::swap,
    net::{IpAddr, SocketAddr},
    path::PathBuf,
    sync::{atomic::Ordering, Arc},
    time::{Duration, Instant},
};

// Necessary until a hop_limit method is added to UdpSocket

use tokio::sync::{oneshot, Mutex};

use env_logger::Env;
use log::{debug, info, log_enabled, warn};

use nom_derive::Parse;
use rand::prelude::*;
use tokio::io;
use tokio::net::UdpSocket;
use tokio::signal::unix::{signal, SignalKind};
use tokio::sync::mpsc;
use tokio::task::JoinHandle;
use tokio::time;

use atomic::Atomic;

const CONTROL_PORT: u16 = 3784;
const ECHO_PORT: u16 = 3785;

#[derive(Parser, Debug)]
#[command(version, about = "BFD (RFC 5880) daemon")]
struct Args {
    /// YAML configuration file [default: /etc/rust-bfd/config.yaml, ignored if absent]
    #[arg(short, long, value_name = "FILE", env = "BFD_CONFIG")]
    config: Option<PathBuf>,

    /// Local IP address to bind to and use as the source of BFD control packets (repeatable). Replaces any bind
    /// addresses from the config file or environment.
    #[arg(short, long, value_name = "IP")]
    bind: Vec<IpAddr>,

    /// Peer IP address to establish a session with, using the default timers (repeatable). Added to the peers from
    /// the config file.
    #[arg(short, long = "peer", value_name = "IP")]
    peers: Vec<IpAddr>,
}

// RFC 5881 section 4: the source port MUST be in the range 49152 to 65535.
const SOURCE_PORT_RANGE: std::ops::RangeInclusive<u16> = 49152..=65535;
const SOURCE_PORT_BIND_ATTEMPTS: usize = 32;

/// Bind a UDP socket on a random unused port in the RFC 5881 source port range, with TTL/hop limit 255.
async fn bind_control_socket(local_addr: IpAddr) -> io::Result<UdpSocket> {
    for _ in 0..SOURCE_PORT_BIND_ATTEMPTS {
        let port = rand::thread_rng().gen_range(SOURCE_PORT_RANGE);
        match UdpSocket::bind(SocketAddr::new(local_addr, port)).await {
            Ok(sock) => {
                set_ttl_or_hop_limit(&sock, 255)?;
                return Ok(sock);
            }
            Err(e) if e.kind() == ErrorKind::AddrInUse => continue,
            Err(e) => return Err(e),
        }
    }
    Err(io::Error::new(
        ErrorKind::AddrInUse,
        format!(
            "no free source port found for {local_addr} after {SOURCE_PORT_BIND_ATTEMPTS} attempts"
        ),
    ))
}

const ORDERING: Ordering = Ordering::Relaxed;

#[derive(Debug)]
pub enum BfdError {
    // field, value
    InvalidFieldValue(&'static str, &'static str),
}
impl Display for BfdError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            Self::InvalidFieldValue(field, value) => {
                write!(f, "invalid value `{}` for field `{}`", value, field)
            }
        }
    }
}
impl Error for BfdError {}

/// Data structure to store the state of the Bfd machine. The impl does *not*
/// implement any Bfd logic, it is merely a thread-safe data structure
#[derive(Debug)]
struct BfdSessionState {
    control_sock: Arc<UdpSocket>,                  // immutable
    local_ip: IpAddr,                              // immutable and Copy
    peer_addr: IpAddr,                             // immutable and Copy
    session_state: Atomic<BfdState>,               // mutable by any thread
    remote_session_state: Atomic<BfdState>,        // mutable by rx thread
    local_discr: BfdDiscriminator,                 // immutable and Copy
    remote_discr: Atomic<BfdDiscriminator>,        // mutable by rx thread
    local_diag: Atomic<BfdDiagnostic>,             // mutable by any thread
    desired_min_tx_interval: Atomic<BfdInterval>,  // mutable by any thread
    required_min_rx_interval: Atomic<BfdInterval>, // mutable by any thread
    remote_min_rx_interval: Atomic<BfdInterval>,   // mutable by rx thread
    demand_mode: Atomic<bool>,                     // mutable by rx thread
    remote_demand_mode: Atomic<bool>,              // mutable by rx thread
    detect_mult: Atomic<u8>,                       // mutable by control thread
    auth_type: Atomic<BfdAuthType>,                // mutable by control thread
    rcv_auth_seq: Atomic<u32>,                     // mutable by rx thread
    xmit_auth_seq: Atomic<u32>,                    // mutable by tx thread?
    auth_seq_known: Atomic<bool>,                  // mutable by rx thread?

    detection_time: Atomic<BfdInterval>, // mutable by rx thread
    poll_mode: Atomic<bool>,             // mutable by rx thread

    rx_stats: AtomicIntervalStats, // reset each time the session enters Up
    tx_stats: AtomicIntervalStats, // reset each time the session enters Up
    last_up: AtomicInstant,
    last_down: AtomicInstant,
    last_state_change: Mutex<Instant>,
    last_remote_diag: Atomic<BfdDiagnostic>,
}

// TODO: hide the internal data structure behind a trait
impl BfdSessionState {
    fn control_sock(&self) -> Arc<UdpSocket> {
        self.control_sock.clone()
    }
    fn local_ip(&self) -> IpAddr {
        self.local_ip
    }
    fn peer_addr(&self) -> IpAddr {
        self.peer_addr
    }

    fn session_state(&self) -> BfdState {
        self.session_state.load(ORDERING)
    }
    // Note: This does not handle state transition
    fn set_session_state(&self, value: BfdState) {
        self.session_state.store(value, ORDERING)
    }
    fn remote_session_state(&self) -> BfdState {
        self.remote_session_state.load(ORDERING)
    }
    fn set_remote_session_state(&self, value: BfdState) {
        self.remote_session_state.store(value, ORDERING);
    }
    fn local_discr(&self) -> BfdDiscriminator {
        self.local_discr
    }
    fn remote_discr(&self) -> BfdDiscriminator {
        self.remote_discr.load(ORDERING)
    }
    fn set_remote_discr(&self, value: BfdDiscriminator) {
        self.remote_discr.store(value, ORDERING);
    }
    fn demand_mode(&self) -> bool {
        self.demand_mode.load(ORDERING)
    }
    fn set_demand_mode(&self, value: bool) {
        self.demand_mode.store(value, ORDERING)
    }
    fn remote_demand_mode(&self) -> bool {
        self.remote_demand_mode.load(ORDERING)
    }
    fn set_remote_demand_mode(&self, value: bool) {
        self.remote_demand_mode.store(value, ORDERING);
    }
    fn remote_min_rx_interval(&self) -> BfdInterval {
        self.remote_min_rx_interval.load(ORDERING)
    }
    fn set_remote_min_rx_interval(&self, value: BfdInterval) {
        self.remote_min_rx_interval.store(value, ORDERING);
    }
    fn local_diag(&self) -> BfdDiagnostic {
        self.local_diag.load(ORDERING)
    }
    fn set_local_diag(&self, value: BfdDiagnostic) {
        self.local_diag.store(value, ORDERING);
    }
    fn auth_type(&self) -> BfdAuthType {
        self.auth_type.load(ORDERING)
    }
    fn detect_mult(&self) -> u8 {
        self.detect_mult.load(ORDERING)
    }
    fn set_detect_mult(&self, value: u8) {
        self.detect_mult.store(value, ORDERING)
    }
    fn poll_mode(&self) -> bool {
        self.poll_mode.load(ORDERING)
    }
    fn set_poll_mode(&self, value: bool) {
        self.poll_mode.store(value, ORDERING)
    }
    fn desired_min_tx_interval(&self) -> BfdInterval {
        self.desired_min_tx_interval.load(ORDERING)
    }
    fn set_desired_min_tx_interval(&self, value: BfdInterval) {
        self.desired_min_tx_interval.store(value, ORDERING);
    }
    fn required_min_rx_interval(&self) -> BfdInterval {
        self.required_min_rx_interval.load(ORDERING)
    }
    fn set_required_min_rx_interval(&self, value: BfdInterval) {
        self.required_min_rx_interval.store(value, ORDERING)
    }
    fn detection_time(&self) -> BfdInterval {
        self.detection_time.load(ORDERING)
    }
    fn set_detection_time(&self, value: BfdInterval) {
        self.detection_time.store(value, ORDERING)
    }
    fn record_rx(&self) {
        self.rx_stats.record(Instant::now());
    }
    fn record_tx(&self) {
        self.tx_stats.record(Instant::now());
    }
    fn mark_up(&self) {
        self.rx_stats.reset();
        self.tx_stats.reset();
        self.last_up.store(Instant::now());
    }
    fn mark_down(&self) {
        self.last_down.store(Instant::now());
    }
    fn last_state_change(&self) -> Instant {
        *self.last_state_change.try_lock().unwrap()
    }
    fn last_remote_diag(&self) -> BfdDiagnostic {
        self.last_remote_diag.load(ORDERING)
    }
    fn set_last_remote_diag(&self, value: BfdDiagnostic) {
        self.last_remote_diag.store(value, ORDERING)
    }
}

enum WatchdogReset {
    ReceivedPacket,
    Terminate,
}

struct BfdSession {
    state: BfdSessionState,
    periodic: BfdPeriodicSenderHandle,
    rx: mpsc::Receiver<SessionControlCommand>,
    tx: mpsc::Sender<SessionControlCommand>,
    rx_watchdog: Option<mpsc::Sender<WatchdogReset>>,
    cur_interval: BfdInterval,
    // Configured DesiredMinTxInterval; advertised only once Up (RFC 5880 6.8.3 requires >= 1s otherwise)
    configured_desired_min_tx: BfdInterval,
    // DesiredMinTxInterval actually used for the transmit rate; lags the advertised value during a Poll Sequence
    effective_desired_min_tx: BfdInterval,
    event_dispatcher: EventDispatcherHandle,
}

impl From<&BfdSessionState> for BfdSessionStats {
    fn from(state: &BfdSessionState) -> Self {
        Self {
            local_ip: state.local_ip(),
            remote_ip: state.peer_addr(),
            local_discr: state.local_discr(),
            remote_discr: state.remote_discr(),
            state: state.session_state(),
            last_diag: state.last_remote_diag(),
            rx: state.rx_stats.snapshot(),
            tx: state.tx_stats.snapshot(),
            last_up: state.last_up.load(),
            last_down: state.last_down.load(),
            last_change: state.last_state_change(),
            detect_time: state.detection_time(),
            base_interval: BfdInterval::from_secs(0), // TODO: This is a hack and will be patched up in the session
        }
    }
}
enum PeriodicControlCommand {
    Stop,
    Start,
    Quit,
    SetMinInterval(BfdInterval),
    SendPollResponse,
}

enum SessionControlCommand {
    RxPacket(BfdPacket),
    TxControlPacket {
        poll_response: bool,
    },
    GetSessionStats {
        respond_to: oneshot::Sender<BfdSessionStats>,
    },
    ReceiveTimeout,
    Quit,
}

#[derive(Clone)]
struct BfdSessionHandle {
    tx: mpsc::Sender<SessionControlCommand>,
}

impl BfdSessionHandle {
    async fn new(
        local_addr: IpAddr,
        peer: ResolvedPeer,
        local_discr: BfdDiscriminator,
        event_dispatcher: EventDispatcherHandle,
    ) -> io::Result<Self> {
        let control_sock = bind_control_socket(local_addr).await?;

        let (tx, rx) = mpsc::channel(32);
        let new_self = Self { tx };

        let mut session = BfdSession::new(
            rx,
            new_self.clone(),
            control_sock,
            local_addr,
            peer,
            local_discr,
            event_dispatcher,
        );
        tokio::spawn(async move { session.run().await });

        Ok(new_self)
    }
    async fn transmit_control_packet(
        &self,
        poll_response: bool,
    ) -> Result<(), mpsc::error::SendError<SessionControlCommand>> {
        self.tx
            .send(SessionControlCommand::TxControlPacket { poll_response })
            .await
    }
    async fn get_stats(
        &self,
    ) -> Result<BfdSessionStats, mpsc::error::SendError<SessionControlCommand>> {
        let (send, recv) = oneshot::channel();
        self.tx
            .send(SessionControlCommand::GetSessionStats { respond_to: send })
            .await?;
        Ok(recv.await.expect("Unable to receive stats from session"))
    }
}

#[derive(Clone, Debug)]
struct BfdPeriodicSenderHandle {
    tx: mpsc::Sender<PeriodicControlCommand>,
}

impl BfdPeriodicSenderHandle {
    fn new(bfd: BfdSessionHandle, interval: BfdInterval, detect_mult: u8) -> Self {
        let (tx, rx) = mpsc::channel(32);
        let mut periodic_sender = BfdPeriodicSender::new(rx, bfd, interval, detect_mult);
        tokio::spawn(async move { periodic_sender.run().await });

        Self { tx }
    }
}

struct BfdPeriodicSender {
    rx: mpsc::Receiver<PeriodicControlCommand>,
    bfd: BfdSessionHandle,
    cur_interval: BfdInterval,
    detect_mult: u8,
}

fn jittered_delay(interval: time::Duration, detect_mult: u8) -> time::Duration {
    // https://datatracker.ietf.org/doc/html/rfc5880#section-6.8.7
    let min_percent = if detect_mult == 1 { 90 } else { 75 };
    rand::thread_rng().gen_range(interval * min_percent / 100..=interval)
}

impl BfdPeriodicSender {
    fn new(
        rx: mpsc::Receiver<PeriodicControlCommand>,
        bfd: BfdSessionHandle,
        interval: BfdInterval,
        detect_mult: u8,
    ) -> Self {
        Self {
            rx,
            bfd,
            cur_interval: interval,
            detect_mult,
        }
    }

    fn next_delay(&self) -> time::Duration {
        jittered_delay(self.cur_interval.into(), self.detect_mult)
    }
    async fn run(&mut self) {
        let mut running = true;
        let mut next_tx = time::Instant::now() + self.next_delay();

        loop {
            tokio::select! {
             _ = time::sleep_until(next_tx), if running => {
                        self.bfd.transmit_control_packet(false).await.unwrap();
                        next_tx = time::Instant::now() + self.next_delay();
                    },
                    Some(cmd) = self.rx.recv() => {
                        match cmd {
                            PeriodicControlCommand::Quit => { debug!("Closing periodic packet sender"); return},
                            PeriodicControlCommand::Stop => {
                                debug!("Stopping periodic packets");
                                running= false}
                            ,
                            PeriodicControlCommand::Start => {
                                debug!("Starting periodic packets");
                                running = true;
                                next_tx = time::Instant::now() + self.next_delay();
                            },
                            PeriodicControlCommand::SetMinInterval(i) => {
                                self.cur_interval = i;
                                debug!("Updating transmit interval to {}ms", Duration::from(i).as_millis());
                                next_tx = time::Instant::now() + self.next_delay();
                            },
                            PeriodicControlCommand::SendPollResponse => {
                                debug!("Sending poll response");
                                self.bfd.transmit_control_packet(true).await.unwrap();
                            }
                        }
                    }
            };
        }
    }
}

impl BfdSession {
    fn new(
        rx: mpsc::Receiver<SessionControlCommand>,
        handle: BfdSessionHandle,
        control_sock: UdpSocket,
        local_addr: IpAddr,
        peer: ResolvedPeer,
        local_discr: BfdDiscriminator,
        event_dispatcher: EventDispatcherHandle,
    ) -> Self {
        let mut rng = rand::thread_rng();
        let periodic = BfdPeriodicSenderHandle::new(
            handle.clone(),
            time::Duration::from_secs(1).into(),
            peer.detect_mult,
        );
        let configured_desired_min_tx = peer.desired_min_tx;
        let slow_desired_min_tx =
            std::cmp::max(configured_desired_min_tx, BfdInterval::from_secs(1));

        // control_sock
        //     .connect(SocketAddr::new(remote_addr, CONTROL_PORT))
        //     .await?;

        // Incoming packets will come over the channel from the mux, since they don't send to the reciprocal port
        Self {
            rx,
            tx: handle.tx,
            periodic,
            rx_watchdog: None,
            cur_interval: BfdInterval::from_secs(1),
            configured_desired_min_tx,
            effective_desired_min_tx: slow_desired_min_tx,
            event_dispatcher,
            state: BfdSessionState {
                control_sock: Arc::new(control_sock),
                local_ip: local_addr,
                peer_addr: peer.address,
                session_state: Atomic::new(BfdState::default()),
                remote_session_state: Atomic::new(BfdState::default()),
                local_discr,
                remote_discr: Atomic::new(BfdDiscriminator(0)),
                local_diag: Atomic::new(BfdDiagnostic::None),
                desired_min_tx_interval: Atomic::new(slow_desired_min_tx),
                required_min_rx_interval: Atomic::new(peer.required_min_rx),
                remote_min_rx_interval: Atomic::new(BfdInterval::from_millis(300)),
                demand_mode: Atomic::new(false),
                remote_demand_mode: Atomic::new(false),
                detect_mult: Atomic::new(peer.detect_mult),
                auth_type: Atomic::new(BfdAuthType::None),
                rcv_auth_seq: Atomic::new(0),
                xmit_auth_seq: Atomic::new(rng.gen()),
                auth_seq_known: Atomic::new(false),
                detection_time: Atomic::new(BfdInterval::from_secs(0)),
                poll_mode: Atomic::new(false),

                rx_stats: AtomicIntervalStats::new(),
                tx_stats: AtomicIntervalStats::new(),
                last_up: AtomicInstant::new(),
                last_down: AtomicInstant::new(),
                last_state_change: Mutex::new(Instant::now()),
                last_remote_diag: Atomic::new(BfdDiagnostic::None),
            },
        }
    }

    async fn transition(&mut self, new_state: BfdState) {
        if self.state.session_state() == new_state {
            return;
        };
        self.event_dispatcher
            .dispatch(EventMessage::StateChange(StateChangeEvent {
                local_discr: self.state.local_discr(),
                remote_discr: self.state.remote_discr(),
                local_ip: self.state.local_ip(),
                remote_ip: self.state.peer_addr(),
                from_state: self.state.session_state(),
                to_state: new_state,
            }))
            .await;
        info!(
            "Peer {} state change {} -> {}",
            self.state.peer_addr(),
            self.state.session_state(),
            new_state
        );
        *self.state.last_state_change.lock().await = Instant::now();
        match (self.state.session_state(), new_state) {
            (BfdState::Up, BfdState::AdminDown | BfdState::Down | BfdState::Init) => {
                self.state.mark_down();
                if let Some(watchdog) = self.rx_watchdog.as_mut() {
                    watchdog.send(WatchdogReset::Terminate).await.unwrap();
                };
                //   When bfd.SessionState is not Up, the system MUST set bfd.DesiredMinTxInterval to a value of not
                //    less than one second (1,000,000 microseconds).  This is intended to ensure that the bandwidth
                //    consumed by BFD sessions that are not Up is negligible, particularly in the case where a neighbor
                //    may not be running BFD.
                self.state.set_desired_min_tx_interval(std::cmp::max(
                    self.configured_desired_min_tx,
                    BfdInterval::from_secs(1),
                ));
                self.state.set_poll_mode(false);
                self.set_effective_desired_min_tx(self.state.desired_min_tx_interval());
                self.update_transmit_interval().await;
            }
            (BfdState::AdminDown | BfdState::Down | BfdState::Init, BfdState::Up) => {
                self.state.mark_up();
                self.start_watchdog().await;
                self.state.set_local_diag(BfdDiagnostic::None); // This doesn't seem to be required by the spec but seems sensible and at least FRR does it.
                                                                // Advertise the configured interval now, but per section 6.8.3 keep transmitting at the old (slower)
                                                                // rate until a Poll Sequence confirms the peer has seen the change.
                self.state
                    .set_desired_min_tx_interval(self.configured_desired_min_tx);
                if self.configured_desired_min_tx < self.effective_desired_min_tx {
                    self.state.set_poll_mode(true);
                } else {
                    self.set_effective_desired_min_tx(self.configured_desired_min_tx);
                    self.update_transmit_interval().await;
                }
            }
            _ => {}
        }
        self.state.set_session_state(new_state);
        // https://datatracker.ietf.org/doc/html/rfc5880#section-6.8.7
        //   A BFD Control packet SHOULD be transmitted during the interval between periodic Control packet
        //   transmissions when the contents of that packet would differ from that in the previously transmitted packet
        self.transmit_control_packet(false).await;
    }

    async fn run(&mut self) {
        while let Some(cmd) = self.rx.recv().await {
            match cmd {
                SessionControlCommand::Quit => return,
                SessionControlCommand::RxPacket(packet) => {
                    self.receive_control_packet(&packet).await;
                }
                SessionControlCommand::GetSessionStats { respond_to } => {
                    let stats =
                        BfdSessionStats::from(&self.state).with_base_interval(self.cur_interval);
                    respond_to.send(stats).unwrap()
                }
                SessionControlCommand::TxControlPacket { poll_response } => {
                    self.transmit_control_packet(poll_response).await
                }
                SessionControlCommand::ReceiveTimeout => {
                    self.state.set_local_diag(BfdDiagnostic::TimeExpired);
                    self.transition(BfdState::Down).await;
                }
            }
        }
        println!("Session command channel closed");
    }

    async fn transmit_control_packet(&self, poll_response: bool) {
        let packet = BfdPacket {
            flags: BfdFlags(0)
                .with_vers(1)
                .with_diag(self.state.local_diag().into())
                .with_state(self.state.session_state().into())
                .with_poll(if poll_response {
                    false
                } else {
                    self.state.poll_mode()
                })
                .with_final_(poll_response)
                .with_cpi(false)
                .with_demand(self.state.demand_mode())
                .with_detect_mult(self.state.detect_mult())
                .with_length(24),
            my_disc: self.state.local_discr(),
            your_disc: self.state.remote_discr(),
            desired_min_tx: self.state.desired_min_tx_interval(),
            required_min_rx: self.state.required_min_rx_interval(),
            required_min_echo_rx: BfdInterval::from_secs(0),
            auth: None,
        };
        let socket = self.state.control_sock.clone();
        let dest = self.state.peer_addr;
        self.state.record_tx();
        debug!("tx packet: {:?}", packet);
        if let Err(e) = socket
            .send_to(
                packet.serialize().unwrap().as_ref(),
                SocketAddr::new(dest, CONTROL_PORT),
            )
            .await
        {
            warn!(
                "Error sending packet to {:?}:{} ({:?})",
                dest, CONTROL_PORT, e
            );
        }
    }

    async fn start_watchdog(&mut self) {
        let (tx, mut rx) = mpsc::channel(1);
        let cmd_tx = self.tx.clone();
        let duration = self.state.detection_time().into();
        let peer_ip = self.state.peer_addr();

        tokio::spawn(async move {
            let mut running = true;
            while running {
                match tokio::time::timeout(duration, async { rx.recv().await }).await {
                    Ok(Some(WatchdogReset::ReceivedPacket)) => {
                        debug!("Packet received, watchdog ok")
                    }
                    Ok(Some(WatchdogReset::Terminate)) | Ok(None) => {
                        debug!("Watchdog terminating");
                        running = false
                    }
                    Err(_) => {
                        info!(
                            "Peer {} timed out after {}ms",
                            peer_ip,
                            duration.as_millis()
                        );
                        cmd_tx
                            .send(SessionControlCommand::ReceiveTimeout)
                            .await
                            .unwrap()
                    }
                }
            }
        });
        let mut temporary = Some(tx);
        swap(&mut temporary, &mut self.rx_watchdog);
        if let Some(old_watchdog) = temporary.as_mut() {
            debug!(
                "Updating old watchdog with new detection time ({}ms)",
                duration.as_millis()
            );
            old_watchdog
                .send(WatchdogReset::Terminate)
                .await
                .unwrap_or(())
        }
    }

    // https://datatracker.ietf.org/doc/html/rfc5880#section-6.8.6
    async fn receive_control_packet(&mut self, p: &BfdPacket) {
        // https://datatracker.ietf.org/doc/html/rfc5880#section-6.8.4
        //
        // If Demand mode is not active, and a period of time equal to the
        // Detection Time passes without receiving a BFD Control packet from the
        // remote system, and bfd.SessionState is Init or Up, the session has
        // gone down -- the local system MUST set bfd.SessionState to Down and
        // bfd.LocalDiag to 1 (Control Detection Time Expired).
        if let Some(watchdog) = self.rx_watchdog.as_ref() {
            let _ = watchdog.send(WatchdogReset::ReceivedPacket).await;
        }
        self.state.record_rx();
        let received_state = match p.flags.state() {
            Err(_) => {
                warn!("Invalid state, discarding");
                return;
            }
            Ok(v) => v,
        };
        // If the version number is not correct (1), the packet MUST be discarded.
        if p.flags.vers() != 1 {
            warn!("Invalid version {}, discarding", p.flags.vers());
            return;
        }
        //   If the Length field is less than the minimum correct value (24 if the A bit is clear, or 26 if the A bit is
        //   set), the packet MUST be discarded.
        if p.flags.length() < 24 || (p.flags.length() < 26 && p.flags.auth_present()) {
            warn!("Invalid packet length {}, discarding", p.flags.length());
            return;
        }

        // If the Detect Mult field is zero, the packet MUST be discarded.
        if p.flags.detect_mult() == 0 {
            warn!("Invalid detect mult {}, discarding", p.flags.detect_mult());
            return;
        }

        //If the Multipoint (M) bit is nonzero, the packet MUST be discarded.
        if p.flags.multipoint() {
            warn!("Invalid multipoint enabled, discarding");
            return;
        }

        // If the My Discriminator field is zero, the packet MUST be discarded.
        if p.my_disc == BfdDiscriminator(0) {
            warn!("Invalid my discriminator {:?}, discarding", p.my_disc);
            return;
        }

        // If the Your Discriminator field is nonzero, it MUST be used to select the session with which this BFD packet
        //   is associated.  If no session is found, the packet MUST be discarded.
        if p.your_disc != BfdDiscriminator(0) && p.your_disc != self.state.local_discr() {
            warn!(
                "Received unexpected discriminator {:?}, discarding",
                p.your_disc
            );
            return;
        }

        // If the Your Discriminator field is zero and the State field is not Down or AdminDown, the packet MUST be
        //   discarded.
        if p.your_disc == BfdDiscriminator(0)
            && (received_state != BfdState::Down && received_state != BfdState::AdminDown)
        {
            warn!(
                "Got packet with zero discriminator and invalid state {:?}, discarding",
                received_state
            );
            return;
        }

        // If the A bit is set and no authentication is in use (bfd.AuthType is zero), the packet MUST be discarded.
        if p.flags.auth_present() && self.state.auth_type() == BfdAuthType::None {
            warn!("Got packet with auth enabled when we disagree, discarding");
            return;
        }

        // If the A bit is clear and authentication is in use (bfd.AuthType is nonzero), the packet MUST be discarded.
        if !p.flags.auth_present() && self.state.auth_type() != BfdAuthType::None {
            warn!("Got packet without auth when we expect it, discarding");
            return;
        }

        // If the A bit is set, the packet MUST be authenticated under the rules of section 6.7, based on the
        //   authentication type in use (bfd.AuthType).  This may cause the packet to be discarded.
        if p.flags.auth_present() {
            unimplemented!("Authentication is not implemented");
        }

        // Set bfd.RemoteDiscr to the value of My Discriminator.
        self.state.set_remote_discr(p.my_disc);
        // Set bfd.RemoteState to the value of the State (Sta) field.
        self.state.set_remote_session_state(received_state);
        // Set bfd.RemoteDemandMode to the value of the Demand (D) bit.
        self.state.set_remote_demand_mode(p.flags.demand());
        // Set bfd.RemoteMinRxInterval to the value of Required Min RX Interval.
        self.state.set_remote_min_rx_interval(p.required_min_rx);
        // Not in spec: store the received diagnostic
        self.state
            .set_last_remote_diag(p.flags.diag().unwrap_or(BfdDiagnostic::Reserved));

        // If the Required Min Echo RX Interval field is zero, the transmission of Echo packets, if any, MUST cease.
        if p.required_min_echo_rx == BfdInterval::from_secs(0) {
            // TODO: implement echo thread
        }
        // If a Poll Sequence is being transmitted by the local system and the Final (F) bit in the received packet is
        //   set, the Poll Sequence MUST be terminated.
        if self.state.poll_mode() && p.flags.final_() {
            self.state.set_poll_mode(false);
            self.set_effective_desired_min_tx(self.state.desired_min_tx_interval());
        }

        // Update the transmit interval as described in section 6.8.2.
        self.update_transmit_interval().await;

        // Update the Detection Time as described in section 6.8.4.
        self.update_detection_time(p).await;

        // If bfd.SessionState is AdminDown
        //   Discard the packet
        if self.state.session_state() == BfdState::AdminDown {
            return;
        }

        // If received state is AdminDown
        if received_state == BfdState::AdminDown {
            // If bfd.SessionState is not Down
            if self.state.session_state() != BfdState::Down {
                // Set bfd.LocalDiag to 3 (Neighbor signaled session down)
                self.state.set_local_diag(BfdDiagnostic::NeighborDown);
                // Set bfd.SessionState to Down
                self.transition(BfdState::Down).await;
            }
        } else {
            // If bfd.SessionState ...
            match self.state.session_state() {
                BfdState::Down => {
                    // If received State is Down
                    if received_state == BfdState::Down {
                        // Set bfd.SessionState to Init
                        self.transition(BfdState::Init).await;
                    // Else if received State is Init
                    } else if received_state == BfdState::Init {
                        // Set bfd.SessionState to Up
                        self.transition(BfdState::Up).await;
                    }
                }
                BfdState::Init => {
                    // If received State is Init or Up
                    if received_state == BfdState::Init || received_state == BfdState::Up {
                        // Set bfd.SessionState to Up
                        self.transition(BfdState::Up).await;
                    }
                }
                BfdState::Up => {
                    // If received State is Down
                    if received_state == BfdState::Down {
                        // Set bfd.LocalDiag to 3 (Neighbor signaled session down)
                        self.state.set_local_diag(BfdDiagnostic::NeighborDown);
                        // Set bfd.SessionState to Down
                        self.transition(BfdState::Down).await;
                    }
                }
                BfdState::AdminDown => unreachable!("unexpected AdminDown"), // AdminDown is discarded earlier
            }

            // Check to see if Demand mode should become active or not (see section 6.6).
            // - we don't use demand mode

            // If bfd.RemoteDemandMode is 1, bfd.SessionState is Up, and bfd.RemoteSessionState is Up, Demand mode is
            // active on the remote system and
            if self.state.remote_demand_mode()
                && self.state.session_state() == BfdState::Up
                && self.state.remote_session_state() == BfdState::Up
            {
                // the local system MUST cease the periodic transmission of BFD Control packets (see section 6.8.7).
                self.periodic
                    .tx
                    .send(PeriodicControlCommand::Stop)
                    .await
                    .unwrap();
            }

            // If the Poll (P) bit is set, send a BFD Control packet to the remote system with the Poll (P) bit clear,
            // and the Final (F) bit set (see section 6.8.7).
            if p.flags.poll() {
                self.transmit_control_packet(true).await;
            }
            // If the packet was not discarded, it has been received for purposes of the Detection Time expiration rules
            // in section 6.8.4.
        }
    }

    fn set_effective_desired_min_tx(&mut self, value: BfdInterval) {
        if value != self.effective_desired_min_tx {
            info!(
                "Peer {} effective desired-min-tx {} -> {}",
                self.state.peer_addr(),
                self.effective_desired_min_tx,
                value
            );
        }
        self.effective_desired_min_tx = value;
    }

    // https://datatracker.ietf.org/doc/html/rfc5880#section-6.8.2
    async fn update_transmit_interval(&mut self) {
        let new_base_interval = std::cmp::max(
            self.effective_desired_min_tx,
            self.state.remote_min_rx_interval(),
        );
        if new_base_interval != self.cur_interval {
            info!(
                "Peer {} transmit interval {} -> {}",
                self.state.peer_addr(),
                self.cur_interval,
                new_base_interval
            );
            self.periodic
                .tx
                .send(PeriodicControlCommand::SetMinInterval(new_base_interval))
                .await
                .unwrap()
        }
        self.cur_interval = new_base_interval;
    }

    // https://datatracker.ietf.org/doc/html/rfc5880#section-6.8.4
    async fn update_detection_time(&mut self, p: &BfdPacket) {
        let old = self.state.detection_time();
        // Remote-supplied values can overflow the 32-bit microsecond range; saturate rather than panic.
        let saturating = |mult: u8, interval: BfdInterval| {
            interval
                .checked_mul(mult as u32)
                .unwrap_or(BfdInterval::from_micros(u32::MAX))
        };
        self.state.set_detection_time(if !self.state.demand_mode() {
            saturating(
                p.flags.detect_mult(),
                std::cmp::max(self.state.required_min_rx_interval(), p.desired_min_tx),
            )
        } else {
            saturating(
                self.state.detect_mult(),
                std::cmp::max(
                    self.state.desired_min_tx_interval(),
                    self.state.remote_min_rx_interval(),
                ),
            )
        });
        if old != self.state.detection_time() {
            info!(
                "Peer {} detection time {} -> {}",
                self.state.peer_addr(),
                old,
                self.state.detection_time()
            );
            self.start_watchdog().await
        }
    }
}

#[derive(Debug)]
enum EventDispatcherCommand {
    AddSink(Box<dyn EventMessageSink>),
    DispatchMessage(EventMessage),
}

struct EventDispatcher {
    event_sinks: Vec<mpsc::Sender<EventMessage>>,
    event_sink_handles: Vec<JoinHandle<()>>,
    tx: mpsc::Sender<EventDispatcherCommand>,
    rx: mpsc::Receiver<EventDispatcherCommand>, // TODO: there must be a better way to do this
}

impl EventDispatcher {
    fn new(
        channel_tx: mpsc::Sender<EventDispatcherCommand>,
        channel_rx: mpsc::Receiver<EventDispatcherCommand>,
    ) -> Self {
        let event_sinks = Vec::new();
        let event_sink_handles = Vec::new();

        Self {
            event_sinks,
            event_sink_handles,
            tx: channel_tx,
            rx: channel_rx,
        }
    }
    fn add_sink(&mut self, sink: Box<dyn EventMessageSink>) {
        // Save the transmit channel
        self.event_sinks.push(sink.channel());
        // Run the sink
        self.event_sink_handles.push(sink.run());
    }
    async fn dispatch(&self, event: EventMessage) {
        debug!(
            "EventDispatcher dispatching event {:?} to sinks {:?}",
            event, self.event_sinks
        );
        for sink in &self.event_sinks {
            sink.send(event.clone()).await.unwrap();
        }
    }
    async fn run(mut self) {
        info!("Event dispatcher running");
        while let Some(cmd) = self.rx.recv().await {
            debug!("EventDispatcher got command {:?}", cmd);
            match cmd {
                EventDispatcherCommand::AddSink(sink) => self.add_sink(sink),
                EventDispatcherCommand::DispatchMessage(event) => self.dispatch(event).await,
            }
        }
    }
}

#[derive(Clone)]
struct EventDispatcherHandle {
    channel_tx: mpsc::Sender<EventDispatcherCommand>,
}

impl EventDispatcherHandle {
    async fn new() -> Self {
        let (channel_tx, channel_rx) = mpsc::channel(16);
        let dispatcher = EventDispatcher::new(channel_tx.clone(), channel_rx);
        tokio::spawn(async move { dispatcher.run().await });
        Self { channel_tx }
    }
    async fn dispatch(&self, event: EventMessage) {
        self.channel_tx
            .send(EventDispatcherCommand::DispatchMessage(event))
            .await
            .unwrap();
    }
    async fn add_sink(&self, sink: Box<dyn EventMessageSink>) {
        self.channel_tx
            .send(EventDispatcherCommand::AddSink(sink))
            .await
            .unwrap()
    }
}

struct Bfdd {
    local: SocketAddr,
    peers: Vec<ResolvedPeer>,
    sessions_by_ip: HashMap<(IpAddr, IpAddr), BfdDiscriminator>,
    sessions_by_discr: HashMap<BfdDiscriminator, BfdSessionHandle>,
    // Sources we have already warned about unmatched packets from; cleared per source once a packet matches again
    unmatched_sources: std::sync::Mutex<HashSet<IpAddr>>,

    control_socket: Arc<UdpSocket>,

    event_dispatcher: EventDispatcherHandle,
}

impl Bfdd {
    async fn new(config: ResolvedConfig) -> Result<Self, Box<dyn Error>> {
        let local = match config.bind.as_slice() {
            [addr] => SocketAddr::new(*addr, CONTROL_PORT),
            addrs => {
                return Err(format!(
                    "{} bind addresses configured, but only one is currently supported",
                    addrs.len()
                )
                .into())
            }
        };
        let peers = config.peers;
        let sessions_by_ip = HashMap::new();
        let sessions_by_discr = HashMap::new();

        // TODO: Bind only the sockets we need for the configuration, including multihop and run them in different tasks.
        info!("Binding on {:?}", local);
        let control_socket = Arc::new(UdpSocket::bind(local).await?);
        // If BFD authentication is not in use on a session, all BFD Control packets for the session MUST be sent with a
        // Time to Live (TTL) or Hop Limit value of 255.
        set_ttl_or_hop_limit(&control_socket, 255)?;
        let _echo_socket = Arc::new(UdpSocket::bind(SocketAddr::new(local.ip(), ECHO_PORT)).await?);

        let event_dispatcher = EventDispatcherHandle::new().await;
        event_dispatcher
            .add_sink(Box::new(ScriptHookSink::with_script_dir(config.hooks_dir)))
            .await;

        Ok(Self {
            local,
            peers,
            sessions_by_ip,
            sessions_by_discr,
            unmatched_sources: std::sync::Mutex::new(HashSet::new()),
            control_socket,
            event_dispatcher,
        })
    }
    async fn start_peers(&mut self) -> io::Result<()> {
        let mut rng = rand::thread_rng();
        for peer in &self.peers {
            let mut local_discr = BfdDiscriminator(0);
            while local_discr.0 == 0 || self.sessions_by_discr.contains_key(&local_discr) {
                local_discr = BfdDiscriminator(rng.gen());
            }
            let session = BfdSessionHandle::new(
                self.local.ip(),
                *peer,
                local_discr,
                self.event_dispatcher.clone(),
            )
            .await?;

            self.sessions_by_ip
                .insert((self.local.ip(), peer.address), local_discr);
            self.sessions_by_discr.insert(local_discr, session);
        }
        Ok(())
    }
    fn session_for_packet(
        &self,
        addr: &SocketAddr,
        packet: &BfdPacket,
    ) -> Option<&BfdSessionHandle> {
        // Try to find a configured session for the remote IP if it doesn't send our discriminator, otherwise
        // use the discriminator that was sent.

        // https://datatracker.ietf.org/doc/html/rfc5880#section-6.3
        //
        // Once the remote end echoes back the local discriminator, all further received packets are
        // demultiplexed based on the Your Discriminator field only (which means that, among other things, the
        // source address field can change or the interface over which the packets are received can change, but
        // the packets will still be associated with the proper session).
        let our_disc = if packet.your_disc == BfdDiscriminator(0) {
            if let Some(discr) = self.sessions_by_ip.get(&(self.local.ip(), addr.ip())) {
                // A restarted peer starts over with Your Discriminator of zero. We deliberately keep the existing
                // session: the RFC 5880 section 6.8.6 state machine takes it Down (diag: Neighbor signaled session
                // down) and bfd.RemoteDiscr is relearned from the new My Discriminator.
                debug!(
                    "Found session for unknown discriminator from {}: {}",
                    addr.ip(),
                    discr
                );
                discr
            } else {
                self.log_unmatched(
                    addr.ip(),
                    log::Level::Warn,
                    "no session configured for this address",
                );
                return None;
            }
        } else {
            &packet.your_disc
        };
        let session = self.sessions_by_discr.get(our_disc);
        if session.is_some() {
            self.unmatched_sources.lock().unwrap().remove(&addr.ip());
        } else {
            self.log_unmatched(
                addr.ip(),
                log::Level::Info,
                &format!("unknown Your Discriminator {our_disc}"),
            );
        }
        session
    }
    /// Log the first unmatched packet from a source at `level`, then at debug level until one matches again.
    fn log_unmatched(&self, source: IpAddr, level: log::Level, reason: &str) {
        if self.unmatched_sources.lock().unwrap().insert(source) {
            log::log!(
                level,
                "Unable to match packet from {source}: {reason} (repeats are logged at debug level)"
            );
        } else {
            debug!("Unable to match packet from {source}: {reason}");
        }
    }
    async fn rx_packet(&self, addr: SocketAddr, packet: BfdPacket) {
        if let Some(session) = self.session_for_packet(&addr, &packet) {
            session
                .tx
                .send(SessionControlCommand::RxPacket(packet))
                .await
                .unwrap();
        }
    }
    async fn session_report(&self) -> String {
        future::join_all(
            self.sessions_by_discr
                .values()
                .map(|session| async { session.get_stats().await.unwrap().to_string() }),
        )
        .await
        .join("\n\t")
    }
    async fn handle_datagram(&self, data: &[u8], addr: SocketAddr) {
        let Ok((_leftover, packet)) = BfdPacket::parse(data) else {
            warn!("Failed to parse packet");
            return;
        };
        debug!("rx packet ({:?}): {:?}", addr.ip(), packet);
        // If the Length field is greater than the payload of the encapsulating protocol, the packet MUST be
        // discarded.
        if packet.flags.length() as usize > data.len() {
            warn!("discarding packet with length > payload");
            return;
        }
        self.rx_packet(addr, packet).await
    }
    async fn run(mut self) -> io::Result<()> {
        info!("rust-bfd starting up");
        self.start_peers().await?;
        info!("configured sessions: {}", self.session_report().await);

        let mut sigusr1 = signal(SignalKind::user_defined1())?;
        let mut buf = [0; 1024];
        loop {
            // TODO: All received BFD Control packets that are demultiplexed to the session MUST be discarded if
            // the received TTL or Hop Limit is not equal to 255. Need raw sockets for this.
            //
            // TODO: Maybe it is okay to receive in the session-specific socket instead of this global one? But
            // the RFC mentions that we must demux based on discriminator only.
            tokio::select! {
                result = self.control_socket.recv_from(&mut buf) => match result {
                    Ok((len, addr)) => {
                        self.handle_datagram(&buf[..len], addr).await;
                        if log_enabled!(log::Level::Debug) {
                            debug!("configured sessions: {}", self.session_report().await);
                        }
                    }
                    Err(e) => warn!("Error receiving from socket {e:?}"),
                },
                _ = sigusr1.recv() => {
                    info!("SIGUSR1 received");
                    info!("configured sessions: {}", self.session_report().await);
                }
            }
        }
    }
}

fn load_config(args: Args) -> Result<ResolvedConfig, ConfigError> {
    let mut config = Config::load(args.config.as_deref())?;
    config.apply_env(|name| std::env::var(name).ok())?;
    config.apply_cli(args.bind, args.peers);
    config.resolve()
}

#[tokio::main]
async fn main() -> io::Result<()> {
    let args = Args::parse();

    env_logger::Builder::from_env(Env::default().default_filter_or("info"))
        .format_timestamp_micros()
        .init();

    let config = load_config(args).map_err(|e| std::io::Error::other(e.to_string()))?;
    if config.peers.is_empty() {
        warn!("No peers configured");
    }

    match Bfdd::new(config).await {
        Ok(bfdd) => bfdd.run().await,
        Err(e) => Err(std::io::Error::other(e.to_string())),
    }
}

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

    #[test]
    fn jitter_is_uniform_over_rfc_range() {
        // RFC 5880 6.8.7: each interval is reduced by a random 0-25% (0-10% when DetectMult is 1), so delays are
        // spread over [floor, interval] with a mean at the midpoint, not clustered at either end.
        let interval = time::Duration::from_millis(300);
        const SAMPLES: u32 = 20_000;
        for (mult, floor_ms, mean_ms) in [(3u8, 225.0, 262.5), (1, 270.0, 285.0)] {
            let delays: Vec<f64> = (0..SAMPLES)
                .map(|_| jittered_delay(interval, mult).as_secs_f64() * 1000.0)
                .collect();
            let mean = delays.iter().sum::<f64>() / f64::from(SAMPLES);
            let min = delays.iter().copied().fold(f64::MAX, f64::min);
            let max = delays.iter().copied().fold(f64::MIN, f64::max);
            let width = 300.0 - floor_ms;
            assert!(
                (mean - mean_ms).abs() < 1.0,
                "mult {mult}: mean {mean}ms, expected {mean_ms}ms"
            );
            assert!(
                min < floor_ms + width * 0.02,
                "mult {mult}: never approached the floor (min {min}ms)"
            );
            assert!(
                max > 300.0 - width * 0.02,
                "mult {mult}: never approached the interval (max {max}ms)"
            );
        }
    }

    #[tokio::test]
    async fn periodic_sender_gaps_respect_rfc_floor() {
        // Gaps are measured between actual sends, so jitter must not be layered onto a shorter fixed clock.
        // Only the lower bound is asserted: scheduler overshoot can legitimately stretch a gap.
        let interval = time::Duration::from_millis(40);
        let floor = interval * 3 / 4;
        let (tx, mut rx) = mpsc::channel(32);
        let sender = BfdPeriodicSenderHandle::new(BfdSessionHandle { tx }, interval.into(), 3);

        let mut sends = Vec::new();
        while sends.len() < 40 {
            match rx.recv().await {
                Some(SessionControlCommand::TxControlPacket {
                    poll_response: false,
                }) => sends.push(time::Instant::now()),
                other => panic!("unexpected command (channel open: {})", other.is_some()),
            }
        }
        sender.tx.send(PeriodicControlCommand::Quit).await.unwrap();

        for pair in sends.windows(2) {
            let gap = pair[1] - pair[0];
            assert!(
                gap + time::Duration::from_millis(2) >= floor,
                "gap {gap:?} below {floor:?}"
            );
        }
    }

    #[test]
    fn jitter_within_rfc_bounds() {
        let interval = time::Duration::from_millis(300);
        for (mult, floor_ms) in [(3, 225), (1, 270)] {
            for _ in 0..1000 {
                let d = jittered_delay(interval, mult);
                assert!(
                    d >= time::Duration::from_millis(floor_ms),
                    "{d:?} below floor for mult {mult}"
                );
                assert!(d <= interval, "{d:?} above interval");
            }
        }
    }

    #[test]
    fn repeated_bind_and_peer_flags() {
        let args = Args::try_parse_from([
            "rust-bfdd",
            "-b",
            "192.0.2.1",
            "--bind",
            "2001:db8::1",
            "-p",
            "192.0.2.2",
            "--peer",
            "2001:db8::2",
        ])
        .unwrap();
        assert_eq!(args.bind.len(), 2);
        assert_eq!(args.peers.len(), 2);
    }

    use std::net::Ipv4Addr;

    const LOCAL: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1));
    const PEER: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 2));
    const LOCAL_DISCR: u32 = 1000;
    const REMOTE_DISCR: u32 = 2000;

    #[derive(Debug)]
    struct CaptureSink {
        tx: mpsc::Sender<EventMessage>,
        rx: mpsc::Receiver<EventMessage>,
        out: mpsc::UnboundedSender<StateChangeEvent>,
    }
    impl EventMessageSink for CaptureSink {
        fn channel(&self) -> mpsc::Sender<EventMessage> {
            self.tx.clone()
        }
        fn run(mut self: Box<Self>) -> JoinHandle<()> {
            tokio::spawn(async move {
                while let Some(EventMessage::StateChange(ev)) = self.rx.recv().await {
                    let _ = self.out.send(ev);
                }
            })
        }
    }

    async fn capture_dispatcher() -> (
        EventDispatcherHandle,
        mpsc::UnboundedReceiver<StateChangeEvent>,
    ) {
        let dispatcher = EventDispatcherHandle::new().await;
        let (tx, rx) = mpsc::channel(16);
        let (out, events) = mpsc::unbounded_channel();
        dispatcher
            .add_sink(Box::new(CaptureSink { tx, rx, out }))
            .await;
        (dispatcher, events)
    }

    fn peer_cfg(desired_ms: u32, required_ms: u32, detect_mult: u8) -> ResolvedPeer {
        ResolvedPeer {
            address: PEER,
            desired_min_tx: BfdInterval::from_millis(desired_ms),
            required_min_rx: BfdInterval::from_millis(required_ms),
            detect_mult,
        }
    }

    fn pkt(state: BfdState, my: u32, your: u32) -> BfdPacket {
        BfdPacket {
            flags: BfdFlags(0)
                .with_vers(1)
                .with_state(state as u8)
                .with_detect_mult(3)
                .with_length(24),
            my_disc: BfdDiscriminator(my),
            your_disc: BfdDiscriminator(your),
            desired_min_tx: BfdInterval::from_millis(300),
            required_min_rx: BfdInterval::from_millis(300),
            required_min_echo_rx: BfdInterval::from_micros(0),
            auth: None,
        }
    }

    fn with_flags(mut p: BfdPacket, f: impl FnOnce(BfdFlags) -> BfdFlags) -> BfdPacket {
        p.flags = f(p.flags);
        p
    }
    async fn session(
        peer: ResolvedPeer,
    ) -> (BfdSession, mpsc::UnboundedReceiver<StateChangeEvent>) {
        let (dispatcher, events) = capture_dispatcher().await;
        let (tx, rx) = mpsc::channel(32);
        let sock = UdpSocket::bind((LOCAL, 0)).await.unwrap();
        let s = BfdSession::new(
            rx,
            BfdSessionHandle { tx },
            sock,
            LOCAL,
            peer,
            BfdDiscriminator(LOCAL_DISCR),
            dispatcher,
        );
        (s, events)
    }

    async fn next_event(
        events: &mut mpsc::UnboundedReceiver<StateChangeEvent>,
    ) -> StateChangeEvent {
        time::timeout(Duration::from_secs(2), events.recv())
            .await
            .expect("timed out waiting for event")
            .expect("event channel closed")
    }

    async fn assert_no_event(events: &mut mpsc::UnboundedReceiver<StateChangeEvent>) {
        time::sleep(Duration::from_millis(50)).await;
        assert!(events.try_recv().is_err(), "unexpected state change event");
    }

    async fn session_in(
        state: BfdState,
    ) -> (BfdSession, mpsc::UnboundedReceiver<StateChangeEvent>) {
        let (mut s, mut events) = session(peer_cfg(300, 300, 3)).await;
        let path: &[BfdState] = match state {
            BfdState::Down => &[],
            BfdState::Init => &[BfdState::Down],
            _ => &[BfdState::Down, BfdState::Init],
        };
        for (i, remote) in path.iter().enumerate() {
            let your = if i == 0 { 0 } else { LOCAL_DISCR };
            s.receive_control_packet(&pkt(*remote, REMOTE_DISCR, your))
                .await;
            next_event(&mut events).await;
        }
        assert_eq!(s.state.session_state(), state);
        (s, events)
    }

    #[tokio::test]
    async fn new_session_starts_down_and_slow() {
        let (s, _events) = session(peer_cfg(300, 200, 3)).await;
        assert_eq!(s.state.session_state(), BfdState::Down);
        assert_eq!(s.state.remote_discr(), BfdDiscriminator(0));
        assert_eq!(s.state.local_diag(), BfdDiagnostic::None);
        assert_eq!(s.state.desired_min_tx_interval(), BfdInterval::from_secs(1));
        assert_eq!(
            s.state.required_min_rx_interval(),
            BfdInterval::from_millis(200)
        );
        assert_eq!(s.cur_interval, BfdInterval::from_secs(1));
        assert_eq!(s.state.detect_mult(), 3);
        assert!(!s.state.poll_mode());
    }

    #[tokio::test]
    async fn new_session_keeps_configured_interval_above_one_second() {
        let (s, _events) = session(peer_cfg(2500, 300, 3)).await;
        assert_eq!(
            s.state.desired_min_tx_interval(),
            BfdInterval::from_millis(2500)
        );
        assert_eq!(s.effective_desired_min_tx, BfdInterval::from_millis(2500));
    }

    #[tokio::test]
    async fn three_way_handshake_and_poll_sequence() {
        let (mut s, mut events) = session(peer_cfg(300, 300, 3)).await;

        s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, 0))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Init);
        assert_eq!(s.state.remote_discr(), BfdDiscriminator(REMOTE_DISCR));
        let ev = next_event(&mut events).await;
        assert_eq!(
            (ev.from_state, ev.to_state),
            (BfdState::Down, BfdState::Init)
        );
        assert_eq!(ev.local_discr, BfdDiscriminator(LOCAL_DISCR));
        assert_eq!(ev.remote_discr, BfdDiscriminator(REMOTE_DISCR));
        assert_eq!((ev.local_ip, ev.remote_ip), (LOCAL, PEER));

        s.receive_control_packet(&pkt(BfdState::Init, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Up);
        let ev = next_event(&mut events).await;
        assert_eq!((ev.from_state, ev.to_state), (BfdState::Init, BfdState::Up));
        assert!(s.state.last_up.load().is_some());
        assert_eq!(s.state.local_diag(), BfdDiagnostic::None);
        // Advertise the fast interval now, but keep transmitting slowly until the poll sequence completes
        assert_eq!(
            s.state.desired_min_tx_interval(),
            BfdInterval::from_millis(300)
        );
        assert!(s.state.poll_mode());
        assert_eq!(s.cur_interval, BfdInterval::from_secs(1));

        let fin = with_flags(pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR), |f| {
            f.with_final_(true)
        });
        s.receive_control_packet(&fin).await;
        assert!(!s.state.poll_mode());
        assert_eq!(s.effective_desired_min_tx, BfdInterval::from_millis(300));
        assert_eq!(s.cur_interval, BfdInterval::from_millis(300));
        assert_eq!(s.state.session_state(), BfdState::Up);
    }

    #[tokio::test]
    async fn up_without_poll_when_configured_interval_is_not_faster() {
        let (mut s, _events) = session(peer_cfg(2000, 300, 3)).await;
        s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, 0))
            .await;
        s.receive_control_packet(&pkt(BfdState::Init, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Up);
        assert!(!s.state.poll_mode());
        assert_eq!(s.cur_interval, BfdInterval::from_secs(2));
    }

    #[tokio::test]
    async fn down_goes_straight_to_up_when_remote_is_init() {
        let (mut s, mut events) = session(peer_cfg(300, 300, 3)).await;
        s.receive_control_packet(&pkt(BfdState::Init, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Up);
        let ev = next_event(&mut events).await;
        assert_eq!((ev.from_state, ev.to_state), (BfdState::Down, BfdState::Up));
    }

    #[tokio::test]
    async fn down_ignores_remote_up_and_admin_down() {
        let (mut s, mut events) = session(peer_cfg(300, 300, 3)).await;
        s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Down);
        s.receive_control_packet(&pkt(BfdState::AdminDown, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Down);
        assert_no_event(&mut events).await;
    }

    #[tokio::test]
    async fn init_transitions() {
        let (mut s, mut events) = session_in(BfdState::Init).await;
        s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Init);
        assert_no_event(&mut events).await;
        s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Up);

        let (mut s, mut events) = session_in(BfdState::Init).await;
        s.receive_control_packet(&pkt(BfdState::AdminDown, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Down);
        assert_eq!(s.state.local_diag(), BfdDiagnostic::NeighborDown);
        let ev = next_event(&mut events).await;
        assert_eq!(
            (ev.from_state, ev.to_state),
            (BfdState::Init, BfdState::Down)
        );
    }
    #[tokio::test]
    async fn up_goes_down_when_remote_goes_down() {
        let (mut s, mut events) = session_in(BfdState::Up).await;
        s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_no_event(&mut events).await;

        s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Down);
        assert_eq!(s.state.local_diag(), BfdDiagnostic::NeighborDown);
        let ev = next_event(&mut events).await;
        assert_eq!((ev.from_state, ev.to_state), (BfdState::Up, BfdState::Down));
        assert!(s.state.last_down.load().is_some());
        // Back to the slow rate and no poll sequence in flight
        assert_eq!(s.state.desired_min_tx_interval(), BfdInterval::from_secs(1));
        assert_eq!(s.cur_interval, BfdInterval::from_secs(1));
        assert!(!s.state.poll_mode());
    }

    #[tokio::test]
    async fn up_goes_down_on_remote_admin_down() {
        let (mut s, mut events) = session_in(BfdState::Up).await;
        s.receive_control_packet(&pkt(BfdState::AdminDown, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Down);
        assert_eq!(s.state.local_diag(), BfdDiagnostic::NeighborDown);
        assert_eq!(next_event(&mut events).await.to_state, BfdState::Down);
    }

    #[tokio::test]
    async fn restarted_peer_relearns_discriminator() {
        let (mut s, mut events) = session_in(BfdState::Up).await;
        s.receive_control_packet(&pkt(BfdState::Down, 3000, 0))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Down);
        assert_eq!(s.state.remote_discr(), BfdDiscriminator(3000));
        assert_eq!(next_event(&mut events).await.to_state, BfdState::Down);
        s.receive_control_packet(&pkt(BfdState::Down, 3000, 0))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Init);
    }

    #[tokio::test]
    async fn up_resets_stats_and_down_records_time() {
        let (mut s, _events) = session_in(BfdState::Init).await;
        s.state.record_tx();
        s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.session_state(), BfdState::Up);
        // The Up transition resets the counters, then sends one packet announcing the new state
        assert_eq!(s.state.rx_stats.snapshot().count(), 0);
        assert_eq!(s.state.tx_stats.snapshot().count(), 1);
        s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.rx_stats.snapshot().count(), 1);
    }

    #[tokio::test]
    async fn invalid_packets_are_discarded() {
        type Mutation = fn(BfdPacket) -> BfdPacket;
        let cases: [(&str, Mutation); 8] = [
            ("version", |p| with_flags(p, |f| f.with_vers(2))),
            ("short length", |p| with_flags(p, |f| f.with_length(23))),
            ("zero detect mult", |p| {
                with_flags(p, |f| f.with_detect_mult(0))
            }),
            ("multipoint", |p| with_flags(p, |f| f.with_multipoint(true))),
            ("zero my discr", |mut p| {
                p.my_disc = BfdDiscriminator(0);
                p
            }),
            ("wrong your discr", |mut p| {
                p.your_disc = BfdDiscriminator(LOCAL_DISCR + 1);
                p
            }),
            ("zero your discr, not down", |mut p| {
                p.your_disc = BfdDiscriminator(0);
                p.flags = p.flags.with_state(BfdState::Up as u8);
                p
            }),
            ("auth without auth in use", |p| {
                with_flags(p, |f| f.with_auth_present(true).with_length(26))
            }),
        ];
        for (name, mutate) in cases {
            let (mut s, mut events) = session(peer_cfg(300, 300, 3)).await;
            s.receive_control_packet(&mutate(pkt(BfdState::Down, REMOTE_DISCR, 0)))
                .await;
            assert_eq!(s.state.session_state(), BfdState::Down, "{name}");
            assert_eq!(s.state.remote_discr(), BfdDiscriminator(0), "{name}");
            assert_eq!(
                s.state.remote_min_rx_interval(),
                BfdInterval::from_millis(300),
                "{name}"
            );
            assert_no_event(&mut events).await;
        }
    }

    #[tokio::test]
    async fn auth_length_minimum_is_26() {
        let (mut s, _events) = session(peer_cfg(300, 300, 3)).await;
        let p = with_flags(pkt(BfdState::Down, REMOTE_DISCR, 0), |f| {
            f.with_auth_present(true).with_length(25)
        });
        s.receive_control_packet(&p).await;
        assert_eq!(s.state.remote_discr(), BfdDiscriminator(0));
    }

    #[tokio::test]
    async fn unknown_diagnostic_does_not_panic() {
        let (mut s, _events) = session(peer_cfg(300, 300, 3)).await;
        let p = with_flags(pkt(BfdState::Down, REMOTE_DISCR, 0), |f| f.with_diag(20));
        s.receive_control_packet(&p).await;
        assert_eq!(s.state.session_state(), BfdState::Init);
    }

    #[tokio::test]
    async fn remote_diag_is_recorded() {
        let (mut s, _events) = session(peer_cfg(300, 300, 3)).await;
        let p = with_flags(pkt(BfdState::Down, REMOTE_DISCR, 0), |f| {
            f.with_diag(BfdDiagnostic::PathDown as u8)
        });
        s.receive_control_packet(&p).await;
        assert_eq!(s.state.last_remote_diag(), BfdDiagnostic::PathDown);
    }

    #[tokio::test]
    async fn poll_bit_elicits_final_response() {
        let (mut s, _events) = session_in(BfdState::Up).await;
        let before = s.state.tx_stats.snapshot().count();
        s.receive_control_packet(&pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR))
            .await;
        assert_eq!(s.state.tx_stats.snapshot().count(), before);
        let poll = with_flags(pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR), |f| {
            f.with_poll(true)
        });
        s.receive_control_packet(&poll).await;
        assert_eq!(s.state.tx_stats.snapshot().count(), before + 1);
    }

    #[tokio::test]
    async fn state_change_sends_immediate_packet() {
        let (mut s, _events) = session(peer_cfg(300, 300, 3)).await;
        s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, 0))
            .await;
        assert_eq!(s.state.tx_stats.snapshot().count(), 1);
        s.receive_control_packet(&pkt(BfdState::Down, REMOTE_DISCR, 0))
            .await;
        assert_eq!(s.state.tx_stats.snapshot().count(), 1);
    }

    #[tokio::test]
    async fn transmit_interval_follows_remote_min_rx() {
        let (mut s, _events) = session(peer_cfg(300, 300, 3)).await;
        let mut p = pkt(BfdState::Down, REMOTE_DISCR, 0);
        p.required_min_rx = BfdInterval::from_secs(2);
        s.receive_control_packet(&p).await;
        assert_eq!(s.state.remote_min_rx_interval(), BfdInterval::from_secs(2));
        assert_eq!(s.cur_interval, BfdInterval::from_secs(2));
        p.required_min_rx = BfdInterval::from_millis(100);
        s.receive_control_packet(&p).await;
        // never faster than our own effective desired-min-tx, which is 1s while not Up
        assert_eq!(s.cur_interval, BfdInterval::from_secs(1));
    }

    #[tokio::test]
    async fn detection_time_saturates_on_remote_overflow() {
        let (mut s, _events) = session(peer_cfg(300, 300, 3)).await;
        let mut p = pkt(BfdState::Down, REMOTE_DISCR, 0);
        p.flags = p.flags.with_detect_mult(255);
        p.desired_min_tx = BfdInterval::from_micros(u32::MAX);
        s.receive_control_packet(&p).await;
        assert_eq!(s.state.detection_time(), BfdInterval::from_micros(u32::MAX));
    }

    #[tokio::test]
    async fn detection_time_uses_remote_mult_and_larger_interval() {
        let (mut s, _events) = session(peer_cfg(300, 500, 3)).await;
        let mut p = pkt(BfdState::Down, REMOTE_DISCR, 0);
        p.flags = p.flags.with_detect_mult(5);
        p.desired_min_tx = BfdInterval::from_millis(400);
        s.receive_control_packet(&p).await;
        assert_eq!(s.state.detection_time(), BfdInterval::from_millis(2500));
        p.desired_min_tx = BfdInterval::from_millis(800);
        s.receive_control_packet(&p).await;
        assert_eq!(s.state.detection_time(), BfdInterval::from_millis(4000));
    }
    #[tokio::test]
    async fn watchdog_expires_only_without_packets() {
        let (dispatcher, mut events) = capture_dispatcher().await;
        let handle = BfdSessionHandle::new(
            LOCAL,
            peer_cfg(20, 20, 3),
            BfdDiscriminator(LOCAL_DISCR),
            dispatcher,
        )
        .await
        .unwrap();
        let fast = |state, your| {
            let mut p = pkt(state, REMOTE_DISCR, your);
            p.flags = p.flags.with_detect_mult(10);
            p.desired_min_tx = BfdInterval::from_millis(20);
            p.required_min_rx = BfdInterval::from_millis(20);
            p
        };
        let send = |p| handle.tx.send(SessionControlCommand::RxPacket(p));
        send(fast(BfdState::Down, 0)).await.unwrap();
        send(fast(BfdState::Init, LOCAL_DISCR)).await.unwrap();
        assert_eq!(next_event(&mut events).await.to_state, BfdState::Init);
        assert_eq!(next_event(&mut events).await.to_state, BfdState::Up);

        // detection time is 10 x 20ms; keep-alives at 20ms must hold the session up
        for _ in 0..15 {
            send(fast(BfdState::Up, LOCAL_DISCR)).await.unwrap();
            time::sleep(Duration::from_millis(20)).await;
        }
        assert!(
            events.try_recv().is_err(),
            "session dropped despite keep-alives"
        );

        let ev = next_event(&mut events).await;
        assert_eq!((ev.from_state, ev.to_state), (BfdState::Up, BfdState::Down));
        let stats = handle.get_stats().await.unwrap();
        assert_eq!(stats.state, BfdState::Down);
        assert!(stats.last_down.is_some());
        handle.tx.send(SessionControlCommand::Quit).await.unwrap();
    }

    #[test]
    fn bfd_error_display_orders_value_and_field() {
        let e = BfdError::InvalidFieldValue("state", "bogus");
        assert_eq!(e.to_string(), "invalid value `bogus` for field `state`");
    }

    #[tokio::test]
    async fn control_socket_uses_source_port_range_and_ttl_255() {
        let a = bind_control_socket(LOCAL).await.unwrap();
        let b = bind_control_socket(LOCAL).await.unwrap();
        for s in [&a, &b] {
            let addr = s.local_addr().unwrap();
            assert!(SOURCE_PORT_RANGE.contains(&addr.port()), "{addr}");
            assert_eq!(addr.ip(), LOCAL);
            assert_eq!(s.ttl().unwrap(), 255);
        }
        assert_ne!(
            a.local_addr().unwrap().port(),
            b.local_addr().unwrap().port()
        );
    }

    #[tokio::test]
    async fn control_socket_bind_error_is_propagated() {
        let err = bind_control_socket(IpAddr::V4(Ipv4Addr::new(203, 0, 113, 1)))
            .await
            .unwrap_err();
        assert_ne!(err.kind(), ErrorKind::AddrInUse);
    }

    const OTHER: IpAddr = IpAddr::V4(Ipv4Addr::new(127, 0, 0, 3));

    async fn bfdd() -> (Bfdd, mpsc::Receiver<SessionControlCommand>) {
        let (tx, rx) = mpsc::channel(8);
        let discr = BfdDiscriminator(LOCAL_DISCR);
        let d = Bfdd {
            local: SocketAddr::new(LOCAL, CONTROL_PORT),
            peers: vec![peer_cfg(300, 300, 3)],
            sessions_by_ip: HashMap::from([((LOCAL, PEER), discr)]),
            sessions_by_discr: HashMap::from([(discr, BfdSessionHandle { tx })]),
            unmatched_sources: std::sync::Mutex::new(HashSet::new()),
            control_socket: Arc::new(UdpSocket::bind((LOCAL, 0)).await.unwrap()),
            event_dispatcher: EventDispatcherHandle::new().await,
        };
        (d, rx)
    }

    fn from(ip: IpAddr) -> SocketAddr {
        SocketAddr::new(ip, 49152)
    }

    #[tokio::test]
    async fn demux_by_source_when_your_discr_is_zero() {
        let (d, _rx) = bfdd().await;
        let p = pkt(BfdState::Down, REMOTE_DISCR, 0);
        assert!(d.session_for_packet(&from(PEER), &p).is_some());
        assert!(d.session_for_packet(&from(OTHER), &p).is_none());
    }

    #[tokio::test]
    async fn demux_by_discriminator_ignores_source_address() {
        let (d, _rx) = bfdd().await;
        let p = pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR);
        assert!(d.session_for_packet(&from(PEER), &p).is_some());
        assert!(d.session_for_packet(&from(OTHER), &p).is_some());
        let p = pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR + 1);
        assert!(d.session_for_packet(&from(PEER), &p).is_none());
    }

    #[tokio::test]
    async fn unmatched_sources_are_tracked_until_they_match() {
        let (d, _rx) = bfdd().await;
        let miss = pkt(BfdState::Down, REMOTE_DISCR, 0);
        let hit = pkt(BfdState::Up, REMOTE_DISCR, LOCAL_DISCR);
        assert!(d.session_for_packet(&from(OTHER), &miss).is_none());
        assert!(d.unmatched_sources.lock().unwrap().contains(&OTHER));
        assert!(d.session_for_packet(&from(OTHER), &hit).is_some());
        assert!(!d.unmatched_sources.lock().unwrap().contains(&OTHER));
    }

    #[tokio::test]
    async fn datagram_is_parsed_and_delivered_to_session() {
        let (d, mut rx) = bfdd().await;
        let bytes = pkt(BfdState::Down, REMOTE_DISCR, 0).serialize().unwrap();
        d.handle_datagram(&bytes, from(PEER)).await;
        match rx.try_recv() {
            Ok(SessionControlCommand::RxPacket(p)) => {
                assert_eq!(p.my_disc, BfdDiscriminator(REMOTE_DISCR))
            }
            _ => panic!("expected RxPacket"),
        }
    }

    #[tokio::test]
    async fn bad_datagrams_are_dropped() {
        let (d, mut rx) = bfdd().await;
        let good = pkt(BfdState::Down, REMOTE_DISCR, 0).serialize().unwrap();
        d.handle_datagram(&good[..23], from(PEER)).await;
        d.handle_datagram(&[], from(PEER)).await;
        d.handle_datagram(&[0xff; 7], from(PEER)).await;
        let long = with_flags(pkt(BfdState::Down, REMOTE_DISCR, 0), |f| f.with_length(48));
        d.handle_datagram(&long.serialize().unwrap(), from(PEER))
            .await;
        let unknown = pkt(BfdState::Down, REMOTE_DISCR, 0);
        d.handle_datagram(&unknown.serialize().unwrap(), from(OTHER))
            .await;
        assert!(rx.try_recv().is_err());
    }

    #[tokio::test]
    async fn bfdd_rejects_multiple_bind_addresses() {
        let cfg = ResolvedConfig {
            bind: vec![LOCAL, PEER],
            hooks_dir: PathBuf::from("/nonexistent"),
            peers: vec![],
        };
        let err = Bfdd::new(cfg).await.err().expect("should fail");
        assert!(err.to_string().contains("only one"), "{err}");
    }
}