diff --git a/crates/codex-migrate/src/engine.rs b/crates/codex-migrate/src/engine.rs index 3b495e95..6b0b3d92 100644 --- a/crates/codex-migrate/src/engine.rs +++ b/crates/codex-migrate/src/engine.rs @@ -9,8 +9,13 @@ use std::io::{BufRead, Write}; use anyhow::Result; -use futures::TryStreamExt; -use sea_orm::{ConnectionTrait, EntityTrait, IntoActiveModel, PaginatorTrait, StreamTrait}; +use futures::stream::BoxStream; +use futures::{StreamExt, TryStreamExt}; +use sea_orm::sea_query::Iden; +use sea_orm::{ + ColumnTrait, ColumnType, ConnectionTrait, DatabaseBackend, DbErr, EntityTrait, IntoActiveModel, + Iterable, PaginatorTrait, Statement, StreamTrait, +}; use serde::Serialize; use serde::de::DeserializeOwned; @@ -18,6 +23,59 @@ use serde::de::DeserializeOwned; /// under parameter limits for the widest tables. pub const DEFAULT_BATCH_SIZE: usize = 1000; +/// Open a `Model` stream over `E`'s rows, normalizing UUID storage on SQLite. +/// +/// SeaORM (via sqlx) reads a SQLite `Uuid` column strictly as a 16-byte blob. +/// Databases written by older toolchains may instead store UUIDs as 36-char +/// hyphenated text, which then fails to decode. For a SQLite source we +/// therefore read through a query that coerces each UUID column back to a blob +/// (text → `unhex(replace(col,'-',''))`, blobs pass through untouched), so both +/// storage formats decode identically. Other backends use the normal path. +async fn open_source_stream<'a, E, C>( + conn: &'a C, +) -> Result>> +where + E: EntityTrait, + C: ConnectionTrait + StreamTrait, +{ + if conn.get_database_backend() == DatabaseBackend::Sqlite { + let stmt = sqlite_uuid_normalizing_select::(); + Ok(E::find().from_raw_sql(stmt).stream(conn).await?.boxed()) + } else { + Ok(E::find().stream(conn).await?.boxed()) + } +} + +/// Build `SELECT FROM ` for SQLite where every UUID column is +/// coerced to a 16-byte blob regardless of whether it was stored as a blob or +/// as hyphenated text. Requires SQLite ≥ 3.41 for `unhex` (bundled with sqlx). +fn sqlite_uuid_normalizing_select() -> Statement { + let cols: Vec = ::iter() + .map(|col| { + let name = iden_string(&col); + if matches!(col.def().get_column_type(), ColumnType::Uuid) { + format!( + "CASE WHEN typeof(\"{name}\") = 'text' \ + THEN unhex(replace(\"{name}\", '-', '')) ELSE \"{name}\" END AS \"{name}\"" + ) + } else { + format!("\"{name}\"") + } + }) + .collect(); + + let table = iden_string(&E::default()); + let sql = format!("SELECT {} FROM \"{table}\"", cols.join(", ")); + Statement::from_string(DatabaseBackend::Sqlite, sql) +} + +/// The unquoted identifier string for a column or table. +fn iden_string(iden: &I) -> String { + let mut buf = String::new(); + iden.unquoted(&mut buf); + buf +} + /// 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 @@ -27,7 +85,7 @@ where C: ConnectionTrait + StreamTrait, W: Write, { - let mut stream = E::find().stream(conn).await?; + let mut stream = open_source_stream::(conn).await?; let mut count = 0u64; while let Some(model) = stream.try_next().await? { serde_json::to_writer(&mut *out, &model)?; @@ -75,7 +133,7 @@ where S: ConnectionTrait + StreamTrait, D: ConnectionTrait, { - let mut stream = E::find().stream(src).await?; + let mut stream = open_source_stream::(src).await?; let mut batch: Vec = Vec::with_capacity(batch_size); let mut count = 0u64; while let Some(model) = stream.try_next().await? { diff --git a/crates/codex-migrate/tests/transfer_roundtrip.rs b/crates/codex-migrate/tests/transfer_roundtrip.rs index 43931e33..d479308f 100644 --- a/crates/codex-migrate/tests/transfer_roundtrip.rs +++ b/crates/codex-migrate/tests/transfer_roundtrip.rs @@ -200,3 +200,47 @@ async fn registry_covers_every_migration_table() { "registry drift.\n tables missing from registry: {missing_from_registry:?}\n registry tables not in schema: {unknown_in_registry:?}" ); } + +#[tokio::test] +async fn transfer_reads_text_stored_uuids_from_sqlite() { + // Some databases written by older toolchains store UUIDs as 36-char + // hyphenated text rather than 16-byte blobs. The reader must handle both. + let (src, _s) = create_test_db().await; + let (dst, _d) = create_test_db().await; + let sconn = src.sea_orm_connection(); + + let id = "550e8400-e29b-41d4-a716-446655440000"; + sconn + .execute_unprepared(&format!( + "INSERT INTO genres (id, name, normalized_name, created_at) \ + VALUES ('{id}','Action','action','2020-01-01 00:00:00+00:00')" + )) + .await + .unwrap(); + + // Sanity: the id really is stored as text, not a blob. + let row = sconn + .query_one(Statement::from_string( + sea_orm::DatabaseBackend::Sqlite, + "SELECT typeof(id) AS t FROM genres".to_string(), + )) + .await + .unwrap() + .unwrap(); + assert_eq!(row.try_get::("", "t").unwrap(), "text"); + + codex_migrate::transfer(sconn, dst.sea_orm_connection()) + .await + .expect("transfer must read text-stored UUIDs"); + + let uuid = Uuid::parse_str(id).unwrap(); + let genre = genres::Entity::find_by_id(uuid) + .one(dst.sea_orm_connection()) + .await + .unwrap(); + assert!( + genre.is_some(), + "genre with a text-stored UUID should transfer" + ); + assert_eq!(genre.unwrap().name, "Action"); +} diff --git a/tests/migrate/mod.rs b/tests/migrate/mod.rs index a8e6af94..16225cf5 100644 --- a/tests/migrate/mod.rs +++ b/tests/migrate/mod.rs @@ -217,3 +217,160 @@ async fn sqlite_to_postgres_roundtrip_mirrors_all_data() { assert_eq!(src_lib.name, pg_lib.name); assert_eq!(src_lib.series_config, pg_lib.series_config); } + +// --------------------------------------------------------------------------- +// Export/import across every engine pair. +// --------------------------------------------------------------------------- + +use codex::db::entities::genres; +use codex::migrate::archive::{export_archive, import_archive}; +use codex::migrate::database_config_from_url; +use common::setup_test_db; +use sea_orm::{ConnectionTrait, DatabaseConnection}; + +/// Seed an engine-neutral fixture on any connection: one genre (UUID + text + +/// timestamp) and one user (UUID + JSON permissions + bool). No FKs, so it +/// inserts on SQLite or Postgres identically. +async fn seed_min(conn: &DatabaseConnection) { + let gid = Uuid::new_v4(); + genres::ActiveModel { + id: Set(gid), + name: Set("Action".to_string()), + normalized_name: Set("action".to_string()), + created_at: Set(Utc::now()), + } + .insert(conn) + .await + .unwrap(); + + let uid = Uuid::new_v4(); + users::ActiveModel { + id: Set(uid), + username: Set(format!("u-{uid}")), + email: Set(format!("{uid}@example.com")), + password_hash: Set("h".to_string()), + role: Set("reader".to_string()), + is_active: Set(true), + email_verified: Set(false), + permissions: Set(json!(["books:read"])), + created_at: Set(Utc::now()), + updated_at: Set(Utc::now()), + last_login_at: Set(None), + } + .insert(conn) + .await + .unwrap(); +} + +/// Create a fresh, migrated PostgreSQL database named `name` on the test +/// server. Returns `None` if PostgreSQL is unreachable. +async fn fresh_pg(name: &str) -> Option { + let base = std::env::var("POSTGRES_TEST_URL") + .unwrap_or_else(|_| "postgres://codex:codex@localhost:54321/codex_test".to_string()); + + // Connect to the default test DB to issue CREATE DATABASE. + let admin = Database::new(&database_config_from_url(&base).ok()?) + .await + .ok()?; + let ac = admin.sea_orm_connection(); + ac.execute_unprepared(&format!("DROP DATABASE IF EXISTS {name} WITH (FORCE)")) + .await + .ok()?; + ac.execute_unprepared(&format!("CREATE DATABASE {name}")) + .await + .ok()?; + + let mut cfg = database_config_from_url(&base).ok()?; + cfg.postgres.as_mut().unwrap().database_name = name.to_string(); + let db = Database::new(&cfg).await.ok()?; + db.run_migrations().await.ok()?; + Some(db.sea_orm_connection().clone()) +} + +async fn drop_pg(name: &str) { + let base = std::env::var("POSTGRES_TEST_URL") + .unwrap_or_else(|_| "postgres://codex:codex@localhost:54321/codex_test".to_string()); + if let Ok(cfg) = database_config_from_url(&base) + && let Ok(admin) = Database::new(&cfg).await + { + let _ = admin + .sea_orm_connection() + .execute_unprepared(&format!("DROP DATABASE IF EXISTS {name} WITH (FORCE)")) + .await; + } +} + +/// Export `src` to an archive and import it into `tgt`, then assert the target +/// mirrors the source (row-count parity) and the JSON permissions survived. +async fn export_import_pair(src: &DatabaseConnection, tgt: &DatabaseConnection, label: &str) { + let dir = tempfile::tempdir().unwrap(); + let archive = dir.path().join("export.tar.gz"); + + export_archive(src, &archive, &[]) + .await + .unwrap_or_else(|e| panic!("{label}: export failed: {e:#}")); + import_archive(tgt, &archive, &[]) + .await + .unwrap_or_else(|e| panic!("{label}: import failed: {e:#}")); + + let src_counts = registry::count_all(src).await.unwrap(); + let tgt_counts = registry::count_all(tgt).await.unwrap(); + let mismatches = verify::compare(&src_counts, &tgt_counts); + assert!( + mismatches.is_empty(), + "{label}: count mismatch: {mismatches:?}" + ); + + // JSON permissions (text JSON <-> JSONB) survived across the pair. + let user = users::Entity::find() + .one(tgt) + .await + .unwrap() + .unwrap_or_else(|| panic!("{label}: seeded user missing after import")); + assert_eq!( + user.permissions, + json!(["books:read"]), + "{label}: permissions" + ); +} + +#[tokio::test] +#[ignore] // Postgres pairs require a test database (see `make test-up`) +async fn export_import_across_all_engine_pairs() { + // --- SQLite -> SQLite (always runs). --- + { + let (src, _sd) = setup_test_db().await; + let (tgt, _td) = setup_test_db().await; + seed_min(&src).await; + export_import_pair(&src, &tgt, "sqlite->sqlite").await; + } + + // --- Pairs involving Postgres (skip when unavailable). --- + let Some(pg_a) = fresh_pg("codex_test_pair_a").await else { + eprintln!("PostgreSQL unavailable; ran sqlite->sqlite only"); + return; + }; + let pg_b = fresh_pg("codex_test_pair_b") + .await + .expect("second Postgres database"); + seed_min(&pg_a).await; + + // SQLite -> Postgres + { + let (src, _sd) = setup_test_db().await; + seed_min(&src).await; + export_import_pair(&src, &pg_b, "sqlite->postgres").await; + } + // Postgres -> SQLite + { + let (tgt, _td) = setup_test_db().await; + export_import_pair(&pg_a, &tgt, "postgres->sqlite").await; + } + // Postgres -> Postgres + export_import_pair(&pg_a, &pg_b, "postgres->postgres").await; + + drop(pg_a); + drop(pg_b); + drop_pg("codex_test_pair_a").await; + drop_pg("codex_test_pair_b").await; +}