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