//! Oxygen Strata (docs/stores.md): the storage a repository can be kept

//! on, as a trait, so that a bucket, a directory or a hub's disk all look

//! the same to the code that pushes to and fetches from them.

//!

//! A store has two halves. **Objects** are written once and never

//! change: packs and their indexes, named by content, which any store

//! can hold safely. **References** are the one mutable record (bookmarks,

//! tags, the policy), updated only by compare-and-swap on a version the

//! store hands out: an ETag on S3, a generation on GCS, a content hash on

//! a directory. Two writers can therefore share a store with no lock

//! server: the loser of a race reads again, merges, and tries again.

//!

//! A repository's own objects (`.oxy/objects`, see [`crate::store`]) are

//! kept on a store too, a directory one, with the same keys a bucket

//! would hold; a remote is a store as well. This module holds the trait,

//! a directory store, an in-memory store for tests, and the conformance

//! check every store must pass. Cloud stores live in a crate of their

//! own, because of what they pull in.

use std::collections::BTreeMap;

use std::fs;

use std::io::{self, Read, Seek, SeekFrom, Write};

use std::ops::Range;

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

use std::sync::Mutex;

use std::time::{Duration, Instant};

/// The version of a reference record as the store reports it: opaque,

/// compared only for equality, and only ever valid for the store that

/// issued it. Writing the same bytes again may keep the version (S3's

/// ETag and a directory's content hash are both of the content): a

/// compare-and-swap guards against losing a *different* record, which is

/// all it needs to.

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct Version(pub String);

/// One key in a listing.

#[derive(Clone, Debug, PartialEq, Eq)]

pub struct Entry {

    pub key: String,

    pub size: u64,

    /// When the object was written, in wall milliseconds, where the store

    /// knows (garbage collection's grace period reads it).

    pub modified_ms: Option<u64>,

}

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

pub enum Error {

    #[error("{0}: not in the store")]

    NotFound(String),

    #[error("{0}: already in the store")]

    Exists(String),

    #[error("{0}: changed since it was read")]

    Conflict(String),

    #[error("bad store key {0:?}: {1}")]

    Key(String, &'static str),

    #[error("{0}: busy (another writer holds it)")]

    Busy(String),

    #[error(transparent)]

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

    #[error("{0}")]

    Backend(String),

}

pub type Result<T> = std::result::Result<T, Error>;

/// Longest key a store accepts; S3 allows 1024 bytes.

pub const MAX_KEY_LEN: usize = 1024;

/// Refuse a key that is not a plain relative path: `/`-separated

/// segments of `[A-Za-z0-9._-]`, none empty, none `.` or `..`. The same

/// rule on every store, so a key written to one can be read from another,

/// and so a key can never escape a directory store's root.

pub fn check_key(key: &str) -> Result<()> {

    let bad = |why| Error::Key(key.to_string(), why);

    if key.is_empty() {

        return Err(bad("empty"));

    }

    if key.len() > MAX_KEY_LEN {

        return Err(bad("longer than 1024 bytes"));

    }

    for seg in key.split('/') {

        if seg.is_empty() {

            return Err(bad(

                "an empty segment (a leading, trailing or doubled slash)",

            ));

        }

        if seg == "." || seg == ".." {

            return Err(bad("a . or .. segment"));

        }

        if !seg

            .bytes()

            .all(|b| b.is_ascii_alphanumeric() || b == b'.' || b == b'_' || b == b'-')

        {

            return Err(bad("a character outside A-Z a-z 0-9 . _ -"));

        }

    }

    Ok(())

}

/// Where a repository's packs and references are kept. Every method is

/// synchronous and may be called from several threads at once.

pub trait Store: Send + Sync {

    /// A name for messages: the URL, or the directory.

    fn describe(&self) -> String;

    // ---- objects: written once, never changed

    fn has(&self, key: &str) -> Result<bool>;

    /// The whole object; `NotFound` when there is none.

    fn get(&self, key: &str) -> Result<Vec<u8>>;

    /// Bytes `range` of an object; an error when the range runs past its end.

    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>>;

    /// Write an object that does not exist yet; `Exists` when it does, and

    /// then nothing is changed. Readers never see a partial object.

    fn put_new(&self, key: &str, bytes: &[u8]) -> Result<()>;

    /// Every key under `prefix` (a key prefix, not necessarily a whole

    /// segment), sorted, with sizes.

    fn list(&self, prefix: &str) -> Result<Vec<Entry>>;

    /// Remove an object; fine when it is already gone.

    fn delete(&self, key: &str) -> Result<()>;

    /// One object's listing entry without listing its neighbours; `None`

    /// when there is none.

    fn stat(&self, key: &str) -> Result<Option<Entry>> {

        Ok(self.list(key)?.into_iter().find(|e| e.key == key))

    }

    /// Write several objects that do not exist yet; one that does is left

    /// as it is, which is not an error (objects are named by content). A

    /// store that can batch its syncs does; this does them one at a time.

    fn put_many_new(&self, objects: &[(String, Vec<u8>)]) -> Result<()> {

        for (key, bytes) in objects {

            match self.put_new(key, bytes) {

                Ok(()) | Err(Error::Exists(_)) => {}

                Err(e) => return Err(e),

            }

        }

        Ok(())

    }

    /// `put_new` of a file's contents: a store on the same disk can move

    /// the file into place instead of copying it.

    fn put_new_file(&self, key: &str, path: &Path) -> Result<()> {

        self.put_new(key, &fs::read(path)?)

    }

    // ---- references: one small record, updated by compare-and-swap

    /// The record and its current version, or `None` when it was never

    /// written.

    fn read_ref(&self, key: &str) -> Result<Option<(Vec<u8>, Version)>>;

    /// Replace the record only if its version is still `expected`

    /// (`None`: only if it does not exist yet). `Conflict` otherwise, and

    /// then nothing is changed. Returns the new version.

    fn write_ref(&self, key: &str, bytes: &[u8], expected: Option<&Version>) -> Result<Version>;

}

/// A reference record's version on stores that have no native one: the

/// hash of its bytes, which two different records never share.

fn content_version(bytes: &[u8]) -> Version {

    Version(blake3::hash(bytes).to_hex().to_string())

}

fn check_expected(key: &str, current: Option<&Version>, expected: Option<&Version>) -> Result<()> {

    if current == expected {

        Ok(())

    } else {

        Err(Error::Conflict(key.to_string()))

    }

}

// ---------------------------------------------------------------- directory

/// A store in a directory: each key a file under the root. Create-only

/// writes go through a hard link, which the file system refuses when the

/// name is taken, and a reference is updated under a lock file beside it.

/// Several processes on one machine can share it; it is also what the

/// tests and `oxy remote check` use.

pub struct LocalStore {

    root: PathBuf,

}

/// How long a reference's lock may be held before it counts as abandoned

/// by a process that died, and how long a writer waits for it.

const LOCK_STALE: Duration = Duration::from_secs(60);

const LOCK_WAIT: Duration = Duration::from_secs(10);

impl LocalStore {

    /// Open the store at `root`, creating the directory if needed.

    pub fn open(root: impl Into<PathBuf>) -> Result<LocalStore> {

        let root = root.into();

        fs::create_dir_all(&root)?;

        Ok(LocalStore { root })

    }

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

        &self.root

    }

    fn path(&self, key: &str) -> Result<PathBuf> {

        check_key(key)?;

        Ok(self.root.join(key))

    }

    fn temp_beside(&self, path: &Path) -> PathBuf {

        let mut nonce = [0u8; 6];

        rand::RngCore::fill_bytes(&mut rand::rngs::OsRng, &mut nonce);

        let name = path

            .file_name()

            .map(|n| n.to_string_lossy().into_owned())

            .unwrap_or_default();

        path.with_file_name(format!(".{name}.tmp-{}", hex::encode(nonce)))

    }

    fn list_dir(&self, dir: &Path, rel: &str, prefix: &str, out: &mut Vec<Entry>) -> Result<()> {

        let entries = match fs::read_dir(dir) {

            Ok(e) => e,

            Err(e) if e.kind() == io::ErrorKind::NotFound => return Ok(()),

            Err(e) => return Err(e.into()),

        };

        for entry in entries {

            let entry = entry?;

            let name = entry.file_name().to_string_lossy().into_owned();

            // Temporary and lock files are not objects.

            if name.starts_with('.') {

                continue;

            }

            let key = if rel.is_empty() {

                name.clone()

            } else {

                format!("{rel}/{name}")

            };

            let ty = entry.file_type()?;

            if ty.is_dir() {

                // Descend only where the prefix could still match.

                let dir_prefix = format!("{key}/");

                if prefix.starts_with(&dir_prefix) || dir_prefix.starts_with(prefix) {

                    self.list_dir(&entry.path(), &key, prefix, out)?;

                }

            } else if ty.is_file() && key.starts_with(prefix) {

                let meta = entry.metadata()?;

                out.push(Entry {

                    key,

                    size: meta.len(),

                    modified_ms: modified_ms(&meta),

                });

            }

        }

        Ok(())

    }

    /// Take `path.lock`, waiting for another writer, and removing a lock

    /// a dead process left behind.

    fn lock(&self, key: &str, path: &Path) -> Result<PathBuf> {

        let lock = path.with_file_name(format!(

            ".{}.lock",

            path.file_name()

                .map(|n| n.to_string_lossy().into_owned())

                .unwrap_or_default()

        ));

        let started = Instant::now();

        loop {

            match fs::OpenOptions::new()

                .write(true)

                .create_new(true)

                .open(&lock)

            {

                Ok(mut f) => {

                    let _ = writeln!(f, "{}", std::process::id());

                    return Ok(lock);

                }

                Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {

                    if let Ok(m) = fs::metadata(&lock) {

                        let age = m.modified().ok().and_then(|t| t.elapsed().ok());

                        if age.is_some_and(|a| a > LOCK_STALE) {

                            let _ = fs::remove_file(&lock);

                            continue;

                        }

                    }

                    if started.elapsed() > LOCK_WAIT {

                        return Err(Error::Busy(key.to_string()));

                    }

                    std::thread::sleep(Duration::from_millis(10));

                }

                Err(e) => return Err(e.into()),

            }

        }

    }

}

/// A file's modification time as wall milliseconds.

fn modified_ms(meta: &fs::Metadata) -> Option<u64> {

    meta.modified()

        .ok()

        .and_then(|t| t.duration_since(std::time::UNIX_EPOCH).ok())

        .map(|d| d.as_millis() as u64)

}

impl LocalStore {

    /// Link `tmp` into `path`, create-only; the directory is synced and

    /// `tmp` removed either way. `Exists` when the name is taken.

    fn link_into_place(&self, key: &str, tmp: &Path, path: &Path) -> Result<()> {

        let linked = fs::hard_link(tmp, path);

        let _ = fs::remove_file(tmp);

        match linked {

            Ok(()) => {

                if let Some(dir) = path.parent() {

                    crate::durable::sync_dir(dir)?;

                }

                Ok(())

            }

            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {

                Err(Error::Exists(key.to_string()))

            }

            Err(e) => Err(e.into()),

        }

    }

    /// Create `path`'s directory, syncing the first directory that was

    /// made so its entry is durable too.

    fn ensure_dir(&self, path: &Path) -> Result<()> {

        let Some(dir) = path.parent() else {

            return Ok(());

        };

        if dir.is_dir() {

            return Ok(());

        }

        let mut made = dir.to_path_buf();

        while let Some(parent) = made.parent() {

            if parent.is_dir() || parent == self.root {

                break;

            }

            made = parent.to_path_buf();

        }

        fs::create_dir_all(dir)?;

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

            crate::durable::sync_dir(parent)?;

        }

        Ok(())

    }

}

impl Store for LocalStore {

    fn describe(&self) -> String {

        self.root.display().to_string()

    }

    fn stat(&self, key: &str) -> Result<Option<Entry>> {

        match fs::metadata(self.path(key)?) {

            Ok(m) if m.is_file() => Ok(Some(Entry {

                key: key.to_string(),

                size: m.len(),

                modified_ms: modified_ms(&m),

            })),

            Ok(_) => Ok(None),

            Err(e) if e.kind() == io::ErrorKind::NotFound => Ok(None),

            Err(e) => Err(e.into()),

        }

    }

    /// Every file written and synced (on every core), then each linked

    /// into place and each directory synced once: durable when this

    /// returns, as one at a time would be, with far fewer waits.

    fn put_many_new(&self, objects: &[(String, Vec<u8>)]) -> Result<()> {

        use rayon::prelude::*;

        let mut todo: Vec<(PathBuf, PathBuf, &[u8])> = Vec::new();

        for (key, bytes) in objects {

            let path = self.path(key)?;

            if path.exists() || todo.iter().any(|(p, ..)| *p == path) {

                continue;

            }

            self.ensure_dir(&path)?;

            todo.push((path.clone(), self.temp_beside(&path), bytes));

        }

        if todo.is_empty() {

            return Ok(());

        }

        crate::durable::step("write loose objects")?;

        todo.par_iter()

            .try_for_each(|(_, tmp, bytes)| -> io::Result<()> {

                let mut f = fs::File::create(tmp)?;

                f.write_all(bytes)?;

                f.sync_all()

            })?;

        crate::durable::step("rename loose objects")?;

        let mut dirs: std::collections::BTreeSet<PathBuf> = std::collections::BTreeSet::new();

        for (path, tmp, _) in &todo {

            let linked = fs::hard_link(tmp, path);

            let _ = fs::remove_file(tmp);

            match linked {

                Ok(()) => {

                    if let Some(d) = path.parent() {

                        dirs.insert(d.to_path_buf());

                    }

                }

                Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {}

                Err(e) => return Err(e.into()),

            }

        }

        dirs.par_iter()

            .try_for_each(|d| crate::durable::sync_dir(d))?;

        Ok(())

    }

    /// The file is linked into place when it is on this file system,

    /// else copied; it is removed either way.

    fn put_new_file(&self, key: &str, file: &Path) -> Result<()> {

        let path = self.path(key)?;

        if path.exists() {

            let _ = fs::remove_file(file);

            return Err(Error::Exists(key.to_string()));

        }

        self.ensure_dir(&path)?;

        match fs::hard_link(file, &path) {

            Ok(()) => {

                let _ = fs::remove_file(file);

                if let Some(dir) = path.parent() {

                    crate::durable::sync_dir(dir)?;

                }

                Ok(())

            }

            Err(e) if e.kind() == io::ErrorKind::AlreadyExists => {

                let _ = fs::remove_file(file);

                Err(Error::Exists(key.to_string()))

            }

            Err(_) => {

                // Another file system: copy beside the name, then link.

                let tmp = self.temp_beside(&path);

                let copied = (|| {

                    let mut src = fs::File::open(file)?;

                    let mut dst = fs::File::create(&tmp)?;

                    io::copy(&mut src, &mut dst)?;

                    dst.sync_all()

                })();

                let _ = fs::remove_file(file);

                copied?;

                self.link_into_place(key, &tmp, &path)

            }

        }

    }

    fn has(&self, key: &str) -> Result<bool> {

        Ok(self.path(key)?.is_file())

    }

    fn get(&self, key: &str) -> Result<Vec<u8>> {

        let path = self.path(key)?;

        match fs::read(&path) {

            Ok(b) => Ok(b),

            Err(e) if e.kind() == io::ErrorKind::NotFound => Err(Error::NotFound(key.to_string())),

            Err(e) => Err(e.into()),

        }

    }

    fn get_range(&self, key: &str, range: Range<u64>) -> Result<Vec<u8>> {

        let path = self.path(key)?;

        let mut f = match fs::File::open(&path) {

            Ok(f) => f,

            Err(e) if e.kind() == io::ErrorKind::NotFound => {

                return Err(Error::NotFound(key.to_string()))

            }

            Err(e) => return Err(e.into()),

        };

        let len = f.metadata()?.len();

        if range.end > len || range.start > range.end {