laser-wire 0.10.1

use crate::content::ContentType;
use crate::destination::BackendBinding;
use crate::error::InvalidError;
use crate::limits::{
    MAX_FILTER_SCHEMA_BYTES, MAX_FILTER_SCHEMA_DEPTH, MAX_GRAPH_LABEL_BYTES, MAX_INDEX_FIELD_BYTES,
    MAX_PROJECTION_ID_BYTES, MAX_PROJECTION_NAME_BYTES, MAX_SCHEMA_NAME_BYTES,
};
use crate::validate::{Validate, json_depth_within, validate_bounded_text};
use serde::{Deserialize, Serialize};
use std::str::FromStr;

/// Prefix no projection id may start with: activity keys an index-only
/// binding under it, so it must never name a projection.
pub const RESERVED_PROJECTION_ID_PREFIX: &str = "index:";

/// The one JSON Schema dialect the registry accepts, named or implied.
pub const JSON_SCHEMA_DIALECT: &str = "https://json-schema.org/draft/2020-12/schema";

/// One indexed scalar: the index key `name` and the RFC-6901 JSON `pointer` into
/// the payload it is extracted from.
///
/// **Why declared-once extraction exists.** Stamping `agdx.idx.host_id=node-7`
/// on every Iggy message duplicates the field name on the wire and couples the
/// producer to projection details. With a schema declared once on the
/// projector, producers just push the payload (single or batched) and the
/// worker derives the index from the body. The `agdx.idx.*` header path remains
/// as the escape hatch for raw payloads and for schema-first bodies whose
/// writer schema is not registered. Explicit headers always win over
/// schema-extracted values for the same field name.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct IndexField {
    pub name: String,
    pub pointer: String,
    /// Optional field type: the query result type of the field, a stored value
    /// that does not fit reading as null, and the column type of a wide-column
    /// backend. Unset, the field is a `variant` and keeps each stored type.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub field_type: Option<FieldType>,
}

impl IndexField {
    /// An indexed field `name` extracted from the payload at JSON `pointer`.
    pub fn new(name: impl Into<String>, pointer: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            pointer: pointer.into(),
            field_type: None,
        }
    }

    /// An indexed field with a type its query results carry.
    pub fn typed(
        name: impl Into<String>,
        pointer: impl Into<String>,
        field_type: FieldType,
    ) -> Self {
        Self {
            name: name.into(),
            pointer: pointer.into(),
            field_type: Some(field_type),
        }
    }
}

/// The type of an indexed field. The projector stores whatever scalar the
/// payload carries, and a query reads the field as this type.
#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    strum::Display,
    strum::EnumString,
    strum::VariantArray,
)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
#[non_exhaustive]
pub enum FieldType {
    Text,
    Int,
    Float,
    Bool,
}

/// The indexed fields (and optional vector field) a projection extracts.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct IndexSchema {
    pub fields: Vec<IndexField>,
    // Optional JSON pointer to a vector field (`[f32]`) inside the payload.
    // Default is `/embedding` at the root, which matches what producers
    // writing JSON bodies produce today.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub vector_field: Option<String>,
    // Whether the projector inlines the payload bytes alongside the row by
    // default. The producer can still override per record, but a sensible
    // default means batch-publishers do not need to stamp the directive on
    // every message.
    #[serde(default)]
    pub inline_payload: bool,
}

impl IndexSchema {
    /// Start building an index schema.
    pub fn builder() -> IndexSchemaBuilder {
        IndexSchemaBuilder::default()
    }
}

/// Fluent builder for an `IndexSchema`.
#[derive(Default)]
pub struct IndexSchemaBuilder {
    schema: IndexSchema,
}

impl IndexSchemaBuilder {
    /// Index the JSON value at the root-level field `name`. The index key and
    /// the JSON path are the same in the common case, so `field("host")`
    /// expands to "extract `/host` from the payload and store it under
    /// the index key `host`". For nested or renamed extraction use
    /// [`field_at`](Self::field_at).
    pub fn field(mut self, name: impl Into<String>) -> Self {
        let name = name.into();
        let pointer = format!("/{name}");
        self.schema.fields.push(IndexField::new(name, pointer));
        self
    }

    /// Index the JSON value at `pointer` (RFC-6901) under the index key
    /// `name`. Use when the index column name differs from the payload field,
    /// or when the value lives in a nested structure -
    /// `field_at("cpu_pct", "/cpu/value")` indexes `cpu.value` as
    /// `cpu_pct`.
    pub fn field_at(mut self, name: impl Into<String>, pointer: impl Into<String>) -> Self {
        self.schema.fields.push(IndexField::new(name, pointer));
        self
    }

    /// Point the vector extractor at `pointer` instead of the default
    /// (`/embedding` at the payload root). Pointer is RFC-6901.
    pub fn vector_field(mut self, pointer: impl Into<String>) -> Self {
        self.schema.vector_field = Some(pointer.into());
        self
    }

    /// Default `inline_payload = true` for every record projected through this
    /// schema. Per-record overrides stay valid.
    pub fn inline_payload(mut self) -> Self {
        self.schema.inline_payload = true;
        self
    }

    /// Finish the index schema.
    pub fn build(self) -> IndexSchema {
        self.schema
    }
}

/// Opaque projection identifier. Stable string the producer stamps on the
/// wire via the `agdx.ref` header and the worker keys its catalog by.
/// Recommended shape: `"<name>.v<version>"`, e.g. `"reading.v1"`. Distinct from
/// `schema_id`, which selects a codec's writer schema rather than a
/// materialization rule.
#[derive(Clone, Debug, Default, PartialEq, Eq, PartialOrd, Ord, Hash, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(transparent)]
pub struct ProjectionId(String);

impl ProjectionId {
    /// A projection id from a string.
    pub fn new(value: impl Into<String>) -> Self {
        Self(value.into())
    }

    /// The id as a string slice.
    pub fn as_str(&self) -> &str {
        &self.0
    }
}

impl std::fmt::Display for ProjectionId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.write_str(&self.0)
    }
}

impl AsRef<str> for ProjectionId {
    fn as_ref(&self) -> &str {
        &self.0
    }
}

impl std::borrow::Borrow<str> for ProjectionId {
    fn borrow(&self) -> &str {
        &self.0
    }
}

impl FromStr for ProjectionId {
    type Err = InvalidError;

    fn from_str(s: &str) -> Result<Self, Self::Err> {
        validate_projection_id(s)?;
        Ok(Self(s.to_owned()))
    }
}

impl From<&str> for ProjectionId {
    fn from(value: &str) -> Self {
        Self(value.to_owned())
    }
}

impl From<String> for ProjectionId {
    fn from(value: String) -> Self {
        Self(value)
    }
}

#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ProjectionHistory {
    #[default]
    Append,
    Keyed {
        key: ProjectionKey,
        /// The header whose presence deletes the record's key. Without one the view never deletes.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        delete_header: Option<String>,
    },
}

impl ProjectionHistory {
    pub fn is_append(&self) -> bool {
        matches!(self, Self::Append)
    }
}

#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ProjectionKey {
    Header { name: String },
    Pointer { pointer: String },
}

/// What a projection materializes: queryable rows, or a knowledge graph of
/// nodes and edges. Rides the wire as a u8 code (the growable-dictionary pattern)
/// so a future kind flows through an old reader as
/// [`Unrecognized`](Self::Unrecognized) rather than failing the listing.
#[derive(Clone, Copy, Debug, Default, PartialEq, Eq, Hash, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(from = "u8", into = "u8")]
pub enum ProjectionKind {
    /// Materialize queryable rows (the default).
    #[default]
    Row,
    /// Materialize a knowledge graph (nodes, edges, triplets).
    Graph,
    /// Materialize every routed record whole: its body, headers and source
    /// position, with no extraction.
    Record,
    /// A code this build does not know, passed through.
    Unrecognized(u8),
}

impl ProjectionKind {
    /// The pinned wire code.
    pub const fn code(self) -> u8 {
        match self {
            ProjectionKind::Row => 0,
            ProjectionKind::Graph => 1,
            ProjectionKind::Record => 2,
            ProjectionKind::Unrecognized(code) => code,
        }
    }

    /// The kind for a wire code (unknown codes become
    /// [`Unrecognized`](Self::Unrecognized)).
    pub const fn from_code(code: u8) -> Self {
        match code {
            0 => ProjectionKind::Row,
            1 => ProjectionKind::Graph,
            2 => ProjectionKind::Record,
            other => ProjectionKind::Unrecognized(other),
        }
    }

    /// Whether this is the default `Row` kind (omitted on the wire).
    pub const fn is_row(&self) -> bool {
        matches!(self, ProjectionKind::Row)
    }
}

impl From<u8> for ProjectionKind {
    fn from(code: u8) -> Self {
        Self::from_code(code)
    }
}

impl From<ProjectionKind> for u8 {
    fn from(kind: ProjectionKind) -> u8 {
        kind.code()
    }
}

/// The declared plan for extracting nodes and edges from a payload. Pointer-based
/// (RFC 6901) and deterministic, so no model call is needed. Recorded on a graph
/// projection for discovery and as the extraction contract. Graph data is written
/// via the graph upsert op. Applying this plan to a bound source is managed-side.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct EntitySchema {
    pub nodes: Vec<NodeExtract>,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub edges: Vec<EdgeExtract>,
}

/// One node-extraction rule: the node's label and the pointer to its canonical
/// value (which content-addresses its id, so the same entity converges).
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct NodeExtract {
    pub label: String,
    pub value_pointer: String,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub embedding_pointer: Option<String>,
}

/// One edge-extraction rule: the edge type, the pointers to its endpoint values,
/// and optional pointers to the bitemporal valid-time window.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct EdgeExtract {
    pub edge_type: String,
    pub from_pointer: String,
    pub to_pointer: String,
    /// RFC-6901 pointer to the valid-time start (epoch micros) for the extracted
    /// edge, making it a bitemporal fact. `None` leaves the window open.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub valid_from_pointer: Option<String>,
    /// RFC-6901 pointer to the valid-time end (epoch micros). `None` leaves the
    /// edge open-ended (still valid).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub valid_to_pointer: Option<String>,
}

/// Global reusable projection definition. Names the extraction rules that turn
/// a payload into a queryable row. Not attached to a topic on its own:
/// bindings ([`ProjectionBinding`]) declare where projections may apply.
///
/// # Storage model
///
/// A published record lives in up to three places, controlled by the projection:
///
/// 1. **Iggy log** (always): the original wire bytes, partitioned, replayable
///    from offset 0. The source of truth.
/// 2. **Indexed columns** (always): the scalar fields declared via
///    `field` / `field_at`, extracted from the payload at materialize time
///    and stored in the row. These drive filters, ordering, and aggregates.
/// 3. **Inline body** (opt-in via `inline_payload`, on by default through
///    `Projection::builder`): a copy of the full original payload alongside
///    the row, so typed fetches can decode it without going back to the log.
///
/// The body may carry fields that are not indexed. Only the declared fields
/// are queryable. Everything else rides through as part of the inlined body
/// (when on) or is reachable only by Iggy replay (when off).
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct Projection {
    /// Stable id used on the wire as `agdx.ref`.
    pub id: ProjectionId,
    /// Human-readable name (e.g. `"reading"`). Distinct from `id` so the same
    /// logical projection can rename without breaking the wire ref.
    pub name: String,
    /// Schema-evolution version. Bump on incompatible changes. The wire id
    /// usually encodes this (`reading.v2`).
    pub version: u32,
    /// What this projection materializes: rows (default) or a graph.
    #[serde(default, skip_serializing_if = "ProjectionKind::is_row")]
    pub kind: ProjectionKind,
    #[serde(default, skip_serializing_if = "ProjectionHistory::is_append")]
    pub history: ProjectionHistory,
    /// Expected payload codec. `Any` means best-effort decode: the extraction
    /// plan tolerates an opaque payload.
    pub content_type: ContentType,
    /// Field extraction plan.
    pub extraction: IndexSchema,
    /// Node/edge extraction plan for a `Graph` projection. `None` for a row
    /// projection.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub entity_schema: Option<EntitySchema>,
    /// Default `inline_payload` for records routed through this projection,
    /// overridable per record.
    #[serde(default)]
    pub inline_payload_default: bool,
}

impl Projection {
    /// Start building a projection with this id.
    pub fn builder(id: impl Into<ProjectionId>) -> ProjectionBuilder {
        ProjectionBuilder {
            projection: Self {
                id: id.into(),
                name: String::new(),
                version: 1,
                kind: ProjectionKind::Row,
                history: ProjectionHistory::Append,
                content_type: ContentType::Any,
                // Default: inline the body alongside the indexed row so typed
                // fetches can decode it without going back to the Iggy log.
                // Opt out with `.index_only()` for high-volume projections
                // where duplication is not worth it.
                extraction: IndexSchema {
                    fields: Vec::new(),
                    vector_field: None,
                    inline_payload: true,
                },
                entity_schema: None,
                inline_payload_default: true,
            },
        }
    }
}

/// Fluent builder for a `Projection`.
pub struct ProjectionBuilder {
    projection: Projection,
}

impl ProjectionBuilder {
    /// Keep the current row of each key, deleted by a record carrying `delete_header` when one is named.
    pub fn keyed(mut self, key: ProjectionKey, delete_header: Option<&str>) -> Self {
        self.projection.history = ProjectionHistory::Keyed {
            key,
            delete_header: delete_header.map(str::to_owned),
        };
        self
    }

    /// Set the human-readable projection name.
    pub fn name(mut self, value: impl Into<String>) -> Self {
        self.projection.name = value.into();
        self
    }

    /// Set the projection version.
    pub fn version(mut self, value: u32) -> Self {
        self.projection.version = value;
        self
    }

    /// Set the wire codec the projector decodes records with.
    pub fn content_type(mut self, value: ContentType) -> Self {
        self.projection.content_type = value;
        self
    }

    /// Set the full index schema (instead of `field`/`vector_field`).
    pub fn extraction(mut self, value: IndexSchema) -> Self {
        self.projection.extraction = value;
        self
    }

    /// Make this a whole-record projection: every routed record materializes
    /// with its body, headers and source position, and nothing is extracted.
    pub fn record(mut self) -> Self {
        self.projection.kind = ProjectionKind::Record;
        self.projection.extraction = IndexSchema::default();
        self.projection.entity_schema = None;
        self.projection.inline_payload_default = false;
        self
    }

    /// Make this a graph projection with the given node/edge extraction plan.
    pub fn graph(mut self, schema: EntitySchema) -> Self {
        self.projection.kind = ProjectionKind::Graph;
        self.projection.entity_schema = Some(schema);
        self
    }

    /// Index a field by name (extracted at its matching JSON pointer).
    pub fn field(mut self, name: impl Into<String>) -> Self {
        let name = name.into();
        let pointer = format!("/{name}");
        self.projection
            .extraction
            .fields
            .push(IndexField::new(name, pointer));
        self
    }

    /// Bulk version of [`field`](Self::field): declare many top-level indexed
    /// fields in one call. Each name `n` lands at JSON pointer `/n`. For nested
    /// pointers use [`field_at`](Self::field_at) instead.
    pub fn fields<I, S>(mut self, names: I) -> Self
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        for name in names {
            let name = name.into();
            let pointer = format!("/{name}");
            self.projection
                .extraction
                .fields
                .push(IndexField::new(name, pointer));
        }
        self
    }

    /// Index a field `name` from an explicit JSON `pointer`.
    pub fn field_at(mut self, name: impl Into<String>, pointer: impl Into<String>) -> Self {
        self.projection
            .extraction
            .fields
            .push(IndexField::new(name, pointer));
        self
    }

    /// Like [`field`](Self::field) but with a storage-type hint for columnar
    /// backends (`FieldType::Int`, `Float`, `Bool`, `Text`). The embedded
    /// engine ignores the hint.
    pub fn field_typed(mut self, name: impl Into<String>, field_type: FieldType) -> Self {
        let name = name.into();
        let pointer = format!("/{name}");
        self.projection
            .extraction
            .fields
            .push(IndexField::typed(name, pointer, field_type));
        self
    }

    /// Like [`field_at`](Self::field_at) but with a storage-type hint.
    pub fn field_at_typed(
        mut self,
        name: impl Into<String>,
        pointer: impl Into<String>,
        field_type: FieldType,
    ) -> Self {
        self.projection
            .extraction
            .fields
            .push(IndexField::typed(name, pointer, field_type));
        self
    }

    /// Extract the embedding vector from this JSON pointer.
    pub fn vector_field(mut self, pointer: impl Into<String>) -> Self {
        self.projection.extraction.vector_field = Some(pointer.into());
        self
    }

    /// Inline the original payload alongside the materialized row. This is
    /// the default for projections built through `Projection::builder`, so
    /// calling it is redundant. It stays in the API for callers who want to be
    /// explicit. To opt out, use [`index_only`](Self::index_only).
    pub fn inline_payload(mut self) -> Self {
        self.projection.inline_payload_default = true;
        self.projection.extraction.inline_payload = true;
        self
    }

    /// Opt out of inlining the body. The materialized row keeps only the
    /// indexed scalars declared via `field` / `field_at` (and any vector
    /// extracted via `vector_field`). The full payload is not duplicated into
    /// the index. Use this when:
    ///
    /// - the body is large and you already store it elsewhere (object store,
    ///   OLAP table) and only need a fast queryable secondary index, or
    /// - you are pushing extreme throughput and the index DB cost of a body
    ///   copy is not worth it.
    ///
    /// The Iggy log retains the original bytes either way, so a future
    /// projector rebuild can re-inline. With `index_only`, typed fetches
    /// return rows whose `payload` is `None`. Callers either decode from the
    /// indexed columns or replay from the log.
    pub fn index_only(mut self) -> Self {
        self.projection.inline_payload_default = false;
        self.projection.extraction.inline_payload = false;
        self
    }

    /// Finish the projection.
    pub fn build(self) -> Projection {
        self.projection
    }
}

/// How long a binding's materialized rows live, **decoupled from the source
/// topic's Iggy `message_expiry`**. Lets a projection outlive (or undershoot)
/// the log it was built from.
///
/// The canonical case: a topic with aggressive expiry (short-lived partitions
/// for cheap storage and fast replay) whose derived index must survive forever.
/// Set the binding to [`RetentionPolicy::Keep`].
#[derive(Clone, Copy, Debug, PartialEq, Eq, Default, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum RetentionPolicy {
    /// Follow the log: rows are pruned once Iggy drops the messages that
    /// produced them. The default. Like `KeepUntilSourceDeleted`, it also
    /// deletes the projection when the source topic is deleted.
    #[default]
    MirrorLog,
    /// Keep rows forever, regardless of the source log *or its deletion*.
    Keep,
    /// Keep rows forever while the source topic exists, but drop the whole
    /// projection when the source topic is **deleted**. Ignores message expiry
    /// like `Keep`, but is tied to the topic's existence like `MirrorLog`.
    KeepUntilSourceDeleted,
    /// Keep rows for `ttl_micros` after they were materialized, independent of
    /// the log. Older rows are swept, and survivors outlive an expired source.
    TimeToLive { ttl_micros: u64 },
    /// Keep the newest `rows` rows for the target table, independent of the log.
    MaxRows { rows: u64 },
    /// A policy `kind` a newer server named that this build does not know. The
    /// decode degrades to this rather than failing the whole reply, so an older
    /// client keeps reading bindings it cannot fully interpret (it must not
    /// re-apply one: re-serializing loses the original `kind` and its fields).
    /// The same forward-compat shape the `ResultCode` and u8 dictionaries use.
    #[serde(other)]
    Unknown,
}

/// Declares where a `Projection` is allowed to materialize. Bindings, not
/// projections, tell the worker what to consume. A binding pairs a source
/// selector (stream and topic) with a set of allowed projection refs, an
/// optional default, and one operational index. The index uses the replica-local
/// backend unless an exact backend resource generation is bound.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct ProjectionBinding {
    pub source: SourceSelector,
    /// Projections allowed to fire on messages from this source. A record
    /// stamped with an `agdx.ref` outside this set falls back to the default
    /// projection, and is skipped when there is none.
    #[serde(default)]
    pub allowed_projections: Vec<ProjectionId>,
    /// Default projection applied to records that do not carry an `agdx.ref`.
    /// `None` means such records are skipped.
    #[serde(default)]
    pub default_projection: Option<ProjectionId>,
    /// Exact operational backend resource generation. `None` selects the
    /// replica-local backend.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub backend: Option<BackendBinding>,
    /// Operational index name served by `QueryTarget::Operational`.
    pub index: String,
    /// Opt this binding into the change feed: after each committed projector
    /// batch the managed backend publishes one `ChangeRecord` on the changes topic.
    /// Default off and skipped on the wire, so a deployment that never opts in
    /// emits nothing and a pre-feed binding encodes byte-identically.
    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
    pub notify: bool,
    /// Row lifetime for this binding, independent of the source topic's Iggy
    /// retention. `None` inherits LaserData Cloud's fleet-wide default
    /// (`MirrorLog` unless an operator changed it). Set it via
    /// [`retention`](ProjectionBindingBuilder::retention).
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub retention: Option<RetentionPolicy>,
    /// What happens to a record the destination or the extraction
    /// deterministically rejects. `Block`, the default, holds the partition at
    /// the record until an operator skips it. `DeadLetter` skips it through an
    /// ordered control record, so every replica skips the same record, and
    /// sends it to the dead-letter topic. A destination rejection counts only
    /// after the destination accepted other records of the batch and refused
    /// this one twice. Skipped on the wire at its default.
    #[serde(default, skip_serializing_if = "InvalidRecordPolicy::is_block")]
    pub invalid_records: InvalidRecordPolicy,
    /// What a whole-record binding does with a body larger than its backend
    /// stores. `Block`, the default, holds the partition at the record.
    /// `Reference` keeps the row without the body, which the source log still
    /// holds, and requires `MirrorLog` retention. Skipped on the wire at its
    /// default.
    #[serde(default, skip_serializing_if = "OversizePolicy::is_block")]
    pub oversize: OversizePolicy,
}

/// What a binding does with a record it proved invalid.
#[derive(
    Clone,
    Copy,
    Debug,
    Default,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    strum::Display,
    strum::EnumString,
)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
#[non_exhaustive]
pub enum InvalidRecordPolicy {
    /// Hold the partition at the record until an operator skips it.
    #[default]
    Block,
    /// Skip the record and send it to the dead-letter topic.
    DeadLetter,
    /// A policy a newer server named that this build does not know.
    #[serde(other)]
    #[strum(disabled)]
    Unknown,
}

impl InvalidRecordPolicy {
    /// Whether this is the default `Block` policy (omitted on the wire).
    pub fn is_block(&self) -> bool {
        matches!(self, Self::Block)
    }
}

/// What a whole-record binding does with a body its backend cannot store.
#[derive(
    Clone,
    Copy,
    Debug,
    Default,
    PartialEq,
    Eq,
    Serialize,
    Deserialize,
    strum::Display,
    strum::EnumString,
)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(rename_all = "snake_case")]
#[strum(serialize_all = "snake_case")]
#[non_exhaustive]
pub enum OversizePolicy {
    /// Hold the partition at the record.
    #[default]
    Block,
    /// Keep the row without its body, which the source log still holds.
    Reference,
    /// A policy a newer server named that this build does not know.
    #[serde(other)]
    #[strum(disabled)]
    Unknown,
}

impl OversizePolicy {
    /// Whether this is the default `Block` policy (omitted on the wire).
    pub fn is_block(&self) -> bool {
        matches!(self, Self::Block)
    }
}

impl ProjectionBinding {
    /// Start building a binding (which topic feeds which projection).
    pub fn builder() -> ProjectionBindingBuilder {
        ProjectionBindingBuilder::default()
    }
}

/// Fluent builder for a `ProjectionBinding`.
#[derive(Default)]
pub struct ProjectionBindingBuilder {
    source: Option<SourceSelector>,
    allowed: Vec<ProjectionId>,
    default_projection: Option<ProjectionId>,
    backend: Option<BackendBinding>,
    index: Option<String>,
    notify: bool,
    retention: Option<RetentionPolicy>,
    invalid_records: InvalidRecordPolicy,
    oversize: OversizePolicy,
}

impl ProjectionBindingBuilder {
    /// Set the binding source. A source is always a `(stream, topic)` pair,
    /// and a topic only exists within a stream, so both are required.
    pub fn source(mut self, stream: impl Into<String>, topic: impl Into<String>) -> Self {
        self.source = Some(SourceSelector::new(stream, topic));
        self
    }

    /// Set the binding source from a pre-built [`SourceSelector`].
    pub fn selector(mut self, source: SourceSelector) -> Self {
        self.source = Some(source);
        self
    }

    /// Allow records to route to this projection.
    pub fn allow(mut self, projection: impl Into<ProjectionId>) -> Self {
        self.allowed.push(projection.into());
        self
    }

    /// Projection used when a record carries no `agdx.ref`.
    pub fn default_projection(mut self, projection: impl Into<ProjectionId>) -> Self {
        self.default_projection = Some(projection.into());
        self
    }

    /// Bind the operational index to an exact backend resource generation.
    /// Omit this for the replica-local backend.
    pub fn backend(mut self, backend: BackendBinding) -> Self {
        self.backend = Some(backend);
        self
    }

    /// Set how long the materialized rows live, independent of the source
    /// topic's Iggy retention. Omit to inherit the managed backend's default.
    ///
    /// ```no_run
    /// # use laser_wire::control::{ProjectionBinding, RetentionPolicy};
    /// // Short-lived topic, permanent index:
    /// let binding = ProjectionBinding::builder()
    ///     .source("_agdx", "telemetry")
    ///     .allow("telemetry.v1")
    ///     .index("telemetry_rows")
    ///     .retention(RetentionPolicy::Keep)
    ///     .build();
    /// ```
    pub fn retention(mut self, retention: RetentionPolicy) -> Self {
        self.retention = Some(retention);
        self
    }

    /// Set the operational index name. Omit it to use the source topic name.
    pub fn index(mut self, index: impl Into<String>) -> Self {
        self.index = Some(index.into());
        self
    }

    /// Opt into the change feed: one `ChangeRecord` per committed projector
    /// batch for this binding, on the changes topic.
    #[must_use]
    pub fn notify(mut self) -> Self {
        self.notify = true;
        self
    }

    /// Set what happens to a record the binding proved invalid. See
    /// [`ProjectionBinding::invalid_records`].
    #[must_use]
    pub fn invalid_records(mut self, policy: InvalidRecordPolicy) -> Self {
        self.invalid_records = policy;
        self
    }

    /// Set what a whole-record binding does with an oversized body. See
    /// [`ProjectionBinding::oversize`].
    #[must_use]
    pub fn oversize(mut self, policy: OversizePolicy) -> Self {
        self.oversize = policy;
        self
    }

    /// Build the binding. Panics if `.source(..)` / `.selector(..)` was not set
    /// (programmer error, not user input). Prefer [`try_build`](Self::try_build)
    /// for code paths that handle config-load failures gracefully.
    pub fn build(self) -> ProjectionBinding {
        self.try_build()
            .expect("ProjectionBinding requires a source - call .source(stream, topic)")
    }

    /// Build the binding, returning an error if required fields are missing
    /// instead of panicking.
    pub fn try_build(self) -> Result<ProjectionBinding, InvalidError> {
        let source = self
            .source
            .ok_or_else(|| InvalidError::new("ProjectionBinding requires a source"))?;
        let index = self.index.unwrap_or_else(|| source.topic.clone());
        Ok(ProjectionBinding {
            source,
            allowed_projections: self.allowed,
            default_projection: self.default_projection,
            backend: self.backend,
            index,
            notify: self.notify,
            retention: self.retention,
            invalid_records: self.invalid_records,
            oversize: self.oversize,
        })
    }
}

/// Selects a single source `(stream, topic)` for v1. Both are required, since
/// a topic only exists within a stream. Future versions can extend this to
/// prefix, glob, or multi-stream selectors. Today the model is
/// one-topic-per-binding to keep routing simple.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct SourceSelector {
    pub stream: String,
    pub topic: String,
}

impl SourceSelector {
    /// A `(stream, topic)` source for a binding.
    pub fn new(stream: impl Into<String>, topic: impl Into<String>) -> Self {
        Self {
            stream: stream.into(),
            topic: topic.into(),
        }
    }
}

/// A registered writer schema, keyed by the `id` a producer stamps on
/// `agdx.sid`. Avro and Protobuf schemas decode their schema-first bodies. A
/// JSON Schema validates the decoded payload of a self-describing codec
/// (JSON, MessagePack, CBOR, BSON), which otherwise needs no entry here. Ids
/// are permanent: registering an occupied id with a different definition is
/// rejected managed-side, and dropping tombstones the definition.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct SchemaDef {
    pub id: u32,
    pub source: SchemaSource,
    /// Optional human label, pure metadata. LaserData Cloud stores and returns
    /// it but never dispatches on it (`agdx.sid` carries the id). Uniqueness is
    /// not enforced. Absent on the wire when unset.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    /// Optional caller-tracked schema version, pure metadata. LaserData Cloud
    /// stores and returns it but never dispatches on it (the `id` alone selects
    /// the decoder). Absent on the wire when unset, so pre-version registries
    /// decode unchanged.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub version: Option<u32>,
}

/// A writer schema whose id the control fold chooses at the record's ordered log
/// position: an identical definition of the same registry is reused, otherwise
/// the lowest free id is taken. The id an explicit [`SchemaDef`] names is never
/// renumbered, so the two stay distinct commands.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct SchemaAllocation {
    pub source: SchemaSource,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub version: Option<u32>,
}

impl SchemaAllocation {
    /// The definition this allocation registers under `id`.
    pub fn definition(&self, id: u32) -> SchemaDef {
        SchemaDef {
            id,
            source: self.source.clone(),
            name: self.name.clone(),
            version: self.version,
        }
    }
}

impl SchemaDef {
    /// The codec this schema applies to. A JSON Schema validates any
    /// self-describing codec and reports as `Json`.
    pub fn content_type(&self) -> ContentType {
        match self.source {
            SchemaSource::Avro { .. } => ContentType::Avro,
            SchemaSource::Protobuf { .. } => ContentType::Protobuf,
            SchemaSource::JsonSchema { .. } => ContentType::Json,
            // An unknown source kind from a newer server: best-effort.
            SchemaSource::Unknown => ContentType::Any,
        }
    }
}

/// The schema payload for a schema-first codec. Internally tagged on `kind`
/// (`{"kind":"avro","schema":...}`) so an older client tolerates a newer
/// server's unknown source kind: an unrecognized `kind` decodes to
/// [`SchemaSource::Unknown`] instead of failing the whole reply.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
#[serde(tag = "kind", rename_all = "snake_case")]
#[non_exhaustive]
pub enum SchemaSource {
    /// An Avro writer schema as its canonical JSON text.
    Avro { schema: String },
    /// A Protobuf `FileDescriptorSet` (one or more compiled `.proto` files)
    /// plus the fully-qualified message type to decode, e.g. `"fleet.Reading"`.
    Protobuf {
        #[cfg_attr(
            feature = "schemars",
            schemars(schema_with = "crate::encoding::bytes_schema")
        )]
        #[serde(with = "crate::encoding::bin_bytes")]
        descriptor_set: Vec<u8>,
        message_type: String,
    },
    /// A JSON Schema (draft 2020-12) as its JSON text. Validation-only: a
    /// self-describing record stamping this schema's id on `agdx.sid` has its
    /// decoded payload validated by LaserData Cloud. Mismatches are counted,
    /// optionally dead-lettered, and never materialize body fields.
    JsonSchema { schema: String },
    /// A schema `kind` a newer server named that this build does not know. The
    /// decode degrades to this rather than failing, so an older client can still
    /// read a schema registry that holds a source kind it cannot decode against
    /// (it must not re-register one: the original `kind` and its fields are
    /// lost). Same forward-compat shape as [`RetentionPolicy::Unknown`].
    #[serde(other)]
    Unknown,
}

/// One control command on the control topic. The client SDK (or any tool
/// driving LaserData Cloud) publishes these to register projections,
/// bindings, and schemas.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub enum ControlCommand {
    RegisterProjection(Projection),
    DropProjection(String),
    ApplyBinding(ProjectionBinding),
    RemoveBinding {
        source: SourceSelector,
        projection_ref: Option<String>,
        /// Remove the binding only when it belongs to this incarnation of its
        /// source stream or an earlier one. `None` removes it whatever its
        /// incarnation.
        #[serde(default, skip_serializing_if = "Option::is_none")]
        generation: Option<ControlStreamGeneration>,
    },
    /// Register (or replace by `id`) a writer schema for a schema-first codec.
    RegisterSchema(SchemaDef),
    /// Drop the schema registered under this id.
    DropSchema(u32),
    /// Register a writer schema under the id the control fold chooses. The
    /// record's operation identity keeps the chosen id, so a retry returns it.
    AllocateSchema(SchemaAllocation),
    /// Register (or replace by id) a graph projection: a [`Projection`] with
    /// `kind = Graph` and an `entity_schema`. A distinct command from
    /// [`RegisterProjection`](Self::RegisterProjection) so a deployment can gate
    /// graph registration separately.
    RegisterGraph(Projection),
    /// Drop the graph projection registered under this id.
    DropGraph(String),
    /// Register the named session topics in one stream, or all session topics.
    RegisterSessionSource {
        stream: String,
        topics: SessionTopics,
    },
    /// Stop folding session records from one stream.
    RemoveSessionSource {
        stream: String,
    },
    /// One consumer-filter catalog mutation, stamped by the streaming server.
    /// Its outcome (applied or rejected) is recorded under the operation id.
    FilterCatalog(Box<crate::filter::FilterCatalogCommand>),
    /// An operator accepts that one source partition lost history to
    /// retention. A projection that resumes past the gap continues, and keeps
    /// the gap as a record that its history is incomplete. Only the managed
    /// backend's operator path appends it. The streaming server never forwards
    /// it from a client.
    AcceptGap(GapAcceptance),
    /// The managed backend skips one record that a binding's destination or extraction
    /// deterministically rejected. Ordered on the control topic, so every
    /// replica skips the same record. Appended by the backend itself, for a
    /// binding that opted in or after an operator asked. The streaming server
    /// never forwards it from a client.
    SkipRecord(RecordSkip),
    /// The managed backend observed that one source partition incarnation
    /// retains no record below `first_offset`. Record skips and accepted gaps
    /// below it can never be reached again and leave the registry. Appended by
    /// the backend itself. The streaming server never forwards it from a
    /// client.
    SourceRetained(SourceRetention),
}

/// The first offset one source partition incarnation still retains.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct SourceRetention {
    pub source: SourceSelector,
    pub partition_id: u32,
    /// Creation time of the source topic incarnation the floor belongs to.
    pub topic_created_at_micros: u64,
    /// No record below this offset can be read again, `u64::MAX` when the
    /// incarnation is gone.
    pub first_offset: u64,
}

/// One record a binding skips, on one topic incarnation.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct RecordSkip {
    pub source: SourceSelector,
    pub partition_id: u32,
    /// Creation time of the source topic incarnation. A recreated topic never
    /// inherits a skip.
    pub topic_created_at_micros: u64,
    pub offset: u64,
    /// The operational index or graph the record failed to reach. The
    /// record's other effects still apply.
    pub target: String,
    /// What the destination or the extraction said about the record.
    pub reason: String,
}

impl RecordSkip {
    /// Whether this skip covers `offset` of `partition_id` for `target` on
    /// the topic incarnation `generation`.
    pub fn covers(
        &self,
        stream: &str,
        topic: &str,
        partition_id: u32,
        generation: u64,
        offset: u64,
        target: &str,
    ) -> bool {
        self.source.stream == stream
            && self.source.topic == topic
            && self.partition_id == partition_id
            && self.topic_created_at_micros == generation
            && self.offset == offset
            && self.target == target
    }
}

/// The exact missing range an operator accepted for one source partition.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct GapAcceptance {
    pub source: SourceSelector,
    pub partition_id: u32,
    /// Creation time of the source topic incarnation the gap belongs to. A
    /// recreated topic needs its own acceptance.
    pub topic_created_at_micros: u64,
    /// First missing offset, the resume point the projection expected.
    pub from_offset: u64,
    /// First offset the log could still serve when the gap was accepted.
    pub to_offset: u64,
}

impl GapAcceptance {
    /// Whether this acceptance covers a resume at `cursor` whose first readable
    /// record is `first_available`, on the topic incarnation `generation`.
    pub fn covers(&self, generation: u64, cursor: u64, first_available: u64) -> bool {
        self.topic_created_at_micros == generation
            && self.from_offset <= cursor
            && self.to_offset >= first_available
    }
}

/// The topics a session-source registration covers.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub enum SessionTopics {
    All,
    Named(Vec<String>),
}

/// Versioned wrapper around a [`ControlCommand`], CBOR-named on the wire.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct ControlEnvelope {
    pub v: u32,
    pub timestamp_micros: u64,
    pub command: ControlCommand,
    /// The stream the command's resource names belong to, set by a client that
    /// scopes its resources to a stream on the projection, binding, graph, and
    /// schema commands. The writer-schema registry keys `RegisterSchema` and
    /// `DropSchema` by `(stream, id)` when it is set. Absent on the wire when
    /// unset, which addresses the deployment-wide registry.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream: Option<String>,
    /// Identity the managed backend binds to the record before it appends it
    /// to the control log. A client never sets it. The backend seals the whole
    /// encoded envelope, stamp included, so a record cannot be moved to
    /// another deployment, actor, operation or stream generation.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stamp: Option<ControlStamp>,
}

/// Who appended a control record and for which resource generation.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct ControlStamp {
    /// The deployment whose backend appended the record. Zero for a
    /// standalone backend without a deployment identity.
    pub deployment_id: u64,
    /// Stable identity of the logical control operation. A retry keeps it, so
    /// a second append of the same operation applies once.
    pub operation_id: u128,
    /// The authenticated stream generation, present when the request was
    /// scoped to a stream.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream_generation: Option<ControlStreamGeneration>,
    /// Identifies the sealing key without revealing it.
    pub key_id: u64,
    /// The client that asked for the record, the only actor the record names.
    /// Absent on a record the backend or the streaming server issues on its
    /// own, so the root user, id zero, is never mistaken for them.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub caller: Option<crate::forward::Caller>,
}

/// One incarnation of a stream: a deleted and recreated stream has a new
/// creation time even when it reuses its name and id.
#[derive(Clone, Copy, Debug, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
#[cfg_attr(feature = "schemars", derive(schemars::JsonSchema))]
pub struct ControlStreamGeneration {
    pub stream_id: u32,
    pub created_at_micros: u64,
}

impl Validate for ProjectionId {
    fn validate(&self) -> Result<(), InvalidError> {
        validate_projection_id(&self.0)
    }
}

impl Validate for Projection {
    fn validate(&self) -> Result<(), InvalidError> {
        self.id.validate()?;
        if let ProjectionHistory::Keyed { key, delete_header } = &self.history {
            if self.kind == ProjectionKind::Graph {
                return Err(InvalidError::new(
                    "graph projections do not support keyed history",
                ));
            }
            if let Some(delete_header) = delete_header {
                validate_bounded_text("deletion header", delete_header, MAX_INDEX_FIELD_BYTES)?;
            }
            match key {
                ProjectionKey::Header { name } => {
                    validate_bounded_text("key header", name, MAX_INDEX_FIELD_BYTES)?
                }
                ProjectionKey::Pointer { pointer } => {
                    if self.kind != ProjectionKind::Row {
                        return Err(InvalidError::new(
                            "payload key pointers require a row projection",
                        ));
                    }
                    if !pointer.is_empty() {
                        validate_bounded_text("key pointer", pointer, MAX_INDEX_FIELD_BYTES)?;
                        if !pointer.starts_with('/') {
                            return Err(InvalidError::new("key pointer must be a JSON pointer"));
                        }
                    }
                    let mut bytes = pointer.bytes();
                    while let Some(byte) = bytes.next() {
                        if byte == b'~' && !matches!(bytes.next(), Some(b'0' | b'1')) {
                            return Err(InvalidError::new("key pointer has an invalid escape"));
                        }
                    }
                }
            }
        }

        if self.kind == ProjectionKind::Record
            && (!self.extraction.fields.is_empty()
                || self.extraction.vector_field.is_some()
                || self.entity_schema.is_some())
        {
            return Err(InvalidError::new(
                "a whole-record projection extracts nothing, so it declares no fields, vector field or entity schema",
            ));
        }
        if !self.name.is_empty() {
            validate_bounded_text("projection name", &self.name, MAX_PROJECTION_NAME_BYTES)?;
        }
        for field in &self.extraction.fields {
            validate_bounded_text("index field name", &field.name, MAX_INDEX_FIELD_BYTES)?;
            let mut characters = field.name.chars();
            let leads = characters
                .next()
                .is_some_and(|first| first.is_ascii_alphabetic() || first == '_');
            if !leads || !characters.all(|rest| rest.is_ascii_alphanumeric() || rest == '_') {
                return Err(InvalidError::new(format!(
                    "index field name `{}` must start with an ASCII letter or `_` and hold only ASCII letters, digits and `_`",
                    field.name
                )));
            }
            validate_bounded_text("index field pointer", &field.pointer, MAX_INDEX_FIELD_BYTES)?;
        }
        if let Some(vector_field) = &self.extraction.vector_field {
            validate_bounded_text("index vector field", vector_field, MAX_INDEX_FIELD_BYTES)?;
        }
        if let Some(entity_schema) = &self.entity_schema {
            for node in &entity_schema.nodes {
                validate_bounded_text("graph node label", &node.label, MAX_GRAPH_LABEL_BYTES)?;
            }
            for edge in &entity_schema.edges {
                validate_bounded_text("graph edge type", &edge.edge_type, MAX_GRAPH_LABEL_BYTES)?;
            }
        }
        Ok(())
    }
}

impl Validate for ProjectionBinding {
    fn validate(&self) -> Result<(), InvalidError> {
        for projection in &self.allowed_projections {
            projection.validate()?;
        }
        if let Some(projection) = &self.default_projection {
            projection.validate()?;
        }
        if self.invalid_records == InvalidRecordPolicy::Unknown
            || self.oversize == OversizePolicy::Unknown
            || self.retention == Some(RetentionPolicy::Unknown)
        {
            return Err(InvalidError::new(
                "a binding cannot apply a policy this build does not know",
            ));
        }
        if matches!(
            self.retention,
            Some(
                RetentionPolicy::TimeToLive { ttl_micros: 0 }
                    | RetentionPolicy::MaxRows { rows: 0 }
            )
        ) {
            return Err(InvalidError::new(
                "a time to live retention needs a lifetime above zero, and a max rows retention at least one row",
            ));
        }
        if self.oversize == OversizePolicy::Reference
            && self
                .retention
                .is_some_and(|retention| retention != RetentionPolicy::MirrorLog)
        {
            return Err(InvalidError::new(
                "a binding that keeps body references needs mirror_log retention, so no row outlives its record",
            ));
        }
        Ok(())
    }
}

impl Validate for SchemaDef {
    fn validate(&self) -> Result<(), InvalidError> {
        if self.id == 0 {
            return Err(InvalidError::new("writer schema ids start at 1"));
        }
        validate_schema_metadata(&self.source, self.name.as_deref())
    }
}

impl Validate for SchemaAllocation {
    fn validate(&self) -> Result<(), InvalidError> {
        validate_schema_metadata(&self.source, self.name.as_deref())
    }
}

impl Validate for SchemaSource {
    fn validate(&self) -> Result<(), InvalidError> {
        match self {
            Self::Avro { schema } => validate_schema_text("Avro schema", schema),
            Self::JsonSchema { schema } => {
                validate_schema_text("JSON Schema", schema)?;
                let value: serde_json::Value = serde_json::from_str(schema).map_err(|error| {
                    InvalidError::new(format!("JSON Schema is not valid JSON: {error}"))
                })?;
                match value.get("$schema") {
                    None => Ok(()),
                    Some(serde_json::Value::String(dialect))
                        if dialect.trim_end_matches('#') == JSON_SCHEMA_DIALECT =>
                    {
                        Ok(())
                    }
                    Some(other) => Err(InvalidError::new(format!(
                        "a JSON Schema uses draft 2020-12, not {other}"
                    ))),
                }
            }
            Self::Protobuf {
                descriptor_set,
                message_type,
            } => {
                if descriptor_set.len() > MAX_FILTER_SCHEMA_BYTES {
                    return Err(InvalidError::new(format!(
                        "Protobuf descriptor set exceeds {MAX_FILTER_SCHEMA_BYTES} bytes"
                    )));
                }
                validate_bounded_text("Protobuf message type", message_type, MAX_SCHEMA_NAME_BYTES)
            }
            Self::Unknown => Err(InvalidError::new(
                "a schema source of an unknown kind cannot be registered",
            )),
        }
    }
}

impl Validate for ControlCommand {
    fn validate(&self) -> Result<(), InvalidError> {
        match self {
            Self::RegisterProjection(projection) | Self::RegisterGraph(projection) => {
                projection.validate()
            }
            Self::DropProjection(id) | Self::DropGraph(id) => validate_projection_id(id),
            Self::ApplyBinding(binding) => binding.validate(),
            Self::RemoveBinding { projection_ref, .. } => projection_ref
                .as_deref()
                .map_or(Ok(()), validate_projection_id),
            Self::RegisterSchema(schema) => schema.validate(),
            Self::AllocateSchema(allocation) => allocation.validate(),
            Self::FilterCatalog(command) => command.mutation.validate(),
            Self::DropSchema(_)
            | Self::RegisterSessionSource { .. }
            | Self::RemoveSessionSource { .. }
            | Self::AcceptGap(_)
            | Self::SkipRecord(_)
            | Self::SourceRetained(_) => Ok(()),
        }
    }
}

fn validate_projection_id(id: &str) -> Result<(), InvalidError> {
    validate_bounded_text("projection id", id, MAX_PROJECTION_ID_BYTES)?;
    if id.starts_with(RESERVED_PROJECTION_ID_PREFIX) {
        return Err(InvalidError::new(format!(
            "projection id must not start with {RESERVED_PROJECTION_ID_PREFIX:?}"
        )));
    }
    Ok(())
}

fn validate_schema_metadata(source: &SchemaSource, name: Option<&str>) -> Result<(), InvalidError> {
    if let Some(name) = name {
        validate_bounded_text("schema name", name, MAX_SCHEMA_NAME_BYTES)?;
    }
    source.validate()
}

fn validate_schema_text(label: &str, text: &str) -> Result<(), InvalidError> {
    if text.len() > MAX_FILTER_SCHEMA_BYTES {
        return Err(InvalidError::new(format!(
            "{label} exceeds {MAX_FILTER_SCHEMA_BYTES} bytes"
        )));
    }
    if !json_depth_within(text.as_bytes(), MAX_FILTER_SCHEMA_DEPTH) {
        return Err(InvalidError::new(format!(
            "{label} nests deeper than {MAX_FILTER_SCHEMA_DEPTH}"
        )));
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::limits::{
        MAX_GRAPH_LABEL_BYTES, MAX_INDEX_FIELD_BYTES, MAX_PROJECTION_ID_BYTES,
        MAX_PROJECTION_NAME_BYTES, MAX_SCHEMA_NAME_BYTES,
    };
    use std::borrow::Borrow;

    #[test]
    fn given_keyed_projection_when_validated_then_should_check_kind_and_pointer() {
        for pointer in ["", "/id", "/a~1b/~0"] {
            Projection::builder("keyed")
                .keyed(
                    ProjectionKey::Pointer {
                        pointer: pointer.to_owned(),
                    },
                    Some("deleted"),
                )
                .build()
                .validate()
                .expect("valid pointer");
        }
        for pointer in ["id", "/a~", "/a~2"] {
            assert!(
                Projection::builder("keyed")
                    .keyed(
                        ProjectionKey::Pointer {
                            pointer: pointer.to_owned()
                        },
                        Some("deleted")
                    )
                    .build()
                    .validate()
                    .is_err()
            );
        }
        assert!(
            Projection::builder("keyed")
                .record()
                .keyed(
                    ProjectionKey::Pointer {
                        pointer: "/id".to_owned()
                    },
                    Some("deleted")
                )
                .build()
                .validate()
                .is_err()
        );
        assert!(
            Projection::builder("keyed")
                .graph(EntitySchema::default())
                .keyed(
                    ProjectionKey::Header {
                        name: "id".to_owned()
                    },
                    Some("deleted")
                )
                .build()
                .validate()
                .is_err()
        );
        Projection::builder("keyed")
            .record()
            .keyed(
                ProjectionKey::Header {
                    name: "id".to_owned(),
                },
                Some("deleted"),
            )
            .build()
            .validate()
            .expect("record header key");
    }

    #[test]
    fn given_an_invalid_projection_id_when_any_command_names_it_then_should_refuse_it() {
        let projection = |id: &str| {
            Projection::builder(id)
                .name("reading")
                .field("host")
                .build()
        };
        let at_cap = "p".repeat(MAX_PROJECTION_ID_BYTES);
        projection(&at_cap)
            .validate()
            .expect("a projection id at the cap is valid");
        ControlCommand::DropProjection(at_cap.clone())
            .validate()
            .expect("a drop of a projection id at the cap is valid");
        ControlCommand::RemoveBinding {
            source: SourceSelector::new("fleet", "readings"),
            projection_ref: None,
            generation: None,
        }
        .validate()
        .expect("a whole-binding removal names no projection");
        for invalid in [
            String::new(),
            "p".repeat(MAX_PROJECTION_ID_BYTES + 1),
            "reading\n.v1".to_owned(),
            format!("{RESERVED_PROJECTION_ID_PREFIX}reading"),
        ] {
            assert!(projection(&invalid).validate().is_err(), "{invalid:?}");
            assert!(
                ControlCommand::RegisterProjection(projection(&invalid))
                    .validate()
                    .is_err()
            );
            assert!(
                ControlCommand::DropProjection(invalid.clone())
                    .validate()
                    .is_err()
            );
            assert!(
                ControlCommand::DropGraph(invalid.clone())
                    .validate()
                    .is_err()
            );
            let allowed = ProjectionBinding::builder()
                .source("fleet", "readings")
                .allow(invalid.as_str())
                .build();
            assert!(ControlCommand::ApplyBinding(allowed).validate().is_err());
            let defaulted = ProjectionBinding::builder()
                .source("fleet", "readings")
                .default_projection(invalid.as_str())
                .build();
            assert!(defaulted.validate().is_err());
            assert!(
                ControlCommand::RemoveBinding {
                    source: SourceSelector::new("fleet", "readings"),
                    projection_ref: Some(invalid.clone()),
                    generation: None,
                }
                .validate()
                .is_err()
            );
            assert!(invalid.parse::<ProjectionId>().is_err());
        }
    }

    #[test]
    fn given_a_projection_name_over_the_cap_when_validated_then_should_refuse_it_and_accept_none() {
        let named = |name: &str| Projection::builder("reading.v1").name(name).build();
        named("").validate().expect("a projection name is optional");
        named(&"n".repeat(MAX_PROJECTION_NAME_BYTES))
            .validate()
            .expect("a projection name at the cap is valid");
        assert!(
            named(&"n".repeat(MAX_PROJECTION_NAME_BYTES + 1))
                .validate()
                .is_err()
        );
        assert!(named("read\ting").validate().is_err());
    }

    #[test]
    fn given_an_invalid_index_field_when_a_projection_is_validated_then_should_refuse_it() {
        let with_field = |name: &str, pointer: &str| {
            Projection::builder("reading.v1")
                .field_at(name, pointer)
                .build()
        };
        with_field("host", "/host")
            .validate()
            .expect("a plain index field is valid");
        let long = "f".repeat(MAX_INDEX_FIELD_BYTES + 1);
        assert!(with_field(&long, "/host").validate().is_err());
        assert!(with_field("host", &long).validate().is_err());
        assert!(with_field("", "/host").validate().is_err());
        assert!(with_field("host", "").validate().is_err());
        assert!(with_field("ho\u{7}st", "/host").validate().is_err());
        assert!(
            Projection::builder("reading.v1")
                .vector_field(long)
                .build()
                .validate()
                .is_err()
        );
        assert!(
            Projection::builder("reading.v1")
                .vector_field("")
                .build()
                .validate()
                .is_err()
        );
    }

    #[test]
    fn given_an_invalid_schema_name_when_registered_then_should_refuse_it() {
        let schema = |name: Option<String>| SchemaDef {
            id: 7,
            source: SchemaSource::JsonSchema {
                schema: "{}".to_owned(),
            },
            name,
            version: None,
        };
        schema(None)
            .validate()
            .expect("a schema without a name is valid");
        ControlCommand::RegisterSchema(schema(Some("s".repeat(MAX_SCHEMA_NAME_BYTES))))
            .validate()
            .expect("a schema name at the cap is valid");
        for invalid in [
            String::new(),
            "s".repeat(MAX_SCHEMA_NAME_BYTES + 1),
            "read\ning".to_owned(),
        ] {
            assert!(schema(Some(invalid.clone())).validate().is_err());
            assert!(
                ControlCommand::RegisterSchema(schema(Some(invalid)))
                    .validate()
                    .is_err()
            );
        }
    }

    #[test]
    fn given_an_invalid_graph_label_when_a_graph_projection_is_validated_then_should_refuse_it() {
        let graph = |label: &str, edge_type: &str| {
            Projection::builder("people.v1")
                .graph(EntitySchema {
                    nodes: vec![NodeExtract {
                        label: label.to_owned(),
                        value_pointer: "/name".to_owned(),
                        embedding_pointer: None,
                    }],
                    edges: vec![EdgeExtract {
                        edge_type: edge_type.to_owned(),
                        from_pointer: "/name".to_owned(),
                        to_pointer: "/company".to_owned(),
                        valid_from_pointer: None,
                        valid_to_pointer: None,
                    }],
                })
                .build()
        };
        let at_cap = "l".repeat(MAX_GRAPH_LABEL_BYTES);
        ControlCommand::RegisterGraph(graph(&at_cap, &at_cap))
            .validate()
            .expect("a label and an edge type at the cap are valid");
        let long = "l".repeat(MAX_GRAPH_LABEL_BYTES + 1);
        assert!(graph(&long, "works_at").validate().is_err());
        assert!(graph("Person", &long).validate().is_err());
        assert!(graph("", "works_at").validate().is_err());
        assert!(graph("Person", "").validate().is_err());
        assert!(
            ControlCommand::RegisterGraph(graph("Per\nson", "works_at"))
                .validate()
                .is_err()
        );
    }

    #[test]
    fn given_binding_policies_when_encoded_then_should_carry_them_only_when_not_the_default() {
        let plain = ProjectionBinding::builder().source("s", "t").build();
        assert_eq!(plain.invalid_records, InvalidRecordPolicy::Block);
        assert_eq!(plain.oversize, OversizePolicy::Block);
        let plain_json = serde_json::to_value(&plain).expect("encode a default binding");
        assert!(plain_json.get("invalid_records").is_none());
        assert!(plain_json.get("oversize").is_none());
        let opted = ProjectionBinding::builder()
            .source("s", "t")
            .invalid_records(InvalidRecordPolicy::DeadLetter)
            .oversize(OversizePolicy::Reference)
            .build();
        let json = serde_json::to_value(&opted).expect("encode an opted binding");
        assert_eq!(json["invalid_records"], "dead_letter");
        assert_eq!(json["oversize"], "reference");
        let decoded: ProjectionBinding =
            serde_json::from_value(json).expect("decode an opted binding");
        assert_eq!(decoded, opted);
    }

    #[test]
    fn given_an_unknown_binding_policy_when_decoded_then_should_degrade_and_refuse_to_apply_it() {
        let binding: ProjectionBinding = serde_json::from_value(serde_json::json!({
            "source": {"stream": "s", "topic": "t"},
            "index": "t",
            "invalid_records": "quarantine",
            "oversize": "truncate",
        }))
        .expect("an unknown policy decodes to Unknown");
        assert_eq!(binding.invalid_records, InvalidRecordPolicy::Unknown);
        assert_eq!(binding.oversize, OversizePolicy::Unknown);
        assert!(binding.validate().is_err());
    }

    #[test]
    fn given_body_references_when_retention_outlives_the_log_then_should_refuse_the_binding() {
        let binding = |retention: Option<RetentionPolicy>| {
            let builder = ProjectionBinding::builder()
                .source("s", "t")
                .oversize(OversizePolicy::Reference);
            match retention {
                Some(retention) => builder.retention(retention),
                None => builder,
            }
            .build()
        };
        binding(None)
            .validate()
            .expect("the fleet default is checked by the backend");
        binding(Some(RetentionPolicy::MirrorLog))
            .validate()
            .expect("references live as long as their records");
        assert!(binding(Some(RetentionPolicy::Keep)).validate().is_err());
        assert!(
            binding(Some(RetentionPolicy::TimeToLive { ttl_micros: 1 }))
                .validate()
                .is_err()
        );
    }

    #[test]
    fn given_a_record_projection_when_validated_then_should_refuse_extraction_rules() {
        let record = Projection::builder("orders").record().build();
        assert_eq!(record.kind, ProjectionKind::Record);
        assert_eq!(ProjectionKind::from_code(2), ProjectionKind::Record);
        assert_eq!(ProjectionKind::Record.code(), 2);
        record
            .validate()
            .expect("a bare record projection is valid");
        let with_field = Projection::builder("orders").record().field("id").build();
        assert!(with_field.validate().is_err());
        let with_vector = Projection::builder("orders")
            .record()
            .vector_field("/embedding")
            .build();
        assert!(with_vector.validate().is_err());
    }

    #[test]
    fn given_a_record_skip_when_matched_then_should_cover_only_its_exact_record() {
        let skip = RecordSkip {
            source: SourceSelector::new("s", "t"),
            partition_id: 2,
            topic_created_at_micros: 7,
            offset: 40,
            target: "rows".to_owned(),
            reason: "rejected".to_owned(),
        };
        assert!(skip.covers("s", "t", 2, 7, 40, "rows"));
        assert!(
            !skip.covers("s", "t", 2, 8, 40, "rows"),
            "another incarnation"
        );
        assert!(!skip.covers("s", "t", 2, 7, 41, "rows"));
        assert!(!skip.covers("s", "t", 3, 7, 40, "rows"));
        assert!(!skip.covers("s", "t", 2, 7, 40, "other"));
        assert!(!skip.covers("s", "u", 2, 7, 40, "rows"));
    }

    #[test]
    fn given_an_unknown_schema_source_kind_when_decoded_then_should_degrade_not_fail() {
        // A newer server names a `kind` this build does not know. The decode
        // must degrade to `Unknown` (the forward-compat catch-all) rather than
        // failing the whole reply, and any extra fields are ignored.
        let source: SchemaSource =
            serde_json::from_str(r#"{"kind":"parquet_descriptor","blob":[1,2,3]}"#)
                .expect("an unknown schema kind decodes to Unknown, not an error");
        assert_eq!(source, SchemaSource::Unknown);
        // A `SchemaDef` wrapping it reports the best-effort content type.
        let def = SchemaDef {
            id: 1,
            source,
            name: None,
            version: None,
        };
        assert_eq!(def.content_type(), ContentType::Any);
    }

    #[test]
    fn given_an_unknown_retention_kind_when_decoded_then_should_degrade_not_fail() {
        let policy: RetentionPolicy = serde_json::from_str(r#"{"kind":"keep_for_eras","eras":3}"#)
            .expect("an unknown retention kind decodes to Unknown, not an error");
        assert_eq!(policy, RetentionPolicy::Unknown);
    }

    #[test]
    fn given_an_explicit_operational_index_when_built_then_should_keep_one_index() {
        let binding = ProjectionBinding::builder()
            .source("fleet", "readings")
            .allow("reading.v1")
            .index("readings_rows")
            .build();
        assert_eq!(binding.index, "readings_rows");
        assert_eq!(binding.backend, None);
    }

    #[test]
    fn given_an_operational_backend_when_built_then_should_keep_exact_generation() {
        let backend = BackendBinding {
            resource_id: crate::destination::BackendResourceId::from_u128(7),
            generation: 3,
        };
        let binding = ProjectionBinding::builder()
            .source("fleet", "readings")
            .backend(backend.clone())
            .build();
        assert_eq!(binding.backend, Some(backend));
    }

    #[test]
    fn given_no_operational_index_when_built_then_should_use_the_topic_name() {
        let binding = ProjectionBinding::builder()
            .source("fleet", "readings")
            .allow("reading.v1")
            .build();
        assert_eq!(binding.index, "readings");
    }

    #[test]
    fn given_no_retention_when_built_then_should_default_to_none() {
        let binding = ProjectionBinding::builder()
            .source("fleet", "readings")
            .index("readings_rows")
            .build();
        assert_eq!(binding.retention, None);
    }

    #[test]
    fn given_no_source_when_try_built_then_should_error() {
        assert!(ProjectionBinding::builder().try_build().is_err());
    }

    #[test]
    fn given_a_projection_built_with_the_default_when_inspected_then_should_inline_payload() {
        let projection = Projection::builder("api.call.v1")
            .name("api.call")
            .version(1)
            .fields(["endpoint", "status"])
            .build();
        assert!(
            projection.inline_payload_default,
            "Projection::builder default should inline payload"
        );
        assert!(
            projection.extraction.inline_payload,
            "Projection::builder default should mark extraction.inline_payload too"
        );
    }

    #[test]
    fn given_a_projection_with_index_only_when_inspected_then_should_skip_inlining() {
        let projection = Projection::builder("api.call.v1")
            .name("api.call")
            .version(1)
            .fields(["endpoint"])
            .index_only()
            .build();
        assert!(
            !projection.inline_payload_default,
            "index_only must clear inline_payload_default"
        );
        assert!(
            !projection.extraction.inline_payload,
            "index_only must clear extraction.inline_payload"
        );
    }

    #[test]
    fn given_an_empty_projection_id_when_parsed_then_should_error() {
        assert!("".parse::<ProjectionId>().is_err());
        assert_eq!(
            "reading.v1"
                .parse::<ProjectionId>()
                .expect("non-empty id parses")
                .as_str(),
            "reading.v1"
        );
    }

    #[test]
    fn given_an_index_schema_builder_when_built_then_should_keep_every_declared_rule() {
        let schema = IndexSchema::builder()
            .field("host")
            .field_at("cpu_pct", "/cpu/value")
            .vector_field("/vectors/text")
            .inline_payload()
            .build();
        assert_eq!(
            schema,
            IndexSchema {
                fields: vec![
                    IndexField::new("host", "/host"),
                    IndexField::new("cpu_pct", "/cpu/value"),
                ],
                vector_field: Some("/vectors/text".to_owned()),
                inline_payload: true,
            }
        );
    }

    #[test]
    fn given_a_projection_id_from_a_string_when_read_as_text_then_should_expose_that_string() {
        let id = ProjectionId::from("reading.v1".to_owned());
        let viewed: &str = id.as_ref();
        let borrowed: &str = id.borrow();
        assert_eq!(id, ProjectionId::new("reading.v1"));
        assert_eq!(
            (id.to_string(), viewed, borrowed),
            ("reading.v1".to_owned(), "reading.v1", "reading.v1")
        );
    }

    #[test]
    fn given_a_keyed_record_projection_when_encoded_then_should_round_trip_and_omit_defaults() {
        let keyed = Projection::builder("orders.v1")
            .record()
            .keyed(
                ProjectionKey::Header {
                    name: "order_id".to_owned(),
                },
                Some("deleted"),
            )
            .build();
        let json = serde_json::to_value(&keyed).expect("encode a keyed record projection");
        let decoded: Projection =
            serde_json::from_value(json).expect("decode a keyed record projection");
        assert_eq!(decoded, keyed);
        let plain = serde_json::to_value(Projection::builder("orders.v1").build())
            .expect("encode a default projection");
        assert_eq!((plain.get("kind"), plain.get("history")), (None, None));
    }

    #[test]
    fn given_a_keyed_projection_without_a_delete_header_when_decoded_then_should_validate_and_encode_without_it()
     {
        let keyed = Projection::builder("orders.v1")
            .record()
            .keyed(
                ProjectionKey::Header {
                    name: "order_id".to_owned(),
                },
                Some("deleted"),
            )
            .build();
        let mut json = serde_json::to_value(&keyed).expect("encode a keyed projection");
        json["history"]
            .as_object_mut()
            .expect("a keyed history object")
            .remove("delete_header");
        let decoded: Projection = serde_json::from_value(json.clone())
            .expect("decode a keyed projection without a delete header");
        decoded
            .validate()
            .expect("a keyed view may name no delete header");
        assert_eq!(
            serde_json::to_value(&decoded).expect("encode it again"),
            json
        );
        json["history"]["delete_header"] = serde_json::Value::String(String::new());
        let empty: Projection =
            serde_json::from_value(json).expect("decode an empty delete header");
        assert!(
            empty.validate().is_err(),
            "a named delete header must not be empty"
        );
    }

    #[test]
    fn given_a_selector_and_notify_when_a_binding_is_built_then_should_keep_both() {
        let binding = ProjectionBinding::builder()
            .notify()
            .selector(SourceSelector::new("fleet", "readings"))
            .try_build()
            .expect("a selector is a source");
        assert_eq!(
            binding,
            ProjectionBinding {
                source: SourceSelector::new("fleet", "readings"),
                allowed_projections: Vec::new(),
                default_projection: None,
                backend: None,
                index: "readings".to_owned(),
                notify: true,
                retention: None,
                invalid_records: InvalidRecordPolicy::Block,
                oversize: OversizePolicy::Block,
            }
        );
    }

    #[test]
    fn given_a_gap_acceptance_when_matched_then_should_cover_only_its_incarnation_and_range() {
        let gap = GapAcceptance {
            source: SourceSelector::new("s", "t"),
            partition_id: 2,
            topic_created_at_micros: 7,
            from_offset: 10,
            to_offset: 20,
        };
        assert!(gap.covers(7, 10, 20), "a resume at both accepted edges");
        assert!(!gap.covers(8, 10, 20), "another incarnation");
        assert!(!gap.covers(7, 9, 20), "a cursor before the accepted range");
        assert!(
            !gap.covers(7, 10, 21),
            "a first record past the accepted range"
        );
    }

    #[test]
    fn given_an_index_field_name_that_is_not_an_identifier_when_validated_then_should_refuse_it() {
        for name in ["user-id", "order.total", "9lives", "caf\u{e9}"] {
            let projection = Projection::builder("orders")
                .field_at(name, "/value")
                .build();
            assert!(projection.validate().is_err(), "`{name}` is refused");
        }
        for name in ["user_id", "_private", "total2"] {
            let projection = Projection::builder("orders")
                .field_at(name, "/value")
                .build();
            assert!(projection.validate().is_ok(), "`{name}` is accepted");
        }
    }

    #[test]
    fn given_a_retention_that_never_or_always_prunes_when_validated_then_should_refuse_the_binding()
    {
        for retention in [
            RetentionPolicy::Unknown,
            RetentionPolicy::TimeToLive { ttl_micros: 0 },
            RetentionPolicy::MaxRows { rows: 0 },
        ] {
            let binding = ProjectionBinding::builder()
                .source("s", "t")
                .retention(retention)
                .build();
            assert!(binding.validate().is_err(), "{retention:?} is refused");
        }
        let bounded = ProjectionBinding::builder()
            .source("s", "t")
            .retention(RetentionPolicy::MaxRows { rows: 1 })
            .build();
        assert!(bounded.validate().is_ok());
    }

    #[test]
    fn given_one_unknown_binding_policy_when_validated_then_should_refuse_the_binding() {
        let binding = ProjectionBinding {
            invalid_records: InvalidRecordPolicy::Unknown,
            ..ProjectionBinding::builder().source("s", "t").build()
        };
        assert_eq!(binding.oversize, OversizePolicy::Block);
        assert!(binding.validate().is_err());
    }
}

#[cfg(all(test, feature = "cbor"))]
mod wire_tests {
    use super::*;
    use crate::codes::CONTROL_OP_VERSION;
    use crate::content::ContentType;
    use crate::framing::{decode_named, encode_named};

    #[test]
    fn given_an_apply_binding_when_round_tripped_then_should_preserve_index_and_version() {
        let binding = ProjectionBinding::builder()
            .source("fleet", "readings")
            .allow("reading.v1")
            .default_projection("reading.v1")
            .index("readings_rows")
            .build();
        let envelope = ControlEnvelope {
            v: CONTROL_OP_VERSION,
            timestamp_micros: 42,
            command: ControlCommand::ApplyBinding(binding),
            stream: None,
            stamp: None,
        };
        let bytes = encode_named(&envelope).expect("envelope serializes");
        let back: ControlEnvelope = decode_named(&bytes).expect("envelope deserializes");
        assert_eq!(back.v, CONTROL_OP_VERSION);
        let ControlCommand::ApplyBinding(decoded) = back.command else {
            panic!("expected ApplyBinding");
        };
        assert_eq!(decoded.index, "readings_rows");
        assert_eq!(decoded.backend, None);
        // Retention left unset stays unset across the wire (inherits LaserData Cloud
        // default), and is omitted from the encoding entirely.
        assert_eq!(decoded.retention, None);
    }

    #[test]
    fn given_a_register_schema_when_round_tripped_then_should_preserve_source() {
        let envelope = ControlEnvelope {
            v: CONTROL_OP_VERSION,
            timestamp_micros: 7,
            command: ControlCommand::RegisterSchema(SchemaDef {
                id: 11,
                source: SchemaSource::Avro {
                    schema: r#"{"type":"record","name":"Reading","fields":[]}"#.to_owned(),
                },
                name: None,
                version: None,
            }),
            stream: None,
            stamp: None,
        };
        let bytes = encode_named(&envelope).expect("envelope serializes");
        let back: ControlEnvelope = decode_named(&bytes).expect("envelope deserializes");
        assert_eq!(back.stream, None);
        let ControlCommand::RegisterSchema(decoded) = back.command else {
            panic!("expected RegisterSchema");
        };
        assert_eq!(decoded.id, 11);
        assert_eq!(decoded.content_type(), ContentType::Avro);
    }

    #[test]
    fn given_a_stream_scoped_drop_schema_when_round_tripped_then_should_preserve_the_stream() {
        let envelope = ControlEnvelope {
            v: CONTROL_OP_VERSION,
            timestamp_micros: 7,
            command: ControlCommand::DropSchema(11),
            stream: Some("acme".to_owned()),
            stamp: None,
        };
        let back: ControlEnvelope =
            decode_named(&encode_named(&envelope).expect("envelope serializes"))
                .expect("envelope deserializes");
        assert_eq!(back.stream.as_deref(), Some("acme"));
        assert!(matches!(back.command, ControlCommand::DropSchema(11)));
        let unscoped = ControlEnvelope {
            stream: None,
            ..envelope
        };
        let json = serde_json::to_string(&unscoped).expect("envelope serializes");
        assert!(
            !json.contains("stream"),
            "an unset stream is omitted: {json}"
        );
    }

    #[test]
    fn given_a_stamped_envelope_when_round_tripped_then_should_keep_the_identity_and_omit_an_unset_stamp()
     {
        let envelope = ControlEnvelope {
            v: CONTROL_OP_VERSION,
            timestamp_micros: 7,
            command: ControlCommand::DropSchema(11),
            stream: Some("acme".to_owned()),
            stamp: Some(ControlStamp {
                deployment_id: 42,
                operation_id: u128::MAX - 1,
                stream_generation: Some(ControlStreamGeneration {
                    stream_id: 9,
                    created_at_micros: 1_700_000_000_000_000,
                }),
                key_id: u64::MAX,
                caller: Some(crate::forward::Caller {
                    user_id: 0,
                    agent: None,
                    on_behalf_of: None,
                }),
            }),
        };
        let back: ControlEnvelope =
            decode_named(&encode_named(&envelope).expect("envelope serializes"))
                .expect("envelope deserializes");
        assert_eq!(back.stamp, envelope.stamp);
        let unstamped = ControlEnvelope {
            stamp: None,
            ..envelope
        };
        let json = serde_json::to_string(&unstamped).expect("envelope serializes");
        assert!(
            !json.contains("\"stamp\""),
            "an unset stamp is omitted: {json}"
        );
    }

    #[test]
    fn given_a_protobuf_schema_source_when_round_tripped_then_should_preserve_bytes() {
        let source = SchemaSource::Protobuf {
            descriptor_set: vec![10, 20, 30],
            message_type: "fleet.Reading".to_owned(),
        };
        let bytes = encode_named(&source).expect("serializes");
        let back: SchemaSource = decode_named(&bytes).expect("deserializes");
        assert_eq!(back, source);
    }

    #[test]
    fn given_retention_set_when_round_tripped_then_should_preserve_policy() {
        for policy in [
            // Explicit `MirrorLog` is the default *value* but not the default
            // *state* (`None`). Set explicitly it must survive as `Some(MirrorLog)`,
            // distinct from the unset binding that omits the field entirely.
            RetentionPolicy::MirrorLog,
            RetentionPolicy::Keep,
            RetentionPolicy::KeepUntilSourceDeleted,
            RetentionPolicy::TimeToLive {
                ttl_micros: 3_600_000_000,
            },
            RetentionPolicy::MaxRows { rows: 10_000 },
        ] {
            let binding = ProjectionBinding::builder()
                .source("fleet", "telemetry")
                .allow("telemetry.v1")
                .index("telemetry_rows")
                .retention(policy)
                .build();
            assert_eq!(binding.retention, Some(policy));
            let bytes = encode_named(&binding).expect("binding serializes");
            let back: ProjectionBinding = decode_named(&bytes).expect("binding deserializes");
            assert_eq!(back.retention, Some(policy));
        }
    }
}

#[cfg(all(test, feature = "codecs"))]
mod schema_tests {
    use super::*;
    use crate::framing::{decode_named, encode_named};

    #[test]
    fn given_a_schema_def_with_version_when_round_tripped_then_should_preserve_it() {
        let def = SchemaDef {
            id: 7,
            source: SchemaSource::Avro {
                schema: "{}".to_owned(),
            },
            name: Some("readings".to_owned()),
            version: Some(2),
        };
        let bytes = encode_named(&def).expect("serializes");
        let back: SchemaDef = decode_named(&bytes).expect("deserializes");
        assert_eq!(back.version, Some(2));
        // Unset version is omitted from the wire entirely (back-compat with
        // pre-version registries).
        let unversioned = SchemaDef {
            name: None,
            version: None,
            ..def
        };
        let json = serde_json::to_string(&unversioned).expect("serializes");
        assert!(
            !json.contains("version"),
            "unset version must be omitted: {json}"
        );
    }
}