arc_core/
projection.rs

1//! # Projection Module
2//!
3//! Three-trait architecture for building read models from event streams:
4//!
5//! - **[`Projector`]** — stateless event handler (the "machine"). Contains the pure
6//!   logic for transforming events into read model writes.
7//! - **[`Projection`]** — composed read model unit (the "output"). Ties a projector
8//!   to its storage backend.
9//! - **[`ReadModelStore`](crate::read_model_store::ReadModelStore)** — persistence
10//!   layer for projections. Backend-agnostic storage (defined in `read_model_store` module).
11//!
12//! ## Design Principles
13//!
14//! - **Separation of concerns**: Handler logic (projector) is separate from storage
15//!   (read model store) and orchestration (projection engine)
16//! - **Stateless projectors**: Projectors take `&self`, not `&mut self`. All mutable
17//!   state lives in the `ReadModelStore` via interior mutability.
18//! - **Rebuildable**: Projections can be rebuilt from scratch by replaying events
19//! - **Idempotent**: Handling the same event multiple times should be safe
20//! - **Composable**: One projector per read model concern; swap backends freely
21//!
22//! ## Example
23//!
24//! ```rust,ignore
25//! use arc_core::projection::{Projector, Projection, ProjectionUnit, ProjectionEngine};
26//! use arc_core::read_model_store::{ReadModelStore, InMemoryReadModelStore};
27//! use arc_core::event::Event;
28//! use std::sync::Arc;
29//!
30//! struct UserListProjector;
31//!
32//! #[async_trait]
33//! impl Projector for UserListProjector {
34//!     fn name(&self) -> &str { "UserList" }
35//!
36//!     fn handles(&self) -> Vec<String> {
37//!         vec!["UserCreated".to_string(), "ProfileUpdated".to_string()]
38//!     }
39//!
40//!     async fn apply(&self, event: &Event, store: &dyn ReadModelStore) -> ProjectionResult<()> {
41//!         match event.event_type.as_str() {
42//!             "UserCreated" => {
43//!                 store.upsert(Upsert::new("users_view", &event.aggregate_id, event.payload.clone())).await
44//!                     .map_err(|e| ProjectionError::handle_failed("UserList", &event.event_type, &event.event_id.to_string(), e.to_string()))?;
45//!             }
46//!             _ => {}
47//!         }
48//!         Ok(())
49//!     }
50//! }
51//!
52//! // Compose: projector + store = projection
53//! let store = Arc::new(InMemoryReadModelStore::new());
54//! let projection = ProjectionUnit::new(Box::new(UserListProjector), store, "users_view");
55//!
56//! // Register with engine
57//! let mut engine = ProjectionEngine::new(event_store);
58//! engine.register(Box::new(projection));
59//! engine.process(&event).await?;
60//! ```
61
62use crate::event::Event;
63#[cfg(test)]
64use crate::event::NewEvent;
65use crate::event_bus::EventHandler;
66use crate::event_store::EventStore;
67use crate::read_model_store::ReadModelStore;
68use async_trait::async_trait;
69use std::sync::Arc;
70use thiserror::Error;
71
72// ---------------------------------------------------------------------------
73// Errors
74// ---------------------------------------------------------------------------
75
76/// Errors that can occur during projection operations.
77#[derive(Debug, Error)]
78pub enum ProjectionError {
79    /// Error handling an event
80    #[error(
81        "Projection '{name}' failed to handle event {event_type} (event_id: {event_id}): {message}"
82    )]
83    HandleFailed {
84        name: String,
85        event_type: String,
86        event_id: String,
87        message: String,
88    },
89
90    /// Error clearing projection state
91    #[error("Projection '{name}' failed to clear: {message}")]
92    ClearFailed { name: String, message: String },
93
94    /// Error rebuilding projection
95    #[error("Projection '{name}' failed to rebuild: {message}")]
96    RebuildFailed { name: String, message: String },
97
98    /// Event store error during rebuild
99    #[error("Failed to load events for rebuild: {0}")]
100    EventStoreError(String),
101
102    /// Read model store error
103    #[error("Read model store error in projection '{name}': {message}")]
104    ReadModelError { name: String, message: String },
105
106    /// Other errors
107    #[error("Projection error: {message}")]
108    Other { message: String },
109}
110
111impl ProjectionError {
112    /// Create a handle failed error.
113    pub fn handle_failed(
114        name: impl Into<String>,
115        event_type: impl Into<String>,
116        event_id: impl Into<String>,
117        message: impl Into<String>,
118    ) -> Self {
119        ProjectionError::HandleFailed {
120            name: name.into(),
121            event_type: event_type.into(),
122            event_id: event_id.into(),
123            message: message.into(),
124        }
125    }
126
127    /// Create a clear failed error.
128    pub fn clear_failed(name: impl Into<String>, message: impl Into<String>) -> Self {
129        ProjectionError::ClearFailed {
130            name: name.into(),
131            message: message.into(),
132        }
133    }
134
135    /// Create a rebuild failed error.
136    pub fn rebuild_failed(name: impl Into<String>, message: impl Into<String>) -> Self {
137        ProjectionError::RebuildFailed {
138            name: name.into(),
139            message: message.into(),
140        }
141    }
142
143    /// Create a read model error.
144    pub fn read_model_error(name: impl Into<String>, message: impl Into<String>) -> Self {
145        ProjectionError::ReadModelError {
146            name: name.into(),
147            message: message.into(),
148        }
149    }
150
151    /// Create a generic error.
152    pub fn other(message: impl Into<String>) -> Self {
153        ProjectionError::Other {
154            message: message.into(),
155        }
156    }
157}
158
159/// Result type for projection operations.
160pub type ProjectionResult<T> = Result<T, ProjectionError>;
161
162// ---------------------------------------------------------------------------
163// Projector trait — the stateless event handler
164// ---------------------------------------------------------------------------
165
166/// A projector contains the pure event-handling logic for building a read model.
167///
168/// Projectors are stateless — they receive events and translate them into write
169/// operations against a [`ReadModelStore`]. They do not own the store or the
170/// read model state.
171///
172/// # Design
173///
174/// - **Stateless**: all state lives in the `ReadModelStore`
175/// - **Deterministic**: same events + empty store = same read model
176/// - **Composable**: one projector per read model concern
177/// - **`&self`**: safe to share across threads
178///
179/// # Idempotency
180///
181/// `apply()` should be idempotent — handling the same event twice must produce
182/// the same result. Use UPSERT, check event_id, or make operations naturally
183/// idempotent (SET vs INCREMENT).
184///
185/// # Example
186///
187/// ```rust,ignore
188/// struct UserListProjector;
189///
190/// #[async_trait]
191/// impl Projector for UserListProjector {
192///     fn name(&self) -> &str { "UserList" }
193///
194///     fn handles(&self) -> Vec<String> {
195///         vec!["UserCreated".to_string()]
196///     }
197///
198///     async fn apply(&self, event: &Event, store: &dyn ReadModelStore) -> ProjectionResult<()> {
199///         store.upsert(Upsert::new("users_view", &event.aggregate_id, event.payload.clone())).await
200///             .map_err(|e| ProjectionError::handle_failed(
201///                 "UserList", &event.event_type, &event.event_id.to_string(), e.to_string()
202///             ))?;
203///         Ok(())
204///     }
205/// }
206/// ```
207#[async_trait]
208pub trait Projector: Send + Sync {
209    /// Unique name identifying this projector.
210    ///
211    /// Used for logging, monitoring, position tracking, and rebuild targeting.
212    fn name(&self) -> &str;
213
214    /// Event types this projector handles.
215    ///
216    /// Only events whose `event_type` is in this list will be passed to `apply()`.
217    fn handles(&self) -> Vec<String>;
218
219    /// Apply a single event to the read model via the store.
220    ///
221    /// This method should be idempotent: applying the same event twice
222    /// must produce the same result.
223    async fn apply(&self, event: &Event, store: &dyn ReadModelStore) -> ProjectionResult<()>;
224
225    /// Initialize the read model schema (CREATE TABLE IF NOT EXISTS, etc.).
226    ///
227    /// Called once when the projector is first registered and before rebuilds.
228    /// Default implementation does nothing (for stores that don't need schema setup).
229    async fn init(&self, _store: &dyn ReadModelStore) -> ProjectionResult<()> {
230        Ok(())
231    }
232}
233
234// ---------------------------------------------------------------------------
235// Projection trait — the composed read model unit
236// ---------------------------------------------------------------------------
237
238/// A projection is the composed unit of a projector + its read model store.
239///
240/// It represents a complete, self-contained read model: the logic that transforms
241/// events into state, paired with the storage where that state lives.
242///
243/// Most users don't implement this trait directly. Instead, implement [`Projector`]
244/// and compose it with a [`ReadModelStore`] via [`ProjectionUnit`].
245///
246/// # `&self` not `&mut self`
247///
248/// All methods take `&self`. Mutable state lives in the `ReadModelStore`, which
249/// handles interior mutability via connection pools, `Mutex`, etc.
250#[async_trait]
251pub trait Projection: Send + Sync {
252    /// Projection name (delegates to the projector).
253    fn name(&self) -> &str;
254
255    /// Event types this projection handles (delegates to the projector).
256    fn handles(&self) -> Vec<String>;
257
258    /// Handle a single event by applying it through the projector to the store.
259    async fn handle(&self, event: &Event) -> ProjectionResult<()>;
260
261    /// Clear all read model state for this projection.
262    async fn clear(&self) -> ProjectionResult<()>;
263
264    /// Rebuild from a set of events: clear, then replay matching events.
265    async fn rebuild(&self, events: Vec<Event>) -> ProjectionResult<()> {
266        self.clear().await?;
267        for event in events {
268            if self.handles().contains(&event.event_type) {
269                self.handle(&event).await?;
270            }
271        }
272        Ok(())
273    }
274}
275
276// ---------------------------------------------------------------------------
277// ProjectionUnit — standard composition glue
278// ---------------------------------------------------------------------------
279
280/// Standard composition of a [`Projector`] and a [`ReadModelStore`].
281///
282/// This is the typical way to create a [`Projection`]: provide the event-handling
283/// logic (projector) and the storage backend (store), and `ProjectionUnit` wires
284/// them together.
285///
286/// # Example
287///
288/// ```rust,ignore
289/// let projector = Box::new(UserListProjector);
290/// let store: Arc<dyn ReadModelStore> = Arc::new(SqliteReadModelStore::new(pool));
291/// let projection = ProjectionUnit::new(projector, store, "users_view");
292/// engine.register(Box::new(projection));
293/// ```
294pub struct ProjectionUnit {
295    projector: Box<dyn Projector>,
296    store: Arc<dyn ReadModelStore>,
297    /// Table/collection name used for `clear()` (truncate target).
298    table: String,
299}
300
301impl ProjectionUnit {
302    /// Create a new projection unit.
303    ///
304    /// # Arguments
305    ///
306    /// - `projector`: The stateless event handler
307    /// - `store`: The read model storage backend
308    /// - `table`: Table/collection name to truncate on `clear()`
309    pub fn new(
310        projector: Box<dyn Projector>,
311        store: Arc<dyn ReadModelStore>,
312        table: impl Into<String>,
313    ) -> Self {
314        Self {
315            projector,
316            store,
317            table: table.into(),
318        }
319    }
320}
321
322#[async_trait]
323impl Projection for ProjectionUnit {
324    fn name(&self) -> &str {
325        self.projector.name()
326    }
327
328    fn handles(&self) -> Vec<String> {
329        self.projector.handles()
330    }
331
332    async fn handle(&self, event: &Event) -> ProjectionResult<()> {
333        self.projector.apply(event, self.store.as_ref()).await
334    }
335
336    async fn clear(&self) -> ProjectionResult<()> {
337        self.store
338            .truncate(&self.table)
339            .await
340            .map_err(|e| ProjectionError::clear_failed(self.projector.name(), e.to_string()))
341    }
342}
343
344// ---------------------------------------------------------------------------
345// ProjectionEngine — orchestrates multiple projections
346// ---------------------------------------------------------------------------
347
348/// Engine for managing multiple projections.
349///
350/// The `ProjectionEngine`:
351/// - Registers fully composed [`Projection`] instances
352/// - Routes events to interested projections
353/// - Rebuilds projections from the event store
354/// - Provides convenience registration via [`register_projector`](Self::register_projector)
355///
356/// # Example
357///
358/// ```rust,ignore
359/// let event_store = Box::new(sqlite_event_store);
360/// let mut engine = ProjectionEngine::new(event_store);
361///
362/// // Option 1: register a pre-composed projection
363/// engine.register(Box::new(projection_unit));
364///
365/// // Option 2: convenience — register projector + store directly
366/// engine.register_projector(Box::new(UserListProjector), store.clone(), "users_view");
367///
368/// // Process events
369/// engine.process(&event).await?;
370///
371/// // Rebuild all projections from event store
372/// engine.rebuild_all().await?;
373/// ```
374pub struct ProjectionEngine {
375    projections: Vec<Box<dyn Projection>>,
376    event_store: Box<dyn EventStore>,
377}
378
379impl ProjectionEngine {
380    /// Create a new projection engine.
381    pub fn new(event_store: Box<dyn EventStore>) -> Self {
382        Self {
383            projections: Vec::new(),
384            event_store,
385        }
386    }
387
388    /// Register a fully composed projection.
389    pub fn register(&mut self, projection: Box<dyn Projection>) {
390        tracing::info!("Registering projection: {}", projection.name());
391        self.projections.push(projection);
392    }
393
394    /// Convenience: register a projector + store as a [`ProjectionUnit`].
395    pub fn register_projector(
396        &mut self,
397        projector: Box<dyn Projector>,
398        store: Arc<dyn ReadModelStore>,
399        table: impl Into<String>,
400    ) {
401        let unit = ProjectionUnit::new(projector, store, table);
402        self.register(Box::new(unit));
403    }
404
405    /// Process a single event through all interested projections.
406    ///
407    /// Routes the event to projections whose `handles()` includes the event type.
408    pub async fn process(&self, event: &Event) -> ProjectionResult<()> {
409        for projection in &self.projections {
410            if projection.handles().contains(&event.event_type) {
411                tracing::debug!(
412                    "Processing event {} ({}) in projection {}",
413                    event.event_type,
414                    event.event_id,
415                    projection.name()
416                );
417
418                projection.handle(event).await.map_err(|e| {
419                    ProjectionError::handle_failed(
420                        projection.name(),
421                        &event.event_type,
422                        event.event_id.to_string(),
423                        e.to_string(),
424                    )
425                })?;
426            }
427        }
428        Ok(())
429    }
430
431    /// Process multiple events in sequence.
432    pub async fn process_batch(&self, events: Vec<Event>) -> ProjectionResult<()> {
433        for event in events {
434            self.process(&event).await?;
435        }
436        Ok(())
437    }
438
439    /// Rebuild all registered projections from the event store.
440    pub async fn rebuild_all(&self) -> ProjectionResult<()> {
441        tracing::info!("Rebuilding all projections");
442
443        let events = self
444            .event_store
445            .stream_all(0)
446            .await
447            .map_err(|e| ProjectionError::EventStoreError(e.to_string()))?;
448
449        tracing::info!("Loaded {} events for rebuild", events.len());
450
451        for projection in &self.projections {
452            tracing::info!("Rebuilding projection: {}", projection.name());
453
454            projection
455                .rebuild(events.clone())
456                .await
457                .map_err(|e| ProjectionError::rebuild_failed(projection.name(), e.to_string()))?;
458
459            tracing::info!("Rebuilt projection: {}", projection.name());
460        }
461
462        Ok(())
463    }
464
465    /// Rebuild a specific projection by name.
466    pub async fn rebuild_projection(&self, name: &str) -> ProjectionResult<()> {
467        tracing::info!("Rebuilding projection: {}", name);
468
469        let projection = self
470            .projections
471            .iter()
472            .find(|p| p.name() == name)
473            .ok_or_else(|| ProjectionError::other(format!("Projection not found: {}", name)))?;
474
475        let events = self
476            .event_store
477            .stream_all(0)
478            .await
479            .map_err(|e| ProjectionError::EventStoreError(e.to_string()))?;
480
481        projection
482            .rebuild(events)
483            .await
484            .map_err(|e| ProjectionError::rebuild_failed(name, e.to_string()))?;
485
486        tracing::info!("Rebuilt projection: {}", name);
487        Ok(())
488    }
489
490    /// Get number of registered projections.
491    pub fn projection_count(&self) -> usize {
492        self.projections.len()
493    }
494
495    /// Get names of all registered projections.
496    pub fn projection_names(&self) -> Vec<String> {
497        self.projections
498            .iter()
499            .map(|p| p.name().to_string())
500            .collect()
501    }
502
503    /// Union of every event type any registered projection handles. Used by
504    /// [`ProjectionEngineHandler`] to declare its `handles()` set when
505    /// subscribing to an [`EventBus`](crate::event_bus::EventBus).
506    pub fn all_handled_event_types(&self) -> Vec<String> {
507        let mut all: Vec<String> = self.projections.iter().flat_map(|p| p.handles()).collect();
508        all.sort();
509        all.dedup();
510        all
511    }
512}
513
514// ---------------------------------------------------------------------------
515// EventBus adapter — drive the engine from an in-process bus
516// ---------------------------------------------------------------------------
517
518/// Adapter that lets a [`ProjectionEngine`] subscribe to an
519/// [`EventBus`](crate::event_bus::EventBus). Wraps the engine in an
520/// [`EventHandler`] that routes every relevant event through
521/// [`ProjectionEngine::process`]. The engine stays accessible from outside
522/// (e.g. for `rebuild_all`) via the same [`Arc`].
523///
524/// Lives in `arc-core` because the adapter needs nothing app-specific —
525/// any aggregate's projector can be driven through it.
526pub struct ProjectionEngineHandler {
527    engine: Arc<ProjectionEngine>,
528    handles: Vec<String>,
529}
530
531impl ProjectionEngineHandler {
532    pub fn new(engine: Arc<ProjectionEngine>) -> Self {
533        let handles = engine.all_handled_event_types();
534        Self { engine, handles }
535    }
536}
537
538#[async_trait]
539impl EventHandler for ProjectionEngineHandler {
540    fn handles(&self) -> Vec<String> {
541        self.handles.clone()
542    }
543
544    async fn handle(&self, event: &Event) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
545        self.engine
546            .process(event)
547            .await
548            .map_err(|e| -> Box<dyn std::error::Error + Send + Sync> { Box::new(e) })
549    }
550}
551
552// ===========================================================================
553// Tests
554// ===========================================================================
555
556#[cfg(test)]
557mod tests {
558    use super::*;
559    use crate::event_store::{EventStore, EventStoreResult, VersionCheck};
560    use crate::read_model_store::InMemoryReadModelStore;
561    use std::sync::{Arc, Mutex};
562
563    // -----------------------------------------------------------------------
564    // Mock event store (unchanged — needed for ProjectionEngine)
565    // -----------------------------------------------------------------------
566
567    struct MockEventStore {
568        events: Arc<Mutex<Vec<Event>>>,
569    }
570
571    impl MockEventStore {
572        fn new() -> Self {
573            Self {
574                events: Arc::new(Mutex::new(Vec::new())),
575            }
576        }
577
578        fn add_event(&self, event: Event) {
579            self.events.lock().unwrap().push(event);
580        }
581    }
582
583    #[async_trait]
584    impl EventStore for MockEventStore {
585        async fn append(
586            &self,
587            _aggregate_id: &str,
588            _version_check: VersionCheck,
589            events: Vec<Event>,
590        ) -> EventStoreResult<()> {
591            self.events.lock().unwrap().extend(events);
592            Ok(())
593        }
594
595        async fn load(&self, aggregate_id: &str) -> EventStoreResult<Vec<Event>> {
596            Ok(self
597                .events
598                .lock()
599                .unwrap()
600                .iter()
601                .filter(|e| e.aggregate_id == aggregate_id)
602                .cloned()
603                .collect())
604        }
605
606        async fn load_from(
607            &self,
608            aggregate_id: &str,
609            from_sequence: i64,
610        ) -> EventStoreResult<Vec<Event>> {
611            Ok(self
612                .events
613                .lock()
614                .unwrap()
615                .iter()
616                .filter(|e| e.aggregate_id == aggregate_id && e.sequence >= from_sequence)
617                .cloned()
618                .collect())
619        }
620
621        async fn stream_all(&self, _from_position: i64) -> EventStoreResult<Vec<Event>> {
622            Ok(self.events.lock().unwrap().clone())
623        }
624
625        async fn get_version(&self, aggregate_id: &str) -> EventStoreResult<i64> {
626            Ok(self
627                .events
628                .lock()
629                .unwrap()
630                .iter()
631                .filter(|e| e.aggregate_id == aggregate_id)
632                .map(|e| e.sequence)
633                .max()
634                .unwrap_or(0))
635        }
636    }
637
638    // -----------------------------------------------------------------------
639    // Mock projector — stateless, writes to ReadModelStore
640    // -----------------------------------------------------------------------
641
642    struct MockProjector {
643        name: String,
644        handles_types: Vec<String>,
645    }
646
647    impl MockProjector {
648        fn new(name: &str, handles: Vec<String>) -> Self {
649            Self {
650                name: name.to_string(),
651                handles_types: handles,
652            }
653        }
654    }
655
656    #[async_trait]
657    impl Projector for MockProjector {
658        fn name(&self) -> &str {
659            &self.name
660        }
661
662        fn handles(&self) -> Vec<String> {
663            self.handles_types.clone()
664        }
665
666        async fn apply(&self, event: &Event, store: &dyn ReadModelStore) -> ProjectionResult<()> {
667            use crate::read_model_store::Upsert;
668            store
669                .upsert(Upsert::new(
670                    "test_table",
671                    event.event_id.to_string(),
672                    serde_json::json!({
673                        "id": event.event_id.to_string(),
674                        "event_type": event.event_type,
675                        "version": event.sequence,
676                    }),
677                ))
678                .await
679                .map_err(|e| {
680                    ProjectionError::handle_failed(
681                        &self.name,
682                        &event.event_type,
683                        event.event_id.to_string(),
684                        e.to_string(),
685                    )
686                })?;
687            Ok(())
688        }
689    }
690
691    // -----------------------------------------------------------------------
692    // Helper to build a projection from mock projector + in-memory store
693    // -----------------------------------------------------------------------
694
695    fn make_projection(
696        name: &str,
697        handles: Vec<String>,
698        store: Arc<InMemoryReadModelStore>,
699    ) -> Box<ProjectionUnit> {
700        Box::new(ProjectionUnit::new(
701            Box::new(MockProjector::new(name, handles)),
702            store,
703            "test_table",
704        ))
705    }
706
707    // -----------------------------------------------------------------------
708    // Tests
709    // -----------------------------------------------------------------------
710
711    #[tokio::test]
712    async fn test_projection_engine_new() {
713        let store = Box::new(MockEventStore::new());
714        let engine = ProjectionEngine::new(store);
715        assert_eq!(engine.projection_count(), 0);
716    }
717
718    #[tokio::test]
719    async fn test_register_projection() {
720        let store = Box::new(MockEventStore::new());
721        let mut engine = ProjectionEngine::new(store);
722
723        let rm_store = Arc::new(InMemoryReadModelStore::new());
724        let projection = make_projection("Test", vec!["TestEvent".to_string()], rm_store);
725        engine.register(projection);
726
727        assert_eq!(engine.projection_count(), 1);
728        assert_eq!(engine.projection_names(), vec!["Test"]);
729    }
730
731    #[tokio::test]
732    async fn test_process_event() {
733        let store = Box::new(MockEventStore::new());
734        let mut engine = ProjectionEngine::new(store);
735
736        let rm_store = Arc::new(InMemoryReadModelStore::new());
737        let projection = make_projection("Test", vec!["UserCreated".to_string()], rm_store.clone());
738        engine.register(projection);
739
740        let event = Event::new(NewEvent {
741            aggregate_type: "User",
742            aggregate_id: "user-1",
743            sequence: 1,
744            event_type: "UserCreated",
745            payload: serde_json::json!({"name": "Alice"}),
746        });
747
748        engine.process(&event).await.unwrap();
749
750        assert_eq!(rm_store.get_rows("test_table").len(), 1);
751    }
752
753    #[tokio::test]
754    async fn test_projection_filtering() {
755        let store = Box::new(MockEventStore::new());
756        let mut engine = ProjectionEngine::new(store);
757
758        let rm_store = Arc::new(InMemoryReadModelStore::new());
759        let projection = make_projection("Test", vec!["UserCreated".to_string()], rm_store.clone());
760        engine.register(projection);
761
762        // Event that should be handled
763        let event1 = Event::new(NewEvent {
764            aggregate_type: "User",
765            aggregate_id: "user-1",
766            sequence: 1,
767            event_type: "UserCreated",
768            payload: serde_json::json!({}),
769        });
770        engine.process(&event1).await.unwrap();
771
772        // Event that should be filtered out
773        let event2 = Event::new(NewEvent {
774            aggregate_type: "User",
775            aggregate_id: "user-1",
776            sequence: 2,
777            event_type: "UserDeleted",
778            payload: serde_json::json!({}),
779        });
780        engine.process(&event2).await.unwrap();
781
782        assert_eq!(rm_store.get_rows("test_table").len(), 1);
783    }
784
785    #[tokio::test]
786    async fn test_rebuild_all() {
787        let event_store = MockEventStore::new();
788
789        event_store.add_event(Event::new(NewEvent {
790            aggregate_type: "User",
791            aggregate_id: "user-1",
792            sequence: 1,
793            event_type: "UserCreated",
794            payload: serde_json::json!({}),
795        }));
796        event_store.add_event(Event::new(NewEvent {
797            aggregate_type: "User",
798            aggregate_id: "user-2",
799            sequence: 1,
800            event_type: "UserCreated",
801            payload: serde_json::json!({}),
802        }));
803
804        let mut engine = ProjectionEngine::new(Box::new(event_store));
805
806        let rm_store = Arc::new(InMemoryReadModelStore::new());
807        let projection = make_projection("Test", vec!["UserCreated".to_string()], rm_store.clone());
808        engine.register(projection);
809
810        engine.rebuild_all().await.unwrap();
811
812        assert_eq!(rm_store.get_rows("test_table").len(), 2);
813    }
814
815    #[tokio::test]
816    async fn test_multiple_projections() {
817        let store = Box::new(MockEventStore::new());
818        let mut engine = ProjectionEngine::new(store);
819
820        let rm_store1 = Arc::new(InMemoryReadModelStore::new());
821        let rm_store2 = Arc::new(InMemoryReadModelStore::new());
822
823        let proj1 = make_projection(
824            "Projection1",
825            vec!["UserCreated".to_string()],
826            rm_store1.clone(),
827        );
828        let proj2 = make_projection(
829            "Projection2",
830            vec!["UserCreated".to_string(), "UserDeleted".to_string()],
831            rm_store2.clone(),
832        );
833
834        engine.register(proj1);
835        engine.register(proj2);
836
837        let event = Event::new(NewEvent {
838            aggregate_type: "User",
839            aggregate_id: "user-1",
840            sequence: 1,
841            event_type: "UserCreated",
842            payload: serde_json::json!({}),
843        });
844        engine.process(&event).await.unwrap();
845
846        assert_eq!(rm_store1.get_rows("test_table").len(), 1);
847        assert_eq!(rm_store2.get_rows("test_table").len(), 1);
848    }
849
850    #[tokio::test]
851    async fn test_process_batch() {
852        let store = Box::new(MockEventStore::new());
853        let mut engine = ProjectionEngine::new(store);
854
855        let rm_store = Arc::new(InMemoryReadModelStore::new());
856        let projection = make_projection("Test", vec!["UserCreated".to_string()], rm_store.clone());
857        engine.register(projection);
858
859        let events = vec![
860            Event::new(NewEvent {
861                aggregate_type: "User",
862                aggregate_id: "user-1",
863                sequence: 1,
864                event_type: "UserCreated",
865                payload: serde_json::json!({}),
866            }),
867            Event::new(NewEvent {
868                aggregate_type: "User",
869                aggregate_id: "user-2",
870                sequence: 1,
871                event_type: "UserCreated",
872                payload: serde_json::json!({}),
873            }),
874            Event::new(NewEvent {
875                aggregate_type: "User",
876                aggregate_id: "user-3",
877                sequence: 1,
878                event_type: "UserCreated",
879                payload: serde_json::json!({}),
880            }),
881        ];
882
883        engine.process_batch(events).await.unwrap();
884
885        assert_eq!(rm_store.get_rows("test_table").len(), 3);
886    }
887
888    #[tokio::test]
889    async fn test_projector_init_default() {
890        let projector = MockProjector::new("Test", vec![]);
891        let store = InMemoryReadModelStore::new();
892        // Default init() should succeed (no-op)
893        projector.init(&store).await.unwrap();
894    }
895
896    #[tokio::test]
897    async fn test_register_projector_convenience() {
898        let event_store = Box::new(MockEventStore::new());
899        let mut engine = ProjectionEngine::new(event_store);
900
901        let rm_store: Arc<dyn ReadModelStore> = Arc::new(InMemoryReadModelStore::new());
902        engine.register_projector(
903            Box::new(MockProjector::new("Convenient", vec!["X".to_string()])),
904            rm_store,
905            "my_table",
906        );
907
908        assert_eq!(engine.projection_count(), 1);
909        assert_eq!(engine.projection_names(), vec!["Convenient"]);
910    }
911}
arc_core/projection.rs

arc_core/
projection.rs
