laser_sdk/
projections.rs

1use crate::error::LaserError;
2use crate::laser::Laser;
3use laser_wire::browse::{
4    BrowseOutcome, BrowseReply, GetProjection, GetSchema, ListProjections, ListSchemas,
5    ProjectionInfo, RegisterSchema, SchemaInfo,
6};
7use laser_wire::codes::{
8    AGDX_GET_PROJECTION_CODE, AGDX_GET_SCHEMA_CODE, AGDX_LIST_PROJECTIONS_CODE,
9    AGDX_LIST_SCHEMAS_CODE, AGDX_REGISTER_SCHEMA_CODE, QUERY_OP_VERSION,
10};
11use laser_wire::control::{
12    ControlCommand, Projection, ProjectionBinding, ProjectionId, ProjectionKind, SchemaSource,
13    SourceSelector,
14};
15use laser_wire::framing::encode_named;
16use laser_wire::query::QueryError;
17use serde::Serialize;
18
19impl Laser {
20    /// Handle to the projection registry: `.register(projection)`,
21    /// `.drop(id)`, `.get(id).await`, and the filterable `.list()` browse.
22    /// Cheap to create, it borrows the connection.
23    pub fn projections(&self) -> Projections<'_> {
24        Projections { laser: self }
25    }
26
27    /// Handle to the binding surface: `.apply(binding)` routes a `(stream,
28    /// topic)` source into registered projections, `.remove(source, ..)` stops
29    /// it. Cheap to create, it borrows the connection.
30    pub fn bindings(&self) -> Bindings<'_> {
31        Bindings { laser: self }
32    }
33
34    /// Handle to the writer-schema registry (Avro, Protobuf):
35    /// `.register(def)`, `.drop(id)`, `.get(id).await`, `.list().await`.
36    /// Cheap to create, it borrows the connection.
37    pub fn schemas(&self) -> Schemas<'_> {
38        Schemas { laser: self }
39    }
40
41    // Read-only registry browse over the server's managed bridge, off the log. Gated
42    // on `managed_host`: browse has no topic transport, so anything but Laser Streams
43    // (including Apache Iggy) returns `Unsupported`.
44    async fn execute_browse(
45        &self,
46        code: u32,
47        request: &impl Serialize,
48    ) -> Result<BrowseOutcome, LaserError> {
49        let capabilities = self.capabilities().await;
50        if !capabilities.managed {
51            return Err(LaserError::unsupported(
52                "projections",
53                "projection browse is not served by this deployment",
54            ));
55        }
56        // Browse rides the query envelope version. Fail fast on advertised skew.
57        if let Some(versions) = capabilities.versions
58            && versions.query != QUERY_OP_VERSION
59        {
60            return Err(QueryError::Version {
61                expected: versions.query,
62                got: QUERY_OP_VERSION,
63            }
64            .into());
65        }
66        let payload = encode_named(request)
67            .map_err(|error| LaserError::Codec(format!("encode browse request: {error}")))?;
68        let payload = self.send_raw_with_response(code, payload).await?;
69        match crate::error::decode_managed_reply::<BrowseReply>(&payload)? {
70            BrowseReply::Ok(outcome) => Ok(outcome),
71            BrowseReply::Err(error) => Err(error.into()),
72            _ => Err(LaserError::Protocol(
73                "browse: unknown reply variant".to_owned(),
74            )),
75        }
76    }
77
78    // A control command whose resource names belong to this handle's stream.
79    async fn publish_scoped_control(&self, command: ControlCommand) -> Result<(), LaserError> {
80        let stream = self.resource_stream().map(str::to_owned);
81        self.publish_control_in(command, stream.as_deref()).await
82    }
83
84    fn scoped_projection_id(&self, id: &ProjectionId) -> ProjectionId {
85        ProjectionId::new(self.resource_name(id.as_str()))
86    }
87
88    fn local_projection_id(&self, id: ProjectionId) -> ProjectionId {
89        match self.local_resource_name(id.as_str()) {
90            Some(local) if local != id.as_str() => ProjectionId::new(local),
91            _ => id,
92        }
93    }
94
95    fn scoped_projection(&self, mut projection: Projection) -> Projection {
96        projection.id = self.scoped_projection_id(&projection.id);
97        projection
98    }
99
100    fn scoped_binding(&self, mut binding: ProjectionBinding) -> ProjectionBinding {
101        for id in &mut binding.allowed_projections {
102            *id = self.scoped_projection_id(id);
103        }
104        binding.default_projection = binding
105            .default_projection
106            .map(|id| self.scoped_projection_id(&id));
107        binding.index = self.resource_name(&binding.index);
108        binding
109    }
110
111    fn local_binding(&self, mut binding: ProjectionBinding) -> ProjectionBinding {
112        binding.allowed_projections = binding
113            .allowed_projections
114            .into_iter()
115            .map(|id| self.local_projection_id(id))
116            .collect();
117        binding.default_projection = binding
118            .default_projection
119            .map(|id| self.local_projection_id(id));
120        if let Some(local) = self.local_resource_name(&binding.index) {
121            binding.index = local.to_owned();
122        }
123        binding
124    }
125
126    fn local_projection_info(&self, mut info: ProjectionInfo) -> ProjectionInfo {
127        info.projection.id = self.local_projection_id(info.projection.id);
128        info.bindings = info
129            .bindings
130            .into_iter()
131            .map(|binding| self.local_binding(binding))
132            .collect();
133        info
134    }
135}
136
137/// Handle to the projection registry, built with
138/// [`Laser::projections`](crate::laser::Laser::projections). Writes
139/// (`register` / `drop`) publish control commands to `<ops>/control.commands`
140/// and are applied asynchronously by LaserData Cloud (202-accepted semantics: poll
141/// `.get(id)` to observe the apply). Reads (`get` / `list`) are managed-command
142/// browses served off the registry snapshot.
143pub struct Projections<'a> {
144    laser: &'a Laser,
145}
146
147impl<'a> Projections<'a> {
148    /// Register a [`Projection`] on LaserData Cloud by publishing a
149    /// `RegisterProjection` control command. LaserData Cloud creates the backing
150    /// table and starts applying the projection.
151    ///
152    /// Rejects a graph projection (one built with
153    /// [`ProjectionBuilder::graph`](crate::query::ProjectionBuilder::graph)) with
154    /// [`LaserError::Invalid`]: a graph materializes nodes and edges, not a
155    /// queryable row table, so it registers through
156    /// [`register_graph`](Self::register_graph) instead.
157    pub async fn register(&self, projection: Projection) -> Result<(), LaserError> {
158        if projection.kind == ProjectionKind::Graph {
159            return Err(LaserError::Invalid(format!(
160                "projection `{}` is a graph projection. Register it with `register_graph`",
161                projection.id
162            )));
163        }
164        let projection = self.laser.scoped_projection(projection);
165        self.laser
166            .publish_scoped_control(ControlCommand::RegisterProjection(projection))
167            .await
168    }
169
170    /// Drop a projection by publishing a `DropProjection` control command.
171    /// LaserData Cloud stops applying it, and existing materialized rows are
172    /// left untouched. `id` is the projection ref (e.g. `"reading.v1"`).
173    pub async fn drop(&self, id: impl Into<String>) -> Result<(), LaserError> {
174        let id = self.laser.resource_name(&id.into());
175        self.laser
176            .publish_scoped_control(ControlCommand::DropProjection(id))
177            .await
178    }
179
180    /// Register a graph projection: a [`Projection`] with `kind = Graph` and an
181    /// entity schema (build it with [`Projection::builder`]`.graph(schema)`). It
182    /// records the named knowledge graph so it is discoverable, and declares the
183    /// node and edge extraction plan. A distinct command from
184    /// [`register`](Self::register) so a backend can gate graph registration on
185    /// the `graph` capability.
186    ///
187    /// Graph data is written content-addressed through the `graph` feature's
188    /// `GraphHandle::upsert`, or by the managed projector when this projection
189    /// is bound to a source topic via [`Bindings::apply`] (it applies the
190    /// entity schema to each record and upserts the extracted nodes and edges).
191    /// Registering a graph projection creates no row table. Rejects a non-graph
192    /// projection with [`LaserError::Invalid`].
193    pub async fn register_graph(&self, projection: Projection) -> Result<(), LaserError> {
194        if projection.kind != ProjectionKind::Graph || projection.entity_schema.is_none() {
195            return Err(LaserError::Invalid(format!(
196                "projection `{}` is not a graph projection. Build it with `Projection::builder(..).graph(schema)` or register it with `register`",
197                projection.id
198            )));
199        }
200        let projection = self.laser.scoped_projection(projection);
201        self.laser
202            .publish_scoped_control(ControlCommand::RegisterGraph(projection))
203            .await
204    }
205
206    /// Drop the graph projection registered under `id` by publishing a
207    /// `DropGraph` control command. The materialized nodes and edges are left
208    /// untouched, the same as [`drop`](Self::drop) for a row projection.
209    pub async fn drop_graph(&self, id: impl Into<String>) -> Result<(), LaserError> {
210        let id = self.laser.resource_name(&id.into());
211        self.laser
212            .publish_scoped_control(ControlCommand::DropGraph(id))
213            .await
214    }
215
216    /// Read one projection's details by `id` (schema, content type, indexed
217    /// fields, bindings), or `None` when no projection has that id. Names
218    /// under this handle's stream come back as the caller wrote them.
219    pub async fn get(&self, id: impl Into<String>) -> Result<Option<ProjectionInfo>, LaserError> {
220        let request = GetProjection {
221            v: QUERY_OP_VERSION,
222            id: self.laser.resource_name(&id.into()),
223        };
224        match self
225            .laser
226            .execute_browse(AGDX_GET_PROJECTION_CODE, &request)
227            .await?
228        {
229            BrowseOutcome::Projection(info) => {
230                Ok(info.map(|info| self.laser.local_projection_info(info)))
231            }
232            _ => Err(LaserError::Protocol(
233                "get: unexpected browse outcome".to_owned(),
234            )),
235        }
236    }
237
238    /// Browse the registry. Returns a builder: narrow with `.for_topic` /
239    /// `.name_contains` / `.id_prefix`, then `.fetch().await` for the matching
240    /// projections (each with its extraction schema, content type, indexed
241    /// fields, and bindings). No filter lists every projection.
242    pub fn list(&self) -> ProjectionsRequest<'a> {
243        ProjectionsRequest {
244            laser: self.laser,
245            topics: Vec::new(),
246            name_contains: None,
247            id_prefix: None,
248            search: None,
249        }
250    }
251}
252
253/// Handle to the binding surface, built with
254/// [`Laser::bindings`](crate::laser::Laser::bindings). Writes publish control
255/// commands to `<ops>/control.commands` (202-accepted, applied asynchronously
256/// by LaserData Cloud). Bindings are browsed through the projections they route to:
257/// `laser.projections().get(id)` returns each projection with its bindings.
258pub struct Bindings<'a> {
259    laser: &'a Laser,
260}
261
262impl Bindings<'_> {
263    /// Apply a [`ProjectionBinding`] by publishing an `ApplyBinding` control
264    /// command. LaserData Cloud starts a projector for the binding's `(stream,
265    /// topic)` source. Register the referenced projection first, or LaserData Cloud
266    /// rejects the binding.
267    pub async fn apply(&self, binding: ProjectionBinding) -> Result<(), LaserError> {
268        let binding = self.laser.scoped_binding(binding);
269        self.laser
270            .publish_scoped_control(ControlCommand::ApplyBinding(binding))
271            .await
272    }
273
274    /// Remove a binding by publishing a `RemoveBinding` control command.
275    /// LaserData Cloud stops the projector for `source`, so new records on that
276    /// `(stream, topic)` no longer materialize. Rows already written stay.
277    /// `projection_ref` scopes the removal to a single allowed projection when
278    /// set, else removes the whole binding.
279    pub async fn remove(
280        &self,
281        source: SourceSelector,
282        projection_ref: Option<String>,
283    ) -> Result<(), LaserError> {
284        let projection_ref = projection_ref.map(|id| self.laser.resource_name(&id));
285        self.laser
286            .publish_scoped_control(ControlCommand::RemoveBinding {
287                source,
288                projection_ref,
289            })
290            .await
291    }
292}
293
294/// Handle to the writer-schema registry (Avro, Protobuf), built with
295/// [`Laser::schemas`](crate::laser::Laser::schemas). Writes (`register` /
296/// `drop`) publish control commands and are applied asynchronously by the
297/// LaserData Cloud (202-accepted semantics: poll `.get(id)` until `Some` to observe the
298/// apply). Reads are managed-command browses.
299pub struct Schemas<'a> {
300    laser: &'a Laser,
301}
302
303impl Schemas<'_> {
304    /// Register a writer schema. Synchronous: LaserData Cloud validates that the
305    /// definition compiles, allocates the next free id (ids are
306    /// LaserData-Cloud-allocated and permanent, concurrent callers never collide),
307    /// durably appends the control event, and `.send()` returns the id.
308    /// Producers then stamp it on `agdx.sid` via the publish builders'
309    /// `.schema_id(..)`. Browse visibility follows within the apply latency.
310    pub fn register(&self, source: SchemaSource) -> RegisterSchemaRequest<'_> {
311        RegisterSchemaRequest {
312            laser: self.laser,
313            source,
314            name: None,
315            version: None,
316        }
317    }
318
319    /// Drop the writer schema registered under `id` by publishing a
320    /// `DropSchema` control command. **Drop tombstones, it does not delete**:
321    /// the id leaves the active set but records stamped with it keep
322    /// decoding, and the id stays reserved against re-registration with a
323    /// different definition. Dropping an unknown id is a no-op.
324    pub async fn drop(&self, id: u32) -> Result<(), LaserError> {
325        self.laser
326            .publish_scoped_control(ControlCommand::DropSchema(id))
327            .await
328    }
329
330    /// Read the writer schema occupying `id` (active or tombstoned, see
331    /// [`SchemaInfo::dropped`]), or `None` when the id is free. Read-only browse.
332    pub async fn get(&self, id: u32) -> Result<Option<SchemaInfo>, LaserError> {
333        let request = GetSchema {
334            v: QUERY_OP_VERSION,
335            id,
336            stream: self.laser.resource_stream().map(str::to_owned),
337        };
338        match self
339            .laser
340            .execute_browse(AGDX_GET_SCHEMA_CODE, &request)
341            .await?
342        {
343            BrowseOutcome::Schema(info) => Ok(info),
344            _ => Err(LaserError::Protocol(
345                "schema: unexpected browse outcome".to_owned(),
346            )),
347        }
348    }
349
350    /// List every known writer schema, active and tombstoned (see
351    /// [`SchemaInfo::dropped`]). Read-only browse, listing every schema.
352    pub async fn list(&self) -> Result<Vec<SchemaInfo>, LaserError> {
353        let request = ListSchemas {
354            v: QUERY_OP_VERSION,
355            name_contains: None,
356            stream: self.laser.resource_stream().map(str::to_owned),
357        };
358        match self
359            .laser
360            .execute_browse(AGDX_LIST_SCHEMAS_CODE, &request)
361            .await?
362        {
363            BrowseOutcome::Schemas(list) => Ok(list),
364            _ => Err(LaserError::Protocol(
365                "schemas: unexpected browse outcome".to_owned(),
366            )),
367        }
368    }
369}
370
371/// Builder for the synchronous schema register. Terminal `.send().await`
372/// returns LaserData Cloud-allocated id.
373#[must_use = "call .send().await to register the schema"]
374pub struct RegisterSchemaRequest<'a> {
375    laser: &'a Laser,
376    source: SchemaSource,
377    name: Option<String>,
378    version: Option<u32>,
379}
380
381impl RegisterSchemaRequest<'_> {
382    /// Optional human label, pure metadata: stored and returned by the
383    /// LaserData Cloud, never dispatched on and not unique.
384    pub fn name(mut self, name: impl Into<String>) -> Self {
385        self.name = Some(name.into());
386        self
387    }
388
389    /// Optional caller-tracked schema version, pure metadata: stored and
390    /// returned by LaserData Cloud, never dispatched on.
391    pub fn version(mut self, version: u32) -> Self {
392        self.version = Some(version);
393        self
394    }
395
396    /// Execute the register and return LaserData Cloud-allocated id. Typed errors:
397    /// an uncompilable definition answers `Unsupported` (nothing allocated),
398    /// transport/server trouble answers `Backend`.
399    pub async fn send(self) -> Result<u32, LaserError> {
400        let request = RegisterSchema {
401            v: QUERY_OP_VERSION,
402            source: self.source,
403            name: self.name,
404            version: self.version,
405            stream: self.laser.resource_stream().map(str::to_owned),
406        };
407        match self
408            .laser
409            .execute_browse(AGDX_REGISTER_SCHEMA_CODE, &request)
410            .await?
411        {
412            BrowseOutcome::SchemaRegistered(id) => Ok(id),
413            _ => Err(LaserError::Protocol(
414                "register schema: unexpected outcome".to_owned(),
415            )),
416        }
417    }
418}
419
420/// Fluent builder for [`Projections::list`]. Narrow the registry browse with
421/// `.for_topic` / `.for_topics` / `.name_contains` / `.id_prefix`, then
422/// `.fetch().await`. No filter lists every projection.
423#[must_use = "call .fetch().await to list the projections"]
424pub struct ProjectionsRequest<'a> {
425    laser: &'a Laser,
426    topics: Vec<String>,
427    name_contains: Option<String>,
428    id_prefix: Option<String>,
429    search: Option<String>,
430}
431
432impl<'a> ProjectionsRequest<'a> {
433    /// Keep only projections bound to `topic`. Repeatable.
434    pub fn for_topic(mut self, topic: impl Into<String>) -> Self {
435        self.topics.push(topic.into());
436        self
437    }
438
439    /// Keep only projections bound to any of `topics`.
440    pub fn for_topics<S: Into<String>>(mut self, topics: impl IntoIterator<Item = S>) -> Self {
441        self.topics.extend(topics.into_iter().map(Into::into));
442        self
443    }
444
445    /// Keep only projections whose name contains `substring`.
446    pub fn name_contains(mut self, substring: impl Into<String>) -> Self {
447        self.name_contains = Some(substring.into());
448        self
449    }
450
451    /// Keep only projections whose id starts with `prefix`, the caller's own
452    /// name for it.
453    pub fn id_prefix(mut self, prefix: impl Into<String>) -> Self {
454        self.id_prefix = Some(prefix.into());
455        self
456    }
457
458    /// Keep only projections whose name or id matches `substring`, the
459    /// single-box convenience filter. Composes with the narrower filters.
460    pub fn search(mut self, substring: impl Into<String>) -> Self {
461        self.search = Some(substring.into());
462        self
463    }
464
465    /// Run the browse and return the matching projections. A handle that
466    /// scopes its resources to a stream lists only that stream's projections,
467    /// under the ids the caller gave them.
468    pub async fn fetch(self) -> Result<Vec<ProjectionInfo>, LaserError> {
469        let id_prefix = match self.laser.resource_stream() {
470            Some(stream) => Some(laser_wire::authz::scoped_resource(
471                stream,
472                self.id_prefix.as_deref().unwrap_or_default(),
473            )),
474            None => self.id_prefix,
475        };
476        let request = ListProjections {
477            v: QUERY_OP_VERSION,
478            topics: self.topics,
479            name_contains: self.name_contains,
480            id_prefix,
481            search: self.search,
482        };
483        match self
484            .laser
485            .execute_browse(AGDX_LIST_PROJECTIONS_CODE, &request)
486            .await?
487        {
488            BrowseOutcome::Projections(list) => Ok(list
489                .into_iter()
490                .filter(|info| {
491                    self.laser
492                        .local_resource_name(info.projection.id.as_str())
493                        .is_some()
494                })
495                .map(|info| self.laser.local_projection_info(info))
496                .collect()),
497            _ => Err(LaserError::Protocol(
498                "list: unexpected browse outcome".to_owned(),
499            )),
500        }
501    }
502}
503
504#[cfg(test)]
505mod tests {
506    use crate::laser::{Laser, ResourceNaming};
507    use laser_wire::browse::ProjectionInfo;
508    use laser_wire::control::{Projection, ProjectionBinding};
509
510    fn laser() -> Laser {
511        Laser::from_client(crate::iggy::prelude::IggyClient::default()).with_default_stream("acme")
512    }
513
514    fn binding() -> ProjectionBinding {
515        ProjectionBinding::builder()
516            .source("acme", "readings")
517            .allow("reading.v1")
518            .default_projection("reading.v1")
519            .index("readings_rows")
520            .build()
521    }
522
523    #[test]
524    fn given_a_default_stream_when_scoping_a_binding_then_should_scope_refs_and_index() {
525        let scoped = laser().scoped_binding(binding());
526        assert_eq!(
527            scoped.allowed_projections[0].as_str(),
528            "stream:acme/reading.v1"
529        );
530        assert_eq!(
531            scoped.default_projection.as_ref().map(|id| id.as_str()),
532            Some("stream:acme/reading.v1")
533        );
534        assert_eq!(scoped.index, "stream:acme/readings_rows");
535        assert_eq!(scoped.source.stream, "acme", "the source is an Iggy stream");
536    }
537
538    #[test]
539    fn given_a_scoped_projection_info_when_localized_then_should_return_the_caller_names() {
540        let laser = laser();
541        let projection = laser.scoped_projection(Projection::builder("reading.v1").build());
542        assert_eq!(projection.id.as_str(), "stream:acme/reading.v1");
543        let info = laser.local_projection_info(ProjectionInfo {
544            projection,
545            bindings: vec![laser.scoped_binding(binding())],
546        });
547        assert_eq!(info.projection.id.as_str(), "reading.v1");
548        assert_eq!(
549            info.bindings[0].allowed_projections[0].as_str(),
550            "reading.v1"
551        );
552        assert_eq!(info.bindings[0].index, "readings_rows");
553    }
554
555    #[test]
556    fn given_bare_naming_when_scoping_a_binding_then_should_leave_it_unchanged() {
557        let laser = laser().with_resource_naming(ResourceNaming::Bare);
558        assert_eq!(laser.scoped_binding(binding()), binding());
559    }
560}
laser_sdk/projections.rs

laser_sdk/
projections.rs
