//! Transport-agnostic protocol (spec §16): what a client asks of a remote and

//! what a server does with a push. `LocalRemote` serves another repository

//! directory in-process, `HttpRemote` speaks to an `oxy-server` over HTTP(S),

//! and the server side reuses `Repo::receive_push` and friends unchanged.

//!

//! Wire messages are deterministic CBOR (`wire` module); over HTTP each is a

//! request or response body with content type `application/oxy-cbor`.

use crate::attest::Attest;

use crate::cbor::{self, Value};

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

use crate::identity::Signature;

use crate::objects::{self as o, Change, ObjectError};

use crate::policy::Policy;

use crate::repo::{Repo, RepoError, Result};

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

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

pub const CONTENT_TYPE: &str = "application/oxy-cbor";

pub const PROTOCOL_VERSION: u64 = 1;

/// An attestation together with the signatures over its ObjectRef.

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

pub struct SignedAttest {

    pub attest: Attest,

    pub sigs: Vec<Signature>,

}

/// A policy together with the root signatures over its ObjectRef.

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

pub struct SignedPolicy {

    pub policy: Policy,

    pub sigs: Vec<Signature>,

}

impl SignedPolicy {

    /// The proposal-file encoding (`wire::policy_value`).

    pub fn encode(&self) -> Vec<u8> {

        crate::cbor::encode(&wire::policy_value(self))

    }

    pub fn decode(b: &[u8]) -> Result<SignedPolicy> {

        wire::policy_from(

            &crate::cbor::decode(b).map_err(|e| RepoError::Object(ObjectError::from(e)))?,

        )

    }

}

/// What `refs` returns: every bookmark, every tag, and the current policy.

#[derive(Clone, Debug, Default)]

pub struct RefsResponse {

    pub refs: Vec<RemoteRef>,

    pub policy: Option<SignedPolicy>,

    pub tags: Vec<TagRef>,

    /// Earlier policies, ascending by serial, so a client behind by more

    /// than one update can verify each step (spec §14.5).

    pub history: Vec<SignedPolicy>,

    /// The bookmark a clone checks out (spec §16.1), when the remote names one.

    pub default_bookmark: Option<String>,

    /// The host's latest transparency-log checkpoint, a signed note (spec

    /// §14.8), against which `RemoteRef::logged` proofs are made (key 6,

    /// only when the host keeps a log).

    pub log: Option<String>,

    /// The witnesses the host asks to cosign its checkpoints: each one's

    /// name in cosignatures and its identity's bytes (key 7, only when it

    /// has any), which a client pins (spec §14.8, *Integrated time*).

    pub witnesses: Vec<(String, Vec<u8>)>,

}

/// Where a reference attestation sits in the host's transparency log

/// (spec §14.8): its entry number, the entry, and the inclusion proof

/// against the checkpoint served beside it.

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

pub struct Logged {

    pub index: u64,

    pub entry: Vec<u8>,

    pub proof: Vec<[u8; 32]>,

}

/// A checkpoint witnesses cosigned that includes a logged entry (spec

/// §14.8, *Integrated time*): the cosigned note, and the entry's inclusion

/// proof against it.

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

pub struct Integration {

    pub note: String,

    pub proof: Vec<[u8; 32]>,

}

/// A tag as a remote reports it (spec §10.1).

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

pub struct TagRef {

    pub name: String,

    pub state: ObjectRef,

    pub heads: Vec<ChangeHash>,

    pub attestation: Option<SignedAttest>,

    /// The acceptance of `attestation` by its signer's certificate issuer

    /// when a short-lived key signed it (spec §14.7; key 4, only when there

    /// is one): the time a fetch checks the certificate at.

    pub accepted: Option<SignedAttest>,

    /// The attestation's place in the host's transparency log (key 5,

    /// only when it has one), as a bookmark's.

    pub logged: Option<Logged>,

    /// The cosigned checkpoints that first included `logged`'s entry (key

    /// 6, only when there are any).

    pub integration: Vec<Integration>,

}

impl TagRef {

    /// A tag with no acceptance or log place (yet).

    pub fn new(

        name: String,

        state: ObjectRef,

        heads: Vec<ChangeHash>,

        attestation: Option<SignedAttest>,

    ) -> TagRef {

        TagRef {

            name,

            state,

            heads,

            attestation,

            accepted: None,

            logged: None,

            integration: vec![],

        }

    }

}

/// One reference as a remote reports it.

#[derive(Clone, Debug)]

pub struct RemoteRef {

    pub name: String,

    pub state: ObjectRef,

    pub heads: Vec<ChangeHash>,

    /// The latest attestation, only when it describes `state`. An unattested

    /// push (accepted while no trust policy exists) leaves it `None`.

    pub attestation: Option<SignedAttest>,

    /// Serial of the latest attestation ever recorded, stale or not, so the

    /// next attestation continues the sequence.

    pub last_serial: Option<u64>,

    /// The host's acceptance of `attestation` when a short-lived key signed

    /// it (spec §14.7): the time its certificate is checked at.

    pub accepted: Option<SignedAttest>,

    /// The attestation's place in the host's transparency log (key 6, only

    /// when it has one).

    pub logged: Option<Logged>,

    /// The cosigned checkpoints that first included `logged`'s entry (key

    /// 7, only when there are any): what its integrated time is read from.

    pub integration: Vec<Integration>,

}

/// Everything needed to add a set of changes to a repository.

#[derive(Clone, Debug, Default)]

pub struct Bundle {

    /// In dependency order, each with its individual signatures.

    pub changes: Vec<(Change, Vec<Signature>)>,

    /// Seals covering the changes.

    pub attests: Vec<SignedAttest>,

    /// Chunks, blobs, identities, and state objects referenced by the changes.

    pub objects: Vec<(ObjectType, Vec<u8>)>,

    /// Policies between the receiver's and the one sent with the push,

    /// ascending by serial (spec §14.5).

    pub policies: Vec<SignedPolicy>,

    /// Each change's hash, in the order of `changes`, when whoever made

    /// the bundle knows it: the store it came from, or the bytes it was

    /// decoded from. Never read from the wire. Empty when unknown.

    pub hashes: Vec<ChangeHash>,

    /// Changes as stored, with their signatures, in dependency order, for a

    /// fetch between repositories on one machine: their text blobs and

    /// chunks are among `objects`. Never on the wire.

    pub raw_changes: Vec<(Vec<u8>, Vec<Signature>)>,

    /// Certificates of short-lived keys that signed what the bundle carries

    /// (spec §14.7), their issuers' identities among `objects`.

    pub certificates: Vec<crate::cert::SignedCertificate>,

}

impl Bundle {

    pub fn is_empty(&self) -> bool {

        self.changes.is_empty() && self.attests.is_empty() && self.objects.is_empty()

    }

}

/// A requested bookmark move (spec §10.2, §16.3).

#[derive(Clone, Debug)]

pub struct RefUpdate {

    pub name: String,

    pub old_state: ObjectRef,

    pub new_state: ObjectRef,

    pub force: bool,

    pub attestation: Option<SignedAttest>,

}

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

pub struct RefResult {

    pub name: String,

    pub ok: bool,

    pub reason: String,

}

pub trait Remote {

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

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

    /// Changes in the closure of `want` but not of `have`, in dependency order.

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

    /// The changes with their signatures, seals, identities, and blobs; blobs

    /// larger than `max_blob` bytes are left out for lazy fetching (spec §13.5).

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

    /// Individual objects by reference (spec §16.1 `objects`); a blob comes

    /// with its chunks.

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

    /// A checkpoint of exactly `heads` (spec §16.1 `checkpoint`): the

    /// checkpoint object, its payload blob, and the blob's chunks.

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

    /// A checkpoint of exactly `heads` without its graph blobs (spec §16.1

    /// `checkpoint-index`, oxygen#137), which the client then asks for in

    /// parts with `objects`. A remote that cannot answer it sends the

    /// whole checkpoint, which does as well.

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

        self.checkpoint(heads)

    }

    /// The changes within `depth` dependency steps of `heads` (spec §17).

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

    /// Push changes, bookmark updates, new tags, and optionally a newer

    /// policy. The policy is applied first; a result named `policy` reports

    /// it, and each tag reports as `tags/<name>`.

    fn push(

        &self,

        bundle: Bundle,

        updates: Vec<RefUpdate>,

        policy: Option<SignedPolicy>,

        tags: Vec<TagRef>,

    ) -> Result<Vec<RefResult>>;

}

// ---------------------------------------------------------------- local

/// Another repository on this machine, served in-process.

pub struct LocalRemote {

    pub repo: Repo,

    /// The pushing repository's id, which binds an unbound target on the

    /// first push, as `Oxy-Repo-Id` does over HTTP.

    pub claimed_id: Option<[u8; 16]>,

}

impl LocalRemote {

    pub fn open(path: &Path) -> Result<LocalRemote> {

        Ok(LocalRemote {

            repo: Repo::open(path)?,

            claimed_id: None,

        })

    }

}

impl Remote for LocalRemote {

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

        self.repo.id()

    }

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

        self.repo.remote_refs()

    }

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

        self.repo.negotiate(have, want)

    }

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

        self.repo.bundle_raw_for(changes, max_blob)

    }

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

        self.repo.objects_for(refs)

    }

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

        self.repo.checkpoint_bundle_for(heads)

    }

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

        self.repo.checkpoint_index_for(heads)

    }

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

        self.repo.shallow_changes(heads, depth)

    }

    fn push(

        &self,

        bundle: Bundle,

        updates: Vec<RefUpdate>,

        policy: Option<SignedPolicy>,

        tags: Vec<TagRef>,

    ) -> Result<Vec<RefResult>> {

        if self.repo.is_unbound() {

            match self.claimed_id {

                Some(id) => self.repo.bind_id(id)?,

                None => {

                    return Err(RepoError::Msg(

                        "the remote is unbound and this client did not say which repository it is"

                            .into(),

                    ))

                }

            }

        }

        self.repo.receive_push(bundle, updates, policy, tags)

    }

}

// ---------------------------------------------------------------- http

/// How to reach a remote: credentials and trust.

#[derive(Clone, Debug, Default)]

pub struct RemoteOptions {

    /// Bearer token for pushes.

    pub token: Option<String>,

    /// PEM file with extra root certificates to trust for `https://` remotes

    /// (self-hosted servers); the system web roots are always trusted.

    pub ca: Option<PathBuf>,

    /// The local repository's id, sent as `Oxy-Repo-Id` so an unbound

    /// hosted repository can adopt it on the first push (spec §16.1).

    pub repo_id: Option<[u8; 16]>,

    /// Where a remote kept on a store (`dir://`, `s3://`, …) keeps its

    /// mirror and what it has seen: the remote's directory under

    /// `.oxy/remotes`. None (a clone, before there is a repository) means

    /// a temporary directory.

    pub cache_dir: Option<PathBuf>,

    /// Other tokens for the same hub, tried in turn when `token` is

    /// refused (401, 403 or 404): those of enclosing checkouts, when

    /// `token` came from the credential store rather than from this

    /// remote's own configuration (oxygen#112).

    pub fallback_tokens: Vec<String>,

}

/// Opens a store for a URL it understands, or passes (`None`). Stores that

/// need more than the standard library (the cloud ones) are added by the

/// program that links them, with [`register_store_opener`].

pub type StoreOpener =

    fn(&str, &RemoteOptions) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>>;

static STORE_OPENERS: Mutex<Vec<StoreOpener>> = Mutex::new(Vec::new());

pub fn register_store_opener(opener: StoreOpener) {

    STORE_OPENERS

        .lock()

        .unwrap_or_else(|p| p.into_inner())

        .push(opener);

}

/// A store for `url`, from `dir://` or any registered opener.

pub fn open_store(

    url: &str,

    options: &RemoteOptions,

) -> Option<crate::strata::Result<Box<dyn crate::strata::Store>>> {

    if let Some(path) = url.strip_prefix("dir://") {

        return Some(

            crate::strata::LocalStore::open(path)

                .map(|s| Box::new(s) as Box<dyn crate::strata::Store>),

        );

    }

    let openers = STORE_OPENERS.lock().unwrap_or_else(|p| p.into_inner());

    openers.iter().find_map(|open| open(url, options))

}

/// The request header carrying the client's repository id.

pub const REPO_ID_HEADER: &str = "Oxy-Repo-Id";

/// The id of a hosted repository that has not adopted one yet.

pub const UNBOUND_ID: [u8; 16] = [0; 16];

/// The request header in which a client declares the protocol features it

/// understands beyond the 0.5.0-era set: a comma-separated list, which a

/// server may tailor its answers to (docs/protocol.md, *Client

/// capabilities*). A request without it is a client of that era.

pub const FEATURES_HEADER: &str = "Oxy-Features";

/// The feature of a client that authorizes a tag signed by a short-lived,

/// certified key by its certificate at the time the host accepted it

/// (oxygen#129). A client without it refuses such a tag, and with it the

/// whole fetch, so servers leave those tags out of its `refs` (oxygen#131).

pub const FEATURE_CERTIFIED_TAGS: &str = "certified-tags";

/// The features this client declares in [`FEATURES_HEADER`].

pub const CLIENT_FEATURES: &[&str] = &[FEATURE_CERTIFIED_TAGS];

/// The `User-Agent` every request of this client sends: `oxy/<version>`.

pub fn user_agent() -> String {

    format!("oxy/{}", env!("CARGO_PKG_VERSION"))

}

/// Whether a [`FEATURES_HEADER`] value (absent: `None`) declares `feature`.

/// Entries are comma-separated, compared without surrounding blanks and

/// regardless of case; unknown ones are ignored.

pub fn declares_feature(header: Option<&str>, feature: &str) -> bool {

    header.is_some_and(|h| h.split(',').any(|f| f.trim().eq_ignore_ascii_case(feature)))

}

/// A remote served by `oxy-server` at `<base>/oxy/v1/...`.

pub struct HttpRemote {

    base: String,

    /// The token in use, and those still to try when it is refused.

    token: Mutex<Option<String>>,

    fallbacks: Mutex<Vec<String>>,

    repo_id: Option<[u8; 16]>,

    agent: ureq::Agent,

}

/// A TLS client configuration trusting the web roots plus `ca`, if given.

fn tls_client_config(ca: Option<&Path>) -> Result<Arc<rustls::ClientConfig>> {

    let mut roots = rustls::RootCertStore::empty();

    roots.extend(webpki_roots::TLS_SERVER_ROOTS.iter().cloned());

    if let Some(path) = ca {

        let pem = std::fs::read(path)

            .map_err(|e| RepoError::Msg(format!("reading CA file {}: {e}", path.display())))?;

        let certs = rustls_pemfile::certs(&mut pem.as_slice())

            .collect::<std::result::Result<Vec<_>, _>>()

            .map_err(|e| RepoError::Msg(format!("parsing CA file {}: {e}", path.display())))?;

        if certs.is_empty() {

            return Err(RepoError::Msg(format!(

                "CA file {} holds no certificates",

                path.display()

            )));

        }

        for c in certs {

            roots

                .add(c)

                .map_err(|e| RepoError::Msg(format!("CA file {}: {e}", path.display())))?;

        }

    }

    let cfg = rustls::ClientConfig::builder_with_provider(Arc::new(

        rustls::crypto::ring::default_provider(),

    ))

    .with_safe_default_protocol_versions()

    .map_err(|e| RepoError::Msg(e.to_string()))?

    .with_root_certificates(roots)

    .with_no_client_auth();

    Ok(Arc::new(cfg))

}

impl HttpRemote {

    pub fn new(base: &str, options: &RemoteOptions) -> Result<HttpRemote> {

        refuse_plain_token(base, options)?;

        let agent = ureq::AgentBuilder::new()

            .tls_config(tls_client_config(options.ca.as_deref())?)

            .user_agent(&user_agent())

            .build();

        Ok(HttpRemote {

            base: base.trim_end_matches('/').to_string(),

            token: Mutex::new(options.token.clone()),

            fallbacks: Mutex::new(options.fallback_tokens.iter().rev().cloned().collect()),

            repo_id: options.repo_id,

            agent,

        })

    }

    /// A request, tried again with the next fallback token while the

    /// hub refuses the one in use (oxygen#112).

    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> Result<Vec<u8>> {

        loop {

            let token = self.token.lock().unwrap_or_else(|p| p.into_inner()).clone();

            match self.call_once(post, endpoint, body, token.as_deref()) {

                Err(RepoError::Msg(m)) if token.is_some() && refused(&m) => {

                    let next = self

                        .fallbacks

                        .lock()

                        .unwrap_or_else(|p| p.into_inner())

                        .pop();

                    let Some(next) = next else {

                        return Err(RepoError::Msg(m));

                    };

                    crate::credentials::warn_once(&format!(

                        "{}: the stored token was refused; using the one an enclosing checkout has for this hub",

                        self.base

                    ));

                    *self.token.lock().unwrap_or_else(|p| p.into_inner()) = Some(next);

                }

                other => return other,

            }

        }

    }

    fn call_once(

        &self,

        post: bool,

        endpoint: &str,

        body: Option<&[u8]>,

        token: Option<&str>,

    ) -> Result<Vec<u8>> {

        let url = format!("{}/oxy/v1/{endpoint}", self.base);

        let mut req = if post {

            self.agent.post(&url)

        } else {

            self.agent.get(&url)

        };

        req = req

            .set("Accept", CONTENT_TYPE)

            .set(FEATURES_HEADER, &CLIENT_FEATURES.join(","));

        if let Some(t) = token {

            req = req.set("Authorization", &format!("Bearer {t}"));

        }

        if let Some(id) = &self.repo_id {

            req = req.set(REPO_ID_HEADER, &hex::encode(id));

        }

        let resp = match body {

            Some(b) => req.set("Content-Type", CONTENT_TYPE).send_bytes(b),

            None => req.call(),

        };

        match resp {

            Ok(r) => {

                let mut out = Vec::new();

                r.into_reader()

                    .read_to_end(&mut out)

                    .map_err(|e| RepoError::Msg(format!("{url}: {e}")))?;

                Ok(out)

            }

            Err(ureq::Error::Status(code, r)) => {

                let text = r.into_string().unwrap_or_default();

                // The hint comes from the status alone: the server's text

                // (a host's reason for refusing a push, several lines