soaprs_memory/
lib.rs

1//! Reference in-memory implementation of soaprs ports.
2
3mod auth;
4mod cqrs;
5mod events;
6
7pub use auth::MemorySessionStore;
8pub use cqrs::{
9    MemoryEventOutbox, MemoryEventStore, MemoryInboxStore, MemoryOutboxStore,
10    MemoryProjectionCheckpointStore, MemorySagaActionStore, MemorySagaOutbox, MemorySagaStore,
11    MemorySnapshotStore, MemoryStoredEventStore, MemoryTransactionalProjection,
12};
13pub use events::{MemoryEventBus, SubscriptionId};
14
15use std::{cmp::Ordering, sync::RwLock};
16
17use soaprs_core::{BoxFuture, Entity, SoapError, SoapResult};
18use soaprs_repository::{
19    Condition, FieldName, FindParams, LogicalOperator, Operator, ReadRepository, ScalarValue, Sort,
20    SortDirection, WriteRepository,
21};
22
23/// Provides logical field values to the in-memory query evaluator.
24pub trait Queryable {
25    /// Reports whether the entity exposes this logical query field.
26    fn supports_field(field: &FieldName) -> bool;
27
28    /// Returns the value of a supported logical field.
29    ///
30    /// Nullable fields return [`ScalarValue::Null`]. An unknown field must
31    /// return [`SoapError::validation`], even though the repository validates
32    /// field support before evaluating entities.
33    fn field_value(&self, field: &FieldName) -> SoapResult<ScalarValue>;
34}
35
36/// Thread-safe in-memory repository used as a reference adapter.
37#[derive(Debug)]
38pub struct MemoryRepository<E> {
39    entities: RwLock<Vec<E>>,
40}
41
42impl<E> MemoryRepository<E> {
43    /// Creates an empty repository.
44    pub const fn new() -> Self {
45        Self {
46            entities: RwLock::new(Vec::new()),
47        }
48    }
49
50    fn read_entities(&self) -> SoapResult<std::sync::RwLockReadGuard<'_, Vec<E>>> {
51        self.entities
52            .read()
53            .map_err(|_| SoapError::infrastructure("in-memory repository read lock poisoned"))
54    }
55
56    fn write_entities(&self) -> SoapResult<std::sync::RwLockWriteGuard<'_, Vec<E>>> {
57        self.entities
58            .write()
59            .map_err(|_| SoapError::infrastructure("in-memory repository write lock poisoned"))
60    }
61}
62
63impl<E> Default for MemoryRepository<E> {
64    fn default() -> Self {
65        Self::new()
66    }
67}
68
69impl<E> ReadRepository<E> for MemoryRepository<E>
70where
71    E: Entity + Queryable + Clone + 'static,
72{
73    fn find(&self, params: FindParams) -> BoxFuture<'_, SoapResult<Vec<E>>> {
74        Box::pin(async move {
75            params.validate()?;
76            validate_find_fields::<E>(&params)?;
77            let entities = self.read_entities()?;
78            let mut matched = entities
79                .iter()
80                .filter_map(
81                    |entity| match matches_condition(entity, params.condition.as_ref()) {
82                        Ok(true) => Some(Ok(entity.clone())),
83                        Ok(false) => None,
84                        Err(error) => Some(Err(error)),
85                    },
86                )
87                .collect::<SoapResult<Vec<_>>>()?;
88
89            for sort in params.sort.iter().rev() {
90                sort_entities(&mut matched, sort)?;
91            }
92
93            let available = matched.len().saturating_sub(params.offset);
94            let take = params.limit.unwrap_or(available);
95            Ok(matched.into_iter().skip(params.offset).take(take).collect())
96        })
97    }
98
99    fn get<'a>(&'a self, id: &'a E::Id) -> BoxFuture<'a, SoapResult<Option<E>>> {
100        Box::pin(async move {
101            let entities = self.read_entities()?;
102            Ok(entities.iter().find(|entity| entity.id() == id).cloned())
103        })
104    }
105
106    fn count(&self, params: FindParams) -> BoxFuture<'_, SoapResult<u64>> {
107        Box::pin(async move {
108            params.validate()?;
109            if let Some(condition) = &params.condition {
110                validate_condition_fields::<E>(condition)?;
111            }
112            let entities = self.read_entities()?;
113            let mut count = 0_u64;
114            for entity in entities.iter() {
115                if matches_condition(entity, params.condition.as_ref())? {
116                    count = count.saturating_add(1);
117                }
118            }
119            Ok(count)
120        })
121    }
122}
123
124impl<E> WriteRepository<E> for MemoryRepository<E>
125where
126    E: Entity + Queryable + Clone + 'static,
127{
128    fn insert(&self, entity: E) -> BoxFuture<'_, SoapResult<()>> {
129        Box::pin(async move {
130            let mut entities = self.write_entities()?;
131            if entities.iter().any(|existing| existing.id() == entity.id()) {
132                return Err(SoapError::conflict("duplicate entity identifier"));
133            }
134            entities.push(entity);
135            Ok(())
136        })
137    }
138
139    fn replace(&self, entity: E) -> BoxFuture<'_, SoapResult<()>> {
140        Box::pin(async move {
141            let mut entities = self.write_entities()?;
142            let Some(index) = entities
143                .iter()
144                .position(|existing| existing.id() == entity.id())
145            else {
146                return Err(SoapError::not_found("entity identifier"));
147            };
148            entities[index] = entity;
149            Ok(())
150        })
151    }
152
153    fn remove<'a>(&'a self, id: &'a E::Id) -> BoxFuture<'a, SoapResult<bool>> {
154        Box::pin(async move {
155            let mut entities = self.write_entities()?;
156            let Some(index) = entities.iter().position(|entity| entity.id() == id) else {
157                return Ok(false);
158            };
159            entities.remove(index);
160            Ok(true)
161        })
162    }
163}
164
165fn matches_condition<E>(entity: &E, condition: Option<&Condition>) -> SoapResult<bool>
166where
167    E: Queryable,
168{
169    match condition {
170        None => Ok(true),
171        Some(Condition::Group {
172            operator,
173            conditions,
174        }) => match operator {
175            LogicalOperator::And => {
176                for condition in conditions {
177                    if !matches_condition(entity, Some(condition))? {
178                        return Ok(false);
179                    }
180                }
181                Ok(true)
182            }
183            LogicalOperator::Or => {
184                for condition in conditions {
185                    if matches_condition(entity, Some(condition))? {
186                        return Ok(true);
187                    }
188                }
189                Ok(false)
190            }
191        },
192        Some(Condition::Predicate {
193            field,
194            operator,
195            value,
196        }) => evaluate_predicate(entity.field_value(field)?, *operator, value.as_ref()),
197    }
198}
199
200fn evaluate_predicate(
201    actual: ScalarValue,
202    operator: Operator,
203    expected: Option<&ScalarValue>,
204) -> SoapResult<bool> {
205    validate_entity_value(&actual)?;
206    match operator {
207        Operator::IsNull => Ok(matches!(actual, ScalarValue::Null)),
208        Operator::IsNotNull => Ok(!matches!(actual, ScalarValue::Null)),
209        _ if matches!(actual, ScalarValue::Null) => Ok(false),
210        Operator::Eq | Operator::Ne => {
211            let expected = expected
212                .ok_or_else(|| SoapError::validation("equality requires a non-null value"))?;
213            let equivalent = actual
214                .compare(expected)
215                .ok_or_else(|| SoapError::validation("equality requires compatible value types"))?
216                .is_eq();
217            Ok(if operator == Operator::Eq {
218                equivalent
219            } else {
220                !equivalent
221            })
222        }
223        Operator::Gt | Operator::Gte | Operator::Lt | Operator::Lte => {
224            let ordering = comparable_ordering(Some(&actual), expected)?;
225            Ok(match operator {
226                Operator::Gt => ordering == Ordering::Greater,
227                Operator::Gte => ordering != Ordering::Less,
228                Operator::Lt => ordering == Ordering::Less,
229                Operator::Lte => ordering != Ordering::Greater,
230                _ => false,
231            })
232        }
233        Operator::In | Operator::NotIn => {
234            let Some(ScalarValue::List(values)) = expected else {
235                return Err(SoapError::validation("set operators require a list value"));
236            };
237            if let Some(first) = values.first() {
238                if actual.compare(first).is_none() {
239                    return Err(SoapError::validation(
240                        "set membership requires compatible value types",
241                    ));
242                }
243            }
244            let contains = values.iter().any(|item| actual.equivalent(item));
245            Ok(if operator == Operator::In {
246                contains
247            } else {
248                !contains
249            })
250        }
251        Operator::Like => {
252            let (ScalarValue::String(actual), Some(ScalarValue::String(pattern))) =
253                (&actual, expected)
254            else {
255                return Err(SoapError::validation("LIKE requires string values"));
256            };
257            Ok(like_matches(actual, pattern))
258        }
259    }
260}
261
262fn validate_find_fields<E>(params: &FindParams) -> SoapResult<()>
263where
264    E: Queryable,
265{
266    if let Some(condition) = &params.condition {
267        validate_condition_fields::<E>(condition)?;
268    }
269    for sort in &params.sort {
270        validate_field::<E>(&sort.field)?;
271    }
272    Ok(())
273}
274
275fn validate_condition_fields<E>(condition: &Condition) -> SoapResult<()>
276where
277    E: Queryable,
278{
279    match condition {
280        Condition::Predicate { field, .. } => validate_field::<E>(field),
281        Condition::Group { conditions, .. } => {
282            for condition in conditions {
283                validate_condition_fields::<E>(condition)?;
284            }
285            Ok(())
286        }
287    }
288}
289
290fn validate_field<E>(field: &FieldName) -> SoapResult<()>
291where
292    E: Queryable,
293{
294    if E::supports_field(field) {
295        Ok(())
296    } else {
297        Err(SoapError::validation(format!(
298            "unknown query field `{field}`"
299        )))
300    }
301}
302
303fn sort_entities<E>(entities: &mut [E], sort: &Sort) -> SoapResult<()>
304where
305    E: Queryable,
306{
307    for entity in entities.iter() {
308        validate_entity_value(&entity.field_value(&sort.field)?)?;
309    }
310
311    let mut failure = None;
312    entities.sort_by(|left, right| {
313        if failure.is_some() {
314            return Ordering::Equal;
315        }
316
317        match compare_sort_values(
318            left.field_value(&sort.field),
319            right.field_value(&sort.field),
320        ) {
321            Ok(ordering) => match sort.direction {
322                SortDirection::Ascending => ordering,
323                SortDirection::Descending => ordering.reverse(),
324            },
325            Err(error) => {
326                failure = Some(error);
327                Ordering::Equal
328            }
329        }
330    });
331
332    match failure {
333        Some(error) => Err(error),
334        None => Ok(()),
335    }
336}
337
338fn compare_sort_values(
339    left: SoapResult<ScalarValue>,
340    right: SoapResult<ScalarValue>,
341) -> SoapResult<Ordering> {
342    let left = left?;
343    let right = right?;
344    validate_entity_value(&left)?;
345    validate_entity_value(&right)?;
346
347    match (&left, &right) {
348        (ScalarValue::Null, ScalarValue::Null) => Ok(Ordering::Equal),
349        (ScalarValue::Null, _) => Ok(Ordering::Less),
350        (_, ScalarValue::Null) => Ok(Ordering::Greater),
351        _ => left
352            .compare(&right)
353            .ok_or_else(|| SoapError::validation("sorting requires compatible field value types")),
354    }
355}
356
357fn validate_entity_value(value: &ScalarValue) -> SoapResult<()> {
358    value.validate()?;
359    if matches!(value, ScalarValue::List(_)) {
360        Err(SoapError::validation(
361            "entity query fields must contain scalar values",
362        ))
363    } else {
364        Ok(())
365    }
366}
367
368fn comparable_ordering(
369    actual: Option<&ScalarValue>,
370    expected: Option<&ScalarValue>,
371) -> SoapResult<Ordering> {
372    let (Some(actual), Some(expected)) = (actual, expected) else {
373        return Err(SoapError::validation(
374            "comparison requires two non-null values",
375        ));
376    };
377    actual
378        .compare(expected)
379        .ok_or_else(|| SoapError::validation("comparison requires compatible value types"))
380}
381
382fn like_matches(value: &str, pattern: &str) -> bool {
383    let value: Vec<_> = value.chars().collect();
384    let pattern: Vec<_> = pattern.chars().collect();
385    let mut table = vec![vec![false; pattern.len() + 1]; value.len() + 1];
386    table[0][0] = true;
387    for pattern_index in 1..=pattern.len() {
388        if pattern[pattern_index - 1] == '%' {
389            table[0][pattern_index] = table[0][pattern_index - 1];
390        }
391    }
392    for value_index in 1..=value.len() {
393        for pattern_index in 1..=pattern.len() {
394            table[value_index][pattern_index] = match pattern[pattern_index - 1] {
395                '%' => {
396                    table[value_index][pattern_index - 1] || table[value_index - 1][pattern_index]
397                }
398                '_' => table[value_index - 1][pattern_index - 1],
399                character => {
400                    character == value[value_index - 1] && table[value_index - 1][pattern_index - 1]
401                }
402            };
403        }
404    }
405    table[value.len()][pattern.len()]
406}
407
408#[cfg(test)]
409mod tests {
410    use std::cmp::Ordering;
411
412    use soaprs_core::{SoapError, SoapErrorKind};
413    use soaprs_repository::{Operator, ScalarValue};
414
415    use super::{comparable_ordering, evaluate_predicate, like_matches};
416
417    #[test]
418    fn evaluates_all_comparison_operators() {
419        let actual = ScalarValue::I64(10);
420
421        assert_eq!(
422            evaluate_predicate(actual.clone(), Operator::Eq, Some(&ScalarValue::U64(10))).ok(),
423            Some(true)
424        );
425        assert_eq!(
426            evaluate_predicate(actual.clone(), Operator::Ne, Some(&ScalarValue::I64(11))).ok(),
427            Some(true)
428        );
429        assert_eq!(
430            evaluate_predicate(actual.clone(), Operator::Gt, Some(&ScalarValue::I64(9))).ok(),
431            Some(true)
432        );
433        assert_eq!(
434            evaluate_predicate(actual.clone(), Operator::Gte, Some(&ScalarValue::I64(10))).ok(),
435            Some(true)
436        );
437        assert_eq!(
438            evaluate_predicate(actual.clone(), Operator::Lt, Some(&ScalarValue::I64(11))).ok(),
439            Some(true)
440        );
441        assert_eq!(
442            evaluate_predicate(actual, Operator::Lte, Some(&ScalarValue::F64(10.0))).ok(),
443            Some(true)
444        );
445    }
446
447    #[test]
448    fn evaluates_set_and_null_operators() {
449        let values = ScalarValue::List(vec![ScalarValue::I64(1), ScalarValue::U64(2)]);
450
451        assert_eq!(
452            evaluate_predicate(ScalarValue::I64(2), Operator::In, Some(&values)).ok(),
453            Some(true)
454        );
455        assert_eq!(
456            evaluate_predicate(ScalarValue::I64(3), Operator::NotIn, Some(&values)).ok(),
457            Some(true)
458        );
459        assert_eq!(
460            evaluate_predicate(ScalarValue::Null, Operator::IsNull, None).ok(),
461            Some(true)
462        );
463        assert_eq!(
464            evaluate_predicate(ScalarValue::Null, Operator::IsNotNull, None).ok(),
465            Some(false)
466        );
467        assert_eq!(
468            evaluate_predicate(ScalarValue::Null, Operator::Ne, Some(&ScalarValue::I64(1))).ok(),
469            Some(false)
470        );
471    }
472
473    #[test]
474    fn like_uses_characters_instead_of_utf8_bytes() {
475        assert!(like_matches("Łódź", "_ód%"));
476        assert!(!like_matches("Łódź", "__ód%"));
477        assert!(like_matches(r"a\b", r"a\b"));
478    }
479
480    #[test]
481    fn incompatible_comparisons_return_validation_errors() {
482        let result = comparable_ordering(
483            Some(&ScalarValue::String("10".into())),
484            Some(&ScalarValue::I64(10)),
485        );
486
487        assert_eq!(
488            result.as_ref().map_err(SoapError::kind),
489            Err(SoapErrorKind::Validation)
490        );
491        assert_ne!(result.ok(), Some(Ordering::Equal));
492    }
493}
soaprs_memory/lib.rs

soaprs_memory/
lib.rs
