Transactions

forgen-org · Feb 10, 2024 · 8bf3b34 · 8bf3b34
1 parent e1d3f25
commit 8bf3b34
Show file tree

Hide file tree

Showing 21 changed files with 186 additions and 156 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/auth/Cargo.toml b/auth/Cargo.toml
@@ -18,7 +18,6 @@ reqwest = { version = "0.11", features = ["json"],  optional = true }
 serde = "1.0.196"
 serde_urlencoded = { version = "0.7.1", optional = true }
 sha2 = "0.10.8"
-tokio-retry = { version = "0.3", optional = true }
 urlencoding = { version = "2.1.3", optional = true }
 uuid = { version = "1.7.0", features = ["serde", "v4"] }
 wasm-bindgen = { version = "0.2.90", optional = true }
@@ -32,7 +31,7 @@ tokio = { version = "1.0", features = ["full"] }
 
 [features]
 default = ["jwt", "linkedin"]
-axum = ["dep:axum", "dep:tokio-retry"]
+axum = ["dep:axum"]
 jwt = ["dep:hmac", "dep:jwt"]
 mongodb = ["dep:mongodb"]
 linkedin = ["dep:reqwest", "dep:serde_urlencoded"]

diff --git a/auth/src/application/command.rs b/auth/src/application/command.rs
@@ -1,5 +1,4 @@
-// use super::event_bus::EventBus;
-use crate::application::port::*;
+use super::port::*;
 use crate::domain;
 use crate::domain::scalar::*;
 use framework::*;
@@ -17,7 +16,7 @@ where
 {
     type Error = CommandError;
 
-    async fn execute(&self, runtime: &R) -> Result<(), CommandError> {
+    async fn execute(&self, runtime: &R) -> Result<TransactionId, Self::Error> {
         let user_id = EventStore::identify_by_email(runtime, &self.email).await?;
 
         let existing_events = match user_id {
@@ -31,9 +30,7 @@ where
         })?;
 
         EventStore::push(runtime, &new_events).await?;
-        EventBus::publish(runtime, new_events);
-
-        Ok(())
+        Ok(EventBus::publish(runtime, new_events))
     }
 }
 
@@ -50,7 +47,7 @@ where
 {
     type Error = CommandError;
 
-    async fn execute(&self, runtime: &R) -> Result<(), CommandError> {
+    async fn execute(&self, runtime: &R) -> Result<TransactionId, Self::Error> {
         let user_id = EventStore::identify_by_email(runtime, &self.email).await?;
 
         let existing_events = match user_id {
@@ -64,9 +61,7 @@ where
         })?;
 
         EventStore::push(runtime, &new_events).await?;
-        EventBus::publish(runtime, new_events);
-
-        Ok(())
+        Ok(EventBus::publish(runtime, new_events))
     }
 }
 

diff --git a/auth/src/application/listener.rs b/auth/src/application/listener.rs
@@ -1,32 +1,28 @@
-use self::event::Event;
-
 use super::port::*;
 use super::projection::User;
+use super::transaction::Transaction;
 use crate::domain::scalar::*;
-use crate::*;
+use crate::domain::Event;
 use framework::*;
 use futures::stream::StreamExt;
 use std::collections::HashMap;
 
 pub struct RecomputeUserProjections(pub Vec<Event>);
 
-#[async_trait]
-impl<R> Execute<R> for RecomputeUserProjections
-where
-    R: EventStore + UserRepository,
-    R: Send + Sync,
-{
-    type Error = UnexpectedError;
-
-    async fn execute(&self, runtime: &R) -> Result<(), UnexpectedError> {
+impl RecomputeUserProjections {
+    async fn reflect<R>(&self, runtime: &R) -> Result<Vec<UserId>, UnexpectedError>
+    where
+        R: EventStore + TransactionBus + UserRepository,
+        R: Send + Sync,
+    {
         // Caching projections
         let mut users = HashMap::<UserId, User>::new();
 
         // Applying events
         for event in self.0.iter() {
             let user_id = match event {
-                domain::Event::Registered { user_id, .. } => user_id.clone(),
-                domain::Event::LoggedIn { user_id, .. } => user_id.clone(),
+                Event::Registered { user_id, .. } => user_id.clone(),
+                Event::LoggedIn { user_id, .. } => user_id.clone(),
                 _ => continue,
             };
 
@@ -46,28 +42,41 @@ where
                 },
             };
 
-            user.apply(event);
+            user.apply(&event);
         }
 
         // Save projections
         for user in users.values() {
             UserRepository::save(runtime, user).await?;
         }
 
-        Ok(())
+        Ok(users.keys().cloned().collect())
     }
 }
 
 #[async_trait]
 impl<R> Listen<R> for RecomputeUserProjections
 where
-    R: EventBus + EventStore + UserRepository,
+    R: EventBus + EventStore + TransactionBus + UserRepository,
     R: Send + Sync,
 {
     async fn listen(runtime: &R) {
-        while let Some(events) = EventBus::subscribe(runtime).next().await {
-            if let Err(err) = RecomputeUserProjections(events).execute(runtime).await {
-                error!("Failed to recompute user projections: {}", err);
+        while let Some((id, events)) = EventBus::subscribe(runtime).next().await {
+            match RecomputeUserProjections(events).reflect(runtime).await {
+                Ok(user_ids) => {
+                    for user_id in user_ids {
+                        TransactionBus::publish(
+                            runtime,
+                            Transaction::UserProjected {
+                                id: id.clone(),
+                                user_id,
+                            },
+                        );
+                    }
+                }
+                Err(err) => {
+                    error!("Failed to recompute user projections: {}", err);
+                }
             }
         }
     }

diff --git a/auth/src/application/mod.rs b/auth/src/application/mod.rs
@@ -3,6 +3,7 @@ pub mod listener;
 pub mod port;
 pub mod projection;
 pub mod query;
+pub mod transaction;
 
 pub use crate::domain::event;
 pub use crate::domain::scalar;

diff --git a/auth/src/application/port.rs b/auth/src/application/port.rs
@@ -1,32 +1,38 @@
-use std::pin::Pin;
-
 use super::projection::User;
+use super::transaction::Transaction;
+use crate::domain;
 use crate::domain::scalar::*;
 use crate::domain::Event;
 use framework::*;
-use futures::stream::Stream;
-
-pub type EventStream = Pin<Box<dyn Stream<Item = Vec<Event>> + Send>>;
 
 #[async_trait]
 #[delegate]
 pub trait EventBus {
-    fn publish(&self, events: Vec<Event>);
-    fn subscribe(&self) -> EventStream;
+    fn publish(&self, events: Vec<Event>) -> TransactionId;
+    fn subscribe(&self) -> EventStream<Event>;
+}
+
+#[async_trait]
+#[delegate]
+pub trait TransactionBus {
+    fn publish(&self, transaction: Transaction);
+    fn subscribe(&self) -> TransactionStream<Transaction>;
 }
 
 #[async_trait]
 #[delegate]
 pub trait EventStore {
     async fn identify_by_email(&self, email: &Email) -> Result<Option<UserId>, UnexpectedError>;
-    async fn pull_by_user_id(&self, user_id: &UserId) -> Result<Vec<Event>, UnexpectedError>;
-    async fn push(&self, events: &[Event]) -> Result<(), UnexpectedError>;
+    async fn pull_by_user_id(
+        &self,
+        user_id: &UserId,
+    ) -> Result<Vec<domain::Event>, UnexpectedError>;
+    async fn push(&self, events: &[domain::Event]) -> Result<(), UnexpectedError>;
 }
 
 #[async_trait]
 #[delegate]
 pub trait UserRepository {
-    async fn find_by_email(&self, email: &Email) -> Result<Option<User>, UnexpectedError>;
     async fn find_by_user_id(&self, user_id: &UserId) -> Result<Option<User>, UnexpectedError>;
     async fn save(&self, projection: &User) -> Result<(), UnexpectedError>;
 }

diff --git a/auth/src/application/query.rs b/auth/src/application/query.rs
@@ -2,12 +2,12 @@ use super::port::*;
 use crate::domain::scalar::*;
 use framework::*;
 
-pub struct GetJwtByEmail {
-    pub email: Email,
+pub struct GetJwtByUserId {
+    pub user_id: UserId,
 }
 
 #[async_trait]
-impl<R> Fetch<R> for GetJwtByEmail
+impl<R> Fetch<R> for GetJwtByUserId
 where
     R: JwtPort + UserRepository,
     R: Send + Sync,
@@ -16,7 +16,7 @@ where
     type Error = GetJwtByEmailError;
 
     async fn fetch(&self, runtime: &R) -> Result<Jwt, GetJwtByEmailError> {
-        let user = UserRepository::find_by_email(runtime, &self.email)
+        let user = UserRepository::find_by_user_id(runtime, &self.user_id)
             .await?
             .ok_or(GetJwtByEmailError::UserNotFound)?;
         let jwt = JwtPort::sign(runtime, &user)?;

diff --git a/auth/src/application/test_runtime.rs b/auth/src/application/test_runtime.rs
@@ -24,13 +24,6 @@ impl EventStore for TestRuntime {
 
 #[async_trait]
 impl UserRepository for TestRuntime {
-    async fn find_by_email(&self, email: &Email) -> Result<Option<User>, UnexpectedError> {
-        Ok(Some(User {
-            email: email.to_string(),
-            user_id: UserId::default().to_string(),
-        }))
-    }
-
     async fn find_by_user_id(&self, user_id: &UserId) -> Result<Option<User>, UnexpectedError> {
         Ok(Some(User {
             email: "[email protected]".to_string(),
@@ -54,5 +47,3 @@ impl JwtPort for TestRuntime {
         })
     }
 }
-
-impl Framework for TestRuntime {}
diff --git a/auth/src/application/transaction.rs b/auth/src/application/transaction.rs
@@ -0,0 +1,7 @@
+use crate::domain::scalar::UserId;
+use framework::TransactionId;
+
+#[derive(Clone, PartialEq)]
+pub enum Transaction {
+    UserProjected { id: TransactionId, user_id: UserId },
+}
diff --git a/auth/src/clients/axum/api/login.rs b/auth/src/clients/axum/api/login.rs
@@ -1,8 +1,11 @@
-use crate::application::{
-    command::Login,
-    port::*,
-    query::GetJwtByEmail,
-    scalar::{Email, Password},
+use crate::{
+    application::{
+        command::Login,
+        port::*,
+        query::GetJwtByUserId,
+        scalar::{Email, Password},
+    },
+    transaction::Transaction,
 };
 use axum::{
     extract::State,
@@ -11,6 +14,7 @@ use axum::{
     Json,
 };
 use framework::*;
+use futures::StreamExt;
 use serde::Deserialize;
 use std::sync::Arc;
 
@@ -19,19 +23,29 @@ pub async fn handler<R>(
     Json(payload): Json<Payload>,
 ) -> Result<String, Response>
 where
-    R: Framework,
-    R: EventBus + EventStore + JwtPort + UserRepository,
+    R: EventBus + EventStore + JwtPort + UserRepository + TransactionBus,
+    R: Send + Sync,
 {
     let command = Login::try_from(payload)?;
-    let email = command.email.clone();
 
-    runtime
-        .execute(command)
+    let transaction_id = command
+        .execute(runtime.as_ref())
         .await
         .map_err(|err| (StatusCode::UNAUTHORIZED, format!("{}", err)).into_response())?;
 
-    let jwt = runtime
-        .fetch(GetJwtByEmail { email })
+    let user_id = loop {
+        if let Some(transaction) = TransactionBus::subscribe(runtime.as_ref()).next().await {
+            let Transaction::UserProjected { id, user_id } = transaction;
+            if id == transaction_id {
+                break user_id;
+            }
+        }
+    };
+
+    let query = GetJwtByUserId { user_id };
+
+    let jwt = query
+        .fetch(runtime.as_ref())
         .await
         .map_err(|err| (StatusCode::INTERNAL_SERVER_ERROR, format!("{}", err)).into_response())?;