laser_wire/
browse.rs

1use crate::control::{Projection, ProjectionBinding, SchemaDef, SchemaSource};
2use crate::query::QueryError;
3use serde::{Deserialize, Serialize};
4
5/// A registered projection plus the bindings that route topics into it. The full
6/// picture of one materialized view: its extraction schema, expected content
7/// type, indexed fields, and where it applies.
8#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
9pub struct ProjectionInfo {
10    pub projection: Projection,
11    #[serde(default, skip_serializing_if = "Vec::is_empty")]
12    pub bindings: Vec<ProjectionBinding>,
13}
14
15/// A writer schema plus its lifecycle state. A `dropped` schema is hidden
16/// from the active set and its id rejects re-registration with a different
17/// definition, but records stamped with the id keep decoding (ids are
18/// permanent). Wire mirrors stay constructible (wire-stability-bound, not
19/// API-stability-bound).
20#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
21pub struct SchemaInfo {
22    pub schema: SchemaDef,
23    #[serde(default)]
24    pub dropped: bool,
25}
26
27/// Request to read one projection's details by id.
28#[derive(Clone, Debug, Serialize, Deserialize)]
29pub struct GetProjection {
30    pub v: u32,
31    pub id: String,
32}
33
34/// Request to list registered projections, optionally filtered. Empty filters
35/// list every projection. `topics` keeps projections bound to any of the named
36/// source topics. `name_contains` keeps those whose name contains the substring.
37/// `id_prefix` keeps those whose id starts with it. `search` is the single-box
38/// convenience that matches the substring against the name OR the id. The
39/// filters compose (AND).
40#[derive(Clone, Debug, Serialize, Deserialize)]
41pub struct ListProjections {
42    pub v: u32,
43    #[serde(default, skip_serializing_if = "Vec::is_empty")]
44    pub topics: Vec<String>,
45    #[serde(default, skip_serializing_if = "Option::is_none")]
46    pub name_contains: Option<String>,
47    #[serde(default, skip_serializing_if = "Option::is_none")]
48    pub id_prefix: Option<String>,
49    #[serde(default, skip_serializing_if = "Option::is_none")]
50    pub search: Option<String>,
51}
52
53/// Request to read one registered writer schema by id.
54#[derive(Clone, Debug, Serialize, Deserialize)]
55pub struct GetSchema {
56    pub v: u32,
57    pub id: u32,
58    /// The stream whose registry holds `id`. Absent for the deployment-wide
59    /// registry.
60    #[serde(default, skip_serializing_if = "Option::is_none")]
61    pub stream: Option<String>,
62}
63
64/// Request to list registered writer schemas, optionally filtered.
65/// `name_contains` keeps those whose optional name contains the substring, and
66/// an absent filter lists every schema. The filter lives here, not only on the
67/// HTTP query, so the server pushes it down rather than the client paging the
68/// whole set.
69#[derive(Clone, Debug, Serialize, Deserialize)]
70pub struct ListSchemas {
71    pub v: u32,
72    #[serde(default, skip_serializing_if = "Option::is_none")]
73    pub name_contains: Option<String>,
74    /// The stream whose registry is listed. Absent for the deployment-wide
75    /// registry.
76    #[serde(default, skip_serializing_if = "Option::is_none")]
77    pub stream: Option<String>,
78}
79
80/// The synchronous register request: no id, LaserData Cloud validates the
81/// definition, allocates the next free id, durably appends the control
82/// event, and replies `SchemaRegistered(id)`. Distinct from
83/// `ControlCommand::RegisterSchema`, the durable log form that carries the
84/// allocated id.
85#[derive(Clone, Debug, Serialize, Deserialize)]
86pub struct RegisterSchema {
87    pub v: u32,
88    pub source: SchemaSource,
89    #[serde(default, skip_serializing_if = "Option::is_none")]
90    pub name: Option<String>,
91    #[serde(default, skip_serializing_if = "Option::is_none")]
92    pub version: Option<u32>,
93    /// The stream whose registry allocates the id. Absent for the
94    /// deployment-wide registry.
95    #[serde(default, skip_serializing_if = "Option::is_none")]
96    pub stream: Option<String>,
97}
98
99/// Request to decode one record body under the schema registered for `id`,
100/// the same decode the projector runs on `agdx.sid`-stamped records. Read-only
101/// convenience for consoles and tools that hold a schema-first (Avro/Protobuf)
102/// payload and want its JSON form without re-implementing the codecs.
103#[derive(Clone, Debug, Serialize, Deserialize)]
104pub struct DecodeRecord {
105    pub v: u32,
106    pub id: u32,
107    #[serde(with = "crate::encoding::bin_bytes")]
108    pub payload: Vec<u8>,
109    /// The stream whose registry holds `id`. Absent for the deployment-wide
110    /// registry.
111    #[serde(default, skip_serializing_if = "Option::is_none")]
112    pub stream: Option<String>,
113}
114
115/// Reply to a registry browse: `Ok` with the result, or `Err`.
116#[derive(Clone, Debug, Serialize, Deserialize)]
117#[non_exhaustive]
118pub enum BrowseReply {
119    Ok(BrowseOutcome),
120    Err(QueryError),
121}
122
123/// The result of a registry browse, shaped per request.
124#[derive(Clone, Debug, Serialize, Deserialize)]
125#[non_exhaustive]
126pub enum BrowseOutcome {
127    /// `list_projections`: every registered projection.
128    Projections(Vec<ProjectionInfo>),
129    /// `get projection`: the projection with the requested id, or `None`.
130    Projection(Option<ProjectionInfo>),
131    /// `list schemas`: every known writer schema, active and tombstoned.
132    Schemas(Vec<SchemaInfo>),
133    /// `get schema`: the schema occupying the requested id, or `None`.
134    Schema(Option<SchemaInfo>),
135    /// `register schema`: the LaserData-Cloud-allocated id, already durably appended
136    /// to the control topic (visibility follows within the apply latency).
137    SchemaRegistered(u32),
138    /// `decode record`: the payload's JSON form under the requested schema,
139    /// or `None` when the body does not decode under it. A `serde_json::Value`
140    /// because the decoded shape is arbitrary (object, array, scalar) and the
141    /// LaserData Cloud encodes exactly that value into the reply.
142    Decoded(Option<serde_json::Value>),
143}
144
145#[cfg(all(test, feature = "cbor"))]
146mod tests {
147    use super::*;
148    use crate::codes::QUERY_OP_VERSION;
149    use crate::content::ContentType;
150    use crate::control::ProjectionBinding;
151    use crate::framing::{decode_named, encode_named};
152
153    #[test]
154    fn given_a_browse_reply_when_round_tripped_then_should_preserve_projection_details() {
155        let info = ProjectionInfo {
156            projection: Projection::builder("reading.v1")
157                .name("reading")
158                .version(1)
159                .content_type(ContentType::Json)
160                .fields(["reading_id", "cpu"])
161                .build(),
162            bindings: vec![
163                ProjectionBinding::builder()
164                    .source("fleet", "readings")
165                    .allow("reading.v1")
166                    .default_projection("reading.v1")
167                    .index("readings_rows")
168                    .build(),
169            ],
170        };
171        let reply = BrowseReply::Ok(BrowseOutcome::Projections(vec![info]));
172        let bytes = encode_named(&reply).expect("the reply serializes");
173        let back: BrowseReply = decode_named(&bytes).expect("the reply deserializes");
174        let BrowseReply::Ok(BrowseOutcome::Projections(list)) = back else {
175            panic!("expected an Ok(Projections) browse reply");
176        };
177        assert_eq!(list.len(), 1);
178        assert_eq!(list[0].projection.id.as_str(), "reading.v1");
179        assert_eq!(list[0].projection.extraction.fields.len(), 2);
180        assert_eq!(list[0].bindings[0].index, "readings_rows");
181    }
182
183    #[test]
184    fn given_a_decode_record_when_round_tripped_then_should_preserve_payload_bytes() {
185        let request = DecodeRecord {
186            v: QUERY_OP_VERSION,
187            id: 7,
188            payload: vec![0xff, 0x00, 0x10],
189            stream: None,
190        };
191        let bytes = encode_named(&request).expect("serializes");
192        let back: DecodeRecord = decode_named(&bytes).expect("deserializes");
193        assert_eq!(back.id, 7);
194        assert_eq!(back.payload, vec![0xff, 0x00, 0x10]);
195        assert_eq!(back.stream, None);
196    }
197
198    #[test]
199    fn given_schema_requests_with_a_stream_when_round_tripped_then_should_preserve_it() {
200        let get = GetSchema {
201            v: QUERY_OP_VERSION,
202            id: 7,
203            stream: Some("acme".to_owned()),
204        };
205        let back: GetSchema =
206            decode_named(&encode_named(&get).expect("serializes")).expect("deserializes");
207        assert_eq!((back.id, back.stream.as_deref()), (7, Some("acme")));
208        let list = ListSchemas {
209            v: QUERY_OP_VERSION,
210            name_contains: None,
211            stream: Some("acme".to_owned()),
212        };
213        let back: ListSchemas =
214            decode_named(&encode_named(&list).expect("serializes")).expect("deserializes");
215        assert_eq!(back.stream.as_deref(), Some("acme"));
216        let register = RegisterSchema {
217            v: QUERY_OP_VERSION,
218            source: SchemaSource::JsonSchema {
219                schema: "{}".to_owned(),
220            },
221            name: None,
222            version: None,
223            stream: Some("acme".to_owned()),
224        };
225        let back: RegisterSchema =
226            decode_named(&encode_named(&register).expect("serializes")).expect("deserializes");
227        assert_eq!(back.stream.as_deref(), Some("acme"));
228        let decode = DecodeRecord {
229            v: QUERY_OP_VERSION,
230            id: 7,
231            payload: vec![1],
232            stream: Some("acme".to_owned()),
233        };
234        let back: DecodeRecord =
235            decode_named(&encode_named(&decode).expect("serializes")).expect("deserializes");
236        assert_eq!(back.stream.as_deref(), Some("acme"));
237    }
238
239    #[test]
240    fn given_schema_requests_without_a_stream_when_encoded_then_should_omit_it() {
241        let get = GetSchema {
242            v: QUERY_OP_VERSION,
243            id: 7,
244            stream: None,
245        };
246        let list = ListSchemas {
247            v: QUERY_OP_VERSION,
248            name_contains: None,
249            stream: None,
250        };
251        for json in [
252            serde_json::to_string(&get).expect("serializes"),
253            serde_json::to_string(&list).expect("serializes"),
254        ] {
255            assert!(
256                !json.contains("stream"),
257                "an unset stream is omitted: {json}"
258            );
259        }
260    }
261}
laser_wire/browse.rs

laser_wire/
browse.rs
