feat(client): #54 add verify hash function

Author: Snapstromegon

Cargo.lock

@@ -27,7 +27,7 @@ futures-util = "0.3.31"
 jsonschema = "0.33.0"
 reqwest = { version = "0.12.23", features = ["json", "stream"] }
 serde = { version = "1.0.219", features = ["derive"] }
-serde_json = "1.0.143"
+serde_json = { version = "1.0.143", features = ["raw_value"] }
 testcontainers = { version = "0.25.0", features = [
   "http_wait",
 ], optional = true }
@@ -37,6 +37,8 @@ tokio-util = { version = "0.7.16", features = ["io"] }
 tokio-stream = { version = "0.1.17", features = ["io-util"] }
 typed-builder = "0.21.2"
 url = "2.5.4"
+sha2 = "0.10.9"
+hex = "0.4.3"
 
 [dev-dependencies]
 testcontainers = { version = "0.25.0", features = ["http_wait"] }

Cargo.toml

@@ -69,25 +69,30 @@ pub trait OneShotRequest: ClientRequest {
 
 /// A line in any json-nd stream coming from the database
 #[derive(Deserialize, Debug)]
-#[serde(tag = "type", content = "payload", rename_all = "camelCase")]
+#[serde(untagged)]
 enum StreamLineItem<T> {
-    /// An error occured during the request
-    Error { error: String },
-    /// A heardbeat message was sent to keep the connection alive.
-    /// This is only used when observing events, but it does not hurt to have it everywhere.
-    Heartbeat(Value),
+    Predefined(PredefinedStreamLineItem),
     /// A successful response from the database
     /// Since the exact type of the payload is not known at this point, we use this as a fallback case.
     /// Every request item gets put in here and the type can be checked later on.
     /// The type name checking is only for semantic reasons, as the payload is already parsed as the correct type at this point.
-    #[serde(untagged)]
     Ok {
         #[serde(rename = "type")]
         ty: String,
         payload: T,
     },
 }
 
+#[derive(Deserialize, Debug)]
+#[serde(tag = "type", content = "payload", rename_all = "camelCase")]
+enum PredefinedStreamLineItem {
+    /// An error occured during the request
+    Error { error: String },
+    /// A heardbeat message was sent to keep the connection alive.
+    /// This is only used when observing events, but it does not hurt to have it everywhere.
+    Heartbeat(Value),
+}
+
 /// Represents a request to the database that expects a stream of responses
 pub trait StreamingRequest: ClientRequest {
     type ItemType: DeserializeOwned;
@@ -113,11 +118,13 @@ pub trait StreamingRequest: ClientRequest {
                 .filter_map(|o| async {
                     match o {
                         // An error was passed by the database, so we forward it as an error.
-                        Ok(StreamLineItem::Error { error }) => {
-                            Some(Err(ClientError::DBError(error)))
-                        }
+                        Ok(StreamLineItem::Predefined(PredefinedStreamLineItem::Error {
+                            error,
+                        })) => Some(Err(ClientError::DBError(error))),
                         // A heartbeat message was sent, which we ignore.
-                        Ok(StreamLineItem::Heartbeat(_value)) => None,
+                        Ok(StreamLineItem::Predefined(PredefinedStreamLineItem::Heartbeat(
+                            _value,
+                        ))) => None,
                         // A successful response was sent with the correct type.
                         Ok(StreamLineItem::Ok { payload, ty }) if ty == Self::ITEM_TYPE_NAME => {
                             Some(Ok(payload))

src/client/client_request.rs

@@ -70,4 +70,12 @@ pub enum EventError {
     #[cfg(feature = "cloudevents")]
     #[error("The passed cloudevent is invalid")]
     InvalidCloudevent,
+    /// Hash verification failed
+    #[error("Hash verification failed")]
+    HashVerificationFailed {
+        /// Expected hash as in the DB
+        expected: String,
+        /// Actual hash as computed
+        actual: String,
+    },
 }

src/error.rs

@@ -1,15 +1,53 @@
 use chrono::{DateTime, Utc};
 use serde::{Deserialize, Serialize};
-use serde_json::Value;
+use serde_json::value::{RawValue, Value};
 
-use crate::event::{EventCandidate, trace_info::TraceInfo};
+use crate::{
+    error::EventError,
+    event::{EventCandidate, trace_info::TraceInfo},
+};
 #[cfg(feature = "cloudevents")]
 use cloudevents::EventBuilder;
+use sha2::{Digest, Sha256};
+
+#[derive(Debug, Clone)]
+pub struct CustomValue {
+    parsed: Value,
+    raw: Box<RawValue>,
+}
+
+impl PartialEq for CustomValue {
+    fn eq(&self, other: &Self) -> bool {
+        self.parsed == other.parsed
+    }
+}
+
+impl Eq for CustomValue {}
+
+impl<'de> Deserialize<'de> for CustomValue {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        let raw = Box::<RawValue>::deserialize(deserializer)?;
+        let parsed: Value = serde_json::from_str(raw.get()).map_err(serde::de::Error::custom)?;
+        Ok(Self { parsed, raw })
+    }
+}
+
+impl Serialize for CustomValue {
+    fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error>
+    where
+        S: serde::Serializer,
+    {
+        self.raw.serialize(serializer)
+    }
+}
 
 /// Represents an event that has been received from the DB.
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
 pub struct Event {
-    data: Value,
+    data: CustomValue,
     datacontenttype: String,
     hash: String,
     id: String,
@@ -28,7 +66,7 @@ impl Event {
     /// Get the data of an event.
     #[must_use]
     pub fn data(&self) -> &Value {
-        &self.data
+        &self.data.parsed
     }
     /// Get the data content type of an event.
     #[must_use]
@@ -92,12 +130,74 @@ impl Event {
     pub fn ty(&self) -> &str {
         &self.ty
     }
+
+    /// Verify the hash of an event.
+    ///
+    /// ```
+    /// use eventsourcingdb::event::EventCandidate;
+    /// # use serde_json::json;
+    /// # tokio_test::block_on(async {
+    /// # let container = eventsourcingdb::container::Container::start_preview().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::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())
+    ///        .ty("io.eventsourcingdb.test".to_string())
+    ///        .build()
+    /// ];
+    /// let written_events = client.write_events(candidates, vec![]).await.expect("Failed to write events");
+    /// let event = &written_events[0];
+    /// event.verify_hash().expect("Hash verification failed");
+    /// # })
+    /// ```
+    ///
+    /// # Errors
+    /// Returns an error if the hash verification fails.
+    pub fn verify_hash(&self) -> Result<(), EventError> {
+        let metadata = format!(
+            "{}|{}|{}|{}|{}|{}|{}|{}",
+            self.specversion,
+            self.id,
+            self.predecessorhash,
+            self.time
+                .to_rfc3339_opts(chrono::SecondsFormat::Nanos, true),
+            self.source,
+            self.subject,
+            self.ty,
+            self.datacontenttype,
+        );
+
+        let metadata_hash = Sha256::digest(metadata.as_bytes());
+        let metadata_hash_hex = hex::encode(metadata_hash);
+
+        let data_hash = Sha256::digest(self.data.raw.get());
+        let data_hash_hex = hex::encode(data_hash);
+
+        let final_hash_input = format!("{metadata_hash_hex}{data_hash_hex}");
+        let final_hash = Sha256::digest(final_hash_input.as_bytes());
+        let final_hash_hex = hex::encode(final_hash);
+
+        if final_hash_hex == self.hash {
+            Ok(())
+        } else {
+            Err(EventError::HashVerificationFailed {
+                expected: self.hash.clone(),
+                actual: final_hash_hex,
+            })
+        }
+    }
 }
 
 impl From<Event> for EventCandidate {
     fn from(event: Event) -> Self {
         Self {
-            data: event.data,
+            data: event.data.parsed,
             source: event.source,
             subject: event.subject,
             ty: event.ty,