oma-fetch 0.40.0

use crate::{
    CompressType, DownloadSource, Event, SingleDownloadErrorHelper, checksum::ChecksumValidator,
    send_request,
};
use std::{
    io::{self, SeekFrom},
    path::Path,
    pin::Pin,
    sync::atomic::{AtomicUsize, Ordering},
    task::{Context, Poll},
    time::Duration,
};

use async_compression::futures::bufread::{
    BzDecoder, GzipDecoder, Lz4Decoder, LzmaDecoder, XzDecoder, ZstdDecoder,
};
use bon::bon;
use futures::{AsyncBufRead, AsyncRead, TryStreamExt, io::BufReader};
use headers::{ContentLength, ContentRange, HeaderMapExt};
use reqwest::{Method, StatusCode, header::RANGE};
use reqwest_middleware::ClientWithMiddleware;
use serde::{Deserialize, Deserializer, Serialize, Serializer};
use snafu::{ResultExt, Snafu};
use tokio::{
    fs::{self, File},
    io::{AsyncBufReadExt as _, AsyncReadExt as _, AsyncSeekExt, AsyncWriteExt},
    time::timeout,
};

use oma_logger::{debug, trace};
use tokio_util::compat::{FuturesAsyncReadCompatExt, TokioAsyncReadCompatExt};

use crate::{DownloadEntry, DownloadSourceType};

const READ_FILE_BUFSIZE: usize = 65536;
const DOWNLOAD_BUFSIZE: usize = 8192;

#[derive(Debug, Snafu)]
pub enum BuilderError {
    #[snafu(display("Download task {file_name} sources is empty"))]
    EmptySource { file_name: String },
    #[snafu(display("Not allow set illegal download threads: {count}"))]
    IllegalDownloadThread { count: usize },
}

pub(crate) struct SingleDownloader {
    client: ClientWithMiddleware,
    pub entry: DownloadEntry,
    total: usize,
    retry_times: usize,
    download_list_index: usize,
    timeout: Duration,
}

pub enum DownloadResult {
    Success(SuccessSummary),
    Failed { file_name: String },
}

#[derive(Debug)]
pub struct SuccessSummary {
    pub file_name: String,
    pub index: usize,
    pub wrote: bool,
    pub url: String,
}

#[derive(Debug, Snafu)]
pub enum SingleDownloadError {
    #[snafu(display("Failed to set permission"))]
    SetPermission { source: io::Error },
    #[snafu(display("Failed to open file as rw mode"))]
    OpenAsWriteMode { source: io::Error },
    #[snafu(display("Failed to open file"))]
    Open { source: io::Error },
    #[snafu(display("Failed to create file"))]
    Create { source: io::Error },
    #[snafu(display("Failed to seek file"))]
    Seek { source: io::Error },
    #[snafu(display("Failed to write file"))]
    Write { source: io::Error },
    #[snafu(display("Failed to flush file"))]
    Flush { source: io::Error },
    #[snafu(display("Failed to remove file"))]
    Remove { source: io::Error },
    #[snafu(display("Failed to create symlink"))]
    CreateSymlink { source: io::Error },
    #[snafu(display("Request Error"))]
    ReqwestMiddlewareError { source: reqwest_middleware::Error },
    #[snafu(display("Broken pipe"))]
    BrokenPipe { source: io::Error },
    #[snafu(display("Send request timeout"))]
    SendRequestTimeout,
    #[snafu(display("Download file timeout"))]
    DownloadTimeout,
    #[snafu(display("checksum mismatch"))]
    ChecksumMismatch,
    #[snafu(display("semaphore acquire error"))]
    AcquireError,
}

impl Serialize for SingleDownloadError {
    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        let helper = match self {
            Self::SetPermission { source } => SingleDownloadErrorHelper::SetPermission {
                source: source.to_string(),
            },
            Self::OpenAsWriteMode { source } => SingleDownloadErrorHelper::OpenAsWriteMode {
                source: source.to_string(),
            },
            Self::Open { source } => SingleDownloadErrorHelper::Open {
                source: source.to_string(),
            },
            Self::Create { source } => SingleDownloadErrorHelper::Create {
                source: source.to_string(),
            },
            Self::Seek { source } => SingleDownloadErrorHelper::Seek {
                source: source.to_string(),
            },
            Self::Write { source } => SingleDownloadErrorHelper::Write {
                source: source.to_string(),
            },
            Self::Flush { source } => SingleDownloadErrorHelper::Flush {
                source: source.to_string(),
            },
            Self::Remove { source } => SingleDownloadErrorHelper::Remove {
                source: source.to_string(),
            },
            Self::CreateSymlink { source } => SingleDownloadErrorHelper::CreateSymlink {
                source: source.to_string(),
            },
            Self::ReqwestMiddlewareError { source } => {
                SingleDownloadErrorHelper::ReqwestMiddlewareError {
                    source: source.to_string(),
                }
            }
            Self::BrokenPipe { source } => SingleDownloadErrorHelper::BrokenPipe {
                source: source.to_string(),
            },
            Self::SendRequestTimeout => SingleDownloadErrorHelper::SendRequestTimeout,
            Self::DownloadTimeout => SingleDownloadErrorHelper::DownloadTimeout,
            Self::ChecksumMismatch => SingleDownloadErrorHelper::ChecksumMismatch,
            Self::AcquireError => SingleDownloadErrorHelper::AcquireError,
        };

        helper.serialize(serializer)
    }
}

impl<'de> Deserialize<'de> for SingleDownloadError {
    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
    where
        D: Deserializer<'de>,
    {
        let helper = SingleDownloadErrorHelper::deserialize(deserializer)?;

        let error = match helper {
            SingleDownloadErrorHelper::SetPermission { source } => Self::SetPermission {
                source: io::Error::other(source),
            },
            SingleDownloadErrorHelper::OpenAsWriteMode { source } => Self::OpenAsWriteMode {
                source: io::Error::other(source),
            },
            SingleDownloadErrorHelper::Open { source } => Self::Open {
                source: io::Error::other(source),
            },
            SingleDownloadErrorHelper::Create { source } => Self::Create {
                source: io::Error::other(source),
            },
            SingleDownloadErrorHelper::Seek { source } => Self::Seek {
                source: io::Error::other(source),
            },
            SingleDownloadErrorHelper::Write { source } => Self::Write {
                source: io::Error::other(source),
            },
            SingleDownloadErrorHelper::Flush { source } => Self::Flush {
                source: io::Error::other(source),
            },
            SingleDownloadErrorHelper::Remove { source } => Self::Remove {
                source: io::Error::other(source),
            },
            SingleDownloadErrorHelper::CreateSymlink { source } => Self::CreateSymlink {
                source: io::Error::other(source),
            },

            // reqwest_middleware::Error 无法简单 new，但它通常支持从标准 Error 转换，或者转为自定义格式
            // 这里可以通过 anyhow 或标准映射转换为中间状态错误
            SingleDownloadErrorHelper::ReqwestMiddlewareError { source } => {
                Self::ReqwestMiddlewareError {
                    source: reqwest_middleware::Error::Middleware(anyhow::anyhow!(source)),
                }
            }

            SingleDownloadErrorHelper::BrokenPipe { source } => Self::BrokenPipe {
                source: io::Error::new(io::ErrorKind::BrokenPipe, source),
            },
            SingleDownloadErrorHelper::SendRequestTimeout => Self::SendRequestTimeout,
            SingleDownloadErrorHelper::DownloadTimeout => Self::DownloadTimeout,
            SingleDownloadErrorHelper::ChecksumMismatch => Self::ChecksumMismatch,
            SingleDownloadErrorHelper::AcquireError => Self::AcquireError,
        };

        Ok(error)
    }
}

/// An error from a download attempt together with the bytes that attempt had
/// already reported to the global progress bar, so the caller can undo them
/// before retrying or falling back to the next source.
pub(crate) struct ProgressedError {
    pub(crate) error: SingleDownloadError,
    pub(crate) bytes: u64,
}

impl ProgressedError {
    pub(crate) fn new(error: SingleDownloadError, bytes: u64) -> Self {
        Self { error, bytes }
    }
}

impl From<SingleDownloadError> for ProgressedError {
    fn from(error: SingleDownloadError) -> Self {
        Self { error, bytes: 0 }
    }
}

#[bon]
impl SingleDownloader {
    #[builder]
    pub(crate) fn new(
        client: ClientWithMiddleware,
        entry: DownloadEntry,
        total: usize,
        retry_times: usize,
        download_list_index: usize,
        timeout: Duration,
    ) -> Result<SingleDownloader, BuilderError> {
        if entry.source.is_empty() {
            return Err(BuilderError::EmptySource {
                file_name: entry.filename.to_string(),
            });
        }

        Ok(Self {
            client,
            entry,
            total,
            retry_times,
            download_list_index,
            timeout,
        })
    }

    pub(crate) async fn try_download<F, Fut>(self, callback: &F) -> DownloadResult
    where
        F: Fn(Event) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = ()> + Send + 'static,
    {
        if let Err(e) = tokio::fs::create_dir_all(&self.entry.dir).await {
            callback(Event::Failed {
                file_name: self.entry.filename.clone(),
                error: SingleDownloadError::Create { source: e },
            })
            .await;
            return DownloadResult::Failed {
                file_name: self.entry.filename.to_string(),
            };
        }

        let msg = self.entry.msg.as_deref().unwrap_or(&*self.entry.filename);

        if let Some(ref final_dir) = self.entry.final_dir {
            let local_file_in_formal = final_dir.join(&*self.entry.filename);

            if local_file_in_formal.is_file()
                && let Some(ref hash) = self.entry.hash
            {
                let mut validator = hash.get_validator();

                if let Ok(mut f) = tokio::fs::File::open(&local_file_in_formal).await {
                    let (read, finish) =
                        crate::download::checksum(callback, &mut f, &mut validator).await;

                    if finish {
                        callback(Event::DownloadDone {
                            index: self.download_list_index,
                            msg: msg.into(),
                        })
                        .await;

                        return DownloadResult::Success(SuccessSummary {
                            file_name: self.entry.filename.to_string(),
                            url: self.entry.source.first().unwrap().url.to_string(),
                            index: self.download_list_index,
                            wrote: false,
                        });
                    }

                    // The checksum helper already reported the stale file's
                    // bytes; undo them so the bar only counts the re-download.
                    if read != 0 {
                        callback(Event::GlobalProgressSub(read)).await;
                    }
                }
            }
        }

        let mut sources = self.entry.source.clone();
        assert!(!sources.is_empty());

        // Use a stable sort so that sources with the same priority keep their
        // given order: the primary source first, fallbacks after it.
        sources.sort_by(|a, b| b.source_type.cmp(&a.source_type));

        for (index, c) in sources.iter().enumerate() {
            let download_res = match &c.source_type {
                DownloadSourceType::Http => self.try_http_download(c, callback).await,
                DownloadSourceType::Local(as_symlink) => {
                    self.download_local(c, *as_symlink, callback).await
                }
            };

            match download_res {
                Ok(wrote) => {
                    if let Some(ref final_dir) = self.entry.final_dir {
                        let current_path = self.entry.dir.join(&*self.entry.filename);
                        let target_path = final_dir.join(&*self.entry.filename);

                        if !final_dir.is_dir()
                            && let Err(e) = tokio::fs::create_dir_all(final_dir).await
                        {
                            callback(Event::Failed {
                                file_name: final_dir.to_string_lossy().to_string(),
                                error: SingleDownloadError::Create { source: e },
                            })
                            .await;
                            return DownloadResult::Failed {
                                file_name: self.entry.filename.to_string(),
                            };
                        }

                        if current_path.is_file() {
                            trace!(
                                "Moving completed file from {} to {}",
                                current_path.display(),
                                target_path.display()
                            );
                            if let Err(e) = tokio::fs::rename(&current_path, &target_path).await {
                                callback(Event::Failed {
                                    file_name: self.entry.filename.clone(),
                                    error: SingleDownloadError::Write { source: e },
                                })
                                .await;
                                return DownloadResult::Failed {
                                    file_name: self.entry.filename.to_string(),
                                };
                            }
                        }
                    }

                    callback(Event::DownloadDone {
                        index: self.download_list_index,
                        msg: msg.into(),
                    })
                    .await;

                    return DownloadResult::Success(SuccessSummary {
                        file_name: self.entry.filename.to_string(),
                        url: c.url.clone(),
                        index: self.download_list_index,
                        wrote,
                    });
                }
                Err(e) => {
                    if index == sources.len() - 1 {
                        callback(Event::Failed {
                            file_name: self.entry.filename.clone(),
                            error: e,
                        })
                        .await;
                        return DownloadResult::Failed {
                            file_name: self.entry.filename.to_string(),
                        };
                    }
                    callback(Event::NextUrl {
                        index: self.download_list_index,
                        file_name: self.entry.filename.to_string(),
                        err: e,
                        by_hash: self.entry.by_hash_fallback,
                    })
                    .await;
                }
            }
        }

        unreachable!()
    }

    /// Download file with retry (http)
    async fn try_http_download<F, Fut>(
        &self,
        source: &DownloadSource,
        callback: &F,
    ) -> Result<bool, SingleDownloadError>
    where
        F: Fn(Event) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = ()> + Send + 'static,
    {
        let mut times = 1;
        let mut allow_resume = self.entry.allow_resume;
        loop {
            match self.http_download(allow_resume, source, callback).await {
                Ok(s) => {
                    return Ok(s);
                }
                Err(ProgressedError { error: e, bytes }) => {
                    // Undo this attempt's global progress contribution before
                    // retrying or handing the failure up, so the retry (or the
                    // next source) starts from a clean bar.
                    if bytes != 0 {
                        callback(Event::GlobalProgressSub(bytes)).await;
                    }
                    match e {
                        SingleDownloadError::ChecksumMismatch => {
                            if self.retry_times == times {
                                return Err(e);
                            }

                            if times > 1 {
                                callback(Event::ChecksumMismatch {
                                    index: self.download_list_index,
                                    filename: self.entry.filename.to_string(),
                                    times,
                                })
                                .await;
                            }

                            times += 1;
                            allow_resume = false;
                        }
                        SingleDownloadError::DownloadTimeout => {
                            if self.retry_times == times {
                                return Err(e);
                            }

                            if times > 1 {
                                callback(Event::Timeout {
                                    filename: self.entry.filename.to_string(),
                                    times,
                                })
                                .await;
                            }

                            times += 1;
                        }
                        e => {
                            return Err(e);
                        }
                    }
                }
            }
        }
    }

    /// Inner download attempt. On error the [`ProgressedError`] carries the
    /// bytes this attempt already reported to the global progress bar, so the
    /// retry loop can undo them before retrying / trying the next source.
    async fn http_download<F, Fut>(
        &self,
        allow_resume: bool,
        source: &DownloadSource,
        callback: &F,
    ) -> Result<bool, ProgressedError>
    where
        F: Fn(Event) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = ()> + Send + 'static,
    {
        let file = self.entry.dir.join(&*self.entry.filename);
        let file_exist = file.exists();
        let file_size = file.metadata().ok().map(|x| x.len()).unwrap_or(0);

        trace!("{} Exist file size is: {file_size}", file.display());
        trace!("{} download url is: {}", file.display(), source.url);
        let is_symlink = file.is_symlink();

        debug!("file {} is symlink = {}", file.display(), is_symlink);

        // Net bytes this attempt has actually reported to the global progress
        // bar so far. Starts at the checksum-seeded bytes of an existing file
        // (the checksum helper adds them), then mirrors downloaded_size once
        // progress updates begin. Reported on every return so the wrapper can
        // undo it.
        let mut reported: u64 = 0;

        // downloaded HTTP content representation size
        let mut downloaded_size: u64 = 0;
        let mut old_downloaded_size: u64 = 0;

        if is_symlink {
            tokio::fs::remove_file(&file).await.context(RemoveSnafu)?;
        }

        let mut validator = self
            .entry
            .hash
            .as_ref()
            .map(|hash| hash.get_validator())
            .unwrap_or(ChecksumValidator::None);

        if file_exist && !is_symlink {
            trace!(
                "File {} already exists, verifying checksum ...",
                self.entry.filename
            );
            downloaded_size = file_size;

            if let Some(hash) = &self.entry.hash {
                trace!("Hash {} exists for the existing file.", hash);

                let mut f = tokio::fs::OpenOptions::new()
                    .read(true)
                    .open(&file)
                    .await
                    .context(OpenSnafu)?;

                let (read, finish) = checksum(callback, &mut f, &mut validator).await;

                if finish {
                    trace!("checksum of {} matches, cache hit!", self.entry.filename);
                    callback(Event::ProgressDone(self.download_list_index)).await;
                    return Ok(false);
                }

                debug!(
                    "checksum mismatch, initiating re-download for file {} ...",
                    self.entry.filename
                );
                old_downloaded_size = read;
                // The checksum helper already added these seeded bytes to the
                // global bar; account for them so a failure before the first
                // progress update undoes them too (otherwise they leak).
                reported = read;
            }

            if source.file_type != CompressType::None {
                downloaded_size = 0;
            }
        }

        callback(Event::NewProgressSpinner {
            index: self.download_list_index,
            msg: self.download_message(),
            total: self.total,
        })
        .await;

        // open destination file
        let mut dest = match tokio::fs::OpenOptions::new()
            .write(true)
            .read(true)
            .create(true)
            .truncate(false)
            .open(&file)
            .await
        {
            Ok(f) => f,
            Err(e) => {
                callback(Event::ProgressDone(self.download_list_index)).await;
                return Err(ProgressedError::new(
                    SingleDownloadError::Create { source: e },
                    reported,
                ));
            }
        };

        let mut buf = Vec::with_capacity(DOWNLOAD_BUFSIZE);
        #[allow(clippy::uninit_vec)]
        unsafe {
            buf.set_len(DOWNLOAD_BUFSIZE)
        };
        let mut total_size: Option<u64> = None;
        let mut first_request = true;
        'download: while if let Some(total_size) = total_size {
            downloaded_size < total_size
        } else {
            true
        } {
            if !allow_resume || source.file_type != CompressType::None {
                // restart download if resume is disallowed
                downloaded_size = 0;
            }

            let old_total_size = total_size;

            // send request
            let mut req = self.client.request(Method::GET, &source.url);

            if downloaded_size != 0 {
                // request for resume
                // assume reqwest's automatic decompression is disabled
                debug!("sending partial request ...");
                req = req.header(RANGE, format!("bytes={downloaded_size}-"));
            } else {
                debug!("sending complete request ...");
            }
            let resp = timeout(self.timeout, send_request(req)).await;
            let resp = match resp {
                Ok(Ok(resp)) => resp,
                Ok(Err(e)) => match e.status() {
                    Some(StatusCode::RANGE_NOT_SATISFIABLE) => {
                        debug!("range not satisfiable from server, restarting ...");
                        downloaded_size = 0;
                        continue 'download;
                    }
                    Some(StatusCode::BAD_REQUEST) => {
                        // some servers reply with Bad Request when Range is invalid
                        // so retry once
                        if downloaded_size == 0 {
                            callback(Event::ProgressDone(self.download_list_index)).await;
                            return Err(ProgressedError::new(
                                SingleDownloadError::ReqwestMiddlewareError { source: e },
                                reported,
                            ));
                        } else {
                            debug!("HTTP Bad Request from server, restarting ...");
                            downloaded_size = 0;
                            continue 'download;
                        }
                    }
                    _ => {
                        callback(Event::ProgressDone(self.download_list_index)).await;
                        return Err(ProgressedError::new(
                            SingleDownloadError::ReqwestMiddlewareError { source: e },
                            reported,
                        ));
                    }
                },
                Err(_) => {
                    callback(Event::ProgressDone(self.download_list_index)).await;
                    return Err(ProgressedError::new(
                        SingleDownloadError::SendRequestTimeout,
                        reported,
                    ));
                }
            };

            // check resume
            let resp_headers = resp.headers();
            'resume: {
                if resp.status() == StatusCode::PARTIAL_CONTENT {
                    match resp_headers.typed_get::<ContentRange>() {
                        Some(range) => {
                            // update total size if possible
                            if let Some(new_total_size) = range.bytes_len() {
                                debug!(
                                    "extracted complete length from Content-Range: {new_total_size}"
                                );
                                total_size = Some(new_total_size);
                            }

                            // check returned range is the expected
                            if let Some((returned_start, _)) = range.bytes_range() {
                                if returned_start != downloaded_size {
                                    // The server didn't send us the request range
                                    // Implementing part combination is too complex, just restart it
                                    debug!("incomplete Content-Range, restarting ...");
                                    downloaded_size = 0;
                                    continue 'download;
                                }
                                debug!("partial request succeeded");
                                break 'resume;
                            } else {
                                // Unsatisfiable Content-Range should never appear in HTTP 206
                                // per RFC 9110. The server implementation is violating RFC.
                                debug!("unsatisfiable Content-Range in HTTP 206, restarting ...");
                                downloaded_size = 0;
                                continue 'download;
                            }
                        }
                        None => {
                            debug!("multi-parts are not supported, restarting ...");
                            downloaded_size = 0;
                            continue 'download;
                        }
                    }
                }
                if resp.status() == StatusCode::OK {
                    // update total size if possible.
                    // Content-Length is the complete length for OK but it may not be the case for other statuses
                    if let Some(length) = resp_headers.typed_get::<ContentLength>() {
                        let length = length.0;
                        debug!("extracted complete length from Content-Length: {length}");
                        total_size = Some(length);
                    }
                }
                if downloaded_size != 0 {
                    // requested partial response, but not getting expected response
                    debug!("range request failed");
                    downloaded_size = 0;
                    // no need to re-send request in this case, the body is already complete
                }
            }
            debug!("response body is at {downloaded_size}/{total_size:?}");
            let is_complete = resp.status() == StatusCode::OK;

            // seek to expected location
            if downloaded_size != old_downloaded_size {
                assert!(downloaded_size == 0 || source.file_type == CompressType::None);
                debug!("moving writer from {old_downloaded_size} to {downloaded_size}");
                if let Err(e) = dest.seek(SeekFrom::Start(0)).await {
                    callback(Event::ProgressDone(self.download_list_index)).await;
                    return Err(ProgressedError::new(
                        SingleDownloadError::Seek { source: e },
                        reported,
                    ));
                }

                if downloaded_size == 0 {
                    validator.reset();
                } else {
                    {
                        // refresh hasher state
                        let mut dest_buf = Vec::with_capacity(downloaded_size.try_into().unwrap());
                        if let Err(e) = dest.read_to_end(&mut dest_buf).await {
                            callback(Event::ProgressDone(self.download_list_index)).await;
                            return Err(ProgressedError::new(
                                SingleDownloadError::Seek { source: e },
                                reported,
                            ));
                        }
                        validator.reset();
                        validator.update(dest_buf);
                    }

                    if let Err(e) = dest.seek(SeekFrom::Start(downloaded_size)).await {
                        callback(Event::ProgressDone(self.download_list_index)).await;
                        return Err(ProgressedError::new(
                            SingleDownloadError::Seek { source: e },
                            reported,
                        ));
                    }
                }
            } else if let Err(e) = dest.seek(SeekFrom::Start(downloaded_size)).await {
                callback(Event::ProgressDone(self.download_list_index)).await;
                return Err(ProgressedError::new(
                    SingleDownloadError::Seek { source: e },
                    reported,
                ));
            }

            // truncate file
            if let Err(e) = dest.set_len(downloaded_size).await {
                callback(Event::ProgressDone(self.download_list_index)).await;
                return Err(ProgressedError::new(
                    SingleDownloadError::Write { source: e },
                    reported,
                ));
            }

            // update progress
            if old_total_size != total_size
                || old_downloaded_size > downloaded_size
                || first_request
            {
                // recreate the progress bar if:
                // 1. total size updated
                // 2. offset moved backwards
                // 3. is the first request (the previous bar is a spinner)
                first_request = false;
                callback(Event::ProgressDone(self.download_list_index)).await;
                callback(Event::NewProgressBar {
                    index: self.download_list_index,
                    msg: self.download_message(),
                    total: self.total,
                    size: total_size.unwrap_or(0),
                })
                .await;
                callback(Event::ProgressInc {
                    index: self.download_list_index,
                    size: downloaded_size,
                })
                .await;
                if old_downloaded_size != downloaded_size {
                    callback(Event::GlobalProgressSub(old_downloaded_size)).await;
                    callback(Event::GlobalProgressAdd(downloaded_size)).await;
                }
            } else if old_downloaded_size < downloaded_size {
                let new_offset = downloaded_size - old_downloaded_size;
                callback(Event::ProgressInc {
                    index: self.download_list_index,
                    size: new_offset,
                })
                .await;
                callback(Event::GlobalProgressAdd(new_offset)).await;
            }

            reported = downloaded_size;
            old_downloaded_size = downloaded_size;

            let stream = resp
                .bytes_stream()
                .map_err(io::Error::other)
                .into_async_read();
            let mut stream = BufReader::new(stream);

            // initialize decompressor
            let reader: &mut (dyn AsyncRead + Unpin + Send) = match source.file_type {
                CompressType::Xz => &mut XzDecoder::new(&mut stream),
                CompressType::Gzip => &mut GzipDecoder::new(&mut stream),
                CompressType::Bz2 => &mut BzDecoder::new(&mut stream),
                CompressType::Zstd => &mut ZstdDecoder::new(&mut stream),
                CompressType::Lzma => &mut LzmaDecoder::new(&mut stream),
                CompressType::Lz4 => &mut Lz4Decoder::new(&mut stream),
                CompressType::None => &mut stream,
            };

            let stream_counter = AtomicUsize::new(0);
            let counted_reader = Counter::new(reader, &stream_counter);
            let mut reader = counted_reader.compat();

            // copy data
            loop {
                let buf_size = match timeout(self.timeout, reader.read(&mut buf[..])).await {
                    Ok(Ok(size)) => size,
                    Ok(Err(e)) => {
                        callback(Event::ProgressDone(self.download_list_index)).await;
                        return Err(ProgressedError::new(
                            SingleDownloadError::BrokenPipe { source: e },
                            reported,
                        ));
                    }
                    Err(_) => {
                        callback(Event::ProgressDone(self.download_list_index)).await;
                        return Err(ProgressedError::new(
                            SingleDownloadError::DownloadTimeout,
                            reported,
                        ));
                    }
                };
                if buf_size == 0 {
                    break; // EOF
                }
                if let Err(e) = dest.write_all(&buf[..buf_size]).await {
                    callback(Event::ProgressDone(self.download_list_index)).await;
                    return Err(ProgressedError::new(
                        SingleDownloadError::Write { source: e },
                        reported,
                    ));
                }
                validator.update(&buf[..buf_size]);

                let http_size = stream_counter.swap(0, Ordering::AcqRel);
                let http_size: u64 = http_size.try_into().unwrap();
                downloaded_size += http_size;
                callback(Event::ProgressInc {
                    index: self.download_list_index,
                    size: http_size,
                })
                .await;
                callback(Event::GlobalProgressAdd(http_size)).await;
                reported = downloaded_size;
            }

            debug!("downloaded {} bytes", downloaded_size - old_downloaded_size);
            if downloaded_size == old_downloaded_size {
                // this should not happen ...
                break 'download;
            }

            if is_complete || total_size.is_none() {
                // total size is unknown, we have to assume that the body is complete
                break 'download;
            }

            old_downloaded_size = downloaded_size;
        }
        debug!("download end, {downloaded_size} bytes");
        callback(Event::ProgressDone(self.download_list_index)).await;

        // verify checksum
        if !validator.finish() {
            debug!("checksum mismatch for {}", self.entry.filename);
            callback(Event::ProgressDone(self.download_list_index)).await;

            // truncate file, avoid attempts to reuse it in retries
            if let Err(e) = dest.set_len(0).await {
                callback(Event::ProgressDone(self.download_list_index)).await;
                return Err(ProgressedError::new(
                    SingleDownloadError::Write { source: e },
                    reported,
                ));
            }

            return Err(ProgressedError::new(
                SingleDownloadError::ChecksumMismatch,
                reported,
            ));
        }
        if matches!(validator, ChecksumValidator::None) {
            trace!(
                "checksum verification succeeded for {}",
                self.entry.filename
            );
        }

        // flush
        if let Err(e) = dest.shutdown().await {
            callback(Event::ProgressDone(self.download_list_index)).await;
            return Err(ProgressedError::new(
                SingleDownloadError::Flush { source: e },
                reported,
            ));
        }

        callback(Event::ProgressDone(self.download_list_index)).await;
        Ok(true)
    }

    /// Download local source file
    async fn download_local<F, Fut>(
        &self,
        source: &DownloadSource,
        as_symlink: bool,
        callback: &F,
    ) -> Result<bool, SingleDownloadError>
    where
        F: Fn(Event) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = ()> + Send + 'static,
    {
        debug!("{:?}", self.entry);

        let url = source.url.strip_prefix("file:").unwrap();

        let url_path = Path::new(url);

        let total_size = tokio::fs::metadata(url_path)
            .await
            .context(OpenSnafu)?
            .len();

        let file = self.entry.dir.join(&*self.entry.filename);
        if file.is_symlink() || (as_symlink && file.is_file()) {
            tokio::fs::remove_file(&file).await.context(RemoveSnafu)?;
        }

        if as_symlink {
            if let Some(hash) = &self.entry.hash {
                self.checksum_local(callback, url_path, hash).await?;
            }

            tokio::fs::symlink(url_path, file)
                .await
                .context(CreateSymlinkSnafu)?;

            return Ok(true);
        }

        callback(Event::NewProgressBar {
            index: self.download_list_index,
            total: self.total,
            msg: self.download_message(),
            size: total_size,
        })
        .await;

        trace!("Path for file: {}", url_path.display());

        let from = File::open(&url_path).await.context(CreateSnafu)?;
        let from = tokio::io::BufReader::new(from).compat();

        trace!("Successfully opened file: {}", url_path.display());

        let mut to = File::create(self.entry.dir.join(&*self.entry.filename))
            .await
            .context(CreateSnafu)?;

        let reader: &mut (dyn AsyncRead + Unpin + Send) = match source.file_type {
            CompressType::Xz => &mut XzDecoder::new(BufReader::new(from)),
            CompressType::Gzip => &mut GzipDecoder::new(BufReader::new(from)),
            CompressType::Bz2 => &mut BzDecoder::new(BufReader::new(from)),
            CompressType::Zstd => &mut ZstdDecoder::new(BufReader::new(from)),
            CompressType::Lzma => &mut LzmaDecoder::new(BufReader::new(from)),
            CompressType::Lz4 => &mut Lz4Decoder::new(BufReader::new(from)),
            CompressType::None => &mut BufReader::new(from),
        };

        let mut reader = reader.compat();

        trace!(
            "Successfully created file: {}",
            self.entry.dir.join(&*self.entry.filename).display()
        );

        // Single cleanup point: undo the local copy's global progress on
        // failure, so a fallback to the next source starts from a clean bar.
        match self
            .download_local_copy(&mut to, &mut reader, callback)
            .await
        {
            Ok(()) => Ok(true),
            Err(ProgressedError { error, bytes }) => {
                if bytes != 0 {
                    callback(Event::GlobalProgressSub(bytes)).await;
                }
                // Undo this source's bar: the copy loop cannot report a
                // `ProgressDone` itself (it never mounted a bar), so do it
                // here before handing the error up.
                callback(Event::ProgressDone(self.download_list_index)).await;
                Err(error)
            }
        }
    }

    /// Copy loop for a local source. On error the [`ProgressedError`] carries
    /// the bytes already reported to the global progress bar, so the caller
    /// can undo them.
    async fn download_local_copy<R, F, Fut>(
        &self,
        to: &mut File,
        reader: &mut R,
        callback: &F,
    ) -> Result<(), ProgressedError>
    where
        R: tokio::io::AsyncRead + Unpin + Send,
        F: Fn(Event) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = ()> + Send + 'static,
    {
        let mut v = self
            .entry
            .hash
            .as_ref()
            .map(|v| v.get_validator())
            .unwrap_or(ChecksumValidator::None);

        let mut buf = vec![0u8; READ_FILE_BUFSIZE];
        let mut self_progress = 0;

        loop {
            let size = match reader.read(&mut buf[..]).await {
                Ok(size) => size,
                Err(e) => {
                    callback(Event::ProgressDone(self.download_list_index)).await;
                    return Err(ProgressedError::new(
                        SingleDownloadError::BrokenPipe { source: e },
                        self_progress as u64,
                    ));
                }
            };

            if size == 0 {
                break;
            }

            if let Err(e) = to.write_all(&buf[..size]).await {
                callback(Event::ProgressDone(self.download_list_index)).await;
                return Err(ProgressedError::new(
                    SingleDownloadError::Write { source: e },
                    self_progress as u64,
                ));
            }

            self_progress += size;

            callback(Event::ProgressInc {
                index: self.download_list_index,
                size: size as u64,
            })
            .await;

            v.update(&buf[..size]);
            callback(Event::GlobalProgressAdd(size as u64)).await;
        }

        if !v.finish() {
            callback(Event::ProgressDone(self.download_list_index)).await;
            return Err(ProgressedError::new(
                SingleDownloadError::ChecksumMismatch,
                self_progress as u64,
            ));
        }

        callback(Event::ProgressDone(self.download_list_index)).await;

        Ok(())
    }

    async fn checksum_local<F, Fut>(
        &self,
        callback: &F,
        url_path: &Path,
        hash: &crate::checksum::Checksum,
    ) -> Result<(), SingleDownloadError>
    where
        F: Fn(Event) -> Fut + Send + Sync + 'static,
        Fut: std::future::Future<Output = ()> + Send + 'static,
    {
        let mut f = fs::File::open(url_path).await.context(OpenSnafu)?;
        let (size, finish) = checksum(callback, &mut f, &mut hash.get_validator()).await;

        if !finish {
            callback(Event::GlobalProgressSub(size)).await;
            callback(Event::ProgressDone(self.download_list_index)).await;
            return Err(SingleDownloadError::ChecksumMismatch);
        }

        Ok(())
    }

    fn download_message(&self) -> String {
        self.entry
            .msg
            .as_deref()
            .unwrap_or(&self.entry.filename)
            .to_string()
    }
}

async fn checksum<F, Fut>(callback: &F, f: &mut File, v: &mut ChecksumValidator) -> (u64, bool)
where
    F: Fn(Event) -> Fut + Send + Sync + 'static,
    Fut: std::future::Future<Output = ()> + Send + 'static,
{
    let mut reader = tokio::io::BufReader::with_capacity(READ_FILE_BUFSIZE, f);

    let mut read = 0;

    loop {
        let buffer = match reader.fill_buf().await {
            Ok([]) => break,
            Ok(buffer) => buffer,
            Err(e) => {
                debug!("Error while reading file: {e}");
                break;
            }
        };

        v.update(buffer);

        callback(Event::GlobalProgressAdd(buffer.len() as u64)).await;
        read += buffer.len() as u64;
        let len = buffer.len();

        reader.consume(len);
    }

    (read, v.finish())
}

pub struct Counter<'a, D> {
    inner: D,
    bytes: &'a AtomicUsize,
}

impl<'a, D> Counter<'a, D> {
    #[inline]
    pub const fn new(inner: D, bytes: &'a AtomicUsize) -> Self {
        Self { inner, bytes }
    }

    #[inline]
    pub fn take_read_bytes(&self) -> usize {
        self.bytes.swap(0, Ordering::AcqRel)
    }
}

impl<R: AsyncRead + Unpin> AsyncRead for Counter<'_, R> {
    fn poll_read(
        self: Pin<&mut Self>,
        ctx: &mut Context<'_>,
        buf: &mut [u8],
    ) -> Poll<std::io::Result<usize>> {
        let counter = self.get_mut();
        let pin = Pin::new(&mut counter.inner);

        let poll = pin.poll_read(ctx, buf);
        if let Poll::Ready(Ok(bytes)) = poll {
            counter.bytes.fetch_add(bytes, Ordering::AcqRel);
        }

        poll
    }
}

impl<R: AsyncBufRead + Unpin> AsyncBufRead for Counter<'_, R> {
    fn poll_fill_buf(self: Pin<&mut Self>, ctx: &mut Context<'_>) -> Poll<std::io::Result<&[u8]>> {
        let counter = self.get_mut();
        let pin = Pin::new(&mut counter.inner);
        pin.poll_fill_buf(ctx)
    }

    fn consume(self: Pin<&mut Self>, amt: usize) {
        let counter = self.get_mut();
        counter.bytes.fetch_add(amt, Ordering::AcqRel);
        let pin = Pin::new(&mut counter.inner);
        pin.consume(amt);
    }
}