haematite_cli/serve/query/

continuation.rs

use haematite_query::page::{Origin, Resume};
use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;

use super::{Binding, Seal, codec};
use crate::serve::contracts::identifier;
use crate::serve::protocol::Version;
use crate::serve::storage::Failure;

const FORMAT: u32 = 1;

#[derive(Debug)]
pub struct Continuation {
    pub branch: String,
    pub version: Version,
    pub resume: Resume,
    /// Later pages use this value; only the first page reads the clock.
    pub now: Box<RawValue>,
    binding: Binding,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Cursor {
    records_read: u64,
    pending: Origin,
    discard_outputs: u64,
    passed: bool,
    #[serde(default)]
    passed_counts: Vec<u64>,
}

#[derive(Deserialize)]
#[serde(deny_unknown_fields)]
struct Wire {
    format: u32,
    branch: String,
    version: Version,
    resume: Cursor,
    now: Box<RawValue>,
    binding: Binding,
}

#[derive(Serialize)]
struct CursorView<'a> {
    records_read: u64,
    pending: &'a Origin,
    discard_outputs: u64,
    passed: bool,
    #[serde(skip_serializing_if = "<[u64]>::is_empty")]
    passed_counts: &'a [u64],
}

#[derive(Serialize)]
struct WireView<'a> {
    format: u32,
    branch: &'a str,
    version: Version,
    resume: CursorView<'a>,
    now: &'a RawValue,
    binding: &'a Binding,
}

impl Continuation {
    /// Construct and validate a captured query continuation.
    ///
    /// # Errors
    /// Returns `query_continuation_invalid` for an invalid branch, cursor or score, or propagates an invalid binding.
    pub fn new(
        branch: String,
        version: Version,
        resume: Resume,
        now: Box<RawValue>,
        binding: Binding,
    ) -> Result<Self, Failure> {
        let token = Self {
            branch,
            version,
            resume,
            now,
            binding,
        };
        token.validate()?;
        Ok(token)
    }

    fn validate(&self) -> Result<(), Failure> {
        identifier(&self.branch).map_err(|_| codec::invalid())?;
        // A page that read only records the caller may not open passes a
        // place having given none.
        if (self.resume.records_read == 0 && !self.resume.passed)
            || self
                .resume
                .pending
                .score
                .is_some_and(|score| !score.is_finite())
        {
            return Err(codec::invalid());
        }
        self.binding.validate()
    }

    /// The token sealed by `seal`: nothing in it can be read or changed by
    /// whoever holds it.
    ///
    /// # Errors
    ///
    /// As [`Self::plain`], and when the token cannot be sealed.
    pub fn encode(&self, seal: &Seal) -> Result<String, Failure> {
        seal.seal(&self.plain()?)
    }

    /// The token opened by `seal`, then checked as [`Self::from_plain`].
    ///
    /// # Errors
    ///
    /// `query_continuation_invalid` for a token `seal` did not seal or that
    /// was altered; then every refusal of [`Self::from_plain`].
    pub fn decode(encoded: &str, seal: &Seal) -> Result<Self, Failure> {
        Self::from_plain(&seal.open(encoded)?)
    }

    /// The token's own canonical bytes, before sealing.
    ///
    /// # Errors
    ///
    /// When the token is not valid.
    pub fn plain(&self) -> Result<Vec<u8>, Failure> {
        self.validate()?;
        let wire = WireView {
            format: FORMAT,
            branch: &self.branch,
            version: self.version,
            resume: CursorView {
                records_read: self.resume.records_read,
                pending: &self.resume.pending,
                discard_outputs: self.resume.discard_outputs,
                passed: self.resume.passed,
                passed_counts: &self.resume.passed_counts,
            },
            now: &self.now,
            binding: &self.binding,
        };
        serde_json::to_vec(&wire).map_err(|_| codec::invalid())
    }

    /// The token whose canonical bytes are `bytes`, opened and not yet checked
    /// against a request.
    ///
    /// # Errors
    ///
    /// `query_continuation_format` for another format, and
    /// `query_continuation_invalid` for bytes that are not a valid token in
    /// their canonical form.
    pub fn from_plain(bytes: &[u8]) -> Result<Self, Failure> {
        let wire: Wire = serde_json::from_slice(bytes).map_err(|_| codec::invalid())?;
        if wire.format != FORMAT {
            return Err(Failure::new(
                "query_continuation_format",
                "continuation format is unsupported",
            ));
        }
        let token = Self::new(
            wire.branch,
            wire.version,
            Resume {
                records_read: wire.resume.records_read,
                pending: wire.resume.pending,
                discard_outputs: wire.resume.discard_outputs,
                passed: wire.resume.passed,
                passed_counts: wire.resume.passed_counts,
            },
            wire.now,
            wire.binding,
        )?;
        if token.plain()? != bytes {
            return Err(codec::invalid());
        }
        Ok(token)
    }

    /// Check the continuation against the next page's request binding.
    ///
    /// # Errors
    /// Propagates invalid or changed binding component refusals.
    pub fn check_binding(&self, expected: &Binding) -> Result<(), Failure> {
        self.binding.check(expected)
    }

    /// Check the requested branch and optional version against the captured moment.
    ///
    /// # Errors
    /// Returns `query_continuation_moment` if either differs from the token.
    pub fn check_moment(&self, branch: &str, version: Option<Version>) -> Result<(), Failure> {
        if self.branch != branch || version.is_some_and(|version| version != self.version) {
            return Err(Failure::new(
                "query_continuation_moment",
                "continuation is bound to a different branch or captured version",
            ));
        }
        Ok(())
    }
}