lift-migration 0.1.7

#![cfg(feature = "sqlite")]

use std::{
    fs,
    path::{Path, PathBuf},
    time::{SystemTime, UNIX_EPOCH},
};

use quex::{self, FromRow, Pool, Row, SqliteConnectOptions};

use crate::{AlterTableBlueprint, MigrationError, Migrator, SchemaDialect, TableBlueprint};

#[derive(Debug)]
pub struct UserCount {
    value: i64,
}

impl FromRow for UserCount {
    fn from_row(row: &Row) -> quex::Result<Self> {
        Ok(Self {
            value: row.get("value")?,
        })
    }
}

#[derive(Debug)]
struct ForeignKeyRule {
    on_delete: String,
}

impl FromRow for ForeignKeyRule {
    fn from_row(row: &Row) -> quex::Result<Self> {
        Ok(Self {
            on_delete: row.get("on_delete")?,
        })
    }
}

#[derive(Debug)]
struct TableColumn {
    data_type: String,
}

impl FromRow for TableColumn {
    fn from_row(row: &Row) -> quex::Result<Self> {
        Ok(Self {
            data_type: row.get("type")?,
        })
    }
}

#[derive(Debug)]
struct UserEmailValues {
    email: String,
    normalized_email: String,
}

impl FromRow for UserEmailValues {
    fn from_row(row: &Row) -> quex::Result<Self> {
        Ok(Self {
            email: row.get("email")?,
            normalized_email: row.get("normalized_email")?,
        })
    }
}

#[derive(Debug)]
struct PrimaryKeyColumn {
    name: String,
    pk: i64,
}

impl FromRow for PrimaryKeyColumn {
    fn from_row(row: &Row) -> quex::Result<Self> {
        Ok(Self {
            name: row.get("name")?,
            pk: row.get("pk")?,
        })
    }
}

async fn sqlite_pool() -> quex::Result<Pool> {
    quex::Pool::connect(SqliteConnectOptions::new().in_memory())?
        .max_size(1)
        .build()
        .await
}

fn temp_migration_dir(name: &str) -> PathBuf {
    let mut path = std::env::temp_dir();
    let nanos = SystemTime::now()
        .duration_since(UNIX_EPOCH)
        .unwrap()
        .as_nanos();
    path.push(format!("lift-migration-{name}-{nanos}"));
    fs::create_dir_all(&path).unwrap();
    path
}

fn write_migration(dir: &Path, name: &str, contents: &str) {
    fs::write(dir.join(name), contents).unwrap();
}

#[test]
fn schema_builder_renders_sqlite_and_postgres_sql() {
    let mut table = TableBlueprint::new("users");
    table.id();
    table.string("name");
    table.unique(["name"]);
    table.check("length(name) > 0");
    table.index("users_name_idx", ["name"]);
    table.unique_index("users_name_unique_idx", ["name"]);
    table
        .foreign_id("team_id")
        .constrained()
        .cascade_on_delete()
        .cascade_on_update();
    table.timestamps();

    assert_eq!(
        table.create_sql(SchemaDialect::Sqlite),
        r#"create table "users" ("id" integer primary key autoincrement, "name" varchar(255) not null, "team_id" bigint not null, "created_at" datetime not null default current_timestamp, "updated_at" datetime not null default current_timestamp, foreign key ("team_id") references "teams"("id") on delete cascade on update cascade, unique ("name"), check (length(name) > 0));"#
    );
    assert_eq!(
        table.create_sql(SchemaDialect::Postgres),
        r#"create table "users" ("id" bigserial primary key, "name" varchar(255) not null, "team_id" bigint not null, "created_at" timestamp not null default current_timestamp, "updated_at" timestamp not null default current_timestamp, foreign key ("team_id") references "teams"("id") on delete cascade on update cascade, unique ("name"), check (length(name) > 0));"#
    );
}

#[test]
fn alter_table_builder_renders_drop_columns_for_each_dialect() {
    let mut table = AlterTableBlueprint::new("users");
    table.drop_column("nickname");
    table.drop_timestamps();

    assert_eq!(
        table.sql_statements(SchemaDialect::Sqlite),
        vec![
            r#"alter table "users" drop column "nickname";"#.to_owned(),
            r#"alter table "users" drop column "created_at";"#.to_owned(),
            r#"alter table "users" drop column "updated_at";"#.to_owned(),
        ]
    );
    assert_eq!(
        table.sql_statements(SchemaDialect::Postgres),
        vec![
            r#"alter table "users" drop column "nickname", drop column "created_at", drop column "updated_at";"#.to_owned(),
        ]
    );
}

#[test]
fn alter_table_builder_renders_add_columns() {
    let mut table = AlterTableBlueprint::new("users");
    table.string("slug").nullable().unique();

    assert_eq!(
        table.sql_statements(SchemaDialect::Sqlite),
        vec![r#"alter table "users" add column "slug" varchar(255) unique;"#.to_owned()]
    );
}

#[tokio::test]
async fn sqlite_migrator_runs_lift_files_and_rolls_back() {
    let dir = temp_migration_dir("run-and-rollback");
    write_migration(
        &dir,
        "202604030001_create_users.lift",
        r#"
create users {
  id
  email text unique
  team_id bigint
}

down {
  drop users
}
"#,
    );

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);

    let report = migrator.run().await.unwrap();
    assert_eq!(report.batch, Some(1));
    assert_eq!(report.applied, vec![202604030001]);

    quex::query("insert into users(email, team_id) values(?, ?)")
        .bind("alice@example.com")
        .bind(1_i64)
        .execute(&pool)
        .await
        .unwrap();

    let users = quex::query("select count(*) as value from users")
        .one::<UserCount>(&pool)
        .await
        .unwrap();
    assert_eq!(users.value, 1);

    let rollback = migrator.rollback_last_batch().await.unwrap();
    assert_eq!(rollback.batch, Some(1));
    assert_eq!(rollback.rolled_back, vec![202604030001]);

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_detects_checksum_drift() {
    let dir = temp_migration_dir("checksum");
    let file = dir.join("202604030002_create_users.lift");
    fs::write(
        &file,
        r#"
create users {
  id primary
  email text
}

down {
  drop users
}
"#,
    )
    .unwrap();

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);
    migrator.run().await.unwrap();

    fs::write(
        &file,
        r#"
create users {
  id primary
  email text unique
}

down {
  drop users
}
"#,
    )
    .unwrap();

    let err = migrator.run().await.err().unwrap();
    assert!(matches!(
        err,
        MigrationError::ChecksumMismatch { id: 202604030002 }
    ));

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_runs_lift_alter_and_rename_table() {
    let dir = temp_migration_dir("alter-and-rename");
    write_migration(
        &dir,
        "202604030003_create_and_rename_users.lift",
        r#"
create users {
  id primary
  email text unique
}

alter users {
  add column timezone text default "UTC"
  rename column email to login
  add index (timezone)
}

rename users to accounts

down {
  rename accounts to users

  alter users {
    drop index users_timezone_idx
    rename column login to email
    drop column timezone
  }

  drop users
}
"#,
    );

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);

    migrator.run().await.unwrap();

    let table_count = quex::query(
        "select count(*) as value from sqlite_master where type = 'table' and name = ?",
    )
    .bind("accounts")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(table_count.value, 1);

    let column_count =
        quex::query("select count(*) as value from pragma_table_info('accounts') where name = ?")
            .bind("login")
            .one::<UserCount>(&pool)
            .await
            .unwrap();
    assert_eq!(column_count.value, 1);

    let index_count = quex::query(
        "select count(*) as value from sqlite_master where type = 'index' and name = ?",
    )
    .bind("users_timezone_idx")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(index_count.value, 1);

    migrator.rollback_last_batch().await.unwrap();

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_executes_enum_statements() {
    let dir = temp_migration_dir("enum-ops");
    write_migration(
        &dir,
        "202604030004_create_status_enum.lift",
        r#"
create enum post_status { draft, published }

create posts {
  id primary
  status post_status
}

alter enum post_status {
  add value archived
}

rename enum post_status to article_status

down {
  drop posts
  drop enum article_status
}
"#,
    );

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);
    migrator.run().await.unwrap();

    let column = quex::query("select type from pragma_table_info('posts') where name = ?")
        .bind("status")
        .one::<TableColumn>(&pool)
        .await
        .unwrap();
    assert_eq!(column.data_type.to_ascii_lowercase(), "post_status");

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_executes_composite_primary_keys() {
    let dir = temp_migration_dir("composite-primary");
    write_migration(
        &dir,
        "202604030005_create_memberships.lift",
        r#"
create memberships {
  user_id uuid
  org_id uuid
  primary (user_id, org_id)
}
"#,
    );

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);
    migrator.run().await.unwrap();

    let primary = quex::query(
        "select name, pk from pragma_table_info('memberships') where pk > 0 order by pk",
    )
    .all::<PrimaryKeyColumn>(&pool)
    .await
    .unwrap();
    assert_eq!(primary.len(), 2);
    assert_eq!(primary[0].name, "user_id");
    assert_eq!(primary[0].pk, 1);
    assert_eq!(primary[1].name, "org_id");
    assert_eq!(primary[1].pk, 2);

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_supports_implicit_remember_token_helper() {
    let dir = temp_migration_dir("remember-token-helper");
    write_migration(
        &dir,
        "202604030006_create_users.lift",
        r#"
create users {
  id primary
  remember_token
}
"#,
    );

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);
    migrator.run().await.unwrap();

    let column_type = quex::query(
        "select count(*) as value from pragma_table_info('users') where name = ? and type = ?",
    )
    .bind("remember_token")
    .bind("varchar(100)")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(column_type.value, 1);

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_infers_relation_type_and_auto_indexes() {
    let dir = temp_migration_dir("relation-defaults");
    let pool = sqlite_pool().await.unwrap();

    quex::query("create table countries (code varchar(2) primary key not null)")
        .execute(&pool)
        .await
        .unwrap();

    write_migration(
        &dir,
        "202604030007_create_addresses.lift",
        r#"
create addresses {
  id primary
  country_code ref countries.code
}
"#,
    );

    let migrator = Migrator::new(&pool).path(&dir);
    migrator.run().await.unwrap();

    let column = quex::query("select type from pragma_table_info('addresses') where name = ?")
        .bind("country_code")
        .one::<TableColumn>(&pool)
        .await
        .unwrap();
    assert_eq!(column.data_type.to_ascii_lowercase(), "varchar(2)");

    let index = quex::query(
        "select count(*) as value from sqlite_master where type = 'index' and name = ?",
    )
    .bind("addresses_country_code_idx")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(index.value, 1);

    let fk = quex::query("select on_delete from pragma_foreign_key_list('addresses') limit 1")
        .one::<ForeignKeyRule>(&pool)
        .await
        .unwrap();
    assert_eq!(fk.on_delete.to_ascii_lowercase(), "restrict");

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_honors_relation_no_index_override() {
    let dir = temp_migration_dir("relation-no-index");
    let pool = sqlite_pool().await.unwrap();

    quex::query("create table users (id integer primary key not null)")
        .execute(&pool)
        .await
        .unwrap();

    write_migration(
        &dir,
        "202604030008_create_posts.lift",
        r#"
create posts {
  id primary
  author_id ref users no index
}
"#,
    );

    let migrator = Migrator::new(&pool).path(&dir);
    migrator.run().await.unwrap();

    let index = quex::query(
        "select count(*) as value from sqlite_master where type = 'index' and name = ?",
    )
    .bind("posts_author_id_idx")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(index.value, 0);

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_supports_relation_add_column_in_alter_table() {
    let dir = temp_migration_dir("relation-alter-add");
    let pool = sqlite_pool().await.unwrap();

    quex::query("create table users (id integer primary key not null)")
        .execute(&pool)
        .await
        .unwrap();

    write_migration(
        &dir,
        "202604030009_create_posts.lift",
        r#"
create posts {
  id primary
}

alter posts {
  add column author_id ref users
}
"#,
    );

    let migrator = Migrator::new(&pool).path(&dir);
    migrator.run().await.unwrap();

    let column = quex::query("select type from pragma_table_info('posts') where name = ?")
        .bind("author_id")
        .one::<TableColumn>(&pool)
        .await
        .unwrap();
    assert_eq!(column.data_type.to_ascii_lowercase(), "integer");

    let fk = quex::query("select on_delete from pragma_foreign_key_list('posts') limit 1")
        .one::<ForeignKeyRule>(&pool)
        .await
        .unwrap();
    assert_eq!(fk.on_delete.to_ascii_lowercase(), "restrict");

    let index = quex::query(
        "select count(*) as value from sqlite_master where type = 'index' and name = ?",
    )
    .bind("posts_author_id_idx")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(index.value, 1);

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_runs_forward_only_sql_migration() {
    let dir = temp_migration_dir("sql-forward-only");
    write_migration(
        &dir,
        "202604030009_create_logs.sql",
        "create table logs (id integer primary key, message text not null);",
    );

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);
    let report = migrator.run().await.unwrap();
    assert_eq!(report.applied, vec![202604030009]);

    let table = quex::query(
        "select count(*) as value from sqlite_master where type = 'table' and name = ?",
    )
    .bind("logs")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(table.value, 1);

    let rollback = migrator.rollback_last_batch().await.unwrap();
    assert_eq!(rollback.rolled_back, vec![202604030009]);

    let table = quex::query(
        "select count(*) as value from sqlite_master where type = 'table' and name = ?",
    )
    .bind("logs")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(table.value, 1);

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_runs_sql_migration_pair_and_rolls_back() {
    let dir = temp_migration_dir("sql-pair");
    write_migration(
        &dir,
        "202604030010_create_tags.sql",
        "create table tags (id integer primary key, name text not null);",
    );
    write_migration(
        &dir,
        "202604030010_create_tags.down.sql",
        "drop table tags;",
    );

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);
    migrator.run().await.unwrap();

    let table = quex::query(
        "select count(*) as value from sqlite_master where type = 'table' and name = ?",
    )
    .bind("tags")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(table.value, 1);

    let rollback = migrator.rollback_last_batch().await.unwrap();
    assert_eq!(rollback.rolled_back, vec![202604030010]);

    let table = quex::query(
        "select count(*) as value from sqlite_master where type = 'table' and name = ?",
    )
    .bind("tags")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(table.value, 0);

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_retries_failed_sql_migration_after_fix() {
    let dir = temp_migration_dir("retry-failed-sql");
    let file = dir.join("202604030011_create_logs.sql");
    fs::write(
        &file,
        r#"
create table logs (
  id integer primary key,
  message text not null
);
create index logs_missing_idx on logs (missing_column);
"#,
    )
    .unwrap();

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);

    let err = migrator.run().await.err().unwrap();
    assert!(!matches!(err, MigrationError::FailedMigration { .. }));

    let table = quex::query(
        "select count(*) as value from sqlite_master where type = 'table' and name = ?",
    )
    .bind("logs")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(table.value, 0);

    fs::write(
        &file,
        r#"
create table logs (
  id integer primary key,
  message text not null
);
"#,
    )
    .unwrap();

    let report = migrator.run().await.unwrap();
    assert_eq!(report.applied, vec![202604030011]);

    let table = quex::query(
        "select count(*) as value from sqlite_master where type = 'table' and name = ?",
    )
    .bind("logs")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(table.value, 1);

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_rejects_down_sql_without_forward_pair() {
    let dir = temp_migration_dir("sql-down-only");
    write_migration(
        &dir,
        "202604030012_create_audit_logs.down.sql",
        "drop table audit_logs;",
    );

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);
    let err = migrator.run().await.err().unwrap();
    assert!(matches!(err, MigrationError::InvalidMigrationFile { .. }));

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_runs_inline_sql_blocks_in_lift() {
    let dir = temp_migration_dir("inline-sql");
    write_migration(
        &dir,
        "202604030013_inline_sql.lift",
        r#"
sql """
create table logs (
  id integer primary key,
  message text not null
);
"""

down {
  sql """
drop table logs;
"""
}
"#,
    );

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);
    migrator.run().await.unwrap();

    let table = quex::query(
        "select count(*) as value from sqlite_master where type = 'table' and name = ?",
    )
    .bind("logs")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(table.value, 1);

    migrator.rollback_last_batch().await.unwrap();

    let table = quex::query(
        "select count(*) as value from sqlite_master where type = 'table' and name = ?",
    )
    .bind("logs")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(table.value, 0);

    fs::remove_dir_all(dir).unwrap();
}

#[tokio::test]
async fn sqlite_migrator_runs_backfill_blocks_in_lift() {
    let dir = temp_migration_dir("backfill");
    write_migration(
        &dir,
        "202604030014_backfill_users.lift",
        r#"
create users {
  id primary
  email text
}

sql """
insert into users (email) values ('Alice@Example.com');
"""

alter users {
  add column normalized_email text nullable
}

backfill {
  update users
  set normalized_email = lower(email)
  where normalized_email is null;
}

down {
  backfill {
    update users
    set normalized_email = null;
  }

  alter users {
    drop column normalized_email
  }

  drop users
}
"#,
    );

    let pool = sqlite_pool().await.unwrap();
    let migrator = Migrator::new(&pool).path(&dir);
    migrator.run().await.unwrap();

    let user = quex::query("select email, normalized_email from users limit 1")
        .one::<UserEmailValues>(&pool)
        .await
        .unwrap();
    assert_eq!(user.email, "Alice@Example.com");
    assert_eq!(user.normalized_email, "alice@example.com");

    migrator.rollback_last_batch().await.unwrap();

    let table = quex::query(
        "select count(*) as value from sqlite_master where type = 'table' and name = ?",
    )
    .bind("users")
    .one::<UserCount>(&pool)
    .await
    .unwrap();
    assert_eq!(table.value, 0);

    fs::remove_dir_all(dir).unwrap();
}