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

}

/// What `refs` returns: every bookmark plus the current policy, if any.

#[derive(Clone, Debug, Default)]

pub struct RefsResponse {

    pub refs: Vec<RemoteRef>,

    pub policy: Option<SignedPolicy>,

}

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

}

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

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

    /// Push changes, bookmark updates, and optionally a newer policy. The

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

    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>, policy: Option<SignedPolicy>) -> Result<Vec<RefResult>>;

}

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

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

pub struct LocalRemote {

    pub repo: Repo,

}

impl LocalRemote {

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

        Ok(LocalRemote { repo: Repo::open(path)? })

    }

}

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]) -> Result<Bundle> {

        self.repo.bundle_for(changes)

    }

    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>, policy: Option<SignedPolicy>) -> Result<Vec<RefResult>> {

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

    }

}

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

}

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

pub struct HttpRemote {

    base: String,

    token: Option<String>,

    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(), 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}"));

        }

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

                Err(RepoError::Msg(format!("{url}: HTTP {code}: {}", text.trim())))

            }

            Err(e) => Err(RepoError::Msg(format!("{url}: {e}"))),

        }

    }

}

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]) -> Result<Bundle> {

        wire::decode_bundle(&self.call(true, "fetch", Some(&wire::encode_hashes(changes)))?)

    }

    fn push(&self, bundle: Bundle, updates: Vec<RefUpdate>, policy: Option<SignedPolicy>) -> Result<Vec<RefResult>> {

        wire::decode_push_response(&self.call(true, "push", Some(&wire::encode_push(&bundle, &updates, policy.as_ref())))?)

    }

}

/// 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:?}")));

    }

    Ok(Box::new(LocalRemote::open(Path::new(path))?))

}

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

    }

    fn sigs_from(v: &Value) -> Result<Vec<Signature>> {

        Ok(o::array(v, "signatures")?.iter().map(Signature::from_value).collect::<std::result::Result<Vec<_>, _>>()?)

    }

    fn hashes_value(hs: &[ObjectRef]) -> Value {

        Value::Array(hs.iter().map(|h| Value::Bytes(h.0.to_vec())).collect())

    }

    fn hashes_from(v: &Value) -> Result<Vec<ObjectRef>> {

        Ok(o::array(v, "hashes")?.iter().map(|h| o::oref(h, "hash")).collect::<std::result::Result<Vec<_>, _>>()?)

    }

    fn attest_value(sa: &SignedAttest) -> Value {

        Value::Array(vec![Value::Bytes(sa.attest.encode()), sigs_value(&sa.sigs)])

    }

    fn attest_from(v: &Value) -> Result<SignedAttest> {

        let a = o::array(v, "signed attest")?;

        if a.len() != 2 {

            return Err(schema("signed attest: expected [attest, sigs]"));

        }

        Ok(SignedAttest { attest: Attest::decode(o::bytes(&a[0], "attest")?)?, sigs: sigs_from(&a[1])? })

    }

    fn opt_attest_value(sa: &Option<SignedAttest>) -> Value {

        sa.as_ref().map(attest_value).unwrap_or(Value::Null)

    }

    fn opt_attest_from(v: &Value) -> Result<Option<SignedAttest>> {

        if v.is_null() { Ok(None) } else { attest_from(v).map(Some) }

    }

    fn opt_policy_value(sp: Option<&SignedPolicy>) -> Value {

        match sp {

            Some(sp) => Value::Array(vec![Value::Bytes(sp.policy.encode()), sigs_value(&sp.sigs)]),

            None => Value::Null,

        }

    }

    fn opt_policy_from(v: &Value) -> Result<Option<SignedPolicy>> {

        if v.is_null() {

            return Ok(None);

        }

        let a = o::array(v, "signed policy")?;

        if a.len() != 2 {

            return Err(schema("signed policy: expected [policy, sigs]"));

        }

        Ok(Some(SignedPolicy { policy: Policy::decode(o::bytes(&a[0], "policy")?)?, sigs: sigs_from(&a[1])? }))

    }

    fn map<'a>(v: &'a Value, what: &str) -> Result<&'a o::Map> {

        Ok(o::map(v, what)?)

    }

    fn field<'a>(m: &'a o::Map, k: u64, what: &str) -> Result<&'a Value> {

        Ok(o::field(m, k, what)?)

    }

    pub fn bundle_value(b: &Bundle) -> Value {

        Value::map([

            (

                0,

                Value::Array(b.changes.iter().map(|(c, s)| Value::Array(vec![Value::Bytes(c.encode()), sigs_value(s)])).collect()),

            ),

            (1, Value::Array(b.attests.iter().map(attest_value).collect())),

            (

                2,

                Value::Array(

                    b.objects.iter().map(|(t, body)| Value::Array(vec![Value::Text(t.name().into()), Value::Bytes(body.clone())])).collect(),

                ),

            ),

        ])

    }

    pub fn bundle_from_value(v: &Value) -> Result<Bundle> {

        let m = map(v, "bundle")?;

        let mut b = Bundle::default();

        for c in o::array(field(m, 0, "changes")?, "changes")? {

            let pair = o::array(c, "change entry")?;

            if pair.len() != 2 {

                return Err(schema("change entry: expected [change, sigs]"));

            }

            b.changes.push((Change::decode(o::bytes(&pair[0], "change")?)?, sigs_from(&pair[1])?));

        }

        for a in o::array(field(m, 1, "attests")?, "attests")? {

            b.attests.push(attest_from(a)?);

        }

        for ob in o::array(field(m, 2, "objects")?, "objects")? {

            let pair = o::array(ob, "object")?;

            if pair.len() != 2 {

                return Err(schema("object: expected [type, body]"));

            }

            let ty = ObjectType::from_name(o::text(&pair[0], "object type")?).ok_or_else(|| schema("unknown object type"))?;

            b.objects.push((ty, o::bytes(&pair[1], "object body")?.to_vec()));

        }

        Ok(b)

    }

    pub fn encode_bundle(b: &Bundle) -> Vec<u8> {

        cbor::encode(&bundle_value(b))

    }

    pub fn decode_bundle(bytes: &[u8]) -> Result<Bundle> {

        bundle_from_value(&cbor::decode(bytes).map_err(ObjectError::from)?)

    }

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

        cbor::encode(&Value::map([

            (0, Value::Array(vec![Value::Uint(PROTOCOL_VERSION)])),

            (1, Value::Bytes(id.to_vec())),

            (2, Value::Array(vec![Value::Text("seal".into())])),

        ]))

    }

    pub fn decode_capabilities(bytes: &[u8]) -> Result<[u8; 16]> {

        let v = cbor::decode(bytes).map_err(ObjectError::from)?;

        let m = map(&v, "capabilities")?;

        let versions = o::array(field(m, 0, "versions")?, "versions")?;

        if !versions.iter().any(|x| x.as_uint() == Some(PROTOCOL_VERSION)) {

            return Err(RepoError::Msg("remote does not speak protocol version 1".into()));

        }

        o::bytes(field(m, 1, "repo id")?, "repo id")?.try_into().map_err(|_| schema("repo id: expected 16 bytes"))

    }

    pub fn encode_refs(id: [u8; 16], resp: &RefsResponse) -> Vec<u8> {

        cbor::encode(&Value::map([

            (

                0,

                Value::Array(

                    resp.refs

                        .iter()

                        .map(|r| {

                            Value::map([

                                (0, Value::Text(r.name.clone())),

                                (1, Value::Bytes(r.state.0.to_vec())),

                                (2, hashes_value(&r.heads)),

                                (3, opt_attest_value(&r.attestation)),

                                (4, r.last_serial.map(Value::Uint).unwrap_or(Value::Null)),

                            ])

                        })

                        .collect(),

                ),

            ),

            (1, Value::Bytes(id.to_vec())),

            (2, opt_policy_value(resp.policy.as_ref())),

        ]))

    }

    pub fn decode_refs(bytes: &[u8]) -> Result<([u8; 16], RefsResponse)> {

        let v = cbor::decode(bytes).map_err(ObjectError::from)?;

        let m = map(&v, "refs")?;

        let id: [u8; 16] = o::bytes(field(m, 1, "repo id")?, "repo id")?.try_into().map_err(|_| schema("repo id: expected 16 bytes"))?;

        let mut refs = Vec::new();

        for r in o::array(field(m, 0, "refs")?, "refs")? {

            let rm = map(r, "ref")?;

            let ls = field(rm, 4, "last serial")?;

            refs.push(RemoteRef {

                name: o::text(field(rm, 0, "name")?, "name")?.to_owned(),