wasm-dbms 0.10.0

use std::cmp::Ordering;

use wasm_dbms_api::prelude::{
    ColumnDef, Database as _, DeleteBehavior, Filter, InsertRecord as _, Nullable, OrderDirection,
    Query, TableSchema as _, Text, Uint32, UpdateRecord as _, Value,
};
use wasm_dbms_macros::{DatabaseSchema, Table};
use wasm_dbms_memory::prelude::HeapMemoryProvider;

use super::sort_values_with_direction;
use crate::prelude::{DbmsContext, WasmDbmsDatabase};
use crate::schema::DatabaseSchema as _;

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "users"]
pub struct User {
    #[primary_key]
    pub id: Uint32,
    pub name: Text,
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "posts"]
pub struct Post {
    #[primary_key]
    pub id: Uint32,
    pub title: Text,
    #[foreign_key(entity = "User", table = "users", column = "id")]
    pub user_id: Uint32,
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "contracts"]
pub struct Contract {
    #[primary_key]
    pub id: Uint32,
    #[unique]
    pub code: Text,
    #[autoincrement]
    pub order: Uint32,
    #[foreign_key(entity = "User", table = "users", column = "id")]
    pub user_id: Uint32,
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "sales"]
pub struct Sale {
    #[primary_key]
    pub id: Uint32,
    pub category: Text,
    pub price: Uint32,
    pub bonus: Nullable<Uint32>,
}

#[derive(DatabaseSchema)]
#[tables(User = "users", Post = "posts", Contract = "contracts", Sale = "sales")]
pub struct TestSchema;

fn setup() -> DbmsContext<HeapMemoryProvider> {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    TestSchema::register_tables(&ctx).unwrap();
    ctx
}

fn insert_user(db: &WasmDbmsDatabase<'_, HeapMemoryProvider>, id: u32, name: &str) {
    let insert = UserInsertRequest::from_values(&[
        (User::columns()[0], Value::Uint32(Uint32(id))),
        (User::columns()[1], Value::Text(Text(name.to_string()))),
    ])
    .unwrap();
    db.insert::<User>(insert).unwrap();
}

fn insert_contract(
    db: &WasmDbmsDatabase<'_, HeapMemoryProvider>,
    id: u32,
    code: &str,
    user_id: u32,
) {
    let insert = ContractInsertRequest::from_values(&[
        (Contract::columns()[0], Value::Uint32(Uint32(id))),
        (Contract::columns()[1], Value::Text(Text(code.to_string()))),
        // columns()[2] is `order` (autoincrement) — omitted so DBMS auto-generates it
        (Contract::columns()[3], Value::Uint32(Uint32(user_id))),
    ])
    .unwrap();
    db.insert::<Contract>(insert).unwrap();
}

fn insert_post(db: &WasmDbmsDatabase<'_, HeapMemoryProvider>, id: u32, title: &str, user_id: u32) {
    let insert = PostInsertRequest::from_values(&[
        (Post::columns()[0], Value::Uint32(Uint32(id))),
        (Post::columns()[1], Value::Text(Text(title.to_string()))),
        (Post::columns()[2], Value::Uint32(Uint32(user_id))),
    ])
    .unwrap();
    db.insert::<Post>(insert).unwrap();
}

// -- sort_values_with_direction tests --

#[test]
fn test_sort_values_ascending() {
    let a = Value::Uint32(Uint32(1));
    let b = Value::Uint32(Uint32(2));
    assert_eq!(
        sort_values_with_direction(Some(&a), Some(&b), OrderDirection::Ascending),
        Ordering::Less
    );
}

#[test]
fn test_sort_values_descending() {
    let a = Value::Uint32(Uint32(1));
    let b = Value::Uint32(Uint32(2));
    assert_eq!(
        sort_values_with_direction(Some(&a), Some(&b), OrderDirection::Descending),
        Ordering::Greater
    );
}

#[test]
fn test_sort_values_some_none() {
    let a = Value::Uint32(Uint32(1));
    assert_eq!(
        sort_values_with_direction(Some(&a), None, OrderDirection::Ascending),
        Ordering::Greater
    );
    assert_eq!(
        sort_values_with_direction(None, Some(&a), OrderDirection::Ascending),
        Ordering::Less
    );
}

#[test]
fn test_sort_values_none_none() {
    assert_eq!(
        sort_values_with_direction(None, None, OrderDirection::Ascending),
        Ordering::Equal
    );
}

// -- select with ordering --

#[test]
fn test_select_with_order_by_ascending() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 3, "charlie");
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");

    let rows = db
        .select::<User>(Query::builder().all().order_by_asc("name").build())
        .unwrap();
    assert_eq!(rows.len(), 3);
    assert_eq!(rows[0].name, Some(Text("alice".to_string())));
    assert_eq!(rows[1].name, Some(Text("bob".to_string())));
    assert_eq!(rows[2].name, Some(Text("charlie".to_string())));
}

#[test]
fn test_select_with_order_by_descending() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");
    insert_user(&db, 3, "charlie");

    let rows = db
        .select::<User>(Query::builder().all().order_by_desc("name").build())
        .unwrap();
    assert_eq!(rows.len(), 3);
    assert_eq!(rows[0].name, Some(Text("charlie".to_string())));
    assert_eq!(rows[1].name, Some(Text("bob".to_string())));
    assert_eq!(rows[2].name, Some(Text("alice".to_string())));
}

// -- select with offset and limit --

#[test]
fn test_select_with_limit() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");
    insert_user(&db, 3, "charlie");

    let rows = db
        .select::<User>(Query::builder().all().limit(2).build())
        .unwrap();
    assert_eq!(rows.len(), 2);
}

#[test]
fn test_select_with_zero_limit_returns_no_rows() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let rows = db
        .select::<User>(Query::builder().all().limit(0).build())
        .unwrap();
    assert!(rows.is_empty());
}

#[test]
fn test_index_select_with_zero_limit_returns_no_rows() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let rows = db
        .select::<User>(
            Query::builder()
                .all()
                .filter(Some(Filter::eq("id", Value::from(1u32))))
                .limit(0)
                .build(),
        )
        .unwrap();
    assert!(rows.is_empty());
}

#[test]
fn test_select_with_maximum_limit_does_not_preallocate_the_limit() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

    let rows = db
        .select::<User>(Query::builder().all().limit(usize::MAX).build())
        .unwrap();
    assert!(rows.is_empty());
}

#[test]
fn test_select_with_offset() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");
    insert_user(&db, 3, "charlie");

    let rows = db
        .select::<User>(Query::builder().all().offset(1).build())
        .unwrap();
    assert_eq!(rows.len(), 2);
}

#[test]
fn test_select_with_offset_and_limit() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");
    insert_user(&db, 3, "charlie");

    let rows = db
        .select::<User>(Query::builder().all().offset(1).limit(1).build())
        .unwrap();
    assert_eq!(rows.len(), 1);
}

#[test]
fn test_select_with_order_by_and_limit() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "charlie");
    insert_user(&db, 2, "alice");
    insert_user(&db, 3, "bob");

    // ORDER BY name ASC with LIMIT 2 should return the first 2 sorted rows
    let rows = db
        .select::<User>(Query::builder().all().order_by_asc("name").limit(2).build())
        .unwrap();
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].name, Some(Text("alice".to_string())));
    assert_eq!(rows[1].name, Some(Text("bob".to_string())));
}

#[test]
fn test_select_with_order_by_and_offset_and_limit() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "charlie");
    insert_user(&db, 2, "alice");
    insert_user(&db, 3, "bob");
    insert_user(&db, 4, "dave");

    // ORDER BY name ASC with OFFSET 1 LIMIT 2 should skip "alice" and return "bob", "charlie"
    let rows = db
        .select::<User>(
            Query::builder()
                .all()
                .order_by_asc("name")
                .offset(1)
                .limit(2)
                .build(),
        )
        .unwrap();
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].name, Some(Text("bob".to_string())));
    assert_eq!(rows[1].name, Some(Text("charlie".to_string())));
}

// -- select with filter --

#[test]
fn test_select_with_filter() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");

    let rows = db
        .select::<User>(
            Query::builder()
                .all()
                .and_where(Filter::eq("name", Value::Text(Text("alice".to_string()))))
                .build(),
        )
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].name, Some(Text("alice".to_string())));
}

// -- select with column selection --

#[test]
fn test_select_with_column_selection() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let rows = TestSchema
        .select(&db, "users", Query::builder().field("name").build())
        .unwrap();
    assert_eq!(rows.len(), 1);
    // Should only have the "name" column
    assert_eq!(rows[0].len(), 1);
    assert_eq!(rows[0][0].0.name, "name");
}

// -- update operations --

#[test]
fn test_update_record() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[1], Value::Text(Text("alicia".to_string())))],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    let count = db.update::<User>(patch).unwrap();
    assert_eq!(count, 1);

    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows[0].name, Some(Text("alicia".to_string())));
}

// -- oversized values --

/// Builds an insert request for a user whose name is `name_len` bytes long.
fn user_insert_with_name_len(id: u32, name_len: usize) -> UserInsertRequest {
    UserInsertRequest::from_values(&[
        (User::columns()[0], Value::Uint32(Uint32(id))),
        (User::columns()[1], Value::Text(Text("x".repeat(name_len)))),
    ])
    .unwrap()
}

#[test]
fn test_insert_rejects_oversized_text_before_storage() {
    use wasm_dbms_api::prelude::{DbmsError, MemoryError};

    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

    // 65_533 bytes still fits the Text prefix but not a record page;
    // 65_534 and 65_536 do not fit the 2-byte length prefix at all.
    for name_len in [65_533, 65_534, 65_536, 70_000] {
        let result = db.insert::<User>(user_insert_with_name_len(1, name_len));
        assert!(
            matches!(
                result,
                Err(DbmsError::Memory(MemoryError::DataTooLarge { .. }))
            ),
            "{name_len}-byte name: expected DataTooLarge, got {result:?}"
        );
    }

    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert!(rows.is_empty());

    // the table is still usable afterwards
    insert_user(&db, 1, "alice");
    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
}

#[test]
fn test_insert_round_trips_largest_storable_text() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

    // id(4) + prefix(2) + name + record header(2), padded to 32, must fit
    // an MSize: 4 + 2 + 65_494 + 2 = 65_502 -> 65_504.
    let name = "x".repeat(65_494);
    let insert = UserInsertRequest::from_values(&[
        (User::columns()[0], Value::Uint32(Uint32(1))),
        (User::columns()[1], Value::Text(Text(name.clone()))),
    ])
    .unwrap();
    db.insert::<User>(insert).unwrap();

    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows[0].name, Some(Text(name)));
}

#[test]
fn test_update_rejects_oversized_text_and_keeps_original_row() {
    use wasm_dbms_api::prelude::{DbmsError, MemoryError};

    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[1], Value::Text(Text("x".repeat(70_000))))],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    let result = db.update::<User>(patch);
    assert!(
        matches!(
            result,
            Err(DbmsError::Memory(MemoryError::DataTooLarge { .. }))
        ),
        "expected DataTooLarge, got {result:?}"
    );

    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].name, Some(Text("alice".to_string())));
}

#[test]
fn test_transaction_commit_rejects_oversized_text() {
    use wasm_dbms_api::prelude::{DbmsError, MemoryError};

    let ctx = setup();
    let tx_id = ctx.begin_transaction();
    let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);

    db.insert::<User>(user_insert_with_name_len(1, 70_000))
        .unwrap();
    let result = db.commit();
    assert!(
        matches!(
            result,
            Err(DbmsError::Memory(MemoryError::DataTooLarge { .. }))
        ),
        "expected DataTooLarge, got {result:?}"
    );

    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert!(rows.is_empty());
}

/// Table whose 256-byte name does not fit the one-byte snapshot field.
#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "tttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttttt"]
pub struct LongNameTable {
    #[primary_key]
    pub id: Uint32,
}

#[test]
fn test_register_table_rejects_table_name_above_255_bytes() {
    use wasm_dbms_api::prelude::{DbmsError, MemoryError};

    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    let pages_before = ctx.mm.borrow().pages_count();

    let result = ctx.register_table::<LongNameTable>();

    assert!(
        matches!(
            result,
            Err(DbmsError::Memory(MemoryError::ConstraintViolation(_)))
        ),
        "expected ConstraintViolation, got {result:?}"
    );
    assert!(!ctx.has_table(LongNameTable::table_name()));
    assert_eq!(ctx.mm.borrow().pages_count(), pages_before);
}

#[test]
fn test_update_no_matching_records() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[1], Value::Text(Text("bob".to_string())))],
        Some(Filter::eq("id", Value::Uint32(Uint32(999)))),
    )
    .unwrap();
    let count = db.update::<User>(patch).unwrap();
    assert_eq!(count, 0);
}

#[test]
fn test_update_request_default_all_none() {
    let patch = UserUpdateRequest::default();
    assert!(patch.id.is_none());
    assert!(patch.name.is_none());
    assert!(patch.where_clause.is_none());
    assert!(patch.update_values().is_empty());
    assert!(patch.where_clause().is_none());
}

#[test]
fn test_update_request_default_with_struct_update() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let patch = UserUpdateRequest {
        name: Some(Text("alicia".to_string())),
        where_clause: Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        ..Default::default()
    };
    let count = db.update::<User>(patch).unwrap();
    assert_eq!(count, 1);

    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows[0].name, Some(Text("alicia".to_string())));
}

// -- delete operations --

#[test]
fn test_delete_with_filter() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");

    let count = db
        .delete::<User>(
            DeleteBehavior::Restrict,
            Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        )
        .unwrap();
    assert_eq!(count, 1);

    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(2)));
}

#[test]
fn test_delete_restrict_with_fk_reference_fails() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_post(&db, 10, "post1", 1);

    let result = db.delete::<User>(DeleteBehavior::Restrict, None);
    assert!(result.is_err());
}

#[test]
fn test_delete_cascade_removes_referencing_records() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_post(&db, 10, "post1", 1);

    let count = db.delete::<User>(DeleteBehavior::Cascade, None).unwrap();
    // 1 user + 1 cascaded post
    assert_eq!(count, 2);

    let users = db.select::<User>(Query::builder().build()).unwrap();
    assert!(users.is_empty());
    let posts = db.select::<Post>(Query::builder().build()).unwrap();
    assert!(posts.is_empty());
}

// -- commit without transaction --

#[test]
fn test_commit_without_transaction_returns_error() {
    let ctx = setup();
    let mut db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    let result = db.commit();
    assert!(result.is_err());
}

// -- transaction commit with update --

#[test]
fn test_transaction_update_and_commit() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let tx_id = ctx.begin_transaction();
    let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);

    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[1], Value::Text(Text("alicia".to_string())))],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    db.update::<User>(patch).unwrap();
    db.commit().unwrap();

    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows[0].name, Some(Text("alicia".to_string())));
}

// -- transaction delete and commit --

#[test]
fn test_transaction_delete_and_commit() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");

    let tx_id = ctx.begin_transaction();
    let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);

    db.delete::<User>(
        DeleteBehavior::Restrict,
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    db.commit().unwrap();

    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(2)));
}

// -- transaction PK update then subsequent update and commit (#65) --

#[test]
fn test_transaction_pk_update_then_column_update_and_commit() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let tx_id = ctx.begin_transaction();
    let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);

    // Step 1: update PK from 1 to 10
    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[0], Value::Uint32(Uint32(10)))],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    db.update::<User>(patch).unwrap();

    // Step 2: update name on the same row (now keyed by id=10)
    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[1], Value::Text(Text("alicia".to_string())))],
        Some(Filter::eq("id", Value::Uint32(Uint32(10)))),
    )
    .unwrap();
    db.update::<User>(patch).unwrap();

    // Verify within transaction: select should return the row with both updates
    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(10)));
    assert_eq!(rows[0].name, Some(Text("alicia".to_string())));

    // Commit and verify persistence
    db.commit().unwrap();

    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(10)));
    assert_eq!(rows[0].name, Some(Text("alicia".to_string())));
}

#[test]
fn test_transaction_pk_update_then_delete() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let tx_id = ctx.begin_transaction();
    let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);

    // Step 1: update PK from 1 to 10
    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[0], Value::Uint32(Uint32(10)))],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    db.update::<User>(patch).unwrap();

    // Step 2: delete the row by new PK
    db.delete::<User>(
        DeleteBehavior::Restrict,
        Some(Filter::eq("id", Value::Uint32(Uint32(10)))),
    )
    .unwrap();

    // Verify within transaction: row should be gone
    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 0);

    // Commit and verify
    db.commit().unwrap();

    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 0);
}

// -- transaction insert/delete/reinsert of the same PK (#117) --

#[test]
fn test_transaction_insert_delete_reinsert_same_pk_is_visible_once() {
    let ctx = setup();
    let tx_id = ctx.begin_transaction();
    let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);

    insert_user(&db, 1, "first");
    let deleted = db
        .delete::<User>(
            DeleteBehavior::Restrict,
            Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        )
        .unwrap();
    assert_eq!(deleted, 1);
    insert_user(&db, 1, "second");

    // full-scan path
    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(1)));
    assert_eq!(rows[0].name, Some(Text("second".to_string())));

    // index path (PK is always indexed)
    let rows = db
        .select::<User>(
            Query::builder()
                .all()
                .and_where(Filter::eq("id", Value::Uint32(Uint32(1))))
                .build(),
        )
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].name, Some(Text("second".to_string())));

    db.commit().unwrap();

    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(1)));
    assert_eq!(rows[0].name, Some(Text("second".to_string())));
}

#[test]
fn test_transaction_delete_committed_row_then_reinsert_same_pk_is_visible_once() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "first");

    let tx_id = ctx.begin_transaction();
    let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);
    db.delete::<User>(
        DeleteBehavior::Restrict,
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    insert_user(&db, 1, "second");

    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(1)));
    assert_eq!(rows[0].name, Some(Text("second".to_string())));

    db.commit().unwrap();

    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].name, Some(Text("second".to_string())));
}

#[test]
fn test_transaction_pk_update_then_reinsert_old_pk_yields_both_rows() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let tx_id = ctx.begin_transaction();
    let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);

    // move alice from id 1 to id 2, then reuse id 1 for a new row
    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[0], Value::Uint32(Uint32(2)))],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    db.update::<User>(patch).unwrap();
    insert_user(&db, 1, "bob");

    let rows = db
        .select::<User>(Query::builder().all().order_by_asc("id").build())
        .unwrap();
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].id, Some(Uint32(1)));
    assert_eq!(rows[0].name, Some(Text("bob".to_string())));
    assert_eq!(rows[1].id, Some(Uint32(2)));
    assert_eq!(rows[1].name, Some(Text("alice".to_string())));

    db.commit().unwrap();

    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    let rows = db
        .select::<User>(Query::builder().all().order_by_asc("id").build())
        .unwrap();
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].name, Some(Text("bob".to_string())));
    assert_eq!(rows[1].name, Some(Text("alice".to_string())));
}

#[test]
fn test_transaction_insert_pk_update_then_reinsert_old_pk_yields_both_rows() {
    let ctx = setup();
    let tx_id = ctx.begin_transaction();
    let db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);

    // insert alice as id 1 inside the transaction, move her to id 2, reuse id 1
    insert_user(&db, 1, "alice");
    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[0], Value::Uint32(Uint32(2)))],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    db.update::<User>(patch).unwrap();
    insert_user(&db, 1, "bob");

    let rows = db
        .select::<User>(Query::builder().all().order_by_asc("id").build())
        .unwrap();
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].id, Some(Uint32(1)));
    assert_eq!(rows[0].name, Some(Text("bob".to_string())));
    assert_eq!(rows[1].id, Some(Uint32(2)));
    assert_eq!(rows[1].name, Some(Text("alice".to_string())));
}

// -- transactional update applies only to rows captured at staging (#118) --

#[test]
fn test_transaction_update_commit_ignores_rows_inserted_after_staging() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "old");

    let tx_id = ctx.begin_transaction();
    let mut tx_db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);
    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[1], Value::Text(Text("new".to_string())))],
        Some(Filter::eq("name", Value::Text(Text("old".to_string())))),
    )
    .unwrap();
    let staged = tx_db.update::<User>(patch).unwrap();
    assert_eq!(staged, 1);

    // another caller inserts a row matching the staged filter before commit
    insert_user(&db, 2, "old");

    tx_db.commit().unwrap();

    let rows = db
        .select::<User>(Query::builder().all().order_by_asc("id").build())
        .unwrap();
    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].id, Some(Uint32(1)));
    assert_eq!(rows[0].name, Some(Text("new".to_string())));
    assert_eq!(rows[1].id, Some(Uint32(2)));
    assert_eq!(rows[1].name, Some(Text("old".to_string())));

    let updated = db
        .select::<User>(
            Query::builder()
                .all()
                .and_where(Filter::eq("name", Value::Text(Text("new".to_string()))))
                .build(),
        )
        .unwrap();
    assert_eq!(updated.len() as u64, staged);
}

#[test]
fn test_transaction_update_commit_includes_rows_inserted_in_same_transaction() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "old");

    let tx_id = ctx.begin_transaction();
    let mut tx_db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);
    insert_user(&tx_db, 2, "old");
    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[1], Value::Text(Text("new".to_string())))],
        Some(Filter::eq("name", Value::Text(Text("old".to_string())))),
    )
    .unwrap();
    let staged = tx_db.update::<User>(patch).unwrap();
    assert_eq!(staged, 2);

    // a row inserted outside the transaction after staging must stay untouched
    insert_user(&db, 3, "old");

    tx_db.commit().unwrap();

    let rows = db
        .select::<User>(Query::builder().all().order_by_asc("id").build())
        .unwrap();
    assert_eq!(rows.len(), 3);
    assert_eq!(rows[0].name, Some(Text("new".to_string())));
    assert_eq!(rows[1].name, Some(Text("new".to_string())));
    assert_eq!(rows[2].name, Some(Text("old".to_string())));
}

#[test]
fn test_transaction_update_matching_no_rows_does_not_update_later_insertions() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

    let tx_id = ctx.begin_transaction();
    let mut tx_db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);
    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[1], Value::Text(Text("new".to_string())))],
        Some(Filter::eq("name", Value::Text(Text("old".to_string())))),
    )
    .unwrap();
    assert_eq!(tx_db.update::<User>(patch).unwrap(), 0);

    insert_user(&db, 1, "old");

    tx_db.commit().unwrap();

    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].name, Some(Text("old".to_string())));
}

#[test]
fn test_transaction_delete_commit_ignores_rows_inserted_after_staging() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "old");

    let tx_id = ctx.begin_transaction();
    let mut tx_db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);
    let staged = tx_db
        .delete::<User>(
            DeleteBehavior::Restrict,
            Some(Filter::eq("name", Value::Text(Text("old".to_string())))),
        )
        .unwrap();
    assert_eq!(staged, 1);

    insert_user(&db, 2, "old");

    tx_db.commit().unwrap();

    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(2)));
}

#[test]
fn test_transaction_update_conflicts_when_captured_primary_key_is_reused() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "original");

    let tx_id = ctx.begin_transaction();
    let mut tx_db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);
    let patch = UserUpdateRequest::from_values(
        &[(User::columns()[1], Value::Text(Text("updated".to_string())))],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    assert_eq!(tx_db.update::<User>(patch).unwrap(), 1);

    db.delete::<User>(
        DeleteBehavior::Restrict,
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    insert_user(&db, 1, "replacement");

    assert!(tx_db.commit().is_err());

    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(1)));
    assert_eq!(rows[0].name, Some(Text("replacement".to_string())));
}

#[test]
fn test_transaction_delete_conflicts_when_captured_primary_key_is_reused() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "original");

    let tx_id = ctx.begin_transaction();
    let mut tx_db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);
    assert_eq!(
        tx_db
            .delete::<User>(
                DeleteBehavior::Restrict,
                Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
            )
            .unwrap(),
        1
    );

    db.delete::<User>(
        DeleteBehavior::Restrict,
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    insert_user(&db, 1, "replacement");

    assert!(tx_db.commit().is_err());

    let rows = db.select::<User>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(1)));
    assert_eq!(rows[0].name, Some(Text("replacement".to_string())));
}

// -- select_raw --

#[test]
fn test_select_raw() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let rows = db.select_raw("users", Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0][0].1, Value::Uint32(Uint32(1)));
}

// -- select with distinct --

#[test]
fn test_select_distinct_by_single_column() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");
    insert_user(&db, 3, "alice");
    insert_user(&db, 4, "charlie");
    insert_user(&db, 5, "bob");

    let rows = db
        .select::<User>(
            Query::builder()
                .all()
                .distinct(&["name"])
                .order_by_asc("name")
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 3);
    assert_eq!(rows[0].name, Some(Text("alice".to_string())));
    assert_eq!(rows[1].name, Some(Text("bob".to_string())));
    assert_eq!(rows[2].name, Some(Text("charlie".to_string())));
}

#[test]
fn test_select_distinct_keeps_first_encountered() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "alice");
    insert_user(&db, 3, "alice");

    let rows = db
        .select::<User>(Query::builder().all().distinct(&["name"]).build())
        .unwrap();

    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(1)));
}

#[test]
fn test_select_distinct_by_multiple_columns() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");
    // Posts where (title, user_id) pairs deliberately duplicate.
    insert_post(&db, 10, "hello", 1);
    insert_post(&db, 11, "world", 1);
    insert_post(&db, 12, "hello", 2);
    insert_post(&db, 13, "hello", 1);

    let rows = db
        .select::<Post>(
            Query::builder()
                .all()
                .distinct(&["user_id"])
                .order_by_asc("id")
                .build(),
        )
        .unwrap();
    // Distinct by user_id alone => 2 rows (one per user).
    assert_eq!(rows.len(), 2);

    let rows = db
        .select::<Post>(
            Query::builder()
                .all()
                .distinct(&["title", "user_id"])
                .order_by_asc("id")
                .build(),
        )
        .unwrap();
    // Distinct by (title, user_id) => 3 unique pairs:
    // ("hello",1), ("world",1), ("hello",2). The fourth row dupes ("hello",1).
    assert_eq!(rows.len(), 3);
}

#[test]
fn test_select_distinct_with_no_duplicates() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");
    insert_user(&db, 3, "charlie");

    let rows = db
        .select::<User>(Query::builder().all().distinct(&["name"]).build())
        .unwrap();
    assert_eq!(rows.len(), 3);
}

#[test]
fn test_select_distinct_empty_distinct_by_is_noop() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "alice");
    insert_user(&db, 3, "bob");

    let empty: &[&str] = &[];
    let rows = db
        .select::<User>(Query::builder().all().distinct(empty).build())
        .unwrap();
    assert_eq!(rows.len(), 3);
}

#[test]
fn test_select_distinct_with_filter() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "alice");
    insert_user(&db, 3, "bob");
    insert_user(&db, 4, "charlie");

    let rows = db
        .select::<User>(
            Query::builder()
                .all()
                .and_where(Filter::ne("name", Value::Text(Text("charlie".to_string()))))
                .distinct(&["name"])
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 2);
    let names: Vec<_> = rows.iter().filter_map(|r| r.name.clone()).collect();
    assert!(names.contains(&Text("alice".to_string())));
    assert!(names.contains(&Text("bob".to_string())));
}

#[test]
fn test_select_distinct_with_limit_and_offset() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "alice");
    insert_user(&db, 3, "bob");
    insert_user(&db, 4, "charlie");
    insert_user(&db, 5, "dave");

    // 4 distinct names -> alice, bob, charlie, dave; offset 1, limit 2 -> bob, charlie
    let rows = db
        .select::<User>(
            Query::builder()
                .all()
                .distinct(&["name"])
                .order_by_asc("name")
                .offset(1)
                .limit(2)
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].name, Some(Text("bob".to_string())));
    assert_eq!(rows[1].name, Some(Text("charlie".to_string())));
}

#[test]
fn test_select_distinct_pagination_applies_after_dedup() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    // Many duplicates; without distinct LIMIT 2 would yield duplicates.
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "alice");
    insert_user(&db, 3, "alice");
    insert_user(&db, 4, "bob");
    insert_user(&db, 5, "bob");

    let rows = db
        .select::<User>(
            Query::builder()
                .all()
                .distinct(&["name"])
                .order_by_asc("name")
                .limit(2)
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 2);
    assert_eq!(rows[0].name, Some(Text("alice".to_string())));
    assert_eq!(rows[1].name, Some(Text("bob".to_string())));
}

#[test]
fn test_select_distinct_on_unknown_column_collapses_to_one() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");
    insert_user(&db, 3, "charlie");

    // Unknown column -> all keys are Null -> every row hashes to the same key
    let rows = db
        .select::<User>(Query::builder().all().distinct(&["nonexistent"]).build())
        .unwrap();

    assert_eq!(rows.len(), 1);
}

#[test]
fn test_select_distinct_via_select_raw() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "alice");
    insert_user(&db, 3, "bob");

    let rows = db
        .select_raw("users", Query::builder().all().distinct(&["name"]).build())
        .unwrap();
    assert_eq!(rows.len(), 2);
}

// -- select with join returns error on typed select --

#[test]
fn test_typed_select_with_join_returns_error() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

    let query = Query::builder()
        .all()
        .inner_join("posts", "id", "user_id")
        .build();
    let result = db.select::<User>(query);
    assert!(result.is_err());
}

// -- insert_index tests --

/// A table with a single-column index on `email`, used for index tests.
#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "indexed_users"]
pub struct IndexedUser {
    #[primary_key]
    pub id: Uint32,
    #[index]
    pub email: Text,
}

/// A table with a composite index on `(first_name, last_name)`.
#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "composite_users"]
pub struct CompositeUser {
    #[primary_key]
    pub id: Uint32,
    #[index(group = "idx_full_name")]
    pub first_name: Text,
    #[index(group = "idx_full_name")]
    pub last_name: Text,
}

#[derive(DatabaseSchema)]
#[tables(IndexedUser = "indexed_users", CompositeUser = "composite_users")]
pub struct IndexedTestSchema;

fn setup_indexed() -> DbmsContext<HeapMemoryProvider> {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    IndexedTestSchema::register_tables(&ctx).unwrap();
    ctx
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "name_indexed_users"]
pub struct NameIndexedUser {
    #[primary_key]
    pub id: Uint32,
    #[index]
    pub name: Text,
    pub age: Uint32,
}

#[derive(DatabaseSchema)]
#[tables(NameIndexedUser = "name_indexed_users")]
pub struct NameIndexedTestSchema;

fn setup_name_indexed() -> DbmsContext<HeapMemoryProvider> {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    NameIndexedTestSchema::register_tables(&ctx).unwrap();
    ctx
}

fn insert_name_indexed_user(
    db: &WasmDbmsDatabase<'_, HeapMemoryProvider>,
    id: u32,
    name: &str,
    age: u32,
) {
    let insert = NameIndexedUserInsertRequest::from_values(&[
        (NameIndexedUser::columns()[0], Value::Uint32(Uint32(id))),
        (
            NameIndexedUser::columns()[1],
            Value::Text(Text(name.to_string())),
        ),
        (NameIndexedUser::columns()[2], Value::Uint32(Uint32(age))),
    ])
    .unwrap();
    db.insert::<NameIndexedUser>(insert).unwrap();
}

#[test]
fn test_select_eq_on_indexed_column() {
    let ctx = setup_name_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema);
    insert_name_indexed_user(&db, 1, "alice", 20);
    insert_name_indexed_user(&db, 2, "bob", 25);
    insert_name_indexed_user(&db, 3, "alice", 30);

    let rows = db
        .select::<NameIndexedUser>(
            Query::builder()
                .all()
                .and_where(Filter::eq("name", Value::Text(Text("alice".to_string()))))
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 2);
    assert!(
        rows.iter()
            .all(|row| row.name == Some(Text("alice".to_string())))
    );
}

#[test]
fn test_select_range_on_indexed_column() {
    let ctx = setup_name_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema);
    insert_name_indexed_user(&db, 1, "alice", 20);
    insert_name_indexed_user(&db, 2, "bob", 25);
    insert_name_indexed_user(&db, 3, "charlie", 30);

    let rows = db
        .select::<NameIndexedUser>(
            Query::builder()
                .all()
                .and_where(Filter::gt("name", Value::Text(Text("alice".to_string()))))
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 2);
    assert!(
        rows.iter()
            .all(|row| row.name != Some(Text("alice".to_string())))
    );
}

/// Seeds `name_indexed_users` with the names `b`, `m`, and `z`.
fn seed_name_indexed_bounds(db: &WasmDbmsDatabase<'_, HeapMemoryProvider>) {
    insert_name_indexed_user(db, 1, "b", 20);
    insert_name_indexed_user(db, 2, "m", 25);
    insert_name_indexed_user(db, 3, "z", 30);
}

/// Selects `name_indexed_users` with `filter` and returns the sorted names.
fn select_name_indexed_names(
    db: &WasmDbmsDatabase<'_, HeapMemoryProvider>,
    filter: Filter,
) -> Vec<String> {
    let mut names: Vec<String> = db
        .select::<NameIndexedUser>(Query::builder().all().and_where(filter).build())
        .unwrap()
        .into_iter()
        .map(|row| row.name.expect("name should be selected").0)
        .collect();
    names.sort();
    names
}

fn name_text(value: &str) -> Value {
    Value::Text(Text(value.to_string()))
}

#[test]
fn test_select_indexed_and_intersects_repeated_bounds_in_both_orders() {
    let ctx = setup_name_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema);
    seed_name_indexed_bounds(&db);

    let cases: [(Filter, Filter, &[&str]); 7] = [
        (
            Filter::ge("name", name_text("b")),
            Filter::ge("name", name_text("m")),
            &["m", "z"],
        ),
        (
            Filter::gt("name", name_text("b")),
            Filter::gt("name", name_text("m")),
            &["z"],
        ),
        (
            Filter::le("name", name_text("z")),
            Filter::le("name", name_text("m")),
            &["b", "m"],
        ),
        (
            Filter::lt("name", name_text("z")),
            Filter::lt("name", name_text("m")),
            &["b"],
        ),
        (
            Filter::ge("name", name_text("m")),
            Filter::gt("name", name_text("m")),
            &["z"],
        ),
        (
            Filter::le("name", name_text("m")),
            Filter::lt("name", name_text("m")),
            &["b"],
        ),
        (
            Filter::ge("name", name_text("c")),
            Filter::le("name", name_text("y")),
            &["m"],
        ),
    ];

    for (left, right, expected) in cases {
        for filter in [
            left.clone().and(right.clone()),
            right.clone().and(left.clone()),
        ] {
            assert_eq!(
                select_name_indexed_names(&db, filter.clone()),
                expected.to_vec(),
                "unexpected rows for {filter:?}"
            );
        }
    }
}

#[test]
fn test_select_indexed_and_with_disjoint_bounds_returns_nothing() {
    let ctx = setup_name_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema);
    seed_name_indexed_bounds(&db);

    let lower = Filter::ge("name", name_text("n"));
    let upper = Filter::le("name", name_text("m"));
    for filter in [
        lower.clone().and(upper.clone()),
        upper.clone().and(lower.clone()),
    ] {
        assert!(
            select_name_indexed_names(&db, filter.clone()).is_empty(),
            "disjoint bounds must not match rows for {filter:?}"
        );
    }
}

#[test]
fn test_select_in_on_indexed_column() {
    let ctx = setup_name_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema);
    insert_name_indexed_user(&db, 1, "alice", 20);
    insert_name_indexed_user(&db, 2, "bob", 25);
    insert_name_indexed_user(&db, 3, "charlie", 30);

    let rows = db
        .select::<NameIndexedUser>(
            Query::builder()
                .all()
                .and_where(Filter::in_list(
                    "name",
                    vec![
                        Value::Text(Text("alice".to_string())),
                        Value::Text(Text("charlie".to_string())),
                    ],
                ))
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 2);
    assert!(rows.iter().all(|row| {
        matches!(
            row.name.as_ref(),
            Some(Text(name)) if name == "alice" || name == "charlie"
        )
    }));
}

#[test]
fn test_select_on_non_indexed_column_falls_back_to_scan() {
    let ctx = setup_name_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema);
    insert_name_indexed_user(&db, 1, "alice", 20);
    insert_name_indexed_user(&db, 2, "bob", 25);
    insert_name_indexed_user(&db, 3, "charlie", 30);

    let rows = db
        .select::<NameIndexedUser>(
            Query::builder()
                .all()
                .and_where(Filter::eq("age", Value::Uint32(Uint32(25))))
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].name, Some(Text("bob".to_string())));
}

#[test]
fn test_select_eq_on_indexed_column_in_transaction() {
    let ctx = setup_name_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema);
    insert_name_indexed_user(&db, 1, "alice", 20);
    insert_name_indexed_user(&db, 2, "bob", 25);

    let tx_id = ctx.begin_transaction();
    let tx_db = WasmDbmsDatabase::from_transaction(&ctx, NameIndexedTestSchema, tx_id);
    insert_name_indexed_user(&tx_db, 3, "alice", 35);

    let rows = tx_db
        .select::<NameIndexedUser>(
            Query::builder()
                .all()
                .and_where(Filter::eq("name", Value::Text(Text("alice".to_string()))))
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 2);
}

#[test]
fn test_select_eq_on_indexed_column_after_delete_and_reinsert_in_transaction() {
    let ctx = setup_name_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema);
    insert_name_indexed_user(&db, 1, "alice", 20);

    let tx_id = ctx.begin_transaction();
    let tx_db = WasmDbmsDatabase::from_transaction(&ctx, NameIndexedTestSchema, tx_id);
    let deleted = tx_db
        .delete::<NameIndexedUser>(
            DeleteBehavior::Restrict,
            Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        )
        .unwrap();
    assert_eq!(deleted, 1);
    insert_name_indexed_user(&tx_db, 2, "alice", 99);

    let rows = tx_db
        .select::<NameIndexedUser>(
            Query::builder()
                .all()
                .and_where(Filter::eq("name", Value::Text(Text("alice".to_string()))))
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(2)));
    assert_eq!(rows[0].age, Some(Uint32(99)));
}

#[test]
fn test_update_on_indexed_column_filter() {
    let ctx = setup_name_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema);
    insert_name_indexed_user(&db, 1, "alice", 20);
    insert_name_indexed_user(&db, 2, "bob", 25);
    insert_name_indexed_user(&db, 3, "alice", 30);

    let patch = NameIndexedUserUpdateRequest::from_values(
        &[(NameIndexedUser::columns()[2], Value::Uint32(Uint32(99)))],
        Some(Filter::eq("name", Value::Text(Text("alice".to_string())))),
    )
    .unwrap();
    let count = db.update::<NameIndexedUser>(patch).unwrap();
    assert_eq!(count, 2);

    let rows = db
        .select::<NameIndexedUser>(
            Query::builder()
                .all()
                .and_where(Filter::eq("name", Value::Text(Text("alice".to_string()))))
                .build(),
        )
        .unwrap();

    assert!(rows.iter().all(|row| row.age == Some(Uint32(99))));
}

#[test]
fn test_delete_on_indexed_column_filter() {
    let ctx = setup_name_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema);
    insert_name_indexed_user(&db, 1, "alice", 20);
    insert_name_indexed_user(&db, 2, "bob", 25);
    insert_name_indexed_user(&db, 3, "alice", 30);

    let count = db
        .delete::<NameIndexedUser>(
            DeleteBehavior::Restrict,
            Some(Filter::eq("name", Value::Text(Text("alice".to_string())))),
        )
        .unwrap();
    assert_eq!(count, 2);

    let rows = db
        .select::<NameIndexedUser>(Query::builder().all().build())
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].name, Some(Text("bob".to_string())));
}

#[test]
fn test_insert_index_populates_single_column_index() {
    use wasm_dbms_memory::RecordAddress;

    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    // Load the table registry and manually call insert_index.
    let mut table_registry = db.load_table_registry::<IndexedUser>().unwrap();
    let record_address = RecordAddress::new(100, 0);
    let values = vec![
        (IndexedUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            IndexedUser::columns()[1],
            Value::Text(Text("alice@example.com".to_string())),
        ),
    ];

    let mut mm = db.ctx.mm.borrow_mut();
    db.insert_index::<IndexedUser>(&mut table_registry, record_address, &values, &mut *mm)
        .unwrap();

    // Search the index for the inserted key.
    let key = vec![Value::Text(Text("alice@example.com".to_string()))];
    let results = table_registry
        .index_ledger()
        .search(&["email"], &key, &mut *mm)
        .unwrap();
    assert_eq!(results.len(), 1);
    assert_eq!(results[0], record_address);
}

#[test]
fn test_insert_index_populates_composite_index() {
    use wasm_dbms_memory::RecordAddress;

    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    let mut table_registry = db.load_table_registry::<CompositeUser>().unwrap();
    let record_address = RecordAddress::new(200, 16);
    let values = vec![
        (CompositeUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            CompositeUser::columns()[1],
            Value::Text(Text("Alice".to_string())),
        ),
        (
            CompositeUser::columns()[2],
            Value::Text(Text("Smith".to_string())),
        ),
    ];

    let mut mm = db.ctx.mm.borrow_mut();
    db.insert_index::<CompositeUser>(&mut table_registry, record_address, &values, &mut *mm)
        .unwrap();

    // Search the composite index with the correct key order.
    let key = vec![
        Value::Text(Text("Alice".to_string())),
        Value::Text(Text("Smith".to_string())),
    ];
    let results = table_registry
        .index_ledger()
        .search(&["first_name", "last_name"], &key, &mut *mm)
        .unwrap();
    assert_eq!(results.len(), 1);
    assert_eq!(results[0], record_address);
}

#[test]
fn test_insert_index_missing_column_defaults_to_null() {
    use wasm_dbms_memory::RecordAddress;

    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    let mut table_registry = db.load_table_registry::<IndexedUser>().unwrap();
    let record_address = RecordAddress::new(300, 0);
    // Provide only the PK, omit the indexed `email` column.
    let values = vec![(IndexedUser::columns()[0], Value::Uint32(Uint32(1)))];

    let mut mm = db.ctx.mm.borrow_mut();
    db.insert_index::<IndexedUser>(&mut table_registry, record_address, &values, &mut *mm)
        .unwrap();

    // The index should contain a Null key.
    let key = vec![Value::Null];
    let results = table_registry
        .index_ledger()
        .search(&["email"], &key, &mut *mm)
        .unwrap();
    assert_eq!(results.len(), 1);
    assert_eq!(results[0], record_address);
}

#[test]
fn test_insert_index_always_includes_pk_index() {
    use wasm_dbms_memory::RecordAddress;

    // User has no explicit #[index] but the PK is always auto-indexed.
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

    let mut table_registry = db.load_table_registry::<User>().unwrap();
    let record_address = RecordAddress::new(100, 0);
    let values = vec![
        (User::columns()[0], Value::Uint32(Uint32(42))),
        (User::columns()[1], Value::Text(Text("alice".to_string()))),
    ];

    let mut mm = db.ctx.mm.borrow_mut();
    db.insert_index::<User>(&mut table_registry, record_address, &values, &mut *mm)
        .unwrap();

    // The PK index should be searchable.
    let pk_key = vec![Value::Uint32(Uint32(42))];
    let results = table_registry
        .index_ledger()
        .search(&["id"], &pk_key, &mut *mm)
        .unwrap();
    assert_eq!(results.len(), 1);
    assert_eq!(results[0], record_address);
}

// -- insert flow populates index --

#[test]
fn test_insert_populates_single_column_index() {
    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    let insert = IndexedUserInsertRequest::from_values(&[
        (IndexedUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            IndexedUser::columns()[1],
            Value::Text(Text("alice@example.com".to_string())),
        ),
    ])
    .unwrap();
    db.insert::<IndexedUser>(insert).unwrap();

    // Load the table registry and verify the index was populated.
    let table_registry = db.load_table_registry::<IndexedUser>().unwrap();
    let mut mm = db.ctx.mm.borrow_mut();
    let key = vec![Value::Text(Text("alice@example.com".to_string()))];
    let results = table_registry
        .index_ledger()
        .search(&["email"], &key, &mut *mm)
        .unwrap();
    assert_eq!(results.len(), 1);
}

#[test]
fn test_insert_multiple_records_populates_index() {
    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    for (id, email) in [(1, "alice@example.com"), (2, "bob@example.com")] {
        let insert = IndexedUserInsertRequest::from_values(&[
            (IndexedUser::columns()[0], Value::Uint32(Uint32(id))),
            (
                IndexedUser::columns()[1],
                Value::Text(Text(email.to_string())),
            ),
        ])
        .unwrap();
        db.insert::<IndexedUser>(insert).unwrap();
    }

    let table_registry = db.load_table_registry::<IndexedUser>().unwrap();
    let mut mm = db.ctx.mm.borrow_mut();

    // Both entries should be individually searchable.
    let alice_key = vec![Value::Text(Text("alice@example.com".to_string()))];
    let bob_key = vec![Value::Text(Text("bob@example.com".to_string()))];
    assert_eq!(
        table_registry
            .index_ledger()
            .search(&["email"], &alice_key, &mut *mm)
            .unwrap()
            .len(),
        1
    );
    assert_eq!(
        table_registry
            .index_ledger()
            .search(&["email"], &bob_key, &mut *mm)
            .unwrap()
            .len(),
        1
    );
}

#[test]
fn test_insert_populates_composite_index() {
    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    let insert = CompositeUserInsertRequest::from_values(&[
        (CompositeUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            CompositeUser::columns()[1],
            Value::Text(Text("Alice".to_string())),
        ),
        (
            CompositeUser::columns()[2],
            Value::Text(Text("Smith".to_string())),
        ),
    ])
    .unwrap();
    db.insert::<CompositeUser>(insert).unwrap();

    let table_registry = db.load_table_registry::<CompositeUser>().unwrap();
    let mut mm = db.ctx.mm.borrow_mut();
    let key = vec![
        Value::Text(Text("Alice".to_string())),
        Value::Text(Text("Smith".to_string())),
    ];
    let results = table_registry
        .index_ledger()
        .search(&["first_name", "last_name"], &key, &mut *mm)
        .unwrap();
    assert_eq!(results.len(), 1);
}

// -- index_key tests --

#[test]
fn test_index_key_extracts_matching_columns() {
    let values = vec![
        (IndexedUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            IndexedUser::columns()[1],
            Value::Text(Text("alice@example.com".to_string())),
        ),
    ];
    let key = super::index_key(&["email"], &values);
    assert_eq!(
        key,
        vec![Value::Text(Text("alice@example.com".to_string()))]
    );
}

#[test]
fn test_index_key_missing_column_defaults_to_null() {
    let values = vec![(IndexedUser::columns()[0], Value::Uint32(Uint32(1)))];
    let key = super::index_key(&["email"], &values);
    assert_eq!(key, vec![Value::Null]);
}

#[test]
fn test_index_key_composite() {
    let values = vec![
        (CompositeUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            CompositeUser::columns()[1],
            Value::Text(Text("Alice".to_string())),
        ),
        (
            CompositeUser::columns()[2],
            Value::Text(Text("Smith".to_string())),
        ),
    ];
    let key = super::index_key(&["first_name", "last_name"], &values);
    assert_eq!(
        key,
        vec![
            Value::Text(Text("Alice".to_string())),
            Value::Text(Text("Smith".to_string())),
        ]
    );
}

// -- delete_index unit tests --

#[test]
fn test_delete_index_removes_entry() {
    use wasm_dbms_memory::RecordAddress;

    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    let mut table_registry = db.load_table_registry::<IndexedUser>().unwrap();
    let record_address = RecordAddress::new(100, 0);
    let values = vec![
        (IndexedUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            IndexedUser::columns()[1],
            Value::Text(Text("alice@example.com".to_string())),
        ),
    ];

    let mut mm = db.ctx.mm.borrow_mut();
    db.insert_index::<IndexedUser>(&mut table_registry, record_address, &values, &mut *mm)
        .unwrap();

    // Delete the index entry.
    db.delete_index::<IndexedUser>(&mut table_registry, record_address, &values, &mut *mm)
        .unwrap();

    // The index should now be empty for this key.
    let key = vec![Value::Text(Text("alice@example.com".to_string()))];
    let results = table_registry
        .index_ledger()
        .search(&["email"], &key, &mut *mm)
        .unwrap();
    assert!(results.is_empty());
}

#[test]
fn test_delete_index_removes_pk_index() {
    use wasm_dbms_memory::RecordAddress;

    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

    let mut table_registry = db.load_table_registry::<User>().unwrap();
    let record_address = RecordAddress::new(100, 0);
    let values = vec![
        (User::columns()[0], Value::Uint32(Uint32(42))),
        (User::columns()[1], Value::Text(Text("alice".to_string()))),
    ];

    let mut mm = db.ctx.mm.borrow_mut();
    db.insert_index::<User>(&mut table_registry, record_address, &values, &mut *mm)
        .unwrap();

    db.delete_index::<User>(&mut table_registry, record_address, &values, &mut *mm)
        .unwrap();

    // The PK index entry should be gone.
    let pk_key = vec![Value::Uint32(Uint32(42))];
    let results = table_registry
        .index_ledger()
        .search(&["id"], &pk_key, &mut *mm)
        .unwrap();
    assert!(results.is_empty());
}

// -- update_index unit tests --

#[test]
fn test_update_index_same_key_updates_pointer() {
    use wasm_dbms_memory::RecordAddress;

    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    let mut table_registry = db.load_table_registry::<IndexedUser>().unwrap();
    let old_address = RecordAddress::new(100, 0);
    let new_address = RecordAddress::new(200, 32);
    let values = vec![
        (IndexedUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            IndexedUser::columns()[1],
            Value::Text(Text("alice@example.com".to_string())),
        ),
    ];

    let mut mm = db.ctx.mm.borrow_mut();
    db.insert_index::<IndexedUser>(&mut table_registry, old_address, &values, &mut *mm)
        .unwrap();

    // Update with same values (only address changed).
    db.update_index::<IndexedUser>(
        &mut table_registry,
        old_address,
        new_address,
        &values,
        &values,
        &mut *mm,
    )
    .unwrap();

    // The key should now point to the new address.
    let key = vec![Value::Text(Text("alice@example.com".to_string()))];
    let results = table_registry
        .index_ledger()
        .search(&["email"], &key, &mut *mm)
        .unwrap();
    assert_eq!(results.len(), 1);
    assert_eq!(results[0], new_address);
}

#[test]
fn test_update_index_changed_key_replaces_entry() {
    use wasm_dbms_memory::RecordAddress;

    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    let mut table_registry = db.load_table_registry::<IndexedUser>().unwrap();
    let old_address = RecordAddress::new(100, 0);
    let new_address = RecordAddress::new(200, 32);
    let old_values = vec![
        (IndexedUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            IndexedUser::columns()[1],
            Value::Text(Text("old@example.com".to_string())),
        ),
    ];
    let new_values = vec![
        (IndexedUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            IndexedUser::columns()[1],
            Value::Text(Text("new@example.com".to_string())),
        ),
    ];

    let mut mm = db.ctx.mm.borrow_mut();
    db.insert_index::<IndexedUser>(&mut table_registry, old_address, &old_values, &mut *mm)
        .unwrap();

    // Update with a changed indexed column value.
    db.update_index::<IndexedUser>(
        &mut table_registry,
        old_address,
        new_address,
        &old_values,
        &new_values,
        &mut *mm,
    )
    .unwrap();

    // The old key should be gone.
    let old_key = vec![Value::Text(Text("old@example.com".to_string()))];
    let results = table_registry
        .index_ledger()
        .search(&["email"], &old_key, &mut *mm)
        .unwrap();
    assert!(results.is_empty());

    // The new key should point to the new address.
    let new_key = vec![Value::Text(Text("new@example.com".to_string()))];
    let results = table_registry
        .index_ledger()
        .search(&["email"], &new_key, &mut *mm)
        .unwrap();
    assert_eq!(results.len(), 1);
    assert_eq!(results[0], new_address);
}

// -- delete flow removes index --

#[test]
fn test_delete_removes_index_entry() {
    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    let insert = IndexedUserInsertRequest::from_values(&[
        (IndexedUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            IndexedUser::columns()[1],
            Value::Text(Text("alice@example.com".to_string())),
        ),
    ])
    .unwrap();
    db.insert::<IndexedUser>(insert).unwrap();

    db.delete::<IndexedUser>(DeleteBehavior::Restrict, None)
        .unwrap();

    let table_registry = db.load_table_registry::<IndexedUser>().unwrap();
    let mut mm = db.ctx.mm.borrow_mut();
    let key = vec![Value::Text(Text("alice@example.com".to_string()))];
    let results = table_registry
        .index_ledger()
        .search(&["email"], &key, &mut *mm)
        .unwrap();
    assert!(results.is_empty());
}

#[test]
fn test_delete_with_filter_removes_only_matching_index_entries() {
    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    for (id, email) in [(1, "alice@example.com"), (2, "bob@example.com")] {
        let insert = IndexedUserInsertRequest::from_values(&[
            (IndexedUser::columns()[0], Value::Uint32(Uint32(id))),
            (
                IndexedUser::columns()[1],
                Value::Text(Text(email.to_string())),
            ),
        ])
        .unwrap();
        db.insert::<IndexedUser>(insert).unwrap();
    }

    db.delete::<IndexedUser>(
        DeleteBehavior::Restrict,
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();

    let table_registry = db.load_table_registry::<IndexedUser>().unwrap();
    let mut mm = db.ctx.mm.borrow_mut();

    // alice's index entry should be gone.
    let alice_key = vec![Value::Text(Text("alice@example.com".to_string()))];
    assert!(
        table_registry
            .index_ledger()
            .search(&["email"], &alice_key, &mut *mm)
            .unwrap()
            .is_empty()
    );

    // bob's index entry should remain.
    let bob_key = vec![Value::Text(Text("bob@example.com".to_string()))];
    assert_eq!(
        table_registry
            .index_ledger()
            .search(&["email"], &bob_key, &mut *mm)
            .unwrap()
            .len(),
        1
    );
}

// -- update flow updates index --

#[test]
fn test_update_non_indexed_column_preserves_index() {
    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    let insert = IndexedUserInsertRequest::from_values(&[
        (IndexedUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            IndexedUser::columns()[1],
            Value::Text(Text("alice@example.com".to_string())),
        ),
    ])
    .unwrap();
    db.insert::<IndexedUser>(insert).unwrap();

    // Update a non-indexed column (id is the PK, not indexed by #[index]).
    // IndexedUser only has email indexed, so changing the PK should not affect
    // the email index. However, the PK is the primary key, not a patchable
    // column in normal usage. Instead, let's just verify that the index is
    // still searchable after an update that does not touch the indexed column.
    // We update `email` to the same value, which should preserve the index.
    let patch = IndexedUserUpdateRequest::from_values(
        &[(
            IndexedUser::columns()[1],
            Value::Text(Text("alice@example.com".to_string())),
        )],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    db.update::<IndexedUser>(patch).unwrap();

    let table_registry = db.load_table_registry::<IndexedUser>().unwrap();
    let mut mm = db.ctx.mm.borrow_mut();
    let key = vec![Value::Text(Text("alice@example.com".to_string()))];
    let results = table_registry
        .index_ledger()
        .search(&["email"], &key, &mut *mm)
        .unwrap();
    assert_eq!(results.len(), 1);
}

#[test]
fn test_update_indexed_column_updates_index() {
    let ctx = setup_indexed();
    let db = WasmDbmsDatabase::oneshot(&ctx, IndexedTestSchema);

    let insert = IndexedUserInsertRequest::from_values(&[
        (IndexedUser::columns()[0], Value::Uint32(Uint32(1))),
        (
            IndexedUser::columns()[1],
            Value::Text(Text("old@example.com".to_string())),
        ),
    ])
    .unwrap();
    db.insert::<IndexedUser>(insert).unwrap();

    // Update the indexed column to a new value.
    let patch = IndexedUserUpdateRequest::from_values(
        &[(
            IndexedUser::columns()[1],
            Value::Text(Text("new@example.com".to_string())),
        )],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    db.update::<IndexedUser>(patch).unwrap();

    let table_registry = db.load_table_registry::<IndexedUser>().unwrap();
    let mut mm = db.ctx.mm.borrow_mut();

    // The old key should no longer be in the index.
    let old_key = vec![Value::Text(Text("old@example.com".to_string()))];
    assert!(
        table_registry
            .index_ledger()
            .search(&["email"], &old_key, &mut *mm)
            .unwrap()
            .is_empty()
    );

    // The new key should be present.
    let new_key = vec![Value::Text(Text("new@example.com".to_string()))];
    assert_eq!(
        table_registry
            .index_ledger()
            .search(&["email"], &new_key, &mut *mm)
            .unwrap()
            .len(),
        1
    );
}

#[test]
fn test_contract_should_have_unique_code() {
    let columns = Contract::columns();
    let code_column = columns
        .iter()
        .find(|col| col.name == "code")
        .expect("code column");
    assert!(
        code_column.unique,
        "Contract.code should be marked as unique"
    );
    // check primary key
    let pk_column = columns
        .iter()
        .find(|col| col.name == "id")
        .expect("id column");
    assert!(
        pk_column.primary_key,
        "Contract.id should be marked as primary key"
    );
    assert!(pk_column.unique, "Contract.id should be unique");
    // check user id
    let user_id_column = columns
        .iter()
        .find(|col| col.name == "user_id")
        .expect("user_id column");
    assert!(
        !user_id_column.unique,
        "Contract.user_id should not be unique"
    );

    // check indexes
    let indexes = Contract::indexes();
    // code column must be indexed
    indexes
        .iter()
        .find(|idx| idx.columns() == ["code"])
        .expect("index on code column");
    // pk must be index
    indexes
        .iter()
        .find(|idx| idx.columns() == ["id"])
        .expect("index on id column");
}

// -- unique constraint tests --

#[test]
fn test_insert_contract_with_unique_code_succeeds() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_contract(&db, 1, "CONTRACT-001", 1);
    insert_contract(&db, 2, "CONTRACT-002", 1);

    let rows = db.select::<Contract>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 2);
}

#[test]
fn test_insert_contract_with_duplicate_code_fails() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_contract(&db, 1, "CONTRACT-001", 1);

    let insert = ContractInsertRequest::from_values(&[
        (Contract::columns()[0], Value::Uint32(Uint32(2))),
        (
            Contract::columns()[1],
            Value::Text(Text("CONTRACT-001".to_string())),
        ),
        // columns()[2] is `order` (autoincrement) — omitted
        (Contract::columns()[3], Value::Uint32(Uint32(1))),
    ])
    .unwrap();
    let result = db.insert::<Contract>(insert);
    assert!(matches!(
        result,
        Err(wasm_dbms_api::prelude::DbmsError::Query(
            wasm_dbms_api::prelude::QueryError::UniqueConstraintViolation { .. }
        ))
    ));
}

#[test]
fn test_update_contract_code_to_unique_value_succeeds() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_contract(&db, 1, "CONTRACT-001", 1);

    let patch = ContractUpdateRequest::from_values(
        &[(
            Contract::columns()[1],
            Value::Text(Text("CONTRACT-999".to_string())),
        )],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    db.update::<Contract>(patch).unwrap();

    let rows = db
        .select::<Contract>(
            Query::builder()
                .and_where(Filter::Eq("id".to_string(), Value::Uint32(Uint32(1))))
                .build(),
        )
        .unwrap();
    assert_eq!(rows[0].code, Some(Text("CONTRACT-999".to_string())));
}

#[test]
fn test_update_contract_keeping_same_code_succeeds() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_user(&db, 2, "bob");
    insert_contract(&db, 1, "CONTRACT-001", 1);

    // Change user_id but keep the same unique code
    let patch = ContractUpdateRequest::from_values(
        &[(Contract::columns()[2], Value::Uint32(Uint32(2)))],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    db.update::<Contract>(patch).unwrap();
}

#[test]
fn test_update_contract_code_to_existing_value_fails() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_contract(&db, 1, "CONTRACT-001", 1);
    insert_contract(&db, 2, "CONTRACT-002", 1);

    let patch = ContractUpdateRequest::from_values(
        &[(
            Contract::columns()[1],
            Value::Text(Text("CONTRACT-001".to_string())),
        )],
        Some(Filter::eq("id", Value::Uint32(Uint32(2)))),
    )
    .unwrap();
    let result = db.update::<Contract>(patch);
    assert!(matches!(
        result,
        Err(wasm_dbms_api::prelude::DbmsError::Query(
            wasm_dbms_api::prelude::QueryError::UniqueConstraintViolation { .. }
        ))
    ));
}

#[test]
fn test_unique_constraint_with_transaction_commit() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_contract(&db, 1, "CONTRACT-001", 1);

    // Insert a second contract within a transaction
    let tx_id = ctx.begin_transaction();
    let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);
    insert_contract(&db, 2, "CONTRACT-002", 1);
    db.commit().unwrap();

    // After commit, inserting a duplicate should fail
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    let insert = ContractInsertRequest::from_values(&[
        (Contract::columns()[0], Value::Uint32(Uint32(3))),
        (
            Contract::columns()[1],
            Value::Text(Text("CONTRACT-002".to_string())),
        ),
        // columns()[2] is `order` (autoincrement) — omitted
        (Contract::columns()[3], Value::Uint32(Uint32(1))),
    ])
    .unwrap();
    assert!(matches!(
        db.insert::<Contract>(insert),
        Err(wasm_dbms_api::prelude::DbmsError::Query(
            wasm_dbms_api::prelude::QueryError::UniqueConstraintViolation { .. }
        ))
    ));
}

#[test]
fn test_unique_constraint_after_delete_allows_reuse() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");
    insert_contract(&db, 1, "CONTRACT-001", 1);

    // Delete the contract
    db.delete::<Contract>(
        DeleteBehavior::Restrict,
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();

    // Now inserting a new contract with the same code should succeed
    insert_contract(&db, 2, "CONTRACT-001", 1);
}

// -- autoincrement tests --

#[test]
fn test_autoincrement_auto_generates_sequential_values() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    insert_contract(&db, 1, "C-001", 1);
    insert_contract(&db, 2, "C-002", 1);
    insert_contract(&db, 3, "C-003", 1);

    let rows = db
        .select::<Contract>(Query::builder().order_by_asc("id").build())
        .unwrap();

    assert_eq!(rows.len(), 3);
    assert_eq!(rows[0].order, Some(Uint32(1)));
    assert_eq!(rows[1].order, Some(Uint32(2)));
    assert_eq!(rows[2].order, Some(Uint32(3)));
}

#[test]
fn test_autoincrement_explicit_value_overrides_auto() {
    use wasm_dbms_api::prelude::Autoincrement;

    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    // Insert with an explicit autoincrement value
    let insert = ContractInsertRequest {
        id: Uint32(1),
        code: Text("C-001".to_string()),
        order: Autoincrement::Value(Uint32(42)),
        user_id: Uint32(1),
    };
    db.insert::<Contract>(insert).unwrap();

    let rows = db.select::<Contract>(Query::builder().build()).unwrap();

    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].order, Some(Uint32(42)));
}

#[test]
fn test_autoincrement_does_not_recycle_after_delete() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    insert_contract(&db, 1, "C-001", 1); // order = 1
    insert_contract(&db, 2, "C-002", 1); // order = 2

    // Delete the first contract
    db.delete::<Contract>(
        DeleteBehavior::Restrict,
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();

    // Next insert should get order = 3, not 1
    insert_contract(&db, 3, "C-003", 1);

    let rows = db
        .select::<Contract>(
            Query::builder()
                .and_where(Filter::eq("id", Value::Uint32(Uint32(3))))
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].order, Some(Uint32(3)));
}

#[test]
fn test_autoincrement_with_transaction_commit() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    // Insert outside transaction
    insert_contract(&db, 1, "C-001", 1); // order = 1

    // Insert inside transaction
    let tx_id = ctx.begin_transaction();
    let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);
    insert_contract(&db, 2, "C-002", 1); // order = 2
    db.commit().unwrap();

    // After commit, next auto value should be 3
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_contract(&db, 3, "C-003", 1);

    let rows = db
        .select::<Contract>(Query::builder().order_by_asc("id").build())
        .unwrap();

    assert_eq!(rows.len(), 3);
    assert_eq!(rows[0].order, Some(Uint32(1)));
    assert_eq!(rows[1].order, Some(Uint32(2)));
    assert_eq!(rows[2].order, Some(Uint32(3)));
}

#[test]
fn test_autoincrement_with_transaction_rollback_does_not_revert_counter() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    insert_contract(&db, 1, "C-001", 1); // order = 1

    // Insert inside transaction then rollback
    let tx_id = ctx.begin_transaction();
    let mut db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);
    insert_contract(&db, 2, "C-002", 1); // order = 2 (consumed)
    db.rollback().unwrap();

    // After rollback, counter should still have advanced (order = 2 is consumed)
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_contract(&db, 3, "C-003", 1);

    let rows = db
        .select::<Contract>(Query::builder().order_by_asc("id").build())
        .unwrap();

    assert_eq!(rows.len(), 2); // only C-001 and C-003
    assert_eq!(rows[0].order, Some(Uint32(1)));
    assert_eq!(rows[1].order, Some(Uint32(3))); // 2 was consumed by the rolled-back tx
}

#[test]
fn test_autoincrement_from_values_with_auto_variant() {
    use wasm_dbms_api::prelude::Autoincrement;

    // from_values without the autoincrement column should produce Auto
    let insert = ContractInsertRequest::from_values(&[
        (Contract::columns()[0], Value::Uint32(Uint32(1))),
        (
            Contract::columns()[1],
            Value::Text(Text("C-001".to_string())),
        ),
        (Contract::columns()[3], Value::Uint32(Uint32(1))),
    ])
    .unwrap();

    assert_eq!(insert.order, Autoincrement::Auto);
}

#[test]
fn test_autoincrement_from_values_with_value_variant() {
    use wasm_dbms_api::prelude::Autoincrement;

    // from_values with the autoincrement column should produce Value
    let insert = ContractInsertRequest::from_values(&[
        (Contract::columns()[0], Value::Uint32(Uint32(1))),
        (
            Contract::columns()[1],
            Value::Text(Text("C-001".to_string())),
        ),
        (Contract::columns()[2], Value::Uint32(Uint32(99))),
        (Contract::columns()[3], Value::Uint32(Uint32(1))),
    ])
    .unwrap();

    assert_eq!(insert.order, Autoincrement::Value(Uint32(99)));
}

#[test]
fn test_autoincrement_into_values_skips_auto() {
    use wasm_dbms_api::prelude::{Autoincrement, InsertRecord as _};

    let insert = ContractInsertRequest {
        id: Uint32(1),
        code: Text("C-001".to_string()),
        order: Autoincrement::Auto,
        user_id: Uint32(1),
    };
    let values = insert.into_values();

    // Should have 3 values (id, code, user_id) — order is skipped
    assert_eq!(values.len(), 3);
    assert!(values.iter().all(|(col, _)| col.name != "order"));
}

#[test]
fn test_autoincrement_into_values_includes_explicit_value() {
    use wasm_dbms_api::prelude::{Autoincrement, InsertRecord as _};

    let insert = ContractInsertRequest {
        id: Uint32(1),
        code: Text("C-001".to_string()),
        order: Autoincrement::Value(Uint32(42)),
        user_id: Uint32(1),
    };
    let values = insert.into_values();

    // Should have 4 values (id, code, order, user_id)
    assert_eq!(values.len(), 4);
    let order_val = values.iter().find(|(col, _)| col.name == "order");
    assert!(order_val.is_some());
    assert_eq!(order_val.unwrap().1, Value::Uint32(Uint32(42)));
}

#[test]
fn test_autoincrement_select_filter_on_autoincrement_column() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    insert_contract(&db, 1, "C-001", 1); // order = 1
    insert_contract(&db, 2, "C-002", 1); // order = 2
    insert_contract(&db, 3, "C-003", 1); // order = 3

    let rows = db
        .select::<Contract>(
            Query::builder()
                .and_where(Filter::eq("order", Value::Uint32(Uint32(2))))
                .build(),
        )
        .unwrap();

    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(2)));
    assert_eq!(rows[0].order, Some(Uint32(2)));
}

// -- aggregate tests --

mod aggregate_tests {
    use rust_decimal::Decimal as RustDecimal;
    use wasm_dbms_api::prelude::{
        AggregateFunction, AggregatedValue, Database as _, Decimal, Filter, OrderDirection, Query,
        Text, Uint32, Uint64, Value,
    };

    use super::{Post, TestSchema, User, insert_post, insert_user, setup};
    use crate::prelude::WasmDbmsDatabase;

    fn seed(db: &WasmDbmsDatabase<'_, wasm_dbms_memory::prelude::HeapMemoryProvider>) {
        insert_user(db, 1, "alice");
        insert_user(db, 2, "bob");
        insert_user(db, 3, "carol");
        // Posts: alice has 3, bob has 1, carol has 0.
        insert_post(db, 10, "p1", 1);
        insert_post(db, 20, "p2", 1);
        insert_post(db, 30, "p3", 1);
        insert_post(db, 40, "p4", 2);
    }

    fn dec(s: &str) -> Value {
        Value::Decimal(Decimal(RustDecimal::from_str_exact(s).unwrap()))
    }

    #[test]
    fn aggregate_count_all_no_group_by() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let result = db
            .aggregate::<Post>(Query::default(), &[AggregateFunction::Count(None)])
            .unwrap();

        assert_eq!(result.len(), 1);
        assert!(result[0].group_keys.is_empty());
        assert_eq!(result[0].values, vec![AggregatedValue::Count(4)]);
    }

    #[test]
    fn aggregate_count_column_skips_nulls() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        // user_id is non-null on every Post, so COUNT(user_id) == COUNT(*).
        let result = db
            .aggregate::<Post>(
                Query::default(),
                &[AggregateFunction::Count(Some("user_id".into()))],
            )
            .unwrap();

        assert_eq!(result[0].values, vec![AggregatedValue::Count(4)]);
    }

    #[test]
    fn aggregate_sum_avg_min_max_no_group_by() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let aggs = vec![
            AggregateFunction::Sum("user_id".into()),
            AggregateFunction::Avg("user_id".into()),
            AggregateFunction::Min("user_id".into()),
            AggregateFunction::Max("user_id".into()),
        ];

        let result = db.aggregate::<Post>(Query::default(), &aggs).unwrap();
        assert_eq!(result.len(), 1);
        let v = &result[0].values;
        // sum(1+1+1+2) = 5, avg = 5/4 = 1.25
        assert_eq!(v[0], AggregatedValue::Sum(dec("5")));
        assert_eq!(v[1], AggregatedValue::Avg(dec("1.25")));
        assert_eq!(v[2], AggregatedValue::Min(Value::Uint32(Uint32(1))));
        assert_eq!(v[3], AggregatedValue::Max(Value::Uint32(Uint32(2))));
    }

    #[test]
    fn aggregate_group_by_single_column() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let query = Query::builder().group_by(&["user_id"]).build();
        let aggs = vec![AggregateFunction::Count(None)];

        let mut result = db.aggregate::<Post>(query, &aggs).unwrap();
        // Sort by group key for deterministic assertion.
        result.sort_by(|a, b| a.group_keys[0].cmp(&b.group_keys[0]));

        assert_eq!(result.len(), 2);
        assert_eq!(result[0].group_keys, vec![Value::Uint32(Uint32(1))]);
        assert_eq!(result[0].values, vec![AggregatedValue::Count(3)]);
        assert_eq!(result[1].group_keys, vec![Value::Uint32(Uint32(2))]);
        assert_eq!(result[1].values, vec![AggregatedValue::Count(1)]);
    }

    #[test]
    fn aggregate_having_filters_groups() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        // GROUP BY user_id, HAVING agg0 > 1 — keep only alice's group.
        let query = Query::builder()
            .group_by(&["user_id"])
            .having(Filter::gt("agg0", Value::Uint64(Uint64(1))))
            .build();
        let aggs = vec![AggregateFunction::Count(None)];

        let result = db.aggregate::<Post>(query, &aggs).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].group_keys, vec![Value::Uint32(Uint32(1))]);
        assert_eq!(result[0].values, vec![AggregatedValue::Count(3)]);
    }

    #[test]
    fn aggregate_having_on_group_key_column() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let query = Query::builder()
            .group_by(&["user_id"])
            .having(Filter::eq("user_id", Value::Uint32(Uint32(2))))
            .build();
        let aggs = vec![AggregateFunction::Count(None)];

        let result = db.aggregate::<Post>(query, &aggs).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].group_keys, vec![Value::Uint32(Uint32(2))]);
    }

    #[test]
    fn aggregate_order_by_aggregate_output_then_limit() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let query = Query::builder()
            .group_by(&["user_id"])
            .order_by_desc("agg0")
            .limit(1)
            .build();
        let aggs = vec![AggregateFunction::Count(None)];

        let result = db.aggregate::<Post>(query, &aggs).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].group_keys, vec![Value::Uint32(Uint32(1))]);
        assert_eq!(result[0].values, vec![AggregatedValue::Count(3)]);
    }

    #[test]
    fn aggregate_offset_skips_rows() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let query = Query::builder()
            .group_by(&["user_id"])
            .order_by_asc("user_id")
            .offset(1)
            .build();
        let aggs = vec![AggregateFunction::Count(None)];

        let result = db.aggregate::<Post>(query, &aggs).unwrap();
        assert_eq!(result.len(), 1);
        assert_eq!(result[0].group_keys, vec![Value::Uint32(Uint32(2))]);
    }

    #[test]
    fn aggregate_with_where_pre_filters_rows() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let query = Query::builder()
            .and_where(Filter::eq("user_id", Value::Uint32(Uint32(1))))
            .build();
        let aggs = vec![AggregateFunction::Count(None)];

        let result = db.aggregate::<Post>(query, &aggs).unwrap();
        assert_eq!(result[0].values, vec![AggregatedValue::Count(3)]);
    }

    #[test]
    fn aggregate_empty_table_no_group_by() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        // No inserts.

        let aggs = vec![
            AggregateFunction::Count(None),
            AggregateFunction::Sum("user_id".into()),
            AggregateFunction::Avg("user_id".into()),
            AggregateFunction::Min("user_id".into()),
            AggregateFunction::Max("user_id".into()),
        ];

        let result = db.aggregate::<Post>(Query::default(), &aggs).unwrap();
        assert_eq!(result.len(), 1);
        let v = &result[0].values;
        assert_eq!(v[0], AggregatedValue::Count(0));
        assert_eq!(v[1], AggregatedValue::Sum(Value::Null));
        assert_eq!(v[2], AggregatedValue::Avg(Value::Null));
        assert_eq!(v[3], AggregatedValue::Min(Value::Null));
        assert_eq!(v[4], AggregatedValue::Max(Value::Null));
    }

    #[test]
    fn aggregate_empty_table_group_by_returns_no_rows() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

        let query = Query::builder().group_by(&["user_id"]).build();
        let aggs = vec![AggregateFunction::Count(None)];

        let result = db.aggregate::<Post>(query, &aggs).unwrap();
        assert!(result.is_empty());
    }

    #[test]
    fn aggregate_sum_on_non_numeric_column_errors() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let aggs = vec![AggregateFunction::Sum("name".into())];
        let err = db
            .aggregate::<User>(Query::default(), &aggs)
            .expect_err("SUM on Text must be rejected");
        let msg = err.to_string();
        assert!(msg.contains("numeric"), "unexpected error: {msg}");
    }

    #[test]
    fn aggregate_avg_on_non_numeric_column_errors() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let aggs = vec![AggregateFunction::Avg("name".into())];
        let err = db
            .aggregate::<User>(Query::default(), &aggs)
            .expect_err("AVG on Text must be rejected");
        assert!(err.to_string().contains("numeric"));
    }

    #[test]
    fn aggregate_unknown_column_errors() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let aggs = vec![AggregateFunction::Sum("not_a_column".into())];
        let err = db
            .aggregate::<User>(Query::default(), &aggs)
            .expect_err("unknown column must be rejected");
        assert!(err.to_string().contains("not_a_column"));
    }

    #[test]
    fn aggregate_group_by_unknown_column_errors() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

        let query = Query::builder().group_by(&["nope"]).build();
        let err = db
            .aggregate::<Post>(query, &[AggregateFunction::Count(None)])
            .expect_err("unknown group_by column must be rejected");
        assert!(err.to_string().contains("nope"));
    }

    #[test]
    fn aggregate_having_unknown_aggregate_errors() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        // Only one aggregate (agg0); reference agg1.
        let query = Query::builder()
            .group_by(&["user_id"])
            .having(Filter::gt("agg1", Value::Uint64(Uint64(0))))
            .build();
        let err = db
            .aggregate::<Post>(query, &[AggregateFunction::Count(None)])
            .expect_err("HAVING on unknown agg must be rejected");
        assert!(err.to_string().contains("agg1"));
    }

    #[test]
    fn aggregate_having_unknown_column_errors() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let query = Query::builder()
            .group_by(&["user_id"])
            .having(Filter::eq("foo", Value::Uint32(Uint32(0))))
            .build();
        let err = db
            .aggregate::<Post>(query, &[AggregateFunction::Count(None)])
            .expect_err("HAVING on unknown column must be rejected");
        assert!(err.to_string().contains("foo"));
    }

    #[test]
    fn aggregate_order_by_unknown_aggregate_errors() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let query = Query::builder()
            .group_by(&["user_id"])
            .order_by_asc("agg7")
            .build();
        let err = db
            .aggregate::<Post>(query, &[AggregateFunction::Count(None)])
            .expect_err("ORDER BY on unknown agg must be rejected");
        assert!(err.to_string().contains("agg7"));
    }

    #[test]
    fn aggregate_having_like_rejected() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        seed(&db);

        let query = Query::builder()
            .group_by(&["user_id"])
            .having(Filter::like("user_id", "%"))
            .build();
        let err = db
            .aggregate::<Post>(query, &[AggregateFunction::Count(None)])
            .expect_err("LIKE in HAVING must be rejected");
        assert!(err.to_string().contains("LIKE"));
    }

    #[test]
    fn aggregate_avg_decimal_division_precision() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        // Insert posts so SUM(user_id) / 3 = 7/3 = 2.333...
        insert_user(&db, 1, "a");
        insert_user(&db, 2, "b");
        insert_user(&db, 4, "c");
        insert_post(&db, 1, "p", 1);
        insert_post(&db, 2, "p", 2);
        insert_post(&db, 3, "p", 4);

        let aggs = vec![AggregateFunction::Avg("user_id".into())];
        let result = db.aggregate::<Post>(Query::default(), &aggs).unwrap();
        let AggregatedValue::Avg(Value::Decimal(d)) = &result[0].values[0] else {
            panic!("expected Decimal avg");
        };
        // 7 / 3 truncated to RustDecimal precision should round-trip via string.
        assert!(d.0.to_string().starts_with("2.33"));
    }

    #[test]
    fn aggregate_join_query_rejected() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

        let query = Query::builder()
            .inner_join("users", "user_id", "id")
            .build();
        let err = db
            .aggregate::<Post>(query, &[AggregateFunction::Count(None)])
            .expect_err("joins must be rejected");
        assert!(err.to_string().contains("join"));
    }

    #[test]
    fn aggregate_eager_relations_rejected() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

        let query = Query::builder().with("users").build();
        let err = db
            .aggregate::<Post>(query, &[AggregateFunction::Count(None)])
            .expect_err("eager relations must be rejected");
        assert!(err.to_string().contains("eager"));
    }

    #[test]
    fn aggregate_orders_by_descending() {
        let _ = OrderDirection::Descending; // silence unused import warning if any
    }

    // -- extra coverage: multi-col group_by, NULLs, Decimal AVG, transactions,
    //    combined order, JSON-in-HAVING rejection --

    use wasm_dbms_api::prelude::{InsertRecord as _, JsonFilter, Nullable, TableSchema as _};

    use super::{Sale, SaleInsertRequest};

    fn insert_sale(
        db: &WasmDbmsDatabase<'_, wasm_dbms_memory::prelude::HeapMemoryProvider>,
        id: u32,
        category: &str,
        price: u32,
        bonus: Option<u32>,
    ) {
        let bonus: Nullable<Uint32> = bonus.map(Uint32).into();
        let req = SaleInsertRequest::from_values(&[
            (Sale::columns()[0], Value::Uint32(Uint32(id))),
            (Sale::columns()[1], Value::Text(Text(category.to_string()))),
            (Sale::columns()[2], Value::Uint32(Uint32(price))),
            (Sale::columns()[3], bonus.into()),
        ])
        .unwrap();
        db.insert::<Sale>(req).unwrap();
    }

    #[test]
    fn aggregate_multi_column_group_by() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_sale(&db, 1, "books", 10, None);
        insert_sale(&db, 2, "books", 20, None);
        insert_sale(&db, 3, "toys", 5, None);
        insert_sale(&db, 4, "toys", 5, None);
        insert_sale(&db, 5, "books", 10, None);

        let query = Query::builder()
            .group_by(&["category", "price"])
            .order_by_asc("category")
            .order_by_asc("price")
            .build();
        let result = db
            .aggregate::<Sale>(query, &[AggregateFunction::Count(None)])
            .unwrap();

        // Groups: (books, 10) x2, (books, 20) x1, (toys, 5) x2
        assert_eq!(result.len(), 3);
        assert_eq!(
            result[0].group_keys,
            vec![Value::Text(Text("books".into())), Value::Uint32(Uint32(10))]
        );
        assert_eq!(result[0].values, vec![AggregatedValue::Count(2)]);
        assert_eq!(result[1].values, vec![AggregatedValue::Count(1)]);
        assert_eq!(result[2].values, vec![AggregatedValue::Count(2)]);
    }

    #[test]
    fn aggregate_count_column_with_actual_nulls() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_sale(&db, 1, "a", 1, Some(10));
        insert_sale(&db, 2, "a", 1, None);
        insert_sale(&db, 3, "a", 1, Some(20));

        let result = db
            .aggregate::<Sale>(
                Query::default(),
                &[
                    AggregateFunction::Count(None),
                    AggregateFunction::Count(Some("bonus".into())),
                ],
            )
            .unwrap();
        let v = &result[0].values;
        assert_eq!(v[0], AggregatedValue::Count(3));
        assert_eq!(v[1], AggregatedValue::Count(2));
    }

    #[test]
    fn aggregate_sum_avg_skips_nulls_decimal_output() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        // Use the nullable `bonus` column so SUM/AVG must skip nulls.
        insert_sale(&db, 1, "x", 1, Some(10));
        insert_sale(&db, 2, "x", 1, None);
        insert_sale(&db, 3, "x", 1, Some(20));

        let result = db
            .aggregate::<Sale>(
                Query::default(),
                &[
                    AggregateFunction::Sum("bonus".into()),
                    AggregateFunction::Avg("bonus".into()),
                ],
            )
            .unwrap();
        assert_eq!(result[0].values[0], AggregatedValue::Sum(dec("30")));
        // avg = 30 / 2 = 15
        assert_eq!(result[0].values[1], AggregatedValue::Avg(dec("15")));
    }

    #[test]
    fn aggregate_inside_transaction_sees_writes() {
        let ctx = setup();
        let tx_id = ctx.begin_transaction();
        let db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx_id);

        // Inside the transaction, insert rows then aggregate.
        insert_user(&db, 1, "a");
        insert_user(&db, 2, "b");
        insert_post(&db, 10, "p", 1);
        insert_post(&db, 11, "p", 1);
        insert_post(&db, 12, "p", 2);

        let result = db
            .aggregate::<Post>(
                Query::builder().group_by(&["user_id"]).build(),
                &[AggregateFunction::Count(None)],
            )
            .unwrap();
        // 2 distinct user_ids in the overlay
        assert_eq!(result.len(), 2);
    }

    #[test]
    fn aggregate_order_by_combines_group_key_and_aggregate() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
        insert_user(&db, 1, "a");
        insert_user(&db, 2, "b");
        insert_user(&db, 3, "c");
        insert_post(&db, 10, "p", 1);
        insert_post(&db, 20, "p", 2);
        insert_post(&db, 21, "p", 2);
        insert_post(&db, 30, "p", 3);

        // Order by COUNT desc then user_id asc.
        let query = Query::builder()
            .group_by(&["user_id"])
            .order_by_desc("agg0")
            .order_by_asc("user_id")
            .build();
        let result = db
            .aggregate::<Post>(query, &[AggregateFunction::Count(None)])
            .unwrap();

        assert_eq!(result.len(), 3);
        // user 2 has 2 posts (highest); users 1 and 3 each have 1, tie broken by user_id asc.
        assert_eq!(result[0].group_keys, vec![Value::Uint32(Uint32(2))]);
        assert_eq!(result[1].group_keys, vec![Value::Uint32(Uint32(1))]);
        assert_eq!(result[2].group_keys, vec![Value::Uint32(Uint32(3))]);
    }

    #[test]
    fn aggregate_having_json_filter_rejected() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

        let json_filter = JsonFilter::extract_eq("$.foo", Value::Text(Text("bar".into())));
        let query = Query::builder()
            .group_by(&["user_id"])
            .having(Filter::json("user_id", json_filter))
            .build();
        let err = db
            .aggregate::<Post>(query, &[AggregateFunction::Count(None)])
            .expect_err("JSON in HAVING must be rejected");
        assert!(err.to_string().contains("JSON"));
    }
}

#[test]
fn select_with_group_by_is_rejected() {
    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);
    insert_user(&db, 1, "alice");

    let query = Query::builder().group_by(&["id"]).build();
    let err = db
        .select::<User>(query)
        .expect_err("GROUP BY must be rejected on select");
    assert!(err.to_string().contains("GROUP BY"));
}

#[test]
fn select_with_having_is_rejected() {
    use wasm_dbms_api::prelude::Filter as F;

    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

    let query = Query::builder()
        .having(F::gt("agg0", Value::Uint32(Uint32(0))))
        .build();
    let err = db
        .select::<User>(query)
        .expect_err("HAVING must be rejected on select");
    assert!(err.to_string().contains("GROUP BY"));
}

mod migration_e2e {
    use std::borrow::Cow;

    use wasm_dbms_api::prelude::{
        ColumnSnapshot, DataTypeSnapshot, Database as _, DbmsError, Encode as _, IndexSnapshot,
        InsertRecord as _, MigrationError, MigrationOp, MigrationPolicy, Query, TableSchema as _,
        Text, Uint32, Value,
    };
    use wasm_dbms_macros::{DatabaseSchema, Table};
    use wasm_dbms_memory::MemoryAccess;
    use wasm_dbms_memory::prelude::HeapMemoryProvider;

    use crate::database::migration::snapshots;
    use crate::prelude::{DbmsContext, WasmDbmsDatabase};
    use crate::schema::DatabaseSchema;

    #[derive(Debug, Table, Clone, PartialEq, Eq)]
    #[table = "users"]
    pub struct User {
        #[primary_key]
        pub id: Uint32,
        pub name: Text,
    }

    #[derive(DatabaseSchema)]
    #[tables(User = "users")]
    pub struct UserSchema;

    #[derive(Debug, Table, Clone, PartialEq, Eq)]
    #[table = "users"]
    #[migrate]
    pub struct UserV2 {
        #[primary_key]
        pub id: Uint32,
        pub name: Text,
        pub email: Text,
    }

    impl wasm_dbms_api::prelude::Migrate for UserV2 {
        fn default_value(column: &str) -> Option<Value> {
            match column {
                "email" => Some(Value::Text(Text("dynamic@example.com".to_string()))),
                _ => None,
            }
        }
    }

    #[derive(DatabaseSchema)]
    #[tables(UserV2 = "users")]
    pub struct UserSchemaV2;

    fn setup() -> DbmsContext<HeapMemoryProvider> {
        let ctx = DbmsContext::new(HeapMemoryProvider::default());
        UserSchema::register_tables(&ctx).unwrap();
        ctx
    }

    fn tamper_snapshot_to_force_drift(ctx: &DbmsContext<HeapMemoryProvider>) {
        let snapshot_page = {
            let sr = ctx.schema_registry.borrow();
            sr.table_registry_page::<User>()
                .unwrap()
                .schema_snapshot_page
        };
        let mut tampered = User::schema_snapshot();
        tampered.indexes.push(IndexSnapshot {
            columns: vec!["name".to_string()],
            unique: false,
        });
        ctx.mm
            .borrow_mut()
            .write_at(snapshot_page, 0, &tampered)
            .unwrap();
        let mut schema_registry = ctx.schema_registry.borrow_mut();
        let mut mm = ctx.mm.borrow_mut();
        schema_registry.refresh_schema_hash(&mut *mm).unwrap();
        schema_registry.save(&mut *mm).unwrap();
        ctx.clear_drift();
    }

    #[test]
    fn test_has_drift_returns_false_for_matching_schemas() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);
        assert!(!db.has_drift().unwrap());
    }

    #[test]
    fn test_database_creation_eagerly_caches_drift_flag() {
        let ctx = setup();
        let compiled_hash = snapshots::compute_hash(<UserSchema as DatabaseSchema<
            HeapMemoryProvider,
        >>::compiled_snapshots());
        assert_eq!(ctx.cached_drift_for(compiled_hash), None);
        let _db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);
        assert_eq!(ctx.cached_drift_for(compiled_hash), Some(false));
    }

    #[test]
    fn test_has_drift_returns_true_after_persisted_snapshot_diverges() {
        let ctx = setup();
        tamper_snapshot_to_force_drift(&ctx);
        let db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);
        assert!(db.has_drift().unwrap());
    }

    #[test]
    fn test_pending_migrations_is_empty_when_schemas_match() {
        let ctx = setup();
        let db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);
        let ops = db.pending_migrations().unwrap();
        assert!(ops.is_empty());
    }

    #[test]
    fn test_pending_migrations_produces_drop_index_for_extra_persisted_index() {
        let ctx = setup();
        tamper_snapshot_to_force_drift(&ctx);
        let db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);
        let ops = db.pending_migrations().unwrap();
        assert_eq!(ops.len(), 1);
        assert!(matches!(
            &ops[0],
            MigrationOp::DropIndex { table, index }
                if table == "users" && index.columns == vec!["name".to_string()]
        ));
    }

    #[test]
    fn test_migrate_clears_drift_and_persists_compiled_snapshot() {
        let ctx = setup();
        tamper_snapshot_to_force_drift(&ctx);
        let mut db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);

        assert!(db.has_drift().unwrap());
        db.migrate(MigrationPolicy::default()).unwrap();
        assert!(!db.has_drift().unwrap());

        let stored = ctx
            .schema_registry
            .borrow()
            .stored_snapshots(&mut *ctx.mm.borrow_mut())
            .unwrap();
        let users = stored.iter().find(|s| s.name == "users").unwrap();
        assert!(
            !users
                .indexes
                .iter()
                .any(|i| i.columns == vec!["name".to_string()]),
            "extra index should be removed by the migration"
        );
    }

    #[test]
    fn test_select_returns_schema_drift_error_while_drift_is_active() {
        let ctx = setup();
        tamper_snapshot_to_force_drift(&ctx);
        let db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);

        let result = db.select::<User>(Query::builder().build());
        assert!(matches!(
            result,
            Err(DbmsError::Migration(MigrationError::SchemaDrift))
        ));
    }

    #[test]
    fn test_insert_returns_schema_drift_error_while_drift_is_active() {
        let ctx = setup();
        tamper_snapshot_to_force_drift(&ctx);
        let db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);

        let insert = UserInsertRequest::from_values(&[
            (User::columns()[0], Value::Uint32(Uint32(1))),
            (User::columns()[1], Value::Text(Text("alice".to_string()))),
        ])
        .unwrap();
        let result = db.insert::<User>(insert);
        assert!(matches!(
            result,
            Err(DbmsError::Migration(MigrationError::SchemaDrift))
        ));
    }

    #[test]
    fn test_crud_resumes_after_successful_migration() {
        let ctx = setup();

        {
            let pre_db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);
            let insert = UserInsertRequest::from_values(&[
                (User::columns()[0], Value::Uint32(Uint32(1))),
                (User::columns()[1], Value::Text(Text("alice".to_string()))),
            ])
            .unwrap();
            pre_db.insert::<User>(insert).unwrap();
        }

        tamper_snapshot_to_force_drift(&ctx);

        let mut db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);
        let blocked = db.select::<User>(Query::builder().build());
        assert!(matches!(
            blocked,
            Err(DbmsError::Migration(MigrationError::SchemaDrift))
        ));

        db.migrate(MigrationPolicy::default()).unwrap();

        let rows = db.select::<User>(Query::builder().build()).unwrap();
        assert_eq!(rows.len(), 1);
        assert_eq!(rows[0].id, Some(Uint32(1)));
        assert_eq!(rows[0].name, Some(Text("alice".to_string())));
    }

    #[test]
    fn test_migrate_rejects_destructive_drop_table_under_default_policy() {
        let ctx = setup();
        let stale_snapshot = wasm_dbms_api::prelude::TableSchemaSnapshot {
            version: wasm_dbms_api::prelude::TableSchemaSnapshot::latest_version(),
            name: "stale_table".to_string(),
            primary_key: "id".to_string(),
            alignment: 8,
            columns: vec![ColumnSnapshot {
                name: "id".to_string(),
                data_type: DataTypeSnapshot::Uint32,
                nullable: false,
                auto_increment: false,
                unique: true,
                primary_key: true,
                foreign_key: None,
                default: None,
            }],
            indexes: vec![],
        };
        ctx.schema_registry
            .borrow_mut()
            .register_table_from_snapshot(&stale_snapshot, &mut *ctx.mm.borrow_mut())
            .unwrap();
        ctx.clear_drift();

        let mut db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);
        assert!(db.has_drift().unwrap());

        let result = db.migrate(MigrationPolicy::default());
        assert!(matches!(
            result,
            Err(DbmsError::Migration(MigrationError::DestructiveOpDenied { ref op })) if op.contains("DropTable")
        ));

        let after = db.select::<User>(Query::builder().build());
        assert!(matches!(
            after,
            Err(DbmsError::Migration(MigrationError::SchemaDrift))
        ));
    }

    #[test]
    fn test_migrate_with_destructive_policy_drops_stale_table() {
        let ctx = setup();
        let stale_snapshot = wasm_dbms_api::prelude::TableSchemaSnapshot {
            version: wasm_dbms_api::prelude::TableSchemaSnapshot::latest_version(),
            name: "stale_table".to_string(),
            primary_key: "id".to_string(),
            alignment: 8,
            columns: vec![ColumnSnapshot {
                name: "id".to_string(),
                data_type: DataTypeSnapshot::Uint32,
                nullable: false,
                auto_increment: false,
                unique: true,
                primary_key: true,
                foreign_key: None,
                default: None,
            }],
            indexes: vec![],
        };
        ctx.schema_registry
            .borrow_mut()
            .register_table_from_snapshot(&stale_snapshot, &mut *ctx.mm.borrow_mut())
            .unwrap();
        ctx.clear_drift();

        let mut db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);
        db.migrate(MigrationPolicy {
            allow_destructive: true,
        })
        .unwrap();

        let stored = ctx
            .schema_registry
            .borrow()
            .stored_snapshots(&mut *ctx.mm.borrow_mut())
            .unwrap();
        assert!(!stored.iter().any(|s| s.name == "stale_table"));
        assert!(stored.iter().any(|s| s.name == "users"));
    }

    #[test]
    fn test_migrate_accepts_dynamic_default_for_non_nullable_added_column() {
        let ctx = setup();
        {
            let db = WasmDbmsDatabase::oneshot(&ctx, UserSchema);
            let insert = UserInsertRequest::from_values(&[
                (User::columns()[0], Value::Uint32(Uint32(1))),
                (User::columns()[1], Value::Text(Text("alice".to_string()))),
            ])
            .unwrap();
            db.insert::<User>(insert).unwrap();
        }

        let mut db = WasmDbmsDatabase::oneshot(&ctx, UserSchemaV2);
        assert!(db.has_drift().unwrap());
        db.migrate(MigrationPolicy::default()).unwrap();

        let snapshot = {
            let mut mm = ctx.mm.borrow_mut();
            ctx.schema_registry
                .borrow()
                .stored_snapshots(&mut *mm)
                .unwrap()
                .into_iter()
                .find(|snapshot| snapshot.name == "users")
                .unwrap()
        };
        let pages = ctx
            .schema_registry
            .borrow()
            .table_registry_page_by_name("users")
            .unwrap();
        let raw_body = {
            let mut mm = ctx.mm.borrow_mut();
            let registry = wasm_dbms_memory::TableRegistry::load(pages, &mut *mm).unwrap();
            let mut reader = registry.iter_raw(snapshot.alignment as u16, &mut *mm);
            let raw = reader.try_next().unwrap().expect("migrated row");
            raw.bytes
        };
        let decoded_snapshot =
            crate::database::migration::codec::decode_record_by_snapshot(&raw_body, &snapshot)
                .unwrap();
        assert_eq!(decoded_snapshot.len(), 3);
        let decoded_row = UserV2::decode(Cow::Borrowed(&raw_body)).unwrap();
        assert_eq!(decoded_row.email, Text("dynamic@example.com".to_string()));
        {
            let mut mm = ctx.mm.borrow_mut();
            let registry = wasm_dbms_memory::TableRegistry::load(pages, &mut *mm).unwrap();
            let mut reader = registry.read::<UserV2, _>(&mut *mm);
            let first = reader.try_next().unwrap().expect("typed migrated row");
            assert_eq!(first.record.email, Text("dynamic@example.com".to_string()));
            let second = reader.try_next();
            assert!(
                second.is_ok(),
                "typed reader should not error after migrated row"
            );
            assert!(
                second.unwrap().is_none(),
                "typed reader should stop after one row"
            );
        }
        let rows = db.select_raw("users", Query::builder().build()).unwrap();
        assert_eq!(rows.len(), 1);
        let email = rows[0]
            .iter()
            .find(|(column, _)| column.name == "email")
            .map(|(_, value)| value)
            .expect("email column should be present after migration");
        assert_eq!(email, &Value::Text(Text("dynamic@example.com".to_string())));
    }
}

#[test]
fn select_join_with_group_by_is_rejected() {
    use wasm_dbms_api::prelude::Database as _;

    let ctx = setup();
    let db = WasmDbmsDatabase::oneshot(&ctx, TestSchema);

    let query = Query::builder()
        .inner_join("posts", "id", "user_id")
        .group_by(&["id"])
        .build();
    let err = db
        .select_join("users", query)
        .expect_err("GROUP BY must be rejected on select_join");
    assert!(err.to_string().contains("GROUP BY"));
}

use wasm_dbms_api::prelude::Decimal;

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "invoices"]
pub struct Invoice {
    #[primary_key]
    pub id: Uint32,
    pub discount: Nullable<Decimal>,
}

#[derive(DatabaseSchema)]
#[tables(Invoice = "invoices")]
pub struct InvoiceTestSchema;

#[test]
fn nullable_decimal_column_round_trips_value_and_null() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    InvoiceTestSchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, InvoiceTestSchema);

    let discount = Decimal(rust_decimal::Decimal::new(1250, 2));
    for (id, discount) in [(1, Nullable::Value(discount)), (2, Nullable::Null)] {
        let insert = InvoiceInsertRequest::from_values(&[
            (Invoice::columns()[0], Value::Uint32(Uint32(id))),
            (Invoice::columns()[1], discount.into()),
        ])
        .unwrap();
        db.insert::<Invoice>(insert).unwrap();
    }

    let rows = db
        .select_raw("invoices", Query::builder().order_by_asc("id").build())
        .unwrap();
    let discounts: Vec<&Value> = rows
        .iter()
        .map(|row| {
            row.iter()
                .find(|(column, _)| column.name == "discount")
                .map(|(_, value)| value)
                .expect("discount column should be present")
        })
        .collect();
    assert_eq!(discounts, vec![&Value::Decimal(discount), &Value::Null]);
}

use wasm_dbms_api::prelude::{ClampSanitizer, Int32, RoundToScaleSanitizer};

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "payments"]
pub struct Payment {
    #[primary_key]
    pub id: Uint32,
    #[sanitizer(ClampSanitizer, min = 0, max = 100)]
    pub score: Int32,
    #[sanitizer(RoundToScaleSanitizer(2))]
    pub amount: Decimal,
}

#[derive(DatabaseSchema)]
#[tables(Payment = "payments")]
pub struct PaymentTestSchema;

#[test]
fn tuple_and_named_sanitizer_attributes_apply_on_insert() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    PaymentTestSchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, PaymentTestSchema);

    let insert = PaymentInsertRequest::from_values(&[
        (Payment::columns()[0], Value::Uint32(Uint32(1))),
        (Payment::columns()[1], Value::Int32(Int32(250))),
        (
            Payment::columns()[2],
            Value::Decimal(Decimal(rust_decimal::Decimal::new(12_346, 3))),
        ),
    ])
    .unwrap();
    db.insert::<Payment>(insert).unwrap();

    let rows = db.select::<Payment>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].score, Some(Int32(100)));
    assert_eq!(
        rows[0].amount,
        Some(Decimal(rust_decimal::Decimal::new(1_235, 2)))
    );
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "employees"]
pub struct Employee {
    #[primary_key]
    pub id: Uint32,
    pub name: Text,
    #[foreign_key(entity = "Employee", table = "employees", column = "id")]
    pub manager_id: Nullable<Uint32>,
}

#[derive(DatabaseSchema)]
#[tables(Employee = "employees")]
pub struct EmployeeTestSchema;

fn insert_employee(
    db: &WasmDbmsDatabase<'_, HeapMemoryProvider>,
    id: u32,
    name: &str,
    manager_id: Option<u32>,
) -> wasm_dbms_api::prelude::DbmsResult<()> {
    let manager_id: Nullable<Uint32> = manager_id.map(Uint32).into();
    let insert = EmployeeInsertRequest::from_values(&[
        (Employee::columns()[0], Value::Uint32(Uint32(id))),
        (Employee::columns()[1], Value::Text(Text(name.to_string()))),
        (Employee::columns()[2], manager_id.into()),
    ])
    .unwrap();
    db.insert::<Employee>(insert)
}

#[test]
fn nullable_foreign_key_accepts_null_and_eager_loads_present_relation() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    EmployeeTestSchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, EmployeeTestSchema);

    insert_employee(&db, 1, "boss", None).unwrap();
    insert_employee(&db, 2, "worker", Some(1)).unwrap();
    let err = insert_employee(&db, 3, "orphan", Some(99)).unwrap_err();
    assert!(matches!(
        err,
        wasm_dbms_api::prelude::DbmsError::Query(
            wasm_dbms_api::prelude::QueryError::BrokenForeignKeyReference { .. }
        )
    ));

    let rows = db
        .select_raw("employees", Query::builder().order_by_asc("id").build())
        .unwrap();
    let managers: Vec<&Value> = rows
        .iter()
        .map(|row| {
            row.iter()
                .find(|(column, _)| column.name == "manager_id")
                .map(|(_, value)| value)
                .expect("manager_id column should be present")
        })
        .collect();
    assert_eq!(managers, vec![&Value::Null, &Value::Uint32(Uint32(1))]);

    let employees = db
        .select::<Employee>(
            Query::builder()
                .all()
                .with("employees")
                .order_by_asc("id")
                .build(),
        )
        .unwrap();
    assert_eq!(employees.len(), 2);
    assert_eq!(employees[0].manager_id, None);
    let manager = employees[1]
        .manager_id
        .as_ref()
        .expect("manager should be eager loaded");
    assert_eq!(manager.id, Some(Uint32(1)));
    assert_eq!(manager.name, Some(Text("boss".to_string())));
}

fn update_manager(
    db: &WasmDbmsDatabase<'_, HeapMemoryProvider>,
    id: u32,
    manager_id: Value,
) -> wasm_dbms_api::prelude::DbmsResult<u64> {
    let patch = EmployeeUpdateRequest::from_values(
        &[(Employee::columns()[2], manager_id)],
        Some(Filter::eq("id", Value::Uint32(Uint32(id)))),
    )
    .unwrap();
    db.update::<Employee>(patch)
}

fn manager_of(db: &WasmDbmsDatabase<'_, HeapMemoryProvider>, id: u32) -> Value {
    let rows = db
        .select_raw(
            "employees",
            Query::builder()
                .filter(Some(Filter::eq("id", Value::Uint32(Uint32(id)))))
                .build(),
        )
        .unwrap();
    rows[0]
        .iter()
        .find(|(column, _)| column.name == "manager_id")
        .map(|(_, value)| value.clone())
        .expect("manager_id column should be present")
}

#[test]
fn nullable_foreign_key_update_skips_existence_check_for_null() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    EmployeeTestSchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, EmployeeTestSchema);

    insert_employee(&db, 1, "boss", None).unwrap();
    insert_employee(&db, 2, "worker", Some(1)).unwrap();

    // set a present relation to null
    assert_eq!(update_manager(&db, 2, Value::Null).unwrap(), 1);
    assert_eq!(manager_of(&db, 2), Value::Null);

    // set null back to an existing relation
    assert_eq!(update_manager(&db, 2, Value::Uint32(Uint32(1))).unwrap(), 1);
    assert_eq!(manager_of(&db, 2), Value::Uint32(Uint32(1)));

    // a non-null reference to a missing record is still rejected
    let err = update_manager(&db, 2, Value::Uint32(Uint32(99))).unwrap_err();
    assert!(matches!(
        err,
        wasm_dbms_api::prelude::DbmsError::Query(
            wasm_dbms_api::prelude::QueryError::BrokenForeignKeyReference { .. }
        )
    ));
    assert_eq!(manager_of(&db, 2), Value::Uint32(Uint32(1)));
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "qualified_types"]
pub struct QualifiedTypes {
    #[primary_key]
    #[autoincrement]
    pub id: wasm_dbms_api::prelude::Uint32,
    pub name: ::wasm_dbms_api::prelude::Text,
    pub nickname: wasm_dbms_api::prelude::Nullable<wasm_dbms_api::prelude::Text>,
    pub score: Nullable<wasm_dbms_api::prelude::Uint32>,
}

#[derive(DatabaseSchema)]
#[tables(QualifiedTypes = "qualified_types")]
pub struct QualifiedTypesTestSchema;

#[test]
fn qualified_field_types_map_to_built_in_columns() {
    use wasm_dbms_api::prelude::DataTypeKind;

    let kinds: Vec<(&str, DataTypeKind, bool)> = QualifiedTypes::columns()
        .iter()
        .map(|column| (column.name, column.data_type, column.nullable))
        .collect();
    assert_eq!(
        kinds,
        vec![
            ("id", DataTypeKind::Uint32, false),
            ("name", DataTypeKind::Text, false),
            ("nickname", DataTypeKind::Text, true),
            ("score", DataTypeKind::Uint32, true),
        ]
    );

    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    QualifiedTypesTestSchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, QualifiedTypesTestSchema);

    let insert = QualifiedTypesInsertRequest::from_values(&[
        (
            QualifiedTypes::columns()[1],
            Value::Text(Text("alice".to_string())),
        ),
        (QualifiedTypes::columns()[2], Value::Null),
        (QualifiedTypes::columns()[3], Value::Uint32(Uint32(7))),
    ])
    .unwrap();
    db.insert::<QualifiedTypes>(insert).unwrap();

    let rows = db
        .select::<QualifiedTypes>(Query::builder().build())
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert!(rows[0].id.is_some());
    assert_eq!(rows[0].name, Some(Text("alice".to_string())));
    assert_eq!(rows[0].nickname, Some(Nullable::Null));
    assert_eq!(rows[0].score, Some(Nullable::Value(Uint32(7))));
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "private_rows"]
struct PrivateRow {
    #[primary_key]
    id: Uint32,
    label: Text,
}

#[derive(DatabaseSchema)]
#[tables(PrivateRow = "private_rows")]
struct PrivateRowTestSchema;

#[test]
fn private_table_struct_round_trips_through_insert_and_select() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    PrivateRowTestSchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, PrivateRowTestSchema);

    db.insert::<PrivateRow>(PrivateRowInsertRequest {
        id: Uint32(1),
        label: Text("hidden".to_string()),
    })
    .unwrap();

    let rows = db.select::<PrivateRow>(Query::builder().build()).unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(1)));
    assert_eq!(rows[0].label, Some(Text("hidden".to_string())));
}

/// Custom data type used to instantiate the generic table fixtures.
#[derive(
    Debug,
    Clone,
    Default,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    serde::Serialize,
    serde::Deserialize,
    wasm_dbms_macros::Encode,
    wasm_dbms_macros::CustomDataType,
)]
#[type_tag = "priority"]
pub struct Priority {
    level: wasm_dbms_api::prelude::Uint8,
}

impl std::fmt::Display for Priority {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "priority {level}", level = self.level.0)
    }
}

impl wasm_dbms_api::prelude::DataType for Priority {}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "generic_rows"]
pub struct GenericRow<T>
where
    T: wasm_dbms_api::prelude::CustomDataType,
{
    #[primary_key]
    pub id: Uint32,
    #[custom_type]
    pub payload: T,
}

#[derive(DatabaseSchema)]
#[tables(GenericRow<T> = "generic_rows")]
pub struct GenericRowTestSchema<T>
where
    T: wasm_dbms_api::prelude::CustomDataType,
{
    _payload: std::marker::PhantomData<T>,
}

impl<T> GenericRowTestSchema<T>
where
    T: wasm_dbms_api::prelude::CustomDataType,
{
    fn new() -> Self {
        Self {
            _payload: std::marker::PhantomData,
        }
    }
}

#[test]
fn generic_table_and_schema_round_trip_through_insert_and_select() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    GenericRowTestSchema::<Priority>::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, GenericRowTestSchema::<Priority>::new());

    let payload = Priority {
        level: wasm_dbms_api::prelude::Uint8(3),
    };
    db.insert::<GenericRow<Priority>>(GenericRowInsertRequest {
        id: Uint32(1),
        payload: payload.clone(),
    })
    .unwrap();

    let rows = db
        .select::<GenericRow<Priority>>(Query::builder().build())
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0].id, Some(Uint32(1)));
    assert_eq!(rows[0].payload, Some(payload.clone()));

    // dynamic dispatch through the generic schema reaches the same table
    let raw = db
        .select_raw("generic_rows", Query::builder().build())
        .unwrap();
    assert_eq!(raw.len(), 1);
    assert!(
        raw[0]
            .iter()
            .any(|(column, value)| column.name == "payload"
                && *value == Value::from(payload.clone()))
    );
}

/// Table whose column names match identifiers used inside the generated code.
#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "shadowing_rows"]
pub struct ShadowingRow {
    #[primary_key]
    pub id: Uint32,
    pub data: Text,
    pub values: Text,
    pub offset: Uint32,
    pub column: Nullable<Text>,
    pub this_record_values: Text,
    #[custom_type]
    pub cv: Priority,
    #[custom_type]
    pub decoded: Nullable<Priority>,
    #[foreign_key(entity = "User", table = "users", column = "id")]
    pub has_fk_values: Uint32,
}

#[derive(DatabaseSchema)]
#[tables(User = "users", ShadowingRow = "shadowing_rows")]
pub struct ShadowingTestSchema;

/// `Encode` struct whose field names match the generated decoder locals.
#[derive(Debug, Clone, PartialEq, Eq, wasm_dbms_macros::Encode)]
pub struct EncodedRow {
    data: Text,
    offset: Uint32,
    number: Uint32,
}

#[test]
fn encode_derive_round_trips_fields_named_like_decoder_locals() {
    use wasm_dbms_api::prelude::Encode as _;

    let row = EncodedRow {
        data: Text("payload".to_string()),
        offset: Uint32(7),
        number: Uint32(42),
    };
    let encoded = row.encode();
    assert_eq!(EncodedRow::decode(encoded).unwrap(), row);
}

#[test]
fn table_fields_named_like_generated_locals_round_trip() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    ShadowingTestSchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, ShadowingTestSchema);
    insert_user(&db, 1, "owner");

    let priority = |level: u8| Priority {
        level: wasm_dbms_api::prelude::Uint8(level),
    };
    let columns = ShadowingRow::columns();
    let insert = ShadowingRowInsertRequest::from_values(&[
        (columns[0], Value::Uint32(Uint32(1))),
        (columns[1], Value::Text(Text("data".to_string()))),
        (columns[2], Value::Text(Text("values".to_string()))),
        (columns[3], Value::Uint32(Uint32(3))),
        (columns[4], Value::Text(Text("column".to_string()))),
        (columns[5], Value::Text(Text("this".to_string()))),
        (columns[6], Value::from(priority(1))),
        (columns[7], Value::from(priority(2))),
        (columns[8], Value::Uint32(Uint32(1))),
    ])
    .unwrap();
    db.insert::<ShadowingRow>(insert).unwrap();

    let patch = ShadowingRowUpdateRequest::from_values(
        &[
            (columns[1], Value::Text(Text("new data".to_string()))),
            (columns[2], Value::Text(Text("new values".to_string()))),
            (columns[4], Value::Null),
            (columns[6], Value::from(priority(5))),
        ],
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    )
    .unwrap();
    assert_eq!(db.update::<ShadowingRow>(patch).unwrap(), 1);

    let rows = db
        .select::<ShadowingRow>(Query::builder().all().with("users").build())
        .unwrap();
    assert_eq!(rows.len(), 1);
    let row = &rows[0];
    assert_eq!(row.data, Some(Text("new data".to_string())));
    assert_eq!(row.values, Some(Text("new values".to_string())));
    assert_eq!(row.offset, Some(Uint32(3)));
    assert_eq!(row.column, Some(Nullable::Null));
    assert_eq!(row.this_record_values, Some(Text("this".to_string())));
    assert_eq!(row.cv, Some(priority(5)));
    assert_eq!(row.decoded, Some(Nullable::Value(priority(2))));
    let owner = row
        .has_fk_values
        .as_ref()
        .expect("owner should be eager loaded");
    assert_eq!(owner.name, Some(Text("owner".to_string())));
}

/// Named struct without fields.
#[derive(Debug, Clone, PartialEq, Eq, wasm_dbms_macros::Encode)]
pub struct EmptyNamed {}

/// Unit struct.
#[derive(Debug, Clone, PartialEq, Eq, wasm_dbms_macros::Encode)]
pub struct EmptyUnit;

/// Struct embedding a zero-size field next to a fixed-size one.
#[derive(Debug, Clone, PartialEq, Eq, wasm_dbms_macros::Encode)]
pub struct WithEmptyField {
    empty: EmptyNamed,
    number: Uint32,
}

#[test]
fn encode_derive_supports_empty_structs() {
    use wasm_dbms_api::prelude::{DataSize, Encode};
    use wasm_dbms_memory::align_up;

    assert_eq!(<EmptyNamed as Encode>::SIZE, DataSize::Fixed(0));
    assert_eq!(<EmptyNamed as Encode>::ALIGNMENT, 1);
    assert_eq!(<EmptyUnit as Encode>::SIZE, DataSize::Fixed(0));
    assert_eq!(<EmptyUnit as Encode>::ALIGNMENT, 1);
    // a non-zero alignment keeps page math well defined
    assert_eq!(align_up::<EmptyNamed>(5), 5);

    let named = EmptyNamed {};
    assert_eq!(named.size(), 0);
    let encoded = named.encode();
    assert!(encoded.is_empty());
    assert_eq!(EmptyNamed::decode(encoded).unwrap(), named);

    let unit = EmptyUnit;
    assert_eq!(unit.size(), 0);
    assert_eq!(EmptyUnit::decode(unit.encode()).unwrap(), unit);

    let nested = WithEmptyField {
        empty: EmptyNamed {},
        number: Uint32(9),
    };
    assert_eq!(<WithEmptyField as Encode>::SIZE, DataSize::Fixed(4));
    assert_eq!(<WithEmptyField as Encode>::ALIGNMENT, 4);
    assert_eq!(WithEmptyField::decode(nested.encode()).unwrap(), nested);
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "key_users"]
pub struct KeyUser {
    #[primary_key]
    pub id: Uint32,
    #[unique]
    pub code: Uint32,
    pub nickname: Nullable<Text>,
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "key_posts"]
pub struct KeyPost {
    #[primary_key]
    pub id: Uint32,
    #[foreign_key(entity = "KeyUser", table = "key_users", column = "code")]
    pub owner: Uint32,
}

#[derive(DatabaseSchema)]
#[tables(KeyUser = "key_users", KeyPost = "key_posts")]
pub struct KeySchema;

fn key_user_code(db: &WasmDbmsDatabase<'_, HeapMemoryProvider>) -> Option<Uint32> {
    db.select::<KeyUser>(Query::builder().build()).unwrap()[0].code
}

#[test]
fn dynamic_update_rejects_a_wrong_typed_field() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    KeySchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, KeySchema);
    db.insert::<KeyUser>(KeyUserInsertRequest {
        id: 1.into(),
        code: 10.into(),
        nickname: Nullable::Null,
    })
    .unwrap();

    let result = KeySchema.update(
        &db,
        "key_users",
        &[(KeyUser::columns()[1], Value::Text("wrong type".into()))],
        None,
    );
    assert!(matches!(
        result,
        Err(wasm_dbms_api::prelude::DbmsError::Query(
            wasm_dbms_api::prelude::QueryError::InvalidQuery(_)
        ))
    ));
    assert_eq!(key_user_code(&db), Some(Uint32(10)));

    // null is only accepted by nullable columns
    let result = KeySchema.update(
        &db,
        "key_users",
        &[(KeyUser::columns()[1], Value::Null)],
        None,
    );
    assert!(result.is_err());
    assert_eq!(key_user_code(&db), Some(Uint32(10)));
    let result = KeySchema.update(
        &db,
        "key_users",
        &[
            (KeyUser::columns()[1], Value::Uint32(Uint32(11))),
            (KeyUser::columns()[2], Value::Null),
        ],
        None,
    );
    assert_eq!(result.unwrap(), 1);
    assert_eq!(key_user_code(&db), Some(Uint32(11)));
}

#[test]
fn dynamic_update_rejects_a_custom_value_of_another_type() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    GenericRowTestSchema::<Priority>::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, GenericRowTestSchema::<Priority>::new());
    let payload = Priority {
        level: wasm_dbms_api::prelude::Uint8(1),
    };
    db.insert::<GenericRow<Priority>>(GenericRowInsertRequest {
        id: Uint32(1),
        payload: payload.clone(),
    })
    .unwrap();

    let columns = GenericRow::<Priority>::columns();
    let wrong_tag = Value::Custom(wasm_dbms_api::prelude::CustomValue {
        type_tag: "not_a_priority".to_string(),
        encoded: vec![7],
        display: "7".to_string(),
    });
    for wrong in [wrong_tag, Value::Uint32(Uint32(7))] {
        let result = GenericRowTestSchema::<Priority>::new().update(
            &db,
            "generic_rows",
            &[(columns[1], wrong)],
            None,
        );
        assert!(result.is_err());
    }

    let rows = db
        .select::<GenericRow<Priority>>(Query::builder().build())
        .unwrap();
    assert_eq!(rows[0].payload, Some(payload));
}

#[test]
fn foreign_key_uses_the_declared_referenced_column() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    KeySchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, KeySchema);
    db.insert::<KeyUser>(KeyUserInsertRequest {
        id: 1.into(),
        code: 10.into(),
        nickname: Nullable::Null,
    })
    .unwrap();

    assert!(
        db.insert::<KeyPost>(KeyPostInsertRequest {
            id: 1.into(),
            owner: 10.into(),
        })
        .is_ok()
    );
    assert!(
        db.insert::<KeyPost>(KeyPostInsertRequest {
            id: 2.into(),
            owner: 1.into(),
        })
        .is_err()
    );

    // eager loading resolves the relation through `code`
    let posts = db
        .select::<KeyPost>(Query::builder().all().with("key_users").build())
        .unwrap();
    assert_eq!(posts.len(), 1);
    let owner = posts[0]
        .owner
        .as_ref()
        .expect("owner should be eager loaded");
    assert_eq!(owner.id, Some(Uint32(1)));
    assert_eq!(owner.code, Some(Uint32(10)));
}

#[test]
fn updating_a_referenced_non_primary_column_rewrites_the_references() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    KeySchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, KeySchema);
    db.insert::<KeyUser>(KeyUserInsertRequest {
        id: 1.into(),
        code: 10.into(),
        nickname: Nullable::Null,
    })
    .unwrap();
    db.insert::<KeyPost>(KeyPostInsertRequest {
        id: 1.into(),
        owner: 10.into(),
    })
    .unwrap();

    let patch = KeyUserUpdateRequest {
        code: Some(Uint32(99)),
        where_clause: Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        ..Default::default()
    };
    // the user and the post referencing its code
    assert_eq!(db.update::<KeyUser>(patch).unwrap(), 2);

    let posts = db
        .select::<KeyPost>(Query::builder().all().with("key_users").build())
        .expect("references must stay valid");
    let owner = posts[0]
        .owner
        .as_ref()
        .expect("owner should be eager loaded");
    assert_eq!(owner.code, Some(Uint32(99)));

    // an update leaving the referenced column unchanged does not touch the references
    let patch = KeyUserUpdateRequest {
        nickname: Some(Nullable::Value("nick".into())),
        where_clause: Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        ..Default::default()
    };
    assert_eq!(db.update::<KeyUser>(patch).unwrap(), 1);
}

#[test]
fn foreign_key_on_a_non_primary_column_guards_deletes_and_primary_key_updates() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    KeySchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, KeySchema);
    for (id, code) in [(1, 10), (10, 20)] {
        db.insert::<KeyUser>(KeyUserInsertRequest {
            id: id.into(),
            code: code.into(),
            nickname: Nullable::Null,
        })
        .unwrap();
    }
    db.insert::<KeyPost>(KeyPostInsertRequest {
        id: 1.into(),
        owner: 10.into(),
    })
    .unwrap();

    // user 1 (code 10) is referenced by the post: restrict must reject the delete
    let restricted = db.delete::<KeyUser>(
        DeleteBehavior::Restrict,
        Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
    );
    assert!(restricted.is_err());

    // changing the primary key of user 10 (code 20) must not rewrite the post, whose owner
    // value 10 is a `code`, not a primary key
    let patch = KeyUserUpdateRequest {
        id: Some(Uint32(7)),
        where_clause: Some(Filter::eq("id", Value::Uint32(Uint32(10)))),
        ..Default::default()
    };
    assert_eq!(db.update::<KeyUser>(patch).unwrap(), 1);
    let raw = db
        .select_raw("key_posts", Query::builder().build())
        .unwrap();
    assert!(
        raw[0]
            .iter()
            .any(|(column, value)| column.name == "owner" && *value == Value::Uint32(Uint32(10)))
    );

    // user 7 (code 20) is not referenced
    assert_eq!(
        db.delete::<KeyUser>(
            DeleteBehavior::Restrict,
            Some(Filter::eq("id", Value::Uint32(Uint32(7)))),
        )
        .unwrap(),
        1
    );

    // cascade removes the post referencing code 10
    assert_eq!(
        db.delete::<KeyUser>(
            DeleteBehavior::Cascade,
            Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        )
        .unwrap(),
        2
    );
    assert!(
        db.select::<KeyPost>(Query::builder().build())
            .unwrap()
            .is_empty()
    );
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "validated_names"]
pub struct ValidatedName {
    #[primary_key]
    #[autoincrement]
    pub id: Uint32,
    #[validate(wasm_dbms_api::prelude::MinStrlenValidator(2))]
    #[validate(wasm_dbms_api::prelude::MaxStrlenValidator(20))]
    #[sanitizer(wasm_dbms_api::prelude::TrimSanitizer)]
    #[sanitizer(wasm_dbms_api::prelude::LowerCaseSanitizer)]
    pub name: Text,
}

#[derive(DatabaseSchema)]
#[tables(ValidatedName = "validated_names")]
pub struct ValidatedNameTestSchema;

#[test]
fn repeated_validators_and_sanitizers_all_run_in_source_order() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    ValidatedNameTestSchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, ValidatedNameTestSchema);
    let insert_name = |name: String| {
        db.insert::<ValidatedName>(ValidatedNameInsertRequest {
            id: wasm_dbms_api::prelude::Autoincrement::Auto,
            name: Text(name.clone()),
        })?;
        let rows = db.select::<ValidatedName>(
            Query::builder()
                .filter(Some(Filter::eq(
                    "name",
                    Value::Text(Text(name.trim().to_lowercase())),
                )))
                .build(),
        )?;
        Ok::<_, wasm_dbms_api::prelude::DbmsError>(rows[0].clone())
    };

    assert!(insert_name("x".to_string()).is_err());
    assert!(insert_name("x".repeat(21)).is_err());
    assert_eq!(
        insert_name("  ALICE  ".to_string()).unwrap().name,
        Some(Text("alice".into()))
    );
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "priced_items"]
pub struct PricedItem {
    #[primary_key]
    pub id: Uint32,
    pub price: wasm_dbms_api::prelude::Decimal,
    pub quantity: Uint32,
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "access_tokens"]
pub struct AccessToken {
    #[primary_key]
    pub id: Uint32,
    pub token: wasm_dbms_api::prelude::Uuid,
    pub uses: Uint32,
}

#[derive(Debug, Table, Clone, PartialEq, Eq)]
#[table = "coupons"]
pub struct Coupon {
    #[primary_key]
    pub id: Uint32,
    pub discount: wasm_dbms_api::prelude::Decimal,
    pub code: wasm_dbms_api::prelude::Uuid,
    pub fee: wasm_dbms_api::prelude::Decimal,
    pub owner: wasm_dbms_api::prelude::Uuid,
    pub label: Text,
}

#[derive(DatabaseSchema)]
#[tables(
    PricedItem = "priced_items",
    AccessToken = "access_tokens",
    Coupon = "coupons"
)]
pub struct FixedWidthColumnSchema;

fn fixed_width_uuid(byte: u8) -> wasm_dbms_api::prelude::Uuid {
    use wasm_dbms_api::prelude::Encode as _;

    wasm_dbms_api::prelude::Uuid::decode(std::borrow::Cow::Owned(vec![byte; 16]))
        .expect("16 bytes should decode as a UUID")
}

#[test]
fn decimal_column_can_be_followed_by_another_column() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    FixedWidthColumnSchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, FixedWidthColumnSchema);
    let price = Value::Decimal(wasm_dbms_api::prelude::Decimal("10.50".parse().unwrap()));
    let insert = PricedItemInsertRequest::from_values(&[
        (PricedItem::columns()[0], Value::Uint32(Uint32(1))),
        (PricedItem::columns()[1], price.clone()),
        (PricedItem::columns()[2], Value::Uint32(Uint32(3))),
    ])
    .unwrap();
    db.insert::<PricedItem>(insert).unwrap();

    let rows = db
        .select_raw("priced_items", Query::builder().all().build())
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0][1].1, price);
    assert_eq!(rows[0][2].1, Value::Uint32(Uint32(3)));
}

#[test]
fn uuid_column_can_be_followed_by_another_column() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    FixedWidthColumnSchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, FixedWidthColumnSchema);
    let token = Value::Uuid(fixed_width_uuid(0xab));
    let insert = AccessTokenInsertRequest::from_values(&[
        (AccessToken::columns()[0], Value::Uint32(Uint32(1))),
        (AccessToken::columns()[1], token.clone()),
        (AccessToken::columns()[2], Value::Uint32(Uint32(7))),
    ])
    .unwrap();
    db.insert::<AccessToken>(insert).unwrap();

    let rows = db
        .select_raw("access_tokens", Query::builder().all().build())
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0][1].1, token);
    assert_eq!(rows[0][2].1, Value::Uint32(Uint32(7)));
}

#[test]
fn several_decimal_and_uuid_columns_round_trip_in_one_table() {
    let ctx = DbmsContext::new(HeapMemoryProvider::default());
    FixedWidthColumnSchema::register_tables(&ctx).unwrap();
    let db = WasmDbmsDatabase::oneshot(&ctx, FixedWidthColumnSchema);
    let discount = Value::Decimal(wasm_dbms_api::prelude::Decimal("0.15".parse().unwrap()));
    let code = Value::Uuid(fixed_width_uuid(0x11));
    let fee = Value::Decimal(wasm_dbms_api::prelude::Decimal("-2.75".parse().unwrap()));
    let owner = Value::Uuid(fixed_width_uuid(0x22));
    let label = Value::Text(Text("spring".to_string()));
    let insert = CouponInsertRequest::from_values(&[
        (Coupon::columns()[0], Value::Uint32(Uint32(1))),
        (Coupon::columns()[1], discount.clone()),
        (Coupon::columns()[2], code.clone()),
        (Coupon::columns()[3], fee.clone()),
        (Coupon::columns()[4], owner.clone()),
        (Coupon::columns()[5], label.clone()),
    ])
    .unwrap();
    db.insert::<Coupon>(insert).unwrap();

    let rows = db
        .select_raw("coupons", Query::builder().all().build())
        .unwrap();
    assert_eq!(rows.len(), 1);
    let values: Vec<&Value> = rows[0].iter().map(|(_, value)| value).collect();
    assert_eq!(
        values,
        vec![
            &Value::Uint32(Uint32(1)),
            &discount,
            &code,
            &fee,
            &owner,
            &label
        ]
    );
}

#[test]
fn transaction_primary_key_lookup_sees_its_own_update() {
    let ctx = setup();
    insert_user(&WasmDbmsDatabase::oneshot(&ctx, TestSchema), 1, "Alice");

    let tx = ctx.begin_transaction();
    let db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx);
    let updated = TestSchema
        .update(
            &db,
            "users",
            &[(User::columns()[1], Value::Text(Text("Bob".to_string())))],
            Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        )
        .unwrap();
    assert_eq!(updated, 1);

    let all_rows = db
        .select_raw("users", Query::builder().all().build())
        .unwrap();
    assert_eq!(all_rows[0][1].1, Value::Text(Text("Bob".to_string())));

    let by_primary_key = db
        .select_raw(
            "users",
            Query::builder()
                .all()
                .filter(Some(Filter::eq("id", Value::Uint32(Uint32(1)))))
                .build(),
        )
        .unwrap();
    assert_eq!(by_primary_key.len(), 1);
    assert_eq!(by_primary_key[0][1].1, Value::Text(Text("Bob".to_string())));

    // the remaining filter is evaluated on the updated row, not on the stored one
    let select_by_id_and_name = |name: &str| {
        db.select_raw(
            "users",
            Query::builder()
                .all()
                .filter(Some(
                    Filter::eq("id", Value::Uint32(Uint32(1)))
                        .and(Filter::eq("name", Value::Text(Text(name.to_string())))),
                ))
                .build(),
        )
        .unwrap()
    };
    assert_eq!(select_by_id_and_name("Bob").len(), 1);
    assert!(select_by_id_and_name("Alice").is_empty());
}

#[test]
fn transaction_update_by_primary_key_builds_on_its_previous_update() {
    let ctx = setup();
    insert_user(&WasmDbmsDatabase::oneshot(&ctx, TestSchema), 1, "Alice");

    let tx = ctx.begin_transaction();
    let db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx);
    let rename = |from: &str, to: &str| {
        TestSchema
            .update(
                &db,
                "users",
                &[(User::columns()[1], Value::Text(Text(to.to_string())))],
                Some(
                    Filter::eq("id", Value::Uint32(Uint32(1)))
                        .and(Filter::eq("name", Value::Text(Text(from.to_string())))),
                ),
            )
            .unwrap()
    };

    assert_eq!(rename("Alice", "Bob"), 1);
    assert_eq!(rename("Bob", "Carol"), 1);
    assert_eq!(rename("Alice", "Dave"), 0);

    let rows = db
        .select_raw("users", Query::builder().all().build())
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0][1].1, Value::Text(Text("Carol".to_string())));
}

fn select_name_indexed(
    db: &WasmDbmsDatabase<'_, HeapMemoryProvider>,
    filter: Filter,
) -> Vec<Vec<(ColumnDef, Value)>> {
    db.select_raw(
        "name_indexed_users",
        Query::builder().all().filter(Some(filter)).build(),
    )
    .unwrap()
}

fn name_eq(name: &str) -> Filter {
    Filter::eq("name", Value::Text(Text(name.to_string())))
}

fn update_name_indexed_in_tx(
    db: &WasmDbmsDatabase<'_, HeapMemoryProvider>,
    patch: &[(ColumnDef, Value)],
    filter: Filter,
) {
    let updated = NameIndexedTestSchema
        .update(db, "name_indexed_users", patch, Some(filter))
        .unwrap();
    assert_eq!(updated, 1);
}

#[test]
fn transaction_index_lookup_sees_update_of_non_indexed_column() {
    let ctx = setup_name_indexed();
    insert_name_indexed_user(
        &WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema),
        1,
        "alice",
        30,
    );

    let tx = ctx.begin_transaction();
    let db = WasmDbmsDatabase::from_transaction(&ctx, NameIndexedTestSchema, tx);
    update_name_indexed_in_tx(
        &db,
        &[(NameIndexedUser::columns()[2], Value::Uint32(Uint32(31)))],
        name_eq("alice"),
    );

    let rows = select_name_indexed(&db, name_eq("alice"));
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0][2].1, Value::Uint32(Uint32(31)));

    // the remaining (non-indexed) filter sees the updated value only
    let age_is = |age: u32| {
        select_name_indexed(
            &db,
            name_eq("alice").and(Filter::eq("age", Value::Uint32(Uint32(age)))),
        )
    };
    assert_eq!(age_is(31).len(), 1);
    assert!(age_is(30).is_empty());
}

#[test]
fn transaction_index_lookup_follows_update_of_indexed_column() {
    let ctx = setup_name_indexed();
    insert_name_indexed_user(
        &WasmDbmsDatabase::oneshot(&ctx, NameIndexedTestSchema),
        1,
        "alice",
        30,
    );

    let tx = ctx.begin_transaction();
    let db = WasmDbmsDatabase::from_transaction(&ctx, NameIndexedTestSchema, tx);
    update_name_indexed_in_tx(
        &db,
        &[(
            NameIndexedUser::columns()[1],
            Value::Text(Text("bob".to_string())),
        )],
        name_eq("alice"),
    );
    update_name_indexed_in_tx(
        &db,
        &[(NameIndexedUser::columns()[2], Value::Uint32(Uint32(32)))],
        name_eq("bob"),
    );

    assert!(select_name_indexed(&db, name_eq("alice")).is_empty());
    let rows = select_name_indexed(&db, name_eq("bob"));
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0][0].1, Value::Uint32(Uint32(1)));
    assert_eq!(rows[0][2].1, Value::Uint32(Uint32(32)));
}

#[test]
fn transaction_primary_key_lookup_follows_primary_key_update() {
    let ctx = setup();
    insert_user(&WasmDbmsDatabase::oneshot(&ctx, TestSchema), 1, "Alice");

    let tx = ctx.begin_transaction();
    let db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx);
    let updated = TestSchema
        .update(
            &db,
            "users",
            &[
                (User::columns()[0], Value::Uint32(Uint32(2))),
                (User::columns()[1], Value::Text(Text("Bob".to_string()))),
            ],
            Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        )
        .unwrap();
    assert_eq!(updated, 1);

    let by_id = |id: u32| {
        db.select_raw(
            "users",
            Query::builder()
                .all()
                .filter(Some(Filter::eq("id", Value::Uint32(Uint32(id)))))
                .build(),
        )
        .unwrap()
    };
    assert!(by_id(1).is_empty());
    let rows = by_id(2);
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0][1].1, Value::Text(Text("Bob".to_string())));
}

#[test]
fn transaction_primary_key_lookup_does_not_see_deleted_row() {
    let ctx = setup();
    insert_user(&WasmDbmsDatabase::oneshot(&ctx, TestSchema), 1, "Alice");

    let tx = ctx.begin_transaction();
    let db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx);
    TestSchema
        .update(
            &db,
            "users",
            &[(User::columns()[1], Value::Text(Text("Bob".to_string())))],
            Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        )
        .unwrap();
    let deleted = TestSchema
        .delete(
            &db,
            "users",
            DeleteBehavior::Restrict,
            Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        )
        .unwrap();
    assert_eq!(deleted, 1);

    let rows = db
        .select_raw(
            "users",
            Query::builder()
                .all()
                .filter(Some(Filter::eq("id", Value::Uint32(Uint32(1)))))
                .build(),
        )
        .unwrap();
    assert!(rows.is_empty());
}

#[test]
fn transaction_primary_key_lookup_returns_reinserted_row_after_delete() {
    let ctx = setup();
    insert_user(&WasmDbmsDatabase::oneshot(&ctx, TestSchema), 1, "Alice");

    let tx = ctx.begin_transaction();
    let db = WasmDbmsDatabase::from_transaction(&ctx, TestSchema, tx);
    TestSchema
        .delete(
            &db,
            "users",
            DeleteBehavior::Restrict,
            Some(Filter::eq("id", Value::Uint32(Uint32(1)))),
        )
        .unwrap();
    insert_user(&db, 1, "Bob");

    let rows = db
        .select_raw(
            "users",
            Query::builder()
                .all()
                .filter(Some(Filter::eq("id", Value::Uint32(Uint32(1)))))
                .build(),
        )
        .unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0][1].1, Value::Text(Text("Bob".to_string())));
}