feat(cli): blob put and blob clear

.github/workflows/ci.yml

@@ -752,6 +752,7 @@ jobs:
             system_remote::remote_read_preserves_projection_order_in_json_and_csv \
             system_remote::remote_branch_create_list_merge_flow \
             system_remote::remote_merge_delivery_loss_never_replays_committed_effect \
+            system_remote::remote_blob_write_delivery_loss_never_replays_committed_effect \
             system_remote::remote_merge_statement_delivery_loss_never_replays_committed_effect \
             system_remote::remote_deployment_delivery_loss_preserves_owned_completion \
             system_remote::remote_branch_delete_removes_branch \

changelog.d/blob-cell-put-clear.added.md

@@ -9,5 +9,5 @@
   is the one a read at that commit reports. One put holds at most the
   session's `write_max_bytes` (32 MiB by default), inclusive; a larger one
   fails before any table is opened with resource `Blob write payload bytes`.
-  The HTTP server exposes them as `PUT` and `DELETE /blob`; the CLI does not
-  expose them yet.
+  The HTTP server exposes them as `PUT` and `DELETE /blob`, and the CLI as
+  `blob put` and `blob clear`.

changelog.d/blob-cli-put-clear.added.md

@@ -0,0 +1,11 @@
+- `omnigraph blob put ENTITY TYPE ID PROPERTY` stores the raw bytes of
+  `--file PATH` or stdin in one Blob cell, and `omnigraph blob clear` sets a
+  nullable cell to null, on a `--store` graph or a served one. Both take
+  `--branch`, `--if-match TAG` (`*` or entity tags) and `--json`, and print the
+  receipt the server returns: the selector, branch, state, size and ETag of a
+  managed value, the actor and the exact commit, which is `null` for a clear of
+  a cell that was already null. Input over 32 MiB and a malformed `--if-match`
+  are refused before the graph is addressed, and an interactive stdin is
+  refused rather than waited on. `--as` names the actor of a `--store` write.
+  A served Blob precondition failure exits 4, as a graph-commit one does; an
+  embedded one exits 1. A write is sent once and never retried.

crates/omnigraph-cli/Cargo.toml

@@ -13,6 +13,7 @@ name = "omnigraph"
 path = "src/main.rs"
 
 [dependencies]
+bytes = { workspace = true }
 omnigraph = { package = "omnigraph-engine", path = "../omnigraph", version = "0.13.0" }
 omnigraph-compiler = { path = "../omnigraph-compiler", version = "0.13.0" }
 omnigraph-api-types = { path = "../omnigraph-api-types", version = "0.13.0" }

crates/omnigraph-cli/src/blob_cli.rs

@@ -1,20 +1,20 @@
-//! CLI-only shaping for graph-level Blob reads.
+//! CLI-only shaping for graph-level Blob reads and writes.
 //!
 //! The engine and server own descriptor interpretation, snapshot fencing, and
 //! bounded payload reads. This module keeps the command line's selector,
 //! range, and error vocabulary identical across the embedded and served arms.
 
 use std::fs::File;
-use std::io::{self, Write};
+use std::io::{self, IsTerminal, Read, Write};
 use std::ops::Range;
-use std::path::PathBuf;
+use std::path::{Path, PathBuf};
 
 use color_eyre::Report;
 use color_eyre::eyre::{Result, eyre};
 use omnigraph::db::{ReadTarget, SnapshotId};
 use omnigraph::error::{ManifestErrorKind, OmniError};
-use omnigraph::{BlobCell, EntityKind, ExternalBlobRef};
-use omnigraph_api_types::{BlobEntityKind, BlobReadQuery};
+use omnigraph::{BlobCell, BlobPrecondition, EntityKind, ExternalBlobRef};
+use omnigraph_api_types::{BlobEntityKind, BlobReadQuery, BlobWriteQuery};
 use reqwest::StatusCode;
 use reqwest::header::{CONTENT_LENGTH, ETAG, HeaderMap, HeaderName};
 
@@ -164,6 +164,104 @@ pub(crate) fn blob_url(base_url: &str, query: &BlobReadQuery) -> Result<String>
     crate::helpers::remote_url(base_url, &["blob"], &params)
 }
 
+pub(crate) fn blob_write_query(
+    entity: BlobEntityKind,
+    type_name: String,
+    id: String,
+    property: String,
+    branch: Option<String>,
+) -> BlobWriteQuery {
+    BlobWriteQuery {
+        entity,
+        r#type: type_name,
+        id,
+        property,
+        branch,
+    }
+}
+
+pub(crate) fn blob_write_cell(query: &BlobWriteQuery) -> BlobCell {
+    BlobCell {
+        entity: match query.entity {
+            BlobEntityKind::Node => EntityKind::Node,
+            BlobEntityKind::Edge => EntityKind::Edge,
+        },
+        type_name: query.r#type.clone(),
+        id: query.id.clone(),
+        property: query.property.clone(),
+    }
+}
+
+pub(crate) fn blob_write_url(base_url: &str, query: &BlobWriteQuery) -> Result<String> {
+    let entity = match query.entity {
+        BlobEntityKind::Node => "node",
+        BlobEntityKind::Edge => "edge",
+    };
+    let mut params = vec![
+        ("entity", entity),
+        ("type", query.r#type.as_str()),
+        ("id", query.id.as_str()),
+        ("property", query.property.as_str()),
+    ];
+    if let Some(branch) = query.branch.as_deref() {
+        params.push(("branch", branch));
+    }
+    crate::helpers::remote_url(base_url, &["blob"], &params)
+}
+
+/// Parse `--if-match` with the parser the server applies to the `If-Match`
+/// header, so a malformed value is refused before any graph is addressed.
+pub(crate) fn parse_if_match(raw: Option<&str>) -> Result<Option<BlobPrecondition>> {
+    omnigraph_api_types::parse_blob_if_match(raw.map(str::as_bytes)).map_err(|error| eyre!(error))
+}
+
+/// Read `blob put`'s bytes from `--file` or stdin, never both, before any
+/// graph is addressed. Input over the put limit is refused with the resource
+/// the engine and the server report, without reading past one more byte;
+/// an interactive stdin is refused rather than waited on.
+pub(crate) fn read_put_input(file: Option<&Path>) -> Result<bytes::Bytes> {
+    let limit = omnigraph::BLOB_WRITE_MAX_BYTES;
+    let too_large = |actual: u64| -> Report {
+        OmniError::resource_limit(omnigraph::BLOB_WRITE_PAYLOAD_RESOURCE, limit, actual).into()
+    };
+    let mut buffer = Vec::new();
+    match file {
+        Some(path) => {
+            let file = File::open(path)
+                .map_err(|error| eyre!("cannot read --file {}: {error}", path.display()))?;
+            let length = file
+                .metadata()
+                .map_err(|error| eyre!("cannot read --file {}: {error}", path.display()))?
+                .len();
+            if length > limit {
+                return Err(too_large(length));
+            }
+            buffer.reserve(usize::try_from(length).unwrap_or(0));
+            file.take(limit + 1)
+                .read_to_end(&mut buffer)
+                .map_err(|error| eyre!("cannot read --file {}: {error}", path.display()))?;
+        }
+        None => {
+            let stdin = io::stdin();
+            if stdin.is_terminal() {
+                return Err(eyre!(
+                    "blob put reads its bytes from stdin; pipe them in or pass --file PATH"
+                ));
+            }
+            stdin
+                .lock()
+                .take(limit + 1)
+                .read_to_end(&mut buffer)
+                .map_err(|error| eyre!("cannot read stdin: {error}"))?;
+        }
+    }
+    let read = buffer.len() as u64;
+    if read > limit {
+        return Err(too_large(read));
+    }
+    Ok(bytes::Bytes::from(buffer))
+}
+
 pub(crate) fn whole_external_uri(reference: &ExternalBlobRef) -> Result<&str> {
     reference
         .whole_object_uri()

crates/omnigraph-cli/src/cli.rs

@@ -247,7 +247,7 @@ pub(crate) enum Command {
         #[arg(long = "type")]
         type_names: Vec<String>,
     },
-    /// Read one logical node or edge Blob cell.
+    /// Read, replace or clear one logical node or edge Blob cell.
     Blob {
         #[command(subcommand)]
         command: BlobCommand,
@@ -516,6 +516,61 @@ pub(crate) enum BlobCommand {
         #[arg(long)]
         json: bool,
     },
+    /// Replace one Blob value of an existing node or edge with raw bytes from
+    /// a file or stdin, as one graph commit.
+    Put {
+        /// Logical entity namespace.
+        #[arg(value_name = "ENTITY")]
+        entity: BlobEntityArg,
+        /// Accepted-schema node or edge type.
+        #[arg(value_name = "TYPE")]
+        type_name: String,
+        /// Logical entity id; the entity must exist.
+        #[arg(value_name = "ID")]
+        id: String,
+        /// Blob property name.
+        #[arg(value_name = "PROPERTY")]
+        property: String,
+        /// Write a named branch (defaults to main).
+        #[arg(long)]
+        branch: Option<String>,
+        /// Read the bytes from this file instead of stdin.
+        #[arg(long, value_name = "PATH")]
+        file: Option<PathBuf>,
+        /// Write only while the cell's ETag is one of these entity tags, or,
+        /// as `*`, while the cell holds any value.
+        #[arg(long, value_name = "TAG")]
+        if_match: Option<String>,
+        /// Emit the receipt as JSON.
+        #[arg(long)]
+        json: bool,
+    },
+    /// Set one nullable Blob value of an existing node or edge to null, as one
+    /// graph commit; a cell that is already null publishes nothing.
+    Clear {
+        /// Logical entity namespace.
+        #[arg(value_name = "ENTITY")]
+        entity: BlobEntityArg,
+        /// Accepted-schema node or edge type.
+        #[arg(value_name = "TYPE")]
+        type_name: String,
+        /// Logical entity id; the entity must exist.
+        #[arg(value_name = "ID")]
+        id: String,
+        /// Blob property name.
+        #[arg(value_name = "PROPERTY")]
+        property: String,
+        /// Write a named branch (defaults to main).
+        #[arg(long)]
+        branch: Option<String>,
+        /// Write only while the cell's ETag is one of these entity tags, or,
+        /// as `*`, while the cell holds any value.
+        #[arg(long, value_name = "TAG")]
+        if_match: Option<String>,
+        /// Emit the receipt as JSON.
+        #[arg(long)]
+        json: bool,
+    },
 }
 
 #[derive(Debug, Subcommand)]