kmp_adapter_embedded/adapter/
migration.rs

1//! Compatibility API for store-layout migration receipts.
2//!
3//! Current KMP has one supported layout, SQLite format 3, so there is no live
4//! in-process migration path. These APIs remain available to avoid breaking
5//! downstream Rust callers: a current source reports that migration is
6//! unnecessary, and an unsupported source is preserved with the same generic
7//! external export/import recovery contract as the open gate.
8
9use std::fs;
10use std::path::Path;
11
12use kmp_domain::{ContextUpdatedEvent, PortError, ProjectionMutation};
13use serde::{Deserialize, Serialize};
14
15use super::engine::{Key, Table};
16use super::format_version::{self, StorageEngine};
17use super::store::EmbeddedKernelStore;
18
19/// What a completed historical store migration recorded in its destination.
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21pub struct StoreMigrationReceipt {
22    pub source_format: u32,
23    pub source_sha256: String,
24    pub destination_format: u32,
25    pub events_migrated: u64,
26    pub mutations_applied: u64,
27    pub kernel_version: String,
28}
29
30impl StoreMigrationReceipt {
31    /// Single key used by historical migration receipts.
32    pub const MIGRATION_ID: &'static str = "store-format-migration";
33}
34
35impl EmbeddedKernelStore {
36    /// Compatibility entry point. No current layout requires migration.
37    pub async fn migrate_data_dir<F>(
38        source_dir: &Path,
39        destination_dir: &Path,
40        derive: F,
41    ) -> Result<(Self, StoreMigrationReceipt), PortError>
42    where
43        F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
44    {
45        Self::migrate_data_dir_to(source_dir, destination_dir, StorageEngine::Sqlite, derive).await
46    }
47
48    /// Compatibility entry point with an explicit destination engine.
49    pub async fn migrate_data_dir_to<F>(
50        source_dir: &Path,
51        destination_dir: &Path,
52        destination_engine: StorageEngine,
53        _derive: F,
54    ) -> Result<(Self, StoreMigrationReceipt), PortError>
55    where
56        F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
57    {
58        if same_file(source_dir, destination_dir) {
59            return Err(PortError::InvalidState(
60                "migration source and destination are the same data directory".to_string(),
61            ));
62        }
63        let source_format = format_version::read_stamped_version(source_dir)?;
64        if source_format > StorageEngine::NEWEST_KNOWN_FORMAT_VERSION {
65            return Err(PortError::InvalidState(format!(
66                "migration source `{}` uses format version {source_format}, newer than this \
67                 binary supports ({}); upgrade the binary",
68                source_dir.display(),
69                StorageEngine::NEWEST_KNOWN_FORMAT_VERSION
70            )));
71        }
72        if source_format == StorageEngine::Sqlite.format_version() {
73            return Err(PortError::Unavailable(format!(
74                "migration from a SQLite format-3 store is unnecessary and unsupported; the \
75                 source at `{}` is left untouched",
76                source_dir.display()
77            )));
78        }
79        let _ = destination_engine;
80        Err(PortError::InvalidState(format!(
81            "migration source `{}` uses unsupported format version {source_format}; current \
82             KMP left it untouched. Old contracts are not migrated by this redesign; \
83             use a compatible binary to inspect the source and start a fresh current store",
84            source_dir.display()
85        )))
86    }
87
88    /// Reopen a completed destination or apply the compatibility migration.
89    pub async fn open_or_migrate_data_dir<F>(
90        source_dir: &Path,
91        destination_dir: &Path,
92        derive: F,
93    ) -> Result<(Self, Option<StoreMigrationReceipt>), PortError>
94    where
95        F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
96    {
97        Self::open_or_migrate_data_dir_to(
98            source_dir,
99            destination_dir,
100            StorageEngine::Sqlite,
101            derive,
102        )
103        .await
104    }
105
106    /// Reopen a completed destination or apply the compatibility migration.
107    pub async fn open_or_migrate_data_dir_to<F>(
108        source_dir: &Path,
109        destination_dir: &Path,
110        destination_engine: StorageEngine,
111        derive: F,
112    ) -> Result<(Self, Option<StoreMigrationReceipt>), PortError>
113    where
114        F: Fn(&ContextUpdatedEvent) -> Result<Vec<ProjectionMutation>, PortError> + Send + 'static,
115    {
116        if format_version::existing_store_file(destination_dir).is_some() {
117            let store = Self::open(destination_dir)?;
118            let receipt = store.migration_receipt().await?;
119            return Ok((store, receipt));
120        }
121        let (store, receipt) =
122            Self::migrate_data_dir_to(source_dir, destination_dir, destination_engine, derive)
123                .await?;
124        Ok((store, Some(receipt)))
125    }
126
127    /// A historical receipt stored by the migration that produced this store.
128    pub async fn migration_receipt(&self) -> Result<Option<StoreMigrationReceipt>, PortError> {
129        self.run(|store| {
130            let tx = store.begin_read()?;
131            let Some(raw) = tx.get(
132                Table::Migrations,
133                Key::Str(StoreMigrationReceipt::MIGRATION_ID),
134            )?
135            else {
136                return Ok(None);
137            };
138            let receipt = serde_json::from_slice(&raw).map_err(|error| {
139                PortError::InvalidState(format!("migration receipt is unreadable: {error}"))
140            })?;
141            Ok(Some(receipt))
142        })
143        .await
144    }
145}
146
147fn same_file(left: &Path, right: &Path) -> bool {
148    match (fs::canonicalize(left), fs::canonicalize(right)) {
149        (Ok(left), Ok(right)) => left == right,
150        _ => left == right,
151    }
152}
kmp_adapter_embedded/adapter/migration.rs

kmp_adapter_embedded/adapter/
migration.rs
