dex_sdk/
persistence.rs

1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Super Durable Source License 1.0.
4// You may not use this file except in compliance with the License.
5// See the LICENSE file in the repository root.
6//
7// SPDX-License-Identifier: LicenseRef-Super-Durable-1.0
8
9use crate::{Attribute, AttributeIndex, AttributeMap, Channel, ChannelMap, Stream};
10
11#[derive(Clone, Debug, Default, Eq, PartialEq)]
12/// Declares the Attributes, Channels, and Streams a Flow owns.
13///
14/// Return a schema from [`crate::Flow::persistence`]. Definitions must have unique names,
15/// and all values accessed by Step or RPC code must appear in the schema.
16///
17/// # Examples
18///
19/// ```
20/// use dex_sdk::{Attribute, Channel, PersistenceSchema};
21///
22/// let status = Attribute::<String>::new("status");
23/// let commands = Channel::<String>::new("commands");
24/// let schema = PersistenceSchema::new().attribute(&status).channel(&commands);
25/// ```
26pub struct PersistenceSchema {
27    definitions: Vec<PersistenceDefinition>,
28}
29
30#[derive(Clone, Debug, Eq, PartialEq)]
31pub(crate) struct PersistenceDefinition {
32    pub(crate) name: String,
33    pub(crate) kind: PersistenceKind,
34    pub(crate) index: Option<AttributeIndex>,
35    pub(crate) sync_to_attribute_store: bool,
36    pub(crate) stream_identity: Option<usize>,
37    pub(crate) max_estimated_bytes: Option<i64>,
38}
39
40#[derive(Clone, Copy, Debug, Eq, PartialEq)]
41pub(crate) enum PersistenceKind {
42    Attribute,
43    AttributeMap,
44    Channel,
45    ChannelMap,
46    Stream,
47}
48
49impl PersistenceSchema {
50    /// Creates an empty schema.
51    pub fn new() -> Self {
52        Self::default()
53    }
54
55    /// Adds one single-value Attribute definition.
56    pub fn attribute<T>(mut self, attribute: &Attribute<T>) -> Self {
57        self.add(
58            attribute.name(),
59            PersistenceKind::Attribute,
60            attribute.index().cloned(),
61            attribute.is_sync_to_attribute_store(),
62            None,
63            None,
64        );
65        self
66    }
67
68    /// Adds one keyed Attribute-map definition.
69    pub fn attribute_map<T>(mut self, attribute: &AttributeMap<T>) -> Self {
70        self.add(
71            attribute.name(),
72            PersistenceKind::AttributeMap,
73            attribute.index().cloned(),
74            attribute.is_sync_to_attribute_store(),
75            None,
76            None,
77        );
78        self
79    }
80
81    /// Adds one Channel definition.
82    pub fn channel<T>(mut self, channel: &Channel<T>) -> Self {
83        self.add(
84            channel.name(),
85            PersistenceKind::Channel,
86            None,
87            false,
88            None,
89            None,
90        );
91        self
92    }
93
94    /// Adds one keyed Channel-map definition.
95    pub fn channel_map<T>(mut self, channel: &ChannelMap<T>) -> Self {
96        self.add(
97            channel.name(),
98            PersistenceKind::ChannelMap,
99            None,
100            false,
101            None,
102            None,
103        );
104        self
105    }
106
107    /// Adds one best-effort Stream definition.
108    pub fn stream<T>(mut self, stream: &Stream<T>) -> Self {
109        self.add(
110            stream.name(),
111            PersistenceKind::Stream,
112            None,
113            false,
114            Some(stream.identity()),
115            Some(stream.max_estimated_bytes()),
116        );
117        self
118    }
119
120    pub(crate) fn definitions(&self) -> &[PersistenceDefinition] {
121        &self.definitions
122    }
123
124    fn add(
125        &mut self,
126        name: &str,
127        kind: PersistenceKind,
128        index: Option<AttributeIndex>,
129        sync_to_attribute_store: bool,
130        stream_identity: Option<usize>,
131        max_estimated_bytes: Option<i64>,
132    ) {
133        self.definitions.push(PersistenceDefinition {
134            name: name.to_string(),
135            kind,
136            index,
137            sync_to_attribute_store,
138            stream_identity,
139            max_estimated_bytes,
140        });
141    }
142}
143
144#[cfg(test)]
145mod tests {
146    use super::PersistenceSchema;
147    use crate::{Attribute, AttributeMap, Stream};
148
149    #[test]
150    fn registration_metadata_retains_sync_configuration() {
151        let attribute = Attribute::<String>::new("plain");
152        let attribute_map = AttributeMap::<String>::new("map").sync_to_attribute_store();
153        let schema = PersistenceSchema::new()
154            .attribute(&attribute)
155            .attribute_map(&attribute_map);
156
157        assert!(!schema.definitions[0].sync_to_attribute_store);
158        assert!(schema.definitions[1].sync_to_attribute_store);
159    }
160
161    #[test]
162    fn registration_metadata_retains_stream_identity_and_capacity() {
163        let stream = Stream::<String>::new("thinking", 1_048_576);
164        let schema = PersistenceSchema::new().stream(&stream);
165
166        assert_eq!(
167            schema.definitions[0].stream_identity,
168            Some(stream.identity())
169        );
170        assert_eq!(schema.definitions[0].max_estimated_bytes, Some(1_048_576));
171    }
172}
dex_sdk/persistence.rs

dex_sdk/
persistence.rs
