diff --git a/README.md b/README.md index 4b3baa71f..6f8018bf5 100644 --- a/README.md +++ b/README.md @@ -249,8 +249,15 @@ impl Todo { Create events at a specific version for [upcasting](#event-upcasting--versioning): ```rust +type InitV1 = (String, String); +type InitV2 = (String, String, u8); + +fn upcast_init_v1_v2((id, task): InitV1) -> InitV2 { + (id, task, 0) +} + #[sourced(entity, upcasters( - ("Initialized", 1 => 2, upcast_init_v1_v2), + ("Initialized", 1 => 2, InitV1 => InitV2, upcast_init_v1_v2), ))] impl TodoV2 { #[event("Initialized", version = 2)] @@ -381,11 +388,18 @@ aggregate!(Todo, entity { With [upcasters](#event-upcasting--versioning) for event schema evolution: ```rust +type InitV1 = (String, String); +type InitV2 = (String, String, u8); + +fn upcast_initialized_v1_v2((id, task): InitV1) -> InitV2 { + (id, task, 0) +} + aggregate!(Todo, entity { "Initialized"(id, task, priority) => initialize, "Completed"() => complete(), } upcasters [ - ("Initialized", 1 => 2, upcast_initialized_v1_v2), + ("Initialized", 1 => 2, InitV1 => InitV2, upcast_initialized_v1_v2), ]); ``` @@ -1251,17 +1265,19 @@ let todo = repo.get("todo-1")?.unwrap(); ## Event Upcasting / Versioning -Event schemas evolve over time. When you add a field to an event (e.g., `priority` to `Initialized`), old serialized events in storage can't deserialize into the new type — especially with bitcode's rigid binary format. **Upcasters** solve this: pure functions that transform old event payloads into the current format at read time, without modifying stored data. +Event schemas evolve over time. When you add a field to an event (e.g., `priority` to `Initialized`), old serialized events in storage can't deserialize into the new type. **Upcasters** solve this: typed functions that transform old event payload shapes into the current format at read time, without modifying stored data. ### Defining an Upcaster -An upcaster is a plain function that converts a payload from one version to the next: +An upcaster is a plain function that converts a typed payload from one version to the next. The crate handles payload decoding and encoding: ```rust +type InitV1 = (String, String); +type InitV2 = (String, String, u8); + /// Upcasts Initialized v1 (id, task) → v2 (id, task, priority) -fn upcast_init_v1_v2(payload: &[u8]) -> Vec { - let (id, task): (String, String) = bitcode::deserialize(payload).unwrap(); - bitcode::serialize(&(id, task, 0u8)).unwrap() // default priority = 0 +fn upcast_init_v1_v2((id, task): InitV1) -> InitV2 { + (id, task, 0) } ``` @@ -1279,7 +1295,7 @@ struct Todo { } #[sourced(entity, upcasters( - ("Initialized", 1 => 2, upcast_init_v1_v2), + ("Initialized", 1 => 2, InitV1 => InitV2, upcast_init_v1_v2), ))] impl Todo { #[event("Initialized", version = 2)] @@ -1308,7 +1324,7 @@ aggregate!(Todo, entity { "Initialized"(id, task, priority) => initialize, "Completed"() => complete(), } upcasters [ - ("Initialized", 1 => 2, upcast_init_v1_v2), + ("Initialized", 1 => 2, InitV1 => InitV2, upcast_init_v1_v2), ]); ``` @@ -1319,19 +1335,21 @@ Old events stored as `(id, task)` at v1 get transparently upcasted to `(id, task Upcasters chain automatically. Each transforms one version to the next (v1->v2->v3): ```rust -fn upcast_init_v1_v2(payload: &[u8]) -> Vec { - let (id, task): (String, String) = bitcode::deserialize(payload).unwrap(); - bitcode::serialize(&(id, task, 0u8)).unwrap() +type InitV1 = (String, String); +type InitV2 = (String, String, u8); +type InitV3 = (String, String, u8, String); + +fn upcast_init_v1_v2((id, task): InitV1) -> InitV2 { + (id, task, 0) } -fn upcast_init_v2_v3(payload: &[u8]) -> Vec { - let (id, task, priority): (String, String, u8) = bitcode::deserialize(payload).unwrap(); - bitcode::serialize(&(id, task, priority, String::new())).unwrap() // add due_date +fn upcast_init_v2_v3((id, task, priority): InitV2) -> InitV3 { + (id, task, priority, String::new()) } #[sourced(entity, upcasters( - ("Initialized", 1 => 2, upcast_init_v1_v2), - ("Initialized", 2 => 3, upcast_init_v2_v3), + ("Initialized", 1 => 2, InitV1 => InitV2, upcast_init_v1_v2), + ("Initialized", 2 => 3, InitV2 => InitV3, upcast_init_v2_v3), ))] impl Todo { #[event("Initialized", version = 3)] @@ -1351,26 +1369,16 @@ A v1 event automatically chains through v1->v2->v3. A v2 event only goes through - **No stored data modified**: Upcasters are read-time transformations. The event store is never touched. - **Zero overhead when unused**: If an aggregate has no upcasters, `hydrate()` takes the fast path with no extra allocation. -### The `EventUpcaster` Struct +### Direct Upcasting -Under the hood, each upcaster is a plain struct with a function pointer — no traits, no boxing: +You can also use `upcast_events()` directly with an aggregate's registered upcasters for custom hydration logic: ```rust -pub struct EventUpcaster { - pub event_type: &'static str, - pub from_version: u64, - pub to_version: u64, - pub transform: fn(payload: &[u8]) -> Vec, -} -``` - -You can also use `upcast_events()` directly for custom hydration logic: +use sourced_rust::{upcast_events, Aggregate, EventRecord, UpcastError}; -```rust -use sourced_rust::{upcast_events, EventUpcaster}; - -let upcasters: &[EventUpcaster] = &[/* ... */]; -let upcasted = upcast_events(events, upcasters); +fn upcast_for_replay(events: Vec) -> Result, UpcastError> { + upcast_events(events, Todo::upcasters()) +} ``` ## Project Structure diff --git a/sourced_rust_macros/src/lib.rs b/sourced_rust_macros/src/lib.rs index a0bd442f8..76c79cd56 100644 --- a/sourced_rust_macros/src/lib.rs +++ b/sourced_rust_macros/src/lib.rs @@ -179,6 +179,85 @@ fn generate_enqueue_call( } } +fn upcaster_wrapper_prefix(owner: &Ident) -> String { + owner + .to_string() + .trim_start_matches("r#") + .to_ascii_lowercase() +} + +fn generate_upcaster_tokens( + owner: &Ident, + upcasters: &[UpcasterDef], +) -> (proc_macro2::TokenStream, proc_macro2::TokenStream) { + if upcasters.is_empty() { + return (quote! {}, quote! {}); + } + + let prefix = upcaster_wrapper_prefix(owner); + let wrapper_names: Vec<_> = upcasters + .iter() + .enumerate() + .map(|(idx, _)| format_ident!("__sourced_upcast_{}_{}", prefix, idx)) + .collect(); + + let wrapper_defs = upcasters + .iter() + .zip(wrapper_names.iter()) + .map(|(u, wrapper)| { + let source_type = &u.source_type; + let target_type = &u.target_type; + let to_version = &u.to_version; + let transform_fn = &u.transform_fn; + quote! { + fn #wrapper( + event: &sourced_rust::EventRecord, + ) -> Result, sourced_rust::UpcastError> { + sourced_rust::upcast_payload::<#source_type, #target_type>( + event, + #to_version, + #transform_fn, + ) + } + } + }); + + let upcaster_entries = upcasters + .iter() + .zip(wrapper_names.iter()) + .map(|(u, wrapper)| { + let event_name = &u.event_name; + let from_version = &u.from_version; + let to_version = &u.to_version; + quote! { + sourced_rust::EventUpcaster { + event_type: #event_name, + from_version: #from_version, + to_version: #to_version, + transform: #owner::#wrapper, + } + } + }); + + let upcasters_method = quote! { + fn upcasters() -> &'static [sourced_rust::EventUpcaster] { + static UPCASTERS: &[sourced_rust::EventUpcaster] = &[ + #(#upcaster_entries),* + ]; + UPCASTERS + } + }; + + ( + quote! { + impl #owner { + #(#wrapper_defs)* + } + }, + upcasters_method, + ) +} + // ============================================================================ // #[enqueue] attribute macro // ============================================================================ @@ -543,35 +622,12 @@ pub fn aggregate(input: TokenStream) -> TokenStream { } }); - // Generate upcasters() method if upcasters are defined - let upcasters_method = if input.upcasters.is_empty() { - quote! {} - } else { - let upcaster_entries = input.upcasters.iter().map(|u| { - let event_name = &u.event_name; - let from_version = &u.from_version; - let to_version = &u.to_version; - let transform_fn = &u.transform_fn; - quote! { - sourced_rust::EventUpcaster { - event_type: #event_name, - from_version: #from_version, - to_version: #to_version, - transform: #transform_fn, - } - } - }); - quote! { - fn upcasters() -> &'static [sourced_rust::EventUpcaster] { - static UPCASTERS: &[sourced_rust::EventUpcaster] = &[ - #(#upcaster_entries),* - ]; - UPCASTERS - } - } - }; + let (upcaster_wrappers, upcasters_method) = + generate_upcaster_tokens(agg_name, &input.upcasters); let expanded = quote! { + #upcaster_wrappers + impl sourced_rust::Aggregate for #agg_name { type ReplayError = String; @@ -605,6 +661,8 @@ struct UpcasterDef { event_name: LitStr, from_version: syn::LitInt, to_version: syn::LitInt, + source_type: syn::Type, + target_type: syn::Type, transform_fn: syn::Path, } @@ -681,7 +739,7 @@ impl Parse for AggregateInput { syn::bracketed!(upcaster_content in input); while !upcaster_content.is_empty() { - // Parse: ("EventName", from => to, transform_fn) + // Parse: ("EventName", from => to, SourceType => TargetType, transform_fn) let inner; syn::parenthesized!(inner in upcaster_content); @@ -691,12 +749,18 @@ impl Parse for AggregateInput { inner.parse::]>()?; let to_version: syn::LitInt = inner.parse()?; inner.parse::()?; + let source_type: syn::Type = inner.parse()?; + inner.parse::]>()?; + let target_type: syn::Type = inner.parse()?; + inner.parse::()?; let transform_fn: syn::Path = inner.parse()?; upcasters.push(UpcasterDef { event_name, from_version, to_version, + source_type, + target_type, transform_fn, }); @@ -764,11 +828,17 @@ fn parse_sourced_args(input: ParseStream) -> syn::Result { inner.parse::]>()?; let to_ver: syn::LitInt = inner.parse()?; inner.parse::()?; + let source_type: syn::Type = inner.parse()?; + inner.parse::]>()?; + let target_type: syn::Type = inner.parse()?; + inner.parse::()?; let transform: syn::Path = inner.parse()?; upcasters.push(UpcasterDef { event_name: ev_name, from_version: from_ver, to_version: to_ver, + source_type, + target_type, transform_fn: transform, }); if upcaster_content.peek(Token![,]) { @@ -873,7 +943,7 @@ struct EventMethodInfo { /// Options: /// - `#[sourced(entity)]` - entity field name /// - `#[sourced(entity, events = "CustomName")]` - custom enum name -/// - `#[sourced(entity, upcasters(("EventName", 1 => 2, upcast_fn)))]` - upcasters +/// - `#[sourced(entity, upcasters(("EventName", 1 => 2, OldPayload => NewPayload, upcast_fn)))]` - upcasters #[proc_macro_attribute] pub fn sourced(attr: TokenStream, item: TokenStream) -> TokenStream { let args = parse_macro_input!(attr with parse_sourced_args); @@ -1075,32 +1145,8 @@ pub fn sourced(attr: TokenStream, item: TokenStream) -> TokenStream { } }); - let upcasters_method = if args.upcasters.is_empty() { - quote! {} - } else { - let upcaster_entries = args.upcasters.iter().map(|u| { - let ev_name = &u.event_name; - let from_v = &u.from_version; - let to_v = &u.to_version; - let transform = &u.transform_fn; - quote! { - sourced_rust::EventUpcaster { - event_type: #ev_name, - from_version: #from_v, - to_version: #to_v, - transform: #transform, - } - } - }); - quote! { - fn upcasters() -> &'static [sourced_rust::EventUpcaster] { - static UPCASTERS: &[sourced_rust::EventUpcaster] = &[ - #(#upcaster_entries),* - ]; - UPCASTERS - } - } - }; + let (upcaster_wrappers, upcasters_method) = + generate_upcaster_tokens(&struct_name, &args.upcasters); let aggregate_impl = quote! { impl sourced_rust::Aggregate for #struct_name { @@ -1134,6 +1180,7 @@ pub fn sourced(attr: TokenStream, item: TokenStream) -> TokenStream { #enum_def #event_name_impl #try_from_impl + #upcaster_wrappers #aggregate_impl }; diff --git a/src/entity/mod.rs b/src/entity/mod.rs index daea0383a..ee7041107 100644 --- a/src/entity/mod.rs +++ b/src/entity/mod.rs @@ -13,4 +13,4 @@ pub use event_record::{ BITCODE_PAYLOAD_CODEC_VERSION, }; pub use local_event::LocalEvent; -pub use upcaster::{upcast_events, EventUpcaster, UpcastError}; +pub use upcaster::{upcast_events, upcast_payload, EventUpcaster, UpcastError}; diff --git a/src/entity/upcaster.rs b/src/entity/upcaster.rs index 8aa6396b5..d60975308 100644 --- a/src/entity/upcaster.rs +++ b/src/entity/upcaster.rs @@ -1,7 +1,9 @@ use std::collections::HashSet; use std::fmt; -use super::EventRecord; +use serde::{de::DeserializeOwned, Serialize}; + +use super::{BitcodePayloadCodec, EventRecord, EventRecordError, PayloadCodec}; /// A stateless, pure transformation that converts an event payload from one version to another. /// @@ -11,7 +13,7 @@ pub struct EventUpcaster { pub event_type: &'static str, pub from_version: u64, pub to_version: u64, - pub transform: fn(payload: &[u8]) -> Vec, + pub transform: fn(event: &EventRecord) -> Result, UpcastError>, } /// Error returned when an upcaster chain cannot make safe forward progress. @@ -30,6 +32,12 @@ pub enum UpcastError { event_type: String, version: u64, }, + PayloadTransform { + event_type: String, + from: u64, + to: u64, + source: EventRecordError, + }, } impl fmt::Display for UpcastError { @@ -57,11 +65,57 @@ impl fmt::Display for UpcastError { f, "upcaster chain for event {event_type} cycles back to version {version}" ), + UpcastError::PayloadTransform { + event_type, + from, + to, + source, + } => write!( + f, + "failed to upcast event {event_type} from version {from} to {to}: {source}" + ), + } + } +} + +impl std::error::Error for UpcastError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + UpcastError::PayloadTransform { source, .. } => Some(source), + UpcastError::SameVersionTransition { .. } + | UpcastError::BackwardTransition { .. } + | UpcastError::CycleDetected { .. } => None, } } } -impl std::error::Error for UpcastError {} +/// Decode an event payload, transform it with typed Rust values, and encode the result. +pub fn upcast_payload( + event: &EventRecord, + to_version: u64, + transform: fn(From) -> To, +) -> Result, UpcastError> +where + From: DeserializeOwned, + To: Serialize, +{ + let decoded = event + .decode::() + .map_err(|source| UpcastError::PayloadTransform { + event_type: event.event_name.clone(), + from: event.event_version, + to: to_version, + source, + })?; + + let transformed = transform(decoded); + BitcodePayloadCodec::encode(&transformed).map_err(|source| UpcastError::PayloadTransform { + event_type: event.event_name.clone(), + from: event.event_version, + to: to_version, + source: EventRecordError::encode(source), + }) +} /// Apply upcasters to a list of events. Chains automatically (v1->v2->v3). pub fn upcast_events( @@ -106,7 +160,7 @@ fn upcast_one( } let next_version = u.to_version; - event.payload = (u.transform)(&event.payload); + event.payload = (u.transform)(&event)?; event.event_version = next_version; if !seen_versions.insert(next_version) { return Err(UpcastError::CycleDetected { @@ -129,6 +183,36 @@ fn upcast_one( mod tests { use super::*; + fn event_payload(value: &T) -> Vec { + BitcodePayloadCodec::encode(value).unwrap() + } + + fn add_default_priority((id, task): (String, String)) -> (String, String, u8) { + (id, task, 0) + } + + fn add_empty_due_date( + (id, task, priority): (String, String, u8), + ) -> (String, String, u8, String) { + (id, task, priority, String::new()) + } + + fn upcast_test_event_v1_v2(event: &EventRecord) -> Result, UpcastError> { + upcast_payload::<(String, String), (String, String, u8)>(event, 2, add_default_priority) + } + + fn upcast_test_event_v2_v3(event: &EventRecord) -> Result, UpcastError> { + upcast_payload::<(String, String, u8), (String, String, u8, String)>( + event, + 3, + add_empty_due_date, + ) + } + + fn passthrough(event: &EventRecord) -> Result, UpcastError> { + Ok(event.payload.clone()) + } + #[test] fn no_upcasters_leaves_events_unchanged() { let event = EventRecord::new("TestEvent", vec![1, 2, 3], 1); @@ -139,19 +223,22 @@ mod tests { #[test] fn single_upcaster_transforms_matching_event() { - let event = EventRecord::new("TestEvent", vec![1, 2], 1); + let event = EventRecord::new( + "TestEvent", + event_payload(&("id".to_string(), "task".to_string())), + 1, + ); let upcasters = [EventUpcaster { event_type: "TestEvent", from_version: 1, to_version: 2, - transform: |payload| { - let mut new = payload.to_vec(); - new.push(99); - new - }, + transform: upcast_test_event_v1_v2, }]; let events = upcast_events(vec![event], &upcasters).unwrap(); - assert_eq!(events[0].payload, vec![1, 2, 99]); + assert_eq!( + events[0].decode::<(String, String, u8)>().unwrap(), + ("id".to_string(), "task".to_string(), 0) + ); assert_eq!(events[0].event_version, 2); } @@ -162,7 +249,7 @@ mod tests { event_type: "TestEvent", from_version: 1, to_version: 2, - transform: |_| vec![99], + transform: upcast_test_event_v1_v2, }]; let events = upcast_events(vec![event], &upcasters).unwrap(); assert_eq!(events[0].payload, vec![1, 2]); @@ -171,63 +258,117 @@ mod tests { #[test] fn chained_upcasters_v1_to_v3() { - let event = EventRecord::new("TestEvent", vec![1], 1); + let event = EventRecord::new( + "TestEvent", + event_payload(&("id".to_string(), "task".to_string())), + 1, + ); let upcasters = [ EventUpcaster { event_type: "TestEvent", from_version: 1, to_version: 2, - transform: |payload| { - let mut new = payload.to_vec(); - new.push(2); - new - }, + transform: upcast_test_event_v1_v2, }, EventUpcaster { event_type: "TestEvent", from_version: 2, to_version: 3, - transform: |payload| { - let mut new = payload.to_vec(); - new.push(3); - new - }, + transform: upcast_test_event_v2_v3, }, ]; let events = upcast_events(vec![event], &upcasters).unwrap(); - assert_eq!(events[0].payload, vec![1, 2, 3]); + assert_eq!( + events[0].decode::<(String, String, u8, String)>().unwrap(), + ("id".to_string(), "task".to_string(), 0, String::new()) + ); assert_eq!(events[0].event_version, 3); } #[test] fn mixed_events_some_upcasted_some_not() { let events = vec![ - EventRecord::new("A", vec![10], 1), + EventRecord::new( + "A", + event_payload(&("id".to_string(), "task".to_string())), + 1, + ), EventRecord::new("B", vec![20], 1), - EventRecord::new_versioned("A", vec![10, 99], 3, 2), + EventRecord::new_versioned( + "A", + event_payload(&("id".to_string(), "task".to_string(), 99u8)), + 3, + 2, + ), ]; let upcasters = [EventUpcaster { event_type: "A", from_version: 1, to_version: 2, - transform: |payload| { - let mut new = payload.to_vec(); - new.push(99); - new - }, + transform: upcast_test_event_v1_v2, }]; let result = upcast_events(events, &upcasters).unwrap(); // First A: upcasted from v1 to v2 - assert_eq!(result[0].payload, vec![10, 99]); + assert_eq!( + result[0].decode::<(String, String, u8)>().unwrap(), + ("id".to_string(), "task".to_string(), 0) + ); assert_eq!(result[0].event_version, 2); // B: untouched assert_eq!(result[1].payload, vec![20]); assert_eq!(result[1].event_version, 1); // Second A already at v2: untouched - assert_eq!(result[2].payload, vec![10, 99]); + assert_eq!( + result[2].decode::<(String, String, u8)>().unwrap(), + ("id".to_string(), "task".to_string(), 99) + ); assert_eq!(result[2].event_version, 2); } + #[test] + fn upcast_events_returns_payload_error_when_decode_fails() { + let event = EventRecord::new("A", vec![10], 1); + let upcasters = [EventUpcaster { + event_type: "A", + from_version: 1, + to_version: 2, + transform: upcast_test_event_v1_v2, + }]; + + let err = upcast_events(vec![event], &upcasters).unwrap_err(); + + match err { + UpcastError::PayloadTransform { + event_type, + from, + to, + .. + } => { + assert_eq!(event_type, "A"); + assert_eq!(from, 1); + assert_eq!(to, 2); + } + other => panic!("expected payload transform error, got {other:?}"), + } + } + + #[test] + fn payload_transform_error_exposes_source() { + let source = EventRecordError { + message: "decode failed".to_string(), + }; + let err = UpcastError::PayloadTransform { + event_type: "A".to_string(), + from: 1, + to: 2, + source: source.clone(), + }; + + let chained = std::error::Error::source(&err).unwrap(); + + assert_eq!(chained.to_string(), source.to_string()); + } + #[test] fn upcast_events_rejects_same_version_transition() { let event = EventRecord::new("A", vec![10], 1); @@ -235,7 +376,7 @@ mod tests { event_type: "A", from_version: 1, to_version: 1, - transform: |payload| payload.to_vec(), + transform: passthrough, }]; let err = upcast_events(vec![event], &upcasters).unwrap_err(); @@ -256,7 +397,7 @@ mod tests { event_type: "A", from_version: 3, to_version: 2, - transform: |payload| payload.to_vec(), + transform: passthrough, }]; let err = upcast_events(vec![event], &upcasters).unwrap_err(); @@ -279,13 +420,13 @@ mod tests { event_type: "A", from_version: 1, to_version: 2, - transform: |payload| payload.to_vec(), + transform: passthrough, }, EventUpcaster { event_type: "A", from_version: 2, to_version: 1, - transform: |payload| payload.to_vec(), + transform: passthrough, }, ]; diff --git a/src/lib.rs b/src/lib.rs index 3812ef14a..cc9a49b9f 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -21,8 +21,8 @@ pub mod snapshot; // Re-export entity types at crate root for convenience pub use entity::{ - upcast_events, BitcodePayloadCodec, Committable, Entity, Event, EventRecord, EventRecordError, - EventUpcaster, LocalEvent, PayloadCodec, UpcastError, BITCODE_PAYLOAD_CODEC, + upcast_events, upcast_payload, BitcodePayloadCodec, Committable, Entity, Event, EventRecord, + EventRecordError, EventUpcaster, LocalEvent, PayloadCodec, UpcastError, BITCODE_PAYLOAD_CODEC, BITCODE_PAYLOAD_CODEC_VERSION, }; diff --git a/tests/sourced_upcasting/aggregate.rs b/tests/sourced_upcasting/aggregate.rs index 39879fa6f..1ecd9ffe9 100644 --- a/tests/sourced_upcasting/aggregate.rs +++ b/tests/sourced_upcasting/aggregate.rs @@ -1,5 +1,9 @@ use sourced_rust::{sourced, Entity}; +pub type InitializedV1 = (String, String, String); +pub type InitializedV2 = (String, String, String, u8); +pub type InitializedV3 = (String, String, String, u8, String); + // ============================================================================= // V1 aggregate: original schema // ============================================================================= @@ -31,9 +35,8 @@ impl TodoV1 { // V2 aggregate: added priority field + upcaster // ============================================================================= -pub fn upcast_initialized_v1_v2(payload: &[u8]) -> Vec { - let (id, user_id, task): (String, String, String) = bitcode::deserialize(payload).unwrap(); - bitcode::serialize(&(id, user_id, task, 0u8)).unwrap() +pub fn upcast_initialized_v1_v2((id, user_id, task): InitializedV1) -> InitializedV2 { + (id, user_id, task, 0) } #[derive(Default)] @@ -46,9 +49,13 @@ pub struct TodoV2 { } #[sourced(entity, upcasters( - ("Initialized", 1 => 2, upcast_initialized_v1_v2), + ("Initialized", 1 => 2, InitializedV1 => InitializedV2, Self::upcast_initialized_v1_v2), ))] impl TodoV2 { + pub fn upcast_initialized_v1_v2(payload: InitializedV1) -> InitializedV2 { + upcast_initialized_v1_v2(payload) + } + #[event("Initialized", version = 2)] pub fn initialize(&mut self, id: String, user_id: String, task: String, priority: u8) { self.entity.set_id(&id); @@ -67,10 +74,8 @@ impl TodoV2 { // V3 aggregate: added due_date field + chained upcasters // ============================================================================= -pub fn upcast_initialized_v2_v3(payload: &[u8]) -> Vec { - let (id, user_id, task, priority): (String, String, String, u8) = - bitcode::deserialize(payload).unwrap(); - bitcode::serialize(&(id, user_id, task, priority, String::new())).unwrap() +pub fn upcast_initialized_v2_v3((id, user_id, task, priority): InitializedV2) -> InitializedV3 { + (id, user_id, task, priority, String::new()) } #[derive(Default)] @@ -84,8 +89,8 @@ pub struct TodoV3 { } #[sourced(entity, upcasters( - ("Initialized", 1 => 2, upcast_initialized_v1_v2), - ("Initialized", 2 => 3, upcast_initialized_v2_v3), + ("Initialized", 1 => 2, InitializedV1 => InitializedV2, upcast_initialized_v1_v2), + ("Initialized", 2 => 3, InitializedV2 => InitializedV3, upcast_initialized_v2_v3), ))] impl TodoV3 { #[event("Initialized", version = 3)] diff --git a/tests/upcasting/aggregate.rs b/tests/upcasting/aggregate.rs index e76c9cf1f..a02adf93f 100644 --- a/tests/upcasting/aggregate.rs +++ b/tests/upcasting/aggregate.rs @@ -1,6 +1,10 @@ use serde::{Deserialize, Serialize}; use sourced_rust::{digest, Entity, Snapshottable}; +pub type InitializedV1 = (String, String, String); +pub type InitializedV2 = (String, String, String, u8); +pub type InitializedV3 = (String, String, String, u8, String); + // ============================================================================= // V1 aggregate: original schema // ============================================================================= @@ -38,9 +42,8 @@ sourced_rust::aggregate!(TodoV1, entity { // ============================================================================= /// Upcasts Initialized v1 (id, user_id, task) → v2 (id, user_id, task, priority) -pub fn upcast_initialized_v1_v2(payload: &[u8]) -> Vec { - let (id, user_id, task): (String, String, String) = bitcode::deserialize(payload).unwrap(); - bitcode::serialize(&(id, user_id, task, 0u8)).unwrap() +pub fn upcast_initialized_v1_v2((id, user_id, task): InitializedV1) -> InitializedV2 { + (id, user_id, task, 0) } #[derive(Default)] @@ -71,7 +74,7 @@ sourced_rust::aggregate!(TodoV2, entity { "Initialized"(id, user_id, task, priority) => initialize, "Completed"() => complete(), } upcasters [ - ("Initialized", 1 => 2, upcast_initialized_v1_v2), + ("Initialized", 1 => 2, InitializedV1 => InitializedV2, upcast_initialized_v1_v2), ]); // ============================================================================= @@ -79,11 +82,8 @@ sourced_rust::aggregate!(TodoV2, entity { // ============================================================================= /// Upcasts Initialized v2 (id, user_id, task, priority) → v3 (id, user_id, task, priority, due_date) -pub fn upcast_initialized_v2_v3(payload: &[u8]) -> Vec { - let (id, user_id, task, priority): (String, String, String, u8) = - bitcode::deserialize(payload).unwrap(); - // Default due_date: empty string meaning "no due date" - bitcode::serialize(&(id, user_id, task, priority, String::new())).unwrap() +pub fn upcast_initialized_v2_v3((id, user_id, task, priority): InitializedV2) -> InitializedV3 { + (id, user_id, task, priority, String::new()) } #[derive(Default)] @@ -123,8 +123,8 @@ sourced_rust::aggregate!(TodoV3, entity { "Initialized"(id, user_id, task, priority, due_date) => initialize, "Completed"() => complete(), } upcasters [ - ("Initialized", 1 => 2, upcast_initialized_v1_v2), - ("Initialized", 2 => 3, upcast_initialized_v2_v3), + ("Initialized", 1 => 2, InitializedV1 => InitializedV2, upcast_initialized_v1_v2), + ("Initialized", 2 => 3, InitializedV2 => InitializedV3, upcast_initialized_v2_v3), ]); // ============================================================================= diff --git a/tests/upcasting/main.rs b/tests/upcasting/main.rs index a19601985..ab9297e90 100644 --- a/tests/upcasting/main.rs +++ b/tests/upcasting/main.rs @@ -2,12 +2,13 @@ mod aggregate; use aggregate::{TodoV1, TodoV2, TodoV3}; use sourced_rust::{ - hydrate, upcast_events, Aggregate, AggregateBuilder, Commit, Entity, EventRecord, - EventUpcaster, HashMapRepository, RepositoryError, SnapshotStore, + hydrate, hydrate_from_snapshot, upcast_events, Aggregate, AggregateBuilder, Commit, Entity, + EventRecord, EventUpcaster, HashMapRepository, RepositoryError, SnapshotRecord, SnapshotStore, + UpcastError, }; -fn identity_payload(payload: &[u8]) -> Vec { - payload.to_vec() +fn identity_payload(event: &EventRecord) -> Result, UpcastError> { + Ok(event.payload.clone()) } #[derive(Debug, Default)] @@ -286,14 +287,7 @@ fn upcast_events_standalone() { bitcode::serialize(&("id1".to_string(), "user1".to_string(), "task1".to_string())).unwrap(); let event = EventRecord::new("Initialized", payload_v1, 1); - let upcasters: &[EventUpcaster] = &[EventUpcaster { - event_type: "Initialized", - from_version: 1, - to_version: 2, - transform: aggregate::upcast_initialized_v1_v2, - }]; - - let result = upcast_events(vec![event], upcasters).unwrap(); + let result = upcast_events(vec![event], TodoV2::upcasters()).unwrap(); assert_eq!(result[0].event_version, 2); let (id, user, task, priority): (String, String, String, u8) = @@ -319,6 +313,20 @@ fn hydrate_rejects_invalid_same_version_upcaster() { } } +#[test] +fn hydrate_returns_replay_error_when_typed_upcaster_decode_fails() { + let mut entity = Entity::new(); + entity.load_from_history(vec![EventRecord::new("Initialized", vec![0xff], 1)]); + + match hydrate::(entity) { + Err(RepositoryError::Replay(message)) => { + assert!(message.contains("failed to upcast event Initialized")); + } + Err(other) => panic!("expected replay error, got {other:?}"), + Ok(_) => panic!("expected replay error"), + } +} + // ============================================================================= // Snapshot + upcasting // ============================================================================= @@ -370,3 +378,32 @@ fn snapshot_repo_with_v1_events_upcasted_on_hydrate() { assert_eq!(loaded.priority, 0); // upcasted default assert!(loaded.completed); } + +#[test] +fn hydrate_from_snapshot_returns_replay_error_when_post_snapshot_upcaster_decode_fails() { + let snapshot = SnapshotRecord { + aggregate_id: "t1".to_string(), + version: 1, + data: bitcode::serialize(&aggregate::TodoV2Snapshot { + id: "t1".to_string(), + user_id: "iris".to_string(), + task: "Plan".to_string(), + priority: 1, + completed: false, + }) + .unwrap(), + }; + let mut invalid_event = EventRecord::new("Initialized", vec![0xff], 2); + invalid_event.sequence = 2; + + let mut entity = Entity::new(); + entity.load_from_history(vec![invalid_event]); + + match hydrate_from_snapshot::(entity, snapshot) { + Err(RepositoryError::Replay(message)) => { + assert!(message.contains("failed to upcast event Initialized")); + } + Err(other) => panic!("expected replay error, got {other:?}"), + Ok(_) => panic!("expected replay error"), + } +}