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::any::TypeId;
10use std::marker::PhantomData;
11
12use dex_protocol::dex::{AttributeSyncConfig, AttributeWrite, IndexConfig, value};
13
14use crate::{Context, HandlerError, HandlerResult, Value};
15
16/// Selects one AttributeMap instance to load for a handler invocation.
17///
18/// Create instance loads with [`AttributeMap::load`], then attach them with
19/// [`Rpc::load_attribute_map_instance`](crate::Rpc::load_attribute_map_instance).
20#[derive(Clone, Debug, Eq, PartialEq)]
21pub struct AttributeMapLoad {
22 pub(crate) name: String,
23 pub(crate) instance: Option<String>,
24}
25
26/// Selects one AttributeMap instance for [`Client::get_attributes`](crate::Client::get_attributes).
27///
28/// Since 1.31.0.
29///
30/// Create it with [`AttributeMap::instance`] and pass a reference to it in the tuple of reads.
31/// It carries the AttributeMap definition, the instance key, and the value type `T`.
32/// The read returns `Option<T>`, which is `None` when the instance is not set.
33///
34/// It is a different type from [`AttributeMapLoad`].
35/// An [`AttributeMap::load`] result selects an instance to load into a handler Context.
36/// [`Client::get_attributes`](crate::Client::get_attributes) does not accept it.
37///
38/// Creating an instance selector does not validate the key.
39/// [`Client::get_attributes`](crate::Client::get_attributes) rejects an empty key or a key that
40/// contains `/` with [`SdkError::InvalidArgument`](crate::SdkError::InvalidArgument).
41///
42/// # Examples
43///
44/// ```no_run
45/// use dex_sdk::{AttributeMap, Client};
46/// use std::sync::LazyLock;
47///
48/// static LINE_QUANTITIES: LazyLock<AttributeMap<u32>> =
49/// LazyLock::new(|| AttributeMap::new("line-quantities"));
50///
51/// fn read_line_quantity(client: &Client, order_id: &str) -> dex_sdk::SdkResult<u32> {
52/// let (quantity,) = client.get_attributes(order_id, (&LINE_QUANTITIES.instance("line-1"),))?;
53/// Ok(quantity.unwrap_or(0))
54/// }
55/// ```
56pub struct AttributeMapInstance<T> {
57 attribute_map: AttributeMap<T>,
58 instance: String,
59}
60
61impl<T> AttributeMapInstance<T> {
62 /// Returns the AttributeMap definition this selector reads.
63 pub fn attribute_map(&self) -> &AttributeMap<T> {
64 &self.attribute_map
65 }
66
67 /// Returns the instance key exactly as it was passed to [`AttributeMap::instance`].
68 pub fn instance(&self) -> &str {
69 &self.instance
70 }
71}
72
73impl<T> Clone for AttributeMapInstance<T> {
74 fn clone(&self) -> Self {
75 Self {
76 attribute_map: self.attribute_map.clone(),
77 instance: self.instance.clone(),
78 }
79 }
80}
81
82/// Defines one durable, typed value stored with a Flow.
83///
84/// Declare Attributes on the Flow, add them to [`crate::PersistenceSchema`], and reuse the same
85/// definition from Step and RPC code. Values use Serde serialization. `get` distinguishes a missing
86/// value with `None`; `get_required` converts absence into [`HandlerError`].
87///
88/// Register definitions once, preferably as module-level [`std::sync::LazyLock`] statics.
89/// Singleton Attributes are loaded into each handler Context. AttributeMap values require explicit
90/// instance or whole-map loads. Lock every cooperating read-modify-write operation on the same
91/// Attribute or map instance; a read alone does not serialize concurrent writers.
92/// Indexing and Attribute Store synchronization are opt-in and independent. Attribute Store
93/// projection is asynchronous; use the Context or Client to read committed Dex state directly.
94///
95/// # Examples
96///
97/// ```
98/// use dex_sdk::{Attribute, AttributeIndex};
99///
100/// let status = Attribute::<String>::new("status")
101/// .indexed(AttributeIndex::keyword());
102/// ```
103pub struct Attribute<T> {
104 name: String,
105 index: Option<AttributeIndex>,
106 sync_to_attribute_store: bool,
107 is_hard_delete_on_null: bool,
108 numeric_or_boolean_type_name: Option<&'static str>,
109 marker: PhantomData<fn() -> T>,
110}
111
112impl<T> Attribute<T> {
113 /// Defines an unindexed Attribute with a stable `name`.
114 ///
115 /// Slash is prohibited because it is a reserved character.
116 pub fn new(name: impl Into<String>) -> Self {
117 Self {
118 name: name.into(),
119 index: None,
120 sync_to_attribute_store: false,
121 is_hard_delete_on_null: false,
122 numeric_or_boolean_type_name: None,
123 marker: PhantomData,
124 }
125 }
126
127 /// Enables Flow-search indexing with the selected index type.
128 pub fn indexed(mut self, index: AttributeIndex) -> Self {
129 self.index = Some(index);
130 self
131 }
132
133 /// Enables asynchronous latest-state projection to the Flow's selected Attribute Store.
134 ///
135 /// The Flow must select a server-configured store with
136 /// [`FlowConfig::attribute_store_names`](crate::FlowConfig::attribute_store_names). Deletions
137 /// project SQL `NULL`, and projection failures never roll back the Flow Attribute write.
138 #[must_use]
139 pub fn sync_to_attribute_store(mut self) -> Self {
140 self.sync_to_attribute_store = true;
141 self
142 }
143
144 /// Stores this Attribute's values only in the Dex blob store and permanently deletes the stored
145 /// value when the Attribute is deleted or set to null.
146 ///
147 /// Use this option for short-lived secrets, such as API keys or access tokens. Dex never writes
148 /// the value into Flow history, regardless of its size. Flow history keeps only a reference to
149 /// the stored value. [`Self::delete`], [`Context::delete_attribute`], and a write whose value
150 /// encodes to null permanently delete the stored value before the write commits. A deleted
151 /// value reads as an absent Attribute, so [`Self::get`] returns `Ok(None)`.
152 ///
153 /// Step handlers, RPC handlers, and Attribute reads receive the value inline. The SDK never
154 /// stores the value in the local [`BlobCache`](crate::BlobCache).
155 ///
156 /// # Supported values
157 ///
158 /// Only values that encode to a string or an object are supported. Strings, Serde structs,
159 /// maps, sequences, and bytes are supported. Numeric and Boolean values are rejected.
160 /// [`Registry::register`](crate::Registry::register) returns
161 /// [`SdkError::FlowDefinition`](crate::SdkError::FlowDefinition) when `T` is a Rust integer,
162 /// floating-point, or `bool` type, or an `Option` of one. Every write also checks its encoded
163 /// value. A value of another type that encodes to a number or a Boolean, such as a newtype
164 /// around `u64`, fails when it is written. Deletion and null values are always allowed.
165 ///
166 /// To store a numeric or Boolean secret, wrap it in a struct so that it encodes as an object:
167 ///
168 /// ```
169 /// use dex_sdk::Attribute;
170 /// use serde::{Deserialize, Serialize};
171 ///
172 /// #[derive(Deserialize, Serialize)]
173 /// struct OneTimeCode {
174 /// code: u32,
175 /// }
176 ///
177 /// let one_time_code = Attribute::<OneTimeCode>::new("one-time-code").hard_delete_on_null();
178 /// ```
179 ///
180 /// # Restrictions
181 ///
182 /// - This option cannot be combined with [`Self::indexed`] or
183 /// [`Self::sync_to_attribute_store`]. Registration returns
184 /// [`SdkError::FlowDefinition`](crate::SdkError::FlowDefinition) for either combination.
185 /// - [`Client::wait_for_attribute_match`](crate::Client::wait_for_attribute_match) rejects a
186 /// hard-delete Attribute with [`SdkError::InvalidArgument`](crate::SdkError::InvalidArgument)
187 /// before it contacts Dex.
188 /// - Action and Action permission conditions cannot read a hard-delete Attribute. Dex
189 /// matches those conditions against stored Attribute values, which hard-delete Attributes
190 /// never keep in Flow history. Use an ordinary Attribute as the condition source instead.
191 /// - Every write must use a definition whose hard-delete setting matches the definition that
192 /// the Flow registered. A mismatched write is rejected.
193 /// - The Dex Server must have an enabled blob store. The Server rejects writes when its blob
194 /// store is disabled.
195 /// - The Dex Server must support protocol version 3 or later. A Worker learns the protocol
196 /// version when it starts. A Client requests it once, before its first hard-delete write,
197 /// and reuses the result. Writes fail with an error when the Server is older.
198 ///
199 /// # Limitations
200 ///
201 /// - Each Flow keeps one stored copy per Attribute key. Every write overwrites it in place.
202 /// - A Step or RPC attempt writes or deletes the stored value before Dex commits the attempt's
203 /// result. A retried attempt, other Steps, and Attribute reads can therefore observe a write
204 /// or deletion from an attempt that later failed.
205 /// - When concurrent Steps or RPCs write the same key, the final stored value may not be the
206 /// value of the last committed write.
207 /// - A Flow start can fail because a Flow with the same Flow ID is already running. That
208 /// failed start still overwrites the running Flow's stored value. A null initial value
209 /// deletes it. This applies to
210 /// [`Client::start_flow_with_options`](crate::Client::start_flow_with_options) and to SubFlow
211 /// starts.
212 /// - Executions that reuse a Flow ID share the stored value.
213 /// - Resetting a Flow does not restore a deleted value. A deleted value reads as absent.
214 /// - A Flow retry or a restarted SubFlow starts from the original initial value. If an
215 /// earlier run deleted that value, the new run reads it as absent.
216 /// - Removing `hard_delete_on_null` from a definition leaves the stored value in place until
217 /// Dex removes the Flow. Delete the value with the hard-delete definition before removing it.
218 /// - Stored values remain until they are deleted, or until Dex removes the Flow through
219 /// retention and blob cleanup.
220 /// - Blob store backups and bucket versioning can retain copies of stored values.
221 ///
222 /// # Examples
223 ///
224 /// ```
225 /// use dex_sdk::{Attribute, Context, HandlerResult, PersistenceSchema, StartFlowOptions};
226 /// use std::sync::LazyLock;
227 ///
228 /// static API_KEY: LazyLock<Attribute<String>> =
229 /// LazyLock::new(|| Attribute::new("api-key").hard_delete_on_null());
230 ///
231 /// // Register the definition in the Flow's persistence schema.
232 /// let schema = PersistenceSchema::new().attribute(&API_KEY);
233 ///
234 /// // Supply the first secret when the Flow starts.
235 /// let options = StartFlowOptions::new().initial_attribute(&API_KEY, "key-1".to_owned());
236 ///
237 /// // A Step reads the secret inline and rotates it.
238 /// fn rotate_api_key(context: &mut Context) -> HandlerResult<()> {
239 /// let current = API_KEY.get_required(context)?;
240 /// API_KEY.set(context, format!("{current}-rotated"))
241 /// }
242 ///
243 /// // A later Step permanently deletes the stored secret.
244 /// fn revoke_api_key(context: &mut Context) -> HandlerResult<()> {
245 /// API_KEY.delete(context)
246 /// }
247 /// # let _ = (schema, options, rotate_api_key, revoke_api_key);
248 /// ```
249 #[must_use]
250 pub fn hard_delete_on_null(mut self) -> Self
251 where
252 T: 'static,
253 {
254 self.is_hard_delete_on_null = true;
255 self.numeric_or_boolean_type_name = numeric_or_boolean_type_name::<T>();
256 self
257 }
258
259 /// Reads the current value from a Step or RPC invocation.
260 ///
261 /// Returns `Ok(None)` when no value is stored.
262 ///
263 /// # Errors
264 ///
265 /// Returns [`HandlerError`] when the payload cannot be decoded.
266 pub fn get(&self, context: &Context) -> HandlerResult<Option<T>>
267 where
268 T: Value,
269 {
270 context.get_attribute(self)
271 }
272
273 /// Reads the current value and treats absence as an error.
274 ///
275 /// # Errors
276 ///
277 /// Returns [`HandlerError`] when the Attribute is missing or cannot be decoded.
278 pub fn get_required(&self, context: &Context) -> HandlerResult<T>
279 where
280 T: Value,
281 {
282 self.get(context)?.ok_or_else(|| {
283 HandlerError::new(
284 "dex_sdk::HandlerError",
285 format!("attribute {} is missing", self.name),
286 )
287 })
288 }
289
290 /// Stages `value` for durable persistence in the current invocation.
291 ///
292 /// # Errors
293 ///
294 /// Returns [`HandlerError`] when the value cannot be encoded or indexed.
295 pub fn set(&self, context: &mut Context, value: T) -> HandlerResult<()>
296 where
297 T: Value,
298 {
299 context.set_attribute(self, value)
300 }
301
302 /// Stages deletion of the current value.
303 ///
304 /// Deleting a missing Attribute is valid.
305 pub fn delete(&self, context: &mut Context) -> HandlerResult<()> {
306 context.delete_attribute(self)
307 }
308
309 /// Creates a lock request for Step options or an RPC definition.
310 pub fn lock(&self) -> AttributeLock {
311 AttributeLock {
312 attribute: self.name.clone(),
313 instance: None,
314 }
315 }
316
317 pub(crate) fn name(&self) -> &str {
318 &self.name
319 }
320
321 pub(crate) fn index(&self) -> Option<&AttributeIndex> {
322 self.index.as_ref()
323 }
324
325 pub(crate) fn sync_config(&self) -> Option<AttributeSyncConfig> {
326 self.sync_to_attribute_store
327 .then_some(AttributeSyncConfig { enabled: true })
328 }
329
330 pub(crate) fn is_sync_to_attribute_store(&self) -> bool {
331 self.sync_to_attribute_store
332 }
333
334 pub(crate) fn is_hard_delete_on_null(&self) -> bool {
335 self.is_hard_delete_on_null
336 }
337
338 pub(crate) fn numeric_or_boolean_type_name(&self) -> Option<&'static str> {
339 self.numeric_or_boolean_type_name
340 }
341
342 pub(crate) fn write_settings(&self) -> AttributeWriteSettings {
343 AttributeWriteSettings {
344 index_config: self.index().map(|index| index.proto_config(false)),
345 sync_config: self.sync_config(),
346 is_hard_delete_on_null: self.is_hard_delete_on_null,
347 }
348 }
349}
350
351impl<T> Clone for Attribute<T> {
352 fn clone(&self) -> Self {
353 Self {
354 name: self.name.clone(),
355 index: self.index.clone(),
356 sync_to_attribute_store: self.sync_to_attribute_store,
357 is_hard_delete_on_null: self.is_hard_delete_on_null,
358 numeric_or_boolean_type_name: self.numeric_or_boolean_type_name,
359 marker: PhantomData,
360 }
361 }
362}
363
364/// Defines keyed durable values sharing one Attribute name.
365///
366/// Each `instance` identifies an independent stored value and lock. Add the map definition once to
367/// [`crate::PersistenceSchema`], then pass an instance on each access.
368/// Slash is prohibited in instance keys because it is a reserved character.
369pub struct AttributeMap<T> {
370 name: String,
371 index: Option<AttributeIndex>,
372 sync_to_attribute_store: bool,
373 is_hard_delete_on_null: bool,
374 numeric_or_boolean_type_name: Option<&'static str>,
375 marker: PhantomData<fn() -> T>,
376}
377
378impl<T> AttributeMap<T> {
379 /// Defines an unindexed Attribute map with a stable `name`.
380 ///
381 /// Slash is prohibited because it is a reserved character.
382 pub fn new(name: impl Into<String>) -> Self {
383 Self {
384 name: name.into(),
385 index: None,
386 sync_to_attribute_store: false,
387 is_hard_delete_on_null: false,
388 numeric_or_boolean_type_name: None,
389 marker: PhantomData,
390 }
391 }
392
393 /// Enables search indexing for every instance using the selected index type.
394 pub fn indexed(mut self, index: AttributeIndex) -> Self {
395 self.index = Some(index);
396 self
397 }
398
399 /// Enables asynchronous latest-state projection for every AttributeMap instance.
400 ///
401 /// Each instance maps by its physical Attribute name. The Flow must select a server-configured
402 /// stores with [`FlowConfig::attribute_store_names`](crate::FlowConfig::attribute_store_names).
403 /// Deletions project SQL `NULL`, and projection failures never roll back Flow Attribute writes.
404 #[must_use]
405 pub fn sync_to_attribute_store(mut self) -> Self {
406 self.sync_to_attribute_store = true;
407 self
408 }
409
410 /// Stores every instance's values only in the Dex blob store and permanently deletes an
411 /// instance's stored value when that instance is deleted or set to null.
412 ///
413 /// Use this option for keyed short-lived secrets, such as one API key per tenant. Each
414 /// instance has its own stored value. Dex never writes instance values into Flow history.
415 /// [`Self::delete`] and a write whose value encodes to null permanently delete the instance's
416 /// stored value. The instance then reads as absent. Handlers and Attribute reads receive
417 /// values inline. The SDK never stores them in the local [`BlobCache`](crate::BlobCache).
418 ///
419 /// Only string and object values, including bytes, are supported. Numeric and Boolean values
420 /// are rejected. This option cannot be combined with [`Self::indexed`] or
421 /// [`Self::sync_to_attribute_store`].
422 /// [`Client::wait_for_attribute_map_instance_match`](crate::Client::wait_for_attribute_map_instance_match)
423 /// rejects it before it contacts Dex. Action and Action permission conditions cannot read
424 /// it. Writes require a Server with an enabled blob store and
425 /// protocol version 3 or later. Each instance key keeps one stored copy, and later writes
426 /// overwrite it in place.
427 ///
428 /// Supported values, restrictions, Server requirements, and limitations are otherwise the same
429 /// as for [`Attribute::hard_delete_on_null`]. Read that method before storing secrets.
430 ///
431 /// # Examples
432 ///
433 /// ```
434 /// use dex_sdk::{AttributeMap, Context, HandlerResult, PersistenceSchema, StartFlowOptions};
435 /// use std::sync::LazyLock;
436 ///
437 /// static TENANT_API_KEYS: LazyLock<AttributeMap<String>> =
438 /// LazyLock::new(|| AttributeMap::new("tenant-api-keys").hard_delete_on_null());
439 ///
440 /// let schema = PersistenceSchema::new().attribute_map(&TENANT_API_KEYS);
441 /// let options = StartFlowOptions::new().initial_attribute_map(
442 /// &TENANT_API_KEYS,
443 /// "tenant-1",
444 /// "key-1".to_owned(),
445 /// );
446 ///
447 /// // The Step or RPC must load the instance before reading it.
448 /// fn revoke_tenant_api_key(context: &mut Context, tenant: &str) -> HandlerResult<()> {
449 /// if TENANT_API_KEYS.get(context, tenant)?.is_some() {
450 /// TENANT_API_KEYS.delete(context, tenant)?;
451 /// }
452 /// Ok(())
453 /// }
454 /// # let _ = (schema, options, revoke_tenant_api_key);
455 /// ```
456 #[must_use]
457 pub fn hard_delete_on_null(mut self) -> Self
458 where
459 T: 'static,
460 {
461 self.is_hard_delete_on_null = true;
462 self.numeric_or_boolean_type_name = numeric_or_boolean_type_name::<T>();
463 self
464 }
465
466 /// Selects one slash-free map `instance` for loading into a handler Context.
467 ///
468 /// The instance remains readable for the handler even when it is absent; absence is returned as
469 /// `None`. Load the map definition when the RPC must enumerate every instance.
470 pub fn load(&self, instance: impl Into<String>) -> AttributeMapLoad {
471 AttributeMapLoad {
472 name: self.name.clone(),
473 instance: Some(instance.into()),
474 }
475 }
476
477 /// Selects one map `instance` for [`Client::get_attributes`](crate::Client::get_attributes).
478 ///
479 /// Use it to read one AttributeMap instance from outside the Flow.
480 /// The selector copies this definition and the key.
481 /// It does not contact Dex or validate the key.
482 /// [`Client::get_attributes`](crate::Client::get_attributes) rejects an empty key or a key
483 /// that contains `/`.
484 /// Use [`Self::load`] instead to load an instance into a Step or RPC Context.
485 /// See [`AttributeMapInstance`] for an example.
486 ///
487 /// # Parameters
488 ///
489 /// - `instance` is the map instance key. The SDK percent-encodes it into the physical key.
490 ///
491 /// Returns a typed selector whose read yields `Option<T>`.
492 ///
493 /// Since 1.31.0.
494 pub fn instance(&self, instance: impl Into<String>) -> AttributeMapInstance<T> {
495 AttributeMapInstance {
496 attribute_map: self.clone(),
497 instance: instance.into(),
498 }
499 }
500
501 /// Reads one instance, returning `Ok(None)` when it is absent.
502 ///
503 /// # Errors
504 ///
505 /// Returns [`HandlerError`] when the payload cannot be decoded.
506 pub fn get(&self, context: &Context, instance: &str) -> HandlerResult<Option<T>>
507 where
508 T: Value,
509 {
510 context.get_attribute_map(self, instance)
511 }
512
513 /// Reads one instance and treats absence as an error.
514 ///
515 /// # Errors
516 ///
517 /// Returns [`HandlerError`] when the instance is missing or cannot be decoded.
518 pub fn get_required(&self, context: &Context, instance: &str) -> HandlerResult<T>
519 where
520 T: Value,
521 {
522 self.get(context, instance)?.ok_or_else(|| {
523 HandlerError::new(
524 "dex_sdk::HandlerError",
525 format!("attribute {}[{instance}] is missing", self.name),
526 )
527 })
528 }
529
530 /// Stages one instance value for durable persistence.
531 ///
532 /// # Errors
533 ///
534 /// Returns [`HandlerError`] when the value cannot be encoded or indexed.
535 pub fn set(&self, context: &mut Context, instance: &str, value: T) -> HandlerResult<()>
536 where
537 T: Value,
538 {
539 context.set_attribute_map(self, instance, value)
540 }
541
542 /// Stages deletion of one instance; missing instances are valid.
543 pub fn delete(&self, context: &mut Context, instance: &str) -> HandlerResult<()> {
544 context.delete_attribute_map(self, instance)
545 }
546
547 /// Returns the effective number of instances, including buffered writes.
548 pub fn map_size(&self, context: &Context) -> HandlerResult<usize> {
549 Ok(self.all_instance_keys(context)?.len())
550 }
551
552 /// Returns decoded effective instance keys in ascending string order.
553 pub fn all_instance_keys(&self, context: &Context) -> HandlerResult<Vec<String>> {
554 context.attribute_map_keys(self)
555 }
556
557 /// Creates a lock request scoped to one map instance.
558 pub fn lock(&self, instance: impl Into<String>) -> AttributeLock {
559 let instance = instance.into();
560 crate::registry::assert_map_instance(&instance);
561 AttributeLock {
562 attribute: self.name.clone(),
563 instance: Some(instance),
564 }
565 }
566
567 pub(crate) fn name(&self) -> &str {
568 &self.name
569 }
570
571 pub(crate) fn index(&self) -> Option<&AttributeIndex> {
572 self.index.as_ref()
573 }
574
575 pub(crate) fn sync_config(&self) -> Option<AttributeSyncConfig> {
576 self.sync_to_attribute_store
577 .then_some(AttributeSyncConfig { enabled: true })
578 }
579
580 pub(crate) fn is_sync_to_attribute_store(&self) -> bool {
581 self.sync_to_attribute_store
582 }
583
584 pub(crate) fn is_hard_delete_on_null(&self) -> bool {
585 self.is_hard_delete_on_null
586 }
587
588 pub(crate) fn numeric_or_boolean_type_name(&self) -> Option<&'static str> {
589 self.numeric_or_boolean_type_name
590 }
591
592 pub(crate) fn write_settings(&self) -> AttributeWriteSettings {
593 AttributeWriteSettings {
594 index_config: self.index().map(|index| index.proto_config(true)),
595 sync_config: self.sync_config(),
596 is_hard_delete_on_null: self.is_hard_delete_on_null,
597 }
598 }
599}
600
601impl<T> Clone for AttributeMap<T> {
602 fn clone(&self) -> Self {
603 Self {
604 name: self.name.clone(),
605 index: self.index.clone(),
606 sync_to_attribute_store: self.sync_to_attribute_store,
607 is_hard_delete_on_null: self.is_hard_delete_on_null,
608 numeric_or_boolean_type_name: self.numeric_or_boolean_type_name,
609 marker: PhantomData,
610 }
611 }
612}
613
614/// Carries the definition-derived fields copied onto every write of one Attribute definition.
615#[derive(Clone, Debug, Default)]
616pub(crate) struct AttributeWriteSettings {
617 pub(crate) index_config: Option<IndexConfig>,
618 pub(crate) sync_config: Option<AttributeSyncConfig>,
619 pub(crate) is_hard_delete_on_null: bool,
620}
621
622impl AttributeWriteSettings {
623 pub(crate) fn attribute_write(
624 self,
625 key: String,
626 value: dex_protocol::dex::Value,
627 ) -> AttributeWrite {
628 AttributeWrite {
629 key,
630 value: Some(value),
631 index_config: self.index_config,
632 sync_config: self.sync_config,
633 is_hard_delete_on_null: self.is_hard_delete_on_null,
634 }
635 }
636}
637
638/// Returns `Ok(())` for a valid hard-delete write and for every write without the flag.
639///
640/// Returns an error when a hard-delete write also enables indexing or Attribute Store
641/// synchronization, or when its value encodes to an integer, floating-point, or Boolean value.
642pub(crate) fn validate_hard_delete_attribute_write(write: &AttributeWrite) -> Result<(), String> {
643 if !write.is_hard_delete_on_null {
644 return Ok(());
645 }
646 let key = &write.key;
647 if write
648 .index_config
649 .as_ref()
650 .is_some_and(|config| config.enable)
651 {
652 return Err(format!(
653 "hard-delete Attribute {key} cannot be combined with an index"
654 ));
655 }
656 if write
657 .sync_config
658 .as_ref()
659 .is_some_and(|config| config.enabled)
660 {
661 return Err(format!(
662 "hard-delete Attribute {key} cannot be combined with sync_to_attribute_store"
663 ));
664 }
665 match write.value.as_ref().and_then(|value| value.kind.as_ref()) {
666 Some(value::Kind::IntValue(_))
667 | Some(value::Kind::DoubleValue(_))
668 | Some(value::Kind::BoolValue(_)) => Err(format!(
669 "hard-delete Attribute {key} supports only string and object values; numeric and \
670 Boolean values are rejected"
671 )),
672 _ => Ok(()),
673 }
674}
675
676/// Returns `Ok(())` when a write's hard-delete setting matches the registered definition of
677/// `attribute_name`, and an error naming the Attribute otherwise.
678pub(crate) fn require_registered_hard_delete_setting(
679 attribute_name: &str,
680 is_registered_hard_delete_on_null: bool,
681 is_write_hard_delete_on_null: bool,
682) -> Result<(), String> {
683 if is_registered_hard_delete_on_null == is_write_hard_delete_on_null {
684 return Ok(());
685 }
686 Err(format!(
687 "Attribute {attribute_name} hard_delete_on_null setting does not match its registered \
688 definition"
689 ))
690}
691
692/// Returns the type name when `T` is a Rust numeric or Boolean type, or an `Option` of one.
693///
694/// Returns `None` for every other type, including newtypes that still encode to numbers.
695fn numeric_or_boolean_type_name<T: 'static>() -> Option<&'static str> {
696 macro_rules! scalar_type_ids {
697 ($($scalar:ty),* $(,)?) => {
698 [$(TypeId::of::<$scalar>(), TypeId::of::<Option<$scalar>>()),*]
699 };
700 }
701 let scalar_type_ids = scalar_type_ids!(
702 i8, i16, i32, i64, i128, isize, u8, u16, u32, u64, u128, usize, f32, f64, bool,
703 );
704 scalar_type_ids
705 .contains(&TypeId::of::<T>())
706 .then(std::any::type_name::<T>)
707}
708
709#[derive(Clone, Debug, Eq, PartialEq)]
710/// Describes how Dex projects an Attribute value into its Flow-search index.
711///
712/// Without [`Self::with_key`], a single Attribute uses its name as the physical key. Attribute maps
713/// derive per-instance physical keys unless an explicit key is set.
714pub struct AttributeIndex {
715 kind: AttributeIndexKind,
716 key: Option<String>,
717}
718
719impl AttributeIndex {
720 fn new(kind: AttributeIndexKind) -> Self {
721 Self { kind, key: None }
722 }
723
724 /// Indexes one exact UTF-8 string for equality and aggregation queries.
725 pub fn keyword() -> Self {
726 Self::new(AttributeIndexKind::Keyword)
727 }
728
729 /// Indexes analyzed UTF-8 text for full-text queries.
730 pub fn full_text() -> Self {
731 Self::new(AttributeIndexKind::FullText)
732 }
733
734 /// Indexes an array of exact UTF-8 strings.
735 pub fn keyword_array() -> Self {
736 Self::new(AttributeIndexKind::KeywordArray)
737 }
738
739 /// Indexes a signed 64-bit integer.
740 pub fn int() -> Self {
741 Self::new(AttributeIndexKind::Int)
742 }
743
744 /// Indexes a finite 64-bit floating-point number.
745 pub fn double() -> Self {
746 Self::new(AttributeIndexKind::Double)
747 }
748
749 /// Indexes a Boolean value.
750 pub fn bool() -> Self {
751 Self::new(AttributeIndexKind::Bool)
752 }
753
754 /// Indexes an RFC 3339 date-time value.
755 pub fn date_time() -> Self {
756 Self::new(AttributeIndexKind::DateTime)
757 }
758
759 /// Uses an explicit physical search-index key instead of the default Attribute-derived key.
760 pub fn with_key(mut self, key: impl Into<String>) -> Self {
761 self.key = Some(key.into());
762 self
763 }
764}
765
766#[derive(Clone, Copy, Debug, Eq, PartialEq)]
767pub(crate) enum AttributeIndexKind {
768 Keyword,
769 FullText,
770 KeywordArray,
771 Int,
772 Double,
773 Bool,
774 DateTime,
775}
776
777#[derive(Clone, Debug, Eq, PartialEq)]
778/// Identifies an Attribute or Attribute-map instance lock.
779///
780/// Create locks with [`Attribute::lock`] or [`AttributeMap::lock`], then attach them to
781/// [`crate::StepOptions`] or [`crate::RpcDefinition`](crate::Rpc).
782pub struct AttributeLock {
783 attribute: String,
784 instance: Option<String>,
785}
786
787impl AttributeIndex {
788 pub(crate) fn proto_config(&self, dynamic: bool) -> IndexConfig {
789 IndexConfig {
790 enable: true,
791 r#type: self.kind.proto_value(),
792 index_key: self
793 .key
794 .clone()
795 .or_else(|| dynamic.then(String::new))
796 .unwrap_or_default(),
797 }
798 }
799
800 pub(crate) fn key(&self) -> Option<&str> {
801 self.key.as_deref()
802 }
803
804 pub(crate) fn proto_value(&self) -> i32 {
805 self.kind.proto_value()
806 }
807}
808
809impl AttributeIndexKind {
810 pub(crate) fn proto_value(self) -> i32 {
811 use dex_protocol::dex::IndexType;
812
813 match self {
814 Self::Keyword => IndexType::Keyword.into(),
815 Self::FullText => IndexType::Text.into(),
816 Self::KeywordArray => IndexType::KeywordArray.into(),
817 Self::Int => IndexType::Int.into(),
818 Self::Double => IndexType::Double.into(),
819 Self::Bool => IndexType::Bool.into(),
820 Self::DateTime => IndexType::Datetime.into(),
821 }
822 }
823}
824
825impl AttributeLock {
826 pub(crate) fn attribute_name(&self) -> &str {
827 &self.attribute
828 }
829
830 pub(crate) fn instance(&self) -> Option<&str> {
831 self.instance.as_deref()
832 }
833
834 pub(crate) fn physical_name(&self) -> String {
835 match self.instance.as_deref() {
836 Some(instance) => crate::registry::physical_name(&self.attribute, instance),
837 None => self.attribute.clone(),
838 }
839 }
840}
841
842#[cfg(test)]
843mod tests {
844 use super::{Attribute, AttributeMap};
845
846 #[test]
847 fn sync_configuration_is_opt_in_and_survives_clone() {
848 let plain = Attribute::<String>::new("plain");
849 let synced = Attribute::<String>::new("synced").sync_to_attribute_store();
850 let synced_map = AttributeMap::<String>::new("synced_map").sync_to_attribute_store();
851
852 assert!(plain.sync_config().is_none());
853 assert_eq!(
854 synced.sync_config().map(|config| config.enabled),
855 Some(true)
856 );
857 assert_eq!(
858 synced_map
859 .clone()
860 .sync_config()
861 .map(|config| config.enabled),
862 Some(true)
863 );
864 }
865}