soaprs-memory 0.2.0

//! Reference in-memory implementation of soaprs ports.

mod cqrs;
mod events;

pub use cqrs::{
    MemoryEventOutbox, MemoryEventStore, MemoryInboxStore, MemoryOutboxStore,
    MemoryProjectionCheckpointStore, MemorySagaActionStore, MemorySagaOutbox, MemorySagaStore,
    MemorySnapshotStore, MemoryStoredEventStore, MemoryTransactionalProjection,
};
pub use events::{MemoryEventBus, SubscriptionId};

use std::{cmp::Ordering, sync::RwLock};

use soaprs_core::{BoxFuture, Entity, SoapError, SoapResult};
use soaprs_repository::{
    Condition, FieldName, FindParams, LogicalOperator, Operator, ReadRepository, ScalarValue, Sort,
    SortDirection, WriteRepository,
};

/// Provides logical field values to the in-memory query evaluator.
pub trait Queryable {
    /// Reports whether the entity exposes this logical query field.
    fn supports_field(field: &FieldName) -> bool;

    /// Returns the value of a supported logical field.
    ///
    /// Nullable fields return [`ScalarValue::Null`]. An unknown field must
    /// return [`SoapError::validation`], even though the repository validates
    /// field support before evaluating entities.
    fn field_value(&self, field: &FieldName) -> SoapResult<ScalarValue>;
}

/// Thread-safe in-memory repository used as a reference adapter.
#[derive(Debug)]
pub struct MemoryRepository<E> {
    entities: RwLock<Vec<E>>,
}

impl<E> MemoryRepository<E> {
    /// Creates an empty repository.
    pub const fn new() -> Self {
        Self {
            entities: RwLock::new(Vec::new()),
        }
    }

    fn read_entities(&self) -> SoapResult<std::sync::RwLockReadGuard<'_, Vec<E>>> {
        self.entities
            .read()
            .map_err(|_| SoapError::infrastructure("in-memory repository read lock poisoned"))
    }

    fn write_entities(&self) -> SoapResult<std::sync::RwLockWriteGuard<'_, Vec<E>>> {
        self.entities
            .write()
            .map_err(|_| SoapError::infrastructure("in-memory repository write lock poisoned"))
    }
}

impl<E> Default for MemoryRepository<E> {
    fn default() -> Self {
        Self::new()
    }
}

impl<E> ReadRepository<E> for MemoryRepository<E>
where
    E: Entity + Queryable + Clone + 'static,
{
    fn find(&self, params: FindParams) -> BoxFuture<'_, SoapResult<Vec<E>>> {
        Box::pin(async move {
            params.validate()?;
            validate_find_fields::<E>(&params)?;
            let entities = self.read_entities()?;
            let mut matched = entities
                .iter()
                .filter_map(
                    |entity| match matches_condition(entity, params.condition.as_ref()) {
                        Ok(true) => Some(Ok(entity.clone())),
                        Ok(false) => None,
                        Err(error) => Some(Err(error)),
                    },
                )
                .collect::<SoapResult<Vec<_>>>()?;

            for sort in params.sort.iter().rev() {
                sort_entities(&mut matched, sort)?;
            }

            let available = matched.len().saturating_sub(params.offset);
            let take = params.limit.unwrap_or(available);
            Ok(matched.into_iter().skip(params.offset).take(take).collect())
        })
    }

    fn get<'a>(&'a self, id: &'a E::Id) -> BoxFuture<'a, SoapResult<Option<E>>> {
        Box::pin(async move {
            let entities = self.read_entities()?;
            Ok(entities.iter().find(|entity| entity.id() == id).cloned())
        })
    }

    fn count(&self, params: FindParams) -> BoxFuture<'_, SoapResult<u64>> {
        Box::pin(async move {
            params.validate()?;
            if let Some(condition) = &params.condition {
                validate_condition_fields::<E>(condition)?;
            }
            let entities = self.read_entities()?;
            let mut count = 0_u64;
            for entity in entities.iter() {
                if matches_condition(entity, params.condition.as_ref())? {
                    count = count.saturating_add(1);
                }
            }
            Ok(count)
        })
    }
}

impl<E> WriteRepository<E> for MemoryRepository<E>
where
    E: Entity + Queryable + Clone + 'static,
{
    fn insert(&self, entity: E) -> BoxFuture<'_, SoapResult<()>> {
        Box::pin(async move {
            let mut entities = self.write_entities()?;
            if entities.iter().any(|existing| existing.id() == entity.id()) {
                return Err(SoapError::conflict("duplicate entity identifier"));
            }
            entities.push(entity);
            Ok(())
        })
    }

    fn replace(&self, entity: E) -> BoxFuture<'_, SoapResult<()>> {
        Box::pin(async move {
            let mut entities = self.write_entities()?;
            let Some(index) = entities
                .iter()
                .position(|existing| existing.id() == entity.id())
            else {
                return Err(SoapError::not_found("entity identifier"));
            };
            entities[index] = entity;
            Ok(())
        })
    }

    fn remove<'a>(&'a self, id: &'a E::Id) -> BoxFuture<'a, SoapResult<bool>> {
        Box::pin(async move {
            let mut entities = self.write_entities()?;
            let Some(index) = entities.iter().position(|entity| entity.id() == id) else {
                return Ok(false);
            };
            entities.remove(index);
            Ok(true)
        })
    }
}

fn matches_condition<E>(entity: &E, condition: Option<&Condition>) -> SoapResult<bool>
where
    E: Queryable,
{
    match condition {
        None => Ok(true),
        Some(Condition::Group {
            operator,
            conditions,
        }) => match operator {
            LogicalOperator::And => {
                for condition in conditions {
                    if !matches_condition(entity, Some(condition))? {
                        return Ok(false);
                    }
                }
                Ok(true)
            }
            LogicalOperator::Or => {
                for condition in conditions {
                    if matches_condition(entity, Some(condition))? {
                        return Ok(true);
                    }
                }
                Ok(false)
            }
        },
        Some(Condition::Predicate {
            field,
            operator,
            value,
        }) => evaluate_predicate(entity.field_value(field)?, *operator, value.as_ref()),
    }
}

fn evaluate_predicate(
    actual: ScalarValue,
    operator: Operator,
    expected: Option<&ScalarValue>,
) -> SoapResult<bool> {
    validate_entity_value(&actual)?;
    match operator {
        Operator::IsNull => Ok(matches!(actual, ScalarValue::Null)),
        Operator::IsNotNull => Ok(!matches!(actual, ScalarValue::Null)),
        _ if matches!(actual, ScalarValue::Null) => Ok(false),
        Operator::Eq | Operator::Ne => {
            let expected = expected
                .ok_or_else(|| SoapError::validation("equality requires a non-null value"))?;
            let equivalent = actual
                .compare(expected)
                .ok_or_else(|| SoapError::validation("equality requires compatible value types"))?
                .is_eq();
            Ok(if operator == Operator::Eq {
                equivalent
            } else {
                !equivalent
            })
        }
        Operator::Gt | Operator::Gte | Operator::Lt | Operator::Lte => {
            let ordering = comparable_ordering(Some(&actual), expected)?;
            Ok(match operator {
                Operator::Gt => ordering == Ordering::Greater,
                Operator::Gte => ordering != Ordering::Less,
                Operator::Lt => ordering == Ordering::Less,
                Operator::Lte => ordering != Ordering::Greater,
                _ => false,
            })
        }
        Operator::In | Operator::NotIn => {
            let Some(ScalarValue::List(values)) = expected else {
                return Err(SoapError::validation("set operators require a list value"));
            };
            if let Some(first) = values.first() {
                if actual.compare(first).is_none() {
                    return Err(SoapError::validation(
                        "set membership requires compatible value types",
                    ));
                }
            }
            let contains = values.iter().any(|item| actual.equivalent(item));
            Ok(if operator == Operator::In {
                contains
            } else {
                !contains
            })
        }
        Operator::Like => {
            let (ScalarValue::String(actual), Some(ScalarValue::String(pattern))) =
                (&actual, expected)
            else {
                return Err(SoapError::validation("LIKE requires string values"));
            };
            Ok(like_matches(actual, pattern))
        }
    }
}

fn validate_find_fields<E>(params: &FindParams) -> SoapResult<()>
where
    E: Queryable,
{
    if let Some(condition) = &params.condition {
        validate_condition_fields::<E>(condition)?;
    }
    for sort in &params.sort {
        validate_field::<E>(&sort.field)?;
    }
    Ok(())
}

fn validate_condition_fields<E>(condition: &Condition) -> SoapResult<()>
where
    E: Queryable,
{
    match condition {
        Condition::Predicate { field, .. } => validate_field::<E>(field),
        Condition::Group { conditions, .. } => {
            for condition in conditions {
                validate_condition_fields::<E>(condition)?;
            }
            Ok(())
        }
    }
}

fn validate_field<E>(field: &FieldName) -> SoapResult<()>
where
    E: Queryable,
{
    if E::supports_field(field) {
        Ok(())
    } else {
        Err(SoapError::validation(format!(
            "unknown query field `{field}`"
        )))
    }
}

fn sort_entities<E>(entities: &mut [E], sort: &Sort) -> SoapResult<()>
where
    E: Queryable,
{
    for entity in entities.iter() {
        validate_entity_value(&entity.field_value(&sort.field)?)?;
    }

    let mut failure = None;
    entities.sort_by(|left, right| {
        if failure.is_some() {
            return Ordering::Equal;
        }

        match compare_sort_values(
            left.field_value(&sort.field),
            right.field_value(&sort.field),
        ) {
            Ok(ordering) => match sort.direction {
                SortDirection::Ascending => ordering,
                SortDirection::Descending => ordering.reverse(),
            },
            Err(error) => {
                failure = Some(error);
                Ordering::Equal
            }
        }
    });

    match failure {
        Some(error) => Err(error),
        None => Ok(()),
    }
}

fn compare_sort_values(
    left: SoapResult<ScalarValue>,
    right: SoapResult<ScalarValue>,
) -> SoapResult<Ordering> {
    let left = left?;
    let right = right?;
    validate_entity_value(&left)?;
    validate_entity_value(&right)?;

    match (&left, &right) {
        (ScalarValue::Null, ScalarValue::Null) => Ok(Ordering::Equal),
        (ScalarValue::Null, _) => Ok(Ordering::Less),
        (_, ScalarValue::Null) => Ok(Ordering::Greater),
        _ => left
            .compare(&right)
            .ok_or_else(|| SoapError::validation("sorting requires compatible field value types")),
    }
}

fn validate_entity_value(value: &ScalarValue) -> SoapResult<()> {
    value.validate()?;
    if matches!(value, ScalarValue::List(_)) {
        Err(SoapError::validation(
            "entity query fields must contain scalar values",
        ))
    } else {
        Ok(())
    }
}

fn comparable_ordering(
    actual: Option<&ScalarValue>,
    expected: Option<&ScalarValue>,
) -> SoapResult<Ordering> {
    let (Some(actual), Some(expected)) = (actual, expected) else {
        return Err(SoapError::validation(
            "comparison requires two non-null values",
        ));
    };
    actual
        .compare(expected)
        .ok_or_else(|| SoapError::validation("comparison requires compatible value types"))
}

fn like_matches(value: &str, pattern: &str) -> bool {
    let value: Vec<_> = value.chars().collect();
    let pattern: Vec<_> = pattern.chars().collect();
    let mut table = vec![vec![false; pattern.len() + 1]; value.len() + 1];
    table[0][0] = true;
    for pattern_index in 1..=pattern.len() {
        if pattern[pattern_index - 1] == '%' {
            table[0][pattern_index] = table[0][pattern_index - 1];
        }
    }
    for value_index in 1..=value.len() {
        for pattern_index in 1..=pattern.len() {
            table[value_index][pattern_index] = match pattern[pattern_index - 1] {
                '%' => {
                    table[value_index][pattern_index - 1] || table[value_index - 1][pattern_index]
                }
                '_' => table[value_index - 1][pattern_index - 1],
                character => {
                    character == value[value_index - 1] && table[value_index - 1][pattern_index - 1]
                }
            };
        }
    }
    table[value.len()][pattern.len()]
}

#[cfg(test)]
mod tests {
    use std::cmp::Ordering;

    use soaprs_core::{SoapError, SoapErrorKind};
    use soaprs_repository::{Operator, ScalarValue};

    use super::{comparable_ordering, evaluate_predicate, like_matches};

    #[test]
    fn evaluates_all_comparison_operators() {
        let actual = ScalarValue::I64(10);

        assert_eq!(
            evaluate_predicate(actual.clone(), Operator::Eq, Some(&ScalarValue::U64(10))).ok(),
            Some(true)
        );
        assert_eq!(
            evaluate_predicate(actual.clone(), Operator::Ne, Some(&ScalarValue::I64(11))).ok(),
            Some(true)
        );
        assert_eq!(
            evaluate_predicate(actual.clone(), Operator::Gt, Some(&ScalarValue::I64(9))).ok(),
            Some(true)
        );
        assert_eq!(
            evaluate_predicate(actual.clone(), Operator::Gte, Some(&ScalarValue::I64(10))).ok(),
            Some(true)
        );
        assert_eq!(
            evaluate_predicate(actual.clone(), Operator::Lt, Some(&ScalarValue::I64(11))).ok(),
            Some(true)
        );
        assert_eq!(
            evaluate_predicate(actual, Operator::Lte, Some(&ScalarValue::F64(10.0))).ok(),
            Some(true)
        );
    }

    #[test]
    fn evaluates_set_and_null_operators() {
        let values = ScalarValue::List(vec![ScalarValue::I64(1), ScalarValue::U64(2)]);

        assert_eq!(
            evaluate_predicate(ScalarValue::I64(2), Operator::In, Some(&values)).ok(),
            Some(true)
        );
        assert_eq!(
            evaluate_predicate(ScalarValue::I64(3), Operator::NotIn, Some(&values)).ok(),
            Some(true)
        );
        assert_eq!(
            evaluate_predicate(ScalarValue::Null, Operator::IsNull, None).ok(),
            Some(true)
        );
        assert_eq!(
            evaluate_predicate(ScalarValue::Null, Operator::IsNotNull, None).ok(),
            Some(false)
        );
        assert_eq!(
            evaluate_predicate(ScalarValue::Null, Operator::Ne, Some(&ScalarValue::I64(1))).ok(),
            Some(false)
        );
    }

    #[test]
    fn like_uses_characters_instead_of_utf8_bytes() {
        assert!(like_matches("Łódź", "_ód%"));
        assert!(!like_matches("Łódź", "__ód%"));
        assert!(like_matches(r"a\b", r"a\b"));
    }

    #[test]
    fn incompatible_comparisons_return_validation_errors() {
        let result = comparable_ordering(
            Some(&ScalarValue::String("10".into())),
            Some(&ScalarValue::I64(10)),
        );

        assert_eq!(
            result.as_ref().map_err(SoapError::kind),
            Err(SoapErrorKind::Validation)
        );
        assert_ne!(result.ok(), Some(Ordering::Equal));
    }
}