//! 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.

//!

//! The working repository (`.oxy/`) is not a `Store`; it is read and

//! written directly, as before. A store is what a *remote* is kept on.

//! 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,

}

#[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<()>;

    // ---- 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) {

                out.push(Entry {

                    key,

                    size: entry.metadata()?.len(),

                });

            }

        }

        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()),

            }

        }

    }

}

impl Store for LocalStore {

    fn describe(&self) -> String {

        self.root.display().to_string()

    }

    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 {

            return Err(Error::Backend(format!(

                "{key}: bytes {}..{} asked of an object of {len} bytes",

                range.start, range.end

            )));

        }

        f.seek(SeekFrom::Start(range.start))?;

        let mut out = vec![0u8; (range.end - range.start) as usize];

        f.read_exact(&mut out)?;

        Ok(out)

    }

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

        let path = self.path(key)?;

        if path.exists() {

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

        }

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

            fs::create_dir_all(dir)?;

        }

        // Written beside its name, then linked into it: the link fails if

        // the name was taken meanwhile, and a reader never sees a part.

        let tmp = self.temp_beside(&path);

        let written = (|| {

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

            f.write_all(bytes)?;

            f.sync_all()

        })();

        let linked = written.and_then(|()| 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()),

        }

    }

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

        let mut out = Vec::new();

        self.list_dir(&self.root, "", prefix, &mut out)?;

        out.sort_by(|a, b| a.key.cmp(&b.key));

        Ok(out)

    }

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

        let path = self.path(key)?;

        match fs::remove_file(&path) {

            Ok(()) => {}

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

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

        }

        // A directory left empty goes too, up to the root, so the store

        // holds nothing but its objects.

        let mut dir = path.parent();

        while let Some(d) = dir {

            if d == self.root || fs::remove_dir(d).is_err() {

                break;

            }

            dir = d.parent();

        }

        Ok(())

    }

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

        let path = self.path(key)?;

        match fs::read(&path) {

            Ok(b) => {

                let v = content_version(&b);

                Ok(Some((b, v)))

            }

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

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

        }

    }

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

        let path = self.path(key)?;

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

            fs::create_dir_all(dir)?;

        }

        let lock = self.lock(key, &path)?;

        let result = (|| {

            let current = self.read_ref(key)?.map(|(_, v)| v);

            check_expected(key, current.as_ref(), expected)?;

            crate::durable::write_atomic(&path, bytes)?;

            Ok(content_version(bytes))

        })();

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

        result

    }

}

// ---------------------------------------------------------------- memory

/// A store in memory, for tests: the conditional writes behave as the

/// cloud stores' do.

#[derive(Default)]

pub struct MemoryStore {

    objects: Mutex<BTreeMap<String, Vec<u8>>>,

}

impl MemoryStore {

    pub fn new() -> MemoryStore {

        MemoryStore::default()

    }

    fn map(&self) -> std::sync::MutexGuard<'_, BTreeMap<String, Vec<u8>>> {

        self.objects.lock().unwrap_or_else(|p| p.into_inner())

    }

}

impl Store for MemoryStore {

    fn describe(&self) -> String {

        "memory".into()

    }

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

        check_key(key)?;

        Ok(self.map().contains_key(key))

    }

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

        check_key(key)?;

        self.map()

            .get(key)

            .cloned()

            .ok_or_else(|| Error::NotFound(key.to_string()))

    }

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

        let all = self.get(key)?;

        let len = all.len() as u64;

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

            return Err(Error::Backend(format!(

                "{key}: bytes {}..{} asked of an object of {len} bytes",

                range.start, range.end

            )));

        }

        Ok(all[range.start as usize..range.end as usize].to_vec())

    }

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

        check_key(key)?;

        let mut map = self.map();

        if map.contains_key(key) {

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

        }

        map.insert(key.to_string(), bytes.to_vec());

        Ok(())

    }

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

        Ok(self

            .map()

            .range(prefix.to_string()..)

            .take_while(|(k, _)| k.starts_with(prefix))

            .map(|(k, v)| Entry {

                key: k.clone(),

                size: v.len() as u64,

            })

            .collect())

    }

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

        check_key(key)?;

        self.map().remove(key);

        Ok(())

    }

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