dex_sdk/
persistence.rs

1// Copyright (c) 2026 Super Durable, Inc.
2//
3// Licensed under the Sustainable Use 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-Sustainable-Use-1.0
8
9use crate::{Attribute, AttributeIndex, AttributeMap, Channel, ChannelMap, Stream};
10
11#[derive(Clone, Debug, Default)]
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// Schema equality predates value type tracking, so it keeps ignoring value types.
31impl PartialEq for PersistenceSchema {
32    fn eq(&self, other: &Self) -> bool {
33        self.definitions.len() == other.definitions.len()
34            && self.definitions.iter().zip(&other.definitions).all(
35                |(definition, other_definition)| {
36                    definition.is_equal_ignoring_value_type(other_definition)
37                },
38            )
39    }
40}
41
42impl Eq for PersistenceSchema {}
43
44#[derive(Clone, Debug, Eq, PartialEq)]
45pub(crate) struct PersistenceDefinition {
46    pub(crate) name: String,
47    pub(crate) kind: PersistenceKind,
48    pub(crate) index: Option<AttributeIndex>,
49    pub(crate) sync_to_attribute_store: bool,
50    pub(crate) is_hard_delete_on_null: bool,
51    pub(crate) numeric_or_boolean_type_name: Option<&'static str>,
52    pub(crate) value_type_name: Option<&'static str>,
53    pub(crate) stream_identity: Option<usize>,
54    pub(crate) stream_capacity_bytes: Option<i64>,
55}
56
57impl PersistenceDefinition {
58    pub(crate) fn new(name: &str, kind: PersistenceKind) -> Self {
59        Self {
60            name: name.to_string(),
61            kind,
62            index: None,
63            sync_to_attribute_store: false,
64            is_hard_delete_on_null: false,
65            numeric_or_boolean_type_name: None,
66            value_type_name: None,
67            stream_identity: None,
68            stream_capacity_bytes: None,
69        }
70    }
71
72    fn is_equal_ignoring_value_type(&self, other: &Self) -> bool {
73        let Self {
74            name,
75            kind,
76            index,
77            sync_to_attribute_store,
78            is_hard_delete_on_null,
79            numeric_or_boolean_type_name,
80            value_type_name: _,
81            stream_identity,
82            stream_capacity_bytes,
83        } = self;
84        *name == other.name
85            && *kind == other.kind
86            && *index == other.index
87            && *sync_to_attribute_store == other.sync_to_attribute_store
88            && *is_hard_delete_on_null == other.is_hard_delete_on_null
89            && *numeric_or_boolean_type_name == other.numeric_or_boolean_type_name
90            && *stream_identity == other.stream_identity
91            && *stream_capacity_bytes == other.stream_capacity_bytes
92    }
93
94    pub(crate) fn for_attribute<T>(attribute: &Attribute<T>) -> Self {
95        Self {
96            index: attribute.index().cloned(),
97            sync_to_attribute_store: attribute.is_sync_to_attribute_store(),
98            is_hard_delete_on_null: attribute.is_hard_delete_on_null(),
99            numeric_or_boolean_type_name: attribute.numeric_or_boolean_type_name(),
100            value_type_name: Some(std::any::type_name::<T>()),
101            ..Self::new(attribute.name(), PersistenceKind::Attribute)
102        }
103    }
104
105    pub(crate) fn for_attribute_map<T>(attribute_map: &AttributeMap<T>) -> Self {
106        Self {
107            index: attribute_map.index().cloned(),
108            sync_to_attribute_store: attribute_map.is_sync_to_attribute_store(),
109            is_hard_delete_on_null: attribute_map.is_hard_delete_on_null(),
110            numeric_or_boolean_type_name: attribute_map.numeric_or_boolean_type_name(),
111            value_type_name: Some(std::any::type_name::<T>()),
112            ..Self::new(attribute_map.name(), PersistenceKind::AttributeMap)
113        }
114    }
115}
116
117#[derive(Clone, Copy, Debug, Eq, PartialEq)]
118pub(crate) enum PersistenceKind {
119    Attribute,
120    AttributeMap,
121    Channel,
122    ChannelMap,
123    Stream,
124}
125
126impl PersistenceSchema {
127    /// Creates an empty schema.
128    pub fn new() -> Self {
129        Self::default()
130    }
131
132    /// Adds one single-value Attribute definition.
133    pub fn attribute<T>(mut self, attribute: &Attribute<T>) -> Self {
134        self.definitions
135            .push(PersistenceDefinition::for_attribute(attribute));
136        self
137    }
138
139    /// Adds one keyed Attribute-map definition.
140    pub fn attribute_map<T>(mut self, attribute: &AttributeMap<T>) -> Self {
141        self.definitions
142            .push(PersistenceDefinition::for_attribute_map(attribute));
143        self
144    }
145
146    /// Adds one Channel definition.
147    pub fn channel<T>(mut self, channel: &Channel<T>) -> Self {
148        self.definitions.push(PersistenceDefinition::new(
149            channel.name(),
150            PersistenceKind::Channel,
151        ));
152        self
153    }
154
155    /// Adds one keyed Channel-map definition.
156    pub fn channel_map<T>(mut self, channel: &ChannelMap<T>) -> Self {
157        self.definitions.push(PersistenceDefinition::new(
158            channel.name(),
159            PersistenceKind::ChannelMap,
160        ));
161        self
162    }
163
164    /// Adds one best-effort Stream definition.
165    pub fn stream<T>(mut self, stream: &Stream<T>) -> Self {
166        self.definitions.push(PersistenceDefinition {
167            stream_identity: Some(stream.identity()),
168            stream_capacity_bytes: Some(stream.stream_capacity_bytes()),
169            ..PersistenceDefinition::new(stream.name(), PersistenceKind::Stream)
170        });
171        self
172    }
173
174    pub(crate) fn definitions(&self) -> &[PersistenceDefinition] {
175        &self.definitions
176    }
177}
178
179#[cfg(test)]
180mod tests {
181    use super::PersistenceSchema;
182    use crate::{Attribute, AttributeMap, Stream};
183
184    #[test]
185    fn registration_metadata_retains_sync_configuration() {
186        let attribute = Attribute::<String>::new("plain");
187        let attribute_map = AttributeMap::<String>::new("map").sync_to_attribute_store();
188        let schema = PersistenceSchema::new()
189            .attribute(&attribute)
190            .attribute_map(&attribute_map);
191
192        assert!(!schema.definitions[0].sync_to_attribute_store);
193        assert!(schema.definitions[1].sync_to_attribute_store);
194    }
195
196    #[test]
197    fn registration_metadata_retains_hard_delete_configuration() {
198        let plain = Attribute::<String>::new("plain");
199        let secret = Attribute::<String>::new("secret").hard_delete_on_null();
200        let counter = Attribute::<Option<u64>>::new("counter").hard_delete_on_null();
201        let secrets = AttributeMap::<Vec<u8>>::new("secrets").hard_delete_on_null();
202        let schema = PersistenceSchema::new()
203            .attribute(&plain)
204            .attribute(&secret)
205            .attribute(&counter)
206            .attribute_map(&secrets.clone());
207
208        assert!(!schema.definitions[0].is_hard_delete_on_null);
209        assert!(schema.definitions[1].is_hard_delete_on_null);
210        assert_eq!(schema.definitions[1].numeric_or_boolean_type_name, None);
211        assert!(
212            schema.definitions[2]
213                .numeric_or_boolean_type_name
214                .is_some_and(|type_name| type_name.contains("u64"))
215        );
216        assert!(schema.definitions[3].is_hard_delete_on_null);
217        assert_eq!(schema.definitions[3].numeric_or_boolean_type_name, None);
218    }
219
220    #[test]
221    fn schema_equality_ignores_attribute_value_types() {
222        let text = PersistenceSchema::new().attribute(&Attribute::<String>::new("status"));
223        let number = PersistenceSchema::new().attribute(&Attribute::<i64>::new("status"));
224        let other_name = PersistenceSchema::new().attribute(&Attribute::<String>::new("state"));
225
226        assert_eq!(text, number);
227        assert_ne!(text, other_name);
228    }
229
230    #[test]
231    fn registration_metadata_retains_stream_identity_and_capacity() {
232        let stream = Stream::<String>::new("thinking", 1_048_576);
233        let schema = PersistenceSchema::new().stream(&stream);
234
235        assert_eq!(
236            schema.definitions[0].stream_identity,
237            Some(stream.identity())
238        );
239        assert_eq!(schema.definitions[0].stream_capacity_bytes, Some(1_048_576));
240    }
241}
dex_sdk/persistence.rs

dex_sdk/
persistence.rs
