//! Repositories on disk (hub spec §4): naming, the registry of open

//! handles and their locks, and the page-ready views built from `oxy-core`.

//!

//! Every `oxy-core` call is synchronous and runs on the blocking pool under

//! the repository's lock (§16.2): pushes and other writes take the write

//! half, page reads the read half.

pub mod archive;

pub mod cache;

pub mod edits;

pub mod functions;

pub mod names;

pub mod proposals;

pub mod search;

pub mod view;

pub use cache::{Loaded, StateCache};

pub use names::RepoName;

use oxy_core::repo::Repo;

use std::collections::HashMap;

use std::path::{Path, PathBuf};

use std::sync::{Arc, Mutex};

use tokio::sync::{OwnedMutexGuard, OwnedRwLockReadGuard, OwnedRwLockWriteGuard, RwLock};

/// A repository's locks (oxygenhub#249). Readers never wait for a push or

/// a gc: the store gives every handle one head and finds what a repack

/// moved (oxy-core, "readers never wait"). Writers take `writer` in turn;

/// everyone holds `life` shared, and only deleting or moving the

/// repository holds it alone.

#[derive(Default)]

struct Locks {

    life: Arc<RwLock<()>>,

    writer: Arc<tokio::sync::Mutex<()>>,

}

/// A reader's hold: the repository is not deleted or moved under it.

pub type ReadGuard = OwnedRwLockReadGuard<()>;

/// A writer's hold: no other writer runs, and the repository stays put.

/// Readers carry on.

pub struct WriteGuard {

    _writer: OwnedMutexGuard<()>,

    _life: OwnedRwLockReadGuard<()>,

}

/// Deleting or moving a repository: nothing else runs on it.

pub type ExclusiveGuard = OwnedRwLockWriteGuard<()>;

/// Where repositories live and how their access is serialised.

pub struct Repos {

    root: PathBuf,

    locks: Mutex<HashMap<String, Arc<Locks>>>,

    cache: Arc<StateCache>,

    pub searches: search::SearchCache,

}

#[derive(Debug, thiserror::Error)]

pub enum RepoError {

    #[error("no such repository")]

    NotFound,

    #[error("{0} already exists")]

    Exists(String),

    #[error("{0}")]

    Oxy(#[from] oxy_core::repo::RepoError),

    #[error("{0}")]

    Store(#[from] oxy_core::store::StoreError),

    #[error("{0}")]

    Io(#[from] std::io::Error),

}

impl Repos {

    pub fn new(root: PathBuf) -> Repos {

        // States are bounded by their estimated bytes (`cache::budget`); the

        // count only stops many tiny states piling up. At 64 it was always

        // full and evicting long before the bytes were (#145).

        let briefs = root.parent().unwrap_or(&root).join("cache").join("summaries").join(root.file_name().unwrap_or_default());

        let cache = Arc::new(StateCache::new(256).with_briefs(briefs));

        Repos { root, locks: Mutex::new(HashMap::new()), cache, searches: search::SearchCache::default() }

    }

    pub fn cache(&self) -> &Arc<StateCache> {

        &self.cache

    }

    pub fn root(&self) -> &Path {

        &self.root

    }

    /// `<root>/<owner>/<name>`, holding `.oxy/` and nothing else.

    pub fn dir(&self, name: &RepoName) -> PathBuf {

        self.root.join(&name.owner).join(&name.name)

    }

    pub fn exists(&self, name: &RepoName) -> bool {

        self.dir(name).join(oxy_core::repo::OXY_DIR).is_dir()

    }

    /// Open a hosted repository; it is always bare.

    pub fn open(&self, name: &RepoName) -> Result<Repo, RepoError> {

        if !self.exists(name) {

            return Err(RepoError::NotFound);

        }

        Ok(Repo::open(&self.dir(name))?)

    }

    /// Create an empty bare repository. With no `id` it is **unbound**: the

    /// first push binds it to the pusher's repository id, so an existing

    /// repository keeps its identity when it moves here (hub spec §4.1).

    pub fn create(&self, name: &RepoName, id: Option<[u8; 16]>) -> Result<Repo, RepoError> {

        let dir = self.dir(name);

        if dir.exists() {

            return Err(RepoError::Exists(name.to_string()));

        }

        Ok(match id {

            Some(id) => Repo::init_bare_with_id(&dir, id)?,

            None => Repo::init_bare_unbound(&dir)?,

        })

    }

    /// `create` off the async workers (#335): a bare init makes about

    /// sixteen fsyncs, and on a busy disk each waits behind every other

    /// write, so a create once held a worker for 26 s and stalled the

    /// requests queued behind it.

    pub async fn create_async(self: &Arc<Self>, name: &RepoName, id: Option<[u8; 16]>) -> Result<(), RepoError> {

        let (repos, name) = (self.clone(), name.clone());

        tokio::task::spawn_blocking(move || repos.create(&name, id).map(drop))

            .await

            .map_err(|e| RepoError::Io(std::io::Error::other(e.to_string())))?

    }

    /// Create an empty, unbound bare repository whose objects and

    /// operation log are kept in the store at `url` (oxygenhub#96); the

    /// store's credentials must already be known to the opener.

    pub fn create_on(&self, name: &RepoName, url: &str) -> Result<Repo, RepoError> {

        let dir = self.dir(name);

        if dir.exists() {

            return Err(RepoError::Exists(name.to_string()));

        }

        let made = (|| -> Result<Repo, RepoError> {

            drop(Repo::init_bare_on(&dir, [0; 16], Some(url))?);

            // Unbound as `create` makes them: oxy-core's marker file in

            // place of the id, so the first push gives the id.

            let oxy = dir.join(oxy_core::repo::OXY_DIR);

            std::fs::remove_file(oxy.join("id"))?;

            std::fs::write(oxy.join("unbound"), b"")?;

            Ok(Repo::open(&dir)?)

        })();

        if made.is_err() {

            let _ = std::fs::remove_dir_all(&dir);

        }

        made

    }

    /// Fork `src` into `dest` (hub spec §4): a new bare repository with its

    /// own id, the source's objects hard-linked where the file system allows

    /// (copied otherwise; objects are immutable, so sharing them is safe),

    /// and the source's bookmarks. The trust policy is not copied: the

    /// fork's owner sets their own. Runs under the source's writer lock, so

    /// no push lands while its files are linked.

    pub async fn fork(&self, src: &RepoName, dest: &RepoName) -> Result<(), RepoError> {

        let guard = self.write_guard(src).await;

        let (from, to) = (self.dir(src), self.dir(dest));

        if !self.exists(src) {

            return Err(RepoError::NotFound);

        }

        if to.exists() {

            return Err(RepoError::Exists(dest.to_string()));

        }

        if Repo::open(&from).is_ok_and(|r| r.config.storage_url.is_some()) {

            return Err(RepoError::Io(std::io::Error::other(

                "a repository kept in a bucket cannot be forked yet: clone it and push the clone to a new repository",

            )));

        }

        tokio::task::spawn_blocking(move || {

            let _guard = guard;

            let made = (|| -> Result<(), RepoError> {

                drop(Repo::init_bare(&to)?);

                let objects = |d: &Path| d.join(oxy_core::repo::OXY_DIR).join("objects");

                link_tree(&objects(&from), &objects(&to))?;

                let source = Repo::open(&from)?;

                let fork = Repo::open(&to)?;

                let existing: Vec<String> = fork.bookmarks()?.into_iter().map(|b| b.name).collect();

                for b in source.bookmarks()? {

                    if existing.contains(&b.name) {

                        fork.bookmark_set(&b.name, &b.heads)?;

                    } else {

                        fork.bookmark_create(&b.name, Some(&b.heads))?;

                    }

                }

                Ok(())

            })();

            if made.is_err() {

                let _ = std::fs::remove_dir_all(&to);

            }

            made

        })

        .await

        .map_err(|e| RepoError::Io(std::io::Error::other(e.to_string())))?

    }

    /// Move a repository to another name (a transfer, #108): the directory,

    /// and the lock the old name held.

    pub fn rename(&self, from: &RepoName, to: &RepoName) -> Result<(), RepoError> {

        let (src, dest) = (self.dir(from), self.dir(to));

        if !src.exists() {

            return Err(RepoError::NotFound);

        }

        if dest.exists() {

            return Err(RepoError::Exists(to.to_string()));

        }

        if let Some(parent) = dest.parent() {

            std::fs::create_dir_all(parent)?;

        }

        std::fs::rename(&src, &dest)?;

        let mut locks = self.locks.lock().unwrap();

        if let Some(l) = locks.remove(&from.to_string()) {

            locks.insert(to.to_string(), l);

        }

        Ok(())

    }

    /// Move a repository to the trash (hub spec §4.1); a job empties it later.

    pub fn trash(&self, name: &RepoName) -> Result<PathBuf, RepoError> {

        let dir = self.dir(name);

        if !dir.exists() {

            return Err(RepoError::NotFound);

        }

        let trash = self.root.parent().unwrap_or(&self.root).join("trash").join(&name.owner);

        // A drop's name is `drops/ID`: its own directory under the owner's.

        let dest = trash.join(format!("{}-{}", name.name, crate::db::now_ms()));

        std::fs::create_dir_all(dest.parent().unwrap_or(&trash))?;

        std::fs::rename(&dir, &dest)?;

        Ok(dest)

    }

    fn lock(&self, name: &RepoName) -> Arc<Locks> {

        self.locks.lock().unwrap().entry(name.to_string()).or_default().clone()

    }

    /// Read while `f` runs on the blocking pool: never waiting for a

    /// writer, only for the repository being deleted or moved.

    pub async fn read<T: Send + 'static>(

        &self,

        name: &RepoName,

        f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static,

    ) -> Result<T, RepoError> {

        let guard = self.read_guard(name).await;

        let dir = self.dir(name);

        let exists = self.exists(name);

        let cache = self.cache.clone();

        tokio::task::spawn_blocking(move || {

            let _guard = guard;

            if !exists {

                return Err(RepoError::NotFound);

            }

            let repo = Repo::open(&dir)?;

            f(&repo, &cache)

        })

        .await

        .map_err(|e| RepoError::Io(std::io::Error::other(e)))?

    }

    /// Write while `f` runs on the blocking pool, after the writers ahead.

    pub async fn write<T: Send + 'static>(

        &self,

        name: &RepoName,

        f: impl FnOnce(&Repo, &StateCache) -> Result<T, RepoError> + Send + 'static,

    ) -> Result<T, RepoError> {

        let guard = self.write_guard(name).await;

        let dir = self.dir(name);

        let exists = self.exists(name);

        let cache = self.cache.clone();

        tokio::task::spawn_blocking(move || {

            let _guard = guard;

            if !exists {

                return Err(RepoError::NotFound);

            }

            let repo = Repo::open(&dir)?;

            f(&repo, &cache)

        })

        .await

        .map_err(|e| RepoError::Io(std::io::Error::other(e)))?

    }

    /// The writer's hold alone, for callers that open the repository

    /// themselves (the protocol server's push, a backup's last pass). The

    /// writer lock first, so a writer queued behind others holds nothing a

    /// delete would wait for.

    pub async fn write_guard(&self, name: &RepoName) -> WriteGuard {

        let locks = self.lock(name);

        let writer = locks.writer.clone().lock_owned().await;

        WriteGuard { _writer: writer, _life: locks.life.clone().read_owned().await }

    }

    pub async fn read_guard(&self, name: &RepoName) -> ReadGuard {

        self.lock(name).life.clone().read_owned().await

    }

    /// Deleting or moving the repository: waits for every reader and

    /// writer, and holds off new ones.

    pub async fn exclusive_guard(&self, name: &RepoName) -> ExclusiveGuard {

        self.lock(name).life.clone().write_owned().await

    }

}

/// Hard-link every file under `from` into the same place under `to`,

/// copying when a link is not possible (another file system). Temporary

/// files of writers in progress are skipped.

fn link_tree(from: &Path, to: &Path) -> std::io::Result<()> {

    std::fs::create_dir_all(to)?;

    for e in std::fs::read_dir(from)? {

        let e = e?;

        let name = e.file_name();

        let n = name.to_string_lossy();

        if n.starts_with("tmp-") || n.starts_with("sigs-tmp-") || n.contains(".tmp") {

            continue;

        }

        let (src, dst) = (e.path(), to.join(&name));

        if e.file_type()?.is_dir() {

            link_tree(&src, &dst)?;

        } else if !dst.exists() && std::fs::hard_link(&src, &dst).is_err() {

            std::fs::copy(&src, &dst)?;

        }

    }

    Ok(())

}