feat(client): add writing events to client adhering to the "Writing Events" compliance criteria (#7)

* feat(client): add writing events to client adhering to the "Writing Events" compliance criteria

* feat(client): rework write events based on forward looking learnings in #8

Signed-off-by: Raphael Höser <[email protected]>

* chore(client): restructure event module to avoid allowing clippy warnings

---------

Signed-off-by: Raphael Höser <[email protected]>

Author: Snapstromegon

Cargo.lock

@@ -19,6 +19,7 @@ testcontainers = { version = "0.24.0", features = [
   "http_wait",
 ], optional = true }
 thiserror = "2.0.12"
+typed-builder = "0.21.0"
 
 [dev-dependencies]
 eventsourcingdb-client-rust = { path = ".", features = ["testcontainer"] }

Cargo.toml

@@ -8,7 +8,7 @@ This is a work in progress and not yet ready for production use.
 Based on the [compliance criteria](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/) the SDK covers these criteria:
 
 - 🚀 [Essentials](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#essentials)
-- ❌ [Writing Events](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#writing-events)
+- 🚀 [Writing Events](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#writing-events)
 - ❌ [Reading Events](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#reading-events)
 - ❌ [Using EventQL](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#using-eventql)
 - ❌ [Observing Events](https://docs.eventsourcingdb.io/client-sdks/compliance-criteria/#observing-events)

README.md

@@ -18,13 +18,20 @@
 //! If this works, it means that the client is correctly configured and you can use it to make requests to the DB.
 
 mod client_request;
+mod precondition;
 
-use client_request::{ClientRequest, PingRequest, VerifyApiTokenRequest};
+use client_request::{
+    ClientRequest, OneShotRequest, PingRequest, VerifyApiTokenRequest, WriteEventsRequest,
+};
 
+pub use precondition::Precondition;
 use reqwest;
 use url::Url;
 
-use crate::error::ClientError;
+use crate::{
+    error::ClientError,
+    event::{Event, EventCandidate},
+};
 
 /// Client for an [EventsourcingDB](https://www.eventsourcingdb.io/) instance.
 #[derive(Debug)]
@@ -72,9 +79,14 @@ impl Client {
 
     /// Utility function to request an endpoint of the API.
     ///
+    /// This function will return a [`reqwest::RequestBuilder`] which can be used to send the request.
+    ///
     /// # Errors
     /// This function will return an error if the request fails or if the URL is invalid.
-    async fn request<R: ClientRequest>(&self, endpoint: R) -> Result<R::Response, ClientError> {
+    fn build_request<R: ClientRequest>(
+        &self,
+        endpoint: &R,
+    ) -> Result<reqwest::RequestBuilder, ClientError> {
         let url = self
             .base_url
             .join(endpoint.url_path())
@@ -93,15 +105,27 @@ impl Client {
         } else {
             request
         };
+        Ok(request)
+    }
 
-        let response = request.send().await?;
+    /// Utility function to request an endpoint of the API as a oneshot.
+    ///
+    /// This means, that the response is not streamed, but returned as a single value.
+    ///
+    /// # Errors
+    /// This function will return an error if the request fails or if the URL is invalid.
+    async fn request_oneshot<R: OneShotRequest>(
+        &self,
+        endpoint: R,
+    ) -> Result<R::Response, ClientError> {
+        let response = self.build_request(&endpoint)?.send().await?;
 
         if response.status().is_success() {
             let result = response.json().await?;
             endpoint.validate_response(&result)?;
             Ok(result)
         } else {
-            Err(ClientError::DBError(
+            Err(ClientError::DBApiError(
                 response.status(),
                 response.text().await.unwrap_or_default(),
             ))
@@ -125,7 +149,7 @@ impl Client {
     /// # Errors
     /// This function will return an error if the request fails or if the URL is invalid.
     pub async fn ping(&self) -> Result<(), ClientError> {
-        let _ = self.request(PingRequest).await?;
+        let _ = self.request_oneshot(PingRequest).await?;
         Ok(())
     }
 
@@ -146,7 +170,45 @@ impl Client {
     /// # Errors
     /// This function will return an error if the request fails or if the URL is invalid.
     pub async fn verify_api_token(&self) -> Result<(), ClientError> {
-        let _ = self.request(VerifyApiTokenRequest).await?;
+        let _ = self.request_oneshot(VerifyApiTokenRequest).await?;
         Ok(())
     }
+
+    /// Writes events to the DB instance.
+    ///
+    /// ```
+    /// use eventsourcingdb_client_rust::event::EventCandidate;
+    /// # use serde_json::json;
+    /// # tokio_test::block_on(async {
+    /// # let container = eventsourcingdb_client_rust::container::Container::start_default().await.unwrap();
+    /// let db_url = "http://localhost:3000/";
+    /// let api_token = "secrettoken";
+    /// # let db_url = container.get_base_url().await.unwrap();
+    /// # let api_token = container.get_api_token();
+    /// let client = eventsourcingdb_client_rust::client::Client::new(db_url, api_token);
+    /// let candidates = vec![
+    ///     EventCandidate::builder()
+    ///        .source("https://www.eventsourcingdb.io".to_string())
+    ///        .data(json!({"value": 1}))
+    ///        .subject("/test".to_string())
+    ///        .r#type("io.eventsourcingdb.test".to_string())
+    ///        .build()
+    /// ];
+    /// let written_events = client.write_events(candidates, vec![]).await.expect("Failed to write events");
+    /// # })
+    /// ```
+    ///
+    /// # Errors
+    /// This function will return an error if the request fails or if the URL is invalid.
+    pub async fn write_events(
+        &self,
+        events: Vec<EventCandidate>,
+        preconditions: Vec<Precondition>,
+    ) -> Result<Vec<Event>, ClientError> {
+        self.request_oneshot(WriteEventsRequest {
+            events,
+            preconditions,
+        })
+        .await
+    }
 }

src/client.rs

@@ -1,15 +1,21 @@
 //! This is a purely internal module to represent client requests to the database.
 
-use reqwest::Method;
-use serde_json::Value;
+mod ping;
+mod verify_api_token;
+mod write_events;
+
+pub use ping::PingRequest;
+pub use verify_api_token::VerifyApiTokenRequest;
+pub use write_events::WriteEventsRequest;
 
-use crate::{error::ClientError, event::ManagementEvent};
+use crate::error::ClientError;
+use reqwest::Method;
+use serde::{Serialize, de::DeserializeOwned};
 
 /// Represents a request to the database client
 pub trait ClientRequest {
     const URL_PATH: &'static str;
     const METHOD: Method;
-    type Response: serde::de::DeserializeOwned;
 
     /// Returns the URL path for the request
     fn url_path(&self) -> &'static str {
@@ -22,44 +28,17 @@ pub trait ClientRequest {
     }
 
     /// Returns the body for the request
-    fn body(&self) -> Option<Result<Value, ClientError>> {
-        None
+    fn body(&self) -> Option<Result<impl Serialize, ClientError>> {
+        None::<Result<(), _>>
     }
+}
+
+/// Represents a request to the database that expects a single response
+pub trait OneShotRequest: ClientRequest {
+    type Response: DeserializeOwned;
 
     /// Validate the response from the database
     fn validate_response(&self, _response: &Self::Response) -> Result<(), ClientError> {
         Ok(())
     }
 }
-
-/// Ping the Database instance
-#[derive(Debug, Clone, Copy)]
-pub struct PingRequest;
-
-impl ClientRequest for PingRequest {
-    const URL_PATH: &'static str = "/api/v1/ping";
-    const METHOD: Method = Method::GET;
-    type Response = ManagementEvent;
-
-    fn validate_response(&self, response: &Self::Response) -> Result<(), ClientError> {
-        (response.ty() == "io.eventsourcingdb.api.ping-received")
-            .then_some(())
-            .ok_or(ClientError::PingFailed)
-    }
-}
-
-/// Verify the API token
-#[derive(Debug, Clone, Copy)]
-pub struct VerifyApiTokenRequest;
-
-impl ClientRequest for VerifyApiTokenRequest {
-    const URL_PATH: &'static str = "/api/v1/verify-api-token";
-    const METHOD: Method = Method::POST;
-    type Response = ManagementEvent;
-
-    fn validate_response(&self, response: &Self::Response) -> Result<(), ClientError> {
-        (response.ty() == "io.eventsourcingdb.api.api-token-verified")
-            .then_some(())
-            .ok_or(ClientError::APITokenInvalid)
-    }
-}

src/client/client_request.rs

@@ -0,0 +1,24 @@
+use reqwest::Method;
+
+use crate::{error::ClientError, event::ManagementEvent};
+
+use super::{ClientRequest, OneShotRequest};
+
+/// Ping the Database instance
+#[derive(Debug, Clone, Copy)]
+pub struct PingRequest;
+
+impl ClientRequest for PingRequest {
+    const URL_PATH: &'static str = "/api/v1/ping";
+    const METHOD: Method = Method::GET;
+}
+
+impl OneShotRequest for PingRequest {
+    type Response = ManagementEvent;
+
+    fn validate_response(&self, response: &Self::Response) -> Result<(), ClientError> {
+        (response.ty() == "io.eventsourcingdb.api.ping-received")
+            .then_some(())
+            .ok_or(ClientError::PingFailed)
+    }
+}

src/client/client_request/ping.rs

@@ -0,0 +1,24 @@
+use reqwest::Method;
+
+use crate::{error::ClientError, event::ManagementEvent};
+
+use super::{ClientRequest, OneShotRequest};
+
+/// Verify the API token
+#[derive(Debug, Clone, Copy)]
+pub struct VerifyApiTokenRequest;
+
+impl ClientRequest for VerifyApiTokenRequest {
+    const URL_PATH: &'static str = "/api/v1/verify-api-token";
+    const METHOD: Method = Method::POST;
+}
+
+impl OneShotRequest for VerifyApiTokenRequest {
+    type Response = ManagementEvent;
+
+    fn validate_response(&self, response: &Self::Response) -> Result<(), ClientError> {
+        (response.ty() == "io.eventsourcingdb.api.api-token-verified")
+            .then_some(())
+            .ok_or(ClientError::APITokenInvalid)
+    }
+}

src/client/client_request/verify_api_token.rs

@@ -0,0 +1,26 @@
+use super::{ClientRequest, OneShotRequest};
+use crate::{
+    client::Precondition,
+    error::ClientError,
+    event::{Event, EventCandidate},
+};
+use reqwest::Method;
+use serde::Serialize;
+
+#[derive(Debug, Serialize)]
+pub struct WriteEventsRequest {
+    pub events: Vec<EventCandidate>,
+    pub preconditions: Vec<Precondition>,
+}
+
+impl ClientRequest for WriteEventsRequest {
+    const URL_PATH: &'static str = "/api/v1/write-events";
+    const METHOD: Method = Method::POST;
+
+    fn body(&self) -> Option<Result<impl Serialize, ClientError>> {
+        Some(Ok(self))
+    }
+}
+impl OneShotRequest for WriteEventsRequest {
+    type Response = Vec<Event>;
+}

src/client/client_request/write_events.rs

@@ -0,0 +1,22 @@
+use serde::Serialize;
+
+/// Enum for different preconditions that can be used when writing events
+#[derive(Debug, Serialize)]
+#[serde(tag = "type", content = "payload")]
+pub enum Precondition {
+    /// Check if the subject with the given path has no other events
+    #[serde(rename = "isSubjectPristine")]
+    IsSubjectPristine {
+        /// The subject to check
+        subject: String,
+    },
+    /// Check if the subject with the given path has no other events
+    #[serde(rename = "isSubjectOnEventId")]
+    IsSubjectOnEventId {
+        /// The subject to check
+        subject: String,
+        /// The event ID to check against
+        #[serde(rename = "eventId")]
+        event_id: String,
+    },
+}

src/client/precondition.rs

@@ -26,7 +26,7 @@ pub enum ClientError {
     SerdeJsonError(#[from] serde_json::Error),
     /// The DB returned an error+
     #[error("The DB returned an error: {0}")]
-    DBError(StatusCode, String),
+    DBApiError(StatusCode, String),
     /// There was a problem with the `cloudevents` message
     #[cfg(feature = "cloudevents")]
     #[error("The CloudEvents message is invalid: {0}")]
@@ -46,3 +46,12 @@ pub enum ContainerError {
     #[error("URL parsing error: {0}")]
     URLParseError(#[from] url::ParseError),
 }
+
+/// Error type for the event
+#[derive(Debug, thiserror::Error)]
+pub enum EventError {
+    /// The passed cloudevent is invalid
+    #[cfg(feature = "cloudevents")]
+    #[error("The passed cloudevent is invalid")]
+    InvalidCloudevent,
+}