//! A repository kept on a [`Store`] (Oxygen Strata, docs/stores.md): what

//! `oxy push dir://…` or `oxy clone s3://…` talk to.

//!

//! The store holds three kinds of key:

//!

//! - `oxygen.toml`: the layout's format and the repository's id, written

//!   once by the first push;

//! - `bundles/<seq>-<hash>.bundle`: each push's changes, signatures,

//!   seals, objects and policies, as the wire protocol encodes a bundle,

//!   written once and named by content;

//! - `refs`: every bookmark with its heads and attestation, every tag, the

//!   policy and its history, encoded as the protocol's `refs` response,

//!   moved by compare-and-swap.

//!

//! A store cannot run code, so the client does what a hub does on a push,

//! against a **mirror**: a bare repository of its own, kept beside the

//! remote's other state under `.oxy/remotes/<name>/strata/`, brought level

//! with the store before every use (the record read, the bundles it has

//! not seen imported, the view set from the record), and then used as a

//! [`LocalRemote`] would be. A push runs the hub's own checks

//! (`receive_push`) on the mirror, uploads the bundle, and moves `refs`

//! from the version it read; if the record moved meanwhile, the mirror is

//! put back and the push reports it, exactly as a hub reports a stale

//! push. Nothing in the store is trusted: every object is re-hashed as it

//! is imported, and the fetch that follows verifies each bookmark's

//! attestation under the policy, as it does for any remote.

use super::{fail, Repo, RepoError, Result};

use crate::hash::{ChangeHash, ObjectRef, ObjectType};

use crate::objects::StateObject;

use crate::protocol::{

    wire, Bundle, RefResult, RefUpdate, RefsResponse, Remote, RemoteOptions, SignedPolicy, TagRef,

    UNBOUND_ID,

};

use crate::strata::{Error as StoreError, Store, Version};

use std::cell::RefCell;

use std::collections::BTreeSet;

use std::fs;

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

const MARKER_KEY: &str = "oxygen.toml";

const REFS_KEY: &str = "refs";

const BUNDLES_PREFIX: &str = "bundles/";

const FORMAT: u32 = 1;

impl From<StoreError> for RepoError {

    fn from(e: StoreError) -> RepoError {

        RepoError::Msg(e.to_string())

    }

}

/// The `oxygen.toml` marker: two lines, written once.

fn marker(id: [u8; 16]) -> Vec<u8> {

    format!("format = {FORMAT}\nrepo = \"{}\"\n", hex::encode(id)).into_bytes()

}

fn parse_marker(bytes: &[u8], what: &str) -> Result<[u8; 16]> {

    let text = String::from_utf8_lossy(bytes);

    let mut format = None;

    let mut repo = None;

    for line in text.lines() {

        let Some((k, v)) = line.split_once('=') else {

            continue;

        };

        let v = v.trim().trim_matches('"');

        match k.trim() {

            "format" => format = v.parse::<u32>().ok(),

            "repo" => {

                repo = hex::decode(v)

                    .ok()

                    .and_then(|b| <[u8; 16]>::try_from(b).ok())

            }

            _ => {}

        }

    }

    match (format, repo) {

        (Some(f), _) if f > FORMAT => fail(format!(

            "{what} holds a repository in Strata format {f}, newer than this oxy understands (format {FORMAT}): upgrade oxy"

        )),

        (Some(_), Some(id)) => Ok(id),

        _ => fail(format!("{what}: its oxygen.toml is not a Strata marker")),

    }

}

/// A remote kept on a store.

pub struct BucketRemote {

    store: Box<dyn Store>,

    /// Where the mirror lives: the remote's directory under `.oxy/remotes`,

    /// or a temporary directory when there is no repository yet (a clone),

    /// removed on drop.

    dir: PathBuf,

    temporary: bool,

    claimed_id: Option<[u8; 16]>,

    mirror: RefCell<Option<Repo>>,

}

impl Drop for BucketRemote {

    fn drop(&mut self) {

        if self.temporary {

            let _ = fs::remove_dir_all(&self.dir);

        }

    }

}

/// What a sync found: the record's version (to move it from) and the

/// repository id the store holds, if any.

struct Synced {

    version: Option<Version>,

    id: Option<[u8; 16]>,

}

impl BucketRemote {

    pub fn open(store: Box<dyn Store>, options: &RemoteOptions) -> Result<BucketRemote> {

        let (dir, temporary) = match &options.cache_dir {

            Some(d) => (d.clone(), false),

            None => {

                let mut nonce = [0u8; 6];

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

                (

                    std::env::temp_dir().join(format!(

                        "oxy-strata-{}-{}",

                        std::process::id(),

                        hex::encode(nonce)

                    )),

                    true,

                )

            }

        };

        Ok(BucketRemote {

            store,

            dir,

            temporary,

            claimed_id: options.repo_id,

            mirror: RefCell::new(None),

        })

    }

    fn what(&self) -> String {

        self.store.describe()

    }

    /// The repository id the store holds: `None` until a first push.

    fn stored_id(&self) -> Result<Option<[u8; 16]>> {

        match self.store.get(MARKER_KEY) {

            Ok(b) => Ok(Some(parse_marker(&b, &self.what())?)),

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

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

        }

    }

    fn mirror_dir(&self) -> PathBuf {

        self.dir.join("mirror")

    }

    fn seen_path(&self) -> PathBuf {

        self.dir.join("seen")

    }

    /// The bundles already imported into the mirror.

    fn seen(&self) -> BTreeSet<String> {

        fs::read_to_string(self.seen_path())

            .map(|s| s.lines().map(str::to_string).collect())

            .unwrap_or_default()

    }

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

        let mut seen = self.seen();

        if seen.insert(key.to_string()) {

            let text: String = seen.iter().map(|k| format!("{k}\n")).collect();

            crate::durable::write_atomic(&self.seen_path(), text.as_bytes())?;

        }

        Ok(())

    }

    /// Open the mirror for repository `id`, creating it on first use.

    fn open_mirror(&self, id: [u8; 16]) -> Result<()> {

        if self.mirror.borrow().is_some() {

            return Ok(());

        }

        let dir = self.mirror_dir();

        let repo = if dir.join(super::OXY_DIR).is_dir() {

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

            if repo.id()? != id {

                // The store was emptied and reused for another repository:

                // start the mirror over.

                drop(repo);

                fs::remove_dir_all(&self.dir)?;

                fs::create_dir_all(&self.dir)?;

                Repo::init_bare_with_id(&dir, id)?

            } else {

                repo

            }

        } else {

            fs::create_dir_all(&self.dir)?;

            Repo::init_bare_with_id(&dir, id)?

        };

        *self.mirror.borrow_mut() = Some(repo);

        Ok(())

    }

    fn with_mirror<T>(&self, f: impl FnOnce(&Repo) -> Result<T>) -> Result<T> {

        let m = self.mirror.borrow();

        match m.as_ref() {

            Some(repo) => f(repo),

            None => fail(format!(

                "{} holds no repository yet: push one to it first",

                self.what()

            )),

        }

    }

    /// Bring the mirror level with the store.

    fn sync(&self) -> Result<Synced> {

        let Some(id) = self.stored_id()? else {

            return Ok(Synced {

                version: None,

                id: None,

            });

        };

        self.open_mirror(id)?;

        let record = self.store.read_ref(REFS_KEY)?;

        let (resp, version) = match record {

            Some((bytes, version)) => {

                let (rid, resp) = wire::decode_refs(&bytes)?;

                if rid != id {

                    return fail(format!(

                        "{}: its refs record names a different repository than its oxygen.toml; the store is damaged",

                        self.what()

                    ));

                }

                (resp, Some(version))

            }

            None => (RefsResponse::default(), None),

        };

        self.with_mirror(|mirror| {

            // Another process may have brought the mirror level since this

            // handle read its head.

            mirror.refresh()?;

            // The policy first: bundles are then imported under it.

            if let Some(sp) = &resp.policy {

                let have = mirror.current_policy()?.map(|(_, p)| p.policy.serial);

                if have.is_none_or(|s| s < sp.policy.serial) {

                    mirror

                        .adopt_policy_chain(&resp.history, sp, "strata sync")

                        .map_err(|e| RepoError::Msg(format!("policy from {}: {e}", self.what())))?;

                }

            }

            self.import_new_bundles(mirror)?;

            self.set_view(mirror, &resp)?;

            Ok(())

        })?;

        Ok(Synced {

            version,

            id: Some(id),

        })

    }

    /// Import every bundle in the store the mirror has not seen. Bundles

    /// are named in the order they were pushed, but two pushes from the

    /// same base may interleave, so a bundle whose dependencies are not

    /// there yet is tried again after the others.

    fn import_new_bundles(&self, mirror: &Repo) -> Result<()> {

        let seen = self.seen();

        let mut pending: Vec<String> = self

            .store

            .list(BUNDLES_PREFIX)?

            .into_iter()

            .map(|e| e.key)

            .filter(|k| !seen.contains(k))

            .collect();

        let strict = mirror.current_policy()?.is_some();

        while !pending.is_empty() {

            let mut deferred = Vec::new();

            let mut last_error = None;

            let before = pending.len();

            for key in pending.drain(..) {

                let bytes = match self.store.get(&key) {

                    Ok(b) => b,

                    // Listed, then gone: a gc merged it into a newer

                    // bundle, which this listing holds too.

                    Err(StoreError::NotFound(_)) => continue,

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

                };

                let bundle = wire::decode_bundle(&bytes)

                    .map_err(|e| RepoError::Msg(format!("{}: {key}: {e}", self.what())))?;

                match mirror.import_bundle(&bundle, strict) {

                    Ok(_) => self.mark_seen(&key)?,

                    Err(RepoError::Msg(m)) if m.contains("neither stored nor in the bundle") => {

                        last_error = Some(m);

                        deferred.push(key);

                    }

                    Err(e) => return Err(RepoError::Msg(format!("{}: {key}: {e}", self.what()))),

                }

            }

            if deferred.len() == before {

                // No progress: something those bundles need was never

                // pushed, or a push is still in progress.

                return fail(format!(

                    "{}: {} cannot be imported ({}); a push may be in progress, so try again, else the store is missing changes",

                    self.what(),

                    deferred.join(", "),

                    last_error.unwrap_or_default()

                ));

            }

            pending = deferred;

        }

        Ok(())

    }

    /// Make the mirror's view say what the record says.

    fn set_view(&self, mirror: &Repo, resp: &RefsResponse) -> Result<()> {

        let (op_ref, op) = mirror.head_op()?;

        let mut view = op.view.clone();

        view.bookmarks.clear();

        for r in &resp.refs {

            for h in &r.heads {

                if !mirror.store.has(h) {

                    return fail(format!(

                        "{}: bookmark {:?} names change {}, which no bundle there holds; a push may be in progress, so try again",

                        self.what(),

                        r.name,

                        h.short(12)

                    ));

                }

            }

            let state = mirror.store.put_state(&StateObject::new(r.heads.clone()))?;

            if state != r.state {

                return fail(format!(

                    "{}: bookmark {:?} names a state that does not match its heads; the store is damaged",

                    self.what(),

                    r.name

                ));

            }

            if let Some(sa) = &r.attestation {

                let current = mirror.latest_ref_attest(&r.name)?.map(|(_, a)| a);

                if current.as_ref() != Some(sa) {

                    let a = mirror.store.put_attest(&sa.attest)?;

                    for sig in &sa.sigs {

                        mirror.store.put_signature(&a, sig)?;

                    }

                    mirror.set_ref_attest(&r.name, &a)?;

                }

            }

            view.bookmarks.insert(r.name.clone(), state);

        }

        let policy = mirror.current_policy()?.map(|(_, sp)| sp.policy);

        for t in &resp.tags {

            mirror.store.put_state(&StateObject::new(t.heads.clone()))?;

            mirror

                .adopt_tag(

                    &mut view,

                    t,

                    policy.as_ref(),

                    &mut super::tags::TagCheck::Mirrored,

                )

                .map_err(|e| RepoError::Msg(format!("{}: {}", self.what(), e.reason)))?;

        }

        if view != op.view {

            mirror.record_op(&[op_ref], "strata sync", view)?;

        }

        Ok(())

    }

    /// The record for the mirror's current references.

    fn record(&self, id: [u8; 16]) -> Result<Vec<u8>> {

        self.with_mirror(|mirror| {

            let mut resp = mirror.remote_refs()?;

            // A mirror has no checkout; the store's clone picks a bookmark

            // as any other remote's does.

            resp.default_bookmark = None;

            Ok(wire::encode_refs(id, &resp))

        })

    }

    /// Store a bundle under the next sequence number; returns its key.

    fn upload(&self, bundle: &Bundle) -> Result<String> {

        let bytes = wire::encode_bundle(bundle);

        let hash = blake3::hash(&bytes).to_hex();

        let seq = self.store.list(BUNDLES_PREFIX)?.len() + 1;

        let key = format!("{BUNDLES_PREFIX}{seq:08}-{}.bundle", &hash[..16]);

        match self.store.put_new(&key, &bytes) {

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

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

        }

        self.mark_seen(&key)?;

        Ok(key)

    }

}

impl Remote for BucketRemote {

    fn repo_id(&self) -> Result<[u8; 16]> {

        Ok(self.stored_id()?.unwrap_or(UNBOUND_ID))

    }

    fn refs(&self) -> Result<RefsResponse> {

        let synced = self.sync()?;

        if synced.id.is_none() {

            return Ok(RefsResponse::default());

        }

        self.with_mirror(|mirror| {

            let mut resp = mirror.remote_refs()?;

            resp.default_bookmark = None;

            Ok(resp)

        })

    }

    fn negotiate(&self, have: &[ChangeHash], want: &[ChangeHash]) -> Result<Vec<ChangeHash>> {

        self.with_mirror(|m| m.negotiate(have, want))

    }

    fn fetch(&self, changes: &[ChangeHash], max_blob: Option<u64>) -> Result<Bundle> {

        self.with_mirror(|m| m.bundle_raw_for(changes, max_blob))

    }

    fn objects(&self, refs: &[ObjectRef]) -> Result<Vec<(ObjectType, Vec<u8>)>> {

        self.with_mirror(|m| m.objects_for(refs))

    }

    fn checkpoint(&self, heads: &[ChangeHash]) -> Result<Bundle> {

        self.with_mirror(|m| m.checkpoint_bundle_for(heads))

    }

    fn shallow(&self, heads: &[ChangeHash], depth: u32) -> Result<Vec<ChangeHash>> {

        self.with_mirror(|m| m.shallow_changes(heads, depth))

    }

    fn push(

        &self,

        bundle: Bundle,

        updates: Vec<RefUpdate>,

        policy: Option<SignedPolicy>,

        tags: Vec<TagRef>,

    ) -> Result<Vec<RefResult>> {

        let mut synced = self.sync()?;

        let id = match synced.id {

            Some(id) => id,

            None => {

                // The first push binds the store to this repository. Two

                // first pushes race on the marker; the loser reads the

                // winner's and goes on only if it is the same repository.

                let Some(id) = self.claimed_id else {

                    return fail(format!(

                        "{} is empty, and this client did not say which repository it is",

                        self.what()

                    ));

                };

                match self.store.put_new(MARKER_KEY, &marker(id)) {

                    Ok(()) => {}

                    Err(StoreError::Exists(_)) => {}

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

                }

                synced = self.sync()?;

                match synced.id {

                    Some(stored) if stored == id => id,

                    _ => {

                        return fail(format!(

                            "{} was just bound to another repository by a concurrent push",

                            self.what()

                        ))

                    }