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

dex_sdk/
attribute.rs
