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