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, 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) is_hard_delete_on_null: bool,
37    pub(crate) numeric_or_boolean_type_name: Option<&'static str>,
38    pub(crate) stream_identity: Option<usize>,
39    pub(crate) stream_capacity_bytes: Option<i64>,
40}
41
42impl PersistenceDefinition {
43    pub(crate) fn new(name: &str, kind: PersistenceKind) -> Self {
44        Self {
45            name: name.to_string(),
46            kind,
47            index: None,
48            sync_to_attribute_store: false,
49            is_hard_delete_on_null: false,
50            numeric_or_boolean_type_name: None,
51            stream_identity: None,
52            stream_capacity_bytes: None,
53        }
54    }
55}
56
57#[derive(Clone, Copy, Debug, Eq, PartialEq)]
58pub(crate) enum PersistenceKind {
59    Attribute,
60    AttributeMap,
61    Channel,
62    ChannelMap,
63    Stream,
64}
65
66impl PersistenceSchema {
67    /// Creates an empty schema.
68    pub fn new() -> Self {
69        Self::default()
70    }
71
72    /// Adds one single-value Attribute definition.
73    pub fn attribute<T>(mut self, attribute: &Attribute<T>) -> Self {
74        self.definitions.push(PersistenceDefinition {
75            index: attribute.index().cloned(),
76            sync_to_attribute_store: attribute.is_sync_to_attribute_store(),
77            is_hard_delete_on_null: attribute.is_hard_delete_on_null(),
78            numeric_or_boolean_type_name: attribute.numeric_or_boolean_type_name(),
79            ..PersistenceDefinition::new(attribute.name(), PersistenceKind::Attribute)
80        });
81        self
82    }
83
84    /// Adds one keyed Attribute-map definition.
85    pub fn attribute_map<T>(mut self, attribute: &AttributeMap<T>) -> Self {
86        self.definitions.push(PersistenceDefinition {
87            index: attribute.index().cloned(),
88            sync_to_attribute_store: attribute.is_sync_to_attribute_store(),
89            is_hard_delete_on_null: attribute.is_hard_delete_on_null(),
90            numeric_or_boolean_type_name: attribute.numeric_or_boolean_type_name(),
91            ..PersistenceDefinition::new(attribute.name(), PersistenceKind::AttributeMap)
92        });
93        self
94    }
95
96    /// Adds one Channel definition.
97    pub fn channel<T>(mut self, channel: &Channel<T>) -> Self {
98        self.definitions.push(PersistenceDefinition::new(
99            channel.name(),
100            PersistenceKind::Channel,
101        ));
102        self
103    }
104
105    /// Adds one keyed Channel-map definition.
106    pub fn channel_map<T>(mut self, channel: &ChannelMap<T>) -> Self {
107        self.definitions.push(PersistenceDefinition::new(
108            channel.name(),
109            PersistenceKind::ChannelMap,
110        ));
111        self
112    }
113
114    /// Adds one best-effort Stream definition.
115    pub fn stream<T>(mut self, stream: &Stream<T>) -> Self {
116        self.definitions.push(PersistenceDefinition {
117            stream_identity: Some(stream.identity()),
118            stream_capacity_bytes: Some(stream.stream_capacity_bytes()),
119            ..PersistenceDefinition::new(stream.name(), PersistenceKind::Stream)
120        });
121        self
122    }
123
124    pub(crate) fn definitions(&self) -> &[PersistenceDefinition] {
125        &self.definitions
126    }
127}
128
129#[cfg(test)]
130mod tests {
131    use super::PersistenceSchema;
132    use crate::{Attribute, AttributeMap, Stream};
133
134    #[test]
135    fn registration_metadata_retains_sync_configuration() {
136        let attribute = Attribute::<String>::new("plain");
137        let attribute_map = AttributeMap::<String>::new("map").sync_to_attribute_store();
138        let schema = PersistenceSchema::new()
139            .attribute(&attribute)
140            .attribute_map(&attribute_map);
141
142        assert!(!schema.definitions[0].sync_to_attribute_store);
143        assert!(schema.definitions[1].sync_to_attribute_store);
144    }
145
146    #[test]
147    fn registration_metadata_retains_hard_delete_configuration() {
148        let plain = Attribute::<String>::new("plain");
149        let secret = Attribute::<String>::new("secret").hard_delete_on_null();
150        let counter = Attribute::<Option<u64>>::new("counter").hard_delete_on_null();
151        let secrets = AttributeMap::<Vec<u8>>::new("secrets").hard_delete_on_null();
152        let schema = PersistenceSchema::new()
153            .attribute(&plain)
154            .attribute(&secret)
155            .attribute(&counter)
156            .attribute_map(&secrets.clone());
157
158        assert!(!schema.definitions[0].is_hard_delete_on_null);
159        assert!(schema.definitions[1].is_hard_delete_on_null);
160        assert_eq!(schema.definitions[1].numeric_or_boolean_type_name, None);
161        assert!(
162            schema.definitions[2]
163                .numeric_or_boolean_type_name
164                .is_some_and(|type_name| type_name.contains("u64"))
165        );
166        assert!(schema.definitions[3].is_hard_delete_on_null);
167        assert_eq!(schema.definitions[3].numeric_or_boolean_type_name, None);
168    }
169
170    #[test]
171    fn registration_metadata_retains_stream_identity_and_capacity() {
172        let stream = Stream::<String>::new("thinking", 1_048_576);
173        let schema = PersistenceSchema::new().stream(&stream);
174
175        assert_eq!(
176            schema.definitions[0].stream_identity,
177            Some(stream.identity())
178        );
179        assert_eq!(schema.definitions[0].stream_capacity_bytes, Some(1_048_576));
180    }
181}
dex_sdk/persistence.rs

dex_sdk/
persistence.rs
