asic-rs 0.8.1

use std::{
    any::Any,
    collections::{HashMap, HashSet},
    future::Future,
    net::{IpAddr, Ipv4Addr, SocketAddr},
    panic::AssertUnwindSafe,
    pin::Pin,
    str::FromStr,
    sync::Arc,
    time::Duration,
};

use anyhow::Result;
use asic_rs_core::{
    data::command::MinerCommand,
    traits::{
        entry::FirmwareEntry,
        identification::WebResponse,
        miner::{Miner, MinerAuth},
    },
    util::{send_rpc_command, send_web_command},
};
use futures::{
    Stream, StreamExt,
    future::FutureExt,
    pin_mut,
    stream::{self, FuturesUnordered},
};
use ipnet::IpNet;
use rand::seq::SliceRandom;
use tokio::{net::TcpStream, sync::Semaphore, time::timeout};

const IDENTIFICATION_TIMEOUT: Duration = Duration::from_secs(10);
const CONNECTIVITY_TIMEOUT: Duration = Duration::from_secs(1);
const CONNECTIVITY_RETRIES: u32 = 3;
const CONNECTIVITY_RETRY_BACKOFF: Duration = Duration::from_millis(100);
const MAX_CONNECTIVITY_RETRY_BACKOFF: Duration = Duration::from_secs(2);
const MINER_PORTS: [u16; 4] = [80, 4028, 4029, 8889];
const NOFILE_PER_CONCURRENCY: u64 = 8;
const MIN_NOFILE_LIMIT: u64 = 2048;

fn calculate_optimal_concurrency(ip_count: usize) -> usize {
    match ip_count {
        0..=1000 => 1000,
        1001..=5000 => 2500,
        5001..=10000 => 5000,
        _ => 10000,
    }
}

fn calculate_desired_nofile_limit(concurrency: usize) -> u64 {
    (concurrency as u64)
        .saturating_mul(NOFILE_PER_CONCURRENCY)
        .max(MIN_NOFILE_LIMIT)
}

async fn check_port_open(ip: IpAddr, port: u16, connectivity_timeout: Duration) -> bool {
    let addr: SocketAddr = (ip, port).into();
    let stream = match timeout(connectivity_timeout, TcpStream::connect(addr)).await {
        Ok(Ok(stream)) => stream,
        _ => return false,
    };
    let _ = stream.set_nodelay(true);
    true
}

async fn with_connectivity_permit<Fut>(permits: Arc<Semaphore>, probe: Fut) -> bool
where
    Fut: Future<Output = bool>,
{
    let Ok(_permit) = permits.acquire_owned().await else {
        return false;
    };
    probe.await
}

async fn race_miner_ports<F, Fut>(mut probe: F) -> bool
where
    F: FnMut(u16) -> Fut,
    Fut: Future<Output = bool>,
{
    let mut probes = FuturesUnordered::new();
    for port in MINER_PORTS {
        probes.push(probe(port));
    }
    while let Some(open) = probes.next().await {
        if open {
            return true;
        }
    }
    false
}

async fn check_miner_ports(
    ip: IpAddr,
    connectivity_timeout: Duration,
    permits: Arc<Semaphore>,
) -> bool {
    race_miner_ports(|port| {
        with_connectivity_permit(
            Arc::clone(&permits),
            check_port_open(ip, port, connectivity_timeout),
        )
    })
    .await
}

fn connectivity_retry_delay(retry_index: u32, initial_backoff: Duration) -> Duration {
    let multiplier = 1_u32 << retry_index.min(31);
    initial_backoff
        .saturating_mul(multiplier)
        .min(MAX_CONNECTIVITY_RETRY_BACKOFF)
}

async fn retry_connectivity<F, Fut>(retries: u32, initial_backoff: Duration, mut probe: F) -> bool
where
    F: FnMut() -> Fut,
    Fut: Future<Output = bool>,
{
    for attempt in 0..=retries {
        if attempt > 0 {
            tokio::time::sleep(connectivity_retry_delay(attempt - 1, initial_backoff)).await;
        }
        if probe().await {
            return true;
        }
    }
    false
}

async fn get_miner_type_from_command(
    ip: IpAddr,
    command: MinerCommand,
    registry: Arc<[Arc<dyn FirmwareEntry>]>,
) -> Option<Arc<dyn FirmwareEntry>> {
    match command {
        MinerCommand::RPC { command, .. } => {
            let response = send_rpc_command(&ip, command).await?;
            let upper = response.to_string().to_uppercase();
            registry.iter().find(|fw| fw.identify_rpc(&upper)).cloned()
        }
        MinerCommand::WebAPI { command, .. } => {
            let (body, headers, status) = send_web_command(&ip, command).await?;
            let auth_header = headers
                .get("www-authenticate")
                .and_then(|h| h.to_str().ok())
                .unwrap_or("");
            let algo_header = headers
                .get("algorithm")
                .and_then(|h| h.to_str().ok())
                .unwrap_or("");
            let redirect_header = headers
                .get("location")
                .and_then(|h| h.to_str().ok())
                .unwrap_or("");
            let web_resp = WebResponse {
                body: &body,
                auth_header,
                algo_header,
                redirect_header,
                status: status.as_u16(),
            };
            registry
                .iter()
                .find(|fw| fw.identify_web(&web_resp))
                .cloned()
        }
        _ => None,
    }
}

fn panic_message(panic_info: &(dyn Any + Send)) -> &str {
    if let Some(message) = panic_info.downcast_ref::<&str>() {
        message
    } else if let Some(message) = panic_info.downcast_ref::<String>() {
        message.as_str()
    } else {
        "unknown panic"
    }
}

async fn get_miner_type_from_command_catch_unwind(
    ip: IpAddr,
    command: MinerCommand,
    registry: Arc<[Arc<dyn FirmwareEntry>]>,
) -> Option<Arc<dyn FirmwareEntry>> {
    match AssertUnwindSafe(get_miner_type_from_command(ip, command, registry))
        .catch_unwind()
        .await
    {
        Ok(result) => result,
        Err(panic_info) => {
            tracing::warn!(
                "discovery command panicked for {ip}: {}",
                panic_message(&*panic_info)
            );
            None
        }
    }
}

/// Build the default firmware registry, gated by feature flags.
///
/// Non-stock firmwares are listed first so they take priority over stock
/// when multiple responses are received for the same device.
#[allow(clippy::vec_init_then_push)]
pub fn default_firmware_registry() -> Vec<Arc<dyn FirmwareEntry>> {
    let mut registry: Vec<Arc<dyn FirmwareEntry>> = vec![];

    #[cfg(feature = "braiins")]
    registry.push(Arc::new(
        asic_rs_firmwares_braiins::firmware::BraiinsFirmware::default(),
    ));

    #[cfg(feature = "luxminer")]
    registry.push(Arc::new(
        asic_rs_firmwares_luxminer::firmware::LuxMinerFirmware::default(),
    ));

    #[cfg(feature = "marathon")]
    registry.push(Arc::new(
        asic_rs_firmwares_marathon::firmware::MarathonFirmware::default(),
    ));

    #[cfg(feature = "vnish")]
    registry.push(Arc::new(
        asic_rs_firmwares_vnish::firmware::VnishFirmware::default(),
    ));

    #[cfg(feature = "volcminer")]
    registry.push(Arc::new(
        asic_rs_firmwares_volcminer::firmware::VolcMinerStockFirmware::default(),
    ));

    #[cfg(feature = "elphapex")]
    registry.push(Arc::new(
        asic_rs_firmwares_elphapex::firmware::ElphapexStockFirmware::default(),
    ));

    #[cfg(feature = "epic")]
    registry.push(Arc::new(
        asic_rs_firmwares_epic::firmware::EPicFirmware::default(),
    ));

    // Stock firmwares — checked last so non-stock take priority
    #[cfg(feature = "futurebit")]
    registry.push(Arc::new(
        asic_rs_firmwares_futurebit::firmware::ApolloFirmware::default(),
    ));

    #[cfg(feature = "whatsminer")]
    registry.push(Arc::new(
        asic_rs_firmwares_whatsminer::firmware::WhatsMinerFirmware::default(),
    ));

    #[cfg(feature = "antminer")]
    registry.push(Arc::new(
        asic_rs_firmwares_antminer::firmware::AntMinerStockFirmware::default(),
    ));

    #[cfg(feature = "sealminer")]
    registry.push(Arc::new(
        asic_rs_firmwares_sealminer::firmware::SealMinerStockFirmware::default(),
    ));

    #[cfg(feature = "avalonminer")]
    registry.push(Arc::new(
        asic_rs_firmwares_avalonminer::firmware::AvalonStockFirmware::default(),
    ));

    #[cfg(feature = "auradine")]
    registry.push(Arc::new(
        asic_rs_firmwares_auradine::firmware::AuradineFirmware::default(),
    ));

    // NerdAxe before Bitaxe — both check web root but NerdAxe is more specific
    #[cfg(feature = "nerdaxe")]
    registry.push(Arc::new(
        asic_rs_firmwares_nerdaxe::firmware::NerdAxeFirmware::default(),
    ));

    #[cfg(feature = "proto")]
    registry.push(Arc::new(
        asic_rs_firmwares_proto::firmware::ProtoFirmware::default(),
    ));

    #[cfg(feature = "bitaxe")]
    registry.push(Arc::new(
        asic_rs_firmwares_bitaxe::firmware::BitaxeFirmware::default(),
    ));

    registry
}

#[derive(Clone)]
/// Discovers ASIC miners and constructs firmware-specific miner handles.
///
/// A factory owns the IP addresses to scan, the firmware registry used for
/// identification, and discovery tuning such as timeouts and concurrency.
/// Constructors like [`Self::from_subnet`], [`Self::from_octets`], and
/// [`Self::from_range`] create a factory with an initial search range. The
/// matching `with_*` methods append additional addresses and return the updated
/// factory for chaining.
pub struct MinerFactory {
    search_firmwares: Option<Vec<Arc<dyn FirmwareEntry>>>,
    ips: Vec<IpAddr>,
    discovery_auth_by_firmware: HashMap<String, MinerAuth>,
    identification_timeout: Duration,
    connectivity_timeout: Duration,
    connectivity_retries: u32,
    concurrent: Option<usize>,
    nofile_limit: Option<u64>,
    nofile_adjustment: bool,
    check_port: bool,
}

impl std::fmt::Debug for MinerFactory {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("MinerFactory")
            .field("ips", &self.ips.len())
            .field(
                "search_firmwares",
                &self.search_firmwares.as_ref().map(|v| v.len()),
            )
            .field(
                "discovery_auth_by_firmware",
                &self.discovery_auth_by_firmware.len(),
            )
            .field("identification_timeout", &self.identification_timeout)
            .field("connectivity_timeout", &self.connectivity_timeout)
            .field("connectivity_retries", &self.connectivity_retries)
            .field("concurrent", &self.concurrent)
            .field("nofile_limit", &self.nofile_limit)
            .field("nofile_adjustment", &self.nofile_adjustment)
            .field("check_port", &self.check_port)
            .finish()
    }
}

impl Default for MinerFactory {
    fn default() -> Self {
        Self::new()
    }
}

impl MinerFactory {
    #[tracing::instrument(level = "debug", skip(self))]
    pub async fn scan_miner(&self, ip: IpAddr) -> Result<Option<Box<dyn Miner>>> {
        let connection_limit = self
            .concurrent
            .unwrap_or(calculate_optimal_concurrency(self.ips.len().max(1)));
        self.scan_miner_with_connection_limit(ip, Arc::new(Semaphore::new(connection_limit.max(1))))
            .await
    }

    async fn scan_miner_with_connection_limit(
        &self,
        ip: IpAddr,
        connection_limit: Arc<Semaphore>,
    ) -> Result<Option<Box<dyn Miner>>> {
        if !self.check_port {
            return self.get_miner(ip).await;
        }
        if retry_connectivity(
            self.connectivity_retries,
            CONNECTIVITY_RETRY_BACKOFF,
            || check_miner_ports(ip, self.connectivity_timeout, Arc::clone(&connection_limit)),
        )
        .await
        {
            return self.get_miner(ip).await;
        }
        tracing::trace!("no response from any miner-specific ports");
        Ok(None)
    }

    /// Discover and construct a miner at the given IP.
    ///
    /// Uses backend default credentials during discovery/build unless
    /// overridden via [`Self::with_firmware_discovery_auth`].
    #[tracing::instrument(level = "debug", skip(self))]
    pub async fn get_miner(&self, ip: IpAddr) -> Result<Option<Box<dyn Miner>>> {
        let discovery = AssertUnwindSafe(self.get_miner_inner(ip)).catch_unwind();
        match timeout(self.identification_timeout, discovery).await {
            Ok(Ok(result)) => result,
            Ok(Err(panic_info)) => {
                let msg = panic_message(&*panic_info);
                tracing::error!("panic during miner discovery for {ip}: {msg}");
                Err(anyhow::anyhow!(
                    "internal panic during miner discovery: {msg}"
                ))
            }
            Err(_) => {
                tracing::debug!(
                    timeout_ms = self.identification_timeout.as_millis(),
                    "miner discovery timed out for {ip}"
                );
                Ok(None)
            }
        }
    }

    async fn get_miner_inner(&self, ip: IpAddr) -> Result<Option<Box<dyn Miner>>> {
        let registry: Arc<[Arc<dyn FirmwareEntry>]> = Arc::from(
            self.search_firmwares
                .clone()
                .unwrap_or_else(default_firmware_registry)
                .as_slice(),
        );

        let found = {
            let mut commands: HashSet<MinerCommand> = HashSet::new();
            for fw in registry.iter() {
                for cmd in fw.get_discovery_commands() {
                    commands.insert(cmd);
                }
            }

            let mut discovery_tasks = FuturesUnordered::new();
            for command in commands {
                let reg = registry.clone();
                discovery_tasks.push(get_miner_type_from_command_catch_unwind(ip, command, reg));
            }

            let mut found: Option<Arc<dyn FirmwareEntry>> = None;

            loop {
                if discovery_tasks.is_empty() {
                    break;
                }
                if let Some(Some(fw)) = discovery_tasks.next().await {
                    found = Some(fw);
                    break;
                }
            }

            // If we found a stock firmware, wait a short window for non-stock to respond
            if found.as_ref().map(|f| f.is_stock()).unwrap_or(false) {
                let upgrade_window = tokio::time::sleep(Duration::from_millis(300)).fuse();
                pin_mut!(upgrade_window);

                loop {
                    if discovery_tasks.is_empty() {
                        break;
                    }
                    tokio::select! {
                        _ = &mut upgrade_window => break,
                        r = discovery_tasks.next() => {
                            if let Some(Some(fw)) = r
                                && !fw.is_stock()
                            {
                                found = Some(fw);
                                break;
                            }
                        }
                    }
                }
            }

            found
        };

        match found {
            Some(fw) => {
                let auth = self.discovery_auth_by_firmware.get(&fw.to_string());
                match fw.build_miner(ip, auth).await {
                    Ok(miner) => Ok(Some(miner)),
                    Err(e) => {
                        tracing::debug!("failed to build miner for {ip}: {e}");
                        Ok(None)
                    }
                }
            }
            None => {
                tracing::debug!("failed to identify {ip}");
                Ok(None)
            }
        }
    }

    /// Create an empty factory.
    ///
    /// Use one of the `with_*` range methods before calling [`Self::scan`], or
    /// call [`Self::get_miner`] directly when a single IP address is known.
    pub fn new() -> MinerFactory {
        MinerFactory {
            search_firmwares: None,
            ips: Vec::new(),
            discovery_auth_by_firmware: HashMap::new(),
            identification_timeout: IDENTIFICATION_TIMEOUT,
            connectivity_timeout: CONNECTIVITY_TIMEOUT,
            connectivity_retries: CONNECTIVITY_RETRIES,
            concurrent: None,
            nofile_limit: None,
            nofile_adjustment: true,
            check_port: true,
        }
    }

    /// Enable or disable the quick TCP port check before miner identification.
    ///
    /// Port checking reduces wasted identification attempts during scans by
    /// probing common miner ports first. Disable it when a network filters TCP
    /// probes but still responds to the firmware-specific discovery requests.
    pub fn with_port_check(mut self, enabled: bool) -> Self {
        self.check_port = enabled;
        self
    }

    /// Set credentials for a specific firmware entry to use during
    /// miner construction after identification.
    pub fn with_firmware_discovery_auth(
        mut self,
        firmware: &dyn FirmwareEntry,
        auth: MinerAuth,
    ) -> Self {
        self.discovery_auth_by_firmware
            .insert(firmware.to_string(), auth);
        self
    }

    /// Set the maximum number of addresses scanned at the same time.
    ///
    /// This also caps active TCP connectivity probes across the scan. If unset,
    /// scan concurrency is chosen from the number of queued hosts.
    pub fn with_concurrent_limit(mut self, limit: usize) -> Self {
        self.concurrent = Some(limit);
        self
    }

    /// Set the desired process file descriptor limit before large scans.
    ///
    /// This is best-effort. If the operating system rejects the requested
    /// value, scanning continues with the existing limit.
    pub fn with_nofile_limit(mut self, limit: u64) -> Self {
        self.nofile_limit = Some(limit);
        self
    }

    /// Enable or disable automatic file descriptor limit adjustment.
    ///
    /// Automatic adjustment is enabled by default and is only attempted before
    /// scans. The operation is fail-open.
    pub fn with_nofile_adjustment(mut self, enabled: bool) -> Self {
        self.nofile_adjustment = enabled;
        self
    }

    /// Choose scan concurrency from the number of queued hosts.
    ///
    /// This is normally unnecessary because [`Self::scan`] and streaming scans
    /// already use adaptive concurrency when no explicit limit is set.
    pub fn with_adaptive_concurrency(mut self) -> Self {
        self.concurrent = Some(calculate_optimal_concurrency(self.ips.len()));
        self
    }

    /// Populate the concurrency limit if it has not already been set.
    pub fn update_adaptive_concurrency(&mut self) {
        if self.concurrent.is_none() {
            self.concurrent = Some(calculate_optimal_concurrency(self.ips.len()));
        }
    }

    /// Set the maximum time spent identifying and constructing a miner.
    ///
    /// The deadline covers all discovery commands and firmware-specific miner
    /// construction after connectivity has been established.
    pub fn with_identification_timeout(mut self, timeout: Duration) -> Self {
        self.identification_timeout = timeout;
        self
    }

    /// Set the end-to-end identification and construction timeout in seconds.
    pub fn with_identification_timeout_secs(mut self, timeout_secs: u64) -> Self {
        self.identification_timeout = Duration::from_secs(timeout_secs);
        self
    }

    /// Set the timeout for quick connectivity probes during scans.
    pub fn with_connectivity_timeout(mut self, timeout: Duration) -> Self {
        self.connectivity_timeout = timeout;
        self
    }

    /// Set the connectivity probe timeout in seconds.
    pub fn with_connectivity_timeout_secs(mut self, timeout_secs: u64) -> Self {
        self.connectivity_timeout = Duration::from_secs(timeout_secs);
        self
    }

    /// Set the number of connectivity retries after the initial attempt.
    ///
    /// Each address is probed at least once. Retries use bounded exponential
    /// backoff and remain inside the initial scan's concurrency bound.
    pub fn with_connectivity_retries(mut self, retries: u32) -> Self {
        self.connectivity_retries = retries;
        self
    }

    /// Override the firmware registry with a custom list.
    pub fn with_firmwares(mut self, firmwares: Vec<Arc<dyn FirmwareEntry>>) -> Self {
        self.search_firmwares = Some(firmwares);
        self
    }

    /// Create a factory populated with all addresses from a CIDR subnet.
    ///
    /// Both IPv4 and IPv6 CIDR strings are supported.
    pub fn from_subnet(subnet: &str) -> Result<Self> {
        Self::new().with_subnet(subnet)
    }

    /// Append all addresses from a CIDR subnet to this factory.
    pub fn with_subnet(mut self, subnet: &str) -> Result<Self> {
        let ips = self.hosts_from_subnet(subnet)?;
        self.ips.extend(ips);
        self.shuffle_ips();
        Ok(self)
    }

    /// Replace this factory's queued addresses with all addresses from a CIDR subnet.
    pub fn set_subnet(&mut self, subnet: &str) -> Result<&Self> {
        let ips = self.hosts_from_subnet(subnet)?;
        self.ips = ips;
        self.shuffle_ips();
        Ok(self)
    }

    fn hosts_from_subnet(&self, subnet: &str) -> Result<Vec<IpAddr>> {
        let network = IpNet::from_str(subnet)?;
        let hosts = match network {
            IpNet::V4(network_v4) => {
                let start = u32::from(network_v4.network());
                let end = u32::from(network_v4.broadcast());

                (start..=end)
                    .map(Ipv4Addr::from)
                    .map(IpAddr::V4)
                    .collect::<Vec<IpAddr>>()
            }
            IpNet::V6(network_v6) => network_v6.hosts().map(IpAddr::V6).collect(),
        };

        Ok(hosts)
    }

    fn shuffle_ips(&mut self) {
        let mut rng = rand::rng();
        self.ips.shuffle(&mut rng);
    }

    /// Create a factory from four IPv4 octet selectors.
    ///
    /// Each octet may be a single value such as `"192"` or an inclusive range
    /// such as `"1-254"`.
    pub fn from_octets(octet1: &str, octet2: &str, octet3: &str, octet4: &str) -> Result<Self> {
        Self::new().with_octets(octet1, octet2, octet3, octet4)
    }

    /// Append addresses generated from four IPv4 octet selectors.
    pub fn with_octets(
        mut self,
        octet1: &str,
        octet2: &str,
        octet3: &str,
        octet4: &str,
    ) -> Result<Self> {
        let ips = self.hosts_from_octets(octet1, octet2, octet3, octet4)?;
        self.ips.extend(ips);
        self.shuffle_ips();
        Ok(self)
    }

    /// Replace this factory's queued addresses with four IPv4 octet selectors.
    pub fn set_octets(
        &mut self,
        octet1: &str,
        octet2: &str,
        octet3: &str,
        octet4: &str,
    ) -> Result<&Self> {
        let ips = self.hosts_from_octets(octet1, octet2, octet3, octet4)?;
        self.ips = ips;
        self.shuffle_ips();
        Ok(self)
    }

    fn hosts_from_octets(
        &self,
        octet1: &str,
        octet2: &str,
        octet3: &str,
        octet4: &str,
    ) -> Result<Vec<IpAddr>> {
        let octet1_range = parse_octet_range(octet1)?;
        let octet2_range = parse_octet_range(octet2)?;
        let octet3_range = parse_octet_range(octet3)?;
        let octet4_range = parse_octet_range(octet4)?;

        Ok(generate_ips_from_ranges(
            &octet1_range,
            &octet2_range,
            &octet3_range,
            &octet4_range,
        ))
    }

    /// Create a factory from an IPv4 range string.
    ///
    /// Range strings use dotted octets where any octet may be a single value or
    /// an inclusive range, for example `"192.168.1.1-254"`.
    pub fn from_range(range_str: &str) -> Result<Self> {
        Self::new().with_range(range_str)
    }

    /// Append addresses generated from an IPv4 range string.
    pub fn with_range(mut self, range_str: &str) -> Result<Self> {
        let ips = self.hosts_from_range(range_str)?;
        self.ips.extend(ips);
        self.shuffle_ips();
        Ok(self)
    }

    /// Replace this factory's queued addresses with an IPv4 range string.
    pub fn set_range(&mut self, range_str: &str) -> Result<&Self> {
        let ips = self.hosts_from_range(range_str)?;
        self.ips = ips;
        self.shuffle_ips();
        Ok(self)
    }

    fn hosts_from_range(&self, range_str: &str) -> Result<Vec<IpAddr>> {
        let parts: Vec<&str> = range_str.split('.').collect();
        if parts.len() != 4 {
            return Err(anyhow::anyhow!(
                "Invalid IP range format. Expected format: 10.1-199.0.1-199"
            ));
        }

        let octet1_range = parse_octet_range(parts[0])?;
        let octet2_range = parse_octet_range(parts[1])?;
        let octet3_range = parse_octet_range(parts[2])?;
        let octet4_range = parse_octet_range(parts[3])?;

        Ok(generate_ips_from_ranges(
            &octet1_range,
            &octet2_range,
            &octet3_range,
            &octet4_range,
        ))
    }

    /// Return the queued scan addresses.
    pub fn hosts(&self) -> Vec<IpAddr> {
        self.ips.clone()
    }

    /// Return the number of queued scan addresses.
    pub fn len(&self) -> usize {
        self.ips.len()
    }

    /// Return whether this factory has no queued scan addresses.
    pub fn is_empty(&self) -> bool {
        self.ips.is_empty()
    }

    /// Scan all queued addresses and return every successfully identified miner.
    ///
    /// Unsupported hosts and failed identification attempts are skipped. An
    /// error is returned only when the factory has no queued IP addresses.
    pub async fn scan(&self) -> Result<Vec<Box<dyn Miner>>> {
        if self.ips.is_empty() {
            return Err(anyhow::anyhow!(
                "No IPs to scan. Use with_subnet, with_octets, or with_range to set IPs."
            ));
        }

        let concurrency = self
            .concurrent
            .unwrap_or(calculate_optimal_concurrency(self.ips.len()));

        if let Some(desired_nofile) = self.nofile_limit.or_else(|| {
            self.nofile_adjustment
                .then(|| calculate_desired_nofile_limit(concurrency))
        }) {
            maybe_adjust_nofile_limit(desired_nofile);
        }

        let connection_limit = Arc::new(Semaphore::new(concurrency.max(1)));
        let miners: Vec<Box<dyn Miner>> = stream::iter(self.ips.iter().copied())
            .map(|ip| {
                let connection_limit = Arc::clone(&connection_limit);
                async move {
                    self.scan_miner_with_connection_limit(ip, connection_limit)
                        .await
                        .ok()
                        .flatten()
                }
            })
            .buffer_unordered(concurrency)
            .filter_map(|miner_opt| async move { miner_opt })
            .collect()
            .await;

        Ok(miners)
    }

    /// Scan queued addresses as a stream of successfully identified miners.
    ///
    /// Use this when callers should process miners as soon as they are found
    /// instead of waiting for the full scan to finish.
    pub fn scan_stream(&self) -> Pin<Box<impl Stream<Item = Box<dyn Miner>> + Send + use<>>> {
        let concurrency = self
            .concurrent
            .unwrap_or(calculate_optimal_concurrency(self.ips.len()));

        if let Some(desired_nofile) = self.nofile_limit.or_else(|| {
            self.nofile_adjustment
                .then(|| calculate_desired_nofile_limit(concurrency))
        }) {
            maybe_adjust_nofile_limit(desired_nofile);
        }

        let factory = Arc::new(self.clone());
        let ips: Arc<[IpAddr]> = Arc::from(self.ips.as_slice());
        let connection_limit = Arc::new(Semaphore::new(concurrency.max(1)));

        let ip_count = ips.len();
        let stream = stream::iter(0..ip_count)
            .map(move |i| {
                let factory = Arc::clone(&factory);
                let ips = Arc::clone(&ips);
                let connection_limit = Arc::clone(&connection_limit);
                async move {
                    factory
                        .scan_miner_with_connection_limit(ips[i], connection_limit)
                        .await
                        .ok()
                        .flatten()
                }
            })
            .buffer_unordered(concurrency)
            .filter_map(|miner_opt| async move { miner_opt });

        Box::pin(stream)
    }

    /// Scan queued addresses as a stream that preserves every attempted IP.
    ///
    /// Stream items are `(ip, miner)` pairs. `miner` is `None` when the host did
    /// not identify as a supported miner.
    #[allow(clippy::type_complexity)]
    pub fn scan_stream_with_ip(
        &self,
    ) -> Pin<Box<impl Stream<Item = (IpAddr, Option<Box<dyn Miner>>)> + Send + use<>>> {
        let concurrency = self
            .concurrent
            .unwrap_or(calculate_optimal_concurrency(self.ips.len()));

        if let Some(desired_nofile) = self.nofile_limit.or_else(|| {
            self.nofile_adjustment
                .then(|| calculate_desired_nofile_limit(concurrency))
        }) {
            maybe_adjust_nofile_limit(desired_nofile);
        }

        let factory = Arc::new(self.clone());
        let ips: Arc<[IpAddr]> = Arc::from(self.ips.as_slice());
        let connection_limit = Arc::new(Semaphore::new(concurrency.max(1)));

        let ip_count = ips.len();
        let stream = stream::iter(0..ip_count)
            .map(move |i| {
                let factory = Arc::clone(&factory);
                let ips = Arc::clone(&ips);
                let connection_limit = Arc::clone(&connection_limit);
                async move {
                    (
                        ips[i],
                        factory
                            .scan_miner_with_connection_limit(ips[i], connection_limit)
                            .await
                            .ok()
                            .flatten(),
                    )
                }
            })
            .buffer_unordered(concurrency);

        Box::pin(stream)
    }

    /// Append an octet range, scan it, and return identified miners.
    pub async fn scan_by_octets(
        self,
        octet1: &str,
        octet2: &str,
        octet3: &str,
        octet4: &str,
    ) -> Result<Vec<Box<dyn Miner>>> {
        self.with_octets(octet1, octet2, octet3, octet4)?
            .scan()
            .await
    }

    /// Append an IPv4 range string, scan it, and return identified miners.
    pub async fn scan_by_range(self, range_str: &str) -> Result<Vec<Box<dyn Miner>>> {
        self.with_range(range_str)?.scan().await
    }
}

#[cfg(unix)]
fn maybe_adjust_nofile_limit(desired: u64) {
    if let Err(err) = rlimit::increase_nofile_limit(desired) {
        tracing::warn!("failed to raise RLIMIT_NOFILE to {desired}: {err}");
    }
}

#[cfg(windows)]
fn maybe_adjust_nofile_limit(desired: u64) {
    let current = rlimit::getmaxstdio() as u64;
    if current >= desired {
        return;
    }

    let target = desired.min(u32::MAX as u64) as u32;
    if let Err(err) = rlimit::setmaxstdio(target) {
        tracing::warn!("failed to raise maxstdio from {current} to {target}: {err}");
    }
}

#[cfg(not(any(unix, windows)))]
fn maybe_adjust_nofile_limit(_desired: u64) {}

fn parse_octet_range(range_str: &str) -> Result<Vec<u8>> {
    if range_str.contains('-') {
        let parts: Vec<&str> = range_str.split('-').collect();
        if parts.len() != 2 {
            return Err(anyhow::anyhow!("Invalid range format: {}", range_str));
        }

        let start: u8 = parts[0].parse()?;
        let end: u8 = parts[1].parse()?;

        if start > end {
            return Err(anyhow::anyhow!(
                "Invalid range: start > end in {}",
                range_str
            ));
        }

        Ok((start..=end).collect())
    } else {
        let value: u8 = range_str.parse()?;
        Ok(vec![value])
    }
}

fn generate_ips_from_ranges(
    octet1_range: &[u8],
    octet2_range: &[u8],
    octet3_range: &[u8],
    octet4_range: &[u8],
) -> Vec<IpAddr> {
    let mut ips = Vec::new();

    for &o1 in octet1_range {
        for &o2 in octet2_range {
            for &o3 in octet3_range {
                for &o4 in octet4_range {
                    ips.push(IpAddr::V4(Ipv4Addr::new(o1, o2, o3, o4)));
                }
            }
        }
    }

    ips
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::atomic::{AtomicUsize, Ordering};

    #[tokio::test]
    async fn port_race_checks_every_port_when_none_succeed() {
        let checked = Arc::new(std::sync::Mutex::new(Vec::new()));

        let connected = race_miner_ports(|port| {
            checked.lock().unwrap().push(port);
            std::future::ready(false)
        })
        .await;

        assert!(!connected);
        let mut checked = checked.lock().unwrap().clone();
        checked.sort_unstable();
        let mut expected = MINER_PORTS.to_vec();
        expected.sort_unstable();
        assert_eq!(checked, expected);
    }

    #[tokio::test]
    async fn port_race_cancels_remaining_probes_after_success() {
        struct CancellationGuard {
            cancelled: Arc<AtomicUsize>,
            completed: bool,
        }

        impl Drop for CancellationGuard {
            fn drop(&mut self) {
                if !self.completed {
                    self.cancelled.fetch_add(1, Ordering::SeqCst);
                }
            }
        }

        let permits = Arc::new(Semaphore::new(MINER_PORTS.len()));
        let barrier = Arc::new(tokio::sync::Barrier::new(MINER_PORTS.len()));
        let cancelled = Arc::new(AtomicUsize::new(0));

        let connected = race_miner_ports(|port| {
            let permits = Arc::clone(&permits);
            let barrier = Arc::clone(&barrier);
            let cancelled = Arc::clone(&cancelled);
            async move {
                with_connectivity_permit(permits, async move {
                    let mut guard = CancellationGuard {
                        cancelled,
                        completed: false,
                    };
                    barrier.wait().await;
                    if port == MINER_PORTS[0] {
                        guard.completed = true;
                        true
                    } else {
                        std::future::pending().await
                    }
                })
                .await
            }
        })
        .await;

        assert!(connected);
        assert_eq!(cancelled.load(Ordering::SeqCst), MINER_PORTS.len() - 1);
        assert_eq!(permits.available_permits(), MINER_PORTS.len());
    }

    #[tokio::test]
    async fn port_probes_and_retries_share_global_connection_limit() {
        let permits = Arc::new(Semaphore::new(2));
        let active = Arc::new(AtomicUsize::new(0));
        let max_active = Arc::new(AtomicUsize::new(0));

        let outcomes: Vec<bool> = stream::iter(0..4)
            .map(|_| {
                let permits = Arc::clone(&permits);
                let active = Arc::clone(&active);
                let max_active = Arc::clone(&max_active);
                async move {
                    retry_connectivity(1, Duration::ZERO, || {
                        let permits = Arc::clone(&permits);
                        let active = Arc::clone(&active);
                        let max_active = Arc::clone(&max_active);
                        race_miner_ports(move |_| {
                            let permits = Arc::clone(&permits);
                            let active = Arc::clone(&active);
                            let max_active = Arc::clone(&max_active);
                            with_connectivity_permit(permits, async move {
                                let current = active.fetch_add(1, Ordering::SeqCst) + 1;
                                max_active.fetch_max(current, Ordering::SeqCst);
                                tokio::time::sleep(Duration::from_millis(2)).await;
                                active.fetch_sub(1, Ordering::SeqCst);
                                false
                            })
                        })
                    })
                    .await
                }
            })
            .buffer_unordered(4)
            .collect()
            .await;

        assert_eq!(outcomes, vec![false; 4]);
        assert_eq!(max_active.load(Ordering::SeqCst), 2);
        assert_eq!(active.load(Ordering::SeqCst), 0);
        assert_eq!(permits.available_permits(), 2);
    }

    #[tokio::test]
    #[cfg(feature = "whatsminer")]
    async fn identification_timeout_bounds_miner_construction() -> anyhow::Result<()> {
        use asic_rs_firmwares_whatsminer::firmware::WhatsMinerFirmware;
        use tokio::{
            io::{AsyncReadExt, AsyncWriteExt},
            net::TcpListener,
            sync::oneshot,
        };

        let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, 4028)).await?;
        let (construction_started, mut construction_started_rx) = oneshot::channel();
        let server = tokio::spawn(async move {
            let (mut identification_socket, _) = listener.accept().await?;
            let mut request = [0_u8; 256];
            let _bytes_read = identification_socket.read(&mut request).await?;
            identification_socket
                .write_all(
                    b"{\"STATUS\":[{\"STATUS\":\"S\"}],\"DEVDETAILS\":[{\"Driver\":\"bitmicro\"}]}\0",
                )
                .await?;

            let (_construction_socket, _) = listener.accept().await?;
            let _ = construction_started.send(());
            tokio::time::sleep(Duration::from_secs(2)).await;
            std::io::Result::Ok(())
        });
        let factory = MinerFactory::new()
            .with_firmwares(vec![Arc::new(WhatsMinerFirmware::default())])
            .with_identification_timeout(Duration::from_millis(250));
        let started = tokio::time::Instant::now();

        let miner = factory.get_miner(IpAddr::V4(Ipv4Addr::LOCALHOST)).await?;

        assert!(miner.is_none());
        assert!(started.elapsed() < Duration::from_secs(1));
        assert!(construction_started_rx.try_recv().is_ok());
        server.abort();
        Ok(())
    }

    #[tokio::test]
    async fn zero_connectivity_retries_still_probes_once() {
        let mut attempts = 0;

        let connected = retry_connectivity(0, Duration::ZERO, || {
            attempts += 1;
            std::future::ready(false)
        })
        .await;

        assert!(!connected);
        assert_eq!(attempts, 1);
    }

    #[tokio::test]
    async fn connectivity_retry_recovers_within_configured_attempts() {
        let mut outcomes = [false, false, true].into_iter();
        let mut attempts = 0;

        let connected = retry_connectivity(3, Duration::ZERO, || {
            attempts += 1;
            std::future::ready(outcomes.next().unwrap_or(false))
        })
        .await;

        assert!(connected);
        assert_eq!(attempts, 3);
    }

    #[tokio::test]
    async fn connectivity_retry_exhaustion_uses_initial_attempt_plus_retries() {
        let mut attempts = 0;

        let connected = retry_connectivity(2, Duration::ZERO, || {
            attempts += 1;
            std::future::ready(false)
        })
        .await;

        assert!(!connected);
        assert_eq!(attempts, 3);
    }

    #[test]
    fn connectivity_retry_backoff_is_exponential_and_bounded() {
        assert_eq!(
            connectivity_retry_delay(0, CONNECTIVITY_RETRY_BACKOFF),
            Duration::from_millis(100)
        );
        assert_eq!(
            connectivity_retry_delay(1, CONNECTIVITY_RETRY_BACKOFF),
            Duration::from_millis(200)
        );
        assert_eq!(
            connectivity_retry_delay(4, CONNECTIVITY_RETRY_BACKOFF),
            Duration::from_millis(1600)
        );
        assert_eq!(
            connectivity_retry_delay(5, CONNECTIVITY_RETRY_BACKOFF),
            MAX_CONNECTIVITY_RETRY_BACKOFF
        );
        assert_eq!(
            connectivity_retry_delay(u32::MAX, CONNECTIVITY_RETRY_BACKOFF),
            MAX_CONNECTIVITY_RETRY_BACKOFF
        );
    }

    #[test]
    fn connectivity_retry_default_preserves_upstream_value() {
        assert_eq!(MinerFactory::new().connectivity_retries, 3);
    }

    #[test]
    #[cfg(feature = "whatsminer")]
    fn test_identify_whatsminer_rpc() {
        use asic_rs_core::traits::identification::FirmwareIdentification;
        use asic_rs_firmwares_whatsminer::firmware::WhatsMinerFirmware;

        const RAW_DATA: &str = r#"{"STATUS": [{"STATUS": "S", "Msg": "Device Details"}], "DEVDETAILS": [{"DEVDETAILS": 0, "Name": "SM", "ID": 0, "Driver": "bitmicro", "Kernel": "", "Model": "M30S+_VE40"}, {"DEVDETAILS": 1, "Name": "SM", "ID": 1, "Driver": "bitmicro", "Kernel": "", "Model": "M30S+_VE40"}, {"DEVDETAILS": 2, "Name": "SM", "ID": 2, "Driver": "bitmicro", "Kernel": "", "Model": "M30S+_VE40"}], "id": 1}"#;
        let fw = WhatsMinerFirmware::default();
        assert!(fw.identify_rpc(&RAW_DATA.to_uppercase()));
        assert!(fw.is_stock());
    }

    #[test]
    #[cfg(feature = "whatsminer")]
    fn test_identify_whatsminer_web_redirect() {
        use asic_rs_core::traits::identification::{FirmwareIdentification, WebResponse};
        use asic_rs_firmwares_whatsminer::firmware::WhatsMinerFirmware;

        let web_resp = WebResponse {
            body: "",
            auth_header: "",
            algo_header: "",
            redirect_header: "https://example.com/",
            status: 307,
        };
        let fw = WhatsMinerFirmware::default();
        assert!(fw.identify_web(&web_resp));
    }

    #[test]
    fn test_parse_octet_range() {
        let result = parse_octet_range("10").unwrap();
        assert_eq!(result, vec![10]);

        let result = parse_octet_range("1-5").unwrap();
        assert_eq!(result, vec![1, 2, 3, 4, 5]);

        let result = parse_octet_range("200-255").unwrap();
        assert_eq!(result, (200..=255).collect::<Vec<u8>>());

        let result = parse_octet_range("200-100");
        assert!(result.is_err());

        let result = parse_octet_range("1-5-10");
        assert!(result.is_err());

        let result = parse_octet_range("300");
        assert!(result.is_err());
    }

    #[test]
    fn test_generate_ips_from_ranges() {
        let octet1 = vec![192];
        let octet2 = vec![168];
        let octet3 = vec![1];
        let octet4 = vec![1, 2];

        let ips = generate_ips_from_ranges(&octet1, &octet2, &octet3, &octet4);

        assert_eq!(ips.len(), 2);
        assert!(ips.contains(&IpAddr::V4(Ipv4Addr::new(192, 168, 1, 1))));
        assert!(ips.contains(&IpAddr::V4(Ipv4Addr::new(192, 168, 1, 2))));
    }

    #[test]
    #[cfg(feature = "nerdaxe")]
    fn identify_nerdaxe_web() {
        use asic_rs_core::traits::identification::{FirmwareIdentification, WebResponse};
        use asic_rs_firmwares_nerdaxe::firmware::NerdAxeFirmware;

        #[track_caller]
        fn case(body: &str) {
            let response = WebResponse {
                body,
                auth_header: "",
                algo_header: "",
                redirect_header: "",
                status: 200,
            };
            assert!(NerdAxeFirmware::default().identify_web(&response));
        }

        case("<html><title>NerdAxe</title></html>");
        case("<html><title>NerdQAxe</title></html>");
        case("<html><title>NerdMiner</title></html>");
    }

    #[test]
    #[cfg(all(feature = "bitaxe", feature = "nerdaxe"))]
    fn identify_bitaxe_not_nerdaxe() {
        use asic_rs_core::traits::identification::{FirmwareIdentification, WebResponse};
        use asic_rs_firmwares_bitaxe::firmware::BitaxeFirmware;
        use asic_rs_firmwares_nerdaxe::firmware::NerdAxeFirmware;

        let response = WebResponse {
            body: "<html><title>AxeOS</title></html>",
            auth_header: "",
            algo_header: "",
            redirect_header: "",
            status: 200,
        };
        assert!(BitaxeFirmware::default().identify_web(&response));
        assert!(!NerdAxeFirmware::default().identify_web(&response));
    }
}