// 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.

use crate::{
    ErasureAlgo, ErasureInfo, Error, FileInfo, FileInfoVersions, InlineData, NULL_VERSION_ID, ObjectPartInfo, RawFileInfo,
    ReplicationState, ReplicationStatusType, Result, VersionPurgeStatusType, is_restored_object_on_disk,
    replication_statuses_map, version_purge_statuses_map,
};
use byteorder::ByteOrder;
use bytes::Bytes;
use rustfs_utils::http::headers::{
    AMZ_META_UNENCRYPTED_CONTENT_LENGTH, AMZ_META_UNENCRYPTED_CONTENT_MD5, AMZ_RESTORE_EXPIRY_DAYS, AMZ_RESTORE_REQUEST_DATE,
    AMZ_STORAGE_CLASS,
};
use rustfs_utils::http::{
    AMZ_BUCKET_REPLICATION_STATUS, SUFFIX_CRC, SUFFIX_DATA_MOV, SUFFIX_HEALING, SUFFIX_PURGESTATUS, SUFFIX_REPLICA_STATUS,
    SUFFIX_REPLICA_TIMESTAMP, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, has_internal_suffix, insert_bytes,
    is_internal_key,
};
use s3s::header::X_AMZ_RESTORE;
use serde::{Deserialize, Serialize};
use std::cmp::Ordering;
use std::convert::TryFrom;
use std::hash::Hasher;
use std::io::{Read, Write};
use std::{collections::HashMap, io::Cursor};
use time::OffsetDateTime;
use time::format_description::well_known::Rfc3339;
use tokio::io::AsyncRead;
use tracing::{error, warn};
use uuid::Uuid;
use xxhash_rust::xxh64;

// XL header specifies the format
pub static XL_FILE_HEADER: [u8; 4] = [b'X', b'L', b'2', b' '];
// pub static XL_FILE_VERSION_CURRENT: [u8; 4] = [0; 4];

// Current version being written.
// static XL_FILE_VERSION: [u8; 4] = [1, 0, 3, 0];
static XL_FILE_VERSION_MAJOR: u16 = 1;
static XL_FILE_VERSION_MINOR: u16 = 3;
static XL_HEADER_VERSION: u8 = 3;
pub static XL_META_VERSION: u8 = 3;
/// Legacy format (main branch): meta_ver=2 with file/header versions 1.3.3.
static XXHASH_SEED: u64 = 0;

const XL_FLAG_FREE_VERSION: u8 = 1 << 0;
// const XL_FLAG_USES_DATA_DIR: u8 = 1 << 1;
const _XL_FLAG_INLINE_DATA: u8 = 1 << 2;

const META_DATA_READ_DEFAULT: usize = 4 << 10;
const MSGP_UINT32_SIZE: usize = 5;

/// Max object versions per object, default is 10000
const DEFAULT_OBJECT_MAX_VERSIONS: usize = 10000;

/// Returns the inline data map key for a version_id. "null" for null version.
pub(crate) fn data_key_for_version(version_id: Option<Uuid>) -> String {
    if version_id.is_none() || version_id == Some(Uuid::nil()) {
        NULL_VERSION_ID.to_string()
    } else {
        version_id.unwrap_or_default().to_string()
    }
}

fn legacy_data_key_for_version(version_id: Option<Uuid>) -> Option<String> {
    if version_id.is_none() || version_id == Some(Uuid::nil()) {
        Some(Uuid::nil().to_string())
    } else {
        None
    }
}

pub const TRANSITION_COMPLETE: &str = "complete";
pub const TRANSITION_PENDING: &str = "pending";

pub const FREE_VERSION: &str = "free-version";

pub const TRANSITION_STATUS: &str = "transition-status";
pub const TRANSITIONED_OBJECTNAME: &str = "transitioned-object";
pub const TRANSITIONED_VERSION_ID: &str = "transitioned-versionID";
pub const TRANSITION_TIER: &str = "transition-tier";

/// Returns true if the key is a transient internal flag that should not be persisted to meta_sys.
pub fn is_skip_meta_key(key: &str) -> bool {
    has_internal_suffix(key, SUFFIX_HEALING) || has_internal_suffix(key, SUFFIX_DATA_MOV)
}

mod codec;
mod inline_data;
mod msgp_decode;
mod validation;
mod version;

pub use validation::{DetailedVersionStats, VersionStats};
pub use version::*;

// type ScanHeaderVersionFn = Box<dyn Fn(usize, &[u8], &[u8]) -> Result<()>>;

#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct FileMeta {
    pub versions: Vec<FileMetaShallowVersion>,
    pub data: InlineData, // TODO: xlMetaInlineData
    pub meta_ver: u8,
}

impl FileMeta {
    pub fn new() -> Self {
        Self {
            meta_ver: XL_META_VERSION,
            data: InlineData::new(),
            ..Default::default()
        }
    }

    fn get_idx(&self, idx: usize) -> Result<FileMetaVersion> {
        if idx > self.versions.len() {
            return Err(Error::FileNotFound);
        }

        FileMetaVersion::try_from(self.versions[idx].meta.as_slice())
    }

    fn set_idx(&mut self, idx: usize, ver: FileMetaVersion) -> Result<()> {
        if idx >= self.versions.len() {
            return Err(Error::FileNotFound);
        }

        // TODO: use old buf
        let meta_buf = ver.marshal_msg()?;

        let pre_mod_time = self.versions[idx].header.mod_time;

        self.versions[idx].header = ver.header();
        self.versions[idx].meta = meta_buf;

        if pre_mod_time != self.versions[idx].header.mod_time {
            self.sort_by_mod_time();
        }

        Ok(())
    }

    fn sort_by_mod_time(&mut self) {
        if self.versions.len() <= 1 {
            return;
        }

        self.versions.sort_by(|a, b| {
            if a.header.mod_time != b.header.mod_time {
                b.header.mod_time.cmp(&a.header.mod_time)
            } else if a.header.version_type != b.header.version_type {
                b.header.version_type.cmp(&a.header.version_type)
            } else if a.header.version_id != b.header.version_id {
                b.header.version_id.cmp(&a.header.version_id)
            } else if a.header.flags != b.header.flags {
                b.header.flags.cmp(&a.header.flags)
            } else {
                b.cmp(a)
            }
        });
    }

    fn find_inline_data_for_version(&self, version_id: Option<Uuid>) -> Result<Option<Vec<u8>>> {
        let key = data_key_for_version(version_id);
        if let Some(data) = self.data.find(key.as_str())? {
            return Ok(Some(data));
        }

        if let Some(legacy_key) = legacy_data_key_for_version(version_id)
            && legacy_key != key
        {
            return self.data.find(legacy_key.as_str());
        }

        Ok(None)
    }

    // Find version
    pub fn find_version(&self, vid: Option<Uuid>) -> Result<(usize, FileMetaVersion)> {
        let vid = vid.unwrap_or_default();
        for (i, fver) in self.versions.iter().enumerate() {
            if fver.header.version_id == Some(vid) {
                let version = self.get_idx(i)?;
                return Ok((i, version));
            }
        }

        Err(Error::FileVersionNotFound)
    }

    pub fn update_object_version(&mut self, fi: FileInfo) -> Result<()> {
        for version in self.versions.iter_mut() {
            match version.header.version_type {
                VersionType::Invalid | VersionType::Legacy => (),
                VersionType::Object => {
                    // For non-versioned buckets, treat None as Uuid::nil()
                    let fi_vid = fi.version_id.or(Some(Uuid::nil()));
                    let ver_vid = version.header.version_id.or(Some(Uuid::nil()));

                    if ver_vid == fi_vid {
                        let mut ver = FileMetaVersion::try_from(version.meta.as_slice())?;

                        if let Some(ref mut obj) = ver.object {
                            for (k, v) in fi.metadata.iter() {
                                // Split metadata into meta_user and meta_sys based on prefix
                                // This logic must match From<FileInfo> for MetaObject
                                let is_system = is_internal_key(k);
                                if is_system {
                                    // Skip internal flags that shouldn't be persisted
                                    if is_skip_meta_key(k) {
                                        continue;
                                    }
                                    // Insert into meta_sys
                                    obj.meta_sys.insert(k.clone(), v.as_bytes().to_vec());
                                } else {
                                    // Insert into meta_user
                                    obj.meta_user.insert(k.clone(), v.clone());
                                }
                            }

                            if let Some(mod_time) = fi.mod_time {
                                obj.mod_time = Some(mod_time);
                            }

                            if let Some(content_hash) = fi.checksum.as_ref() {
                                insert_bytes(&mut obj.meta_sys, SUFFIX_CRC, content_hash.to_vec());
                            }
                        }

                        // Update
                        version.header = ver.header();
                        version.meta = ver.marshal_msg()?;
                    }
                }
                VersionType::Delete => {
                    if version.header.version_id == fi.version_id {
                        return Err(Error::MethodNotAllowed);
                    }
                }
            }
        }

        self.versions.sort_by(|a, b| {
            if a.header.mod_time != b.header.mod_time {
                b.header.mod_time.cmp(&a.header.mod_time)
            } else if a.header.version_type != b.header.version_type {
                b.header.version_type.cmp(&a.header.version_type)
            } else if a.header.version_id != b.header.version_id {
                b.header.version_id.cmp(&a.header.version_id)
            } else if a.header.flags != b.header.flags {
                b.header.flags.cmp(&a.header.flags)
            } else {
                b.cmp(a)
            }
        });
        Ok(())
    }

    pub fn add_version(&mut self, mut fi: FileInfo) -> Result<()> {
        // empty version_id means "null" (versioning disabled/suspended)
        if fi.version_id.is_none() {
            fi.version_id = Some(Uuid::nil());
        }

        let version_key = data_key_for_version(fi.version_id);
        let mut next_data = self.data.clone();

        if let Some(ref data) = fi.data {
            next_data.replace(&version_key, data.to_vec())?;
        } else {
            let _ = next_data.remove_key(&version_key)?;
        }

        let version = FileMetaVersion::from(fi);

        self.add_version_filemata(version)?;
        self.data = next_data;

        Ok(())
    }

    pub fn add_version_filemata(&mut self, version: FileMetaVersion) -> Result<()> {
        if !version.valid() {
            return Err(Error::other("file meta version invalid"));
        }

        // check max versions limit
        if self.versions.len() + 1 > DEFAULT_OBJECT_MAX_VERSIONS {
            return Err(Error::other(
                "You've exceeded the limit on the number of versions you can create on this object",
            ));
        }

        if self.versions.is_empty() {
            self.versions.push(FileMetaShallowVersion::try_from(version)?);
            return Ok(());
        }

        let vid = version.get_version_id();

        // Match existing version for replace; null version: None and Some(nil) are equivalent
        let matches = |h: &Option<Uuid>| {
            let v_null = vid.is_none() || vid == Some(Uuid::nil());
            let h_null = h.is_none() || *h == Some(Uuid::nil());
            (v_null && h_null) || (vid == *h)
        };

        if let Some(fidx) = self.versions.iter().position(|v| matches(&v.header.version_id)) {
            return self.set_idx(fidx, version);
        }

        // append placeholder to find insert position
        let placeholder = FileMetaShallowVersion {
            header: FileMetaVersionHeader {
                mod_time: None, // None sorts before any real mod_time
                ..Default::default()
            },
            meta: Vec::new(),
        };
        self.versions.push(placeholder);

        let mod_time = version.get_mod_time();
        let new_shallow = FileMetaShallowVersion::try_from(version)?;

        for (idx, exist) in self.versions.iter().enumerate() {
            let ex_mt = exist.header.mod_time;
            let insert_here = match (ex_mt, mod_time) {
                (None, _) => true, // placeholder: always insert before
                (Some(em), Some(nm)) => em <= nm,
                (Some(_), None) => false,
            };
            if insert_here {
                self.versions.insert(idx, new_shallow);
                self.versions.pop(); // remove placeholder
                return Ok(());
            }
        }
        self.versions.pop(); // remove placeholder on fallback
        Err(Error::other("add_version failed"))

        // if !ver.valid() {
        //     return Err(Error::other("attempted to add invalid version"));
        // }

        // if self.versions.len() + 1 >= 100 {
        //     return Err(Error::other(
        //         "You've exceeded the limit on the number of versions you can create on this object",
        //     ));
        // }

        // let mod_time = ver.get_mod_time();
        // let encoded = ver.marshal_msg()?;
        // let new_version = FileMetaShallowVersion {
        //     header: ver.header(),
        //     meta: encoded,
        // };

        // // Find the insertion position: insert before the first element with mod_time >= new mod_time
        // // This maintains descending order by mod_time (newest first)
        // let insert_pos = self
        //     .versions
        //     .iter()
        //     .position(|existing| existing.header.mod_time <= mod_time)
        //     .unwrap_or(self.versions.len());
        // self.versions.insert(insert_pos, new_version);
        // Ok(())
    }

    // delete_version deletes version, returns data_dir
    #[tracing::instrument(level = "debug", skip(self))]
    pub fn delete_version(&mut self, fi: &FileInfo) -> Result<Option<Uuid>> {
        let vid = Some(fi.version_id.unwrap_or(Uuid::nil()));
        let target_is_delete_marker = self
            .versions
            .iter()
            .find(|ver| ver.header.version_id == vid)
            .is_some_and(|ver| ver.header.version_type == VersionType::Delete);

        let mut ventry = FileMetaVersion::default();
        if fi.deleted {
            ventry.version_type = VersionType::Delete;
            ventry.delete_marker = Some(MetaDeleteMarker {
                version_id: vid,
                mod_time: fi.mod_time,
                ..Default::default()
            });

            if !fi.is_valid() {
                return Err(Error::other("invalid file meta version"));
            }
        }

        let mut update_version = false;
        if fi.version_purge_status().is_empty()
            && (fi.delete_marker_replication_status() == ReplicationStatusType::Replica
                || fi.delete_marker_replication_status() == ReplicationStatusType::Empty)
        {
            update_version = fi.mark_deleted;
        } else {
            if fi.deleted
                && fi.version_purge_status() != VersionPurgeStatusType::Complete
                && (!fi.version_purge_status().is_empty() || fi.delete_marker_replication_status().is_empty())
            {
                update_version = true;
            }

            if !fi.version_purge_status().is_empty() && fi.version_purge_status() != VersionPurgeStatusType::Complete {
                update_version = true;
            }
        }

        if target_is_delete_marker && !fi.deleted && !fi.version_purge_status().is_empty() {
            update_version = false;
        }

        if fi.deleted {
            if !fi.delete_marker_replication_status().is_empty()
                && let Some(delete_marker) = ventry.delete_marker.as_mut()
            {
                if fi.delete_marker_replication_status() == ReplicationStatusType::Replica {
                    insert_bytes(
                        &mut delete_marker.meta_sys,
                        SUFFIX_REPLICA_STATUS,
                        fi.replication_state_internal
                            .as_ref()
                            .map(|v| v.replica_status.clone())
                            .unwrap_or_default()
                            .as_str()
                            .as_bytes()
                            .to_vec(),
                    );
                    insert_bytes(
                        &mut delete_marker.meta_sys,
                        SUFFIX_REPLICA_TIMESTAMP,
                        fi.replication_state_internal
                            .as_ref()
                            .map(|v| v.replica_timestamp.unwrap_or(OffsetDateTime::UNIX_EPOCH).to_string())
                            .unwrap_or_default()
                            .as_bytes()
                            .to_vec(),
                    );
                } else {
                    insert_bytes(
                        &mut delete_marker.meta_sys,
                        SUFFIX_REPLICATION_STATUS,
                        fi.replication_state_internal
                            .as_ref()
                            .map(|v| v.replication_status_internal.clone().unwrap_or_default())
                            .unwrap_or_default()
                            .as_bytes()
                            .to_vec(),
                    );
                    insert_bytes(
                        &mut delete_marker.meta_sys,
                        SUFFIX_REPLICATION_TIMESTAMP,
                        fi.replication_state_internal
                            .as_ref()
                            .map(|v| v.replication_timestamp.unwrap_or(OffsetDateTime::UNIX_EPOCH).to_string())
                            .unwrap_or_default()
                            .as_bytes()
                            .to_vec(),
                    );
                }
            }

            if !fi.version_purge_status().is_empty()
                && let Some(delete_marker) = ventry.delete_marker.as_mut()
            {
                let value = fi
                    .replication_state_internal
                    .as_ref()
                    .map(|v| v.version_purge_status_internal.clone().unwrap_or_default())
                    .unwrap_or_default()
                    .as_bytes()
                    .to_vec();
                insert_bytes(&mut delete_marker.meta_sys, SUFFIX_PURGESTATUS, value);
            }

            if let Some(delete_marker) = ventry.delete_marker.as_mut() {
                for (k, v) in fi
                    .replication_state_internal
                    .as_ref()
                    .map(|v| v.reset_statuses_map.clone())
                    .unwrap_or_default()
                {
                    delete_marker.meta_sys.insert(k.clone(), v.clone().as_bytes().to_vec());
                }
            }
        }

        let mut found_index = None;

        for (i, ver) in self.versions.iter().enumerate() {
            if ver.header.version_id != vid {
                continue;
            }

            match ver.header.version_type {
                VersionType::Invalid | VersionType::Legacy => return Err(Error::other("invalid file meta version")),
                VersionType::Delete => {
                    if update_version {
                        let mut v = self.get_idx(i)?;
                        if v.delete_marker.is_none() {
                            v.delete_marker = Some(MetaDeleteMarker {
                                version_id: vid,
                                mod_time: fi.mod_time,
                                meta_sys: HashMap::new(),
                            });
                        }

                        if let Some(delete_marker) = v.delete_marker.as_mut() {
                            if !fi.delete_marker_replication_status().is_empty() {
                                if fi.delete_marker_replication_status() == ReplicationStatusType::Replica {
                                    insert_bytes(
                                        &mut delete_marker.meta_sys,
                                        SUFFIX_REPLICA_STATUS,
                                        fi.replication_state_internal
                                            .as_ref()
                                            .map(|v| v.replica_status.clone())
                                            .unwrap_or_default()
                                            .as_str()
                                            .as_bytes()
                                            .to_vec(),
                                    );
                                    insert_bytes(
                                        &mut delete_marker.meta_sys,
                                        SUFFIX_REPLICA_TIMESTAMP,
                                        fi.replication_state_internal
                                            .as_ref()
                                            .map(|v| v.replica_timestamp.unwrap_or(OffsetDateTime::UNIX_EPOCH).to_string())
                                            .unwrap_or_default()
                                            .as_bytes()
                                            .to_vec(),
                                    );
                                } else {
                                    insert_bytes(
                                        &mut delete_marker.meta_sys,
                                        SUFFIX_REPLICATION_STATUS,
                                        fi.replication_state_internal
                                            .as_ref()
                                            .map(|v| v.replication_status_internal.clone().unwrap_or_default())
                                            .unwrap_or_default()
                                            .as_bytes()
                                            .to_vec(),
                                    );
                                    insert_bytes(
                                        &mut delete_marker.meta_sys,
                                        SUFFIX_REPLICATION_TIMESTAMP,
                                        fi.replication_state_internal
                                            .as_ref()
                                            .map(|v| v.replication_timestamp.unwrap_or(OffsetDateTime::UNIX_EPOCH).to_string())
                                            .unwrap_or_default()
                                            .as_bytes()
                                            .to_vec(),
                                    );
                                }
                            }

                            for (k, v) in fi
                                .replication_state_internal
                                .as_ref()
                                .map(|v| v.reset_statuses_map.clone())
                                .unwrap_or_default()
                            {
                                delete_marker.meta_sys.insert(k.clone(), v.clone().as_bytes().to_vec());
                            }
                        }

                        self.set_idx(i, v)?;
                        return Ok(None);
                    }
                    self.versions.remove(i);

                    if (fi.mark_deleted && fi.version_purge_status() != VersionPurgeStatusType::Complete)
                        || (fi.deleted && vid == Some(Uuid::nil()))
                    {
                        self.add_version_filemata(ventry)?;
                    }

                    return Ok(None);
                }
                VersionType::Object => {
                    if update_version && !fi.deleted {
                        let mut v = self.get_idx(i)?;

                        if let Some(obj) = v.object.as_mut() {
                            let value = fi
                                .replication_state_internal
                                .as_ref()
                                .map(|v| v.version_purge_status_internal.clone().unwrap_or_default())
                                .unwrap_or_default()
                                .as_bytes()
                                .to_vec();
                            insert_bytes(&mut obj.meta_sys, SUFFIX_PURGESTATUS, value);
                            for (k, v) in fi
                                .replication_state_internal
                                .as_ref()
                                .map(|v| v.reset_statuses_map.clone())
                                .unwrap_or_default()
                            {
                                obj.meta_sys.insert(k.clone(), v.clone().as_bytes().to_vec());
                            }
                        }

                        let old_dir = v.object.as_ref().map(|v| v.data_dir).unwrap_or_default();
                        self.set_idx(i, v)?;

                        return Ok(old_dir);
                    }
                    found_index = Some(i);
                }
            }
        }

        let Some(i) = found_index else {
            if fi.deleted {
                self.add_version_filemata(ventry)?;
                return Ok(None);
            }
            return Err(Error::FileVersionNotFound);
        };

        let mut ver = self.get_idx(i)?;

        let Some(obj) = &mut ver.object else {
            if fi.deleted {
                self.add_version_filemata(ventry)?;
                return Ok(None);
            }
            return Err(Error::FileVersionNotFound);
        };

        let obj_version_id = obj.version_id;
        let obj_data_dir = obj.data_dir;

        let mut err = if fi.expire_restored {
            obj.remove_restore_hdrs();
            self.set_idx(i, ver).err()
        } else if fi.transition_status == TRANSITION_COMPLETE {
            obj.set_transition(fi);
            obj.reset_inline_data();
            self.set_idx(i, ver).err()
        } else {
            self.versions.remove(i);

            let (free_version, to_free) = obj.init_free_version(fi);

            if to_free {
                self.add_version_filemata(free_version).err()
            } else {
                None
            }
        };

        if fi.deleted {
            err = self.add_version_filemata(ventry).err();
        }

        if self.shared_data_dir_count(obj_version_id, obj_data_dir) > 0 {
            return Ok(None);
        }

        if let Some(e) = err {
            return Err(e);
        }

        Ok(obj_data_dir)
    }

    pub fn into_fileinfo(
        &self,
        volume: &str,
        path: &str,
        version_id: &str,
        read_data: bool,
        include_free_versions: bool,
        all_parts: bool,
    ) -> Result<FileInfo> {
        let vid = {
            if !version_id.is_empty() {
                Uuid::parse_str(version_id)?
            } else {
                Uuid::nil()
            }
        };

        let mut is_latest = true;
        let mut succ_mod_time = None;
        let mut non_free_versions = self.versions.len();

        let mut found = false;
        let mut found_free_version = None;
        let mut found_fi = None;

        for ver in self.versions.iter() {
            let header = &ver.header;

            // TODO: freeVersion
            if header.free_version() {
                non_free_versions -= 1;
                if include_free_versions
                    && found_free_version.is_none()
                    && let Ok(found_free_fi) = ver.parse_version_meta()
                    && found_free_fi.version_type != VersionType::Invalid
                {
                    let mut free_fi = found_free_fi.into_fileinfo(volume, path, all_parts);
                    free_fi.is_latest = true;
                    found_free_version = Some(free_fi);
                }

                if header.version_id != Some(vid) {
                    continue;
                }
            }

            if found {
                continue;
            }

            if !version_id.is_empty() && header.version_id != Some(vid) {
                is_latest = false;
                succ_mod_time = header.mod_time;
                continue;
            }

            found = true;

            let mut fi = ver.into_fileinfo(volume, path, all_parts)?;
            fi.is_latest = is_latest;

            if let Some(_d) = succ_mod_time {
                fi.successor_mod_time = succ_mod_time;
            }

            if read_data && fi.inline_data() {
                fi.data = self.find_inline_data_for_version(fi.version_id)?.map(bytes::Bytes::from);
            }

            found_fi = Some(fi);
        }

        if !found {
            if version_id.is_empty() {
                if include_free_versions
                    && non_free_versions == 0
                    && let Some(free_version) = found_free_version
                {
                    return Ok(free_version);
                }
                return Err(Error::FileNotFound);
            } else {
                return Err(Error::FileVersionNotFound);
            }
        }

        if let Some(mut fi) = found_fi {
            fi.num_versions = non_free_versions;

            return Ok(fi);
        }

        if version_id.is_empty() {
            Err(Error::FileNotFound)
        } else {
            Err(Error::FileVersionNotFound)
        }
    }

    pub fn get_file_info_versions(&self, volume: &str, path: &str, include_free_versions: bool) -> Result<FileInfoVersions> {
        let mut versions = self.into_file_info_versions(volume, path, true)?;

        let mut n = 0;

        let mut versions_vec = Vec::new();

        for fi in versions.versions.iter() {
            if fi.tier_free_version() {
                if !include_free_versions {
                    versions.free_versions.push(fi.clone());
                }
            } else {
                if !include_free_versions {
                    versions_vec.push(fi.clone());
                }
                n += 1;
            }
        }

        if !include_free_versions {
            versions.versions = versions_vec;
        }

        for fi in versions.free_versions.iter_mut() {
            fi.num_versions = n;
        }

        Ok(versions)
    }

    pub fn get_all_file_info_versions(&self, volume: &str, path: &str, all_parts: bool) -> Result<FileInfoVersions> {
        self.into_file_info_versions(volume, path, all_parts)
    }

    pub fn into_file_info_versions(&self, volume: &str, path: &str, all_parts: bool) -> Result<FileInfoVersions> {
        let mut versions = Vec::new();
        for version in self.versions.iter() {
            let fi = version.into_fileinfo(volume, path, all_parts)?;
            versions.push(fi);
        }

        let num = versions.len();
        let mut prev_mod_time = None;
        for (i, fi) in versions.iter_mut().enumerate() {
            if i == 0 {
                fi.is_latest = true;
            } else {
                fi.successor_mod_time = prev_mod_time;
            }
            fi.num_versions = num;
            prev_mod_time = fi.mod_time;
        }

        if versions.is_empty() {
            versions.push(FileInfo {
                name: path.to_string(),
                volume: volume.to_string(),
                deleted: true,
                is_latest: true,
                ..Default::default()
            });
        }

        Ok(FileInfoVersions {
            volume: volume.to_string(),
            name: path.to_string(),
            latest_mod_time: versions[0].mod_time,
            versions,
            ..Default::default()
        })
    }

    pub fn latest_mod_time(&self) -> Option<OffsetDateTime> {
        if self.versions.is_empty() {
            return None;
        }

        self.versions.first().unwrap().header.mod_time
    }

    /// Load or convert from buffer. Handles both current (meta_ver=3) and legacy (meta_ver=2) formats.
    pub fn load_or_convert(buf: &[u8]) -> Result<Self> {
        Self::load(buf)
    }

    /// List all versions as FileInfo
    pub fn list_versions(&self, volume: &str, path: &str, all_parts: bool) -> Result<Vec<FileInfo>> {
        let mut file_infos = Vec::new();
        for (i, version) in self.versions.iter().enumerate() {
            let mut fi = version.into_fileinfo(volume, path, all_parts)?;
            fi.is_latest = i == 0;
            file_infos.push(fi);
        }
        Ok(file_infos)
    }

    /// Check if all versions are hidden
    pub fn all_hidden(&self, top_delete_marker: bool) -> bool {
        if self.versions.is_empty() {
            return true;
        }

        if top_delete_marker && self.versions[0].header.version_type != VersionType::Delete {
            return false;
        }

        // Check if all versions are either delete markers or free versions
        self.versions
            .iter()
            .all(|v| v.header.version_type == VersionType::Delete || v.header.free_version())
    }

    /// Append metadata to buffer
    pub fn append_to(&self, dst: &mut Vec<u8>) -> Result<()> {
        let data = self.marshal_msg()?;
        dst.extend_from_slice(&data);
        Ok(())
    }

    /// Find version by string ID
    pub fn find_version_str(&self, version_id: &str) -> Result<(usize, FileMetaVersion)> {
        if version_id.is_empty() {
            return Err(Error::other("empty version ID"));
        }

        let uuid = Uuid::parse_str(version_id)?;
        self.find_version(Some(uuid))
    }
}

#[cfg(test)]
mod test {
    use super::*;
    use crate::test_data::*;

    #[test]
    fn test_new_file_meta() {
        let mut fm = FileMeta::new();

        let (m, n) = (3, 2);

        for i in 0..5 {
            let mut fi = FileInfo::new(i.to_string().as_str(), m, n);
            fi.mod_time = Some(OffsetDateTime::now_utc());

            fm.add_version(fi).unwrap();
        }

        let buff = fm.marshal_msg().unwrap();

        let mut newfm = FileMeta::default();
        newfm.unmarshal_msg(&buff).unwrap();

        assert_eq!(fm, newfm)
    }

    #[test]
    fn test_marshal_metaobject() {
        let obj = MetaObject {
            data_dir: Some(Uuid::new_v4()),
            ..Default::default()
        };

        // println!("obj {:?}", &obj);

        let encoded = obj.marshal_msg().unwrap();

        let mut obj2 = MetaObject::default();
        obj2.unmarshal_msg(&encoded).unwrap();

        // println!("obj2 {:?}", &obj2);

        assert_eq!(obj, obj2);
        assert_eq!(obj.data_dir, obj2.data_dir);
    }

    #[test]
    fn test_marshal_metadeletemarker() {
        let obj = MetaDeleteMarker {
            version_id: Some(Uuid::new_v4()),
            ..Default::default()
        };

        // println!("obj {:?}", &obj);

        let encoded = obj.marshal_msg().unwrap();

        let mut obj2 = MetaDeleteMarker::default();
        obj2.unmarshal_msg(&encoded).unwrap();

        // println!("obj2 {:?}", &obj2);

        assert_eq!(obj, obj2);
        assert_eq!(obj.version_id, obj2.version_id);
    }

    #[test]
    fn test_marshal_metaversion() {
        let mut fi = FileInfo::new("tset", 3, 2);
        fi.version_id = Some(Uuid::new_v4());
        fi.mod_time = Some(OffsetDateTime::from_unix_timestamp(OffsetDateTime::now_utc().unix_timestamp()).unwrap());
        let mut obj = FileMetaVersion::from(fi);
        obj.write_version = 110;

        // println!("obj {:?}", &obj);

        let encoded = obj.marshal_msg().unwrap();

        let mut obj2 = FileMetaVersion::default();
        obj2.unmarshal_msg(&encoded).unwrap();

        // println!("obj2 {:?}", &obj2);

        // Timestamp inconsistency
        assert_eq!(obj, obj2);
        assert_eq!(obj.get_version_id(), obj2.get_version_id());
        assert_eq!(obj.write_version, obj2.write_version);
        assert_eq!(obj.write_version, 110);
    }

    #[test]
    fn test_marshal_metaversionheader() {
        let mut obj = FileMetaVersionHeader::default();
        let vid = Some(Uuid::new_v4());
        obj.version_id = vid;

        let encoded = obj.marshal_msg().unwrap();

        let mut obj2 = FileMetaVersionHeader::default();
        obj2.unmarshal_msg(&encoded).unwrap();

        // Timestamp inconsistency
        assert_eq!(obj, obj2);
        assert_eq!(obj.version_id, obj2.version_id);
        assert_eq!(obj.version_id, vid);
    }

    #[test]
    fn test_real_xlmeta_compatibility() {
        // Test compatibility with real xl.meta formats
        let data = create_real_xlmeta().expect("Failed to create realistic test data");

        // Verify the file header
        assert_eq!(&data[0..4], b"XL2 ", "File header should be 'XL2 '");
        assert_eq!(&data[4..8], &[1, 0, 3, 0], "Version number should be 1.3.0");

        // Parse metadata
        let fm = FileMeta::load(&data).expect("Failed to parse realistic data");

        // Verify basic properties
        assert_eq!(fm.meta_ver, XL_META_VERSION);
        assert_eq!(
            fm.versions.len(),
            3,
            "Should have three versions (one object, one delete marker, one Legacy)"
        );

        // Verify version types
        let mut object_count = 0;
        let mut delete_count = 0;
        let mut legacy_count = 0;

        for version in &fm.versions {
            match version.header.version_type {
                VersionType::Object => object_count += 1,
                VersionType::Delete => delete_count += 1,
                VersionType::Legacy => legacy_count += 1,
                VersionType::Invalid => panic!("No invalid versions should be present"),
            }
        }

        assert_eq!(object_count, 1, "Should have one object version");
        assert_eq!(delete_count, 1, "Should have one delete marker");
        assert_eq!(legacy_count, 1, "Should have one Legacy version");

        // Verify compatibility
        assert!(fm.is_compatible_with_meta(), "Should be compatible with the xl format");

        // Verify integrity
        fm.validate_integrity().expect("Integrity validation failed");

        // Verify version statistics
        let stats = fm.get_version_stats();
        assert_eq!(stats.total_versions, 3);
        assert_eq!(stats.object_versions, 1);
        assert_eq!(stats.delete_markers, 1);
        assert_eq!(stats.invalid_versions, 1); // Legacy is counted as invalid
    }

    #[test]
    fn test_issue_2288_legacy_xlmeta_compatibility() {
        let data = create_issue_2288_legacy_xlmeta().expect("Failed to load issue #2288 fixture");
        let (major, minor, header_ver, meta_ver) = FileMeta::read_format_versions(&data).unwrap();
        assert_eq!((major, minor, header_ver, meta_ver), (1, 3, 2, 1));

        let fm = FileMeta::load(&data).expect("Failed to parse legacy issue #2288 xl.meta");
        assert_eq!(fm.meta_ver, 1);
        assert_eq!(fm.versions.len(), 1);
        assert_eq!(fm.versions[0].header.version_type, VersionType::Object);
        assert_eq!(fm.versions[0].header.signature, [0x96, 0x33, 0x4c, 0x78]);
        assert_eq!(fm.versions[0].header.ec_n, 0);
        assert_eq!(fm.versions[0].header.ec_m, 0);

        let fi = fm
            .into_fileinfo("viscom", "test.txt", "", true, false, true)
            .expect("Failed to extract file info from legacy issue #2288 xl.meta");
        assert_eq!(fi.size, 35);
        assert_eq!(fi.num_versions, 1);
        assert!(fi.is_latest);
    }

    #[test]
    fn test_issue_2265_legacy_meta_v2_object_compatibility() {
        let data = create_issue_2265_legacy_meta_v2_object_xlmeta().expect("Failed to load issue #2265 object fixture");
        let (major, minor, header_ver, meta_ver) = FileMeta::read_format_versions(&data).unwrap();
        assert_eq!((major, minor, header_ver, meta_ver), (1, 3, 3, 2));

        let fm = FileMeta::load(&data).expect("Failed to parse legacy issue #2265 object xl.meta");
        assert_eq!(fm.meta_ver, 2);
        assert_eq!(fm.versions.len(), 1);
        assert_eq!(fm.versions[0].header.version_type, VersionType::Object);
        assert_eq!(fm.versions[0].header.ec_n, 0);
        assert_eq!(fm.versions[0].header.ec_m, 1);

        let fi = fm
            .into_fileinfo("bucket", ".metadata.bin", "", true, false, true)
            .expect("Failed to extract file info from legacy issue #2265 object xl.meta");
        assert_eq!(fi.size, 707);
        assert_eq!(fi.num_versions, 1);
        assert_eq!(fi.metadata.get("etag").map(String::as_str), Some("4359404618e32a0bd8944e9ff6802f53"));
        assert_eq!(
            fi.data_dir.map(|id| id.to_string()).as_deref(),
            Some("04bee19a-6eea-40c4-96fd-1f39257fcbdc")
        );
        assert!(fi.uses_legacy_checksum);
        assert!(fi.is_latest);
    }

    #[test]
    fn test_issue_2265_legacy_meta_v2_config_compatibility() {
        let data = create_issue_2265_legacy_meta_v2_config_xlmeta().expect("Failed to load issue #2265 config fixture");
        let (major, minor, header_ver, meta_ver) = FileMeta::read_format_versions(&data).unwrap();
        assert_eq!((major, minor, header_ver, meta_ver), (1, 3, 3, 2));

        let fm = FileMeta::load(&data).expect("Failed to parse legacy issue #2265 config xl.meta");
        assert_eq!(fm.meta_ver, 2);
        assert_eq!(fm.versions.len(), 1);
        assert_eq!(fm.versions[0].header.version_type, VersionType::Object);

        let fi = fm
            .into_fileinfo("config", "format.json", "", true, false, true)
            .expect("Failed to extract file info from legacy issue #2265 config xl.meta");
        assert_eq!(fi.size, 74);
        assert_eq!(fi.num_versions, 1);
        assert_eq!(fi.metadata.get("etag").map(String::as_str), Some("12b368ce52e496e61ac47b366c7c3b66"));
        assert_eq!(
            fi.data_dir.map(|id| id.to_string()).as_deref(),
            Some("fba8e4c3-3f42-4242-94e0-5ab84b83ae97")
        );
        assert!(fi.uses_legacy_checksum);
        assert!(fi.is_latest);
    }

    #[test]
    fn test_issue_2434_legacy_meta_v2_pool_compatibility() {
        let data = create_issue_2434_legacy_meta_v2_pool_xlmeta().expect("Failed to load issue #2434 pool fixture");
        let (major, minor, header_ver, meta_ver) = FileMeta::read_format_versions(&data).unwrap();
        assert_eq!((major, minor, header_ver, meta_ver), (1, 3, 3, 2));

        let fm = FileMeta::load(&data).expect("Failed to parse legacy issue #2434 pool xl.meta");
        assert_eq!(fm.meta_ver, 2);
        assert_eq!(fm.versions.len(), 1);
        assert_eq!(fm.versions[0].header.version_type, VersionType::Object);

        let fi = fm
            .into_fileinfo(".rustfs.sys", "pool.bin", "", true, false, true)
            .expect("Failed to extract file info from legacy issue #2434 pool xl.meta");
        assert_eq!(fi.size, 48);
        assert_eq!(fi.num_versions, 1);
        assert_eq!(fi.version_id, None);
        assert_eq!(fi.metadata.get("etag").map(String::as_str), Some("8d270d7a184cfa30cc0bf09ea74fd964"));
        assert_eq!(
            fi.data_dir.map(|id| id.to_string()).as_deref(),
            Some("2bcefaca-44dd-4f01-a79e-63eeb0dda396")
        );
        assert!(fi.uses_legacy_checksum);
        assert!(fi.is_latest);
    }

    #[test]
    fn test_legacy_v1_object_xlmeta_compatibility() {
        let data = create_legacy_v1_object_xlmeta().expect("Failed to create legacy v1 object xl.meta");
        let (major, minor, header_ver, meta_ver) = FileMeta::read_format_versions(&data).unwrap();
        assert_eq!((major, minor, header_ver, meta_ver), (1, 3, 1, 1));

        let fm = FileMeta::load(&data).expect("Failed to parse legacy v1 object xl.meta");
        assert_eq!(fm.meta_ver, 1);
        assert_eq!(fm.versions.len(), 1);
        assert_eq!(fm.versions[0].header.version_type, VersionType::Legacy);
        assert_eq!(fm.versions[0].header.ec_n, 0);
        assert_eq!(fm.versions[0].header.ec_m, 0);

        let fi = fm
            .into_fileinfo("bucket", "hello.txt", "", true, false, true)
            .expect("Failed to extract file info from legacy v1 object xl.meta");
        assert_eq!(fi.size, 11);
        assert_eq!(fi.num_versions, 1);
        assert_eq!(fi.mode, Some(0o644));
        assert_eq!(fi.parts.len(), 1);
        assert_eq!(fi.parts[0].etag, "etag-1");
        assert_eq!(fi.parts[0].size, 11);
        assert_eq!(fi.erasure.data_blocks, 4);
        assert_eq!(fi.erasure.parity_blocks, 2);
        assert_eq!(fi.metadata.get("content-type").map(String::as_str), Some("text/plain"));
        assert!(fi.is_latest);
    }

    #[test]
    fn test_complex_xlmeta_handling() {
        // Test complex xl.meta files with many versions
        let data = create_complex_xlmeta().expect("Failed to create complex test data");
        let fm = FileMeta::load(&data).expect("Failed to parse complex data");

        // Verify version count
        assert!(fm.versions.len() >= 10, "Should have at least 10 versions");

        // Verify version ordering
        assert!(fm.is_sorted_by_mod_time(), "Versions should be sorted by modification time");

        // Verify presence of different version types
        let stats = fm.get_version_stats();
        assert!(stats.object_versions > 0, "Should include object versions");
        assert!(stats.delete_markers > 0, "Should include delete markers");

        // Test version merge functionality
        let merged = merge_file_meta_versions(1, false, 0, std::slice::from_ref(&fm.versions));
        assert!(!merged.is_empty(), "Merged output should contain versions");
    }

    #[test]
    fn test_inline_data_handling() {
        // Test inline data handling
        let data = create_xlmeta_with_inline_data().expect("Failed to create inline test data");
        let fm = FileMeta::load(&data).expect("Failed to parse inline data");

        assert_eq!(fm.versions.len(), 1, "Should have one version");
        assert!(!fm.data.as_slice().is_empty(), "Should contain inline data");

        // Verify inline data contents
        let inline_data = fm.data.as_slice();
        assert!(!inline_data.is_empty(), "Inline data should not be empty");
    }

    #[test]
    fn test_into_fileinfo_reads_legacy_nil_uuid_inline_key() {
        let mut fm = FileMeta::new();
        let mut fi = FileInfo::new("test", 2, 1);
        fi.mod_time = Some(OffsetDateTime::now_utc());
        fi.data = Some(Bytes::from_static(b"legacy-inline"));
        fi.set_inline_data();
        fm.add_version(fi).unwrap();

        let current_key = data_key_for_version(Some(Uuid::nil()));
        let legacy_key = legacy_data_key_for_version(Some(Uuid::nil())).unwrap();
        let payload = fm.data.find(current_key.as_str()).unwrap().unwrap();

        fm.data.remove_key(current_key.as_str()).unwrap();
        fm.data.replace(legacy_key.as_str(), payload.clone()).unwrap();

        let restored = fm.into_fileinfo("bucket", "test", "", true, false, true).unwrap();

        assert!(restored.inline_data());
        assert_eq!(restored.data, Some(Bytes::from(payload)));
    }

    #[test]
    fn test_error_handling_and_recovery() {
        // Test error handling and recovery
        let corrupted_data = create_corrupted_xlmeta();
        let result = FileMeta::load(&corrupted_data);
        assert!(result.is_err(), "Corrupted data should fail to parse");

        // Test handling of empty files
        let empty_data = create_empty_xlmeta().expect("Failed to create empty test data");
        let fm = FileMeta::load(&empty_data).expect("Failed to parse empty data");
        assert_eq!(fm.versions.len(), 0, "An empty file should have no versions");
    }

    #[test]
    fn test_version_type_legacy_support() {
        // Validate support for Legacy version types
        assert_eq!(VersionType::Legacy.to_u8(), 3);
        assert_eq!(VersionType::from_u8(3), VersionType::Legacy);
        assert!(VersionType::Legacy.valid(), "Legacy type should be valid");

        // Exercise creation and handling of Legacy versions
        let legacy_version = FileMetaVersion {
            version_type: VersionType::Legacy,
            legacy_object: None,
            object: None,
            delete_marker: None,
            write_version: 1,
            uses_legacy_checksum: false,
        };

        assert!(legacy_version.is_legacy(), "Should be recognized as a Legacy version");
    }

    #[test]
    fn test_signature_calculation() {
        // Test signature calculation
        let data = create_real_xlmeta().expect("Failed to create test data");
        let fm = FileMeta::load(&data).expect("Parsing failed");

        for version in &fm.versions {
            let signature = version.header.get_signature();
            assert_eq!(signature.len(), 4, "Signature should be 4 bytes");

            // Verify signature consistency for identical versions
            let signature2 = version.header.get_signature();
            assert_eq!(signature, signature2, "Identical versions should produce identical signatures");
        }
    }

    #[test]
    fn test_metadata_validation() {
        // Test metadata validation
        let data = create_real_xlmeta().expect("Failed to create test data");
        let fm = FileMeta::load(&data).expect("Parsing failed");

        // Test integrity validation
        fm.validate_integrity().expect("Integrity validation should succeed");

        // Test compatibility checks
        assert!(fm.is_compatible_with_meta(), "Should be compatible with the xl format");

        // Test version ordering checks
        assert!(fm.is_sorted_by_mod_time(), "Versions should be time-ordered");
    }

    #[test]
    fn test_round_trip_serialization() {
        // Test round-trip serialization consistency
        let original_data = create_real_xlmeta().expect("Failed to create original test data");
        let fm = FileMeta::load(&original_data).expect("Failed to parse original data");

        // Serialize again
        let serialized_data = fm.marshal_msg().expect("Re-serialization failed");

        // Parse again
        let fm2 = FileMeta::load(&serialized_data).expect("Failed to parse serialized data");

        // Verify consistency
        assert_eq!(fm.versions.len(), fm2.versions.len(), "Version counts should match");
        assert_eq!(fm.meta_ver, fm2.meta_ver, "Metadata versions should match");

        // Verify version content consistency
        for (v1, v2) in fm.versions.iter().zip(fm2.versions.iter()) {
            assert_eq!(v1.header.version_type, v2.header.version_type, "Version types should match");
            assert_eq!(v1.header.version_id, v2.header.version_id, "Version IDs should match");
        }
    }

    #[test]
    fn test_performance_with_large_metadata() {
        // Test performance with large metadata files
        use std::time::Instant;

        let start = Instant::now();
        let data = create_complex_xlmeta().expect("Failed to create large test data");
        let creation_time = start.elapsed();

        let start = Instant::now();
        let fm = FileMeta::load(&data).expect("Failed to parse large data");
        let parsing_time = start.elapsed();

        let start = Instant::now();
        let _serialized = fm.marshal_msg().expect("Serialization failed");
        let serialization_time = start.elapsed();

        println!("Performance results:");
        println!("  Creation time: {creation_time:?}");
        println!("  Parsing time: {parsing_time:?}");
        println!("  Serialization time: {serialization_time:?}");

        // Basic performance assertions (adjust as needed for real workloads)
        assert!(parsing_time.as_millis() < 100, "Parsing time should be under 100 ms");
        assert!(serialization_time.as_millis() < 100, "Serialization time should be under 100 ms");
    }

    #[test]
    fn test_edge_cases() {
        // Test edge cases

        // 1. Test empty version IDs
        let mut fm = FileMeta::new();
        let version = FileMetaVersion {
            version_type: VersionType::Object,
            legacy_object: None,
            object: Some(MetaObject {
                version_id: None, // Empty version ID
                data_dir: None,
                erasure_algorithm: crate::fileinfo::ErasureAlgo::ReedSolomon,
                erasure_m: 1,
                erasure_n: 1,
                erasure_block_size: 64 * 1024,
                erasure_index: 0,
                erasure_dist: vec![0],
                bitrot_checksum_algo: ChecksumAlgo::HighwayHash,
                part_numbers: vec![1],
                part_etags: Vec::new(),
                part_sizes: vec![0],
                part_actual_sizes: Vec::new(),
                part_indices: Vec::new(),
                size: 0,
                mod_time: None,
                meta_sys: HashMap::new(),
                meta_user: HashMap::new(),
            }),
            delete_marker: None,
            write_version: 1,
            uses_legacy_checksum: false,
        };

        let shallow_version = FileMetaShallowVersion::try_from(version).expect("Conversion failed");
        fm.versions.push(shallow_version);

        // Should support serialization and deserialization
        let data = fm.marshal_msg().expect("Serialization failed");
        let fm2 = FileMeta::load(&data).expect("Parsing failed");
        assert_eq!(fm2.versions.len(), 1);

        // 2. Test extremely large file sizes
        let large_object = MetaObject {
            size: i64::MAX,
            part_sizes: vec![usize::MAX],
            ..Default::default()
        };

        // Should handle very large numbers
        assert_eq!(large_object.size, i64::MAX);
    }

    #[tokio::test]
    async fn test_concurrent_operations() {
        // Test thread safety for concurrent operations
        use std::sync::Arc;
        use std::sync::Mutex;

        let fm = Arc::new(Mutex::new(FileMeta::new()));
        let mut handles = vec![];

        // Add versions concurrently
        for i in 0..10 {
            let fm_clone: Arc<Mutex<FileMeta>> = Arc::clone(&fm);
            let handle = tokio::spawn(async move {
                let mut fi = crate::fileinfo::FileInfo::new(&format!("test-{i}"), 2, 1);
                fi.version_id = Some(Uuid::new_v4());
                fi.mod_time = Some(OffsetDateTime::now_utc());

                let mut fm_guard = fm_clone.lock().unwrap();
                fm_guard.add_version(fi).unwrap();
            });
            handles.push(handle);
        }

        // Wait for all tasks to finish
        for handle in handles {
            handle.await.unwrap();
        }

        let fm_guard = fm.lock().unwrap();
        assert_eq!(fm_guard.versions.len(), 10);
    }

    #[test]
    fn test_memory_efficiency() {
        // Test memory efficiency
        use std::mem;

        // Measure memory usage for empty structs
        let empty_fm = FileMeta::new();
        let empty_size = mem::size_of_val(&empty_fm);
        println!("Empty FileMeta size: {empty_size} bytes");

        // Measure memory usage with many versions
        let mut large_fm = FileMeta::new();
        for i in 0..100 {
            let mut fi = crate::fileinfo::FileInfo::new(&format!("test-{i}"), 2, 1);
            fi.version_id = Some(Uuid::new_v4());
            fi.mod_time = Some(OffsetDateTime::now_utc());
            large_fm.add_version(fi).unwrap();
        }

        let large_size = mem::size_of_val(&large_fm);
        println!("Large FileMeta size: {large_size} bytes");

        // Ensure memory usage is reasonable (size_of_val covers only stack allocations)
        // For structs containing Vec, size_of_val may match because capacity lives on the heap
        println!("Number of versions: {}", large_fm.versions.len());
        assert!(!large_fm.versions.is_empty(), "Should contain version data");
    }

    #[test]
    fn test_version_ordering_edge_cases() {
        // Test boundary cases for version ordering
        let mut fm = FileMeta::new();

        // Add versions with identical timestamps
        let same_time = OffsetDateTime::now_utc();
        for i in 0..5 {
            let mut fi = crate::fileinfo::FileInfo::new(&format!("test-{i}"), 2, 1);
            fi.version_id = Some(Uuid::new_v4());
            fi.mod_time = Some(same_time);
            fm.add_version(fi).unwrap();
        }

        // Verify stable ordering
        let original_order = fm.versions.iter().map(|v| v.header.version_id).len();
        fm.sort_by_mod_time();
        let sorted_order = fm.versions.iter().map(|v| v.header.version_id).len();

        // Sorting should remain stable for identical timestamps
        assert_eq!(original_order, sorted_order);
    }

    #[test]
    fn test_checksum_algorithms() {
        // Test different checksum algorithms
        let algorithms = vec![ChecksumAlgo::Invalid, ChecksumAlgo::HighwayHash];

        for algo in algorithms {
            let obj = MetaObject {
                bitrot_checksum_algo: algo.clone(),
                ..Default::default()
            };

            // Verify checksum validation logic
            match algo {
                ChecksumAlgo::Invalid => assert!(!algo.valid()),
                ChecksumAlgo::HighwayHash => assert!(algo.valid()),
            }

            // Verify serialization and deserialization
            let data = obj.marshal_msg().unwrap();
            let mut obj2 = MetaObject::default();
            obj2.unmarshal_msg(&data).unwrap();
            assert_eq!(obj.bitrot_checksum_algo.to_u8(), obj2.bitrot_checksum_algo.to_u8());
        }
    }

    #[test]
    fn test_erasure_coding_parameters() {
        // Test combinations of erasure coding parameters
        let test_cases = vec![
            (1, 1), // Minimum configuration
            (2, 1), // Common configuration
            (4, 2), // Standard configuration
            (8, 4), // High redundancy configuration
        ];

        for (data_blocks, parity_blocks) in test_cases {
            let obj = MetaObject {
                erasure_m: data_blocks,
                erasure_n: parity_blocks,
                erasure_dist: (0..(data_blocks + parity_blocks)).map(|i| i as u8).collect(),
                ..Default::default()
            };

            // Verify parameter validity
            assert!(obj.erasure_m > 0, "Data block count must be greater than 0");
            assert!(obj.erasure_n > 0, "Parity block count must be greater than 0");
            assert_eq!(obj.erasure_dist.len(), data_blocks + parity_blocks);

            // Verify serialization and deserialization
            let data = obj.marshal_msg().unwrap();
            let mut obj2 = MetaObject::default();
            obj2.unmarshal_msg(&data).unwrap();
            assert_eq!(obj.erasure_m, obj2.erasure_m);
            assert_eq!(obj.erasure_n, obj2.erasure_n);
            assert_eq!(obj.erasure_dist, obj2.erasure_dist);
        }
    }

    #[test]
    fn test_metadata_size_limits() {
        // Test metadata size limits
        let mut obj = MetaObject::default();

        // Test moderate amounts of user metadata
        for i in 0..10 {
            obj.meta_user
                .insert(format!("key-{i:04}"), format!("value-{:04}-{}", i, "x".repeat(10)));
        }

        // Verify metadata can be serialized
        let data = obj.marshal_msg().unwrap();
        assert!(data.len() > 100, "Serialized data should have a reasonable size");

        // Verify deserialization succeeds
        let mut obj2 = MetaObject::default();
        obj2.unmarshal_msg(&data).unwrap();
        assert_eq!(obj.meta_user.len(), obj2.meta_user.len());
    }

    #[test]
    fn test_version_statistics_accuracy() {
        // Test accuracy of version statistics
        let mut fm = FileMeta::new();

        // Add different version types
        let object_count = 3;
        let delete_count = 2;

        // Add object versions
        for i in 0..object_count {
            let mut fi = crate::fileinfo::FileInfo::new(&format!("obj-{i}"), 2, 1);
            fi.version_id = Some(Uuid::new_v4());
            fi.mod_time = Some(OffsetDateTime::now_utc());
            fm.add_version(fi).unwrap();
        }

        // Add delete markers
        for i in 0..delete_count {
            let delete_marker = MetaDeleteMarker {
                version_id: Some(Uuid::new_v4()),
                mod_time: Some(OffsetDateTime::now_utc()),
                meta_sys: HashMap::new(),
            };

            let delete_version = FileMetaVersion {
                version_type: VersionType::Delete,
                legacy_object: None,
                object: None,
                delete_marker: Some(delete_marker),
                write_version: (i + 100) as u64,
                uses_legacy_checksum: false,
            };

            let shallow_version = FileMetaShallowVersion::try_from(delete_version).unwrap();
            fm.versions.push(shallow_version);
        }

        // Verify overall statistics
        let stats = fm.get_version_stats();
        assert_eq!(stats.total_versions, object_count + delete_count);
        assert_eq!(stats.object_versions, object_count);
        assert_eq!(stats.delete_markers, delete_count);

        // Verify detailed statistics
        let detailed_stats = fm.get_detailed_version_stats();
        assert_eq!(detailed_stats.total_versions, object_count + delete_count);
        assert_eq!(detailed_stats.object_versions, object_count);
        assert_eq!(detailed_stats.delete_markers, delete_count);
    }

    #[test]
    fn test_cross_platform_compatibility() {
        // Test cross-platform compatibility (endianness, separators, etc.)
        let mut fm = FileMeta::new();

        // Use platform-specific path styles
        let paths = vec![
            "unix/style/path",
            "windows\\style\\path",
            "mixed/style\\path",
            "unicode/path/test",
        ];

        for path in paths {
            let mut fi = crate::fileinfo::FileInfo::new(path, 2, 1);
            fi.version_id = Some(Uuid::new_v4());
            fi.mod_time = Some(OffsetDateTime::now_utc());
            fm.add_version(fi).unwrap();
        }

        // Verify serialization/deserialization consistency across platforms
        let data = fm.marshal_msg().unwrap();
        let mut fm2 = FileMeta::default();
        fm2.unmarshal_msg(&data).unwrap();

        assert_eq!(fm.versions.len(), fm2.versions.len());

        // Verify UUID endianness consistency
        for (v1, v2) in fm.versions.iter().zip(fm2.versions.iter()) {
            assert_eq!(v1.header.version_id, v2.header.version_id);
        }
    }

    #[test]
    fn delete_version_removes_delete_marker_during_version_purge_replication() {
        let version_id = Uuid::new_v4();
        let mut fm = FileMeta::new();
        fm.add_version_filemata(FileMetaVersion {
            version_type: VersionType::Delete,
            legacy_object: None,
            object: None,
            delete_marker: Some(MetaDeleteMarker {
                version_id: Some(version_id),
                mod_time: Some(OffsetDateTime::now_utc()),
                meta_sys: HashMap::new(),
            }),
            write_version: 1,
            uses_legacy_checksum: false,
        })
        .unwrap();

        let fi = FileInfo {
            deleted: false,
            mark_deleted: false,
            version_id: Some(version_id),
            replication_state_internal: Some(ReplicationState {
                version_purge_status_internal: Some("target=PENDING;".to_string()),
                purge_targets: version_purge_statuses_map("target=PENDING;"),
                ..Default::default()
            }),
            ..Default::default()
        };

        let result = fm.delete_version(&fi).unwrap();

        assert!(result.is_none());
        assert!(
            fm.versions.is_empty(),
            "delete-marker version purge should remove the local marker instead of rewriting purge metadata onto it"
        );
    }

    #[test]
    fn test_data_integrity_validation() {
        // Test data integrity checks
        let mut fm = FileMeta::new();

        // Add a normal version
        let mut fi = crate::fileinfo::FileInfo::new("test", 2, 1);
        fi.version_id = Some(Uuid::new_v4());
        fi.mod_time = Some(OffsetDateTime::now_utc());
        fm.add_version(fi).unwrap();

        // Verify integrity under normal conditions
        assert!(fm.validate_integrity().is_ok());
    }

    #[test]
    fn test_add_version_clears_stale_inline_data_for_null_version() {
        let mut fm = FileMeta::new();

        let mut inline_fi = crate::fileinfo::FileInfo::new("test", 2, 1);
        inline_fi.mod_time = Some(OffsetDateTime::now_utc());
        inline_fi.data = Some(Bytes::new());
        inline_fi.set_inline_data();
        fm.add_version(inline_fi).unwrap();

        let inline_version = fm.into_fileinfo("bucket", "test", "", true, false, true).unwrap();
        assert!(inline_version.inline_data());
        assert_eq!(inline_version.data, Some(Bytes::new()));

        let mut disk_fi = crate::fileinfo::FileInfo::new("test", 2, 1);
        disk_fi.mod_time = Some(OffsetDateTime::now_utc() + time::Duration::seconds(1));
        disk_fi.size = 1024;
        fm.add_version(disk_fi).unwrap();

        let latest = fm.into_fileinfo("bucket", "test", "", true, false, true).unwrap();
        assert!(!latest.inline_data());
        assert!(latest.data.is_none());
        assert!(
            fm.data
                .find(data_key_for_version(latest.version_id).as_str())
                .unwrap()
                .is_none()
        );
    }

    #[test]
    fn test_add_version_keeps_inline_data_when_version_insert_fails() {
        let mut fm = FileMeta::new();

        let mut inline_fi = crate::fileinfo::FileInfo::new("test", 2, 1);
        inline_fi.mod_time = Some(OffsetDateTime::now_utc());
        inline_fi.data = Some(Bytes::from_static(b"inline"));
        inline_fi.set_inline_data();
        fm.add_version(inline_fi).unwrap();

        let before = fm.data.find(data_key_for_version(Some(Uuid::nil())).as_str()).unwrap();
        assert_eq!(before, Some(Bytes::from_static(b"inline").to_vec()));

        let invalid_disk_fi = crate::fileinfo::FileInfo::new("test", 2, 1);
        let err = fm.add_version(invalid_disk_fi).unwrap_err();
        assert!(err.to_string().contains("file meta version invalid"));

        let after = fm.data.find(data_key_for_version(Some(Uuid::nil())).as_str()).unwrap();
        assert_eq!(after, Some(Bytes::from_static(b"inline").to_vec()));
    }

    #[test]
    fn test_update_object_version_persists_checksum_metadata() {
        let mut fm = FileMeta::new();
        let version_id = Some(Uuid::new_v4());

        let mut fi = crate::fileinfo::FileInfo::new("test", 2, 1);
        fi.version_id = version_id;
        fi.mod_time = Some(OffsetDateTime::now_utc());
        fm.add_version(fi).unwrap();

        let checksum = Bytes::from_static(b"resolved-checksum");
        let mut update = crate::fileinfo::FileInfo::new("test", 2, 1);
        update.version_id = version_id;
        update.metadata.insert("x-amz-meta-owner".to_string(), "alice".to_string());
        update.checksum = Some(checksum.clone());

        fm.update_object_version(update).unwrap();

        let (_, version) = fm.find_version(version_id).unwrap();
        let stored = version.into_fileinfo("bucket", "test", true);
        assert_eq!(stored.metadata.get("x-amz-meta-owner"), Some(&"alice".to_string()));
        assert_eq!(stored.checksum, Some(checksum));
    }

    #[test]
    fn test_version_merge_scenarios() {
        // Test various version merge scenarios
        let mut versions1 = vec![];
        let mut versions2 = vec![];

        // Create two distinct sets of versions
        for i in 0..3 {
            let mut fi1 = crate::fileinfo::FileInfo::new(&format!("test1-{i}"), 2, 1);
            fi1.version_id = Some(Uuid::new_v4());
            fi1.mod_time = Some(OffsetDateTime::from_unix_timestamp(1000 + i * 10).unwrap());

            let mut fi2 = crate::fileinfo::FileInfo::new(&format!("test2-{i}"), 2, 1);
            fi2.version_id = Some(Uuid::new_v4());
            fi2.mod_time = Some(OffsetDateTime::from_unix_timestamp(1005 + i * 10).unwrap());

            let version1 = FileMetaVersion::from(fi1);
            let version2 = FileMetaVersion::from(fi2);

            versions1.push(FileMetaShallowVersion::try_from(version1).unwrap());
            versions2.push(FileMetaShallowVersion::try_from(version2).unwrap());
        }

        // Test a simple merge scenario
        let merged = merge_file_meta_versions(1, false, 0, &[versions1.clone()]);
        assert!(!merged.is_empty(), "Merging a single version list should not be empty");

        // Test merging multiple version lists
        let merged = merge_file_meta_versions(1, false, 0, &[versions1.clone(), versions2.clone()]);
        // Merge results may be empty depending on compatibility, which is acceptable
        println!("Merge result count: {}", merged.len());
    }

    #[test]
    fn test_flags_operations() {
        // Test flag operations
        let flags = vec![Flags::FreeVersion, Flags::UsesDataDir, Flags::InlineData];

        for flag in flags {
            let flag_value = flag as u8;
            assert!(flag_value > 0, "Flag value should be greater than 0");

            // Test flag combinations
            let combined = Flags::FreeVersion as u8 | Flags::UsesDataDir as u8;
            // For bitwise operations, combined values may not exceed individual ones; this is normal
            assert!(combined > 0, "Combined flag value should be greater than 0");
        }
    }

    #[test]
    fn test_uuid_handling_edge_cases() {
        // Test UUID edge cases
        let test_uuids = vec![
            Uuid::new_v4(), // Random UUID
        ];

        for uuid in test_uuids {
            let obj = MetaObject {
                version_id: Some(uuid),
                data_dir: Some(uuid),
                ..Default::default()
            };

            // Verify serialization and deserialization
            let data = obj.marshal_msg().unwrap();
            let mut obj2 = MetaObject::default();
            obj2.unmarshal_msg(&data).unwrap();

            assert_eq!(obj.version_id, obj2.version_id);
            assert_eq!(obj.data_dir, obj2.data_dir);
        }

        // Test nil UUID separately because serialization converts it to None
        let obj = MetaObject {
            version_id: Some(Uuid::nil()),
            data_dir: Some(Uuid::nil()),
            ..Default::default()
        };

        let data = obj.marshal_msg().unwrap();
        let mut obj2 = MetaObject::default();
        obj2.unmarshal_msg(&data).unwrap();

        // nil UUIDs may be converted to None during serialization; this is expected
        // Inspect the actual serialization behavior
        println!("Original version_id: {:?}", obj.version_id);
        println!("Deserialized version_id: {:?}", obj2.version_id);
        // Consider the test successful as long as deserialization succeeds
    }

    #[test]
    fn test_part_handling_edge_cases() {
        // Test edge cases for shard handling
        let mut obj = MetaObject::default();

        // Test an empty shard list
        assert!(obj.part_numbers.is_empty());
        assert!(obj.part_etags.is_empty());
        assert!(obj.part_sizes.is_empty());

        // Test a single shard
        obj.part_numbers = vec![1];
        obj.part_etags = vec!["etag1".to_string()];
        obj.part_sizes = vec![1024];
        obj.part_actual_sizes = vec![1024];

        let data = obj.marshal_msg().unwrap();
        let mut obj2 = MetaObject::default();
        obj2.unmarshal_msg(&data).unwrap();

        assert_eq!(obj.part_numbers, obj2.part_numbers);
        assert_eq!(obj.part_etags, obj2.part_etags);
        assert_eq!(obj.part_sizes, obj2.part_sizes);
        assert_eq!(obj.part_actual_sizes, obj2.part_actual_sizes);

        // Test multiple shards
        obj.part_numbers = vec![1, 2, 3];
        obj.part_etags = vec!["etag1".to_string(), "etag2".to_string(), "etag3".to_string()];
        obj.part_sizes = vec![1024, 2048, 512];
        obj.part_actual_sizes = vec![1024, 2048, 512];

        let data = obj.marshal_msg().unwrap();
        let mut obj2 = MetaObject::default();
        obj2.unmarshal_msg(&data).unwrap();

        assert_eq!(obj.part_numbers, obj2.part_numbers);
        assert_eq!(obj.part_etags, obj2.part_etags);
        assert_eq!(obj.part_sizes, obj2.part_sizes);
        assert_eq!(obj.part_actual_sizes, obj2.part_actual_sizes);
    }

    #[test]
    fn test_version_header_validation() {
        // Test version header validation
        let mut header = FileMetaVersionHeader {
            version_type: VersionType::Object,
            mod_time: Some(OffsetDateTime::now_utc()),
            ec_m: 2,
            ec_n: 1,
            ..Default::default()
        };
        assert!(header.is_valid());

        // Test invalid version types
        header.version_type = VersionType::Invalid;
        assert!(!header.is_valid());

        // Reset to a valid state
        header.version_type = VersionType::Object;
        assert!(header.is_valid());

        // Test invalid erasure coding parameters
        // When ec_m = 0, has_ec() returns false so parity parameters are skipped
        header.ec_m = 0;
        header.ec_n = 1;
        assert!(header.is_valid()); // Valid because erasure coding is disabled

        // Enable erasure coding with invalid parameters
        header.ec_m = 2;
        header.ec_n = 0;
        // When ec_n = 0, has_ec() returns false so parity parameters are skipped
        assert!(header.is_valid()); // This remains valid because has_ec() returns false

        // Reset to a valid state
        header.ec_n = 1;
        assert!(header.is_valid());
    }

    #[test]
    fn test_special_characters_in_metadata() {
        // Test special character handling in metadata
        let mut obj = MetaObject::default();

        // Test various special characters
        let special_cases = vec![
            ("empty", ""),
            ("unicode", "test🚀🎉"),
            ("newlines", "line1\nline2\nline3"),
            ("tabs", "col1\tcol2\tcol3"),
            ("quotes", "\"quoted\" and 'single'"),
            ("backslashes", "path\\to\\file"),
            ("mixed", "Mixed: Chinese, English, 123, !@#$%"),
        ];

        for (key, value) in special_cases {
            obj.meta_user.insert(key.to_string(), value.to_string());
        }

        // Verify serialization and deserialization
        let data = obj.marshal_msg().unwrap();
        let mut obj2 = MetaObject::default();
        obj2.unmarshal_msg(&data).unwrap();

        assert_eq!(obj.meta_user, obj2.meta_user);

        // Verify each special character is correctly saved
        for (key, expected_value) in [
            ("empty", ""),
            ("unicode", "test🚀🎉"),
            ("newlines", "line1\nline2\nline3"),
            ("tabs", "col1\tcol2\tcol3"),
            ("quotes", "\"quoted\" and 'single'"),
            ("backslashes", "path\\to\\file"),
            ("mixed", "Mixed: Chinese, English, 123, !@#$%"),
        ] {
            assert_eq!(obj2.meta_user.get(key), Some(&expected_value.to_string()));
        }
    }
}

#[tokio::test]
async fn test_read_xl_meta_no_data() {
    use tokio::fs::File;
    use tokio::io::AsyncWriteExt;

    let mut fm = FileMeta::new();

    let (m, n) = (3, 2);

    for i in 0..5 {
        let mut fi = FileInfo::new(i.to_string().as_str(), m, n);
        fi.mod_time = Some(OffsetDateTime::now_utc());

        fm.add_version(fi).unwrap();
    }

    let mut buff = fm.marshal_msg().unwrap();

    buff.resize(buff.len() + 100, 0);

    // Use tempfile to avoid conflicts with parallel tests or previous runs
    let dir = tempfile::tempdir().unwrap();
    let filepath = dir.path().join("test_xl.meta");

    let mut file = File::create(&filepath).await.unwrap();
    // Write string data
    file.write_all(&buff).await.unwrap();
    file.flush().await.unwrap();

    let mut f = File::open(&filepath).await.unwrap();

    let stat = f.metadata().await.unwrap();

    let data = read_xl_meta_no_data(&mut f, stat.len() as usize).await.unwrap();

    let mut newfm = FileMeta::default();
    newfm.unmarshal_msg(&data).unwrap();

    assert_eq!(fm, newfm)
}
