zenkey_fleet/bus/

body.rs

//! Preparing a body for the wire (issue #97) — the encode half of the codec
//! seam, on the path a write actually takes.
//!
//! Until this module, both frontends *validated* an outgoing body by encoding
//! it against the producer's served schema and then **threw the encoded bytes
//! away**, putting the operator's JSON text on the wire. Everything downstream
//! was therefore a lie for any subject whose declared encoding was not JSON: a
//! `application/protobuf` subject could be described, refined, decoded — and
//! not published to. The fix is one seam, here, so that "the body was checked"
//! and "the body was encoded" stop being two different things.
//!
//! Three obligations this owes its callers, all of them RFC 09 §5.1 O4 in
//! different clothes:
//!
//! - a body that was **not** encoded says so ([`BodySource`]) — publishing
//!   as-typed is a legitimate outcome, silently publishing as-typed is not;
//! - the wire `Encoding` is resolved from what was *declared*, never sniffed
//!   off the operator's text ([`encode_encoding`]);
//! - a refusal happens **before** the bus, and the caller can opt out of the
//!   refusal ([`PrepareMode`]) without opting out of the labelling.

use crate::{Error, Result};
use zenkey::schema::{SchemaKind, TypeSchema, WireEncoding};
use zenoh::Session;

use crate::model::decode::SchemaStore;
use crate::model::registry::SliceSet;

/// How the bytes on the wire came to be — carried out of [`prepare_publish`]
/// so a frontend can say it, not guess it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum BodySource {
    /// Encoded through the producer's served schema for this type name.
    Encoded { type_name: String },
    /// No schema resolved — an unregistered key, an untyped subject, or a
    /// producer that serves no `describe`. The body ships as the caller typed
    /// it, which is honest only because it is labelled.
    AsTyped,
    /// The caller asked for verbatim bytes ([`PrepareMode::Raw`]).
    Raw,
}

/// What to do when a schema resolves and the body does not fit it.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum PrepareMode {
    /// Refuse before the bus (`zenctl` default).
    Encode,
    /// Try to encode; on failure ship the body as typed, with a note
    /// (`--no-validate`). "Do not refuse" — never "do not tell me".
    Lenient,
    /// Never encode; ship verbatim (`--raw`).
    Raw,
}

/// A body ready for the wire, with the provenance a caller must surface.
#[derive(Debug, Clone)]
pub struct PreparedBody {
    pub bytes: Vec<u8>,
    /// The wire `Encoding` to set, when one is known. `None` means nothing was
    /// declared anywhere — the publisher says nothing rather than guessing.
    pub encoding: Option<String>,
    pub source: BodySource,
    /// A line for the caller to print/render verbatim. `None` when the
    /// ordinary thing happened (encoded against a served schema).
    pub note: Option<String>,
}

impl PreparedBody {
    fn raw(bytes: Vec<u8>, encoding: Option<String>, note: Option<String>) -> PreparedBody {
        PreparedBody {
            bytes,
            encoding,
            source: BodySource::Raw,
            note,
        }
    }
}

/// The encoding a body should be **encoded into**, which is a different
/// question from the one [`crate::model::decode::resolve_encoding`] answers.
///
/// Decoding resolves *sample > registry > sniff* because a received payload
/// has bytes to sniff. An outgoing body has none that mean anything — the
/// operator typed JSON whatever the subject carries, so sniffing it would
/// label every protobuf subject `application/json`. The ladder is therefore
/// **declared flag > registry `encoding` > the schema kind's native
/// encoding**, and the kind is the authority of last resort precisely because
/// it is the one thing that cannot be wrong.
pub fn encode_encoding(
    declared: Option<&str>,
    registry: Option<&WireEncoding>,
    schema: Option<&TypeSchema>,
) -> Option<String> {
    if let Some(e) = declared {
        return Some(e.to_string());
    }
    if let Some(e) = registry {
        return Some(e.as_encoding_str().to_string());
    }
    schema.and_then(|s| match s.kind().as_str() {
        SchemaKind::JSON_SCHEMA => Some("application/json".to_string()),
        SchemaKind::PROTOBUF => Some("application/protobuf".to_string()),
        SchemaKind::CDR => Some("application/cdr".to_string()),
        // An unknown kind's native framing is exactly what this tool does not
        // know. Saying nothing beats naming the wrong one.
        _ => None,
    })
}

/// What to encode — the operator's half of a prepare, shared by both entry
/// points below.
///
/// The other half is what to encode it *against*, which is exactly what
/// differs between them: [`prepare_request`] is told the producer and type
/// outright, [`prepare_publish`] refines them out of a wire key. That split
/// is why this is one spec and not two: everything in it means the same
/// thing on both paths.
#[derive(Debug, Clone, Copy)]
pub struct PrepareSpec<'a> {
    /// The wire encoding the caller stated (`--encoding`), when they did.
    pub declared_encoding: Option<&'a str>,
    /// The bytes as the operator supplied them, before any encode.
    pub body: &'a [u8],
    /// Encode against the served schema, or ship verbatim.
    pub mode: PrepareMode,
}

/// Encode `spec.body` against `producer`'s served schema for `type_name`, or
/// explain why it could not.
///
/// `Ok` with [`BodySource::AsTyped`] when no schema resolves — that is not a
/// failure, and it is not silence either (RFC 08 §7 is a SHOULD; a producer
/// that serves no `describe` has said nothing about this type, which is
/// different from having said "any bytes will do").
///
/// Base-less, so a bare `&Session` rather than a [`crate::Fleet`]: nothing
/// here composes a key — the producer is named, and the schema comes through
/// the store, which carries the base already. [`prepare_publish`] takes a
/// `Fleet` because it *refines a wire key*, which needs one.
pub async fn prepare_request(
    session: &Session,
    store: &SchemaStore,
    producer: &str,
    type_name: &str,
    registry_encoding: Option<&WireEncoding>,
    spec: PrepareSpec<'_>,
) -> Result<PreparedBody> {
    let PrepareSpec {
        declared_encoding,
        body,
        mode,
    } = spec;
    if mode == PrepareMode::Raw {
        return Ok(PreparedBody::raw(
            body.to_vec(),
            encode_encoding(declared_encoding, registry_encoding, None),
            Some("raw: bytes sent verbatim, not encoded against the served schema".into()),
        ));
    }
    let schema = store.schema_for(session, producer, type_name).await;
    let encoding = encode_encoding(declared_encoding, registry_encoding, schema.as_ref());
    let Some(schema) = schema else {
        return Ok(PreparedBody {
            bytes: body.to_vec(),
            encoding,
            source: BodySource::AsTyped,
            note: Some(format!(
                "{producer} serves no schema for {type_name} — body sent as typed, unchecked \
                 (RFC 08 §7 describe is a SHOULD; \"not served\" is not \"anything goes\")"
            )),
        });
    };

    let lenient = mode == PrepareMode::Lenient;
    let value: serde_json::Value = match serde_json::from_slice(body) {
        Ok(v) => v,
        Err(e) if lenient => {
            return Ok(PreparedBody {
                bytes: body.to_vec(),
                encoding,
                source: BodySource::AsTyped,
                note: Some(format!(
                    "body is not JSON, so it could not be encoded as {type_name} ({e}) — \
                     sent as typed"
                )),
            });
        }
        Err(e) => {
            // The caller handed us this body.
            return Err(Error::unaskable(
                "body",
                format!(
                    "is not JSON but {producer} declares schema-validated type \
                     {type_name} — {e}"
                ),
            ));
        }
    };

    let target = encoding
        .as_deref()
        .map(WireEncoding::from_encoding_str)
        // With nothing declared anywhere the target is the schema's own kind,
        // and for `json-schema` that is JSON — the framing an operator typed.
        .unwrap_or(WireEncoding::Json);
    match store.encode(&schema, &value, &target) {
        Ok(bytes) => Ok(PreparedBody {
            bytes,
            encoding,
            source: BodySource::Encoded {
                type_name: type_name.to_string(),
            },
            note: None,
        }),
        Err(e) if lenient => Ok(PreparedBody {
            bytes: body.to_vec(),
            encoding,
            source: BodySource::AsTyped,
            note: Some(format!(
                "body rejected by {type_name}'s served schema ({e}) — sent as typed anyway"
            )),
        }),
        Err(e) => Err(Error::unaskable(
            "body",
            format!("rejected by {type_name}'s served schema: {e}"),
        )),
    }
}

/// The publish-side entry point: refine a **full wire key** against the loaded
/// slices, then [`prepare_request`] on whatever type it names.
///
/// An unregistered key is not an error — it is the ordinary case on a bus this
/// convention does not govern, and the note says which case happened.
pub async fn prepare_publish(
    fleet: &crate::Fleet<'_>,
    store: &SchemaStore,
    slices: Option<&SliceSet>,
    wire_key: &str,
    spec: PrepareSpec<'_>,
) -> Result<PreparedBody> {
    let (session, base) = (fleet.session(), fleet.base());
    let PrepareSpec {
        declared_encoding,
        body,
        mode,
    } = spec;
    if mode == PrepareMode::Raw {
        return Ok(PreparedBody::raw(
            body.to_vec(),
            declared_encoding.map(str::to_string),
            Some("raw: bytes sent verbatim, not encoded against the served schema".into()),
        ));
    }

    let description = crate::model::facts::describe_key(base, wire_key, slices);
    let crate::model::facts::Registration::Registered(subject) = &description.facts.registration
    else {
        return Ok(PreparedBody {
            bytes: body.to_vec(),
            encoding: declared_encoding.map(str::to_string),
            source: BodySource::AsTyped,
            note: Some(match slices {
                // O4: with no slices loaded the tool has not asked, and
                // "not asked" is not "unregistered".
                None => format!(
                    "no registry loaded, so {wire_key} was never classified — body sent as typed"
                ),
                Some(_) => format!(
                    "{wire_key} is not a registered subject ({:?}) — body sent as typed",
                    description.facts.registration
                ),
            }),
        });
    };
    let Some(producer) = subject_producer(&description) else {
        return Ok(PreparedBody {
            bytes: body.to_vec(),
            encoding: encode_encoding(declared_encoding, subject.encoding.as_ref(), None),
            source: BodySource::AsTyped,
            note: Some(format!(
                "{wire_key} refines to a registered subject with no producer chunk to ask for a \
                 schema — body sent as typed"
            )),
        });
    };
    if subject.type_name.is_empty() {
        return Ok(PreparedBody {
            bytes: body.to_vec(),
            encoding: encode_encoding(declared_encoding, subject.encoding.as_ref(), None),
            source: BodySource::AsTyped,
            note: Some(format!(
                "{wire_key} is registered but declares no payload type — body sent as typed"
            )),
        });
    }

    prepare_request(
        session,
        store,
        &producer,
        &subject.type_name,
        subject.encoding.as_ref(),
        spec,
    )
    .await
}

/// The producer a registered description refined through. `SubjectFacts` does
/// not carry it (a service slice's name is not a key chunk), so it is derived
/// from the key shape.
pub fn subject_producer(description: &crate::model::facts::KeyDescription) -> Option<String> {
    match &description.facts.shape {
        crate::model::facts::KeyShape::V1(v) => v.producer.clone(),
        _ => None,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use serde_json::json;

    #[test]
    fn the_encode_ladder_never_sniffs_the_operators_text() {
        let protobuf = TypeSchema::protobuf("t.Blob", b"\x0a\x00");
        // Nothing declared: the kind decides — not the JSON the operator typed.
        assert_eq!(
            encode_encoding(None, None, Some(&protobuf)).as_deref(),
            Some("application/protobuf")
        );
        // The registry outranks the kind…
        assert_eq!(
            encode_encoding(None, Some(&WireEncoding::Cbor), Some(&protobuf)).as_deref(),
            Some("application/cbor")
        );
        // …and the flag outranks the registry.
        assert_eq!(
            encode_encoding(Some("application/json"), Some(&WireEncoding::Cbor), None).as_deref(),
            Some("application/json")
        );
        // An unknown kind's framing is unknown, and saying nothing is the
        // honest answer (O4).
        let json = TypeSchema::json_schema(json!({"type": "object"}));
        assert_eq!(
            encode_encoding(None, None, Some(&json)).as_deref(),
            Some("application/json")
        );
        assert_eq!(encode_encoding(None, None, None), None);
    }
}