use std::collections::{BTreeSet, HashMap};
use std::ffi::{OsStr, OsString};
use std::fs::File;
use std::fs::OpenOptions;
use std::io;
use std::os::fd::{AsFd, AsRawFd};
use std::os::unix::ffi::{OsStrExt, OsStringExt};
use std::os::unix::fs::FileExt;
use std::os::unix::fs::OpenOptionsExt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::{Arc, Mutex, RwLock};
use std::time::{Duration, SystemTime, UNIX_EPOCH};

use fuser::{
    BsdFileFlags, Config, CopyFileRangeFlags, Errno, FileAttr, FileHandle, FileType, Filesystem,
    FopenFlags, Generation, INodeNo, InitFlags, KernelConfig, LockFlags, LockOwner, MountOption,
    OpenFlags, RenameFlags as FuseRenameFlags, ReplyAttr, ReplyCreate, ReplyData, ReplyDirectory,
    ReplyEmpty, ReplyEntry, ReplyLock, ReplyLseek, ReplyOpen, ReplyStatfs, ReplyWrite, ReplyXattr,
    Request, TimeOrNow, WriteFlags,
};
use nix::fcntl::{FcntlArg, fcntl};
use rustix::fs::{
    FallocateFlags, FlockOperation, Mode, OFlags, RenameFlags, SeekFrom, Timestamps, UTIME_NOW,
    UTIME_OMIT, XattrFlags, copy_file_range, fallocate, fchmod, fchown, fgetxattr, flistxattr,
    flock as bsd_flock, fremovexattr, fsetxattr, fstat, fstatvfs, ftruncate, futimens, mknodat,
    seek,
};
use rustix::process::{Gid, Uid};
use rustix::time::Timespec;

use crate::backing::{BackingRoot, validate_name, validate_relative};
use crate::classifier::tracked_paths_from_index;
use crate::classifier::{Classification, ClassificationSnapshot, GitClassifier};
use crate::migration::{Provenance, reserved};
use nodes::{
    DirectoryEntry, DirectoryStream, NodeTable, OpenFile, PosixLockKey, ResolvedEntry, errno,
    file_type, open_flags, system_time,
};

const TTL: Duration = Duration::ZERO;

#[derive(Clone, Debug)]
pub struct MountConfiguration {
    pub workspace: PathBuf,
    pub isolated_state: PathBuf,
    pub mountpoint: PathBuf,
    pub global_excludes: Option<PathBuf>,
    pub threads: usize,
}

pub(crate) struct RouterConfiguration {
    pub workspace_label: PathBuf,
    pub workspace: File,
    pub isolated_state: File,
    pub mountpoint: PathBuf,
    pub global_excludes: Option<PathBuf>,
    pub isolate_ignored: bool,
    pub threads: usize,
}

pub struct IgnoredWorkspace {
    host: BackingRoot,
    isolated: BackingRoot,
    classifier: GitClassifier,
    isolate_ignored: bool,
    nodes: Mutex<NodeTable>,
    files: Mutex<HashMap<u64, Arc<OpenFile>>>,
    directories: Mutex<HashMap<u64, DirectoryStream>>,
    posix_lock_files: Mutex<HashMap<PosixLockKey, Arc<File>>>,
    namespace: RwLock<()>,
    provenance: Mutex<Provenance>,
    next_handle: AtomicU64,
}

fn mount_config(threads: usize) -> Config {
    let mut config = Config::default();
    config.mount_options.extend([
        MountOption::FSName("lazaret-ignored-workspace".to_owned()),
        MountOption::RW,
        MountOption::Exec,
        MountOption::NoSuid,
        MountOption::NoDev,
        MountOption::DefaultPermissions,
    ]);
    config.n_threads = Some(threads.max(1));
    config.clone_fd = true;
    config
}

impl IgnoredWorkspace {
    /// # Errors
    /// Refuses roots that are not distinct, descriptor-openable directories in a Git workspace.
    pub fn open(
        workspace: &Path,
        isolated_state: &Path,
        global_excludes: Option<PathBuf>,
    ) -> io::Result<Self> {
        let workspace = workspace.canonicalize()?;
        let isolated_state = isolated_state.canonicalize()?;
        if workspace == isolated_state {
            return Err(io::Error::other("workspace and isolated roots must differ"));
        }
        let host = BackingRoot::open(&workspace)?;
        let isolated = BackingRoot::open(&isolated_state)?;
        Self::from_roots(workspace, host, isolated, global_excludes)
    }

    /// # Errors
    /// Refuses aliased or non-directory authorities and unreadable Git policy inputs.
    pub fn from_authorities(
        workspace_label: PathBuf,
        workspace: File,
        isolated_state: File,
        global_excludes: Option<PathBuf>,
    ) -> io::Result<Self> {
        let host = BackingRoot::from_file(workspace)?;
        let isolated = BackingRoot::from_file(isolated_state)?;
        Self::from_roots(workspace_label, host, isolated, global_excludes)
    }

    /// # Errors
    /// Refuses aliased or non-directory authorities and unreadable Git policy inputs.
    pub fn from_authorities_with_policy(
        workspace_label: PathBuf,
        workspace: File,
        isolated_state: File,
        global_excludes: Option<PathBuf>,
        isolate_ignored: bool,
    ) -> io::Result<Self> {
        let host = BackingRoot::from_file(workspace)?;
        let isolated = BackingRoot::from_file(isolated_state)?;
        Self::from_roots_with_policy(
            workspace_label,
            host,
            isolated,
            global_excludes,
            isolate_ignored,
        )
    }

    fn from_roots(
        workspace_label: PathBuf,
        host: BackingRoot,
        isolated: BackingRoot,
        global_excludes: Option<PathBuf>,
    ) -> io::Result<Self> {
        Self::from_roots_with_policy(workspace_label, host, isolated, global_excludes, true)
    }

    fn from_roots_with_policy(
        workspace_label: PathBuf,
        host: BackingRoot,
        isolated: BackingRoot,
        global_excludes: Option<PathBuf>,
        isolate_ignored: bool,
    ) -> io::Result<Self> {
        if host.identity()? == isolated.identity()? {
            return Err(io::Error::other("workspace and isolated authorities alias"));
        }
        crate::migration::initialize(&host, &isolated)?;
        let classifier_root = BackingRoot::from_file(host.duplicate()?)?;
        let classifier =
            GitClassifier::from_root(workspace_label, classifier_root, global_excludes)?;
        Ok(Self {
            host,
            isolated,
            classifier,
            isolate_ignored,
            nodes: Mutex::new(NodeTable::new()),
            files: Mutex::new(HashMap::new()),
            directories: Mutex::new(HashMap::new()),
            posix_lock_files: Mutex::new(HashMap::new()),
            namespace: RwLock::new(()),
            provenance: Mutex::new(Provenance::default()),
            next_handle: AtomicU64::new(1),
        })
    }

    /// # Errors
    /// Returns mount or daemon I/O failures.
    pub fn mount(configuration: MountConfiguration) -> io::Result<()> {
        let filesystem = Self::open(
            &configuration.workspace,
            &configuration.isolated_state,
            configuration.global_excludes,
        )?;
        let config = mount_config(configuration.threads);
        fuser::mount2(filesystem, configuration.mountpoint, &config)
    }

    /// # Errors
    /// Returns mount, readiness, control-channel, unmount, or daemon I/O failures.
    pub(crate) fn serve_from_authorities(
        configuration: RouterConfiguration,
        ready: impl FnOnce() -> io::Result<()>,
    ) -> io::Result<()> {
        let filesystem = Self::from_authorities_with_policy(
            configuration.workspace_label,
            configuration.workspace,
            configuration.isolated_state,
            configuration.global_excludes,
            configuration.isolate_ignored,
        )?;
        let session = fuser::spawn_mount2(
            filesystem,
            configuration.mountpoint,
            &mount_config(configuration.threads),
        )?;
        if let Err(error) = ready() {
            drop(session);
            return Err(error);
        }
        std::io::copy(&mut std::io::stdin().lock(), &mut std::io::sink())?;
        drop(session);
        Ok(())
    }

    fn root(&self, backing: Classification) -> &BackingRoot {
        match backing {
            Classification::Host => &self.host,
            Classification::Isolated => &self.isolated,
        }
    }

    fn path_for(&self, inode: INodeNo) -> io::Result<PathBuf> {
        self.nodes
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .path(inode)
            .ok_or_else(|| io::Error::from_raw_os_error(libc::ENOENT))
    }

    fn child_path(&self, parent: INodeNo, name: &OsStr) -> io::Result<PathBuf> {
        validate_name(name)?;
        let path = self.path_for(parent)?.join(name);
        validate_relative(&path)?;
        if reserved(&path) {
            return Err(io::Error::from_raw_os_error(libc::ENOENT));
        }
        Ok(path)
    }

    fn inode_for(&self, entry: &ResolvedEntry) -> INodeNo {
        if entry.path.as_os_str().is_empty() {
            return INodeNo::ROOT;
        }
        self.nodes
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .inode(entry)
    }

    fn resolve(&self, snapshot: &ClassificationSnapshot, path: &Path) -> io::Result<ResolvedEntry> {
        if reserved(path) {
            return Err(io::Error::from_raw_os_error(libc::ENOENT));
        }
        let host = self.host.stat(path).ok();
        let isolated = self.isolated.stat(path).ok();
        let directory = isolated
            .as_ref()
            .or(host.as_ref())
            .is_some_and(|stat| stat.st_mode & libc::S_IFMT == libc::S_IFDIR);
        let backing = self.classify(snapshot, path, directory);
        if snapshot.is_tracked(path) {
            if let (Some(host), Some(isolated)) = (&host, &isolated)
                && host.st_mode & libc::S_IFMT != isolated.st_mode & libc::S_IFMT
            {
                return Err(io::Error::from_raw_os_error(libc::EIO));
            }
            if backing == Classification::Host && host.is_none() && isolated.is_some() {
                return Err(io::Error::from_raw_os_error(libc::EIO));
            }
        }
        let stat = match backing {
            Classification::Host => host,
            Classification::Isolated => isolated,
        };
        let (backing, stat) = if let Some(stat) = stat {
            (backing, stat)
        } else if backing == Classification::Isolated
            && host.as_ref().is_some_and(|stat| {
                stat.st_mode & libc::S_IFMT == libc::S_IFDIR
                    && self
                        .provenance
                        .lock()
                        .unwrap_or_else(std::sync::PoisonError::into_inner)
                        .matches(path, &crate::migration::ObjectStamp::from_stat(stat))
            })
        {
            (Classification::Host, host.expect("checked host directory"))
        } else {
            return Err(io::Error::from_raw_os_error(libc::ENOENT));
        };
        Ok(ResolvedEntry {
            path: path.to_path_buf(),
            backing,
            stat,
        })
    }

    fn create_backing(
        &self,
        snapshot: &ClassificationSnapshot,
        path: &Path,
        directory: bool,
    ) -> io::Result<Classification> {
        let backing = self.classify(snapshot, path, directory);
        if backing == Classification::Isolated {
            self.isolated.ensure_parent_directories(path)?;
        }
        Ok(backing)
    }

    fn classify(
        &self,
        snapshot: &ClassificationSnapshot,
        path: &Path,
        directory: bool,
    ) -> Classification {
        if self.isolate_ignored {
            snapshot.classify(path, directory)
        } else {
            Classification::Host
        }
    }

    fn open_file(&self, inode: INodeNo, flags: i32) -> io::Result<u64> {
        let path = self.path_for(inode)?;
        let snapshot = self.classifier.snapshot_for(&path);
        let entry = self.resolve(&snapshot, &path)?;
        let file = self
            .root(entry.backing)
            .open_entry(&path, open_flags(flags), Mode::empty())?;
        if flags & libc::O_TRUNC != 0 {
            self.record_changed(entry.backing, &path);
        }
        let handle = self.next_handle.fetch_add(1, Ordering::Relaxed);
        self.files
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .insert(
                handle,
                Arc::new(OpenFile {
                    file,
                    backing: entry.backing,
                    path,
                    access_mode: flags & libc::O_ACCMODE,
                    migration_quiesced: AtomicBool::new(flags & libc::O_ACCMODE == libc::O_RDONLY),
                }),
            );
        Ok(handle)
    }

    fn visible_entries(
        &self,
        directory: &Path,
        snapshot: &ClassificationSnapshot,
    ) -> io::Result<Vec<DirectoryEntry>> {
        let mut names = BTreeSet::new();
        for root in [&self.host, &self.isolated] {
            match root.entries(directory) {
                Ok(entries) => names.extend(entries),
                Err(error) if error.kind() == io::ErrorKind::NotFound => {}
                Err(error) => return Err(error),
            }
        }
        let mut entries = Vec::new();
        for name in names {
            let path = directory.join(&name);
            if let Ok(entry) = self.resolve(snapshot, &path) {
                entries.push(DirectoryEntry {
                    inode: self.inode_for(&entry),
                    kind: file_type(entry.stat.st_mode)?,
                    name,
                });
            }
        }
        Ok(entries)
    }

    fn record_created(&self, backing: Classification, path: &Path) -> io::Result<()> {
        if backing != Classification::Host {
            return Ok(());
        }
        let stat = self.host.stat(path)?;
        let mut provenance = self
            .provenance
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        provenance.record(path.to_path_buf(), &stat);
        let mut parent = path.parent();
        while let Some(path) = parent {
            if let Ok(stat) = self.host.stat(path) {
                provenance.refresh(path, &stat);
            }
            parent = path.parent();
        }
        Ok(())
    }

    fn record_changed(&self, backing: Classification, path: &Path) {
        if backing != Classification::Host {
            return;
        }
        let mut provenance = self
            .provenance
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if let Ok(stat) = self.host.stat(path) {
            provenance.refresh(path, &stat);
        }
    }

    fn has_unquiesced_descendant(&self, source: &Path) -> bool {
        self.files
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .values()
            .any(|open| {
                (open.path == source || open.path.starts_with(source))
                    && !open.migration_quiesced.load(Ordering::Acquire)
            })
    }

    fn wait_for_unquiesced_descendants(&self, source: &Path) -> bool {
        for _ in 0..100 {
            if !self.has_unquiesced_descendant(source) {
                return false;
            }
            std::thread::sleep(Duration::from_millis(1));
        }
        self.has_unquiesced_descendant(source)
    }

    fn reply_entry(&self, path: &Path, reply: ReplyEntry) {
        let snapshot = self.classifier.snapshot_for(path);
        match self
            .resolve(&snapshot, path)
            .and_then(|entry| self.attributes(&entry))
        {
            Ok(attributes) => reply.entry(&TTL, &attributes, Generation(0)),
            Err(error) => reply.error(errno(&error)),
        }
    }

    fn attributes(&self, entry: &ResolvedEntry) -> io::Result<FileAttr> {
        let stat = entry.stat;
        Ok(FileAttr {
            ino: self.inode_for(entry),
            size: u64::try_from(stat.st_size)
                .map_err(|_| io::Error::other("negative file size"))?,
            blocks: u64::try_from(stat.st_blocks)
                .map_err(|_| io::Error::other("negative block count"))?,
            atime: system_time(stat.st_atime, stat.st_atime_nsec),
            mtime: system_time(stat.st_mtime, stat.st_mtime_nsec),
            ctime: system_time(stat.st_ctime, stat.st_ctime_nsec),
            crtime: UNIX_EPOCH,
            kind: file_type(stat.st_mode)?,
            perm: (stat.st_mode & 0o7777) as u16,
            nlink: u32::try_from(stat.st_nlink).unwrap_or(u32::MAX),
            uid: stat.st_uid,
            gid: stat.st_gid,
            rdev: u32::try_from(stat.st_rdev).unwrap_or(u32::MAX),
            blksize: u32::try_from(stat.st_blksize).unwrap_or(u32::MAX),
            flags: 0,
        })
    }

    fn with_file<T>(
        &self,
        handle: FileHandle,
        operation: impl FnOnce(&OpenFile) -> io::Result<T>,
    ) -> io::Result<T> {
        let open = self
            .files
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .get(&handle.0)
            .cloned()
            .ok_or_else(|| io::Error::from_raw_os_error(libc::EBADF))?;
        operation(&open)
    }

    fn posix_lock_file(&self, open: &OpenFile, owner: LockOwner) -> io::Result<Arc<File>> {
        let stat = fstat(open.file.as_fd())?;
        let key = PosixLockKey {
            backing: open.backing,
            device: stat.st_dev,
            inode: stat.st_ino,
            owner: owner.0,
        };
        let mut files = self
            .posix_lock_files
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner);
        if let Some(file) = files.get(&key) {
            return Ok(Arc::clone(file));
        }
        let mut options = OpenOptions::new();
        match open.access_mode {
            libc::O_RDONLY => {
                options.read(true);
            }
            libc::O_WRONLY => {
                options.write(true);
            }
            libc::O_RDWR => {
                options.read(true).write(true);
            }
            _ => return Err(io::Error::from_raw_os_error(libc::EINVAL)),
        }
        let file = Arc::new(
            options
                .custom_flags(libc::O_CLOEXEC)
                .open(format!("/proc/self/fd/{}", open.file.as_raw_fd()))?,
        );
        let reopened = fstat(file.as_fd())?;
        if (reopened.st_dev, reopened.st_ino) != (stat.st_dev, stat.st_ino) {
            return Err(io::Error::other("lock descriptor identity changed"));
        }
        files.insert(key, Arc::clone(&file));
        Ok(file)
    }

    fn release_posix_locks(&self, open: &OpenFile, owner: LockOwner) -> io::Result<()> {
        let stat = fstat(open.file.as_fd())?;
        self.posix_lock_files
            .lock()
            .unwrap_or_else(std::sync::PoisonError::into_inner)
            .remove(&PosixLockKey {
                backing: open.backing,
                device: stat.st_dev,
                inode: stat.st_ino,
                owner: owner.0,
            });
        Ok(())
    }
}

mod nodes;
mod operations;

#[cfg(test)]
mod tests;