fast-xray/src/easy/mod.rs

//! `easy`: find one IP fast enough to hit a target speed, then stop.
//!
//! A [`producer`] keeps a validated *queue* of IPs (those that pass ping +
//! latency) topped up between a low and high watermark. The [`consumer`] runs
//! `CONCURRENCY` independent screens off that queue: under n-way contention each
//! stream gets roughly link/n, so anything reaching `target/n` is worth a solo
//! confirm and anything already at `target` is a guaranteed pass. Promising IPs
//! go onto a separate confirm queue; once the in-flight screens wind down,
//! they're re-tested single-threaded (accurate, no contention) until one clears
//! `target`. Every so many failures we re-measure the direct link in case it
//! drifted. A fixed [`ui`] panel shows what's under test and the last few
//! results, sharing one mutex-guarded [`state`] bucket between the three tasks.

mod consumer;
mod engine;
mod producer;
mod state;
mod ui;

pub(crate) const CHUAN_NODE: &str = "vless://6c1a367a-b67f-49e4-9224-a372fc44beb7@cloudcone.pchuan.top:443?encryption=none&security=tls&sni=cloudcone.pchuan.top&insecure=0&allowInsecure=0&type=ws&host=cloudcone.pchuan.top&path=%2Fdemo#CF-VLESS-WS-TLS-er2r80yifx";

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;

use anyhow::{Result, anyhow};
use console::style;
use indicatif::ProgressBar;

use crate::cli::EasyArgs;
use crate::cloudflare::{self, Family};
use crate::report::{print_easy_found, resolve_node, write_file};

use ui::{Panel, render_loop};

// Surface the headless engine and observable state for the web server, which
// builds an alternate (SSE) UI around the same search.
pub(crate) use engine::{measure_baseline, resolve_target, search};
pub(crate) use state::{FeedStage, FeedState, Found, Outcome, Phase, Snapshot, State};

// Fixed knobs — the point of `easy` is to not expose these.
const CONCURRENCY: usize = 5; // independent screens in flight, also the screen floor divisor
const QUEUE_HIGH: usize = 50; // validated-queue refill target
const QUEUE_LOW: usize = 30; // refill kicks in once the queue drops below this
const PRODUCE_BATCH: usize = 50; // IPs per discovery round — one full latency wave (matches LAT_CONCURRENCY)
const RECENT: usize = 3; // finished results kept on screen for review
const FEED: usize = 5; // live discovery lines (ping/latency) shown while finding IPs

const PING_CONCURRENCY: usize = 200;
const PING_TIMEOUT: f64 = 3.0;
const PING_MIN_MS: f64 = 10.0;
const PING_MAX_MS: f64 = 200.0;

const LAT_CONCURRENCY: usize = 50;
const LAT_TIMEOUT: f64 = 5.0;
const LAT_MAX_MS: f64 = 300.0;

const SPEED_TIMEOUT: f64 = 5.0;
const SPEED_BYTES: u64 = 50_000_000; // file is large; time/steady-state caps first

const MAX_BARREN_ROUNDS: usize = 8; // give up after this many empty discovery rounds
const DEFAULT_SPEED_FRACTION: f64 = 0.80; // default target = this × direct speed

const DIRECT_TIMEOUT: f64 = 8.0; // one longer direct measurement (the mirror is too bimodal to sample)
const FAIL_RECALIBRATE: usize = 100; // re-measure the direct link every this many failures

const POLL: Duration = Duration::from_millis(50); // producer/consumer idle poll
const FRAME: Duration = Duration::from_millis(120); // render tick
const FRAMES: [&str; 10] = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"];

pub(crate) async fn run(args: EasyArgs) -> Result<()> {
    run_inner(args, false).await
}

pub(crate) async fn run_with_fallback(args: EasyArgs) -> Result<()> {
    run_inner(args, true).await
}

async fn run_inner(args: EasyArgs, fallback_best: bool) -> Result<()> {
    let node = Arc::new(resolve_node(&args.node_file, &args.node)?);
    if matches!(args.speed, Some(s) if s <= 0.0) {
        return Err(anyhow!("--speed must be greater than 0"));
    }

    let state = State::new();

    // 0. Direct (no-proxy) speed via a domestic mirror — the link ceiling, and
    //    the source of the default target when --speed is omitted.
    let spinner = spin("Measuring direct (no-proxy) download speed…");
    let baseline = engine::measure_baseline().await;
    state.set_baseline(baseline);
    match baseline {
        Some(mbs) => spinner.finish_with_message(format!(
            "{} direct download speed: {} MB/s",
            style("✓").green().bold(),
            style(format!("{mbs:.2}")).cyan()
        )),
        None => spinner.finish_with_message(format!(
            "{} direct speed unavailable (direct access blocked?)",
            style("!").yellow().bold()
        )),
    }

    // Resolve the target: explicit --speed, else a fraction of the measured
    // direct speed. Reject a target the local link can't reach; with neither a
    // target nor a measured link there's nothing to aim for.
    let (target, note, explicit_target) = engine::resolve_target(args.speed, baseline)?;
    state.set_target(target);

    let family = if args.ipv6 { Family::Both } else { Family::V4 };
    let spinner = spin("Fetching Cloudflare ranges…");
    let ranges = Arc::new(cloudflare::fetch_ranges(family).await?);
    spinner.finish_with_message(format!(
        "{} Cloudflare ranges: {} v4, {} v6",
        style("✓").green().bold(),
        ranges.v4.len(),
        ranges.v6.len()
    ));
    eprintln!(
        "{} target ≥ {} MB/s{}, up to {} valid IPs",
        style("easy").bold().cyan(),
        style(format!("{target:.2}")).cyan(),
        style(note).dim(),
        args.max
    );

    let stop = Arc::new(AtomicBool::new(false));
    let ui = tokio::spawn(render_loop(state.clone(), Panel::new(), args.max, stop.clone()));

    let hit =
        engine::search(
            node.clone(),
            ranges,
            args.max,
            args.ipv6,
            explicit_target,
            fallback_best,
            state.clone(),
        )
            .await;

    stop.store(true, Ordering::Relaxed);
    let _ = ui.await; // render loop clears the view before returning
    state.set_phase(Phase::Done);

    match hit {
        Some(Found { ip, latency, mbs }) => {
            let alias = format!("{mbs:.2}MB-{:.0}ms-{}", latency.as_secs_f64() * 1000.0, ip);
            let url = node.to_url(ip, &alias);
            print_easy_found(ip, mbs, latency, &url);
            write_file(&args.output.join("result.txt"), &format!("{url}\n"))?;
        }
        None => {
            let (final_target, valid) = (state.target(), state.valid());
            eprintln!(
                "{} no IP reached {final_target:.2} MB/s after testing {valid} valid IP(s)",
                style("✗").red().bold()
            );
        }
    }
    Ok(())
}

fn spin(msg: &str) -> ProgressBar {
    let pb = ProgressBar::new_spinner();
    pb.enable_steady_tick(Duration::from_millis(90));
    pb.set_message(msg.to_string());
    pb
}

fn secs(s: f64) -> Duration {
    Duration::from_secs_f64(s)
}

fn millis(ms: f64) -> Duration {
    Duration::from_secs_f64(ms / 1000.0)
}