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

storage_kit/state/
record.rs
