//! 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. With `replace`, the

/// rows already here (a fresh hub's, a rehearsal's) are emptied first, in

/// the same transaction; without it, a database with rows is refused.

pub async fn from_sqlite(pool: &Pool, path: &Path, replace: bool) -> 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?;

    if !replace {

        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, or give --replace to empty this one first");

            }

        }

    }

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

    if replace {

        let all: Vec<String> = tables.iter().chain(std::iter::once(&"search".to_string())).map(|t| format!("\"{t}\"")).collect();

        sqlx::query(&format!("TRUNCATE {} RESTART IDENTITY CASCADE", all.join(", "))).execute(&mut *tx).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)

}