odl 1.1.0

use crate::{
    credentials::Credentials,
    download_metadata::{DownloadMetadata, FileChecksum, PartDetails},
    error::MetadataError,
    fs_utils,
    hash::HashDigest,
    response_info::ResponseInfo,
};
use chrono::{DateTime, Utc};
use derive_builder::{Builder, UninitializedFieldError};
use reqwest::{
    Proxy, Url,
    header::{HeaderMap, HeaderName, HeaderValue},
};
use std::{
    collections::HashMap,
    path::{self, PathBuf},
    str::FromStr,
};
use thiserror::Error;
use tokio::sync::Semaphore;
use ulid::Ulid;

/// Represents a download instruction.
///
/// `Download` models all information needed to perform a single file download:
/// the remote `url`, the local `download_dir` used to store part files and
/// metadata, the `save_dir` and `filename` for the final assembled file, and
/// per-download options such as `max_connections` and server-provided
/// attributes (ETag, last-modified, size, hashes).
///
/// Use `Download::from_response_info` to construct a new instruction from a
/// `ResponseInfo` (returned after probing the remote URL), or
/// `Download::from_metadata` to recreate an instruction from persisted
/// metadata on disk.
///
/// Examples
///
/// The canonical way to obtain a `Download` is to probe the remote URL and
/// construct it from the HTTP response information. Consumers typically use
/// `DownloadManager::evaluate` which performs the probe and returns a
/// `Download` instruction ready for `DownloadManager::download`.
///
/// The example below is illustrative and intentionally ignored for doctests
/// because the required `ResponseInfo` type is internal to the crate.
///
/// ```ignore
/// // pseudo-code:
/// // let instr = manager
/// //     .evaluate(EvaluateRequest::new(url, save_dir, &save_resolver))
/// //     .await?;
/// // manager.download(DownloadRequest::new(instr, &server_resolver)).await?;
/// ```
#[derive(Builder, Debug, Clone)]
#[builder(build_fn(validate = "Self::validate", error = "DownloadBuilderError"))]
pub struct Download {
    /// Download directory used to store a small metadata file and parts
    download_dir: path::PathBuf,
    /// URL of the file to download.
    url: Url,
    /// Whether the server supports using range requests (i.e., resumable downloads).
    #[builder(default = false)]
    is_resumable: bool,
    /// Should we use the last-modified value server has sent us for the final file ?
    #[builder(default = false)]
    use_server_time: bool,
    /// The final file name to use to save on disk.
    filename: String,
    /// Where to save the final file
    save_dir: path::PathBuf,
    /// File size reported by the server in bytes. can be unknown until download is actually finished.
    #[builder(default = None)]
    size: Option<u64>,
    /// File size reported by the server in bytes. can be unknown until download is actually finished.
    #[builder(default = Vec::new())]
    checksums: Vec<HashDigest>,
    /// the e-tag the server has sent us, if any
    #[builder(default = None)]
    etag: Option<String>,
    /// the last-modified value the server has sent us, if any
    #[builder(default = None)]
    last_modified: Option<i64>,
    /// did the server ask us to authenticate?
    #[builder(default = false)]
    requires_auth: bool,
    /// did the server ask us to authenticate using basic auth?
    #[builder(default = false)]
    requires_basic_auth: bool,
    /// username and password to use, when requires_auth is true and this is provided
    #[builder(default = None)]
    credentials: Option<Credentials>,
    /// proxy to use, if provided
    #[builder(default = None)]
    proxy: Option<Proxy>,
    #[builder(default = None)]
    headers: Option<HeaderMap>,
    /// Preferred number of connections for this download.
    /// This will determine the initial number of parts, which will not be decreased after determination,
    /// even if the max_connections is decreased.
    #[builder(default = 6)]
    max_connections: u64,
    /// The parts determined based on max_connections and size. if size is unknown, this will only contain one element
    parts: HashMap<String, PartDetails>,
    /// Is the download finished?
    #[builder(default = false)]
    finished: bool,
}

/// Result of [`Download::compute_split`]: how to resize an existing
/// part (`new_left_size`) and where the new right-hand part begins
/// (`new_right_offset` / `new_right_size`).
#[derive(Debug, Clone, Copy)]
pub struct PartSplit {
    pub new_left_size: u64,
    pub new_right_offset: u64,
    pub new_right_size: u64,
}

impl Download {
    // Getters for Download fields
    const METADATA_FILENAME: &'static str = "metadata.pb";
    const METADATA_TEMP_FILENAME: &'static str = "metadata.pb.temp";
    const LOCK_FILENAME: &'static str = "odl.lock";
    pub const PART_EXTENSION: &'static str = "part";
    pub const MIN_PART_SIZE: u64 = 300 * 1024; // 300 KB
    /// Sentinel value stored in `PartDetails.size` when the server did not
    /// report a total length (no `Content-Length`, no `Content-Range`).
    /// The downloader treats such a part as "stream until EOF" and never
    /// sends a `Range` header. Resumption / dynamic-split / grow_parts all
    /// skip parts carrying this sentinel.
    pub const UNKNOWN_PART_SIZE: u64 = u64::MAX;
    /// Assumed filesystem cluster size used to keep part boundaries aligned
    /// so the assembler can reflink parts into the final file.
    ///
    /// 4 KiB matches the page/cluster size on btrfs, xfs, ext4 and ReFS.
    /// On filesystems with larger clusters (e.g. ZFS recordsize 128 KiB)
    /// reflink fails the alignment check and the assembler falls back to
    /// a buffered copy — correct, just no CoW share.
    ///
    /// Three call sites depend on this constant; keep them in sync:
    ///   * `Download::split_parts` — initial split offsets/sizes
    ///   * `Downloader::try_split_dynamic` — mid-flight split boundary
    ///   * `download_manager::io::assemble_blocking` — reflink alignment check
    pub const ASSEMBLY_CLUSTER_SIZE: u64 = 4096;

    // Split logic assumes the minimum part size is at least one cluster,
    // otherwise the base-size round-down could produce zero and emit
    // empty leading parts.
    const _ASSERT_MIN_PART_GE_CLUSTER: () =
        assert!(Self::MIN_PART_SIZE >= Self::ASSEMBLY_CLUSTER_SIZE);

    pub fn download_dir(&self) -> &path::PathBuf {
        &self.download_dir
    }

    pub fn part_path(&self, ulid: &str) -> path::PathBuf {
        self.download_dir
            .join(format!("{}.{}", ulid, Self::PART_EXTENSION))
    }

    pub fn set_download_dir(&mut self, path: PathBuf) {
        self.download_dir = path
    }

    pub fn lockfile_path(&self) -> path::PathBuf {
        self.download_dir.join(Self::LOCK_FILENAME)
    }

    pub fn metadata_path(&self) -> path::PathBuf {
        self.download_dir.join(Self::METADATA_FILENAME)
    }

    pub fn metadata_temp_path(&self) -> path::PathBuf {
        self.download_dir.join(Self::METADATA_TEMP_FILENAME)
    }

    pub fn final_file_path(&self) -> path::PathBuf {
        self.save_dir.join(&self.filename)
    }

    pub fn url(&self) -> &Url {
        &self.url
    }

    pub fn is_resumable(&self) -> bool {
        self.is_resumable
    }

    pub fn use_server_time(&self) -> bool {
        self.use_server_time
    }

    pub fn filename(&self) -> &str {
        &self.filename
    }

    pub fn set_filename(&mut self, filename: String) {
        self.filename = filename;
    }

    /// Merge additional expected checksums (e.g. user-supplied via CLI)
    /// into the instruction, skipping any already present. These are
    /// persisted to metadata and verified against the assembled file
    /// alongside any server-advertised checksums.
    pub fn add_checksums(&mut self, extra: impl IntoIterator<Item = HashDigest>) {
        for c in extra {
            if !self.checksums.contains(&c) {
                self.checksums.push(c);
            }
        }
    }

    pub fn save_dir(&self) -> &path::PathBuf {
        &self.save_dir
    }

    pub fn set_save_dir(&mut self, path: PathBuf) {
        self.save_dir = path
    }

    pub fn size(&self) -> Option<u64> {
        self.size
    }

    pub fn etag(&self) -> &Option<String> {
        &self.etag
    }

    pub fn last_modified(&self) -> Option<i64> {
        self.last_modified
    }

    pub fn last_modified_as_date(&self) -> Option<DateTime<Utc>> {
        self.last_modified
            .and_then(|x| chrono::DateTime::from_timestamp(x, 0))
    }

    pub fn requires_auth(&self) -> bool {
        self.requires_auth
    }

    pub fn requires_basic_auth(&self) -> bool {
        self.requires_basic_auth
    }

    pub fn credentials(&self) -> &Option<Credentials> {
        &self.credentials
    }

    pub fn proxy(&self) -> &Option<Proxy> {
        &self.proxy
    }

    pub fn headers(&self) -> &Option<HeaderMap> {
        &self.headers
    }

    pub fn max_connections(&self) -> u64 {
        self.max_connections
    }

    pub fn parts(&self) -> &HashMap<String, PartDetails> {
        &self.parts
    }

    pub fn finished(&self) -> bool {
        self.finished
    }

    pub fn from_metadata(
        download_dir: path::PathBuf,
        metadata: DownloadMetadata,
    ) -> Result<Download, MetadataError> {
        let url = Url::parse(&metadata.url).map_err(|e| MetadataError::Other {
            message: e.to_string(),
        })?;

        Ok(Self {
            download_dir,
            url,
            is_resumable: metadata.is_resumable,
            use_server_time: metadata.use_server_time,
            filename: metadata.filename, // is cleaned up before its stored as metadata, by from_response
            save_dir: PathBuf::from(metadata.save_dir),
            etag: metadata.last_etag,
            last_modified: metadata.last_modified,
            size: metadata.size,
            checksums: metadata
                .checksums
                .into_iter()
                .map(|c| c.try_into())
                .collect::<Result<Vec<HashDigest>, _>>()
                .unwrap_or_default(),
            credentials: None,
            requires_auth: metadata.requires_auth,
            requires_basic_auth: metadata.requires_basic_auth,
            proxy: None,
            headers: if metadata.headers.is_empty() {
                None
            } else {
                let mut map = HeaderMap::new();
                for (k, v) in metadata.headers {
                    if let (Ok(header_name), Ok(header_value)) =
                        (HeaderName::from_str(&k), HeaderValue::from_str(&v))
                    {
                        map.insert(header_name, header_value);
                    }
                }
                Some(map)
            },
            max_connections: metadata.max_connections,
            parts: metadata.parts,
            finished: metadata.finished,
        })
    }

    pub fn as_metadata(&self) -> DownloadMetadata {
        DownloadMetadata {
            url: self.url.to_string(),
            filename: self.filename.clone(),
            save_dir: self.save_dir.to_string_lossy().into_owned(),
            is_resumable: self.is_resumable,
            use_server_time: self.use_server_time,
            last_modified: self.last_modified,
            last_etag: self.etag.clone(),
            size: self.size,
            checksums: self
                .checksums
                .iter()
                .map(|h| h.clone().into())
                .collect::<Vec<FileChecksum>>(),
            requires_auth: self.requires_auth,
            requires_basic_auth: self.requires_basic_auth,
            headers: self
                .headers
                .as_ref()
                .map(|h| {
                    h.iter()
                        .map(|(k, v)| (k.to_string(), v.to_str().unwrap_or("").to_string()))
                        .collect()
                })
                .unwrap_or_default(),
            max_connections: self.max_connections,
            parts: self.parts.clone(),
            finished: self.finished,
        }
    }

    #[allow(clippy::too_many_arguments)]
    pub fn from_response_info(
        download_dir: &std::path::Path,
        save_dir: path::PathBuf,
        response_info: ResponseInfo,
        max_connections: u64,
        use_server_time: bool,
        credentials: Option<Credentials>,
        proxy: Option<Proxy>,
        headers: Option<HeaderMap>,
    ) -> Download {
        let filename = fs_utils::cleanup_filename(response_info.extract_filename().as_str());
        Self {
            download_dir: download_dir.join(&filename),
            url: response_info.url().clone(),
            is_resumable: response_info.is_resumable(),
            use_server_time,
            filename,
            save_dir,
            etag: response_info.etag(),
            last_modified: response_info.parse_last_modified(),
            size: response_info.total_length(),
            checksums: response_info.extract_hashes(),
            credentials,
            requires_auth: response_info.requires_auth(),
            requires_basic_auth: response_info.requires_basic_auth(),
            proxy,
            headers,
            max_connections,
            parts: Download::determine_parts(
                response_info.total_length(),
                if response_info.is_resumable() {
                    max_connections
                } else {
                    1
                },
            ),
            finished: false,
        }
    }

    /// Compute a cluster-aligned split point for a part with `offset`,
    /// `size`, and `already_consumed` bytes already written/scheduled.
    /// The split favours the right half (new part) absorbing roughly the
    /// remaining bytes / 2; the left half (current part) keeps every
    /// byte up to `already_consumed`, so no progress is invalidated.
    ///
    /// Returns `None` when the resulting halves wouldn't both clear
    /// `min_part_size`, the new boundary wouldn't move past
    /// `already_consumed`, or the input `offset` is not on a cluster
    /// boundary (which would break reflink-based assembly).
    ///
    /// Reflink invariant kept:
    /// - `new_left_size` is rounded down to a multiple of
    ///   `ASSEMBLY_CLUSTER_SIZE` so the left half ends on a cluster
    ///   boundary → its reflink range stays aligned.
    /// - `new_right_offset = offset + new_left_size` stays cluster-
    ///   aligned because `offset` is required to be aligned on entry
    ///   and `new_left_size` is a cluster multiple.
    /// - `new_right_size = size - new_left_size` inherits any tail
    ///   unalignment from `size`. The original tail-unaligned part is
    ///   always the LAST in absolute-offset order, so its split right
    ///   child remains last too — Linux's `ficlonerange` allows an
    ///   unaligned tail on the final reflink range (Windows falls back
    ///   to a byte copy, same as before).
    ///
    /// Both callers — mid-flight dynamic splits in
    /// `Downloader::split_task` and the static resume-time grow in
    /// `download_manager::grow_parts` — share this geometry; only the
    /// minimum-size threshold differs.
    pub fn compute_split(
        offset: u64,
        size: u64,
        already_consumed: u64,
        min_part_size: u64,
    ) -> Option<PartSplit> {
        if !offset.is_multiple_of(Self::ASSEMBLY_CLUSTER_SIZE) {
            // Caller bug: a part whose absolute offset isn't cluster-
            // aligned can't be assembled via reflink, so refuse to split
            // it (preserving the current bad state is strictly better
            // than producing two bad parts).
            debug_assert!(
                false,
                "compute_split: offset {offset:#x} not cluster-aligned",
            );
            return None;
        }
        if already_consumed >= size {
            return None;
        }
        let remaining = size - already_consumed;
        if remaining < min_part_size * 2 {
            return None;
        }
        let candidate = already_consumed + remaining / 2;
        // Round down to a multiple of ASSEMBLY_CLUSTER_SIZE so the new
        // boundary lands on a cluster edge (reflink requirement).
        let new_left_size = candidate - candidate % Self::ASSEMBLY_CLUSTER_SIZE;
        if new_left_size <= already_consumed {
            return None;
        }
        let new_right_size = size - new_left_size;
        if new_right_size < min_part_size || new_left_size - already_consumed < min_part_size {
            return None;
        }
        Some(PartSplit {
            new_left_size,
            new_right_offset: offset + new_left_size,
            new_right_size,
        })
    }

    pub fn determine_parts(
        size: Option<u64>,
        max_connections: u64,
    ) -> HashMap<String, PartDetails> {
        let mut parts = HashMap::new();

        let max_connections = if max_connections > 0 {
            max_connections
        } else {
            1
        };

        // Unknown total length (no Content-Length / Content-Range from the
        // server). Emit a single "stream until EOF" part marked with the
        // UNKNOWN_PART_SIZE sentinel. The downloader skips Range, drains
        // the body, and rewrites part.size to the actual byte count when
        // it completes.
        if size.is_none() {
            let ulid = Ulid::new().to_string();
            parts.insert(
                ulid.clone(),
                PartDetails {
                    offset: 0,
                    size: Self::UNKNOWN_PART_SIZE,
                    ulid,
                    finished: false,
                },
            );
            return parts;
        }

        let size = size.unwrap_or(0);

        // Always return at least one part, even if size is 0
        // If the size is small (<= MIN_PART_SIZE) we keep a single part to
        // avoid fragmenting the download into many very small requests.
        if size <= Self::MIN_PART_SIZE {
            let ulid = Ulid::new().to_string();
            parts.insert(
                ulid.clone(),
                PartDetails {
                    offset: 0,
                    size,
                    ulid,
                    finished: size == 0,
                },
            );
            return parts;
        }

        let mut actual_connections = max_connections;
        let min_connections = size.div_ceil(Self::MIN_PART_SIZE);
        if actual_connections > min_connections {
            actual_connections = min_connections;
        }

        // Round each middle part's size down to a cluster multiple so
        // the assembler can reflink it at its absolute offset.
        let raw_base = size / actual_connections;
        let base_size = raw_base - raw_base % Self::ASSEMBLY_CLUSTER_SIZE;
        let mut offset = 0;

        // Cluster-aligned base size lets the assembler reflink each part at its
        // offset. Remainder + alignment slack go on the last part, whose tail
        // is allowed to be unaligned (the trailing copy handles it).
        for i in 0..actual_connections {
            let part_size = if i == actual_connections - 1 {
                size - offset
            } else {
                base_size
            };
            let ulid = Ulid::new().to_string();
            parts.insert(
                ulid.clone(),
                PartDetails {
                    offset,
                    size: part_size,
                    ulid,
                    finished: false,
                },
            );
            offset += part_size;
        }

        parts
    }
}

impl PartialEq for Download {
    fn eq(&self, other: &Self) -> bool {
        self.url == other.url
            && self.download_dir == other.download_dir
            && self.filename == other.filename
    }
}

impl DownloadBuilder {
    fn validate(&self) -> Result<(), DownloadBuilderError> {
        if self.download_dir.is_none() {
            return Err(DownloadBuilderError::MissingDownloadDir);
        }
        if self.save_dir.is_none() {
            return Err(DownloadBuilderError::MissingSaveDir);
        }
        if self.url.is_none() {
            return Err(DownloadBuilderError::MissingUrl);
        }
        if self.filename.is_none() {
            return Err(DownloadBuilderError::MissingFilename);
        }
        if self
            .max_connections
            .is_none_or(|x| x == 0 || x >= Semaphore::MAX_PERMITS.try_into().unwrap_or(1_000_000))
        {
            return Err(DownloadBuilderError::InvalidNumConnections);
        }
        Ok(())
    }
}

#[derive(Error, Debug)]
pub enum DownloadBuilderError {
    #[error("download_dir is required")]
    MissingDownloadDir,
    #[error("save_dir is required")]
    MissingSaveDir,
    #[error("url is required")]
    MissingUrl,
    #[error("filename is required")]
    MissingFilename,
    #[error("max_connections must be at least 1")]
    InvalidNumConnections,
    /// Uninitialized field
    #[error("uninitialized field: {0}")]
    UninitializedField(String),
    /// Custom validation error
    #[error("validation error: {0}")]
    ValidationError(String),
}

impl From<String> for DownloadBuilderError {
    fn from(s: String) -> Self {
        Self::ValidationError(s)
    }
}

impl From<UninitializedFieldError> for DownloadBuilderError {
    fn from(ufe: UninitializedFieldError) -> Self {
        Self::UninitializedField(ufe.to_string())
    }
}
#[cfg(test)]
mod tests {
    use super::*;

    fn test_download(checksums: Vec<HashDigest>) -> Download {
        DownloadBuilder::default()
            .download_dir(PathBuf::from("/tmp/dl"))
            .save_dir(PathBuf::from("/tmp/save"))
            .url(Url::parse("https://example.com/file").unwrap())
            .filename("file".to_string())
            .max_connections(1)
            .checksums(checksums)
            .parts(Download::determine_parts(Some(0), 1))
            .build()
            .unwrap()
    }

    #[test]
    fn add_checksums_merges_and_dedups() {
        use crate::hash::HashEncoding;
        // Instruction already carries a server-advertised SHA256.
        let server = HashDigest::SHA256("aa".repeat(32), HashEncoding::Hex);
        let mut dl = test_download(vec![server.clone()]);

        // User supplies the identical SHA256 again (must dedup against the
        // seeded one) plus a brand-new MD5 (must be kept).
        let user_new = HashDigest::MD5("bb".repeat(16), HashEncoding::Hex);
        dl.add_checksums(vec![server.clone(), user_new.clone()]);

        assert_eq!(dl.checksums, vec![server, user_new]);
    }

    #[test]
    fn add_checksums_into_empty() {
        use crate::hash::HashEncoding;
        let mut dl = test_download(vec![]);
        let c = HashDigest::SHA512("cc".repeat(64), HashEncoding::Hex);
        dl.add_checksums(vec![c.clone()]);
        assert_eq!(dl.checksums, vec![c]);
    }

    #[test]
    fn test_determine_parts_unknown_size_streams_until_eof() {
        // No Content-Length / Content-Range from the server (typical for
        // chunked HTML, gzipped responses where reqwest strips the length,
        // etc.) must yield a single unfinished part flagged with the
        // UNKNOWN_PART_SIZE sentinel so the downloader streams to EOF
        // instead of treating it as a zero-byte file already complete.
        let parts = Download::determine_parts(None, 4);
        assert_eq!(parts.len(), 1);
        let part = parts.values().next().unwrap();
        assert_eq!(part.offset, 0);
        assert_eq!(part.size, Download::UNKNOWN_PART_SIZE);
        assert!(!part.finished);
    }

    #[test]
    fn test_determine_parts_zero_size() {
        let parts = Download::determine_parts(Some(0), 4);
        assert_eq!(parts.len(), 1);
        let part_vec: Vec<_> = parts.values().collect();
        let part = part_vec[0];
        assert_eq!(part.offset, 0);
        assert_eq!(part.size, 0);
        assert!(part.finished);
    }

    #[test]
    fn test_determine_parts_zero_connections() {
        let parts = Download::determine_parts(Some(1024 * 1024), 0);
        assert_eq!(parts.len(), 1);
        let part_vec: Vec<_> = parts.values().collect();
        let part = part_vec[0];
        assert_eq!(part.offset, 0);
        assert_eq!(part.size, 1024 * 1024);
        assert!(!part.finished);
    }

    #[test]
    fn test_determine_parts_small_file() {
        // File smaller than MIN_PART_SIZE (300 KB)
        let size = 200 * 1024;
        let parts = Download::determine_parts(Some(size), 4);
        assert_eq!(parts.len(), 1);
        let part_vec: Vec<_> = parts.values().collect();
        let part = part_vec[0];
        assert_eq!(part.offset, 0);
        assert_eq!(part.size, size);
        assert!(!part.finished);
    }

    #[test]
    fn test_determine_parts_exact_min_part_size() {
        let size = 300 * 1024;
        let parts = Download::determine_parts(Some(size), 4);
        assert_eq!(parts.len(), 1);
        let part_vec: Vec<_> = parts.values().collect();
        let part = part_vec[0];
        assert_eq!(part.offset, 0);
        assert_eq!(part.size, size);
        assert!(!part.finished);
    }

    #[test]
    fn test_determine_parts_even_split() {
        // 1 MB file, 4 connections
        let size = 1024 * 1024;
        let max_connections = 4;
        let parts = Download::determine_parts(Some(size), max_connections);
        assert_eq!(parts.len(), max_connections as usize);
        let mut part_vec: Vec<_> = parts.values().collect();
        part_vec.sort_by_key(|p| p.offset);
        let total: u64 = part_vec.iter().map(|p| p.size).sum();
        assert_eq!(total, size);
        assert_eq!(part_vec[0].offset, 0);
        assert_eq!(part_vec[1].offset, part_vec[0].size);
        assert_eq!(part_vec[2].offset, part_vec[0].size + part_vec[1].size);
        assert_eq!(
            part_vec[3].offset,
            part_vec[0].size + part_vec[1].size + part_vec[2].size
        );
    }

    #[test]
    fn test_determine_parts_uneven_split() {
        // 1 MB + 123 bytes, 3 connections
        let size = 1024 * 1024 + 123;
        let max_connections = 3;
        let parts = Download::determine_parts(Some(size), max_connections);
        assert_eq!(parts.len(), max_connections as usize);
        let mut part_vec: Vec<_> = parts.values().collect();
        part_vec.sort_by_key(|p| p.offset);
        let total: u64 = part_vec.iter().map(|p| p.size).sum();
        assert_eq!(total, size);
        // The last part absorbs the remainder so middle offsets stay
        // cluster-aligned for reflink-based assembly.
        assert!(part_vec[2].size >= part_vec[1].size);
        assert_eq!(part_vec[0].size, part_vec[1].size);
    }

    #[test]
    fn test_determine_parts_too_many_connections() {
        // File size is such that min_connections < max_connections
        let size = 900 * 1024; // 900 KB
        let max_connections = 10;
        let parts = Download::determine_parts(Some(size), max_connections);
        // Should not exceed min_connections (3)
        assert_eq!(parts.len(), 3);
        let mut part_vec: Vec<_> = parts.values().collect();
        part_vec.sort_by_key(|p| p.offset);
        let total: u64 = part_vec.iter().map(|p| p.size).sum();
        assert_eq!(total, size);
    }

    #[test]
    fn test_determine_parts_800kb_file() {
        // 800 KB file, should be split into 3 parts (since MIN_PART_SIZE is 300 KB)
        let size = 800 * 1024;
        let max_connections = 10; // More than needed, should be capped by min_connections
        let parts = Download::determine_parts(Some(size), max_connections);
        // 800 KB / 300 KB = 2.66..., so should be 3 parts
        assert_eq!(parts.len(), 3);
        let mut part_vec: Vec<_> = parts.values().collect();
        part_vec.sort_by_key(|p| p.offset);
        let total: u64 = part_vec.iter().map(|p| p.size).sum();
        assert_eq!(total, size);

        // Check offsets are correct and contiguous
        assert_eq!(part_vec[0].offset, 0);
        assert_eq!(part_vec[1].offset, part_vec[0].offset + part_vec[0].size);
        assert_eq!(part_vec[2].offset, part_vec[1].offset + part_vec[1].size);

        // The last part absorbs the remainder; preceding parts share the
        // cluster-aligned base size.
        assert_eq!(part_vec[0].size, part_vec[1].size);
        assert!(part_vec[2].size >= part_vec[1].size);
        assert_eq!(part_vec[0].offset % Download::ASSEMBLY_CLUSTER_SIZE, 0);
        assert_eq!(part_vec[1].offset % Download::ASSEMBLY_CLUSTER_SIZE, 0);
        assert_eq!(part_vec[2].offset % Download::ASSEMBLY_CLUSTER_SIZE, 0);
    }

    #[test]
    fn compute_split_returns_none_when_remaining_below_double_min() {
        // remaining = size - already = MIN_PART_SIZE * 2 - 1
        let size = Download::MIN_PART_SIZE * 2 - 1;
        assert!(Download::compute_split(0, size, 0, Download::MIN_PART_SIZE).is_none());
    }

    #[test]
    fn compute_split_aligns_boundary_and_preserves_total() {
        let size = Download::MIN_PART_SIZE * 8;
        let split = Download::compute_split(1024 * 1024, size, 0, Download::MIN_PART_SIZE)
            .expect("split expected");
        // Left half aligned to cluster boundary
        assert_eq!(split.new_left_size % Download::ASSEMBLY_CLUSTER_SIZE, 0);
        // Total bytes preserved
        assert_eq!(split.new_left_size + split.new_right_size, size);
        // New offset = base offset + left size
        assert_eq!(split.new_right_offset, 1024 * 1024 + split.new_left_size);
        // Both halves above min
        assert!(split.new_left_size >= Download::MIN_PART_SIZE);
        assert!(split.new_right_size >= Download::MIN_PART_SIZE);
    }

    #[test]
    fn compute_split_keeps_offsets_cluster_aligned_for_reflink() {
        // Start from a determine_parts result (which guarantees all
        // middle offsets are cluster-aligned) and recursively split the
        // largest unfinished candidate. Every produced offset must stay
        // cluster-aligned so the assembler can reflink.
        let size = 50 * 1024 * 1024 + 1234; // 50 MiB + unaligned tail
        let mut parts = Download::determine_parts(Some(size), 4);

        for _ in 0..8 {
            let candidate = parts
                .values()
                .filter_map(|p| {
                    Download::compute_split(p.offset, p.size, 0, Download::MIN_PART_SIZE)
                        .map(|s| (p.ulid.clone(), p.offset, p.size, s))
                })
                .max_by_key(|(_, _, _, s)| s.new_right_size);
            let Some((ulid, _, _, split)) = candidate else {
                break;
            };
            // Update left
            if let Some(p) = parts.get_mut(&ulid) {
                p.size = split.new_left_size;
            }
            // Insert right
            let new_ulid = ulid::Ulid::new().to_string();
            parts.insert(
                new_ulid.clone(),
                crate::download_metadata::PartDetails {
                    offset: split.new_right_offset,
                    size: split.new_right_size,
                    ulid: new_ulid,
                    finished: false,
                },
            );
        }

        // All offsets must be cluster-aligned for reflink.
        for p in parts.values() {
            assert_eq!(
                p.offset % Download::ASSEMBLY_CLUSTER_SIZE,
                0,
                "offset {} broke cluster alignment after split",
                p.offset
            );
        }
        // Coverage preserved.
        let total: u64 = parts.values().map(|p| p.size).sum();
        assert_eq!(total, size);

        // Among the parts, the last-by-offset is the only one allowed
        // to have unaligned size. Every other must be cluster-aligned
        // size to keep its reflink range fully aligned.
        let mut sorted: Vec<_> = parts.values().collect();
        sorted.sort_by_key(|p| p.offset);
        for p in &sorted[..sorted.len() - 1] {
            assert_eq!(
                p.size % Download::ASSEMBLY_CLUSTER_SIZE,
                0,
                "non-last part size {} broke cluster alignment",
                p.size
            );
        }
    }

    #[test]
    fn compute_split_respects_already_consumed_floor() {
        // Already consumed half; remainder must still be splittable.
        let size = Download::MIN_PART_SIZE * 8;
        let consumed = Download::MIN_PART_SIZE * 4;
        let split = Download::compute_split(0, size, consumed, Download::MIN_PART_SIZE)
            .expect("split expected");
        assert!(
            split.new_left_size > consumed,
            "boundary must move past already-consumed prefix"
        );
        assert_eq!(split.new_left_size + split.new_right_size, size);
    }
}