fix query

Author: luca-iachini

.env

@@ -1 +1 @@
-DATABASE_URL=postgres://postgres:postgres@localhost:5432/event_store
+DATABASE_URL=postgres://postgres:postgres@localhost:5432/postgres

disintegrate-postgres/src/event_store.rs

@@ -106,12 +106,9 @@ where
         <QE as TryFrom<E>>::Error: StdError + Send + Sync + 'static,
     {
         let sql = format!(
-            r#"
-            WITH epoch AS (
-             SELECT MAX(event_id) as __epoch FROM event
-            )
-            SELECT event_id, payload, __epoch 
-            FROM event CROSS JOIN epoch WHERE ({criteria}) AND event_id <= __epoch
+            r#"SELECT event.event_id, event.payload, epoch.__epoch_id
+            FROM (SELECT MAX(event_id) AS __epoch_id FROM event) AS epoch
+            LEFT JOIN event ON event.event_id <= epoch.__epoch_id AND ({criteria})
             ORDER BY event_id ASC"#,
             criteria = CriteriaBuilder::new(query).build()
         );
@@ -121,13 +118,15 @@ where
             let mut epoch_id: PgEventId = 0;
             while let Some(row) = rows.next().await {
                 let row = row?;
-                let event_id = row.get(0);
+                let event_id: Option<i64> = row.get(0);
                 epoch_id = row.get(2);
-                let payload = self.serde.deserialize(row.get(1))?;
-                let payload: QE = payload
+                if let Some(event_id) = event_id {
+                    let payload = self.serde.deserialize(row.get(1))?;
+                    let payload: QE = payload
                         .try_into()
                         .map_err(|e| Error::QueryEventMapping(Box::new(e)))?;
-                yield Ok(StreamItem::Event(PersistedEvent::new(event_id, payload)));
+                    yield Ok(StreamItem::Event(PersistedEvent::new(event_id, payload)));
+                }
             }
             yield Ok(StreamItem::End(epoch_id));
         }
@@ -176,20 +175,18 @@ where
 
     /// Appends new events to the event store.
     ///
-    /// This function inserts the provided `events` into the PostgreSQL event store by performing
-    /// two separate inserts. First, it inserts the events into the `event_sequence` table to reclaim
-    /// a set of IDs for the events. Then, it inserts the events into the `event` table along with
-    /// their IDs, event types, domain identifiers, and payloads. Finally, it marks the event IDs as `consumed`
-    /// in the event sequence table. If marking the event IDs as consumed fails (e.g., another process has already consumed the IDs),
-    /// a conflict error is raised. This conflict indicates that the data retrieved by the query is stale,
-    /// meaning that the events generated are no longer valid due to being generated from an old version
-    /// of the event store.
+    /// This function inserts the provided `events` into the PostgreSQL-backed event store.
+    /// Before inserting, it queries the `event` table to ensure that no events have been
+    /// appended since the given `version`. If newer events are found, a concurrency error
+    /// is returned to prevent invalid state transitions.
+    ///
+    /// If the concurrency check succeeds, the events are inserted into the `event` table.
     ///
     /// # Arguments
     ///
-    /// * `events` - A vector of events to be appended.
-    /// * `query` - The stream query specifying the criteria for filtering events.
-    /// * `version` - The ID of the last consumed event.
+    /// * `events` - The events to append to the event store.
+    /// * `query` - The stream query that identifies the target event stream.
+    /// * `version` - The ID of the last consumed event, used for optimistic concurrency control.
     ///
     /// # Returns
     ///
@@ -206,13 +203,16 @@ where
         QE: Event + Clone + Send + Sync,
     {
         let mut tx = self.pool.begin().await?;
-        sqlx::query("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE").execute(&mut *tx).await?;
-        let inv_events: i64 = sqlx::query_scalar(&format!("SELECT count(*) FROM event WHERE {}", CriteriaBuilder::new(&query.change_origin(version)).build()))
-            .fetch_one(&mut *tx)
-            .await
-            .map_err(map_concurrency_err)?;
+        sqlx::query("SET TRANSACTION ISOLATION LEVEL SERIALIZABLE")
+            .execute(&mut *tx)
+            .await?;
 
-        if inv_events > 0 {
+        if sqlx::query_scalar(&format!(
+            "SELECT EXISTS (SELECT 1 FROM event WHERE {})",
+            CriteriaBuilder::new(&query.change_origin(version)).build()
+        ))
+        .fetch_one(&mut *tx)
+        .await? {
             return Err(Error::Concurrency);
         }
 
@@ -231,16 +231,55 @@ where
             .map(|(event_id, event)| PersistedEvent::new(*event_id, event))
             .collect::<Vec<_>>();
 
-        tx.commit().await?;
+        tx.commit().await.map_err(map_concurrency_err)?;
 
         Ok(persisted_events)
     }
 
+    /// Appends a batch of events to the PostgreSQL-backed event store **without** verifying
+    /// whether new events have been added since the last read.
+    ///
+    /// # Arguments
+    ///
+    /// * `events` - A vector of events to be appended.
+    ///
+    /// # Returns
+    ///
+    /// A `Result` containing a vector of `PersistedEvent` representing the appended events,
+    /// or an error of type `Self::Error`.
+    async fn append_without_validation(
+        &self,
+        events: Vec<E>,
+    ) -> Result<Vec<PersistedEvent<PgEventId, E>>, Self::Error>
+    where
+        E: Clone + 'async_trait,
+    {
+        let mut tx = self.pool.begin().await?;
+
+        let mut sequence_insert = InsertEventsBuilder::new(&events, &self.serde);
+        let event_ids: Vec<PgEventId> = sequence_insert
+            .build()
+            .fetch_all(&mut *tx)
+            .await?
+            .into_iter()
+            .map(|r| r.get(0))
+            .collect();
+
+        let persisted_events = event_ids
+            .iter()
+            .zip(events)
+            .map(|(event_id, event)| PersistedEvent::new(*event_id, event))
+            .collect::<Vec<_>>();
+
+        tx.commit().await?;
+
+        Ok(persisted_events)
+    }
 }
 
 fn map_concurrency_err(err: sqlx::Error) -> Error {
     if let sqlx::Error::Database(ref description) = err {
-        if description.code().as_deref() == Some("23514") {
+        if description.code().as_deref() == Some("40001") {
             return Error::Concurrency;
         }
     }

disintegrate-postgres/src/event_store/append.rs

@@ -1,6 +1,6 @@
 use std::collections::BTreeSet;
 
-use disintegrate::{Event, Identifier, PersistedEvent};
+use disintegrate::{Event, Identifier};
 use disintegrate_serde::Serde;
 use sqlx::postgres::PgArguments;
 use sqlx::query::Query;
@@ -83,4 +83,95 @@ where
 
 #[cfg(test)]
 mod tests {
+    use disintegrate::{
+        domain_identifiers, ident, DomainIdentifierInfo, DomainIdentifierSet, EventInfo,
+        EventSchema, IdentifierType,
+    };
+    use serde::{Deserialize, Serialize};
+    use sqlx::Execute;
+
+    use super::*;
+
+    #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+    #[serde(tag = "event_type", rename_all = "snake_case")]
+    enum ShoppingCartEvent {
+        Added {
+            product_id: String,
+            cart_id: String,
+            quantity: i64,
+        },
+        Removed {
+            product_id: String,
+            cart_id: String,
+            quantity: i64,
+        },
+    }
+
+    impl Event for ShoppingCartEvent {
+        const SCHEMA: EventSchema = EventSchema {
+            events: &["ShoppingCartAdded", "ShoppingCartRemoved"],
+            events_info: &[
+                &EventInfo {
+                    name: "ShoppingCartAdded",
+                    domain_identifiers: &[&ident!(#product_id), &ident!(#cart_id)],
+                },
+                &EventInfo {
+                    name: "ShoppingCartRemoved",
+                    domain_identifiers: &[&ident!(#product_id), &ident!(#cart_id)],
+                },
+            ],
+            domain_identifiers: &[
+                &DomainIdentifierInfo {
+                    ident: ident!(#cart_id),
+                    type_info: IdentifierType::String,
+                },
+                &DomainIdentifierInfo {
+                    ident: ident!(#product_id),
+                    type_info: IdentifierType::String,
+                },
+            ],
+        };
+        fn name(&self) -> &'static str {
+            match self {
+                ShoppingCartEvent::Added { .. } => "ShoppingCartAdded",
+                ShoppingCartEvent::Removed { .. } => "ShoppingCartRemoved",
+            }
+        }
+        fn domain_identifiers(&self) -> DomainIdentifierSet {
+            match self {
+                ShoppingCartEvent::Added {
+                    product_id,
+                    cart_id,
+                    ..
+                } => domain_identifiers! {product_id: product_id, cart_id: cart_id},
+                ShoppingCartEvent::Removed {
+                    product_id,
+                    cart_id,
+                    ..
+                } => domain_identifiers! {product_id: product_id, cart_id: cart_id},
+            }
+        }
+    }
+
+    #[test]
+    fn it_builds_event_insert() {
+        let events = &[
+            ShoppingCartEvent::Added {
+                product_id: "product_1".into(),
+                cart_id: "cart_1".into(),
+                quantity: 10,
+            },
+            ShoppingCartEvent::Removed {
+                product_id: "product_1".into(),
+                cart_id: "cart_1".into(),
+                quantity: 10,
+            },
+        ];
+        let serde = disintegrate_serde::serde::json::Json::default();
+        let mut insert_query = InsertEventsBuilder::new(events, &serde);
+        assert_eq!(
+            insert_query.build().sql(),
+            "INSERT INTO event (event_type,payload,cart_id,product_id) VALUES ($1, $2, $3, $4), ($5, $6, $7, $8) RETURNING (event_id)"
+        );
+    }
 }

disintegrate-postgres/src/event_store/tests.rs

@@ -1,8 +1,8 @@
-use super::append::{InsertEventSequenceBuilder, InsertEventsBuilder};
+use super::append::InsertEventsBuilder;
 use crate::{Error, PgEventId, PgEventStore};
 use disintegrate::{
     domain_identifiers, ident, query, DomainIdentifierInfo, DomainIdentifierSet, Event, EventInfo,
-    EventSchema, EventStore, IdentifierType, PersistedEvent,
+    EventSchema, EventStore, IdentifierType,
 };
 use disintegrate_serde::serde::json::Json;
 use disintegrate_serde::Deserializer;
@@ -270,24 +270,8 @@ pub async fn insert_events<E: Event + Clone + Serialize + DeserializeOwned>(
     pool: &PgPool,
     events: &[E],
 ) {
-    let mut event_sequence_insert = InsertEventSequenceBuilder::new(events)
-        .with_consumed(true)
-        .with_committed(true);
-    let event_ids: Vec<PgEventId> = event_sequence_insert
-        .build()
-        .fetch_all(pool)
-        .await
-        .unwrap()
-        .into_iter()
-        .map(|r| r.get(0))
-        .collect();
-    let persisted_events = event_ids
-        .into_iter()
-        .zip(events.iter())
-        .map(|(id, event)| PersistedEvent::new(id, event.clone()))
-        .collect::<Vec<_>>();
     let serde = disintegrate_serde::serde::json::Json::default();
-    InsertEventsBuilder::new(persisted_events.as_slice(), &serde)
+    InsertEventsBuilder::new(events, &serde)
         .build()
         .execute(pool)
         .await

disintegrate-postgres/src/listener.rs

@@ -539,7 +539,6 @@ where
         };
 
         let result = self.handle_events_from(last_processed_id, &mut tx).await;
-
         self.release_listener(result, tx).await
     }
 

disintegrate-postgres/src/listener/tests.rs

@@ -216,6 +216,7 @@ async fn it_runs_event_listeners(pool: PgPool) {
 
     assert!(append_result.is_ok());
     let carts = Cart::carts(&pool).await.unwrap();
+    dbg!(&carts);
     assert_eq!(carts.len(), 1);
     let first_row = carts.first().unwrap();
     assert_eq!("cart_1", &first_row.cart_id);

disintegrate-postgres/src/migrator.rs

@@ -43,12 +43,7 @@ where
 
     /// Init `PgEventStore` database
     pub async fn init_event_store(&self) -> Result<(), Error> {
-        const RESERVED_NAMES: &[&str] = &[
-            "event_id",
-            "payload",
-            "event_type",
-            "inserted_at",
-        ];
+        const RESERVED_NAMES: &[&str] = &["event_id", "payload", "event_type", "inserted_at"];
 
         sqlx::query(include_str!("event_store/sql/table_event.sql"))
             .execute(&self.event_store.pool)

disintegrate/src/event_store.rs

@@ -69,6 +69,29 @@ where
     where
         E: Clone + 'async_trait,
         QE: Event + 'static + Clone + Send + Sync;
+
+    /// Appends a batch of events to the event store **without** verifying if
+    /// new events have been added since the last read.
+    ///
+    /// This method is useful when you are certain that no other process
+    /// has modified the event store in a way that would make your logic stale.
+    ///
+    /// If you need to guarantee that no duplicate events are added,
+    /// use the `append` method instead, providing a query that ensures uniqueness.
+    ///
+    /// # Arguments
+    ///
+    /// * `events` - A vector of events to append to the event store.
+    ///
+    ///# Returns
+    ///
+    /// A `Result` containing a vector of `PersistedEvent` representing the appended events, or an error.
+    async fn append_without_validation(
+        &self,
+        events: Vec<E>,
+    ) -> Result<Vec<PersistedEvent<ID, E>>, Self::Error>
+    where
+        E: Clone + 'async_trait;
 }
 
 /// An item in the event stream.