feat: add event codec and per-entity DEK helper

Cargo.toml

@@ -3,7 +3,7 @@ name = "es-entity"
 description = "Event Sourcing Entity Framework"
 repository = "https://github.com/GaloyMoney/es-entity"
 documentation = "https://docs.rs/es-entity"
-version = "0.10.36-dev"
+version = "0.10.35"
 edition = "2024"
 license = "Apache-2.0"
 categories = ["data-structures", "database"]
@@ -62,7 +62,7 @@ members = [
 
 [workspace.dependencies]
 
-es-entity-macros = { path = "es-entity-macros", version = "0.10.36-dev" }
+es-entity-macros = { path = "es-entity-macros", version = "0.10.35" }
 
 anyhow = "1.0"
 async-graphql = { version = "8.0.0-rc.4", default-features = false }

es-entity-macros/Cargo.toml

@@ -2,7 +2,7 @@
 name = "es-entity-macros"
 description = "Proc macros for es-entity"
 repository = "https://github.com/GaloyMoney/cala"
-version = "0.10.36-dev"
+version = "0.10.35"
 edition = "2024"
 license = "Apache-2.0"
 categories = ["data-structures", "database"]

es-entity-macros/src/repo/mod.rs

@@ -191,6 +191,7 @@ impl ToTokens for EsRepo<'_> {
         let modify_error = self.opts.modify_error();
         let find_error = self.opts.find_error();
         let query_error = self.opts.query_error();
+        let event_payload_codec = self.opts.event_payload_codec();
         let error_types = self.error_types.generate();
         let map_constraint_fn = self.error_types.generate_map_constraint_fn();
 
@@ -255,6 +256,7 @@ impl ToTokens for EsRepo<'_> {
 
             impl #impl_generics es_entity::EsRepo for #repo #ty_generics #where_clause {
                 type Entity = #entity;
+                type EventPayloadCodec = #event_payload_codec;
                 type CreateError = #create_error;
                 type ModifyError = #modify_error;
                 type FindError = #find_error;

es-entity-macros/src/repo/options/mod.rs

@@ -9,6 +9,20 @@ use quote::quote;
 pub use columns::*;
 pub use delete::*;
 
+#[derive(Debug, Clone)]
+pub struct EventPayloadCodecConfig {
+    pub path: syn::Path,
+}
+
+impl FromMeta for EventPayloadCodecConfig {
+    fn from_string(value: &str) -> darling::Result<Self> {
+        Ok(Self {
+            path: syn::parse_str(value)
+                .map_err(|e| darling::Error::custom(format!("invalid codec path: {e}")))?,
+        })
+    }
+}
+
 #[derive(Debug, Clone)]
 pub struct PostPersistHookConfig {
     pub method: syn::Ident,
@@ -237,6 +251,8 @@ pub struct RepositoryOptions {
 
     #[darling(default)]
     persist_event_context: Option<bool>,
+    #[darling(default)]
+    event_payload_codec: Option<EventPayloadCodecConfig>,
 }
 
 impl RepositoryOptions {
@@ -309,6 +325,15 @@ impl RepositoryOptions {
         }
     }
 
+    pub fn event_payload_codec(&self) -> proc_macro2::TokenStream {
+        if let Some(codec) = &self.event_payload_codec {
+            let path = &codec.path;
+            quote! { #path }
+        } else {
+            quote! { es_entity::JsonEventPayloadCodec }
+        }
+    }
+
     pub fn events_table_name(&self) -> &str {
         self.events_table_name
             .as_ref()

es-entity-macros/src/repo/persist_events_batch_fn.rs

@@ -91,7 +91,7 @@ impl ToTokens for PersistEventsBatchFn<'_> {
                     let id = events.id();
                     let offset = events.len_persisted() + 1;
                     let types = events.new_event_types();
-                    let serialized = events.serialize_new_events();
+                    let serialized = events.serialize_new_events_with_codec::<<Self as es_entity::EsRepo>::EventPayloadCodec>();
                     #ctx_extend
 
                     let n_events = serialized.len();
@@ -168,7 +168,7 @@ mod tests {
                     let id = events.id();
                     let offset = events.len_persisted() + 1;
                     let types = events.new_event_types();
-                    let serialized = events.serialize_new_events();
+                    let serialized = events.serialize_new_events_with_codec::<<Self as es_entity::EsRepo>::EventPayloadCodec>();
                     let contexts = events.serialize_new_event_contexts();
                     if let Some(contexts) = contexts {
                         all_contexts.extend(contexts);
@@ -248,7 +248,7 @@ mod tests {
                     let id = events.id();
                     let offset = events.len_persisted() + 1;
                     let types = events.new_event_types();
-                    let serialized = events.serialize_new_events();
+                    let serialized = events.serialize_new_events_with_codec::<<Self as es_entity::EsRepo>::EventPayloadCodec>();
 
                     let n_events = serialized.len();
                     all_serialized.extend(serialized);

es-entity-macros/src/repo/persist_events_fn.rs

@@ -78,7 +78,7 @@ impl ToTokens for PersistEventsFn<'_> {
                 let id = events.id();
                 let offset = events.len_persisted();
                 let events_types = events.new_event_types();
-                let serialized_events = events.serialize_new_events();
+                let serialized_events = events.serialize_new_events_with_codec::<<Self as es_entity::EsRepo>::EventPayloadCodec>();
                 #ctx_var
                 let now = op.maybe_now();
 
@@ -144,7 +144,7 @@ mod tests {
                 let id = events.id();
                 let offset = events.len_persisted();
                 let events_types = events.new_event_types();
-                let serialized_events = events.serialize_new_events();
+                let serialized_events = events.serialize_new_events_with_codec::<<Self as es_entity::EsRepo>::EventPayloadCodec>();
                 let contexts = events.serialize_new_event_contexts();
                 let now = op.maybe_now();
 
@@ -207,7 +207,7 @@ mod tests {
                 let id = events.id();
                 let offset = events.len_persisted();
                 let events_types = events.new_event_types();
-                let serialized_events = events.serialize_new_events();
+                let serialized_events = events.serialize_new_events_with_codec::<<Self as es_entity::EsRepo>::EventPayloadCodec>();
                 let now = op.maybe_now();
 
                 let rows = sqlx::query!(

es-entity-macros/src/repo/populate_nested.rs

@@ -77,7 +77,7 @@ impl ToTokens for PopulateNested<'_> {
                         ).fetch_all(op.as_executor()).await?
                     };
                     let n = rows.len();
-                    let (mut res, _) = es_entity::EntityEvents::load_n::<<Self as EsRepo>::Entity>(rows.into_iter(), n)?;
+                    let (mut res, _) = es_entity::EntityEvents::load_n_with_codec::<<Self as es_entity::EsRepo>::Entity, <Self as es_entity::EsRepo>::EventPayloadCodec>(rows.into_iter(), n)?;
                     Self::load_all_nested_in_op_include_deleted::<_, __EsErr>(op, &mut res).await?;
                     for entity in res.into_iter() {
                         let parent = lookup.get_mut(&entity.#accessor).expect("parent not present");
@@ -111,7 +111,7 @@ impl ToTokens for PopulateNested<'_> {
                         ).fetch_all(op.as_executor()).await?
                     };
                     let n = rows.len();
-                    let (mut res, _) = es_entity::EntityEvents::load_n::<<Self as EsRepo>::Entity>(rows.into_iter(), n)?;
+                    let (mut res, _) = es_entity::EntityEvents::load_n_with_codec::<<Self as es_entity::EsRepo>::Entity, <Self as es_entity::EsRepo>::EventPayloadCodec>(rows.into_iter(), n)?;
                     Self::load_all_nested_in_op::<_, __EsErr>(op, &mut res).await?;
                     for entity in res.into_iter() {
                         let parent = lookup.get_mut(&entity.#accessor).expect("parent not present");

src/crypto_shred.rs

@@ -0,0 +1,80 @@
+//! Helpers for event-sourced entities that use per-entity encrypted data keys.
+//!
+//! The crate deliberately does not prescribe a cryptography implementation.
+//! Applications can store their own encrypted data-key type in mutable
+//! projection state and use this trait to expose the key lifecycle consistently.
+
+/// Whether an entity still has an encrypted data key or has been crypto-shredded.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum PerEntityDataKeyState {
+    Present,
+    Shredded,
+}
+
+/// Contract for entities whose sensitive payloads are encrypted by a
+/// per-entity data key.
+///
+/// Implementors should keep the encrypted data key outside immutable event
+/// payloads, attach it to hydrated state when the projection row is loaded, and
+/// return `None` after the data key has been intentionally removed. Historical
+/// ciphertext may remain in the event stream, but without the data key the
+/// application should treat sensitive payloads as unrecoverable.
+pub trait PerEntityDataKey {
+    type EncryptedDataKey;
+
+    fn encrypted_data_key(&self) -> Option<&Self::EncryptedDataKey>;
+
+    fn attach_encrypted_data_key(&mut self, encrypted_data_key: Self::EncryptedDataKey);
+
+    fn clear_encrypted_data_key(&mut self);
+
+    fn data_key_state(&self) -> PerEntityDataKeyState {
+        if self.encrypted_data_key().is_some() {
+            PerEntityDataKeyState::Present
+        } else {
+            PerEntityDataKeyState::Shredded
+        }
+    }
+
+    fn is_crypto_shredded(&self) -> bool {
+        self.data_key_state() == PerEntityDataKeyState::Shredded
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    struct EntityWithKey(Option<String>);
+
+    impl PerEntityDataKey for EntityWithKey {
+        type EncryptedDataKey = String;
+
+        fn encrypted_data_key(&self) -> Option<&Self::EncryptedDataKey> {
+            self.0.as_ref()
+        }
+
+        fn attach_encrypted_data_key(&mut self, encrypted_data_key: Self::EncryptedDataKey) {
+            self.0 = Some(encrypted_data_key);
+        }
+
+        fn clear_encrypted_data_key(&mut self) {
+            self.0 = None;
+        }
+    }
+
+    #[test]
+    fn data_key_state_tracks_key_lifecycle() {
+        let mut entity = EntityWithKey(None);
+        assert_eq!(entity.data_key_state(), PerEntityDataKeyState::Shredded);
+        assert!(entity.is_crypto_shredded());
+
+        entity.attach_encrypted_data_key("wrapped-key".to_string());
+        assert_eq!(entity.data_key_state(), PerEntityDataKeyState::Present);
+        assert!(!entity.is_crypto_shredded());
+
+        entity.clear_encrypted_data_key();
+        assert_eq!(entity.data_key_state(), PerEntityDataKeyState::Shredded);
+        assert!(entity.is_crypto_shredded());
+    }
+}