haematite_cli/door/

forward.rs

//! Studio's verbs and head follows, carried to their store's socket.
//!
//! The session's Lys pass is presented first (ACCESS-002 R2); a branch's
//! head is followed the same way, the store pushing each change
//! ([`follow`]).
//!
//! Each verb is one connection: hello, then `session.pass` with the
//! session's pass, then the verb. A pass the store refuses ends the request
//! there and the verb is never sent. `hello` touches no resource, so it is
//! answered with the store's own hello. `session.pass` belongs to the door,
//! so a browser can never present a pass of its own.

use std::path::Path;
use std::time::Duration;

use serde_json::value::RawValue;
use serde_json::{Value, json};

use crate::client::{Client, Ended, Refusal, Reply};
use crate::serve::pass::now;

/// What a carried verb came to.
#[derive(Debug)]
pub enum Carried {
    /// The bytes the store sent for `result`.
    Answered(Box<RawValue>),
    /// The store's refusal, or the door's for a verb it keeps to itself.
    Refused(Refusal),
}

/// Carry `method` with `params` to the store at `socket`, presenting
/// `token` first, each step bounded by `timeout`.
///
/// # Errors
///
/// [`Ended`] as the client names it: nothing asked, or, from the verb's
/// sending on, an outcome that is not known.
pub fn carry(
    socket: &Path,
    timeout: Duration,
    token: &str,
    method: &str,
    params: Option<&RawValue>,
) -> Result<Carried, Ended> {
    if method == "session.pass" {
        return Ok(Carried::Refused(Refusal {
            code: "invalid_operation".to_owned(),
            message: "session.pass is the door's own and is never carried".to_owned(),
            path: None,
            // The door refuses this itself; no store sent a receipt.
            verified: None,
        }));
    }
    let mut client = Client::connect(socket, timeout)?;
    if method == "hello" {
        return Ok(Carried::Answered(client.hello().to_owned()));
    }
    let pass = serde_json::value::to_raw_value(&json!({ "pass": token }))
        .map_err(|error| Ended::NotAsked(format!("the pass could not be written: {error}")))?;
    if let Reply::Refused(refusal) = client.call("session.pass", Some(&pass))? {
        return Ok(Carried::Refused(refusal));
    }
    Ok(match client.call(method, params)? {
        Reply::Answered(answer) => Carried::Answered(answer),
        Reply::Refused(refusal) => Carried::Refused(refusal),
    })
}

/// What asking a store to follow a branch's head came to.
pub enum Followed {
    /// The store took the follow and sent the head it starts from.
    Following(Following),
    /// The store refused the session's pass or the follow.
    Refused(Refusal),
}

/// A follow a store took: the first head it sent, and the connection every
/// later frame arrives on.
pub struct Following {
    /// The bytes the store sent for the first head's `result`.
    pub first: Box<RawValue>,
    client: Client,
    /// When the presented pass expires, in seconds since the epoch, as the
    /// store answered `session.pass`.
    expires: Option<u64>,
    timeout: Duration,
}

impl Following {
    /// The store's next frame of the follow: a head, or the refusal that
    /// ends it. The store sends a frame only when the head moves, so the
    /// read waits as long as that takes, bounded by the pass's expiry and
    /// `timeout` past it: the store ends every follow at that expiry, so a
    /// read still waiting then means a store that stopped answering. With no
    /// expiry known the read is unbounded.
    ///
    /// # Errors
    ///
    /// [`Ended`] as the client names it when no frame could be read.
    pub fn next_frame(&mut self) -> Result<Reply, Ended> {
        self.client.next_reply()
    }

    pub(super) fn disconnect_handle(&self) -> std::io::Result<std::os::unix::net::UnixStream> {
        self.client.follow_socket()
    }

    /// Bound every later read by the pass's expiry and `timeout` past it,
    /// set once when the follow starts. A store that has already closed the
    /// connection refuses the bound on some systems (macOS answers EINVAL);
    /// what it sent before closing is still read, and the read itself then
    /// names the closed connection, so that refusal is not the follow's end.
    fn bound_reads(&self) {
        let bound = self.expires.map(|expires| {
            let left = Duration::from_secs(expires.saturating_sub(now()));
            left.saturating_add(self.timeout)
        });
        if let Err(ended) = self.client.read_within(bound) {
            eprintln!("haematite door: a follow's read bound was not set: {ended}");
        }
    }
}

/// Ask the store at `socket` to follow `branch`'s head, presenting `token`
/// first, each step up to the first head bounded by `timeout`.
///
/// # Errors
///
/// [`Ended`] as the client names it. A follow changes nothing, so an
/// unknown outcome only means no head was read.
pub fn follow(
    socket: &Path,
    timeout: Duration,
    token: &str,
    branch: &str,
) -> Result<Followed, Ended> {
    let mut client = Client::connect(socket, timeout)?;
    let pass = raw(&json!({ "pass": token }))?;
    let expires = match client.call("session.pass", Some(&pass))? {
        Reply::Refused(refusal) => return Ok(Followed::Refused(refusal)),
        Reply::Answered(answer) => expiry(&answer),
    };
    let params = raw(&json!({ "branch": branch }))?;
    Ok(match client.call("head.follow", Some(&params))? {
        Reply::Answered(first) => {
            let following = Following {
                first,
                client,
                expires,
                timeout,
            };
            following.bound_reads();
            Followed::Following(following)
        }
        Reply::Refused(refusal) => Followed::Refused(refusal),
    })
}

fn raw(value: &Value) -> Result<Box<RawValue>, Ended> {
    serde_json::value::to_raw_value(value)
        .map_err(|error| Ended::NotAsked(format!("the request could not be written: {error}")))
}

/// The expiry, in seconds since the epoch, a store answered `session.pass`
/// with: `{holder, expires}`.
fn expiry(answer: &RawValue) -> Option<u64> {
    let answer: Value = serde_json::from_str(answer.get()).ok()?;
    answer.get("expires")?.as_u64()
}

#[cfg(test)]
#[path = "forward_tests.rs"]
mod tests;