| 1 | //! Schema migration framework for `~/.deepseek/` persisted records. |
| 2 | //! |
| 3 | //! Every persistence layer in `crates/tui/src/` (sessions, threads, |
| 4 | //! tasks, automations, offline queue) gates `schema_version > CURRENT_*` |
| 5 | //! to prevent silent truncation when an older binary tries to load a |
| 6 | //! record from a newer one. What was missing — and what this module |
| 7 | //! fixes — is the **upgrade path**: when `schema_version < CURRENT_*`, |
| 8 | //! the load function should run forward migrations rather than loading |
| 9 | //! a partially-correct record. |
| 10 | //! |
| 11 | //! ## Domain registration |
| 12 | //! |
| 13 | //! Each persistence type implements [`SchemaMigration`]: |
| 14 | //! |
| 15 | //! ```ignore |
| 16 | //! pub struct SessionMigration; |
| 17 | //! |
| 18 | //! impl SchemaMigration for SessionMigration { |
| 19 | //! const CURRENT_VERSION: u32 = 1; |
| 20 | //! const DOMAIN: &'static str = "session"; |
| 21 | //! const MIGRATIONS: &'static [MigrationFn] = &[ |
| 22 | //! // index i migrates from version (i+1) to (i+2) |
| 23 | //! migrate_session_v1_to_v2, |
| 24 | //! ]; |
| 25 | //! } |
| 26 | //! ``` |
| 27 | //! |
| 28 | //! ## Load-site usage |
| 29 | //! |
| 30 | //! Inside the load function, after deserialization: |
| 31 | //! |
| 32 | //! ```ignore |
| 33 | //! if record.schema_version < SessionMigration::CURRENT_VERSION { |
| 34 | //! let mut value: serde_json::Value = serde_json::from_str(&raw)?; |
| 35 | //! let _final = SessionMigration::migrate( |
| 36 | //! &mut value, |
| 37 | //! record.schema_version, |
| 38 | //! )?; |
| 39 | //! backup_before_migrate(&path, SessionMigration::DOMAIN); |
| 40 | //! write_atomic(&path, serde_json::to_string_pretty(&value)?.as_bytes())?; |
| 41 | //! // Re-deserialize with the migrated value into the up-to-date struct. |
| 42 | //! record = serde_json::from_value(value)?; |
| 43 | //! } |
| 44 | //! ``` |
| 45 | //! |
| 46 | //! ## Migration step contract |
| 47 | //! |
| 48 | //! Each step takes a mutable JSON value at version `N` and mutates it |
| 49 | //! into version `N+1`. Steps must be idempotent in the sense that a |
| 50 | //! re-run of the migration on an already-migrated value should be a |
| 51 | //! no-op (because `serde_json::Value` is cheap to introspect, this |
| 52 | //! usually means "if field already exists with the new shape, skip"). |
| 53 | //! |
| 54 | //! Steps must NOT call `write_atomic` themselves — the framework writes |
| 55 | //! once at the end. They must NOT log credentials or other sensitive |
| 56 | //! material from the value being migrated. |
| 57 | |
| 58 | use std::fs; |
| 59 | use std::path::{Path, PathBuf}; |
| 60 | |
| 61 | /// Result returned when a migration step fails. |
| 62 | #[derive(Debug)] |
| 63 | pub struct MigrationError { |
| 64 | pub from_version: u32, |
| 65 | pub to_version: u32, |
| 66 | pub reason: String, |
| 67 | } |
| 68 | |
| 69 | impl std::fmt::Display for MigrationError { |
| 70 | fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { |
| 71 | write!( |
| 72 | f, |
| 73 | "schema migration {} → {} failed: {}", |
| 74 | self.from_version, self.to_version, self.reason |
| 75 | ) |
| 76 | } |
| 77 | } |
| 78 | |
| 79 | impl std::error::Error for MigrationError {} |
| 80 | |
| 81 | /// Signature of a single forward migration step. |
| 82 | #[allow(dead_code)] // Public surface; first concrete migrator lands when v2 ships. |
| 83 | pub type MigrationFn = fn(&mut serde_json::Value) -> Result<(), MigrationError>; |
| 84 | |
| 85 | /// Each persistence domain implements this trait. |
| 86 | /// |
| 87 | /// `MIGRATIONS[i]` migrates from version `i + 1` to version `i + 2`. So |
| 88 | /// `MIGRATIONS[0]` is the v1 → v2 step, `MIGRATIONS[1]` is v2 → v3, etc. |
| 89 | /// `CURRENT_VERSION` must equal `MIGRATIONS.len() + 1` (i.e. the version |
| 90 | /// produced by running every step in sequence starting from version 1). |
| 91 | #[allow(dead_code)] // Public surface; first concrete domain lands when v2 ships. |
| 92 | pub trait SchemaMigration { |
| 93 | /// The current schema version for this domain. |
| 94 | const CURRENT_VERSION: u32; |
| 95 | |
| 96 | /// Human-readable domain label for logging (e.g. "session", "thread"). |
| 97 | const DOMAIN: &'static str; |
| 98 | |
| 99 | /// Ordered list of migration step functions. |
| 100 | const MIGRATIONS: &'static [MigrationFn]; |
| 101 | |
| 102 | /// Run all required migrations to bring `version` up to |
| 103 | /// [`CURRENT_VERSION`](SchemaMigration::CURRENT_VERSION). |
| 104 | /// |
| 105 | /// Returns the final stamped version. Stamps each intermediate |
| 106 | /// version onto `value["schema_version"]` so a partial migration |
| 107 | /// failure leaves a record at a known state rather than mixed. |
| 108 | fn migrate(value: &mut serde_json::Value, version: u32) -> Result<u32, MigrationError> { |
| 109 | if version > Self::CURRENT_VERSION { |
| 110 | // Caller's responsibility to reject newer-than-supported |
| 111 | // records — the framework's job is forward migration only. |
| 112 | return Err(MigrationError { |
| 113 | from_version: version, |
| 114 | to_version: Self::CURRENT_VERSION, |
| 115 | reason: format!( |
| 116 | "{} record at v{version} is newer than current v{}", |
| 117 | Self::DOMAIN, |
| 118 | Self::CURRENT_VERSION |
| 119 | ), |
| 120 | }); |
| 121 | } |
| 122 | |
| 123 | let mut current = version; |
| 124 | for (idx, step) in Self::MIGRATIONS.iter().enumerate() { |
| 125 | let step_from = (idx as u32) + 1; |
| 126 | if current > step_from { |
| 127 | // Already past this step — the value was loaded at a |
| 128 | // newer-than-step version, skip. |
| 129 | continue; |
| 130 | } |
| 131 | if current < step_from { |
| 132 | // Underflow: Self's MIGRATIONS are dense from 1, and |
| 133 | // the loop should never see a record older than the |
| 134 | // first step. If we get here the const list is misordered. |
| 135 | return Err(MigrationError { |
| 136 | from_version: current, |
| 137 | to_version: step_from + 1, |
| 138 | reason: format!( |
| 139 | "{} migration list is non-contiguous at index {idx}", |
| 140 | Self::DOMAIN |
| 141 | ), |
| 142 | }); |
| 143 | } |
| 144 | step(value)?; |
| 145 | current = step_from + 1; |
| 146 | value["schema_version"] = serde_json::json!(current); |
| 147 | } |
| 148 | |
| 149 | if current != Self::CURRENT_VERSION { |
| 150 | return Err(MigrationError { |
| 151 | from_version: version, |
| 152 | to_version: Self::CURRENT_VERSION, |
| 153 | reason: format!( |
| 154 | "{} migrated to v{current} but expected v{}", |
| 155 | Self::DOMAIN, |
| 156 | Self::CURRENT_VERSION |
| 157 | ), |
| 158 | }); |
| 159 | } |
| 160 | |
| 161 | Ok(current) |
| 162 | } |
| 163 | } |
| 164 | |
| 165 | /// Create a `.bak` copy of `path` before mutation. Returns the backup |
| 166 | /// path. Errors are logged at warn level and ignored — the migration |
| 167 | /// proceeds because [`crate::utils::write_atomic`] is itself crash-safe. |
| 168 | /// |
| 169 | /// The `.bak` file is left on disk after a successful migration so a |
| 170 | /// user who notices a regression can manually restore it. No automatic |
| 171 | /// garbage collection — bak files are user-visible recovery artifacts. |
| 172 | #[allow(dead_code)] // Public surface; first call site lands when v2 ships. |
| 173 | pub fn backup_before_migrate(path: &Path, domain: &str) -> PathBuf { |
| 174 | let bak = path.with_extension( |
| 175 | path.extension() |
| 176 | .map(|ext| format!("{}.bak", ext.to_string_lossy())) |
| 177 | .unwrap_or_else(|| "bak".to_string()), |
| 178 | ); |
| 179 | match fs::copy(path, &bak) { |
| 180 | Ok(_) => tracing::info!( |
| 181 | domain, |
| 182 | from = %path.display(), |
| 183 | to = %bak.display(), |
| 184 | "schema backup created" |
| 185 | ), |
| 186 | Err(e) => tracing::warn!( |
| 187 | domain, |
| 188 | from = %path.display(), |
| 189 | error = %e, |
| 190 | "schema backup failed (continuing — migration is crash-safe)" |
| 191 | ), |
| 192 | } |
| 193 | bak |
| 194 | } |
| 195 | |
| 196 | /// Per-domain migration registrations. |
| 197 | /// |
| 198 | /// Each persistence type below points at the same `CURRENT_*` constant |
| 199 | /// the original module already gates on. The `MIGRATIONS` list is empty |
| 200 | /// today because no schema bumps have shipped yet — but the framework is |
| 201 | /// in place so the next bump only needs to: |
| 202 | /// |
| 203 | /// 1. Add a `migrate_<domain>_v<N>_to_v<N+1>` function in this module. |
| 204 | /// 2. Append it to the matching `MIGRATIONS` list. |
| 205 | /// 3. Bump `CURRENT_VERSION` to match. |
| 206 | /// 4. Wire `<Domain>Migration::migrate(...)` into the load function in |
| 207 | /// the owning module. |
| 208 | pub mod registry { |
| 209 | use super::{MigrationFn, SchemaMigration}; |
| 210 | |
| 211 | /// Sessions: `~/.deepseek/sessions/<id>.json` and the latest |
| 212 | /// checkpoint at `~/.deepseek/sessions/checkpoints/latest.json`. |
| 213 | pub struct SessionMigration; |
| 214 | impl SchemaMigration for SessionMigration { |
| 215 | const CURRENT_VERSION: u32 = 1; |
| 216 | const DOMAIN: &'static str = "session"; |
| 217 | const MIGRATIONS: &'static [MigrationFn] = &[]; |
| 218 | } |
| 219 | |
| 220 | /// Offline queue: `~/.deepseek/sessions/checkpoints/offline_queue.json`. |
| 221 | pub struct OfflineQueueMigration; |
| 222 | impl SchemaMigration for OfflineQueueMigration { |
| 223 | const CURRENT_VERSION: u32 = 1; |
| 224 | const DOMAIN: &'static str = "offline_queue"; |
| 225 | const MIGRATIONS: &'static [MigrationFn] = &[]; |
| 226 | } |
| 227 | |
| 228 | /// Runtime threads / turns / items / events / store state — all |
| 229 | /// share `CURRENT_RUNTIME_SCHEMA_VERSION`. |
| 230 | pub struct RuntimeMigration; |
| 231 | impl SchemaMigration for RuntimeMigration { |
| 232 | const CURRENT_VERSION: u32 = 2; |
| 233 | const DOMAIN: &'static str = "runtime"; |
| 234 | const MIGRATIONS: &'static [MigrationFn] = &[]; |
| 235 | } |
| 236 | |
| 237 | /// Durable tasks under `~/.deepseek/tasks/`. |
| 238 | pub struct TaskMigration; |
| 239 | impl SchemaMigration for TaskMigration { |
| 240 | const CURRENT_VERSION: u32 = 2; |
| 241 | const DOMAIN: &'static str = "task"; |
| 242 | const MIGRATIONS: &'static [MigrationFn] = &[]; |
| 243 | } |
| 244 | |
| 245 | /// Automation records and their per-run history. |
| 246 | pub struct AutomationMigration; |
| 247 | impl SchemaMigration for AutomationMigration { |
| 248 | const CURRENT_VERSION: u32 = 1; |
| 249 | const DOMAIN: &'static str = "automation"; |
| 250 | const MIGRATIONS: &'static [MigrationFn] = &[]; |
| 251 | } |
| 252 | |
| 253 | pub struct AutomationRunMigration; |
| 254 | impl SchemaMigration for AutomationRunMigration { |
| 255 | const CURRENT_VERSION: u32 = 1; |
| 256 | const DOMAIN: &'static str = "automation_run"; |
| 257 | const MIGRATIONS: &'static [MigrationFn] = &[]; |
| 258 | } |
| 259 | } |
| 260 | |
| 261 | #[cfg(test)] |
| 262 | mod tests { |
| 263 | use super::*; |
| 264 | |
| 265 | /// Test harness: a fake "thread" domain at v3 with two migrations |
| 266 | /// (v1 → v2 adds an `archived` field; v2 → v3 adds a `kind` field). |
| 267 | struct TestThreadMigration; |
| 268 | |
| 269 | fn add_archived_field(value: &mut serde_json::Value) -> Result<(), MigrationError> { |
| 270 | if value.get("archived").is_none() { |
| 271 | value["archived"] = serde_json::json!(false); |
| 272 | } |
| 273 | Ok(()) |
| 274 | } |
| 275 | |
| 276 | fn add_kind_field(value: &mut serde_json::Value) -> Result<(), MigrationError> { |
| 277 | if value.get("kind").is_none() { |
| 278 | value["kind"] = serde_json::json!("standard"); |
| 279 | } |
| 280 | Ok(()) |
| 281 | } |
| 282 | |
| 283 | impl SchemaMigration for TestThreadMigration { |
| 284 | const CURRENT_VERSION: u32 = 3; |
| 285 | const DOMAIN: &'static str = "test_thread"; |
| 286 | const MIGRATIONS: &'static [MigrationFn] = &[add_archived_field, add_kind_field]; |
| 287 | } |
| 288 | |
| 289 | #[test] |
| 290 | fn migrate_no_op_when_already_current() { |
| 291 | let mut value = serde_json::json!({ |
| 292 | "schema_version": 3, |
| 293 | "id": "abc", |
| 294 | "archived": true, |
| 295 | "kind": "feature_branch" |
| 296 | }); |
| 297 | let final_version = TestThreadMigration::migrate(&mut value, 3).expect("ok"); |
| 298 | assert_eq!(final_version, 3); |
| 299 | // Existing values must be untouched (we don't reset to defaults). |
| 300 | assert_eq!(value["archived"], serde_json::json!(true)); |
| 301 | assert_eq!(value["kind"], serde_json::json!("feature_branch")); |
| 302 | } |
| 303 | |
| 304 | #[test] |
| 305 | fn migrate_runs_all_steps_from_v1() { |
| 306 | let mut value = serde_json::json!({ |
| 307 | "schema_version": 1, |
| 308 | "id": "abc" |
| 309 | }); |
| 310 | let final_version = TestThreadMigration::migrate(&mut value, 1).expect("ok"); |
| 311 | assert_eq!(final_version, 3); |
| 312 | assert_eq!(value["schema_version"], serde_json::json!(3)); |
| 313 | assert_eq!(value["archived"], serde_json::json!(false)); |
| 314 | assert_eq!(value["kind"], serde_json::json!("standard")); |
| 315 | } |
| 316 | |
| 317 | #[test] |
| 318 | fn migrate_runs_only_remaining_steps_from_v2() { |
| 319 | let mut value = serde_json::json!({ |
| 320 | "schema_version": 2, |
| 321 | "id": "abc", |
| 322 | "archived": true |
| 323 | }); |
| 324 | let final_version = TestThreadMigration::migrate(&mut value, 2).expect("ok"); |
| 325 | assert_eq!(final_version, 3); |
| 326 | // archived was already set; migration must NOT overwrite to default. |
| 327 | assert_eq!(value["archived"], serde_json::json!(true)); |
| 328 | assert_eq!(value["kind"], serde_json::json!("standard")); |
| 329 | } |
| 330 | |
| 331 | #[test] |
| 332 | fn migrate_rejects_newer_than_current() { |
| 333 | let mut value = serde_json::json!({ |
| 334 | "schema_version": 99 |
| 335 | }); |
| 336 | let err = TestThreadMigration::migrate(&mut value, 99).expect_err("must reject"); |
| 337 | assert_eq!(err.from_version, 99); |
| 338 | assert_eq!(err.to_version, 3); |
| 339 | assert!(err.reason.contains("newer than current")); |
| 340 | } |
| 341 | |
| 342 | #[test] |
| 343 | fn backup_creates_bak_file_alongside_original() { |
| 344 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 345 | let path = tmp.path().join("session_abc.json"); |
| 346 | std::fs::write(&path, r#"{"id":"abc"}"#).expect("write"); |
| 347 | let bak = backup_before_migrate(&path, "test_session"); |
| 348 | assert!(bak.exists(), "bak file must exist at {}", bak.display()); |
| 349 | assert_eq!( |
| 350 | std::fs::read_to_string(&bak).expect("read bak"), |
| 351 | r#"{"id":"abc"}"# |
| 352 | ); |
| 353 | // Bak is path.json.bak (extension append, not replace). |
| 354 | assert!( |
| 355 | bak.to_string_lossy().ends_with(".json.bak"), |
| 356 | "bak suffix must be `.json.bak`; got {}", |
| 357 | bak.display() |
| 358 | ); |
| 359 | } |
| 360 | |
| 361 | #[test] |
| 362 | fn backup_failure_does_not_panic_or_propagate() { |
| 363 | // Pointing at a non-existent source: copy fails, but the function |
| 364 | // returns the bak path it would have used and logs a warning. |
| 365 | let tmp = tempfile::tempdir().expect("tempdir"); |
| 366 | let path = tmp.path().join("does_not_exist.json"); |
| 367 | let bak = backup_before_migrate(&path, "test_session"); |
| 368 | // The path is well-formed even though copy failed. |
| 369 | assert!(bak.to_string_lossy().ends_with(".json.bak")); |
| 370 | } |
| 371 | } |
| 372 |