/*
This file is part of Yama.

Yama is free software: you can redistribute it and/or modify
it under the terms of the GNU General Public License as published by
the Free Software Foundation, either version 3 of the License, or
(at your option) any later version.

Yama 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 Yama.  If not, see <https://www.gnu.org/licenses/>.
*/


use std::fs::File;
use std::io;
use std::io::{Read, Write};
use std::path::Path;
use std::sync::Arc;

use anyhow::{anyhow, bail, Context};
use clap::crate_version;
use log::warn;

use crate::chunking::{RecursiveChunker, RecursiveUnchunker, SENSIBLE_THRESHOLD};
use crate::definitions::{PointerData, RecursiveChunkRef, RootTreeNode, TreeNode};
use crate::pile::compression::{CompressionSettings, RawPileCompressor};
use crate::pile::integrity::RawPileIntegrityChecker;
use crate::pile::local_sqlitebloblogs::SqliteBloblogPile;
use crate::pile::{Pile, PileDescriptor, PileStorage, RawPile};
use crate::tree::{integrate_node_in_place, merge_uid_or_gid_tables};
use crate::utils::get_number_of_workers;

pub fn init(dir: &Path) -> anyhow::Result<()> {
    let yama_toml = dir.join("yama.toml");
    if yama_toml.exists() {
        bail!("yama.toml already exists. Cannot create yama pile here.");
    }

    /*
    let pile_db = sled::open(dir.join("pile.sled"))?;
    pile_db.flush()?;
     */

    let mut file = File::create(yama_toml)?;

    let desc = PileDescriptor {
        yama_version: crate_version!().to_owned(),
        storage: PileStorage::SqliteIndexedBloblog,
        compression: Some(12),
    };

    file.write_all(&toml::to_vec(&desc)?)?;

    Ok(())
}

pub fn load_pile_descriptor(dir: &Path) -> anyhow::Result<PileDescriptor> {
    let yama_toml = dir.join("yama.toml");
    if !yama_toml.exists() {
        bail!("yama.toml does not exist here. Is this an existing pile?");
    }

    let mut file = File::open(yama_toml)?;
    let mut buf = Vec::new();
    file.read_to_end(&mut buf)?;
    Ok(toml::from_slice(&buf)?)
}

pub fn open_pile(dir: &Path, desc: &PileDescriptor) -> anyhow::Result<Pile<Box<dyn RawPile>>> {
    let num_compressors = get_number_of_workers("YAMA_COMPRESSORS");
    let num_decompressors = get_number_of_workers("YAMA_DECOMPRESSORS");

    match desc.storage {
        PileStorage::RemoteOnly => {
            bail!("This is a remote-only pile. No local storage allowed.");
        }
        PileStorage::SqliteIndexedBloblog => {
            let blob_raw_pile = RawPileIntegrityChecker::new(SqliteBloblogPile::open(dir)?);
            let raw_pile: Box<dyn RawPile> = match desc.compression {
                None => Box::new(blob_raw_pile),
                Some(comp_level) => {
                    let mut dictionary = Vec::new();
                    let dict_path = dir.join("important_zstd.dict");
                    File::open(dict_path)
                        .context("You need important_zstd.dict in your pile folder.")?
                        .read_to_end(&mut dictionary)?;

                    let (compressed_pile, _handles) = RawPileCompressor::new(
                        blob_raw_pile,
                        CompressionSettings {
                            dictionary: Arc::new(dictionary),
                            level: comp_level as i32,
                            num_compressors: num_compressors as u32,
                            num_decompressors: num_decompressors as u32,
                        },
                    )?;

                    Box::new(compressed_pile)
                }
            };
            Ok(Pile::new(raw_pile))
        }
    }
}

pub fn store_tree_node<RP: RawPile>(
    pile: &Pile<RP>,
    root_tree_node: &RootTreeNode,
) -> anyhow::Result<RecursiveChunkRef> {
    let serialised = serde_bare::to_vec(root_tree_node)?;
    let mut chunker = RecursiveChunker::new(SENSIBLE_THRESHOLD, pile);
    io::copy(&mut (&serialised[..]), &mut chunker)?;
    let chunk_ref = chunker.finish()?;
    Ok(chunk_ref)
}

pub fn retrieve_tree_node<RP: RawPile>(
    pile: &Pile<RP>,
    chunk_ref: RecursiveChunkRef,
) -> anyhow::Result<RootTreeNode> {
    let mut serialised = Vec::new();
    let mut unchunker = RecursiveUnchunker::new(pile, chunk_ref);
    io::copy(&mut unchunker, &mut serialised)?;
    Ok(serde_bare::from_slice(&serialised)?)
    /*
    let unchunker = RecursiveUnchunker::new(pile, chunk_ref);
    Ok(serde_bare::from_reader(unchunker)?)
     */
}

pub fn fully_integrate_pointer_node<RP: RawPile>(
    pile: &Pile<RP>,
    tree_node: &mut TreeNode,
    pointer: &mut PointerData,
) -> anyhow::Result<()> {
    if let Some(parent_name) = &pointer.parent_pointer {
        let mut parent = pile
            .read_pointer(parent_name.as_str())?
            .ok_or_else(|| anyhow!("Parent pointer {:?} not found.", parent_name))?;
        let mut parent_node = retrieve_tree_node(pile, parent.chunk_ref.clone())?.node;

        fully_integrate_pointer_node(pile, &mut parent_node, &mut parent)?;
        integrate_node_in_place(tree_node, &mut parent_node)?;

        // merge in the UID and GID tables when integrating.
        if !merge_uid_or_gid_tables(&mut pointer.uid_lookup, &parent.uid_lookup) {
            warn!(
                "Overlap when merging parent:{:?}'s UID table into child.",
                parent_name
            );
        }
        if !merge_uid_or_gid_tables(&mut pointer.gid_lookup, &parent.gid_lookup) {
            warn!(
                "Overlap when merging parent:{:?}'s GID table into child.",
                parent_name
            );
        }

        pointer.parent_pointer = None;
    }
    Ok(())
}

pub fn fully_load_pointer<RP: RawPile>(
    pile: &Pile<RP>,
    pointer_name: &str,
) -> anyhow::Result<(PointerData, RootTreeNode)> {
    let mut pointer_data = pile
        .read_pointer(pointer_name)?
        .ok_or_else(|| anyhow!("Pointer {:?} not found.", pointer_name))?;
    let mut root_node = retrieve_tree_node(pile, pointer_data.chunk_ref.clone())?;

    fully_integrate_pointer_node(pile, &mut root_node.node, &mut pointer_data)?;

    Ok((pointer_data, root_node))
}