//! S3 “single object, many files” blob layout (`MultiFileStream` /
//! `MultiFilePreamble`).
//!
//! Wire format (V1), little-endian:
//!
//! ```text
//!   ┌─────────────── 20-byte preamble ───────────────┐
//!   │ magic[8]       = "TICIMF\0\0"                 │
//!   │ version        = u16 LE = 1                    │
//!   │ flags          = u16 LE = 0 (reserved; must 0)  │
//!   │ manifest_len   = u32 LE                      │
//!   │ manifest_crc32c = u32 LE (CRC32C of manifest) │
//!   └────────────────────────────────────────────────┘
//!   ┌── manifest_len bytes: prost(MultiFileManifest) ┐
//!   │   • frag_id                                     │
//!   │   • files[]: relative_path, offset, size        │
//!   │     offsets packed contiguously from 0, order  │
//!   │     matches payload concatenation order below    │
//!   └────────────────────────────────────────────────┘
//!   ┌── raw file bytes: Σ files[i].size ────────────┐
//!   │   concatenate local file contents in manifest │
//!   │   order (same order as `files` / encoding)    │
//!   └────────────────────────────────────────────────┘
//! ```
//!
//! Readers verify `manifest_crc32c` against the manifest blob before `prost`
//! decode. `MultiFileStream::concat` appends full V1 packages: `[pkg0][pkg1]…`
//! (each pkg is preamble + manifest + payloads).

use std::{io, path::Path, pin::Pin, sync::Arc, task::Poll, time::Instant};

use anyhow::{Context, Result, bail, ensure};
use async_stream::stream;
use bytes::Bytes;
use futures::{Stream, StreamExt};
use lazy_static::lazy_static;
use log::info;
use prost::Message;
use regex::Regex;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWriteExt};
use tokio_util::io::ReaderStream;

use crate::{
    multi_file_proto::{FileEntry, MultiFileManifest},
    s3_error::{DownloadError, UploadError, retry_with_backoff},
    storage::{BackendStorage, ReadRange},
};

/// Hard limit on serialized `MultiFileManifest` size to guard against OOM on
/// corrupt or malicious preamble (the on-wire `manifest_len` field is
/// attacker-controlled).  4 MiB accommodates hundreds of thousands of file
/// entries, far beyond realistic usage, while keeping `vec![0u8; len]` safe.
const MAX_MANIFEST_SIZE: usize = 4 * 1024 * 1024;
const MULTI_FILE_MAGIC_LEN: usize = 8;
const MULTI_FILE_PREAMBLE_LEN: usize = 20;
const MULTI_FILE_SUPPORTED_FLAGS: u16 = 0;
const MULTI_FILE_VERSION_V1: u16 = 1;

lazy_static! {
    static ref MULTI_FILE_RELATIVE_PATH_RE: Regex =
        Regex::new(r"^[A-Za-z0-9._-]+(/[A-Za-z0-9._-]+)*$").unwrap();
}

#[derive(Debug, Clone)]
pub struct FileMeta {
    pub relative_path: String,
    pub size: u64,
}

/// Fixed little-endian header at the start of a multi-file object, followed by
/// `manifest_len` bytes of protobuf `MultiFileManifest`, then the concatenated
/// file payloads.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct MultiFilePreamble {
    /// Format magic (`TICIMF\0\0`) so readers recognize this wrapper and do not
    /// mis-parse arbitrary blobs.
    magic: [u8; MULTI_FILE_MAGIC_LEN],
    /// Wire format version; readers branch on this for forward compatibility.
    /// Only `MULTI_FILE_VERSION_V1` is defined today.
    version: u16,
    /// Reserved feature bits (compression, checksum policy, etc.); unknown bits
    /// are rejected in `validate`. Currently must be 0.
    flags: u16,
    /// Serialized manifest length in bytes, immediately after the preamble;
    /// used to read the protobuf blob with a single `read_exact`.
    manifest_len: u32,
    /// CRC32C over the raw manifest bytes; compared after reading to detect
    /// truncation or corruption before decode.
    manifest_crc32c: u32,
}

impl MultiFilePreamble {
    const MAGIC: [u8; MULTI_FILE_MAGIC_LEN] = *b"TICIMF\0\0";

    fn from_manifest(manifest_bytes: &[u8]) -> Result<Self> {
        let manifest_len = u32::try_from(manifest_bytes.len())
            .with_context(|| format!("Manifest too large: {}", manifest_bytes.len()))?;
        Ok(Self {
            magic: Self::MAGIC,
            version: MULTI_FILE_VERSION_V1,
            flags: MULTI_FILE_SUPPORTED_FLAGS,
            manifest_len,
            manifest_crc32c: crc32c::crc32c(manifest_bytes),
        })
    }

    fn encode(self) -> [u8; MULTI_FILE_PREAMBLE_LEN] {
        let mut bytes = [0u8; MULTI_FILE_PREAMBLE_LEN];
        bytes[..MULTI_FILE_MAGIC_LEN].copy_from_slice(&self.magic);
        bytes[8..10].copy_from_slice(&self.version.to_le_bytes());
        bytes[10..12].copy_from_slice(&self.flags.to_le_bytes());
        bytes[12..16].copy_from_slice(&self.manifest_len.to_le_bytes());
        bytes[16..20].copy_from_slice(&self.manifest_crc32c.to_le_bytes());
        bytes
    }

    fn parse(bytes: &[u8; MULTI_FILE_PREAMBLE_LEN]) -> Result<Self> {
        let mut magic = [0u8; MULTI_FILE_MAGIC_LEN];
        magic.copy_from_slice(&bytes[..MULTI_FILE_MAGIC_LEN]);
        let preamble = Self {
            magic,
            version: u16::from_le_bytes([bytes[8], bytes[9]]),
            flags: u16::from_le_bytes([bytes[10], bytes[11]]),
            manifest_len: u32::from_le_bytes([bytes[12], bytes[13], bytes[14], bytes[15]]),
            manifest_crc32c: u32::from_le_bytes([bytes[16], bytes[17], bytes[18], bytes[19]]),
        };
        preamble.validate()?;
        Ok(preamble)
    }

    fn validate(&self) -> Result<()> {
        if self.magic != Self::MAGIC {
            bail!("Invalid multi-file magic: {:?}", self.magic);
        }
        if self.version != MULTI_FILE_VERSION_V1 {
            bail!("Unsupported multi-file version: {}", self.version);
        }
        let unsupported_flags = self.flags & !MULTI_FILE_SUPPORTED_FLAGS;
        if unsupported_flags != 0 {
            bail!("Unsupported multi-file flags: 0x{unsupported_flags:04x}");
        }
        let manifest_len =
            usize::try_from(self.manifest_len).context("Manifest length does not fit in usize")?;
        if manifest_len > MAX_MANIFEST_SIZE {
            bail!("Manifest too large: {}", manifest_len);
        }
        Ok(())
    }
}

impl MultiFileManifest {
    fn parse(manifest_bytes: &[u8]) -> Result<Self> {
        Self::decode(manifest_bytes).with_context(|| {
            format!(
                "Failed to deserialize MultiFileManifest protobuf, len={}",
                manifest_bytes.len()
            )
        })
    }

    fn from_bundle(bundle: &FileBundle) -> Result<Self> {
        Ok(Self {
            frag_id: bundle.frag_id,
            files: build_file_entries(&bundle.files)?,
        })
    }

    fn validate_and_compute_total_size(&self) -> Result<u64> {
        validate_file_entries(&self.files)
    }
}

#[derive(Debug, Clone)]
struct DecodedMultiFileMetadata {
    manifest: MultiFileManifest,
    metadata_size: u64,
    total_file_size: u64,
}

/// Metadata for one payload file inside a multi-file object.
///
/// This is the public decoded entry type returned by
/// `read_multi_file_metadata_from_*`, not the prost-generated wire type
/// `multi_file_proto::FileEntry`. Wire `FileEntry` is crate-internal and
/// describes the serialized manifest: its `offset` is relative to the payload
/// area (see `multi_file.proto`). Callers doing range reads or seeks need the
/// absolute `payload_offset` from the object start, plus sibling fields on
/// [`MultiFileObjectMetadata`] (`metadata_size`, `total_object_size`, etc.),
/// so we convert at decode time in [`MultiFileObjectMetadata::from_decoded`]
/// rather than exposing the protobuf shape.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MultiFileObjectEntry {
    /// File path relative to the original bundle base directory.
    pub relative_path: String,
    /// Absolute byte offset of this file payload from the start of the
    /// multi-file object.
    pub payload_offset: u64,
    /// File payload size in bytes.
    pub size: u64,
}

/// Decoded metadata for a complete multi-file object.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MultiFileObjectMetadata {
    /// Fragment id stored in the multi-file manifest.
    pub frag_id: u64,
    /// Serialized preamble plus manifest size in bytes.
    pub metadata_size: u64,
    /// Total size of all payload files in bytes.
    pub total_file_size: u64,
    /// Total object size in bytes, including metadata and payloads.
    pub total_object_size: u64,
    /// Payload files stored in manifest order.
    pub files: Vec<MultiFileObjectEntry>,
}

impl MultiFileObjectMetadata {
    fn from_decoded(decoded: DecodedMultiFileMetadata) -> Result<Self> {
        let files = decoded
            .manifest
            .files
            .into_iter()
            .map(|file| {
                let payload_offset = decoded
                    .metadata_size
                    .checked_add(file.offset)
                    .with_context(|| {
                        format!(
                            "Multi-file payload offset overflow: path={}",
                            file.relative_path
                        )
                    })?;
                Ok(MultiFileObjectEntry {
                    relative_path: file.relative_path,
                    payload_offset,
                    size: file.size,
                })
            })
            .collect::<Result<Vec<_>>>()?;
        let total_object_size = decoded
            .metadata_size
            .checked_add(decoded.total_file_size)
            .context("Multi-file object size overflow")?;
        Ok(Self {
            frag_id: decoded.manifest.frag_id,
            metadata_size: decoded.metadata_size,
            total_file_size: decoded.total_file_size,
            total_object_size,
            files,
        })
    }
}

// Wire `FileEntry.offset` counts from the first payload byte (immediately after
// preamble + manifest), not from the object start. Payloads are concatenated
// in manifest order with no gaps, so offsets start at 0 and advance by each
// file's size.
fn build_file_entries(files: &[FileMeta]) -> Result<Vec<FileEntry>> {
    let mut entries = Vec::with_capacity(files.len());
    let mut offset = 0u64;
    for file in files {
        entries.push(FileEntry {
            relative_path: file.relative_path.clone(),
            offset,
            size: file.size,
        });
        offset = offset
            .checked_add(file.size)
            .with_context(|| format!("File size overflow: path={}", file.relative_path))?;
    }
    Ok(entries)
}

fn is_valid_multi_file_relative_path(path: &str) -> bool {
    MULTI_FILE_RELATIVE_PATH_RE.is_match(path)
}

fn validate_file_entries(files: &[FileEntry]) -> Result<u64> {
    if files.is_empty() {
        bail!("MultiFileManifest files list is empty");
    }
    let mut expected_offset = 0u64;
    for file in files {
        if !is_valid_multi_file_relative_path(&file.relative_path) {
            bail!(
                "Invalid relative_path (allowed: ASCII letters, digits, . _ -, \
                 slash-separated; pattern ^[A-Za-z0-9._-]+(/[A-Za-z0-9._-]+)*$): {:?}",
                file.relative_path
            );
        }
        if file.offset != expected_offset {
            bail!(
                "Invalid file offset: path={}, got={}, expected={}",
                file.relative_path,
                file.offset,
                expected_offset
            );
        }
        expected_offset = expected_offset
            .checked_add(file.size)
            .with_context(|| format!("File size overflow: path={}", file.relative_path))?;
    }
    Ok(expected_offset)
}

#[derive(Debug, Clone)]
pub struct FileBundle {
    base_dir: String,
    frag_id: u64,
    files: Vec<FileMeta>,
}

impl FileBundle {
    pub fn new(base_dir: impl Into<String>, frag_id: u64, files: Vec<FileMeta>) -> Self {
        Self {
            base_dir: base_dir.into(),
            frag_id,
            files,
        }
    }

    fn encode_manifest(&self) -> Result<(Vec<u8>, u64)> {
        let manifest = MultiFileManifest::from_bundle(self)?;
        let total_file_size = manifest.validate_and_compute_total_size()?;
        let manifest_bytes = manifest.encode_to_vec();
        if manifest_bytes.len() > MAX_MANIFEST_SIZE {
            bail!("Manifest too large: {}", manifest_bytes.len());
        }
        Ok((manifest_bytes, total_file_size))
    }
}

type FileStream = Pin<Box<dyn futures::Stream<Item = Result<Bytes, io::Error>> + Send>>;
// A multi-file byte stream that implements `Body`,
// allowing multiple files to be streamed as a single HTTP body.
pub struct MultiFileStream {
    inner: Arc<tokio::sync::Mutex<FileStream>>,
    size: u64,
}

impl Stream for MultiFileStream {
    type Item = Result<Bytes, io::Error>;

    fn poll_next(
        self: Pin<&mut Self>,
        cx: &mut std::task::Context<'_>,
    ) -> Poll<Option<Self::Item>> {
        let binding = self.inner.clone();
        let guard_fut = binding.lock();
        futures::pin_mut!(guard_fut);

        let mut stream_guard = match guard_fut.as_mut().poll(cx) {
            Poll::Ready(g) => g,
            Poll::Pending => return Poll::Pending,
        };

        stream_guard.as_mut().poll_next(cx)
    }
}

impl MultiFileStream {
    pub fn size(&self) -> u64 {
        self.size
    }

    /// Convert to a Stream compatible with BackendStorage::put_object
    pub fn into_stream(
        self,
    ) -> Pin<Box<dyn Stream<Item = Result<Bytes, io::Error>> + Send + Sync + 'static>> {
        Box::pin(self)
    }

    pub fn new(bundle: &FileBundle) -> Result<MultiFileStream, std::io::Error> {
        let (manifest, total_file_size) = bundle
            .encode_manifest()
            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
        let preamble = MultiFilePreamble::from_manifest(&manifest)
            .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?;
        let preamble_bytes = preamble.encode();

        let multi_file_size =
            MULTI_FILE_PREAMBLE_LEN as u64 + manifest.len() as u64 + total_file_size;

        let base_dir = bundle.base_dir.clone();
        let files = bundle.files.clone();

        let s = stream! {
            // 1. Emit fixed preamble.
            yield Ok(Bytes::copy_from_slice(&preamble_bytes));
            // 2. Emit protobuf manifest.
            yield Ok(Bytes::from(manifest));
            // 3. Emit file contents.
            for file in files {
                let path = format!("{}/{}", base_dir, file.relative_path);
                let reader = tokio::fs::File::open(&path)
                    .await
                    .map_err(|e| io::Error::new(e.kind(), format!("Open file failed: path={}, err={:?}", path, e)))?;
                let mut reader_stream = ReaderStream::new(reader);
                while let Some(chunk_res) = reader_stream.next().await {
                    let chunk = chunk_res.map_err(|e| io::Error::new(e.kind(), format!("Read file failed: path={}, err={:?}", path, e)))?;
                    yield Ok(chunk);
                }
            }
        };

        let body = MultiFileStream {
            inner: Arc::new(tokio::sync::Mutex::new(Box::pin(s))),
            size: multi_file_size,
        };

        Ok(body)
    }

    // Concatenate multiple `MultiFileStream` into one.
    // The concatenated stream will have the format:
    // [Header1][Files1][Header2][Files2]...
    //
    // The `stream_offsets` vector will contain the offset of each stream in the
    // concatenated stream.
    pub fn concat(streams: Vec<MultiFileStream>) -> (MultiFileStream, Vec<u64>) {
        let mut stream_offsets = Vec::with_capacity(streams.len());
        let mut offset = 0;
        for stream in &streams {
            stream_offsets.push(offset);
            offset += stream.size;
        }
        let total_size = offset;

        let s = stream! {
            for stream in streams {
                let inner = stream.inner.clone();
                let mut guard = inner.lock().await;
                while let Some(chunk_res) = guard.next().await {
                    yield chunk_res;
                }
            }
        };

        (
            MultiFileStream {
                inner: Arc::new(tokio::sync::Mutex::new(Box::pin(s))),
                size: total_size,
            },
            stream_offsets,
        )
    }
}

fn multi_file_manifest_len(preamble: MultiFilePreamble) -> Result<usize> {
    usize::try_from(preamble.manifest_len).context("Manifest length does not fit in usize")
}

fn parse_decoded_multi_file_metadata(
    preamble: MultiFilePreamble,
    manifest_bytes: Vec<u8>,
) -> Result<DecodedMultiFileMetadata> {
    let manifest_len = multi_file_manifest_len(preamble)?;
    ensure!(
        manifest_bytes.len() == manifest_len,
        "Manifest length mismatch: got={}, expected={}",
        manifest_bytes.len(),
        manifest_len
    );
    let manifest_crc32c = crc32c::crc32c(&manifest_bytes);
    ensure!(
        manifest_crc32c == preamble.manifest_crc32c,
        "Manifest crc32c mismatch: got {}, expected {}",
        manifest_crc32c,
        preamble.manifest_crc32c
    );
    let manifest = MultiFileManifest::parse(&manifest_bytes)?;
    let total_file_size = manifest.validate_and_compute_total_size()?;
    Ok(DecodedMultiFileMetadata {
        manifest,
        metadata_size: MULTI_FILE_PREAMBLE_LEN as u64 + manifest_len as u64,
        total_file_size,
    })
}

/// Read and validate multi-file object metadata from a blocking reader.
///
/// This reads only the preamble and manifest, then validates the manifest
/// checksum and entries. It does not read payload bytes or verify the reader's
/// total length.
pub fn read_multi_file_metadata_from_reader<R: std::io::Read>(
    reader: &mut R,
) -> Result<MultiFileObjectMetadata> {
    let mut preamble_bytes = [0u8; MULTI_FILE_PREAMBLE_LEN];
    reader
        .read_exact(&mut preamble_bytes)
        .context("Read multi-file preamble failed")?;
    let preamble = MultiFilePreamble::parse(&preamble_bytes)?;
    let manifest_len = multi_file_manifest_len(preamble)?;
    let mut manifest_bytes = vec![0u8; manifest_len];
    reader
        .read_exact(&mut manifest_bytes)
        .context("Read multi-file manifest failed")?;
    MultiFileObjectMetadata::from_decoded(parse_decoded_multi_file_metadata(
        preamble,
        manifest_bytes,
    )?)
}

/// Read and validate multi-file object metadata from a local object file.
///
/// After parsing the preamble and manifest, this verifies that the opened
/// file's length equals `total_object_size`.
pub fn read_multi_file_metadata_from_path(
    path: impl AsRef<Path>,
) -> Result<MultiFileObjectMetadata> {
    let path = path.as_ref();
    let mut file =
        std::fs::File::open(path).with_context(|| format!("Open multi-file object {:?}", path))?;
    let metadata = read_multi_file_metadata_from_reader(&mut file)
        .with_context(|| format!("Parse multi-file object metadata {:?}", path))?;
    let file_len = file
        .metadata()
        .with_context(|| format!("Stat multi-file object {:?}", path))?
        .len();
    ensure!(
        file_len == metadata.total_object_size,
        "Multi-file object length mismatch: path={:?}, got={}, expected={}",
        path,
        file_len,
        metadata.total_object_size
    );
    Ok(metadata)
}

async fn read_multi_file_metadata<R>(
    reader: &mut R,
) -> Result<DecodedMultiFileMetadata, DownloadError>
where
    R: AsyncRead + Unpin,
{
    let mut preamble_bytes = [0u8; MULTI_FILE_PREAMBLE_LEN];
    reader.read_exact(&mut preamble_bytes).await.map_err(|e| {
        DownloadError::Io(io::Error::new(
            e.kind(),
            format!("Read multi-file preamble failed: {:?}", e),
        ))
    })?;
    let preamble = MultiFilePreamble::parse(&preamble_bytes)
        .map_err(|e| DownloadError::CorruptedData(format!("{e:#}")))?;
    let manifest_len = multi_file_manifest_len(preamble)
        .map_err(|e| DownloadError::CorruptedData(format!("{e:#}")))?;
    let mut manifest_bytes = vec![0u8; manifest_len];
    reader.read_exact(&mut manifest_bytes).await.map_err(|e| {
        DownloadError::Io(io::Error::new(
            e.kind(),
            format!("Read multi-file manifest failed: {:?}", e),
        ))
    })?;
    parse_decoded_multi_file_metadata(preamble, manifest_bytes)
        .map_err(|e| DownloadError::CorruptedData(format!("{e:#}")))
}

// Upload multiple files as one single object using ObjectStore.
async fn upload_multi_files_as_one_object(
    backend_storage: &dyn BackendStorage,
    bucket: &str,
    key: &str,
    bundle: &FileBundle,
) -> Result<u64, UploadError> {
    let start = Instant::now();

    let body = MultiFileStream::new(bundle).map_err(UploadError::Io)?;
    let merged_file_size = body.size();
    let stream = body.into_stream();

    backend_storage
        .put_object(bucket, key, stream, merged_file_size)
        .await
        .map_err(UploadError::Backend)?;
    info!(
        "upload_multi_files_as_one_object: bundle={:?}, key={}, size={}, cost={:.3}s",
        bundle,
        key,
        merged_file_size,
        start.elapsed().as_secs_f64()
    );
    Ok(merged_file_size)
}

pub async fn upload_multi_files_as_one_object_with_retry(
    backend_storage: &dyn BackendStorage,
    bucket: &str,
    key: &str,
    bundle: &FileBundle,
) -> Result<u64> {
    retry_with_backoff(move || {
        upload_multi_files_as_one_object(backend_storage, bucket, key, bundle)
    })
    .await
    .with_context(|| {
        format!(
            "upload_multi_files_as_one_object failed: bundle={:?}, key={}",
            bundle, key
        )
    })
}

// Download a single object which is actually a multi-file package
// created by `MultiFileStream`, and reconstruct the files on disk.
async fn download_one_object_as_multi_files(
    backend_storage: &dyn BackendStorage,
    bucket: &str,
    key: &str,
    range: Option<ReadRange>,
    target_base_dir: &str,
) -> Result<(), DownloadError> {
    let start = Instant::now();
    let (mut reader, meta) = backend_storage
        .get_object(bucket, key, range)
        .await
        .map_err(DownloadError::Backend)?;

    let metadata = read_multi_file_metadata(&mut reader).await?;

    // Verify size
    if let Some(content_len) = meta.content_length {
        let expected_len = metadata.metadata_size + metadata.total_file_size;
        // If content_len is negative try_from will fail and return None
        // If content_len is positive, it will be converted to u64 and compared with
        // expected_len
        if u64::try_from(content_len).ok() != Some(expected_len) {
            return Err(DownloadError::CorruptedData(format!(
                "Content length mismatch: got {}, expected {} (bucket={}, key={})",
                content_len, expected_len, bucket, key
            )));
        }
    } else {
        info!(
            "Content length is None, skipping verify. metadata_size={}, file_total_size={}, bucket={}, key={}",
            metadata.metadata_size, metadata.total_file_size, bucket, key
        );
    }

    // Rebuild files on disk
    const BUF_SIZE: usize = 65536;
    let mut copy_buf = vec![0u8; BUF_SIZE];
    for file in metadata.manifest.files.iter() {
        // Safety check: ensure relative_path is indeed relative and does not contain ..
        let path_obj = Path::new(&file.relative_path);
        if path_obj.is_absolute()
            || path_obj
                .components()
                .any(|c| matches!(c, std::path::Component::ParentDir))
        {
            return Err(DownloadError::CorruptedData(format!(
                "Security warning: Invalid relative path '{}'",
                file.relative_path
            )));
        }

        let local_path = Path::new(target_base_dir).join(&file.relative_path);

        if let Some(parent) = local_path.parent() {
            tokio::fs::create_dir_all(parent).await.map_err(|e| {
                DownloadError::Io(io::Error::new(
                    e.kind(),
                    format!("create_dir_all: dir={:?}, {:?}", parent, e),
                ))
            })?;
        }

        let mut writer = tokio::fs::File::create(&local_path).await.map_err(|e| {
            DownloadError::Io(io::Error::new(
                e.kind(),
                format!("create_file: file={:?}, {:?}", local_path, e),
            ))
        })?;

        let mut remaining = file.size as usize;
        while remaining > 0 {
            let to_read = remaining.min(BUF_SIZE);
            let content = &mut copy_buf[..to_read];
            reader.read_exact(content).await.map_err(|e| {
                DownloadError::Io(io::Error::new(
                    e.kind(),
                    format!(
                        "Read content: file={:?}, remaining={}, {:?}",
                        local_path, remaining, e
                    ),
                ))
            })?;
            writer.write_all(content).await.map_err(|e| {
                DownloadError::Io(io::Error::new(
                    e.kind(),
                    format!("Write file content: file={:?}, {:?}", local_path, e),
                ))
            })?;
            remaining -= to_read;
        }
        writer.flush().await?;
        writer.sync_data().await?;
    }
    info!(
        "download_one_object_as_multi_files key={}, target_base_dir={}, manifest={:?}, cost={:.3}s",
        key,
        target_base_dir,
        metadata.manifest,
        start.elapsed().as_secs_f64()
    );
    Ok(())
}

pub async fn download_one_object_as_multi_files_with_retry(
    backend_storage: &dyn BackendStorage,
    bucket: &str,
    key: &str,
    range: Option<ReadRange>,
    base_dir: &str,
) -> Result<()> {
    retry_with_backoff(|| {
        download_one_object_as_multi_files(backend_storage, bucket, key, range, base_dir)
    })
    .await
    .with_context(|| {
        format!(
            "Download object: bucket={}, key={}, range={:?}, base_dir={}",
            bucket, key, range, base_dir
        )
    })
}

// Group files into buckets such that the total size in each group
// is close to (but not exceeding) the given target size.
pub fn group_files(
    base_dir: &str,
    relative_paths: &[String],
    target_size: u64,
) -> Result<Vec<Vec<FileMeta>>> {
    // Collect metadata (file size) for each file
    let mut files = Vec::with_capacity(relative_paths.len());
    for relative_path in relative_paths {
        let full_path = Path::new(base_dir).join(relative_path);
        let meta = std::fs::metadata(&full_path)
            .with_context(|| format!("Get metadata of {:?}", full_path))?;
        files.push(FileMeta {
            relative_path: relative_path.clone(),
            size: meta.len(),
        });
    }
    Ok(group_file_metas(files, target_size))
}

// Sort files by descending size, then place each file into the "best-fit"
// group.
pub fn group_file_metas(mut files: Vec<FileMeta>, target_size: u64) -> Vec<Vec<FileMeta>> {
    // Sort files from largest to smallest
    files.sort_by(|a, b| b.size.cmp(&a.size));
    // groups[i] = list of files in group i
    let mut groups: Vec<Vec<FileMeta>> = Vec::new();
    // group_sums[i] = total size of group i
    let mut group_sums: Vec<u64> = Vec::new();
    // Place each file into an existing group (best fit), or create a new one
    for file in files.iter() {
        let mut best_fit = None; // Index of best-fit group
        let mut min_diff = u64::MAX; // How much space would be left if placed here
        // Try to fit file into an existing group
        for (i, &sum) in group_sums.iter().enumerate() {
            if sum + file.size <= target_size {
                let diff = target_size - (sum + file.size);
                if diff < min_diff {
                    min_diff = diff;
                    best_fit = Some(i);
                }
            }
        }
        // If found a group that can fit this file, put it there.
        // Otherwise, start a new group.
        match best_fit {
            Some(idx) => {
                groups[idx].push(file.clone());
                group_sums[idx] += file.size;
            }
            None => {
                groups.push(vec![file.clone()]);
                group_sums.push(file.size);
            }
        }
    }
    groups
}

#[cfg(test)]
mod tests {

    use std::io::Write;

    use anyhow::anyhow;
    use tokio::io::AsyncReadExt;

    use super::*;

    // --- Negative tests for MultiFilePreamble::parse / validate ---

    fn make_valid_preamble(
        manifest_len: u32,
        manifest_crc32c: u32,
    ) -> [u8; MULTI_FILE_PREAMBLE_LEN] {
        let preamble = MultiFilePreamble {
            magic: MultiFilePreamble::MAGIC,
            version: MULTI_FILE_VERSION_V1,
            flags: MULTI_FILE_SUPPORTED_FLAGS,
            manifest_len,
            manifest_crc32c,
        };
        preamble.encode()
    }

    #[test]
    fn test_preamble_parse_corrupted_magic() {
        let mut bytes = make_valid_preamble(10, 0);
        bytes[0] ^= 0xFF; // corrupt first magic byte
        let err = MultiFilePreamble::parse(&bytes).unwrap_err();
        assert!(err.to_string().contains("Invalid multi-file magic"));
    }

    #[test]
    fn test_preamble_parse_unsupported_version() {
        let mut preamble = MultiFilePreamble {
            magic: MultiFilePreamble::MAGIC,
            version: MULTI_FILE_VERSION_V1,
            flags: MULTI_FILE_SUPPORTED_FLAGS,
            manifest_len: 0,
            manifest_crc32c: 0,
        };
        preamble.version = 99; // unsupported version
        let bytes = preamble.encode();
        let err = MultiFilePreamble::parse(&bytes).unwrap_err();
        assert!(err.to_string().contains("Unsupported multi-file version"));
    }

    #[test]
    fn test_preamble_parse_unsupported_flags() {
        let mut preamble = MultiFilePreamble {
            magic: MultiFilePreamble::MAGIC,
            version: MULTI_FILE_VERSION_V1,
            flags: MULTI_FILE_SUPPORTED_FLAGS,
            manifest_len: 0,
            manifest_crc32c: 0,
        };
        preamble.flags = 0x0001; // unsupported flag bit
        let bytes = preamble.encode();
        let err = MultiFilePreamble::parse(&bytes).unwrap_err();
        assert!(err.to_string().contains("Unsupported multi-file flags"));
    }

    #[test]
    fn test_preamble_parse_manifest_too_large() {
        let preamble = MultiFilePreamble {
            magic: MultiFilePreamble::MAGIC,
            version: MULTI_FILE_VERSION_V1,
            flags: MULTI_FILE_SUPPORTED_FLAGS,
            manifest_len: MAX_MANIFEST_SIZE as u32 + 1,
            manifest_crc32c: 0,
        };
        let bytes = preamble.encode();
        let err = MultiFilePreamble::parse(&bytes).unwrap_err();
        assert!(err.to_string().contains("Manifest too large"));
    }

    // --- Negative tests for manifest CRC mismatch ---

    #[tokio::test]
    async fn test_read_metadata_crc_mismatch() {
        // Build a valid preamble with a CRC that doesn't match the manifest bytes.
        let manifest_bytes = b"some manifest data".to_vec();
        let bad_crc = crc32c::crc32c(&manifest_bytes) ^ 0xDEADBEEF;
        let preamble = MultiFilePreamble {
            magic: MultiFilePreamble::MAGIC,
            version: MULTI_FILE_VERSION_V1,
            flags: MULTI_FILE_SUPPORTED_FLAGS,
            manifest_len: manifest_bytes.len() as u32,
            manifest_crc32c: bad_crc,
        };

        let mut data = Vec::new();
        data.extend_from_slice(&preamble.encode());
        data.extend_from_slice(&manifest_bytes);

        let mut cursor = std::io::Cursor::new(&data);
        let err = read_multi_file_metadata(&mut cursor).await.unwrap_err();
        assert!(matches!(err, DownloadError::CorruptedData(_)));
        assert!(err.to_string().contains("crc32c mismatch"));
    }

    #[test]
    fn test_read_multi_file_metadata_from_reader_sync() -> Result<()> {
        let bytes = build_multi_file_test_bytes()?;

        let metadata = read_multi_file_metadata_from_reader(&mut std::io::Cursor::new(&bytes))?;

        assert_eq!(metadata.frag_id, 42);
        assert_eq!(metadata.files.len(), 2);
        assert_eq!(metadata.files[0].relative_path, "index/meta.json");
        assert_eq!(metadata.files[0].payload_offset, metadata.metadata_size);
        assert_eq!(metadata.files[0].size, br#"{\"segments\":[]}"#.len() as u64);
        assert_eq!(
            metadata.files[1].payload_offset,
            metadata.metadata_size + metadata.files[0].size
        );
        assert_eq!(
            metadata.total_file_size,
            metadata.files[0].size + metadata.files[1].size
        );
        assert_eq!(metadata.total_object_size, bytes.len() as u64);
        Ok(())
    }

    #[test]
    fn test_read_multi_file_metadata_from_path_sync() -> Result<()> {
        let bytes = build_multi_file_test_bytes()?;
        let temp_dir = tempfile::tempdir()?;
        let object_path = temp_dir.path().join("multi.42");
        std::fs::write(&object_path, &bytes)?;

        let reader_metadata =
            read_multi_file_metadata_from_reader(&mut std::io::Cursor::new(&bytes))?;
        let path_metadata = read_multi_file_metadata_from_path(&object_path)?;

        assert_eq!(
            path_metadata.total_object_size,
            reader_metadata.total_object_size
        );
        assert_eq!(path_metadata.files, reader_metadata.files);
        Ok(())
    }

    #[test]
    fn test_read_multi_file_metadata_from_path_length_mismatch() -> Result<()> {
        let mut bytes = build_multi_file_test_bytes()?;
        bytes.push(0);
        let temp_dir = tempfile::tempdir()?;
        let object_path = temp_dir.path().join("multi.42");
        std::fs::write(&object_path, &bytes)?;

        let err = read_multi_file_metadata_from_path(&object_path).unwrap_err();
        assert!(
            err.to_string()
                .contains("Multi-file object length mismatch"),
            "err={err:#}"
        );
        Ok(())
    }

    fn build_multi_file_test_bytes() -> Result<Vec<u8>> {
        let rt = tokio::runtime::Runtime::new()?;
        let temp_dir = tempfile::tempdir()?;
        let base_path = temp_dir
            .path()
            .to_str()
            .context("temporary directory path is not valid UTF-8")?;
        let index_dir = temp_dir.path().join("index");
        std::fs::create_dir_all(&index_dir)?;
        std::fs::write(index_dir.join("meta.json"), br#"{\"segments\":[]}"#)?;
        std::fs::write(index_dir.join("seg.term"), b"term-bytes")?;

        let files = vec![
            FileMeta {
                relative_path: "index/meta.json".to_string(),
                size: std::fs::metadata(index_dir.join("meta.json"))?.len(),
            },
            FileMeta {
                relative_path: "index/seg.term".to_string(),
                size: std::fs::metadata(index_dir.join("seg.term"))?.len(),
            },
        ];
        let bundle = FileBundle::new(base_path, 42, files);
        rt.block_on(stream_to_bytes(MultiFileStream::new(&bundle)?))
    }

    // --- Negative tests for truncated reads ---

    #[tokio::test]
    async fn test_read_metadata_truncated_preamble() {
        let data = vec![0u8; 3]; // too short for a full preamble
        let mut cursor = std::io::Cursor::new(&data);
        let err = read_multi_file_metadata(&mut cursor).await.unwrap_err();
        assert!(matches!(err, DownloadError::Io(_)));
    }

    #[tokio::test]
    async fn test_read_metadata_truncated_manifest() {
        // Preamble claims 100 manifest bytes but we only supply 10.
        let preamble = MultiFilePreamble {
            magic: MultiFilePreamble::MAGIC,
            version: MULTI_FILE_VERSION_V1,
            flags: MULTI_FILE_SUPPORTED_FLAGS,
            manifest_len: 100,
            manifest_crc32c: 0,
        };
        let mut data = Vec::new();
        data.extend_from_slice(&preamble.encode());
        data.extend_from_slice(&[0u8; 10]); // only 10 bytes

        let mut cursor = std::io::Cursor::new(&data);
        let err = read_multi_file_metadata(&mut cursor).await.unwrap_err();
        assert!(matches!(err, DownloadError::Io(_)));
    }

    // --- validate_file_entries relative_path
    // (^[A-Za-z0-9._-]+(/[A-Za-z0-9._-]+)*$) ---

    #[test]
    fn test_validate_file_entries_rejects_empty_files() {
        let err = MultiFileManifest {
            frag_id: 0,
            files: vec![],
        }
        .validate_and_compute_total_size()
        .unwrap_err();
        assert!(err.to_string().contains("files list is empty"));
    }

    #[test]
    fn test_validate_file_entries_rejects_invalid_relative_path() {
        for bad in ["", "/a", "a/", "a//b", "a b", "a\tb", "a\\b", "café", "a:1"] {
            let files = vec![FileEntry {
                relative_path: bad.into(),
                offset: 0,
                size: 1,
            }];
            let err = MultiFileManifest { frag_id: 0, files }
                .validate_and_compute_total_size()
                .unwrap_err();
            assert!(
                err.to_string().contains("Invalid relative_path"),
                "path={bad:?} err={err}"
            );
        }
    }

    #[test]
    fn test_validate_file_entries_accepts_valid_relative_path() {
        for good in ["a", "A.z_9", "dir_a/sub_b/file.c", "a-b_c.d"] {
            let files = vec![FileEntry {
                relative_path: good.into(),
                offset: 0,
                size: 1,
            }];
            MultiFileManifest { frag_id: 0, files }
                .validate_and_compute_total_size()
                .unwrap();
        }
    }

    // --- Negative tests for validate_file_entries ---

    #[test]
    fn test_validate_file_entries_offset_discontinuity() {
        let files = vec![
            FileEntry {
                relative_path: "a".into(),
                offset: 0,
                size: 10,
            },
            FileEntry {
                relative_path: "b".into(),
                offset: 5, // should be 10
                size: 10,
            },
        ];
        let err = MultiFileManifest { frag_id: 0, files }
            .validate_and_compute_total_size()
            .unwrap_err();
        assert!(err.to_string().contains("Invalid file offset"));
    }

    #[test]
    fn test_validate_file_entries_overflow() {
        let files = vec![
            FileEntry {
                relative_path: "a".into(),
                offset: 0,
                size: u64::MAX,
            },
            FileEntry {
                relative_path: "b".into(),
                offset: u64::MAX,
                size: 1,
            },
        ];
        let err = MultiFileManifest { frag_id: 0, files }
            .validate_and_compute_total_size()
            .unwrap_err();
        assert!(err.to_string().contains("File size overflow"));
    }

    // --- Existing tests ---

    #[test]
    fn test_group_files_advanced() {
        let files = vec![
            FileMeta {
                relative_path: "a".to_string(),
                size: 100,
            },
            FileMeta {
                relative_path: "b".to_string(),
                size: 200,
            },
            FileMeta {
                relative_path: "c".to_string(),
                size: 150,
            },
            FileMeta {
                relative_path: "d".to_string(),
                size: 80,
            },
            FileMeta {
                relative_path: "e".to_string(),
                size: 50,
            },
            FileMeta {
                relative_path: "f".to_string(),
                size: 70,
            },
            FileMeta {
                relative_path: "g".to_string(),
                size: 30,
            },
            FileMeta {
                relative_path: "h".to_string(),
                size: 60,
            },
        ];
        let groups = group_file_metas(files.clone(), 256);
        for group in groups.iter() {
            let sum: u64 = group.iter().map(|f| f.size).sum();
            assert!(sum <= 256);
        }
        let mut group_files = Vec::with_capacity(files.len());
        for group in groups.iter() {
            group_files.extend(group.iter().cloned());
        }
        group_files.sort_by(|a, b| a.relative_path.cmp(&b.relative_path));
        assert_eq!(files.len(), group_files.len());
        for (i, file) in files.iter().enumerate() {
            assert_eq!(file.relative_path, group_files[i].relative_path);
            assert_eq!(file.size, group_files[i].size);
        }
    }

    #[test]
    fn test_group_files_all_in_one() {
        let files = vec![
            FileMeta {
                relative_path: "a".into(),
                size: 50,
            },
            FileMeta {
                relative_path: "b".into(),
                size: 60,
            },
            FileMeta {
                relative_path: "c".into(),
                size: 40,
            },
        ];
        let groups = group_file_metas(files.clone(), 200);
        assert_eq!(groups.len(), 1);
        let total: u64 = groups[0].iter().map(|f| f.size).sum();
        assert_eq!(total, 150);
    }

    #[test]
    fn test_group_files_all_separate() {
        let files = vec![
            FileMeta {
                relative_path: "a".into(),
                size: 300,
            },
            FileMeta {
                relative_path: "b".into(),
                size: 400,
            },
            FileMeta {
                relative_path: "c".into(),
                size: 500,
            },
        ];
        let groups = group_file_metas(files.clone(), 256);
        assert_eq!(groups.len(), files.len());
        for group in groups.iter() {
            assert_eq!(group.len(), 1);
            let file = files
                .iter()
                .find(|f| f.relative_path == group[0].relative_path)
                .unwrap();
            assert_eq!(group[0].relative_path, file.relative_path);
            assert_eq!(group[0].size, file.size);
        }
    }

    #[test]
    fn test_group_files_exact_target() {
        let files = vec![
            FileMeta {
                relative_path: "a".into(),
                size: 256,
            },
            FileMeta {
                relative_path: "b".into(),
                size: 100,
            },
            FileMeta {
                relative_path: "c".into(),
                size: 150,
            },
        ];
        let groups = group_file_metas(files.clone(), 256);

        let group_a = groups
            .iter()
            .find(|g| g.iter().any(|f| f.relative_path == "a"))
            .unwrap();
        assert_eq!(group_a.len(), 1);
        assert_eq!(group_a[0].size, 256);

        let other_files: Vec<_> = groups
            .iter()
            .filter(|g| !g.iter().any(|f| f.relative_path == "a"))
            .flatten()
            .map(|f| f.relative_path.clone())
            .collect();

        assert!(other_files.contains(&"b".to_string()));
        assert!(other_files.contains(&"c".to_string()));
    }

    async fn stream_to_bytes(stream: MultiFileStream) -> Result<Vec<u8>> {
        let mut stream = stream.into_stream();
        let mut bytes = Vec::new();
        while let Some(chunk) = stream.next().await {
            bytes.extend_from_slice(&chunk.map_err(|e| anyhow!(e))?);
        }
        Ok(bytes)
    }

    #[tokio::test]
    async fn test_multi_file_stream_creation_and_content() -> Result<()> {
        let temp_dir = tempfile::tempdir()?;
        let base_path = temp_dir
            .path()
            .to_str()
            .context("temp dir path is not valid utf-8")?;

        let file1_name = "f1.txt";
        let file1_content = b"Hello World";
        let file1_path = temp_dir.path().join(file1_name);
        {
            let mut f = std::fs::File::create(&file1_path)?;
            f.write_all(file1_content)?;
        }

        let file2_name = "sub/f2.txt";
        let file2_content = b"Rust Programming";
        let file2_path = temp_dir.path().join(file2_name);
        std::fs::create_dir_all(file2_path.parent().context("file2_path parent is None")?)?;
        {
            let mut f = std::fs::File::create(&file2_path)?;
            f.write_all(file2_content)?;
        }

        let files = vec![
            FileMeta {
                relative_path: file1_name.to_string(),
                size: file1_content.len() as u64,
            },
            FileMeta {
                relative_path: file2_name.to_string(),
                size: file2_content.len() as u64,
            },
        ];

        let bundle = FileBundle::new(base_path, 1001, files.clone());

        let (expected_manifest_bytes, _) = bundle.encode_manifest()?;
        let expected_preamble = MultiFilePreamble::from_manifest(&expected_manifest_bytes)?;

        let stream = MultiFileStream::new(&bundle)?;

        let content_len = (file1_content.len() + file2_content.len()) as u64;
        assert_eq!(
            stream.size(),
            MULTI_FILE_PREAMBLE_LEN as u64 + expected_manifest_bytes.len() as u64 + content_len
        );

        let output_bytes = stream_to_bytes(stream).await?;

        let mut cursor = std::io::Cursor::new(&output_bytes);
        let metadata = read_multi_file_metadata(&mut cursor).await?;
        let mut preamble_arr = [0u8; MULTI_FILE_PREAMBLE_LEN];
        preamble_arr.copy_from_slice(&output_bytes[..MULTI_FILE_PREAMBLE_LEN]);
        assert_eq!(MultiFilePreamble::parse(&preamble_arr)?, expected_preamble);
        assert_eq!(
            metadata.metadata_size,
            output_bytes.len() as u64 - content_len
        );
        assert_eq!(metadata.manifest.frag_id, 1001);
        assert_eq!(metadata.manifest.files.len(), 2);
        assert_eq!(metadata.manifest.files[0].relative_path, file1_name);
        assert_eq!(metadata.manifest.files[0].offset, 0);
        assert_eq!(metadata.manifest.files[0].size, file1_content.len() as u64);
        assert_eq!(metadata.manifest.files[1].relative_path, file2_name);
        assert_eq!(
            metadata.manifest.files[1].offset,
            file1_content.len() as u64
        );
        assert_eq!(metadata.manifest.files[1].size, file2_content.len() as u64);

        let mut content_buf = Vec::new();
        cursor.read_to_end(&mut content_buf).await?;
        let expected_content = [file1_content.as_slice(), file2_content.as_slice()].concat();
        assert_eq!(content_buf, expected_content);

        Ok(())
    }

    #[tokio::test]
    async fn test_multi_file_stream_concat() -> Result<()> {
        let temp_dir = tempfile::tempdir()?;
        let base_path = temp_dir
            .path()
            .to_str()
            .context("temporary directory path is not valid UTF-8")?;

        let fname = "test.txt";
        let fpath = temp_dir.path().join(fname);
        std::fs::write(&fpath, b"123")?;

        let meta = vec![FileMeta {
            relative_path: fname.into(),
            size: 3,
        }];

        let bundle1 = FileBundle::new(base_path, 1, meta.clone());
        let bundle2 = FileBundle::new(base_path, 2, meta.clone());

        let s1 = MultiFileStream::new(&bundle1)?;
        let s1_size = s1.size();
        let s2 = MultiFileStream::new(&bundle2)?;
        let s2_size = s2.size();

        let (merged_stream, offsets) = MultiFileStream::concat(vec![s1, s2]);

        assert_eq!(offsets.len(), 2);
        assert_eq!(offsets[0], 0);
        assert_eq!(offsets[1], s1_size);
        assert_eq!(merged_stream.size(), s1_size + s2_size);

        let bytes = stream_to_bytes(merged_stream).await?;
        assert_eq!(bytes.len() as u64, s1_size + s2_size);

        Ok(())
    }
}