feat(client): add metadata and discovery compliance matching

This implementation is incomplete and lacking tests

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

Author: Snapstromegon

Cargo.lock

@@ -12,13 +12,19 @@ testcontainer = ["dep:testcontainers"]
 chrono = { version = "0.4.41", features = ["serde"] }
 cloudevents-sdk = { version = "0.8.0", features = ["reqwest"], optional = true }
 url = "2.5.4"
-reqwest = { version = "0.12.15", features = ["json"] }
+reqwest = { version = "0.12.15", features = ["json", "stream"] }
 serde = { version = "1.0.219", features = ["derive"] }
 serde_json = "1.0.140"
 testcontainers = { version = "0.24.0", features = [
   "http_wait",
 ], optional = true }
 thiserror = "2.0.12"
+jsonschema = "0.30.0"
+futures-util = "0.3.31"
+tokio-util = { version = "0.7.15", features = ["io"] }
+tokio-stream = { version = "0.1.17", features = ["io-util"] }
+futures = "0.3.31"
+tokio = { version = "1.44.2", features = ["io-util"] }
 
 [dev-dependencies]
 eventsourcingdb-client-rust = { path = ".", features = ["testcontainer"] }

Cargo.toml

@@ -19,12 +19,15 @@
 
 mod client_request;
 
-use client_request::{ClientRequest, PingRequest, VerifyApiTokenRequest};
+use client_request::{
+    list_event_types::EventType, ClientRequest, ListEventTypesRequest, ListSubjectsRequest, OneShotRequest, PingRequest, RegisterEventSchemaRequest, StreamingRequest, VerifyApiTokenRequest
+};
 
+use futures::Stream;
 use reqwest;
 use url::Url;
 
-use crate::error::ClientError;
+use crate::{error::ClientError, event::ManagementEvent};
 
 /// Client for an [EventsourcingDB](https://www.eventsourcingdb.io/) instance.
 #[derive(Debug)]
@@ -72,9 +75,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 +101,49 @@ 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(),
+            ))
+        }
+    }
+
+    /// Utility function to request an endpoint of the API as a stream.
+    ///
+    /// This means, that the response is streamed and returned as a stream of values.
+    ///
+    /// # Errors
+    /// This function will return an error if the request fails or if the URL is invalid.
+    async fn request_streaming<R: StreamingRequest>(
+        &self,
+        endpoint: R,
+    ) -> Result<impl Stream<Item = Result<R::ItemType, ClientError>>, ClientError> {
+        let response = self.build_request(&endpoint)?.send().await?;
+
+        if response.status().is_success() {
+            Ok(endpoint.build_stream(response))
+        } else {
+            Err(ClientError::DBApiError(
                 response.status(),
                 response.text().await.unwrap_or_default(),
             ))
@@ -125,7 +167,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 +188,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(())
     }
+
+    /// Registers an event schema with the DB instance.
+    ///
+    /// # Errors
+    /// This function will return an error if the request fails or if the provided schema is invalid.
+    pub async fn register_event_schema(
+        &self,
+        event_type: &str,
+        schema: &serde_json::Value,
+    ) -> Result<ManagementEvent, ClientError> {
+        self.request_oneshot(RegisterEventSchemaRequest::try_new(event_type, schema)?)
+            .await
+    }
+
+    /// List all subjects in the DB instance.
+    ///
+    /// # Errors
+    /// This function will return an error if the request fails or if the URL is invalid.
+    pub async fn list_subjects(
+        &self,
+        base_subject: Option<&str>,
+    ) -> Result<impl Stream<Item = Result<String, ClientError>>, ClientError> {
+        let response = self
+            .request_streaming(ListSubjectsRequest { base_subject })
+            .await?;
+        Ok(response)
+    }
+
+    /// List all event types in the DB instance.
+    ///
+    /// # Errors
+    /// This function will return an error if the request fails or if the URL is invalid.
+    pub async fn list_event_types(
+        &self,
+    ) -> Result<impl Stream<Item = Result<EventType, ClientError>>, ClientError> {
+        let response = self.request_streaming(ListEventTypesRequest).await?;
+        Ok(response)
+    }
 }

src/client.rs

@@ -1,15 +1,31 @@
 //! This is a purely internal module to represent client requests to the database.
 
-use reqwest::Method;
-use serde_json::Value;
+pub mod list_event_types;
+mod list_subjects;
+mod ping;
+mod register_event_schema;
+mod verify_api_token;
+
+pub use list_event_types::ListEventTypesRequest;
+pub use list_subjects::ListSubjectsRequest;
+pub use ping::PingRequest;
+pub use register_event_schema::RegisterEventSchemaRequest;
+pub use verify_api_token::VerifyApiTokenRequest;
 
-use crate::{error::ClientError, event::ManagementEvent};
+use crate::error::ClientError;
+use futures::{Stream, stream::TryStreamExt};
+use futures_util::io;
+use reqwest::Method;
+use serde::Serialize;
+use serde::de::DeserializeOwned;
+use tokio::io::{AsyncBufReadExt, BufReader};
+use tokio_stream::wrappers::LinesStream;
+use tokio_util::io::StreamReader;
 
 /// 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 +38,40 @@ 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;
+/// Represents a request to the database that expects a stream of responses
+pub trait StreamingRequest: ClientRequest {
+    type ItemType: DeserializeOwned;
 
-impl ClientRequest for VerifyApiTokenRequest {
-    const URL_PATH: &'static str = "/api/v1/verify-api-token";
-    const METHOD: Method = Method::POST;
-    type Response = ManagementEvent;
+    fn build_stream(
+        self,
+        response: reqwest::Response,
+    ) -> impl Stream<Item = Result<Self::ItemType, ClientError>>;
 
-    fn validate_response(&self, response: &Self::Response) -> Result<(), ClientError> {
-        (response.ty() == "io.eventsourcingdb.api.api-token-verified")
-            .then_some(())
-            .ok_or(ClientError::APITokenInvalid)
+    fn lines_stream(
+        response: reqwest::Response,
+    ) -> impl Stream<Item = Result<String, ClientError>> {
+        let bytes = response.bytes_stream().map_err(|err| {
+            io::Error::new(
+                io::ErrorKind::Other,
+                format!("Failed to read response stream: {err}"),
+            )
+        });
+        let stream_reader = StreamReader::new(bytes);
+        LinesStream::new(BufReader::new(stream_reader).lines()).map_err(ClientError::from)
     }
 }

src/client/client_request.rs

@@ -0,0 +1,56 @@
+use futures::{Stream, stream::StreamExt};
+use reqwest::Method;
+use serde::{Deserialize, Serialize};
+use serde_json::Value;
+
+use crate::error::ClientError;
+
+use super::{ClientRequest, StreamingRequest};
+#[derive(Deserialize, Debug)]
+pub struct EventType {
+    pub name: String,
+    pub is_phantom: bool,
+    pub schema: Option<Value>,
+}
+
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ListEventTypesRequest;
+
+impl ClientRequest for ListEventTypesRequest {
+    const URL_PATH: &'static str = "/api/v1/read-event-types";
+    const METHOD: Method = Method::POST;
+
+    fn body(&self) -> Option<Result<impl Serialize, ClientError>> {
+        Some(Ok(self))
+    }
+}
+impl StreamingRequest for ListEventTypesRequest {
+    type ItemType = EventType;
+
+    fn build_stream(
+        self,
+        response: reqwest::Response,
+    ) -> impl Stream<Item = Result<Self::ItemType, ClientError>> {
+        #[derive(Deserialize, Debug)]
+        enum LineItem {
+            Error(String),
+            EventType(EventType),
+        }
+
+        impl From<LineItem> for Result<EventType, ClientError> {
+            fn from(item: LineItem) -> Self {
+                match item {
+                    LineItem::Error(err) => Err(ClientError::DBError(err)),
+                    LineItem::EventType(event_type) => Ok(event_type),
+                }
+            }
+        }
+
+        Self::lines_stream(response).map(|line| {
+            let line = line?;
+            let item: LineItem = serde_json::from_str(line.as_str())?;
+            item.into()
+        })
+    }
+}

src/client/client_request/list_event_types.rs

@@ -0,0 +1,48 @@
+use futures::{Stream, stream::StreamExt};
+use reqwest::Method;
+use serde::{Deserialize, Serialize};
+
+use crate::error::ClientError;
+
+use super::{ClientRequest, StreamingRequest};
+
+#[derive(Debug, Clone, Serialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ListSubjectsRequest<'a> {
+    pub base_subject: Option<&'a str>,
+}
+
+impl<'a> ClientRequest for ListSubjectsRequest<'a> {
+    const URL_PATH: &'static str = "/api/v1/read-subjects";
+    const METHOD: Method = Method::POST;
+
+    fn body(&self) -> Option<Result<impl Serialize, ClientError>> {
+        Some(Ok(self))
+    }
+}
+impl<'a> StreamingRequest for ListSubjectsRequest<'a> {
+    type ItemType = String;
+
+    fn build_stream(
+        self,
+        response: reqwest::Response,
+    ) -> impl Stream<Item = Result<Self::ItemType, ClientError>> {
+        #[derive(Deserialize, Debug)]
+        struct LineItem {
+            payload: LineItemPayload,
+            r#type: String,
+        }
+        #[derive(Deserialize, Debug)]
+        struct LineItemPayload {
+            subject: String,
+        }
+        Self::lines_stream(response).map(|line| {
+            let line = line?;
+            let item: LineItem = serde_json::from_str(line.as_str())?;
+            if item.r#type != "subject" {
+                return Err(ClientError::InvalidEventType);
+            }
+            Ok(item.payload.subject)
+        })
+    }
+}

src/client/client_request/list_subjects.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)
+    }
+}