kmp_adapter_embedded/adapter/
portability.rs

1//! Export/import (E6): the append-only event log is the portable form of an
2//! embedded store. Export dumps it in sequence order; import replays it into
3//! an empty store, reproducing identical revisions, idempotency outcomes and
4//! projections — temporal reads and relation proof survive the round trip by
5//! construction.
6
7use kmp_domain::{ContextEventStore, ContextUpdatedEvent, PortError, ProjectionMutation};
8use serde::{Deserialize, Serialize};
9
10use super::replay::ProjectionRebuildReport;
11use super::store::EmbeddedKernelStore;
12
13/// First line of a bundle file: integrity metadata for fail-fast import.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15pub struct BundleHeader {
16    pub bundle_format: u32,
17    pub store_format: u32,
18    pub event_count: u64,
19    pub kernel_version: String,
20}
21
22pub const BUNDLE_FORMAT_VERSION: u32 = 1;
23
24/// Outcome of an import: events replayed and projections rebuilt.
25#[derive(Debug, Clone, Copy, PartialEq, Eq)]
26pub struct ImportReport {
27    pub events_imported: u64,
28    pub rebuild: ProjectionRebuildReport,
29}
30
31impl EmbeddedKernelStore {
32    /// Serializes the full event log as a JSON-Lines bundle: one header line
33    /// followed by one event per line, in sequence order.
34    pub async fn export_bundle(&self) -> Result<String, PortError> {
35        let events = self.run(EmbeddedKernelStore::read_event_log).await?;
36        let header = BundleHeader {
37            bundle_format: BUNDLE_FORMAT_VERSION,
38            store_format: super::format_version::EVENT_FORMAT_VERSION,
39            event_count: events.len() as u64,
40            kernel_version: env!("CARGO_PKG_VERSION").to_string(),
41        };
42        let mut out = String::new();
43        out.push_str(&encode_line("bundle header", &header)?);
44        for event in &events {
45            out.push_str(&encode_line("bundle event", event)?);
46        }
47        Ok(out)
48    }
49
50    /// Replays a bundle into this store. Fail-fast rules: the store must be
51    /// empty (no merge semantics in v1 — ADR-011 rationale applies), the
52    /// header must match supported formats, and every event must reproduce
53    /// exactly the revision it was exported with.
54    pub async fn import_bundle<F>(&self, bundle: &str, derive: F) -> Result<ImportReport, PortError>
55    where
56        F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
57    {
58        let (log_length, _) = self.event_log_stats().await?;
59        if log_length != 0 {
60            return Err(PortError::Conflict(format!(
61                "import requires an empty store; this store already holds {log_length} events \
62                 (merging bundles is not supported)"
63            )));
64        }
65
66        let mut lines = bundle.lines().filter(|line| !line.trim().is_empty());
67        let header: BundleHeader = decode_line(
68            "bundle header",
69            lines.next().ok_or_else(|| {
70                PortError::InvalidState("bundle is empty: missing header line".to_string())
71            })?,
72        )?;
73        if header.bundle_format != BUNDLE_FORMAT_VERSION {
74            return Err(PortError::InvalidState(format!(
75                "bundle format {} is not supported (this binary reads {})",
76                header.bundle_format, BUNDLE_FORMAT_VERSION
77            )));
78        }
79        if header.store_format != super::format_version::EVENT_FORMAT_VERSION {
80            return Err(PortError::InvalidState(format!(
81                "bundle was exported from store format {}, this binary supports {}",
82                header.store_format,
83                super::format_version::EVENT_FORMAT_VERSION
84            )));
85        }
86
87        let mut events = Vec::new();
88        for line in lines {
89            events.push(decode_line::<ContextUpdatedEvent>("bundle event", line)?);
90        }
91        let events_imported = self.replay_event_stream(events).await?;
92        if events_imported != header.event_count {
93            return Err(PortError::InvalidState(format!(
94                "bundle header declares {} events but {} were present",
95                header.event_count, events_imported
96            )));
97        }
98
99        let rebuild = self.rebuild_projections(derive).await?;
100        Ok(ImportReport {
101            events_imported,
102            rebuild,
103        })
104    }
105}
106
107impl EmbeddedKernelStore {
108    /// Replays a history into this store, in order, checking that every
109    /// event lands on the revision it was recorded with.
110    ///
111    /// That check is the whole point: a replay that silently renumbers
112    /// history would produce a store that reads plausibly and cites
113    /// revisions that never existed. Shared by import and migration, which
114    /// are the same operation seen from two different distances.
115    pub(crate) async fn replay_event_stream<I>(&self, events: I) -> Result<u64, PortError>
116    where
117        I: IntoIterator<Item = ContextUpdatedEvent>,
118    {
119        let mut replayed = 0u64;
120        for event in events {
121            let recorded_revision = event.revision;
122            let expected_previous = recorded_revision.checked_sub(1).ok_or_else(|| {
123                PortError::InvalidState("event carries revision 0; the log is corrupt".to_string())
124            })?;
125            let assigned = self.append(event, expected_previous).await?;
126            if assigned != recorded_revision {
127                return Err(PortError::Conflict(format!(
128                    "replay integrity violation: assigned revision {assigned}, \
129                     history recorded {recorded_revision}"
130                )));
131            }
132            replayed += 1;
133        }
134        Ok(replayed)
135    }
136}
137
138fn encode_line<T: Serialize>(what: &str, value: &T) -> Result<String, PortError> {
139    let mut line = serde_json::to_string(value)
140        .map_err(|error| PortError::InvalidState(format!("could not encode {what}: {error}")))?;
141    line.push('\n');
142    Ok(line)
143}
144
145fn decode_line<T: for<'de> Deserialize<'de>>(what: &str, line: &str) -> Result<T, PortError> {
146    serde_json::from_str(line)
147        .map_err(|error| PortError::InvalidState(format!("could not decode {what}: {error}")))
148}
kmp_adapter_embedded/adapter/portability.rs

kmp_adapter_embedded/adapter/
portability.rs
