kmp_adapter_embedded/adapter/
replay.rs

1use kmp_domain::{ContextUpdatedEvent, PortError, ProjectionMutation};
2
3use super::engine::Table;
4use super::projection_write::apply_mutations_in_transaction;
5use super::store::EmbeddedKernelStore;
6
7/// Outcome of a projection rebuild from the append-only event log.
8#[derive(Debug, Clone, Copy, PartialEq, Eq)]
9pub struct ProjectionRebuildReport {
10    pub events_replayed: u64,
11    pub mutations_applied: u64,
12}
13
14impl EmbeddedKernelStore {
15    /// Drops every projection table and rebuilds them by replaying the event
16    /// log in sequence order — the recovery and migration story in one.
17    ///
18    /// The mutation derivation is injected so this adapter stays free of
19    /// application-layer dependencies; the composition root passes
20    /// `kmp_application::projection_mutations_for_context_event`.
21    /// The whole rebuild is one transaction: a crash mid-rebuild leaves the
22    /// previous projections intact.
23    pub async fn rebuild_projections<F>(
24        &self,
25        derive: F,
26    ) -> Result<ProjectionRebuildReport, PortError>
27    where
28        F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
29    {
30        self.run(move |store| {
31            // Freeze the event frontier under the same write lock as the
32            // rebuild. A concurrent condense cannot commit between the read
33            // of the log and replacement of its card projections.
34            let mut tx = store.begin_write()?;
35            let events = tx.scan_u64(Table::EventLog)?;
36            let events_replayed = events.len() as u64;
37            tx.clear(Table::Nodes)?;
38            tx.clear(Table::Relations)?;
39            tx.clear(Table::RelationsByTarget)?;
40            tx.clear(Table::Details)?;
41            // Cleared and rebuilt with Details, in this same transaction: the
42            // two tables are never observable out of step.
43            tx.clear(Table::DetailHeaders)?;
44            tx.clear(Table::Anchors)?;
45            tx.clear(Table::Cards)?;
46            tx.clear(Table::CardVersions)?;
47
48            let mut mutations_applied = 0;
49            for (_, raw) in events {
50                let event = super::serdes::decode::<ContextUpdatedEvent>("replay event", &raw)?;
51                mutations_applied += apply_mutations_in_transaction(tx.as_mut(), derive(&event)?)?;
52            }
53            tx.commit()?;
54
55            Ok(ProjectionRebuildReport {
56                events_replayed,
57                mutations_applied,
58            })
59        })
60        .await
61    }
62}
kmp_adapter_embedded/adapter/replay.rs

kmp_adapter_embedded/adapter/
replay.rs
