diff --git a/Cargo.lock b/Cargo.lock index 496c2908..e97504dc 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -821,6 +821,7 @@ dependencies = [ "codex-config", "codex-db", "codex-events", + "codex-migrate", "codex-models", "codex-parsers", "codex-scanner", @@ -990,6 +991,28 @@ dependencies = [ "uuid", ] +[[package]] +name = "codex-migrate" +version = "1.39.14" +dependencies = [ + "anyhow", + "chrono", + "codex-config", + "codex-db", + "flate2", + "futures", + "migration", + "sea-orm", + "serde", + "serde_json", + "tar", + "tempfile", + "tokio", + "tracing", + "url", + "uuid", +] + [[package]] name = "codex-models" version = "1.39.14" @@ -1970,6 +1993,16 @@ version = "0.2.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "28dea519a9695b9977216879a3ebfddf92f1c08c05d984f8996aecd6ecdc811d" +[[package]] +name = "filetime" +version = "0.2.29" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "5c287a33c7f0a620c38e641e7f60827713987b3c0f26e8ddc9462cc69cf75759" +dependencies = [ + "cfg-if", + "libc", +] + [[package]] name = "find-msvc-tools" version = "0.1.9" @@ -6155,6 +6188,17 @@ version = "1.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369" +[[package]] +name = "tar" +version = "0.4.46" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3f6221d9a6003c78398e3b239969f352578258df48c8eb051caadae0015bc840" +dependencies = [ + "filetime", + "libc", + "xattr", +] + [[package]] name = "tempfile" version = "3.25.0" @@ -7758,6 +7802,16 @@ dependencies = [ "tap", ] +[[package]] +name = "xattr" +version = "1.6.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "32e45ad4206f6d2479085147f02bc2ef834ac85886624a23575ae137c8aa8156" +dependencies = [ + "libc", + "rustix", +] + [[package]] name = "xmlwriter" version = "0.1.0" diff --git a/Cargo.toml b/Cargo.toml index 4f4799d3..2cfe4949 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -47,6 +47,7 @@ members = [ "crates/codex-utils", "crates/codex-parsers", "crates/codex-db", + "crates/codex-migrate", "crates/codex-services", "crates/codex-search", "crates/codex-scanner", @@ -81,6 +82,7 @@ codex-cli-common = { path = "crates/codex-cli-common", default-features = false codex-config = { path = "crates/codex-config" } codex-db = { path = "crates/codex-db" } codex-events = { path = "crates/codex-events" } +codex-migrate = { path = "crates/codex-migrate" } codex-models = { path = "crates/codex-models" } codex-parsers = { path = "crates/codex-parsers", default-features = false } codex-scanner = { path = "crates/codex-scanner", default-features = false } @@ -114,6 +116,7 @@ codex-cli-common = { workspace = true } codex-config = { workspace = true } codex-db = { workspace = true } codex-events = { workspace = true } +codex-migrate = { workspace = true } codex-models = { workspace = true } codex-parsers = { workspace = true } codex-scanner = { workspace = true } diff --git a/crates/codex-migrate/Cargo.toml b/crates/codex-migrate/Cargo.toml new file mode 100644 index 00000000..93fbb6e6 --- /dev/null +++ b/crates/codex-migrate/Cargo.toml @@ -0,0 +1,51 @@ +[package] +name = "codex-migrate" +version.workspace = true +edition.workspace = true +publish = false + +[lib] +name = "codex_migrate" +path = "src/lib.rs" + +[dependencies] +anyhow = { workspace = true } +chrono = { workspace = true } +serde = { workspace = true } +tokio = { workspace = true } +tracing = { workspace = true } +uuid = { workspace = true } + +codex-db = { workspace = true } +codex-config = { workspace = true } +migration = { path = "../../migration" } + +# SeaORM — same backend feature set as codex-db so both drivers stay supported. +sea-orm = { version = "1.1", features = [ + "sqlx-postgres", + "sqlx-sqlite", + "runtime-tokio-rustls", + "macros", + "with-chrono", + "with-uuid", +] } + +# NDJSON row serialization. +serde_json = "1.0" +# Streaming rows out of `Entity::find().stream()`. +futures = "0.3" +# Archive packing: gzip-compressed tar for the portable export bundle. +tar = "0.4" +flate2 = "1.0" +# Parsing postgres:// connection URLs for the `copy` command. +url = "2" +# Staging directories for export/import at runtime. +tempfile = { workspace = true } + +[dev-dependencies] +codex-db = { workspace = true, features = ["test-utils"] } +migration = { path = "../../migration" } +tempfile = { workspace = true } +tokio = { workspace = true } +uuid = { workspace = true } +chrono = { workspace = true } diff --git a/crates/codex-migrate/src/archive.rs b/crates/codex-migrate/src/archive.rs new file mode 100644 index 00000000..41cafc88 --- /dev/null +++ b/crates/codex-migrate/src/archive.rs @@ -0,0 +1,258 @@ +//! Portable export archive: a gzip-compressed tar bundling the database +//! (as NDJSON) plus the on-disk artifacts the database only references by path. +//! +//! Layout: +//! ```text +//! manifest.json +//! db/.ndjson +//! thumbnails/ (files.thumbnail_dir, when bundled) +//! uploads/ (files.uploads_dir, when bundled) +//! plugins/ (files.plugins_dir, when bundled) +//! cache/ (pdf.cache_dir, only with --include-cache) +//! ``` + +use std::path::{Path, PathBuf}; + +use anyhow::{Context, Result}; +use sea_orm::{ConnectionTrait, DatabaseConnection}; + +use crate::manifest::{ + ARCHIVE_FORMAT_VERSION, ArtifactEntry, ArtifactGroup, Manifest, TableCount, backend_name, + schema_version, +}; +use crate::reroot::{self, RerootStats}; +use crate::{TransferReport, registry}; + +/// An artifact tree to bundle on export: which group, and the source +/// instance's directory that currently holds those files. +#[derive(Clone, Debug)] +pub struct ArtifactSource { + pub group: ArtifactGroup, + pub source_dir: PathBuf, +} + +/// Where an artifact group's files should be written on import, for the target +/// instance. +#[derive(Clone, Debug)] +pub struct ArtifactTarget { + pub group: ArtifactGroup, + pub target_dir: PathBuf, +} + +/// Outcome of [`import_archive`]. +#[derive(Debug)] +pub struct ImportOutcome { + pub manifest: Manifest, + pub report: TransferReport, + pub reroot: RerootStats, +} + +/// Export `conn` and the given artifact trees into a `.tar.gz` at `out_path`. +/// Returns the manifest that was written. Artifact sources whose directory does +/// not exist are silently skipped (and omitted from the manifest). +pub async fn export_archive( + conn: &DatabaseConnection, + out_path: &Path, + artifacts: &[ArtifactSource], +) -> Result { + let staging = tempfile::tempdir().context("failed to create export staging dir")?; + let db_dir = staging.path().join("db"); + std::fs::create_dir_all(&db_dir)?; + + let counts = registry::dump_all_to_dir(conn, &db_dir) + .await + .context("failed to dump database tables")?; + let total_rows = counts.iter().map(|c| c.rows).sum(); + + let present: Vec<&ArtifactSource> = + artifacts.iter().filter(|a| a.source_dir.exists()).collect(); + + let manifest = Manifest { + format_version: ARCHIVE_FORMAT_VERSION, + source_backend: backend_name(conn.get_database_backend()).to_string(), + schema_version: schema_version(conn).await?, + tables: counts + .into_iter() + .map(|c| TableCount { + table: c.table, + rows: c.rows, + }) + .collect(), + total_rows, + artifacts: present + .iter() + .map(|a| ArtifactEntry { + group: a.group, + source_base_dir: a.source_dir.to_string_lossy().into_owned(), + }) + .collect(), + created_at: chrono::Utc::now().to_rfc3339(), + }; + + std::fs::write( + staging.path().join("manifest.json"), + serde_json::to_vec_pretty(&manifest)?, + ) + .context("failed to write manifest.json")?; + + write_archive(out_path, staging.path(), &present) + .with_context(|| format!("failed to write archive {}", out_path.display()))?; + + Ok(manifest) +} + +/// Import a `.tar.gz` produced by [`export_archive`] into `conn`, unpacking +/// bundled artifacts to the given targets and re-rooting DB paths accordingly. +/// +/// This performs the load and re-root only. The schema-version and +/// fresh-target safety checks belong to the CLI layer, which should validate +/// the returned/[`extract`]ed manifest before committing to an import. +pub async fn import_archive( + conn: &DatabaseConnection, + in_path: &Path, + targets: &[ArtifactTarget], +) -> Result { + let staging = tempfile::tempdir().context("failed to create import staging dir")?; + let manifest = extract(in_path, staging.path())?; + + let report = crate::load_from_dir(conn, &staging.path().join("db")) + .await + .context("failed to load database from archive")?; + + // Unpack each bundled artifact group to its target dir and record the + // base-dir remapping for path re-rooting. + let mut thumbnail_remap: Option<(String, String)> = None; + let mut uploads_remap: Option<(String, String)> = None; + + for entry in &manifest.artifacts { + let Some(target) = targets.iter().find(|t| t.group == entry.group) else { + continue; + }; + let src = staging.path().join(entry.group.archive_dir()); + if src.exists() { + copy_dir_all(&src, &target.target_dir).with_context(|| { + format!( + "failed to unpack {} into {}", + entry.group.archive_dir(), + target.target_dir.display() + ) + })?; + } + let to = target.target_dir.to_string_lossy().into_owned(); + match entry.group { + ArtifactGroup::Thumbnails => { + thumbnail_remap = Some((entry.source_base_dir.clone(), to)) + } + ArtifactGroup::Uploads => uploads_remap = Some((entry.source_base_dir.clone(), to)), + ArtifactGroup::Plugins | ArtifactGroup::Cache => {} + } + } + + let reroot = reroot::reroot_all( + conn, + thumbnail_remap + .as_ref() + .map(|(f, t)| (f.as_str(), t.as_str())), + uploads_remap + .as_ref() + .map(|(f, t)| (f.as_str(), t.as_str())), + ) + .await + .context("failed to re-root artifact paths")?; + + Ok(ImportOutcome { + manifest, + report, + reroot, + }) +} + +/// Extract an archive into `dest` and return its (validated-format) manifest. +/// Exposed so the CLI can inspect the manifest before importing. +pub fn extract(in_path: &Path, dest: &Path) -> Result { + let file = std::fs::File::open(in_path) + .with_context(|| format!("failed to open archive {}", in_path.display()))?; + let decoder = flate2::read::GzDecoder::new(file); + let mut archive = tar::Archive::new(decoder); + archive + .unpack(dest) + .context("failed to extract archive (corrupt or not a codex export?)")?; + + let manifest_file = std::fs::File::open(dest.join("manifest.json")) + .context("archive is missing manifest.json")?; + let manifest: Manifest = serde_json::from_reader(std::io::BufReader::new(manifest_file)) + .context("failed to parse manifest.json")?; + + check_format_version(&manifest)?; + Ok(manifest) +} + +/// Read only the `manifest.json` from an archive without unpacking it, for +/// pre-flight checks (schema version, artifact list) before a heavy import. +/// `manifest.json` is written first, so this stops after the first entry. +pub fn read_manifest(in_path: &Path) -> Result { + let file = std::fs::File::open(in_path) + .with_context(|| format!("failed to open archive {}", in_path.display()))?; + let decoder = flate2::read::GzDecoder::new(file); + let mut archive = tar::Archive::new(decoder); + + for entry in archive.entries().context("archive is not a valid tar")? { + let mut entry = entry?; + if entry.path()?.as_ref() == Path::new("manifest.json") { + let manifest: Manifest = + serde_json::from_reader(&mut entry).context("failed to parse manifest.json")?; + check_format_version(&manifest)?; + return Ok(manifest); + } + } + anyhow::bail!("archive is missing manifest.json"); +} + +fn check_format_version(manifest: &Manifest) -> Result<()> { + if manifest.format_version != ARCHIVE_FORMAT_VERSION { + anyhow::bail!( + "unsupported archive format version {} (this build reads version {})", + manifest.format_version, + ARCHIVE_FORMAT_VERSION + ); + } + Ok(()) +} + +/// Pack the staged manifest + `db/` tree and the artifact source dirs into a +/// gzip tar. Artifacts are streamed directly from their source locations (no +/// intermediate copy). +fn write_archive(out: &Path, staging: &Path, artifacts: &[&ArtifactSource]) -> Result<()> { + if let Some(parent) = out.parent() + && !parent.as_os_str().is_empty() + { + std::fs::create_dir_all(parent).ok(); + } + let file = std::fs::File::create(out)?; + let encoder = flate2::write::GzEncoder::new(file, flate2::Compression::default()); + let mut builder = tar::Builder::new(encoder); + + builder.append_path_with_name(staging.join("manifest.json"), "manifest.json")?; + builder.append_dir_all("db", staging.join("db"))?; + for a in artifacts { + builder.append_dir_all(a.group.archive_dir(), &a.source_dir)?; + } + + builder.into_inner()?.finish()?; + Ok(()) +} + +/// Recursively copy `src` into `dst`, creating `dst` and intermediate dirs. +fn copy_dir_all(src: &Path, dst: &Path) -> Result<()> { + std::fs::create_dir_all(dst)?; + for entry in std::fs::read_dir(src)? { + let entry = entry?; + let target = dst.join(entry.file_name()); + if entry.file_type()?.is_dir() { + copy_dir_all(&entry.path(), &target)?; + } else { + std::fs::copy(entry.path(), &target)?; + } + } + Ok(()) +} diff --git a/crates/codex-migrate/src/conn.rs b/crates/codex-migrate/src/conn.rs new file mode 100644 index 00000000..d383c064 --- /dev/null +++ b/crates/codex-migrate/src/conn.rs @@ -0,0 +1,106 @@ +//! Build a [`DatabaseConfig`] from a connection URL, for the `copy` command's +//! `--from` / `--to` flags and the `CODEX_SOURCE_/TARGET_DATABASE_URL` env vars. +//! +//! Supported forms: +//! - `sqlite://` (also `sqlite:`) — e.g. `sqlite:///abs/codex.db` +//! or `sqlite://relative/codex.db` +//! - `postgres://user:pass@host:port/dbname` (also `postgresql://`) +//! +//! Passwords with characters that need URL-escaping are fiddly to pass on a +//! command line; prefer `--from-config` / `--to-config` (or env) for those. + +use anyhow::{Context, Result, bail}; +use codex_config::{DatabaseConfig, DatabaseType, PostgresConfig, SQLiteConfig}; + +/// Parse a database connection URL into a [`DatabaseConfig`]. Pool settings and +/// other fields use their defaults. +pub fn database_config_from_url(raw: &str) -> Result { + if let Some(rest) = raw + .strip_prefix("sqlite://") + .or_else(|| raw.strip_prefix("sqlite:")) + { + let path = rest.split('?').next().unwrap_or(rest); + if path.is_empty() { + bail!("sqlite URL is missing a file path: {raw}"); + } + return Ok(DatabaseConfig { + db_type: DatabaseType::SQLite, + postgres: None, + sqlite: Some(SQLiteConfig { + path: path.to_string(), + ..SQLiteConfig::default() + }), + }); + } + + if raw.starts_with("postgres://") || raw.starts_with("postgresql://") { + let parsed = + url::Url::parse(raw).with_context(|| format!("invalid postgres URL: {raw}"))?; + let host = parsed.host_str().unwrap_or("localhost").to_string(); + let port = parsed.port().unwrap_or(5432); + let username = parsed.username().to_string(); + let password = parsed.password().unwrap_or("").to_string(); + let database_name = parsed.path().trim_start_matches('/').to_string(); + if database_name.is_empty() { + bail!("postgres URL is missing a database name: {raw}"); + } + return Ok(DatabaseConfig { + db_type: DatabaseType::Postgres, + postgres: Some(PostgresConfig { + host, + port, + username, + password, + database_name, + ..PostgresConfig::default() + }), + sqlite: None, + }); + } + + bail!("unsupported database URL scheme (expected sqlite:// or postgres://): {raw}"); +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn parses_absolute_sqlite() { + let cfg = database_config_from_url("sqlite:///var/lib/codex/codex.db").unwrap(); + assert_eq!(cfg.db_type, DatabaseType::SQLite); + assert_eq!(cfg.sqlite.unwrap().path, "/var/lib/codex/codex.db"); + } + + #[test] + fn parses_relative_sqlite_and_strips_query() { + let cfg = database_config_from_url("sqlite://data/codex.db?mode=rwc").unwrap(); + assert_eq!(cfg.sqlite.unwrap().path, "data/codex.db"); + } + + #[test] + fn parses_postgres() { + let cfg = + database_config_from_url("postgres://codex:secret@db.internal:5433/codexdb").unwrap(); + assert_eq!(cfg.db_type, DatabaseType::Postgres); + let pg = cfg.postgres.unwrap(); + assert_eq!(pg.host, "db.internal"); + assert_eq!(pg.port, 5433); + assert_eq!(pg.username, "codex"); + assert_eq!(pg.password, "secret"); + assert_eq!(pg.database_name, "codexdb"); + } + + #[test] + fn postgres_defaults_port_when_absent() { + let cfg = database_config_from_url("postgresql://u@localhost/db").unwrap(); + assert_eq!(cfg.postgres.unwrap().port, 5432); + } + + #[test] + fn rejects_unknown_scheme() { + assert!(database_config_from_url("mysql://x/y").is_err()); + assert!(database_config_from_url("postgres://u@h/").is_err()); + assert!(database_config_from_url("sqlite://").is_err()); + } +} diff --git a/crates/codex-migrate/src/engine.rs b/crates/codex-migrate/src/engine.rs new file mode 100644 index 00000000..3b495e95 --- /dev/null +++ b/crates/codex-migrate/src/engine.rs @@ -0,0 +1,127 @@ +//! Generic, entity-driven transfer primitives. +//! +//! Every function here is generic over a SeaORM [`EntityTrait`]. Because +//! SeaORM performs the engine-specific type mapping (UUID blob ↔ native +//! `uuid`, text JSON ↔ JSONB, 0/1 ↔ `bool`, timestamps) when it materializes a +//! typed `Model`, moving rows as `Model` values is correct by construction +//! across SQLite and PostgreSQL — no hand-written cast rules, no raw bytes. + +use std::io::{BufRead, Write}; + +use anyhow::Result; +use futures::TryStreamExt; +use sea_orm::{ConnectionTrait, EntityTrait, IntoActiveModel, PaginatorTrait, StreamTrait}; +use serde::Serialize; +use serde::de::DeserializeOwned; + +/// Default insert batch size. Bounds memory and round-trips while staying well +/// under parameter limits for the widest tables. +pub const DEFAULT_BATCH_SIZE: usize = 1000; + +/// Stream every row of `E` and write it as one NDJSON line to `out`. +/// Returns the number of rows written. +pub async fn dump_table(conn: &C, out: &mut W) -> Result +where + E: EntityTrait, + E::Model: Serialize, + C: ConnectionTrait + StreamTrait, + W: Write, +{ + let mut stream = E::find().stream(conn).await?; + let mut count = 0u64; + while let Some(model) = stream.try_next().await? { + serde_json::to_writer(&mut *out, &model)?; + out.write_all(b"\n")?; + count += 1; + } + Ok(count) +} + +/// Read NDJSON lines from `reader`, deserialize each into `E::Model`, and +/// insert them into `conn` in batches of `batch_size`. Returns rows inserted. +pub async fn load_table(conn: &C, reader: R, batch_size: usize) -> Result +where + E: EntityTrait, + E::Model: DeserializeOwned + IntoActiveModel, + C: ConnectionTrait, + R: BufRead, +{ + let mut batch: Vec = Vec::with_capacity(batch_size); + let mut count = 0u64; + for line in reader.lines() { + let line = line?; + if line.trim().is_empty() { + continue; + } + let model: E::Model = serde_json::from_str(&line)?; + batch.push(model.into_active_model()); + if batch.len() >= batch_size { + count += insert_batch::(conn, std::mem::take(&mut batch)).await?; + } + } + if !batch.is_empty() { + count += insert_batch::(conn, batch).await?; + } + Ok(count) +} + +/// Stream every row of `E` directly from `src` into `dst` in batches, without +/// an intermediate serialized form. This is the path used by the direct +/// database-to-database `copy`. +pub async fn copy_table(src: &S, dst: &D, batch_size: usize) -> Result +where + E: EntityTrait, + E::Model: IntoActiveModel, + S: ConnectionTrait + StreamTrait, + D: ConnectionTrait, +{ + let mut stream = E::find().stream(src).await?; + let mut batch: Vec = Vec::with_capacity(batch_size); + let mut count = 0u64; + while let Some(model) = stream.try_next().await? { + batch.push(model.into_active_model()); + if batch.len() >= batch_size { + count += insert_batch::(dst, std::mem::take(&mut batch)).await?; + } + } + if !batch.is_empty() { + count += insert_batch::(dst, batch).await?; + } + Ok(count) +} + +/// Count rows in `E`. +pub async fn count_table(conn: &C) -> Result +where + E: EntityTrait, + E::Model: Send + Sync, + C: ConnectionTrait, +{ + Ok(E::find().count(conn).await?) +} + +/// Delete every row of `E`. Used by `--replace` before a load. Callers must +/// have foreign-key enforcement disabled (see [`crate::fk`]). +pub async fn truncate_table(conn: &C) -> Result +where + E: EntityTrait, + C: ConnectionTrait, +{ + Ok(E::delete_many().exec(conn).await?.rows_affected) +} + +/// Insert one batch, using `exec_without_returning` to avoid a per-row +/// RETURNING clause on bulk loads. No-op (and no DB round-trip) when empty. +async fn insert_batch(conn: &C, batch: Vec) -> Result +where + E: EntityTrait, + E::Model: IntoActiveModel, + C: ConnectionTrait, +{ + let n = batch.len() as u64; + if n == 0 { + return Ok(0); + } + E::insert_many(batch).exec_without_returning(conn).await?; + Ok(n) +} diff --git a/crates/codex-migrate/src/fk.rs b/crates/codex-migrate/src/fk.rs new file mode 100644 index 00000000..cb5fb8cc --- /dev/null +++ b/crates/codex-migrate/src/fk.rs @@ -0,0 +1,35 @@ +//! Foreign-key enforcement control for bulk loads. +//! +//! A 1:1 load inserts tables in an arbitrary order, so per-row FK enforcement +//! would reject rows whose parents are inserted later. Rather than maintain a +//! drift-prone topological insert order across ~60 tables (some +//! self-referential), we disable enforcement for the duration of the load and +//! rely on the source data already being internally consistent. +//! +//! These calls are transaction-scoped and must run on the same connection that +//! performs the load — pass a [`sea_orm::DatabaseTransaction`]. +//! +//! - **SQLite:** `PRAGMA defer_foreign_keys = ON` defers all FK checks until +//! COMMIT (and auto-resets there), so a complete, consistent dataset is still +//! validated as a whole when the transaction commits. +//! - **PostgreSQL:** `SET LOCAL session_replication_role = replica` suppresses +//! FK (and other) triggers for the transaction. This requires the connection +//! role to be a superuser or the table owner. There is no commit-time +//! re-check, so integrity rests on source consistency plus the row-count +//! verification performed after the load. + +use anyhow::{Context, Result}; +use sea_orm::{ConnectionTrait, DatabaseBackend}; + +/// Disable foreign-key enforcement for the current transaction. +pub async fn disable(conn: &C) -> Result<()> { + let sql = match conn.get_database_backend() { + DatabaseBackend::Sqlite => "PRAGMA defer_foreign_keys = ON", + DatabaseBackend::Postgres => "SET LOCAL session_replication_role = replica", + DatabaseBackend::MySql => "SET FOREIGN_KEY_CHECKS = 0", + }; + conn.execute_unprepared(sql) + .await + .with_context(|| format!("failed to disable foreign-key enforcement ({sql})"))?; + Ok(()) +} diff --git a/crates/codex-migrate/src/guard.rs b/crates/codex-migrate/src/guard.rs new file mode 100644 index 00000000..d164a352 --- /dev/null +++ b/crates/codex-migrate/src/guard.rs @@ -0,0 +1,26 @@ +//! Safety gate for destructive loads. +//! +//! `import` and `copy` always truncate the target (a faithful mirror must +//! overwrite migration-seeded rows), so before doing that we refuse a target +//! that already holds *user* data unless the caller passed `--replace`. +//! +//! "User data" is deliberately a small, high-signal set of top-level tables. +//! Migration-seeded tables (e.g. `settings`) are non-empty on any freshly +//! migrated database and must NOT be treated as user data, or a fresh target +//! would always look occupied. + +use anyhow::Result; +use codex_db::entities::{books, libraries, series, users}; +use sea_orm::ConnectionTrait; + +use crate::engine::count_table; + +/// Returns `true` if the target already contains user-owned content +/// (libraries, series, books, or users). A freshly migrated, unused database +/// returns `false`. +pub async fn has_user_data(conn: &C) -> Result { + Ok(count_table::(conn).await? > 0 + || count_table::(conn).await? > 0 + || count_table::(conn).await? > 0 + || count_table::(conn).await? > 0) +} diff --git a/crates/codex-migrate/src/lib.rs b/crates/codex-migrate/src/lib.rs new file mode 100644 index 00000000..9a00ff60 --- /dev/null +++ b/crates/codex-migrate/src/lib.rs @@ -0,0 +1,106 @@ +//! Portable, engine-agnostic, entity-driven data transfer for Codex. +//! +//! Moves every row of every table between two SeaORM backends by materializing +//! typed `Model` values, so SQLite ↔ PostgreSQL type differences (UUID blob vs +//! native `uuid`, text JSON vs JSONB, 0/1 vs `bool`) are handled by SeaORM +//! rather than by hand-written casts. This crate is the shared engine behind +//! the `export`, `import`, and `copy` subcommands. +//! +//! Loads run inside a single destination transaction with foreign-key +//! enforcement disabled ([`fk`]), then commit as a unit. On SQLite the commit +//! re-validates the whole dataset (deferred FK check); on PostgreSQL integrity +//! rests on source consistency plus the row-count verification in [`verify`]. + +pub mod archive; +pub mod conn; +pub mod engine; +pub mod fk; +pub mod guard; +pub mod manifest; +pub mod registry; +pub mod reroot; +pub mod verify; + +pub use conn::database_config_from_url; + +use std::path::Path; + +use anyhow::{Context, Result}; +use sea_orm::{DatabaseConnection, TransactionTrait}; + +pub use manifest::Manifest; +pub use registry::{TableRows, table_names}; + +/// Outcome of a [`transfer`], carrying per-table and total row counts. +#[derive(Debug, Clone)] +pub struct TransferReport { + pub tables: Vec, + pub total_rows: u64, +} + +/// Copy all data from `src` into `dst`, producing a faithful 1:1 mirror. +/// +/// The destination must already have the schema applied (run migrations +/// first). Because migrations seed rows (e.g. default `settings`), a freshly +/// migrated database is *not* empty; every table is therefore truncated before +/// the load so the source's own rows are authoritative. The entire operation +/// (disable FK → truncate → load) runs in one destination transaction and +/// commits atomically — on SQLite the commit re-validates all foreign keys. +/// +/// This is the low-level primitive. The destructive-overwrite safety gate +/// (refuse a target that already holds user data unless `--replace`) lives in +/// the CLI layer, not here. +pub async fn transfer( + src: &DatabaseConnection, + dst: &DatabaseConnection, +) -> Result { + let txn = dst + .begin() + .await + .context("failed to open destination transaction")?; + + fk::disable(&txn).await?; + + registry::truncate_all(&txn) + .await + .context("failed to clear destination before load")?; + + let tables = registry::copy_all(src, &txn, engine::DEFAULT_BATCH_SIZE) + .await + .context("failed while copying tables")?; + + txn.commit() + .await + .context("failed to commit destination transaction (foreign-key check may have failed)")?; + + let total_rows = tables.iter().map(|t| t.rows).sum(); + Ok(TransferReport { tables, total_rows }) +} + +/// Load a directory of `
.ndjson` files into `dst`, producing a faithful +/// 1:1 mirror. Same transaction semantics as [`transfer`] (disable FK → +/// truncate → load → commit); this is the archive-backed load used by +/// `import`. +pub async fn load_from_dir(dst: &DatabaseConnection, db_dir: &Path) -> Result { + let txn = dst + .begin() + .await + .context("failed to open destination transaction")?; + + fk::disable(&txn).await?; + + registry::truncate_all(&txn) + .await + .context("failed to clear destination before load")?; + + let tables = registry::load_all_from_dir(&txn, db_dir, engine::DEFAULT_BATCH_SIZE) + .await + .context("failed while loading tables from archive")?; + + txn.commit() + .await + .context("failed to commit destination transaction (foreign-key check may have failed)")?; + + let total_rows = tables.iter().map(|t| t.rows).sum(); + Ok(TransferReport { tables, total_rows }) +} diff --git a/crates/codex-migrate/src/manifest.rs b/crates/codex-migrate/src/manifest.rs new file mode 100644 index 00000000..3709fb95 --- /dev/null +++ b/crates/codex-migrate/src/manifest.rs @@ -0,0 +1,92 @@ +//! Archive manifest: the `manifest.json` at the root of an export bundle. +//! +//! Records enough to validate an import (format + schema version), to verify +//! it (per-table row counts), and to re-root on-disk artifact paths (which +//! artifact groups are present and the source base directory each came from). + +use anyhow::{Context, Result}; +use migration::{Migrator, MigratorTrait}; +use sea_orm::{ConnectionTrait, DatabaseBackend}; +use serde::{Deserialize, Serialize}; + +/// Bumped when the on-disk archive layout or manifest shape changes +/// incompatibly. Import refuses a mismatched major format. +pub const ARCHIVE_FORMAT_VERSION: u32 = 1; + +/// A bundleable tree of on-disk files that the database only references by +/// path. Each maps to a top-level directory inside the archive. +#[derive(Serialize, Deserialize, Clone, Copy, Debug, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum ArtifactGroup { + /// Generated book/series thumbnails (`files.thumbnail_dir`). + Thumbnails, + /// Uploaded / extracted / plugin covers (`files.uploads_dir`). + Uploads, + /// Plugin data and credentials on disk (`files.plugins_dir`). + Plugins, + /// Rendered PDF page cache (`pdf.cache_dir`) — reproducible, opt-in. + Cache, +} + +impl ArtifactGroup { + /// The archive-relative top-level directory for this group. + pub fn archive_dir(self) -> &'static str { + match self { + ArtifactGroup::Thumbnails => "thumbnails", + ArtifactGroup::Uploads => "uploads", + ArtifactGroup::Plugins => "plugins", + ArtifactGroup::Cache => "cache", + } + } +} + +/// Which artifact group is present in the archive and the source instance's +/// base directory it was captured from (needed to re-root DB paths on import). +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct ArtifactEntry { + pub group: ArtifactGroup, + /// The source instance's configured base dir (e.g. `data/thumbnails`). + pub source_base_dir: String, +} + +/// Row count for one table at export time. +#[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq)] +pub struct TableCount { + pub table: String, + pub rows: u64, +} + +/// The archive's `manifest.json`. +#[derive(Serialize, Deserialize, Clone, Debug)] +pub struct Manifest { + pub format_version: u32, + /// `"sqlite"` or `"postgres"` — the backend the data was exported from. + pub source_backend: String, + /// Name of the last applied migration at export time, or `None` if the + /// source had no migration bookkeeping. + pub schema_version: Option, + pub tables: Vec, + pub total_rows: u64, + pub artifacts: Vec, + /// RFC 3339 timestamp of the export. + pub created_at: String, +} + +/// Render a backend as its manifest string. +pub fn backend_name(backend: DatabaseBackend) -> &'static str { + match backend { + DatabaseBackend::Sqlite => "sqlite", + DatabaseBackend::Postgres => "postgres", + DatabaseBackend::MySql => "mysql", + } +} + +/// The current schema version of `conn`: the lexicographically-last applied +/// migration name (migration names are timestamp-prefixed, so this is the most +/// recent). Returns `None` if no migrations are applied. +pub async fn schema_version(conn: &C) -> Result> { + let applied = Migrator::get_applied_migrations(conn) + .await + .context("failed to read applied migrations")?; + Ok(applied.into_iter().map(|m| m.name().to_string()).max()) +} diff --git a/crates/codex-migrate/src/registry.rs b/crates/codex-migrate/src/registry.rs new file mode 100644 index 00000000..e0f68482 --- /dev/null +++ b/crates/codex-migrate/src/registry.rs @@ -0,0 +1,208 @@ +//! The entity registry: every table listed exactly once. +//! +//! [`for_each_entity!`] is an x-macro that expands a caller-supplied per-entity +//! macro once for each entity module. All collective operations +//! (count / copy / dump / load / truncate) are generated from this single list, +//! so adding a table means adding one line here — and the drift-guard test +//! fails loudly if a migration adds a table that is missing from this list. + +use std::path::Path; + +use anyhow::Result; +use sea_orm::{ConnectionTrait, EntityName, StreamTrait}; + +use crate::engine; + +/// Row count for a single table, produced by the collective operations. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct TableRows { + pub table: String, + pub rows: u64, +} + +/// Expand `$op!(module_name)` once per entity, in `entities/mod.rs` order. +/// +/// Order is irrelevant to correctness: loads run with foreign-key enforcement +/// disabled (see [`crate::fk`]), so parents and children may be inserted in any +/// order. +macro_rules! for_each_entity { + ($op:ident) => { + $op!(api_keys); + $op!(book_covers); + $op!(book_duplicates); + $op!(book_external_ids); + $op!(book_external_links); + $op!(book_genres); + $op!(book_metadata); + $op!(book_tags); + $op!(books); + $op!(email_verification_tokens); + $op!(libraries); + $op!(library_jobs); + $op!(metadata_sources); + $op!(pages); + $op!(plugin_failures); + $op!(plugins); + $op!(read_progress); + $op!(refresh_tokens); + $op!(scheduled_firing_claims); + $op!(series); + $op!(settings); + $op!(settings_history); + $op!(task_metrics); + $op!(tasks); + $op!(users); + $op!(oidc_connections); + $op!(plugin_data); + $op!(user_plugin_data); + $op!(user_plugins); + $op!(filter_presets); + $op!(genres); + $op!(release_ledger); + $op!(release_sources); + $op!(series_aliases); + $op!(series_alternate_titles); + $op!(series_covers); + $op!(series_duplicates); + $op!(series_exports); + $op!(series_external_ids); + $op!(series_external_links); + $op!(series_external_ratings); + $op!(series_genres); + $op!(series_metadata); + $op!(series_tags); + $op!(series_tracking); + $op!(tags); + $op!(user_preferences); + $op!(user_series_ratings); + $op!(series_sharing_tags); + $op!(sharing_tags); + $op!(user_sharing_tags); + $op!(access_group_oidc_mappings); + $op!(access_group_sharing_tags); + $op!(access_groups); + $op!(user_access_groups); + $op!(collection_series); + $op!(collections); + $op!(read_list_books); + $op!(read_lists); + $op!(want_to_read); + }; +} + +/// The table name of every registered entity, in registry order. +pub fn table_names() -> Vec { + let mut names = Vec::new(); + macro_rules! push_name { + ($ent:ident) => {{ + names.push( + ::default() + .table_name() + .to_string(), + ); + }}; + } + for_each_entity!(push_name); + names +} + +/// Count rows in every table. +pub async fn count_all(conn: &C) -> Result> +where + C: ConnectionTrait, +{ + let mut rows = Vec::new(); + macro_rules! count_one { + ($ent:ident) => {{ + type E = codex_db::entities::$ent::Entity; + let table = ::default().table_name().to_string(); + let n = engine::count_table::(conn).await?; + rows.push(TableRows { table, rows: n }); + }}; + } + for_each_entity!(count_one); + Ok(rows) +} + +/// Delete every row of every table. Callers must disable FK enforcement first +/// (used by `--replace`). +pub async fn truncate_all(conn: &C) -> Result<()> +where + C: ConnectionTrait, +{ + macro_rules! truncate_one { + ($ent:ident) => {{ + type E = codex_db::entities::$ent::Entity; + engine::truncate_table::(conn).await?; + }}; + } + for_each_entity!(truncate_one); + Ok(()) +} + +/// Stream every table directly from `src` into `dst` (the direct copy path). +pub async fn copy_all(src: &S, dst: &D, batch_size: usize) -> Result> +where + S: ConnectionTrait + StreamTrait, + D: ConnectionTrait, +{ + let mut rows = Vec::new(); + macro_rules! copy_one { + ($ent:ident) => {{ + type E = codex_db::entities::$ent::Entity; + let table = ::default().table_name().to_string(); + let n = engine::copy_table::(src, dst, batch_size).await?; + rows.push(TableRows { table, rows: n }); + }}; + } + for_each_entity!(copy_one); + Ok(rows) +} + +/// Dump every table to `dir` as one `
.ndjson` file each (the archive +/// payload). `dir` must already exist. +pub async fn dump_all_to_dir(conn: &C, dir: &Path) -> Result> +where + C: ConnectionTrait + StreamTrait, +{ + use std::io::Write as _; + let mut rows = Vec::new(); + macro_rules! dump_one { + ($ent:ident) => {{ + type E = codex_db::entities::$ent::Entity; + let table = ::default().table_name().to_string(); + let path = dir.join(format!("{table}.ndjson")); + let mut writer = std::io::BufWriter::new(std::fs::File::create(&path)?); + let n = engine::dump_table::(conn, &mut writer).await?; + writer.flush()?; + rows.push(TableRows { table, rows: n }); + }}; + } + for_each_entity!(dump_one); + Ok(rows) +} + +/// Load every table from a directory of `
.ndjson` files. Missing files +/// are treated as empty tables. +pub async fn load_all_from_dir(conn: &C, dir: &Path, batch_size: usize) -> Result> +where + C: ConnectionTrait, +{ + let mut rows = Vec::new(); + macro_rules! load_one { + ($ent:ident) => {{ + type E = codex_db::entities::$ent::Entity; + let table = ::default().table_name().to_string(); + let path = dir.join(format!("{table}.ndjson")); + let n = if path.exists() { + let reader = std::io::BufReader::new(std::fs::File::open(&path)?); + engine::load_table::(conn, reader, batch_size).await? + } else { + 0 + }; + rows.push(TableRows { table, rows: n }); + }}; + } + for_each_entity!(load_one); + Ok(rows) +} diff --git a/crates/codex-migrate/src/reroot.rs b/crates/codex-migrate/src/reroot.rs new file mode 100644 index 00000000..2bd76c7a --- /dev/null +++ b/crates/codex-migrate/src/reroot.rs @@ -0,0 +1,178 @@ +//! Rewrite on-disk file paths stored in the database after an import. +//! +//! Codex stores image paths with the configured base directory baked in and +//! reads them back with `fs::read(&path)` (no re-join). When an archive is +//! imported into an instance whose `files.*_dir` differ from the source's, the +//! stored paths must be re-rooted from the source base dir (recorded in the +//! manifest) to the target base dir, or the images won't resolve. +//! +//! Affected columns: +//! - `books.thumbnail_path` — under `files.thumbnail_dir` +//! - `book_covers.path` — under `files.uploads_dir` +//! - `series_covers.path` — under `files.uploads_dir` + +use anyhow::Result; +use codex_db::entities::{book_covers, books, series_covers}; +use sea_orm::sea_query::Expr; +use sea_orm::{ColumnTrait, ConnectionTrait, EntityTrait, QueryFilter, QuerySelect}; +use uuid::Uuid; + +/// How many stored paths were rewritten, per column family. +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct RerootStats { + pub thumbnails: u64, + pub covers: u64, +} + +/// Rewrite `path` if it sits under `from`, replacing that prefix with `to`. +/// Returns `None` when there is nothing to do (prefix doesn't match, or the +/// base dirs are identical). Matching respects path boundaries so +/// `data/thumb` does not match `data/thumbnails`. +pub fn reroot_path(path: &str, from: &str, to: &str) -> Option { + let from = from.trim_end_matches('/'); + let to = to.trim_end_matches('/'); + if from == to || from.is_empty() { + return None; + } + let rest = path.strip_prefix(from)?; + if rest.is_empty() || rest.starts_with('/') { + Some(format!("{to}{rest}")) + } else { + None + } +} + +/// Re-root all affected columns. `thumbnail` and `uploads` are each an optional +/// `(from, to)` base-dir pair; `None` skips that family. +pub async fn reroot_all( + conn: &C, + thumbnail: Option<(&str, &str)>, + uploads: Option<(&str, &str)>, +) -> Result { + let mut stats = RerootStats::default(); + + if let Some((from, to)) = thumbnail { + stats.thumbnails = reroot_books_thumbnails(conn, from, to).await?; + } + if let Some((from, to)) = uploads { + stats.covers = reroot_book_covers(conn, from, to).await? + + reroot_series_covers(conn, from, to).await?; + } + + Ok(stats) +} + +async fn reroot_books_thumbnails( + conn: &C, + from: &str, + to: &str, +) -> Result { + let rows: Vec<(Uuid, Option)> = books::Entity::find() + .select_only() + .column(books::Column::Id) + .column(books::Column::ThumbnailPath) + .filter(books::Column::ThumbnailPath.is_not_null()) + .into_tuple() + .all(conn) + .await?; + + let mut n = 0; + for (id, path) in rows { + let Some(path) = path else { continue }; + if let Some(new) = reroot_path(&path, from, to) { + books::Entity::update_many() + .col_expr(books::Column::ThumbnailPath, Expr::value(new)) + .filter(books::Column::Id.eq(id)) + .exec(conn) + .await?; + n += 1; + } + } + Ok(n) +} + +async fn reroot_book_covers(conn: &C, from: &str, to: &str) -> Result { + let rows: Vec<(Uuid, String)> = book_covers::Entity::find() + .select_only() + .column(book_covers::Column::Id) + .column(book_covers::Column::Path) + .into_tuple() + .all(conn) + .await?; + + let mut n = 0; + for (id, path) in rows { + if let Some(new) = reroot_path(&path, from, to) { + book_covers::Entity::update_many() + .col_expr(book_covers::Column::Path, Expr::value(new)) + .filter(book_covers::Column::Id.eq(id)) + .exec(conn) + .await?; + n += 1; + } + } + Ok(n) +} + +async fn reroot_series_covers(conn: &C, from: &str, to: &str) -> Result { + let rows: Vec<(Uuid, String)> = series_covers::Entity::find() + .select_only() + .column(series_covers::Column::Id) + .column(series_covers::Column::Path) + .into_tuple() + .all(conn) + .await?; + + let mut n = 0; + for (id, path) in rows { + if let Some(new) = reroot_path(&path, from, to) { + series_covers::Entity::update_many() + .col_expr(series_covers::Column::Path, Expr::value(new)) + .filter(series_covers::Column::Id.eq(id)) + .exec(conn) + .await?; + n += 1; + } + } + Ok(n) +} + +#[cfg(test)] +mod tests { + use super::reroot_path; + + #[test] + fn rewrites_matching_prefix() { + assert_eq!( + reroot_path( + "data/thumbnails/books/ab/x.jpg", + "data/thumbnails", + "/srv/thumbs" + ), + Some("/srv/thumbs/books/ab/x.jpg".to_string()) + ); + } + + #[test] + fn respects_path_boundaries() { + // `data/thumb` must NOT match `data/thumbnails/...`. + assert_eq!( + reroot_path("data/thumbnails/x.jpg", "data/thumb", "/new"), + None + ); + } + + #[test] + fn ignores_trailing_slashes_and_identical_dirs() { + assert_eq!( + reroot_path("/a/b/c.jpg", "/a/b/", "/x/y"), + Some("/x/y/c.jpg".to_string()) + ); + assert_eq!(reroot_path("/a/b/c.jpg", "/a/b", "/a/b"), None); + } + + #[test] + fn returns_none_when_prefix_absent() { + assert_eq!(reroot_path("/other/c.jpg", "/a/b", "/x/y"), None); + } +} diff --git a/crates/codex-migrate/src/verify.rs b/crates/codex-migrate/src/verify.rs new file mode 100644 index 00000000..cb2f687b --- /dev/null +++ b/crates/codex-migrate/src/verify.rs @@ -0,0 +1,45 @@ +//! Post-load verification: per-table row-count parity between source and +//! target. A mismatch means rows were dropped or duplicated and the transfer +//! must be treated as failed. + +use crate::registry::TableRows; + +/// A table whose source and target row counts disagree. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct CountMismatch { + pub table: String, + pub source: u64, + pub target: u64, +} + +/// Compare two per-table count sets (as produced by +/// [`crate::registry::count_all`]) and return every table that differs. +/// Tables present in only one set are reported with `0` for the missing side. +pub fn compare(source: &[TableRows], target: &[TableRows]) -> Vec { + let mut mismatches = Vec::new(); + for s in source { + let target_rows = target + .iter() + .find(|t| t.table == s.table) + .map(|t| t.rows) + .unwrap_or(0); + if target_rows != s.rows { + mismatches.push(CountMismatch { + table: s.table.clone(), + source: s.rows, + target: target_rows, + }); + } + } + // Tables that exist on the target but not the source. + for t in target { + if !source.iter().any(|s| s.table == t.table) && t.rows != 0 { + mismatches.push(CountMismatch { + table: t.table.clone(), + source: 0, + target: t.rows, + }); + } + } + mismatches +} diff --git a/crates/codex-migrate/tests/archive_roundtrip.rs b/crates/codex-migrate/tests/archive_roundtrip.rs new file mode 100644 index 00000000..dce30649 --- /dev/null +++ b/crates/codex-migrate/tests/archive_roundtrip.rs @@ -0,0 +1,215 @@ +//! End-to-end archive round-trip: export a source DB plus on-disk artifacts to +//! a `.tar.gz`, import into a fresh DB whose artifact dirs differ, and assert +//! the data mirrors and every stored file path is re-rooted and resolvable. + +use std::fs; +use std::path::Path; + +use chrono::Utc; +use codex_db::entities::{book_covers, books, series_covers}; +use codex_db::test_helpers::create_test_db; +use codex_migrate::archive::{ArtifactSource, ArtifactTarget, export_archive, import_archive}; +use codex_migrate::manifest::ArtifactGroup; +use codex_migrate::{registry, verify}; +use sea_orm::{ActiveModelTrait, EntityTrait, Set}; +use uuid::Uuid; + +struct Seeded { + book_id: Uuid, + book_cover_id: Uuid, + series_cover_id: Uuid, + thumb_stored: String, + book_cover_stored: String, + series_cover_stored: String, +} + +/// Write a small file at `abs`, creating parents. Returns the path as stored. +fn touch(abs: &Path) -> String { + fs::create_dir_all(abs.parent().unwrap()).unwrap(); + fs::write(abs, b"BINARY").unwrap(); + abs.to_string_lossy().into_owned() +} + +/// Seed a library → series → book, a book cover and a series cover, with their +/// image files placed under `thumb_dir` / `uploads_dir` and the DB storing the +/// absolute paths (mirroring an absolute-`files.*_dir` config). +async fn seed(db: &codex_db::Database, thumb_dir: &Path, uploads_dir: &Path) -> Seeded { + let library = db + .create_library("Comics", "/library", codex_db::ScanningStrategy::Default) + .await + .unwrap(); + let series = db.create_series(library.id, "Saga").await.unwrap(); + let conn = db.sea_orm_connection(); + + let book_id = Uuid::new_v4(); + let thumb_abs = thumb_dir.join(format!("books/{}/{book_id}.jpg", &book_id.to_string()[..2])); + let thumb_stored = touch(&thumb_abs); + + books::ActiveModel { + id: Set(book_id), + series_id: Set(series.id), + library_id: Set(library.id), + path: Set("/library/Saga/001.cbz".to_string()), + file_name: Set("001.cbz".to_string()), + file_size: Set(1234), + file_hash: Set("hash".to_string()), + partial_hash: Set("phash".to_string()), + format: Set("cbz".to_string()), + page_count: Set(20), + deleted: Set(false), + analyzed: Set(true), + modified_at: Set(Utc::now()), + created_at: Set(Utc::now()), + updated_at: Set(Utc::now()), + thumbnail_path: Set(Some(thumb_stored.clone())), + ..Default::default() + } + .insert(conn) + .await + .unwrap(); + + let book_cover_id = Uuid::new_v4(); + let bc_abs = uploads_dir.join(format!("covers/books/{book_id}/{book_cover_id}.jpg")); + let book_cover_stored = touch(&bc_abs); + book_covers::ActiveModel { + id: Set(book_cover_id), + book_id: Set(book_id), + source: Set("custom".to_string()), + path: Set(book_cover_stored.clone()), + is_selected: Set(true), + width: Set(None), + height: Set(None), + created_at: Set(Utc::now()), + updated_at: Set(Utc::now()), + } + .insert(conn) + .await + .unwrap(); + + let series_cover_id = Uuid::new_v4(); + let sc_abs = uploads_dir.join(format!("covers/series/{}/{series_cover_id}.jpg", series.id)); + let series_cover_stored = touch(&sc_abs); + series_covers::ActiveModel { + id: Set(series_cover_id), + series_id: Set(series.id), + source: Set("custom".to_string()), + path: Set(series_cover_stored.clone()), + is_selected: Set(true), + width: Set(None), + height: Set(None), + created_at: Set(Utc::now()), + updated_at: Set(Utc::now()), + } + .insert(conn) + .await + .unwrap(); + + Seeded { + book_id, + book_cover_id, + series_cover_id, + thumb_stored, + book_cover_stored, + series_cover_stored, + } +} + +#[tokio::test] +async fn archive_roundtrip_mirrors_data_and_reroots_paths() { + let scratch = tempfile::tempdir().unwrap(); + let src_thumbs = scratch.path().join("src/thumbnails"); + let src_uploads = scratch.path().join("src/uploads"); + let dst_thumbs = scratch.path().join("dst/thumbnails"); + let dst_uploads = scratch.path().join("dst/uploads"); + let archive_path = scratch.path().join("export.tar.gz"); + + // --- Source: DB + artifact files on disk. --- + let (src, _src_dir) = create_test_db().await; + let seeded = seed(&src, &src_thumbs, &src_uploads).await; + + let manifest = export_archive( + src.sea_orm_connection(), + &archive_path, + &[ + ArtifactSource { + group: ArtifactGroup::Thumbnails, + source_dir: src_thumbs.clone(), + }, + ArtifactSource { + group: ArtifactGroup::Uploads, + source_dir: src_uploads.clone(), + }, + ], + ) + .await + .expect("export should succeed"); + + assert!(archive_path.exists(), "archive file written"); + assert_eq!(manifest.source_backend, "sqlite"); + assert!(manifest.schema_version.is_some()); + assert_eq!(manifest.artifacts.len(), 2); + assert!(manifest.total_rows >= 5); + + // --- Target: fresh DB, DIFFERENT artifact dirs. --- + let (dst, _dst_dir) = create_test_db().await; + let outcome = import_archive( + dst.sea_orm_connection(), + &archive_path, + &[ + ArtifactTarget { + group: ArtifactGroup::Thumbnails, + target_dir: dst_thumbs.clone(), + }, + ArtifactTarget { + group: ArtifactGroup::Uploads, + target_dir: dst_uploads.clone(), + }, + ], + ) + .await + .expect("import should succeed"); + + // Row-count parity across every table. + let src_counts = registry::count_all(src.sea_orm_connection()).await.unwrap(); + let dst_counts = registry::count_all(dst.sea_orm_connection()).await.unwrap(); + assert!( + verify::compare(&src_counts, &dst_counts).is_empty(), + "row counts must mirror" + ); + assert_eq!(outcome.reroot.thumbnails, 1); + assert_eq!(outcome.reroot.covers, 2); + + // Stored paths were re-rooted from the source base dirs to the target's. + let dst_conn = dst.sea_orm_connection(); + let book = books::Entity::find_by_id(seeded.book_id) + .one(dst_conn) + .await + .unwrap() + .unwrap(); + let new_thumb = book.thumbnail_path.expect("thumbnail path present"); + assert!( + new_thumb.starts_with(dst_thumbs.to_string_lossy().as_ref()), + "thumbnail re-rooted: {new_thumb}" + ); + assert_ne!(new_thumb, seeded.thumb_stored, "path actually changed"); + // ...and the bundled file was unpacked to the new location and resolves. + assert!(Path::new(&new_thumb).exists(), "thumbnail file unpacked"); + + let bc = book_covers::Entity::find_by_id(seeded.book_cover_id) + .one(dst_conn) + .await + .unwrap() + .unwrap(); + assert!(bc.path.starts_with(dst_uploads.to_string_lossy().as_ref())); + assert_ne!(bc.path, seeded.book_cover_stored); + assert!(Path::new(&bc.path).exists(), "book cover unpacked"); + + let sc = series_covers::Entity::find_by_id(seeded.series_cover_id) + .one(dst_conn) + .await + .unwrap() + .unwrap(); + assert!(sc.path.starts_with(dst_uploads.to_string_lossy().as_ref())); + assert_ne!(sc.path, seeded.series_cover_stored); + assert!(Path::new(&sc.path).exists(), "series cover unpacked"); +} diff --git a/crates/codex-migrate/tests/transfer_roundtrip.rs b/crates/codex-migrate/tests/transfer_roundtrip.rs new file mode 100644 index 00000000..43931e33 --- /dev/null +++ b/crates/codex-migrate/tests/transfer_roundtrip.rs @@ -0,0 +1,202 @@ +//! Engine-level round-trip: populate a SQLite source with an FK-connected +//! dataset, transfer it into a fresh SQLite target, and assert the target is a +//! byte-identical mirror. Also guards the entity registry against schema drift. + +use codex_db::entities::{genres, series_genres}; +use codex_db::test_helpers::create_test_db; +use codex_migrate::{registry, verify}; +use sea_orm::{ActiveModelTrait, ConnectionTrait, EntityTrait, Set, Statement}; +use uuid::Uuid; + +/// Seed a small but FK-connected dataset: library → series, plus a genre and a +/// series↔genre junction row. Exercises UUID FKs, a JSON config column +/// (libraries), and a two-FK junction table. Returns the ids for later checks. +async fn seed_source(db: &codex_db::Database) -> (Uuid, Uuid, Uuid) { + let library = db + .create_library( + "Comics", + "/library/comics", + codex_db::ScanningStrategy::Default, + ) + .await + .unwrap(); + let series = db.create_series(library.id, "Saga").await.unwrap(); + + let conn = db.sea_orm_connection(); + let genre_id = Uuid::new_v4(); + genres::ActiveModel { + id: Set(genre_id), + name: Set("Action".to_string()), + normalized_name: Set("action".to_string()), + created_at: Set(chrono::Utc::now()), + } + .insert(conn) + .await + .unwrap(); + + series_genres::ActiveModel { + series_id: Set(series.id), + genre_id: Set(genre_id), + } + .insert(conn) + .await + .unwrap(); + + (library.id, series.id, genre_id) +} + +#[tokio::test] +async fn transfer_mirrors_source_into_empty_target() { + let (src, _src_dir) = create_test_db().await; + let (dst, _dst_dir) = create_test_db().await; + + let (library_id, series_id, genre_id) = seed_source(&src).await; + + let report = codex_migrate::transfer(src.sea_orm_connection(), dst.sea_orm_connection()) + .await + .expect("transfer should succeed"); + + // Every non-empty table copied at least what we seeded. + assert!( + report.total_rows >= 4, + "expected >=4 rows, got {}", + report.total_rows + ); + + // Row-count parity across every table: no drops, no duplicates. + let src_counts = registry::count_all(src.sea_orm_connection()).await.unwrap(); + let dst_counts = registry::count_all(dst.sea_orm_connection()).await.unwrap(); + let mismatches = verify::compare(&src_counts, &dst_counts); + assert!(mismatches.is_empty(), "count mismatches: {mismatches:?}"); + + // Spot-check that specific rows, UUID FKs, and the JSON column survived. + let dst_conn = dst.sea_orm_connection(); + + let library = codex_db::entities::libraries::Entity::find_by_id(library_id) + .one(dst_conn) + .await + .unwrap() + .expect("library present in target"); + assert_eq!(library.name, "Comics"); + + let series = codex_db::entities::series::Entity::find_by_id(series_id) + .one(dst_conn) + .await + .unwrap() + .expect("series present in target"); + assert_eq!(series.name, "Saga"); + assert_eq!(series.library_id, library_id, "UUID FK preserved"); + + let genre = genres::Entity::find_by_id(genre_id) + .one(dst_conn) + .await + .unwrap() + .expect("genre present in target"); + assert_eq!(genre.normalized_name, "action"); + + // Junction row transferred with both FKs intact. + let link = series_genres::Entity::find_by_id((series_id, genre_id)) + .one(dst_conn) + .await + .unwrap(); + assert!(link.is_some(), "series↔genre junction row preserved"); +} + +#[tokio::test] +async fn transfer_preserves_library_json_config_exactly() { + let (src, _src_dir) = create_test_db().await; + let (dst, _dst_dir) = create_test_db().await; + let (library_id, _, _) = seed_source(&src).await; + + codex_migrate::transfer(src.sea_orm_connection(), dst.sea_orm_connection()) + .await + .unwrap(); + + // Full-model equality confirms JSON config columns round-trip verbatim. + let src_lib = codex_db::entities::libraries::Entity::find_by_id(library_id) + .one(src.sea_orm_connection()) + .await + .unwrap() + .unwrap(); + let dst_lib = codex_db::entities::libraries::Entity::find_by_id(library_id) + .one(dst.sea_orm_connection()) + .await + .unwrap() + .unwrap(); + assert_eq!(src_lib, dst_lib); +} + +#[tokio::test] +async fn transfer_overwrites_existing_target_data() { + let (src, _src_dir) = create_test_db().await; + let (dst, _dst_dir) = create_test_db().await; + + // Source has one library; target starts with unrelated content. + seed_source(&src).await; + let (dst_only_lib, _, _) = seed_source(&dst).await; + + // First transfer replaces the target's content with the source's. + codex_migrate::transfer(src.sea_orm_connection(), dst.sea_orm_connection()) + .await + .unwrap(); + + // The target's original library is gone; the target now mirrors the source. + let stale = codex_db::entities::libraries::Entity::find_by_id(dst_only_lib) + .one(dst.sea_orm_connection()) + .await + .unwrap(); + assert!(stale.is_none(), "pre-existing target row should be cleared"); + + let src_counts = registry::count_all(src.sea_orm_connection()).await.unwrap(); + let after_first = registry::count_all(dst.sea_orm_connection()).await.unwrap(); + assert!(verify::compare(&src_counts, &after_first).is_empty()); + + // A second transfer is idempotent (truncate → reload leaves the mirror). + codex_migrate::transfer(src.sea_orm_connection(), dst.sea_orm_connection()) + .await + .unwrap(); + let after_second = registry::count_all(dst.sea_orm_connection()).await.unwrap(); + assert!(verify::compare(&src_counts, &after_second).is_empty()); +} + +/// The registry must list exactly the tables the migrations create (minus +/// SeaORM's own bookkeeping table). Fails loudly if a new table is added to +/// the schema but not to the registry — or vice versa. +#[tokio::test] +async fn registry_covers_every_migration_table() { + let (db, _dir) = create_test_db().await; + let conn = db.sea_orm_connection(); + + let rows = conn + .query_all(Statement::from_string( + sea_orm::DatabaseBackend::Sqlite, + "SELECT name FROM sqlite_master WHERE type='table' \ + AND name NOT LIKE 'sqlite_%' AND name <> 'seaql_migrations'" + .to_string(), + )) + .await + .unwrap(); + + let mut schema_tables: Vec = rows + .iter() + .map(|r| r.try_get::("", "name").unwrap()) + .collect(); + schema_tables.sort(); + + let mut registry_tables = registry::table_names(); + registry_tables.sort(); + + let missing_from_registry: Vec<_> = schema_tables + .iter() + .filter(|t| !registry_tables.contains(t)) + .collect(); + let unknown_in_registry: Vec<_> = registry_tables + .iter() + .filter(|t| !schema_tables.contains(t)) + .collect(); + + assert!( + missing_from_registry.is_empty() && unknown_in_registry.is_empty(), + "registry drift.\n tables missing from registry: {missing_from_registry:?}\n registry tables not in schema: {unknown_in_registry:?}" + ); +} diff --git a/docs/docs/deployment/backup-and-migration.md b/docs/docs/deployment/backup-and-migration.md new file mode 100644 index 00000000..221e6bc3 --- /dev/null +++ b/docs/docs/deployment/backup-and-migration.md @@ -0,0 +1,186 @@ +--- +sidebar_position: 8 +--- + +# Backup & Migration + +Codex can export its entire dataset to a portable archive and load it back into +any supported database engine. This powers three workflows: + +- **Backup / restore** — snapshot the database (and the on-disk artifacts it + references) to a single `.tar.gz`, and restore it later. +- **SQLite → PostgreSQL migration** — move an existing single-node instance to a + distributed, PostgreSQL-backed deployment with a faithful 1:1 copy of all + data (metadata, custom data, ratings, reading progress, uploaded covers, + plugin state). +- **Direct instance-to-instance copy** — stream one database's rows straight + into another. + +All three are driven by the database's own entity definitions, so +engine-specific representations (UUIDs, JSON, booleans, timestamps) are +translated correctly between SQLite and PostgreSQL — something a raw SQL dump or +generic converter cannot guarantee. + +:::info Why not just copy the SQLite file or use `pg_dump`? +The SQLite file only works on SQLite. `pg_dump` only reads PostgreSQL. Neither +crosses engines: SQLite stores UUIDs as 16-byte blobs and JSON as text, while +PostgreSQL uses native `uuid` and `jsonb`. The `export`/`import`/`copy` commands +translate these correctly. +::: + +## Commands + +### `export` + +Writes the database and its on-disk artifacts to a `.tar.gz`. + +```bash +codex export --config config/codex.yaml --output codex-backup.tar.gz +``` + +By default the archive bundles the database plus **thumbnails**, **uploaded +covers**, and **plugin data**. Flags: + +| Flag | Effect | +|------|--------| +| `--include-cache` | Also bundle the rendered PDF page cache (reproducible, can be large) | +| `--db-only` | Bundle the database only; no on-disk artifacts | +| `--no-thumbnails` | Skip generated thumbnails | +| `--no-uploads` | Skip uploaded/extracted covers | +| `--no-plugins` | Skip plugin data | + +The archive contains a `manifest.json` (format and schema version, per-table row +counts, bundled artifact groups), one `db/
.ndjson` per table, and the +bundled artifact directories. + +### `import` + +Loads an archive into the current instance. Runs migrations on the target first, +then validates and loads. + +```bash +codex import --config config/codex.yaml --input codex-backup.tar.gz +``` + +Import **refuses to run** if: + +- the archive's schema version does not match this instance's schema (import + with a Codex build whose schema matches the archive), or +- the target database already contains user data (libraries, series, books, or + users) — unless you pass `--replace`. + +```bash +# Overwrite an existing instance with the archive's contents: +codex import --input codex-backup.tar.gz --replace +``` + +On import, file paths stored in the database are **re-rooted** to this +instance's configured directories, so an archive from an instance with different +`files.*_dir` paths still resolves its images. + +### `copy` + +Streams database rows directly from one database to another, without an +intermediate file. + +```bash +# Run on the destination: pull the old SQLite database into the local (Postgres) config +codex copy --from "sqlite:///var/lib/codex/codex.db" + +# Run on the source: push into a new instance +codex copy --to "postgres://codex:secret@db:5432/codex" + +# Both sides explicit +codex copy --from "sqlite:///old/codex.db" --to "postgres://codex@db:5432/codex" +``` + +Each side resolves in this order: an explicit `--from` / `--to` URL → +`CODEX_SOURCE_DATABASE_URL` / `CODEX_TARGET_DATABASE_URL` → a `--from-config` / +`--to-config` file → the local instance config (`--config`) when that side is +omitted. At least one side must be non-local. + +:::caution `copy` moves rows only +`copy` transfers database rows, not files. On-disk artifacts (thumbnails, +covers, plugin data) are **not** moved — sync them separately (e.g. `rsync` or a +volume copy). For a self-contained move including files, use `export` + `import`. +::: + +To avoid leaking a password via the process list, prefer the env vars or +`--from-config` / `--to-config` over a `postgres://user:pass@…` URL on the +command line. + +## Migrate SQLite → PostgreSQL (Kubernetes) + +A step-by-step runbook for moving a single-node SQLite instance to a +PostgreSQL-backed, worker-separated deployment. + +1. **Quiesce the source.** Stop writes to the running instance (scale it down or + take it offline). The export reads a consistent snapshot; new writes during + the export would be lost. + +2. **Export on the source**, including artifacts (the default): + + ```bash + codex export --config config/codex.yaml --output codex-migration.tar.gz + ``` + +3. **Provision PostgreSQL** and configure the new deployment to use it. **Carry + over the encryption key** (see the caution below) into the new instance's + config. + +4. **Import on the new instance.** The target is fresh, so no `--replace` is + needed: + + ```bash + codex import --config config/codex.yaml --input codex-migration.tar.gz + ``` + + Migrations run, rows load, artifacts unpack, and file paths are re-rooted to + the new instance's directories. + +5. **Verify.** Review the import summary (per-table row counts, re-rooted path + counts). Bring up the server and worker, open the app, and confirm covers, + thumbnails, and reading progress are intact. + +:::danger Carry over the encryption key +Encrypted values (such as plugin credentials) are copied as **ciphertext** and +are never decrypted during migration. The destination instance must be +configured with the **same encryption key** as the source, or those values +cannot be decrypted after the move. +::: + +:::note PostgreSQL privileges +`import` and `copy` temporarily disable foreign-key enforcement during the bulk +load (`SET session_replication_role = replica`), which requires the target +connection to be a **superuser or the database owner**. This is normally the +case when you provision the database yourself. +::: + +## Backup & restore + +The same tooling doubles as backup: + +```bash +# Nightly backup +codex export --output "backups/codex-$(date +%F).tar.gz" + +# Restore into a fresh instance +codex import --input backups/codex-2026-07-04.tar.gz + +# Restore over an existing instance (destructive) +codex import --input backups/codex-2026-07-04.tar.gz --replace +``` + +Because export/import are engine-agnostic, a backup taken from PostgreSQL can be +restored into SQLite (e.g. to pull production down to a local file for +debugging) and vice versa. + +## What is and isn't included + +**Included:** every database table, and (by default) generated thumbnails, +uploaded/extracted covers, and plugin data. + +**Not included:** your **library files** themselves (the CBZ/EPUB/PDF on disk) — +those live on a volume the new instance is expected to mount. The reproducible +PDF page cache is excluded unless you pass `--include-cache`, and user-generated +export files are not bundled. diff --git a/docs/sidebars.ts b/docs/sidebars.ts index bdfb655d..17712dad 100644 --- a/docs/sidebars.ts +++ b/docs/sidebars.ts @@ -88,6 +88,7 @@ const sidebars: SidebarsConfig = { "deployment/database", "deployment/performance", "deployment/operations", + "deployment/backup-and-migration", ], }, "troubleshooting", diff --git a/src/commands/copy.rs b/src/commands/copy.rs new file mode 100644 index 00000000..85a3145c --- /dev/null +++ b/src/commands/copy.rs @@ -0,0 +1,213 @@ +use anyhow::{Context, Result, bail}; +use codex_cli_common::{init_tracing, load_config}; +use codex_config::DatabaseConfig; +use codex_db::Database; +use codex_migrate::{database_config_from_url, guard, manifest, transfer}; +use std::path::{Path, PathBuf}; +use tracing::{info, warn}; + +/// Copy all database rows directly from one instance to another. +/// +/// Each side resolves independently, in precedence order: explicit URL flag → +/// `CODEX_SOURCE_/TARGET_DATABASE_URL` env → `--from-config`/`--to-config` +/// file → the local instance config (`--config`) when that side is omitted. +/// At least one side must be non-local. +/// +/// `copy` moves rows only; on-disk files are not transferred. +pub async fn copy_command( + config_path: PathBuf, + from: Option, + to: Option, + from_config: Option, + to_config: Option, + replace: bool, +) -> Result<()> { + // Local config: used for tracing and as the fallback for an omitted side. + let (local_config, _created) = load_config(config_path.clone())?; + let _tracing = init_tracing(&local_config)?; + + let source_explicit = resolve_side( + from.as_deref(), + from_config.as_deref(), + "CODEX_SOURCE_DATABASE_URL", + )?; + let target_explicit = resolve_side( + to.as_deref(), + to_config.as_deref(), + "CODEX_TARGET_DATABASE_URL", + )?; + + if source_explicit.is_none() && target_explicit.is_none() { + bail!( + "copy needs at least one explicit side: pass --from/--to (or --from-config/--to-config, \ + or set CODEX_SOURCE_DATABASE_URL / CODEX_TARGET_DATABASE_URL). With neither, both sides \ + would be the local config — a copy onto itself." + ); + } + + let source_cfg = source_explicit.unwrap_or_else(|| local_config.database.clone()); + let target_cfg = target_explicit.unwrap_or_else(|| local_config.database.clone()); + + info!("Connecting to source and target databases..."); + let source = Database::new(&source_cfg) + .await + .context("Failed to connect to source database")?; + let target = Database::new(&target_cfg) + .await + .context("Failed to connect to target database")?; + + // Ensure the target has the schema, then verify source and target agree. + target + .run_migrations() + .await + .context("Failed to run migrations on the target database")?; + + let source_version = manifest::schema_version(source.sea_orm_connection()).await?; + let target_version = manifest::schema_version(target.sea_orm_connection()).await?; + if source_version != target_version { + bail!( + "schema version mismatch: source is at {:?}, target is at {:?}. \ + Both instances must be on the same Codex schema.", + source_version, + target_version + ); + } + + if !replace && guard::has_user_data(target.sea_orm_connection()).await? { + bail!( + "target database already contains data (libraries/series/books/users). \ + Refusing to overwrite. Re-run with --replace to replace it." + ); + } + + info!("Copying database rows from source to target..."); + let report = transfer(source.sea_orm_connection(), target.sea_orm_connection()) + .await + .context("Copy failed")?; + + info!("========================================"); + info!( + "✓ Copy complete: {} rows across {} tables", + report.total_rows, + report.tables.len() + ); + warn!( + "copy transfers database rows only — on-disk files (thumbnails, covers, plugin data) are \ + NOT moved. Sync those separately (rsync / volume copy), and ensure the target uses the \ + same encryption key as the source." + ); + info!("========================================"); + Ok(()) +} + +/// Resolve one side of the copy. Returns `Some(config)` when an explicit source +/// is given (flag → env → config file), or `None` to signal "use local". +fn resolve_side( + url: Option<&str>, + config_file: Option<&Path>, + env_key: &str, +) -> Result> { + if let Some(u) = url { + return Ok(Some(database_config_from_url(u)?)); + } + if let Ok(u) = std::env::var(env_key) + && !u.is_empty() + { + return Ok(Some(database_config_from_url(&u)?)); + } + if let Some(path) = config_file { + let (config, _created) = load_config(path.to_path_buf())?; + return Ok(Some(config.database)); + } + Ok(None) +} + +#[cfg(test)] +mod tests { + use super::*; + use codex_config::DatabaseType; + use codex_db::Database; + use tempfile::TempDir; + + fn write_config(dir: &std::path::Path, name: &str) -> PathBuf { + let cfg = dir.join(format!("{name}.yaml")); + let content = format!( + r#" +application: + host: "127.0.0.1" + port: 8080 +database: + db_type: sqlite + sqlite: + path: "{db}" +files: + thumbnail_dir: "{base}/{name}-thumbnails" + uploads_dir: "{base}/{name}-uploads" + plugins_dir: "{base}/{name}-plugins" +"#, + db = dir.join(format!("{name}.db")).display(), + base = dir.display(), + name = name, + ); + std::fs::write(&cfg, content).unwrap(); + cfg + } + + #[test] + fn resolve_side_prefers_explicit_url_and_defaults_to_local() { + let from_url = resolve_side(Some("sqlite:///tmp/x.db"), None, "CODEX_UNSET_XYZ") + .unwrap() + .expect("explicit url resolves to Some"); + assert_eq!(from_url.db_type, DatabaseType::SQLite); + + // No url, no env, no config file → None (meaning "use local"). + let none = resolve_side(None, None, "CODEX_UNSET_XYZ").unwrap(); + assert!(none.is_none()); + } + + #[tokio::test] + async fn copy_pulls_source_url_into_local_target() { + let dir = TempDir::new().unwrap(); + let src_cfg = write_config(dir.path(), "src"); + let tgt_cfg = write_config(dir.path(), "tgt"); + let src_db_path = dir.path().join("src.db"); + + // Seed the source. + { + let (config, _) = load_config(src_cfg.clone()).unwrap(); + let db = Database::new(&config.database).await.unwrap(); + db.run_migrations().await.unwrap(); + db.create_library("Manga", "/lib", codex_db::ScanningStrategy::Default) + .await + .unwrap(); + } + + // copy --from sqlite://src into the local (tgt) config. + copy_command( + tgt_cfg.clone(), + Some(format!("sqlite://{}", src_db_path.display())), + None, + None, + None, + false, + ) + .await + .expect("copy should succeed"); + + let (config, _) = load_config(tgt_cfg).unwrap(); + let db = Database::new(&config.database).await.unwrap(); + let libs = db.list_libraries().await.unwrap(); + assert_eq!(libs.len(), 1); + assert_eq!(libs[0].name, "Manga"); + } + + #[tokio::test] + async fn copy_requires_at_least_one_explicit_side() { + let dir = TempDir::new().unwrap(); + let cfg = write_config(dir.path(), "only"); + let err = copy_command(cfg, None, None, None, None, false) + .await + .expect_err("copy with no explicit side must error"); + assert!(err.to_string().contains("at least one explicit side")); + } +} diff --git a/src/commands/export.rs b/src/commands/export.rs new file mode 100644 index 00000000..f9502f68 --- /dev/null +++ b/src/commands/export.rs @@ -0,0 +1,83 @@ +use anyhow::{Context, Result}; +use codex_cli_common::{init_tracing, load_config}; +use codex_db::Database; +use codex_migrate::archive::{ArtifactSource, export_archive}; +use codex_migrate::manifest::ArtifactGroup; +use std::path::PathBuf; +use tracing::info; + +/// Export the current instance's database and on-disk artifacts to a portable +/// `.tar.gz` archive. The database is written as one NDJSON file per table; the +/// default artifact bundle is thumbnails + uploads + plugin data. +#[allow(clippy::too_many_arguments)] +pub async fn export_command( + config_path: PathBuf, + output: PathBuf, + include_cache: bool, + db_only: bool, + no_thumbnails: bool, + no_uploads: bool, + no_plugins: bool, +) -> Result<()> { + let (config, _created) = load_config(config_path.clone())?; + let _tracing = init_tracing(&config)?; + info!("Loading configuration from {:?}", config_path); + + let db = Database::new(&config.database) + .await + .context("Failed to connect to database")?; + + let mut artifacts = Vec::new(); + if !db_only { + if !no_thumbnails { + artifacts.push(ArtifactSource { + group: ArtifactGroup::Thumbnails, + source_dir: config.files.thumbnail_dir.clone().into(), + }); + } + if !no_uploads { + artifacts.push(ArtifactSource { + group: ArtifactGroup::Uploads, + source_dir: config.files.uploads_dir.clone().into(), + }); + } + if !no_plugins { + artifacts.push(ArtifactSource { + group: ArtifactGroup::Plugins, + source_dir: config.files.plugins_dir.clone().into(), + }); + } + if include_cache { + artifacts.push(ArtifactSource { + group: ArtifactGroup::Cache, + source_dir: config.pdf.cache_dir.clone().into(), + }); + } + } + + info!("Exporting database to {}", output.display()); + let manifest = export_archive(db.sea_orm_connection(), &output, &artifacts) + .await + .context("Export failed")?; + + info!("========================================"); + info!( + "✓ Export complete: {} rows across {} tables", + manifest.total_rows, + manifest.tables.len() + ); + if manifest.artifacts.is_empty() { + info!(" (database only — no artifacts bundled)"); + } else { + for entry in &manifest.artifacts { + info!( + " bundled {} from {}", + entry.group.archive_dir(), + entry.source_base_dir + ); + } + } + info!("Archive written to {}", output.display()); + info!("========================================"); + Ok(()) +} diff --git a/src/commands/import.rs b/src/commands/import.rs new file mode 100644 index 00000000..d8b91300 --- /dev/null +++ b/src/commands/import.rs @@ -0,0 +1,206 @@ +use anyhow::{Context, Result, bail}; +use codex_cli_common::{init_tracing, load_config}; +use codex_config::Config; +use codex_db::Database; +use codex_migrate::archive::{ArtifactTarget, import_archive, read_manifest}; +use codex_migrate::manifest::ArtifactGroup; +use codex_migrate::{guard, manifest}; +use std::path::PathBuf; +use tracing::{info, warn}; + +/// Import a `.tar.gz` produced by `export` into the current instance. Runs +/// migrations on the target, verifies the archive matches this schema, refuses +/// a target that already holds user data (unless `--replace`), then loads the +/// data, unpacks artifacts, and re-roots stored file paths. +pub async fn import_command(config_path: PathBuf, input: PathBuf, replace: bool) -> Result<()> { + let (config, _created) = load_config(config_path.clone())?; + let _tracing = init_tracing(&config)?; + info!("Loading configuration from {:?}", config_path); + + if !input.exists() { + bail!("archive not found: {}", input.display()); + } + + let db = Database::new(&config.database) + .await + .context("Failed to connect to database")?; + + // The target must have the schema before we can load into it. + db.run_migrations() + .await + .context("Failed to run migrations on the target database")?; + let conn = db.sea_orm_connection(); + + // Pre-flight guards, before any destructive change. + let archive_manifest = read_manifest(&input).context("Failed to read archive manifest")?; + let target_version = manifest::schema_version(conn).await?; + if archive_manifest.schema_version != target_version { + bail!( + "schema version mismatch: archive was exported at {:?}, this instance is at {:?}. \ + Import with a Codex build whose schema matches the archive.", + archive_manifest.schema_version, + target_version + ); + } + + if !replace && guard::has_user_data(conn).await? { + bail!( + "target database already contains data (libraries/series/books/users). \ + Refusing to overwrite. Re-run with --replace to replace it with the archive." + ); + } + + info!("Importing {} ...", input.display()); + let outcome = import_archive(conn, &input, &artifact_targets(&config)) + .await + .context("Import failed")?; + + info!("========================================"); + info!( + "✓ Import complete: {} rows across {} tables", + outcome.report.total_rows, + outcome.report.tables.len() + ); + info!( + " re-rooted {} thumbnail path(s), {} cover path(s)", + outcome.reroot.thumbnails, outcome.reroot.covers + ); + warn!( + "Reminder: encrypted values (e.g. plugin credentials) were copied as ciphertext. \ + This instance must be configured with the SAME encryption key as the source, or they \ + cannot be decrypted." + ); + info!("========================================"); + Ok(()) +} + +/// Where each artifact group should be unpacked on this instance. `import` +/// only writes the groups actually present in the archive. +fn artifact_targets(config: &Config) -> Vec { + vec![ + ArtifactTarget { + group: ArtifactGroup::Thumbnails, + target_dir: config.files.thumbnail_dir.clone().into(), + }, + ArtifactTarget { + group: ArtifactGroup::Uploads, + target_dir: config.files.uploads_dir.clone().into(), + }, + ArtifactTarget { + group: ArtifactGroup::Plugins, + target_dir: config.files.plugins_dir.clone().into(), + }, + ArtifactTarget { + group: ArtifactGroup::Cache, + target_dir: config.pdf.cache_dir.clone().into(), + }, + ] +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::commands::export::export_command; + use codex_db::Database; + use std::path::Path; + use tempfile::TempDir; + + /// Write a minimal SQLite config with artifact dirs under `dir`. + fn write_config(dir: &Path, name: &str) -> PathBuf { + let cfg = dir.join(format!("{name}.yaml")); + let content = format!( + r#" +application: + host: "127.0.0.1" + port: 8080 +database: + db_type: sqlite + sqlite: + path: "{db}" +files: + thumbnail_dir: "{base}/{name}-thumbnails" + uploads_dir: "{base}/{name}-uploads" + plugins_dir: "{base}/{name}-plugins" +"#, + db = dir.join(format!("{name}.db")).display(), + base = dir.display(), + name = name, + ); + std::fs::write(&cfg, content).unwrap(); + cfg + } + + async fn seed_library(config_path: &Path, name: &str) { + let (config, _) = load_config(config_path.to_path_buf()).unwrap(); + let db = Database::new(&config.database).await.unwrap(); + db.run_migrations().await.unwrap(); + db.create_library(name, "/lib", codex_db::ScanningStrategy::Default) + .await + .unwrap(); + } + + async fn library_names(config_path: &Path) -> Vec { + let (config, _) = load_config(config_path.to_path_buf()).unwrap(); + let db = Database::new(&config.database).await.unwrap(); + db.list_libraries() + .await + .unwrap() + .into_iter() + .map(|l| l.name) + .collect() + } + + #[tokio::test] + async fn export_then_import_roundtrips_and_guards_nonfresh_target() { + let dir = TempDir::new().unwrap(); + let src_cfg = write_config(dir.path(), "src"); + let tgt_cfg = write_config(dir.path(), "tgt"); + let archive = dir.path().join("export.tar.gz"); + + seed_library(&src_cfg, "Comics").await; + + export_command( + src_cfg.clone(), + archive.clone(), + false, + false, + false, + false, + false, + ) + .await + .expect("export should succeed"); + assert!(archive.exists(), "archive written"); + + // Import into a fresh target. + import_command(tgt_cfg.clone(), archive.clone(), false) + .await + .expect("import into fresh target should succeed"); + assert_eq!(library_names(&tgt_cfg).await, vec!["Comics".to_string()]); + + // A second import without --replace is refused (target now has data). + let err = import_command(tgt_cfg.clone(), archive.clone(), false) + .await + .expect_err("import into non-fresh target should be refused"); + assert!( + err.to_string().contains("already contains data"), + "unexpected error: {err}" + ); + + // With --replace it succeeds and still mirrors the source. + import_command(tgt_cfg.clone(), archive.clone(), true) + .await + .expect("import --replace should succeed"); + assert_eq!(library_names(&tgt_cfg).await, vec!["Comics".to_string()]); + } + + #[tokio::test] + async fn import_rejects_missing_archive() { + let dir = TempDir::new().unwrap(); + let cfg = write_config(dir.path(), "tgt"); + let err = import_command(cfg, dir.path().join("nope.tar.gz"), false) + .await + .expect_err("missing archive should error"); + assert!(err.to_string().contains("archive not found")); + } +} diff --git a/src/commands/mod.rs b/src/commands/mod.rs index 33f9ba9c..8debb085 100644 --- a/src/commands/mod.rs +++ b/src/commands/mod.rs @@ -1,3 +1,6 @@ +pub mod copy; +pub mod export; +pub mod import; pub mod migrate; pub mod openapi; pub mod scan; @@ -7,6 +10,9 @@ pub mod tasks; pub mod wait_for_migrations; pub mod worker; +pub use copy::copy_command; +pub use export::export_command; +pub use import::import_command; pub use migrate::migrate_command; pub use openapi::{OpenApiFormat, openapi_command}; pub use scan::scan_command; diff --git a/src/lib.rs b/src/lib.rs index 3c46160d..43f5eaea 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -6,6 +6,7 @@ pub use codex_api::web; pub use codex_config as config; pub use codex_db as db; pub use codex_events as events; +pub use codex_migrate as migrate; pub use codex_models as models; pub use codex_parsers as parsers; pub use codex_scanner as scanner; diff --git a/src/main.rs b/src/main.rs index 3e011479..53deb0bd 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,8 +2,9 @@ mod commands; use clap::{Parser, Subcommand}; use commands::{ - OpenApiFormat, TasksSubcommand, migrate_command, openapi_command, scan_command, seed_command, - serve_command, tasks_command, wait_for_migrations_command, worker_command, + OpenApiFormat, TasksSubcommand, copy_command, export_command, import_command, migrate_command, + openapi_command, scan_command, seed_command, serve_command, tasks_command, + wait_for_migrations_command, worker_command, }; use std::path::PathBuf; @@ -105,6 +106,79 @@ enum Commands { #[command(subcommand)] command: TasksSubcommand, }, + + /// Export the database and on-disk artifacts to a portable .tar.gz archive + Export { + /// Path to configuration file + #[arg(short, long, default_value = DEFAULT_CONFIG_PATH)] + config: PathBuf, + + /// Output archive path (.tar.gz) + #[arg(short, long)] + output: PathBuf, + + /// Also bundle the (reproducible) rendered PDF page cache + #[arg(long)] + include_cache: bool, + + /// Export the database only; bundle no on-disk artifacts + #[arg(long)] + db_only: bool, + + /// Skip bundling generated thumbnails + #[arg(long)] + no_thumbnails: bool, + + /// Skip bundling uploaded/extracted covers + #[arg(long)] + no_uploads: bool, + + /// Skip bundling plugin data + #[arg(long)] + no_plugins: bool, + }, + + /// Import a .tar.gz archive produced by `export` into this instance + Import { + /// Path to configuration file + #[arg(short, long, default_value = DEFAULT_CONFIG_PATH)] + config: PathBuf, + + /// Input archive path (.tar.gz) + #[arg(short, long)] + input: PathBuf, + + /// Replace existing data in the target instead of refusing a non-empty target + #[arg(long)] + replace: bool, + }, + + /// Copy database rows directly from one instance to another + Copy { + /// Local instance config (tracing + fallback for an omitted side) + #[arg(short, long, default_value = DEFAULT_CONFIG_PATH)] + config: PathBuf, + + /// Source database URL (sqlite://… or postgres://…) + #[arg(long)] + from: Option, + + /// Target database URL (sqlite://… or postgres://…) + #[arg(long)] + to: Option, + + /// Read the source database connection from this config file + #[arg(long)] + from_config: Option, + + /// Read the target database connection from this config file + #[arg(long)] + to_config: Option, + + /// Replace existing data in the target instead of refusing a non-empty target + #[arg(long)] + replace: bool, + }, } #[tokio::main] @@ -148,6 +222,43 @@ async fn main() -> anyhow::Result<()> { Commands::Tasks { config, command } => { tasks_command(config, command).await?; } + Commands::Export { + config, + output, + include_cache, + db_only, + no_thumbnails, + no_uploads, + no_plugins, + } => { + export_command( + config, + output, + include_cache, + db_only, + no_thumbnails, + no_uploads, + no_plugins, + ) + .await?; + } + Commands::Import { + config, + input, + replace, + } => { + import_command(config, input, replace).await?; + } + Commands::Copy { + config, + from, + to, + from_config, + to_config, + replace, + } => { + copy_command(config, from, to, from_config, to_config, replace).await?; + } } Ok(()) diff --git a/tests/it.rs b/tests/it.rs index 9d72cd22..60638432 100644 --- a/tests/it.rs +++ b/tests/it.rs @@ -8,6 +8,7 @@ mod api; mod db; mod event_bridge; +mod migrate; mod parsers; mod scanner; mod scheduler; diff --git a/tests/migrate/mod.rs b/tests/migrate/mod.rs new file mode 100644 index 00000000..a8e6af94 --- /dev/null +++ b/tests/migrate/mod.rs @@ -0,0 +1,219 @@ +//! Cross-engine migration tests. +//! +//! The headline case: a faithful 1:1 transfer from SQLite to PostgreSQL, which +//! exercises the representation differences SeaORM papers over — UUID blobs vs +//! native `uuid`, text JSON vs JSONB, 0/1 vs `bool`, and timestamps. Gated +//! like the other PostgreSQL tests: `#[ignore]`, and skips gracefully when no +//! test database is reachable. + +#[path = "../common/mod.rs"] +mod common; + +use chrono::Utc; +use codex::db::Database; +use codex::db::entities::{api_keys, books, libraries, read_progress, user_series_ratings, users}; +use codex::migrate::{registry, transfer, verify}; +use codex::models::ScanningStrategy; +use common::{setup_test_db_postgres, setup_test_db_wrapper}; +use sea_orm::{ActiveModelTrait, EntityTrait, Set}; +use serde_json::json; +use uuid::Uuid; + +struct Ids { + library: Uuid, + book: Uuid, + user: Uuid, + progress: Uuid, + rating: Uuid, + api_key: Uuid, +} + +/// Seed a connected fixture exercising JSON, bool, floats, ints, UUID FKs, and +/// timestamps: library → series → book, a user (JSON permissions), reading +/// progress, a series rating, and an API key (JSON permissions). +async fn seed_source(db: &Database) -> Ids { + let library = db + .create_library("Comics", "/lib", ScanningStrategy::Default) + .await + .unwrap(); + let series = db.create_series(library.id, "Saga").await.unwrap(); + let conn = db.sea_orm_connection(); + + let book = Uuid::new_v4(); + books::ActiveModel { + id: Set(book), + series_id: Set(series.id), + library_id: Set(library.id), + path: Set("/lib/Saga/1.cbz".to_string()), + file_name: Set("1.cbz".to_string()), + file_size: Set(999), + file_hash: Set("h".to_string()), + partial_hash: Set("p".to_string()), + format: Set("cbz".to_string()), + page_count: Set(10), + deleted: Set(false), + analyzed: Set(true), + modified_at: Set(Utc::now()), + created_at: Set(Utc::now()), + updated_at: Set(Utc::now()), + ..Default::default() + } + .insert(conn) + .await + .unwrap(); + + let user = Uuid::new_v4(); + users::ActiveModel { + id: Set(user), + username: Set("reader1".to_string()), + email: Set("reader1@example.com".to_string()), + password_hash: Set("hash".to_string()), + role: Set("reader".to_string()), + is_active: Set(true), + email_verified: Set(true), + permissions: Set(json!(["books:read"])), + created_at: Set(Utc::now()), + updated_at: Set(Utc::now()), + last_login_at: Set(None), + } + .insert(conn) + .await + .unwrap(); + + let progress = Uuid::new_v4(); + read_progress::ActiveModel { + id: Set(progress), + user_id: Set(user), + book_id: Set(book), + current_page: Set(42), + progress_percentage: Set(Some(0.5)), + completed: Set(false), + started_at: Set(Utc::now()), + updated_at: Set(Utc::now()), + completed_at: Set(None), + r2_progression: Set(None), + } + .insert(conn) + .await + .unwrap(); + + let rating = Uuid::new_v4(); + user_series_ratings::ActiveModel { + id: Set(rating), + user_id: Set(user), + series_id: Set(series.id), + rating: Set(85), + notes: Set(Some("great".to_string())), + created_at: Set(Utc::now()), + updated_at: Set(Utc::now()), + } + .insert(conn) + .await + .unwrap(); + + let api_key = Uuid::new_v4(); + api_keys::ActiveModel { + id: Set(api_key), + user_id: Set(user), + name: Set("cli".to_string()), + key_hash: Set("kh".to_string()), + key_prefix: Set("cdx_".to_string()), + permissions: Set(json!(["books:read"])), + is_active: Set(true), + expires_at: Set(None), + last_used_at: Set(None), + created_at: Set(Utc::now()), + updated_at: Set(Utc::now()), + } + .insert(conn) + .await + .unwrap(); + + Ids { + library: library.id, + book, + user, + progress, + rating, + api_key, + } +} + +#[tokio::test] +#[ignore] // requires a PostgreSQL test database (see `make test-up`) +async fn sqlite_to_postgres_roundtrip_mirrors_all_data() { + let (src, _src_dir) = setup_test_db_wrapper().await; + let Some(pg) = setup_test_db_postgres().await else { + return; // no PostgreSQL available — skip + }; + let ids = seed_source(&src).await; + + transfer(src.sea_orm_connection(), &pg) + .await + .expect("SQLite -> PostgreSQL transfer should succeed"); + + // Row-count parity across every table. + let src_counts = registry::count_all(src.sea_orm_connection()).await.unwrap(); + let pg_counts = registry::count_all(&pg).await.unwrap(); + let mismatches = verify::compare(&src_counts, &pg_counts); + assert!(mismatches.is_empty(), "count mismatches: {mismatches:?}"); + + // JSON columns (text JSON on SQLite -> JSONB on Postgres) survive verbatim. + let user = users::Entity::find_by_id(ids.user) + .one(&pg) + .await + .unwrap() + .expect("user present"); + assert_eq!(user.permissions, json!(["books:read"])); + assert!(user.is_active); + + let key = api_keys::Entity::find_by_id(ids.api_key) + .one(&pg) + .await + .unwrap() + .expect("api key present"); + assert_eq!(key.permissions, json!(["books:read"])); + + // Reading progress: ints, a float, and a bool. + let progress = read_progress::Entity::find_by_id(ids.progress) + .one(&pg) + .await + .unwrap() + .expect("read progress present"); + assert_eq!(progress.current_page, 42); + assert_eq!(progress.progress_percentage, Some(0.5)); + assert!(!progress.completed); + + // Series rating value and note. + let rating = user_series_ratings::Entity::find_by_id(ids.rating) + .one(&pg) + .await + .unwrap() + .expect("rating present"); + assert_eq!(rating.rating, 85); + assert_eq!(rating.notes.as_deref(), Some("great")); + + // Book bool/int columns and UUID FKs resolve. + let book = books::Entity::find_by_id(ids.book) + .one(&pg) + .await + .unwrap() + .expect("book present"); + assert_eq!(book.library_id, ids.library); + assert_eq!(book.page_count, 10); + assert!(book.analyzed && !book.deleted); + + // Library JSON config columns match across engines. + let src_lib = libraries::Entity::find_by_id(ids.library) + .one(src.sea_orm_connection()) + .await + .unwrap() + .unwrap(); + let pg_lib = libraries::Entity::find_by_id(ids.library) + .one(&pg) + .await + .unwrap() + .unwrap(); + assert_eq!(src_lib.name, pg_lib.name); + assert_eq!(src_lib.series_config, pg_lib.series_config); +}