storage_kit/state/
record.rs

1//! Revisioned state records, validated keys, and pagination.
2//!
3//! [`StateKey`](crate::state::record::StateKey) separates a validated namespace from a namespace-local key.
4//! Write a [`StateRecord`](crate::state::record::StateRecord) through [`crate::StateTransactionBackend`] and read
5//! a [`StateRecordSnapshot`](crate::state::record::StateRecordSnapshot) to observe its per-key revision. Listing uses
6//! literal prefixes and exclusive key cursors; multiple pages do not form a
7//! historical snapshot. JSON payload clones support caller-owned copy-on-write
8//! editing through [`Arc::make_mut`](std::sync::Arc::make_mut), without publishing edits automatically.
9
10use std::{num::NonZeroU64, sync::Arc};
11
12use crate::StateStoreError;
13
14const MAX_NAMESPACE_BYTES: usize = 256;
15const MAX_KEY_BYTES: usize = 1024;
16const DEFAULT_LIST_LIMIT: usize = 100;
17const MAX_LIST_LIMIT: usize = 1000;
18
19/// Validated state namespace.
20///
21/// Namespaces are nonempty and contain at most 256 UTF-8 bytes. Leading or
22/// trailing whitespace and control characters are rejected.
23///
24/// # Examples
25///
26/// ```
27/// use storage_kit::state::record::StateNamespace;
28/// let namespace = StateNamespace::new("workflow")?;
29/// assert_eq!(namespace.as_str(), "workflow");
30/// # Ok::<(), storage_kit::StateStoreError>(())
31/// ```
32#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
33pub struct StateNamespace(String);
34
35impl StateNamespace {
36    /// Validates a namespace.
37    ///
38    /// # Errors
39    ///
40    /// Returns [`StateStoreError::InvalidRequest`] for an empty namespace, more
41    /// than 256 UTF-8 bytes, surrounding whitespace, or a control character.
42    ///
43    /// # Examples
44    ///
45    /// ```
46    /// use storage_kit::state::record::StateNamespace;
47    /// assert!(StateNamespace::new(" padded").is_err());
48    /// let namespace = StateNamespace::new("workflow")?;
49    /// assert_eq!(namespace.as_str(), "workflow");
50    /// # Ok::<(), storage_kit::StateStoreError>(())
51    /// ```
52    pub fn new(namespace: impl Into<String>) -> Result<Self, StateStoreError> {
53        Self::validate_fragment("namespace", namespace.into(), MAX_NAMESPACE_BYTES).map(Self)
54    }
55
56    /// Borrows the namespace.
57    ///
58    /// # Examples
59    ///
60    /// ```
61    /// use storage_kit::state::record::StateNamespace;
62    /// let namespace = StateNamespace::new("workflow")?;
63    /// assert_eq!(namespace.as_str(), "workflow");
64    /// # Ok::<(), storage_kit::StateStoreError>(())
65    /// ```
66    #[must_use]
67    pub fn as_str(&self) -> &str {
68        &self.0
69    }
70
71    pub(super) fn validate_fragment(
72        label: &'static str,
73        value: String,
74        maximum: usize,
75    ) -> Result<String, StateStoreError> {
76        if value.is_empty() {
77            return Err(StateStoreError::invalid_request(format!(
78                "state {label} must not be empty"
79            )));
80        }
81        if value.len() > maximum {
82            return Err(StateStoreError::invalid_request(format!(
83                "state {label} must contain at most {maximum} bytes"
84            )));
85        }
86        if value.trim() != value {
87            return Err(StateStoreError::invalid_request(format!(
88                "state {label} must not contain leading or trailing whitespace"
89            )));
90        }
91        if value.chars().any(char::is_control) {
92            return Err(StateStoreError::invalid_request(format!(
93                "state {label} must not contain control characters"
94            )));
95        }
96        Ok(value)
97    }
98}
99
100/// Validated key prefix for list requests.
101///
102/// Prefixes use literal, case-sensitive matching: `%`, `_`, and `*` are not
103/// wildcards. A prefix has the same 1024-byte and character bounds as a key;
104/// omit the prefix to select the entire namespace.
105///
106/// # Examples
107///
108/// ```
109/// use storage_kit::state::record::StateKeyPrefix;
110/// let prefix = StateKeyPrefix::new("run/")?;
111/// assert_eq!(prefix.as_str(), "run/");
112/// # Ok::<(), storage_kit::StateStoreError>(())
113/// ```
114#[derive(Clone, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
115pub struct StateKeyPrefix(String);
116
117impl StateKeyPrefix {
118    /// Validates a key prefix.
119    ///
120    /// # Errors
121    ///
122    /// Returns [`StateStoreError::InvalidRequest`] for an empty prefix, more than
123    /// 1024 UTF-8 bytes, surrounding whitespace, or a control character.
124    ///
125    /// # Examples
126    ///
127    /// ```
128    /// use storage_kit::state::record::StateKeyPrefix;
129    /// let prefix = StateKeyPrefix::new("run/%")?;
130    /// assert_eq!(prefix.as_str(), "run/%");
131    /// assert!(StateKeyPrefix::new("").is_err());
132    /// # Ok::<(), storage_kit::StateStoreError>(())
133    /// ```
134    pub fn new(prefix: impl Into<String>) -> Result<Self, StateStoreError> {
135        StateNamespace::validate_fragment("prefix", prefix.into(), MAX_KEY_BYTES).map(Self)
136    }
137
138    /// Borrows the prefix.
139    ///
140    /// # Examples
141    ///
142    /// ```
143    /// use storage_kit::state::record::StateKeyPrefix;
144    /// let prefix = StateKeyPrefix::new("run/")?;
145    /// assert_eq!(prefix.as_str(), "run/");
146    /// # Ok::<(), storage_kit::StateStoreError>(())
147    /// ```
148    #[must_use]
149    pub fn as_str(&self) -> &str {
150        &self.0
151    }
152}
153
154/// Fully qualified state-record key.
155///
156/// The namespace and namespace-local key are separate values. Keys are
157/// nonempty, contain at most 1024 UTF-8 bytes, and reject surrounding
158/// whitespace and control characters. A slash has no special storage meaning.
159///
160/// # Examples
161///
162/// ```
163/// use storage_kit::state::record::StateKey;
164/// let key = StateKey::new("workflow", "run/1")?;
165/// assert_eq!(key.namespace().as_str(), "workflow");
166/// assert_eq!(key.key(), "run/1");
167/// # Ok::<(), storage_kit::StateStoreError>(())
168/// ```
169#[derive(Clone, Debug, Eq, Hash, PartialEq)]
170pub struct StateKey {
171    namespace: StateNamespace,
172    key: String,
173}
174
175impl StateKey {
176    /// Validates a namespace and key fragment.
177    ///
178    /// # Errors
179    ///
180    /// Returns [`StateStoreError::InvalidRequest`] when the namespace exceeds
181    /// 256 UTF-8 bytes or the key exceeds 1024, or either fragment is empty,
182    /// has surrounding whitespace, or contains a control character.
183    ///
184    /// # Examples
185    ///
186    /// ```
187    /// use storage_kit::state::record::StateKey;
188    /// let key = StateKey::new("workflow", "run/1")?;
189    /// assert_eq!(key.key(), "run/1");
190    /// assert!(StateKey::new("workflow", "run/1\n").is_err());
191    /// # Ok::<(), storage_kit::StateStoreError>(())
192    /// ```
193    pub fn new(
194        namespace: impl Into<String>,
195        key: impl Into<String>,
196    ) -> Result<Self, StateStoreError> {
197        Ok(Self {
198            namespace: StateNamespace::new(namespace)?,
199            key: StateNamespace::validate_fragment("key", key.into(), MAX_KEY_BYTES)?,
200        })
201    }
202
203    /// Returns the namespace.
204    ///
205    /// # Examples
206    ///
207    /// ```
208    /// use storage_kit::state::record::StateKey;
209    /// let key = StateKey::new("workflow", "run/1")?;
210    /// assert_eq!(key.namespace().as_str(), "workflow");
211    /// # Ok::<(), storage_kit::StateStoreError>(())
212    /// ```
213    #[must_use]
214    pub fn namespace(&self) -> &StateNamespace {
215        &self.namespace
216    }
217
218    /// Returns the namespace-local key.
219    ///
220    /// # Examples
221    ///
222    /// ```
223    /// use storage_kit::state::record::StateKey;
224    /// let key = StateKey::new("workflow", "run/1")?;
225    /// assert_eq!(key.key(), "run/1");
226    /// # Ok::<(), storage_kit::StateStoreError>(())
227    /// ```
228    #[must_use]
229    pub fn key(&self) -> &str {
230        &self.key
231    }
232}
233
234/// Positive monotonically increasing state revision.
235///
236/// Revisions belong to individual keys. They are not global commit ids,
237/// workflow sequence numbers, or historical snapshot selectors. Puts and
238/// deletes increment the key revision; tombstones retain it across recreation.
239/// The positive `u64` representation does not promise every backend can
240/// persist its full range.
241///
242/// # Examples
243///
244/// ```
245/// use storage_kit::state::record::StateRevision;
246/// let revision = StateRevision::new(3)?;
247/// assert_eq!(revision.get(), 3);
248/// assert!(StateRevision::new(0).is_err());
249/// # Ok::<(), storage_kit::StateStoreError>(())
250/// ```
251#[derive(Clone, Copy, Debug, Eq, Hash, Ord, PartialEq, PartialOrd)]
252pub struct StateRevision(NonZeroU64);
253
254impl StateRevision {
255    /// Validates a positive revision.
256    ///
257    /// This validates a representation; it does not assign a revision to stored data.
258    ///
259    /// # Errors
260    ///
261    /// Returns [`StateStoreError::InvalidRequest`] when `value` is zero.
262    ///
263    /// # Examples
264    ///
265    /// ```
266    /// use storage_kit::state::record::StateRevision;
267    /// assert!(StateRevision::new(0).is_err());
268    /// assert_eq!(StateRevision::new(1)?, StateRevision::first());
269    /// # Ok::<(), storage_kit::StateStoreError>(())
270    /// ```
271    pub fn new(value: u64) -> Result<Self, StateStoreError> {
272        NonZeroU64::new(value)
273            .map(Self)
274            .ok_or_else(|| StateStoreError::invalid_request("state revision must be positive"))
275    }
276
277    /// Returns revision one.
278    ///
279    /// # Examples
280    ///
281    /// ```
282    /// use storage_kit::state::record::StateRevision;
283    /// assert_eq!(StateRevision::first().get(), 1);
284    /// # Ok::<(), storage_kit::StateStoreError>(())
285    /// ```
286    #[must_use]
287    pub fn first() -> Self {
288        Self(NonZeroU64::MIN)
289    }
290
291    /// Returns the revision value.
292    ///
293    /// # Examples
294    ///
295    /// ```
296    /// use storage_kit::state::record::StateRevision;
297    /// assert_eq!(StateRevision::new(8)?.get(), 8);
298    /// # Ok::<(), storage_kit::StateStoreError>(())
299    /// ```
300    #[must_use]
301    pub fn get(self) -> u64 {
302        self.0.get()
303    }
304
305    #[cfg(any(feature = "redb", feature = "sqlite"))]
306    pub(crate) fn next(self) -> Result<Self, StateStoreError> {
307        self.get()
308            .checked_add(1)
309            .ok_or_else(|| StateStoreError::invalid_request("state revision overflow"))
310            .and_then(Self::new)
311    }
312
313    #[cfg(any(feature = "postgres", feature = "sqlite"))]
314    pub(crate) fn from_i64(value: i64) -> Result<Self, StateStoreError> {
315        u64::try_from(value)
316            .map_err(|error| {
317                StateStoreError::invalid_request(format!(
318                    "state revision conversion failed: {error}"
319                ))
320            })
321            .and_then(Self::new)
322    }
323
324    #[cfg(feature = "redb")]
325    pub(crate) fn from_backend_u64(value: u64) -> Result<Self, StateStoreError> {
326        Self::new(value)
327    }
328
329    #[cfg(feature = "sqlite")]
330    pub(crate) fn to_backend_i64(self) -> Result<i64, StateStoreError> {
331        i64::try_from(self.get()).map_err(|error| {
332            StateStoreError::invalid_request(format!(
333                "state revision exceeds backend signed 64-bit range: {error}"
334            ))
335        })
336    }
337}
338
339/// Caller-authored state record input.
340///
341/// # Examples
342///
343/// ```
344/// use storage_kit::state::record::{StateKey, StateRecord};
345/// let original = StateRecord::new(StateKey::new("workflow", "run/1")?,
346///     serde_json::json!({"status": "pending"}));
347/// let mut edited = original.clone();
348/// std::sync::Arc::make_mut(&mut edited.value)["status"] = serde_json::json!("done");
349/// assert_eq!(original.value["status"], "pending");
350/// assert_eq!(edited.value["status"], "done");
351/// # Ok::<(), storage_kit::StateStoreError>(())
352/// ```
353#[derive(Clone, Debug, Eq, PartialEq)]
354pub struct StateRecord {
355    /// Fully qualified key.
356    pub key: StateKey,
357    /// Shared JSON content. Use [`Arc::make_mut`] to edit a caller-owned copy.
358    ///
359    /// Clones retain their original content when another copy is edited. Stores
360    /// do not promise pointer identity across separately returned observations.
361    pub value: Arc<serde_json::Value>,
362}
363
364impl StateRecord {
365    /// Creates a record input.
366    ///
367    /// # Examples
368    ///
369    /// ```
370    /// use storage_kit::state::record::{StateKey, StateRecord};
371    /// let record = StateRecord::new(StateKey::new("workflow", "run/1")?,
372    ///     serde_json::json!({"status": "pending"}));
373    /// assert_eq!(record.value["status"], "pending");
374    /// # Ok::<(), storage_kit::StateStoreError>(())
375    /// ```
376    #[must_use]
377    pub fn new(key: StateKey, value: serde_json::Value) -> Self {
378        Self {
379            key,
380            value: Arc::new(value),
381        }
382    }
383}
384
385/// Observed record and its backend revision.
386///
387/// This value describes one observed record, not a historical store snapshot.
388/// Constructing it does not publish a record or reserve a backend revision.
389///
390/// # Examples
391///
392/// ```
393/// use storage_kit::state::record::{StateKey, StateRevision, StateRecord, StateRecordSnapshot};
394/// let record = StateRecord::new(StateKey::new("workflow", "run/1")?, serde_json::json!(42));
395/// let snapshot = StateRecordSnapshot::new(record, StateRevision::first());
396/// assert_eq!(snapshot.revision.get(), 1);
397/// # Ok::<(), storage_kit::StateStoreError>(())
398/// ```
399#[derive(Clone, Debug, Eq, PartialEq)]
400pub struct StateRecordSnapshot {
401    /// Stored record.
402    pub record: StateRecord,
403    /// Observed revision.
404    pub revision: StateRevision,
405}
406
407impl StateRecordSnapshot {
408    /// Creates an observed snapshot.
409    ///
410    /// # Examples
411    ///
412    /// ```
413    /// use storage_kit::state::record::{StateKey, StateRevision, StateRecord, StateRecordSnapshot};
414    /// let record = StateRecord::new(StateKey::new("workflow", "run/1")?, serde_json::json!(42));
415    /// let snapshot = StateRecordSnapshot::new(record, StateRevision::new(2)?);
416    /// assert_eq!(snapshot.record.value.as_ref(), &serde_json::json!(42));
417    /// assert_eq!(snapshot.revision.get(), 2);
418    /// # Ok::<(), storage_kit::StateStoreError>(())
419    /// ```
420    #[must_use]
421    pub fn new(record: StateRecord, revision: StateRevision) -> Self {
422        Self { record, revision }
423    }
424}
425
426/// Key-ordered namespace list request.
427///
428/// Results use ascending lexicographic UTF-8 key order within one exact
429/// namespace. The optional prefix is literal and case-sensitive; the cursor
430/// is exclusive. Each page has a consistent read view, but separate pages are
431/// not a historical snapshot. The default page size is 100; accepted sizes
432/// are 1 through 1000 inclusive.
433///
434/// # Examples
435///
436/// ```
437/// use storage_kit::state::record::{StateKeyPrefix, StateListRequest};
438/// let request = StateListRequest::new("workflow")?
439///     .with_key_prefix(StateKeyPrefix::new("run/")?)
440///     .with_limit(25)?
441///     .start_after("run/10")?;
442/// assert_eq!(request.prefix(), Some("run/"));
443/// assert_eq!(request.start_after_key(), Some("run/10"));
444/// # Ok::<(), storage_kit::StateStoreError>(())
445/// ```
446#[derive(Clone, Debug)]
447pub struct StateListRequest {
448    namespace: StateNamespace,
449    prefix: Option<StateKeyPrefix>,
450    limit: usize,
451    start_after: Option<String>,
452}
453
454impl StateListRequest {
455    /// Default page size.
456    pub const DEFAULT_LIMIT: usize = DEFAULT_LIST_LIMIT;
457    /// Maximum page size.
458    pub const MAX_LIMIT: usize = MAX_LIST_LIMIT;
459
460    /// Creates a namespace-scoped request.
461    ///
462    /// The new request has no prefix or cursor and uses [`Self::DEFAULT_LIMIT`].
463    ///
464    /// # Errors
465    ///
466    /// Returns [`StateStoreError::InvalidRequest`] when the namespace does not
467    /// satisfy [`StateNamespace::new`].
468    ///
469    /// # Examples
470    ///
471    /// ```
472    /// use storage_kit::state::record::StateListRequest;
473    /// let request = StateListRequest::new("workflow")?;
474    /// assert_eq!(request.limit(), StateListRequest::DEFAULT_LIMIT);
475    /// assert_eq!(request.start_after_key(), None);
476    /// # Ok::<(), storage_kit::StateStoreError>(())
477    /// ```
478    pub fn new(namespace: impl Into<String>) -> Result<Self, StateStoreError> {
479        Ok(Self {
480            namespace: StateNamespace::new(namespace)?,
481            prefix: None,
482            limit: DEFAULT_LIST_LIMIT,
483            start_after: None,
484        })
485    }
486
487    /// Restricts results to a validated key prefix.
488    ///
489    /// # Examples
490    ///
491    /// ```
492    /// use storage_kit::state::record::{StateKeyPrefix, StateListRequest};
493    /// let request = StateListRequest::new("workflow")?
494    ///     .with_key_prefix(StateKeyPrefix::new("run/%")?);
495    /// assert_eq!(request.prefix(), Some("run/%"));
496    /// # Ok::<(), storage_kit::StateStoreError>(())
497    /// ```
498    #[must_use]
499    pub fn with_key_prefix(mut self, prefix: StateKeyPrefix) -> Self {
500        self.prefix = Some(prefix);
501        self
502    }
503
504    /// Replaces the page size.
505    ///
506    /// # Errors
507    ///
508    /// Returns [`StateStoreError::InvalidRequest`] unless `limit` is between
509    /// 1 and [`Self::MAX_LIMIT`] inclusive.
510    ///
511    /// # Examples
512    ///
513    /// ```
514    /// use storage_kit::state::record::StateListRequest;
515    /// let request = StateListRequest::new("workflow")?.with_limit(25)?;
516    /// assert_eq!(request.limit(), 25);
517    /// assert!(StateListRequest::new("workflow")?.with_limit(0).is_err());
518    /// # Ok::<(), storage_kit::StateStoreError>(())
519    /// ```
520    pub fn with_limit(mut self, limit: usize) -> Result<Self, StateStoreError> {
521        if !(1..=MAX_LIST_LIMIT).contains(&limit) {
522            return Err(StateStoreError::invalid_request(format!(
523                "state list limit must be between 1 and {MAX_LIST_LIMIT}"
524            )));
525        }
526        self.limit = limit;
527        Ok(self)
528    }
529
530    /// Sets the namespace-local exclusive cursor.
531    ///
532    /// Pass the previous response's `next_start_after` unchanged with the same
533    /// namespace and prefix. The cursor need not name a record that still exists.
534    ///
535    /// # Errors
536    ///
537    /// Returns [`StateStoreError::InvalidRequest`] for an empty cursor, more than
538    /// 1024 UTF-8 bytes, surrounding whitespace, or a control character.
539    ///
540    /// # Examples
541    ///
542    /// ```
543    /// use storage_kit::state::record::StateListRequest;
544    /// let request = StateListRequest::new("workflow")?.start_after("run/10")?;
545    /// assert_eq!(request.start_after_key(), Some("run/10"));
546    /// # Ok::<(), storage_kit::StateStoreError>(())
547    /// ```
548    pub fn start_after(mut self, cursor: impl Into<String>) -> Result<Self, StateStoreError> {
549        self.start_after = Some(StateNamespace::validate_fragment(
550            "start_after",
551            cursor.into(),
552            MAX_KEY_BYTES,
553        )?);
554        Ok(self)
555    }
556
557    /// Returns the namespace.
558    ///
559    /// # Examples
560    ///
561    /// ```
562    /// use storage_kit::state::record::StateListRequest;
563    /// let request = StateListRequest::new("workflow")?;
564    /// assert_eq!(request.namespace().as_str(), "workflow");
565    /// # Ok::<(), storage_kit::StateStoreError>(())
566    /// ```
567    #[must_use]
568    pub fn namespace(&self) -> &StateNamespace {
569        &self.namespace
570    }
571
572    /// Returns the optional typed prefix.
573    ///
574    /// # Examples
575    ///
576    /// ```
577    /// use storage_kit::state::record::{StateKeyPrefix, StateListRequest};
578    /// let prefix = StateKeyPrefix::new("run/")?;
579    /// let request = StateListRequest::new("workflow")?.with_key_prefix(prefix.clone());
580    /// assert_eq!(request.key_prefix(), Some(&prefix));
581    /// # Ok::<(), storage_kit::StateStoreError>(())
582    /// ```
583    #[must_use]
584    pub fn key_prefix(&self) -> Option<&StateKeyPrefix> {
585        self.prefix.as_ref()
586    }
587
588    /// Returns the optional prefix string.
589    ///
590    /// # Examples
591    ///
592    /// ```
593    /// use storage_kit::state::record::StateListRequest;
594    /// let request = StateListRequest::new("workflow")?;
595    /// assert_eq!(request.prefix(), None);
596    /// # Ok::<(), storage_kit::StateStoreError>(())
597    /// ```
598    #[must_use]
599    pub fn prefix(&self) -> Option<&str> {
600        self.prefix.as_ref().map(StateKeyPrefix::as_str)
601    }
602
603    /// Returns the page size.
604    ///
605    /// # Examples
606    ///
607    /// ```
608    /// use storage_kit::state::record::StateListRequest;
609    /// let request = StateListRequest::new("workflow")?.with_limit(12)?;
610    /// assert_eq!(request.limit(), 12);
611    /// # Ok::<(), storage_kit::StateStoreError>(())
612    /// ```
613    #[must_use]
614    pub fn limit(&self) -> usize {
615        self.limit
616    }
617
618    /// Returns the optional cursor.
619    ///
620    /// # Examples
621    ///
622    /// ```
623    /// use storage_kit::state::record::StateListRequest;
624    /// let request = StateListRequest::new("workflow")?.start_after("run/1")?;
625    /// assert_eq!(request.start_after_key(), Some("run/1"));
626    /// # Ok::<(), storage_kit::StateStoreError>(())
627    /// ```
628    #[must_use]
629    pub fn start_after_key(&self) -> Option<&str> {
630        self.start_after.as_deref()
631    }
632}
633
634/// One page of observed records.
635///
636/// Records are present values only; tombstones are omitted. A continuation is
637/// the last returned key only when another matching record exists in this
638/// page's read view. An empty result has no continuation. Reusing a cursor
639/// starts a new read view and does not freeze concurrently changing records.
640///
641/// # Examples
642///
643/// ```
644/// use storage_kit::state::record::{StateKey, StateRevision, StateRecord, StateRecordSnapshot, StateListRequest, StateListResponse};
645/// let record = StateRecord::new(StateKey::new("workflow", "run/1")?, serde_json::json!(42));
646/// let page = StateListResponse {
647///     records: vec![StateRecordSnapshot::new(record, StateRevision::first())],
648///     next_start_after: Some("run/1".to_owned()),
649/// };
650/// let next = StateListRequest::new("workflow")?
651///     .start_after(page.next_start_after.as_deref().unwrap())?;
652/// assert_eq!(next.start_after_key(), Some("run/1"));
653/// # Ok::<(), storage_kit::StateStoreError>(())
654/// ```
655#[derive(Clone, Debug, Default)]
656pub struct StateListResponse {
657    /// Returned records.
658    pub records: Vec<StateRecordSnapshot>,
659    /// Cursor for a subsequent page.
660    pub next_start_after: Option<String>,
661}
storage_kit/state/record.rs

storage_kit/state/
record.rs
