//
// Wildland Project
//
// Copyright © 2022 Golem Foundation
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU General Public License version 3 as published by
// the Free Software Foundation.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
// GNU General Public License for more details.
//
// You should have received a copy of the GNU General Public License
// along with this program.  If not, see <https://www.gnu.org/licenses/>.

mod path_resolver;

use std::collections::{HashMap, HashSet};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};

pub use path_resolver::*;
use thiserror::Error;
use uuid::Uuid;

use crate::catlib_service::entities::ContainerPath;
use crate::storage_manager::StorageManager;
use crate::{Container, Storage};

#[derive(Debug, Error, Clone, PartialEq, Eq)]
#[repr(C)]
pub enum ContainerManagerError {
    #[error("The given container has been already mounted")]
    AlreadyMounted,
    #[error("Generic mounting error: {0}")]
    MountingError(String),
    #[error("Container manifest is malformed due to: {0}")]
    MalformedManifest(String),
    #[error("The given container is not mounted")]
    ContainerNotMounted,
    #[error("The given container does not have any valid storages")]
    NoValidStorages,
}

/// Container Manager handle.
///
/// Container Manager by design is a singleton, so every handle should point to the same data.
#[derive(Clone)]
pub struct ContainerManager {
    containers_state: ContainerState,
    storage_manager: StorageManager,
}

impl ContainerManager {
    pub fn new(containers_state: ContainerState, storage_manager: StorageManager) -> Self {
        Self {
            containers_state,
            storage_manager,
        }
    }

    pub fn mount(&self, container: &Container) -> Result<(), ContainerManagerError> {
        let container_uuid = container.uuid();

        if let std::collections::hash_map::Entry::Vacant(e) = self
            .containers_state
            .mounted_containers
            .lock()
            .expect("Poisoned mutex!")
            .entry(container_uuid)
        {
            let container_paths = container
                .get_path()
                .map(PathBuf::from)
                .map_err(|e| ContainerManagerError::MountingError(format!("{e}")))?;

            let mountable_storage = self.get_first_available_storage(container)?;

            self.storage_manager
                .mount(&mountable_storage)
                .map_err(|e| ContainerManagerError::MountingError(format!("{e}")))?;

            e.insert((container.clone(), container_paths));
            Ok(())
        } else {
            Err(ContainerManagerError::AlreadyMounted)
        }
    }

    pub fn unmount(&self, container: &Container) -> Result<(), ContainerManagerError> {
        let container_uuid = container.uuid();
        self.containers_state
            .mounted_containers
            .lock()
            .expect("Poisoned mutex!")
            .remove(&container_uuid)
            .ok_or(ContainerManagerError::ContainerNotMounted)
            .map(|_| ())
    }

    pub fn is_mounted(&self, container: &Container) -> bool {
        let container_uuid = container.uuid();
        self.containers_state
            .mounted_containers
            .lock()
            .expect("Poisoned mutex!")
            .contains_key(&container_uuid)
    }

    fn get_first_available_storage(
        &self,
        container: &Container,
    ) -> Result<Storage, ContainerManagerError> {
        let container_storages = container.get_storages().map_err(|e| {
            ContainerManagerError::MalformedManifest(format!(
                "not being able to parse container storage. {e}"
            ))
        })?;

        if container_storages.is_empty() {
            return Err(ContainerManagerError::MalformedManifest(
                "Not having any storage entity".into(),
            ));
        }

        // TODO: Later on, once we support encryption and/or ACLs
        // there'll be more sophisticated methods to determine which storage
        // should be mounted rather than just taking the first one from the list
        match container_storages.first() {
            Some(storage) => Ok(storage.clone()),
            None => Err(ContainerManagerError::NoValidStorages),
        }
    }
}

#[derive(Default, Clone, Debug)]
pub struct ContainerState {
    mounted_containers: Arc<Mutex<HashMap<Uuid, (Container, ContainerPath)>>>,
}

impl PathResolver for ContainerState {
    fn resolve(&self, input_path: &Path) -> Result<HashSet<ResolvedPath>, PathResolutionError> {
        let mut physical_paths = Vec::new();
        let mut virtual_paths = HashSet::new();

        for (container, path) in self
            .mounted_containers
            .lock()
            .expect("Poisoned mutex!")
            .values()
        {
            if path.starts_with(input_path) && path != input_path {
                virtual_paths.insert(ResolvedPath::VirtualPath(path.clone()));
                continue;
            }

            let stripped_path = match input_path.strip_prefix(path) {
                Ok(stripped_path) => stripped_path,
                Err(_) => continue,
            };

            // only first path, which is considered as a primary, is exposed in a filesystem
            let mut path_within_storage = PathBuf::from("/");
            path_within_storage.push(stripped_path);

            let storage = container
                .get_storages()?
                .into_iter()
                .next()
                .ok_or_else(|| {
                    PathResolutionError::Generic(format!(
                        "No storages found for container {}",
                        container.uuid()
                    ))
                })?;

            physical_paths.push(ResolvedPath::PathWithStorages {
                path_within_storage,
                storage,
            });
        }

        Ok(physical_paths.into_iter().chain(virtual_paths).collect())
    }

    // Returns a claim of the container basing on the list of mounted containers.
    fn get_mount_path(&self, storage_id: Uuid) -> Result<Option<PathBuf>, PathResolutionError> {
        for (container, _) in self
            .mounted_containers
            .lock()
            .expect("Poisoned mutex!")
            .values()
        {
            if container
                .get_storages()?
                .iter()
                .any(|storage| storage.uuid() == storage_id)
            {
                let path = container.clone().get_path()?;
                return Ok(Some(PathBuf::from(path)));
            }
        }
        Ok(None)
    }
}

#[cfg(test)]
mod tests {
    use std::str::FromStr;
    use std::sync::Arc;

    use serde_json::json;
    use uuid::Uuid;

    use crate::catlib_service::entities::MockContainerManifest;
    use crate::dfs::interface::MockDfsFrontend;
    use crate::storage_manager::StorageManager;
    use crate::{Container, ContainerManager, ContainerManagerError, ContainerState, Storage};

    fn dummy_storage() -> Storage {
        Storage::new(
            Uuid::new_v4(),
            None,
            "test backend type".into(),
            json!({}),
            false,
            None,
        )
    }

    fn container_manager() -> ContainerManager {
        let mut dfs_mock = MockDfsFrontend::new();
        dfs_mock.expect_mount().returning(|_| Ok(()));
        let storage_manager = StorageManager::new(Arc::new(dfs_mock));

        ContainerManager::new(ContainerState::default(), storage_manager)
    }

    mod path_resolver_tests {
        use std::collections::HashSet;
        use std::path::PathBuf;
        use std::sync::Arc;

        use pretty_assertions::assert_eq;
        use test_case::test_case;
        use uuid::Uuid;

        use super::dummy_storage;
        use crate::catlib_service::entities::{ContainerPath, MockContainerManifest};
        use crate::{Container, PathResolver, ResolvedPath, Storage};

        fn create_container_with_storage(
            id: u128,
            claimed_path: ContainerPath,
        ) -> (Container, Storage) {
            let storage = dummy_storage();

            let mut container = MockContainerManifest::new();
            container
                .expect_uuid()
                .returning(move || Uuid::from_u128(id));
            container
                .expect_get_path()
                .returning(move || Ok(claimed_path.clone()));
            container.expect_get_storages().returning({
                let storage = storage.clone();
                move || Ok(vec![(&storage).into()])
            });

            let container = Container::new(Arc::new(container));

            (container, storage)
        }

        #[test_case("/".into(), PathBuf::from("/"), Some(PathBuf::from("/")), None; "claimed root")]
        #[test_case("/path".into(), PathBuf::from("/path"), Some(PathBuf::from("/")), None; "resolved root within storage")]
        #[test_case("/path/".into(), PathBuf::from("/path"), Some(PathBuf::from("/")), None; "slash at the end of claimed path")]
        #[test_case("/path".into(), PathBuf::from("/path/"), Some(PathBuf::from("/")), None; "slash at the end of input")]
        #[test_case("/books".into(), PathBuf::from("/books/fantasy"), Some(PathBuf::from("/fantasy")), None; "one nested component")]
        #[test_case("/books/fantasy".into(), PathBuf::from("/books/fantasy/a/lot/of/other/dirs"), Some(PathBuf::from("/a/lot/of/other/dirs")), None; "many nested components")]
        #[test_case("/books/fantasy".into(), PathBuf::from("/books/"), None, Some(PathBuf::from("/books/fantasy")); "virtual path extended with one component")]
        #[test_case("/books/fantasy".into(), PathBuf::from("/books"), None, Some(PathBuf::from("/books/fantasy")); "virtual path extended with one component (no slash at input's end)")]
        #[test_case("/books/fantasy/lord/of/the/rings".into(), PathBuf::from("/books/fantasy"), None, Some(PathBuf::from("/books/fantasy/lord/of/the/rings/")); "virtual path extended with many component")]
        fn test_resolve_with_one_container(
            claimed_path: ContainerPath,
            resolve_arg_path: PathBuf,
            expected_path_within_storage: Option<PathBuf>,
            expected_virtual_path: Option<PathBuf>,
        ) {
            let container_manager = super::container_manager();

            let (container1, storage) = create_container_with_storage(1, claimed_path);

            container_manager.mount(&container1).unwrap();

            let resolved_paths = container_manager
                .containers_state
                .resolve(&resolve_arg_path)
                .unwrap();
            let expected: HashSet<ResolvedPath> = [
                expected_path_within_storage.map(|path_within_storage| {
                    ResolvedPath::PathWithStorages {
                        path_within_storage,
                        storage,
                    }
                }),
                expected_virtual_path.map(ResolvedPath::VirtualPath),
            ]
            .into_iter()
            .flatten()
            .collect();
            assert_eq!(expected, resolved_paths);
        }

        #[test_case(PathBuf::from("/"), "/".into(), "/".into(), vec![PathBuf::from("/")], vec![PathBuf::from("/")], vec![]; "both claiming root")]
        #[test_case(PathBuf::from("/a/b/c"), "/a".into(), "/a/b".into(), vec![PathBuf::from("/b/c")], vec![PathBuf::from("/c")], vec![]; "physical paths from two containers")]
        #[test_case(PathBuf::from("/a/b/c"), "/a/b/c/d".into(), "/a/b".into(), vec![], vec![PathBuf::from("/c")], vec![PathBuf::from("/a/b/c/d")]; "physical and virtual paths from two containers")]
        fn test_resolve_with_two_containers(
            resolve_arg_path: PathBuf,
            claimed_path_1: ContainerPath,
            claimed_path_2: ContainerPath,
            expected_paths_within_storage_1: Vec<PathBuf>,
            expected_paths_within_storage_2: Vec<PathBuf>,
            expected_virtual_paths: Vec<PathBuf>,
        ) {
            let container_manager = super::container_manager();

            let (container_1, storage_1) = create_container_with_storage(1, claimed_path_1);
            container_manager.mount(&container_1).unwrap();

            let (container_2, storage_2) = create_container_with_storage(2, claimed_path_2);
            container_manager.mount(&container_2).unwrap();

            let resolved_paths = container_manager
                .containers_state
                .resolve(&resolve_arg_path)
                .unwrap();
            let expected: HashSet<ResolvedPath> = expected_paths_within_storage_1
                .into_iter()
                .map(|path_within_storage| ResolvedPath::PathWithStorages {
                    path_within_storage,
                    storage: storage_1.clone(),
                })
                .chain(
                    expected_paths_within_storage_2
                        .into_iter()
                        .map(|path_within_storage| ResolvedPath::PathWithStorages {
                            path_within_storage,
                            storage: storage_2.clone(),
                        }),
                )
                .chain(
                    expected_virtual_paths
                        .into_iter()
                        .map(ResolvedPath::VirtualPath),
                )
                .collect();
            assert_eq!(expected, resolved_paths.into_iter().collect());
        }
    }

    #[test]
    fn mount_container() {
        let container_manager = container_manager();
        let mut container1 = MockContainerManifest::new();

        container1
            .expect_get_path()
            .returning(|| Ok("/some/path1".into()));
        container1
            .expect_uuid()
            .returning(|| Uuid::from_str("00000000-0000-0000-0000-000000000001").unwrap());
        container1
            .expect_get_storages()
            .returning(|| Ok(vec![(&dummy_storage()).into()]));
        let container1 = Arc::new(container1) as _;
        container_manager
            .mount(&Container::new(container1))
            .unwrap();
    }

    #[test]
    fn mount_mounted_container_returns_error() {
        let container_manager = container_manager();
        let mut container1 = MockContainerManifest::new();
        container1
            .expect_get_path()
            .returning(|| Ok("/some/path1".into()));
        container1
            .expect_uuid()
            .returning(|| Uuid::from_str("00000000-0000-0000-0000-000000000001").unwrap());
        container1
            .expect_get_storages()
            .returning(|| Ok(vec![(&dummy_storage()).into()]));
        let container1 = Arc::new(container1);
        container_manager
            .mount(&Container::new(container1.clone()))
            .unwrap();
        assert!(matches!(
            container_manager.mount(&Container::new(container1)),
            Err(ContainerManagerError::AlreadyMounted)
        ));
    }

    #[test]
    fn unmount_container() {
        let container_manager = container_manager();
        let mut container1 = MockContainerManifest::new();
        container1
            .expect_get_path()
            .returning(|| Ok("/some/path1".into()));
        container1
            .expect_uuid()
            .returning(|| Uuid::from_str("00000000-0000-0000-0000-000000000001").unwrap());
        container1
            .expect_get_storages()
            .returning(|| Ok(vec![(&dummy_storage()).into()]));
        let container1 = Arc::new(container1);
        container_manager
            .mount(&Container::new(container1.clone()))
            .unwrap();
        container_manager
            .unmount(&Container::new(container1))
            .unwrap();
    }

    #[test]
    fn unmount_not_mounted_container_returns_error() {
        let container_manager = container_manager();
        let mut container1 = MockContainerManifest::new();
        container1
            .expect_get_path()
            .returning(|| Ok("/some/path1".into()));
        container1
            .expect_uuid()
            .returning(|| Uuid::from_str("00000000-0000-0000-0000-000000000001").unwrap());
        container1
            .expect_get_storages()
            .returning(|| Ok(vec![(&dummy_storage()).into()]));
        let container1 = Arc::new(container1);
        assert!(matches!(
            container_manager.unmount(&Container::new(container1)),
            Err(ContainerManagerError::ContainerNotMounted)
        ));
    }
}