armdb 0.4.1

use std::sync::Arc;

use armour_core::Fuid;
use armour_core::{CollectionKind, GetRefs, GetType, OnMissing};
use rapira::Rapira;
use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};

use crate::armour::db::Db;
use crate::key::Key;
use crate::schema::*;
use crate::{CollectionMeta, Config, NoHook};
use crate::{RapiraCodec, typed_map};

armour_core::const_hasher_or!(
    s_user,
    "SUSR",
    "K_zxJmNfxOS4R5DfaA2Dl42iHC4LjNaPTfLha6VKQoU1xtk91gdYVSNyl_o5iw1jqxx2zTqYqrQ",
    brand = "SUSR"
);
armour_core::const_hasher_or!(
    s_group,
    "SGRP",
    "au2voAmcasJn_OEcNTf9OZVhP0FEECk9l59i3On0yirc7FyIHMWeEQWqEui4bjEQtfqZkMRpejE",
    brand = "SGRP"
);

pub type SUserId = Fuid<s_user::Hasher>;
pub type SGroupId = Fuid<s_group::Hasher>;

/// Endian-safe Fuid from a native u64 (round-trips with `get()` / `visit_ids`).
fn fuid_from_native<H>(v: u64) -> Fuid<H> {
    Fuid::from_be_bytes(v.to_be_bytes())
}

fn suser(name: &str, mentor: SUserId) -> SUser {
    SUser {
        name: name.into(),
        mentor,
    }
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
pub struct SUser {
    pub name: String,
    /// Intra-entity FK (same brand as SelfId) — validated in value fields, not skipped.
    pub mentor: SUserId,
}
impl CollectionMeta for SUser {
    type SelfId = SUserId;
    const NAME: &'static str = "s_users";
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
pub struct SGroup {
    pub title: String,
    pub owner: SUserId,
    pub member_count: u32,
}
impl CollectionMeta for SGroup {
    type SelfId = SGroupId;
    const NAME: &'static str = "s_groups";
}

/// Db + typed collections for most schema tests.
#[allow(clippy::type_complexity)]
pub fn open_fixture(
    dir: &std::path::Path,
) -> (
    Arc<Db>,
    Arc<typed_map!(SUser, RapiraCodec)>,
    Arc<typed_map!(SGroup, RapiraCodec)>,
) {
    let db = Arc::new(Db::open_test(dir).expect("open db"));
    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("open users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("open groups");
    (db, users, groups)
}

#[test]
fn schema_mermaid_golden() {
    let dir = tempfile::tempdir().expect("tmp");
    let (_db, users, groups) = open_fixture(dir.path());

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register_entity(&groups);
    let built = reg.finish().expect("finish");

    let out = MermaidEr.format(&built.graph());
    let expected = "\
erDiagram
    s_users {
        Str name
        Fuid mentor
    }
    s_groups {
        Str title
        Fuid owner
        U32 member_count
    }
    s_users }o--|| s_users : \"fk mentor (SUSR)\"
    s_groups }o--|| s_users : \"fk owner (SUSR)\"
";
    assert_eq!(out, expected);
}

#[test]
fn schema_register_and_finish() {
    let dir = tempfile::tempdir().expect("tmp");
    let (_db, users, groups) = open_fixture(dir.path());

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register_entity(&groups);
    let built = reg.finish().expect("finish ok");

    let graph = built.graph();
    assert_eq!(graph.collections.len(), 2);
    let users_node = graph
        .collections
        .iter()
        .find(|c| c.name == "s_users")
        .expect("users node");
    assert_eq!(users_node.kind, CollectionKind::Entity);
    assert_eq!(users_node.self_brand.as_deref(), Some("SUSR"));
    assert!(
        graph
            .relations
            .iter()
            .any(|r| r.from == "s_groups" && r.to == "s_users")
    );
}

#[test]
fn schema_finish_rejects_duplicate_entity_brand() {
    let dir = tempfile::tempdir().expect("tmp");
    let (_db, users, _groups) = open_fixture(dir.path());

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SUserDup {
        pub bio: String,
    }
    impl CollectionMeta for SUserDup {
        type SelfId = SUserId;
        const NAME: &'static str = "s_users_dup";
    }
    let db2 = Db::open_test(dir.path().join("dup"))
        .map(Arc::new)
        .expect("db2");
    let dups = db2
        .open_typed_map::<SUserDup, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("open dup");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register_entity(&dups);
    assert!(
        reg.finish().is_err(),
        "duplicate entity brand must be rejected"
    );
}

#[test]
fn schema_validate_same_brand_value_fk_dangling() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, users, groups) = open_fixture(dir.path());

    let u1 = fuid_from_native::<s_user::Hasher>(1);
    users.insert(&u1, suser("a", u1)).expect("ins");
    users
        .insert(
            &fuid_from_native::<s_user::Hasher>(2),
            suser("b", fuid_from_native::<s_user::Hasher>(999)),
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register_entity(&groups);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::DanglingFk {
                collection,
                target,
                id,
                ..
            } if collection == "s_users" && target == "s_users" && *id == 999
        )),
        "{report}"
    );
}

#[test]
fn schema_validate_dangling_fk() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, users, groups) = open_fixture(dir.path());

    let alive = fuid_from_native::<s_user::Hasher>(101);
    users.insert(&alive, suser("a", alive)).expect("ins");
    groups
        .insert(
            &fuid_from_native::<s_group::Hasher>(1),
            SGroup {
                title: "ok".into(),
                owner: alive,
                member_count: 0,
            },
        )
        .expect("ins");
    groups
        .insert(
            &fuid_from_native::<s_group::Hasher>(2),
            SGroup {
                title: "bad".into(),
                owner: fuid_from_native::<s_user::Hasher>(999),
                member_count: 0,
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register_entity(&groups);
    let built = reg.finish().expect("finish");

    let report = validate(&db, &built);
    let dangling: Vec<_> = report
        .findings
        .iter()
        .filter(|f| matches!(f, Finding::DanglingFk { .. }))
        .collect();
    assert_eq!(dangling.len(), 1, "report: {report}");
    match dangling[0] {
        Finding::DanglingFk {
            collection,
            target,
            id,
            ..
        } => {
            assert_eq!(*collection, "s_groups".to_string());
            assert_eq!(*target, "s_users".to_string());
            assert_eq!(*id, 999);
        }
        _ => unreachable!(),
    }
}

#[test]
fn schema_validate_unclassified_collection() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, users, _groups) = open_fixture(dir.path());

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    // s_groups is open in Db but not registered
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::UnclassifiedCollection { name } if name == "s_groups"))
    );
}

/// Index key: owner -> group (inverse of s_groups.owner).
#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    FromBytes,
    IntoBytes,
    Immutable,
    KnownLayout,
    armour_core::GetRefs,
)]
#[repr(C)]
struct OwnerGroupKey {
    owner: SUserId,
    group: SGroupId,
}
crate::impl_key_zerocopy!(
    OwnerGroupKey,
    armour_core::KeyScheme::Typed(&[armour_core::KeyType::Fuid, armour_core::KeyType::Fuid])
);

#[derive(
    Clone, Copy, Debug, PartialEq, GetType, GetRefs, FromBytes, IntoBytes, Immutable, KnownLayout,
)]
#[repr(C)]
struct OwnerGroupMarker(pub u8);

impl CollectionMeta for OwnerGroupMarker {
    type SelfId = OwnerGroupKey;
    const NAME: &'static str = "s_owner_groups";
}

#[test]
fn schema_index_hook_mutation_matrix() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));

    let idx = db
        .open_zero_tree::<OwnerGroupMarker, { std::mem::size_of::<OwnerGroupMarker>() }, _>(
            Config::default(),
            NoHook,
            &[],
        )
        .expect("open idx");

    let mut reg = SchemaRegistry::new();
    reg.register(&idx, CollectionKind::SecondaryIndex);

    let hook = reg.secondary_index::<SGroup, _, _>(idx.clone(), |k: &SGroupId, v: &SGroup| {
        (
            OwnerGroupKey {
                owner: v.owner,
                group: *k,
            },
            OwnerGroupMarker(1),
        )
    });

    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::default(), hook, &[])
        .expect("open groups");
    reg.register_entity(&groups);

    let g = fuid_from_native::<s_group::Hasher>(1);
    let u1 = fuid_from_native::<s_user::Hasher>(10);
    let u2 = fuid_from_native::<s_user::Hasher>(20);

    groups
        .insert(
            &g,
            SGroup {
                title: "t".into(),
                owner: u1,
                member_count: 0,
            },
        )
        .expect("ins");
    assert!(idx.contains(&OwnerGroupKey {
        owner: u1,
        group: g
    }));

    groups
        .update(&g, |v| SGroup {
            title: "t2".into(),
            ..v.clone()
        })
        .expect("upd");
    assert!(idx.contains(&OwnerGroupKey {
        owner: u1,
        group: g
    }));

    groups
        .update(&g, |v| SGroup {
            owner: u2,
            ..v.clone()
        })
        .expect("upd2");
    assert!(
        !idx.contains(&OwnerGroupKey {
            owner: u1,
            group: g
        }),
        "old index row must be removed"
    );
    assert!(idx.contains(&OwnerGroupKey {
        owner: u2,
        group: g
    }));

    groups.delete(&g).expect("del");
    assert!(!idx.contains(&OwnerGroupKey {
        owner: u2,
        group: g
    }));
}

#[test]
fn schema_validate_index_sync() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let idx = db
        .open_zero_tree::<OwnerGroupMarker, { std::mem::size_of::<OwnerGroupMarker>() }, _>(
            Config::default(),
            NoHook,
            &[],
        )
        .expect("idx");

    let mut reg = SchemaRegistry::new();
    reg.register(&idx, CollectionKind::SecondaryIndex);
    // Hook not attached — desync is introduced manually.
    let _unused_hook = reg.secondary_index::<SGroup, _, _>(idx.clone(), |k, v: &SGroup| {
        (
            OwnerGroupKey {
                owner: v.owner,
                group: *k,
            },
            OwnerGroupMarker(1),
        )
    });
    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("groups");
    reg.register_entity(&users);
    reg.register_entity(&groups);

    let u = fuid_from_native::<s_user::Hasher>(1);
    users.insert(&u, suser("a", u)).expect("i");
    // missing index entry: source exists, index does not
    groups
        .insert(
            &fuid_from_native::<s_group::Hasher>(1),
            SGroup {
                title: "a".into(),
                owner: u,
                member_count: 0,
            },
        )
        .expect("i");
    // orphan index entry: index exists, source does not
    idx.insert(
        &OwnerGroupKey {
            owner: u,
            group: fuid_from_native::<s_group::Hasher>(777),
        },
        &OwnerGroupMarker(1),
    )
    .expect("i");

    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::IndexMissing { .. })),
        "{report}"
    );
    assert!(
        report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::IndexOrphan { .. })),
        "{report}"
    );
}

#[test]
fn schema_validate_unvalidated_edge_for_unknown_brand() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, _users, groups) = open_fixture(dir.path());

    let mut reg = SchemaRegistry::new();
    // s_users NOT registered as entity — SUSR brand has no target
    reg.register_entity(&groups);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::UnvalidatedEdge { .. }))
    );
}

/// Unbranded FK: u64 field referencing a user entity.
#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
pub struct SDevice {
    pub raw_owner: u64,
}
impl CollectionMeta for SDevice {
    type SelfId = [u8; 8];
    const NAME: &'static str = "s_devices";
}

#[test]
fn schema_validate_explicit_fk_unbranded() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, users, _groups) = open_fixture(dir.path());
    let devices = db
        .open_typed_map::<SDevice, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("devices");

    let alive = fuid_from_native::<s_user::Hasher>(5);
    users.insert(&alive, suser("x", alive)).expect("i");
    devices
        .insert(&[1u8; 8], SDevice { raw_owner: 5 })
        .expect("i");
    devices
        .insert(&[2u8; 8], SDevice { raw_owner: 6 })
        .expect("i"); // dangling

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register(&devices, CollectionKind::Standalone);
    reg.fk::<SDevice, SUser, _>(
        OnMissing::Error,
        FkLabel::field("raw_owner"),
        |_k: &[u8; 8], v: &SDevice| Some(fuid_from_native::<s_user::Hasher>(v.raw_owner)),
    );
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    let dangling: Vec<_> = report
        .findings
        .iter()
        .filter(|f| {
            matches!(
                f,
                Finding::DanglingFk {
                    collection,
                    ..
                } if collection == "s_devices"
            )
        })
        .collect();
    assert_eq!(dangling.len(), 1, "{report}");
}

#[test]
fn schema_validate_denormalized() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, users, _groups) = open_fixture(dir.path());

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SUserCard {
        pub name: String,
    }
    impl CollectionMeta for SUserCard {
        type SelfId = SUserId;
        const NAME: &'static str = "s_user_cards";
    }
    let cards = db
        .open_typed_map::<SUserCard, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("cards");

    let u1 = fuid_from_native::<s_user::Hasher>(1);
    let u2 = fuid_from_native::<s_user::Hasher>(2);
    users.insert(&u1, suser("a", u1)).expect("i");
    users.insert(&u2, suser("b", u2)).expect("i");
    cards
        .insert(&u1, SUserCard { name: "a".into() })
        .expect("i"); // ok
    // u2 has no card -> DenormMissing
    let u3 = fuid_from_native::<s_user::Hasher>(3);
    users.insert(&u3, suser("c", u3)).expect("i");
    cards
        .insert(
            &u3,
            SUserCard {
                name: "STALE".into(),
            },
        )
        .expect("i"); // DenormMismatch

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register(&cards, CollectionKind::Denormalized);
    reg.denormalized::<SUser, SUserCard, _, _>(
        |k: &SUserId, _v: &SUser| *k,
        |s: &SUser, d: &SUserCard| s.name == d.name,
    );
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::DenormMissing { .. })),
        "{report}"
    );
    assert!(
        report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::DenormMismatch { .. })),
        "{report}"
    );
}

#[test]
fn schema_validate_counter() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, users, groups) = open_fixture(dir.path());

    /// Membership: key = (group, user) — prefix by group.
    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        Eq,
        PartialOrd,
        Ord,
        Hash,
        FromBytes,
        IntoBytes,
        Immutable,
        KnownLayout,
        armour_core::GetRefs,
    )]
    #[repr(C)]
    struct SMemberKey {
        group: SGroupId,
        user: SUserId,
    }
    crate::impl_key_zerocopy!(
        SMemberKey,
        armour_core::KeyScheme::Typed(&[armour_core::KeyType::Fuid, armour_core::KeyType::Fuid])
    );

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        GetType,
        GetRefs,
        FromBytes,
        IntoBytes,
        Immutable,
        KnownLayout,
    )]
    #[repr(C)]
    struct SMember(pub u8);
    impl CollectionMeta for SMember {
        type SelfId = SMemberKey;
        const NAME: &'static str = "s_members";
    }
    let members = db
        .open_zero_tree::<SMember, { std::mem::size_of::<SMember>() }, _>(
            Config::default(),
            NoHook,
            &[],
        )
        .expect("members");

    let g = fuid_from_native::<s_group::Hasher>(1);
    let u1 = fuid_from_native::<s_user::Hasher>(1);
    users.insert(&u1, suser("a", u1)).expect("i");
    groups
        .insert(
            &g,
            SGroup {
                title: "g".into(),
                owner: u1,
                member_count: 2,
            },
        )
        .expect("i"); // wrong: actually 1 member
    members
        .insert(&SMemberKey { group: g, user: u1 }, &SMember(1))
        .expect("i");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register_entity(&groups);
    reg.register(&members, CollectionKind::SecondaryIndex);
    reg.counter::<SGroup, _, _>(
        "member_count",
        |v: &SGroup| v.member_count as u64,
        "s_members",
        |k: &SGroupId| Key::as_bytes(k).to_vec(),
    );
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    match report
        .findings
        .iter()
        .find(|f| matches!(f, Finding::CounterMismatch { .. }))
    {
        Some(Finding::CounterMismatch {
            stored, expected, ..
        }) => {
            assert_eq!((*stored, *expected), (2, 1));
        }
        other => panic!("expected CounterMismatch, got {other:?}; report: {report}"),
    }
}

/// Several sources feeding one index collection: the orphan pass must run
/// against the UNION of expected keys — rows projected by source B are not
/// orphans of source A.
#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct SGroupArchive {
    pub owner: SUserId,
}
impl CollectionMeta for SGroupArchive {
    type SelfId = SGroupId;
    const NAME: &'static str = "s_groups_archive";
}

#[test]
fn schema_validate_index_sync_multi_source() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let idx = db
        .open_zero_tree::<OwnerGroupMarker, { std::mem::size_of::<OwnerGroupMarker>() }, _>(
            Config::default(),
            NoHook,
            &[],
        )
        .expect("idx");

    let mut reg = SchemaRegistry::new();
    reg.register(&idx, CollectionKind::SecondaryIndex);
    // Two declarations into the same index, hooks not attached — rows inserted manually.
    let _h1 = reg.secondary_index::<SGroup, _, _>(idx.clone(), |k, v: &SGroup| {
        (
            OwnerGroupKey {
                owner: v.owner,
                group: *k,
            },
            OwnerGroupMarker(1),
        )
    });
    let _h2 = reg.secondary_index::<SGroupArchive, _, _>(idx.clone(), |k, v: &SGroupArchive| {
        (
            OwnerGroupKey {
                owner: v.owner,
                group: *k,
            },
            OwnerGroupMarker(1),
        )
    });

    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("groups");
    let archive = db
        .open_typed_map::<SGroupArchive, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("archive");
    reg.register_entity(&users);
    reg.register_entity(&groups);
    reg.register(&archive, CollectionKind::Denormalized);

    let u = fuid_from_native::<s_user::Hasher>(1);
    users.insert(&u, suser("a", u)).expect("i");
    let g1 = fuid_from_native::<s_group::Hasher>(1);
    let g2 = fuid_from_native::<s_group::Hasher>(2);
    groups
        .insert(
            &g1,
            SGroup {
                title: "live".into(),
                owner: u,
                member_count: 0,
            },
        )
        .expect("i");
    archive.insert(&g2, SGroupArchive { owner: u }).expect("i");
    // Index rows for both sources + one true orphan.
    idx.insert(
        &OwnerGroupKey {
            owner: u,
            group: g1,
        },
        &OwnerGroupMarker(1),
    )
    .expect("i");
    idx.insert(
        &OwnerGroupKey {
            owner: u,
            group: g2,
        },
        &OwnerGroupMarker(1),
    )
    .expect("i");
    let g_orphan = fuid_from_native::<s_group::Hasher>(99);
    idx.insert(
        &OwnerGroupKey {
            owner: u,
            group: g_orphan,
        },
        &OwnerGroupMarker(1),
    )
    .expect("i");

    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    let orphans: Vec<_> = report
        .findings
        .iter()
        .filter(|f| matches!(f, Finding::IndexOrphan { .. }))
        .collect();
    assert_eq!(
        orphans.len(),
        1,
        "only the true orphan is flagged: {report}"
    );
    assert!(
        !report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::IndexMissing { .. })),
        "{report}"
    );
}

/// P2 regression: an entity keyed by an unbranded `Fuid<()>` must not register the
/// empty brand `""` as an FK target. Two such entities must not collide as
/// `DuplicateEntityBrand("")`, and neither exports a `self_brand`.
#[test]
fn schema_unbranded_entity_key_is_not_a_brand_target() {
    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SAnonA {
        label: String,
    }
    impl CollectionMeta for SAnonA {
        type SelfId = Fuid<()>;
        const NAME: &'static str = "s_anon_a";
    }
    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SAnonB {
        label: String,
    }
    impl CollectionMeta for SAnonB {
        type SelfId = Fuid<()>;
        const NAME: &'static str = "s_anon_b";
    }

    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let a = db
        .open_typed_map::<SAnonA, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("open a");
    let b = db
        .open_typed_map::<SAnonB, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("open b");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&a);
    reg.register_entity(&b);
    // Before the fix both recorded brand "" and collided here.
    let built = reg
        .finish()
        .expect("unbranded entity keys must not be brand targets");

    let graph = built.graph();
    for name in ["s_anon_a", "s_anon_b"] {
        let node = graph
            .collections
            .iter()
            .find(|c| c.name == name)
            .expect("node");
        assert!(
            node.self_brand.is_none(),
            "unbranded entity must not export a brand: {name}"
        );
    }
}

#[cfg(feature = "var-collections")]
#[test]
fn schema_var_typed_entities_auto_fk() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));

    // users -> VarTypedTree (exercises iter()-based scan)
    let users = db
        .open_var_typed_tree::<SUser, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("open users");
    // groups -> VarTypedMap (exercises for_each-based scan)
    let groups = db
        .open_var_typed_map::<SGroup, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("open groups");

    let alive = fuid_from_native::<s_user::Hasher>(101);
    users.put(&alive, &suser("a", alive)).expect("put user");
    // Dangling mentor 777: only the VarTypedTree (users) iter-based scan can
    // surface this — guards against `VarTypedTree::scan` regressing to a no-op.
    users
        .put(
            &fuid_from_native::<s_user::Hasher>(102),
            &suser("bad-mentor", fuid_from_native::<s_user::Hasher>(777)),
        )
        .expect("put dangling user");
    groups
        .put(
            &fuid_from_native::<s_group::Hasher>(1),
            &SGroup {
                title: "ok".into(),
                owner: alive,
                member_count: 0,
            },
        )
        .expect("put ok group");
    groups
        .put(
            &fuid_from_native::<s_group::Hasher>(2),
            &SGroup {
                title: "bad".into(),
                owner: fuid_from_native::<s_user::Hasher>(999),
                member_count: 0,
            },
        )
        .expect("put bad group");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register_entity(&groups);
    let built = reg.finish().expect("finish");

    let report = validate(&db, &built);
    let dangling: Vec<_> = report
        .findings
        .iter()
        .filter(|f| matches!(f, Finding::DanglingFk { .. }))
        .collect();
    assert_eq!(dangling.len(), 2, "report: {report}");
    // From the users VarTypedTree (iter scan): mentor 777.
    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::DanglingFk { collection, target, id, .. }
                if collection == "s_users" && target == "s_users" && *id == 777
        )),
        "missing users-tree dangling FK; report: {report}"
    );
    // From the groups VarTypedMap (for_each scan): owner 999.
    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::DanglingFk { collection, target, id, .. }
                if collection == "s_groups" && target == "s_users" && *id == 999
        )),
        "missing groups-map dangling FK; report: {report}"
    );
}

/// Codec-serialized index marker (Var indexes use RapiraCodec, not zerocopy).
#[cfg(feature = "var-collections")]
#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct VarMarker(#[rapira(with = rapira::byte_rapira)] pub u8);

#[cfg(feature = "var-collections")]
impl CollectionMeta for VarMarker {
    type SelfId = OwnerGroupKey;
    const NAME: &'static str = "s_var_owner_groups";
}

#[cfg(feature = "var-collections")]
#[test]
fn schema_var_typed_tree_index_target() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));

    let idx = db
        .open_var_typed_tree::<VarMarker, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("open idx");

    let mut reg = SchemaRegistry::new();
    reg.register(&idx, CollectionKind::SecondaryIndex);
    let hook = reg.secondary_index::<SGroup, _, _>(idx.clone(), |k: &SGroupId, v: &SGroup| {
        (
            OwnerGroupKey {
                owner: v.owner,
                group: *k,
            },
            VarMarker(1),
        )
    });

    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::default(), hook, &[])
        .expect("open groups");

    let g = fuid_from_native::<s_group::Hasher>(1);
    let u1 = fuid_from_native::<s_user::Hasher>(10);
    let u2 = fuid_from_native::<s_user::Hasher>(20);

    groups
        .insert(
            &g,
            SGroup {
                title: "t".into(),
                owner: u1,
                member_count: 0,
            },
        )
        .expect("ins");
    assert!(
        idx.contains(&OwnerGroupKey {
            owner: u1,
            group: g
        }),
        "put_entry"
    );

    groups
        .update(&g, |v| SGroup {
            owner: u2,
            ..v.clone()
        })
        .expect("upd");
    assert!(
        !idx.contains(&OwnerGroupKey {
            owner: u1,
            group: g
        }),
        "delete_entry of old key"
    );
    assert!(
        idx.contains(&OwnerGroupKey {
            owner: u2,
            group: g
        }),
        "put_entry of new key"
    );

    groups.delete(&g).expect("del");
    assert!(
        !idx.contains(&OwnerGroupKey {
            owner: u2,
            group: g
        }),
        "delete_entry"
    );
}

#[cfg(feature = "var-collections")]
#[test]
fn schema_var_typed_map_index_target() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));

    let idx = db
        .open_var_typed_map::<VarMarker, RapiraCodec, _>(Config::default(), NoHook, &[])
        .expect("open idx");

    let mut reg = SchemaRegistry::new();
    reg.register(&idx, CollectionKind::SecondaryIndex);
    let hook = reg.secondary_index::<SGroup, _, _>(idx.clone(), |k: &SGroupId, v: &SGroup| {
        (
            OwnerGroupKey {
                owner: v.owner,
                group: *k,
            },
            VarMarker(1),
        )
    });

    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::default(), hook, &[])
        .expect("open groups");

    let g = fuid_from_native::<s_group::Hasher>(1);
    let u1 = fuid_from_native::<s_user::Hasher>(10);
    let u2 = fuid_from_native::<s_user::Hasher>(20);

    groups
        .insert(
            &g,
            SGroup {
                title: "t".into(),
                owner: u1,
                member_count: 0,
            },
        )
        .expect("ins");
    assert!(
        idx.contains(&OwnerGroupKey {
            owner: u1,
            group: g
        }),
        "put_entry"
    );

    // Owner change relocates the index entry: delete old key + put new key.
    groups
        .update(&g, |v| SGroup {
            owner: u2,
            ..v.clone()
        })
        .expect("upd");
    assert!(
        !idx.contains(&OwnerGroupKey {
            owner: u1,
            group: g
        }),
        "delete_entry of old key"
    );
    assert!(
        idx.contains(&OwnerGroupKey {
            owner: u2,
            group: g
        }),
        "put_entry of new key"
    );

    groups.delete(&g).expect("del");
    assert!(
        !idx.contains(&OwnerGroupKey {
            owner: u2,
            group: g
        }),
        "delete_entry"
    );
}

/// Codec that decodes every SUser except those named "POISON" (simulates a
/// row whose value can no longer be decoded).
#[cfg(feature = "var-collections")]
#[derive(Clone, Default)]
struct PoisonNameCodec;

#[cfg(feature = "var-collections")]
impl crate::Codec<SUser> for PoisonNameCodec {
    fn encode_to(&self, value: &SUser, buf: &mut Vec<u8>) -> crate::DbResult<()> {
        buf.clear();
        rapira::extend_vec(value, buf);
        Ok(())
    }

    fn decode_from(&self, bytes: &[u8]) -> crate::DbResult<SUser> {
        let user: SUser = unsafe {
            rapira::deser_unchecked(bytes)
                .map_err(|_| crate::DbError::CorruptedEntry { offset: 0 })?
        };
        if user.name == "POISON" {
            return Err(crate::DbError::CorruptedEntry { offset: 0 });
        }
        Ok(user)
    }
}

#[cfg(feature = "var-collections")]
#[test]
fn schema_var_scan_skips_undecodable_rows() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));

    let users = db
        .open_var_typed_map::<SUser, PoisonNameCodec, _>(Config::default(), NoHook, &[])
        .expect("open users");

    let u1 = fuid_from_native::<s_user::Hasher>(1);
    let u2 = fuid_from_native::<s_user::Hasher>(2);
    let u3 = fuid_from_native::<s_user::Hasher>(3);

    // valid; mentor points at existing u1
    users
        .put(
            &u1,
            &SUser {
                name: "ok".into(),
                mentor: u1,
            },
        )
        .expect("put u1");
    // valid decode, dangling mentor 999 -> MUST be reported
    users
        .put(
            &u3,
            &SUser {
                name: "ok3".into(),
                mentor: fuid_from_native::<s_user::Hasher>(999),
            },
        )
        .expect("put u3");
    // decode fails (name == "POISON"), dangling mentor 888 -> MUST be skipped
    users
        .put(
            &u2,
            &SUser {
                name: "POISON".into(),
                mentor: fuid_from_native::<s_user::Hasher>(888),
            },
        )
        .expect("put u2");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    let built = reg.finish().expect("finish");

    let report = validate(&db, &built);
    let dangling: Vec<u64> = report
        .findings
        .iter()
        .filter_map(|f| match f {
            Finding::DanglingFk { id, .. } => Some(*id),
            _ => None,
        })
        .collect();
    // u3's dangling mentor (999) is found; the POISON row is skipped, so 888 never appears.
    assert_eq!(dangling, vec![999], "report: {report}");
}

#[cfg(feature = "var-collections")]
#[test]
fn schema_var_tree_scan_skips_undecodable_rows() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));

    // VarTypedTree::scan has its own decode loop (as_inner().iter() + warn),
    // distinct from VarTypedMap — exercise its decode-skip path too.
    let users = db
        .open_var_typed_tree::<SUser, PoisonNameCodec, _>(Config::default(), NoHook, &[])
        .expect("open users");

    let u1 = fuid_from_native::<s_user::Hasher>(1);
    let u2 = fuid_from_native::<s_user::Hasher>(2);
    let u3 = fuid_from_native::<s_user::Hasher>(3);

    // valid; mentor points at existing u1
    users
        .put(
            &u1,
            &SUser {
                name: "ok".into(),
                mentor: u1,
            },
        )
        .expect("put u1");
    // valid decode, dangling mentor 999 -> MUST be reported
    users
        .put(
            &u3,
            &SUser {
                name: "ok3".into(),
                mentor: fuid_from_native::<s_user::Hasher>(999),
            },
        )
        .expect("put u3");
    // decode fails (name == "POISON"), dangling mentor 888 -> MUST be skipped
    users
        .put(
            &u2,
            &SUser {
                name: "POISON".into(),
                mentor: fuid_from_native::<s_user::Hasher>(888),
            },
        )
        .expect("put u2");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    let built = reg.finish().expect("finish");

    let report = validate(&db, &built);
    let dangling: Vec<u64> = report
        .findings
        .iter()
        .filter_map(|f| match f {
            Finding::DanglingFk { id, .. } => Some(*id),
            _ => None,
        })
        .collect();
    // u3's dangling mentor (999) is found; the POISON row is skipped, so 888 never appears.
    assert_eq!(dangling, vec![999], "report: {report}");
}

// --- Full storage-class matrix (spec §Тестирование, review R2) -------------
// 8 distinct value types so each opened collection has a unique NAME.
// Keys reuse SUserId (Fuid) — registered as Standalone, so no brand target is
// created and the graph carries no FK edges, isolating the storage assertions.
// Gated on `var-collections` (a default feature) because the matrix opens the
// VarTyped* collections; the fixtures live inside the gated module so the
// `schema`-without-`var-collections` build stays warning-clean.
#[cfg(feature = "var-collections")]
mod storage_matrix {
    use super::*;
    use armour_core::{CollectionShape, StorageBackend, StorageClass, ValueModel};

    use crate::FixedConfig;

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct MTypedTree {
        x: u32,
    }
    impl CollectionMeta for MTypedTree {
        type SelfId = SUserId;
        const NAME: &'static str = "m_typed_tree";
    }

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct MTypedMap {
        x: u32,
    }
    impl CollectionMeta for MTypedMap {
        type SelfId = SUserId;
        const NAME: &'static str = "m_typed_map";
    }

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct MVarTree {
        x: u32,
    }
    impl CollectionMeta for MVarTree {
        type SelfId = SUserId;
        const NAME: &'static str = "m_var_tree";
    }

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct MVarMap {
        x: u32,
    }
    impl CollectionMeta for MVarMap {
        type SelfId = SUserId;
        const NAME: &'static str = "m_var_map";
    }

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        GetType,
        GetRefs,
        FromBytes,
        IntoBytes,
        Immutable,
        KnownLayout,
    )]
    #[repr(C)]
    struct MZeroTreeBc(u32);
    impl CollectionMeta for MZeroTreeBc {
        type SelfId = SUserId;
        const NAME: &'static str = "m_zero_tree_bc";
    }

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        GetType,
        GetRefs,
        FromBytes,
        IntoBytes,
        Immutable,
        KnownLayout,
    )]
    #[repr(C)]
    struct MZeroMapBc(u32);
    impl CollectionMeta for MZeroMapBc {
        type SelfId = SUserId;
        const NAME: &'static str = "m_zero_map_bc";
    }

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        GetType,
        GetRefs,
        FromBytes,
        IntoBytes,
        Immutable,
        KnownLayout,
    )]
    #[repr(C)]
    struct MZeroTreeFx(u32);
    impl CollectionMeta for MZeroTreeFx {
        type SelfId = SUserId;
        const NAME: &'static str = "m_zero_tree_fx";
    }

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        GetType,
        GetRefs,
        FromBytes,
        IntoBytes,
        Immutable,
        KnownLayout,
    )]
    #[repr(C)]
    struct MZeroMapFx(u32);
    impl CollectionMeta for MZeroMapFx {
        type SelfId = SUserId;
        const NAME: &'static str = "m_zero_map_fx";
    }

    #[test]
    fn schema_storage_class_matrix() {
        let dir = tempfile::tempdir().expect("tmp");
        let db = Arc::new(Db::open_test(dir.path()).expect("db"));

        let typed_tree = db
            .open_typed_tree::<MTypedTree, RapiraCodec, _>(Config::default(), NoHook, &[])
            .expect("typed tree");
        let typed_map = db
            .open_typed_map::<MTypedMap, RapiraCodec, _>(Config::default(), NoHook, &[])
            .expect("typed map");
        let var_tree = db
            .open_var_typed_tree::<MVarTree, RapiraCodec, _>(Config::default(), NoHook, &[])
            .expect("var tree");
        let var_map = db
            .open_var_typed_map::<MVarMap, RapiraCodec, _>(Config::default(), NoHook, &[])
            .expect("var map");
        let zero_tree_bc = db
            .open_zero_tree::<MZeroTreeBc, { std::mem::size_of::<MZeroTreeBc>() }, _>(
                Config::default(),
                NoHook,
                &[],
            )
            .expect("zero tree bitcask");
        let zero_map_bc = db
            .open_zero_map::<MZeroMapBc, { std::mem::size_of::<MZeroMapBc>() }, _>(
                Config::default(),
                NoHook,
                &[],
            )
            .expect("zero map bitcask");
        let zero_tree_fx = db
            .open_zero_tree_fixed::<MZeroTreeFx, { std::mem::size_of::<MZeroTreeFx>() }, _>(
                FixedConfig::test(),
                NoHook,
                &[],
            )
            .expect("zero tree fixed");
        let zero_map_fx = db
            .open_zero_map_fixed::<MZeroMapFx, { std::mem::size_of::<MZeroMapFx>() }, _>(
                FixedConfig::test(),
                NoHook,
                &[],
            )
            .expect("zero map fixed");

        let mut reg = SchemaRegistry::new();
        reg.register(&typed_tree, CollectionKind::Standalone);
        reg.register(&typed_map, CollectionKind::Standalone);
        reg.register(&var_tree, CollectionKind::Standalone);
        reg.register(&var_map, CollectionKind::Standalone);
        reg.register(&zero_tree_bc, CollectionKind::Standalone);
        reg.register(&zero_map_bc, CollectionKind::Standalone);
        reg.register(&zero_tree_fx, CollectionKind::Standalone);
        reg.register(&zero_map_fx, CollectionKind::Standalone);
        reg.register_external("m_external", CollectionKind::Standalone);
        let built = reg.finish().expect("finish");
        let graph = built.graph();

        let storage_of = |name: &str| -> Option<StorageClass> {
            graph
                .collections
                .iter()
                .find(|c| c.name == name)
                .unwrap_or_else(|| panic!("node {name} missing"))
                .storage
        };
        let sc = |model, shape, backend| {
            Some(StorageClass {
                model,
                shape,
                backend,
            })
        };

        assert_eq!(
            storage_of("m_typed_tree"),
            sc(
                ValueModel::Typed,
                CollectionShape::Tree,
                StorageBackend::Bitcask
            )
        );
        assert_eq!(
            storage_of("m_typed_map"),
            sc(
                ValueModel::Typed,
                CollectionShape::Map,
                StorageBackend::Bitcask
            )
        );
        assert_eq!(
            storage_of("m_var_tree"),
            sc(
                ValueModel::VarTyped,
                CollectionShape::Tree,
                StorageBackend::Bitcask
            )
        );
        assert_eq!(
            storage_of("m_var_map"),
            sc(
                ValueModel::VarTyped,
                CollectionShape::Map,
                StorageBackend::Bitcask
            )
        );
        assert_eq!(
            storage_of("m_zero_tree_bc"),
            sc(
                ValueModel::Zero,
                CollectionShape::Tree,
                StorageBackend::Bitcask
            )
        );
        assert_eq!(
            storage_of("m_zero_map_bc"),
            sc(
                ValueModel::Zero,
                CollectionShape::Map,
                StorageBackend::Bitcask
            )
        );
        assert_eq!(
            storage_of("m_zero_tree_fx"),
            sc(
                ValueModel::Zero,
                CollectionShape::Tree,
                StorageBackend::Fixed
            )
        );
        assert_eq!(
            storage_of("m_zero_map_fx"),
            sc(
                ValueModel::Zero,
                CollectionShape::Map,
                StorageBackend::Fixed
            )
        );

        // External node: no handle -> None.
        assert_eq!(storage_of("m_external"), None);

        // JSON smoke: model is serializable for a future GUI / RPC GetSchema.
        let json = serde_json::to_string(&graph).expect("serialize graph");
        assert!(json.contains("\"model\""), "{json}");
        assert!(json.contains("\"backend\""), "{json}");
    }
}