//! Virtual Object Layer (VOL) abstraction for pluggable HDF5 backends.
//!
//! The VOL trait provides a backend-agnostic interface for HDF5 file I/O,
//! enabling seamless integration with different storage systems such as
//! local filesystems, S3/object stores, DAOS, and HSDS.
//!
//! # Architecture
//!
//! The VOL separates the HDF5 API from the underlying storage:
//!
//! ```text
//! ┌───────────────────────────┐
//! │ clawhdf5 High-level API  │
//! ├───────────────────────────┤
//! │   VirtualObjectLayer      │  ← trait defined here
//! ├───────┬───────┬───────────┤
//! │ Local │  S3   │   DAOS    │  ← pluggable backends
//! └───────┴───────┴───────────┘
//! ```
//!
//! # Example
//!
//! ```ignore
//! use clawhdf5_io::vol::{VirtualObjectLayer, NativeVol};
//!
//! let vol = NativeVol::open("data.h5").unwrap();
//! let data = vol.read_dataset("sensors/temperature").unwrap();
//! ```

use std::collections::HashMap;
use std::fmt;
use std::io;

/// Capabilities that a VOL connector supports.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum VolCapability {
    /// Can read existing datasets.
    ReadData,
    /// Can write new datasets.
    WriteData,
    /// Can list groups and datasets.
    ListObjects,
    /// Can read/write attributes.
    Attributes,
    /// Supports chunked storage.
    ChunkedStorage,
    /// Supports parallel I/O.
    ParallelIO,
    /// Supports partial/selection reads.
    SelectionRead,
    /// Supports SWMR mode.
    Swmr,
}

/// Error type for VOL operations.
#[derive(Debug)]
pub enum VolError {
    /// I/O error from the underlying storage.
    Io(io::Error),
    /// The requested operation is not supported by this VOL connector.
    Unsupported(String),
    /// Object (dataset, group, attribute) not found.
    NotFound(String),
    /// Data format or parsing error.
    DataError(String),
}

impl fmt::Display for VolError {
    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
        match self {
            VolError::Io(e) => write!(f, "VOL I/O error: {e}"),
            VolError::Unsupported(op) => write!(f, "VOL operation not supported: {op}"),
            VolError::NotFound(name) => write!(f, "VOL object not found: {name}"),
            VolError::DataError(msg) => write!(f, "VOL data error: {msg}"),
        }
    }
}

impl std::error::Error for VolError {
    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
        match self {
            VolError::Io(e) => Some(e),
            _ => None,
        }
    }
}

impl From<io::Error> for VolError {
    fn from(e: io::Error) -> Self {
        VolError::Io(e)
    }
}

/// Information about a dataset within a VOL connector.
#[derive(Debug, Clone)]
pub struct VolDatasetInfo {
    /// Dataset path within the file.
    pub path: String,
    /// Shape (dimensions).
    pub shape: Vec<u64>,
    /// Element type description (e.g., "f64", "i32", "compound{...}").
    pub dtype: String,
    /// Total data size in bytes.
    pub size_bytes: u64,
}

/// Information about a group within a VOL connector.
#[derive(Debug, Clone)]
pub struct VolGroupInfo {
    /// Group path.
    pub path: String,
    /// Number of child datasets.
    pub num_datasets: usize,
    /// Number of child groups.
    pub num_groups: usize,
}

/// Virtual Object Layer trait for pluggable storage backends.
///
/// Implement this trait to add support for a new storage backend.
/// Each method has a default implementation that returns `Unsupported`,
/// so connectors only need to implement the methods they support.
pub trait VirtualObjectLayer: Send + Sync {
    /// Returns the connector name (e.g., "native", "s3", "daos").
    fn name(&self) -> &str;

    /// Returns the set of capabilities this connector supports.
    fn capabilities(&self) -> Vec<VolCapability>;

    /// Check if a specific capability is supported.
    fn supports(&self, cap: VolCapability) -> bool {
        self.capabilities().contains(&cap)
    }

    /// Open a file/container at the given location.
    ///
    /// The `location` interpretation depends on the backend:
    /// - Native: filesystem path
    /// - S3: `s3://bucket/key`
    /// - HSDS: `http://host:port/path`
    fn open(&mut self, location: &str) -> Result<(), VolError>;

    /// Close the current file/container.
    fn close(&mut self) -> Result<(), VolError> {
        Ok(())
    }

    /// Read raw bytes for a dataset at the given path.
    fn read_dataset(&self, _path: &str) -> Result<Vec<u8>, VolError> {
        Err(VolError::Unsupported(format!(
            "{}: read_dataset not supported",
            self.name()
        )))
    }

    /// Get information about a dataset.
    fn dataset_info(&self, _path: &str) -> Result<VolDatasetInfo, VolError> {
        Err(VolError::Unsupported(format!(
            "{}: dataset_info not supported",
            self.name()
        )))
    }

    /// List datasets in a group.
    fn list_datasets(&self, _group_path: &str) -> Result<Vec<String>, VolError> {
        Err(VolError::Unsupported(format!(
            "{}: list_datasets not supported",
            self.name()
        )))
    }

    /// List subgroups in a group.
    fn list_groups(&self, _group_path: &str) -> Result<Vec<String>, VolError> {
        Err(VolError::Unsupported(format!(
            "{}: list_groups not supported",
            self.name()
        )))
    }

    /// Read attributes for an object (dataset or group).
    fn read_attributes(&self, _path: &str) -> Result<HashMap<String, Vec<u8>>, VolError> {
        Err(VolError::Unsupported(format!(
            "{}: read_attributes not supported",
            self.name()
        )))
    }

    /// Write a dataset at the given path.
    fn write_dataset(
        &mut self,
        _path: &str,
        _data: &[u8],
        _shape: &[u64],
        _dtype: &str,
    ) -> Result<(), VolError> {
        Err(VolError::Unsupported(format!(
            "{}: write_dataset not supported",
            self.name()
        )))
    }
}

/// Native (local filesystem) VOL connector.
///
/// Reads and writes HDF5 files on the local filesystem using the
/// clawhdf5-format library for parsing.
#[derive(Debug)]
pub struct NativeVol {
    data: Option<Vec<u8>>,
    location: Option<String>,
    /// Why the bytes given to [`NativeVol::from_bytes`] cannot be read
    /// (what `open` would have refused them for).
    load_error: Option<String>,
}

/// The HDF5 bytes of a whole file and its superblock: from the superblock
/// (addresses are relative to it, so any user block is skipped) to the end
/// of file the superblock records. A file shorter than that is truncated
/// and refused, and nothing past it is read, as in libhdf5.
pub(crate) fn hdf5_view(
    whole: &[u8],
) -> Result<(&[u8], clawhdf5_format::superblock::Superblock), VolError> {
    use clawhdf5_format::{signature::split_user_block, superblock::Superblock};
    let err = |e: clawhdf5_format::error::FormatError| VolError::DataError(e.to_string());
    let (user_block, data) = split_user_block(whole).map_err(err)?;
    let sb = Superblock::parse(data, 0).map_err(err)?;
    let end = sb
        .data_end(user_block.len() as u64, whole.len() as u64)
        .map_err(err)?;
    // data_end is at most the file length less the user block.
    Ok((&data[..end as usize], sb))
}

/// Check a whole file as libhdf5 does when it opens it ([`hdf5_view`], and
/// the superblock extension), and write any metadata cache image over the
/// file's metadata in place: libhdf5 reads the image's entries instead of
/// the file's own bytes at their addresses, which may be stale
/// (`clawhdf5_format::superblock_ext`). A file whose image libhdf5 cannot
/// load is refused: a connector that reads whole datasets has no way to
/// open the file and fail each object instead.
pub(crate) fn load_hdf5(whole: &mut [u8]) -> Result<(), VolError> {
    use clawhdf5_format::superblock_ext::apply_cache_image_in_place;
    let err = |e: clawhdf5_format::error::FormatError| VolError::DataError(e.to_string());
    let (len, sb) = {
        let (data, sb) = hdf5_view(whole)?;
        (data.len(), sb)
    };
    // hdf5_view's bytes start at the superblock, after any user block.
    let base = clawhdf5_format::signature::split_user_block(whole)
        .map_err(err)?
        .0
        .len();
    apply_cache_image_in_place(&mut whole[base..base + len], &sb).map_err(err)
}

impl NativeVol {
    /// Create a new native VOL connector.
    pub fn new() -> Self {
        Self {
            data: None,
            location: None,
            load_error: None,
        }
    }

    /// Create and open a native VOL connector from a file path.
    pub fn open_path(path: &str) -> Result<Self, VolError> {
        let mut vol = Self::new();
        vol.open(path)?;
        Ok(vol)
    }

    /// Create a native VOL connector from bytes already in memory.
    pub fn from_bytes(mut data: Vec<u8>) -> Self {
        let load_error = load_hdf5(&mut data).err().map(|e| e.to_string());
        Self {
            data: Some(data),
            location: Some("<memory>".into()),
            load_error,
        }
    }

    /// Access the raw file bytes, if loaded.
    pub fn as_bytes(&self) -> Option<&[u8]> {
        self.data.as_deref()
    }
}

impl Default for NativeVol {
    fn default() -> Self {
        Self::new()
    }
}

impl VirtualObjectLayer for NativeVol {
    fn name(&self) -> &str {
        "native"
    }

    fn capabilities(&self) -> Vec<VolCapability> {
        vec![
            VolCapability::ReadData,
            VolCapability::WriteData,
            VolCapability::ListObjects,
            VolCapability::Attributes,
            VolCapability::ChunkedStorage,
            VolCapability::SelectionRead,
        ]
    }

    fn open(&mut self, location: &str) -> Result<(), VolError> {
        let mut data = std::fs::read(location)?;
        // Refuse a truncated file at open, as libhdf5 does, and load any
        // metadata cache image.
        load_hdf5(&mut data)?;
        self.data = Some(data);
        self.load_error = None;
        self.location = Some(location.to_string());
        Ok(())
    }

    fn close(&mut self) -> Result<(), VolError> {
        self.data = None;
        self.location = None;
        Ok(())
    }

    fn read_dataset(&self, path: &str) -> Result<Vec<u8>, VolError> {
        let data = self.data.as_ref().ok_or_else(|| {
            VolError::Io(io::Error::new(io::ErrorKind::NotConnected, "file not open"))
        })?;
        if let Some(e) = &self.load_error {
            return Err(VolError::DataError(e.clone()));
        }

        use clawhdf5_format::{
            data_layout::DataLayout, data_read::read_raw_data_full, dataspace::Dataspace,
            datatype::Datatype, filter_pipeline::FilterPipeline, group_v2::resolve_path_any,
            message_type::MessageType, object_header::ObjectHeader,
        };

        let (data, sb) = hdf5_view(data)?;
        let addr = resolve_path_any(data, &sb, path)
            .map_err(|e| VolError::NotFound(format!("{path}: {e}")))?;

        let header = ObjectHeader::parse(data, addr as usize, sb.offset_size, sb.length_size)
            .map_err(|e| VolError::DataError(e.to_string()))?;

        let dt_msg = header
            .messages
            .iter()
            .find(|m| m.msg_type == MessageType::Datatype)
            .ok_or_else(|| VolError::DataError("missing datatype".into()))?;
        let (datatype, _) = Datatype::parse_in_header(&dt_msg.data, header.version)
            .map_err(|e| VolError::DataError(e.to_string()))?;

        let ds_msg = header
            .messages
            .iter()
            .find(|m| m.msg_type == MessageType::Dataspace)
            .ok_or_else(|| VolError::DataError("missing dataspace".into()))?;
        let dataspace = Dataspace::parse(&ds_msg.data, sb.length_size)
            .map_err(|e| VolError::DataError(e.to_string()))?;

        let dl_msg = header
            .messages
            .iter()
            .find(|m| m.msg_type == MessageType::DataLayout)
            .ok_or_else(|| VolError::DataError("missing data layout".into()))?;
        let layout = DataLayout::parse(&dl_msg.data, sb.offset_size, sb.length_size)
            .map_err(|e| VolError::DataError(e.to_string()))?;

        let pipeline = header
            .messages
            .iter()
            .find(|m| m.msg_type == MessageType::FilterPipeline)
            .and_then(|msg| FilterPipeline::parse(&msg.data).ok());

        read_raw_data_full(
            data,
            &layout,
            &dataspace,
            &datatype,
            pipeline.as_ref(),
            sb.offset_size,
            sb.length_size,
        )
        .map_err(|e| VolError::DataError(e.to_string()))
    }
}

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

    #[test]
    fn native_vol_new() {
        let vol = NativeVol::new();
        assert_eq!(vol.name(), "native");
        assert!(vol.supports(VolCapability::ReadData));
        assert!(!vol.supports(VolCapability::Swmr));
    }

    #[test]
    fn native_vol_from_bytes() {
        use clawhdf5_format::file_writer::FileWriter as FmtWriter;

        let mut fw = FmtWriter::new();
        fw.create_dataset("test").with_f64_data(&[1.0, 2.0, 3.0]);
        let bytes = fw.finish().unwrap();

        let vol = NativeVol::from_bytes(bytes);
        assert!(vol.as_bytes().is_some());

        let raw = vol.read_dataset("test").unwrap();
        assert_eq!(raw.len(), 3 * 8); // 3 f64 values
    }

    #[test]
    fn native_vol_not_open_error() {
        let vol = NativeVol::new();
        let result = vol.read_dataset("test");
        assert!(result.is_err());
    }

    #[test]
    fn native_vol_close() {
        use clawhdf5_format::file_writer::FileWriter as FmtWriter;

        let mut fw = FmtWriter::new();
        fw.create_dataset("x").with_f64_data(&[1.0]);
        let bytes = fw.finish().unwrap();

        let mut vol = NativeVol::from_bytes(bytes);
        assert!(vol.as_bytes().is_some());
        vol.close().unwrap();
        assert!(vol.as_bytes().is_none());
    }

    /// As libhdf5 does: a file shorter than the end of file its superblock
    /// records is truncated and refused (at open, and when read from
    /// memory), and bytes appended past that end are not part of the file.
    #[test]
    fn native_vol_refuses_truncated_files_and_ignores_trailing_bytes() {
        use clawhdf5_format::file_writer::FileWriter as FmtWriter;

        let mut fw = FmtWriter::new();
        fw.create_dataset("x").with_f64_data(&[1.0, 2.0, 3.0]);
        let bytes = fw.finish().unwrap();

        let truncated = bytes[..bytes.len() - 8].to_vec();
        let err = NativeVol::from_bytes(truncated.clone())
            .read_dataset("x")
            .unwrap_err();
        assert!(err.to_string().contains("truncated"), "{err}");
        let dir = std::env::temp_dir().join(format!("clawhdf5_vol_trunc_{}", std::process::id()));
        std::fs::create_dir_all(&dir).unwrap();
        let path = dir.join("truncated.h5");
        std::fs::write(&path, &truncated).unwrap();
        let err = NativeVol::open_path(path.to_str().unwrap()).err().unwrap();
        assert!(err.to_string().contains("truncated"), "{err}");
        std::fs::remove_dir_all(&dir).ok();

        let mut appended = bytes.clone();
        appended.extend_from_slice(&[0xAB; 64]);
        let raw = NativeVol::from_bytes(appended).read_dataset("x").unwrap();
        assert_eq!(raw.len(), 24);
    }

    /// libhdf5's `h5clear_mdc_image.h5` (see the clawhdf5 crate's
    /// `tests/metadata_cache_image.rs`): the root group's header exists
    /// only in the file's metadata cache image, so reading the file's own
    /// bytes finds zeros there.
    #[test]
    fn native_vol_reads_through_a_metadata_cache_image() {
        let path = concat!(
            env!("CARGO_MANIFEST_DIR"),
            "/../clawhdf5/tests/fixtures/h5clear_mdc_image.h5"
        );
        let expected: Vec<u8> = (0..50)
            .flat_map(|i| (0..100).map(move |j| i * j))
            .flat_map(i32::to_le_bytes)
            .collect();
        let vol = NativeVol::open_path(path).unwrap();
        assert_eq!(vol.read_dataset("DSET").unwrap(), expected);
        let bytes = std::fs::read(path).unwrap();
        let vol = NativeVol::from_bytes(bytes.clone());
        assert_eq!(vol.read_dataset("DSET").unwrap(), expected);

        // An image libhdf5 cannot load is refused, not read around.
        let mut bad = bytes;
        let at = bad.windows(4).position(|w| w == b"MDCI").unwrap();
        bad[at] = b'X';
        let err = NativeVol::from_bytes(bad.clone())
            .read_dataset("DSET")
            .unwrap_err();
        assert!(err.to_string().contains("cache image"), "{err}");
        let dir = tempfile::tempdir().unwrap();
        let bad_path = dir.path().join("bad_image.h5");
        std::fs::write(&bad_path, &bad).unwrap();
        let err = NativeVol::open_path(bad_path.to_str().unwrap())
            .err()
            .unwrap();
        assert!(err.to_string().contains("cache image"), "{err}");
    }

    #[test]
    fn vol_error_display() {
        let err = VolError::Unsupported("read_dataset".into());
        assert!(err.to_string().contains("not supported"));

        let err = VolError::NotFound("missing".into());
        assert!(err.to_string().contains("not found"));
    }

    #[test]
    fn vol_capabilities() {
        let vol = NativeVol::new();
        let caps = vol.capabilities();
        assert!(caps.contains(&VolCapability::ReadData));
        assert!(caps.contains(&VolCapability::WriteData));
        assert!(caps.contains(&VolCapability::ListObjects));
    }
}