runifold_effect/
store.rs

1use std::{
2    collections::BTreeMap,
3    sync::{Arc, Mutex, MutexGuard},
4};
5
6use runifold_core::{CapabilityId, EffectId};
7
8use crate::{EffectExecutorError, EffectExecutorErrorKind, EffectRecord};
9
10/// Atomic persistence boundary for write-ahead effect records.
11pub trait EffectStore: Send + Sync {
12    /// Loads a record by effect identity.
13    ///
14    /// # Errors
15    ///
16    /// Returns [`EffectExecutorError`] on storage failure.
17    fn load(&self, id: EffectId) -> Result<Option<EffectRecord>, EffectExecutorError>;
18
19    /// Resolves a logical effect by capability-scoped idempotency key.
20    ///
21    /// # Errors
22    ///
23    /// Returns [`EffectExecutorError`] on storage failure.
24    fn find_by_idempotency(
25        &self,
26        capability_id: CapabilityId,
27        key: &str,
28    ) -> Result<Option<EffectRecord>, EffectExecutorError>;
29
30    /// Creates or atomically replaces a record.
31    ///
32    /// `None` is create-only. Updates require the exact current revision and
33    /// a new revision exactly one greater.
34    ///
35    /// # Errors
36    ///
37    /// Returns [`EffectExecutorError`] on storage failure or conflict.
38    fn compare_and_swap(
39        &self,
40        record: &EffectRecord,
41        expected_revision: Option<u64>,
42    ) -> Result<(), EffectExecutorError>;
43}
44
45/// In-memory effect store with atomic idempotency indexing.
46#[derive(Clone, Debug, Default)]
47pub struct InMemoryEffectStore {
48    state: Arc<Mutex<StoreState>>,
49}
50
51#[derive(Debug, Default)]
52struct StoreState {
53    records: BTreeMap<EffectId, EffectRecord>,
54    idempotency: BTreeMap<(CapabilityId, String), EffectId>,
55}
56
57impl InMemoryEffectStore {
58    /// Creates an empty store.
59    pub fn new() -> Self {
60        Self::default()
61    }
62
63    fn state(&self) -> MutexGuard<'_, StoreState> {
64        self.state
65            .lock()
66            .unwrap_or_else(std::sync::PoisonError::into_inner)
67    }
68}
69
70impl EffectStore for InMemoryEffectStore {
71    fn load(&self, id: EffectId) -> Result<Option<EffectRecord>, EffectExecutorError> {
72        Ok(self.state().records.get(&id).cloned())
73    }
74
75    fn find_by_idempotency(
76        &self,
77        capability_id: CapabilityId,
78        key: &str,
79    ) -> Result<Option<EffectRecord>, EffectExecutorError> {
80        let state = self.state();
81        Ok(state
82            .idempotency
83            .get(&(capability_id, key.into()))
84            .and_then(|id| state.records.get(id))
85            .cloned())
86    }
87
88    fn compare_and_swap(
89        &self,
90        record: &EffectRecord,
91        expected_revision: Option<u64>,
92    ) -> Result<(), EffectExecutorError> {
93        let mut state = self.state();
94        let current = state.records.get(&record.request.effect_id);
95        let valid = match (current, expected_revision) {
96            (None, None) => record.revision == 0,
97            (Some(current), Some(expected)) => {
98                current.revision == expected
99                    && expected
100                        .checked_add(1)
101                        .is_some_and(|next| record.revision == next)
102            }
103            _ => false,
104        };
105        if !valid {
106            return Err(EffectExecutorError::new(
107                EffectExecutorErrorKind::Store,
108                "effect record revision precondition failed",
109            ));
110        }
111
112        if let Some(key) = &record.request.idempotency_key {
113            let index = (record.request.capability_id, key.clone());
114            if let Some(existing) = state.idempotency.get(&index)
115                && *existing != record.request.effect_id
116            {
117                return Err(EffectExecutorError::new(
118                    EffectExecutorErrorKind::IdempotencyConflict,
119                    "idempotency key already belongs to another effect",
120                ));
121            }
122            state.idempotency.insert(index, record.request.effect_id);
123        }
124        state
125            .records
126            .insert(record.request.effect_id, record.clone());
127        Ok(())
128    }
129}
runifold_effect/store.rs

runifold_effect/
store.rs
