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

dex_sdk/
attribute.rs
