laser-wire 0.7.0

use crate::control::{Projection, ProjectionBinding, SchemaDef, SchemaSource};
use crate::query::QueryError;
use serde::{Deserialize, Serialize};

/// A registered projection plus the bindings that route topics into it. The full
/// picture of one materialized view: its extraction schema, expected content
/// type, indexed fields, and where it applies.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct ProjectionInfo {
    pub projection: Projection,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub bindings: Vec<ProjectionBinding>,
}

/// A writer schema plus its lifecycle state. A `dropped` schema is hidden
/// from the active set and its id rejects re-registration with a different
/// definition, but records stamped with the id keep decoding (ids are
/// permanent). Wire mirrors stay constructible (wire-stability-bound, not
/// API-stability-bound).
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub struct SchemaInfo {
    pub schema: SchemaDef,
    #[serde(default)]
    pub dropped: bool,
}

/// Request to read one projection's details by id.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct GetProjection {
    pub v: u32,
    pub id: String,
}

/// Request to list registered projections, optionally filtered. Empty filters
/// list every projection. `topics` keeps projections bound to any of the named
/// source topics. `name_contains` keeps those whose name contains the substring.
/// `id_prefix` keeps those whose id starts with it. `search` is the single-box
/// convenience that matches the substring against the name OR the id. The
/// filters compose (AND).
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ListProjections {
    pub v: u32,
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub topics: Vec<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name_contains: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub id_prefix: Option<String>,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub search: Option<String>,
}

/// Request to read one registered writer schema by id.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct GetSchema {
    pub v: u32,
    pub id: u32,
    /// The stream whose registry holds `id`. Absent for the deployment-wide
    /// registry.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream: Option<String>,
}

/// Request to list registered writer schemas, optionally filtered.
/// `name_contains` keeps those whose optional name contains the substring, and
/// an absent filter lists every schema. The filter lives here, not only on the
/// HTTP query, so the server pushes it down rather than the client paging the
/// whole set.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct ListSchemas {
    pub v: u32,
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub name_contains: Option<String>,
    /// The stream whose registry is listed. Absent for the deployment-wide
    /// registry.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream: Option<String>,
}

/// The synchronous register request: no id, LaserData Cloud validates the
/// definition, allocates the next free id, durably appends the control
/// event, and replies `SchemaRegistered(id)`. Distinct from
/// `ControlCommand::RegisterSchema`, the durable log form that carries the
/// allocated id.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct RegisterSchema {
    pub v: u32,
    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>,
    /// The stream whose registry allocates the id. Absent for the
    /// deployment-wide registry.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream: Option<String>,
}

/// Request to decode one record body under the schema registered for `id`,
/// the same decode the projector runs on `agdx.sid`-stamped records. Read-only
/// convenience for consoles and tools that hold a schema-first (Avro/Protobuf)
/// payload and want its JSON form without re-implementing the codecs.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct DecodeRecord {
    pub v: u32,
    pub id: u32,
    #[serde(with = "crate::encoding::bin_bytes")]
    pub payload: Vec<u8>,
    /// The stream whose registry holds `id`. Absent for the deployment-wide
    /// registry.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub stream: Option<String>,
}

/// Reply to a registry browse: `Ok` with the result, or `Err`.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[non_exhaustive]
pub enum BrowseReply {
    Ok(BrowseOutcome),
    Err(QueryError),
}

/// The result of a registry browse, shaped per request.
#[derive(Clone, Debug, Serialize, Deserialize)]
#[non_exhaustive]
pub enum BrowseOutcome {
    /// `list_projections`: every registered projection.
    Projections(Vec<ProjectionInfo>),
    /// `get projection`: the projection with the requested id, or `None`.
    Projection(Option<ProjectionInfo>),
    /// `list schemas`: every known writer schema, active and tombstoned.
    Schemas(Vec<SchemaInfo>),
    /// `get schema`: the schema occupying the requested id, or `None`.
    Schema(Option<SchemaInfo>),
    /// `register schema`: the LaserData-Cloud-allocated id, already durably appended
    /// to the control topic (visibility follows within the apply latency).
    SchemaRegistered(u32),
    /// `decode record`: the payload's JSON form under the requested schema,
    /// or `None` when the body does not decode under it. A `serde_json::Value`
    /// because the decoded shape is arbitrary (object, array, scalar) and the
    /// LaserData Cloud encodes exactly that value into the reply.
    Decoded(Option<serde_json::Value>),
}

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

    #[test]
    fn given_a_browse_reply_when_round_tripped_then_should_preserve_projection_details() {
        let info = ProjectionInfo {
            projection: Projection::builder("reading.v1")
                .name("reading")
                .version(1)
                .content_type(ContentType::Json)
                .fields(["reading_id", "cpu"])
                .build(),
            bindings: vec![
                ProjectionBinding::builder()
                    .source("fleet", "readings")
                    .allow("reading.v1")
                    .default_projection("reading.v1")
                    .index("readings_rows")
                    .build(),
            ],
        };
        let reply = BrowseReply::Ok(BrowseOutcome::Projections(vec![info]));
        let bytes = encode_named(&reply).expect("the reply serializes");
        let back: BrowseReply = decode_named(&bytes).expect("the reply deserializes");
        let BrowseReply::Ok(BrowseOutcome::Projections(list)) = back else {
            panic!("expected an Ok(Projections) browse reply");
        };
        assert_eq!(list.len(), 1);
        assert_eq!(list[0].projection.id.as_str(), "reading.v1");
        assert_eq!(list[0].projection.extraction.fields.len(), 2);
        assert_eq!(list[0].bindings[0].index, "readings_rows");
    }

    #[test]
    fn given_a_decode_record_when_round_tripped_then_should_preserve_payload_bytes() {
        let request = DecodeRecord {
            v: QUERY_OP_VERSION,
            id: 7,
            payload: vec![0xff, 0x00, 0x10],
            stream: None,
        };
        let bytes = encode_named(&request).expect("serializes");
        let back: DecodeRecord = decode_named(&bytes).expect("deserializes");
        assert_eq!(back.id, 7);
        assert_eq!(back.payload, vec![0xff, 0x00, 0x10]);
        assert_eq!(back.stream, None);
    }

    #[test]
    fn given_schema_requests_with_a_stream_when_round_tripped_then_should_preserve_it() {
        let get = GetSchema {
            v: QUERY_OP_VERSION,
            id: 7,
            stream: Some("acme".to_owned()),
        };
        let back: GetSchema =
            decode_named(&encode_named(&get).expect("serializes")).expect("deserializes");
        assert_eq!((back.id, back.stream.as_deref()), (7, Some("acme")));
        let list = ListSchemas {
            v: QUERY_OP_VERSION,
            name_contains: None,
            stream: Some("acme".to_owned()),
        };
        let back: ListSchemas =
            decode_named(&encode_named(&list).expect("serializes")).expect("deserializes");
        assert_eq!(back.stream.as_deref(), Some("acme"));
        let register = RegisterSchema {
            v: QUERY_OP_VERSION,
            source: SchemaSource::JsonSchema {
                schema: "{}".to_owned(),
            },
            name: None,
            version: None,
            stream: Some("acme".to_owned()),
        };
        let back: RegisterSchema =
            decode_named(&encode_named(&register).expect("serializes")).expect("deserializes");
        assert_eq!(back.stream.as_deref(), Some("acme"));
        let decode = DecodeRecord {
            v: QUERY_OP_VERSION,
            id: 7,
            payload: vec![1],
            stream: Some("acme".to_owned()),
        };
        let back: DecodeRecord =
            decode_named(&encode_named(&decode).expect("serializes")).expect("deserializes");
        assert_eq!(back.stream.as_deref(), Some("acme"));
    }

    #[test]
    fn given_schema_requests_without_a_stream_when_encoded_then_should_omit_it() {
        let get = GetSchema {
            v: QUERY_OP_VERSION,
            id: 7,
            stream: None,
        };
        let list = ListSchemas {
            v: QUERY_OP_VERSION,
            name_contains: None,
            stream: None,
        };
        for json in [
            serde_json::to_string(&get).expect("serializes"),
            serde_json::to_string(&list).expect("serializes"),
        ] {
            assert!(
                !json.contains("stream"),
                "an unset stream is omitted: {json}"
            );
        }
    }
}