//! 大文件多连接分块的编排与校验。
//!
//! 分块本身由 `chunked.rs` 负责（纯规划 + 单块抓取），这里负责并发、进度上报、
//! 续传元数据的读写与「整文件重算一次摘要」（分块是乱序写，只能整文件校验）。
//!
//! # 续传（M293：台账 P21 第 2 条 + P19(b)）
//!
//! 此前这里开头无条件 `part.discard()`、失败也 `discard()`，于是 188MB 的 JDK 任何一次失败
//! 都从 0 重下；M289 又给「预算耗尽」加了「不再重试、不再换源」的短路，慢链路上一次晚到的
//! 失败就变成「这个文件下不来」。现在：
//!
//! * 每落完一段就把它连同**这段字节的摘要**写进 `ResumeMeta::chunks`（`ChunkJournal`）；
//! * 失败**不再**丢中间文件——已完成的段是几十分钟的字节，元数据里带着它们的自证记录；
//! * 下次进本函数先过 [`super::chunk_resume::decide`]（身份）与 [`Self::verify_chunks`]
//!   （旧字节自证），两道门都过才复用；任何一道不过就整份丢掉从头再来；
//! * 整文件摘要照旧在传输结束后整份重算，覆盖新下 + 复用的全部字节。
//!
//! 唯一仍然主动 `discard()` 的失败是 `RangeUnsupported`：服务器不支持 Range 时单连接路径
//! 接不住「预分配 + 乱序段」这种形状的文件（它只会截断重写），留着没意义。

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

use super::atomic::PartFile;
use super::cache::now_ms;
use super::chunk_resume::{self, ChunkDecision, ChunkJournal};
use super::chunked::{self, ChunkFetch};
use super::engine::{AttemptOk, DownloadEngine};
use super::error::{DownloadError, DownloadErrorKind};
use super::progress::ProgressEvent;
use super::resume::{ChunkRecord, ResumeMeta};
use super::stall::{StallReason, StallReport, StallStage};
use super::types::{Candidate, DownloadId, DownloadRequest, HashAlgo, Integrity};
use super::verify;

impl DownloadEngine {
    pub(crate) async fn attempt_chunked(
        &self,
        id: DownloadId,
        req: &DownloadRequest,
        candidate: &Candidate,
        part: &mut PartFile,
        total: Option<u64>,
    ) -> Result<AttemptOk, DownloadError> {
        let Some(total) = total.filter(|total| *total > 0) else {
            return Err(DownloadError::new(
                DownloadErrorKind::RangeUnsupported,
                "总长未知，无法分块",
            ));
        };
        let ranges = chunked::plan(total, self.cfg.chunked.effective_connections());
        if ranges.len() < 2 {
            return Err(DownloadError::new(
                DownloadErrorKind::RangeUnsupported,
                "可切分段不足，退化为单连接",
            ));
        }
        let integrity = req.integrity.as_ref();

        // ① 身份：旧分块是不是**这次**下载的产物（纯判定，不碰盘）。
        let meta_path = part.meta_path();
        let previous = ResumeMeta::load(&meta_path);
        let decision = chunk_resume::decide(
            req.resume,
            part.len(),
            previous.as_ref(),
            &candidate.url,
            integrity.map(|item| (item.algo, item.hex.as_str())),
            total,
            &ranges,
        );
        // ② 自证：按记录重读那些字节、重算摘要。任何一段对不上就整份丢掉从头再来
        //    （理由见 `chunk_resume` 模块头：对不上说明这份元数据本身已经不可信）。
        let mut done: Vec<ChunkRecord> = Vec::new();
        match decision {
            ChunkDecision::Reuse { done: records } => {
                if self.verify_chunks(part, &records).await {
                    done = records;
                } else {
                    part.discard();
                }
            }
            ChunkDecision::Restart { .. } => part.discard(),
        }
        let reused: u64 = done.iter().map(ChunkRecord::len).sum();
        // 续传时把旧校验器带上：这是 `If-Range` 的取值来源，也是「远端内容没变」的协议层证据。
        let if_range = if done.is_empty() {
            None
        } else {
            previous
                .as_ref()
                .and_then(|meta| meta.validator().if_range_value().map(str::to_string))
        };

        part.ensure_parent_async()
            .await
            .map_err(|err| io_err("创建下载目录失败", &err))?;
        {
            let mut options = tokio::fs::OpenOptions::new();
            options.create(true).write(true);
            if done.is_empty() {
                // 从头：先把可能存在的旧内容截掉，再预分配整长。
                options.truncate(true);
            }
            let file = options
                .open(part.path())
                .await
                .map_err(|err| io_err("预分配中间文件失败", &err))?;
            // 续传时这一行是「归一化」：分块写是随机偏移，长度本该已经是 total；
            // 若被外部截短/加长过，先摆正，缺的段下面的补抓会填满。
            file.set_len(total)
                .await
                .map_err(|err| io_err("预分配长度失败", &err))?;
        }

        let mut journal = ChunkJournal::begin(
            ResumeMeta {
                url: candidate.url.clone(),
                // 续传沿用旧校验器：新响应可能不给 ETag，而 `If-Range` 要靠它。
                etag: previous.as_ref().and_then(|meta| meta.etag.clone()),
                last_modified: previous
                    .as_ref()
                    .and_then(|meta| meta.last_modified.clone()),
                total: Some(total),
                algo: integrity.map_or(HashAlgo::Sha1, |item| item.algo),
                hash: integrity.map_or_else(String::new, |item| item.hex.clone()),
                saved_at_ms: now_ms(),
                chunks: Some(done),
            },
            meta_path,
        );

        let counter = Arc::new(AtomicU64::new(reused));
        let reporter = self.spawn_reporter(id, Arc::clone(&counter), total);

        let deadlines = self.cfg.deadlines();
        let todo = chunk_resume::remaining(&ranges, journal.done());
        let mut set = tokio::task::JoinSet::new();
        for range in todo {
            let client = self.http.clone();
            let url = candidate.url.clone();
            let path = part.path().to_path_buf();
            let progress = Arc::clone(&counter);
            let if_range = if_range.clone();
            set.spawn(async move {
                chunked::fetch_chunk(
                    &client,
                    &url,
                    range,
                    &path,
                    deadlines,
                    &progress,
                    if_range.as_deref(),
                )
                .await
            });
        }

        let failure = self
            .await_chunks(&mut set, &counter, total, &candidate.url, &mut journal)
            .await;
        reporter.abort();

        if let Some(err) = failure {
            // 已完成的段连同自证摘要都在元数据里（`ChunkJournal` 每段落一次盘），中间文件留着，
            // 下一次（甚至下次启动）接着下。只有「这个源不支持 Range」才把文件丢掉：
            // 单连接路径接不住预分配 + 乱序段的形状。
            if err.kind == DownloadErrorKind::RangeUnsupported {
                part.discard();
            }
            return Err(err.with_source(candidate.source.as_str()));
        }

        // 分块是乱序写，只能整文件重算一次摘要——这一步覆盖新下的字节**与**复用的旧字节。
        self.verify_part(part, integrity, Some(total), id).await?;
        self.emit(ProgressEvent::Progress {
            id,
            downloaded: total,
            total_bytes: Some(total),
        });
        Ok(AttemptOk {
            bytes: total,
            resumed_from: reused,
        })
    }

    /// 回收分块任务，同时盯着「整个作业有没有在动」。
    ///
    /// 两道判据必须分得开（另一处是单块自己的读空闲超时，阶段 = 等字节）：
    /// * **无进度**：`counter` 连续 `chunk_stall_timeout` 没有新字节。分块任务卡在写盘上
    ///   （杀软/网络盘）时单块读超时不会响，只有这道抓得住——现场留给 `join_next` 的正是
    ///   「188MB 落完 99% 之后永久挂住」那个形态；
    /// * **总期限**：`chunk_job_budget`（按平均吞吐下限反推，见 `stall` 模块头）用完。
    ///
    /// 两者都归到阶段 [`StallStage::AwaitChunkJoin`]，靠 [`StallReason`] 区分。
    ///
    /// 成功的段在这里**顺手记账**（[`ChunkJournal::record`]）：这是唯一同时握着「段结果」与
    /// 「段区间」的地方，落到元数据之后，本作业后面怎么挂都不影响这几段的续传价值。
    async fn await_chunks(
        &self,
        set: &mut tokio::task::JoinSet<Result<ChunkFetch, DownloadError>>,
        counter: &AtomicU64,
        total: u64,
        url: &str,
        journal: &mut ChunkJournal,
    ) -> Option<DownloadError> {
        let budget = self.cfg.chunk_job_budget(Some(total));
        let hard_deadline = tokio::time::Instant::now() + budget;
        let mut last_seen = 0_u64;
        let mut last_progress = tokio::time::Instant::now();
        loop {
            let seen = counter.load(Ordering::Relaxed);
            if seen > last_seen {
                last_seen = seen;
                last_progress = tokio::time::Instant::now();
            }
            let wake = (last_progress + self.cfg.chunk_stall_timeout).min(hard_deadline);
            match tokio::time::timeout_at(wake, set.join_next()).await {
                Ok(Some(Ok(Ok(fetched)))) => {
                    // 「这个源给的还是不是同一个文件」：`Content-Range` 的总长必须与我们规划的
                    // 一致，否则写下去的段根本拼不成目标文件（单连接路径同一处判据是收尾的
                    // `written != total`）。
                    if fetched.total.is_some_and(|server| server != total) {
                        set.abort_all();
                        return Some(DownloadError::new(
                            DownloadErrorKind::Network,
                            format!(
                                "分块总长不符：响应声明 {}，本次规划 {total}（{url}）",
                                fetched.total.unwrap_or_default()
                            ),
                        ));
                    }
                    journal.record(fetched);
                }
                Ok(Some(Ok(Err(err)))) => {
                    set.abort_all();
                    return Some(err);
                }
                Ok(Some(Err(err))) => {
                    set.abort_all();
                    return Some(DownloadError::new(
                        DownloadErrorKind::Network,
                        format!("分块任务异常：{err}"),
                    ));
                }
                // 全部回收（成功）。
                Ok(None) => return None,
                Err(_) => {
                    let now = tokio::time::Instant::now();
                    // 窗口内确实有字节进来 ⇒ 续期（「慢但有效」不被误杀的机制就在这一行）。
                    let seen = counter.load(Ordering::Relaxed);
                    if now < hard_deadline && seen > last_seen {
                        last_seen = seen;
                        last_progress = now;
                        continue;
                    }
                    let report = if now >= hard_deadline {
                        StallReport::deadline(
                            StallStage::AwaitChunkJoin,
                            seen,
                            Some(total),
                            now - last_progress,
                        )
                    } else {
                        StallReport::no_progress(
                            StallStage::AwaitChunkJoin,
                            seen,
                            Some(total),
                            now - last_progress,
                        )
                    };
                    set.abort_all();
                    let context = match report.reason {
                        StallReason::Deadline => {
                            format!("分块作业超过总期限 {}s（{url}）", budget.as_secs())
                        }
                        StallReason::NoProgress => format!("分块作业长时间没有新字节（{url}）"),
                    };
                    return Some(DownloadError::stalled(report, context));
                }
            }
        }
    }

    /// 旧分块自证：按记录逐段重读、重算摘要。
    ///
    /// `true` = 每一段都对得上，可以复用；`false` = 证明不了（摘要不符 / 盘上不够长 / 读盘失败 /
    /// 超时），调用方必须整份丢掉。超时也归到「证明不了」：`spawn_blocking` 的线程不可取消，
    /// 拿不到结果就不能拿旧字节去拼文件——宁可再生一遍字节。
    async fn verify_chunks(&self, part: &PartFile, records: &[ChunkRecord]) -> bool {
        let path = part.path().to_path_buf();
        let records = records.to_vec();
        let job = tokio::task::spawn_blocking(move || {
            chunk_resume::verify_records(&path, &records).is_ok()
        });
        matches!(
            tokio::time::timeout(self.cfg.verify_timeout, job).await,
            Ok(Ok(true))
        )
    }

    fn spawn_reporter(
        &self,
        id: DownloadId,
        counter: Arc<AtomicU64>,
        total: u64,
    ) -> tokio::task::JoinHandle<()> {
        let emitter = self.progress.clone();
        let interval = self.cfg.progress_throttle.max(Duration::from_millis(50));
        tokio::spawn(async move {
            let mut ticker = tokio::time::interval(interval);
            loop {
                ticker.tick().await;
                let downloaded = counter.load(Ordering::Relaxed).min(total);
                emitter.emit(ProgressEvent::Progress {
                    id,
                    downloaded,
                    total_bytes: Some(total),
                });
            }
        })
    }

    pub(crate) async fn verify_part(
        &self,
        part: &mut PartFile,
        integrity: Option<&Integrity>,
        expected: Option<u64>,
        id: DownloadId,
    ) -> Result<(), DownloadError> {
        let Some(integrity) = integrity else {
            return Ok(());
        };
        self.emit(ProgressEvent::Verifying {
            id,
            algo: integrity.algo,
            expected_prefix: integrity.short_prefix().to_string(),
        });
        // 整文件重算是同步阻塞 IO：挪到 blocking 池并套上界，否则「校验阶段卡住」
        // （网络盘掉线、杀软锁文件）照样是永久挂住，且它发生在字节全部收完之后，
        // 从进度上看与「下载完成」一模一样。
        //
        // 代价（复审 r1 p3-1，未验）：`spawn_blocking` 的线程**不可取消**——到点后它还在
        // 读文件、还持着句柄，重试会再堆一个（每源最多 3 个）。**假设未验**：Windows 上
        // 那个句柄可能让随后的同名重建（`attempt_chunked` 的 `OpenOptions::create(true)`）
        // 拿到 `Access is denied`，表现为「预分配中间文件失败」而不是干净重试
        // （`PartFile::discard` 忽略错误）。仍然有界、有分类，只是诊断会拐一道。
        let on_disk = part.len();
        let path = part.path().to_path_buf();
        let algo = integrity.algo;
        let digest = tokio::time::timeout(
            self.cfg.verify_timeout,
            tokio::task::spawn_blocking(move || verify::digest_file(&path, algo)),
        )
        .await
        .map_err(|_| {
            DownloadError::stalled(
                StallReport::deadline(
                    StallStage::Verify,
                    on_disk,
                    expected,
                    self.cfg.verify_timeout,
                ),
                format!(
                    "整文件重算摘要超过 {}ms",
                    self.cfg.verify_timeout.as_millis()
                ),
            )
        })?
        .map_err(|err| {
            DownloadError::new(DownloadErrorKind::Io, format!("校验任务异常退出：{err}"))
        })?
        .map_err(|err| io_err("读取中间文件失败", &err))?;
        if verify::matches(&integrity.hex, &digest) {
            return Ok(());
        }
        part.discard();
        Err(DownloadError::new(
            DownloadErrorKind::Integrity,
            format!("校验不符：期望 {} 实得 {digest}", integrity.hex),
        ))
    }
}

fn io_err(context: &str, err: &std::io::Error) -> DownloadError {
    DownloadError::new(DownloadErrorKind::Io, format!("{context}: {err}"))
}