fix(migrate): read SQLite UUIDs stored as text, not just blobs

crates/codex-migrate/src/engine.rs

@@ -9,15 +9,73 @@
 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;
 
 /// 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;
 
+/// 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<BoxStream<'a, std::result::Result<E::Model, DbErr>>>
+where
+    E: EntityTrait,
+    C: ConnectionTrait + StreamTrait,
+{
+    if conn.get_database_backend() == DatabaseBackend::Sqlite {
+        let stmt = sqlite_uuid_normalizing_select::<E>();
+        Ok(E::find().from_raw_sql(stmt).stream(conn).await?.boxed())
+    } else {
+        Ok(E::find().stream(conn).await?.boxed())
+    }
+}
+
+/// Build `SELECT <cols> FROM <table>` 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<E: EntityTrait>() -> Statement {
+    let cols: Vec<String> = <E::Column as Iterable>::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<I: Iden>(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<E, C, W>(conn: &C, out: &mut W) -> Result<u64>
@@ -27,7 +85,7 @@ where
     C: ConnectionTrait + StreamTrait,
     W: Write,
 {
-    let mut stream = E::find().stream(conn).await?;
+    let mut stream = open_source_stream::<E, C>(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::<E, S>(src).await?;
     let mut batch: Vec<E::ActiveModel> = Vec::with_capacity(batch_size);
     let mut count = 0u64;
     while let Some(model) = stream.try_next().await? {

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::<String>("", "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");
+}

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<DatabaseConnection> {
+    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;
+}