haematite-cli 0.21.1

//! `haematite write`: one change to a branch, and always whether it was applied.
//!
//! The change goes through `mutation.apply` under a request id. When the
//! outcome is not known after sending, the client connects once more and asks
//! for the receipt; a missing receipt is never read as "not applied".

use std::io::Read;
use std::path::Path;
use std::process::ExitCode;
use std::time::Duration;

use serde::{Deserialize, Serialize};
use serde_json::value::RawValue;

use super::{Client, Common, Ended, Reply, Text, finish, one_value, socket_of};

const USAGE: &str = "\
usage: haematite write (--socket <path> | --config <serve.json>) [--timeout-ms N] --branch <name>
                  [--expected <generation>.<sequence>] [--request-id <text>] <kind> <value | ->
<kind> is a change kind as the service names it (document.put, vector.put, documents.put_many,
vectors.put_many, edges.put_many, documents.delete_many and the rest); <value> is one JSON value, sent as the text given; - reads it from standard input.
branch.fork: --branch is the source, and the value names the new branch and the source version
it starts from: {\"name\":\"<new>\",\"from\":{\"generation\":\"<g>\",\"sequence\":\"<s>\"}}.
branch.discard: the value names the --branch itself, never another: {\"name\":\"<branch>\"}.
Without --expected, the change is applied to the branch as it stands when the command runs. A
caller who read a record and is changing it gives the version it read as --expected, so a change
made in between is refused rather than overwritten.
Without --request-id, one is made and printed as `haematite: request <id>` before sending. If the
outcome is not known, running the same command with that --request-id is safe to repeat.
Repeated after the change was applied, the same command answers with the same receipt; with
different content under the same --request-id it is refused.
--timeout-ms bounds connecting, writing and reading; the default is 30000.
Exit 0: applied, the receipt on standard output. Exit 1: could not ask; nothing was sent.
Exit 2: refused by the service, its code and message on standard error. Exit 3: the outcome
is not known; standard error gives the request id to repeat it with.";

/// The most branches asked for on one page while finding a head, as the
/// Studio asks; never more than the service's own page.
const BRANCH_PAGE: u32 = 200;

struct Write {
    target: super::Target,
    timeout: Duration,
    branch: String,
    expected: Option<Box<RawValue>>,
    request_id: Option<String>,
    kind: String,
    value: Text,
}

#[derive(Serialize)]
struct Apply<'a> {
    request_id: &'a str,
    branch: &'a str,
    expected: &'a RawValue,
    change: Change<'a>,
}

#[derive(Serialize)]
struct Change<'a> {
    kind: &'a str,
    value: &'a RawValue,
}

#[derive(Serialize)]
struct ReceiptQuery<'a> {
    branch: &'a str,
    request_id: &'a str,
}

#[derive(Serialize)]
struct BranchQuery<'a> {
    #[serde(skip_serializing_if = "Option::is_none")]
    after: Option<&'a str>,
    limit: u32,
}

#[derive(Deserialize)]
struct BranchPage {
    entries: Vec<Branch>,
    next_after: Option<String>,
}

#[derive(Deserialize)]
struct Branch {
    branch: String,
    version: Box<RawValue>,
}

/// `haematite write`, ending in its exit code.
#[must_use]
pub fn run(args: impl Iterator<Item = String>) -> ExitCode {
    finish(write(args))
}

fn write(args: impl Iterator<Item = String>) -> Result<Reply, Ended> {
    let asked = parse(args).map_err(|why| Ended::Usage(format!("{why}\n{USAGE}")))?;
    let value = one_value(asked.value, "value")?;
    let socket = socket_of(&asked.target)?;
    // Only an ordinary change is applied on the version before its receipt's.
    let ordinary = !matches!(asked.kind.as_str(), "branch.fork" | "branch.discard");
    let replays = asked.request_id.is_some() && ordinary;
    let request_id = asked.request_id.map_or_else(made_request_id, Ok)?;
    let mut client = Client::connect(&socket, asked.timeout)?;
    let branch = asked.branch.as_str();
    let expected = match asked.expected {
        Some(expected) => expected,
        None => match origin(&mut client, branch, &request_id, replays)? {
            Ok(origin) => origin,
            Err(refused) => return Ok(refused),
        },
    };
    let change = Change {
        kind: &asked.kind,
        value: &value,
    };
    let apply = Apply {
        request_id: &request_id,
        branch,
        expected: &expected,
        change,
    };
    let params = serde_json::value::to_raw_value(&apply)
        .map_err(|error| Ended::NotAsked(format!("the request could not be written: {error}")))?;
    match client.call("mutation.apply", Some(&params)) {
        Err(Ended::Unknown(why)) => reconcile(&socket, asked.timeout, branch, &request_id, &why),
        other => other,
    }
}

fn parse(mut args: impl Iterator<Item = String>) -> Result<Write, String> {
    let mut common = Common::default();
    let mut branch = None;
    let mut expected = None;
    let mut request_id = None;
    let mut rest = Vec::new();
    while let Some(arg) = args.next() {
        if common.take(&arg, &mut args)? {
            continue;
        }
        match arg.as_str() {
            "--branch" if branch.is_none() => branch = Some(value_of(&arg, &mut args)?),
            "--expected" if expected.is_none() => {
                expected = Some(version(&value_of(&arg, &mut args)?)?);
            }
            "--request-id" if request_id.is_none() => {
                request_id = Some(value_of(&arg, &mut args)?);
            }
            flag if flag.starts_with("--") => {
                return Err(format!("unknown or repeated flag {flag:?}"));
            }
            _ => rest.push(arg),
        }
    }
    let (target, timeout) = common.finish()?;
    let branch = branch.ok_or("name the branch with --branch")?;
    let mut rest = rest.into_iter();
    let kind = rest.next().ok_or("name the change kind")?;
    let value = rest
        .next()
        .map(Text::of)
        .ok_or("give the change's value, or -")?;
    if let Some(extra) = rest.next() {
        return Err(format!("unexpected argument {extra:?}"));
    }
    Ok(Write {
        target,
        timeout,
        branch,
        expected,
        request_id,
        kind,
        value,
    })
}

fn value_of(flag: &str, args: &mut impl Iterator<Item = String>) -> Result<String, String> {
    args.next()
        .filter(|value| !value.is_empty())
        .ok_or_else(|| format!("{flag} needs a value"))
}

/// `<generation>.<sequence>` as the service writes a version.
fn version(text: &str) -> Result<Box<RawValue>, String> {
    let refused = || format!("--expected {text:?} is not <generation>.<sequence>");
    let (generation, sequence) = text.split_once('.').ok_or_else(refused)?;
    let generation: u64 = generation.parse().map_err(|_| refused())?;
    let sequence: u64 = sequence.parse().map_err(|_| refused())?;
    version_value(generation, sequence).map_err(|error| error.to_string())
}

fn version_value(generation: u64, sequence: u64) -> serde_json::Result<Box<RawValue>> {
    RawValue::from_string(format!(
        r#"{{"generation":"{generation}","sequence":"{sequence}"}}"#
    ))
}

/// The version a change without --expected is sent against.
///
/// A given request id may name a change already applied. The service replays
/// its receipt only for the same request, `expected` included, so the version
/// that change was applied on is sent again and the service judges whether
/// the content is the same: the receipt, or `request_conflict` with nothing
/// applied. Otherwise, and for a made id, a fork or a discard, the branch's
/// head, and the service answers.
fn origin(
    client: &mut Client,
    branch: &str,
    request_id: &str,
    replays: bool,
) -> Result<Result<Box<RawValue>, Reply>, Ended> {
    if replays {
        match earlier(client, branch, request_id)? {
            Err(refused) => return Ok(Err(refused)),
            Ok(Some(receipt)) => {
                if let Some(origin) = applied_on(&receipt) {
                    return Ok(Ok(origin));
                }
            }
            Ok(None) => {}
        }
    }
    head(client, branch, request_id)
}

/// The receipt the branch already holds for `request_id`, if any. Nothing that
/// changes anything has been sent, so a failure here is could not ask.
fn earlier(
    client: &mut Client,
    branch: &str,
    request_id: &str,
) -> Result<Result<Option<Box<RawValue>>, Reply>, Ended> {
    let unread = |why: &dyn std::fmt::Display| {
        Ended::NotAsked(format!(
            "an earlier receipt for {request_id} could not be read: {why}"
        ))
    };
    let query = ReceiptQuery { branch, request_id };
    let params = serde_json::value::to_raw_value(&query).map_err(|error| unread(&error))?;
    match client.call("mutation.receipt", Some(&params)) {
        Ok(Reply::Answered(receipt)) if receipt.get() == "null" => Ok(Ok(None)),
        Ok(Reply::Answered(receipt)) => Ok(Ok(Some(receipt))),
        Ok(refused @ Reply::Refused(_)) => Ok(Err(refused)),
        Err(ended) => Err(unread(&ended.why())),
    }
}

#[derive(Deserialize)]
struct Applied {
    version: AppliedVersion,
}

#[derive(Deserialize)]
struct AppliedVersion {
    generation: String,
    sequence: String,
}

/// The version a change was applied on. This depends on the service's rule
/// for an ordinary change (storage/mutations.rs `mutate_with_import_bound`):
/// one applied on `{generation, sequence}` is written as `{generation,
/// sequence + 1}`; `tests/client_write.rs` pins that rule beside this client.
/// It exists because the served receipt carries no `request_origin`; once the
/// attributed admission path is served, its receipt does and is read instead.
/// `None` for sequence 0, or a receipt that does not read as one.
fn applied_on(receipt: &RawValue) -> Option<Box<RawValue>> {
    let applied: Applied = serde_json::from_str(receipt.get()).ok()?;
    let generation: u64 = applied.version.generation.parse().ok()?;
    let sequence: u64 = applied.version.sequence.parse().ok()?;
    version_value(generation, sequence.checked_sub(1)?).ok()
}

/// 128 bits from the operating system's random source, in hex, printed
/// before anything is sent so the caller has it whatever happens next.
fn made_request_id() -> Result<String, Ended> {
    let mut bytes = [0_u8; 16];
    std::fs::File::open("/dev/urandom")
        .and_then(|mut source| source.read_exact(&mut bytes))
        .map_err(|error| Ended::NotAsked(format!("no request id could be made: {error}")))?;
    let made = format!("{:032x}", u128::from_be_bytes(bytes));
    eprintln!("haematite: request {made}");
    Ok(made)
}

/// The branch's head as the service lists it, read the way the Studio reads
/// it. A branch that is not listed is asked for its receipt, a read the
/// service refuses by name for a missing branch; that refusal is passed on.
fn head(
    client: &mut Client,
    branch: &str,
    request_id: &str,
) -> Result<Result<Box<RawValue>, Reply>, Ended> {
    let unread = |why: &dyn std::fmt::Display| {
        Ended::NotAsked(format!("the head of {branch} could not be read: {why}"))
    };
    let limit = BRANCH_PAGE.min(client.max_page().get());
    let mut after = None;
    loop {
        let query = BranchQuery {
            after: after.as_deref(),
            limit,
        };
        let params = serde_json::value::to_raw_value(&query).map_err(|error| unread(&error))?;
        let page = match client.call("branches.list", Some(&params)) {
            Ok(Reply::Answered(page)) => page,
            Ok(refused @ Reply::Refused(_)) => return Ok(Err(refused)),
            Err(ended) => return Err(unread(&ended.why())),
        };
        let page: BranchPage = serde_json::from_str(page.get()).map_err(|error| unread(&error))?;
        if let Some(found) = page
            .entries
            .into_iter()
            .find(|entry| entry.branch == branch)
        {
            return Ok(Ok(found.version));
        }
        let Some(next) = page.next_after else {
            break;
        };
        after = Some(next);
    }
    let query = ReceiptQuery { branch, request_id };
    let params = serde_json::value::to_raw_value(&query).map_err(|error| unread(&error))?;
    match client.call("mutation.receipt", Some(&params)) {
        Ok(refused @ Reply::Refused(_)) => Ok(Err(refused)),
        Ok(Reply::Answered(receipt)) if receipt.get() == "null" => {
            Err(unread(&format!("{branch} is not listed")))
        }
        // A discard that landed: the branch is gone and its receipt remains.
        // Which change it was cannot be checked from here, so nothing is sent.
        Ok(Reply::Answered(_)) => Err(Ended::Unknown(format!(
            "{branch} is not listed, and the service holds a receipt for request \
             {request_id} on it, so a change under this id was applied; whether it was this \
             change is confirmed by running the same command with the --expected first sent"
        ))),
        Err(ended) => Err(unread(&ended.why())),
    }
}

/// The outcome of a request that was sent and not answered: its receipt, on a
/// new connection. A receipt found is the answer; anything else leaves the
/// outcome not known, with the request id to repeat it under.
fn reconcile(
    socket: &Path,
    timeout: Duration,
    branch: &str,
    request_id: &str,
    why: &str,
) -> Result<Reply, Ended> {
    let unknown = |found: String| {
        Ended::Unknown(format!(
            "request {request_id}: {why}; {found}. Running the same command with --request-id \
             {request_id} is safe to repeat"
        ))
    };
    let query = ReceiptQuery { branch, request_id };
    let asked = serde_json::value::to_raw_value(&query)
        .map_err(|error| Ended::NotAsked(error.to_string()))
        .and_then(|params| {
            Client::connect(socket, timeout)?.call("mutation.receipt", Some(&params))
        });
    match asked {
        Ok(Reply::Answered(receipt)) if receipt.get() != "null" => Ok(Reply::Answered(receipt)),
        Ok(Reply::Answered(_)) => Err(unknown("the service has no receipt for it yet".to_owned())),
        Ok(Reply::Refused(refusal)) => Err(unknown(format!(
            "its receipt was refused: {}: {}",
            refusal.code, refusal.message
        ))),
        Err(ended) => Err(unknown(format!(
            "its receipt could not be asked: {}",
            ended.why()
        ))),
    }
}