//! Official JP resource download execution.

use crate::curl_transfer::{run_curl_with_retry_with_proxy, CurlProxyConfig, CurlRetryError};
use crate::downloader::{
    DownloadScheduler, DownloaderBackend, DEFAULT_DOWNLOAD_CONCURRENCY, MAX_DOWNLOAD_CONCURRENCY,
    MIN_DOWNLOAD_CONCURRENCY,
};
use crate::official_pull::OfficialResourcePullPlan;
use crate::path_security::{
    ensure_path_within_root, ensure_safe_directory_path, ensure_safe_file_target,
    read_file_no_symlink, validate_output_root, write_file_atomic, STATE_FILE_MODE,
};
use crate::zip_validation::{
    path_has_zip_extension, url_or_path_has_zip_extension, validate_zip_structure,
};
use bat_adapters::official::yostar_jp::YostarJpResourceEndpointKind;
use bat_adapters::official::{
    destination_under_root, DownloadUrlMapper, OfficialResourceBackend, SidecarHashStrategy,
    XxHash32DecimalSeedZero, YostarJpBackend,
};
use serde::{Deserialize, Serialize};
use std::collections::{BTreeMap, HashSet};
use std::fs::{self, File};
use std::io::Read;
use std::path::{Path, PathBuf};
use std::process::Command;
use std::time::{SystemTime, UNIX_EPOCH};

/// 官方资源下载错误：携带统一错误码，便于 CLI/RPC 归类。
///
/// 下载链路内部大量使用 `Result<_, String>`；这些经 `From<String>` 归入
/// `internal`，只有 curl 下载失败等有明确来源的错误会带上准确的网络域码。
#[derive(Debug)]
pub struct DownloadError {
    code: bat_core::ErrorCode,
    message: String,
}

impl DownloadError {
    pub(crate) fn new(code: bat_core::ErrorCode, message: impl Into<String>) -> Self {
        Self {
            code,
            message: message.into(),
        }
    }

    /// 该错误的统一错误码。
    pub fn code(&self) -> bat_core::ErrorCode {
        self.code
    }
}

impl From<String> for DownloadError {
    fn from(message: String) -> Self {
        Self::new(bat_core::ErrorCode::INTERNAL, message)
    }
}

impl std::fmt::Display for DownloadError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(&self.message)
    }
}

impl std::error::Error for DownloadError {}

const DOWNLOAD_MANIFEST_FILE: &str = "official-download-manifest.json";
const DOWNLOAD_QUARANTINE_FILE: &str = "official-download-quarantine.json";
const DOWNLOAD_MANIFEST_VERSION: u32 = 1;
const DOWNLOAD_QUARANTINE_VERSION: u32 = 1;
const DEFAULT_RETRY_ATTEMPTS: usize = 3;

/// Outcome for one official resource pull item.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OfficialResourcePullStatus {
    /// The destination already existed and passed local manifest validation.
    SkippedExisting,
    /// A partial `.part` file was resumed.
    Resumed,
    /// The resource was downloaded from scratch.
    Downloaded,
}

impl OfficialResourcePullStatus {
    /// Returns a stable label for reports and progress logs.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::SkippedExisting => "skipped_existing",
            Self::Resumed => "resumed",
            Self::Downloaded => "downloaded",
        }
    }
}

/// Progress event kind emitted while executing an official pull plan.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OfficialResourcePullProgressKind {
    /// A URL is about to be checked or downloaded.
    Started,
    /// A URL finished as skipped, resumed, or downloaded.
    Finished,
    /// An official sidecar hash pair passed verification.
    Verification,
    /// A URL failed after retry classification and was quarantined for this run.
    Failed,
}

impl OfficialResourcePullProgressKind {
    /// Returns a stable label for progress logs.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Started => "started",
            Self::Finished => "finished",
            Self::Verification => "verification",
            Self::Failed => "failed",
        }
    }
}

/// Local validation results for one completed resource.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct OfficialResourceVerification {
    /// Expected byte size from the local manifest, when reusing a file.
    pub expected_bytes: Option<u64>,
    /// Actual byte size observed after the transfer.
    pub actual_bytes: u64,
    /// Expected BLAKE3 from the local manifest, when reusing a file.
    pub expected_blake3: Option<String>,
    /// Actual BLAKE3 computed from the completed file.
    pub actual_blake3: String,
    /// Whether ZIP structure validation was required for this resource.
    pub zip_checked: bool,
    /// Whether the required ZIP structure validation passed.
    pub zip_structure_verified: bool,
}

/// Progress for one URL in an official pull plan.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OfficialResourcePullProgress {
    /// Progress event kind.
    pub kind: OfficialResourcePullProgressKind,
    /// Monotonic completed item count for the whole pull.
    ///
    /// `Started` events report the currently completed count before the URL
    /// finishes; `Finished` events report the count after completion. This is
    /// intentionally not the URL's plan position, so status percentages stay
    /// monotonic even if execution order or skip/resume mix changes.
    pub index: usize,
    /// Total URL count in the pull plan.
    pub total: usize,
    /// Official URL currently being processed.
    pub url: String,
    /// Finished pull status, when `kind` is `Finished`.
    pub status: Option<OfficialResourcePullStatus>,
    /// Final local file size, when `kind` is `Finished`.
    pub bytes: Option<u64>,
    /// Bytes transferred during this run, when `kind` is `Finished`.
    pub transferred_bytes: Option<u64>,
    /// Local size/BLAKE3/ZIP validation for a completed URL.
    pub verification: Option<OfficialResourceVerification>,
    /// Official sidecar hash verification completed after this URL.
    pub official_hash: Option<OfficialResourceHashVerification>,
    /// Stable failure kind label, when `kind` is `Failed`.
    pub failure_kind: Option<String>,
    /// HTTP status parsed from curl stderr, when available.
    pub failure_http_status: Option<u16>,
    /// Whether retry policy considered the final failure retryable.
    pub failure_retryable: Option<bool>,
    /// Number of curl attempts that were executed before this failure.
    pub failure_attempts: Option<usize>,
    /// Whether the URL was written to the local quarantine manifest.
    pub quarantined: bool,
}

impl OfficialResourcePullProgress {
    fn started(index: usize, total: usize, url: String) -> Self {
        Self {
            kind: OfficialResourcePullProgressKind::Started,
            index,
            total,
            url,
            status: None,
            bytes: None,
            transferred_bytes: None,
            verification: None,
            official_hash: None,
            failure_kind: None,
            failure_http_status: None,
            failure_retryable: None,
            failure_attempts: None,
            quarantined: false,
        }
    }

    fn finished(
        index: usize,
        total: usize,
        url: String,
        status: OfficialResourcePullStatus,
        bytes: u64,
        transferred_bytes: u64,
        verification: OfficialResourceVerification,
    ) -> Self {
        Self {
            kind: OfficialResourcePullProgressKind::Finished,
            index,
            total,
            url,
            status: Some(status),
            bytes: Some(bytes),
            transferred_bytes: Some(transferred_bytes),
            verification: Some(verification),
            official_hash: None,
            failure_kind: None,
            failure_http_status: None,
            failure_retryable: None,
            failure_attempts: None,
            quarantined: false,
        }
    }

    fn verification(
        index: usize,
        total: usize,
        url: String,
        official_hash: OfficialResourceHashVerification,
    ) -> Self {
        Self {
            kind: OfficialResourcePullProgressKind::Verification,
            index,
            total,
            url,
            status: None,
            bytes: None,
            transferred_bytes: None,
            verification: None,
            official_hash: Some(official_hash),
            failure_kind: None,
            failure_http_status: None,
            failure_retryable: None,
            failure_attempts: None,
            quarantined: false,
        }
    }

    fn failed(index: usize, total: usize, url: String, error: &PullOneError) -> Self {
        Self {
            kind: OfficialResourcePullProgressKind::Failed,
            index,
            total,
            url,
            status: None,
            bytes: None,
            transferred_bytes: None,
            verification: None,
            official_hash: None,
            failure_kind: error.failure_kind().map(ToOwned::to_owned),
            failure_http_status: error.http_status(),
            failure_retryable: error.retryable(),
            failure_attempts: error.attempts(),
            quarantined: true,
        }
    }
}

/// Official hash algorithm used by a verified resource sidecar.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum OfficialResourceHashAlgorithm {
    /// Decimal text form of `xxHash32` with seed `0`.
    #[serde(rename = "xxhash32_decimal")]
    XxHash32Decimal,
}

impl OfficialResourceHashAlgorithm {
    /// Returns a stable algorithm label for reports and logs.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::XxHash32Decimal => "xxhash32_decimal",
        }
    }
}

/// Successful official hash verification for one downloaded resource.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct OfficialResourceHashVerification {
    /// Downloaded data URL that was verified.
    pub data_url: String,
    /// Official `.hash` URL used as the verification source.
    pub hash_url: String,
    /// Official hash algorithm.
    pub algorithm: OfficialResourceHashAlgorithm,
    /// Expected hash value read from the official `.hash` file.
    pub expected: String,
    /// Actual hash value computed from the downloaded data file.
    pub actual: String,
}

/// One downloaded official resource.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OfficialResourcePullItem {
    /// Original URL.
    pub url: String,
    /// Local destination path under the output root.
    pub destination: PathBuf,
    /// Final local file size.
    pub bytes: u64,
    /// Bytes transferred during this pull execution.
    pub transferred_bytes: u64,
    /// Pull outcome.
    pub status: OfficialResourcePullStatus,
}

/// Download report for one executed pull plan.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct OfficialResourcePullReport {
    /// Downloaded resources in pull-plan order, independent of completion order.
    pub items: Vec<OfficialResourcePullItem>,
    /// Official hash sidecars successfully verified after download.
    pub verified_hashes: Vec<OfficialResourceHashVerification>,
}

impl OfficialResourcePullReport {
    /// Returns the total downloaded byte count.
    pub fn total_bytes(&self) -> u64 {
        self.items.iter().map(|item| item.bytes).sum()
    }

    /// Returns bytes transferred by this pull execution.
    pub fn transferred_bytes(&self) -> u64 {
        self.items.iter().map(|item| item.transferred_bytes).sum()
    }

    /// Returns the number of resources reused from the local output tree.
    pub fn skipped_count(&self) -> usize {
        self.items
            .iter()
            .filter(|item| item.status == OfficialResourcePullStatus::SkippedExisting)
            .count()
    }

    /// Returns the number of resources resumed from `.part` files.
    pub fn resumed_count(&self) -> usize {
        self.items
            .iter()
            .filter(|item| item.status == OfficialResourcePullStatus::Resumed)
            .count()
    }

    /// Returns the number of resources downloaded from scratch.
    pub fn downloaded_count(&self) -> usize {
        self.items
            .iter()
            .filter(|item| item.status == OfficialResourcePullStatus::Downloaded)
            .count()
    }

    /// Returns the number of official hash sidecars verified after download.
    pub fn official_hash_verified_count(&self) -> usize {
        self.verified_hashes.len()
    }
}

/// Local download-manifest audit status for one expected official URL.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum OfficialLocalManifestAuditStatus {
    /// The local file exists and matches the recorded manifest size and BLAKE3.
    Verified,
    /// No manifest entry exists for this URL.
    MissingManifestEntry,
    /// The manifest entry URL does not match the expected URL.
    UrlMismatch,
    /// The manifest destination does not match the URL-derived destination.
    DestinationMismatch,
    /// The manifest entry exists but the local file is missing.
    MissingFile,
    /// The local file size differs from the manifest entry.
    SizeMismatch,
    /// The local file BLAKE3 differs from the manifest entry.
    Blake3Mismatch,
    /// The local file has a `.zip` URL or destination but its ZIP structure is
    /// invalid.
    ZipStructureInvalid,
}

impl OfficialLocalManifestAuditStatus {
    /// Returns true when this status proves the local file can be reused.
    pub fn is_verified(self) -> bool {
        self == Self::Verified
    }

    /// Returns a stable status label for CLI and JSON reports.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Verified => "verified",
            Self::MissingManifestEntry => "missing_manifest_entry",
            Self::UrlMismatch => "url_mismatch",
            Self::DestinationMismatch => "destination_mismatch",
            Self::MissingFile => "missing_file",
            Self::SizeMismatch => "size_mismatch",
            Self::Blake3Mismatch => "blake3_mismatch",
            Self::ZipStructureInvalid => "zip_structure_invalid",
        }
    }
}

/// Local download-manifest audit result for one expected official URL.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct OfficialLocalManifestAuditItem {
    /// Expected official URL from the pull plan.
    pub url: String,
    /// URL-derived local destination.
    pub destination: PathBuf,
    /// Audit status.
    pub status: OfficialLocalManifestAuditStatus,
    /// Manifest-recorded destination, when an entry exists.
    pub manifest_destination: Option<String>,
    /// Manifest-recorded byte size, when an entry exists.
    pub expected_bytes: Option<u64>,
    /// Actual local byte size, when the destination exists.
    pub actual_bytes: Option<u64>,
    /// Manifest-recorded BLAKE3, when an entry exists.
    pub expected_blake3: Option<String>,
    /// Actual local BLAKE3, when it could be computed.
    pub actual_blake3: Option<String>,
    /// ZIP structure validation error, when the URL or destination is a ZIP
    /// and the local file did not pass ZIP structure validation.
    pub zip_error: Option<String>,
    /// Whether this item required and passed ZIP structure validation.
    pub zip_structure_verified: bool,
}

/// Local download-manifest audit result for a pull plan.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct OfficialLocalManifestAuditReport {
    /// Per-URL audit items in pull-plan order.
    pub items: Vec<OfficialLocalManifestAuditItem>,
}

impl OfficialLocalManifestAuditReport {
    /// Returns true when every expected URL was locally verified.
    pub fn is_clean(&self) -> bool {
        self.items.iter().all(|item| item.status.is_verified())
    }

    /// Returns the number of locally verified files.
    pub fn verified_count(&self) -> usize {
        self.items
            .iter()
            .filter(|item| item.status.is_verified())
            .count()
    }

    /// Returns the number of files that require repair.
    pub fn repair_needed_count(&self) -> usize {
        self.items.len().saturating_sub(self.verified_count())
    }

    /// Returns the number of files that passed local manifest path, size, and
    /// BLAKE3 validation.
    pub fn manifest_blake3_verified_count(&self) -> usize {
        self.items
            .iter()
            .filter(|item| item.status.is_verified())
            .count()
    }

    /// Returns the number of ZIP files that passed structural validation.
    pub fn zip_structure_verified_count(&self) -> usize {
        self.items
            .iter()
            .filter(|item| item.zip_structure_verified)
            .count()
    }
}

/// Full local verification report for all entries currently recorded in the
/// official download manifest.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct OfficialLocalVerificationReport {
    /// Per-entry manifest audit results.
    pub items: Vec<OfficialLocalManifestAuditItem>,
    /// Official seed hash pairs found in the local manifest.
    pub official_hash_pair_count: usize,
    /// Official seed hash pairs that passed xxHash32 verification.
    pub official_hash_verified_count: usize,
    /// Official seed hash verification failures.
    pub official_hash_errors: Vec<String>,
}

impl OfficialLocalVerificationReport {
    /// Returns true when every local manifest entry and every local official
    /// seed hash pair passed verification.
    pub fn is_clean(&self) -> bool {
        self.items.iter().all(|item| item.status.is_verified())
            && self.official_hash_errors.is_empty()
            && self.official_hash_verified_count == self.official_hash_pair_count
    }

    /// Returns the number of manifest entries that passed verification.
    pub fn verified_count(&self) -> usize {
        self.items
            .iter()
            .filter(|item| item.status.is_verified())
            .count()
    }

    /// Returns the number of manifest entries that need attention.
    pub fn failure_count(&self) -> usize {
        self.items.len().saturating_sub(self.verified_count())
    }

    /// Returns the number of local manifest entries that passed path, size,
    /// and BLAKE3 validation.
    pub fn manifest_blake3_verified_count(&self) -> usize {
        self.items
            .iter()
            .filter(|item| item.status.is_verified())
            .count()
    }

    /// Returns the number of ZIP files that passed structural validation.
    pub fn zip_structure_verified_count(&self) -> usize {
        self.items
            .iter()
            .filter(|item| item.zip_structure_verified)
            .count()
    }
}

/// Existing local state for an official pull plan.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct OfficialLocalResourceState {
    /// Number of expected URLs that already have download-manifest entries.
    pub manifest_entry_count: usize,
    /// Number of expected URL destinations that already exist on disk.
    pub existing_file_count: usize,
}

impl OfficialLocalResourceState {
    /// Returns true when this plan has any reusable or auditable local state.
    pub fn has_any_resources(self) -> bool {
        self.manifest_entry_count > 0 || self.existing_file_count > 0
    }
}

/// Executes an official pull plan with the system `curl` command.
#[derive(Debug, Clone)]
pub struct OfficialResourcePullService {
    output_root: PathBuf,
    backend: YostarJpBackend,
    curl_command: PathBuf,
    curl_proxy: CurlProxyConfig,
    retry_attempts: usize,
    max_concurrency: usize,
}

impl OfficialResourcePullService {
    /// Creates a pull service that uses `curl` from `PATH`.
    pub fn new(output_root: impl Into<PathBuf>) -> Self {
        Self::with_curl_command(output_root, "curl")
    }

    /// Creates a pull service with an explicit `curl` binary path.
    pub fn with_curl_command(
        output_root: impl Into<PathBuf>,
        curl_command: impl Into<PathBuf>,
    ) -> Self {
        Self {
            output_root: output_root.into(),
            backend: YostarJpBackend,
            curl_command: curl_command.into(),
            curl_proxy: CurlProxyConfig::default(),
            retry_attempts: DEFAULT_RETRY_ATTEMPTS,
            max_concurrency: DEFAULT_DOWNLOAD_CONCURRENCY,
        }
    }

    /// Sets the proxy configuration for every `curl` transfer.
    pub fn with_proxy_config(mut self, curl_proxy: CurlProxyConfig) -> Self {
        self.curl_proxy = curl_proxy;
        self
    }

    /// Sets the number of attempts for each `curl` transfer.
    pub fn with_retry_attempts(mut self, retry_attempts: usize) -> Self {
        self.retry_attempts = retry_attempts.max(1);
        self
    }

    /// Sets the bounded number of concurrent downloads.
    ///
    /// The default is [`DEFAULT_DOWNLOAD_CONCURRENCY`]. Values are kept
    /// within the supported `1..=256` range; the public update configuration
    /// validates input before constructing this service.
    pub fn with_max_concurrency(mut self, max_concurrency: usize) -> Self {
        self.max_concurrency =
            max_concurrency.clamp(MIN_DOWNLOAD_CONCURRENCY, MAX_DOWNLOAD_CONCURRENCY);
        self
    }

    /// Returns the configured download concurrency.
    pub fn max_concurrency(&self) -> usize {
        self.max_concurrency
    }

    /// Returns the output root used for downloaded files.
    pub fn output_root(&self) -> &Path {
        &self.output_root
    }

    /// Returns the local manifest path used to validate completed downloads.
    pub fn download_manifest_path(&self) -> PathBuf {
        self.output_root.join(DOWNLOAD_MANIFEST_FILE)
    }

    /// Returns the local quarantine path used to diagnose failed downloads.
    pub fn download_quarantine_path(&self) -> PathBuf {
        self.output_root.join(DOWNLOAD_QUARANTINE_FILE)
    }

    /// Removes resource files recorded by an older manifest but absent from
    /// the current pull plan.
    ///
    /// This is intentionally limited to destinations owned by the previous
    /// download manifest. Management files such as snapshots, version state,
    /// and quarantine records are left untouched.
    pub fn prune_stale_manifest_entries(
        &self,
        plan: &OfficialResourcePullPlan,
    ) -> Result<usize, String> {
        self.ensure_output_root_ready()?;
        let expected_urls = plan.all_urls()?.into_iter().collect::<HashSet<_>>();
        let mut manifest = self.read_download_manifest()?;
        let stale_entries = manifest
            .entries
            .iter()
            .filter(|(url, _)| !expected_urls.contains(*url))
            .map(|(url, entry)| (url.clone(), entry.destination.clone()))
            .collect::<Vec<_>>();

        for (url, relative_destination) in &stale_entries {
            let destination = self.output_root.join(relative_destination);
            self.remove_managed_destination(&destination, url)?;
            self.remove_managed_destination(&partial_path_for(&destination), url)?;
            manifest.entries.remove(url);
        }

        if !stale_entries.is_empty() {
            self.write_download_manifest(&manifest)?;
        }

        Ok(stale_entries.len())
    }

    /// Executes the plan and downloads every official URL to disk.
    pub fn pull(
        &self,
        plan: &OfficialResourcePullPlan,
    ) -> Result<OfficialResourcePullReport, DownloadError> {
        self.pull_with_progress(plan, |_| {})
    }

    /// Executes the plan and calls `progress` for every URL before and after it
    /// is checked or downloaded.
    pub fn pull_with_progress(
        &self,
        plan: &OfficialResourcePullPlan,
        mut progress: impl FnMut(OfficialResourcePullProgress),
    ) -> Result<OfficialResourcePullReport, DownloadError> {
        self.pull_with_progress_and_cancellation(plan, &mut progress, || false)
    }

    /// Executes the plan, emits URL progress, and aborts between URLs when
    /// `should_cancel` returns true.
    pub fn pull_with_progress_and_cancellation(
        &self,
        plan: &OfficialResourcePullPlan,
        mut progress: impl FnMut(OfficialResourcePullProgress),
        mut should_cancel: impl FnMut() -> bool,
    ) -> Result<OfficialResourcePullReport, DownloadError> {
        self.ensure_output_root_ready()?;

        let official_hash_pairs = official_seed_hash_pairs(plan);
        let force_refresh_urls = official_hash_refresh_urls(&official_hash_pairs);
        let mut manifest = self.read_download_manifest()?;
        let urls = plan.all_urls()?;
        let total = urls.len();

        // Phase A：无网络的前置校验（fail-fast）。逐个校验官方性、算目标路径、
        // 建目录，并判定该 URL 是「已验证可跳过」还是「需下载」。任何非官方
        // URL 在发起任何下载前即拒绝。
        if should_cancel() {
            return Err("官方资源拉取已被停止请求中断".to_string().into());
        }
        let mut planned: Vec<PlannedDownload> = Vec::with_capacity(total);
        for url in urls {
            if !self.backend.is_official_url(&url) {
                return Err(DownloadError::new(
                    bat_core::ErrorCode::NON_OFFICIAL_URL,
                    format!("拒绝下载非官方 URL：{url}"),
                ));
            }
            let destination = self.destination_for_url(&url)?;
            if let Some(parent) = destination.parent() {
                ensure_safe_directory_path(parent, "下载目标目录")?;
                fs::create_dir_all(parent).map_err(|error| {
                    format!("创建下载目标目录失败 {}：{error}", parent.display())
                })?;
                ensure_safe_directory_path(parent, "下载目标目录")?;
            }
            ensure_safe_file_target(&self.output_root, &destination, "下载目标文件")?;

            let force_refresh = force_refresh_urls.contains(&url);
            let existing = if force_refresh {
                None
            } else {
                self.validated_existing_file(&url, &destination, &manifest)?
            };
            planned.push(PlannedDownload {
                url,
                destination,
                existing,
            });
        }

        if self.max_concurrency > 1 {
            return self.pull_planned_concurrently(
                planned,
                manifest,
                official_hash_pairs,
                total,
                &mut progress,
                &mut should_cancel,
            );
        }

        // Phase B：按 plan 顺序处理每个 URL。下载或复用完成并写入 manifest 后，
        // 立即尝试校验已经到齐的官方 `.bytes/.hash` pair，避免把文件级问题延后到整轮末尾。
        let mut completed_count = 0usize;
        let mut items = Vec::with_capacity(planned.len());
        let mut verified_hashes = Vec::new();
        let mut verified_hash_urls = HashSet::<String>::new();
        let mut processed_urls = HashSet::<String>::new();

        for item in &planned {
            progress(OfficialResourcePullProgress::started(
                completed_count,
                total,
                item.url.clone(),
            ));
            if should_cancel() {
                return Err("官方资源拉取已被停止请求中断".to_string().into());
            }

            let result = if let Some(existing) = &item.existing {
                self.clear_quarantine_entry(&item.url)?;
                existing.clone()
            } else {
                match self.pull_one(&item.url, &item.destination) {
                    Ok(mut pull_result) => {
                        let verification_result = self
                            .clear_quarantine_entry(&item.url)
                            .and_then(|_| {
                                self.record_download_manifest_entry(
                                    &mut manifest,
                                    &item.url,
                                    &item.destination,
                                )
                            })
                            .and_then(|verification| {
                                self.write_download_manifest(&manifest)
                                    .map(|_| verification)
                            });
                        match verification_result {
                            Ok(verification) => {
                                pull_result.verification = verification;
                            }
                            Err(error) => {
                                let error =
                                    PullOneError::plain(format!("记录下载 manifest 失败：{error}"));
                                self.record_quarantine_entry(&item.url, &item.destination, &error)?;
                                progress(OfficialResourcePullProgress::failed(
                                    completed_count,
                                    total,
                                    item.url.clone(),
                                    &error,
                                ));
                                return Err(DownloadError::new(
                                    error.error_code(),
                                    format!(
                                        "官方资源下载失败：URL 已进入 quarantine，中止本轮同步、不发布不完整资源；url={} quarantine={}；{}",
                                        item.url,
                                        self.download_quarantine_path().display(),
                                        error.message
                                    ),
                                ));
                            }
                        }

                        pull_result
                    }
                    Err(error) => {
                        self.record_quarantine_entry(&item.url, &item.destination, &error)?;
                        progress(OfficialResourcePullProgress::failed(
                            completed_count,
                            total,
                            item.url.clone(),
                            &error,
                        ));
                        return Err(DownloadError::new(
                            error.error_code(),
                            format!(
                                "官方资源下载失败：URL 已进入 quarantine，中止本轮同步、不发布不完整资源；url={} quarantine={}；{}",
                                item.url,
                                self.download_quarantine_path().display(),
                                error.message
                            ),
                        ));
                    }
                }
            };

            completed_count += 1;
            progress(OfficialResourcePullProgress::finished(
                completed_count,
                total,
                item.url.clone(),
                result.status,
                result.bytes,
                result.transferred_bytes,
                result.verification.clone(),
            ));
            processed_urls.insert(item.url.clone());
            let newly_verified_hashes = self.verify_ready_official_hashes(
                &official_hash_pairs,
                &processed_urls,
                &mut verified_hash_urls,
                &mut verified_hashes,
                &mut manifest,
            )?;
            for verification in newly_verified_hashes {
                progress(OfficialResourcePullProgress::verification(
                    completed_count,
                    total,
                    verification.data_url.clone(),
                    verification,
                ));
            }
            items.push(OfficialResourcePullItem {
                url: item.url.clone(),
                destination: item.destination.clone(),
                bytes: result.bytes,
                transferred_bytes: result.transferred_bytes,
                status: result.status,
            });
        }

        self.verify_all_official_hashes_are_complete(&official_hash_pairs, &verified_hash_urls)?;

        Ok(OfficialResourcePullReport {
            items,
            verified_hashes,
        })
    }

    fn pull_planned_concurrently(
        &self,
        planned: Vec<PlannedDownload>,
        mut manifest: OfficialDownloadManifest,
        official_hash_pairs: Vec<OfficialSeedHashPair>,
        total: usize,
        progress: &mut impl FnMut(OfficialResourcePullProgress),
        should_cancel: &mut impl FnMut() -> bool,
    ) -> Result<OfficialResourcePullReport, DownloadError> {
        if should_cancel() {
            return Err("官方资源拉取已被停止请求中断".to_string().into());
        }
        for item in &planned {
            progress(OfficialResourcePullProgress::started(
                0,
                total,
                item.url.clone(),
            ));
        }

        let backend = CurlDownloadBackend { service: self };
        let mut completed_count = 0usize;
        let mut items_by_plan_index: Vec<Option<OfficialResourcePullItem>> =
            (0..total).map(|_| None).collect();
        let mut verified_hashes = Vec::new();
        let mut verified_hash_urls = HashSet::<String>::new();
        let mut processed_urls = HashSet::<String>::new();

        DownloadScheduler::new(self.max_concurrency).execute_with_observer(
            &backend,
            planned.clone(),
            |plan_index, result| {
                let item = &planned[plan_index];
                if should_cancel() {
                    return Err("官方资源拉取已被停止请求中断".to_string().into());
                }

                let needs_manifest = item.existing.is_none();
                let verification = match result {
                    Ok(pull_result) => {
                        let verification_result = self
                            .clear_quarantine_entry(&item.url)
                            .and_then(|_| {
                                if needs_manifest {
                                    self.record_download_manifest_entry(
                                        &mut manifest,
                                        &item.url,
                                        &item.destination,
                                    )
                                } else {
                                    Ok(pull_result.verification.clone())
                                }
                            })
                            .and_then(|verification| {
                                if needs_manifest {
                                    self.write_download_manifest(&manifest)
                                        .map(|_| verification)
                                } else {
                                    Ok(verification)
                                }
                            });
                        match verification_result {
                            Ok(verification) => verification,
                            Err(error) => {
                                let error = PullOneError::plain(format!(
                                    "记录下载 manifest 失败：{error}"
                                ));
                                self.record_quarantine_entry(
                                    &item.url,
                                    &item.destination,
                                    &error,
                                )?;
                                progress(OfficialResourcePullProgress::failed(
                                    completed_count,
                                    total,
                                    item.url.clone(),
                                    &error,
                                ));
                                return Err(DownloadError::new(
                                    error.error_code(),
                                    format!(
                                        "官方资源下载失败：URL 已进入 quarantine，中止本轮同步、不发布不完整资源；url={} quarantine={}；{}",
                                        item.url,
                                        self.download_quarantine_path().display(),
                                        error.message
                                    ),
                                ));
                            }
                        }
                    }
                    Err(error) => {
                        self.record_quarantine_entry(&item.url, &item.destination, error)?;
                        progress(OfficialResourcePullProgress::failed(
                            completed_count,
                            total,
                            item.url.clone(),
                            error,
                        ));
                        return Err(DownloadError::new(
                            error.error_code(),
                            format!(
                                "官方资源下载失败：URL 已进入 quarantine，中止本轮同步、不发布不完整资源；url={} quarantine={}；{}",
                                item.url,
                                self.download_quarantine_path().display(),
                                error.message
                            ),
                        ));
                    }
                };

                completed_count += 1;
                progress(OfficialResourcePullProgress::finished(
                    completed_count,
                    total,
                    item.url.clone(),
                    result
                        .as_ref()
                        .expect("successful result handled above")
                        .status,
                    result
                        .as_ref()
                        .expect("successful result handled above")
                        .bytes,
                    result
                        .as_ref()
                        .expect("successful result handled above")
                        .transferred_bytes,
                    verification,
                ));
                processed_urls.insert(item.url.clone());
                let newly_verified_hashes = self.verify_ready_official_hashes(
                    &official_hash_pairs,
                    &processed_urls,
                    &mut verified_hash_urls,
                    &mut verified_hashes,
                    &mut manifest,
                )?;
                for verification in newly_verified_hashes {
                    progress(OfficialResourcePullProgress::verification(
                        completed_count,
                        total,
                        verification.data_url.clone(),
                        verification,
                    ));
                }
                let pull_result = result
                    .as_ref()
                    .expect("successful result handled above");
                items_by_plan_index[plan_index] = Some(OfficialResourcePullItem {
                    url: item.url.clone(),
                    destination: item.destination.clone(),
                    bytes: pull_result.bytes,
                    transferred_bytes: pull_result.transferred_bytes,
                    status: pull_result.status,
                });
                Ok(())
            },
        )?;

        self.verify_all_official_hashes_are_complete(&official_hash_pairs, &verified_hash_urls)?;
        let items = items_by_plan_index
            .into_iter()
            .enumerate()
            .map(|(index, item)| {
                item.ok_or_else(|| {
                    DownloadError::from(format!(
                        "并发下载结果缺少 plan index={index}，拒绝发布不完整资源"
                    ))
                })
            })
            .collect::<Result<Vec<_>, _>>()?;
        Ok(OfficialResourcePullReport {
            items,
            verified_hashes,
        })
    }

    /// Audits every URL in a pull plan against the local download manifest.
    ///
    /// This performs no network I/O. It checks that each URL has a manifest
    /// entry, that the entry maps to the URL-derived destination, and that the
    /// local file still matches the recorded size and BLAKE3 digest.
    pub fn audit_local_manifest(
        &self,
        plan: &OfficialResourcePullPlan,
    ) -> Result<OfficialLocalManifestAuditReport, String> {
        let manifest = self.read_download_manifest()?;
        let mut items = Vec::new();

        for url in plan.all_urls()? {
            let destination = self.destination_for_url(&url)?;
            items.push(self.audit_one(&url, destination, &manifest)?);
        }

        Ok(OfficialLocalManifestAuditReport { items })
    }

    /// Verifies every entry currently present in the local download manifest.
    ///
    /// This is intentionally network-free. It validates each recorded file
    /// using the local manifest's size and BLAKE3 digest, then verifies every
    /// local seed `.bytes`/`.hash` pair that is present using the official
    /// decimal xxHash32 rule.
    pub fn verify_local_download_manifest(
        &self,
    ) -> Result<OfficialLocalVerificationReport, String> {
        let manifest = self.read_download_manifest()?;
        let urls = manifest.entries.keys().cloned().collect::<Vec<_>>();
        let mut items = Vec::with_capacity(urls.len());

        for url in urls {
            let destination = self.destination_for_url(&url)?;
            items.push(self.audit_one(&url, destination, &manifest)?);
        }

        let pairs = local_manifest_seed_hash_pairs(&manifest);
        let mut official_hash_verified_count = 0;
        let mut official_hash_errors = Vec::new();
        for pair in &pairs {
            match self.verify_official_seed_hash_pair_from_disk(pair) {
                Ok(_) => official_hash_verified_count += 1,
                Err(error) => official_hash_errors.push(error),
            }
        }

        Ok(OfficialLocalVerificationReport {
            items,
            official_hash_pair_count: pairs.len(),
            official_hash_verified_count,
            official_hash_errors,
        })
    }

    /// Returns whether the current output root already contains local state for
    /// this plan. This is intentionally cheaper than a full audit and is used
    /// to choose between audit-then-repair and first-time full pull flows.
    pub fn local_resource_state(
        &self,
        plan: &OfficialResourcePullPlan,
    ) -> Result<OfficialLocalResourceState, String> {
        let manifest = self.read_download_manifest()?;
        let mut manifest_entry_count = 0;
        let mut existing_file_count = 0;

        for url in plan.all_urls()? {
            if manifest.entries.contains_key(&url) {
                manifest_entry_count += 1;
            }

            let destination = self.destination_for_url(&url)?;
            if file_len_if_exists(&destination)?.is_some() {
                existing_file_count += 1;
            }
        }

        Ok(OfficialLocalResourceState {
            manifest_entry_count,
            existing_file_count,
        })
    }

    /// Fetches an official URL into memory.
    ///
    /// This is intended for small discovery inputs such as `server-info`,
    /// `TableCatalog.bytes`, `BundlePackingInfo.bytes`, and
    /// `MediaCatalog.bytes`.
    pub fn fetch_bytes(&self, url: &str) -> Result<Vec<u8>, DownloadError> {
        if !self.backend.is_official_url(url) {
            return Err(DownloadError::new(
                bat_core::ErrorCode::NON_OFFICIAL_URL,
                format!("拒绝拉取非官方 URL：{url}"),
            ));
        }

        let output = run_curl_with_retry_with_proxy(
            &self.curl_command,
            url,
            None,
            self.retry_attempts,
            &self.curl_proxy,
            || {
                let mut command = Command::new(&self.curl_command);
                command
                    .arg("--fail")
                    .arg("--location")
                    .arg("--silent")
                    .arg("--show-error")
                    .arg("--url")
                    .arg(url);
                command
            },
        )
        .map_err(|error| {
            // 保留 curl 失败的准确网络域码；进程类失败归 INTERNAL。
            let code = error
                .final_error_code()
                .unwrap_or(bat_core::ErrorCode::INTERNAL);
            DownloadError::new(code, format!("curl 拉取失败 {url}：{error}"))
        })?;

        Ok(output.stdout)
    }

    fn pull_one(&self, url: &str, destination: &Path) -> Result<PullOneResult, PullOneError> {
        ensure_safe_file_target(&self.output_root, destination, "下载目标文件")
            .map_err(PullOneError::plain)?;
        let partial = partial_path_for(destination);
        ensure_safe_file_target(&self.output_root, &partial, "临时下载文件")
            .map_err(PullOneError::plain)?;
        let partial_len_before = file_len_if_exists(&partial)
            .map_err(PullOneError::plain)?
            .unwrap_or_default();
        let resumed = partial_len_before > 0;
        let mut status = if resumed {
            OfficialResourcePullStatus::Resumed
        } else {
            OfficialResourcePullStatus::Downloaded
        };

        if resumed {
            if let Err(error) = self.download_one(url, &partial, true) {
                let _ = fs::remove_file(&partial);
                self.download_one(url, &partial, false)
                    .map_err(|retry_error| {
                        PullOneError::from_retry(
                            format!("续传失败 {url}：{error}；清理后重新下载也失败：{retry_error}"),
                            retry_error,
                        )
                    })?;
                status = OfficialResourcePullStatus::Downloaded;
            } else if let Err(error) = self.validate_zip_if_needed(url, &partial) {
                let _ = fs::remove_file(&partial);
                self.download_one(url, &partial, false)
                    .map_err(|retry_error| {
                        PullOneError::from_retry(
                            format!("续传后的 ZIP 结构校验失败 {url}：{error}；清理后重新下载也失败：{retry_error}"),
                            retry_error,
                        )
                    })?;
                status = OfficialResourcePullStatus::Downloaded;
            }
        } else {
            self.download_one(url, &partial, false).map_err(|error| {
                PullOneError::from_retry(format!("下载失败 {url}：{error}"), error)
            })?;
        }
        if let Err(error) = self.validate_zip_if_needed(url, &partial) {
            let _ = fs::remove_file(&partial);
            return Err(PullOneError::plain(error));
        }

        ensure_safe_file_target(&self.output_root, destination, "下载目标文件")
            .map_err(PullOneError::plain)?;
        if file_len_if_exists(destination)
            .map_err(PullOneError::plain)?
            .is_some()
        {
            fs::remove_file(destination)
                .map_err(|error| format!("替换已有目标文件失败 {}：{error}", destination.display()))
                .map_err(PullOneError::plain)?;
        }
        ensure_safe_file_target(&self.output_root, destination, "下载目标文件")
            .map_err(PullOneError::plain)?;

        fs::rename(&partial, destination)
            .map_err(|error| {
                format!(
                    "移动临时下载文件失败 {} -> {}：{error}",
                    partial.display(),
                    destination.display()
                )
            })
            .map_err(PullOneError::plain)?;

        let bytes = file_len_if_exists(destination)
            .map_err(PullOneError::plain)?
            .ok_or_else(|| format!("下载完成后目标文件缺失 {}", destination.display()))
            .map_err(PullOneError::plain)?;
        let verification = self
            .local_verification(url, destination, None, None)
            .map_err(PullOneError::plain)?;

        Ok(PullOneResult {
            bytes,
            transferred_bytes: if status == OfficialResourcePullStatus::Resumed {
                bytes.saturating_sub(partial_len_before)
            } else {
                bytes
            },
            status,
            verification,
        })
    }

    fn download_one(
        &self,
        url: &str,
        destination: &Path,
        resume: bool,
    ) -> Result<(), CurlRetryError> {
        run_curl_with_retry_with_proxy(
            &self.curl_command,
            url,
            Some(destination),
            self.retry_attempts,
            &self.curl_proxy,
            || {
                let mut command = Command::new(&self.curl_command);
                command
                    .arg("--fail")
                    .arg("--location")
                    .arg("--silent")
                    .arg("--show-error")
                    .arg("--output")
                    .arg(destination);
                if resume {
                    command.arg("--continue-at").arg("-");
                }
                command.arg("--url").arg(url);
                command
            },
        )
        .map(|_| ())
    }

    fn verify_ready_official_hashes(
        &self,
        pairs: &[OfficialSeedHashPair],
        processed_urls: &HashSet<String>,
        verified_hash_urls: &mut HashSet<String>,
        verified_hashes: &mut Vec<OfficialResourceHashVerification>,
        manifest: &mut OfficialDownloadManifest,
    ) -> Result<Vec<OfficialResourceHashVerification>, String> {
        let mut newly_verified = Vec::new();
        for pair in pairs {
            if verified_hash_urls.contains(&pair.hash_url) {
                continue;
            }

            if !processed_urls.contains(&pair.data_url) || !processed_urls.contains(&pair.hash_url)
            {
                continue;
            }

            if !self.hash_pair_files_exist(pair)? {
                continue;
            }

            let verification = match self.verify_official_seed_hash_pair_from_disk(pair) {
                Ok(verification) => verification,
                Err(error) => {
                    if let Err(cleanup_error) =
                        self.invalidate_download_manifest_hash_pair(manifest, pair)
                    {
                        return Err(format!(
                            "{error}；同时清理官方 hash 对应的本地 manifest 条目失败：{cleanup_error}"
                        ));
                    }

                    return Err(error);
                }
            };

            newly_verified.push(verification.clone());
            verified_hashes.push(verification);
            verified_hash_urls.insert(pair.hash_url.clone());
        }

        Ok(newly_verified)
    }

    fn hash_pair_files_exist(&self, pair: &OfficialSeedHashPair) -> Result<bool, String> {
        let data_path = self.destination_for_url(&pair.data_url)?;
        let hash_path = self.destination_for_url(&pair.hash_url)?;
        Ok(file_len_if_exists(&data_path)?.is_some() && file_len_if_exists(&hash_path)?.is_some())
    }

    fn verify_official_seed_hash_pair_from_disk(
        &self,
        pair: &OfficialSeedHashPair,
    ) -> Result<OfficialResourceHashVerification, String> {
        let data_path = self.destination_for_url(&pair.data_url)?;
        let hash_path = self.destination_for_url(&pair.hash_url)?;
        let data = fs::read(&data_path).map_err(|error| {
            format!(
                "读取官方 hash 对应数据文件失败 {}：{error}",
                data_path.display()
            )
        })?;
        let hash = fs::read(&hash_path).map_err(|error| {
            format!(
                "读取官方 hash sidecar 失败 {}：{error}",
                hash_path.display()
            )
        })?;

        verify_official_seed_catalog_hash(&pair.data_url, &data, &pair.hash_url, &hash)
    }

    fn invalidate_download_manifest_hash_pair(
        &self,
        manifest: &mut OfficialDownloadManifest,
        pair: &OfficialSeedHashPair,
    ) -> Result<(), String> {
        manifest.entries.remove(&pair.data_url);
        manifest.entries.remove(&pair.hash_url);
        self.write_download_manifest(manifest)
    }

    fn verify_all_official_hashes_are_complete(
        &self,
        pairs: &[OfficialSeedHashPair],
        verified_hash_urls: &HashSet<String>,
    ) -> Result<(), String> {
        for pair in pairs {
            if !verified_hash_urls.contains(&pair.hash_url) {
                return Err(format!(
                    "官方 hash 对未完成校验：数据={} hash={}",
                    pair.data_url, pair.hash_url
                ));
            }
        }

        Ok(())
    }

    fn read_download_manifest(&self) -> Result<OfficialDownloadManifest, String> {
        self.ensure_output_root_safe()?;
        let path = self.download_manifest_path();
        ensure_safe_file_target(&self.output_root, &path, "下载 manifest")?;
        let Some(bytes) = read_file_no_symlink(&path, "下载 manifest")? else {
            return Ok(OfficialDownloadManifest::default());
        };

        let manifest: OfficialDownloadManifest = serde_json::from_slice(&bytes)
            .map_err(|error| format!("解析下载 manifest 失败 {}：{error}", path.display()))?;

        if manifest.version != DOWNLOAD_MANIFEST_VERSION {
            return Err(format!(
                "不支持的下载 manifest 版本 {}，文件 {}",
                manifest.version,
                path.display()
            ));
        }

        Ok(manifest)
    }

    fn remove_managed_destination(&self, path: &Path, url: &str) -> Result<(), String> {
        ensure_path_within_root(&self.output_root, path)?;
        ensure_safe_file_target(&self.output_root, path, "旧官方资源清理目标")?;

        match fs::symlink_metadata(path) {
            Ok(metadata) if metadata.file_type().is_symlink() => Err(format!(
                "旧官方资源清理目标不能是 symlink：url={url} path={}",
                path.display()
            )),
            Ok(metadata) if metadata.is_file() => {
                fs::remove_file(path).map_err(|error| {
                    format!(
                        "清理旧官方资源失败：url={url} path={} error={error}",
                        path.display()
                    )
                })?;
                remove_empty_parent_directories(&self.output_root, path.parent())?;
                Ok(())
            }
            Ok(metadata) if metadata.is_dir() => Err(format!(
                "旧官方资源清理目标不能是目录：url={url} path={}",
                path.display()
            )),
            Ok(_) => Err(format!(
                "旧官方资源清理目标不是普通文件：url={url} path={}",
                path.display()
            )),
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(()),
            Err(error) => Err(format!(
                "检查旧官方资源清理目标失败：url={url} path={} error={error}",
                path.display()
            )),
        }
    }

    fn write_download_manifest(&self, manifest: &OfficialDownloadManifest) -> Result<(), String> {
        self.ensure_output_root_ready()?;
        let path = self.download_manifest_path();
        if let Some(parent) = path.parent() {
            ensure_safe_directory_path(parent, "下载 manifest 目录")?;
            fs::create_dir_all(parent).map_err(|error| {
                format!("创建下载 manifest 目录失败 {}：{error}", parent.display())
            })?;
            ensure_safe_directory_path(parent, "下载 manifest 目录")?;
        }
        ensure_safe_file_target(&self.output_root, &path, "下载 manifest")?;

        let bytes = serde_json::to_vec_pretty(manifest)
            .map_err(|error| format!("序列化下载 manifest 失败 {}：{error}", path.display()))?;
        write_file_atomic(&path, &bytes, STATE_FILE_MODE, "下载 manifest")?;

        Ok(())
    }

    fn read_download_quarantine(&self) -> Result<OfficialDownloadQuarantineManifest, String> {
        self.ensure_output_root_safe()?;
        let path = self.download_quarantine_path();
        ensure_safe_file_target(&self.output_root, &path, "下载 quarantine")?;
        let Some(bytes) = read_file_no_symlink(&path, "下载 quarantine")? else {
            return Ok(OfficialDownloadQuarantineManifest::default());
        };

        let manifest: OfficialDownloadQuarantineManifest = serde_json::from_slice(&bytes)
            .map_err(|error| format!("解析下载 quarantine 失败 {}：{error}", path.display()))?;

        if manifest.version != DOWNLOAD_QUARANTINE_VERSION {
            return Err(format!(
                "不支持的下载 quarantine 版本 {}，文件 {}",
                manifest.version,
                path.display()
            ));
        }

        Ok(manifest)
    }

    fn write_download_quarantine(
        &self,
        manifest: &OfficialDownloadQuarantineManifest,
    ) -> Result<(), String> {
        self.ensure_output_root_ready()?;
        let path = self.download_quarantine_path();
        if let Some(parent) = path.parent() {
            ensure_safe_directory_path(parent, "下载 quarantine 目录")?;
            fs::create_dir_all(parent).map_err(|error| {
                format!("创建下载 quarantine 目录失败 {}：{error}", parent.display())
            })?;
            ensure_safe_directory_path(parent, "下载 quarantine 目录")?;
        }
        ensure_safe_file_target(&self.output_root, &path, "下载 quarantine")?;

        let bytes = serde_json::to_vec_pretty(manifest)
            .map_err(|error| format!("序列化下载 quarantine 失败 {}：{error}", path.display()))?;
        write_file_atomic(&path, &bytes, STATE_FILE_MODE, "下载 quarantine")?;

        Ok(())
    }

    fn record_quarantine_entry(
        &self,
        url: &str,
        destination: &Path,
        error: &PullOneError,
    ) -> Result<(), String> {
        let mut manifest = self.read_download_quarantine()?;
        manifest.entries.insert(
            url.to_string(),
            OfficialDownloadQuarantineEntry {
                url: url.to_string(),
                destination: self.relative_destination(destination)?,
                failed_at_unix_seconds: unix_seconds_now(),
                attempts: error.attempts().unwrap_or(0),
                failure_kind: error.failure_kind().map(ToOwned::to_owned),
                http_status: error.http_status(),
                retryable: error.retryable(),
                last_error: error.message.clone(),
            },
        );
        self.write_download_quarantine(&manifest)
    }

    fn clear_quarantine_entry(&self, url: &str) -> Result<(), String> {
        let mut manifest = self.read_download_quarantine()?;
        if manifest.entries.remove(url).is_none() {
            return Ok(());
        }
        self.write_download_quarantine(&manifest)
    }

    fn validated_existing_file(
        &self,
        url: &str,
        destination: &Path,
        manifest: &OfficialDownloadManifest,
    ) -> Result<Option<PullOneResult>, String> {
        let Some(entry) = manifest.entries.get(url) else {
            return Ok(None);
        };

        if entry.url != url {
            return Ok(None);
        }

        if entry.destination != self.relative_destination(destination)? {
            return Ok(None);
        }

        let Some(bytes) = file_len_if_exists(destination)? else {
            return Ok(None);
        };

        if bytes != entry.bytes {
            return Ok(None);
        }

        let digest = blake3_file_hex(destination)?;
        if digest != entry.blake3 {
            return Ok(None);
        }

        if let Err(_error) = self.validate_zip_if_needed(url, destination) {
            return Ok(None);
        }

        let verification = self.local_verification(
            url,
            destination,
            Some(entry.bytes),
            Some(entry.blake3.clone()),
        )?;
        Ok(Some(PullOneResult {
            bytes,
            transferred_bytes: 0,
            status: OfficialResourcePullStatus::SkippedExisting,
            verification,
        }))
    }

    fn local_verification(
        &self,
        url: &str,
        destination: &Path,
        expected_bytes: Option<u64>,
        expected_blake3: Option<String>,
    ) -> Result<OfficialResourceVerification, String> {
        let actual_bytes = file_len_if_exists(destination)?
            .ok_or_else(|| format!("读取资源文件信息失败 {}", destination.display()))?;
        let actual_blake3 = blake3_file_hex(destination)?;
        let zip_checked = url_or_path_has_zip_extension(url) || path_has_zip_extension(destination);
        if zip_checked {
            self.validate_zip_if_needed(url, destination)?;
        }

        Ok(OfficialResourceVerification {
            expected_bytes,
            actual_bytes,
            expected_blake3,
            actual_blake3,
            zip_checked,
            zip_structure_verified: zip_checked,
        })
    }

    fn record_download_manifest_entry(
        &self,
        manifest: &mut OfficialDownloadManifest,
        url: &str,
        destination: &Path,
    ) -> Result<OfficialResourceVerification, String> {
        let mut verification = self.local_verification(url, destination, None, None)?;
        let relative_destination = self.relative_destination(destination)?;

        manifest.entries.insert(
            url.to_string(),
            OfficialDownloadManifestEntry {
                url: url.to_string(),
                destination: relative_destination,
                bytes: verification.actual_bytes,
                blake3: verification.actual_blake3.clone(),
            },
        );

        verification.expected_bytes = Some(verification.actual_bytes);
        verification.expected_blake3 = Some(verification.actual_blake3.clone());
        Ok(verification)
    }

    fn validate_zip_if_needed(&self, url: &str, destination: &Path) -> Result<(), String> {
        ensure_safe_file_target(&self.output_root, destination, "ZIP 校验目标")?;
        if !url_or_path_has_zip_extension(url) && !path_has_zip_extension(destination) {
            return Ok(());
        }

        validate_zip_structure(destination).map(|_report| ())
    }

    fn relative_destination(&self, destination: &Path) -> Result<String, String> {
        let relative = destination
            .strip_prefix(&self.output_root)
            .map_err(|error| {
                format!(
                    "目标路径 {} 不在输出目录 {} 下：{error}",
                    destination.display(),
                    self.output_root.display()
                )
            })?;

        Ok(relative.to_string_lossy().replace('\\', "/"))
    }

    fn audit_one(
        &self,
        url: &str,
        destination: PathBuf,
        manifest: &OfficialDownloadManifest,
    ) -> Result<OfficialLocalManifestAuditItem, String> {
        let Some(entry) = manifest.entries.get(url) else {
            let actual_bytes = file_len_if_exists(&destination)?;
            return Ok(OfficialLocalManifestAuditItem {
                url: url.to_string(),
                destination,
                status: OfficialLocalManifestAuditStatus::MissingManifestEntry,
                manifest_destination: None,
                expected_bytes: None,
                actual_bytes,
                expected_blake3: None,
                actual_blake3: None,
                zip_error: None,
                zip_structure_verified: false,
            });
        };

        let manifest_destination = Some(entry.destination.clone());
        let expected_bytes = Some(entry.bytes);
        let expected_blake3 = Some(entry.blake3.clone());

        if entry.url != url {
            return Ok(OfficialLocalManifestAuditItem {
                url: url.to_string(),
                destination,
                status: OfficialLocalManifestAuditStatus::UrlMismatch,
                manifest_destination,
                expected_bytes,
                actual_bytes: None,
                expected_blake3,
                actual_blake3: None,
                zip_error: None,
                zip_structure_verified: false,
            });
        }

        let expected_destination = self.relative_destination(&destination)?;
        if entry.destination != expected_destination {
            return Ok(OfficialLocalManifestAuditItem {
                url: url.to_string(),
                destination,
                status: OfficialLocalManifestAuditStatus::DestinationMismatch,
                manifest_destination,
                expected_bytes,
                actual_bytes: None,
                expected_blake3,
                actual_blake3: None,
                zip_error: None,
                zip_structure_verified: false,
            });
        }

        let Some(actual_bytes) = file_len_if_exists(&destination)? else {
            return Ok(OfficialLocalManifestAuditItem {
                url: url.to_string(),
                destination,
                status: OfficialLocalManifestAuditStatus::MissingFile,
                manifest_destination,
                expected_bytes,
                actual_bytes: None,
                expected_blake3,
                actual_blake3: None,
                zip_error: None,
                zip_structure_verified: false,
            });
        };

        if actual_bytes != entry.bytes {
            return Ok(OfficialLocalManifestAuditItem {
                url: url.to_string(),
                destination,
                status: OfficialLocalManifestAuditStatus::SizeMismatch,
                manifest_destination,
                expected_bytes,
                actual_bytes: Some(actual_bytes),
                expected_blake3,
                actual_blake3: None,
                zip_error: None,
                zip_structure_verified: false,
            });
        }

        let actual_blake3 = blake3_file_hex(&destination)?;
        if actual_blake3 != entry.blake3 {
            return Ok(OfficialLocalManifestAuditItem {
                url: url.to_string(),
                destination,
                status: OfficialLocalManifestAuditStatus::Blake3Mismatch,
                manifest_destination,
                expected_bytes,
                actual_bytes: Some(actual_bytes),
                expected_blake3,
                actual_blake3: Some(actual_blake3),
                zip_error: None,
                zip_structure_verified: false,
            });
        }

        let zip_structure_required =
            url_or_path_has_zip_extension(url) || path_has_zip_extension(&destination);
        if let Err(error) = self.validate_zip_if_needed(url, &destination) {
            return Ok(OfficialLocalManifestAuditItem {
                url: url.to_string(),
                destination,
                status: OfficialLocalManifestAuditStatus::ZipStructureInvalid,
                manifest_destination,
                expected_bytes,
                actual_bytes: Some(actual_bytes),
                expected_blake3,
                actual_blake3: Some(actual_blake3),
                zip_error: Some(error),
                zip_structure_verified: false,
            });
        }

        Ok(OfficialLocalManifestAuditItem {
            url: url.to_string(),
            destination,
            status: OfficialLocalManifestAuditStatus::Verified,
            manifest_destination,
            expected_bytes,
            actual_bytes: Some(actual_bytes),
            expected_blake3,
            actual_blake3: Some(actual_blake3),
            zip_error: None,
            zip_structure_verified: zip_structure_required,
        })
    }

    fn destination_for_url(&self, url: &str) -> Result<PathBuf, String> {
        self.ensure_output_root_safe()?;
        if !self.backend.is_official_url(url) {
            return Err(format!("URL 不是官方 JP host：{url}"));
        }

        let relative_destination = self.backend.relative_destination(url)?;
        let destination = destination_under_root(&self.output_root, &relative_destination)?;
        ensure_path_within_root(&self.output_root, &destination)?;
        Ok(destination)
    }

    fn ensure_output_root_safe(&self) -> Result<(), String> {
        validate_output_root(&self.output_root)
    }

    fn ensure_output_root_ready(&self) -> Result<(), String> {
        self.ensure_output_root_safe()?;
        ensure_safe_directory_path(&self.output_root, "资源输出目录")?;
        fs::create_dir_all(&self.output_root).map_err(|error| {
            format!(
                "创建资源输出目录失败 {}：{error}",
                self.output_root.display()
            )
        })?;
        ensure_safe_directory_path(&self.output_root, "资源输出目录")
    }
}

/// Verifies an official seed catalog `.bytes` payload against its official
/// `.hash` sidecar.
///
/// The currently verified seed catalogs are `TableCatalog.bytes`,
/// `BundlePackingInfo.bytes`, and `MediaCatalog.bytes`; their `.hash` files are
/// decimal text `xxHash32` values with seed `0`.
pub fn verify_official_seed_catalog_hash(
    data_url: &str,
    data: &[u8],
    hash_url: &str,
    hash_bytes: &[u8],
) -> Result<OfficialResourceHashVerification, String> {
    let strategy = XxHash32DecimalSeedZero;
    let verification = strategy
        .verify(data, hash_bytes)
        .map_err(|error| format!("{error}：{hash_url}"))?;

    if verification.expected != verification.actual {
        return Err(format!(
            "官方 hash 校验失败 {data_url}：期望 {}（来自 {hash_url}），实际 {}",
            verification.expected, verification.actual
        ));
    }

    Ok(OfficialResourceHashVerification {
        data_url: data_url.to_string(),
        hash_url: hash_url.to_string(),
        algorithm: OfficialResourceHashAlgorithm::XxHash32Decimal,
        expected: verification.expected,
        actual: verification.actual,
    })
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct OfficialSeedHashPair {
    data_url: String,
    hash_url: String,
}

fn official_seed_hash_pairs(plan: &OfficialResourcePullPlan) -> Vec<OfficialSeedHashPair> {
    plan.discovery
        .endpoints
        .iter()
        .filter_map(|endpoint| {
            let hash_kind = strong_seed_hash_kind(endpoint.kind)?;
            let hash_endpoint = plan.discovery.endpoints.iter().find(|candidate| {
                candidate.kind == hash_kind && candidate.platform == endpoint.platform
            })?;

            Some(OfficialSeedHashPair {
                data_url: endpoint.url.clone(),
                hash_url: hash_endpoint.url.clone(),
            })
        })
        .collect()
}

fn local_manifest_seed_hash_pairs(
    manifest: &OfficialDownloadManifest,
) -> Vec<OfficialSeedHashPair> {
    manifest
        .entries
        .keys()
        .filter_map(|data_url| {
            let hash_url = data_url
                .strip_suffix(".bytes")
                .map(|prefix| format!("{prefix}.hash"))?;
            if !manifest.entries.contains_key(&hash_url) {
                return None;
            }

            Some(OfficialSeedHashPair {
                data_url: data_url.clone(),
                hash_url,
            })
        })
        .collect()
}

fn official_hash_refresh_urls(pairs: &[OfficialSeedHashPair]) -> HashSet<String> {
    let mut urls = HashSet::new();
    for pair in pairs {
        urls.insert(pair.data_url.clone());
        urls.insert(pair.hash_url.clone());
    }
    urls
}

fn strong_seed_hash_kind(
    kind: YostarJpResourceEndpointKind,
) -> Option<YostarJpResourceEndpointKind> {
    match kind {
        YostarJpResourceEndpointKind::TableCatalog => {
            Some(YostarJpResourceEndpointKind::TableCatalogHash)
        }
        YostarJpResourceEndpointKind::BundlePackingInfo => {
            Some(YostarJpResourceEndpointKind::BundlePackingInfoHash)
        }
        YostarJpResourceEndpointKind::MediaCatalog => {
            Some(YostarJpResourceEndpointKind::MediaCatalogHash)
        }
        YostarJpResourceEndpointKind::AddressablesCatalog
        | YostarJpResourceEndpointKind::TableCatalogHash
        | YostarJpResourceEndpointKind::AddressablesCatalogHash
        | YostarJpResourceEndpointKind::BundlePackingInfoHash
        | YostarJpResourceEndpointKind::MediaCatalogHash => None,
    }
}

/// 官方下载 manifest：按完整 URL 为键记录每个已下载资源的本地校验信息。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct OfficialDownloadManifest {
    /// Manifest 结构版本。
    #[serde(default = "default_download_manifest_version")]
    pub version: u32,
    /// 按 URL 为键的资源条目。
    #[serde(default)]
    pub entries: BTreeMap<String, OfficialDownloadManifestEntry>,
}

impl Default for OfficialDownloadManifest {
    fn default() -> Self {
        Self {
            version: DOWNLOAD_MANIFEST_VERSION,
            entries: BTreeMap::new(),
        }
    }
}

/// 官方下载 manifest 中的单个资源条目。
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct OfficialDownloadManifestEntry {
    /// 官方资源 URL。
    pub url: String,
    /// 相对资源根目录的落盘路径。
    pub destination: String,
    /// 文件字节数。
    pub bytes: u64,
    /// 文件 BLAKE3 摘要（hex）。
    pub blake3: String,
}

/// 读取指定资源根目录下的官方下载 manifest（daemon RPC 只读查询用）。
///
/// 文件缺失返回 `Ok(None)`；符号链接、解析失败或版本不支持返回 `Err`。
pub fn read_download_manifest_at(
    resource_root: &Path,
) -> Result<Option<OfficialDownloadManifest>, String> {
    let path = resource_root.join(DOWNLOAD_MANIFEST_FILE);
    let Some(bytes) = read_file_no_symlink(&path, "下载 manifest")? else {
        return Ok(None);
    };

    let manifest: OfficialDownloadManifest = serde_json::from_slice(&bytes)
        .map_err(|error| format!("解析下载 manifest 失败 {}：{error}", path.display()))?;
    if manifest.version != DOWNLOAD_MANIFEST_VERSION {
        return Err(format!(
            "不支持的下载 manifest 版本 {}，文件 {}",
            manifest.version,
            path.display()
        ));
    }

    Ok(Some(manifest))
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct OfficialDownloadQuarantineManifest {
    #[serde(default = "default_download_quarantine_version")]
    version: u32,
    #[serde(default)]
    entries: BTreeMap<String, OfficialDownloadQuarantineEntry>,
}

impl Default for OfficialDownloadQuarantineManifest {
    fn default() -> Self {
        Self {
            version: DOWNLOAD_QUARANTINE_VERSION,
            entries: BTreeMap::new(),
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
struct OfficialDownloadQuarantineEntry {
    url: String,
    destination: String,
    failed_at_unix_seconds: u64,
    attempts: usize,
    failure_kind: Option<String>,
    http_status: Option<u16>,
    retryable: Option<bool>,
    last_error: String,
}

fn default_download_manifest_version() -> u32 {
    DOWNLOAD_MANIFEST_VERSION
}

fn default_download_quarantine_version() -> u32 {
    DOWNLOAD_QUARANTINE_VERSION
}

/// Phase A 产出的单个下载计划项：URL、目标路径，以及若命中本地 manifest
/// 校验则带上「已验证可跳过」的结果（`existing`）。
#[derive(Debug, Clone)]
struct PlannedDownload {
    url: String,
    destination: PathBuf,
    existing: Option<PullOneResult>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct PullOneResult {
    bytes: u64,
    transferred_bytes: u64,
    status: OfficialResourcePullStatus,
    verification: OfficialResourceVerification,
}

#[derive(Debug, Clone, PartialEq, Eq)]
struct PullOneError {
    message: String,
    retry_error: Option<CurlRetryError>,
}

struct CurlDownloadBackend<'a> {
    service: &'a OfficialResourcePullService,
}

impl DownloaderBackend<PlannedDownload> for CurlDownloadBackend<'_> {
    type Output = PullOneResult;
    type Error = PullOneError;

    fn download(&self, task: PlannedDownload) -> Result<Self::Output, Self::Error> {
        task.existing
            .map(Ok)
            .unwrap_or_else(|| self.service.pull_one(&task.url, &task.destination))
    }
}

impl PullOneError {
    fn plain(message: String) -> Self {
        Self {
            message,
            retry_error: None,
        }
    }

    fn from_retry(message: String, retry_error: CurlRetryError) -> Self {
        Self {
            message,
            retry_error: Some(retry_error),
        }
    }

    fn failure_kind(&self) -> Option<&'static str> {
        self.retry_error
            .as_ref()
            .and_then(CurlRetryError::final_kind_label)
    }

    /// 映射到统一错误码：curl 重试失败取网络域码，其余（进程/续传/替换等）归 INTERNAL。
    fn error_code(&self) -> bat_core::ErrorCode {
        self.retry_error
            .as_ref()
            .and_then(CurlRetryError::final_error_code)
            .unwrap_or(bat_core::ErrorCode::INTERNAL)
    }

    fn http_status(&self) -> Option<u16> {
        self.retry_error
            .as_ref()
            .and_then(CurlRetryError::final_http_status)
    }

    fn retryable(&self) -> Option<bool> {
        self.retry_error
            .as_ref()
            .and_then(CurlRetryError::final_retryable)
    }

    fn attempts(&self) -> Option<usize> {
        self.retry_error.as_ref().map(CurlRetryError::attempt_count)
    }
}

impl std::fmt::Display for PullOneError {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        formatter.write_str(&self.message)
    }
}

impl std::error::Error for PullOneError {}

fn file_len_if_exists(path: &Path) -> Result<Option<u64>, String> {
    match fs::symlink_metadata(path) {
        Ok(metadata) if metadata.file_type().is_symlink() => {
            Err(format!("拒绝跟随 symlink 文件：{}", path.display()))
        }
        Ok(metadata) if metadata.is_file() => Ok(Some(metadata.len())),
        Ok(_) => Err(format!("期望文件路径，但实际不是文件：{}", path.display())),
        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
        Err(error) => Err(format!("读取路径信息失败 {}：{error}", path.display())),
    }
}

fn partial_path_for(destination: &Path) -> PathBuf {
    let mut partial = destination.as_os_str().to_owned();
    partial.push(".part");
    PathBuf::from(partial)
}

fn remove_empty_parent_directories(root: &Path, parent: Option<&Path>) -> Result<(), String> {
    let mut current = parent;
    while let Some(path) = current {
        if path == root {
            break;
        }

        let metadata = match fs::symlink_metadata(path) {
            Ok(metadata) => metadata,
            Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
                current = path.parent();
                continue;
            }
            Err(error) => {
                return Err(format!(
                    "检查旧官方资源父目录失败 {}：{error}",
                    path.display()
                ));
            }
        };
        if metadata.file_type().is_symlink() {
            return Err(format!(
                "旧官方资源父目录不能是 symlink：{}",
                path.display()
            ));
        }
        if !metadata.is_dir() {
            break;
        }

        let mut entries = fs::read_dir(path)
            .map_err(|error| format!("读取旧官方资源父目录失败 {}：{error}", path.display()))?;
        if entries.next().is_some() {
            break;
        }
        fs::remove_dir(path)
            .map_err(|error| format!("清理旧官方资源空目录失败 {}：{error}", path.display()))?;
        current = path.parent();
    }
    Ok(())
}

fn unix_seconds_now() -> u64 {
    SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .map(|duration| duration.as_secs())
        .unwrap_or(0)
}

fn blake3_file_hex(path: &Path) -> Result<String, String> {
    match fs::symlink_metadata(path) {
        Ok(metadata) if metadata.file_type().is_symlink() => {
            return Err(format!("拒绝读取 symlink 文件：{}", path.display()));
        }
        Ok(metadata) if metadata.is_file() => {}
        Ok(_) => return Err(format!("期望文件路径，但实际不是文件：{}", path.display())),
        Err(error) => return Err(format!("读取路径信息失败 {}：{error}", path.display())),
    }
    let mut file =
        File::open(path).map_err(|error| format!("打开文件失败 {}：{error}", path.display()))?;
    let mut hasher = blake3::Hasher::new();
    let mut buffer = [0u8; 64 * 1024];

    loop {
        let read = file
            .read(&mut buffer)
            .map_err(|error| format!("读取文件失败 {}：{error}", path.display()))?;
        if read == 0 {
            break;
        }
        hasher.update(&buffer[..read]);
    }

    Ok(hasher.finalize().to_hex().to_string())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::official_pull::{
        build_official_pull_plan, build_official_pull_plan_for_platforms, OfficialResourcePullPlan,
    };
    use bat_adapters::official::inventory::{
        YostarJpDownloadInventory, YostarJpPlatformDownloadInventory,
    };
    use bat_adapters::official::yostar_jp::{
        PatchPlatform, YostarJpResourceDiscoveryPlan, YostarJpResourceEndpoint,
        YostarJpResourceEndpointKind,
    };
    use std::fs;
    use tempfile::TempDir;

    fn xxhash32(bytes: &[u8]) -> u32 {
        XxHash32DecimalSeedZero
            .digest(bytes)
            .parse()
            .expect("xxHash32 decimal digest")
    }

    #[test]
    fn fetch_bytes_maps_rejections_to_error_codes() {
        // 非官方 URL：安全边界拒绝。
        let service = OfficialResourcePullService::new("/tmp/unused");
        let error = service
            .fetch_bytes("https://evil.example/server_info.json")
            .unwrap_err();
        assert_eq!(error.code(), bat_core::ErrorCode::NON_OFFICIAL_URL);

        // curl 404：保留网络域码。
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_shell_script(
            &curl_path,
            r#"#!/bin/sh
printf 'curl: (22) The requested URL returned error: 404\n' >&2
exit 22
"#,
        );
        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let error = service
            .fetch_bytes("https://prod-clientpatch.bluearchiveyostar.com/r93_token/missing.bytes")
            .unwrap_err();
        assert_eq!(error.code(), bat_core::ErrorCode::HTTP_NOT_FOUND);
    }

    #[test]
    fn read_download_manifest_at_handles_missing_and_bad_version() {
        let temp = TempDir::new().unwrap();
        // 文件缺失：Ok(None)。
        assert!(read_download_manifest_at(temp.path()).unwrap().is_none());

        // 正常 manifest：读取条目。
        let path = temp.path().join(DOWNLOAD_MANIFEST_FILE);
        fs::write(
            &path,
            br#"{"version":1,"entries":{"https://a":{"url":"https://a","destination":"a","bytes":1,"blake3":"aa"}}}"#,
        )
        .unwrap();
        let manifest = read_download_manifest_at(temp.path()).unwrap().unwrap();
        assert_eq!(manifest.entries.len(), 1);

        // 不支持的版本：Err。
        fs::write(&path, br#"{"version":999,"entries":{}}"#).unwrap();
        assert!(read_download_manifest_at(temp.path()).is_err());
    }

    fn discovery_plan() -> YostarJpResourceDiscoveryPlan {
        YostarJpResourceDiscoveryPlan {
            connection_group_name: "Prod-Audit".to_string(),
            app_version: "1.70.0".to_string(),
            bundle_version: Some("s8tloc7lo3".to_string()),
            addressables_root:
                "https://prod-clientpatch.bluearchiveyostar.com/r93_token".to_string(),
            endpoints: vec![
                YostarJpResourceEndpoint {
                    kind: YostarJpResourceEndpointKind::TableCatalog,
                    platform: None,
                    url: "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes".to_string(),
                },
                YostarJpResourceEndpoint {
                    kind: YostarJpResourceEndpointKind::TableCatalogHash,
                    platform: None,
                    url: "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.hash".to_string(),
                },
            ],
        }
    }

    fn inventory() -> YostarJpDownloadInventory {
        YostarJpDownloadInventory::from_catalog_bytes(
            b"FullPatch_000.zip",
            b"ExcelDB.db ExcelDB.db",
            b"GameData\\Audio\\VOC_JP\\JP_Airi.zip",
        )
    }

    fn platform_inventory() -> YostarJpPlatformDownloadInventory {
        YostarJpPlatformDownloadInventory::from_shared_inventory(
            inventory(),
            &[PatchPlatform::Windows],
        )
    }

    fn fake_curl_script() -> String {
        format!(
            r#"#!/bin/sh
set -eu
out=""
url=""
while [ "$#" -gt 0 ]; do
    case "$1" in
        --output)
            out="$2"
            shift 2
            ;;
        --url)
            url="$2"
            shift 2
            ;;
        *)
            shift
            ;;
    esac
done
        if [ -n "$out" ]; then
    mkdir -p "$(dirname "$out")"
    case "$url" in
        */TableBundles/TableCatalog.hash)
            printf '%s' '{table_catalog_hash}' > "$out"
            ;;
        */TableBundles/TableCatalog.bytes)
            printf '%s' 'ExcelDB.db' > "$out"
            ;;
        */Windows_PatchPack/BundlePackingInfo.hash)
            printf '%s' '{windows_bundle_hash}' > "$out"
            ;;
        */Android_PatchPack/BundlePackingInfo.hash)
            printf '%s' '{android_bundle_hash}' > "$out"
            ;;
        */Windows_PatchPack/BundlePackingInfo.bytes)
            printf '%s' 'FullPatch_000.zip' > "$out"
            ;;
        */Android_PatchPack/BundlePackingInfo.bytes)
            printf '%s' 'FullPatch_001.zip' > "$out"
            ;;
        */MediaResources-Windows/Catalog/MediaCatalog.hash)
            printf '%s' '{windows_media_hash}' > "$out"
            ;;
        */MediaResources/Catalog/MediaCatalog.hash)
            printf '%s' '{android_media_hash}' > "$out"
            ;;
        */MediaResources-Windows/Catalog/MediaCatalog.bytes)
            printf '%s' 'GameData\Audio\VOC_JP\JP_Airi_Win.zip' > "$out"
            ;;
        */MediaResources/Catalog/MediaCatalog.bytes)
            printf '%s' 'GameData\Audio\VOC_JP\JP_Airi_Android.zip' > "$out"
            ;;
        *.zip)
            printf '\120\113\003\004\024\000\000\000\000\000\000\000\000\000\000\000\000\000\002\000\000\000\002\000\000\000\010\000\000\000file.txtok\120\113\001\002\024\000\024\000\000\000\000\000\000\000\000\000\000\000\000\000\002\000\000\000\002\000\000\000\010\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000\000file.txt\120\113\005\006\000\000\000\000\001\000\001\000\066\000\000\000\050\000\000\000\000\000' > "$out"
            ;;
        *)
            printf '%s' "$url" > "$out"
            ;;
    esac
else
    printf '%s' "$url"
fi
"#,
            table_catalog_hash = xxhash32(b"ExcelDB.db"),
            windows_bundle_hash = xxhash32(b"FullPatch_000.zip"),
            android_bundle_hash = xxhash32(b"FullPatch_001.zip"),
            windows_media_hash = xxhash32(b"GameData\\Audio\\VOC_JP\\JP_Airi_Win.zip"),
            android_media_hash = xxhash32(b"GameData\\Audio\\VOC_JP\\JP_Airi_Android.zip"),
        )
    }

    fn fake_resume_curl_script() -> &'static str {
        r#"#!/bin/sh
set -eu
out=""
url=""
resume=0
while [ "$#" -gt 0 ]; do
    case "$1" in
        --output)
            out="$2"
            shift 2
            ;;
        --continue-at)
            resume=1
            shift 2
            ;;
        --url)
            url="$2"
            shift 2
            ;;
        *)
            shift
            ;;
    esac
done
if [ -n "$out" ]; then
    mkdir -p "$(dirname "$out")"
    if [ "$resume" -eq 1 ]; then
        printf '%s' "-resumed" >> "$out"
    else
        printf '%s' "$url" > "$out"
    fi
else
    printf '%s' "$url"
fi
"#
    }

    fn fake_flaky_curl_script() -> &'static str {
        r#"#!/bin/sh
set -eu
out=""
url=""
while [ "$#" -gt 0 ]; do
    case "$1" in
        --output)
            out="$2"
            shift 2
            ;;
        --url)
            url="$2"
            shift 2
            ;;
        *)
            shift
            ;;
    esac
done
state="$0.state"
count=0
if [ -f "$state" ]; then
    count="$(cat "$state")"
fi
count=$((count + 1))
printf '%s' "$count" > "$state"
if [ "$count" -eq 1 ]; then
    printf '%s\n' "temporary failure" >&2
    exit 22
fi
if [ -n "$out" ]; then
    mkdir -p "$(dirname "$out")"
    printf '%s' "$url" > "$out"
else
    printf '%s' "$url"
fi
"#
    }

    fn fake_bad_hash_curl_script() -> &'static str {
        r#"#!/bin/sh
set -eu
out=""
url=""
while [ "$#" -gt 0 ]; do
    case "$1" in
        --output)
            out="$2"
            shift 2
            ;;
        --url)
            url="$2"
            shift 2
            ;;
        *)
            shift
            ;;
    esac
done
if [ -n "$out" ]; then
    mkdir -p "$(dirname "$out")"
    case "$url" in
        */TableBundles/TableCatalog.hash)
            printf '%s' '1' > "$out"
            ;;
        */TableBundles/TableCatalog.bytes)
            printf '%s' 'ExcelDB.db' > "$out"
            ;;
        *)
            printf '%s' "$url" > "$out"
            ;;
    esac
else
    printf '%s' "$url"
fi
"#
    }

    fn fake_bad_zip_curl_script() -> &'static str {
        r#"#!/bin/sh
set -eu
out=""
url=""
while [ "$#" -gt 0 ]; do
    case "$1" in
        --output)
            out="$2"
            shift 2
            ;;
        --url)
            url="$2"
            shift 2
            ;;
        *)
            shift
            ;;
    esac
done
if [ -n "$out" ]; then
    mkdir -p "$(dirname "$out")"
    printf '%s' 'not a zip archive' > "$out"
else
    printf '%s' "$url"
fi
"#
    }

    fn fake_http_error_curl_script(status: u16) -> String {
        format!(
            r#"#!/bin/sh
set -eu
url=""
while [ "$#" -gt 0 ]; do
    case "$1" in
        --url)
            url="$2"
            shift 2
            ;;
        *)
            shift
            ;;
    esac
done
state="$0.state"
count=0
if [ -f "$state" ]; then
    count="$(cat "$state")"
fi
count=$((count + 1))
printf '%s' "$count" > "$state"
printf 'curl: (22) The requested URL returned error: {status}\n' >&2
exit 22
"#
        )
    }

    fn write_shell_script(path: &Path, script: &str) {
        let temp_path = path.with_extension("tmp");
        fs::write(&temp_path, script).unwrap();
        #[cfg(unix)]
        {
            use std::os::unix::fs::PermissionsExt;
            let mut permissions = fs::metadata(&temp_path).unwrap().permissions();
            permissions.set_mode(0o755);
            fs::set_permissions(&temp_path, permissions).unwrap();
        }
        fs::rename(temp_path, path).unwrap();
    }

    fn write_fake_curl(path: &Path) {
        write_shell_script(path, &fake_curl_script());
    }

    fn write_fake_resume_curl(path: &Path) {
        write_shell_script(path, fake_resume_curl_script());
    }

    fn write_fake_flaky_curl(path: &Path) {
        write_shell_script(path, fake_flaky_curl_script());
    }

    fn write_fake_bad_hash_curl(path: &Path) {
        write_shell_script(path, fake_bad_hash_curl_script());
    }

    fn write_fake_bad_zip_curl(path: &Path) {
        write_shell_script(path, fake_bad_zip_curl_script());
    }

    fn write_fake_http_error_curl(path: &Path, status: u16) {
        write_shell_script(path, &fake_http_error_curl_script(status));
    }

    fn one_url_plan(url: &str) -> OfficialResourcePullPlan {
        OfficialResourcePullPlan {
            discovery: YostarJpResourceDiscoveryPlan {
                connection_group_name: "Prod-Audit".to_string(),
                app_version: "1.70.0".to_string(),
                bundle_version: None,
                addressables_root: "https://prod-clientpatch.bluearchiveyostar.com/r93_token"
                    .to_string(),
                endpoints: vec![YostarJpResourceEndpoint {
                    kind: YostarJpResourceEndpointKind::TableCatalog,
                    platform: None,
                    url: url.to_string(),
                }],
            },
            inventory: YostarJpPlatformDownloadInventory::from_shared_inventory(
                YostarJpDownloadInventory::from_catalog_bytes(b"", b"", b""),
                &[],
            ),
            platforms: Vec::new(),
        }
    }

    fn one_file_zip(name: &[u8], data: &[u8]) -> Vec<u8> {
        let mut bytes = Vec::new();
        bytes.extend_from_slice(&0x0403_4b50u32.to_le_bytes());
        bytes.extend_from_slice(&20u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u32.to_le_bytes());
        bytes.extend_from_slice(&(data.len() as u32).to_le_bytes());
        bytes.extend_from_slice(&(data.len() as u32).to_le_bytes());
        bytes.extend_from_slice(&(name.len() as u16).to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(name);
        bytes.extend_from_slice(data);

        let central_offset = bytes.len() as u32;
        bytes.extend_from_slice(&0x0201_4b50u32.to_le_bytes());
        bytes.extend_from_slice(&20u16.to_le_bytes());
        bytes.extend_from_slice(&20u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u32.to_le_bytes());
        bytes.extend_from_slice(&(data.len() as u32).to_le_bytes());
        bytes.extend_from_slice(&(data.len() as u32).to_le_bytes());
        bytes.extend_from_slice(&(name.len() as u16).to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u32.to_le_bytes());
        bytes.extend_from_slice(&0u32.to_le_bytes());
        bytes.extend_from_slice(name);
        let central_size = bytes.len() as u32 - central_offset;

        bytes.extend_from_slice(&0x0605_4b50u32.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes.extend_from_slice(&1u16.to_le_bytes());
        bytes.extend_from_slice(&1u16.to_le_bytes());
        bytes.extend_from_slice(&central_size.to_le_bytes());
        bytes.extend_from_slice(&central_offset.to_le_bytes());
        bytes.extend_from_slice(&0u16.to_le_bytes());
        bytes
    }

    #[test]
    fn downloads_verified_pull_plan_to_disk() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let plan = build_official_pull_plan(discovery_plan(), inventory());
        let report = service.pull(&plan).unwrap();

        assert_eq!(report.items.len(), 7);
        assert!(report.total_bytes() > 0);
        assert_eq!(report.downloaded_count(), report.items.len());
        assert_eq!(report.skipped_count(), 0);
        assert_eq!(report.resumed_count(), 0);
        assert_eq!(report.transferred_bytes(), report.total_bytes());
        assert_eq!(report.items[0].url, plan.discovery.endpoints[0].url);
        for item in &report.items {
            assert!(item.destination.exists());
            assert_eq!(item.status, OfficialResourcePullStatus::Downloaded);
            assert_eq!(item.bytes, item.transferred_bytes);
        }
        assert_eq!(report.official_hash_verified_count(), 1);
        assert_eq!(
            report.verified_hashes[0].algorithm,
            OfficialResourceHashAlgorithm::XxHash32Decimal
        );
        assert_eq!(report.verified_hashes[0].expected, "2044170421");
        assert_eq!(report.verified_hashes[0].actual, "2044170421");

        let manifest = service.read_download_manifest().unwrap();
        assert_eq!(manifest.version, DOWNLOAD_MANIFEST_VERSION);
        assert_eq!(manifest.entries.len(), report.items.len());
        for item in &report.items {
            let entry = manifest.entries.get(&item.url).unwrap();
            assert_eq!(entry.url, item.url);
            assert_eq!(
                entry.destination,
                service.relative_destination(&item.destination).unwrap()
            );
            assert_eq!(entry.bytes, item.bytes);
            assert_eq!(entry.blake3, blake3_file_hex(&item.destination).unwrap());
        }
    }

    #[test]
    fn rejects_dangerous_output_root() {
        let service = OfficialResourcePullService::new(PathBuf::from("/"));
        let error = service
            .destination_for_url(
                "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes",
            )
            .unwrap_err();

        assert!(error.to_string().contains("危险路径"));
    }

    #[test]
    fn rejects_resource_url_with_query_or_fragment() {
        let temp = TempDir::new().unwrap();
        let service = OfficialResourcePullService::new(temp.path().join("out"));

        for url in [
            "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes?v=1",
            "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes#frag",
        ] {
            let error = service.destination_for_url(url).unwrap_err();
            assert!(
                error.contains("query 或 fragment"),
                "unexpected error for {url}: {error}"
            );
        }

        // 无 query 的正常资源 URL 仍可映射。
        assert!(service
            .destination_for_url(
                "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes",
            )
            .is_ok());
    }

    #[cfg(unix)]
    #[test]
    fn pull_rejects_symlink_parent_escape() {
        use std::os::unix::fs::symlink;

        let temp = TempDir::new().unwrap();
        let out_dir = temp.path().join("out");
        let escape_dir = temp.path().join("escape");
        let bin_dir = TempDir::new().unwrap();
        fs::create_dir_all(&out_dir).unwrap();
        fs::create_dir_all(&escape_dir).unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(&out_dir, &curl_path);
        let destination = service
            .destination_for_url(
                "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes",
            )
            .unwrap();
        let host_dir = destination
            .ancestors()
            .find(|path| {
                path.file_name().and_then(|name| name.to_str())
                    == Some("prod-clientpatch.bluearchiveyostar.com")
            })
            .unwrap()
            .to_path_buf();
        fs::create_dir_all(&host_dir).unwrap();
        symlink(&escape_dir, host_dir.join("r93_token")).unwrap();

        let error = service
            .pull(&build_official_pull_plan(discovery_plan(), inventory()))
            .unwrap_err();
        assert!(error.to_string().contains("symlink"));
        assert!(fs::read_dir(&escape_dir).unwrap().next().is_none());
    }

    #[cfg(unix)]
    #[test]
    fn pull_rejects_symlink_partial_file() {
        use std::os::unix::fs::symlink;

        let temp = TempDir::new().unwrap();
        let out_dir = temp.path().join("out");
        let escape_file = temp.path().join("escape.txt");
        let bin_dir = TempDir::new().unwrap();
        fs::create_dir_all(&out_dir).unwrap();
        fs::write(&escape_file, b"outside").unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(&out_dir, &curl_path);
        let url = "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes";
        let destination = service.destination_for_url(url).unwrap();
        fs::create_dir_all(destination.parent().unwrap()).unwrap();
        symlink(&escape_file, partial_path_for(&destination)).unwrap();

        let error = service.pull_one(url, &destination).unwrap_err();
        assert!(error.to_string().contains("symlink"));
        assert_eq!(fs::read(&escape_file).unwrap(), b"outside");
    }

    #[test]
    fn local_manifest_audit_detects_and_repairs_corrupted_file() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let plan = build_official_pull_plan(discovery_plan(), inventory());
        let first_report = service.pull(&plan).unwrap();
        let clean_audit = service.audit_local_manifest(&plan).unwrap();
        assert!(clean_audit.is_clean());
        assert_eq!(clean_audit.verified_count(), first_report.items.len());
        assert_eq!(clean_audit.manifest_blake3_verified_count(), 7);
        assert_eq!(clean_audit.zip_structure_verified_count(), 4);

        let corrupted = first_report
            .items
            .iter()
            .find(|item| item.url.ends_with("/Windows_PatchPack/FullPatch_000.zip"))
            .unwrap();
        let mut corrupted_bytes = fs::read(&corrupted.destination).unwrap();
        corrupted_bytes[0] ^= 0xff;
        fs::write(&corrupted.destination, corrupted_bytes).unwrap();

        let corrupted_audit = service.audit_local_manifest(&plan).unwrap();
        assert!(!corrupted_audit.is_clean());
        assert_eq!(corrupted_audit.repair_needed_count(), 1);
        assert!(corrupted_audit.items.iter().any(|item| {
            item.url == corrupted.url
                && item.status == OfficialLocalManifestAuditStatus::Blake3Mismatch
        }));

        let repair_report = service.pull(&plan).unwrap();
        assert!(repair_report.downloaded_count() >= 1);
        assert!(service.audit_local_manifest(&plan).unwrap().is_clean());
    }

    #[test]
    fn full_local_verification_checks_manifest_and_official_hash_pairs() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let plan = build_official_pull_plan(discovery_plan(), inventory());
        service.pull(&plan).unwrap();

        let verification = service.verify_local_download_manifest().unwrap();
        assert!(verification.is_clean());
        assert_eq!(verification.items.len(), 7);
        assert_eq!(verification.verified_count(), 7);
        assert_eq!(verification.manifest_blake3_verified_count(), 7);
        assert_eq!(verification.zip_structure_verified_count(), 4);
        assert_eq!(verification.official_hash_pair_count, 1);
        assert_eq!(verification.official_hash_verified_count, 1);

        let data_path = service
            .destination_for_url(
                "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes",
            )
            .unwrap();
        fs::write(data_path, b"corrupted").unwrap();

        let corrupted = service.verify_local_download_manifest().unwrap();
        assert!(!corrupted.is_clean());
        assert_eq!(corrupted.failure_count(), 1);
        assert_eq!(corrupted.official_hash_errors.len(), 1);
    }

    #[test]
    fn refreshes_official_hash_pairs_and_skips_manifest_verified_content() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);
        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let plan = build_official_pull_plan(discovery_plan(), inventory());
        let mut manifest = OfficialDownloadManifest::default();

        for url in plan.all_urls().unwrap() {
            let destination = service.destination_for_url(&url).unwrap();
            fs::create_dir_all(destination.parent().unwrap()).unwrap();
            let bytes = if url.ends_with("/TableBundles/TableCatalog.bytes") {
                b"ExcelDB.db".to_vec()
            } else if url.ends_with("/TableBundles/TableCatalog.hash") {
                b"2044170421".to_vec()
            } else if url.ends_with(".zip") {
                one_file_zip(b"fixture.txt", b"ok")
            } else {
                url.as_bytes().to_vec()
            };
            fs::write(&destination, bytes).unwrap();
            service
                .record_download_manifest_entry(&mut manifest, &url, &destination)
                .unwrap();
        }
        service.write_download_manifest(&manifest).unwrap();

        let report = service.pull(&plan).unwrap();

        assert_eq!(report.items.len(), 7);
        assert_eq!(report.skipped_count(), 5);
        assert_eq!(report.downloaded_count(), 2);
        assert_eq!(report.resumed_count(), 0);
        assert!(report.transferred_bytes() > 0);
        assert_eq!(report.official_hash_verified_count(), 1);
        assert_eq!(
            fs::read_to_string(
                service
                    .destination_for_url(
                        "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes",
                    )
                    .unwrap()
            )
            .unwrap(),
            "ExcelDB.db"
        );
    }

    #[test]
    fn redownloads_existing_files_without_manifest() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let plan = build_official_pull_plan(discovery_plan(), inventory());

        for url in plan.all_urls().unwrap() {
            let destination = service.destination_for_url(&url).unwrap();
            fs::create_dir_all(destination.parent().unwrap()).unwrap();
            fs::write(destination, b"stale-local-file").unwrap();
        }

        let report = service.pull(&plan).unwrap();

        assert_eq!(report.items.len(), 7);
        assert_eq!(report.skipped_count(), 0);
        assert_eq!(report.downloaded_count(), 7);
        assert_eq!(report.resumed_count(), 0);
        assert_eq!(report.official_hash_verified_count(), 1);
        assert!(report.transferred_bytes() > 0);
        assert!(report
            .items
            .iter()
            .all(|item| item.status == OfficialResourcePullStatus::Downloaded));
    }

    #[test]
    fn local_resource_state_reports_manifest_entries_and_existing_files() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let plan = build_official_pull_plan(discovery_plan(), inventory());
        let urls = plan.all_urls().unwrap();

        let empty_state = service.local_resource_state(&plan).unwrap();
        assert_eq!(empty_state.manifest_entry_count, 0);
        assert_eq!(empty_state.existing_file_count, 0);
        assert!(!empty_state.has_any_resources());

        let first_destination = service.destination_for_url(&urls[0]).unwrap();
        fs::create_dir_all(first_destination.parent().unwrap()).unwrap();
        fs::write(&first_destination, b"local-only").unwrap();

        let file_only_state = service.local_resource_state(&plan).unwrap();
        assert_eq!(file_only_state.manifest_entry_count, 0);
        assert_eq!(file_only_state.existing_file_count, 1);
        assert!(file_only_state.has_any_resources());

        let mut manifest = OfficialDownloadManifest::default();
        service
            .record_download_manifest_entry(&mut manifest, &urls[0], &first_destination)
            .unwrap();
        service.write_download_manifest(&manifest).unwrap();

        let manifest_state = service.local_resource_state(&plan).unwrap();
        assert_eq!(manifest_state.manifest_entry_count, 1);
        assert_eq!(manifest_state.existing_file_count, 1);
        assert!(manifest_state.has_any_resources());
    }

    #[test]
    fn prune_stale_manifest_entries_removes_old_resource_and_partial_only() {
        let out_dir = TempDir::new().unwrap();
        let service = OfficialResourcePullService::new(out_dir.path());
        let plan = build_official_pull_plan(discovery_plan(), inventory());
        let expected_url = plan.all_urls().unwrap().into_iter().next().unwrap();
        let expected_destination = service.destination_for_url(&expected_url).unwrap();
        fs::create_dir_all(expected_destination.parent().unwrap()).unwrap();
        fs::write(&expected_destination, b"keep").unwrap();

        let stale_url =
            "https://prod-clientpatch.bluearchiveyostar.com/r93_token/obsolete/old.bundle";
        let stale_destination = service.destination_for_url(stale_url).unwrap();
        fs::create_dir_all(stale_destination.parent().unwrap()).unwrap();
        fs::write(&stale_destination, b"remove").unwrap();
        fs::write(partial_path_for(&stale_destination), b"remove-partial").unwrap();

        let mut manifest = OfficialDownloadManifest::default();
        service
            .record_download_manifest_entry(&mut manifest, &expected_url, &expected_destination)
            .unwrap();
        manifest.entries.insert(
            stale_url.to_string(),
            OfficialDownloadManifestEntry {
                url: stale_url.to_string(),
                destination: service.relative_destination(&stale_destination).unwrap(),
                bytes: 6,
                blake3: "stale".to_string(),
            },
        );
        service.write_download_manifest(&manifest).unwrap();

        let snapshot_path = out_dir.path().join("official-sync-snapshot.json");
        let quarantine_path = out_dir.path().join(DOWNLOAD_QUARANTINE_FILE);
        fs::write(&snapshot_path, b"snapshot").unwrap();
        fs::write(&quarantine_path, b"quarantine").unwrap();

        assert_eq!(service.prune_stale_manifest_entries(&plan).unwrap(), 1);
        assert!(!stale_destination.exists());
        assert!(!partial_path_for(&stale_destination).exists());
        assert!(!stale_destination.parent().unwrap().exists());
        assert!(expected_destination.exists());
        assert_eq!(fs::read(&snapshot_path).unwrap(), b"snapshot");
        assert_eq!(fs::read(&quarantine_path).unwrap(), b"quarantine");

        let pruned_manifest = service.read_download_manifest().unwrap();
        assert!(pruned_manifest.entries.contains_key(&expected_url));
        assert!(!pruned_manifest.entries.contains_key(stale_url));
    }

    #[cfg(unix)]
    #[test]
    fn prune_stale_manifest_entries_rejects_symlink_destination() {
        use std::os::unix::fs::symlink;

        let out_dir = TempDir::new().unwrap();
        let service = OfficialResourcePullService::new(out_dir.path());
        let plan = build_official_pull_plan(discovery_plan(), inventory());
        let stale_url =
            "https://prod-clientpatch.bluearchiveyostar.com/r93_token/obsolete/old.bundle";
        let stale_destination = service.destination_for_url(stale_url).unwrap();
        fs::create_dir_all(stale_destination.parent().unwrap()).unwrap();
        let outside = out_dir.path().join("outside-resource");
        fs::write(&outside, b"must-keep").unwrap();
        symlink(&outside, &stale_destination).unwrap();

        let mut manifest = OfficialDownloadManifest::default();
        manifest.entries.insert(
            stale_url.to_string(),
            OfficialDownloadManifestEntry {
                url: stale_url.to_string(),
                destination: service.relative_destination(&stale_destination).unwrap(),
                bytes: 9,
                blake3: "stale".to_string(),
            },
        );
        service.write_download_manifest(&manifest).unwrap();

        let error = service.prune_stale_manifest_entries(&plan).unwrap_err();
        assert!(error.contains("symlink"));
        assert!(stale_destination.exists());
        assert_eq!(fs::read(&outside).unwrap(), b"must-keep");
        assert!(service
            .read_download_manifest()
            .unwrap()
            .entries
            .contains_key(stale_url));
    }

    #[test]
    fn redownloads_existing_file_when_manifest_hash_mismatches() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let plan = build_official_pull_plan_for_platforms(
            discovery_plan(),
            inventory(),
            &[PatchPlatform::Windows],
        );

        let first_report = service.pull(&plan).unwrap();
        assert_eq!(first_report.downloaded_count(), first_report.items.len());

        let first_item = &first_report.items[0];
        fs::write(&first_item.destination, b"corrupted").unwrap();

        let second_report = service
            .pull(&OfficialResourcePullPlan {
                discovery: YostarJpResourceDiscoveryPlan {
                    connection_group_name: "Prod-Audit".to_string(),
                    app_version: "1.70.0".to_string(),
                    bundle_version: None,
                    addressables_root: "https://prod-clientpatch.bluearchiveyostar.com/r93_token"
                        .to_string(),
                    endpoints: vec![YostarJpResourceEndpoint {
                        kind: YostarJpResourceEndpointKind::TableCatalog,
                        platform: None,
                        url: first_item.url.clone(),
                    }],
                },
                inventory: YostarJpPlatformDownloadInventory::from_shared_inventory(
                    YostarJpDownloadInventory::from_catalog_bytes(b"", b"", b""),
                    &[],
                ),
                platforms: Vec::new(),
            })
            .unwrap();

        assert_eq!(second_report.items.len(), 1);
        assert_eq!(second_report.skipped_count(), 0);
        assert_eq!(second_report.downloaded_count(), 1);
        assert_eq!(
            fs::read_to_string(&first_item.destination).unwrap(),
            "ExcelDB.db"
        );
    }

    #[test]
    fn rejects_seed_catalog_when_official_hash_mismatches() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_bad_hash_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let plan = OfficialResourcePullPlan {
            discovery: discovery_plan(),
            inventory: YostarJpPlatformDownloadInventory::from_shared_inventory(
                YostarJpDownloadInventory::from_catalog_bytes(b"", b"", b""),
                &[],
            ),
            platforms: Vec::new(),
        };

        let error = service.pull(&plan).unwrap_err();
        assert!(error.to_string().contains("官方 hash 校验失败"));

        let manifest = service.read_download_manifest().unwrap();
        assert!(manifest.entries.is_empty());

        let audit = service.audit_local_manifest(&plan).unwrap();
        assert!(!audit.is_clean());
        assert_eq!(audit.repair_needed_count(), 2);
        assert!(audit
            .items
            .iter()
            .all(|item| { item.status == OfficialLocalManifestAuditStatus::MissingManifestEntry }));
    }

    #[test]
    fn verifies_known_real_seed_catalog_hash_samples() {
        assert_eq!(xxhash32(b""), 46947589);
        assert_eq!(xxhash32(b"ExcelDB.db"), 2044170421);
        assert_eq!(xxhash32(b"FullPatch_000.zip"), 4038880697);
        assert_eq!(
            xxhash32(b"GameData\\Audio\\VOC_JP\\JP_Airi_Win.zip"),
            1865771294
        );
        assert_eq!(xxhash32(b"ExcelDB.db ExcelDB.db"), 2147704569);
    }

    #[test]
    fn rejects_downloaded_zip_when_structure_is_invalid() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_bad_zip_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let zip_url =
            "https://prod-clientpatch.bluearchiveyostar.com/r93_token/Windows_PatchPack/Broken.zip";
        let plan = OfficialResourcePullPlan {
            discovery: YostarJpResourceDiscoveryPlan {
                connection_group_name: "Prod-Audit".to_string(),
                app_version: "1.70.0".to_string(),
                bundle_version: None,
                addressables_root: "https://prod-clientpatch.bluearchiveyostar.com/r93_token"
                    .to_string(),
                endpoints: vec![YostarJpResourceEndpoint {
                    kind: YostarJpResourceEndpointKind::AddressablesCatalog,
                    platform: Some(PatchPlatform::Windows),
                    url: zip_url.to_string(),
                }],
            },
            inventory: YostarJpPlatformDownloadInventory::from_shared_inventory(
                YostarJpDownloadInventory::from_catalog_bytes(b"", b"", b""),
                &[],
            ),
            platforms: Vec::new(),
        };

        let error = service.pull(&plan).unwrap_err();

        assert!(error.to_string().contains("ZIP"));
        assert!(!partial_path_for(&service.destination_for_url(zip_url).unwrap()).exists());
        assert!(service.read_download_manifest().unwrap().entries.is_empty());
    }

    #[test]
    fn local_manifest_audit_rejects_zip_with_matching_size_and_blake3_but_bad_structure() {
        let out_dir = TempDir::new().unwrap();
        let service = OfficialResourcePullService::new(out_dir.path());
        let zip_url =
            "https://prod-clientpatch.bluearchiveyostar.com/r93_token/Windows_PatchPack/Broken.zip";
        let destination = service.destination_for_url(zip_url).unwrap();
        fs::create_dir_all(destination.parent().unwrap()).unwrap();
        fs::write(&destination, b"not a zip archive").unwrap();

        let mut manifest = OfficialDownloadManifest::default();
        let bytes = fs::metadata(&destination).unwrap().len();
        let blake3 = blake3_file_hex(&destination).unwrap();
        manifest.entries.insert(
            zip_url.to_string(),
            OfficialDownloadManifestEntry {
                url: zip_url.to_string(),
                destination: service.relative_destination(&destination).unwrap(),
                bytes,
                blake3,
            },
        );
        service.write_download_manifest(&manifest).unwrap();

        let plan = OfficialResourcePullPlan {
            discovery: YostarJpResourceDiscoveryPlan {
                connection_group_name: "Prod-Audit".to_string(),
                app_version: "1.70.0".to_string(),
                bundle_version: None,
                addressables_root: "https://prod-clientpatch.bluearchiveyostar.com/r93_token"
                    .to_string(),
                endpoints: vec![YostarJpResourceEndpoint {
                    kind: YostarJpResourceEndpointKind::AddressablesCatalog,
                    platform: Some(PatchPlatform::Windows),
                    url: zip_url.to_string(),
                }],
            },
            inventory: YostarJpPlatformDownloadInventory::from_shared_inventory(
                YostarJpDownloadInventory::from_catalog_bytes(b"", b"", b""),
                &[],
            ),
            platforms: Vec::new(),
        };

        let audit = service.audit_local_manifest(&plan).unwrap();
        let verification = service.verify_local_download_manifest().unwrap();

        assert_eq!(
            audit.items[0].status,
            OfficialLocalManifestAuditStatus::ZipStructureInvalid
        );
        assert!(audit.items[0].zip_error.as_ref().unwrap().contains("ZIP"));
        assert!(!verification.is_clean());
        assert_eq!(verification.failure_count(), 1);
    }

    #[test]
    fn resumes_partial_files_before_atomic_completion() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_resume_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let plan = build_official_pull_plan_for_platforms(
            discovery_plan(),
            inventory(),
            &[PatchPlatform::Windows],
        );
        let first_url = plan.all_urls().unwrap().remove(0);
        let destination = service.destination_for_url(&first_url).unwrap();
        fs::create_dir_all(destination.parent().unwrap()).unwrap();
        let partial = partial_path_for(&destination);
        fs::write(&partial, b"partial").unwrap();

        let report = service
            .pull(&OfficialResourcePullPlan {
                discovery: YostarJpResourceDiscoveryPlan {
                    connection_group_name: "Prod-Audit".to_string(),
                    app_version: "1.70.0".to_string(),
                    bundle_version: None,
                    addressables_root: "https://prod-clientpatch.bluearchiveyostar.com/r93_token"
                        .to_string(),
                    endpoints: vec![YostarJpResourceEndpoint {
                        kind: YostarJpResourceEndpointKind::TableCatalog,
                        platform: None,
                        url: first_url,
                    }],
                },
                inventory: YostarJpPlatformDownloadInventory::from_shared_inventory(
                    YostarJpDownloadInventory::from_catalog_bytes(b"", b"", b""),
                    &[],
                ),
                platforms: Vec::new(),
            })
            .unwrap();

        assert_eq!(report.items.len(), 1);
        assert_eq!(report.resumed_count(), 1);
        assert_eq!(report.items[0].status, OfficialResourcePullStatus::Resumed);
        assert_eq!(fs::read_to_string(&destination).unwrap(), "partial-resumed");
        assert!(!partial.exists());

        let manifest = service.read_download_manifest().unwrap();
        assert_eq!(manifest.entries.len(), 1);
        assert!(manifest.entries.contains_key(&report.items[0].url));
    }

    #[test]
    fn pull_emits_started_and_finished_progress_events() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        // 并发下事件顺序不固定，此处只断言与顺序无关的不变量；并发正确性另有
        // downloads_run_concurrently_and_each_url_reports_once 专测。
        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let plan = build_official_pull_plan_for_platforms(
            discovery_plan(),
            inventory(),
            &[PatchPlatform::Windows],
        );
        let expected_urls = plan.all_urls().unwrap().len();
        let mut events = Vec::new();

        let report = service
            .pull_with_progress(&plan, |event| events.push(event))
            .unwrap();

        assert_eq!(report.items.len(), expected_urls);
        let started: Vec<_> = events
            .iter()
            .filter(|event| event.kind == OfficialResourcePullProgressKind::Started)
            .collect();
        let finished: Vec<_> = events
            .iter()
            .filter(|event| event.kind == OfficialResourcePullProgressKind::Finished)
            .collect();
        let verifications: Vec<_> = events
            .iter()
            .filter(|event| event.kind == OfficialResourcePullProgressKind::Verification)
            .collect();
        assert_eq!(started.len(), expected_urls);
        assert_eq!(finished.len(), expected_urls);
        assert_eq!(verifications.len(), report.verified_hashes.len());
        for verification_event in &verifications {
            let hash = verification_event
                .official_hash
                .as_ref()
                .expect("verification event must carry official hash detail");
            let pair_finished_indices = [hash.data_url.as_str(), hash.hash_url.as_str()]
                .into_iter()
                .map(|url| {
                    events
                        .iter()
                        .position(|event| {
                            event.kind == OfficialResourcePullProgressKind::Finished
                                && event.url == url
                        })
                        .expect("hash pair member must finish before verification")
                })
                .collect::<Vec<_>>();
            let verification_index = events
                .iter()
                .position(|event| std::ptr::eq(event, *verification_event))
                .expect("verification event must be present in event stream");
            assert_eq!(
                verification_index,
                pair_finished_indices.into_iter().max().unwrap() + 1,
                "official hash verification must run immediately after the hash pair is complete"
            );
        }

        // 每个 started 的 total 一致；index 表示已完成数量，不能超过总数。
        assert!(started.iter().all(|event| event.index <= expected_urls));
        assert!(started.iter().all(|event| event.total == expected_urls));
        assert!(events.windows(2).all(|pair| pair[0].index <= pair[1].index));
        let finished_indices: Vec<usize> = finished.iter().map(|event| event.index).collect();
        assert_eq!(finished_indices, (1..=expected_urls).collect::<Vec<_>>());

        // started 与 finished 覆盖同一组 URL；finished 均为已下载。
        let started_urls: HashSet<_> = started.iter().map(|event| event.url.clone()).collect();
        let finished_urls: HashSet<_> = finished.iter().map(|event| event.url.clone()).collect();
        assert_eq!(started_urls, finished_urls);
        assert!(finished
            .iter()
            .all(|event| event.status == Some(OfficialResourcePullStatus::Downloaded)));
        assert!(finished.iter().all(|event| {
            event.verification.as_ref().is_some_and(|verification| {
                verification.actual_bytes > 0 && !verification.actual_blake3.is_empty()
            })
        }));
        assert!(verifications.iter().all(|event| {
            event.official_hash.as_ref().is_some_and(|hash| {
                hash.algorithm == OfficialResourceHashAlgorithm::XxHash32Decimal
                    && hash.expected == hash.actual
                    && !hash.hash_url.is_empty()
            })
        }));
        assert!(events
            .iter()
            .any(|event| event.url.ends_with("/TableBundles/ExcelDB.db")));
    }

    #[test]
    fn downloads_run_sequentially_and_each_url_reports_once() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        // 顺序下载：每个 URL 恰好一次 started + 一次 finished，全部文件落盘。
        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path)
            .with_max_concurrency(1);
        let plan = build_official_pull_plan_for_platforms(
            discovery_plan(),
            inventory(),
            &[PatchPlatform::Windows],
        );
        let all_urls = plan.all_urls().unwrap();
        let mut events = Vec::new();
        let report = service
            .pull_with_progress(&plan, |event| events.push(event))
            .unwrap();

        assert_eq!(report.items.len(), all_urls.len());
        for url in &all_urls {
            let started = events
                .iter()
                .filter(|event| {
                    &event.url == url && event.kind == OfficialResourcePullProgressKind::Started
                })
                .count();
            let finished = events
                .iter()
                .filter(|event| {
                    &event.url == url && event.kind == OfficialResourcePullProgressKind::Finished
                })
                .count();
            assert_eq!(started, 1, "url {url} 的 started 次数");
            assert_eq!(finished, 1, "url {url} 的 finished 次数");
        }
        // 所有下载都记入 manifest。
        let manifest = service.read_download_manifest().unwrap();
        assert_eq!(manifest.entries.len(), all_urls.len());
    }

    #[test]
    fn downloads_run_with_bounded_concurrency_and_keep_report_order() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path)
            .with_max_concurrency(3);
        let plan = build_official_pull_plan_for_platforms(
            discovery_plan(),
            inventory(),
            &[PatchPlatform::Windows],
        );
        let all_urls = plan.all_urls().unwrap();
        let mut events = Vec::new();
        let report = service
            .pull_with_progress(&plan, |event| events.push(event))
            .unwrap();

        assert_eq!(service.max_concurrency(), 3);
        assert_eq!(report.items.len(), all_urls.len());
        assert_eq!(
            report
                .items
                .iter()
                .map(|item| &item.url)
                .collect::<Vec<_>>(),
            all_urls.iter().collect::<Vec<_>>()
        );
        for url in &all_urls {
            assert_eq!(
                events
                    .iter()
                    .filter(|event| {
                        event.url == *url && event.kind == OfficialResourcePullProgressKind::Started
                    })
                    .count(),
                1,
                "url {url} 的 started 次数"
            );
            assert_eq!(
                events
                    .iter()
                    .filter(|event| {
                        event.url == *url
                            && event.kind == OfficialResourcePullProgressKind::Finished
                    })
                    .count(),
                1,
                "url {url} 的 finished 次数"
            );
        }
        let finished_indices = events
            .iter()
            .filter(|event| event.kind == OfficialResourcePullProgressKind::Finished)
            .map(|event| event.index)
            .collect::<Vec<_>>();
        assert_eq!(finished_indices, (1..=all_urls.len()).collect::<Vec<_>>());
        assert_eq!(
            service.read_download_manifest().unwrap().entries.len(),
            all_urls.len()
        );
    }

    #[test]
    fn retries_transient_download_failures() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_flaky_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path)
            .with_retry_attempts(2);
        let plan = OfficialResourcePullPlan {
            discovery: YostarJpResourceDiscoveryPlan {
                connection_group_name: "Prod-Audit".to_string(),
                app_version: "1.70.0".to_string(),
                bundle_version: None,
                addressables_root:
                    "https://prod-clientpatch.bluearchiveyostar.com/r93_token".to_string(),
                endpoints: vec![YostarJpResourceEndpoint {
                    kind: YostarJpResourceEndpointKind::TableCatalog,
                    platform: None,
                    url: "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes".to_string(),
                }],
            },
            inventory: YostarJpPlatformDownloadInventory::from_shared_inventory(
                YostarJpDownloadInventory::from_catalog_bytes(b"", b"", b""),
                &[],
            ),
            platforms: Vec::new(),
        };

        let report = service.pull(&plan).unwrap();

        assert_eq!(report.items.len(), 1);
        assert_eq!(report.downloaded_count(), 1);
        assert_eq!(
            fs::read_to_string(&report.items[0].destination).unwrap(),
            report.items[0].url
        );
        assert_eq!(
            fs::read_to_string(curl_path.with_extension("state")).unwrap(),
            "2"
        );
    }

    #[test]
    fn does_not_retry_terminal_http_404_and_records_quarantine() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_http_error_curl(&curl_path, 404);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path)
            .with_retry_attempts(3);
        let url =
            "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/Missing.bytes";
        let plan = one_url_plan(url);
        let mut events = Vec::new();

        let error = service
            .pull_with_progress(&plan, |event| events.push(event))
            .unwrap_err();

        assert!(error.to_string().contains("quarantine"));
        assert!(error.to_string().contains("http_not_found"));
        assert!(error.to_string().contains("retryable=false"));
        // 类型化错误携带准确的 HTTP 404 网络域码。
        assert_eq!(error.code(), bat_core::ErrorCode::HTTP_NOT_FOUND);
        assert_eq!(
            fs::read_to_string(curl_path.with_extension("state")).unwrap(),
            "1"
        );
        let failed = events
            .iter()
            .find(|event| event.kind == OfficialResourcePullProgressKind::Failed)
            .unwrap();
        assert_eq!(failed.failure_kind.as_deref(), Some("http_not_found"));
        assert_eq!(failed.failure_http_status, Some(404));
        assert_eq!(failed.failure_retryable, Some(false));
        assert_eq!(failed.failure_attempts, Some(1));
        assert!(failed.quarantined);

        let quarantine = service.read_download_quarantine().unwrap();
        let entry = quarantine.entries.get(url).unwrap();
        assert_eq!(entry.http_status, Some(404));
        assert_eq!(entry.failure_kind.as_deref(), Some("http_not_found"));
        assert_eq!(entry.retryable, Some(false));
        assert_eq!(entry.attempts, 1);
        assert!(service.read_download_manifest().unwrap().entries.is_empty());
    }

    #[test]
    fn official_http_failure_regression_fixtures_are_enforced() {
        for fixture in [
            include_str!("../tests/fixtures/official_regression/http_403.json"),
            include_str!("../tests/fixtures/official_regression/http_404.json"),
        ] {
            let fixture: serde_json::Value = serde_json::from_str(fixture).unwrap();
            let out_dir = TempDir::new().unwrap();
            let bin_dir = TempDir::new().unwrap();
            let curl_path = bin_dir.path().join("curl");
            let status = fixture["http_status"].as_u64().unwrap() as u16;
            write_fake_http_error_curl(&curl_path, status);

            let service =
                OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path)
                    .with_retry_attempts(3);
            let url = fixture["url"].as_str().unwrap();
            let error = service.pull(&one_url_plan(url)).unwrap_err();
            let expected_kind = fixture["failure_kind"].as_str().unwrap();
            let expected_retryable = fixture["retryable"].as_bool().unwrap();
            let expected_attempts = fixture["expected_attempts"].as_u64().unwrap();

            assert!(error.to_string().contains(expected_kind));
            assert!(error
                .to_string()
                .contains(&format!("retryable={expected_retryable}")));
            assert_eq!(
                fs::read_to_string(curl_path.with_extension("state")).unwrap(),
                expected_attempts.to_string()
            );
            let quarantine = service.read_download_quarantine().unwrap();
            let entry = quarantine.entries.get(url).unwrap();
            assert_eq!(entry.failure_kind.as_deref(), Some(expected_kind));
            assert_eq!(entry.retryable, Some(expected_retryable));
            assert_eq!(entry.attempts, expected_attempts as usize);
        }
    }

    #[test]
    fn official_hash_mismatch_regression_fixture_is_enforced() {
        let fixture: serde_json::Value = serde_json::from_str(include_str!(
            "../tests/fixtures/official_regression/hash_mismatch_catalog.json"
        ))
        .unwrap();
        let error = verify_official_seed_catalog_hash(
            fixture["data_url"].as_str().unwrap(),
            fixture["data"].as_str().unwrap().as_bytes(),
            fixture["hash_url"].as_str().unwrap(),
            fixture["official_hash"].as_str().unwrap().as_bytes(),
        )
        .unwrap_err();

        assert!(error
            .to_string()
            .contains(fixture["expected_error_contains"].as_str().unwrap()));
    }

    #[test]
    fn retries_http_5xx_before_quarantine() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_http_error_curl(&curl_path, 503);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path)
            .with_retry_attempts(3);
        let url =
            "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/Busy.bytes";
        let plan = one_url_plan(url);

        let error = service.pull(&plan).unwrap_err();

        assert!(error.to_string().contains("quarantine"));
        assert!(error.to_string().contains("http_server_error"));
        assert!(error.to_string().contains("retryable=true"));
        assert_eq!(
            fs::read_to_string(curl_path.with_extension("state")).unwrap(),
            "3"
        );
        let quarantine = service.read_download_quarantine().unwrap();
        let entry = quarantine.entries.get(url).unwrap();
        assert_eq!(entry.http_status, Some(503));
        assert_eq!(entry.failure_kind.as_deref(), Some("http_server_error"));
        assert_eq!(entry.retryable, Some(true));
        assert_eq!(entry.attempts, 3);
        assert!(entry.last_error.contains("重试次数已耗尽"));
    }

    #[test]
    fn custom_platform_selection_is_supported() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let plan = build_official_pull_plan_for_platforms(
            discovery_plan(),
            inventory(),
            &[PatchPlatform::Windows, PatchPlatform::Android],
        );
        let report = service.pull(&plan).unwrap();

        assert_eq!(report.items.len(), 7);
        assert!(report
            .items
            .iter()
            .any(|item| item.url.contains("Android_PatchPack")));
    }

    #[test]
    fn fetches_official_seed_bytes_with_curl() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path);
        let bytes = service
            .fetch_bytes(
                "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes",
            )
            .unwrap();

        assert_eq!(
            String::from_utf8(bytes).unwrap(),
            "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes"
        );
    }

    #[test]
    fn retries_transient_fetch_bytes_failures() {
        let out_dir = TempDir::new().unwrap();
        let bin_dir = TempDir::new().unwrap();
        let curl_path = bin_dir.path().join("curl");
        write_fake_flaky_curl(&curl_path);

        let service = OfficialResourcePullService::with_curl_command(out_dir.path(), &curl_path)
            .with_retry_attempts(2);
        let bytes = service
            .fetch_bytes(
                "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes",
            )
            .unwrap();

        assert_eq!(
            String::from_utf8(bytes).unwrap(),
            "https://prod-clientpatch.bluearchiveyostar.com/r93_token/TableBundles/TableCatalog.bytes"
        );
        assert_eq!(
            fs::read_to_string(curl_path.with_extension("state")).unwrap(),
            "2"
        );
    }

    #[test]
    fn rejects_non_official_urls() {
        let service = OfficialResourcePullService::new("/tmp/unused");
        let plan = OfficialResourcePullPlan {
            discovery: YostarJpResourceDiscoveryPlan {
                connection_group_name: "Prod-Audit".to_string(),
                app_version: "1.70.0".to_string(),
                bundle_version: None,
                addressables_root:
                    "https://prod-clientpatch.bluearchiveyostar.com/r93_token".to_string(),
                endpoints: vec![YostarJpResourceEndpoint {
                    kind: YostarJpResourceEndpointKind::TableCatalog,
                    platform: None,
                    url: "https://prod-clientpatch.bluearchive.cafe/r93_token/TableBundles/TableCatalog.bytes".to_string(),
                }],
            },
            inventory: platform_inventory(),
            platforms: vec![PatchPlatform::Windows],
        };

        assert!(service.pull(&plan).is_err());
    }
}