From 564ced0a292cd17a1391c4ab9af65746adb45a69 Mon Sep 17 00:00:00 2001 From: flangator Date: Mon, 16 Mar 2026 22:48:17 +0100 Subject: [PATCH 1/6] feat: initial setup, dependency update and read event implementation --- Cargo.toml | 9 ++- compose.yaml | 21 +++++- src/conversion.rs | 159 ++++++++++++++++++++++++++++++++++++++++ src/errors.rs | 32 ++++++++ src/event_repository.rs | 109 +++++++++++++++++++++++++++ src/lib.rs | 5 ++ src/types.rs | 6 ++ 7 files changed, 337 insertions(+), 4 deletions(-) create mode 100644 src/conversion.rs create mode 100644 src/errors.rs create mode 100644 src/event_repository.rs create mode 100644 src/types.rs diff --git a/Cargo.toml b/Cargo.toml index c52e06e..2274761 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,5 +9,10 @@ license = "Apache-2.0" keywords = ["cqrs", "event-sourcing", "eventsourcingdb", "cqrs-es"] [dependencies] -cqrs-es = "0.4.12" -eventsourcingdb = "1.1.0" +chrono = { version = "0.4.44", features = ["serde"] } +cqrs-es = "0.5.0" +eventsourcingdb = "2.0.1" +futures = "0.3.32" +serde = { version = "1.0.228", features = ["derive"] } +serde_json = "1.0.149" +thiserror = "2.0.18" diff --git a/compose.yaml b/compose.yaml index dfe3ff7..4367d45 100644 --- a/compose.yaml +++ b/compose.yaml @@ -1,8 +1,8 @@ services: eventsourcingdb: - image: thenativeweb/eventsourcingdb + image: thenativeweb/eventsourcingdb:1.2.0 container_name: eventsourcingdb - restart: always + restart: unless-stopped ports: - 3000:3000 command: @@ -12,3 +12,20 @@ services: - --http-enabled - --https-enabled=false - --with-ui + healthcheck: + test: + [ + "CMD-SHELL", + "wget -S --spider http://eventsourcingdb:3000/api/v1/health 2>&1 | grep 'HTTP/1.1 200 OK'", + ] + # ["CMD", "curl", "-f", "http://eventsourcingdb:3000/api/v1/not-exists"] + interval: 5s + timeout: 5s + retries: 5 + # command: "mongod --replSet rs0" + # healthcheck: + # test: | + # mongosh --quiet --eval "try { rs.status().ok } catch (e) { rs.initiate({ _id: 'rs0', members: [{ _id: 0, host: 'localhost:27017' }] }).ok }" + # interval: 5s + # timeout: 5s + # retries: 5 diff --git a/src/conversion.rs b/src/conversion.rs new file mode 100644 index 0000000..18d2650 --- /dev/null +++ b/src/conversion.rs @@ -0,0 +1,159 @@ +use std::str::FromStr; + +use chrono::{DateTime, Utc}; +use cqrs_es::persist::SerializedEvent; +use eventsourcingdb::{Event, TraceInfo}; +use serde::{Deserialize, Serialize}; + +use crate::errors::{EventSourcingDbError, EventSourcingDbResult}; + +fn to_pascal_case(s: &str) -> String { + s.split('-') + .map(|w| { + let mut w = w.to_string(); + w[..1].make_ascii_uppercase(); + w + }) + .collect() +} + +fn pascal_to_kebab_case(s: &str) -> String { + let mut out = String::new(); + + for (i, c) in s.chars().enumerate() { + if c.is_uppercase() { + if i != 0 { + out.push('-'); + } + out.push(c.to_ascii_lowercase()); + } else { + out.push(c); + } + } + + out +} + +// we assume that the subject always has the shape /aggregate_type/aggregate_id +// e.g. /books/42 +pub(crate) fn map_subject_to_aggregate_type_and_id( + subject: &str, +) -> Result<(String, String), EventSourcingDbError> { + let mut parts = subject.trim_start_matches('/').split('/'); + + let aggregate_type = parts + .next() + .ok_or_else(|| EventSourcingDbError::InvalidSubject(subject.into()))?; + + let aggregate_id = parts + .next() + .ok_or_else(|| EventSourcingDbError::InvalidSubject(subject.into()))?; + + if parts.next().is_some() { + return Err(EventSourcingDbError::InvalidSubject(subject.into())); + } + + Ok((aggregate_type.to_string(), aggregate_id.to_string())) +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct ReversedDomain(Vec); + +impl ReversedDomain { + pub fn labels(&self) -> &[String] { + &self.0 + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct QualifiedEventType { + pub reversed_domain: ReversedDomain, + pub event_type: String, + pub event_version: Option, +} + +impl FromStr for QualifiedEventType { + type Err = EventSourcingDbError; + + fn from_str(value: &str) -> Result { + let mut parts: Vec<&str> = value.split('.').collect(); + + if parts.len() < 2 { + return Err(EventSourcingDbError::InvalidEventTypeIdentifier( + value.into(), + )); + } + + // the last part might be a version + let mut version = None; + + if let Some(last) = parts.last() { + if let Some(num) = last.strip_prefix('v') { + version = Some(num.parse()?); + parts.pop(); // version entfernen + } + } + + // the remainder has at least the event type and the reversed domain + if parts.len() < 2 { + return Err(EventSourcingDbError::InvalidEventTypeIdentifier( + value.into(), + )); + } + + let event_type = parts.pop().unwrap().to_string(); + + let reversed_domain = ReversedDomain(parts.into_iter().map(|s| s.to_string()).collect()); + + Ok(QualifiedEventType { + reversed_domain, + event_type, + event_version: version, + }) + } +} + +#[derive(Debug, Clone, Deserialize, Serialize)] +pub struct EventMetadata { + pub source: String, + pub subject: String, + pub time: DateTime, + pub traceinfo: Option, + pub datacontenttype: String, + pub specversion: String, + pub hash: String, + pub predecessorhash: String, + pub signature: Option, +} + +impl From<&Event> for EventMetadata { + fn from(event: &Event) -> Self { + Self { + source: event.source().to_string(), + subject: event.subject().to_string(), + time: event.time().clone(), + traceinfo: event.traceinfo().cloned(), + datacontenttype: event.datacontenttype().to_string(), + specversion: event.specversion().to_string(), + hash: event.hash().to_string(), + predecessorhash: event.predecessorhash().to_string(), + signature: event.signature().map(ToString::to_string), + } + } +} + +pub fn map_event(event: Event) -> EventSourcingDbResult { + let event_type: QualifiedEventType = event.ty().parse()?; + let (aggregate_type, aggregate_id) = map_subject_to_aggregate_type_and_id(event.subject())?; + let meta = EventMetadata::from(&event); + Ok(SerializedEvent { + aggregate_id: aggregate_id, + sequence: event.id().parse()?, + aggregate_type, + event_type: event_type.event_type, + // a non-existing version defaults to 1 + event_version: event_type.event_version.unwrap_or(1 as usize).to_string(), + payload: event.data().clone(), + metadata: serde_json::to_value(&meta)?, + }) +} diff --git a/src/errors.rs b/src/errors.rs new file mode 100644 index 0000000..967e47c --- /dev/null +++ b/src/errors.rs @@ -0,0 +1,32 @@ +use cqrs_es::persist::PersistenceError; +use eventsourcingdb::error::ClientError; +use thiserror::Error; + +#[derive(Debug, Error)] +pub enum EventSourcingDbError { + #[error("{0}")] + SerializationError(#[from] serde_json::Error), + #[error("invalid event type identifier: {0}")] + InvalidEventTypeIdentifier(String), + #[error("invalid subject: {0}")] + InvalidSubject(String), + #[error("invalid sequence number: {0}")] + InvalidSequence(#[from] std::num::ParseIntError), + //TODO: this can be unmangled/ matched + #[error("{0}")] + ClientError(#[from] ClientError), +} + +pub type EventSourcingDbResult = Result; + +impl From for PersistenceError { + //TODO: this conversion can be improved + fn from(value: EventSourcingDbError) -> Self { + match value { + EventSourcingDbError::ClientError(err) => { + PersistenceError::ConnectionError(Box::new(err)) + } + _ => PersistenceError::UnknownError(Box::new(value)), + } + } +} diff --git a/src/event_repository.rs b/src/event_repository.rs new file mode 100644 index 0000000..00e37b3 --- /dev/null +++ b/src/event_repository.rs @@ -0,0 +1,109 @@ +use std::sync::Arc; + +use cqrs_es::{ + Aggregate, + persist::{ + PersistedEventRepository, PersistenceError, ReplayStream, SerializedEvent, + SerializedSnapshot, + }, +}; +use eventsourcingdb::{ + Client, Event, + request_options::{Bound, BoundType, ReadEventsOptions}, +}; +use futures::TryStreamExt; +use serde_json::Value; + +use crate::{ + conversion::{ReversedDomain, map_event}, + errors::EventSourcingDbResult, +}; + +pub struct EventSourcingDbEventRepository { + client: Arc, + domain: ReversedDomain, +} + +impl EventSourcingDbEventRepository { + fn get_subject(id: &str) -> String { + format!("/{}/{}", &A::TYPE, id) + } + + async fn query_events( + &self, + aggregate_id: &str, + min_sequence: usize, + ) -> EventSourcingDbResult> { + let subject = Self::get_subject::(aggregate_id); + let sequence_string = min_sequence.to_string(); + let bound = Bound { + bound_type: BoundType::Inclusive, + id: &sequence_string, + }; + let events: Vec = self + .client + .read_events( + &subject, + Some(ReadEventsOptions { + lower_bound: Some(bound), + ..Default::default() + }), + ) + .await? + .try_collect::>() + .await? + .into_iter() + .map(map_event) + .collect::, _>>()?; + + Ok(events) + } +} + +impl PersistedEventRepository for EventSourcingDbEventRepository { + async fn get_events( + &self, + aggregate_id: &str, + ) -> Result, PersistenceError> { + let events = self.query_events::(aggregate_id, 0).await?; + + Ok(events) + } + + async fn get_last_events( + &self, + aggregate_id: &str, + last_sequence: usize, + ) -> Result, PersistenceError> { + let events = self.query_events::(aggregate_id, last_sequence).await?; + Ok(events) + } + + async fn get_snapshot( + &self, + aggregate_id: &str, + ) -> Result, PersistenceError> { + todo!() + } + + async fn persist( + &self, + events: &[cqrs_es::persist::SerializedEvent], + snapshot_update: Option<(String, Value, usize)>, + ) -> Result<(), PersistenceError> { + todo!() + } + + async fn stream_events( + &self, + aggregate_id: &str, + ) -> Result { + todo!() + } + + async fn stream_all_events( + &self, + ) -> Result { + todo!() + } +} diff --git a/src/lib.rs b/src/lib.rs index b93cf3f..0f04534 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,3 +1,8 @@ +pub mod conversion; +pub mod errors; +pub mod event_repository; +pub mod types; + pub fn add(left: u64, right: u64) -> u64 { left + right } diff --git a/src/types.rs b/src/types.rs new file mode 100644 index 0000000..eabd396 --- /dev/null +++ b/src/types.rs @@ -0,0 +1,6 @@ +use cqrs_es::{CqrsFramework, persist::PersistedEventStore}; + +use crate::event_repository::EventSourcingDbEventRepository; + +pub type EventSourcingDbCqrs = + CqrsFramework>; From 0ae814dbde97afb3881b1817a99ba10993c25dee Mon Sep 17 00:00:00 2001 From: flangator Date: Tue, 17 Mar 2026 00:15:00 +0100 Subject: [PATCH 2/6] feat: map sequences to event ids --- src/conversion.rs | 148 ++++++++++++++++- src/errors.rs | 2 +- src/event_repository.rs | 341 ++++++++++++++++++++++++++++++++++++---- src/lib.rs | 15 -- 4 files changed, 455 insertions(+), 51 deletions(-) diff --git a/src/conversion.rs b/src/conversion.rs index 18d2650..b3aad52 100644 --- a/src/conversion.rs +++ b/src/conversion.rs @@ -4,9 +4,12 @@ use chrono::{DateTime, Utc}; use cqrs_es::persist::SerializedEvent; use eventsourcingdb::{Event, TraceInfo}; use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; use crate::errors::{EventSourcingDbError, EventSourcingDbResult}; +pub const STORED_EVENT_ENVELOPE_MARKER: &str = "eventsourcingdb-es@1"; + fn to_pascal_case(s: &str) -> String { s.split('-') .map(|w| { @@ -60,6 +63,14 @@ pub(crate) fn map_subject_to_aggregate_type_and_id( pub struct ReversedDomain(Vec); impl ReversedDomain { + pub fn new(labels: I) -> Self + where + I: IntoIterator, + S: Into, + { + Self(labels.into_iter().map(Into::into).collect()) + } + pub fn labels(&self) -> &[String] { &self.0 } @@ -142,18 +153,139 @@ impl From<&Event> for EventMetadata { } } -pub fn map_event(event: Event) -> EventSourcingDbResult { +pub fn qualify_event_type( + domain: &ReversedDomain, + event_type: &str, + event_version: &str, +) -> String { + let mut parts: Vec = domain.labels().to_vec(); + parts.push(pascal_to_kebab_case(event_type)); + + if event_version != "1" { + parts.push(format!("v{event_version}")); + } + + parts.join(".") +} + +pub fn wrap_event_data(payload: Value, metadata: Value) -> Value { + json!({ + "_cqrs_es": STORED_EVENT_ENVELOPE_MARKER, + "payload": payload, + "metadata": metadata, + }) +} + +pub fn unwrap_event_data(data: &Value) -> (Value, Value) { + let Some(object) = data.as_object() else { + return (data.clone(), json!({})); + }; + + let is_wrapped = object + .get("_cqrs_es") + .and_then(Value::as_str) + .map(|marker| marker == STORED_EVENT_ENVELOPE_MARKER) + .unwrap_or(false); + + if !is_wrapped { + return (data.clone(), json!({})); + } + + let payload = object.get("payload").cloned().unwrap_or(Value::Null); + let metadata = object.get("metadata").cloned().unwrap_or_else(|| json!({})); + + (payload, metadata) +} + +pub fn map_event( + event: Event, + sequence: usize, +) -> EventSourcingDbResult<(SerializedEvent, EventMetadata)> { let event_type: QualifiedEventType = event.ty().parse()?; let (aggregate_type, aggregate_id) = map_subject_to_aggregate_type_and_id(event.subject())?; + let (payload, metadata) = unwrap_event_data(event.data()); let meta = EventMetadata::from(&event); - Ok(SerializedEvent { - aggregate_id: aggregate_id, - sequence: event.id().parse()?, + let serialized = SerializedEvent { + aggregate_id, + sequence, aggregate_type, - event_type: event_type.event_type, + event_type: to_pascal_case(&event_type.event_type), // a non-existing version defaults to 1 event_version: event_type.event_version.unwrap_or(1 as usize).to_string(), - payload: event.data().clone(), - metadata: serde_json::to_value(&meta)?, - }) + payload, + metadata, + }; + + Ok((serialized, meta)) +} + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn sample_event(data: Value) -> Event { + serde_json::from_value(json!({ + "data": data, + "datacontenttype": "application/json", + "hash": "hash", + "id": "01JXYZOPAQUEID", + "predecessorhash": "predecessor", + "source": "urn:test", + "specversion": "1.0", + "subject": "/BookAggregate/42", + "time": "2026-03-17T10:00:00Z", + "type": "io.eventsourcingdb.book-created.v2", + "signature": null + })) + .expect("event JSON should deserialize") + } + + #[test] + fn qualify_event_type_uses_domain_and_kebab_case() { + let domain = ReversedDomain::new(["io", "eventsourcingdb"]); + let qualified = qualify_event_type(&domain, "BookCreated", "2"); + + assert_eq!(qualified, "io.eventsourcingdb.book-created.v2"); + } + + #[test] + fn wrap_and_unwrap_event_data_round_trip_payload_and_metadata() { + let payload = json!({ "title": "DDD" }); + let metadata = json!({ "request_id": "abc-123" }); + + let wrapped = wrap_event_data(payload.clone(), metadata.clone()); + let (actual_payload, actual_metadata) = unwrap_event_data(&wrapped); + + assert_eq!(actual_payload, payload); + assert_eq!(actual_metadata, metadata); + } + + #[test] + fn unwrap_raw_event_data_keeps_payload_and_defaults_metadata() { + let raw_payload = json!({ "title": "DDD" }); + let (payload, metadata) = unwrap_event_data(&raw_payload); + + assert_eq!(payload, raw_payload); + assert_eq!(metadata, json!({})); + } + + #[test] + fn map_event_uses_logical_sequence_not_opaque_event_id() { + let event = sample_event(wrap_event_data( + json!({ "title": "DDD" }), + json!({ "request_id": "abc-123" }), + )); + + let (serialized, meta) = map_event(event, 7).expect("event should map"); + + assert_eq!(serialized.sequence, 7); + assert_eq!(serialized.aggregate_id, "42"); + assert_eq!(serialized.aggregate_type, "BookAggregate"); + assert_eq!(serialized.event_type, "BookCreated"); + assert_eq!(serialized.event_version, "2"); + assert_eq!(serialized.payload, json!({ "title": "DDD" })); + assert_eq!(serialized.metadata, json!({ "request_id": "abc-123" })); + assert_eq!(meta.subject, "/BookAggregate/42"); + } } diff --git a/src/errors.rs b/src/errors.rs index 967e47c..7b454b2 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -12,7 +12,7 @@ pub enum EventSourcingDbError { InvalidSubject(String), #[error("invalid sequence number: {0}")] InvalidSequence(#[from] std::num::ParseIntError), - //TODO: this can be unmangled/ matched + //TODO: this can be unmangled/ matched, especially OptimisticLocking and serde #[error("{0}")] ClientError(#[from] ClientError), } diff --git a/src/event_repository.rs b/src/event_repository.rs index 00e37b3..3877406 100644 --- a/src/event_repository.rs +++ b/src/event_repository.rs @@ -8,55 +8,207 @@ use cqrs_es::{ }, }; use eventsourcingdb::{ - Client, Event, - request_options::{Bound, BoundType, ReadEventsOptions}, + Client, Event, EventCandidate, Precondition, + request_options::{Bound, BoundType, Ordering, ReadEventsOptions}, }; use futures::TryStreamExt; +use serde::{Deserialize, Serialize}; use serde_json::Value; use crate::{ - conversion::{ReversedDomain, map_event}, - errors::EventSourcingDbResult, + conversion::{ReversedDomain, map_event, qualify_event_type, wrap_event_data}, + errors::{EventSourcingDbError, EventSourcingDbResult}, }; +const SNAPSHOT_EVENT_TYPE: &str = "io.eventsourcingdb.cqrs-es.snapshot-record.v1"; +const SNAPSHOT_EVENT_SOURCE: &str = "urn:eventsourcingdb-es:snapshot"; +const EVENT_SOURCE: &str = "urn:eventsourcingdb-es:event"; + +#[derive(Debug, Clone, Serialize, Deserialize)] +struct EventSourcingDbSnapshotRecord { + aggregate_type: String, + aggregate_id: String, + aggregate: Value, + current_sequence: usize, + current_snapshot: usize, + last_event_id: Option, +} + +#[derive(Debug, Clone)] +struct SubjectState { + logical_sequence_count: usize, + last_event_id: Option, +} + pub struct EventSourcingDbEventRepository { client: Arc, domain: ReversedDomain, } impl EventSourcingDbEventRepository { + pub fn new(client: Arc, domain: ReversedDomain) -> Self { + Self { client, domain } + } + fn get_subject(id: &str) -> String { format!("/{}/{}", &A::TYPE, id) } + fn get_snapshot_subject(id: &str) -> String { + format!("/__snapshots__/{}/{}", &A::TYPE, id) + } + + async fn read_subject_events( + client: Arc, + aggregate_id: String, + ) -> EventSourcingDbResult> { + let subject = Self::get_subject::(&aggregate_id); + Self::read_events_from_subject(client, subject, None).await + } + + async fn read_events_from_subject( + client: Arc, + subject: String, + lower_bound_id: Option, + ) -> EventSourcingDbResult> { + let options = Some(ReadEventsOptions { + lower_bound: lower_bound_id.as_deref().map(|id| Bound { + bound_type: BoundType::Exclusive, + id, + }), + order: Some(Ordering::Chronological), + ..Default::default() + }); + + client + .read_events(&subject, options) + .await? + .try_collect::>() + .await + .map_err(Into::into) + } + + fn map_subject_events( + events: Vec, + _domain: &ReversedDomain, + skip: usize, + ) -> EventSourcingDbResult> { + events + .into_iter() + .enumerate() + .skip(skip) + .map(|(index, event)| map_event(event, index + 1).map(|(event, _)| event)) + .collect() + } + async fn query_events( &self, aggregate_id: &str, - min_sequence: usize, + skip: usize, ) -> EventSourcingDbResult> { - let subject = Self::get_subject::(aggregate_id); - let sequence_string = min_sequence.to_string(); - let bound = Bound { - bound_type: BoundType::Inclusive, - id: &sequence_string, - }; - let events: Vec = self + let events = + Self::read_subject_events::(Arc::clone(&self.client), aggregate_id.to_string()) + .await?; + Self::map_subject_events(events, &self.domain, skip) + } + + async fn load_snapshot_record( + &self, + aggregate_id: &str, + ) -> EventSourcingDbResult> { + let subject = Self::get_snapshot_subject::(aggregate_id); + let mut stream = self .client .read_events( &subject, Some(ReadEventsOptions { - lower_bound: Some(bound), + order: Some(Ordering::Antichronological), ..Default::default() }), ) - .await? - .try_collect::>() - .await? + .await?; + + let Some(event) = stream.try_next().await? else { + return Ok(None); + }; + + let record = serde_json::from_value(event.data().clone())?; + Ok(Some(record)) + } + + async fn read_events_since_snapshot( + &self, + aggregate_id: &str, + snapshot: &EventSourcingDbSnapshotRecord, + ) -> EventSourcingDbResult> { + let Some(last_event_id) = snapshot.last_event_id.clone() else { + return self + .query_events::(aggregate_id, snapshot.current_sequence) + .await; + }; + + let subject = Self::get_subject::(aggregate_id); + let events = + Self::read_events_from_subject(Arc::clone(&self.client), subject, Some(last_event_id)) + .await?; + + events .into_iter() - .map(map_event) - .collect::, _>>()?; + .enumerate() + .map(|(index, event)| { + map_event(event, snapshot.current_sequence + index + 1).map(|(event, _)| event) + }) + .collect() + } + + async fn subject_state( + &self, + aggregate_id: &str, + ) -> EventSourcingDbResult { + let events = + Self::read_subject_events::(Arc::clone(&self.client), aggregate_id.to_string()) + .await?; + let logical_sequence_count = events.len(); + let last_event_id = events.last().map(|event| event.id().to_string()); - Ok(events) + Ok(SubjectState { + logical_sequence_count, + last_event_id, + }) + } + + fn build_event_candidate( + &self, + event: &SerializedEvent, + ) -> EventSourcingDbResult { + Ok(EventCandidate::builder() + .source(EVENT_SOURCE.to_string()) + .subject(Self::get_subject::(&event.aggregate_id)) + .ty(qualify_event_type( + &self.domain, + &event.event_type, + &event.event_version, + )) + .data(wrap_event_data( + event.payload.clone(), + event.metadata.clone(), + )) + .build()) + } + + async fn store_snapshot_record( + &self, + record: EventSourcingDbSnapshotRecord, + ) -> EventSourcingDbResult<()> { + let candidate = EventCandidate::builder() + .source(SNAPSHOT_EVENT_SOURCE.to_string()) + .subject(Self::get_snapshot_subject::(&record.aggregate_id)) + .ty(SNAPSHOT_EVENT_TYPE.to_string()) + .data(serde_json::to_value(record)?) + .build(); + + self.client.write_events(vec![candidate], vec![]).await?; + Ok(()) } } @@ -65,9 +217,9 @@ impl PersistedEventRepository for EventSourcingDbEventRepository { &self, aggregate_id: &str, ) -> Result, PersistenceError> { - let events = self.query_events::(aggregate_id, 0).await?; - - Ok(events) + self.query_events::(aggregate_id, 0) + .await + .map_err(Into::into) } async fn get_last_events( @@ -75,15 +227,35 @@ impl PersistedEventRepository for EventSourcingDbEventRepository { aggregate_id: &str, last_sequence: usize, ) -> Result, PersistenceError> { - let events = self.query_events::(aggregate_id, last_sequence).await?; - Ok(events) + if let Some(snapshot) = self.load_snapshot_record::(aggregate_id).await? { + if snapshot.current_sequence == last_sequence { + return self + .read_events_since_snapshot::(aggregate_id, &snapshot) + .await + .map_err(Into::into); + } + } + + self.query_events::(aggregate_id, last_sequence) + .await + .map_err(Into::into) } async fn get_snapshot( &self, aggregate_id: &str, ) -> Result, PersistenceError> { - todo!() + self.load_snapshot_record::(aggregate_id) + .await + .map(|snapshot| { + snapshot.map(|snapshot| SerializedSnapshot { + aggregate_id: snapshot.aggregate_id, + aggregate: snapshot.aggregate, + current_sequence: snapshot.current_sequence, + current_snapshot: snapshot.current_snapshot, + }) + }) + .map_err(Into::into) } async fn persist( @@ -91,12 +263,73 @@ impl PersistedEventRepository for EventSourcingDbEventRepository { events: &[cqrs_es::persist::SerializedEvent], snapshot_update: Option<(String, Value, usize)>, ) -> Result<(), PersistenceError> { - todo!() + if events.is_empty() { + return Ok(()); + } + + let first_event = &events[0]; + let aggregate_id = first_event.aggregate_id.clone(); + let expected_last_sequence = first_event.sequence.saturating_sub(1); + let subject_state = self.subject_state::(&aggregate_id).await?; + + if subject_state.logical_sequence_count != expected_last_sequence { + return Err(PersistenceError::OptimisticLockError); + } + + let precondition = match subject_state.last_event_id.clone() { + Some(event_id) => Precondition::IsSubjectOnEventId { + subject: Self::get_subject::(&aggregate_id), + event_id, + }, + None => Precondition::IsSubjectPristine { + subject: Self::get_subject::(&aggregate_id), + }, + }; + + let candidates = events + .iter() + .map(|event| self.build_event_candidate::(event)) + .collect::>>()?; + + let written_events = match self + .client + .write_events(candidates, vec![precondition]) + .await + { + Ok(events) => events, + Err(err) => { + return Err(match err { + eventsourcingdb::error::ClientError::DBApiError(status, _) + if matches!(status.as_u16(), 409 | 412) => + { + PersistenceError::OptimisticLockError + } + other => PersistenceError::from(EventSourcingDbError::ClientError(other)), + }); + } + }; + + if let Some((snapshot_aggregate_id, aggregate, current_snapshot)) = snapshot_update { + let last_event_id = written_events.last().map(|event| event.id().to_string()); + let current_sequence = events.last().map(|event| event.sequence).unwrap_or(0); + + self.store_snapshot_record::(EventSourcingDbSnapshotRecord { + aggregate_type: A::TYPE.to_string(), + aggregate_id: snapshot_aggregate_id, + aggregate, + current_sequence, + current_snapshot, + last_event_id, + }) + .await?; + } + + Ok(()) } async fn stream_events( &self, - aggregate_id: &str, + _aggregate_id: &str, ) -> Result { todo!() } @@ -107,3 +340,57 @@ impl PersistedEventRepository for EventSourcingDbEventRepository { todo!() } } + +#[cfg(test)] +mod tests { + use super::*; + use serde_json::json; + + fn sample_event(id: &str, event_type: &str, payload: Value) -> Event { + serde_json::from_value(json!({ + "data": payload, + "datacontenttype": "application/json", + "hash": "hash", + "id": id, + "predecessorhash": "predecessor", + "source": "urn:test", + "specversion": "1.0", + "subject": "/BookAggregate/42", + "time": "2026-03-17T10:00:00Z", + "type": event_type, + "signature": null + })) + .expect("event JSON should deserialize") + } + + #[test] + fn map_subject_events_assigns_logical_sequences_after_skip() { + let domain = ReversedDomain::new(["io", "eventsourcingdb"]); + let events = vec![ + sample_event( + "01JAAA", + "io.eventsourcingdb.book-created", + wrap_event_data(json!({ "number": 1 }), json!({})), + ), + sample_event( + "01JAAB", + "io.eventsourcingdb.book-updated", + wrap_event_data(json!({ "number": 2 }), json!({})), + ), + sample_event( + "01JAAC", + "io.eventsourcingdb.book-updated", + wrap_event_data(json!({ "number": 3 }), json!({})), + ), + ]; + + let serialized = + EventSourcingDbEventRepository::map_subject_events(events, &domain, 1).unwrap(); + + assert_eq!(serialized.len(), 2); + assert_eq!(serialized[0].sequence, 2); + assert_eq!(serialized[0].payload, json!({ "number": 2 })); + assert_eq!(serialized[1].sequence, 3); + assert_eq!(serialized[1].payload, json!({ "number": 3 })); + } +} diff --git a/src/lib.rs b/src/lib.rs index 0f04534..3b0608c 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -2,18 +2,3 @@ pub mod conversion; pub mod errors; pub mod event_repository; pub mod types; - -pub fn add(left: u64, right: u64) -> u64 { - left + right -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn it_works() { - let result = add(2, 2); - assert_eq!(result, 4); - } -} From 317dd5da67669ae7de87b4bc531d5e1047716196 Mon Sep 17 00:00:00 2001 From: flangator Date: Tue, 17 Mar 2026 01:28:11 +0100 Subject: [PATCH 3/6] feat: support for streaming events --- Cargo.toml | 1 + src/conversion.rs | 2 +- src/event_repository.rs | 214 +++++++++++++++++++++++++++++++++++++--- 3 files changed, 202 insertions(+), 15 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index 2274761..44d680c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -16,3 +16,4 @@ futures = "0.3.32" serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.149" thiserror = "2.0.18" +tokio = "1.50.0" diff --git a/src/conversion.rs b/src/conversion.rs index b3aad52..27751f2 100644 --- a/src/conversion.rs +++ b/src/conversion.rs @@ -229,7 +229,7 @@ mod tests { "data": data, "datacontenttype": "application/json", "hash": "hash", - "id": "01JXYZOPAQUEID", + "id": "1", "predecessorhash": "predecessor", "source": "urn:test", "specversion": "1.0", diff --git a/src/event_repository.rs b/src/event_repository.rs index 3877406..f4c3f5e 100644 --- a/src/event_repository.rs +++ b/src/event_repository.rs @@ -1,4 +1,4 @@ -use std::sync::Arc; +use std::{collections::HashMap, sync::Arc}; use cqrs_es::{ Aggregate, @@ -11,7 +11,7 @@ use eventsourcingdb::{ Client, Event, EventCandidate, Precondition, request_options::{Bound, BoundType, Ordering, ReadEventsOptions}, }; -use futures::TryStreamExt; +use futures::{StreamExt, TryStreamExt}; use serde::{Deserialize, Serialize}; use serde_json::Value; @@ -20,9 +20,12 @@ use crate::{ errors::{EventSourcingDbError, EventSourcingDbResult}, }; +//TODO: maybe not the best idea to hardcode that const SNAPSHOT_EVENT_TYPE: &str = "io.eventsourcingdb.cqrs-es.snapshot-record.v1"; const SNAPSHOT_EVENT_SOURCE: &str = "urn:eventsourcingdb-es:snapshot"; const EVENT_SOURCE: &str = "urn:eventsourcingdb-es:event"; +//TODO: make part of struct +const STREAM_CHANNEL_SIZE: usize = 2048; #[derive(Debug, Clone, Serialize, Deserialize)] struct EventSourcingDbSnapshotRecord { @@ -58,6 +61,41 @@ impl EventSourcingDbEventRepository { format!("/__snapshots__/{}/{}", &A::TYPE, id) } + // We enforce the convention /aggregate_type/aggregate_id hard. Every deviation is + // treated as error. + fn aggregate_id_from_subject(subject: &str) -> EventSourcingDbResult { + let mut segments = subject.split('/').filter(|segment| !segment.is_empty()); + let aggregate_type = segments + .next() + .ok_or_else(|| EventSourcingDbError::InvalidSubject(subject.to_string()))?; + let aggregate_id = segments + .next() + .ok_or_else(|| EventSourcingDbError::InvalidSubject(subject.to_string()))?; + + if segments.next().is_some() { + return Err(EventSourcingDbError::InvalidSubject(subject.to_string())); + } + + if aggregate_type.is_empty() || aggregate_id.is_empty() { + return Err(EventSourcingDbError::InvalidSubject(subject.to_string())); + } + + Ok(aggregate_id.to_string()) + } + + fn serialize_stream_event( + event: Event, + sequence: usize, + ) -> Result { + map_event(event, sequence) + .map(|(serialized, _)| serialized) + .map_err(Into::into) + } + + fn stream_client_error(err: eventsourcingdb::error::ClientError) -> PersistenceError { + EventSourcingDbError::ClientError(err).into() + } + async fn read_subject_events( client: Arc, aggregate_id: String, @@ -128,12 +166,12 @@ impl EventSourcingDbEventRepository { ) .await?; - let Some(event) = stream.try_next().await? else { - return Ok(None); - }; - - let record = serde_json::from_value(event.data().clone())?; - Ok(Some(record)) + let record: Option = stream + .try_next() + .await? + .map(|event| serde_json::from_value(event.data().clone())) + .transpose()?; + Ok(record) } async fn read_events_since_snapshot( @@ -165,6 +203,7 @@ impl EventSourcingDbEventRepository { &self, aggregate_id: &str, ) -> EventSourcingDbResult { + //TODO: use a clever eventql query here instead of reading the complete stream let events = Self::read_subject_events::(Arc::clone(&self.client), aggregate_id.to_string()) .await?; @@ -299,6 +338,7 @@ impl PersistedEventRepository for EventSourcingDbEventRepository { Ok(events) => events, Err(err) => { return Err(match err { + //TODO: make part of error conversion eventsourcingdb::error::ClientError::DBApiError(status, _) if matches!(status.as_u16(), 409 | 412) => { @@ -329,21 +369,114 @@ impl PersistedEventRepository for EventSourcingDbEventRepository { async fn stream_events( &self, - _aggregate_id: &str, + aggregate_id: &str, ) -> Result { - todo!() + let client = Arc::clone(&self.client); + let subject = Self::get_subject::(&aggregate_id); + let (mut feed, stream) = ReplayStream::new(STREAM_CHANNEL_SIZE); + + tokio::spawn(async move { + let mut event_stream = match client + .read_events( + &subject, + Some(ReadEventsOptions { + order: Some(Ordering::Chronological), + ..Default::default() + }), + ) + .await + { + Ok(stream) => stream, + Err(err) => { + let _ = feed.push(Err(Self::stream_client_error(err))).await; + return; + } + }; + let mut sequence = 0usize; + + while let Some(result) = event_stream.next().await { + let mapped = match result { + Ok(event) => { + sequence += 1; + Self::serialize_stream_event(event, sequence) + } + Err(err) => Err(Self::stream_client_error(err)), + }; + + if feed.push(mapped).await.is_err() { + break; + } + } + }); + Ok(stream) } async fn stream_all_events( &self, ) -> Result { - todo!() + let client = Arc::clone(&self.client); + + // query all aggregates of this type using the root aggregate subject. + let subject = format!("/{}", A::TYPE); + + let (mut feed, stream) = ReplayStream::new(STREAM_CHANNEL_SIZE); + + tokio::spawn(async move { + let mut event_stream = match client + .read_events( + &subject, + Some(ReadEventsOptions { + recursive: true, + order: Some(Ordering::Chronological), + ..Default::default() + }), + ) + .await + { + Ok(stream) => stream, + Err(err) => { + let _ = feed.push(Err(Self::stream_client_error(err))).await; + return; + } + }; + + let mut sequences: HashMap = HashMap::new(); + + while let Some(result) = event_stream.next().await { + let mapped = match result { + Ok(event) => { + let sequence = match Self::aggregate_id_from_subject(event.subject()) { + Ok(aggregate_id) => { + let next_sequence = sequences.entry(aggregate_id).or_insert(0); + *next_sequence += 1; + *next_sequence + } + Err(err) => { + if feed.push(Err(err.into())).await.is_err() { + break; + } + continue; + } + }; + + Self::serialize_stream_event(event, sequence) + } + Err(err) => Err(Self::stream_client_error(err)), + }; + + if feed.push(mapped).await.is_err() { + break; + } + } + }); + Ok(stream) } } #[cfg(test)] mod tests { use super::*; + use cqrs_es::persist::PersistenceError; use serde_json::json; fn sample_event(id: &str, event_type: &str, payload: Value) -> Event { @@ -368,17 +501,17 @@ mod tests { let domain = ReversedDomain::new(["io", "eventsourcingdb"]); let events = vec![ sample_event( - "01JAAA", + "1", "io.eventsourcingdb.book-created", wrap_event_data(json!({ "number": 1 }), json!({})), ), sample_event( - "01JAAB", + "2", "io.eventsourcingdb.book-updated", wrap_event_data(json!({ "number": 2 }), json!({})), ), sample_event( - "01JAAC", + "3", "io.eventsourcingdb.book-updated", wrap_event_data(json!({ "number": 3 }), json!({})), ), @@ -393,4 +526,57 @@ mod tests { assert_eq!(serialized[1].sequence, 3); assert_eq!(serialized[1].payload, json!({ "number": 3 })); } + + #[test] + fn aggregate_id_from_subject_requires_exact_aggregate_subject() { + let aggregate_id = + EventSourcingDbEventRepository::aggregate_id_from_subject("/BookAggregate/42") + .expect("subject should parse"); + + assert_eq!(aggregate_id, "42"); + assert!(matches!( + EventSourcingDbEventRepository::aggregate_id_from_subject("/BookAggregate/42/chapter"), + Err(EventSourcingDbError::InvalidSubject(_)) + )); + } + + #[test] + fn serialize_stream_event_preserves_supplied_sequence() { + let event = sample_event( + "1", + "io.eventsourcingdb.book-created.v2", + wrap_event_data(json!({ "number": 1 }), json!({ "request_id": "abc" })), + ); + + let serialized = EventSourcingDbEventRepository::serialize_stream_event(event, 5) + .expect("event should serialize"); + + assert_eq!(serialized.sequence, 5); + assert_eq!(serialized.aggregate_id, "42"); + assert_eq!(serialized.event_type, "BookCreated"); + assert_eq!(serialized.event_version, "2"); + } + + #[test] + fn serialize_stream_event_converts_mapping_failures_to_persistence_errors() { + let invalid_event: Event = serde_json::from_value(json!({ + "data": json!({}), + "datacontenttype": "application/json", + "hash": "hash", + "id": "1", + "predecessorhash": "predecessor", + "source": "urn:test", + "specversion": "1.0", + "subject": "/BookAggregate/42", + "time": "2026-03-17T10:00:00Z", + "type": "invalid", + "signature": null + })) + .expect("event JSON should deserialize"); + + let err = EventSourcingDbEventRepository::serialize_stream_event(invalid_event, 1) + .expect_err("invalid event type should fail"); + + assert!(matches!(err, PersistenceError::UnknownError(_))); + } } From d7a410c6ed94e8660d6ce817ab08572c6951b7e0 Mon Sep 17 00:00:00 2001 From: flangator Date: Tue, 17 Mar 2026 19:40:58 +0100 Subject: [PATCH 4/6] feat: example, integration test and proper error mapping --- Cargo.toml | 2 + README.md | 13 +- examples/banking.rs | 319 +++++++++++++++++++++++++++++++++++++ src/conversion.rs | 103 ++++++++++-- src/cqrs.rs | 22 +++ src/errors.rs | 76 ++++++++- src/event_repository.rs | 24 +-- src/lib.rs | 1 + tests/lib.rs | 341 ++++++++++++++++++++++++++++++++++++++++ 9 files changed, 866 insertions(+), 35 deletions(-) create mode 100644 examples/banking.rs create mode 100644 src/cqrs.rs create mode 100644 tests/lib.rs diff --git a/Cargo.toml b/Cargo.toml index 44d680c..84d8308 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -9,6 +9,7 @@ license = "Apache-2.0" keywords = ["cqrs", "event-sourcing", "eventsourcingdb", "cqrs-es"] [dependencies] +async-trait = "0.1.89" chrono = { version = "0.4.44", features = ["serde"] } cqrs-es = "0.5.0" eventsourcingdb = "2.0.1" @@ -17,3 +18,4 @@ serde = { version = "1.0.228", features = ["derive"] } serde_json = "1.0.149" thiserror = "2.0.18" tokio = "1.50.0" +url = "2.5.8" diff --git a/README.md b/README.md index 67694ae..96cbe2f 100644 --- a/README.md +++ b/README.md @@ -6,12 +6,23 @@ An [EventSourcingDB](https://docs.eventsourcingdb.io) implementation of the `Per --- +## Conventions and assumptions + +This adapter follows the naming guidance from the [EventSourcingDB documentation](https://docs.eventsourcingdb.io/) and assumes the following storage conventions: + +- Subjects use the form `//`, for example `/books/42`. +- Event types in EventSourcingDB use reverse-domain notation, kebab-case names, and an explicit major version suffix, for example `io.eventsourcingdb.library.book-acquired.v1`. +- Event type names are mapped to `cqrs-es` event names by converting between kebab-case and PascalCase. +- Only major event versions are stored in EventSourcingDB (`v1`, `v2`, ...). On the `cqrs-es` side they are exposed as `1.0`, `2.0`, and so on. +- Legacy unversioned event types are still read as version `1.0`, but new writes always use an explicit `.v1` suffix. +- Event payload and metadata are stored in an adapter envelope so `cqrs-es` metadata survives round-trips. + ## Usage Add the following to your `Cargo.toml`: ```toml [dependencies] -cqrs-es = "0.4.12" +cqrs-es = "0.5" eventsourcingdb-es = "0.1.0" ``` diff --git a/examples/banking.rs b/examples/banking.rs new file mode 100644 index 0000000..83c6951 --- /dev/null +++ b/examples/banking.rs @@ -0,0 +1,319 @@ +use std::{ + fmt::{Display, Formatter}, + sync::{Arc, RwLock}, +}; + +use async_trait::async_trait; +use cqrs_es::{Aggregate, DomainEvent, EventEnvelope, Query, event_sink::EventSink}; +use eventsourcingdb_es::{conversion::ReversedDomain, cqrs::esdb_cqrs, types::EventSourcingDbCqrs}; +use serde::{Deserialize, Serialize}; + +#[derive(Debug, Serialize, Deserialize)] +pub enum BankAccountCommand { + OpenAccount { account_id: String }, + DepositMoney { amount: f64 }, + WithdrawMoney { amount: f64, atm_id: String }, + WriteCheck { check_number: String, amount: f64 }, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +pub enum BankAccountEvent { + AccountOpened { + account_id: String, + }, + CustomerDepositedMoney { + amount: f64, + balance: f64, + }, + CustomerWithdrewCash { + amount: f64, + balance: f64, + }, + CustomerWroteCheck { + check_number: String, + amount: f64, + balance: f64, + }, +} + +impl DomainEvent for BankAccountEvent { + fn event_type(&self) -> String { + let event_type: &str = match self { + BankAccountEvent::AccountOpened { .. } => "AccountOpened", + BankAccountEvent::CustomerDepositedMoney { .. } => "CustomerDepositedMoney", + BankAccountEvent::CustomerWithdrewCash { .. } => "CustomerWithdrewCash", + BankAccountEvent::CustomerWroteCheck { .. } => "CustomerWroteCheck", + }; + event_type.to_string() + } + + fn event_version(&self) -> String { + "1.0".to_string() + } +} + +#[derive(Debug)] +pub struct BankAccountError(String); + +impl Display for BankAccountError { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + write!(f, "{}", self.0) + } +} + +impl std::error::Error for BankAccountError {} + +impl From<&str> for BankAccountError { + fn from(message: &str) -> Self { + BankAccountError(message.to_string()) + } +} + +pub struct BankAccountServices; + +impl BankAccountServices { + async fn atm_withdrawal(&self, _atm_id: &str, _amount: f64) -> Result<(), AtmError> { + Ok(()) + } + + async fn validate_check(&self, _account: &str, _check: &str) -> Result<(), CheckingError> { + Ok(()) + } +} +pub struct AtmError; +pub struct CheckingError; + +#[derive(Serialize, Default, Deserialize)] +pub struct BankAccount { + account_id: String, + // this is a floating point for our example, don't do this IRL + balance: f64, +} + +impl Aggregate for BankAccount { + // This identifier should be unique to the system. + const TYPE: &'static str = "account"; + type Command = BankAccountCommand; + type Event = BankAccountEvent; + type Error = BankAccountError; + type Services = BankAccountServices; + + // The aggregate logic goes here. Note that this will be the _bulk_ of a CQRS system + // so expect to use helper functions elsewhere to keep the code clean. + async fn handle( + &mut self, + command: Self::Command, + services: &Self::Services, + sink: &EventSink, + ) -> Result<(), Self::Error> { + match command { + BankAccountCommand::OpenAccount { account_id } => { + sink.write(BankAccountEvent::AccountOpened { account_id }, self) + .await; + } + BankAccountCommand::DepositMoney { amount } => { + let balance = self.balance + amount; + sink.write( + BankAccountEvent::CustomerDepositedMoney { amount, balance }, + self, + ) + .await; + } + BankAccountCommand::WithdrawMoney { amount, atm_id } => { + let balance = self.balance - amount; + if balance < 0_f64 { + return Err("funds not available".into()); + } + if services.atm_withdrawal(&atm_id, amount).await.is_err() { + return Err("atm rule violation".into()); + } + sink.write( + BankAccountEvent::CustomerWithdrewCash { amount, balance }, + self, + ) + .await; + } + BankAccountCommand::WriteCheck { + check_number, + amount, + } => { + let balance = self.balance - amount; + if balance < 0_f64 { + return Err("funds not available".into()); + } + if services + .validate_check(&self.account_id, &check_number) + .await + .is_err() + { + return Err("check invalid".into()); + } + sink.write( + BankAccountEvent::CustomerWroteCheck { + check_number, + amount, + balance, + }, + self, + ) + .await; + } + }; + Ok(()) + } + + fn apply(&mut self, event: Self::Event) { + match event { + BankAccountEvent::AccountOpened { account_id } => { + self.account_id = account_id; + } + BankAccountEvent::CustomerDepositedMoney { amount: _, balance } + | BankAccountEvent::CustomerWithdrewCash { amount: _, balance } + | BankAccountEvent::CustomerWroteCheck { + check_number: _, + amount: _, + balance, + } => { + self.balance = balance; + } + } + } +} + +#[derive(Debug, Default, Serialize, Deserialize)] +pub struct BankAccountView { + account_id: String, + balance: f64, + activity: Vec, +} + +#[derive(Clone)] +pub struct BankAccountQuery { + view: Arc>, +} + +impl BankAccountQuery { + fn new(view: Arc>) -> Self { + Self { view } + } +} + +#[async_trait] +impl Query for BankAccountQuery { + async fn dispatch(&self, aggregate_id: &str, events: &[EventEnvelope]) { + let mut view = self.view.write().expect("bank account view lock poisoned"); + + for event in events { + view.account_id = aggregate_id.to_string(); + + match &event.payload { + BankAccountEvent::AccountOpened { account_id } => { + view.account_id = account_id.clone(); + view.activity + .push(format!("opened account {account_id}")); + } + BankAccountEvent::CustomerDepositedMoney { amount, balance } => { + view.balance = *balance; + view.activity + .push(format!("deposited {amount:.2}, balance is now {balance:.2}")); + } + BankAccountEvent::CustomerWithdrewCash { amount, balance } => { + view.balance = *balance; + view.activity + .push(format!("withdrew {amount:.2}, balance is now {balance:.2}")); + } + BankAccountEvent::CustomerWroteCheck { + check_number, + amount, + balance, + } => { + view.balance = *balance; + view.activity.push(format!( + "wrote check {check_number} for {amount:.2}, balance is now {balance:.2}" + )); + } + } + } + } +} + +async fn execute_command( + cqrs: &EventSourcingDbCqrs, + aggregate_id: &str, + label: &str, + command: BankAccountCommand, +) { + cqrs.execute(aggregate_id, command) + .await + .unwrap_or_else(|err| panic!("{label} failed: {err}")); + println!("executed: {label}"); +} + +#[tokio::main(flavor = "current_thread")] +async fn main() { + let client = eventsourcingdb::Client::new( + url::Url::parse("http://localhost:3000").unwrap(), + "secret".to_string(), + ); + + let domain = ReversedDomain::new(vec!["com", "flangator", "banking"]); + let aggregate_id = "account-0001"; + let view = Arc::new(RwLock::new(BankAccountView::default())); + let query = BankAccountQuery::new(Arc::clone(&view)); + let cqrs = esdb_cqrs::(client, domain, vec![Box::new(query)], BankAccountServices); + + execute_command( + &cqrs, + aggregate_id, + "open account", + BankAccountCommand::OpenAccount { + account_id: aggregate_id.to_string(), + }, + ) + .await; + execute_command( + &cqrs, + aggregate_id, + "deposit 250.00", + BankAccountCommand::DepositMoney { amount: 250.0 }, + ) + .await; + execute_command( + &cqrs, + aggregate_id, + "withdraw 40.00 from ATM-7", + BankAccountCommand::WithdrawMoney { + amount: 40.0, + atm_id: "ATM-7".to_string(), + }, + ) + .await; + execute_command( + &cqrs, + aggregate_id, + "write check CHK-1001 for 25.00", + BankAccountCommand::WriteCheck { + check_number: "CHK-1001".to_string(), + amount: 25.0, + }, + ) + .await; + execute_command( + &cqrs, + aggregate_id, + "deposit 10.00", + BankAccountCommand::DepositMoney { amount: 10.0 }, + ) + .await; + + let view = view.read().expect("bank account view lock poisoned"); + + println!(); + println!("projection for {}:", view.account_id); + println!("final balance: {:.2}", view.balance); + println!("expected final balance: 195.00"); + println!("activity:"); + for entry in &view.activity { + println!(" - {entry}"); + } +} diff --git a/src/conversion.rs b/src/conversion.rs index 27751f2..f970d0e 100644 --- a/src/conversion.rs +++ b/src/conversion.rs @@ -83,6 +83,32 @@ pub struct QualifiedEventType { pub event_version: Option, } +fn parse_major_version_str(value: &str) -> Result { + if let Some((major, minor)) = value.split_once('.') { + if minor != "0" { + return Err(EventSourcingDbError::InvalidEventVersion(value.into())); + } + + let major: usize = major.parse()?; + if major == 0 { + return Err(EventSourcingDbError::InvalidEventVersion(value.into())); + } + + return Ok(major); + } + + let major: usize = value.parse()?; + if major == 0 { + return Err(EventSourcingDbError::InvalidEventVersion(value.into())); + } + + Ok(major) +} + +fn format_cqrs_event_version(major: usize) -> String { + format!("{major}.0") +} + impl FromStr for QualifiedEventType { type Err = EventSourcingDbError; @@ -157,15 +183,13 @@ pub fn qualify_event_type( domain: &ReversedDomain, event_type: &str, event_version: &str, -) -> String { +) -> EventSourcingDbResult { + let major = parse_major_version_str(event_version)?; let mut parts: Vec = domain.labels().to_vec(); parts.push(pascal_to_kebab_case(event_type)); + parts.push(format!("v{major}")); - if event_version != "1" { - parts.push(format!("v{event_version}")); - } - - parts.join(".") + Ok(parts.join(".")) } pub fn wrap_event_data(payload: Value, metadata: Value) -> Value { @@ -210,8 +234,8 @@ pub fn map_event( sequence, aggregate_type, event_type: to_pascal_case(&event_type.event_type), - // a non-existing version defaults to 1 - event_version: event_type.event_version.unwrap_or(1 as usize).to_string(), + // a non-existing version defaults to legacy v1. + event_version: format_cqrs_event_version(event_type.event_version.unwrap_or(1)), payload, metadata, }; @@ -244,11 +268,47 @@ mod tests { #[test] fn qualify_event_type_uses_domain_and_kebab_case() { let domain = ReversedDomain::new(["io", "eventsourcingdb"]); - let qualified = qualify_event_type(&domain, "BookCreated", "2"); + let qualified = qualify_event_type(&domain, "BookCreated", "2.0").unwrap(); assert_eq!(qualified, "io.eventsourcingdb.book-created.v2"); } + #[test] + fn qualify_event_type_normalizes_v1_and_major_only_versions() { + let domain = ReversedDomain::new(["io", "eventsourcingdb"]); + + assert_eq!( + qualify_event_type(&domain, "BookCreated", "1").unwrap(), + "io.eventsourcingdb.book-created.v1" + ); + assert_eq!( + qualify_event_type(&domain, "BookCreated", "1.0").unwrap(), + "io.eventsourcingdb.book-created.v1" + ); + assert_eq!( + qualify_event_type(&domain, "BookCreated", "2").unwrap(), + "io.eventsourcingdb.book-created.v2" + ); + } + + #[test] + fn qualify_event_type_rejects_non_major_versions() { + let domain = ReversedDomain::new(["io", "eventsourcingdb"]); + + assert!(matches!( + qualify_event_type(&domain, "BookCreated", "0"), + Err(EventSourcingDbError::InvalidEventVersion(_)) + )); + assert!(matches!( + qualify_event_type(&domain, "BookCreated", "1.1"), + Err(EventSourcingDbError::InvalidEventVersion(_)) + )); + assert!(matches!( + qualify_event_type(&domain, "BookCreated", "1.2.3"), + Err(EventSourcingDbError::InvalidEventVersion(_)) + )); + } + #[test] fn wrap_and_unwrap_event_data_round_trip_payload_and_metadata() { let payload = json!({ "title": "DDD" }); @@ -283,9 +343,32 @@ mod tests { assert_eq!(serialized.aggregate_id, "42"); assert_eq!(serialized.aggregate_type, "BookAggregate"); assert_eq!(serialized.event_type, "BookCreated"); - assert_eq!(serialized.event_version, "2"); + assert_eq!(serialized.event_version, "2.0"); assert_eq!(serialized.payload, json!({ "title": "DDD" })); assert_eq!(serialized.metadata, json!({ "request_id": "abc-123" })); assert_eq!(meta.subject, "/BookAggregate/42"); } + + #[test] + fn map_event_treats_legacy_unversioned_event_type_as_v1() { + let event: Event = serde_json::from_value(json!({ + "data": wrap_event_data(json!({ "title": "DDD" }), json!({})), + "datacontenttype": "application/json", + "hash": "hash", + "id": "1", + "predecessorhash": "predecessor", + "source": "urn:test", + "specversion": "1.0", + "subject": "/BookAggregate/42", + "time": "2026-03-17T10:00:00Z", + "type": "io.eventsourcingdb.book-created", + "signature": null + })) + .expect("event JSON should deserialize"); + + let (serialized, _) = map_event(event, 1).expect("legacy event should map"); + + assert_eq!(serialized.event_type, "BookCreated"); + assert_eq!(serialized.event_version, "1.0"); + } } diff --git a/src/cqrs.rs b/src/cqrs.rs new file mode 100644 index 0000000..1718a18 --- /dev/null +++ b/src/cqrs.rs @@ -0,0 +1,22 @@ +use std::sync::Arc; + +use cqrs_es::{Aggregate, CqrsFramework, Query, persist::PersistedEventStore}; + +use crate::{ + conversion::ReversedDomain, event_repository::EventSourcingDbEventRepository, + types::EventSourcingDbCqrs, +}; + +pub fn esdb_cqrs( + client: eventsourcingdb::Client, + domain: ReversedDomain, + query_processor: Vec>>, + services: A::Services, +) -> EventSourcingDbCqrs +where + A: Aggregate, +{ + let repo = EventSourcingDbEventRepository::new(Arc::new(client), domain); + let store = PersistedEventStore::new_event_store(repo); + CqrsFramework::new(store, query_processor, services) +} diff --git a/src/errors.rs b/src/errors.rs index 7b454b2..14ade56 100644 --- a/src/errors.rs +++ b/src/errors.rs @@ -10,23 +10,87 @@ pub enum EventSourcingDbError { InvalidEventTypeIdentifier(String), #[error("invalid subject: {0}")] InvalidSubject(String), - #[error("invalid sequence number: {0}")] + #[error("invalid numeric suffix: {0}")] InvalidSequence(#[from] std::num::ParseIntError), - //TODO: this can be unmangled/ matched, especially OptimisticLocking and serde + #[error("invalid event version: {0}")] + InvalidEventVersion(String), #[error("{0}")] - ClientError(#[from] ClientError), + Client(ClientError), + #[error("optimistic lock error")] + OptimisticLock, } pub type EventSourcingDbResult = Result; +impl From for EventSourcingDbError { + fn from(value: ClientError) -> Self { + match value { + ClientError::DBApiError(status, _) if status.as_u16() == 409 => { + Self::OptimisticLock + } + other => Self::Client(other), + } + } +} + impl From for PersistenceError { - //TODO: this conversion can be improved fn from(value: EventSourcingDbError) -> Self { match value { - EventSourcingDbError::ClientError(err) => { - PersistenceError::ConnectionError(Box::new(err)) + EventSourcingDbError::OptimisticLock => PersistenceError::OptimisticLockError, + EventSourcingDbError::Client(err) => match err { + ClientError::IoError(_) + | ClientError::ReqwestError(_) + | ClientError::URLParseError(_) + | ClientError::PingFailed => PersistenceError::ConnectionError(Box::new(err)), + ClientError::SerdeJsonError(_) | ClientError::InvalidResponseType(_) => { + PersistenceError::DeserializationError(Box::new(err)) + } + _ => PersistenceError::UnknownError(Box::new(err)), } _ => PersistenceError::UnknownError(Box::new(value)), } } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn client_conflict_maps_to_optimistic_lock() { + let err = EventSourcingDbError::from(ClientError::DBApiError( + "409".parse().expect("409 should parse as status code"), + "conflict".to_string(), + )); + + assert!(matches!(err, EventSourcingDbError::OptimisticLock)); + assert!(matches!( + PersistenceError::from(err), + PersistenceError::OptimisticLockError + )); + } + + #[test] + fn non_conflict_db_api_error_maps_to_unknown_error() { + let err = EventSourcingDbError::from(ClientError::DBApiError( + "400".parse().expect("400 should parse as status code"), + "bad request".to_string(), + )); + + assert!(matches!(err, EventSourcingDbError::Client(_))); + assert!(matches!( + PersistenceError::from(err), + PersistenceError::UnknownError(_) + )); + } + + #[test] + fn invalid_response_type_maps_to_deserialization_error() { + let err = EventSourcingDbError::from(ClientError::InvalidResponseType("text/html".into())); + + assert!(matches!( + PersistenceError::from(err), + PersistenceError::DeserializationError(_) + )); + } +} diff --git a/src/event_repository.rs b/src/event_repository.rs index f4c3f5e..34abcf7 100644 --- a/src/event_repository.rs +++ b/src/event_repository.rs @@ -93,7 +93,7 @@ impl EventSourcingDbEventRepository { } fn stream_client_error(err: eventsourcingdb::error::ClientError) -> PersistenceError { - EventSourcingDbError::ClientError(err).into() + EventSourcingDbError::from(err).into() } async fn read_subject_events( @@ -220,14 +220,12 @@ impl EventSourcingDbEventRepository { &self, event: &SerializedEvent, ) -> EventSourcingDbResult { + let event_type = qualify_event_type(&self.domain, &event.event_type, &event.event_version)?; + Ok(EventCandidate::builder() .source(EVENT_SOURCE.to_string()) .subject(Self::get_subject::(&event.aggregate_id)) - .ty(qualify_event_type( - &self.domain, - &event.event_type, - &event.event_version, - )) + .ty(event_type) .data(wrap_event_data( event.payload.clone(), event.metadata.clone(), @@ -336,17 +334,7 @@ impl PersistedEventRepository for EventSourcingDbEventRepository { .await { Ok(events) => events, - Err(err) => { - return Err(match err { - //TODO: make part of error conversion - eventsourcingdb::error::ClientError::DBApiError(status, _) - if matches!(status.as_u16(), 409 | 412) => - { - PersistenceError::OptimisticLockError - } - other => PersistenceError::from(EventSourcingDbError::ClientError(other)), - }); - } + Err(err) => return Err(PersistenceError::from(EventSourcingDbError::from(err))), }; if let Some((snapshot_aggregate_id, aggregate, current_snapshot)) = snapshot_update { @@ -554,7 +542,7 @@ mod tests { assert_eq!(serialized.sequence, 5); assert_eq!(serialized.aggregate_id, "42"); assert_eq!(serialized.event_type, "BookCreated"); - assert_eq!(serialized.event_version, "2"); + assert_eq!(serialized.event_version, "2.0"); } #[test] diff --git a/src/lib.rs b/src/lib.rs index 3b0608c..caef90e 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -1,4 +1,5 @@ pub mod conversion; +pub mod cqrs; pub mod errors; pub mod event_repository; pub mod types; diff --git a/tests/lib.rs b/tests/lib.rs new file mode 100644 index 0000000..155d0e8 --- /dev/null +++ b/tests/lib.rs @@ -0,0 +1,341 @@ +use std::{ + sync::{ + Arc, + atomic::{AtomicUsize, Ordering}, + }, + time::{SystemTime, UNIX_EPOCH}, +}; + +use cqrs_es::{ + Aggregate, DomainEvent, + EventStore, + doc::{Customer, CustomerEvent}, + persist::{PersistedEventRepository, PersistedEventStore, PersistenceError, SerializedEvent}, +}; +use eventsourcingdb::request_options::{Ordering as EventOrdering, ReadEventsOptions}; +use eventsourcingdb_es::{ + conversion::ReversedDomain, event_repository::EventSourcingDbEventRepository, +}; +use futures::TryStreamExt; +use serde_json::{Value, json}; + +static TEST_ID_COUNTER: AtomicUsize = AtomicUsize::new(0); + +pub fn esdb_client() -> eventsourcingdb::Client { + eventsourcingdb::Client::new( + url::Url::parse("http://localhost:3000").unwrap(), + "secret".to_string(), + ) +} + +fn new_repository() -> EventSourcingDbEventRepository { + EventSourcingDbEventRepository::new( + Arc::new(esdb_client()), + ReversedDomain::new(["io", "eventsourcingdb"]), + ) +} + +fn new_snapshot_store() -> PersistedEventStore { + PersistedEventStore::new_snapshot_store(new_repository(), 2) +} + +fn unique_aggregate_id(prefix: &str) -> String { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .expect("system time should be after unix epoch") + .as_nanos(); + let counter = TEST_ID_COUNTER.fetch_add(1, Ordering::Relaxed); + format!("{prefix}-{nanos}-{counter}") +} + +fn serialized_customer_event( + aggregate_id: &str, + sequence: usize, + event: CustomerEvent, + metadata: Value, +) -> SerializedEvent { + SerializedEvent::new( + aggregate_id.to_string(), + sequence, + Customer::TYPE.to_string(), + event.event_type(), + event.event_version(), + serde_json::to_value(event).expect("customer event should serialize"), + metadata, + ) +} + +#[tokio::test] +async fn persist_and_load_events_round_trip() { + let repository = new_repository(); + let aggregate_id = unique_aggregate_id("persist-and-load"); + + assert_eq!( + 0, + repository + .get_events::(&aggregate_id) + .await + .expect("loading empty stream should succeed") + .len() + ); + + let first_batch = vec![ + serialized_customer_event( + &aggregate_id, + 1, + CustomerEvent::NameAdded { + name: "Alice".to_string(), + }, + json!({ "request_id": "initial-name" }), + ), + serialized_customer_event( + &aggregate_id, + 2, + CustomerEvent::EmailUpdated { + new_email: "alice@example.com".to_string(), + }, + json!({ "request_id": "initial-email" }), + ), + ]; + + repository + .persist::(&first_batch, None) + .await + .expect("initial events should persist"); + + let loaded_after_first_commit = repository + .get_events::(&aggregate_id) + .await + .expect("events should load after first commit"); + + assert_eq!(2, loaded_after_first_commit.len()); + assert_eq!(1, loaded_after_first_commit[0].sequence); + assert_eq!(2, loaded_after_first_commit[1].sequence); + assert_eq!("NameAdded", loaded_after_first_commit[0].event_type); + assert_eq!("1.0", loaded_after_first_commit[0].event_version); + assert_eq!("EmailUpdated", loaded_after_first_commit[1].event_type); + assert_eq!("1.0", loaded_after_first_commit[1].event_version); + assert_eq!( + json!({ "NameAdded": { "name": "Alice" } }), + loaded_after_first_commit[0].payload + ); + assert_eq!( + json!({ "EmailUpdated": { "new_email": "alice@example.com" } }), + loaded_after_first_commit[1].payload + ); + + let second_batch = vec![serialized_customer_event( + &aggregate_id, + 3, + CustomerEvent::EmailUpdated { + new_email: "alice+1@example.com".to_string(), + }, + json!({ "request_id": "follow-up-email" }), + )]; + + repository + .persist::(&second_batch, None) + .await + .expect("follow-up event should persist"); + + let all_events = repository + .get_events::(&aggregate_id) + .await + .expect("full history should load"); + + assert_eq!(3, all_events.len()); + assert_eq!(3, all_events[2].sequence); + assert_eq!("EmailUpdated", all_events[2].event_type); + assert_eq!("1.0", all_events[2].event_version); + assert_eq!( + json!({ "EmailUpdated": { "new_email": "alice+1@example.com" } }), + all_events[2].payload + ); + + let last_events = repository + .get_last_events::(&aggregate_id, 2) + .await + .expect("tail events should load"); + + assert_eq!(1, last_events.len()); + assert_eq!(3, last_events[0].sequence); + assert_eq!("EmailUpdated", last_events[0].event_type); + assert_eq!("1.0", last_events[0].event_version); + assert_eq!( + json!({ "EmailUpdated": { "new_email": "alice+1@example.com" } }), + last_events[0].payload + ); +} + +#[tokio::test] +async fn persist_rejects_stale_sequence_numbers() { + let repository = new_repository(); + let aggregate_id = unique_aggregate_id("optimistic-lock"); + + let initial_events = vec![serialized_customer_event( + &aggregate_id, + 1, + CustomerEvent::NameAdded { + name: "Bob".to_string(), + }, + json!({ "request_id": "create-name" }), + )]; + + repository + .persist::(&initial_events, None) + .await + .expect("initial event should persist"); + + let stale_events = vec![serialized_customer_event( + &aggregate_id, + 1, + CustomerEvent::EmailUpdated { + new_email: "bob@example.com".to_string(), + }, + json!({ "request_id": "stale-update" }), + )]; + + let err = repository + .persist::(&stale_events, None) + .await + .expect_err("stale event sequence should fail"); + + assert!(matches!(err, PersistenceError::OptimisticLockError)); +} + +#[tokio::test] +async fn snapshot_store_persists_snapshots_and_loads_tail_events() { + let repository = new_repository(); + let event_store = new_snapshot_store(); + let aggregate_id = unique_aggregate_id("snapshot-store"); + + let initial_context = event_store + .load_aggregate(&aggregate_id) + .await + .expect("loading empty aggregate should succeed"); + + event_store + .commit( + vec![ + CustomerEvent::NameAdded { + name: "Carol".to_string(), + }, + CustomerEvent::EmailUpdated { + new_email: "carol@example.com".to_string(), + } + ], + initial_context, + Default::default(), + ) + .await + .expect("first commit should write events and snapshot"); + + let snapshot = repository + .get_snapshot::(&aggregate_id) + .await + .expect("snapshot lookup should succeed") + .expect("snapshot should exist after threshold is reached"); + + assert_eq!(aggregate_id, snapshot.aggregate_id); + assert_eq!(2, snapshot.current_sequence); + assert_eq!(1, snapshot.current_snapshot); + assert_eq!( + json!({ + "customer_id": "", + "name": "Carol", + "email": "carol@example.com", + "data_populated": false + }), + snapshot.aggregate + ); + + let context_after_snapshot = event_store + .load_aggregate(&aggregate_id) + .await + .expect("aggregate should load from snapshot"); + + assert_eq!(2, context_after_snapshot.current_sequence); + assert_eq!(Some(1), context_after_snapshot.current_snapshot); + assert_eq!("Carol", context_after_snapshot.aggregate.name); + assert_eq!("carol@example.com", context_after_snapshot.aggregate.email); + + event_store + .commit( + vec![CustomerEvent::EmailUpdated { + new_email: "carol+1@example.com".to_string(), + }], + context_after_snapshot, + Default::default(), + ) + .await + .expect("tail event should persist after snapshot"); + + let tail_events = repository + .get_last_events::(&aggregate_id, snapshot.current_sequence) + .await + .expect("tail events should load from snapshot boundary"); + + assert_eq!(1, tail_events.len()); + assert_eq!(3, tail_events[0].sequence); + assert_eq!("EmailUpdated", tail_events[0].event_type); + assert_eq!("1.0", tail_events[0].event_version); + assert_eq!( + json!({ "EmailUpdated": { "new_email": "carol+1@example.com" } }), + tail_events[0].payload + ); + + let reloaded_context = event_store + .load_aggregate(&aggregate_id) + .await + .expect("aggregate should reload from snapshot plus tail events"); + + assert_eq!(3, reloaded_context.current_sequence); + assert_eq!("Carol", reloaded_context.aggregate.name); + assert_eq!("carol+1@example.com", reloaded_context.aggregate.email); + + let no_new_events = repository + .get_last_events::(&aggregate_id, reloaded_context.current_sequence) + .await + .expect("reading past the end of the stream should succeed"); + + assert!(no_new_events.is_empty()); +} + +#[tokio::test] +async fn persisted_customer_events_use_explicit_v1_event_type_in_eventsourcingdb() { + let repository = new_repository(); + let client = esdb_client(); + let aggregate_id = unique_aggregate_id("wire-shape"); + + let events = vec![serialized_customer_event( + &aggregate_id, + 1, + CustomerEvent::NameAdded { + name: "Dana".to_string(), + }, + json!({ "request_id": "wire-shape" }), + )]; + + repository + .persist::(&events, None) + .await + .expect("customer event should persist"); + + let subject = format!("/Customer/{aggregate_id}"); + let stored_events = client + .read_events( + &subject, + Some(ReadEventsOptions { + order: Some(EventOrdering::Chronological), + ..Default::default() + }), + ) + .await + .expect("stored events should be readable") + .try_collect::>() + .await + .expect("stored events should collect"); + + assert_eq!(1, stored_events.len()); + assert_eq!("io.eventsourcingdb.name-added.v1", stored_events[0].ty()); +} From a1980f55db36f7e655164967a6db5e5b3a712744 Mon Sep 17 00:00:00 2001 From: flangator Date: Tue, 17 Mar 2026 20:02:35 +0100 Subject: [PATCH 5/6] chore: docstrings and explanation of id mapping --- README.md | 4 ++++ src/conversion.rs | 12 +++++++++++- src/event_repository.rs | 11 ++++++++++- 3 files changed, 25 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 96cbe2f..6a54f05 100644 --- a/README.md +++ b/README.md @@ -17,6 +17,10 @@ This adapter follows the naming guidance from the [EventSourcingDB documentation - Legacy unversioned event types are still read as version `1.0`, but new writes always use an explicit `.v1` suffix. - Event payload and metadata are stored in an adapter envelope so `cqrs-es` metadata survives round-trips. +## Mapping EventSourcingDB ids to `cqrs-es` sequences + +`cqrs-es` expects each aggregate stream to have a continuous sequence `1, 2, 3, ...`, while EventSourcingDB ids are global chronological integers encoded as strings for CloudEvents compatibility. This adapter remaps the global id stream to per-aggregate logical sequence numbers, uses the last EventSourcingDB id as the optimistic-write precondition, and stores that id in snapshots so loading after a snapshot can continue from the correct global boundary. + ## Usage Add the following to your `Cargo.toml`: diff --git a/src/conversion.rs b/src/conversion.rs index f970d0e..520b690 100644 --- a/src/conversion.rs +++ b/src/conversion.rs @@ -38,7 +38,7 @@ fn pascal_to_kebab_case(s: &str) -> String { } // we assume that the subject always has the shape /aggregate_type/aggregate_id -// e.g. /books/42 +// e.g. /books/42, everything else will be rejected pub(crate) fn map_subject_to_aggregate_type_and_id( subject: &str, ) -> Result<(String, String), EventSourcingDbError> { @@ -59,6 +59,10 @@ pub(crate) fn map_subject_to_aggregate_type_and_id( Ok((aggregate_type.to_string(), aggregate_id.to_string())) } +/// Reverse-domain prefix used when building persisted EventSourcingDB event types. +/// For example, `ReversedDomain::new(["com", "flangator", "banking"])` makes the +/// CQRS event `CustomerDepositedMoney` version `1.0` persist as +/// `com.flangator.banking.customer-deposited-money.v1`. #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct ReversedDomain(Vec); @@ -76,6 +80,9 @@ impl ReversedDomain { } } +/// The fully qualified event name follows EventSourcingDB [convention](https://docs.eventsourcingdb.io/fundamentals/event-types/#required-field) +/// and wrapts the reversed domain,the actual type as presented to cqrs-es and an optional version. +/// The absence of a version will be treated as 1.0. #[derive(Debug, Clone, PartialEq, Eq, Hash)] pub struct QualifiedEventType { pub reversed_domain: ReversedDomain, @@ -192,6 +199,9 @@ pub fn qualify_event_type( Ok(parts.join(".")) } +/// Since both metadata and data of the cqrs-es event needs to be persisted in the +/// `data` field of [`Event`], a JSON object with a special tag [`STORED_EVENT_ENVELOPE_MARKER`] +/// and two fields is created. pub fn wrap_event_data(payload: Value, metadata: Value) -> Value { json!({ "_cqrs_es": STORED_EVENT_ENVELOPE_MARKER, diff --git a/src/event_repository.rs b/src/event_repository.rs index 34abcf7..05a1ea9 100644 --- a/src/event_repository.rs +++ b/src/event_repository.rs @@ -24,7 +24,7 @@ use crate::{ const SNAPSHOT_EVENT_TYPE: &str = "io.eventsourcingdb.cqrs-es.snapshot-record.v1"; const SNAPSHOT_EVENT_SOURCE: &str = "urn:eventsourcingdb-es:snapshot"; const EVENT_SOURCE: &str = "urn:eventsourcingdb-es:event"; -//TODO: make part of struct +//TODO: make part of struct? const STREAM_CHANNEL_SIZE: usize = 2048; #[derive(Debug, Clone, Serialize, Deserialize)] @@ -32,17 +32,26 @@ struct EventSourcingDbSnapshotRecord { aggregate_type: String, aggregate_id: String, aggregate: Value, + // cqrs-es event sequence current_sequence: usize, current_snapshot: usize, + // EventSourcingDb's event id last_event_id: Option, } +// Required to map betwen EventSourcingDBs internal ids and the continous sequence assumed +// by cqrs-es. #[derive(Debug, Clone)] struct SubjectState { logical_sequence_count: usize, last_event_id: Option, } +/// `cqrs-es` expects per-aggregate events to have a continuous logical sequence `1, 2, 3, ...`, +/// while EventSourcingDB ids are global chronological integers encoded as strings for +/// CloudEvents compatibility. This adapter remaps those global ids to per-aggregate sequences, +/// uses the last stored EventSourcingDB id as the optimistic-write precondition, and stores that +/// id in snapshots so tail reads can resume at the right global boundary. pub struct EventSourcingDbEventRepository { client: Arc, domain: ReversedDomain, From 7d77471436966eeae3f293dce48a4e950674a088 Mon Sep 17 00:00:00 2001 From: flangator Date: Tue, 17 Mar 2026 20:04:20 +0100 Subject: [PATCH 6/6] chore: add implemented features --- README.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/README.md b/README.md index 6a54f05..2ff523b 100644 --- a/README.md +++ b/README.md @@ -30,3 +30,13 @@ Add the following to your `Cargo.toml`: cqrs-es = "0.5" eventsourcingdb-es = "0.1.0" ``` + +## Features implemented + +- [x] Persist events +- [x] Read events by subject (aggregate type and ID) +- [x] Read last events by subject +- [x] Stream events for subject +- [x] Stream all events for aggregate type +- [x] Write snapshots +- [x] Read snapshots