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

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

}

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

}

/// One reference as a remote reports it.

#[derive(Clone, Debug)]

pub struct RemoteRef {

    pub name: String,

    pub state: ObjectRef,

    pub heads: Vec<ChangeHash>,

    pub attestation: Option<SignedAttest>,

}

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

}

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>;

    /// 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_for_filtered(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 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]>,

}

/// 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];

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

pub struct HttpRemote {

    base: String,

    token: Option<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> {

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

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

            .build();

        Ok(HttpRemote {

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

            token: options.token.clone(),

            repo_id: options.repo_id,

            agent,

        })

    }

    fn call(&self, post: bool, endpoint: &str, body: Option<&[u8]>) -> 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);

        if let Some(t) = &self.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();

                let msg = format!("{url}: HTTP {code}: {}", text.trim());

                Err(RepoError::Msg(with_hint(msg)))

            }

            // The transport error already names the URL.

            Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),

        }

    }

}

/// What to do about a failed request, recognised from its message.

pub fn hint_for(message: &str) -> Option<&'static str> {

    let m = message.to_ascii_lowercase();

    let hint = if m.contains("http 401") {

        "the server did not accept the token: set one with `oxy config remote.<remote>.token TOKEN` (a personal access token; pushing needs repo:write)"

    } else if m.contains("http 403") {

        "the token or account lacks permission: pushing needs a token with repo:write and write access to the repository"

    } else if m.contains("http 404") {

        "nothing at this URL: check the owner and repository name, or create the repository on the hub first"

    } else if m.contains("unknownissuer")

        || m.contains("unknown issuer")

        || m.contains("certificate")

        || m.contains("invalidcertificate")

    {

        "the server's certificate is not trusted: for a hub with a self-signed certificate, download it from /.well-known/oxyhub-ca.pem on the hub and run `oxy config remote.<remote>.ca FILE`"

    } else if m.contains("connection refused") {

        "nothing is listening there: check the URL and port, and that the server is running"

    } else if m.contains("dns")

        || m.contains("failed to lookup")

        || m.contains("name or service not known")

    {

        "the host name does not resolve: check the URL"

    } else if m.contains("timed out") || m.contains("timeout") {

        "the server did not answer in time: check the network, or try again"

    } else {

        return None;

    };

    Some(hint)

}

/// `message`, followed by what to do about it when that is known.

pub fn with_hint(message: String) -> String {

    match hint_for(&message) {

        Some(h) => format!("{message}\n  hint: {h}"),

        None => message,

    }

}

/// An authenticated GET of any URL on a host, with the same TLS trust and

/// token as remotes use (for hub APIs: proposals). Returns the body.

pub fn http_get(url: &str, options: &RemoteOptions) -> Result<Vec<u8>> {

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

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

        .build();

    let mut req = agent.get(url).set("Accept", "application/json");

    if let Some(t) = &options.token {

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

    }

    match req.call() {

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

            Err(RepoError::Msg(with_hint(format!(

                "{url}: HTTP {code}: {}",

                text.trim()

            ))))

        }

        // The transport error already names the URL.

        Err(e) => Err(RepoError::Msg(with_hint(e.to_string()))),

    }

}

impl Remote for HttpRemote {

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

        wire::decode_capabilities(&self.call(false, "capabilities", None)?)

    }

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

        Ok(wire::decode_refs(&self.call(false, "refs", None)?)?.1)

    }

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

        wire::decode_hashes(&self.call(

            true,

            "negotiate",

            Some(&wire::encode_negotiate(have, want)),

        )?)

    }

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

        wire::decode_bundle(&self.call(

            true,

            "fetch",

            Some(&wire::encode_fetch(changes, max_blob)),

        )?)

    }

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

        Ok(

            wire::decode_bundle(&self.call(true, "objects", Some(&wire::encode_hashes(refs)))?)?

                .objects,

        )

    }

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

        wire::decode_bundle(&self.call(true, "checkpoint", Some(&wire::encode_hashes(heads)))?)

    }

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

        wire::decode_hashes(&self.call(

            true,

            "shallow",

            Some(&wire::encode_shallow(heads, depth)),

        )?)

    }

    fn push(

        &self,

        bundle: Bundle,

        updates: Vec<RefUpdate>,

        policy: Option<SignedPolicy>,

        tags: Vec<TagRef>,

    ) -> Result<Vec<RefResult>> {

        wire::decode_push_response(&self.call(

            true,

            "push",

            Some(&wire::encode_push(

                &bundle,

                &updates,

                policy.as_ref(),

                &tags,

            )),

        )?)

    }

}

/// Open a remote by URL: a local path (optionally `file://`), or `http(s)://`.

pub fn open_remote(url: &str, options: &RemoteOptions) -> Result<Box<dyn Remote>> {

    if url.starts_with("http://") || url.starts_with("https://") {

        return Ok(Box::new(HttpRemote::new(url, options)?));

    }

    let path = url.strip_prefix("file://").unwrap_or(url);

    if path.contains("://") {

        return Err(RepoError::Msg(format!("unsupported remote URL {url:?}")));

    }

    let mut local = LocalRemote::open(Path::new(path))?;

    local.claimed_id = options.repo_id;

    Ok(Box::new(local))

}

// ---------------------------------------------------------------- wire

/// CBOR encodings of the protocol messages (spec §16.1).

pub mod wire {

    use super::*;

    fn schema(msg: &str) -> RepoError {

        RepoError::Object(ObjectError::Schema(msg.into()))

    }

    fn sigs_value(sigs: &[Signature]) -> Value {

        Value::Array(sigs.iter().map(|s| s.to_value()).collect())