//! 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)
}