zeyn 0.0.6

//! Data layer tests shared by every database: results of `Db::find` are compared with
//! `DataQuery::apply` over the same rows, so SQL and memory data behave the same, then saves,
//! conflicts, unique errors, transactions and streams are checked.

use zeyn::prelude::*;
use zeyn::sql::{Arg, Db, PoolLimits};

/// Hand written statements that differ between databases.
pub struct Raw {
    /// Drops the test tables (and sequences); must not fail when they are missing.
    pub drop: &'static [&'static str],
    /// `UPDATE zeyn_test_items SET qty = <p1> WHERE id = <p2>`.
    pub set_qty: &'static str,
    /// Fills `zeyn_test_stream` with ids 1 to `STREAM_ROWS`.
    pub fill_stream: &'static str,
}

/// Rows the stream test reads.
pub const STREAM_ROWS: u64 = 100_000;
use zeyn::{Aggregate, AggregateRequest, Cursor, Filter, PageRequest, Sort, Value};

#[choice]
pub enum Kind {
    Drink,
    Food,
    Tool,
}

#[entity(table = "zeyn_test_items")]
pub struct Item {
    #[key]
    pub id: ItemId,
    #[rule(required, max_len = 40, unique)]
    pub code: String,
    pub name: String,
    pub note: Option<String>,
    pub kind: Kind,
    pub price: Money,
    pub qty: i64,
    pub small: i32,
    pub active: bool,
    pub day: Date,
    pub at: Option<DateTime>,
    pub clock: Time,
    pub local: LocalDateTime,
    #[readonly]
    pub stock: i64,
    #[version]
    pub version: i64,
}

#[entity(table = "zeyn_test_stream")]
pub struct Line {
    #[key]
    pub id: LineId,
    pub text: String,
}

/// Database URLs from `var`, separated by spaces. Empty means skip, and the test says so.
pub fn urls(var: &str) -> Vec<String> {
    let urls: Vec<String> = std::env::var(var)
        .unwrap_or_default()
        .split_whitespace()
        .map(String::from)
        .collect();
    if urls.is_empty() {
        eprintln!("skipped: set {var} to a database URL to run these tests");
    }
    urls
}

fn item(i: i64) -> Item {
    let names = [
        "Black tea",
        "Green tea",
        "Coffee",
        "Olive oil",
        "Honey",
        "Hazelnuts",
    ];
    let day = Date::new(2026, 1 + (i % 12) as u8, 1 + (i % 28) as u8).unwrap();
    Item {
        id: ItemId(0),
        code: format!("P-{i:04}"),
        name: format!("{} {}", names[(i % 6) as usize], i % 40),
        note: (i % 3 != 0).then(|| format!("Note {}", i % 5)),
        kind: [Kind::Drink, Kind::Food, Kind::Tool][(i % 3) as usize],
        price: Money::tl(i * 1_275 % 9_000),
        qty: i % 17 - 5,
        small: (i % 9) as i32,
        active: i % 4 != 0,
        day,
        at: (i % 5 != 0).then(|| DateTime::from_unix_ms(1_767_225_600_000 + i * 3_600_123)),
        clock: Time::new((i % 24) as u8, (i % 60) as u8, 0).unwrap(),
        local: LocalDateTime::new(day, Time::new(9, (i % 60) as u8, 30).unwrap()),
        stock: 0,
        version: 0,
    }
}

async fn fresh(url: &str, raw: &Raw, limits: PoolLimits) -> Db {
    let db = Db::connect(url, limits).await.unwrap();
    for statement in raw.drop {
        db.execute(statement, vec![])
            .await
            .unwrap();
    }
    db.create_table::<Item>().await.unwrap();
    db.create_table::<Line>().await.unwrap();
    db
}

fn ids(items: &[Item]) -> Vec<i64> {
    items.iter().map(|i| i.id.0).collect()
}

fn close(a: &Value, b: &Value) -> bool {
    match (a, b) {
        (Value::Decimal(x), Value::Decimal(y)) => (x - y).abs() < 1e-9,
        _ => a == b,
    }
}

async fn same(db: &Db, all: &[Item], q: DataQuery) {
    let mem = q.clone().apply(all.to_vec());
    let sql: Page<Item> = db.find(&q).await.unwrap();
    assert_eq!(ids(&sql.items), ids(&mem.items), "rows of {q:?}");
    if q.with_total {
        assert_eq!(sql.total, mem.total, "total of {q:?}");
    }
    assert_eq!(
        sql.totals.len(),
        mem.totals.len(),
        "totals of {q:?}: {:?} vs {:?}",
        sql.totals,
        mem.totals
    );
    for (a, b) in sql.totals.iter().zip(&mem.totals) {
        assert!(
            a.field == b.field && a.op == b.op && close(&a.value, &b.value),
            "{a:?} vs {b:?} in {q:?}"
        );
    }
    assert_eq!(sql.groups.len(), mem.groups.len(), "groups of {q:?}");
    for (a, b) in sql.groups.iter().zip(&mem.groups) {
        assert_eq!((&a.value, a.count), (&b.value, b.count), "group of {q:?}");
        assert_eq!(a.totals.len(), b.totals.len());
        for (x, y) in a.totals.iter().zip(&b.totals) {
            assert!(close(&x.value, &y.value), "{x:?} vs {y:?}");
        }
    }
    assert_eq!(
        sql.next.is_some(),
        mem.next.is_some(),
        "more rows after {q:?}"
    );
}

fn query(filters: Vec<Filter>, sort: Vec<(&str, bool)>, offset: u64, limit: u32) -> DataQuery {
    DataQuery {
        page: PageRequest::Offset { offset, limit },
        sort: sort
            .into_iter()
            .map(|(f, desc)| Sort {
                field: f.into(),
                desc,
            })
            .collect(),
        filters,
        with_total: true,
        ..DataQuery::default()
    }
}

/// Lists, totals, groups, keyset walks, saves, conflicts, unique errors and transactions.
pub async fn data_layer(url: &str, raw: &Raw) {
    let db = fresh(
        url,
        raw,
        PoolLimits {
            max_connections: 3,
            ..PoolLimits::default()
        },
    )
    .await;

    let mut tx = db.begin().await.unwrap();
    for i in 1..=300 {
        tx.insert(&item(i)).await.unwrap();
    }
    tx.execute("UPDATE zeyn_test_items SET stock = MOD(id, 7)", vec![])
        .await
        .unwrap();
    tx.commit().await.unwrap();

    let mut all = Vec::new();
    let mut stream = db.stream::<Item>(&DataQuery::default());
    while let Some(row) = stream.next().await {
        all.push(row.unwrap());
    }
    assert_eq!(all.len(), 300);
    assert_eq!(all[0].code, "P-0001");
    assert_eq!(all[0].version, 1);
    assert!(
        all[3]
            == Item {
                id: ItemId(4),
                stock: 4,
                version: 1,
                ..item(4)
            },
        "every type reads back as written: {:?}",
        all[3]
    );

    same(&db, &all, query(vec![], vec![], 0, 25)).await;
    same(&db, &all, query(vec![], vec![("name", false)], 50, 25)).await;
    same(
        &db,
        &all,
        query(vec![], vec![("note", true), ("price", false)], 0, 60),
    )
    .await;
    same(&db, &all, query(vec![], vec![("at", false)], 280, 30)).await;
    same(
        &db,
        &all,
        query(
            vec![Filter::contains("name", "TEA")],
            vec![("qty", true)],
            0,
            500,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![Filter {
                starts_with: Some("gre".into()),
                ..Filter::contains("name", "")
            }],
            vec![],
            0,
            50,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![Filter {
                equals: Some("p-0042".into()),
                ..Filter::contains("code", "")
            }],
            vec![],
            0,
            50,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![Filter::range("price", Some("10"), Some("45.5"))],
            vec![],
            0,
            500,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![Filter::range("qty", Some("-2.5"), Some("3.5"))],
            vec![("small", true)],
            0,
            500,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![Filter::range("day", Some("2026-03-01"), Some("2026-05-15"))],
            vec![("day", false)],
            0,
            500,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![Filter::range("at", Some("2026-01-02"), Some("2026-01-05"))],
            vec![],
            0,
            500,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![Filter::range("clock", Some("10:00"), Some("14:30"))],
            vec![("clock", true)],
            0,
            500,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![Filter::range("local", Some("2026-02-01"), None)],
            vec![("local", true)],
            0,
            500,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![Filter::any_of("kind", ["Food", "Tool", "Nope"])],
            vec![("kind", false)],
            0,
            500,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![Filter::is("active", false), Filter::blank("note", true)],
            vec![],
            0,
            500,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![
                Filter::blank("at", false),
                Filter::any_of("id", ["3", "30", "300", "x"]),
            ],
            vec![],
            0,
            50,
        ),
    )
    .await;
    same(
        &db,
        &all,
        query(vec![Filter::contains("day", "2026-03")], vec![], 0, 500),
    )
    .await;
    same(
        &db,
        &all,
        query(
            vec![Filter::contains("code", "%"), Filter::contains("code", "_")],
            vec![],
            0,
            500,
        ),
    )
    .await;

    let totals = |field: &str, ops: &[Aggregate]| {
        ops.iter()
            .map(|op| AggregateRequest {
                field: field.into(),
                op: *op,
            })
            .collect::<Vec<_>>()
    };
    let mut q = query(vec![Filter::is("active", true)], vec![], 0, 10);
    q.aggregates = [
        totals(
            "price",
            &[
                Aggregate::Sum,
                Aggregate::Avg,
                Aggregate::Min,
                Aggregate::Max,
                Aggregate::Count,
            ],
        ),
        totals(
            "qty",
            &[
                Aggregate::Sum,
                Aggregate::Avg,
                Aggregate::Min,
                Aggregate::Max,
            ],
        ),
        totals("day", &[Aggregate::Min, Aggregate::Max]),
        totals("at", &[Aggregate::Min, Aggregate::Max, Aggregate::Count]),
        totals("clock", &[Aggregate::Max]),
        totals("note", &[Aggregate::Count]),
    ]
    .concat();
    same(&db, &all, q.clone()).await;
    let mut none = q.clone();
    none.filters = vec![Filter::contains("code", "nothing")];
    same(&db, &all, none).await;

    for (field, desc) in [
        ("kind", false),
        ("active", true),
        ("note", false),
        ("price", false),
    ] {
        let mut g = query(vec![], vec![(field, desc)], 0, 2);
        g.group = Some(field.into());
        g.aggregates = totals("qty", &[Aggregate::Sum, Aggregate::Avg]);
        same(&db, &all, g.clone()).await;
        g.page = PageRequest::Offset {
            offset: 2,
            limit: 2,
        };
        same(&db, &all, g).await;
    }

    for sort in [
        vec![("note", true), ("name", false)],
        vec![("note", false)],
        vec![("at", true)],
        vec![("price", true), ("day", false)],
    ] {
        let mut expected = query(
            vec![Filter::range("qty", Some("-3"), None)],
            sort.clone(),
            0,
            500,
        );
        expected.with_total = false;
        let expected = expected.apply(all.clone());
        let mut walked = Vec::new();
        let mut cursor: Option<Cursor> = None;
        loop {
            let mut q = query(
                vec![Filter::range("qty", Some("-3"), None)],
                sort.clone(),
                0,
                0,
            );
            q.page = PageRequest::After {
                cursor: cursor.clone(),
                limit: 17,
            };
            q.with_total = false;
            let page: Page<Item> = db.find(&q).await.unwrap();
            assert!(page.items.len() <= 17);
            walked.extend(ids(&page.items));
            match page.next {
                Some(next) => cursor = Some(next),
                None => break,
            }
        }
        assert_eq!(walked, ids(&expected.items), "keyset walk by {sort:?}");
    }

    let mut first: Item = db.get(ItemId(5)).await.unwrap();
    assert_eq!(first.stock, 5);
    let stale = first.clone();
    first.name = "Renamed".into();
    first.stock = 999;
    let saved = db.update(&first).await.unwrap();
    assert_eq!(
        (saved.version, saved.name.as_str(), saved.stock),
        (2, "Renamed", 5),
        "readonly stock is not written"
    );
    assert_eq!(db.update(&stale).await.unwrap_err(), Error::Conflict);
    assert_eq!(
        db.update(&Item {
            id: ItemId(99_999),
            version: 1,
            ..item(1)
        })
        .await
        .unwrap_err(),
        Error::Conflict
    );

    let dup = db
        .insert(&Item {
            code: "P-0007".into(),
            ..item(1)
        })
        .await
        .unwrap_err();
    let Error::Validation(errors) = dup else {
        panic!("{dup:?}")
    };
    assert_eq!(errors.errors[0].field, "code");
    assert_eq!(errors.errors[0].rule, zeyn::Rule::Unique);
    let mut taken = db.get::<Item>(ItemId(8)).await.unwrap();
    taken.code = "P-0009".into();
    assert!(matches!(db.update(&taken).await, Err(Error::Validation(_))));

    let mut tx = db.begin().await.unwrap();
    let a = tx.insert(&item(1001)).await.unwrap();
    tx.insert(&item(1002)).await.unwrap();
    tx.rollback().await.unwrap();
    assert_eq!(db.get::<Item>(a.id).await.unwrap_err(), Error::NotFound);
    {
        let mut tx = db.begin().await.unwrap();
        tx.insert(&item(1003)).await.unwrap();
    }
    let count: Page<Item> = db
        .find(&query(
            vec![Filter::contains("code", "P-100")],
            vec![],
            0,
            10,
        ))
        .await
        .unwrap();
    assert_eq!(count.total, Some(0), "a dropped transaction rolls back");
    let mut tx = db.begin().await.unwrap();
    let kept = tx.insert(&item(1004)).await.unwrap();
    let mut changed = tx.get::<Item>(kept.id).await.unwrap();
    changed.qty = 77;
    tx.update(&changed).await.unwrap();
    tx.commit().await.unwrap();
    assert_eq!(
        db.get::<Item>(kept.id)
            .await
            .unwrap()
            .qty,
        77
    );

    db.delete::<Item>(kept.id)
        .await
        .unwrap();
    assert_eq!(
        db.delete::<Item>(kept.id)
            .await
            .unwrap_err(),
        Error::NotFound
    );
    assert_eq!(
        db.get::<Item>(kept.id)
            .await
            .unwrap_err(),
        Error::NotFound
    );

    assert_eq!(
        db.execute(raw.set_qty, vec![Arg::Int(5), Arg::Int(1)])
            .await
            .unwrap(),
        1
    );

    let capped = Db::connect(
        url,
        PoolLimits {
            count_cap: 10,
            ..PoolLimits::default()
        },
    )
    .await
    .unwrap();
    let page: Page<Item> = capped
        .find(&query(vec![], vec![], 0, 5))
        .await
        .unwrap();
    assert_eq!(page.total, Some(10));

    let mut forged = query(vec![], vec![], 0, 5);
    forged.page = PageRequest::After {
        cursor: Some(Cursor("eyJvIjpbXSwidiI6W119".into())),
        limit: 5,
    };
    assert!(matches!(
        db.find::<Item>(&forged).await,
        Err(Error::Validation(_))
    ));
}

/// Entities have no `PartialEq`; compare them through JSON.
impl PartialEq for Item {
    fn eq(&self, other: &Self) -> bool {
        serde_json::to_value(self).unwrap() == serde_json::to_value(other).unwrap()
    }
}

/// Streams a large result in order and drops a stream early without losing the connection.
pub async fn streams(url: &str, raw: &Raw) {
    let db = Db::connect(
        url,
        PoolLimits {
            max_connections: 2,
            ..PoolLimits::default()
        },
    )
    .await
    .unwrap();
    db.execute(raw.fill_stream, vec![])
        .await
        .unwrap();

    let mut stream = db.stream::<Line>(&DataQuery {
        sort: vec![Sort {
            field: "id".into(),
            desc: true,
        }],
        ..DataQuery::default()
    });
    let (mut n, mut last) = (0u64, i64::MAX);
    while let Some(row) = stream.next().await {
        let row = row.unwrap();
        assert!(row.id.0 < last);
        last = row.id.0;
        n += 1;
    }
    assert_eq!(n, STREAM_ROWS);

    let mut early = db.stream::<Line>(&DataQuery::default());
    assert!(early.next().await.unwrap().is_ok());
    drop(early);
    let page: Page<Line> = db
        .find(&DataQuery {
            with_total: true,
            ..DataQuery::default()
        })
        .await
        .unwrap();
    assert_eq!(
        page.total,
        Some(STREAM_ROWS),
        "the pool is usable after a stream is dropped early"
    );
}