feat: add Subscription trait, Transient and Persisted variants (#101)

* feat: add Subscription trait and Transient subscription

* refactor(inmemory): Projector uses a Subscription impl

* feat(postgres): add Persistent subscription implementation

* fix: clippy complaints

Author: ar3s3ru

eventually-core/Cargo.toml

@@ -13,6 +13,7 @@ default = []
 full = ["serde"]
 
 [dependencies]
+anyhow = "1.0"
 futures = { version = "0.3", features = ["async-await"] }
 serde = { version = "1.0", features = ["derive"], optional = true }
 thiserror = "1.0"

eventually-core/src/subscription.rs

@@ -3,10 +3,14 @@
 //!
 //! [`EventStore`]: ../store/trait.EventStore.html
 
-use futures::future::BoxFuture;
-use futures::stream::BoxStream;
+use std::error::Error as StdError;
+use std::sync::atomic::{AtomicU32, Ordering};
+use std::sync::Arc;
 
-use crate::store::Persisted;
+use futures::future::{ok, BoxFuture, FutureExt};
+use futures::stream::{BoxStream, StreamExt, TryStreamExt};
+
+use crate::store::{EventStore, Persisted, Select};
 
 /// Stream of events returned by the [`EventSubscriber::subscribe_all`] method.
 ///
@@ -63,3 +67,184 @@ pub trait EventSubscriber {
     /// [`EventStream`]: type.EventStream.html
     fn subscribe_all(&self) -> BoxFuture<Result<EventStream<Self>, Self::Error>>;
 }
+
+/// Stream of events returned by the [`Subscription::resume`] method.
+///
+/// [`Subscription::resume`]: trait.Subscription.html#method.resume
+pub type SubscriptionStream<'a, S> = BoxStream<
+    'a,
+    Result<
+        Persisted<<S as Subscription>::SourceId, <S as Subscription>::Event>,
+        <S as Subscription>::Error,
+    >,
+>;
+
+/// A Subscription to an [`EventStream`] which can be "checkpointed":
+/// keeps a record of the latest message processed by itself using [`checkpoint`],
+/// and can resume working from such message by using the [`resume`].
+///
+/// [`resume`]: trait.Subscription.html#method.resume
+/// [`checkpoint`]: trait.Subscription.html#method.checkpoint
+pub trait Subscription {
+    /// Type of the Source id, typically an [`AggregateId`].
+    ///
+    /// [`AggregateId`]: ../aggregate/type.AggregateId.html
+    type SourceId: Eq;
+
+    /// Event type stored in the [`EventStore`], typically an [`Aggregate::Event`].
+    ///
+    /// [`Aggregate::Event`]: ../aggregate/trait.Aggregate.html#associatedtype.Event
+    /// [`EventStore`]: ../store/trait.EventStore.html
+    type Event;
+
+    /// Possible errors returned when receiving events from the notification channel.
+    type Error;
+
+    /// Resumes the current state of a `Subscription` by returning the [`EventStream`],
+    /// starting from the last event processed by the `Subscription`.
+    ///
+    /// [`EventStream`]: trait.EventStream.html
+    fn resume(&self) -> BoxFuture<Result<SubscriptionStream<Self>, Self::Error>>;
+
+    /// Saves the provided version (or sequence number) as the latest
+    /// version processed.
+    fn checkpoint(&self, version: u32) -> BoxFuture<Result<(), Self::Error>>;
+}
+
+/// Error type returned by a [`Transient`] Subscription.
+///
+/// [`Transient`]: struct.Transient.html
+#[derive(Debug, thiserror::Error)]
+pub enum Error {
+    /// Error caused by the Subscription's [`EventStore`].
+    ///
+    /// [`EventStore`]: ../store/trait.EventStore.html
+    #[error("error received while listening to the event stream from the store: {0}")]
+    Store(#[source] anyhow::Error),
+
+    /// Error caused by the Subscription's [`EventSubscriber`].
+    ///
+    /// [`EventSubscriber`]: trait.EventSubscriber.html
+    #[error("error received while listening to the event stream subscription: {0}")]
+    Subscription(#[source] anyhow::Error),
+}
+
+/// [`Subscription`] type which gets deleted once the process using it
+/// gets terminated.
+///
+/// Useful for in-memory or one-off [`Projection`]s.
+///
+/// [`Subscription`]: trait.Subscription.html
+/// [`Projection`]: ../projection/trait.Projection.html
+pub struct Transient<Store, Subscriber> {
+    store: Store,
+    subscriber: Subscriber,
+    last_sequence_number: Arc<AtomicU32>,
+}
+
+impl<Store, Subscriber> Transient<Store, Subscriber> {
+    /// Creates a new [`Subscription`] using the specified [`EventStore`]
+    /// and [`EventSubscriber`] to create the [`SubscriptionStream`] from.
+    ///
+    /// [`Subscription`]: trait.Subscription.html
+    /// [`EventStore`]: ../store/trait.EventStore.html
+    /// [`EventSubscriber`]: trait.EventSubscriber.html
+    /// [`SubscriptionStream`]: type.SubscriptionStream.html
+    pub fn new(store: Store, subscriber: Subscriber) -> Self {
+        Self {
+            store,
+            subscriber,
+            last_sequence_number: Default::default(),
+        }
+    }
+
+    /// Specifies the sequence number of the `Event` the [`SubscriptionStream`]
+    /// should start from when calling [`run`].
+    ///
+    /// [`SubscriptionStream`]: type.SubscriptionStream.html
+    /// [`run`]: struct.Transient.html#method.run
+    pub fn from(self, sequence_number: u32) -> Self {
+        self.last_sequence_number
+            .store(sequence_number, Ordering::Relaxed);
+
+        self
+    }
+}
+
+impl<Store, Subscriber> Subscription for Transient<Store, Subscriber>
+where
+    Store: EventStore + Send + Sync,
+    Subscriber: EventSubscriber<
+            SourceId = <Store as EventStore>::SourceId,
+            Event = <Store as EventStore>::Event,
+        > + Send
+        + Sync,
+    <Store as EventStore>::SourceId: Send + Sync,
+    <Store as EventStore>::Event: Send + Sync,
+    <Store as EventStore>::Error: StdError + Send + Sync + 'static,
+    <Subscriber as EventSubscriber>::Error: StdError + Send + Sync + 'static,
+{
+    type SourceId = Store::SourceId;
+    type Event = Store::Event;
+    type Error = Error;
+
+    fn resume(&self) -> BoxFuture<Result<SubscriptionStream<Self>, Self::Error>> {
+        Box::pin(async move {
+            // Create the Subscription first, so that once the future has been resolved
+            // we'll start receiving events right away.
+            //
+            // This is to avoid losing events when waiting for the one-off stream
+            // to resolve its future.
+            //
+            // The impact is that we _might_ get duplicated events from the one-off stream
+            // and the subscription stream. Luckily, we can discard those by
+            // keeping an internal state of the last processed sequence number,
+            // and discard all those events that are found.
+            let subscription = self
+                .subscriber
+                .subscribe_all()
+                .await
+                .map_err(anyhow::Error::from)
+                .map_err(Error::Store)?;
+
+            let one_off_stream = self
+                .store
+                .stream_all(Select::From(
+                    self.last_sequence_number.load(Ordering::Relaxed),
+                ))
+                .await
+                .map_err(anyhow::Error::from)
+                .map_err(Error::Subscription)?;
+
+            let stream = one_off_stream
+                .map_err(anyhow::Error::from)
+                .map_err(Error::Store)
+                .chain(
+                    subscription
+                        .map_err(anyhow::Error::from)
+                        .map_err(Error::Subscription),
+                )
+                .try_filter_map(move |event| async move {
+                    let expected_sequence_number =
+                        self.last_sequence_number.load(Ordering::Relaxed);
+
+                    let event_sequence_number = event.sequence_number();
+
+                    if event_sequence_number < expected_sequence_number {
+                        return Ok(None); // Duplicated event detected, let's skip it.
+                    }
+
+                    Ok(Some(event))
+                })
+                .boxed();
+
+            Ok(stream)
+        })
+    }
+
+    fn checkpoint(&self, version: u32) -> BoxFuture<Result<(), Self::Error>> {
+        // Checkpointing happens in memory on the atomic sequence number checkpoint.
+        self.last_sequence_number.store(version, Ordering::Relaxed);
+        ok(()).boxed()
+    }
+}

eventually-postgres/Cargo.toml

@@ -22,6 +22,7 @@ thiserror = "1.0"
 refinery = { version = "0.3.0", features = ["tokio-postgres"] }
 anyhow = "1.0.32"
 tokio = { version = "0.2", features = ["sync"] }
+tracing = "0.1"
 
 [dev-dependencies]
 testcontainers = "0.9"

eventually-postgres/src/lib.rs

@@ -53,8 +53,21 @@
 //! [`eventually`]: https://docs.rs/eventually
 //! [`EventStore`]: struct.EventStore.html
 
-mod store;
-mod subscriber;
+pub mod store;
+pub mod subscriber;
+pub mod subscription;
 
 pub use store::*;
 pub use subscriber::*;
+pub use subscription::*;
+
+use tokio_postgres::types::ToSql;
+
+/// Adapter type for parameters compatible with `tokio_postgres::Client` methods.
+pub(crate) type Params<'a> = &'a [&'a (dyn ToSql + Sync)];
+
+#[inline]
+#[allow(trivial_casts)]
+pub(crate) fn slice_iter<'a>(s: Params<'a>) -> impl ExactSizeIterator<Item = &'a dyn ToSql> + 'a {
+    s.iter().map(|s| *s as _)
+}

eventually-postgres/src/migrations/V4__subscriptions_table.sql

@@ -0,0 +1,10 @@
+CREATE TABLE subscriptions (
+    name                 TEXT        PRIMARY KEY,
+    aggregate_type_id    TEXT        NOT NULL,
+    last_sequence_number BIGINT      NOT NULL DEFAULT -1,
+    created_at           TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+    updated_at           TIMESTAMPTZ NOT NULL DEFAULT NOW(),
+
+    -- Remove all subscriptions in case the aggregate type is deleted.
+    FOREIGN KEY (aggregate_type_id) REFERENCES aggregate_types(id) ON DELETE CASCADE
+);

eventually-postgres/src/migrations/V5__get_or_create_subscription.sql

@@ -0,0 +1,14 @@
+CREATE OR REPLACE FUNCTION get_or_create_subscription(
+    name              TEXT,
+    aggregate_type_id TEXT
+)
+RETURNS subscriptions
+AS $$
+
+    INSERT INTO subscriptions (name, aggregate_type_id)
+    VALUES (name, aggregate_type_id)
+        ON CONFLICT (name)
+        DO UPDATE SET name=EXCLUDED.name -- Perform update to force row returning.
+    RETURNING *;
+
+$$ LANGUAGE SQL;

eventually-postgres/src/migrations/V6__checkpoint_subscription.sql

@@ -0,0 +1,25 @@
+CREATE OR REPLACE FUNCTION checkpoint_subscription(
+    subscription_name TEXT,
+    aggregate_type    TEXT,
+    sequence_number   BIGINT
+)
+RETURNS VOID
+AS $$
+DECLARE
+    old_sequence_number BIGINT;
+BEGIN
+
+    SELECT s.last_sequence_number INTO old_sequence_number
+    FROM subscriptions s
+    WHERE s."name" = subscription_name AND s.aggregate_type_id = aggregate_type;
+
+    IF old_sequence_number >= sequence_number THEN
+        RAISE EXCEPTION 'invalid sequence number provided: %, should be greater than %', sequence_number, old_sequence_number;
+    END IF;
+
+    UPDATE subscriptions s
+    SET last_sequence_number = sequence_number, updated_at = NOW()
+    WHERE s."name" = subscription_name AND s.aggregate_type_id = aggregate_type;
+
+END;
+$$ LANGUAGE PLPGSQL;

eventually-postgres/src/store.rs

@@ -10,9 +10,10 @@ use futures::stream::{StreamExt, TryStreamExt};
 
 use serde::{Deserialize, Serialize};
 
-use tokio_postgres::types::ToSql;
 use tokio_postgres::Client;
 
+use crate::{slice_iter, Params};
+
 /// Embedded migrations module.
 mod embedded {
     use refinery::embed_migrations;
@@ -171,8 +172,8 @@ impl EventStoreBuilderMigrated {
 /// [`EventStoreBuilder`]: ../../eventually_core/store/trait.EventStoreBuilder.html
 #[derive(Debug, Clone)]
 pub struct EventStore<Id, Event> {
+    pub(crate) type_name: &'static str,
     client: Arc<Client>,
-    type_name: &'static str,
     id: std::marker::PhantomData<Id>,
     payload: std::marker::PhantomData<Event>,
 }
@@ -322,11 +323,3 @@ where
             .boxed())
     }
 }
-
-type Params<'a> = &'a [&'a (dyn ToSql + Sync)];
-
-#[inline]
-#[allow(trivial_casts)]
-fn slice_iter<'a>(s: Params<'a>) -> impl ExactSizeIterator<Item = &'a dyn ToSql> + 'a {
-    s.iter().map(|s| *s as _)
-}