/*
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::ErrorKind;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU32, Ordering};

use anyhow::{anyhow, bail, Context};
use crossbeam_channel::{Receiver, Sender};
use crossbeam_utils::thread;
use indicatif::{ProgressBar, ProgressDrawTarget, ProgressStyle};
use log::{error, warn};

use crate::chunking::{RecursiveChunker, SENSIBLE_THRESHOLD};
use crate::commands;
use crate::commands::{fully_integrate_pointer_node, retrieve_tree_node};
use crate::definitions::{PointerData, RecursiveChunkRef, RootTreeNode, TreeNode};
use crate::pile::{Pile, RawPile};
use crate::tree::{create_uidgid_lookup_tables, differentiate_node_in_place};
use std::collections::BTreeMap;

pub fn store<RP: RawPile>(
    root_path: &Path,
    root: &mut TreeNode,
    pile: &Pile<RP>,
    make_progress_bar: bool,
    num_workers: u8,
) -> anyhow::Result<()> {
    let pbar = if make_progress_bar {
        ProgressBar::with_draw_target(
            root.count_normal_files() as u64,
            ProgressDrawTarget::stdout_with_hz(10),
        )
    } else {
        ProgressBar::hidden()
    };
    pbar.set_style(
        ProgressStyle::default_bar()
            .template("[{elapsed_precise}]/[{eta}] {bar:40.cyan/blue} {pos:>7}/{len:7} {msg}"),
    );
    pbar.set_message("storing");

    let (paths_send, paths_recv) = crossbeam_channel::unbounded();
    let (results_send, results_recv) = crossbeam_channel::bounded(16);

    let critical_failures = AtomicU32::new(0);

    thread::scope(|s| {
        for worker_num in 0..num_workers {
            let paths_recv = paths_recv.clone();
            let results_send = results_send.clone();
            let critical_failures = &critical_failures; // needed because of move
            s.builder()
                .name(format!("yama chunker {}", worker_num))
                .spawn(move |_| {
                    if let Err(e) = store_worker(root_path, pile, paths_recv, results_send) {
                        error!("[critical!] Storage worker {} FAILED: {:?}", worker_num, e);
                        critical_failures.fetch_add(1, Ordering::Relaxed);
                    }
                })
                .expect("Failed to start thread");
        }

        drop(results_send);
        drop(paths_recv);

        if let Err(e) = manager(root, paths_send, results_recv, &pbar) {
            error!("[critical!] Storage manager FAILED: {:?}", e);
            critical_failures.fetch_add(1, Ordering::Relaxed);
        }
    })
    .expect("thread scope failed");

    let critical_failures = critical_failures.load(Ordering::SeqCst);
    if critical_failures > 0 {
        bail!("There were {} critical failures.", critical_failures);
    } else {
        Ok(())
    }
}

pub fn store_worker<RP: RawPile>(
    root: &Path,
    pile: &Pile<RP>,
    paths: Receiver<String>,
    results: Sender<(String, Option<RecursiveChunkRef>)>,
) -> anyhow::Result<()> {
    while let Ok(path) = paths.recv() {
        let full_path = root.join(&path);
        match File::open(&full_path) {
            Ok(mut file) => {
                let mut chunker = RecursiveChunker::new(SENSIBLE_THRESHOLD, &pile);
                // streaming copy from file to chunker, really cool :)
                io::copy(&mut file, &mut chunker)?;
                let chunk_ref = chunker.finish()?;
                results
                    .send((path, Some(chunk_ref)))
                    .or(Err(anyhow!("Failed to send result.")))?;
            }
            Err(err) => match err.kind() {
                ErrorKind::NotFound => {
                    warn!("File vanished: {:?}. Will ignore.", full_path);
                    // send None so the manager knows to remove this from the tree.
                    results
                        .send((path, None))
                        .or(Err(anyhow!("Failed to send result.")))?;
                }
                ErrorKind::PermissionDenied => {
                    // TODO think about if we want a 'skip failed permissions' mode ...
                    error!(
                        "Permission denied to read {:?}; do you need to change user?",
                        full_path
                    );
                    Err(err)?;
                }
                _ => {
                    Err(err)?;
                }
            },
        };
    }

    Ok(())
}

fn delete_node(root: &mut TreeNode, child_path: &str) -> anyhow::Result<()> {
    let path_pieces: Vec<&str> = child_path.split('/').collect();

    let mut this = root;

    for &piece in &path_pieces[0..path_pieces.len() - 1] {
        if let TreeNode::Directory { children, .. } = this {
            match children.get_mut(piece) {
                None => bail!(
                    "Tried to delete {} but {} does not exist.",
                    child_path,
                    piece
                ),
                Some(next) => this = next,
            }
        } else {
            bail!(
                "Tried to delete {} from tree node but '{}' not a directory.",
                child_path,
                piece
            );
        }
    }

    if let TreeNode::Directory { children, .. } = this {
        children.remove(*path_pieces.last().unwrap());
    } else {
        bail!(
            "Tried to delete {} from tree node but parent not a directory.",
            child_path
        );
    }

    Ok(())
}

fn update_node(
    root: &mut TreeNode,
    child_path: &str,
    new_ref: RecursiveChunkRef,
) -> anyhow::Result<()> {
    let mut this = root;
    for piece in child_path.split('/') {
        if let TreeNode::Directory { children, .. } = this {
            this = children
                .get_mut(piece)
                .ok_or_else(|| anyhow!("Tried to update {} but {} not found", child_path, piece))?;
        } else {
            bail!(
                "Tried to update {} but {} not a directory.",
                child_path,
                piece
            );
        }
    }

    if let TreeNode::NormalFile { content, .. } = this {
        *content = new_ref;
    } else {
        bail!("Tried to update {} but it's not a NormalFile.", child_path);
    }

    Ok(())
}

pub fn manager(
    root: &mut TreeNode,
    paths_sender: Sender<String>,
    results_receiver: Receiver<(String, Option<RecursiveChunkRef>)>,
    progress_bar: &ProgressBar,
) -> anyhow::Result<()> {
    root.visit(
        &mut |tree_node, name| {
            if let TreeNode::NormalFile { .. } = tree_node {
                paths_sender
                    .send(name.to_string())
                    .or_else(|_| Err(anyhow!("Unable to send to should-be unbounded channel")))?;
            }
            Ok(())
        },
        "".to_string(),
    )?;

    drop(paths_sender);

    while let Ok((path, opt_chunk_ref)) = results_receiver.recv() {
        progress_bar.inc(1);
        match opt_chunk_ref {
            None => {
                delete_node(root, &path)?;
            }
            Some(new_chunk_ref) => {
                update_node(root, &path, new_chunk_ref)?;
            }
        }
    }

    Ok(())
}

pub fn store_fully(
    pile: &Pile<Box<dyn RawPile>>,
    root_dir: &PathBuf,
    new_pointer_name: &String,
    mut root_node: TreeNode,
    parent: Option<String>,
    num_workers: u8,
) -> anyhow::Result<()> {
    if let Some(parent) = parent.as_ref() {
        let mut parent_pointer = pile.read_pointer(parent)?.ok_or_else(|| {
            anyhow!(
                "Selected parent pointer {:?} didn't exist when tried to retrieve it.",
                parent
            )
        })?;
        let mut parent_node = retrieve_tree_node(&pile, parent_pointer.chunk_ref.clone())?;

        fully_integrate_pointer_node(&pile, &mut parent_node.node, &mut parent_pointer)?;
        differentiate_node_in_place(&mut root_node, &parent_node.node)?;
    }

    store(&root_dir, &mut root_node, &pile, true, num_workers)?;

    let mut uid_lookup = BTreeMap::new();
    let mut gid_lookup = BTreeMap::new();

    create_uidgid_lookup_tables(&root_node, &mut uid_lookup, &mut gid_lookup)
        .context("Failed to build UID and GID lookup tables :(.")?;

    let chunk_ref = commands::store_tree_node(
        &pile,
        &RootTreeNode {
            name: root_dir.file_name().unwrap().to_str().unwrap().to_owned(),
            node: root_node,
        },
    )?;

    let pointer_data = PointerData {
        chunk_ref,
        parent_pointer: parent,
        uid_lookup,
        gid_lookup,
    };

    pile.write_pointer(&new_pointer_name, &pointer_data)?;

    Ok(())
}