icydb-core 0.257.18

mod admission;
mod owned_membership;

use crate::{
    db::predicate::{
        CoercionId, CoercionSpec, CompareOp, ComparePredicate, Predicate, normalize,
        normalize::{
            normalize_accepted_compare_fields_coercion, normalize_compare_value_for_accepted_kind,
            normalize_value_for_accepted_kind,
        },
    },
    db::schema::AcceptedFieldKind,
    value::Value,
};

#[test]
fn normalize_flat_junctions_retain_owned_backing_and_canonical_identity() {
    use crate::db::predicate::fingerprint::predicate_fingerprint;

    let leaf = |field: &str| Predicate::IsNull {
        field: field.into(),
    };
    for conjunction in [true, false] {
        let junction = if conjunction {
            Predicate::And
        } else {
            Predicate::Or
        };
        let neutral = if conjunction {
            Predicate::True
        } else {
            Predicate::False
        };
        let mut children = Vec::with_capacity(16);
        children.extend([
            neutral,
            leaf("b"),
            Predicate::Not(Box::new(Predicate::Not(Box::new(leaf("a"))))),
            leaf("b"),
        ]);
        let backing = children.as_ptr();
        let capacity = children.capacity();
        let expected = junction(vec![leaf("a"), leaf("b")]);
        let mut actual = junction(children);
        for _ in 0..2 {
            actual = normalize(actual);
            assert_eq!(actual, expected);
            assert_eq!(
                predicate_fingerprint(&actual),
                predicate_fingerprint(&expected)
            );
            let (Predicate::And(retained) | Predicate::Or(retained)) = &actual else {
                panic!("two distinct state predicates retain their junction");
            };
            assert_eq!(retained.as_ptr(), backing);
            assert_eq!(retained.capacity(), capacity);
        }
    }
}

#[test]
fn normalize_junctions_flatten_children_exposed_by_double_negation() {
    let leaf = |field: &str| Predicate::IsNull {
        field: field.into(),
    };
    for junction in [Predicate::And, Predicate::Or] {
        let input = junction(vec![
            leaf("c"),
            Predicate::Not(Box::new(Predicate::Not(Box::new(junction(vec![
                leaf("b"),
                leaf("a"),
            ]))))),
            leaf("b"),
        ]);
        assert_eq!(
            normalize(input),
            junction(vec![leaf("a"), leaf("b"), leaf("c")])
        );
    }
}

#[test]
fn normalize_nested_junctions_size_the_flattened_backing_once() {
    let leaf = |index| Predicate::IsNull {
        field: format!("field{index:03}"),
    };
    for conjunction in [true, false] {
        let junction = if conjunction {
            Predicate::And
        } else {
            Predicate::Or
        };
        let neutral = if conjunction {
            Predicate::True
        } else {
            Predicate::False
        };
        for width in [2, 3, 8, 33] {
            let input = junction(vec![
                leaf(0),
                junction((1..=width).rev().map(leaf).collect()),
                // This nested singleton must collapse before destination sizing.
                junction(vec![neutral.clone(), leaf(width + 1)]),
                neutral.clone(),
            ]);
            let expected = junction((0..width + 2).map(leaf).collect());
            let normalized = normalize(input);
            assert_eq!(normalized, expected);
            let (Predicate::And(children) | Predicate::Or(children)) = &normalized else {
                panic!("distinct state predicates retain their junction");
            };
            assert_eq!(children.capacity(), width + 2);
            assert_eq!(normalize(normalized.clone()), normalized);
        }
    }
}

#[test]
fn normalize_orders_varying_length_keys_for_both_connectives() {
    let long = "x".repeat(1024);
    let children = [
        long.clone(),
        "z".into(),
        "a".into(),
        "zz".into(),
        "b".into(),
    ]
    .into_iter()
    .map(|field| Predicate::IsNull { field })
    .collect::<Vec<_>>();
    let expected = ["a".into(), "b".into(), "z".into(), "zz".into(), long]
        .into_iter()
        .map(|field| Predicate::IsNull { field })
        .collect::<Vec<_>>();
    assert_eq!(
        normalize(Predicate::And(children.clone())),
        Predicate::And(expected.clone())
    );
    assert_eq!(normalize(Predicate::Or(children)), Predicate::Or(expected));
}

#[test]
fn normalize_preserves_structurally_distinct_equal_key_order() {
    let leaf = |text: &str| {
        Predicate::Compare(ComparePredicate::with_coercion(
            "name",
            CompareOp::Ne,
            Value::Text(text.to_string()),
            CoercionId::TextCasefold,
        ))
    };
    for children in [vec![leaf("A"), leaf("a")], vec![leaf("a"), leaf("A")]] {
        let predicate = Predicate::Or(children);
        assert_eq!(normalize(predicate.clone()), predicate);
    }
}

#[test]
fn normalize_compact_membership_conjunction_is_idempotent() {
    let predicate = Predicate::And(vec![
        Predicate::Compare(ComparePredicate::with_coercion(
            "stage",
            CompareOp::In,
            Value::List(vec![
                Value::Text("Review".to_string()),
                Value::Text("Draft".to_string()),
            ]),
            CoercionId::Strict,
        )),
        Predicate::eq(
            "collection_id".to_string(),
            Value::Text("01KV5N439P0000000000000000".to_string()),
        ),
        Predicate::Not(Box::new(Predicate::Not(Box::new(Predicate::eq(
            "rank".to_string(),
            Value::Nat64(7),
        ))))),
    ]);

    let normalized = normalize(predicate);
    assert_eq!(normalize(normalized.clone()), normalized);
}

#[test]
fn normalize_moves_nested_text_operands_through_identity_rewrites() {
    let field = "description".to_string();
    let value = "payload".repeat(128);
    let field_backing = field.as_ptr();
    let value_backing = value.as_ptr();
    let predicate = Predicate::Or(vec![
        Predicate::False,
        Predicate::And(vec![
            Predicate::True,
            Predicate::Not(Box::new(Predicate::Not(Box::new(
                Predicate::TextContains {
                    field,
                    value: Value::Text(value),
                },
            )))),
        ]),
    ]);

    let Predicate::TextContains {
        field,
        value: Value::Text(value),
    } = normalize(predicate)
    else {
        panic!("identity rewrites preserve the text predicate");
    };
    assert_eq!(field, "description");
    assert_eq!(value, "payload".repeat(128));
    assert_eq!(field.as_ptr(), field_backing);
    assert_eq!(value.as_ptr(), value_backing);
}

#[test]
fn normalize_nested_negation_preserves_constants_and_canonical_children() {
    let leaf = Predicate::eq("rank".to_string(), Value::Nat64(7));
    for predicate in [
        Predicate::True,
        Predicate::False,
        leaf.clone(),
        Predicate::And(vec![Predicate::True, leaf.clone(), leaf.clone()]),
        Predicate::Or(vec![Predicate::False, leaf.clone(), leaf]),
    ] {
        let expected = normalize(predicate.clone());
        for depth in [1, 2, 3, 4, 32] {
            let mut nested = predicate.clone();
            for _ in 0..depth {
                nested = Predicate::Not(Box::new(nested));
            }
            let normalized = normalize(nested);
            let expected = if depth % 2 == 0 {
                expected.clone()
            } else {
                Predicate::Not(Box::new(expected.clone()))
            };
            assert_eq!(normalized, expected);
            assert_eq!(normalize(normalized.clone()), normalized);
        }
    }
}

#[test]
fn normalize_surviving_negation_retains_its_owned_box_and_identity() {
    use crate::db::predicate::{
        encoding::write_predicate_sort_key, fingerprint::predicate_fingerprint,
    };

    let leaf = Predicate::eq("name".into(), Value::Text("payload".repeat(128)));
    for child in [
        Predicate::True,
        Predicate::False,
        leaf.clone(),
        Predicate::And(vec![Predicate::True, leaf.clone(), leaf.clone()]),
        Predicate::Or(vec![Predicate::False, leaf.clone(), leaf]),
    ] {
        for depth in [1, 3, 31, 125] {
            let canonical_child = normalize(child.clone());
            let expected = Predicate::Not(Box::new(canonical_child));
            let mut nested = child.clone();
            for _ in 1..depth {
                nested = Predicate::Not(Box::new(nested));
            }
            let owned = Box::new(nested);
            let backing = std::ptr::from_ref(owned.as_ref());
            let mut normalized = normalize(Predicate::Not(owned));
            for _ in 0..2 {
                let Predicate::Not(retained) = &normalized else {
                    panic!("odd negation depth retains one NOT");
                };
                assert_eq!(std::ptr::from_ref(retained.as_ref()), backing);
                assert_eq!(normalized, expected);
                let mut actual_bytes = Vec::new();
                let mut expected_bytes = Vec::new();
                write_predicate_sort_key(&mut actual_bytes, &normalized);
                write_predicate_sort_key(&mut expected_bytes, &expected);
                assert_eq!(actual_bytes, expected_bytes);
                assert_eq!(
                    predicate_fingerprint(&normalized),
                    predicate_fingerprint(&expected)
                );
                normalized = normalize(normalized);
            }
        }
    }
}

#[test]
fn normalize_and_dedups_identical_children_and_collapses_to_singleton() {
    let duplicated = Predicate::And(vec![
        Predicate::eq("rank".to_string(), Value::Nat64(7)),
        Predicate::eq("rank".to_string(), Value::Nat64(7)),
    ]);

    let normalized = normalize(duplicated);

    assert_eq!(
        normalized,
        Predicate::Compare(ComparePredicate::eq("rank".to_string(), Value::Nat64(7))),
        "identical AND children should collapse to one predicate",
    );
}

#[test]
fn normalize_or_dedups_identical_children_and_collapses_to_singleton() {
    let duplicated = Predicate::Or(vec![
        Predicate::eq("rank".to_string(), Value::Nat64(7)),
        Predicate::eq("rank".to_string(), Value::Nat64(7)),
    ]);

    let normalized = normalize(duplicated);

    assert_eq!(
        normalized,
        Predicate::Compare(ComparePredicate::eq("rank".to_string(), Value::Nat64(7))),
        "identical OR children should collapse to one predicate",
    );
}

#[test]
fn normalize_and_orders_cheaper_predicates_before_text_contains() {
    let mixed = Predicate::And(vec![
        Predicate::TextContains {
            field: "name".to_string(),
            value: Value::Text("ada".to_string()),
        },
        Predicate::eq("rank".to_string(), Value::Nat64(7)),
    ]);

    let normalized = normalize(mixed);
    let Predicate::And(children) = normalized else {
        panic!("normalized mixed predicate should remain AND with two children");
    };
    assert_eq!(
        children.len(),
        2,
        "mixed AND should keep exactly two children"
    );
    assert!(
        matches!(children[0], Predicate::Compare(_)),
        "cheap compare predicate should be evaluated before text-contains predicate",
    );
    assert!(
        matches!(children[1], Predicate::TextContains { .. }),
        "text-contains predicate should be placed after cheap compare predicate",
    );
}

#[test]
fn normalize_and_orders_scalar_compares_before_membership() {
    let mixed = Predicate::And(vec![
        Predicate::Compare(ComparePredicate::with_coercion(
            "stage",
            CompareOp::In,
            Value::List(vec![
                Value::Text("Draft".to_string()),
                Value::Text("Review".to_string()),
            ]),
            CoercionId::Strict,
        )),
        Predicate::eq(
            "collection_id".to_string(),
            Value::Text("01KV5N439P0000000000000000".to_string()),
        ),
    ]);

    let normalized = normalize(mixed);
    let Predicate::And(children) = normalized else {
        panic!("normalized mixed predicate should remain AND with two children");
    };
    assert_eq!(
        children.len(),
        2,
        "mixed AND should keep exactly two children"
    );
    assert!(
        matches!(
            children[0],
            Predicate::Compare(ComparePredicate {
                op: CompareOp::Eq,
                ..
            })
        ),
        "scalar compare predicate should be evaluated before membership predicate",
    );
    assert!(
        matches!(
            children[1],
            Predicate::Compare(ComparePredicate {
                op: CompareOp::In,
                ..
            })
        ),
        "membership predicate should be placed after scalar compare predicate",
    );
}

#[test]
fn normalize_and_conflicting_eq_literals_collapses_to_false() {
    let predicate = Predicate::And(vec![
        Predicate::eq("rank".to_string(), Value::Nat64(1)),
        Predicate::eq("rank".to_string(), Value::Nat64(2)),
    ]);

    let normalized = normalize(predicate);

    assert_eq!(
        normalized,
        Predicate::False,
        "conflicting equalities in conjunction must collapse to false",
    );
}

#[test]
fn normalize_and_tightens_lower_bounds() {
    let predicate = Predicate::And(vec![
        Predicate::Compare(ComparePredicate::gt("rank".to_string(), Value::Nat64(3))),
        Predicate::Compare(ComparePredicate::gte("rank".to_string(), Value::Nat64(5))),
    ]);

    let normalized = normalize(predicate);

    assert_eq!(
        normalized,
        Predicate::Compare(ComparePredicate::gte("rank".to_string(), Value::Nat64(5))),
        "conjunction should keep the stricter lower bound",
    );
}

#[test]
fn normalize_and_tightens_upper_bounds() {
    let predicate = Predicate::And(vec![
        Predicate::Compare(ComparePredicate::lt("rank".to_string(), Value::Nat64(9))),
        Predicate::Compare(ComparePredicate::lte("rank".to_string(), Value::Nat64(7))),
    ]);

    let normalized = normalize(predicate);

    assert_eq!(
        normalized,
        Predicate::Compare(ComparePredicate::lte("rank".to_string(), Value::Nat64(7))),
        "conjunction should keep the stricter upper bound",
    );
}

#[test]
fn normalize_and_eq_with_satisfied_bound_collapses_to_eq() {
    let predicate = Predicate::And(vec![
        Predicate::eq("rank".to_string(), Value::Nat64(7)),
        Predicate::Compare(ComparePredicate::gt("rank".to_string(), Value::Nat64(5))),
    ]);

    let normalized = normalize(predicate);

    assert_eq!(
        normalized,
        Predicate::Compare(ComparePredicate::eq("rank".to_string(), Value::Nat64(7))),
        "equality should subsume compatible lower-bound constraints",
    );
}

#[test]
fn normalize_and_eq_with_conflicting_bound_collapses_to_false() {
    let predicate = Predicate::And(vec![
        Predicate::eq("rank".to_string(), Value::Nat64(3)),
        Predicate::Compare(ComparePredicate::gt("rank".to_string(), Value::Nat64(5))),
    ]);

    let normalized = normalize(predicate);

    assert_eq!(
        normalized,
        Predicate::False,
        "equality conflicting with a bound must collapse to false",
    );
}

#[test]
fn normalize_and_equal_lower_and_upper_collapse_to_eq() {
    let predicate = Predicate::And(vec![
        Predicate::Compare(ComparePredicate::with_coercion(
            "rank",
            CompareOp::Gte,
            Value::Nat64(11),
            crate::db::predicate::CoercionId::Strict,
        )),
        Predicate::Compare(ComparePredicate::with_coercion(
            "rank",
            CompareOp::Lte,
            Value::Nat64(11),
            crate::db::predicate::CoercionId::Strict,
        )),
    ]);

    let normalized = normalize(predicate);

    assert_eq!(
        normalized,
        Predicate::Compare(ComparePredicate::eq("rank".to_string(), Value::Nat64(11))),
        "matching inclusive lower/upper bounds should collapse to equality",
    );
}

#[test]
fn normalize_and_crossed_bounds_collapse_to_false() {
    let predicate = Predicate::And(vec![
        Predicate::Compare(ComparePredicate::gt("rank".to_string(), Value::Nat64(9))),
        Predicate::Compare(ComparePredicate::lt("rank".to_string(), Value::Nat64(5))),
    ]);

    let normalized = normalize(predicate);

    assert_eq!(
        normalized,
        Predicate::False,
        "crossed lower/upper bounds must collapse to false",
    );
}

#[test]
fn normalize_or_same_field_eq_collapses_to_in() {
    let predicate = Predicate::Or(vec![
        Predicate::Compare(ComparePredicate::with_coercion(
            "tag",
            CompareOp::Eq,
            Value::Text("beta".to_string()),
            CoercionId::Strict,
        )),
        Predicate::Compare(ComparePredicate::with_coercion(
            "tag",
            CompareOp::Eq,
            Value::Text("alpha".to_string()),
            CoercionId::Strict,
        )),
        Predicate::Compare(ComparePredicate::with_coercion(
            "tag",
            CompareOp::Eq,
            Value::Text("beta".to_string()),
            CoercionId::Strict,
        )),
    ]);

    let normalized = normalize(predicate);
    let Predicate::Compare(compare) = normalized else {
        panic!("same-field strict OR-equality should collapse to one IN compare");
    };

    assert_eq!(compare.field, "tag".to_string());
    assert_eq!(compare.op, CompareOp::In);
    assert_eq!(compare.coercion.id, CoercionId::Strict);
    let Value::List(mut values) = compare.value else {
        panic!("collapsed OR-equality compare should carry list literal");
    };
    values.sort_by(Value::canonical_cmp);
    assert_eq!(
        values,
        vec![
            Value::Text("alpha".to_string()),
            Value::Text("beta".to_string()),
        ],
        "same-field strict OR-equality should collapse to deduplicated IN-list members",
    );
}

#[test]
fn normalize_or_mixed_eq_coercions_do_not_collapse_to_in() {
    let predicate = Predicate::Or(vec![
        Predicate::Compare(ComparePredicate::with_coercion(
            "tag",
            CompareOp::Eq,
            Value::Text("alpha".to_string()),
            CoercionId::Strict,
        )),
        Predicate::Compare(ComparePredicate::with_coercion(
            "tag",
            CompareOp::Eq,
            Value::Text("beta".to_string()),
            CoercionId::TextCasefold,
        )),
    ]);

    let normalized = normalize(predicate);
    let Predicate::Or(children) = normalized else {
        panic!("mixed coercion OR-equality should remain OR in canonical form");
    };

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

#[test]
fn normalize_or_list_equality_literals_do_not_collapse_to_in() {
    let predicate = Predicate::Or(vec![
        Predicate::Compare(ComparePredicate::with_coercion(
            "tags",
            CompareOp::Eq,
            Value::List(vec![Value::Text("a".to_string())]),
            CoercionId::Strict,
        )),
        Predicate::Compare(ComparePredicate::with_coercion(
            "tags",
            CompareOp::Eq,
            Value::List(vec![Value::Text("b".to_string())]),
            CoercionId::Strict,
        )),
    ]);

    let normalized = normalize(predicate);
    let Predicate::Or(children) = normalized else {
        panic!("list-literal OR-equality should remain OR in canonical form");
    };

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

#[test]
fn accepted_numeric_membership_normalization_is_canonical() {
    let value = Value::List(vec![
        Value::Int64(3),
        Value::Nat64(1),
        Value::Int64(3),
        Value::Nat64(2),
    ]);
    let coercion = CoercionSpec::new(CoercionId::Strict);

    let normalized = crate::db::query::preparation::with_preparation_work(|work| {
        normalize_compare_value_for_accepted_kind(
            "rank",
            CompareOp::In,
            &value,
            &AcceptedFieldKind::Nat64,
            &coercion,
            work,
        )
    })
    .expect("accepted membership normalization should succeed");

    assert_eq!(
        normalized,
        Value::List(vec![Value::Nat64(1), Value::Nat64(2), Value::Nat64(3)]),
    );
}

#[test]
fn accepted_recursive_set_normalization_is_canonical() {
    let normalized = crate::db::query::preparation::with_preparation_work(|work| {
        normalize_value_for_accepted_kind(
            "tags",
            &Value::List(vec![
                Value::Text("beta".to_string()),
                Value::Text("alpha".to_string()),
                Value::Text("beta".to_string()),
            ]),
            &AcceptedFieldKind::Set(Box::new(AcceptedFieldKind::Text { max_len: None })),
            &CoercionSpec::new(CoercionId::Strict),
            CompareOp::Eq,
            work,
        )
    })
    .expect("accepted set normalization should succeed");

    assert_eq!(
        normalized,
        Value::List(vec![
            Value::Text("alpha".to_string()),
            Value::Text("beta".to_string()),
        ]),
    );
}

#[test]
fn accepted_field_comparison_coercion_uses_accepted_semantics() {
    assert_eq!(
        normalize_accepted_compare_fields_coercion(
            CompareOp::Eq,
            &AcceptedFieldKind::Int64,
            &AcceptedFieldKind::Nat64,
            CoercionId::Strict,
        ),
        CoercionId::NumericWiden,
    );
    assert_eq!(
        normalize_accepted_compare_fields_coercion(
            CompareOp::Lt,
            &AcceptedFieldKind::Text { max_len: None },
            &AcceptedFieldKind::Text { max_len: Some(32) },
            CoercionId::TextCasefold,
        ),
        CoercionId::Strict,
    );
}