haematite-cli 0.21.8

use std::collections::BTreeMap;

use haematite::tree::Hash;
use haematite_query::page::Allowance;
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;

use super::Rows;
use super::canonical;
use crate::serve::contracts::Caller;
use crate::serve::storage::Failure;

const DOMAIN: &[u8] = b"haem.query.continuation.binding\0";
const FORMAT: u32 = 1;

#[derive(Clone, Debug, Serialize)]
pub struct Engine {
    pub protocol: u32,
    pub worker: String,
    pub engine: String,
    pub jaq_core: String,
    pub jaq_std: String,
    pub jaq_json: String,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
struct Components {
    query: String,
    parameters: String,
    engine: String,
    caller: String,
    allowances: String,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(deny_unknown_fields)]
pub struct Binding {
    components: Components,
    aggregate: String,
}

#[derive(Serialize)]
struct PageLimits {
    rows: u64,
    bytes: u64,
    scan: u64,
    /// What each row carries: a token of one form is not one of the other.
    form: Rows,
}

fn invalid() -> Failure {
    Failure::new(
        "query_continuation_invalid",
        "continuation binding is not a canonical digest",
    )
}

fn digest(kind: &str, canonical: &[u8]) -> String {
    let mut bytes = DOMAIN.to_vec();
    bytes.extend_from_slice(&FORMAT.to_be_bytes());
    bytes.extend_from_slice(kind.as_bytes());
    bytes.push(0);
    bytes.extend_from_slice(canonical);
    Hash::of(&bytes).to_string()
}

fn serialized(kind: &str, value: &impl Serialize) -> Result<String, Failure> {
    let bytes = serde_json::to_vec(value).map_err(|_| invalid())?;
    Ok(digest(kind, &bytes))
}

fn canonical_digest(value: &str) -> bool {
    value.len() == 64
        && value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
}

impl Binding {
    /// Bind the caller's parameters; the fixed clock value travels in the token.
    ///
    /// # Errors
    /// Refuses parameters that cannot be canonicalized or components that cannot be serialized for their digests.
    pub fn new(
        query: &str,
        parameters: &BTreeMap<String, Box<RawValue>>,
        engine: &Engine,
        caller: &Caller,
        allowance: &Allowance,
        form: Rows,
    ) -> Result<Self, Failure> {
        let components = Components {
            query: serialized("query", &query)?,
            parameters: digest("parameters", &canonical::parameters(parameters)?),
            engine: serialized("engine", engine)?,
            caller: serialized("caller", caller)?,
            allowances: serialized(
                "allowances",
                &PageLimits {
                    rows: allowance.rows.get(),
                    bytes: allowance.bytes.get(),
                    scan: allowance.scan_records,
                    form,
                },
            )?,
        };
        let aggregate = serialized("aggregate", &components)?;
        Ok(Self {
            components,
            aggregate,
        })
    }

    /// Validate all component digests and their aggregate.
    ///
    /// # Errors
    /// Returns `query_continuation_invalid` for a non-canonical digest or an inconsistent aggregate.
    pub fn validate(&self) -> Result<(), Failure> {
        let components = &self.components;
        if [
            &components.query,
            &components.parameters,
            &components.engine,
            &components.caller,
            &components.allowances,
            &self.aggregate,
        ]
        .into_iter()
        .any(|value| !canonical_digest(value))
            || serialized("aggregate", components)? != self.aggregate
        {
            return Err(invalid());
        }
        Ok(())
    }

    /// Check that this binding belongs to the expected request context.
    ///
    /// # Errors
    /// Refuses invalid bindings or names the query, parameters, engine, caller or allowances that changed.
    pub fn check(&self, expected: &Self) -> Result<(), Failure> {
        self.validate()?;
        expected.validate()?;
        let actual = &self.components;
        let expected = &expected.components;
        for (kind, actual, expected) in [
            ("query", &actual.query, &expected.query),
            ("parameters", &actual.parameters, &expected.parameters),
            ("engine", &actual.engine, &expected.engine),
            ("caller", &actual.caller, &expected.caller),
            ("allowances", &actual.allowances, &expected.allowances),
        ] {
            if actual != expected {
                return Err(Failure::new(
                    &format!("query_continuation_{kind}"),
                    format!("continuation is bound to different {kind}"),
                ));
            }
        }
        Ok(())
    }
}