Postgres, step 4: admin copy-db moves a SQLite hub's records into Postgres 4e2cad00a484

//! Copy a SQLite hub's records into PostgreSQL (#211), for moving a hub:

//! `oxygenhub admin copy-db --from <old data>/hub.db` on the new hub, built

//! with the `postgres` feature and migrated, before it first serves.

//!

//! Everything is copied in one transaction, ids kept, so a failed copy

//! leaves the database as it was. The tables go in the order the schema

//! made them, which is after every table they point at; a column pointing

//! at its own table (a fork's parent) is filled in once all its rows are

//! there. Then each id sequence is moved past the copied ids, the search

//! index is copied, and every table's count is checked against SQLite's.

use super::Pool;

use anyhow::{bail, Context, Result};

use sqlx::sqlite::{SqliteConnectOptions, SqliteRow};

use sqlx::{Connection, Row};

use std::path::Path;

/// Rows sent in one INSERT: well under Postgres's 65535 parameters for the

/// widest table.

const BATCH: usize = 500;

/// One column of a table here, and how to carry its values across.

struct Col {

    name: String,

    kind: Kind,

    /// Points at its own table: copied after the rows.

    self_ref: bool,

}

#[derive(Clone, Copy, PartialEq)]

enum Kind {

    Int,

    Bool,

    Real,

    Blob,

    Text,

}

enum Value {

    Int(Option<i64>),

    Bool(Option<bool>),

    Real(Option<f64>),

    Blob(Option<Vec<u8>>),

    Text(Option<String>),

}

/// What was copied: `(table, rows)`, in copy order.

pub async fn from_sqlite(pool: &Pool, path: &Path) -> Result<Vec<(String, i64)>> {

    let options = SqliteConnectOptions::new().filename(path).read_only(true);

    let mut lite = sqlx::SqliteConnection::connect_with(&options).await.with_context(|| format!("open {}", path.display()))?;

    up_to_date(&mut lite).await?;

    let tables: Vec<String> = sqlx::query_scalar(

        "SELECT CAST(relname AS TEXT) FROM pg_class WHERE relkind = 'r' AND relnamespace = CAST(current_schema() AS regnamespace) \

         AND relname NOT IN ('_sqlx_migrations', 'search') ORDER BY oid",

    )

    .fetch_all(pool)

    .await?;

    for t in &tables {

        let any: bool = sqlx::query_scalar(&format!("SELECT EXISTS (SELECT 1 FROM \"{t}\")")).fetch_one(pool).await?;

        if any {

            bail!("{t} already has rows here: copy into a new, empty database");

        }

    }

    let mut tx = pool.begin().await?;

    let mut copied = Vec::new();

    for t in &tables {

        let cols = columns(&mut tx, t).await?;

        let n = copy_table(&mut lite, &mut tx, t, &cols).await.with_context(|| format!("copy {t}"))?;

        copied.push((t.clone(), n));

    }

    // Ids carry on after the copied ones.

    let numbered: Vec<String> = sqlx::query_scalar(

        "SELECT CAST(t.relname AS TEXT) FROM pg_attribute a JOIN pg_class t ON t.oid = a.attrelid \

         WHERE t.relnamespace = CAST(current_schema() AS regnamespace) AND a.attname = 'id' AND a.attidentity <> ''",

    )

    .fetch_all(&mut *tx)

    .await?;

    for t in &numbered {

        sqlx::query(&format!("SELECT setval(pg_get_serial_sequence($1, 'id'), COALESCE((SELECT MAX(id) FROM \"{t}\"), 0) + 1, false)"))

            .bind(t)

            .execute(&mut *tx)

            .await?;

    }

    let search = copy_search(&mut lite, &mut tx).await.context("copy the search index")?;

    copied.push(("search".into(), search));

    for (t, n) in &copied {

        let there: i64 = sqlx::query_scalar(&format!("SELECT COUNT(*) FROM \"{t}\"")).fetch_one(&mut *tx).await?;

        if there != *n {

            bail!("{t}: {n} rows read but {there} here");

        }

    }

    tx.commit().await?;

    lite.close().await?;

    Ok(copied)

}

/// The SQLite hub must have every migration this hub knows, and no more:

/// the two schemas match only then.

async fn up_to_date(lite: &mut sqlx::SqliteConnection) -> Result<()> {

    let latest = sqlx::migrate!("./migrations").iter().map(|m| m.version).max().unwrap_or(0);

    let have: Option<i64> =

        sqlx::query_scalar("SELECT MAX(version) FROM _sqlx_migrations WHERE success = TRUE").fetch_one(&mut *lite).await?;

    match have {

        Some(v) if v == latest => Ok(()),

        Some(v) => bail!("the SQLite hub is at migration {v}, this one at {latest}: run the same version of the hub on both"),

        None => bail!("not a hub's database: it has no migrations"),

    }

}

async fn columns(tx: &mut sqlx::PgConnection, table: &str) -> Result<Vec<Col>> {

    let rows: Vec<(String, String, bool)> = sqlx::query_as(

        "SELECT CAST(a.attname AS TEXT), format_type(a.atttypid, a.atttypmod), \

           EXISTS (SELECT 1 FROM pg_constraint k WHERE k.conrelid = t.oid AND k.confrelid = t.oid AND k.contype = 'f' AND a.attnum = ANY (k.conkey)) \

         FROM pg_attribute a JOIN pg_class t ON t.oid = a.attrelid \

         WHERE t.relname = $1 AND t.relnamespace = CAST(current_schema() AS regnamespace) \

           AND a.attnum > 0 AND NOT a.attisdropped AND a.attgenerated = '' ORDER BY a.attnum",

    )

    .bind(table)

    .fetch_all(&mut *tx)

    .await?;

    Ok(rows

        .into_iter()

        .map(|(name, ty, self_ref)| {

            let kind = match ty.as_str() {

                "bigint" | "integer" | "smallint" => Kind::Int,

                "boolean" => Kind::Bool,

                "double precision" | "real" => Kind::Real,

                "bytea" => Kind::Blob,

                _ => Kind::Text,

            };

            Col { name, kind, self_ref }

        })

        .collect())

}

fn read(row: &SqliteRow, i: usize, kind: Kind) -> Result<Value> {

    Ok(match kind {

        Kind::Int => Value::Int(row.try_get(i)?),

        Kind::Bool => Value::Bool(row.try_get::<Option<i64>, _>(i)?.map(|v| v != 0)),

        Kind::Real => Value::Real(row.try_get(i)?),

        Kind::Blob => Value::Blob(row.try_get(i)?),

        Kind::Text => Value::Text(row.try_get(i)?),

    })

}

fn list(cols: &[&Col]) -> String {

    cols.iter().map(|c| format!("\"{}\"", c.name)).collect::<Vec<_>>().join(", ")

}

async fn copy_table(lite: &mut sqlx::SqliteConnection, tx: &mut sqlx::PgConnection, table: &str, cols: &[Col]) -> Result<i64> {

    use futures::TryStreamExt;

    let all: Vec<&Col> = cols.iter().collect();

    let now: Vec<&Col> = cols.iter().filter(|c| !c.self_ref).collect();

    let id_at = cols.iter().position(|c| c.name == "id");

    let sql = format!("SELECT {} FROM \"{table}\" ORDER BY rowid", list(&all));

    // Streamed, a batch at a time: a table (Flow's logs) can be large.

    let mut rows = sqlx::query(&sql).fetch(&mut *lite);

    let (mut n, mut batch, mut later) = (0i64, Vec::with_capacity(BATCH), Vec::new());

    loop {

        let row = rows.try_next().await?;

        if let Some(row) = &row {

            let mut values = Vec::with_capacity(now.len());

            let mut refs = Vec::new();

            for (i, c) in cols.iter().enumerate() {

                let v = read(row, i, c.kind).with_context(|| format!("read {table}.{}", c.name))?;

                if c.self_ref {

                    if matches!(v, Value::Int(Some(_))) {

                        refs.push((c.name.clone(), v));

                    }

                } else {

                    values.push(v);

                }

            }

            if !refs.is_empty() {

                let Some(i) = id_at else { bail!("{table} points at itself but has no id") };

                later.push((read(row, i, Kind::Int)?, refs));

            }

            batch.push(values);

            n += 1;

        }

        if batch.len() == BATCH || (row.is_none() && !batch.is_empty()) {

            let mut q = sqlx::QueryBuilder::<super::Db>::new(format!("INSERT INTO \"{table}\" ({}) ", list(&now)));

            q.push_values(batch.drain(..), |mut b, values| {

                for v in values {

                    match v {

                        Value::Int(v) => b.push_bind(v),

                        Value::Bool(v) => b.push_bind(v),

                        Value::Real(v) => b.push_bind(v),

                        Value::Blob(v) => b.push_bind(v),

                        Value::Text(v) => b.push_bind(v),

                    };

                }

            });

            q.build().execute(&mut *tx).await?;

        }

        if row.is_none() {

            break;

        }

    }

    for (id, refs) in later {

        let Value::Int(id) = id else { continue };

        for (name, v) in refs {

            if let Value::Int(v) = v {

                sqlx::query(&format!("UPDATE \"{table}\" SET \"{name}\" = $1 WHERE id = $2")).bind(v).bind(id).execute(&mut *tx).await?;

            }

        }

    }

    Ok(n)

}

async fn copy_search(lite: &mut sqlx::SqliteConnection, tx: &mut sqlx::PgConnection) -> Result<i64> {

    let rows: Vec<(String, i64, i64, i64, String, String)> =

        sqlx::query_as("SELECT kind, repo_id, target_id, number, title, body FROM search").fetch_all(&mut *lite).await?;

    for chunk in rows.chunks(BATCH) {

        let mut q = sqlx::QueryBuilder::<super::Db>::new("INSERT INTO search (kind, repo_id, target_id, number, title, body) ");

        q.push_values(chunk, |mut b, (kind, repo_id, target_id, number, title, body)| {

            b.push_bind(kind).push_bind(repo_id).push_bind(target_id).push_bind(number).push_bind(title).push_bind(body);

        });

        q.build().execute(&mut *tx).await?;

    }

    Ok(rows.len() as i64)

}

#[cfg(feature = "postgres")]

pub mod copy;

    /// Copy a SQLite hub's records into this hub's empty PostgreSQL

    /// database, to move a hub (docs/postgres.md)

    #[cfg(feature = "postgres")]

    CopyDb {

        /// The old hub's `hub.db`, from a stopped hub (or a copy of one)

        #[arg(long)]

        from: PathBuf,

    },

            }

        }

        #[cfg(feature = "postgres")]

        Cmd::Admin { cmd: AdminCmd::CopyDb { from } } => {

            let config = Config::load(&cli.data)?;

            let pool = db::open(&config).await?;

            let copied = db::copy::from_sqlite(&pool, &from).await?;

            for (table, n) in &copied {

                println!("{n:>9}  {table}");

            println!("copied {} rows in {} tables", copied.iter().map(|(_, n)| n).sum::<i64>(), copied.len());

            Ok(())

    let names: Vec<String> = sqlx::query_scalar("SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%'")

        .fetch_all(&mut conn)

        .await

        .unwrap();

/// A SQLite hub with a few of each kind of value, copied into Postgres.

#[tokio::test]

async fn a_sqlite_hub_copies_into_postgres_ids_flags_forks_and_all() {

    use sqlx::Connection;

    let old = tempfile::tempdir().unwrap();

    let path = old.path().join("hub.db");

    let options = sqlx::sqlite::SqliteConnectOptions::new().filename(&path).create_if_missing(true);

    let mut lite = sqlx::SqliteConnection::connect_with(&options).await.unwrap();

    sqlx::migrate!("./migrations").run(&mut lite).await.unwrap();

    for sql in [

        "INSERT INTO accounts (id, username, email, password_hash, session_salt, is_admin, created_ms) VALUES (7, 'Andy', '[email protected]', 'h', 's', 1, 5)",

        "INSERT INTO accounts (id, username, email, password_hash, session_salt, banned, created_ms) VALUES (9, 'bob', '[email protected]', 'h', 's', 1, 6)",

        // The fork comes first: its parent is filled in after.

        "INSERT INTO repositories (id, owner, name, created_ms, updated_ms) VALUES (4, 'andy', 'game', 1, 1)",

        "INSERT INTO repositories (id, owner, name, created_ms, updated_ms, archived) VALUES (3, 'bob', 'engine', 1, 1, 1)",

        "UPDATE repositories SET forked_from = 3 WHERE id = 4",

        "INSERT INTO flow_secrets (scope, scope_id, name, value, updated_ms) VALUES ('repo', 3, 'KEY', X'00FF10', 1)",

        "INSERT INTO search (kind, repo_id, target_id, number, title, body) VALUES ('issue', 3, 1, 1, 'Sprites flicker', 'the sprites flicker on level two')",

    ] {

        sqlx::query(sql).execute(&mut lite).await.unwrap();

    }

    lite.close().await.unwrap();

    let dir = tempfile::tempdir().unwrap();

    let config = Config::load(dir.path()).unwrap();

    let pool = oxygenhub::db::open(&config).await.unwrap();

    let copied = oxygenhub::db::copy::from_sqlite(&pool, &path).await.map_err(|e| format!("{e:#}")).unwrap();

    let count = |t: &str| copied.iter().find(|(c, _)| c == t).map(|(_, n)| *n);

    assert_eq!((count("accounts"), count("repositories"), count("search")), (Some(2), Some(2), Some(1)));

    let flags: Vec<(String, bool, bool)> =

        sqlx::query_as("SELECT username, is_admin, banned FROM accounts WHERE username = 'andy' OR username = 'BOB' ORDER BY id")

            .fetch_all(&pool)

            .await

            .unwrap();

    assert_eq!(flags, [("Andy".to_string(), true, false), ("bob".to_string(), false, true)], "flags, and names without case");

    let game = oxygenhub::db::repos::get(&pool, "andy", "game").await.unwrap().unwrap();

    assert_eq!((game.id, game.forked_from), (4, Some(3)));

    let secret: Vec<u8> = sqlx::query_scalar("SELECT value FROM flow_secrets WHERE name = 'KEY'").fetch_one(&pool).await.unwrap();

    assert_eq!(secret, [0x00, 0xff, 0x10]);

    let hits = oxygenhub::db::notify::search(&pool, "flickering sprite", None, 5).await.unwrap();

    assert_eq!(hits.iter().map(|h| h.title.as_str()).collect::<Vec<_>>(), ["Sprites flicker"]);

    // New rows take ids after the copied ones.

    let next = oxygenhub::db::repos::create(&pool, "andy", "next", "", hub_api::Visibility::Public).await.unwrap();

    assert_eq!(next.id, 5);

    // Never into a database with rows.

    let again = oxygenhub::db::copy::from_sqlite(&pool, &path).await.unwrap_err().to_string();

    assert!(again.contains("already has rows"), "{again}");

}

## Moving a SQLite hub to Postgres

On the new hub (built with `postgres`, `[database]` set, never yet served),

with the old hub stopped and its `hub.db` copied across:

```sh

oxygenhub --data /srv/hub admin copy-db --from /srv/old/hub.db

```

Both hubs must be the same version: the copy refuses a `hub.db` at another

migration, and a database that already has rows. It copies every table in

one transaction with ids kept, moves each id sequence past them, copies the

search index, and checks every table's count before committing. Then move the

data directory's `repos`, `wikis` and `keys` across as they are.