feat(server): PUT and DELETE one Blob cell over HTTP

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

@@ -9,4 +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 CLI and HTTP server do not expose these writes yet.
+  The HTTP server exposes them as `PUT` and `DELETE /blob`; the CLI does not
+  expose them yet.

changelog.d/blob-http-put-delete.added.md

@@ -0,0 +1,12 @@
+- `PUT` and `DELETE /graphs/{graph_id}/blob` replace or clear one Blob value
+  of an existing node or edge, each as one graph commit under the `change`
+  action. `PUT` takes a raw `application/octet-stream` body of at most 32 MiB,
+  inclusive (`415` for another media type, `413` over the limit, refused
+  before reading when `Content-Length` declares it, `408` past the body
+  deadline). Both take `branch`, default `main`, refuse `snapshot`, and accept
+  `If-Match`: `*` or a list of entity tags. A failed `If-Match` is a `412`
+  with the new `blob_precondition_failure.current_etag` detail and an `ETag`
+  header naming the cell's current value. Success returns a receipt with the
+  selector, branch, `kind` (`managed` or `null`), `size` and `etag`, the actor
+  and the exact `commit`, which is `null` only when clearing a cell that was
+  already null. A `PUT` also returns the `ETag` header.

crates/omnigraph-api-types/src/lib.rs

@@ -1056,6 +1056,185 @@ impl BlobStatOutput {
     }
 }
 
+/// Query parameters of `PUT` and `DELETE /graphs/{graph_id}/blob`.
+///
+/// A write names a branch and never a snapshot; an unknown parameter,
+/// `snapshot` included, is refused.
+#[derive(Debug, Clone, Serialize, Deserialize, IntoParams)]
+#[into_params(parameter_in = Query)]
+#[serde(deny_unknown_fields)]
+pub struct BlobWriteQuery {
+    /// Select a logical node or edge cell.
+    pub entity: BlobEntityKind,
+    /// Accepted-schema node or edge type name.
+    pub r#type: String,
+    /// Logical id of an existing entity within the selected type.
+    pub id: String,
+    /// Accepted-schema Blob property name.
+    pub property: String,
+    /// Branch to write. Defaults to `main`.
+    pub branch: Option<String>,
+}
+
+impl From<&BlobWriteQuery> for BlobSelectorOutput {
+    fn from(query: &BlobWriteQuery) -> Self {
+        Self {
+            entity: query.entity,
+            r#type: query.r#type.clone(),
+            id: query.id.clone(),
+            property: query.property.clone(),
+        }
+    }
+}
+
+/// The state a Blob write left its cell in. Separate from
+/// [`BlobContentKindOutput`]: a write can leave the cell null, a stat never
+/// describes one.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ToSchema)]
+#[serde(rename_all = "snake_case")]
+pub enum BlobWriteStateOutput {
+    Managed,
+    Null,
+}
+
+/// Receipt of a Blob put or clear.
+///
+/// A put leaves managed bytes, with their `size` and `etag`; a clear leaves
+/// null, with neither. The `etag` is the one a read at `commit` reports.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct BlobWriteOutput {
+    pub selector: BlobSelectorOutput,
+    /// The branch the write published to.
+    pub branch: String,
+    pub kind: BlobWriteStateOutput,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub size: Option<u64>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub etag: Option<String>,
+    pub actor_id: Option<String>,
+    /// The write's own publication. Always present on the wire; `null` only
+    /// for a clear of a cell that was already null, which publishes nothing.
+    #[serde(deserialize_with = "Option::deserialize")]
+    #[schema(required = true)]
+    pub commit: Option<CommitOutput>,
+}
+
+/// Map an engine Blob write outcome to its receipt. The server and the CLI's
+/// embedded writes share it, so both report the same receipt.
+pub fn blob_write_output(
+    selector: BlobSelectorOutput,
+    branch: impl Into<String>,
+    outcome: &omnigraph::BlobWriteOutcome,
+    actor_id: Option<String>,
+) -> BlobWriteOutput {
+    let branch = branch.into();
+    match outcome {
+        omnigraph::BlobWriteOutcome::Managed {
+            length,
+            etag,
+            commit,
+        } => BlobWriteOutput {
+            selector,
+            branch,
+            kind: BlobWriteStateOutput::Managed,
+            size: Some(*length),
+            etag: Some(etag.as_str().to_string()),
+            actor_id,
+            commit: Some(commit_output(commit)),
+        },
+        omnigraph::BlobWriteOutcome::Null { commit } => BlobWriteOutput {
+            selector,
+            branch,
+            kind: BlobWriteStateOutput::Null,
+            size: None,
+            etag: None,
+            actor_id,
+            commit: commit.as_ref().map(commit_output),
+        },
+    }
+}
+
+/// Parse the `If-Match` field lines of a Blob write into its precondition.
+///
+/// Returns `Ok(None)` when no line is present. A field of exactly `*` is
+/// [`omnigraph::BlobPrecondition::AnyExisting`]; otherwise the comma-separated
+/// entity tags of every line form [`omnigraph::BlobPrecondition::Tags`]. Weak
+/// tags never match a strong comparison, so they are dropped: a list of only
+/// weak tags matches nothing. A malformed field, an empty one, or `*` beside
+/// another element is refused, so a write never runs under a precondition it
+/// misread. The server and the CLI share this parser.
+pub fn parse_blob_if_match<'a>(
+    lines: impl IntoIterator<Item = &'a [u8]>,
+) -> Result<Option<omnigraph::BlobPrecondition>, String> {
+    fn skip_ows(mut rest: &[u8]) -> &[u8] {
+        while let [b' ' | b'\t', tail @ ..] = rest {
+            rest = tail;
+        }
+        rest
+    }
+    let malformed = || "If-Match must be `*` or a list of quoted entity tags".to_string();
+
+    let mut present = false;
+    let mut any = false;
+    let mut elements = 0_usize;
+    let mut tags = Vec::new();
+    for line in lines {
+        present = true;
+        let mut rest = skip_ows(line);
+        while !rest.is_empty() {
+            if let [b',', tail @ ..] = rest {
+                rest = skip_ows(tail);
+                continue;
+            }
+            if let [b'*', tail @ ..] = rest {
+                any = true;
+                rest = tail;
+            } else {
+                let (weak, quoted) = match rest.strip_prefix(b"W/") {
+                    Some(quoted) => (true, quoted),
+                    None => (false, rest),
+                };
+                let [b'"', body @ ..] = quoted else {
+                    return Err(malformed());
+                };
+                let close = body
+                    .iter()
+                    .position(|byte| *byte == b'"')
+                    .ok_or_else(malformed)?;
+                let opaque = &body[..close];
+                if !opaque
+                    .iter()
+                    .all(|byte| matches!(byte, 0x21 | 0x23..=0x7e | 0x80..=0xff))
+                {
+                    return Err(malformed());
+                }
+                // A tag that is not UTF-8 can never equal an engine-issued
+                // ETag, so leaving it out changes no outcome.
+                if !weak && let Ok(opaque) = std::str::from_utf8(opaque) {
+                    tags.push(omnigraph::BlobEtag::from_tag(format!("\"{opaque}\"")));
+                }
+                rest = &body[close + 1..];
+            }
+            elements += 1;
+            rest = skip_ows(rest);
+            match rest {
+                [] => {}
+                [b',', tail @ ..] => rest = skip_ows(tail),
+                _ => return Err(malformed()),
+            }
+        }
+    }
+    if !present {
+        return Ok(None);
+    }
+    match (any, elements) {
+        (_, 0) => Err(malformed()),
+        (true, 1) => Ok(Some(omnigraph::BlobPrecondition::AnyExisting)),
+        (true, _) => Err(malformed()),
+        (false, _) => Ok(Some(omnigraph::BlobPrecondition::Tags(tags))),
+    }
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
 #[serde(deny_unknown_fields)]
 pub struct ChangeRequest {
@@ -1469,6 +1648,17 @@ pub struct RecoveryRequiredOutput {
     pub operation_id: String,
 }
 
+/// A Blob write's `If-Match` did not hold for the cell (HTTP 412). The write
+/// had no effect. `current_etag` is the cell's validator when it holds a
+/// managed value; it is absent for a null or external value, which no tag
+/// matches. Distinct from [`PreconditionFailureOutput`], the graph-commit
+/// precondition.
+#[derive(Debug, Clone, Serialize, Deserialize, ToSchema)]
+pub struct BlobPreconditionFailureOutput {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub current_etag: Option<String>,
+}
+
 /// Structured details for a caller write-precondition failure: HTTP 412, a
 /// mutation carried `Omnigraph-If-Graph-Commit: <commit_id>`, and the branch
 /// head no longer matches that id. The write had no effect; the caller re-reads
@@ -1667,6 +1857,10 @@ pub struct ErrorOutput {
     /// the machine-readable meaning and `code` is omitted.
     #[serde(skip_serializing_if = "Option::is_none")]
     pub precondition_failure: Option<PreconditionFailureOutput>,
+    /// Set with HTTP 412 when a Blob write's `If-Match` did not hold for the
+    /// cell. The `ETag` response header repeats `current_etag` when present.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub blob_precondition_failure: Option<BlobPreconditionFailureOutput>,
     /// Set with HTTP 410 when retained history can no longer reconstruct a
     /// change continuation. Recover via the baseline handshake.
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -1703,6 +1897,7 @@ impl ErrorOutput {
             external_blob_source: None,
             recovery_required: None,
             precondition_failure: None,
+            blob_precondition_failure: None,
             change_feed_gap: None,
             change_diff_refusal: None,
             full_text_index_rebuild_required: None,
@@ -2198,6 +2393,73 @@ mod tests {
     use omnigraph_compiler::settings::SettingScope;
     use serde_json::json;
 
+    #[test]
+    fn blob_if_match_parses_any_tag_lists_and_weak_tags_and_refuses_malformed_fields() {
+        use omnigraph::{BlobEtag, BlobPrecondition};
+        let parse = |lines: &[&str]| parse_blob_if_match(lines.iter().map(|line| line.as_bytes()));
+        let tags = |list: &[&str]| {
+            Some(BlobPrecondition::Tags(
+                list.iter().map(|tag| BlobEtag::from_tag(*tag)).collect(),
+            ))
+        };
+
+        assert_eq!(parse(&[]), Ok(None));
+        assert_eq!(parse(&["*"]), Ok(Some(BlobPrecondition::AnyExisting)));
+        assert_eq!(parse(&[" * "]), Ok(Some(BlobPrecondition::AnyExisting)));
+        assert_eq!(parse(&["\"a\""]), Ok(tags(&["\"a\""])));
+        // Lists across and within lines, empty elements and OWS; a comma
+        // inside a tag is part of the tag.
+        assert_eq!(
+            parse(&["\"a\" ,, \"b,c\"", "\t\"d\""]),
+            Ok(tags(&["\"a\"", "\"b,c\"", "\"d\""]))
+        );
+        // Weak tags never match strongly: dropped, so only-weak matches nothing.
+        assert_eq!(parse(&["W/\"a\", \"b\""]), Ok(tags(&["\"b\""])));
+        assert_eq!(parse(&["W/\"a\""]), Ok(tags(&[])));
+        assert_eq!(parse(&["\"\""]), Ok(tags(&["\"\""])));
+
+        for malformed in [
+            &[""][..],
+            &[" , "],
+            &["a"],
+            &["\"a"],
+            &["\"a\"b"],
+            &["\"a\" \"b\""],
+            &["w/\"a\""],
+            &["*, \"a\""],
+            &["*", "\"a\""],
+            &["*", "*"],
+            &["\"a b\""],
+        ] {
+            assert!(parse(malformed).is_err(), "{malformed:?} must be refused");
+        }
+    }
+
+    #[test]
+    fn blob_write_output_reports_a_managed_put_and_a_no_op_clear() {
+        let selector = BlobSelectorOutput {
+            entity: BlobEntityKind::Node,
+            r#type: "Document".to_string(),
+            id: "readme".to_string(),
+            property: "content".to_string(),
+        };
+        let cleared = blob_write_output(
+            selector.clone(),
+            "main",
+            &omnigraph::BlobWriteOutcome::Null { commit: None },
+            Some("act-alice".to_string()),
+        );
+        let wire = serde_json::to_value(&cleared).unwrap();
+        assert_eq!(wire["kind"], "null");
+        assert!(
+            wire["commit"].is_null(),
+            "a no-op clear reports `commit: null`"
+        );
+        assert!(wire.get("size").is_none() && wire.get("etag").is_none());
+        let read_back: BlobWriteOutput = serde_json::from_value(wire.clone()).unwrap();
+        assert_eq!(serde_json::to_value(&read_back).unwrap(), wire);
+    }
+
     #[test]
     fn diagnostic_suggestions_round_trip_and_old_payloads_remain_valid() {
         let old = json!({"code": "Q002", "expected": "missing parameters", "fix": "query q()"});