soaprs_memory/
lib.rs

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

soaprs_memory/
lib.rs
