ghee 0.6.1

use std::path::PathBuf;

use anyhow::Result;
use path_absolutize::Absolutize;
use walkdir::{DirEntry, WalkDir};

use ghee_lang::{Key, Namespace, Predicate, Value, Xattr};

use crate::{
    best_index, containing_table_info, is_hidden, table_info, xattr_values, xattr_values_from_path,
    Record, TableInfo, DEFAULT_KEY, XATTR_GHEE_LOWER, XATTR_GHEE_UPPER,
};

pub struct PathVisit<'a, 'b, 'c> {
    /// Info about the table to which this path belongs, if any
    pub table_info: Option<&'a TableInfo>,

    /// The path we're visiting
    pub path: &'b PathBuf,

    /// The xattr or field values of the path we're visiting
    pub xattr_values: &'c Record,
}

impl<'a, 'b, 'c> PathVisit<'a, 'b, 'c> {
    pub fn is_table_root(&self) -> bool {
        if let Some(table_info) = self.table_info {
            let table_path = table_info.index_path_abs(table_info.key());
            table_path == self.path
        } else {
            false
        }
    }

    /// The primary key of the table to which this path belongs, if any
    pub fn primary_key(&self) -> Option<&Key> {
        self.table_info.map(|table_info| table_info.key())
    }

    /// Get the xattr values but excluding the `user.ghee` prefix (meta values)
    pub fn xattr_values_no_meta(&self) -> impl Iterator<Item = (&Xattr, &Value)> {
        self.xattr_values
            .iter()
            .filter(|(k, _v)| !(k.namespace == Namespace::User && k.attr.starts_with("ghee")))
    }

    /// The path by which this record is reachable in the original table/index targeted
    pub fn original_path_abs(&self) -> Result<PathBuf> {
        if let Some(table_info) = self.table_info {
            let original_table_key = table_info.key();
            let subkey_values = original_table_key.value_for_record(self.xattr_values)?;

            let mut path = table_info.index_path_abs(original_table_key).clone();

            for sub in subkey_values {
                path.push(sub.to_string());
            }

            Ok(path)
        } else {
            // If there are no indices, this path itself is the original path
            Ok(self.path.clone())
        }
    }
}

/// Walk a path from Ghee's point of view, yielding only the DirEntry's, i.e. _not_
/// loading the xattrs.
///
/// This will prune subpaths which are known to violate predicates, and ignore nested indices.
///
/// WARNING: It is possible this will yield paths which violate the predicates, when this wasn't known
/// by inferring xattr values from subpaths.
/// FIXME Make private
pub(crate) fn walk_paths<'a, 'b>(
    path: &'a PathBuf,
    where_: &'b Vec<Predicate>,
    recursive: bool,
    sort: bool,
) -> Result<impl Iterator<Item = core::result::Result<DirEntry, walkdir::Error>> + 'b> {
    let max_depth = if recursive { usize::MAX } else { 0 };

    let abs_path = path.absolutize().unwrap().to_path_buf();

    let (key, base_path, traverse_path) = if let Some(table_info) = containing_table_info(path)? {
        if table_info.path_abs() == &abs_path {
            // `path` is an initialized table
            let (best_key, best_path) =
                best_index(table_info.indices_abs(), where_, table_info.key());
            (best_key.clone(), best_path.clone(), best_path.clone())
        } else {
            // `path` is a sub-path of an initialized table
            // whether it's a file or a directory, we will traverse from there
            (
                table_info.key().clone(),
                table_info.path_abs().clone(),
                abs_path.clone(),
            )

            //TODO? Use best_index, translating `path` into the chosen index's hierarchy
        }
    } else {
        if path.is_file() {
            (
                DEFAULT_KEY.clone(),
                abs_path.parent().unwrap().to_path_buf(),
                abs_path.clone(),
            )
        } else {
            debug_assert!(path.is_dir());
            (DEFAULT_KEY.clone(), abs_path.clone(), abs_path.clone())
        }
    };

    let mut walker = WalkDir::new(traverse_path).max_depth(max_depth);

    if sort {
        walker = walker.sort_by_file_name();
    }

    let mut first = true;

    Ok(walker.into_iter().filter_entry(move |e| {
        // Ignore nested indices but not the initial path
        // we're walking
        if !first && is_hidden(e) {
            return false;
        }

        first = false;

        let values = xattr_values_from_path(&key, &base_path, e.path()).unwrap();

        // Of all xattr values set on this path, if a relevant predicate is contradicted, proceed no further
        for xattr in values.keys() {
            if let Some(predicate) = where_.iter().find(|p| p.xattr == *xattr) {
                if !predicate.satisfied(&values) {
                    return false;
                }
            }
        }

        true
    }))
}

/**
 * Walk a path from Ghee's point of view, loading and visiting each record, filtered by any provided
 * predicate constraints,
 *
 * A bare directory is considered a table with default primary key components.
 *
 * A file is considered a single record with fields defined by its xattrs, within a table with a key of ['key0']
 *
 * Indices can be provided if the root is a table.
 *
 * Nested indices are ignored completely when not being utilized to speed up the walk.
 *
 * # Arguments
 * * `recursive`: descend recursively into directories
 * * `all`: do not omit the `user.ghee` prefix from returned xattrs
 * * `sort`: walk within directories in sort order; might be slower on large directories
 */
pub fn walk_records<F: Fn(PathVisit) -> Result<()>>(
    path: &PathBuf,
    where_: &Vec<Predicate>,
    recursive: bool,
    all: bool,
    sort: bool,
    visit_empty: bool,
    visitor: &F,
) -> Result<()> {
    let table_info = table_info(path)?;

    'outer: for entry in walk_paths(path, where_, recursive, sort)? {
        let entry = entry?;

        let values = {
            let mut values = xattr_values(entry.path())?;

            if !all {
                // Remove user.ghee prefixed xattrs unless `all` is specified
                let ghee_keys: Vec<Xattr> = values
                    .range(XATTR_GHEE_LOWER.clone()..XATTR_GHEE_UPPER.clone())
                    .map(|(k, _v)| k)
                    .cloned()
                    .collect();

                for k in ghee_keys {
                    values.remove(&k);
                }
            }

            values
        };

        if !values.is_empty() || visit_empty {
            // TODO This redundantly checks predicates covered by the subpath components
            //      We could omit the redundant checks based on the path depth
            for predicate in where_ {
                if !predicate.satisfied(&values) {
                    continue 'outer;
                }
            }

            let visit = PathVisit {
                table_info: table_info.as_ref(),
                path: &entry.path().to_path_buf(),
                xattr_values: &values,
            };

            visitor(visit)?;
        }
    }

    Ok(())
}

#[cfg(test)]
mod test {
    use std::{
        cell::RefCell,
        fs::{create_dir, File},
        path::PathBuf,
        rc::Rc,
    };

    use ghee_lang::{parse_assignment, parse_predicate, Key, Predicate};

    use crate::{
        cmd::{init, set},
        declare_indices,
        test_support::{Scenario, TempDirAuto},
    };

    use super::{walk_paths, walk_records};

    fn visited_paths(
        path: &PathBuf,
        where_: &Vec<Predicate>,
        recursive: bool,
        sort: bool,
    ) -> Vec<PathBuf> {
        let visited: Rc<RefCell<Vec<PathBuf>>> = Rc::new(RefCell::new(Vec::new()));

        walk_paths(path, where_, recursive, sort)
            .unwrap()
            .for_each(|path| {
                visited
                    .borrow_mut()
                    .push(path.unwrap().path().to_path_buf());
            });

        let refcell = Rc::into_inner(visited).unwrap();
        refcell.into_inner()
    }

    fn visited_records(
        path: &PathBuf,
        where_: &Vec<Predicate>,
        recursive: bool,
        all: bool,
        sort: bool,
    ) -> Vec<PathBuf> {
        let visited: Rc<RefCell<Vec<PathBuf>>> = Rc::new(RefCell::new(Vec::new()));

        walk_records(path, where_, recursive, all, sort, false, &|path| {
            visited.borrow_mut().push(path.path.clone());
            Ok(())
        })
        .unwrap();

        let refcell = Rc::into_inner(visited).unwrap();
        refcell.into_inner()
    }

    #[test]
    fn test_walk_bare_dir() {
        let dir = TempDirAuto::new("ghee-test-walk");

        assert_eq!(visited_paths(&dir, &Vec::new(), true, true).len(), 1);

        assert!(visited_records(&dir, &Vec::new(), true, true, false).is_empty());
    }

    #[test]
    fn test_walk_initialized_dir_all() {
        let dir = TempDirAuto::new("ghee-test-walk-dir-all");

        init(&dir, &Key::from_string("test"), false).unwrap();

        assert_eq!(
            visited_records(&dir, &Vec::new(), true, true, false).len(),
            1,
            "Initialized dir wasn't visited even with `all`"
        );
    }

    #[test]
    fn test_walk_initialized_dir_not_all() {
        let dir = TempDirAuto::new("ghee-test-walk-dir-not-all");

        init(&dir, &Key::from_string("test"), false).unwrap();

        let visited = visited_records(&dir, &Vec::new(), true, false, true);

        assert!(visited.is_empty());
    }

    #[test]
    fn test_walk_paths_recursiveness() {
        let dir1 = TempDirAuto::new("ghee-test-walk-paths-dir1");

        set(
            &vec![&dir1],
            &vec![parse_assignment(b"a=1").unwrap().1],
            false,
            false,
        )
        .unwrap();

        let mut dir2 = dir1.clone();
        dir2.push("child");

        create_dir(&dir2).unwrap();

        set(
            &vec![&dir2],
            &vec![parse_assignment(b"a=2").unwrap().1],
            false,
            false,
        )
        .unwrap();

        let visited_flat = visited_paths(&dir1, &Vec::new(), false, true);

        assert_eq!(visited_flat.len(), 1);
        assert!(visited_flat.contains(&dir1));

        let visited_recursive = visited_paths(&dir1, &Vec::new(), true, false);

        assert_eq!(visited_recursive.len(), 2);
        assert!(visited_recursive.contains(&dir1));
        assert!(visited_recursive.contains(&dir2));
    }

    #[test]
    fn test_walk_records_recursiveness() {
        let dir1 = TempDirAuto::new("ghee-test-walk-records-dir1");

        set(
            &vec![&dir1],
            &vec![parse_assignment(b"a=1").unwrap().1],
            false,
            false,
        )
        .unwrap();

        let mut dir2 = dir1.clone();
        dir2.push("child");

        create_dir(&dir2).unwrap();

        set(
            &vec![&dir2],
            &vec![parse_assignment(b"a=2").unwrap().1],
            false,
            false,
        )
        .unwrap();

        let visited_flat = visited_records(&dir1, &Vec::new(), false, false, true);

        assert_eq!(visited_flat.len(), 1);
        assert!(visited_flat.contains(&dir1));

        let visited_recursive = visited_records(&dir1, &Vec::new(), true, false, false);

        assert_eq!(visited_recursive.len(), 2);
        assert!(visited_recursive.contains(&dir1));
        assert!(visited_recursive.contains(&dir2));
    }

    #[test]
    fn test_walk_paths_predicate() {
        let s = Scenario::new("ghee-test-walk-paths-predicate");

        assert_eq!(visited_paths(&s.dir1, &vec![], true, true).len(), 3);

        assert_eq!(
            visited_paths(
                &s.dir1,
                &vec![parse_predicate(b"user.test1=0").unwrap().1],
                true,
                true
            )
            .len(),
            2 // root dir and matching record
        );

        assert_eq!(
            visited_paths(
                &s.dir1,
                &vec![parse_predicate(b"user.test1=10").unwrap().1],
                true,
                true
            )
            .len(),
            2 // root dir and matching record
        );
    }

    #[test]
    fn test_walk_records_predicate() {
        let s = Scenario::new("ghee-test-walk-records-predicate");

        assert_eq!(
            visited_records(&s.dir1, &vec![], true, false, true).len(),
            2
        );
        assert_eq!(visited_records(&s.dir1, &vec![], true, true, true).len(), 3);

        assert_eq!(
            visited_records(
                &s.dir1,
                &vec![parse_predicate(b"user.test1=0").unwrap().1],
                true,
                false,
                true
            )
            .len(),
            1
        );

        assert_eq!(
            visited_records(
                &s.dir1,
                &vec![parse_predicate(b"user.test1=10").unwrap().1],
                true,
                false,
                true
            )
            .len(),
            1
        );
    }

    #[test]
    fn test_walk_paths_file_with_containing_table_info() {
        let dir = TempDirAuto::new("ghee-test-walk-paths-file-with-containing-table-info");

        let key = Key::from_string("blah");

        init(&dir, &key, false).unwrap();

        let file = {
            let mut path = dir.clone();
            path.push("f");
            path
        };

        File::create(&file).unwrap();

        {
            let visited = visited_paths(&file, &Vec::new(), false, false);

            assert_eq!(visited.len(), 1);

            assert_eq!(
                visited[0], file,
                "something other than just the file was visited"
            );
        }

        // Failing predicate
        {
            let visited = visited_paths(
                &file,
                &vec![parse_predicate(b"blah=d").unwrap().1],
                false,
                true,
            );

            assert_eq!(visited.len(), 0);
        }
    }

    #[test]
    fn test_walk_paths_file_without_containing_table_info() {
        let dir = TempDirAuto::new("ghee-test-walk-paths-file-without-containing-table-info");

        let file = {
            let mut path = dir.clone();
            path.push("f");
            path
        };

        File::create(&file).unwrap();

        {
            let visited = visited_paths(&file, &Vec::new(), false, false);

            assert_eq!(visited.len(), 1);

            assert_eq!(visited[0], file);
        }

        // Failing default key predicate
        {
            let visited = visited_paths(
                &file,
                &vec![parse_predicate(b"key0=d").unwrap().1],
                false,
                true,
            );

            assert_eq!(visited.len(), 0, "path returned that fails key0 predicate");
        }
    }

    #[test]
    fn test_walk_paths_rootdir_with_table_info() {
        let dir = TempDirAuto::new("ghee-test-walk-paths-rootdir-with-table-info");

        assert!(dir.is_dir());

        let key = Key::from_string("blah");

        init(&dir, &key, false).unwrap();

        let file = {
            let mut path = dir.clone();
            path.push("f");
            path
        };

        File::create(&file).unwrap();

        {
            assert!(dir.is_dir());

            let visited = visited_paths(&dir, &Vec::new(), true, false);

            assert!(visited.contains(&file));
            assert!(visited.contains(&dir));

            assert_eq!(visited.len(), 2);
        }

        // Failing predicate
        {
            let visited = visited_paths(
                &dir,
                &vec![parse_predicate(b"blah=d").unwrap().1],
                false,
                true,
            );

            assert_eq!(visited.len(), 1);
            assert_eq!(
                visited[0], *dir,
                "dir not returned though it doesn't contradict predicate"
            );
        }
    }

    #[test]
    fn test_walk_paths_subdir_with_containing_table_info() {
        let dir = TempDirAuto::new("ghee-test-walk-paths-subdir-with-containing-table-info");

        let key = Key::from_string("blah,blor");

        init(&dir, &key, false).unwrap();

        let subdir = {
            let mut path = dir.clone();
            path.push("f");
            path
        };

        create_dir(&subdir).unwrap();

        let file = {
            let mut path = subdir.clone();
            path.push("y");
            path
        };

        File::create(&file).unwrap();

        {
            let visited = visited_paths(&subdir, &Vec::new(), true, true);
            assert_eq!(visited.len(), 2);
            assert!(visited.contains(&subdir));
            assert!(visited.contains(&file));
        }

        // Successful predicate on blah
        {
            let visited = visited_paths(
                &subdir,
                &vec![parse_predicate(b"blah=f").unwrap().1],
                true,
                true,
            );
            assert_eq!(visited.len(), 2);
            assert!(visited.contains(&subdir));
            assert!(visited.contains(&file));
        }

        // Successful predicate on blor
        {
            let visited = visited_paths(
                &subdir,
                &vec![parse_predicate(b"blor=y").unwrap().1],
                true,
                true,
            );
            assert_eq!(visited.len(), 2);
            assert!(visited.contains(&subdir));
            assert!(visited.contains(&file));
        }

        // Failed predicate on blah
        {
            let visited = visited_paths(
                &subdir,
                &vec![parse_predicate(b"blah=g").unwrap().1],
                true,
                true,
            );
            assert_eq!(visited.len(), 0, "visited path that fails predicate");
        }

        // Failed predicate on blor
        {
            let visited = visited_paths(
                &subdir,
                &vec![parse_predicate(b"blor=z").unwrap().1],
                true,
                true,
            );
            assert_eq!(visited.len(), 1);
            assert_eq!(visited[0], subdir);
        }
    }

    #[test]
    fn test_walk_paths_dir_without_containing_table_info() {
        let dir = TempDirAuto::new("ghee-test-walk-paths-subdir-with-containing-table-info");

        let file = {
            let mut path = dir.clone();
            path.push("z");
            path
        };

        File::create(&file).unwrap();

        {
            let visited = visited_paths(&dir, &Vec::new(), true, true);
            assert_eq!(visited.len(), 2);
            assert!(visited.contains(&dir));
            assert!(visited.contains(&file));
        }

        // Successful predicate on key0
        {
            let visited = visited_paths(
                &dir,
                &vec![parse_predicate(b"key0=z").unwrap().1],
                true,
                true,
            );
            assert_eq!(visited.len(), 2);
            assert!(visited.contains(&dir));
            assert!(visited.contains(&file));
        }

        // Failed predicate on key0
        {
            let visited = visited_paths(
                &dir,
                &vec![parse_predicate(b"key0=x").unwrap().1],
                true,
                true,
            );
            assert_eq!(visited.len(), 1);
            assert_eq!(visited[0], *dir);
        }
    }

    #[test]
    fn test_walk_paths_prune_by_indices() {
        let dir1 = TempDirAuto::new("ghee-test-walk-paths-prune-by-indices-dir1");
        let dir2 = TempDirAuto::new("ghee-test-walk-paths-prune-by-indices-dir2");

        let key1 = Key::from_string("name");
        let key2 = Key::from_string("state,name");

        let dir1paths: Vec<PathBuf> = vec!["Vivienne", "Drema", "Vella", "Maurita"]
            .into_iter()
            .map(|i| {
                let mut p = dir1.clone();
                p.push(i);

                File::create(&p).unwrap();

                p
            })
            .collect();

        let mut dir2paths: Vec<PathBuf> = Vec::with_capacity(4);

        let dir2sub1 = {
            let mut p = dir2.clone();
            p.push("NM");
            create_dir(&p).unwrap();
            p
        };

        let dir2sub2 = {
            let mut p = dir2.clone();
            p.push("WA");
            create_dir(&p).unwrap();
            p
        };

        vec!["Vivienne", "Drema"]
            .into_iter()
            .map(|name| {
                let mut p = dir2sub1.clone();
                p.push(name);
                File::create(&p).unwrap();
                p
            })
            .for_each(|p| dir2paths.push(p));

        vec!["Vella", "Maurita"]
            .into_iter()
            .map(|name| {
                let mut p = dir2sub2.clone();
                p.push(name);
                File::create(&p).unwrap();
                p
            })
            .for_each(|p| dir2paths.push(p));

        init(&dir1, &key1, false).unwrap();

        init(&dir2, &key2, false).unwrap();

        // Full dir1
        {
            let visited = visited_paths(&dir1, &Vec::new(), true, true);
            assert_eq!(visited.len(), dir1paths.len() + 1); // base path plus four people
            assert!(visited.contains(&dir1));
            for p in &dir1paths {
                assert!(visited.contains(&p));
            }
        }

        // Full dir2
        {
            let visited = visited_paths(&dir2, &Vec::new(), true, true);
            assert_eq!(visited.len(), dir2paths.len() + 2 + 1); // base path plus 2 state subdirs plus four people
            assert!(visited.contains(&dir2));
            assert!(visited.contains(&dir2sub1));
            assert!(visited.contains(&dir2sub2));
            for p in &dir2paths {
                assert!(visited.contains(&p));
            }
        }

        // dir1 pruned by name
        {
            let visited = visited_paths(
                &dir1,
                &vec![parse_predicate(b"name=Vivienne").unwrap().1],
                true,
                true,
            );

            assert_eq!(visited.len(), 2); // base path plus matching file
            assert!(visited.contains(&dir1));
            assert!(visited.contains(&dir1paths[0]));
        }

        // dir2 pruned by state
        {
            let visited = visited_paths(
                &dir2,
                &vec![parse_predicate(b"state=NM").unwrap().1],
                true,
                true,
            );

            assert_eq!(visited.len(), 4); // base path plus NM dir plus NM people
            assert!(visited.contains(&dir2));
            assert!(visited.contains(&dir2sub1));
            assert!(visited.contains(&dir2paths[0]));
            assert!(visited.contains(&dir2paths[1]));
        }

        // dir2 pruned by name
        {
            let visited = visited_paths(
                &dir2,
                &vec![parse_predicate(b"name=Vella").unwrap().1],
                true,
                true,
            );

            assert_eq!(visited.len(), 4); // base path plus NM and WA subdirs plus Vella
            assert!(visited.contains(&dir2));
            assert!(visited.contains(&dir2sub1));
            assert!(visited.contains(&dir2sub2));
            assert!(visited.contains(&dir2paths[2]));
        }

        //--- Demonstrate lack of index acceleration

        // dir1 pruned by state - no pruning happens
        {
            let visited = visited_paths(
                &dir1,
                &vec![parse_predicate(b"state=WA").unwrap().1],
                true,
                true,
            );

            assert_eq!(visited.len(), dir1paths.len() + 1); // base path plus four people
            assert!(visited.contains(&dir1));
            for p in &dir1paths {
                assert!(visited.contains(&p));
            }
        }

        declare_indices(&dir1, &dir2).unwrap();

        // dir1 pruned by state - pruning happens via dir2 index
        {
            let visited = visited_paths(
                &dir1,
                &vec![parse_predicate(b"state=WA").unwrap().1],
                true,
                true,
            );

            assert_eq!(visited.len(), 4); // base path plus WA subdir plus two Washingtonians
            assert!(visited.contains(&dir2));
            assert!(visited.contains(&dir2sub2));
            assert!(visited.contains(&dir2paths[2]));
            assert!(visited.contains(&dir2paths[3]));
        }
    }
}