chunky-bits 0.2.0

use std::{
    convert::TryInto,
    path::Path,
};

use futures::stream::Stream;
use serde::{
    Deserialize,
    Serialize,
};
use tokio::{
    io,
    io::AsyncRead,
};

use crate::{
    cluster::{
        ClusterNodes,
        ClusterProfile,
        ClusterProfiles,
        Destination,
        DestinationInner,
        FileOrDirectory,
        MetadataFormat,
        MetadataTypes,
        Tunables,
    },
    error::{
        ClusterError,
        LocationParseError,
        MetadataReadError,
    },
    file::{
        new_profiler,
        FileReference,
        FileWriteBuilder,
        Location,
        ProfileReport,
        ProfileReporter,
    },
};

#[derive(Clone, Serialize, Deserialize)]
pub struct Cluster {
    #[serde(alias = "destination")]
    #[serde(alias = "nodes")]
    #[serde(alias = "node")]
    pub destinations: ClusterNodes,
    #[serde(alias = "metadata")]
    pub metadata: MetadataTypes,
    pub profiles: ClusterProfiles,
    #[serde(default)]
    #[serde(alias = "tunable")]
    #[serde(alias = "tuning")]
    pub tunables: Tunables,
}

impl Cluster {
    pub async fn from_location(
        location: impl TryInto<Location, Error = impl Into<LocationParseError>>,
    ) -> Result<Cluster, MetadataReadError> {
        MetadataFormat::Yaml.from_location(location).await
    }

    pub fn get_file_writer(&self, profile: &ClusterProfile) -> FileWriteBuilder<Destination> {
        let destination = self.get_destination(profile);
        FileReference::write_builder()
            .destination(destination)
            .chunk_size((1_usize) << profile.get_chunk_size())
            .data_chunks(profile.get_data_chunks())
    }

    pub async fn write_file_ref(
        &self,
        path: impl AsRef<Path>,
        file_ref: &FileReference,
    ) -> Result<(), ClusterError> {
        self.metadata.write(path, &file_ref).await?;
        Ok(())
    }

    pub async fn write_file<R>(
        &self,
        path: impl AsRef<Path>,
        reader: &mut R,
        profile: &ClusterProfile,
        content_type: Option<String>,
    ) -> Result<(), ClusterError>
    where
        R: AsyncRead + Unpin,
    {
        let mut file_ref = self.get_file_writer(profile).write(reader).await?;
        file_ref.content_type = content_type;
        self.metadata.write(path, &file_ref).await.unwrap();
        Ok(())
    }

    pub async fn write_file_with_report<R>(
        &self,
        path: impl AsRef<Path>,
        reader: &mut R,
        profile: &ClusterProfile,
        content_type: Option<String>,
    ) -> (ProfileReport, Result<(), ClusterError>)
    where
        R: AsyncRead + Unpin,
    {
        let (reporter, destination) = self.get_destination_with_profiler(profile);
        let result = FileReference::write_builder()
            .destination(destination)
            .chunk_size((1_usize) << profile.get_chunk_size())
            .data_chunks(profile.get_data_chunks())
            .parity_chunks(profile.get_parity_chunks())
            .write(reader)
            .await;
        match result {
            Ok(mut file_ref) => {
                file_ref.content_type = content_type;
                self.metadata.write(path, &file_ref).await.unwrap();
                (reporter.profile().await, Ok(()))
            },
            Err(err) => (reporter.profile().await, Err(err.into())),
        }
    }

    pub async fn get_file_ref(
        &self,
        path: impl AsRef<Path>,
    ) -> Result<FileReference, MetadataReadError> {
        self.metadata.read(path).await
    }

    pub async fn read_file(
        &self,
        path: impl AsRef<Path>,
    ) -> Result<impl AsyncRead + Unpin, MetadataReadError> {
        let file_ref = self.get_file_ref(path).await?;
        let reader = file_ref.read_builder_owned().reader_owned();
        Ok(reader)
    }

    pub fn get_destination(&self, profile: &ClusterProfile) -> Destination {
        let inner = DestinationInner {
            nodes: self.destinations.clone(),
            location_context: self.tunables.as_ref().clone(),
            profile: profile.clone(),
        };
        Destination(inner.into())
    }

    pub fn get_destination_with_profiler(
        &self,
        profile: &ClusterProfile,
    ) -> (ProfileReporter, Destination) {
        let (profiler, reporter) = new_profiler();
        let location_context = self
            .tunables
            .generate_location_context_builder()
            .profiler(profiler)
            .build();
        (
            reporter,
            Destination(
                DestinationInner {
                    nodes: self.destinations.clone(),
                    location_context,
                    profile: profile.clone(),
                }
                .into(),
            ),
        )
    }

    pub fn get_profile<'a>(
        &self,
        profile: impl Into<Option<&'a str>>,
    ) -> Option<&'_ ClusterProfile> {
        self.profiles.get(profile)
    }

    pub async fn list_files(
        &self,
        path: &Path,
    ) -> Result<impl Stream<Item = io::Result<FileOrDirectory>> + 'static, MetadataReadError> {
        self.metadata.list(path).await
    }
}