//! Object store (spec §13.1, §13.3, §13.4): loose objects, one per key

//! under `loose/<hh>/<rest>`, and packs, `pack-<id>.pack` with a sorted

//! `pack-<id>.idx`, holding zstd-compressed objects. Reads check loose

//! objects first, then every pack; every read re-hashes.

//!

//! Everything is kept on a [`crate::strata::Store`] (docs/stores.md): a

//! directory, which is what `.oxy/objects` is, or a bucket, which is how a

//! repository lives in its owner's region with the hub as one client of

//! it. The keys are the same wherever the store is, so a bucket holds

//! exactly what a directory would, and a directory store is as fast as

//! the files were before there was a trait: a loose object is one file, a

//! packed one is one ranged read of the pack. What must be local, the

//! pack being streamed out and the derived `metas` cache, lives in a

//! scratch directory.

//!

//! Pack file: `OXYPACK1` then entries of

//! `type u8 | flags u8 (bit 0: zstd) | stored_len u32 | raw_len u32 | ref 32 | body`.

//! Index file: `OXYIDX01` then `count u32` then sorted entries of

//! `ref 32 | offset u64 | stored_len u32 | raw_len u32 | type u8 | flags u8`.

//!

//! Beside the objects: `nocompress/<hex>` marks a chunk packs keep raw;

//! `sig/<hh>/<rest>/<key>` holds one signature over an object, written

//! once per signer; `seal/<hh>/<rest>/<attest>` and

//! `receipt/<hh>/<rest>/<attest>` index the seals covering a change and

//! the receipts on a state. Older stores hold these as one list per

//! object (`sigs/`, `seals/`, `receipts/`), which are still read.

//! `pack-retired` records the packs and loose objects a repack replaced,

//! which stay for readers until a collection an hour later (oxygenhub#249).

use crate::attest::Attest;

use crate::hash::{hash_object, header, ObjectRef, ObjectType};

use crate::identity::{Identity, Signature};

use crate::objects::{Blob, Change, ObjectError, StateObject};

use crate::receipt::Receipt;

use crate::strata::{self, Error as BackendError, LocalStore, MemoryStore};

use rand::RngCore;

use std::cell::RefCell;

use std::fs;

use std::io::{self, Write};

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

use std::sync::Arc;

const PACK_MAGIC: &[u8; 8] = b"OXYPACK1";

const IDX_MAGIC: &[u8; 8] = b"OXYIDX01";

const IDX_ENTRY: usize = 32 + 8 + 4 + 4 + 1 + 1;

const FLAG_ZSTD: u8 = 1;

/// Compressed with the pack's trained dictionary (`pack-ID.dict`), which a

/// full repack makes from a sample of the objects it packs: small files

/// share most of their vocabulary, which each one alone cannot use.

const FLAG_DICT: u8 = 2;

/// The size of a pack's dictionary, and the least a repack must pack for

/// one to pay.

const DICT_BYTES: usize = 110 << 10;

const DICT_MIN_OBJECTS: usize = 1000;

/// Objects up to this size are compressed with the dictionary.

const DICT_OBJECT_MAX: usize = 64 << 10;

const ZSTD_LEVEL: i32 = 3;

/// The level of a full repack (`oxy gc`): slower, once, for a smaller store.

pub const REPACK_ZSTD_LEVEL: i32 = 9;

/// Largest object a pack entry may claim; anything bigger is corruption,

/// and refusing it keeps a bad index from driving huge allocations.

const MAX_OBJECT_LEN: u32 = 1 << 30;

/// What reading an entry of a pack whose dictionary is gone says. A pack

/// being deleted can lose its dictionary first, so a read that meets this

/// looks at the packs again before it is believed.

const NO_DICT: &str = "packed with a dictionary the pack lacks";

/// The record of what repacks replaced (oxygenhub#249): packs and loose

/// objects whose contents a newer pack holds, kept for readers that

/// still use them until garbage collection deletes them, at least

/// [`RETIRED_FOR`] later. One line per key, `<retired at, wall ms> <key>`.

/// It is a reference record (updated by compare-and-swap, so concurrent

/// repacks and collections never lose each other's lines), named under

/// `pack-` so the pack listing every open makes shows whether it holds

/// anything.

const RETIRED_KEY: &str = "pack-retired";

/// How long a retired pack or loose object stays in the store: a reader

/// that loaded its list of packs before a repack reads on from the old

/// packs this long, and after that finds the objects in the new pack by

/// looking again (see [`Store::get`]).

pub const RETIRED_FOR: std::time::Duration = std::time::Duration::from_secs(60 * 60);

/// Compare-and-swap rounds for the retired record before giving up.

const RETIRED_TRIES: usize = 32;

/// After a look at the pack listing found nothing new, how long a missed

/// [`Store::has`] answers without looking again: a writer probing for

/// thousands of objects it is about to store would otherwise list the

/// packs for each.

const HAS_RELIST_AFTER: std::time::Duration = std::time::Duration::from_secs(1);

/// A pack listing as compared: each key with its size and time.

type Listed = Vec<(String, u64, Option<u64>)>;

/// The retired record's lines: each key with the time it was retired.

fn parse_retired(bytes: &[u8]) -> std::collections::BTreeMap<String, u64> {

    String::from_utf8_lossy(bytes)

        .lines()

        .filter_map(|l| {

            let (t, k) = l.split_once(' ')?;

            Some((k.to_string(), t.parse().ok()?))

        })

        .collect()

}

fn encode_retired(m: &std::collections::BTreeMap<String, u64>) -> Vec<u8> {

    m.iter()

        .map(|(k, t)| format!("{t} {k}\n"))

        .collect::<String>()

        .into_bytes()

}

/// What deleting retired packs and loose objects did.

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

pub struct RetiredReport {

    /// Retired packs deleted, each with its index and dictionary.

    pub packs: usize,

    /// Retired loose objects deleted.

    pub loose: usize,

    /// Retired entries younger than [`RETIRED_FOR`], left for the next

    /// collection.

    pub kept: usize,

}

/// An object read from the store: its type and body.

pub type Object = (ObjectType, Vec<u8>);

/// Where a long operation reports what it is doing (oxygen#66): a phase,

/// or a count within one, as a line for a spinner or a log.

pub type Progress<'a> = &'a mut dyn FnMut(&str);

/// The storage under a store: objects by key, see [`crate::strata`].

pub type Backend = dyn strata::Store;

fn silent() -> impl FnMut(&str) {

    |_| {}

}

/// A backend error as a store error: what is not found or already there

/// is handled where it matters; the rest is an I/O failure.

fn backend_err(e: BackendError) -> StoreError {

    match e {

        BackendError::Io(e) => StoreError::Io(e),

        other => StoreError::Io(io::Error::other(other.to_string())),

    }

}

/// `put_new` where an object already there is what we wanted.

fn put_new_or_existing(backend: &Backend, key: &str, bytes: &[u8]) -> Result<()> {

    match backend.put_new(key, bytes) {

        Ok(()) | Err(BackendError::Exists(_)) => Ok(()),

        Err(e) => Err(backend_err(e)),

    }

}

/// `get` as an option: `None` when there is no such key.

fn get_optional(backend: &Backend, key: &str) -> Result<Option<Vec<u8>>> {

    match backend.get(key) {

        Ok(b) => Ok(Some(b)),

        Err(BackendError::NotFound(_)) => Ok(None),

        Err(e) => Err(backend_err(e)),

    }

}

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

    backend.delete(key).map_err(backend_err)

}

/// `loose/<hh>/<rest>`.

fn loose_key(r: &ObjectRef) -> String {

    let h = r.to_hex();

    format!("loose/{}/{}", &h[..2], &h[2..])

}

/// The object a loose key names, if it is one.

fn loose_ref(key: &str) -> Option<ObjectRef> {

    let rest = key.strip_prefix("loose/")?;

    let (hh, tail) = rest.split_once('/')?;

    if hh.len() != 2 || tail.contains('/') {

        return None;

    }

    ObjectRef::from_hex(&format!("{hh}{tail}"))

}

/// `<index>/<hh>/<rest>/` for the entries kept under an object.

fn fanout_prefix(index: &str, r: &ObjectRef) -> String {

    let h = r.to_hex();

    format!("{index}/{}/{}/", &h[..2], &h[2..])

}

/// `<index>/<hh>/<rest>`: the one-file-per-object layout of older stores.

fn legacy_key(index: &str, r: &ObjectRef) -> String {

    let h = r.to_hex();

    format!("{index}/{}/{}", &h[..2], &h[2..])

}

fn now_ms() -> u64 {

    std::time::SystemTime::now()

        .duration_since(std::time::UNIX_EPOCH)

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

        .unwrap_or(0)

}

#[derive(Clone, Copy, Debug)]

struct PackEntry {

    offset: u64,

    stored_len: u32,

    raw_len: u32,

    ty: ObjectType,

    flags: u8,

}

/// One pack's index, fully in memory and sorted by ObjectRef.

struct PackIndex {

    /// The pack's key, `pack-<id>.pack`.

    key: String,

    refs: Vec<ObjectRef>,

    entries: Vec<PackEntry>,

    /// The pack's dictionary, prepared once, when it has one.

    dict: Option<Arc<zstd::dict::DecoderDictionary<'static>>>,

}

impl std::fmt::Debug for PackIndex {

    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

        f.debug_struct("PackIndex")

            .field("key", &self.key)

            .field("objects", &self.refs.len())

            .field("dict", &self.dict.is_some())

            .finish()

    }

}

/// Decode a stored pack entry and check it against its reference.

fn decode_entry(

    r: &ObjectRef,

    e: PackEntry,

    stored: Vec<u8>,

    dict: Option<&Arc<zstd::dict::DecoderDictionary<'static>>>,

) -> Result<Object> {

    let body = if e.flags & FLAG_DICT != 0 {

        let dict = dict.ok_or_else(|| StoreError::Corrupt(*r, NO_DICT.into()))?;

        zstd::bulk::Decompressor::with_prepared_dictionary(dict)

            .and_then(|mut d| d.decompress(&stored, e.raw_len as usize))

            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?

    } else if e.flags & FLAG_ZSTD != 0 {

        zstd::bulk::decompress(&stored, e.raw_len as usize)

            .map_err(|err| StoreError::Corrupt(*r, format!("zstd: {err}")))?

    } else {

        stored

    };

    if body.len() != e.raw_len as usize || hash_object(e.ty, &body) != *r {

        return Err(StoreError::Corrupt(

            *r,

            "packed object does not match its reference".into(),

        ));

    }

    Ok((e.ty, body))

}

impl PackIndex {

    /// Load a pack's index; `pack_len` is the pack's size, from the listing

    /// that found it (a stat of it would be a request of its own). `None`

    /// when the index is gone since the listing: the pack was deleted.

    fn load(

        backend: &Backend,

        idx_key: &str,

        pack_key: &str,

        pack_len: u64,

    ) -> Result<Option<PackIndex>> {

        let data = match backend.get(idx_key) {

            Ok(d) => d,

            Err(BackendError::NotFound(_)) => return Ok(None),

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

        };

        let bad = |m: &str| StoreError::Corrupt(ObjectRef::ZERO, format!("{idx_key}: {m}"));

        if data.len() < 12 || &data[..8] != IDX_MAGIC {

            return Err(bad("bad index header"));

        }

        let count = u32::from_le_bytes(data[8..12].try_into().unwrap()) as usize;

        if data.len() != 12 + count * IDX_ENTRY {

            return Err(bad("index length mismatch"));

        }

        let mut refs = Vec::with_capacity(count);

        let mut entries = Vec::with_capacity(count);

        for i in 0..count {

            let e = &data[12 + i * IDX_ENTRY..12 + (i + 1) * IDX_ENTRY];

            let r = ObjectRef::from_slice(&e[..32]).unwrap();

            if let Some(prev) = refs.last() {

                if r <= *prev {

                    return Err(bad("index not sorted"));

                }

            }

            refs.push(r);

            let entry = PackEntry {

                offset: u64::from_le_bytes(e[32..40].try_into().unwrap()),

                stored_len: u32::from_le_bytes(e[40..44].try_into().unwrap()),

                raw_len: u32::from_le_bytes(e[44..48].try_into().unwrap()),

                ty: ObjectType::ALL

                    .get(e[48] as usize)

                    .copied()

                    .ok_or_else(|| bad("unknown object type"))?,

                flags: e[49],

            };

            if entry.stored_len > MAX_OBJECT_LEN || entry.raw_len > MAX_OBJECT_LEN {

                return Err(StoreError::Corrupt(

                    r,

                    format!(

                        "{idx_key}: the index claims an object of {} bytes ({} stored), more than the {MAX_OBJECT_LEN} an object may be",

                        entry.raw_len, entry.stored_len

                    ),

                ));

            }

            if entry.offset.saturating_add(entry.stored_len as u64) > pack_len {

                return Err(bad("index entry points outside the pack"));

            }

            entries.push(entry);

        }

        let dict_key = format!("{}.dict", pack_key.trim_end_matches(".pack"));

        let dict = get_optional(backend, &dict_key)?

            .map(|d| Arc::new(zstd::dict::DecoderDictionary::copy(&d)));

        Ok(Some(PackIndex {

            key: pack_key.to_string(),

            refs,

            entries,

            dict,

        }))

    }

    fn find(&self, r: &ObjectRef) -> Option<PackEntry> {

        self.refs.binary_search(r).ok().map(|i| self.entries[i])

    }

    /// Read one entry from the pack: one ranged read, then decode.

    fn fetch(&self, backend: &Backend, r: &ObjectRef, e: PackEntry) -> Result<Object> {

        let stored = backend

            .get_range(&self.key, e.offset..e.offset + e.stored_len as u64)

            .map_err(|err| match err {

                BackendError::NotFound(_) => StoreError::NotFound(*r),

                other => backend_err(other),

            })?;

        decode_entry(r, e, stored, self.dict.as_ref())

    }

}

fn type_index(ty: ObjectType) -> u8 {

    ObjectType::ALL.iter().position(|t| *t == ty).unwrap() as u8

}

/// What `pack_loose` did.

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

pub struct PackReport {

    pub objects: usize,

    pub raw_bytes: u64,

    pub packed_bytes: u64,

}

/// Packs tolerated before `merge_packs` folds them into one (spec §13.4).

pub const MAX_PACKS: usize = 16;

/// What `prune` did.

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

pub struct PruneReport {

    pub objects: usize,

    pub bytes: u64,

    /// Unreachable objects left alone because they are younger than the grace period.

    pub deferred: usize,

}

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

pub enum StoreError {

    #[error("io: {0}")]

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

    #[error("corrupt object {0}: {1}")]

    Corrupt(ObjectRef, String),

    #[error("object {0}")]

    Object(#[from] ObjectError),

    #[error("object {0} is a {1}, expected {2}")]

    WrongType(ObjectRef, &'static str, &'static str),

    #[error("object {0} not found")]

    NotFound(ObjectRef),

    #[error("a {kind} object of {size} bytes is larger than the {limit} bytes an object may be; nothing was stored. A change this large adds or touches too many files at once: commit it in parts (a directory at a time)")]

    TooLarge {

        kind: &'static str,

        size: usize,

        limit: u32,

    },

}

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

pub struct Store {

    backend: Arc<Backend>,

    /// Temporary files (a pack being streamed) and derived caches (`metas`).

    /// For a store in a directory it is that directory.

    scratch: PathBuf,

    /// The backend is that directory, so the old file paths still hold.

    local: bool,

    packs: RefCell<Vec<Arc<PackIndex>>>,

    /// The pack listing `packs` was loaded from (key, size, time), to tell

    /// whether a look at the store would find anything new.

    packs_listed: RefCell<Listed>,

    /// When a look at the pack listing for a missed [`Store::has`] last

    /// found nothing new.

    has_relisted: std::cell::Cell<Option<std::time::Instant>>,

    /// An open write batch (spec §13.4): objects go here instead of to loose

    /// files, and are written out with one `fsync` when the batch finishes.

    pending: RefCell<Option<Pending>>,

    batch_depth: std::cell::Cell<u32>,

    /// Change metadata already read, from the `metas` cache file.

    metas: RefCell<Option<crate::fasthash::FastMap<ObjectRef, crate::state::ChangeMeta>>>,

}

impl std::fmt::Debug for Store {

    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {

        f.debug_struct("Store")

            .field("backend", &self.backend.describe())

            .field("scratch", &self.scratch)

            .field("packs", &self.packs.borrow().len())

            .finish()

    }

}

/// Objects written since `begin_batch`. Small batches stay in memory and

/// become loose objects; once `SPILL_OBJECTS` or `SPILL_BYTES` is exceeded

/// the batch streams into a pack file instead.

#[derive(Debug)]

struct Pending {

    buffered: Vec<(ObjectType, Vec<u8>)>,

    buffered_bytes: usize,

    lookup: std::collections::HashMap<ObjectRef, PendingEntry>,

    pack: Option<PendingPack>,

    report: PackReport,

}

#[derive(Debug, Clone, Copy)]

enum PendingEntry {

    Buffered(usize),

    Packed(PackEntry),

}

#[derive(Debug)]

struct PendingPack {

    id: String,

    tmp_path: PathBuf,

    writer: io::BufWriter<fs::File>,

    offset: u64,

    index: Vec<(ObjectRef, PackEntry)>,

}

const SPILL_OBJECTS: usize = 256;

const SPILL_BYTES: usize = 8 << 20;

/// Finishes the enclosing batch when dropped; call `finish` to see errors.

pub struct Batch<'a> {

    store: &'a Store,

    done: bool,

}

impl Batch<'_> {

    pub fn finish(mut self) -> Result<()> {

        self.done = true;

        self.store.finish_batch()

    }

}

impl Drop for Batch<'_> {

    fn drop(&mut self) {

        if !self.done {

            let _ = self.store.finish_batch();

        }

    }

}

/// Append one stored object to a pending pack and index it.

fn write_pack_entry(

    pack: &mut PendingPack,

    r: &ObjectRef,

    ty: ObjectType,

    raw_len: usize,

    flags: u8,

    stored: &[u8],

    report: &mut PackReport,

) -> Result<PackEntry> {

    let header_len = 1 + 1 + 4 + 4 + 32;

    pack.writer.write_all(&[type_index(ty), flags])?;

    pack.writer

        .write_all(&(stored.len() as u32).to_le_bytes())?;

    pack.writer.write_all(&(raw_len as u32).to_le_bytes())?;

    pack.writer.write_all(&r.0)?;

    pack.writer.write_all(stored)?;

    let entry = PackEntry {

        offset: pack.offset + header_len,

        stored_len: stored.len() as u32,

        raw_len: raw_len as u32,

        ty,

        flags,

    };

    pack.offset += header_len + stored.len() as u64;

    pack.index.push((*r, entry));

    report.objects += 1;

    report.raw_bytes += raw_len as u64;

    report.packed_bytes += stored.len() as u64;

    Ok(entry)

}

/// A random name for a temporary file or a pack.

fn fresh_id() -> String {

    let mut nonce = [0u8; 8];

    rand::rngs::OsRng.fill_bytes(&mut nonce);

    hex::encode(nonce)

}

/// Under `OXY_STORE_BACKEND=memory` every store opens on a store in

/// memory shared by directory, so the whole test suite can run without a

/// directory store, as it would over a bucket. Objects live for the

/// process.

fn memory_backend_for(dir: &Path) -> Option<Arc<Backend>> {

    use std::collections::HashMap;

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

    if std::env::var_os("OXY_STORE_BACKEND").as_deref() != Some(std::ffi::OsStr::new("memory")) {

        return None;