1use std::collections::{BTreeMap, BTreeSet};
8use std::time::{Duration, SystemTime, UNIX_EPOCH};
9
10use kmp_domain::{ContextEventStore, ContextUpdatedEvent, PortError, ProjectionMutation};
11use serde::{Deserialize, Serialize};
12use sha2::{Digest, Sha256};
13
14use super::replay::ProjectionRebuildReport;
15use super::store::EmbeddedKernelStore;
16
17#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)]
23pub struct BundleEventRange {
24 pub first: Option<u64>,
25 pub last: Option<u64>,
26}
27
28#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
31pub struct BundleHeader {
32 pub bundle_format: u32,
33 pub event_format: u32,
35 pub event_count: u64,
36 pub kernel_version: String,
37 #[serde(default)]
38 pub snapshot_id: String,
39 #[serde(default)]
40 pub created_at_unix_ms: u64,
41 #[serde(default)]
42 pub event_range: BundleEventRange,
43 #[serde(default)]
44 pub abouts: Vec<String>,
45 #[serde(default)]
46 pub content_digest: String,
47}
48
49pub const BUNDLE_FORMAT_VERSION: u32 = 3;
50
51#[derive(Debug, Clone, Copy, PartialEq, Eq)]
53pub struct ImportReport {
54 pub events_imported: u64,
55 pub rebuild: ProjectionRebuildReport,
56}
57
58impl EmbeddedKernelStore {
59 pub fn export_bundle_blocking(&self) -> Result<String, PortError> {
62 encode_bundle(&self.read_event_log()?, None)
63 }
64
65 pub async fn export_bundle(&self) -> Result<String, PortError> {
68 self.run(|store| store.export_bundle_blocking()).await
69 }
70
71 pub async fn export_bundle_for_abouts(
80 &self,
81 requested_abouts: &[String],
82 ) -> Result<String, PortError> {
83 let events = self.run(EmbeddedKernelStore::read_event_log).await?;
84 let events = filter_events_for_abouts(events, requested_abouts)?;
85 encode_bundle(&events, None)
86 }
87
88 pub fn export_bundle_excluding_abouts_blocking(
102 &self,
103 excluded_abouts: &[String],
104 ) -> Result<String, PortError> {
105 let events = self.read_event_log()?;
106 encode_bundle(
107 &filter_events_excluding_abouts(events, excluded_abouts),
108 None,
109 )
110 }
111
112 pub async fn export_bundle_excluding_abouts(
114 &self,
115 excluded_abouts: &[String],
116 ) -> Result<String, PortError> {
117 let events = self.run(EmbeddedKernelStore::read_event_log).await?;
118 encode_bundle(
119 &filter_events_excluding_abouts(events, excluded_abouts),
120 None,
121 )
122 }
123
124 pub async fn export_named_bundle(&self, snapshot_id: &str) -> Result<String, PortError> {
128 if snapshot_id.trim().is_empty() {
129 return Err(PortError::InvalidState(
130 "snapshot id must not be empty".to_string(),
131 ));
132 }
133 let events = self.run(EmbeddedKernelStore::read_event_log).await?;
134 encode_bundle(&events, Some(snapshot_id))
135 }
136
137 pub async fn import_bundle<F>(&self, bundle: &str, derive: F) -> Result<ImportReport, PortError>
142 where
143 F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
144 {
145 let (log_length, _) = self.event_log_stats().await?;
146 if log_length != 0 {
147 return Err(PortError::Conflict(format!(
148 "import requires an empty store; this store already holds {log_length} events \
149 (merging bundles is not supported)"
150 )));
151 }
152
153 let verified = parse_bundle(bundle)?;
154 let header = verified.header;
155 let events = verified.events;
156 validate_revisions(&events)?;
157 for event in &events {
161 derive(event)?;
162 }
163 let events_imported = self.replay_event_stream(events).await?;
164 debug_assert_eq!(events_imported, header.event_count);
165
166 let rebuild = self.rebuild_projections(derive).await?;
167 Ok(ImportReport {
168 events_imported,
169 rebuild,
170 })
171 }
172}
173
174pub fn verify_bundle(bundle: &str) -> Result<BundleHeader, PortError> {
178 parse_bundle(bundle).map(|verified| verified.header)
179}
180
181pub fn bundle_excluding_abouts(
189 bundle: &str,
190 excluded_abouts: &[String],
191) -> Result<String, PortError> {
192 let verified = parse_bundle(bundle)?;
193 encode_bundle(
194 &filter_events_excluding_abouts(verified.events, excluded_abouts),
195 None,
196 )
197}
198
199pub fn merge_bundles(left: &str, right: &str, snapshot_id: &str) -> Result<String, PortError> {
203 if snapshot_id.trim().is_empty() {
204 return Err(PortError::InvalidState(
205 "merged snapshot id must not be empty".to_string(),
206 ));
207 }
208 let left = parse_bundle(left)?;
209 let right = parse_bundle(right)?;
210 let shared = left.events.len().min(right.events.len());
211 if let Some(position) =
212 (0..shared).find(|position| left.events[*position] != right.events[*position])
213 {
214 return Err(PortError::Conflict(format!(
215 "bundle histories diverge at event position {}; KMP only fast-forwards an exact \
216 prefix and will not invent causal order for two branches",
217 position + 1
218 )));
219 }
220 let events = if left.events.len() >= right.events.len() {
221 left.events
222 } else {
223 right.events
224 };
225 encode_bundle(&events, Some(snapshot_id))
226}
227
228struct VerifiedBundle {
229 header: BundleHeader,
230 events: Vec<ContextUpdatedEvent>,
231}
232
233fn parse_bundle(bundle: &str) -> Result<VerifiedBundle, PortError> {
234 let mut lines = bundle.lines().filter(|line| !line.trim().is_empty());
235 let header: BundleHeader = decode_line(
236 "bundle header",
237 lines.next().ok_or_else(|| {
238 PortError::InvalidState("bundle is empty: missing header line".to_string())
239 })?,
240 )?;
241 if header.bundle_format != BUNDLE_FORMAT_VERSION {
242 return Err(PortError::InvalidState(format!(
243 "bundle format {} is not supported (this binary reads {})",
244 header.bundle_format, BUNDLE_FORMAT_VERSION
245 )));
246 }
247 if header.event_format != super::format_version::EVENT_FORMAT_VERSION {
248 return Err(PortError::InvalidState(format!(
249 "bundle carries event format {}, this binary supports {}",
250 header.event_format,
251 super::format_version::EVENT_FORMAT_VERSION
252 )));
253 }
254
255 let mut events = Vec::new();
256 let mut event_payload = String::new();
257 for line in lines {
258 events.push(decode_line::<ContextUpdatedEvent>("bundle event", line)?);
259 event_payload.push_str(line);
260 event_payload.push('\n');
261 }
262 if events.len() as u64 != header.event_count {
263 return Err(PortError::InvalidState(format!(
264 "bundle header declares {} events but {} were present",
265 header.event_count,
266 events.len()
267 )));
268 }
269
270 validate_header(&header, &events, &event_payload)?;
271 Ok(VerifiedBundle { header, events })
272}
273
274fn validate_header(
275 header: &BundleHeader,
276 events: &[ContextUpdatedEvent],
277 event_payload: &str,
278) -> Result<(), PortError> {
279 if header.snapshot_id.trim().is_empty() {
280 return Err(PortError::InvalidState(
281 "bundle format 2 requires snapshot_id".to_string(),
282 ));
283 }
284 if header.created_at_unix_ms == 0 {
285 return Err(PortError::InvalidState(
286 "bundle format 2 requires created_at_unix_ms".to_string(),
287 ));
288 }
289 let expected_range = event_range(events.len());
290 if header.event_range != expected_range {
291 return Err(PortError::InvalidState(format!(
292 "bundle event_range {:?} does not cover its {} events (expected {:?})",
293 header.event_range,
294 events.len(),
295 expected_range
296 )));
297 }
298 let expected_abouts = abouts(events);
299 if header.abouts != expected_abouts {
300 return Err(PortError::InvalidState(format!(
301 "bundle abouts do not match its events (expected {})",
302 expected_abouts.join(", ")
303 )));
304 }
305 let expected_digest = content_digest(event_payload.as_bytes());
306 if header.content_digest != expected_digest {
307 return Err(PortError::InvalidState(format!(
308 "bundle content digest mismatch: header says {}, events produce {expected_digest}",
309 header.content_digest
310 )));
311 }
312 Ok(())
313}
314
315fn encode_bundle(
316 events: &[ContextUpdatedEvent],
317 snapshot_id: Option<&str>,
318) -> Result<String, PortError> {
319 let mut event_payload = String::new();
320 for event in events {
321 event_payload.push_str(&encode_line("bundle event", event)?);
322 }
323 let digest = content_digest(event_payload.as_bytes());
324 let named = snapshot_id.is_some();
325 let snapshot_id = snapshot_id
326 .map(str::to_string)
327 .unwrap_or_else(|| format!("content-{}", &digest[7..23]));
328 let created_at = if named {
333 SystemTime::now()
334 } else {
335 events
336 .iter()
337 .map(|event| event.occurred_at)
338 .max()
339 .unwrap_or(UNIX_EPOCH + Duration::from_millis(1))
340 };
341 let created_at_unix_ms = created_at
342 .duration_since(UNIX_EPOCH)
343 .unwrap_or(Duration::ZERO)
344 .as_millis() as u64;
345 let header = BundleHeader {
346 bundle_format: BUNDLE_FORMAT_VERSION,
347 event_format: super::format_version::EVENT_FORMAT_VERSION,
348 event_count: events.len() as u64,
349 kernel_version: env!("CARGO_PKG_VERSION").to_string(),
350 snapshot_id,
351 created_at_unix_ms,
352 event_range: event_range(events.len()),
353 abouts: abouts(events),
354 content_digest: digest,
355 };
356 let mut out = encode_line("bundle header", &header)?;
357 out.push_str(&event_payload);
358 Ok(out)
359}
360
361fn event_range(event_count: usize) -> BundleEventRange {
362 if event_count == 0 {
363 BundleEventRange::default()
364 } else {
365 BundleEventRange {
366 first: Some(1),
367 last: Some(event_count as u64),
368 }
369 }
370}
371
372fn abouts(events: &[ContextUpdatedEvent]) -> Vec<String> {
373 events
374 .iter()
375 .map(|event| event.root_node_id.clone())
376 .collect::<BTreeSet<_>>()
377 .into_iter()
378 .collect()
379}
380
381fn filter_events_excluding_abouts(
382 events: Vec<ContextUpdatedEvent>,
383 excluded_abouts: &[String],
384) -> Vec<ContextUpdatedEvent> {
385 if excluded_abouts.is_empty() {
386 return events;
387 }
388 let excluded = excluded_abouts.iter().cloned().collect::<BTreeSet<_>>();
389 events
390 .into_iter()
391 .filter(|event| !excluded.contains(&event.root_node_id))
392 .collect()
393}
394
395fn filter_events_for_abouts(
396 events: Vec<ContextUpdatedEvent>,
397 requested_abouts: &[String],
398) -> Result<Vec<ContextUpdatedEvent>, PortError> {
399 if requested_abouts.is_empty() {
400 return Err(PortError::InvalidState(
401 "filtered export requires at least one about".to_string(),
402 ));
403 }
404 let requested = requested_abouts.iter().cloned().collect::<BTreeSet<_>>();
405 let found = events
406 .iter()
407 .filter(|event| requested.contains(&event.root_node_id))
408 .map(|event| event.root_node_id.clone())
409 .collect::<BTreeSet<_>>();
410 let missing = requested.difference(&found).cloned().collect::<Vec<_>>();
411 if !missing.is_empty() {
412 return Err(PortError::InvalidState(format!(
413 "cannot export missing about{}: {}",
414 if missing.len() == 1 { "" } else { "s" },
415 missing
416 .iter()
417 .map(|about| format!("`{about}`"))
418 .collect::<Vec<_>>()
419 .join(", ")
420 )));
421 }
422 Ok(events
423 .into_iter()
424 .filter(|event| requested.contains(&event.root_node_id))
425 .collect())
426}
427
428fn content_digest(bytes: &[u8]) -> String {
429 format!("sha256:{:x}", Sha256::digest(bytes))
430}
431
432fn validate_revisions(events: &[ContextUpdatedEvent]) -> Result<(), PortError> {
433 let mut revisions: BTreeMap<(&str, &str), u64> = BTreeMap::new();
434 for (position, event) in events.iter().enumerate() {
435 let previous = revisions
436 .get(&(event.root_node_id.as_str(), event.role.as_str()))
437 .copied()
438 .unwrap_or(0);
439 let expected = previous + 1;
440 if event.revision != expected {
441 return Err(PortError::InvalidState(format!(
442 "bundle event position {} carries revision {} for ({}, {}), expected {}; no \
443 events were imported",
444 position + 1,
445 event.revision,
446 event.root_node_id,
447 event.role,
448 expected
449 )));
450 }
451 revisions.insert(
452 (event.root_node_id.as_str(), event.role.as_str()),
453 event.revision,
454 );
455 }
456 Ok(())
457}
458
459impl EmbeddedKernelStore {
460 pub(crate) async fn replay_event_stream<I>(&self, events: I) -> Result<u64, PortError>
468 where
469 I: IntoIterator<Item = ContextUpdatedEvent>,
470 {
471 let mut replayed = 0u64;
472 for event in events {
473 let recorded_revision = event.revision;
474 let expected_previous = recorded_revision.checked_sub(1).ok_or_else(|| {
475 PortError::InvalidState("event carries revision 0; the log is corrupt".to_string())
476 })?;
477 let assigned = self.append(event, expected_previous).await?;
478 if assigned != recorded_revision {
479 return Err(PortError::Conflict(format!(
480 "replay integrity violation: assigned revision {assigned}, \
481 history recorded {recorded_revision}"
482 )));
483 }
484 replayed += 1;
485 }
486 Ok(replayed)
487 }
488}
489
490fn encode_line<T: Serialize>(what: &str, value: &T) -> Result<String, PortError> {
491 let mut line = serde_json::to_string(value)
492 .map_err(|error| PortError::InvalidState(format!("could not encode {what}: {error}")))?;
493 line.push('\n');
494 Ok(line)
495}
496
497fn decode_line<T: for<'de> Deserialize<'de>>(what: &str, line: &str) -> Result<T, PortError> {
498 serde_json::from_str(line)
499 .map_err(|error| PortError::InvalidState(format!("could not decode {what}: {error}")))
500}
501
502#[cfg(test)]
503mod tests {
504 use super::*;
505
506 fn event(root: &str, revision: u64, content_hash: &str) -> ContextUpdatedEvent {
507 ContextUpdatedEvent {
508 root_node_id: root.to_string(),
509 role: "agent".to_string(),
510 revision,
511 content_hash: content_hash.to_string(),
512 changes: Vec::new(),
513 idempotency_key: Some(format!("{root}:{revision}")),
514 logical_digest: None,
515 requested_by: Some("portability-test".to_string()),
516 occurred_at: UNIX_EPOCH + Duration::from_secs(revision),
517 }
518 }
519
520 #[test]
521 fn current_format_identifies_and_covers_the_snapshot() {
522 let events = vec![event("project:b", 1, "b"), event("project:a", 1, "a")];
523 let bundle = encode_bundle(&events, Some("pre-release")).expect("bundle");
524 let header = verify_bundle(&bundle).expect("verified");
525
526 assert_eq!(header.bundle_format, BUNDLE_FORMAT_VERSION);
527 assert_eq!(
528 header.event_format,
529 super::super::format_version::EVENT_FORMAT_VERSION
530 );
531 assert_eq!(header.snapshot_id, "pre-release");
532 assert!(header.created_at_unix_ms > 0);
533 assert_eq!(
534 header.event_range,
535 BundleEventRange {
536 first: Some(1),
537 last: Some(2),
538 }
539 );
540 assert_eq!(header.abouts, ["project:a", "project:b"]);
541 assert!(header.content_digest.starts_with("sha256:"));
542 }
543
544 #[test]
545 fn exclusion_keeps_everything_the_excluded_abouts_do_not_root() {
546 let events = vec![
547 event("project:a", 1, "a1"),
548 event("guide:kmp", 1, "g1"),
549 event("project:a", 2, "a2"),
550 event("guide:kmp-agent", 1, "g2"),
551 ];
552 let excluded = vec!["guide:kmp".to_string(), "guide:kmp-agent".to_string()];
553
554 let kept = filter_events_excluding_abouts(events, &excluded);
555
556 assert_eq!(
557 kept.iter()
558 .map(|event| (event.root_node_id.as_str(), event.revision))
559 .collect::<Vec<_>>(),
560 vec![("project:a", 1), ("project:a", 2)]
561 );
562 }
563
564 #[test]
565 fn exclusion_matches_exactly_because_abouts_are_opaque() {
566 let events = vec![
567 event("guide:kmp", 1, "g1"),
568 event("guide:kmp:extra", 1, "x1"),
569 event(" guide:kmp", 1, "s1"),
570 ];
571 let excluded = vec!["guide:kmp".to_string()];
572
573 let kept = filter_events_excluding_abouts(events, &excluded);
574
575 assert_eq!(
576 kept.iter()
577 .map(|event| event.root_node_id.as_str())
578 .collect::<Vec<_>>(),
579 vec!["guide:kmp:extra", " guide:kmp"]
580 );
581 }
582
583 #[test]
584 fn excluding_an_about_the_store_never_had_is_not_an_error() {
585 let events = vec![event("project:a", 1, "a1")];
589 let excluded = vec!["guide:kmp".to_string()];
590
591 let kept = filter_events_excluding_abouts(events, &excluded);
592
593 assert_eq!(kept.len(), 1);
594 assert_eq!(kept[0].root_node_id, "project:a");
595 }
596
597 #[test]
598 fn verified_bundle_exclusion_keeps_unexcluded_events() {
599 let bundle = encode_bundle(
600 &[
601 event("project:a", 1, "a1"),
602 event("guide:kmp-agent", 1, "g1"),
603 event("project:a", 2, "a2"),
604 ],
605 None,
606 )
607 .expect("bundle");
608
609 let filtered = bundle_excluding_abouts(&bundle, &["guide:kmp-agent".to_string()])
610 .expect("filtered verified bundle");
611 let verified = parse_bundle(&filtered).expect("verified filtered bundle");
612
613 assert_eq!(verified.header.event_count, 2);
614 assert_eq!(verified.header.abouts, ["project:a"]);
615 assert_eq!(
616 verified
617 .events
618 .iter()
619 .map(|event| event.content_hash.as_str())
620 .collect::<Vec<_>>(),
621 ["a1", "a2"]
622 );
623 }
624
625 #[test]
626 fn filtered_export_matches_opaque_abouts_exactly_and_renumbers_its_range() {
627 let events = vec![
628 event("project:a", 1, "a1"),
629 event("project:ab", 1, "ab1"),
630 event("project:a", 2, "a2"),
631 ];
632 let filtered = filter_events_for_abouts(events, &["project:a".to_string()])
633 .expect("exact about exists");
634 assert_eq!(filtered.len(), 2);
635 assert!(
636 filtered
637 .iter()
638 .all(|event| event.root_node_id == "project:a")
639 );
640
641 let bundle = encode_bundle(&filtered, None).expect("filtered bundle");
642 let header = verify_bundle(&bundle).expect("filtered bundle verifies");
643 assert_eq!(header.abouts, ["project:a"]);
644 assert_eq!(header.event_count, 2);
645 assert_eq!(
646 header.event_range,
647 BundleEventRange {
648 first: Some(1),
649 last: Some(2),
650 }
651 );
652 }
653
654 #[test]
655 fn filtered_export_names_every_requested_about_that_is_missing() {
656 let error = filter_events_for_abouts(
657 vec![event("project:a", 1, "a")],
658 &["project:a".into(), "project:none".into()],
659 )
660 .expect_err("missing about must fail");
661 assert!(error.to_string().contains("`project:none`"), "{error}");
662 }
663
664 #[test]
665 fn tampering_is_rejected_before_a_bundle_can_be_replayed() {
666 let bundle =
667 encode_bundle(&[event("project:a", 1, "before")], Some("saved")).expect("bundle");
668 let tampered = bundle.replace("\"content_hash\":\"before\"", "\"content_hash\":\"after\"");
669 let error = verify_bundle(&tampered).expect_err("digest catches changed payload");
670 assert!(error.to_string().contains("content digest mismatch"));
671 }
672
673 #[test]
674 fn merge_fast_forwards_an_exact_prefix() {
675 let first = event("project:a", 1, "one");
676 let second = event("project:a", 2, "two");
677 let left = encode_bundle(std::slice::from_ref(&first), Some("left")).expect("left");
678 let right = encode_bundle(&[first, second], Some("right")).expect("right");
679
680 let merged = merge_bundles(&left, &right, "merged").expect("fast forward");
681 let header = verify_bundle(&merged).expect("verified merge");
682 assert_eq!(header.snapshot_id, "merged");
683 assert_eq!(header.event_count, 2);
684 }
685
686 #[test]
687 fn merge_refuses_two_histories_at_the_same_position() {
688 let left = encode_bundle(&[event("project:a", 1, "left")], Some("left")).expect("left");
689 let right = encode_bundle(&[event("project:a", 1, "right")], Some("right")).expect("right");
690
691 let error = merge_bundles(&left, &right, "invented").expect_err("must refuse");
692 assert!(error.to_string().contains("diverge at event position 1"));
693 assert!(error.to_string().contains("will not invent causal order"));
694 }
695
696 #[test]
697 fn unsupported_bundle_and_event_formats_are_rejected() {
698 let bundle = encode_bundle(&[], None).expect("bundle");
699 for old in [1, 2] {
700 let legacy = bundle.replace("\"bundle_format\":3", &format!("\"bundle_format\":{old}"));
701 let error = verify_bundle(&legacy).expect_err("old bundle is unsupported");
702 assert!(error.to_string().contains("is not supported"), "{error}");
703 }
704 let legacy = bundle.replace("\"event_format\":2", "\"event_format\":1");
705 let error = verify_bundle(&legacy).expect_err("old events are unsupported");
706 assert!(
707 error.to_string().contains("bundle carries event format 1"),
708 "{error}"
709 );
710 }
711
712 #[test]
713 fn invalid_later_revision_is_rejected_in_preflight() {
714 let events = [event("project:a", 1, "one"), event("project:a", 3, "three")];
715 let error = validate_revisions(&events).expect_err("revision gap");
716 assert!(error.to_string().contains("position 2"));
717 assert!(error.to_string().contains("no events were imported"));
718 }
719}