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

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,18 @@
 //! 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, PingRequest, VerifyApiTokenRequest, WriteEvents};
 
+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)]
@@ -149,4 +154,42 @@ impl Client {
         let _ = self.request(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(WriteEvents {
+            events,
+            preconditions,
+        })
+        .await
+    }
 }

src/client.rs

@@ -1,9 +1,14 @@
 //! This is a purely internal module to represent client requests to the database.
 
 use reqwest::Method;
-use serde_json::Value;
+use serde::Serialize;
 
-use crate::{error::ClientError, event::ManagementEvent};
+use crate::{
+    error::ClientError,
+    event::{Event, EventCandidate, ManagementEvent},
+};
+
+use super::precondition::Precondition;
 
 /// Represents a request to the database client
 pub trait ClientRequest {
@@ -22,8 +27,8 @@ 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<(), ClientError>>
     }
 
     /// Validate the response from the database
@@ -63,3 +68,19 @@ impl ClientRequest for VerifyApiTokenRequest {
             .ok_or(ClientError::APITokenInvalid)
     }
 }
+
+#[derive(Debug, Serialize)]
+pub struct WriteEvents {
+    pub events: Vec<EventCandidate>,
+    pub preconditions: Vec<Precondition>,
+}
+
+impl ClientRequest for WriteEvents {
+    const URL_PATH: &'static str = "/api/v1/write-events";
+    const METHOD: Method = Method::POST;
+    type Response = Vec<Event>;
+
+    fn body(&self) -> Option<Result<impl Serialize, ClientError>> {
+        Some(Ok(self))
+    }
+}

src/client/client_request.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

@@ -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,
+}

src/error.rs

@@ -1,5 +1,83 @@
 //! This module holds all event types that are send between the client and the database.
 
+// Allow module inception here, since "event" is the expected as the name for both modules.
+// Renaming would be possible, but would probably lead to more confusion.
+#[allow(clippy::module_inception)]
+mod event;
+mod event_candidate;
 mod management_event;
 
+pub use event::Event;
+pub use event_candidate::EventCandidate;
 pub use management_event::ManagementEvent;
+use serde::{Deserialize, Serialize};
+
+#[cfg(feature="cloudevents")]
+use crate::error::EventError;
+
+/// Represents the trace information of an event.
+/// This is used for distributed tracing.
+/// It can either be a traceparent or a traceparent and tracestate.
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(untagged)]
+pub enum TraceInfo {
+    // LEAVE ORDER AS IS
+    // This is important for deserialization as the traceparent is always present
+
+    /// The traceparent and tracestate of the event.
+    /// This is used for distributed tracing.
+    WithState {
+        /// The traceparent of the event.
+        /// This is used for distributed tracing.
+        traceparent: String,
+        /// The tracestate of the event.
+        /// This is used for distributed tracing.
+        tracestate: String,
+    },
+    /// The traceparent of the event.
+    /// This is used for distributed tracing.
+    Traceparent {
+        /// The traceparent of the event.
+        /// This is used for distributed tracing.
+        traceparent: String,
+    },
+}
+
+impl TraceInfo {
+    /// Get the traceparent of the event.
+    #[must_use]
+    pub fn traceparent(&self) -> &str {
+        match self {
+            Self::Traceparent { traceparent } | Self::WithState { traceparent, .. } => traceparent,
+        }
+    }
+    /// Get the tracestate of the event.
+    #[must_use]
+    pub fn tracestate(&self) -> Option<&str> {
+        match self {
+            Self::Traceparent { .. } => None,
+            Self::WithState { tracestate, .. } => Some(tracestate),
+        }
+    }
+
+    /// Create a new `TraceInfo` from a cloudevent.
+    /// This will return None if the cloudevent does not contain a traceparent or tracestate.
+    ///
+    /// # Errors
+    /// If the cloudevent contains a tracestate but no traceparent, an error will be returned.
+    #[cfg(feature="cloudevents")]
+    pub fn from_cloudevent(event: &cloudevents::Event) -> Result<Option<Self>, EventError> {
+        let traceparent = event.extension("traceparent").map(ToString::to_string);
+        let tracestate = event.extension("tracestate").map(ToString::to_string);
+
+        match (traceparent, tracestate) {
+            (Some(traceparent), Some(tracestate)) => Ok(Some(Self::WithState {
+                traceparent,
+                tracestate,
+            })),
+            (Some(traceparent), None) => Ok(Some(Self::Traceparent { traceparent })),
+            (None, None) => Ok(None),
+            (None, Some(_)) => Err(EventError::InvalidCloudevent),
+        }
+    }
+}