soaprs_memory/
lib.rs

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

soaprs_memory/
lib.rs
