dex_sdk/
attribute.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 std::marker::PhantomData;
10
11use dex_protocol::dex::{AttributeSyncConfig, IndexConfig};
12
13use crate::{Context, HandlerError, HandlerResult, Value};
14
15/// Defines one durable, typed value stored with a Flow.
16///
17/// Declare Attributes on the Flow, add them to [`crate::PersistenceSchema`], and reuse the same
18/// definition from Step and RPC code. Values use Serde serialization. `get` distinguishes a missing
19/// value with `None`; `get_required` converts absence into [`HandlerError`].
20///
21/// # Examples
22///
23/// ```
24/// use dex_sdk::{Attribute, AttributeIndex};
25///
26/// let status = Attribute::<String>::new("status")
27///     .indexed(AttributeIndex::keyword());
28/// ```
29pub struct Attribute<T> {
30    name: String,
31    index: Option<AttributeIndex>,
32    sync_to_attribute_store: bool,
33    marker: PhantomData<fn() -> T>,
34}
35
36impl<T> Attribute<T> {
37    /// Defines an unindexed Attribute with stable `name`.
38    pub fn new(name: impl Into<String>) -> Self {
39        Self {
40            name: name.into(),
41            index: None,
42            sync_to_attribute_store: false,
43            marker: PhantomData,
44        }
45    }
46
47    /// Enables Flow-search indexing with the selected index type.
48    pub fn indexed(mut self, index: AttributeIndex) -> Self {
49        self.index = Some(index);
50        self
51    }
52
53    /// Enables asynchronous latest-state projection to the Flow's selected Attribute Store.
54    ///
55    /// The Flow must select a server-configured store with
56    /// [`FlowConfig::attribute_store_name`](crate::FlowConfig::attribute_store_name). Deletions
57    /// project SQL `NULL`, and projection failures never roll back the Flow Attribute write.
58    #[must_use]
59    pub fn sync_to_attribute_store(mut self) -> Self {
60        self.sync_to_attribute_store = true;
61        self
62    }
63
64    /// Reads the current value from a Step or RPC invocation.
65    ///
66    /// Returns `Ok(None)` when no value is stored.
67    ///
68    /// # Errors
69    ///
70    /// Returns [`HandlerError`] when the payload cannot be decoded.
71    pub fn get(&self, context: &Context) -> HandlerResult<Option<T>>
72    where
73        T: Value,
74    {
75        context.get_attribute(self)
76    }
77
78    /// Reads the current value and treats absence as an error.
79    ///
80    /// # Errors
81    ///
82    /// Returns [`HandlerError`] when the Attribute is missing or cannot be decoded.
83    pub fn get_required(&self, context: &Context) -> HandlerResult<T>
84    where
85        T: Value,
86    {
87        self.get(context)?
88            .ok_or_else(|| HandlerError::new(format!("attribute {} is missing", self.name)))
89    }
90
91    /// Stages `value` for durable persistence in the current invocation.
92    ///
93    /// # Errors
94    ///
95    /// Returns [`HandlerError`] when the value cannot be encoded or indexed.
96    pub fn set(&self, context: &mut Context, value: T) -> HandlerResult<()>
97    where
98        T: Value,
99    {
100        context.set_attribute(self, value)
101    }
102
103    /// Stages deletion of the current value.
104    ///
105    /// Deleting a missing Attribute is valid.
106    pub fn delete(&self, context: &mut Context) -> HandlerResult<()> {
107        context.delete_attribute(self)
108    }
109
110    /// Creates a lock request for Step options or an RPC definition.
111    pub fn lock(&self) -> AttributeLock {
112        AttributeLock {
113            attribute: self.name.clone(),
114            instance: None,
115        }
116    }
117
118    pub(crate) fn name(&self) -> &str {
119        &self.name
120    }
121
122    pub(crate) fn index(&self) -> Option<&AttributeIndex> {
123        self.index.as_ref()
124    }
125
126    pub(crate) fn sync_config(&self) -> Option<AttributeSyncConfig> {
127        self.sync_to_attribute_store
128            .then_some(AttributeSyncConfig { enabled: true })
129    }
130
131    pub(crate) fn is_sync_to_attribute_store(&self) -> bool {
132        self.sync_to_attribute_store
133    }
134}
135
136impl<T> Clone for Attribute<T> {
137    fn clone(&self) -> Self {
138        Self {
139            name: self.name.clone(),
140            index: self.index.clone(),
141            sync_to_attribute_store: self.sync_to_attribute_store,
142            marker: PhantomData,
143        }
144    }
145}
146
147/// Defines keyed durable values sharing one Attribute name.
148///
149/// Each `instance` identifies an independent stored value and lock. Add the map definition once to
150/// [`crate::PersistenceSchema`]; pass an instance name on each access.
151pub struct AttributeMap<T> {
152    name: String,
153    index: Option<AttributeIndex>,
154    sync_to_attribute_store: bool,
155    marker: PhantomData<fn() -> T>,
156}
157
158impl<T> AttributeMap<T> {
159    /// Defines an unindexed Attribute map with stable `name`.
160    pub fn new(name: impl Into<String>) -> Self {
161        Self {
162            name: name.into(),
163            index: None,
164            sync_to_attribute_store: false,
165            marker: PhantomData,
166        }
167    }
168
169    /// Enables search indexing for every instance using the selected index type.
170    pub fn indexed(mut self, index: AttributeIndex) -> Self {
171        self.index = Some(index);
172        self
173    }
174
175    /// Enables asynchronous latest-state projection for every AttributeMap instance.
176    ///
177    /// Each instance maps by its physical Attribute name. The Flow must select a server-configured
178    /// store with [`FlowConfig::attribute_store_name`](crate::FlowConfig::attribute_store_name).
179    /// Deletions project SQL `NULL`, and projection failures never roll back Flow Attribute writes.
180    #[must_use]
181    pub fn sync_to_attribute_store(mut self) -> Self {
182        self.sync_to_attribute_store = true;
183        self
184    }
185
186    /// Reads one instance, returning `Ok(None)` when it is absent.
187    ///
188    /// # Errors
189    ///
190    /// Returns [`HandlerError`] when the payload cannot be decoded.
191    pub fn get(&self, context: &Context, instance: &str) -> HandlerResult<Option<T>>
192    where
193        T: Value,
194    {
195        context.get_attribute_map(self, instance)
196    }
197
198    /// Reads one instance and treats absence as an error.
199    ///
200    /// # Errors
201    ///
202    /// Returns [`HandlerError`] when the instance is missing or cannot be decoded.
203    pub fn get_required(&self, context: &Context, instance: &str) -> HandlerResult<T>
204    where
205        T: Value,
206    {
207        self.get(context, instance)?.ok_or_else(|| {
208            HandlerError::new(format!("attribute {}[{instance}] is missing", self.name))
209        })
210    }
211
212    /// Stages one instance value for durable persistence.
213    ///
214    /// # Errors
215    ///
216    /// Returns [`HandlerError`] when the value cannot be encoded or indexed.
217    pub fn set(&self, context: &mut Context, instance: &str, value: T) -> HandlerResult<()>
218    where
219        T: Value,
220    {
221        context.set_attribute_map(self, instance, value)
222    }
223
224    /// Stages deletion of one instance; missing instances are valid.
225    pub fn delete(&self, context: &mut Context, instance: &str) -> HandlerResult<()> {
226        context.delete_attribute_map(self, instance)
227    }
228
229    /// Returns the effective number of instances, including buffered writes.
230    pub fn map_size(&self, context: &Context) -> HandlerResult<usize> {
231        Ok(self.all_instance_keys(context)?.len())
232    }
233
234    /// Returns decoded effective instance keys in ascending string order.
235    pub fn all_instance_keys(&self, context: &Context) -> HandlerResult<Vec<String>> {
236        context.attribute_map_keys(self)
237    }
238
239    /// Creates a lock request scoped to one map instance.
240    pub fn lock(&self, instance: impl Into<String>) -> AttributeLock {
241        AttributeLock {
242            attribute: self.name.clone(),
243            instance: Some(instance.into()),
244        }
245    }
246
247    pub(crate) fn name(&self) -> &str {
248        &self.name
249    }
250
251    pub(crate) fn index(&self) -> Option<&AttributeIndex> {
252        self.index.as_ref()
253    }
254
255    pub(crate) fn sync_config(&self) -> Option<AttributeSyncConfig> {
256        self.sync_to_attribute_store
257            .then_some(AttributeSyncConfig { enabled: true })
258    }
259
260    pub(crate) fn is_sync_to_attribute_store(&self) -> bool {
261        self.sync_to_attribute_store
262    }
263}
264
265impl<T> Clone for AttributeMap<T> {
266    fn clone(&self) -> Self {
267        Self {
268            name: self.name.clone(),
269            index: self.index.clone(),
270            sync_to_attribute_store: self.sync_to_attribute_store,
271            marker: PhantomData,
272        }
273    }
274}
275
276#[derive(Clone, Debug, Eq, PartialEq)]
277/// Describes how Dex projects an Attribute value into its Flow-search index.
278///
279/// Without [`Self::with_key`], a single Attribute uses its name as the physical key. Attribute maps
280/// derive per-instance physical keys unless an explicit key is set.
281pub struct AttributeIndex {
282    kind: AttributeIndexKind,
283    key: Option<String>,
284}
285
286impl AttributeIndex {
287    fn new(kind: AttributeIndexKind) -> Self {
288        Self { kind, key: None }
289    }
290
291    /// Indexes one exact UTF-8 string for equality and aggregation queries.
292    pub fn keyword() -> Self {
293        Self::new(AttributeIndexKind::Keyword)
294    }
295
296    /// Indexes analyzed UTF-8 text for full-text queries.
297    pub fn full_text() -> Self {
298        Self::new(AttributeIndexKind::FullText)
299    }
300
301    /// Indexes an array of exact UTF-8 strings.
302    pub fn keyword_array() -> Self {
303        Self::new(AttributeIndexKind::KeywordArray)
304    }
305
306    /// Indexes a signed 64-bit integer.
307    pub fn int() -> Self {
308        Self::new(AttributeIndexKind::Int)
309    }
310
311    /// Indexes a finite 64-bit floating-point number.
312    pub fn double() -> Self {
313        Self::new(AttributeIndexKind::Double)
314    }
315
316    /// Indexes a Boolean value.
317    pub fn bool() -> Self {
318        Self::new(AttributeIndexKind::Bool)
319    }
320
321    /// Indexes an RFC 3339 date-time value.
322    pub fn date_time() -> Self {
323        Self::new(AttributeIndexKind::DateTime)
324    }
325
326    /// Uses an explicit physical search-index key instead of the default Attribute-derived key.
327    pub fn with_key(mut self, key: impl Into<String>) -> Self {
328        self.key = Some(key.into());
329        self
330    }
331}
332
333#[derive(Clone, Copy, Debug, Eq, PartialEq)]
334pub(crate) enum AttributeIndexKind {
335    Keyword,
336    FullText,
337    KeywordArray,
338    Int,
339    Double,
340    Bool,
341    DateTime,
342}
343
344#[derive(Clone, Debug, Eq, PartialEq)]
345/// Identifies an Attribute or Attribute-map instance lock.
346///
347/// Create locks with [`Attribute::lock`] or [`AttributeMap::lock`], then attach them to
348/// [`crate::StepOptions`] or [`crate::RpcDefinition`](crate::Rpc).
349pub struct AttributeLock {
350    attribute: String,
351    instance: Option<String>,
352}
353
354impl AttributeIndex {
355    pub(crate) fn proto_config(&self, dynamic: bool) -> IndexConfig {
356        IndexConfig {
357            enable: true,
358            r#type: self.kind.proto_value(),
359            index_key: self
360                .key
361                .clone()
362                .or_else(|| dynamic.then(String::new))
363                .unwrap_or_default(),
364        }
365    }
366
367    pub(crate) fn key(&self) -> Option<&str> {
368        self.key.as_deref()
369    }
370
371    pub(crate) fn proto_value(&self) -> i32 {
372        self.kind.proto_value()
373    }
374}
375
376impl AttributeIndexKind {
377    pub(crate) fn proto_value(self) -> i32 {
378        use dex_protocol::dex::IndexType;
379
380        match self {
381            Self::Keyword => IndexType::Keyword.into(),
382            Self::FullText => IndexType::Text.into(),
383            Self::KeywordArray => IndexType::KeywordArray.into(),
384            Self::Int => IndexType::Int.into(),
385            Self::Double => IndexType::Double.into(),
386            Self::Bool => IndexType::Bool.into(),
387            Self::DateTime => IndexType::Datetime.into(),
388        }
389    }
390}
391
392impl AttributeLock {
393    pub(crate) fn physical_name(&self) -> String {
394        match self.instance.as_deref() {
395            Some(instance) => crate::registry::physical_name(&self.attribute, instance),
396            None => self.attribute.clone(),
397        }
398    }
399}
400
401#[cfg(test)]
402mod tests {
403    use super::{Attribute, AttributeMap};
404
405    #[test]
406    fn sync_configuration_is_opt_in_and_survives_clone() {
407        let plain = Attribute::<String>::new("plain");
408        let synced = Attribute::<String>::new("synced").sync_to_attribute_store();
409        let synced_map = AttributeMap::<String>::new("synced_map").sync_to_attribute_store();
410
411        assert!(plain.sync_config().is_none());
412        assert_eq!(
413            synced.sync_config().map(|config| config.enabled),
414            Some(true)
415        );
416        assert_eq!(
417            synced_map
418                .clone()
419                .sync_config()
420                .map(|config| config.enabled),
421            Some(true)
422        );
423    }
424}
dex_sdk/attribute.rs

dex_sdk/
attribute.rs
