rustfs/crates/rio/src/hash_reader.rs

// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

//! HashReader implementation with stream-first construction helpers.
//!
//! `HashReader` still stores a dynamic reader internally so it can preserve
//! capability-aware wrapping behavior. For plain async readers, prefer
//! `HashReader::from_stream(...)`. Use `HashReader::new(...)` when the input is
//! already a `DynReader` or when compatibility with existing boxed wrapping
//! logic matters.
//!
//! ## Construction patterns
//!
//! `HashReader::new(...)` keeps the original dyn-reader behavior:
//!
//! ### Original logic (before generics):
//! ```ignore
//! // Original code would do:
//! // 1. Check if inner is already a HashReader
//! // 2. If size > 0, wrap with HardLimitReader  
//! // 3. If !diskable_md5, wrap with EtagReader
//! // 4. Create HashReader with the wrapped reader
//!
//! let reader = HashReader::new(inner, size, actual_size, etag, diskable_md5)?;
//! ```
//!
//! ### New generic approach:
//! ```rust
//! use rustfs_rio::{HashReader, HardLimitReader, EtagReader};
//! use tokio::io::BufReader;
//! use std::io::Cursor;
//!
//! # tokio_test::block_on(async {
//! let data = b"hello world";
//! let reader = BufReader::new(Cursor::new(&data[..]));
//! let size = data.len() as i64;
//! let actual_size = size;
//! let etag = None;
//! let diskable_md5 = false;
//!
//! // Method 1: Simple creation (recommended for most cases)
//! let hash_reader = HashReader::from_stream(reader, size, actual_size, etag.clone(), None, diskable_md5).expect("operation should succeed");
//!
//! // Method 2: With a capability-aware typed wrapper
//! let reader2 = BufReader::new(Cursor::new(&data[..]));
//! let reader2 = HashReader::from_stream(reader2, size, actual_size, etag.clone(), None, diskable_md5).expect("operation should succeed");
//! let wrapped_reader = EtagReader::new(HardLimitReader::new(reader2, size), etag.clone());
//! let hash_reader2 = HashReader::from_reader(wrapped_reader, size, actual_size, etag.clone(), None, diskable_md5).expect("operation should succeed");
//! # });
//! ```
//!
//! ## HashReader Detection
//!
//! The `HashReaderDetector` trait allows detection of existing HashReader instances:
//!
//! ```rust
//! use rustfs_rio::{HashReader, HashReaderDetector};
//! use tokio::io::BufReader;
//! use std::io::Cursor;
//!
//! # tokio_test::block_on(async {
//! let data = b"test";
//! let reader = BufReader::new(Cursor::new(&data[..]));
//! let hash_reader = HashReader::from_stream(reader, 4, 4, None, None,false).expect("operation should succeed");
//!
//! // Check if a type is a HashReader
//! assert!(hash_reader.is_hash_reader());
//!
//! // `from_stream` is the recommended entry point for plain readers
//! let reader2 = BufReader::new(Cursor::new(&data[..]));
//! let result = HashReader::from_stream(reader2, 4, 4, None, None, false);
//! assert!(result.is_ok());
//! # });
//! ```

use crate::Checksum;
use crate::ChecksumHasher;
use crate::ChecksumType;
use crate::Sha256Hasher;
use crate::compress_index::{Index, TryGetIndex};
use crate::get_content_checksum;
use crate::trailer::SharedTrailerSource;
use crate::{DynReader, EtagReader, EtagResolvable, HardLimitReader, HashReaderDetector, WarpReader, boxed_reader, wrap_reader};

use http::HeaderMap;
use pin_project_lite::pin_project;
use std::collections::HashMap;
use std::io::Cursor;
use std::io::Write;
use std::mem;
use std::pin::Pin;
use std::task::{Context, Poll};
use tokio::io::{AsyncRead, ReadBuf};
use tracing::error;

/// Trait for mutable operations on HashReader
pub trait HashReaderMut {
    fn into_inner(self) -> DynReader;
    fn take_inner(&mut self) -> DynReader;
    fn bytes_read(&self) -> u64;
    fn checksum(&self) -> &Option<String>;
    fn set_checksum(&mut self, checksum: Option<String>);
    fn size(&self) -> i64;
    fn set_size(&mut self, size: i64);
    fn actual_size(&self) -> i64;
    fn set_actual_size(&mut self, actual_size: i64);
    fn content_hash(&self) -> &Option<Checksum>;
    fn content_sha256(&self) -> &Option<String>;
    fn get_trailer(&self) -> Option<&SharedTrailerSource>;
    fn set_trailer(&mut self, trailer: Option<SharedTrailerSource>);
}

pin_project! {

    pub struct HashReader {
        #[pin]
        pub inner: DynReader,
        pub size: i64,
        checksum: Option<String>,
        pub actual_size: i64,
        pub diskable_md5: bool,
        bytes_read: u64,
        content_hash: Option<Checksum>,
        content_hasher: Option<Box<dyn ChecksumHasher>>,
        content_sha256: Option<String>,
        content_sha256_hasher: Option<Sha256Hasher>,
        checksum_on_finish: bool,

        // Read only at EOF; see the timing contract in `crate::trailer`.
        trailer: Option<SharedTrailerSource>,

    }

}

impl HashReader {
    /// Used for transformation layers (compression/encryption)
    pub const SIZE_PRESERVE_LAYER: i64 = -1;

    pub fn from_reader<R>(
        inner: R,
        size: i64,
        actual_size: i64,
        md5hex: Option<String>,
        sha256hex: Option<String>,
        diskable_md5: bool,
    ) -> std::io::Result<Self>
    where
        R: crate::Reader + 'static,
    {
        let inner = if size > 0 {
            let hard_limit_reader = HardLimitReader::new(inner, size);
            if !diskable_md5 {
                boxed_reader(EtagReader::new(hard_limit_reader, md5hex.clone()))
            } else {
                boxed_reader(hard_limit_reader)
            }
        } else if size != Self::SIZE_PRESERVE_LAYER && !diskable_md5 {
            boxed_reader(EtagReader::new(inner, md5hex.clone()))
        } else {
            boxed_reader(inner)
        };

        Ok(Self {
            inner,
            size,
            checksum: md5hex,
            actual_size,
            diskable_md5,
            bytes_read: 0,
            content_hash: None,
            content_hasher: None,
            content_sha256: sha256hex.clone(),
            content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
            checksum_on_finish: false,
            trailer: None,
        })
    }

    pub fn from_stream<R>(
        inner: R,
        size: i64,
        actual_size: i64,
        md5hex: Option<String>,
        sha256hex: Option<String>,
        diskable_md5: bool,
    ) -> std::io::Result<Self>
    where
        R: crate::ReadStream + 'static,
    {
        let inner = WarpReader::new(inner);
        let inner = if size > 0 {
            if !diskable_md5 {
                boxed_reader(EtagReader::new(HardLimitReader::new(inner, size), md5hex.clone()))
            } else {
                boxed_reader(HardLimitReader::new(inner, size))
            }
        } else if size != Self::SIZE_PRESERVE_LAYER && !diskable_md5 {
            boxed_reader(EtagReader::new(inner, md5hex.clone()))
        } else {
            boxed_reader(inner)
        };

        Ok(Self {
            inner,
            size,
            checksum: md5hex,
            actual_size,
            diskable_md5,
            bytes_read: 0,
            content_hash: None,
            content_hasher: None,
            content_sha256: sha256hex.clone(),
            content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
            checksum_on_finish: false,
            trailer: None,
        })
    }

    pub fn new(
        mut inner: DynReader,
        size: i64,
        actual_size: i64,
        md5hex: Option<String>,
        sha256hex: Option<String>,
        diskable_md5: bool,
    ) -> std::io::Result<Self> {
        // Get the innermost HashReader
        if size != Self::SIZE_PRESERVE_LAYER
            && let Some(existing_hash_reader) = inner.as_hash_reader_mut()
        {
            if existing_hash_reader.bytes_read() > 0 {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    "Cannot create HashReader from an already read HashReader",
                ));
            }

            if let Some(checksum) = existing_hash_reader.checksum()
                && let Some(ref md5) = md5hex
                && checksum != md5
            {
                return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "HashReader checksum mismatch"));
            }

            if existing_hash_reader.size() > 0 && size > 0 && existing_hash_reader.size() != size {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::InvalidData,
                    format!("HashReader size mismatch: expected {}, got {}", existing_hash_reader.size(), size),
                ));
            }

            existing_hash_reader.set_checksum(md5hex.clone());

            if existing_hash_reader.size() < 0 && size >= 0 {
                existing_hash_reader.set_size(size);
            }

            if existing_hash_reader.actual_size() <= 0 && actual_size >= 0 {
                existing_hash_reader.set_actual_size(actual_size);
            }

            let size = existing_hash_reader.size();
            let actual_size = existing_hash_reader.actual_size();
            let content_hash = existing_hash_reader.content_hash().clone();
            let content_hasher = existing_hash_reader
                .content_hash()
                .clone()
                .map(|hash| hash.checksum_type.hasher().expect("operation should succeed"));
            let content_sha256 = existing_hash_reader.content_sha256().clone();
            let content_sha256_hasher = existing_hash_reader.content_sha256().clone().map(|_| Sha256Hasher::new());
            let inner = existing_hash_reader.take_inner();

            Ok(Self {
                inner,
                size,
                checksum: md5hex,
                actual_size,
                diskable_md5,
                bytes_read: 0,
                content_sha256,
                content_sha256_hasher,
                content_hash,
                content_hasher,
                checksum_on_finish: false,
                trailer: existing_hash_reader.get_trailer().cloned(),
            })
        } else {
            if size > 0 {
                let hr = HardLimitReader::new(inner, size);
                inner = Box::new(hr);

                if !diskable_md5 && !inner.is_hash_reader() {
                    let er = EtagReader::new(inner, md5hex.clone());
                    inner = Box::new(er);
                }
            } else if size != Self::SIZE_PRESERVE_LAYER && !diskable_md5 {
                let er = EtagReader::new(inner, md5hex.clone());
                inner = Box::new(er);
            }

            Ok(Self {
                inner,
                size,
                checksum: md5hex,
                actual_size,
                diskable_md5,
                bytes_read: 0,
                content_hash: None,
                content_hasher: None,
                content_sha256: sha256hex.clone(),
                content_sha256_hasher: sha256hex.map(|_| Sha256Hasher::new()),
                checksum_on_finish: false,
                trailer: None,
            })
        }
    }

    pub fn into_inner(self) -> DynReader {
        self.inner
    }

    /// Update HashReader parameters
    pub fn update_params(&mut self, size: i64, actual_size: i64, etag: Option<String>) {
        if self.size < 0 && size >= 0 {
            self.size = size;
        }

        if self.actual_size <= 0 && actual_size > 0 {
            self.actual_size = actual_size;
        }

        if etag.is_some() {
            self.checksum = etag;
        }
    }

    pub fn size(&self) -> i64 {
        self.size
    }
    pub fn actual_size(&self) -> i64 {
        self.actual_size
    }

    /// Records the request checksum declared in `headers`. For a trailing
    /// checksum, `trailer` supplies its value once the body has been read.
    pub fn add_checksum(
        &mut self,
        headers: &HeaderMap,
        trailer: Option<SharedTrailerSource>,
        ignore_value: bool,
    ) -> Result<(), std::io::Error> {
        let cs = get_content_checksum(headers)?;

        if ignore_value {
            return Ok(());
        }

        if let Some(checksum) = cs {
            if checksum.checksum_type.trailing() {
                self.trailer = trailer;
            }

            self.content_hash = Some(checksum.clone());

            return self.add_non_trailing_checksum(Some(checksum), ignore_value);
        }

        Ok(())
    }

    pub fn add_checksum_no_trailer(&mut self, header: &HeaderMap, ignore_value: bool) -> Result<(), std::io::Error> {
        let cs = get_content_checksum(header)?;

        if let Some(checksum) = cs {
            self.content_hash = Some(checksum.clone());

            return self.add_non_trailing_checksum(Some(checksum), ignore_value);
        }
        Ok(())
    }

    pub fn add_non_trailing_checksum(&mut self, checksum: Option<Checksum>, ignore_value: bool) -> Result<(), std::io::Error> {
        if let Some(checksum) = checksum {
            self.content_hash = Some(checksum.clone());

            if ignore_value {
                return Ok(());
            }

            if let Some(hasher) = checksum.checksum_type.hasher() {
                self.content_hasher = Some(hasher);
            } else {
                return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid checksum type"));
            }

            tracing::debug!("add_non_trailing_checksum checksum={checksum:?}");
        }
        Ok(())
    }

    pub fn add_calculated_checksum(&mut self, checksum_type: ChecksumType) -> Result<(), std::io::Error> {
        if !checksum_type.is_set() {
            return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid checksum type"));
        }

        let Some(hasher) = checksum_type.hasher() else {
            return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, "Invalid checksum type"));
        };

        self.content_hash = Some(Checksum {
            checksum_type,
            ..Default::default()
        });
        self.content_hasher = Some(hasher);

        Ok(())
    }

    pub fn checksum(&self) -> Option<Checksum> {
        if self
            .content_hash
            .as_ref()
            .is_none_or(|v| !v.checksum_type.is_set() || !v.valid())
        {
            return None;
        }
        self.content_hash.clone()
    }
    pub fn content_crc_type(&self) -> Option<ChecksumType> {
        self.content_hash.as_ref().map(|v| v.checksum_type)
    }

    pub fn content_crc(&self) -> HashMap<String, String> {
        let mut map = HashMap::new();
        if let Some(checksum) = self.content_hash.as_ref() {
            if !checksum.valid() || checksum.checksum_type.is(ChecksumType::NONE) {
                return map;
            }

            if checksum.checksum_type.trailing() {
                if let Some(checksum_str) = trailing_checksum_value(self.trailer.as_ref(), checksum.checksum_type) {
                    map.insert(checksum.checksum_type.to_string(), checksum_str);
                }
                return map;
            }

            map.insert(checksum.checksum_type.to_string(), checksum.encoded.clone());

            return map;
        }
        map
    }
}

/// Returns the received trailer value for `checksum_type`. `Pending` and
/// `Missing` both yield `None`, which leaves a trailing checksum unverified
/// and therefore rejected at EOF.
fn trailing_checksum_value(trailer: Option<&SharedTrailerSource>, checksum_type: ChecksumType) -> Option<String> {
    let key = checksum_type.key()?;
    trailer?.lookup(key).into_present()
}

impl HashReaderMut for HashReader {
    fn into_inner(self) -> DynReader {
        self.inner
    }

    fn take_inner(&mut self) -> DynReader {
        // Replace inner with an empty reader to move it out safely while keeping self valid
        mem::replace(&mut self.inner, wrap_reader(Cursor::new(Vec::new())))
    }

    fn bytes_read(&self) -> u64 {
        self.bytes_read
    }

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

    fn set_checksum(&mut self, checksum: Option<String>) {
        self.checksum = checksum;
    }

    fn size(&self) -> i64 {
        self.size
    }

    fn set_size(&mut self, size: i64) {
        self.size = size;
    }

    fn actual_size(&self) -> i64 {
        self.actual_size
    }

    fn set_actual_size(&mut self, actual_size: i64) {
        self.actual_size = actual_size;
    }

    fn content_hash(&self) -> &Option<Checksum> {
        &self.content_hash
    }

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

    fn get_trailer(&self) -> Option<&SharedTrailerSource> {
        self.trailer.as_ref()
    }

    fn set_trailer(&mut self, trailer: Option<SharedTrailerSource>) {
        self.trailer = trailer;
    }
}

impl AsyncRead for HashReader {
    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
        let this = self.project();

        let before = buf.filled().len();
        match this.inner.poll_read(cx, buf) {
            Poll::Pending => Poll::Pending,
            Poll::Ready(Ok(())) => crate::hp_measure_block!("HashReader::on_chunk", {
                let data = &buf.filled()[before..];
                let filled = data.len();

                *this.bytes_read += filled as u64;

                if filled > 0 {
                    // Update SHA256 hasher
                    if let Some(hasher) = this.content_sha256_hasher
                        && let Err(e) = hasher.write_all(data)
                    {
                        error!("SHA256 hasher write error, error={:?}", e);
                        return Poll::Ready(Err(std::io::Error::other(e)));
                    }

                    // Update content hasher
                    if let Some(hasher) = this.content_hasher
                        && let Err(e) = hasher.write_all(data)
                    {
                        return Poll::Ready(Err(std::io::Error::other(e)));
                    }
                }

                if filled == 0 && !*this.checksum_on_finish {
                    // check SHA256
                    if let (Some(hasher), Some(expected_sha256)) = (this.content_sha256_hasher, this.content_sha256) {
                        let sha256 = hex_simd::encode_to_string(hasher.finalize(), hex_simd::AsciiCase::Lower);
                        if sha256 != *expected_sha256 {
                            error!("SHA256 mismatch, expected={:?}, actual={:?}", expected_sha256, sha256);
                            return Poll::Ready(Err(std::io::Error::new(
                                std::io::ErrorKind::InvalidData,
                                crate::errors::Sha256Mismatch {
                                    expected_sha256: expected_sha256.clone(),
                                    calculated_sha256: sha256,
                                },
                            )));
                        }
                    }

                    // check content hasher
                    if let (Some(hasher), Some(expected_content_hash)) = (this.content_hasher, this.content_hash) {
                        if expected_content_hash.checksum_type.trailing()
                            && let Some(checksum_str) =
                                trailing_checksum_value(this.trailer.as_ref(), expected_content_hash.checksum_type)
                        {
                            expected_content_hash.encoded = checksum_str;
                            expected_content_hash.raw = base64_simd::STANDARD
                                .decode_to_vec(&expected_content_hash.encoded)
                                .map_err(|_| std::io::Error::other("Invalid base64 checksum"))?;

                            if expected_content_hash.raw.is_empty() {
                                return Poll::Ready(Err(std::io::Error::other("Content hash mismatch")));
                            }
                        }

                        let content_hash = hasher.finalize();

                        if expected_content_hash.raw.is_empty()
                            && expected_content_hash.encoded.is_empty()
                            && !expected_content_hash.checksum_type.trailing()
                        {
                            expected_content_hash.raw = content_hash;
                            expected_content_hash.encoded = base64_simd::STANDARD.encode_to_string(&expected_content_hash.raw);
                        } else if content_hash != expected_content_hash.raw {
                            let expected_hex = hex_simd::encode_to_string(&expected_content_hash.raw, hex_simd::AsciiCase::Lower);
                            let actual_hex = hex_simd::encode_to_string(content_hash, hex_simd::AsciiCase::Lower);
                            error!(
                                "Content hash mismatch, type={:?}, encoded={:?}, expected={:?}, actual={:?}",
                                expected_content_hash.checksum_type, expected_content_hash.encoded, expected_hex, actual_hex
                            );
                            // Use ChecksumMismatch error so that API layer can return BadDigest
                            let checksum_err = crate::errors::ChecksumMismatch {
                                want: expected_hex,
                                got: actual_hex,
                            };
                            return Poll::Ready(Err(std::io::Error::new(std::io::ErrorKind::InvalidData, checksum_err)));
                        }
                    }

                    *this.checksum_on_finish = true;
                }
                Poll::Ready(Ok(()))
            }),
            Poll::Ready(Err(e)) => Poll::Ready(Err(e)),
        }
    }
}

impl EtagResolvable for HashReader {
    fn try_resolve_etag(&mut self) -> Option<String> {
        if self.diskable_md5 {
            return None;
        }
        if let Some(etag) = self.inner.try_resolve_etag() {
            return Some(etag);
        }
        // If no etag from inner and we have a stored checksum, return it
        self.checksum.clone()
    }
}

impl HashReaderDetector for HashReader {
    fn is_hash_reader(&self) -> bool {
        true
    }

    fn as_hash_reader_mut(&mut self) -> Option<&mut dyn HashReaderMut> {
        Some(self)
    }
}

impl TryGetIndex for HashReader {
    fn try_get_index(&self) -> Option<&Index> {
        self.inner.try_get_index()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::{DecryptReader, EncryptReader, encrypt_reader, wrap_reader};
    use rand::RngExt;
    use std::io::Cursor;
    use tokio::io::{AsyncReadExt, BufReader};

    #[tokio::test]
    async fn test_hashreader_wrapping_logic() {
        let data = b"hello world";
        let size = data.len() as i64;
        let actual_size = size;
        let etag = None;

        // Test 1: Simple creation
        let reader1 = BufReader::new(Cursor::new(&data[..]));
        let hash_reader1 =
            HashReader::from_stream(reader1, size, actual_size, etag.clone(), None, false).expect("operation should succeed");
        assert_eq!(hash_reader1.size(), size);
        assert_eq!(hash_reader1.actual_size(), actual_size);

        // Test 2: With HardLimitReader wrapping
        let reader2 =
            HashReader::from_stream(BufReader::new(Cursor::new(&data[..])), size, actual_size, etag.clone(), None, false)
                .expect("operation should succeed");
        let hard_limit = HardLimitReader::new(reader2, size);
        let hash_reader2 =
            HashReader::from_reader(hard_limit, size, actual_size, etag.clone(), None, false).expect("operation should succeed");
        assert_eq!(hash_reader2.size(), size);
        assert_eq!(hash_reader2.actual_size(), actual_size);

        // Test 3: With EtagReader wrapping
        let reader3 =
            HashReader::from_stream(BufReader::new(Cursor::new(&data[..])), size, actual_size, etag.clone(), None, false)
                .expect("operation should succeed");
        let etag_reader = EtagReader::new(reader3, etag.clone());
        let hash_reader3 =
            HashReader::from_reader(etag_reader, size, actual_size, etag, None, false).expect("operation should succeed");
        assert_eq!(hash_reader3.size(), size);
        assert_eq!(hash_reader3.actual_size(), actual_size);
    }

    #[test]
    fn test_boxed_reader_capabilities_delegate() {
        let data = b"boxed capabilities";
        let mut boxed_etag_reader =
            Box::new(EtagReader::new(BufReader::new(Cursor::new(&data[..])), Some("boxed_etag".to_string())));
        assert_eq!(boxed_etag_reader.try_resolve_etag(), Some("boxed_etag".to_string()));

        let boxed_hash_reader = Box::new(
            HashReader::from_stream(
                BufReader::new(Cursor::new(&data[..])),
                data.len() as i64,
                data.len() as i64,
                None,
                None,
                false,
            )
            .expect("operation should succeed"),
        );
        assert!(boxed_hash_reader.is_hash_reader());
    }

    #[tokio::test]
    async fn test_from_reader_accepts_boxed_encrypt_reader() {
        let data = b"boxed encrypt reader";
        let inner = HashReader::from_stream(
            BufReader::new(Cursor::new(&data[..])),
            data.len() as i64,
            data.len() as i64,
            None,
            None,
            false,
        )
        .expect("operation should succeed");
        let boxed_encrypt_reader = Box::new(EncryptReader::new(inner, [7u8; 32], [3u8; 12]));

        assert!(boxed_encrypt_reader.is_hash_reader());

        let mut hash_reader = HashReader::from_reader(
            boxed_encrypt_reader,
            HashReader::SIZE_PRESERVE_LAYER,
            data.len() as i64,
            None,
            None,
            false,
        )
        .expect("operation should succeed");
        let mut encrypted = Vec::new();
        hash_reader
            .read_to_end(&mut encrypted)
            .await
            .expect("operation should succeed");

        assert!(!encrypted.is_empty());
        assert_ne!(encrypted, data);
        assert_eq!(hash_reader.actual_size(), data.len() as i64);
    }

    #[tokio::test]
    async fn test_hashreader_etag_basic() {
        let data = b"hello hashreader";
        let reader = BufReader::new(Cursor::new(&data[..]));
        let mut hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false)
            .expect("operation should succeed");
        let mut buf = Vec::new();
        let _ = hash_reader.read_to_end(&mut buf).await.expect("operation should succeed");
        let etag = hash_reader.try_resolve_etag();
        assert!(etag.is_some());
        assert_eq!(buf, data);
    }

    #[tokio::test]
    async fn test_hashreader_diskable_md5() {
        let data = b"no etag";
        let reader = BufReader::new(Cursor::new(&data[..]));
        let mut hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, true)
            .expect("operation should succeed");
        let mut buf = Vec::new();
        let _ = hash_reader.read_to_end(&mut buf).await.expect("operation should succeed");
        // Etag should be None when diskable_md5 is true
        let etag = hash_reader.try_resolve_etag();
        assert!(etag.is_none());
        assert_eq!(buf, data);
    }

    #[tokio::test]
    async fn sha256_mismatch_retains_typed_io_error_source() {
        let data = b"tampered payload";
        let expected_sha256 = "0".repeat(64);
        let reader = BufReader::new(Cursor::new(&data[..]));
        let mut hash_reader =
            HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, Some(expected_sha256.clone()), false)
                .expect("operation should succeed");

        let error = hash_reader
            .read_to_end(&mut Vec::new())
            .await
            .expect_err("SHA256 mismatch should fail");
        let mismatch = error
            .get_ref()
            .and_then(|source| source.downcast_ref::<crate::Sha256Mismatch>())
            .expect("SHA256 mismatch should remain typed");

        assert_eq!(error.kind(), std::io::ErrorKind::InvalidData);
        assert_eq!(mismatch.expected_sha256, expected_sha256);
        assert_eq!(mismatch.calculated_sha256.len(), 64);
        assert_ne!(mismatch.calculated_sha256, mismatch.expected_sha256);
    }

    #[tokio::test]
    async fn test_add_calculated_checksum_records_checksum() {
        let data = b"server-side copy checksum";
        let reader = BufReader::new(Cursor::new(&data[..]));
        let mut hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false)
            .expect("operation should succeed");

        hash_reader
            .add_calculated_checksum(ChecksumType::CRC64_NVME)
            .expect("operation should succeed");

        let mut buf = Vec::new();
        hash_reader.read_to_end(&mut buf).await.expect("operation should succeed");

        let expected = Checksum::new_from_data(ChecksumType::CRC64_NVME, data).expect("operation should succeed");
        let checksums = hash_reader.content_crc();

        assert_eq!(buf, data);
        assert_eq!(hash_reader.content_crc_type(), Some(ChecksumType::CRC64_NVME));
        assert_eq!(checksums.get("CRC64NVME"), Some(&expected.encoded));
    }

    #[tokio::test]
    async fn test_hashreader_new_logic() {
        let data = b"test data";
        let reader = BufReader::new(Cursor::new(&data[..]));
        // Create a HashReader first
        let hash_reader =
            HashReader::from_stream(reader, data.len() as i64, data.len() as i64, Some("test_etag".to_string()), None, false)
                .expect("operation should succeed");
        let hash_reader = wrap_reader(hash_reader);
        // Now try to create another HashReader from the existing one using new
        let result = HashReader::new(
            hash_reader,
            data.len() as i64,
            data.len() as i64,
            Some("test_etag".to_string()),
            None,
            false,
        );

        assert!(result.is_ok());
        let final_reader = result.expect("operation should succeed");
        assert_eq!(final_reader.checksum, Some("test_etag".to_string()));
        assert_eq!(final_reader.size(), data.len() as i64);
    }

    #[tokio::test]
    async fn test_for_wrapping_readers() {
        use crate::{CompressReader, DecompressReader};
        use md5::{Digest, Md5};
        use rand::Rng;
        use rustfs_utils::compress::CompressionAlgorithm;

        // Generate 1MB random data
        let size = 1024 * 1024;
        let mut data = vec![0u8; size];
        rand::rng().fill(&mut data[..]);

        let mut hasher = Md5::new();
        hasher.update(&data);

        let hex = faster_hex::hex_string(hasher.finalize().as_slice());
        let expected = hex.to_string();

        println!("expected: {expected}");

        let reader = Cursor::new(data.clone());
        let reader = BufReader::new(reader);

        // Enable compression test
        let is_compress = true;
        let size = data.len() as i64;
        let actual_size = data.len() as i64;

        let mut hr = HashReader::from_stream(reader, size, actual_size, Some(expected.clone()), None, false)
            .expect("operation should succeed");

        // If compression is enabled, compress data first
        let compressed_data = if is_compress {
            let mut compressed_buf = Vec::new();
            let compress_reader = CompressReader::new(hr, CompressionAlgorithm::Gzip);
            let mut compress_reader = compress_reader;
            compress_reader
                .read_to_end(&mut compressed_buf)
                .await
                .expect("operation should succeed");

            println!("Original size: {}, Compressed size: {}", data.len(), compressed_buf.len());

            compressed_buf
        } else {
            // If not compressing, read original data directly
            let mut buf = Vec::new();
            hr.read_to_end(&mut buf).await.expect("operation should succeed");
            buf
        };

        let mut key = [0u8; 32];
        let mut nonce = [0u8; 12];
        rand::rng().fill_bytes(&mut key);
        rand::rng().fill_bytes(&mut nonce);

        let is_encrypt = true;

        if is_encrypt {
            // Encrypt compressed data
            let encrypt_reader = encrypt_reader::EncryptReader::new(Cursor::new(compressed_data), key, nonce);
            let mut encrypted_data = Vec::new();
            let mut encrypt_reader = encrypt_reader;
            encrypt_reader
                .read_to_end(&mut encrypted_data)
                .await
                .expect("operation should succeed");

            println!("Encrypted size: {}", encrypted_data.len());

            // Decrypt data
            let decrypt_reader = DecryptReader::new(Cursor::new(encrypted_data), key, nonce);
            let mut decrypt_reader = decrypt_reader;
            let mut decrypted_data = Vec::new();
            decrypt_reader
                .read_to_end(&mut decrypted_data)
                .await
                .expect("operation should succeed");

            if is_compress {
                // If compression was used, decompress is needed
                let decompress_reader = DecompressReader::new(Cursor::new(decrypted_data), CompressionAlgorithm::Gzip);
                let mut decompress_reader = decompress_reader;
                let mut final_data = Vec::new();
                decompress_reader
                    .read_to_end(&mut final_data)
                    .await
                    .expect("operation should succeed");

                println!("Final decompressed size: {}", final_data.len());
                assert_eq!(final_data.len() as i64, actual_size);
                assert_eq!(&final_data, &data);
            } else {
                // Without compression we can compare the decrypted bytes directly
                assert_eq!(decrypted_data.len() as i64, actual_size);
                assert_eq!(&decrypted_data, &data);
            }
            return;
        }

        // When encryption is disabled, only handle compression/decompression
        if is_compress {
            let decompress_reader = DecompressReader::new(Cursor::new(compressed_data), CompressionAlgorithm::Gzip);
            let mut decompress_reader = decompress_reader;
            let mut decompressed = Vec::new();
            decompress_reader
                .read_to_end(&mut decompressed)
                .await
                .expect("operation should succeed");

            assert_eq!(decompressed.len() as i64, actual_size);
            assert_eq!(&decompressed, &data);
        } else {
            assert_eq!(compressed_data.len() as i64, actual_size);
            assert_eq!(&compressed_data, &data);
        }

        // Validate the etag (compression alters the payload, so this may require adjustments)
        println!("Test completed successfully with compression: {is_compress}, encryption: {is_encrypt}");
    }

    #[tokio::test]
    async fn test_compression_with_compressible_data() {
        use crate::{CompressReader, DecompressReader};
        use rustfs_utils::compress::CompressionAlgorithm;

        // Create highly compressible data (repeated pattern)
        let pattern = b"Hello, World! This is a test pattern that should compress well. ";
        let repeat_count = 16384; // 16K repetitions
        let mut data = Vec::new();
        for _ in 0..repeat_count {
            data.extend_from_slice(pattern);
        }

        println!("Original data size: {} bytes", data.len());

        let reader = BufReader::new(Cursor::new(data.clone()));
        let hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false)
            .expect("operation should succeed");

        // Test compression
        let compress_reader = CompressReader::new(hash_reader, CompressionAlgorithm::Gzip);
        let mut compressed_data = Vec::new();
        let mut compress_reader = compress_reader;
        compress_reader
            .read_to_end(&mut compressed_data)
            .await
            .expect("operation should succeed");

        println!("Compressed data size: {} bytes", compressed_data.len());
        println!("Compression ratio: {:.2}%", (compressed_data.len() as f64 / data.len() as f64) * 100.0);

        // Verify compression actually reduced size for this compressible data
        assert!(compressed_data.len() < data.len(), "Compression should reduce size for repetitive data");

        // Test decompression
        let decompress_reader = DecompressReader::new(Cursor::new(compressed_data), CompressionAlgorithm::Gzip);
        let mut decompressed_data = Vec::new();
        let mut decompress_reader = decompress_reader;
        decompress_reader
            .read_to_end(&mut decompressed_data)
            .await
            .expect("operation should succeed");

        // Verify decompressed data matches original
        assert_eq!(decompressed_data.len(), data.len());
        assert_eq!(&decompressed_data, &data);

        println!("Compression/decompression test passed successfully!");
    }

    #[tokio::test]
    async fn test_compression_algorithms() {
        use crate::{CompressReader, DecompressReader};
        use rustfs_utils::compress::CompressionAlgorithm;

        let data = b"This is test data for compression algorithm testing. ".repeat(1000);
        println!("Testing with {} bytes of data", data.len());

        let algorithms = vec![
            CompressionAlgorithm::Gzip,
            CompressionAlgorithm::Deflate,
            CompressionAlgorithm::Zstd,
        ];

        for algorithm in algorithms {
            println!("\nTesting algorithm: {algorithm:?}");

            let reader = BufReader::new(Cursor::new(data.clone()));
            let hash_reader = HashReader::from_stream(reader, data.len() as i64, data.len() as i64, None, None, false)
                .expect("operation should succeed");

            // Compress
            let compress_reader = CompressReader::new(hash_reader, algorithm);
            let mut compressed_data = Vec::new();
            let mut compress_reader = compress_reader;
            compress_reader
                .read_to_end(&mut compressed_data)
                .await
                .expect("operation should succeed");

            println!(
                "  Compressed size: {} bytes (ratio: {:.2}%)",
                compressed_data.len(),
                (compressed_data.len() as f64 / data.len() as f64) * 100.0
            );

            // Decompress
            let decompress_reader = DecompressReader::new(Cursor::new(compressed_data), algorithm);
            let mut decompressed_data = Vec::new();
            let mut decompress_reader = decompress_reader;
            decompress_reader
                .read_to_end(&mut decompressed_data)
                .await
                .expect("operation should succeed");

            // Verify
            assert_eq!(decompressed_data.len(), data.len());
            assert_eq!(&decompressed_data, &data);
            println!("  ✓ Algorithm {algorithm:?} test passed");
        }
    }
}

#[cfg(test)]
mod trailer_tests {
    use super::*;
    use crate::{TrailerSource, TrailerValue, is_checksum_mismatch};
    use http::HeaderValue;
    use std::sync::{Arc, Mutex};
    use tokio::io::AsyncReadExt;

    const TYPED: [ChecksumType; 5] = [
        ChecksumType::CRC32,
        ChecksumType::CRC32C,
        ChecksumType::SHA1,
        ChecksumType::SHA256,
        ChecksumType::CRC64_NVME,
    ];

    /// Mirrors an aws-chunked decoder handle: `None` until the trailer section
    /// has been published.
    #[derive(Default)]
    struct FakeTrailer(Mutex<Option<HashMap<String, String>>>);

    impl TrailerSource for FakeTrailer {
        fn lookup(&self, name: &str) -> TrailerValue {
            match self.0.lock().expect("trailer lock").as_ref() {
                None => TrailerValue::Pending,
                Some(fields) => fields.get(name).cloned().map_or(TrailerValue::Missing, TrailerValue::Present),
            }
        }
    }

    /// Body that publishes `fields` (when set) only once it reaches EOF.
    struct PublishOnEof {
        body: Cursor<Vec<u8>>,
        trailer: Arc<FakeTrailer>,
        fields: Option<HashMap<String, String>>,
    }

    impl AsyncRead for PublishOnEof {
        fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
            let this = &mut *self;
            let before = buf.filled().len();
            let result = Pin::new(&mut this.body).poll_read(cx, buf);
            if matches!(result, Poll::Ready(Ok(())))
                && buf.filled().len() == before
                && buf.remaining() > 0
                && let Some(fields) = this.fields.clone()
            {
                *this.trailer.0.lock().expect("trailer lock") = Some(fields);
            }
            result
        }
    }

    fn encoded(checksum_type: ChecksumType, data: &[u8]) -> String {
        let mut hasher = checksum_type.hasher().expect("typed checksum hasher");
        hasher.write_all(data).expect("hash data");
        base64_simd::STANDARD.encode_to_string(hasher.finalize())
    }

    fn trailing_request(checksum_type: ChecksumType) -> HeaderMap {
        let mut headers = HeaderMap::new();
        headers.insert(
            "x-amz-trailer",
            HeaderValue::from_static(checksum_type.key().expect("typed checksum key")),
        );
        headers
    }

    /// `published`: `None` never publishes (still `Pending` at EOF);
    /// `Some(None)` publishes a trailer without the field; `Some(Some(v))`
    /// publishes `v` under the checksum's key.
    fn reader(checksum_type: ChecksumType, data: &[u8], published: Option<Option<String>>) -> (HashReader, Arc<FakeTrailer>) {
        let trailer = Arc::new(FakeTrailer::default());
        let key = checksum_type.key().expect("typed checksum key");
        let fields = published.map(|value| value.into_iter().map(|v| (key.to_string(), v)).collect());
        let body = PublishOnEof {
            body: Cursor::new(data.to_vec()),
            trailer: trailer.clone(),
            fields,
        };
        let size = data.len() as i64;
        let mut reader = HashReader::from_stream(body, size, size, None, None, false).expect("hash reader");
        reader
            .add_checksum(&trailing_request(checksum_type), Some(trailer.clone()), false)
            .expect("declared trailing checksum");
        (reader, trailer)
    }

    async fn read_all(reader: &mut HashReader) -> std::io::Result<Vec<u8>> {
        let mut out = Vec::new();
        reader.read_to_end(&mut out).await.map(|_| out)
    }

    fn assert_checksum_mismatch(err: &std::io::Error) {
        let inner = err.get_ref().expect("typed error source");
        assert!(is_checksum_mismatch(inner), "expected ChecksumMismatch, got {err:?}");
    }

    #[tokio::test]
    async fn trailing_checksum_published_at_eof_is_verified_and_reported() {
        let data = b"trailing checksum payload";
        for checksum_type in TYPED {
            let value = encoded(checksum_type, data);
            let (mut reader, trailer) = reader(checksum_type, data, Some(Some(value.clone())));
            let key = checksum_type.key().expect("typed checksum key");
            assert_eq!(
                trailer.lookup(key),
                TrailerValue::Pending,
                "{checksum_type}: attached before the body ends"
            );

            assert_eq!(read_all(&mut reader).await.expect("verified body"), data);
            assert_eq!(
                reader.content_crc(),
                HashMap::from([(checksum_type.to_string(), value)]),
                "{checksum_type}: verified trailer value is reported for persistence"
            );
        }
    }

    #[tokio::test]
    async fn trailing_checksum_mismatch_is_rejected() {
        let data = b"trailing checksum payload";
        for checksum_type in TYPED {
            let wrong = encoded(checksum_type, b"different payload");
            let (mut reader, _) = reader(checksum_type, data, Some(Some(wrong.clone())));
            let err = read_all(&mut reader).await.expect_err("mismatched trailer must fail");
            assert_checksum_mismatch(&err);
            let mismatch = err
                .get_ref()
                .and_then(|e| e.downcast_ref::<crate::errors::ChecksumMismatch>())
                .expect("mismatch details");
            let wrong_raw = base64_simd::STANDARD.decode_to_vec(&wrong).expect("valid base64");
            assert_eq!(mismatch.want, hex_simd::encode_to_string(wrong_raw, hex_simd::AsciiCase::Lower));
        }
    }

    #[tokio::test]
    async fn trailing_checksum_missing_or_still_pending_at_eof_is_rejected() {
        let data = b"trailing checksum payload";
        for published in [None, Some(None)] {
            let (mut reader, _) = reader(ChecksumType::CRC32C, data, published.clone());
            let err = read_all(&mut reader).await.expect_err("unverifiable trailer must fail");
            assert_checksum_mismatch(&err);
            assert!(reader.content_crc().is_empty(), "{published:?}: nothing to persist");
        }

        let size = data.len() as i64;
        let mut reader = HashReader::from_stream(Cursor::new(data.to_vec()), size, size, None, None, false).expect("hash reader");
        reader
            .add_checksum(&trailing_request(ChecksumType::CRC32C), None, false)
            .expect("declared trailing checksum");
        let err = read_all(&mut reader).await.expect_err("no trailer source must fail");
        assert_checksum_mismatch(&err);
        assert!(reader.content_crc().is_empty());
    }

    #[tokio::test]
    async fn malformed_trailing_checksum_is_an_error_not_a_panic() {
        let data = b"trailing checksum payload";
        for value in ["not base64!", ""] {
            let (mut reader, _) = reader(ChecksumType::SHA256, data, Some(Some(value.to_string())));
            let err = read_all(&mut reader).await.expect_err("malformed trailer must fail");
            assert_eq!(err.kind(), std::io::ErrorKind::Other, "{value:?}: {err:?}");
        }
    }

    #[tokio::test]
    async fn wrapping_hash_reader_keeps_the_trailer_source() {
        let data = b"trailing checksum payload";
        let value = encoded(ChecksumType::CRC32, data);
        let size = data.len() as i64;

        let (inner, _) = reader(ChecksumType::CRC32, data, Some(Some(value.clone())));
        let mut outer = HashReader::new(Box::new(inner), size, size, None, None, false).expect("outer hash reader");
        assert!(outer.get_trailer().is_some(), "wrapping must carry the trailer source");
        assert_eq!(read_all(&mut outer).await.expect("verified body"), data);
        assert_eq!(outer.content_crc(), HashMap::from([("CRC32".to_string(), value)]));

        let wrong = encoded(ChecksumType::CRC32, b"different payload");
        let (inner, _) = reader(ChecksumType::CRC32, data, Some(Some(wrong)));
        let mut outer = HashReader::new(Box::new(inner), size, size, None, None, false).expect("outer hash reader");
        let err = read_all(&mut outer).await.expect_err("mismatch through a wrapper must fail");
        assert_checksum_mismatch(&err);
    }
}