use std::{
    any::Any,
    borrow::Cow,
    collections::BTreeMap,
    fmt::Debug,
    fs::File,
    io::{BufReader, Read, Seek},
    path::Path,
    sync::Arc,
};

use bytes::{Buf, Bytes};
use shared_buffer::OwnedBuffer;

use crate::{compat::Volume, v1::WebC, Version};

/// A version-agnostic read-only WEBC container.
///
/// A `Container` provides a high-level interface for reading and manipulating
/// WEBC container files. It supports multiple versions of WEBC container
/// formats and abstracts the underlying differences between them.
#[derive(Debug, Clone)]
pub struct Container {
    imp: Arc<dyn AbstractWebc + Send + Sync>,
}

#[allow(clippy::result_large_err)]
impl Container {
    /// Load a [`Container`] from disk.
    ///
    /// Where possible, this will try to use a memory-mapped implementation
    /// to reduce memory usage.
    pub fn from_disk(path: impl AsRef<Path>) -> Result<Self, ContainerError> {
        let path = path.as_ref();

        #[cfg(feature = "package")]
        if path.is_dir() {
            let wasmer_toml = path.join("wasmer.toml");
            let pkg = crate::wasmer_package::Package::from_manifest(wasmer_toml)?;
            return Ok(Container::new(pkg));
        }

        let mut f = File::open(path).map_err(|error| ContainerError::Open {
            error,
            path: path.to_path_buf(),
        })?;

        if is_tarball(&mut f) {
            let pkg = crate::wasmer_package::Package::from_tarball(BufReader::new(f))
                .map_err(ContainerError::WasmerPackage)?;
            return Ok(Container::new(pkg));
        }

        match crate::detect(&mut f) {
            Ok(Version::V1) => {
                // We need to explicitly use WebcMmap to get a memory-mapped
                // parser
                let options = crate::v1::ParseOptions::default();
                let webc = crate::v1::WebCMmap::from_file(f, &options)?;
                Ok(Container::new(webc))
            }
            Ok(Version::V2) => {
                // Note: OwnedReader::from_file() will automatically try to
                // use a memory-mapped file when possible.
                let webc = crate::v2::read::OwnedReader::from_file(f)?;
                Ok(Container::new(webc))
            }
            Ok(other) => {
                // fall back to the allocating generic version
                let mut buffer = Vec::new();
                f.rewind()
                    .and_then(|_| f.read_to_end(&mut buffer))
                    .map_err(|error| ContainerError::Read {
                        path: path.to_path_buf(),
                        error,
                    })?;

                Container::from_bytes_and_version(buffer.into(), other)
            }
            Err(e) => Err(ContainerError::Detect(e)),
        }
    }

    /// Load a [`Container`] from bytes in memory.
    pub fn from_bytes(bytes: impl Into<Bytes>) -> Result<Self, ContainerError> {
        let bytes: Bytes = bytes.into();

        if is_tarball(std::io::Cursor::new(&bytes)) {
            let pkg = crate::wasmer_package::Package::from_tarball(bytes.reader())
                .map_err(ContainerError::WasmerPackage)?;
            return Ok(Container::new(pkg));
        }

        let version = crate::detect(bytes.as_ref())?;
        Container::from_bytes_and_version(bytes, version)
    }

    fn new(repr: impl AbstractWebc + Send + Sync + 'static) -> Self {
        Container {
            imp: Arc::new(repr),
        }
    }

    fn from_bytes_and_version(bytes: Bytes, version: Version) -> Result<Self, ContainerError> {
        match version {
            Version::V1 => {
                let options = crate::v1::ParseOptions::default();
                let webc = crate::v1::WebCOwned::parse(bytes, &options)?;
                Ok(Container::new(webc))
            }
            Version::V2 => {
                let reader = crate::v2::read::OwnedReader::parse(bytes)?;
                Ok(Container::new(reader))
            }
            other => Err(ContainerError::UnsupportedVersion(other)),
        }
    }

    /// Get the [`Container`]'s manifest.
    pub fn manifest(&self) -> &crate::metadata::Manifest {
        self.imp.manifest()
    }

    /// Get all atoms stored in the container as a map.
    pub fn atoms(&self) -> BTreeMap<String, OwnedBuffer> {
        let mut atoms = BTreeMap::new();

        for name in self.imp.atom_names() {
            if let Some(atom) = self.imp.get_atom(&name) {
                atoms.insert(name.into_owned(), atom);
            }
        }

        atoms
    }

    /// Get an atom with the given name.
    ///
    /// Returns `None` if the atom does not exist in the container.
    ///
    /// This operation is pretty cheap, typically just a dictionary lookup
    /// followed by reference count bump and some index math.
    pub fn get_atom(&self, name: &str) -> Option<OwnedBuffer> {
        self.imp.get_atom(name)
    }

    /// Get all volumes stored in the container.
    pub fn volumes(&self) -> BTreeMap<String, Volume> {
        let mut volumes = BTreeMap::new();

        for name in self.imp.volume_names() {
            if let Some(atom) = self.imp.get_volume(&name) {
                volumes.insert(name.into_owned(), atom);
            }
        }

        volumes
    }

    /// Get a volume with the given name.
    ///
    /// Returns `None` if the volume does not exist in the container.
    pub fn get_volume(&self, name: &str) -> Option<Volume> {
        self.imp.get_volume(name)
    }

    /// Downcast the [`Container`] a concrete implementation.
    pub fn downcast_ref<T>(&self) -> Option<&T>
    where
        T: 'static,
    {
        self.as_any().downcast_ref()
    }

    /// Downcast the [`Container`] a concrete implementation, returning the
    /// original [`Container`] if the cast fails.
    pub fn downcast<T>(self) -> Result<Arc<T>, Self>
    where
        T: 'static,
    {
        if self.as_any().is::<T>() {
            // Safety: We've just checked that the type matches up.
            unsafe { Ok(Arc::from_raw(Arc::into_raw(self.imp).cast())) }
        } else {
            Err(self)
        }
    }
}

trait AbstractWebc: AsAny + Debug {
    fn manifest(&self) -> &crate::metadata::Manifest;

    fn atom_names(&self) -> Vec<Cow<'_, str>>;
    fn get_atom(&self, name: &str) -> Option<OwnedBuffer>;
    fn volume_names(&self) -> Vec<Cow<'_, str>>;
    fn get_volume(&self, name: &str) -> Option<Volume>;
}

impl AbstractWebc for crate::v2::read::OwnedReader {
    fn manifest(&self) -> &crate::metadata::Manifest {
        self.manifest()
    }

    fn atom_names(&self) -> Vec<Cow<'_, str>> {
        self.atom_names().map(Cow::Borrowed).collect()
    }

    fn get_atom(&self, name: &str) -> Option<OwnedBuffer> {
        self.get_atom(name).cloned().map(OwnedBuffer::from)
    }

    fn volume_names(&self) -> Vec<Cow<'_, str>> {
        crate::v2::read::OwnedReader::volume_names(self)
            .map(Cow::Borrowed)
            .collect()
    }

    fn get_volume(&self, name: &str) -> Option<Volume> {
        self.get_volume(name).ok().map(Volume::new)
    }
}

impl From<crate::v2::read::OwnedReader> for Container {
    fn from(value: crate::v2::read::OwnedReader) -> Self {
        Container::new(value)
    }
}

impl AbstractWebc for crate::v1::WebCMmap {
    fn manifest(&self) -> &crate::metadata::Manifest {
        &self.manifest
    }

    fn atom_names(&self) -> Vec<Cow<'_, str>> {
        self.get_all_atoms().into_keys().map(Cow::Owned).collect()
    }

    fn get_atom(&self, name: &str) -> Option<OwnedBuffer> {
        let atoms = self.get_all_atoms();
        let atom = atoms.get(name)?;
        let range = crate::utils::subslice_offsets(&self.buffer, atom);
        Some(self.buffer.slice(range))
    }

    fn volume_names(&self) -> Vec<Cow<'_, str>> {
        self.volumes
            .keys()
            .map(|s| Cow::Borrowed(s.as_str()))
            .collect()
    }

    fn get_volume(&self, name: &str) -> Option<Volume> {
        let package_name = self.get_package_name();
        let volume = WebC::get_volume(self, &package_name, name)?;
        let buffer = self.buffer.clone();

        Some(Volume::new(crate::volume::VolumeV1 {
            volume: volume.clone(),
            buffer,
        }))
    }
}

impl From<crate::v1::WebCMmap> for Container {
    fn from(value: crate::v1::WebCMmap) -> Self {
        Container::new(value)
    }
}

impl AbstractWebc for crate::v1::WebCOwned {
    fn manifest(&self) -> &crate::metadata::Manifest {
        &self.manifest
    }

    fn atom_names(&self) -> Vec<Cow<'_, str>> {
        self.get_all_atoms().into_keys().map(Cow::Owned).collect()
    }

    fn get_atom(&self, name: &str) -> Option<OwnedBuffer> {
        let atoms = self.get_all_atoms();
        let atom = atoms.get(name)?;
        let range = crate::utils::subslice_offsets(&self.backing_data, atom);
        Some(self.backing_data.slice(range).into())
    }

    fn volume_names(&self) -> Vec<Cow<'_, str>> {
        self.volumes
            .keys()
            .map(|s| Cow::Borrowed(s.as_str()))
            .collect()
    }

    fn get_volume(&self, name: &str) -> Option<Volume> {
        let package_name = self.get_package_name();
        let volume = WebC::get_volume(self, &package_name, name)?.clone();
        let buffer = self.backing_data.clone().into();

        Some(Volume::new(crate::volume::VolumeV1 { buffer, volume }))
    }
}

impl From<crate::v1::WebCOwned> for Container {
    fn from(value: crate::v1::WebCOwned) -> Self {
        Container::new(value)
    }
}

impl AbstractWebc for crate::wasmer_package::Package {
    fn manifest(&self) -> &crate::metadata::Manifest {
        self.manifest()
    }

    fn atom_names(&self) -> Vec<Cow<'_, str>> {
        self.atoms()
            .keys()
            .map(|s| Cow::Borrowed(s.as_str()))
            .collect()
    }

    fn get_atom(&self, name: &str) -> Option<OwnedBuffer> {
        self.atoms().get(name).cloned()
    }

    fn volume_names(&self) -> Vec<Cow<'_, str>> {
        self.volume_names()
    }

    fn get_volume(&self, name: &str) -> Option<Volume> {
        self.get_volume(name).map(Volume::new)
    }
}

impl From<crate::wasmer_package::Package> for Container {
    fn from(value: crate::wasmer_package::Package) -> Self {
        Container::new(value)
    }
}

trait AsAny {
    fn as_any(&self) -> &(dyn Any + 'static);
}

impl<T> AsAny for T
where
    T: Any,
{
    fn as_any(&self) -> &(dyn Any + 'static) {
        self
    }
}

/// Various errors that may occur during [`Container`] operations.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum ContainerError {
    /// Unable to detect the WEBC version.
    #[error("Unable to detect the WEBC version")]
    Detect(#[from] crate::DetectError),
    /// An unsupported WEBC version was found.
    #[error("Unsupported WEBC version, {_0}")]
    UnsupportedVersion(crate::Version),
    /// An error occurred while parsing a v1 WEBC file.
    #[error(transparent)]
    V1(#[from] crate::v1::Error),
    /// An error occurred while parsing a v2 WEBC file.
    #[error(transparent)]
    V2Owned(#[from] crate::v2::read::OwnedReaderError),
    /// an error occurred while loading a Wasmer package from disk.
    #[error(transparent)]
    WasmerPackage(#[from] crate::wasmer_package::WasmerPackageError),
    /// Unable to open a file.
    #[error("Unable to open \"{}\"", path.display())]
    Open {
        /// The file's path.
        path: std::path::PathBuf,
        /// The underlying error.
        #[source]
        error: std::io::Error,
    },
    /// Unable to read a file's contents into memory.
    #[error("Unable to read \"{}\"", path.display())]
    Read {
        /// The file's path.
        path: std::path::PathBuf,
        /// The underlying error.
        #[source]
        error: std::io::Error,
    },
}

/// Check if something looks like a `*.tar.gz` file.
fn is_tarball(mut file: impl Read + Seek) -> bool {
    /// Magic bytes for a `*.tar.gz` file according to
    /// [Wikipedia](https://en.wikipedia.org/wiki/List_of_file_signatures).
    const TAR_GZ_MAGIC_BYTES: [u8; 2] = [0x1F, 0x8B];

    let mut buffer = [0_u8; 2];
    let result = match file.read_exact(&mut buffer) {
        Ok(_) => buffer == TAR_GZ_MAGIC_BYTES,
        Err(_) => false,
    };

    let _ = file.rewind();

    result
}