//! flow-casfs: an Oxygen state mounted read-only (OxygenFlow spec 5.1).
//!
//! The tree comes from the repository's state index; an opaque file's bytes
//! are its stored blob; a text file's bytes are a rendering of its history,
//! made on first read or stat and cached on disk by graph key (equal keys
//! mean equal renderings), so every state in which a file has the same
//! history finds it at once.
//!
//!   flow-casfs mount REPO MOUNTPOINT [--bookmark NAME]
//!
//! It mounts in the foreground and prints how long the mount took; unmount
//! with `fusermount3 -u MOUNTPOINT`.

use anyhow::{bail, Context, Result};
use fuser::{FileAttr, FileType, Filesystem, MountOption, ReplyAttr, ReplyData, ReplyDirectory, ReplyEntry, Request};
use oxy_core::ids::NodeId;
use oxy_core::objects::NodeKind;
use oxy_core::repo::Repo;
use oxy_core::state::State;
use std::collections::HashMap;
use std::ffi::{OsStr, OsString};
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::{Duration, Instant, SystemTime};

/// How long the kernel may keep attributes and names: the mount never
/// changes.
const TTL: Duration = Duration::from_secs(3600);
/// Bytes of files kept in memory; the kernel's page cache keeps the hot
/// ones too, and a file dropped here is cheap to read again.
const MEMORY_BYTES: usize = 256 << 20;

struct Inode {
    parent: u64,
    kind: NodeKind,
    node: NodeId,
    exec: bool,
    opaque: bool,
    /// For a directory: its children by name.
    children: Vec<(OsString, u64)>,
}

struct CasFs {
    repo: Repo,
    state: State,
    inodes: Vec<Inode>,
    /// Bytes already read, by inode, with the order they came in and their
    /// total: the oldest go first beyond `MEMORY_BYTES`.
    bytes: HashMap<u64, Arc<Vec<u8>>>,
    order: std::collections::VecDeque<u64>,
    held: usize,
    cache_dir: PathBuf,
    uid: u32,
    gid: u32,
    mounted: SystemTime,
}

impl CasFs {
    fn new(repo: Repo, state: State) -> CasFs {
        let view = state.tree().view();
        let root = Inode { parent: 1, kind: NodeKind::Dir, node: NodeId::ROOT, exec: false, opaque: false, children: Vec::new() };
        let mut inodes = vec![root];
        let mut by_path: HashMap<&str, u64> = HashMap::with_capacity(view.entries.len());
        by_path.insert("", 1);
        // In path order, so a directory comes before what it holds.
        for (path, e) in view.entries.iter() {
            let (dir, name) = path.rsplit_once('/').unwrap_or(("", path.as_str()));
            let Some(&parent) = by_path.get(dir) else { continue };
            let ino = inodes.len() as u64 + 1;
            inodes.push(Inode { parent, kind: e.kind, node: e.node, exec: e.mode.exec, opaque: e.mode.opaque, children: Vec::new() });
            inodes[parent as usize - 1].children.push((OsString::from(name), ino));
            if e.kind == NodeKind::Dir || e.kind == NodeKind::Subrepo {
                by_path.insert(path.as_str(), ino);
            }
        }
        for i in inodes.iter_mut() {
            i.children.sort();
        }
        let cache_dir = repo.root().join(".oxy").join("cache").join("render");
        CasFs {
            repo,
            state,
            inodes,
            bytes: HashMap::new(),
            order: std::collections::VecDeque::new(),
            held: 0,
            cache_dir,
            uid: unsafe { libc::getuid() },
            gid: unsafe { libc::getgid() },
            mounted: SystemTime::now(),
        }
    }

    fn inode(&self, ino: u64) -> Option<&Inode> {
        self.inodes.get(ino.checked_sub(1)? as usize)
    }

    /// A file's or symlink's bytes: its blob, or its rendering (cached on
    /// disk by graph key).
    fn content(&mut self, ino: u64) -> Result<Arc<Vec<u8>>> {
        if let Some(b) = self.bytes.get(&ino) {
            return Ok(b.clone());
        }
        let i = self.inode(ino).context("no such inode")?;
        let (node, kind, opaque, parent) = (i.node, i.kind, i.opaque, i.parent);
        // A file added once and never edited: read it with every such file
        // of its directory not read yet, in one batch (a directory is
        // usually read whole), on every core.
        if kind == NodeKind::File && !opaque && self.state.single_text(&node).is_some() {
            let siblings: Vec<(u64, oxy_core::objects::TextRef)> = self.inodes[parent as usize - 1]
                .children
                .iter()
                .map(|(_, c)| *c)
                .filter(|c| !self.bytes.contains_key(c))
                .filter_map(|c| {
                    let i = &self.inodes[c as usize - 1];
                    (i.kind == NodeKind::File && !i.opaque).then_some(())?;
                    Some((c, self.state.single_text(&i.node)?))
                })
                .collect();
            let texts: Vec<oxy_core::objects::TextRef> = siblings.iter().map(|(_, t)| *t).collect();
            for ((c, _), b) in siblings.iter().zip(self.repo.single_texts(&texts)?) {
                self.keep(*c, Arc::new(b));
            }
            if let Some(b) = self.bytes.get(&ino) {
                return Ok(b.clone());
            }
        }
        let bytes = if kind == NodeKind::File && !opaque {
            self.rendering(&node)?
        } else {
            match self.state.live_blobs(&node).first() {
                Some(b) => self.repo.read_blob(b)?,
                None => Vec::new(),
            }
        };
        let bytes = Arc::new(bytes);
        self.keep(ino, bytes.clone());
        Ok(bytes)
    }

    /// Keep a file's bytes, dropping the oldest beyond `MEMORY_BYTES`.
    fn keep(&mut self, ino: u64, bytes: Arc<Vec<u8>>) {
        self.held += bytes.len();
        if let Some(old) = self.bytes.insert(ino, bytes) {
            self.held -= old.len();
        } else {
            self.order.push_back(ino);
        }
        while self.held > MEMORY_BYTES && self.order.len() > 1 {
            let Some(oldest) = self.order.pop_front() else { break };
            if let Some(b) = self.bytes.remove(&oldest) {
                self.held -= b.len();
            }
        }
    }

    fn rendering(&self, node: &NodeId) -> Result<Vec<u8>> {
        // A file added once and never edited is its own text: no graph, and
        // nothing to cache.
        if let Some(t) = self.state.single_text(node) {
            return Ok(self.repo.single_texts(&[t])?.pop().unwrap_or_default());
        }
        let key = self.state.graph_key(node);
        let cached = key.map(|k| self.cache_dir.join(hex(&k)));
        if let Some(p) = &cached {
            if let Ok(b) = std::fs::read(p) {
                return Ok(b);
            }
        }
        let graphs = self.repo.build_graphs_detached(&self.state, &[*node])?;
        let bytes = match graphs.get(node) {
            Some(g) => self.state.render_graph(g).0,
            None => Vec::new(),
        };
        if let Some(p) = cached {
            // A cache: written whole or not at all, without waiting for
            // the disk.
            let _ = std::fs::create_dir_all(&self.cache_dir);
            let tmp = p.with_extension(format!("tmp-{}", std::process::id()));
            if std::fs::write(&tmp, &bytes).is_ok() {
                let _ = std::fs::rename(&tmp, &p);
            }
        }
        Ok(bytes)
    }

    fn attr(&mut self, ino: u64) -> Result<FileAttr> {
        let i = self.inode(ino).context("no such inode")?;
        let (kind, perm) = match i.kind {
            NodeKind::Dir | NodeKind::Subrepo => (FileType::Directory, 0o555),
            NodeKind::Symlink => (FileType::Symlink, 0o777),
            _ if i.exec => (FileType::RegularFile, 0o555),
            _ => (FileType::RegularFile, 0o444),
        };
        let nlink = if kind == FileType::Directory { 2 } else { 1 };
        let size = match kind {
            FileType::Directory => 0,
            _ => self.content(ino)?.len() as u64,
        };
        Ok(FileAttr {
            ino,
            size,
            blocks: size.div_ceil(512),
            atime: self.mounted,
            mtime: self.mounted,
            ctime: self.mounted,
            crtime: self.mounted,
            kind,
            perm,
            nlink,
            uid: self.uid,
            gid: self.gid,
            rdev: 0,
            blksize: 4096,
            flags: 0,
        })
    }
}

fn hex(b: &[u8]) -> String {
    b.iter().map(|x| format!("{x:02x}")).collect()
}

impl Filesystem for CasFs {
    fn lookup(&mut self, _req: &Request<'_>, parent: u64, name: &OsStr, reply: ReplyEntry) {
        let found =
            self.inode(parent).and_then(|p| p.children.binary_search_by(|(n, _)| n.as_os_str().cmp(name)).ok().map(|i| p.children[i].1));
        match found.map(|ino| self.attr(ino)) {
            Some(Ok(a)) => reply.entry(&TTL, &a, 0),
            Some(Err(_)) => reply.error(libc::EIO),
            None => reply.error(libc::ENOENT),
        }
    }

    fn getattr(&mut self, _req: &Request<'_>, ino: u64, _fh: Option<u64>, reply: ReplyAttr) {
        match self.attr(ino) {
            Ok(a) => reply.attr(&TTL, &a),
            Err(_) => reply.error(libc::ENOENT),
        }
    }

    fn readlink(&mut self, _req: &Request<'_>, ino: u64, reply: ReplyData) {
        match self.content(ino) {
            Ok(b) => reply.data(&b),
            Err(_) => reply.error(libc::EIO),
        }
    }

    #[allow(clippy::too_many_arguments)]
    fn read(&mut self, _req: &Request<'_>, ino: u64, _fh: u64, offset: i64, size: u32, _flags: i32, _lock: Option<u64>, reply: ReplyData) {
        match self.content(ino) {
            Ok(b) => {
                let start = (offset.max(0) as usize).min(b.len());
                let end = (start + size as usize).min(b.len());
                reply.data(&b[start..end]);
            }
            Err(_) => reply.error(libc::EIO),
        }
    }

    fn readdir(&mut self, _req: &Request<'_>, ino: u64, _fh: u64, offset: i64, mut reply: ReplyDirectory) {
        let Some(dir) = self.inode(ino) else {
            reply.error(libc::ENOENT);
            return;
        };
        let mut entries: Vec<(u64, FileType, OsString)> =
            vec![(ino, FileType::Directory, ".".into()), (dir.parent, FileType::Directory, "..".into())];
        for (name, child) in &dir.children {
            let kind = match self.inodes[*child as usize - 1].kind {
                NodeKind::Dir | NodeKind::Subrepo => FileType::Directory,
                NodeKind::Symlink => FileType::Symlink,
                _ => FileType::RegularFile,
            };
            entries.push((*child, kind, name.clone()));
        }
        for (i, (child, kind, name)) in entries.into_iter().enumerate().skip(offset.max(0) as usize) {
            if reply.add(child, (i + 1) as i64, kind, name) {
                break;
            }
        }
        reply.ok();
    }
}

fn main() -> Result<()> {
    let args: Vec<String> = std::env::args().collect();
    let usage = "usage: flow-casfs mount REPO MOUNTPOINT [--bookmark NAME]";
    if args.len() < 4 || args[1] != "mount" {
        bail!("{usage}");
    }
    let (repo_dir, mountpoint) = (Path::new(&args[2]), Path::new(&args[3]));
    let bookmark = args.iter().position(|a| a == "--bookmark").and_then(|i| args.get(i + 1)).cloned();
    let started = Instant::now();
    let mut repo = Repo::open(repo_dir).with_context(|| format!("open {}", repo_dir.display()))?;
    if let Some(b) = &bookmark {
        repo.config.bookmark = b.clone();
    }
    let heads = repo.bookmark_heads()?;
    let state = repo.load_state_indexed(&heads)?;
    let loaded = started.elapsed();
    let fs = CasFs::new(repo, state);
    eprintln!(
        "flow-casfs: {} entries; state loaded in {:.1} ms, tree built in {:.1} ms",
        fs.inodes.len(),
        loaded.as_secs_f64() * 1000.0,
        (started.elapsed() - loaded).as_secs_f64() * 1000.0
    );
    let options = [MountOption::RO, MountOption::FSName("flow-casfs".into()), MountOption::DefaultPermissions];
    let session = fuser::spawn_mount2(fs, mountpoint, &options).context("mount")?;
    eprintln!("flow-casfs: mounted at {} in {:.1} ms", mountpoint.display(), started.elapsed().as_secs_f64() * 1000.0);
    // Until the mount goes away (fusermount3 -u) or the process is stopped.
    session.guard.join().ok();
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    /// The inode of a path, walking from the root.
    fn ino_of(fs: &CasFs, path: &str) -> Option<u64> {
        let mut ino = 1u64;
        for part in path.split('/') {
            let dir = fs.inode(ino)?;
            let i = dir.children.binary_search_by(|(n, _)| n.as_os_str().cmp(OsStr::new(part))).ok()?;
            ino = dir.children[i].1;
        }
        Some(ino)
    }

    #[test]
    fn a_mounted_state_holds_the_committed_files() {
        let dir = tempfile::tempdir().unwrap();
        std::env::set_var("OXY_HOME", dir.path().join("home"));
        let root = dir.path().join("repo");
        let repo = Repo::init(&root).unwrap();
        let big: String = (0..3000).map(|i| format!("line {i}\n")).collect();
        let files = [("a.txt", "one\ntwo\n".to_string()), ("src/lib.rs", "pub fn f() {}\n".to_string()), ("src/deep/big.txt", big)];
        for (p, c) in &files {
            let full = root.join(p);
            std::fs::create_dir_all(full.parent().unwrap()).unwrap();
            std::fs::write(full, c).unwrap();
        }
        repo.commit("first", &[]).unwrap();
        // An edit, so one file's rendering is a history, not one text.
        std::fs::write(root.join("a.txt"), "one\n2\ntwo\n").unwrap();
        repo.commit("second", &[]).unwrap();
        let heads = repo.bookmark_heads().unwrap();
        let state = repo.load_state_indexed(&heads).unwrap();
        let mut fs = CasFs::new(Repo::open(&root).unwrap(), state);
        for (p, c) in [("a.txt", "one\n2\ntwo\n"), ("src/lib.rs", "pub fn f() {}\n")] {
            let ino = ino_of(&fs, p).unwrap();
            assert_eq!(fs.content(ino).unwrap().as_slice(), c.as_bytes(), "{p}");
        }
        let big = ino_of(&fs, "src/deep/big.txt").unwrap();
        assert_eq!(fs.content(big).unwrap().as_slice(), files[2].1.as_bytes());
        assert_eq!(fs.attr(big).unwrap().size, files[2].1.len() as u64);
        assert_eq!(fs.attr(ino_of(&fs, "src/deep").unwrap()).unwrap().kind, FileType::Directory);
        assert!(ino_of(&fs, "missing.txt").is_none());
        // Only the edited file needed rendering, and its rendering is cached
        // by graph key, so a new mount reads it back; the others are their
        // own text blobs.
        let cached = std::fs::read_dir(&fs.cache_dir).unwrap().count();
        assert_eq!(cached, 1, "renderings cached");
        let state = repo.load_state_indexed(&heads).unwrap();
        let mut again = CasFs::new(Repo::open(&root).unwrap(), state);
        let ino = ino_of(&again, "a.txt").unwrap();
        assert_eq!(again.content(ino).unwrap().as_slice(), b"one\n2\ntwo\n");
    }
}