uta 0.1.2

//! 下载：并发流式下载 + 进度条 + 同名 `.lrc`。

use std::collections::HashSet;
use std::path::{Path, PathBuf};
use std::time::Duration;

use anyhow::{Context, Result, bail};
use futures::{StreamExt, stream};
use indicatif::{MultiProgress, ProgressBar, ProgressStyle};
use reqwest::StatusCode;
use reqwest::header::{CONTENT_RANGE, ETAG, IF_RANGE, RANGE};
use serde::Serialize;
use tokio::io::{AsyncWriteExt, BufWriter};
use tracing::{debug, warn};

use crate::model::{Resolved, Song};
use crate::source::Sources;
use crate::tags::{self, CoverCache, TagData, Written};

/// 一首已解析出直链的歌。
#[derive(Debug, Clone, Serialize)]
pub struct Item {
    #[serde(flatten)]
    pub song: Song,
    #[serde(flatten)]
    pub resolved: Resolved,
}

impl Item {
    pub fn new(song: Song, resolved: Resolved) -> Self {
        Self { song, resolved }
    }
}

#[derive(Debug)]
pub enum Outcome {
    /// `resumed`：从多少字节处续传（0 表示完整下载）
    Downloaded {
        lrc: bool,
        tags: Written,
        resumed: u64,
    },
    Skipped,
    Failed(anyhow::Error),
}

/// 单个文件名分量最大字节数（多数文件系统上限 255，留出扩展名与 `.part` 余量）。
const MAX_STEM_BYTES: usize = 200;
/// 单首下载的尝试次数；每次失败后从已下载处续传。
const ATTEMPTS: usize = 3;

/// 清理文件名：替换路径分隔符与非法字符、去掉控制字符和首尾空白/点，限制长度。
pub fn sanitize_filename(name: &str) -> String {
    let replaced: String = name
        .chars()
        .filter(|c| !c.is_control())
        .map(|c| match c {
            '/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|' => '_',
            c => c,
        })
        .collect();
    let mut s = replaced.trim().trim_end_matches(['.', ' ']).to_string();
    if s.len() > MAX_STEM_BYTES {
        let mut cut = MAX_STEM_BYTES;
        while !s.is_char_boundary(cut) {
            cut -= 1;
        }
        s.truncate(cut);
        s = s.trim_end_matches(['.', ' ']).to_string();
    }
    if s.is_empty() || s.chars().all(|c| c == '.') {
        return "未知".into();
    }
    // Windows 保留名
    let upper = s.split('.').next().unwrap_or("").to_ascii_uppercase();
    let reserved = matches!(upper.as_str(), "CON" | "PRN" | "AUX" | "NUL")
        || (upper.len() == 4
            && (upper.starts_with("COM") || upper.starts_with("LPT"))
            && upper.as_bytes()[3].is_ascii_digit());
    if reserved {
        s.insert(0, '_');
    }
    s
}

/// 文件名主干：`歌名 - 歌手`。
pub fn file_stem(song: &Song) -> String {
    let singers = song.singer_text();
    let raw = if singers.is_empty() {
        song.title.clone()
    } else {
        format!("{} - {}", song.title, singers)
    };
    sanitize_filename(&raw)
}

/// 为一批歌分配输出路径；同批内重名的追加 ` [id]` 区分（跨次运行也稳定）。
pub fn plan_paths(items: &[Item], out_dir: &Path) -> Vec<PathBuf> {
    let mut seen = HashSet::new();
    items
        .iter()
        .map(|it| {
            let ext = &it.resolved.probe.ext;
            let mut stem = file_stem(&it.song);
            if !seen.insert(format!("{stem}.{ext}").to_lowercase()) {
                stem = sanitize_filename(&format!("{stem} [{}]", it.song.id));
                seen.insert(format!("{stem}.{ext}").to_lowercase());
            }
            out_dir.join(format!("{stem}.{ext}"))
        })
        .collect()
}

pub fn part_path(path: &Path) -> PathBuf {
    let mut s = path.as_os_str().to_owned();
    s.push(".part");
    PathBuf::from(s)
}

fn short_name(song: &Song) -> String {
    let name = format!("{} - {}", song.title, song.singer_text());
    let mut out: String = name.chars().take(18).collect();
    if name.chars().count() > 18 {
        out.push('…');
    }
    out
}

/// 保存 ETag 的旁路文件：`<文件>.part.etag`，续传时用于 `If-Range`。
pub fn etag_path(part: &Path) -> PathBuf {
    let mut s = part.as_os_str().to_owned();
    s.push(".etag");
    PathBuf::from(s)
}

/// 解析 `Content-Range: bytes <start>-<end>/<total>`。
pub fn parse_content_range(v: &str) -> Option<(u64, u64, u64)> {
    let rest = v.trim().strip_prefix("bytes ")?;
    let (range, total) = rest.split_once('/')?;
    let (a, b) = range.split_once('-')?;
    Some((
        a.trim().parse().ok()?,
        b.trim().parse().ok()?,
        total.trim().parse().ok()?,
    ))
}

/// 下载到 `part`，校验长度；返回 (总字节数, 续传起点)。
///
/// `part` 已有内容且有 ETag 记录时发 `Range: bytes=<已有>-` + `If-Range: <ETag>` 续传：
/// `referer` 由解析结果给出（各音源 CDN 要求不同）。
/// 实测 QQ 流媒体 ETag 一致时返回 206，不一致时返回 200 整个文件（此时从头写），
/// 起点等于文件大小时返回 416（清除后由调用方重试从头下载）。
async fn fetch_to(
    client: &reqwest::Client,
    url: &str,
    referer: Option<&str>,
    part: &Path,
    pb: &ProgressBar,
) -> Result<(u64, u64)> {
    let etag_file = etag_path(part);
    let have = tokio::fs::metadata(part).await.map_or(0, |m| m.len());
    let etag = if have > 0 {
        tokio::fs::read_to_string(&etag_file)
            .await
            .ok()
            .filter(|e| !e.trim().is_empty())
    } else {
        None
    };
    let mut rb = client.get(url);
    if let Some(r) = referer {
        rb = rb.header("Referer", r);
    }
    if let Some(e) = &etag {
        rb = rb
            .header(RANGE, format!("bytes={have}-"))
            .header(IF_RANGE, e.trim());
    }
    let resp = rb.send().await.context("请求失败")?;
    if resp.status() == StatusCode::RANGE_NOT_SATISFIABLE {
        let _ = tokio::fs::remove_file(part).await;
        let _ = tokio::fs::remove_file(&etag_file).await;
        bail!("续传位置无效，已清除未完成的文件");
    }
    let resp = resp.error_for_status().context("服务器返回错误")?;
    let (start, total) = if resp.status() == StatusCode::PARTIAL_CONTENT {
        let cr = resp
            .headers()
            .get(CONTENT_RANGE)
            .and_then(|v| v.to_str().ok())
            .and_then(parse_content_range)
            .context("206 响应缺少有效的 Content-Range")?;
        if cr.0 != have {
            bail!("续传起点不符：期望 {have}，服务器给出 {}", cr.0);
        }
        (have, Some(cr.2))
    } else {
        if have > 0 {
            debug!(have, "无法续传（服务器文件已变化或不支持 Range），从头下载");
        }
        // 从头下载：记录新的 ETag
        match resp.headers().get(ETAG).and_then(|v| v.to_str().ok()) {
            Some(e) => tokio::fs::write(&etag_file, e).await?,
            None => {
                let _ = tokio::fs::remove_file(&etag_file).await;
            }
        }
        (0, resp.content_length())
    };
    pb.set_length(total.unwrap_or(0));
    pb.set_position(start);
    pb.reset_eta();

    let file = if start > 0 {
        tokio::fs::OpenOptions::new().append(true).open(part).await
    } else {
        tokio::fs::File::create(part).await
    }
    .with_context(|| format!("无法写入 {}", part.display()))?;
    let mut w = BufWriter::with_capacity(256 * 1024, file);
    let mut written = start;
    let mut body = resp.bytes_stream();
    let result = async {
        while let Some(chunk) = body.next().await {
            let chunk = chunk.context("下载中断")?;
            w.write_all(&chunk).await.context("写入文件失败")?;
            written += chunk.len() as u64;
            if total.is_none() {
                pb.set_length(written);
            }
            pb.set_position(written);
        }
        Ok::<_, anyhow::Error>(())
    }
    .await;
    // 出错也把已收到的数据落盘，下次可从这里续传
    w.flush().await?;
    drop(w);
    result?;
    if let Some(n) = total
        && n != written
    {
        bail!("下载不完整：{written}/{n} 字节");
    }
    Ok((written, start))
}

/// 在 `.part` 上写标签（改名前写，避免最终文件处于半写状态）。
/// 写失败但文件仍可解析时只告警；文件被写坏则返回错误。
async fn tag_part(
    part: &Path,
    song: &Song,
    lyric: Option<&str>,
    cover: Option<&tags::Cover>,
) -> Result<Written> {
    let part = part.to_path_buf();
    let song = song.clone();
    let lyric = lyric.map(str::to_string);
    let cover = cover.map(|c| tags::Cover {
        data: c.data.clone(),
        mime: c.mime.clone(),
    });
    tokio::task::spawn_blocking(move || {
        let data = TagData {
            song: &song,
            lyric: lyric.as_deref(),
            cover: cover.as_ref(),
        };
        match tags::write_file(&part, &data) {
            Ok(w) => Ok(w),
            Err(e) if tags::readable(&part) => {
                warn!(id = %song.id, "写标签失败（音频完好，保留无标签文件）: {e:#}");
                Ok(Written::default())
            }
            Err(e) => Err(e.context("写标签后文件损坏")),
        }
    })
    .await
    .context("标签任务异常退出")?
}

/// 写 `.lrc`（调用方保证不覆盖已有文件，与 Python 版一致）。
async fn write_lrc(path: &Path, text: &str) -> bool {
    match tokio::fs::write(path, format!("{text}\n")).await {
        Ok(()) => true,
        Err(e) => {
            warn!("写入歌词 {} 失败: {e}", path.display());
            false
        }
    }
}

async fn download_one(
    client: &reqwest::Client,
    sources: &Sources,
    covers: &CoverCache,
    item: &Item,
    path: &Path,
    pb: &ProgressBar,
) -> Outcome {
    let lrc_path = path.with_extension("lrc");
    let lrc_exists = tokio::fs::try_exists(&lrc_path).await.unwrap_or(false);
    let fetch_lyric = || async {
        match sources.lyric(&item.song).await {
            Ok(l) => l,
            Err(e) => {
                warn!(id = %item.song.id, "获取歌词失败: {e:#}");
                None
            }
        }
    };
    // 已存在的文件不动（包括不补标签），只补 .lrc
    if tokio::fs::try_exists(path).await.unwrap_or(false) {
        if !lrc_exists && let Some(text) = fetch_lyric().await {
            write_lrc(&lrc_path, &text).await;
        }
        return Outcome::Skipped;
    }
    let part = part_path(path);
    let audio = async {
        let mut last = None;
        for attempt in 1..=ATTEMPTS {
            let r = &item.resolved;
            match fetch_to(client, &r.probe.url, r.referer, &part, pb).await {
                Ok((_, start)) => return Ok(start),
                Err(e) => {
                    debug!(id = %item.song.id, attempt, "下载失败: {e:#}");
                    last = Some(e);
                    if attempt < ATTEMPTS {
                        tokio::time::sleep(Duration::from_secs(1)).await;
                    }
                }
            }
        }
        Err(last.expect("至少尝试一次"))
    };
    // 音频、歌词、封面并行获取（歌词即使 .lrc 已存在也要取，用于内嵌）
    let (audio, lyric, cover) = tokio::join!(
        audio,
        fetch_lyric(),
        covers.get(&sources.client, &item.song)
    );
    // 下载失败时保留 .part 与 .etag，重新运行同一命令会从断点续传
    let resumed = match audio {
        Ok(start) => start,
        Err(e) => return Outcome::Failed(e.context("未完成的部分已保留，重新运行可续传")),
    };
    let result = async {
        pb.set_message(format!("{} 写入标签…", short_name(&item.song)));
        let tags = tag_part(&part, &item.song, lyric.as_deref(), cover.as_deref()).await?;
        tokio::fs::rename(&part, path)
            .await
            .with_context(|| format!("无法重命名为 {}", path.display()))?;
        Ok::<_, anyhow::Error>(tags)
    }
    .await;
    let _ = tokio::fs::remove_file(etag_path(&part)).await;
    match result {
        Ok(tags) => {
            let lrc = match &lyric {
                Some(text) if !lrc_exists => write_lrc(&lrc_path, text).await,
                _ => false,
            };
            Outcome::Downloaded { lrc, tags, resumed }
        }
        Err(e) => {
            // 写标签/改名阶段出错，文件可能已损坏，不再续传
            let _ = tokio::fs::remove_file(&part).await;
            Outcome::Failed(e)
        }
    }
}

/// 汇报行的附加说明，如"（.lrc、内嵌歌词、封面）"。
fn extras(lrc: bool, tags: &Written) -> String {
    let parts: Vec<&str> = [
        (lrc, ".lrc"),
        (tags.lyric, "内嵌歌词"),
        (tags.cover, "封面"),
        (tags.track, "曲序"),
        (tags.basic, "基本标签"),
    ]
    .into_iter()
    .filter_map(|(on, s)| on.then_some(s))
    .collect();
    if parts.is_empty() {
        String::new()
    } else {
        format!("（{}）", parts.join("、"))
    }
}

/// 并发下载一批歌，返回与输入同序的结果。
pub async fn download_all(
    client: &reqwest::Client,
    sources: &Sources,
    items: &[Item],
    out_dir: &Path,
    jobs: usize,
) -> Result<Vec<Outcome>> {
    tokio::fs::create_dir_all(out_dir)
        .await
        .with_context(|| format!("无法创建输出目录 {}", out_dir.display()))?;
    let paths = plan_paths(items, out_dir);
    let covers = CoverCache::default();

    let mp = MultiProgress::new();
    let total = mp.add(ProgressBar::new(items.len() as u64));
    total.set_style(
        ProgressStyle::with_template(
            "{prefix:.bold} [{bar:30.green/white}] {pos}/{len} 首  {elapsed}",
        )
        .expect("进度条模板")
        .progress_chars("=> "),
    );
    total.set_prefix("总进度");
    let bar_style = ProgressStyle::with_template(
        "  {msg} [{bar:30.cyan/blue}] {bytes}/{total_bytes} {bytes_per_sec} 剩余 {eta}",
    )
    .expect("进度条模板")
    .progress_chars("=> ");
    let report = |line: String| {
        if mp.is_hidden() {
            eprintln!("{line}");
        } else {
            let _ = mp.println(line);
        }
    };

    let mut results: Vec<(usize, Outcome)> = stream::iter(items.iter().zip(&paths).enumerate())
        .map(|(i, (item, path))| {
            let mp = &mp;
            let bar_style = &bar_style;
            let total = &total;
            let report = &report;
            let covers = &covers;
            async move {
                let pb = mp.insert_before(total, ProgressBar::new(0));
                pb.set_style(bar_style.clone());
                pb.set_message(short_name(&item.song));
                let outcome = download_one(client, sources, covers, item, path, &pb).await;
                pb.finish_and_clear();
                mp.remove(&pb);
                total.inc(1);
                let name = short_name(&item.song);
                report(match &outcome {
                    Outcome::Downloaded { lrc, tags, resumed } => {
                        let from = if *resumed > 0 {
                            format!("（从 {:.1} MB 处续传）", *resumed as f64 / 1048576.0)
                        } else {
                            String::new()
                        };
                        format!("✔ {name}{}{from}", extras(*lrc, tags))
                    }
                    Outcome::Skipped => format!("· {name}（已存在，跳过）"),
                    Outcome::Failed(e) => format!("✘ {name}：{e:#}"),
                });
                (i, outcome)
            }
        })
        .buffer_unordered(jobs.max(1))
        .collect()
        .await;
    total.finish_and_clear();
    results.sort_by_key(|(i, _)| *i);
    Ok(results.into_iter().map(|(_, o)| o).collect())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::model::Level;
    use crate::probe::ProbeResult;

    fn song(id: &str, title: &str, singers: &[&str]) -> Song {
        Song {
            id: id.into(),
            title: title.into(),
            singers: singers.iter().map(|s| s.to_string()).collect(),
            ..Song::default()
        }
    }

    fn item(mid: &str, title: &str, singers: &[&str], ext: &str) -> Item {
        Item::new(
            song(mid, title, singers),
            Resolved {
                via: "vkeys",
                quality: "SQ无损".into(),
                level: Level::Sq,
                probe: ProbeResult {
                    url: "http://x".into(),
                    ext: ext.into(),
                    size: 1,
                },
                referer: None,
                fallback: None,
            },
        )
    }

    #[test]
    fn sanitize_basic() {
        assert_eq!(sanitize_filename("夜曲 - 周杰伦"), "夜曲 - 周杰伦");
        assert_eq!(sanitize_filename("AC/DC: Back?"), "AC_DC_ Back_");
        assert_eq!(sanitize_filename("a\u{0}b\nc"), "abc");
        assert_eq!(sanitize_filename("  name.. "), "name");
        assert_eq!(sanitize_filename(""), "未知");
        assert_eq!(sanitize_filename(".."), "未知");
        assert_eq!(sanitize_filename("con"), "_con");
        assert_eq!(sanitize_filename("COM1.x"), "_COM1.x");
        assert_eq!(sanitize_filename("COMX"), "COMX");
    }

    #[test]
    fn sanitize_length_on_char_boundary() {
        let long = "夜".repeat(100); // 300 字节
        let s = sanitize_filename(&long);
        assert!(s.len() <= MAX_STEM_BYTES);
        assert_eq!(s, "夜".repeat(66));
    }

    #[test]
    fn stems() {
        assert_eq!(
            file_stem(&song("m", "珊瑚海", &["周杰伦", "Lara梁心颐"])),
            "珊瑚海 - 周杰伦, Lara梁心颐"
        );
        assert_eq!(file_stem(&song("m", "无名", &[])), "无名");
    }

    #[test]
    fn plan_paths_dedup() {
        let items = [
            item("a1", "晴天", &["周杰伦"], "flac"),
            item("b2", "晴天", &["周杰伦"], "flac"),
            item("c3", "晴天", &["周杰伦"], "mp3"),
        ];
        let paths = plan_paths(&items, Path::new("/out"));
        assert_eq!(paths[0], Path::new("/out/晴天 - 周杰伦.flac"));
        assert_eq!(paths[1], Path::new("/out/晴天 - 周杰伦 [b2].flac"));
        assert_eq!(paths[2], Path::new("/out/晴天 - 周杰伦.mp3"));
    }

    #[test]
    fn part_and_lrc_paths() {
        let p = Path::new("/out/夜曲 - 周杰伦.flac");
        assert_eq!(part_path(p), Path::new("/out/夜曲 - 周杰伦.flac.part"));
        assert_eq!(p.with_extension("lrc"), Path::new("/out/夜曲 - 周杰伦.lrc"));
    }

    #[test]
    fn extras_text() {
        assert_eq!(extras(false, &Written::default()), "");
        let w = Written {
            basic: false,
            lyric: true,
            cover: true,
            track: true,
        };
        assert_eq!(extras(true, &w), "（.lrc、内嵌歌词、封面、曲序）");
    }

    #[test]
    fn short_names() {
        assert_eq!(short_name(&song("m", "夜曲", &["周杰伦"])), "夜曲 - 周杰伦");
        assert_eq!(
            short_name(&song("m", "一二三四五六七八九十一二三四五六", &["周杰伦"])),
            "一二三四五六七八九十一二三四五六 -…"
        );
    }

    #[test]
    fn content_range() {
        assert_eq!(
            parse_content_range("bytes 1000-1999/26691277"),
            Some((1000, 1999, 26691277))
        );
        assert_eq!(parse_content_range(" bytes 0-0/1 "), Some((0, 0, 1)));
        assert_eq!(parse_content_range("bytes */26691277"), None);
        assert_eq!(parse_content_range("items 0-1/2"), None);
    }

    mod resume {
        use super::super::*;
        use std::sync::Arc;
        use std::sync::atomic::{AtomicUsize, Ordering};
        use tokio::io::{AsyncReadExt, AsyncWriteExt};
        use tokio::net::TcpListener;

        #[derive(Clone, Copy)]
        enum Mode {
            Normal,
            /// 第一次连接只发 n 字节正文就断开
            DropFirstAfter(usize),
            /// 不支持 Range，总是 200
            IgnoreRange,
        }

        fn payload() -> Vec<u8> {
            (0..50_000u32).map(|i| (i % 251) as u8).collect()
        }

        /// 按实测 QQ CDN 的行为响应：If-Range 匹配 → 206；不匹配 → 200；起点 ≥ 长度 → 416。
        async fn server(etag: &'static str, mode: Mode) -> (String, Arc<AtomicUsize>) {
            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = listener.local_addr().unwrap();
            let hits = Arc::new(AtomicUsize::new(0));
            let counter = hits.clone();
            let body = payload();
            tokio::spawn(async move {
                loop {
                    let (mut sock, _) = listener.accept().await.unwrap();
                    let n = counter.fetch_add(1, Ordering::SeqCst);
                    let mut buf = vec![0u8; 4096];
                    let len = sock.read(&mut buf).await.unwrap_or(0);
                    let req = String::from_utf8_lossy(&buf[..len]).to_ascii_lowercase();
                    let header = |name: &str| {
                        req.lines()
                            .find_map(|l| l.strip_prefix(&format!("{name}: ")))
                            .map(|v| v.trim().to_string())
                    };
                    let range = header("range").and_then(|r| {
                        r.strip_prefix("bytes=")?
                            .strip_suffix('-')?
                            .parse::<usize>()
                            .ok()
                    });
                    let if_range_ok = header("if-range").is_none_or(|e| e == etag);
                    let total = body.len();
                    let (head, part): (String, &[u8]) = match (mode, range) {
                        (Mode::IgnoreRange, _) | (_, None) => (
                            format!(
                                "HTTP/1.1 200 OK\r\ncontent-length: {total}\r\netag: {etag}\r\n"
                            ),
                            &body[..],
                        ),
                        (_, Some(_)) if !if_range_ok => (
                            format!(
                                "HTTP/1.1 200 OK\r\ncontent-length: {total}\r\netag: {etag}\r\n"
                            ),
                            &body[..],
                        ),
                        (_, Some(start)) if start >= total => (
                            "HTTP/1.1 416 Range Not Satisfiable\r\ncontent-length: 0\r\n"
                                .to_string(),
                            &[][..],
                        ),
                        (_, Some(start)) => (
                            format!(
                                "HTTP/1.1 206 Partial Content\r\ncontent-length: {}\r\ncontent-range: bytes {start}-{}/{total}\r\netag: {etag}\r\n",
                                total - start,
                                total - 1
                            ),
                            &body[start..],
                        ),
                    };
                    let _ = sock
                        .write_all(format!("{head}connection: close\r\n\r\n").as_bytes())
                        .await;
                    let send = match mode {
                        Mode::DropFirstAfter(k) if n == 0 => &part[..k.min(part.len())],
                        _ => part,
                    };
                    let _ = sock.write_all(send).await;
                    let _ = sock.flush().await;
                }
            });
            (format!("http://{addr}/a.flac"), hits)
        }

        fn client() -> reqwest::Client {
            reqwest::Client::builder()
                .no_proxy()
                .timeout(Duration::from_secs(5))
                .build()
                .unwrap()
        }

        fn temp(name: &str) -> (PathBuf, PathBuf) {
            let dir =
                std::env::temp_dir().join(format!("uta-resume-{name}-{}", std::process::id()));
            let _ = std::fs::remove_dir_all(&dir);
            std::fs::create_dir_all(&dir).unwrap();
            let part = dir.join("a.flac.part");
            (dir, part)
        }

        #[tokio::test]
        async fn fresh_download_saves_etag() {
            let (url, _) = server("\"v1\"", Mode::Normal).await;
            let (dir, part) = temp("fresh");
            let r = fetch_to(&client(), &url, None, &part, &ProgressBar::hidden())
                .await
                .unwrap();
            assert_eq!(r, (50_000, 0));
            assert_eq!(std::fs::read(&part).unwrap(), payload());
            assert_eq!(std::fs::read_to_string(etag_path(&part)).unwrap(), "\"v1\"");
            std::fs::remove_dir_all(dir).unwrap();
        }

        #[tokio::test]
        async fn interrupted_then_resumed() {
            let (url, hits) = server("\"v1\"", Mode::DropFirstAfter(12_345)).await;
            let (dir, part) = temp("interrupt");
            let pb = ProgressBar::hidden();
            assert!(fetch_to(&client(), &url, None, &part, &pb).await.is_err());
            // 断线前收到的数据已落盘
            assert_eq!(std::fs::metadata(&part).unwrap().len(), 12_345);
            let r = fetch_to(&client(), &url, None, &part, &pb).await.unwrap();
            assert_eq!(r, (50_000, 12_345));
            assert_eq!(std::fs::read(&part).unwrap(), payload());
            assert_eq!(hits.load(Ordering::SeqCst), 2);
            std::fs::remove_dir_all(dir).unwrap();
        }

        #[tokio::test]
        async fn etag_changed_restarts() {
            let (url, _) = server("\"new\"", Mode::Normal).await;
            let (dir, part) = temp("etag");
            std::fs::write(&part, vec![9u8; 777]).unwrap();
            std::fs::write(etag_path(&part), "\"old\"").unwrap();
            let r = fetch_to(&client(), &url, None, &part, &ProgressBar::hidden())
                .await
                .unwrap();
            assert_eq!(r, (50_000, 0));
            assert_eq!(std::fs::read(&part).unwrap(), payload());
            assert_eq!(
                std::fs::read_to_string(etag_path(&part)).unwrap(),
                "\"new\""
            );
            std::fs::remove_dir_all(dir).unwrap();
        }

        #[tokio::test]
        async fn part_without_etag_restarts() {
            let (url, _) = server("\"v1\"", Mode::Normal).await;
            let (dir, part) = temp("noetag");
            std::fs::write(&part, vec![9u8; 777]).unwrap();
            let r = fetch_to(&client(), &url, None, &part, &ProgressBar::hidden())
                .await
                .unwrap();
            assert_eq!(r, (50_000, 0));
            assert_eq!(std::fs::read(&part).unwrap(), payload());
            std::fs::remove_dir_all(dir).unwrap();
        }

        #[tokio::test]
        async fn server_ignoring_range_restarts() {
            let (url, _) = server("\"v1\"", Mode::IgnoreRange).await;
            let (dir, part) = temp("ignore");
            std::fs::write(&part, &payload()[..1000]).unwrap();
            std::fs::write(etag_path(&part), "\"v1\"").unwrap();
            let r = fetch_to(&client(), &url, None, &part, &ProgressBar::hidden())
                .await
                .unwrap();
            assert_eq!(r, (50_000, 0));
            assert_eq!(std::fs::read(&part).unwrap(), payload());
            std::fs::remove_dir_all(dir).unwrap();
        }

        #[tokio::test]
        async fn range_not_satisfiable_clears_then_redownloads() {
            let (url, _) = server("\"v1\"", Mode::Normal).await;
            let (dir, part) = temp("416");
            std::fs::write(&part, payload()).unwrap();
            std::fs::write(etag_path(&part), "\"v1\"").unwrap();
            let pb = ProgressBar::hidden();
            assert!(fetch_to(&client(), &url, None, &part, &pb).await.is_err());
            assert!(!part.exists() && !etag_path(&part).exists());
            let r = fetch_to(&client(), &url, None, &part, &pb).await.unwrap();
            assert_eq!(r, (50_000, 0));
            std::fs::remove_dir_all(dir).unwrap();
        }
    }
}