horfimbor_eventsource/

repository.rs

//! the repository mod is where the heavy lifting occurs
//! read / write to the db, play event, add command ...

use std::cmp::Ordering;
use std::fmt::Debug;
use std::marker::PhantomData;

use async_trait::async_trait;
use kurrentdb::{
    AppendToStreamOptions, Client as EventDb, Error, EventData, ReadStreamOptions, StreamPosition,
    StreamState, SubscribeToPersistentSubscriptionOptions,
};
use serde::de::DeserializeOwned;
use serde::{Deserialize, Serialize};

use crate::cache_db::CacheDb;
use crate::helper::create_subscription;
use crate::metadata::{CompleteEvent, Metadata};
use crate::model_key::ModelKey;
use crate::{Dto, EventSourceError, EventSourceStateError};
use crate::{State, Stream};

/// the `DtoRepository` is the reading part of the event storage
/// multiple `DtoRepository` can listen to the event stream but produce
/// different model.
#[derive(Clone)]
#[allow(clippy::module_name_repetitions)]
pub struct DtoRepository<D, C>
where
    D: Dto,
    C: CacheDb<D>,
{
    event_db: EventDb,
    cache_db: C,
    repository_kind: RepositoryKind,
    dto: PhantomData<D>,
}

/// the `StateRepository` is the central piece of the project
/// the update are done with Command on State
/// and the query are done by recomputing all event or reading the `CacheDB`
#[derive(Clone)]
#[allow(clippy::module_name_repetitions)]
pub struct StateRepository<S, C>
where
    S: State,
    C: CacheDb<S>,
{
    event_db: EventDb,
    state_db: C,
    repository_kind: RepositoryKind,
    state: PhantomData<S>,
}

/// `ModelWithPosition` is a helper to add the position to any model
#[derive(Default, Serialize, Deserialize, Debug, Clone)]
pub struct ModelWithPosition<M> {
    position: Option<u64>,
    model: M,
}

impl<M> ModelWithPosition<M>
where
    M: Dto,
{
    /// get the current state
    pub const fn state(&self) -> &M {
        &self.model
    }

    /// play one event to the current model to compute the next
    pub fn play_event(&mut self, event: &M::Event, position: Option<u64>) {
        self.model.play_event(event);

        self.position = position;
    }

    /// the position is the number of event in the event store
    /// `None` indicate that the topic is empty
    pub const fn position(&self) -> Option<u64> {
        self.position
    }
}

/// the enum `RepositoryKind` allow to differentiate the state and dto caches
/// and also to have multiple dto listening to the same events
#[derive(Clone)]
pub enum RepositoryKind {
    /// this repository kind have a cache that is not prefixes
    State,
    /// this is most likely be the name of the dto
    Dto(&'static str),
}

impl RepositoryKind {
    const fn to_cache_prefix(&self) -> Option<&'static str> {
        match self {
            Self::State => None,
            Self::Dto(p) => Some(p),
        }
    }
}

#[allow(missing_docs)]
#[async_trait]
/// the trait `Repository` allow to reconstruct the `Dto` from the `EventDb`
pub trait DtoRepositoryConstructor<D, C>: Clone + Send
where
    D: Dto,
    C: CacheDb<D>,
{
    /// A repository need the `EventDb`, a cache system and a kind,
    /// the cache system can be the `NoCache` provided in `cache_db`,
    /// the kind is used only to avoid cache collision,
    /// the type system take care of everything else.
    fn new(event_db: EventDb, cache_db: C, repository_kind: RepositoryKind) -> Self;
}

#[allow(missing_docs)]
#[async_trait]
/// the trait `Repository` allow to reconstruct the `Dto` from the `EventDb`
pub trait StateRepositoryConstructor<D, C>: Clone + Send
where
    D: Dto,
    C: CacheDb<D>,
{
    /// A repository need the `EventDb`, a cache system and a kind,
    /// the cache system can be the `NoCache` provided in `cache_db`,
    /// the type system take care of everything else.
    fn new(event_db: EventDb, cache_db: C) -> Self;
}

#[allow(missing_docs)]
#[async_trait]
/// the trait `Repository` allow to reconstruct the `Dto` from the `EventDb`
pub trait Repository<D, C>: Clone + Send
where
    D: Dto,
    C: CacheDb<D>,
{
    /// Getter for the `EventDb`
    fn event_db(&self) -> &EventDb;

    /// Getter for the cache
    fn cache_db(&self) -> &C;

    /// Getter for the cache
    fn repository_kind(&self) -> &RepositoryKind;

    async fn get_model(&self, key: &ModelKey) -> Result<ModelWithPosition<D>, EventSourceError>
    where
        D: Dto + DeserializeOwned,
    {
        let value = self
            .cache_db()
            .get(self.repository_kind().to_cache_prefix(), key)
            .map_err(EventSourceError::CacheDbError)?;

        self.complete_from_es(key, &value).await
    }

    async fn complete_from_es(
        &self,
        key: &ModelKey,
        value: &ModelWithPosition<D>,
    ) -> Result<ModelWithPosition<D>, EventSourceError> {
        let mut dto: D = value.model.clone();
        let mut position = value.position;

        let options = ReadStreamOptions::default();
        let options = if let Some(position) = value.position {
            options.position(StreamPosition::Position(position + 1))
        } else {
            options.position(StreamPosition::Start)
        };

        let mut stream = self
            .event_db()
            .read_stream(key.format(), &options)
            .await
            .map_err(EventSourceError::EventStore)?;

        while let Ok(Some(json_event)) = stream.next().await {
            let original_event = json_event.get_original_event();

            let metadata: Metadata =
                serde_json::from_slice(original_event.custom_metadata.as_ref())
                    .map_err(EventSourceError::Serde)?;

            if metadata.is_event() {
                let event = original_event
                    .as_json::<D::Event>()
                    .map_err(EventSourceError::Serde)?;

                dto.play_event(&event);
            }

            position = Some(original_event.revision);
        }

        let result = ModelWithPosition {
            position,
            model: dto,
        };

        Ok(result)
    }

    async fn cache_dto(&self, stream: &Stream, group_name: &str) -> Result<(), EventSourceError> {
        create_subscription(self.event_db(), stream, group_name)
            .await
            .map_err(EventSourceError::EventStore)?;

        let options = SubscribeToPersistentSubscriptionOptions::default().buffer_size(1);

        let mut sub = self
            .event_db()
            .subscribe_to_persistent_subscription(stream.to_string(), group_name, &options)
            .await
            .map_err(EventSourceError::EventStore)?;

        loop {
            let rcv_event = sub.next().await.map_err(EventSourceError::EventStore)?;

            let Some(event) = rcv_event.event.as_ref() else {
                continue;
            };

            let model_key: ModelKey = event
                .stream_id()
                .try_into()
                .map_err(EventSourceError::ModelKey)?;

            let mut model = self
                .cache_db()
                .get(self.repository_kind().to_cache_prefix(), &model_key)
                .map_err(EventSourceError::CacheDbError)?;

            let ordering = if event.revision == 0 {
                if model.position.is_some() {
                    Ordering::Greater
                } else {
                    Ordering::Equal
                }
            } else {
                model
                    .position
                    .map_or(Ordering::Less, |pos| pos.cmp(&(event.revision)))
            };

            match ordering {
                Ordering::Less => {
                    model = self.complete_from_es(&model_key, &model).await?;

                    self.cache_db()
                        .set(&model_key, model, self.repository_kind().to_cache_prefix())
                        .map_err(EventSourceError::CacheDbError)?;
                }
                Ordering::Equal | Ordering::Greater => {}
            }

            sub.ack(&rcv_event)
                .await
                .map_err(EventSourceError::EventStore)?;
        }
    }

    /// # Errors
    ///
    /// Will return `Err` if input is not in the format `index@stream_id`
    fn split_event_id(str: &str) -> Result<(&str, &str), EventSourceError> {
        let mut iter = str.split('@');

        if let (Some(index), Some(stream_id)) = (iter.next(), iter.next()) {
            return Ok((index, stream_id));
        }

        Err(EventSourceError::Position(format!(
            "{str} isnt in the format index@stream_id"
        )))
    }
}

impl<D, C> DtoRepositoryConstructor<D, C> for DtoRepository<D, C>
where
    D: Dto,
    C: CacheDb<D>,
{
    fn new(event_db: EventDb, cache_db: C, repository_kind: RepositoryKind) -> Self {
        Self {
            event_db,
            cache_db,
            repository_kind,
            dto: PhantomData,
        }
    }
}

impl<D, C> Repository<D, C> for DtoRepository<D, C>
where
    D: Dto,
    C: CacheDb<D>,
{
    fn event_db(&self) -> &EventDb {
        &self.event_db
    }
    fn cache_db(&self) -> &C {
        &self.cache_db
    }

    fn repository_kind(&self) -> &RepositoryKind {
        &self.repository_kind
    }
}

impl<S, C> StateRepositoryConstructor<S, C> for StateRepository<S, C>
where
    S: State,
    C: CacheDb<S>,
{
    fn new(event_db: EventDb, state_db: C) -> Self {
        Self {
            event_db,
            state_db,
            repository_kind: RepositoryKind::State,
            state: PhantomData,
        }
    }
}

impl<S, C> Repository<S, C> for StateRepository<S, C>
where
    S: State,
    C: CacheDb<S>,
{
    fn event_db(&self) -> &EventDb {
        &self.event_db
    }
    fn cache_db(&self) -> &C {
        &self.state_db
    }

    fn repository_kind(&self) -> &RepositoryKind {
        &self.repository_kind
    }
}

/// Appending event can resolve with multiple correct behavior
#[derive(Eq, PartialEq)]
pub enum AddedEvent {
    /// Successfully added, nothing more to do
    Success,

    /// The event wasn't added, the command need to be retried
    NeedRetry,
}

impl<S, C> StateRepository<S, C>
where
    S: State,
    C: CacheDb<S>,
{
    /// # Errors
    ///
    /// Will return `Err` if events cannot be added to the eventstore
    pub async fn add_command(
        &self,
        key: &ModelKey,
        command: S::Command,
        previous_metadata: Option<&Metadata>,
    ) -> Result<S, EventSourceStateError>
    where
        S: State,
    {
        let mut model: S;
        let events: Vec<S::Event>;

        loop {
            let (l_model, l_events, retry) = self
                .try_append(key, command.clone(), previous_metadata)
                .await?;
            if retry == AddedEvent::NeedRetry {
                continue;
            }

            model = l_model;
            events = l_events;

            break;
        }

        for event in &events {
            model.play_event(event);
        }

        Ok(model)
    }

    async fn try_append(
        &self,
        key: &ModelKey,
        command: S::Command,
        previous_metadata: Option<&Metadata>,
    ) -> Result<(S, Vec<S::Event>, AddedEvent), EventSourceStateError>
    where
        S: State + Sync,
    {
        let model: ModelWithPosition<S> = self
            .get_model(key)
            .await
            .map_err(EventSourceStateError::EventSourceError)?;

        let state = model.model;

        let events = state
            .try_command(command.clone())
            .map_err(|e| EventSourceStateError::State(format!("{e}")))?;

        let options = model.position.map_or_else(
            || AppendToStreamOptions::default().stream_state(StreamState::NoStream),
            |position| {
                AppendToStreamOptions::default().stream_state(StreamState::StreamRevision(position))
            },
        );

        let command_metadata = CompleteEvent::from_command(&command, previous_metadata)
            .map_err(|e| EventSourceStateError::EventSourceError(EventSourceError::Serde(e)))?;

        let mut events_data = vec![command_metadata.clone()];

        let mut previous_metadata = command_metadata.metadata().to_owned();

        let res_events = events.clone();

        for event in events {
            let event_metadata = CompleteEvent::from_event(&event, &previous_metadata)
                .map_err(|e| EventSourceStateError::EventSourceError(EventSourceError::Serde(e)))?;

            events_data.push(event_metadata.clone());
            event_metadata.metadata().clone_into(&mut previous_metadata);
        }

        let retry = self
            .try_append_event_data(key, &options, events_data)
            .await?;

        Ok((state, res_events, retry))
    }

    async fn try_append_event_data(
        &self,
        key: &ModelKey,
        options: &AppendToStreamOptions,
        events_with_data: Vec<CompleteEvent>,
    ) -> Result<AddedEvent, EventSourceStateError>
    where
        S: State,
    {
        let mut err = Ok(());
        let events: Vec<EventData> = events_with_data
            .into_iter()
            .filter_map(|e| match e.full_event_data() {
                Ok(event) => Some(event),
                Err(e) => {
                    err = Err(EventSourceError::Serde(e));
                    None
                }
            })
            .collect();
        err.map_err(EventSourceStateError::EventSourceError)?;

        let appended = self
            .event_db
            .append_to_stream(key.format(), options, events)
            .await;

        match appended {
            Ok(_) => Ok(AddedEvent::Success),
            Err(Error::WrongExpectedVersion { .. }) => Ok(AddedEvent::NeedRetry),
            Err(e) => Err(EventSourceStateError::EventSourceError(
                EventSourceError::EventStore(e),
            )),
        }
    }
}