armdb 0.12.0

use std::sync::Arc;

use crate::schema::validate::{AmbiguityReason, RefDesc};
use armour_core::Fuid;
use armour_core::{
    CollectionKind, GetRefs, GetType, KeyShape, OnMissing, PathSeg, RefVisitor, RowPart,
};
use rapira::Rapira;
use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};

use crate::armour::db::Db;
use crate::counter_tests::{
    Group, GroupsMap, Member, MemberKey, MembersMap, group_counter_targets,
};
use crate::key::Key;
use crate::schema::*;
use crate::{CollectionMeta, Config, FixedConfig, IndexTargets, NoHook};
use crate::{RapiraCodec, typed_map};
use crate::{UniqueIndex, UniqueTargets};
// Only named in the `var-collections`-gated `VarMembersMap` alias below — a
// non-default build without that feature must not see it as unused.
#[cfg(feature = "var-collections")]
use crate::Projected;

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"
);

armour_core::const_hasher_or!(
    s_chat,
    "KCHAT",
    "K_zxJmNfxOS4R5DfaA2Dl42iHC4LjNaPTfLha6VKQoU1xtk91gdYVSNyl_o5iw1jqxx2zTqYqrQ",
    brand = "chat"
);
armour_core::const_hasher_or!(
    s_msg,
    "KMSG0",
    "au2voAmcasJn_OEcNTf9OZVhP0FEECk9l59i3On0yirc7FyIHMWeEQWqEui4bjEQtfqZkMRpejE",
    brand = "msg"
);
armour_core::const_hasher_or!(
    s_space,
    "SSPC0",
    "K_zxJmNfxOS4R5DfaA2Dl42iHC4LjNaPTfLha6VKQoU1xtk91gdYVSNyl_o5iw1jqxx2zTqYqrQ",
    brand = "space"
);

pub type SUserId = Fuid<s_user::Hasher>;
pub type SGroupId = Fuid<s_group::Hasher>;
pub type SChatId = Fuid<s_chat::Hasher>;
pub type SMsgId = Fuid<s_msg::Hasher>;
pub type SSpaceId = Fuid<s_space::Hasher>;

#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    FromBytes,
    IntoBytes,
    Immutable,
    KnownLayout,
    GetRefs,
    GetType,
    KeyShape,
)]
#[repr(C)]
struct SMessageKey {
    chat_id: SChatId,
    msg_id: SMsgId,
}
crate::impl_key_zerocopy!(
    SMessageKey,
    armour_core::KeyScheme::Typed(&[armour_core::KeyType::Fuid, armour_core::KeyType::Fuid])
);

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct SChat {
    title: String,
}
impl CollectionMeta for SChat {
    type SelfId = SChatId;
    const NAME: &'static str = "s_chats";
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct SMessage {
    #[get_type(ref_with_key)]
    reply_to: Option<SMsgId>,
    text: String,
}
impl CollectionMeta for SMessage {
    type SelfId = SMessageKey;
    const NAME: &'static str = "s_messages";
}

fn open_chat_message_fixture(
    dir: &std::path::Path,
) -> (
    Arc<Db>,
    Arc<typed_map!(SChat, RapiraCodec)>,
    Arc<typed_map!(SMessage, RapiraCodec)>,
) {
    let db = Arc::new(Db::open_test(dir).expect("open db"));
    let chats = db
        .open_typed_map::<SChat, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("open chats");
    let messages = db
        .open_typed_map::<SMessage, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("open messages");
    (db, chats, messages)
}

#[cfg(feature = "var-collections")]
mod index_target_bounds {
    use super::*;

    #[derive(Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct NonCloneVal {
        value: u64,
    }

    impl CollectionMeta for NonCloneVal {
        type SelfId = SUserId;
        const NAME: &'static str = "non_clone_values";
    }

    fn assert_index_target<T: IndexTarget>() {}

    #[test]
    fn non_clone_var_typed_tree_implements_index_target() {
        assert_index_target::<crate::VarTypedTree<SUserId, NonCloneVal, RapiraCodec, NoHook>>();
    }

    #[test]
    fn clone_typed_tree_implements_index_target() {
        assert_index_target::<crate::TypedTree<SUserId, SUser, RapiraCodec, NoHook>>();
    }
}

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

fn assert_no_errors(report: &ValidationReport, context: &str) {
    assert!(report.errors().next().is_none(), "{context}: {report}");
}

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::balanced().build(), NoHook, &[])
        .expect("open users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), 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_shape.as_deref(),
        Some(&["SUSR".to_string()][..])
    );
    assert!(
        graph
            .relations
            .iter()
            .any(|r| r.from == "s_groups" && r.to == "s_users")
    );
}

#[test]
fn schema_finish_allows_multi_home_same_shape() {
    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 dups = db
        .open_typed_map::<SUserDup, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("open dup");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register_entity(&dups);
    reg.register_entity(&groups);
    let built = reg
        .finish()
        .expect("two homes of shape [SUSR] must be allowed");

    let only_in_dup = fuid_from_native::<s_user::Hasher>(42);
    dups.insert(&only_in_dup, SUserDup { bio: "dup".into() })
        .expect("ins");
    groups
        .insert(
            &fuid_from_native::<s_group::Hasher>(7),
            SGroup {
                title: "g".into(),
                owner: only_in_dup,
                member_count: 0,
            },
        )
        .expect("ins");

    let report = validate(&db, &built);
    assert_no_errors(
        &report,
        "FK satisfied via second home when id is absent from first",
    );
}

#[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,
                reference: RefDesc::Assembled { ids, .. },
                ..
            } if collection == "s_users" && ids.as_slice() == [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,
            reference: RefDesc::Assembled { ids, .. },
            homes,
            ..
        } => {
            assert_eq!(*collection, "s_groups".to_string());
            assert_eq!(homes, &vec!["s_users".to_string()]);
            assert_eq!(*ids, vec![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,
    armour_core::GetType,
)]
#[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,
    Default,
    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::balanced().build(),
            NoHook,
            &[],
        )
        .expect("open idx");

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

    let hook = reg.secondary_index::<SGroup, _>(
        IndexTargets::new().with_index(&idx),
        owner_group_projection(Arc::clone(&idx)),
    );

    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), 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::balanced().build(),
            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, _>(
        IndexTargets::new().with_index(&idx),
        owner_group_projection(Arc::clone(&idx)),
    );
    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), 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::balanced().build(), 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::balanced().build(), 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}"
    );
}

// `Other` is a second source type over the same key space as `Member` (the
// shared T9 fixture, `crate::counter_tests`) — a distinct source so two
// `counted` declarations can name the same `(target, field)`.
#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct Other {
    group: [u8; 8],
}
impl CollectionMeta for Other {
    type SelfId = [u8; 8];
    const NAME: &'static str = "others";
}

// Index-target-shaped value, same shape `OwnerGroupMarker` above uses — only
// ever opened, never inserted into, by the cascade test below.
#[derive(
    Clone,
    Copy,
    Debug,
    Default,
    PartialEq,
    GetType,
    GetRefs,
    FromBytes,
    IntoBytes,
    Immutable,
    KnownLayout,
)]
#[repr(C)]
struct Marker(pub u8);
impl CollectionMeta for Marker {
    type SelfId = [u8; 8];
    const NAME: &'static str = "counter_marker";
}

#[test]
fn schema_counted_declaration_rejects_two_sources_for_one_field() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("groups");
    let mut reg = SchemaRegistry::new();
    reg.register(&groups, CollectionKind::Entity);

    // Two different sources declaring the same (target, field) is the case the
    // open-time recount cannot serve: one scan's sum is only an addend.
    let g1 = Arc::clone(&groups);
    let _h1 = reg.counted::<Member, _>(group_counter_targets(&groups), move |_k, v: &Member, s| {
        s.add(&g1, v.group, 1)
    });
    let g2 = Arc::clone(&groups);
    let _h2 = reg.counted::<Other, _>(group_counter_targets(&groups), move |_k, v: &Other, s| {
        s.add(&g2, v.group, 1)
    });

    assert!(matches!(
        reg.finish(),
        Err(RegistryBuildError::DuplicateCounterSource {
            field: "member_count",
            ..
        })
    ));
}

#[test]
fn schema_counted_declaration_rejects_a_cascade() {
    // `groups` is a counter target, so it may not also be a projection source.
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("groups");
    let idx = db
        .open_zero_tree::<Marker, { std::mem::size_of::<Marker>() }, _>(Config::test(), NoHook, &[])
        .expect("idx");
    let mut reg = SchemaRegistry::new();
    let g = Arc::clone(&groups);
    let _counted = reg
        .counted::<Member, _>(group_counter_targets(&groups), move |_k, v: &Member, s| {
            s.add(&g, v.group, 1)
        });
    let res = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        // Same rejection `reject_index_decl` already performs for indexes.
        reg.secondary_index::<Group, _>(IndexTargets::new().with_index(&idx), |_k, _v, _sink| {
            Ok(())
        });
    }));
    assert!(
        res.is_err(),
        "a counter target cannot become a projection source"
    );
}

/// 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::balanced().build(),
            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, _>(
        IndexTargets::new().with_index(&idx),
        owner_group_projection(Arc::clone(&idx)),
    );
    let _h2 = reg.secondary_index::<SGroupArchive, _>(
        IndexTargets::new().with_index(&idx),
        archive_group_projection(Arc::clone(&idx)),
    );

    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("groups");
    let archive = db
        .open_typed_map::<SGroupArchive, RapiraCodec, _>(Config::balanced().build(), 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}"
    );
}

type OwnerGroupIdx = crate::ZeroTree<
    OwnerGroupKey,
    { std::mem::size_of::<OwnerGroupMarker>() },
    OwnerGroupMarker,
    crate::TtlHook<OwnerGroupMarker, NoHook>,
>;

fn open_owner_group_index(db: &Arc<Db>) -> Arc<OwnerGroupIdx> {
    db.open_zero_tree::<OwnerGroupMarker, { std::mem::size_of::<OwnerGroupMarker>() }, _>(
        Config::balanced().build(),
        NoHook,
        &[],
    )
    .expect("idx")
}

fn owner_group_projection(
    idx: Arc<OwnerGroupIdx>,
) -> impl Fn(&SGroupId, &SGroup, &mut crate::ProjectionSink<'_>) -> crate::error::DbResult<()>
+ Clone
+ 'static {
    move |k, v: &SGroup, sink| {
        sink.row(
            &idx,
            OwnerGroupKey {
                owner: v.owner,
                group: *k,
            },
            OwnerGroupMarker(1),
        )?;
        Ok(())
    }
}

fn archive_group_projection(
    idx: Arc<OwnerGroupIdx>,
) -> impl Fn(&SGroupId, &SGroupArchive, &mut crate::ProjectionSink<'_>) -> crate::error::DbResult<()>
+ Clone
+ 'static {
    move |k, v: &SGroupArchive, sink| {
        sink.row(
            &idx,
            OwnerGroupKey {
                owner: v.owner,
                group: *k,
            },
            OwnerGroupMarker(1),
        )?;
        Ok(())
    }
}

#[test]
#[should_panic(expected = "secondary_index: IndexTargets must declare at least one target")]
fn secondary_index_rejects_empty_targets() {
    let mut reg = SchemaRegistry::new();
    let _ = reg.secondary_index::<SGroup, _>(IndexTargets::new(), |_, _, _| Ok(()));
}

#[test]
#[should_panic(expected = "secondary_index: target `s_groups` must not equal source `s_groups`")]
fn secondary_index_rejects_source_as_target() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("groups");
    let mut reg = SchemaRegistry::new();
    let _ =
        reg.secondary_index::<SGroup, _>(IndexTargets::new().with_index(&groups), |_, _, _| Ok(()));
}

#[test]
#[should_panic(
    expected = "secondary_index: target `s_groups` is already a projecting source (cascade forbidden)"
)]
fn secondary_index_rejects_target_that_is_projecting_source() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let idx = open_owner_group_index(&db);
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("groups");
    let mut reg = SchemaRegistry::new();
    let _ = reg.secondary_index::<SGroup, _>(
        IndexTargets::new().with_index(&idx),
        owner_group_projection(Arc::clone(&idx)),
    );
    let _ =
        reg.secondary_index::<SUser, _>(IndexTargets::new().with_index(&groups), |_, _, _| Ok(()));
}

#[test]
#[should_panic(
    expected = "secondary_index: source `s_owner_groups` is already a projection target (cascade forbidden)"
)]
fn secondary_index_rejects_source_that_is_projection_target() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let idx = open_owner_group_index(&db);
    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("users");
    let mut reg = SchemaRegistry::new();
    let _ = reg.secondary_index::<SGroup, _>(
        IndexTargets::new().with_index(&idx),
        owner_group_projection(Arc::clone(&idx)),
    );
    let _ = reg.secondary_index::<OwnerGroupMarker, _>(
        IndexTargets::new().with_index(&users),
        |_, _, _| Ok(()),
    );
}

#[test]
fn index_key_collision_two_source_keys() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let idx = open_owner_group_index(&db);
    let mut reg = SchemaRegistry::new();
    reg.register(&idx, CollectionKind::SecondaryIndex);
    let _ = reg.secondary_index::<SGroup, _>(IndexTargets::new().with_index(&idx), {
        let idx = Arc::clone(&idx);
        move |k: &SGroupId, v: &SGroup, sink| {
            let _ = k;
            let collide = fuid_from_native::<s_group::Hasher>(0);
            sink.row(
                &idx,
                OwnerGroupKey {
                    owner: v.owner,
                    group: collide,
                },
                OwnerGroupMarker(1),
            )?;
            Ok(())
        }
    });
    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), 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");
    let g1 = fuid_from_native::<s_group::Hasher>(1);
    let g2 = fuid_from_native::<s_group::Hasher>(2);
    groups
        .insert(
            &g1,
            SGroup {
                title: "a".into(),
                owner: u,
                member_count: 0,
            },
        )
        .expect("i");
    groups
        .insert(
            &g2,
            SGroup {
                title: "b".into(),
                owner: u,
                member_count: 0,
            },
        )
        .expect("i");

    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::IndexKeyCollision { index, owners, .. }
                if index == "s_owner_groups" && owners.len() == 2
        )),
        "{report}"
    );
}

#[test]
fn index_key_collision_two_source_collections() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let idx = open_owner_group_index(&db);
    let mut reg = SchemaRegistry::new();
    reg.register(&idx, CollectionKind::SecondaryIndex);
    let _ = reg.secondary_index::<SGroup, _>(
        IndexTargets::new().with_index(&idx),
        owner_group_projection(Arc::clone(&idx)),
    );
    let _ = reg.secondary_index::<SGroupArchive, _>(
        IndexTargets::new().with_index(&idx),
        archive_group_projection(Arc::clone(&idx)),
    );
    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("groups");
    let archive = db
        .open_typed_map::<SGroupArchive, RapiraCodec, _>(Config::balanced().build(), 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 g = fuid_from_native::<s_group::Hasher>(42);
    groups
        .insert(
            &g,
            SGroup {
                title: "live".into(),
                owner: u,
                member_count: 0,
            },
        )
        .expect("i");
    archive.insert(&g, SGroupArchive { owner: u }).expect("i");

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

#[test]
fn orphan_is_warning_until_complete_index() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let idx = open_owner_group_index(&db);
    let mut reg = SchemaRegistry::new();
    reg.register(&idx, CollectionKind::SecondaryIndex);
    let _ = reg.secondary_index::<SGroup, _>(
        IndexTargets::new().with_index(&idx),
        owner_group_projection(Arc::clone(&idx)),
    );
    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), 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");
    idx.insert(
        &OwnerGroupKey {
            owner: u,
            group: fuid_from_native::<s_group::Hasher>(99),
        },
        &OwnerGroupMarker(1),
    )
    .expect("orphan");

    let built = reg.finish().expect("finish");
    assert!(!built.is_complete_index(OwnerGroupMarker::NAME));
    assert!(!built.is_complete_index(SGroup::NAME));
    let report = validate(&db, &built);
    let orphan = report
        .findings
        .iter()
        .find(|f| matches!(f, Finding::IndexOrphan { .. }))
        .expect("orphan finding");
    assert!(!orphan.is_error(), "incomplete target: orphan is warning");

    let mut reg2 = SchemaRegistry::new();
    reg2.register(&idx, CollectionKind::SecondaryIndex);
    reg2.complete_index(&idx);
    let _ = reg2.secondary_index::<SGroup, _>(
        IndexTargets::new().with_index(&idx),
        owner_group_projection(Arc::clone(&idx)),
    );
    reg2.register_entity(&users);
    reg2.register_entity(&groups);
    let built2 = reg2.finish().expect("finish");
    assert!(built2.is_complete_index(OwnerGroupMarker::NAME));
    assert!(!built2.is_complete_index(SGroup::NAME));
    let report2 = validate(&db, &built2);
    let orphan2 = report2
        .findings
        .iter()
        .find(|f| matches!(f, Finding::IndexOrphan { .. }))
        .expect("orphan finding");
    assert!(orphan2.is_error(), "complete target: orphan is error");
}

#[test]
fn repair_preserves_orphans_without_opt_in() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let idx = open_owner_group_index(&db);
    let mut reg = SchemaRegistry::new();
    reg.register(&idx, CollectionKind::SecondaryIndex);
    reg.complete_index(&idx);
    let _ = reg.secondary_index::<SGroup, _>(
        IndexTargets::new().with_index(&idx),
        owner_group_projection(Arc::clone(&idx)),
    );
    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), 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");
    let orphan_key = OwnerGroupKey {
        owner: u,
        group: fuid_from_native::<s_group::Hasher>(99),
    };
    idx.insert(&orphan_key, &OwnerGroupMarker(1))
        .expect("orphan");

    let built = reg.finish().expect("finish");
    let report = repair(&db, &built, RepairOptions::default()).expect("repair");
    let stats = report
        .per_index
        .iter()
        .find(|s| s.name == "s_owner_groups")
        .expect("stats");
    assert_eq!(stats.deletes, 0);
    assert_eq!(stats.declined_orphans, 1);
    assert!(idx.contains(&orphan_key));
}

#[test]
fn repair_deletes_orphans_when_complete_and_opt_in() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let idx = open_owner_group_index(&db);
    let mut reg = SchemaRegistry::new();
    reg.register(&idx, CollectionKind::SecondaryIndex);
    reg.complete_index(&idx);
    let _ = reg.secondary_index::<SGroup, _>(
        IndexTargets::new().with_index(&idx),
        owner_group_projection(Arc::clone(&idx)),
    );
    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), 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");
    let orphan_key = OwnerGroupKey {
        owner: u,
        group: fuid_from_native::<s_group::Hasher>(99),
    };
    idx.insert(&orphan_key, &OwnerGroupMarker(1))
        .expect("orphan");

    let built = reg.finish().expect("finish");
    let report = repair(
        &db,
        &built,
        RepairOptions {
            delete_orphans: true,
        },
    )
    .expect("repair");
    let stats = report
        .per_index
        .iter()
        .find(|s| s.name == "s_owner_groups")
        .expect("stats");
    assert_eq!(stats.deletes, 1);
    assert!(!idx.contains(&orphan_key));
}

#[derive(Clone, Debug, Default, PartialEq, Rapira, GetType, GetRefs)]
struct NonCanonVal {
    n: u32,
}

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

#[test]
fn non_canonical_codec_no_false_mismatch() {
    use crate::ProjectableIndex;
    use crate::codec::Codec;
    use std::io::Write;

    #[derive(Clone, Default)]
    struct PrefixCodec;

    impl Codec<NonCanonVal> for PrefixCodec {
        fn encode_to(&self, value: &NonCanonVal, buf: &mut Vec<u8>) -> crate::error::DbResult<()> {
            if value.n % 2 == 0 {
                buf.write_all(&[0])?;
            }
            buf.write_all(&value.n.to_le_bytes())?;
            Ok(())
        }
        fn decode_from(&self, buf: &[u8]) -> crate::error::DbResult<NonCanonVal> {
            let (n, rest) = if !buf.is_empty() && buf[0] == 0 {
                (u32::from_le_bytes(buf[1..5].try_into().unwrap()), &buf[5..])
            } else {
                (u32::from_le_bytes(buf[0..4].try_into().unwrap()), &buf[4..])
            };
            let _ = rest;
            Ok(NonCanonVal { n })
        }
    }

    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let idx = db
        .open_typed_map::<NonCanonVal, PrefixCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("idx");
    let mut reg = SchemaRegistry::new();
    reg.register(&idx, CollectionKind::SecondaryIndex);
    let _ = reg.secondary_index::<SGroup, _>(IndexTargets::new().with_index(&idx), {
        let idx = Arc::clone(&idx);
        move |k, v: &SGroup, sink| {
            sink.row(
                &idx,
                OwnerGroupKey {
                    owner: v.owner,
                    group: *k,
                },
                NonCanonVal { n: 2 },
            )?;
            Ok(())
        }
    });
    let users = db
        .open_typed_map::<SUser, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("users");
    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("groups");
    reg.register_entity(&users);
    reg.register_entity(&groups);

    let u = fuid_from_native::<s_user::Hasher>(1);
    let g = fuid_from_native::<s_group::Hasher>(1);
    users.insert(&u, suser("a", u)).expect("i");
    groups
        .insert(
            &g,
            SGroup {
                title: "g".into(),
                owner: u,
                member_count: 0,
            },
        )
        .expect("i");

    let key = OwnerGroupKey { owner: u, group: g };
    let alt = NonCanonVal { n: 2 };
    let mut buf = Vec::new();
    PrefixCodec.encode_to(&alt, &mut buf).expect("enc");
    idx.put_bytes(Key::as_bytes(&key), &buf)
        .expect("put alt encoding");

    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        !report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::IndexValueMismatch { .. })),
        "{report}"
    );
    let repair_report = repair(&db, &built, RepairOptions::default()).expect("repair");
    let rewrites: u64 = repair_report.per_index.iter().map(|s| s.rewrites).sum();
    assert_eq!(rewrites, 0, "equal values must not rewrite");
}

/// 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
/// duplicate homes, and neither exports a `self_shape`.
#[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::balanced().build(), NoHook, &[])
        .expect("open a");
    let b = db
        .open_typed_map::<SAnonB, RapiraCodec, _>(Config::balanced().build(), 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_shape.is_none(),
            "unbranded entity must not export a shape: {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::balanced().build(), NoHook, &[])
        .expect("open users");
    // groups -> VarTypedMap (exercises for_each-based scan)
    let groups = db
        .open_var_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), 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,
                reference: RefDesc::Assembled { ids, .. },
                ..
            } if collection == "s_users" && ids.as_slice() == [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,
                reference: RefDesc::Assembled { ids, .. },
                ..
            } if collection == "s_groups" && ids.as_slice() == [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::balanced().build(), NoHook, &[])
        .expect("open idx");

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

    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), 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::balanced().build(), NoHook, &[])
        .expect("open idx");

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

    let groups = db
        .open_typed_map::<SGroup, RapiraCodec, _>(Config::balanced().build(), 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> {
        // Test-only codec: rejects "POISON"-named rows so the schema validator's
        // skip-undecodable-rows path can be exercised. Uses checked decode — the
        // stored row is valid rapira, so this succeeds and the name check fires.
        let user: SUser =
            rapira::deserialize(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::balanced().build(), 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 {
                reference: RefDesc::Assembled { ids, .. },
                ..
            } => ids.first().copied(),
            _ => 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::balanced().build(), 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 {
                reference: RefDesc::Assembled { ids, .. },
                ..
            } => ids.first().copied(),
            _ => 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::balanced().build(), NoHook, &[])
            .expect("typed tree");
        let typed_map = db
            .open_typed_map::<MTypedMap, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
            .expect("typed map");
        let var_tree = db
            .open_var_typed_tree::<MVarTree, RapiraCodec, _>(
                Config::balanced().build(),
                NoHook,
                &[],
            )
            .expect("var tree");
        let var_map = db
            .open_var_typed_map::<MVarMap, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
            .expect("var map");
        let zero_tree_bc = db
            .open_zero_tree::<MZeroTreeBc, { std::mem::size_of::<MZeroTreeBc>() }, _>(
                Config::balanced().build(),
                NoHook,
                &[],
            )
            .expect("zero tree bitcask");
        let zero_map_bc = db
            .open_zero_map::<MZeroMapBc, { std::mem::size_of::<MZeroMapBc>() }, _>(
                Config::balanced().build(),
                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 owner_group_idx = db
            .open_zero_tree::<OwnerGroupMarker, { std::mem::size_of::<OwnerGroupMarker>() }, _>(
                Config::balanced().build(),
                NoHook,
                &[],
            )
            .expect("owner group idx");

        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(&owner_group_idx, CollectionKind::SecondaryIndex);
        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
            )
        );

        let key_ty_of = |name: &str| -> Option<armour_core::Typ> {
            graph
                .collections
                .iter()
                .find(|c| c.name == name)
                .unwrap_or_else(|| panic!("node {} not found", name))
                .key_ty
        };

        // 1. `key_ty` of composite key
        let owner_group_idx_key_ty = key_ty_of("s_owner_groups").expect("key_ty should be some");
        let armour_core::Typ::Struct(s) = owner_group_idx_key_ty else {
            panic!("Expected Struct");
        };
        assert_eq!(s.name, "OwnerGroupKey");
        let armour_core::Fields::Named(fields) = s.fields else {
            panic!("Expected named fields");
        };
        assert_eq!(fields.len(), 2);
        assert_eq!(fields[0].0, "owner");
        assert_eq!(fields[1].0, "group");

        // 2. `key_ty` of single branded-ID key
        let typed_tree_key_ty = key_ty_of("m_typed_tree").expect("key_ty should be some");
        assert_eq!(
            typed_tree_key_ty,
            armour_core::Typ::Scalar(armour_core::ScalarTyp::Fuid)
        );

        // 3. External -> key_ty == None
        assert_eq!(key_ty_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}");
    }
}

#[cfg(feature = "fjall")]
#[test]
fn lsm_schema_scan_skips_corrupt_and_keeps_valid() {
    use crate::lsm::{LsmConfig, LsmTree};
    use crate::schema::access::SchemaCollection;

    let dir = tempfile::tempdir().expect("tmp");
    let db = fjall::Database::builder(dir.path())
        .open()
        .expect("fjall db");
    let t: LsmTree<SUserId, SUser, RapiraCodec, NoHook> =
        LsmTree::open(db, LsmConfig::default(), RapiraCodec, NoHook).expect("lsm tree");
    let k1 = fuid_from_native::<s_user::Hasher>(1);
    t.put(&k1, &suser("alice", k1)).expect("put");
    // Corrupt row: value bytes that cannot decode as `SUser`, injected straight
    // into the keyspace. The scan must skip it (with a `warn!`, mirroring the
    // VarTypedTree/VarTypedMap contract) and still surface the valid row.
    t.keyspace()
        .insert(
            crate::key::Key::as_bytes(&fuid_from_native::<s_user::Hasher>(2)),
            b"corrupt",
        )
        .expect("inject corrupt");
    // Collect emitted events so the test observes the `warn!` itself, not just
    // the skip — the previous lossy-iterator implementation also skipped the row
    // silently, so asserting only on `seen` would pass with the fix reverted.
    #[derive(Clone, Default)]
    struct WarnCollector(std::sync::Arc<std::sync::Mutex<Vec<String>>>);
    impl tracing::Subscriber for WarnCollector {
        fn enabled(&self, m: &tracing::Metadata<'_>) -> bool {
            *m.level() <= tracing::Level::WARN
        }
        fn new_span(&self, _: &tracing::span::Attributes<'_>) -> tracing::Id {
            tracing::Id::from_u64(1)
        }
        fn record(&self, _: &tracing::Id, _: &tracing::span::Record<'_>) {}
        fn record_follows_from(&self, _: &tracing::Id, _: &tracing::Id) {}
        fn event(&self, ev: &tracing::Event<'_>) {
            struct Vis(String);
            impl tracing::field::Visit for Vis {
                fn record_debug(&mut self, f: &tracing::field::Field, v: &dyn std::fmt::Debug) {
                    self.0.push_str(&format!("{}={v:?} ", f.name()));
                }
            }
            let mut vis = Vis(String::new());
            ev.record(&mut vis);
            if let Ok(mut g) = self.0.lock() {
                g.push(format!("{} {}", ev.metadata().level(), vis.0));
            }
        }
        fn enter(&self, _: &tracing::Id) {}
        fn exit(&self, _: &tracing::Id) {}
    }

    let collector = WarnCollector::default();
    let events = std::sync::Arc::clone(&collector.0);
    let mut seen: Vec<(SUserId, SUser)> = Vec::new();
    tracing::subscriber::with_default(collector, || {
        SchemaCollection::scan(&t, &mut |k: &SUserId, v: &SUser| seen.push((*k, v.clone())));
    });

    assert_eq!(
        seen.len(),
        1,
        "scan must skip the corrupt row, keep the valid one"
    );
    assert_eq!(seen[0].1.name, "alice");

    let logged = events.lock().expect("events").clone();
    assert!(
        logged
            .iter()
            .any(|e| e.contains("WARN") && e.contains("relations not checked")),
        "the skipped row must be reported at WARN — a silent skip would let a \
         corrupt DB pass CI unnoticed. Got: {logged:?}"
    );
}

/// #103: a row the scan cannot decode must be reported as an ERROR finding, not
/// just a `warn!` — the documented CI recipe asserts on `report.errors()`, so a
/// log-only signal leaves a corrupt collection passing as clean.
#[cfg(feature = "fjall")]
#[test]
fn lsm_validate_reports_undecodable_rows_as_error() {
    use crate::lsm::LsmConfig;
    use armour_core::CollectionKind;

    let dir = tempfile::tempdir().expect("tmp");
    let db = std::sync::Arc::new(crate::armour::Db::open_test(dir.path()).expect("db"));
    let users = db
        .open_lsm_tree::<SUser, RapiraCodec, NoHook>(
            Some(crate::lsm::LsmDbConfig::default()),
            LsmConfig::default(),
            NoHook,
            &[],
        )
        .expect("open lsm");
    let k1 = fuid_from_native::<s_user::Hasher>(1);
    users.put(&k1, &suser("alice", k1)).expect("put");

    let mut reg = SchemaRegistry::new();
    reg.register(&users, CollectionKind::Standalone);
    let built = reg.finish().expect("registry consistent");

    // Clean data ⇒ no undecodable-row finding.
    let report = validate(&db, &built);
    assert!(
        !report
            .findings
            .iter()
            .any(|f| matches!(f, crate::schema::Finding::UndecodableRows { .. })),
        "clean collection must not report undecodable rows: {report}"
    );

    // Inject a value that cannot decode as `SUser`, straight into the keyspace.
    users
        .keyspace()
        .insert(
            crate::key::Key::as_bytes(&fuid_from_native::<s_user::Hasher>(2)),
            b"corrupt",
        )
        .expect("inject");

    let report = validate(&db, &built);
    let found = report
        .findings
        .iter()
        .find_map(|f| match f {
            crate::schema::Finding::UndecodableRows { collection, count } => {
                Some((collection.clone(), *count))
            }
            _ => None,
        })
        .unwrap_or_else(|| panic!("expected an UndecodableRows finding, got: {report}"));
    assert_eq!(found, ("s_users".to_string(), 1));
    assert!(
        report.errors().next().is_some(),
        "the finding must be error-level so the documented CI assert fails"
    );
}

// --- T7 composite-key schema tests (spec Tests section) ----------------------

fn finish_chat_message_registry(
    _db: &Arc<Db>,
    chats: &Arc<typed_map!(SChat, RapiraCodec)>,
    messages: &Arc<typed_map!(SMessage, RapiraCodec)>,
) -> BuiltRegistry {
    let mut reg = SchemaRegistry::new();
    reg.register_entity(chats);
    reg.register_entity(messages);
    reg.finish().expect("finish")
}

#[test]
fn schema_t7_forced_assembly_clean() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());

    let c = fuid_from_native::<s_chat::Hasher>(1);
    let m = fuid_from_native::<s_msg::Hasher>(1);
    chats.insert(&c, SChat { title: "c".into() }).expect("ins");
    messages
        .insert(
            &SMessageKey {
                chat_id: c,
                msg_id: m,
            },
            SMessage {
                reply_to: None,
                text: "hi".into(),
            },
        )
        .expect("ins");

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SReaction {
        chat_id: SChatId,
        msg_id: SMsgId,
    }
    impl CollectionMeta for SReaction {
        type SelfId = [u8; 8];
        const NAME: &'static str = "s_reactions";
    }
    let reactions = db
        .open_typed_map::<SReaction, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("reactions");
    reactions
        .insert(
            &[0u8; 8],
            SReaction {
                chat_id: c,
                msg_id: m,
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register(&reactions, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert_no_errors(&report, "composite forced assembly");
}

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

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SUserAlt {
        nick: String,
    }
    impl CollectionMeta for SUserAlt {
        type SelfId = SUserId;
        const NAME: &'static str = "s_users_alt";
    }
    let alt = db
        .open_typed_map::<SUserAlt, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("alt");

    let only_alt = fuid_from_native::<s_user::Hasher>(55);
    alt.insert(&only_alt, SUserAlt { nick: "a".into() })
        .expect("ins");
    groups
        .insert(
            &fuid_from_native::<s_group::Hasher>(3),
            SGroup {
                title: "g".into(),
                owner: only_alt,
                member_count: 0,
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register_entity(&alt);
    reg.register_entity(&groups);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert_no_errors(&report, "second home satisfies FK");
}

#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    FromBytes,
    IntoBytes,
    Immutable,
    KnownLayout,
    GetRefs,
    GetType,
    KeyShape,
)]
#[repr(C)]
struct SMessageKeyAlt {
    msg_id: SMsgId,
    chat_id: SChatId,
}
crate::impl_key_zerocopy!(
    SMessageKeyAlt,
    armour_core::KeyScheme::Typed(&[armour_core::KeyType::Fuid, armour_core::KeyType::Fuid])
);

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct SMessageAlt {
    text: String,
}
impl CollectionMeta for SMessageAlt {
    type SelfId = SMessageKeyAlt;
    const NAME: &'static str = "s_messages_alt";
}

#[test]
fn schema_t7_opposite_shape_order_second_home() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, _messages) = open_chat_message_fixture(dir.path());

    let alt = db
        .open_typed_map::<SMessageAlt, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("alt messages");

    let c = fuid_from_native::<s_chat::Hasher>(1);
    let m = fuid_from_native::<s_msg::Hasher>(1);
    chats.insert(&c, SChat { title: "c".into() }).expect("ins");
    alt.insert(
        &SMessageKeyAlt {
            msg_id: m,
            chat_id: c,
        },
        SMessageAlt {
            text: "only in alt home".into(),
        },
    )
    .expect("ins");

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SReaction {
        chat_id: SChatId,
        msg_id: SMsgId,
    }
    impl CollectionMeta for SReaction {
        type SelfId = [u8; 8];
        const NAME: &'static str = "s_reactions";
    }
    let reactions = db
        .open_typed_map::<SReaction, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("reactions");
    reactions
        .insert(
            &[0u8; 8],
            SReaction {
                chat_id: c,
                msg_id: m,
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&_messages);
    reg.register_entity(&alt);
    reg.register(&reactions, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert_no_errors(
        &report,
        "FK must probe second home with its SHAPE field order, not first home's",
    );
}

#[test]
fn schema_t7_ambiguous_once() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SDupChat {
        first: SChatId,
        second: SChatId,
        msg_id: SMsgId,
    }
    impl CollectionMeta for SDupChat {
        type SelfId = [u8; 8];
        const NAME: &'static str = "s_dup_chats";
    }
    let dup = db
        .open_typed_map::<SDupChat, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("dup");

    let c1 = fuid_from_native::<s_chat::Hasher>(1);
    let c2 = fuid_from_native::<s_chat::Hasher>(2);
    let m = fuid_from_native::<s_msg::Hasher>(1);
    for i in 0..5u64 {
        dup.insert(
            &i.to_be_bytes(),
            SDupChat {
                first: c1,
                second: c2,
                msg_id: m,
            },
        )
        .expect("ins");
    }

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register(&dup, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    let amb: Vec<_> = report
        .findings
        .iter()
        .filter(|f| matches!(f, Finding::AmbiguousRef(_)))
        .collect();
    assert_eq!(
        amb.len(),
        1,
        "ambiguity reported once, not per row: {report}"
    );
}

#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    FromBytes,
    IntoBytes,
    Immutable,
    KnownLayout,
    GetRefs,
    GetType,
    KeyShape,
)]
#[allow(clippy::duplicated_attributes)] // `chat_id` belongs to both groups
#[get_type(ref(chat_id, parent_msg_id))]
#[get_type(ref(chat_id, reply_msg_id))]
#[repr(C)]
struct SThreadKey {
    chat_id: SChatId,
    parent_msg_id: SMsgId,
    reply_msg_id: SMsgId,
}
crate::impl_key_zerocopy!(
    SThreadKey,
    armour_core::KeyScheme::Typed(&[
        armour_core::KeyType::Fuid,
        armour_core::KeyType::Fuid,
        armour_core::KeyType::Fuid,
    ])
);

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

#[test]
fn schema_t7_declaration_suppresses() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    #[get_type(ref(first, msg_id))]
    struct SDeclaredPair {
        first: SChatId,
        second: SChatId,
        msg_id: SMsgId,
    }
    impl CollectionMeta for SDeclaredPair {
        type SelfId = [u8; 8];
        const NAME: &'static str = "s_declared_pairs";
    }
    let pairs = db
        .open_typed_map::<SDeclaredPair, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("pairs");

    let c1 = fuid_from_native::<s_chat::Hasher>(1);
    let m = fuid_from_native::<s_msg::Hasher>(1);
    chats.insert(&c1, SChat { title: "a".into() }).expect("ins");
    messages
        .insert(
            &SMessageKey {
                chat_id: c1,
                msg_id: m,
            },
            SMessage {
                reply_to: None,
                text: "m".into(),
            },
        )
        .expect("ins");
    pairs
        .insert(
            &[0u8; 8],
            SDeclaredPair {
                first: c1,
                second: c1,
                msg_id: m,
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register(&pairs, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        !report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::AmbiguousRef(_))),
        "declarations must suppress auto ambiguity: {report}"
    );
    assert_no_errors(&report, "declarations suppress auto ambiguity");
}

#[test]
fn schema_t7_ref_with_key() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());

    let c = fuid_from_native::<s_chat::Hasher>(1);
    let parent = fuid_from_native::<s_msg::Hasher>(1);
    let child = fuid_from_native::<s_msg::Hasher>(2);
    chats.insert(&c, SChat { title: "c".into() }).expect("ins");
    messages
        .insert(
            &SMessageKey {
                chat_id: c,
                msg_id: parent,
            },
            SMessage {
                reply_to: None,
                text: "parent".into(),
            },
        )
        .expect("ins");
    messages
        .insert(
            &SMessageKey {
                chat_id: c,
                msg_id: child,
            },
            SMessage {
                reply_to: Some(parent),
                text: "child".into(),
            },
        )
        .expect("ins");

    let built = finish_chat_message_registry(&db, &chats, &messages);
    let report = validate(&db, &built);
    assert_no_errors(&report, "reply_to assembles with key chat");

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        Eq,
        PartialOrd,
        Ord,
        Hash,
        FromBytes,
        IntoBytes,
        Immutable,
        KnownLayout,
        GetRefs,
        GetType,
        KeyShape,
    )]
    #[repr(C)]
    struct DoubleChatKey {
        chat_a: SChatId,
        chat_b: SChatId,
        msg_id: SMsgId,
    }
    crate::impl_key_zerocopy!(
        DoubleChatKey,
        armour_core::KeyScheme::Typed(&[
            armour_core::KeyType::Fuid,
            armour_core::KeyType::Fuid,
            armour_core::KeyType::Fuid,
        ])
    );

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct DoubleChatBody {
        #[get_type(ref_with_key)]
        reply_to: Option<SMsgId>,
    }
    impl CollectionMeta for DoubleChatBody {
        type SelfId = DoubleChatKey;
        const NAME: &'static str = "s_double_chat";
    }
    let double = db
        .open_typed_map::<DoubleChatBody, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("double");
    double
        .insert(
            &DoubleChatKey {
                chat_a: c,
                chat_b: c,
                msg_id: child,
            },
            DoubleChatBody {
                reply_to: Some(parent),
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register(&double, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::AmbiguousRef(amb)
                if amb.part == RowPart::Key
                    && matches!(amb.reason, AmbiguityReason::RepeatedBrand { .. })
        )),
        "key double-brand must surface ambiguity: {report}"
    );
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct SForwardOrigin {
    chat_id: SChatId,
    msg_id: SMsgId,
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct SGroupMessageValue {
    forward: Option<SForwardOrigin>,
    text: String,
}
impl CollectionMeta for SGroupMessageValue {
    type SelfId = SMessageKey;
    const NAME: &'static str = "s_group_messages";
}

#[test]
fn schema_t7_scope_forward_vs_key() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());
    let group_messages = db
        .open_typed_map::<SGroupMessageValue, RapiraCodec, _>(
            Config::balanced().build(),
            NoHook,
            &[],
        )
        .expect("group messages");

    let c = fuid_from_native::<s_chat::Hasher>(1);
    let target = fuid_from_native::<s_msg::Hasher>(1);
    chats.insert(&c, SChat { title: "c".into() }).expect("ins");
    messages
        .insert(
            &SMessageKey {
                chat_id: c,
                msg_id: target,
            },
            SMessage {
                reply_to: None,
                text: "target".into(),
            },
        )
        .expect("ins");
    group_messages
        .insert(
            &SMessageKey {
                chat_id: c,
                msg_id: fuid_from_native::<s_msg::Hasher>(99),
            },
            SGroupMessageValue {
                forward: Some(SForwardOrigin {
                    chat_id: c,
                    msg_id: target,
                }),
                text: "fwd".into(),
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register_entity(&group_messages);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert_no_errors(
        &report,
        "forward pair assembles inside nested scope, not key.chat_id + forward.msg_id",
    );
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct SMessagePeer {
    peer_chat: SChatId,
    peer_msg: SMsgId,
    text: String,
}
impl CollectionMeta for SMessagePeer {
    type SelfId = SMessageKey;
    const NAME: &'static str = "s_message_peers";
}

#[test]
fn schema_t7_self_shape_key_only() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());
    let peers = db
        .open_typed_map::<SMessagePeer, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("peers");

    let c = fuid_from_native::<s_chat::Hasher>(1);
    let m = fuid_from_native::<s_msg::Hasher>(1);
    chats.insert(&c, SChat { title: "c".into() }).expect("ins");
    peers
        .insert(
            &SMessageKey {
                chat_id: c,
                msg_id: m,
            },
            SMessagePeer {
                peer_chat: c,
                peer_msg: fuid_from_native::<s_msg::Hasher>(999),
                text: "peer".into(),
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register_entity(&peers);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::DanglingFk {
                part: RowPart::Value,
                reference: RefDesc::Assembled { ids, .. },
                ..
            } if ids.as_slice() == [1, 999]
        )),
        "value same-shape reference must be checked: {report}"
    );
    assert!(
        !report.findings.iter().any(|f| matches!(
            f,
            Finding::DanglingFk {
                part: RowPart::Key,
                ..
            }
        )),
        "key root self-shape must be suppressed: {report}"
    );
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct SThreadRow {
    mention: SChatId,
}
impl CollectionMeta for SThreadRow {
    type SelfId = SThreadKey;
    const NAME: &'static str = "s_thread_rows";
}

/// Key-side declared references are attributed to `RowPart::Key`; the value root
/// of the same row is a separate scope. Also a regression guard for the
/// self-shape/ambiguity ordering: `s_thread_rows` is the home of its own key
/// shape `[chat, msg, msg]`, whose key root repeats `msg` — reporting the
/// ambiguity before suppressing the self shape made a home accuse itself, so
/// registering it failed validation on day one.
#[test]
fn schema_t7_key_value_root_split() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());
    let rows = db
        .open_typed_map::<SThreadRow, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("rows");

    let c = fuid_from_native::<s_chat::Hasher>(1);
    let p = fuid_from_native::<s_msg::Hasher>(1);
    let r = fuid_from_native::<s_msg::Hasher>(2);
    chats.insert(&c, SChat { title: "c".into() }).expect("ins");
    // The parent exists, the reply does not.
    messages
        .insert(
            &SMessageKey {
                chat_id: c,
                msg_id: p,
            },
            SMessage {
                reply_to: None,
                text: "p".into(),
            },
        )
        .expect("ins");
    rows.insert(
        &SThreadKey {
            chat_id: c,
            parent_msg_id: p,
            reply_msg_id: r,
        },
        SThreadRow { mention: c },
    )
    .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register_entity(&rows);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);

    assert!(
        !report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::AmbiguousRef(_))),
        "a home must not raise an ambiguity about its own key shape: {report}"
    );

    let dangling: Vec<_> = report
        .findings
        .iter()
        .filter(|f| matches!(f, Finding::DanglingFk { .. }))
        .collect();
    assert_eq!(dangling.len(), 1, "{report}");
    assert!(
        matches!(
            dangling[0],
            Finding::DanglingFk {
                part: RowPart::Key,
                path,
                reference: RefDesc::Assembled { ids, .. },
                ..
            } if path.as_slice() == [PathSeg::Field("reply_msg_id")] && ids.as_slice() == [1, 2]
        ),
        "the declared key-side reply reference must dangle, attributed to the key: {report}"
    );
}

#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    FromBytes,
    IntoBytes,
    Immutable,
    KnownLayout,
    GetRefs,
    GetType,
    KeyShape,
)]
#[get_type(ref(space_id, post_id, msg_id))]
#[repr(C)]
struct SForumMsgKeyA {
    space_id: SSpaceId,
    post_id: SMsgId,
    msg_id: SMsgId,
}
crate::impl_key_zerocopy!(
    SForumMsgKeyA,
    armour_core::KeyScheme::Typed(&[
        armour_core::KeyType::Fuid,
        armour_core::KeyType::Fuid,
        armour_core::KeyType::Fuid,
    ])
);

#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    FromBytes,
    IntoBytes,
    Immutable,
    KnownLayout,
    GetRefs,
    GetType,
    KeyShape,
)]
#[get_type(ref(space_id, msg_id, post_id))]
#[repr(C)]
struct SForumMsgKeyB {
    space_id: SSpaceId,
    msg_id: SMsgId,
    post_id: SMsgId,
}
crate::impl_key_zerocopy!(
    SForumMsgKeyB,
    armour_core::KeyScheme::Typed(&[
        armour_core::KeyType::Fuid,
        armour_core::KeyType::Fuid,
        armour_core::KeyType::Fuid,
    ])
);

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

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

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
#[get_type(ref(space_id, post_id, msg_id))]
struct SForumRefRow {
    space_id: SSpaceId,
    post_id: SMsgId,
    msg_id: SMsgId,
}
impl CollectionMeta for SForumRefRow {
    type SelfId = [u8; 8];
    const NAME: &'static str = "s_forum_refs";
}

#[test]
fn schema_t7_repeated_brand_homes() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let a = db
        .open_zero_tree::<SForumMsgA, { std::mem::size_of::<SForumMsgA>() }, _>(
            Config::balanced().build(),
            NoHook,
            &[],
        )
        .expect("a");
    let b = db
        .open_zero_tree::<SForumMsgB, { std::mem::size_of::<SForumMsgB>() }, _>(
            Config::balanced().build(),
            NoHook,
            &[],
        )
        .expect("b");
    let refs = db
        .open_typed_map::<SForumRefRow, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("refs");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&a);
    reg.register_entity(&b);
    reg.register(&refs, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::AmbiguousRef(amb)
                if amb.collection == "s_forum_refs"
                    && matches!(amb.reason, AmbiguityReason::RepeatedBrandHomes { .. })
        )),
        "two homes with repeated-brand multiset must be RepeatedBrandHomes: {report}"
    );
    // ...and the homes themselves must not be accused of their own key shape.
    assert!(
        !report.findings.iter().any(|f| matches!(
            f,
            Finding::AmbiguousRef(amb) if amb.collection != "s_forum_refs"
        )),
        "only the referencing collection may be ambiguous here: {report}"
    );

    let dir2 = tempfile::tempdir().expect("tmp2");
    let db2 = Arc::new(Db::open_test(dir2.path()).expect("db2"));
    let only = db2
        .open_zero_tree::<SForumMsgA, { std::mem::size_of::<SForumMsgA>() }, _>(
            Config::balanced().build(),
            NoHook,
            &[],
        )
        .expect("only");
    let refs2 = db2
        .open_typed_map::<SForumRefRow, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("refs2");

    let s = fuid_from_native::<s_space::Hasher>(1);
    let p = fuid_from_native::<s_msg::Hasher>(2);
    let m = fuid_from_native::<s_msg::Hasher>(3);
    only.insert(
        &SForumMsgKeyA {
            space_id: s,
            post_id: p,
            msg_id: m,
        },
        &SForumMsgA(1),
    )
    .expect("ins");
    refs2
        .insert(
            &[1u8; 8],
            SForumRefRow {
                space_id: s,
                post_id: p,
                msg_id: m,
            },
        )
        .expect("ins");

    let mut reg2 = SchemaRegistry::new();
    reg2.register_entity(&only);
    reg2.register(&refs2, CollectionKind::Standalone);
    let built2 = reg2.finish().expect("finish");
    let report = validate(&db2, &built2);
    assert_no_errors(
        &report,
        "one home resolves repeated-brand declaration positionally",
    );
}

#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    FromBytes,
    IntoBytes,
    Immutable,
    KnownLayout,
    GetRefs,
    GetType,
)]
#[repr(C)]
struct SBrokenKey {
    chat_id: SChatId,
    msg_id: SMsgId,
}

impl KeyShape for SBrokenKey {
    const SHAPE: &'static [&'static str] = &["chat", "msg"];
    fn from_ids(ids: &[u64]) -> Option<Self> {
        if ids.len() < 2 {
            return None;
        }
        if ids[0] == 999 && ids[1] == 999 {
            return None;
        }
        Some(Self {
            chat_id: fuid_from_native(ids[0]),
            msg_id: fuid_from_native(ids[1]),
        })
    }
}
crate::impl_key_zerocopy!(
    SBrokenKey,
    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 SBrokenMarker(pub u8);
impl CollectionMeta for SBrokenMarker {
    type SelfId = SBrokenKey;
    const NAME: &'static str = "s_broken";
}

#[test]
fn schema_t7_from_ids_none_registry_error() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());
    let broken = db
        .open_zero_tree::<SBrokenMarker, { std::mem::size_of::<SBrokenMarker>() }, _>(
            Config::balanced().build(),
            NoHook,
            &[],
        )
        .expect("broken");

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SBrokenRef {
        chat_id: SChatId,
        msg_id: SMsgId,
    }
    impl CollectionMeta for SBrokenRef {
        type SelfId = [u8; 8];
        const NAME: &'static str = "s_broken_refs";
    }
    let refs = db
        .open_typed_map::<SBrokenRef, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("refs");
    let ghost = fuid_from_native::<s_chat::Hasher>(999);
    chats
        .insert(
            &ghost,
            SChat {
                title: "ghost".into(),
            },
        )
        .expect("ins");
    refs.insert(
        &[0u8; 8],
        SBrokenRef {
            chat_id: ghost,
            msg_id: fuid_from_native::<s_msg::Hasher>(999),
        },
    )
    .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register_entity(&broken);
    reg.register(&refs, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::RegistryError { .. })),
        "from_ids None must be RegistryError, not DanglingFk: {report}"
    );
    assert!(
        !report.findings.iter().any(|f| matches!(
            f,
            Finding::DanglingFk {
                reference: RefDesc::Assembled { shape, .. },
                ..
            } if shape == &vec!["chat".to_string(), "msg".to_string()]
        )),
        "{report}"
    );
}

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

    let alive = fuid_from_native::<s_user::Hasher>(5);
    users.insert(&alive, suser("x", alive)).expect("i");

    macro_rules! check_policy {
        ($name:ident, $policy:expr, $expect_finding:expr, $expect_error:expr) => {{
            #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
            struct $name {
                raw_owner: u64,
            }
            impl CollectionMeta for $name {
                type SelfId = [u8; 8];
                const NAME: &'static str = stringify!($name);
            }
            let devices = db
                .open_typed_map::<$name, RapiraCodec, _>(
                    Config::balanced().build(),
                    NoHook,
                    &[],
                )
                .expect("devices");
            devices
                .insert(&[1u8; 8], $name { raw_owner: 6 })
                .expect("i");

            let mut reg = SchemaRegistry::new();
            reg.register_entity(&users);
            reg.register(&devices, CollectionKind::Standalone);
            reg.fk::<$name, SUser, _>(
                $policy,
                FkLabel::field("raw_owner"),
                |_k: &[u8; 8], v: &$name| {
                    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,
                            reference: RefDesc::Explicit { .. },
                            ..
                        } if collection == stringify!($name)
                    )
                })
                .collect();
            assert_eq!(
                !dangling.is_empty(),
                $expect_finding,
                "{}: {report}",
                stringify!($name)
            );
            if $expect_finding {
                match dangling[0] {
                    Finding::DanglingFk {
                        reference: RefDesc::Explicit { label },
                        on_missing,
                        ..
                    } => {
                        assert_eq!(label, "raw_owner");
                        assert_eq!(*on_missing, $policy);
                        assert_eq!(dangling[0].is_error(), $expect_error);
                    }
                    _ => unreachable!(),
                }
            }
        }};
    }

    check_policy!(SDeviceErr, OnMissing::Error, true, true);
    check_policy!(SDeviceWarn, OnMissing::Warn, true, false);
    check_policy!(SDeviceAllow, OnMissing::AllowMissing, false, false);
}

#[test]
fn schema_t7_absent_optional_silent() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());

    let c = fuid_from_native::<s_chat::Hasher>(1);
    let m = fuid_from_native::<s_msg::Hasher>(1);
    chats.insert(&c, SChat { title: "c".into() }).expect("ins");
    messages
        .insert(
            &SMessageKey {
                chat_id: c,
                msg_id: m,
            },
            SMessage {
                reply_to: None,
                text: "no reply".into(),
            },
        )
        .expect("ins");

    let built = finish_chat_message_registry(&db, &chats, &messages);
    let report = validate(&db, &built);
    assert!(
        !report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::DanglingFk { .. } | Finding::AmbiguousRef(_))),
        "absent optional must not emit an edge: {report}"
    );
}

#[test]
fn schema_t7_empty_brand_no_home() {
    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SAnon {
        label: String,
    }
    impl CollectionMeta for SAnon {
        type SelfId = Fuid<()>;
        const NAME: &'static str = "s_anon_empty";
    }

    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let anon = db
        .open_typed_map::<SAnon, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("anon");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&anon);
    let built = reg.finish().expect("finish");
    let graph = built.graph();
    let node = graph
        .collections
        .iter()
        .find(|c| c.name == "s_anon_empty")
        .expect("node");
    assert!(node.self_shape.is_none());
}

struct PathRec(Vec<String>);

impl RefVisitor for PathRec {
    fn enter(&mut self, seg: PathSeg) {
        self.0.push(format!("enter:{seg:?}"));
    }
    fn leave(&mut self) {
        self.0.push("leave".into());
    }
    fn id(&mut self, seg: Option<PathSeg>, brand: &'static str, _: u64) {
        self.0.push(format!("id:{brand}:{seg:?}"));
    }
}

/// Full path of every visited id — the same path the registry's visitor builds
/// (stack + the id's own segment) and the plan looks ids up by.
#[derive(Default)]
struct FullPathRec {
    stack: Vec<PathSeg>,
    seen: Vec<(Vec<PathSeg>, &'static str)>,
}

impl RefVisitor for FullPathRec {
    fn enter(&mut self, seg: PathSeg) {
        self.stack.push(seg);
    }
    fn leave(&mut self) {
        self.stack.pop().expect("leave without enter");
    }
    fn id(&mut self, seg: Option<PathSeg>, brand: &'static str, _: u64) {
        let mut path = self.stack.clone();
        if let Some(seg) = seg {
            path.push(seg);
        }
        self.seen.push((path, brand));
    }
}

fn walked_paths<T: GetRefs>(values: &[&T]) -> Vec<(Vec<PathSeg>, &'static str)> {
    let mut rec = FullPathRec::default();
    for v in values {
        v.visit_refs(&mut rec);
    }
    assert!(rec.stack.is_empty(), "unbalanced enter/leave");
    let mut out = rec.seen;
    out.sort();
    out.dedup();
    out
}

fn declared_paths<T: GetRefs>() -> Vec<(Vec<PathSeg>, &'static str)> {
    let mut out: Vec<_> = T::refs().into_iter().map(|r| (r.path, r.brand)).collect();
    out.sort();
    out.dedup();
    out
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs)]
enum WrapperEnum {
    A { id: SChatId },
    B(SUserId),
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs)]
struct WrapperFixture {
    opt: Option<SChatId>,
    boxed: Box<SChatId>,
    vec: Vec<SUserId>,
    arr: arrayvec::ArrayVec<SMsgId, 2>,
    map: std::collections::BTreeMap<SChatId, SUserId>,
    pair: (SChatId, SMsgId),
    data: WrapperEnum,
}

#[test]
fn schema_t7_wrapper_paths() {
    let refs = WrapperFixture::refs();
    assert!(refs.iter().any(|r| r.path == vec![PathSeg::Field("opt")]));
    assert!(refs.iter().any(|r| r.path == vec![PathSeg::Field("boxed")]));
    assert!(
        refs.iter()
            .any(|r| { r.brand == "SUSR" && r.path == vec![PathSeg::Field("vec"), PathSeg::Elem] })
    );
    assert!(refs.iter().any(|r| {
        r.brand == "msg" && r.path.starts_with(&[PathSeg::Field("arr"), PathSeg::Elem])
    }));
    assert!(refs.iter().any(|r| {
        r.brand == "chat" && r.path == vec![PathSeg::Field("map"), PathSeg::Elem, PathSeg::Index(0)]
    }));
    assert!(
        refs.iter()
            .any(|r| { r.path == vec![PathSeg::Field("pair"), PathSeg::Index(0)] })
    );
    assert!(refs.iter().any(|r| {
        r.path
            == vec![
                PathSeg::Field("data"),
                PathSeg::Variant("A"),
                PathSeg::Field("id"),
            ]
    }));

    let c = fuid_from_native::<s_chat::Hasher>(1);
    let u = fuid_from_native::<s_user::Hasher>(2);
    let m = fuid_from_native::<s_msg::Hasher>(3);
    let mut arr = arrayvec::ArrayVec::new();
    arr.push(m);
    let mut map = std::collections::BTreeMap::new();
    map.insert(c, u);
    let w = WrapperFixture {
        opt: Some(c),
        boxed: Box::new(c),
        vec: vec![u],
        arr,
        map,
        pair: (c, m),
        data: WrapperEnum::A { id: c },
    };
    let mut rec = PathRec(Vec::new());
    w.visit_refs(&mut rec);
    let seq = rec.0.join(",");
    assert!(
        !seq.contains("Field(\"opt\"),enter:Elem"),
        "Option is transparent: {seq}"
    );
    assert!(seq.contains("enter:Field(\"vec\")"), "{seq}");
    assert!(seq.contains("enter:Elem"), "{seq}");
    assert!(
        seq.contains("Field(\"data\")") && seq.contains("Field(\"id\")"),
        "{seq}"
    );

    // The two channels must describe the same paths. The plan is built from
    // `refs()` and filled by `visit_refs`, and the lookup is an exact path
    // match — drift does not fail loudly, it silently drops the reference. The
    // `B` variant is walked too, so the union covers every declared path.
    let w_b = WrapperFixture {
        data: WrapperEnum::B(u),
        ..w.clone()
    };
    assert_eq!(
        declared_paths::<WrapperFixture>(),
        walked_paths(&[&w, &w_b]),
        "static refs() and runtime visit_refs must agree on every path",
    );
}

#[test]
fn schema_t7_graph_edges() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());
    let threads = db
        .open_zero_tree::<SThreadMarker, { std::mem::size_of::<SThreadMarker>() }, _>(
            Config::balanced().build(),
            NoHook,
            &[],
        )
        .expect("threads");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register_entity(&threads);
    let built = reg.finish().expect("finish");
    let graph = built.graph();

    let threads_node = graph
        .collections
        .iter()
        .find(|c| c.name == "s_threads")
        .expect("threads node");
    assert_eq!(
        threads_node.self_shape,
        Some(vec![
            "chat".to_string(),
            "msg".to_string(),
            "msg".to_string()
        ])
    );

    let fk_edges: Vec<_> = graph
        .relations
        .iter()
        .filter(|r| r.from == "s_threads" && r.to == "s_messages")
        .collect();
    assert_eq!(fk_edges.len(), 2, "{:?}", graph.relations);
    let mut last_fields = std::collections::HashSet::new();
    for edge in fk_edges {
        let armour_core::RelationKind::Fk {
            field_path, brand, ..
        } = &edge.kind
        else {
            panic!("expected fk");
        };
        assert_eq!(brand, "chat+msg");
        last_fields.insert(field_path.last().cloned().expect("field_path"));
    }
    assert!(last_fields.contains(&PathSeg::Field("parent_msg_id")));
    assert!(last_fields.contains(&PathSeg::Field("reply_msg_id")));
}

// --- Codex review regressions (A1/B1 + P1/P2 fixes) ---------------------------

#[test]
fn schema_t7_vec_element_scope_flush_dangling() {
    // Without per-Elem flush, IdMap overwrites the same (scope, brand, path) and
    // only the last vector element is validated.
    let dir = tempfile::tempdir().expect("tmp");
    let (db, users, _groups) = open_fixture(dir.path());

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SMentions {
        users: Vec<SUserId>,
    }
    impl CollectionMeta for SMentions {
        type SelfId = [u8; 8];
        const NAME: &'static str = "s_mentions";
    }
    let mentions = db
        .open_typed_map::<SMentions, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("mentions");

    let alive = fuid_from_native::<s_user::Hasher>(1);
    let missing = fuid_from_native::<s_user::Hasher>(99);
    users.insert(&alive, suser("a", alive)).expect("ins");
    mentions
        .insert(
            &[0u8; 8],
            SMentions {
                users: vec![alive, missing],
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register(&mentions, CollectionKind::Standalone);
    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_mentions"
            )
        })
        .collect();
    assert_eq!(
        dangling.len(),
        1,
        "each Vec element must be validated independently: {report}"
    );
}

#[test]
fn schema_t7_thread_key_no_false_repeated_homes() {
    // Decl parts [chat,msg] have no repeats; whole-scope msg×2 must not trip
    // RepeatedBrandHomes when two message homes share the shape.
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());
    let alt = db
        .open_typed_map::<SMessageAlt, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("alt");
    let threads = db
        .open_zero_tree::<SThreadMarker, { std::mem::size_of::<SThreadMarker>() }, _>(
            Config::balanced().build(),
            NoHook,
            &[],
        )
        .expect("threads");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register_entity(&alt);
    reg.register_entity(&threads);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        !report.findings.iter().any(|f| matches!(
            f,
            Finding::AmbiguousRef(amb)
                if matches!(amb.reason, AmbiguityReason::RepeatedBrandHomes { .. })
        )),
        "ThreadKey decls must not false-positive RepeatedBrandHomes: {report}"
    );
}

#[test]
fn schema_t7_len1_decl_does_not_suppress_sibling() {
    // B1: `#[get_type(ref(owner))]` is len-1 and must not suppress auto edges for
    // other same-brand fields in the scope.
    let dir = tempfile::tempdir().expect("tmp");
    let (db, users, _groups) = open_fixture(dir.path());

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    #[get_type(ref(owner))]
    struct SPairOwners {
        owner: SUserId,
        other: SUserId,
    }
    impl CollectionMeta for SPairOwners {
        type SelfId = [u8; 8];
        const NAME: &'static str = "s_pair_owners";
    }
    let pairs = db
        .open_typed_map::<SPairOwners, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("pairs");

    let a = fuid_from_native::<s_user::Hasher>(1);
    let missing = fuid_from_native::<s_user::Hasher>(99);
    users.insert(&a, suser("a", a)).expect("ins");
    pairs
        .insert(
            &[0u8; 8],
            SPairOwners {
                owner: a,
                other: missing,
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register(&pairs, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::DanglingFk {
                collection,
                path,
                ..
            } if collection == "s_pair_owners"
                && path.last() == Some(&PathSeg::Field("other"))
        )),
        "len-1 declaration must leave sibling auto FK intact: {report}"
    );
}

#[test]
fn schema_t7_ref_with_key_plans_every_matching_shape() {
    // A1: `ref_with_key` matching several home shapes plans an edge for each.
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        Eq,
        PartialOrd,
        Ord,
        Hash,
        FromBytes,
        IntoBytes,
        Immutable,
        KnownLayout,
        GetRefs,
        GetType,
        KeyShape,
    )]
    #[repr(C)]
    struct SSpaceMsgKey {
        space_id: SSpaceId,
        msg_id: SMsgId,
    }
    crate::impl_key_zerocopy!(
        SSpaceMsgKey,
        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 SSpaceMsgMarker(pub u8);
    impl CollectionMeta for SSpaceMsgMarker {
        type SelfId = SSpaceMsgKey;
        const NAME: &'static str = "s_space_msgs";
    }

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        Eq,
        PartialOrd,
        Ord,
        Hash,
        FromBytes,
        IntoBytes,
        Immutable,
        KnownLayout,
        GetRefs,
        GetType,
        KeyShape,
    )]
    #[repr(C)]
    struct SDualKey {
        chat_id: SChatId,
        space_id: SSpaceId,
        msg_id: SMsgId,
    }
    crate::impl_key_zerocopy!(
        SDualKey,
        armour_core::KeyScheme::Typed(&[
            armour_core::KeyType::Fuid,
            armour_core::KeyType::Fuid,
            armour_core::KeyType::Fuid,
        ])
    );
    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SDualBody {
        #[get_type(ref_with_key)]
        reply_to: Option<SMsgId>,
    }
    impl CollectionMeta for SDualBody {
        type SelfId = SDualKey;
        const NAME: &'static str = "s_dual";
    }

    let space_msgs = db
        .open_zero_tree::<SSpaceMsgMarker, { std::mem::size_of::<SSpaceMsgMarker>() }, _>(
            Config::balanced().build(),
            NoHook,
            &[],
        )
        .expect("space msgs");
    let dual = db
        .open_typed_map::<SDualBody, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("dual");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register_entity(&space_msgs);
    reg.register(&dual, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let dual_entry = built
        .reg
        .entries
        .iter()
        .find(|e| e.name == "s_dual")
        .expect("dual entry");
    let targets: std::collections::HashSet<&str> = dual_entry
        .match_plan
        .edges
        .iter()
        .flat_map(|e| e.home_names.iter().copied())
        .collect();
    assert!(
        targets.contains("s_messages") && targets.contains("s_space_msgs"),
        "A1 must plan every matching with_key shape, got {targets:?}"
    );
}

#[test]
fn schema_t7_graph_uses_each_home_shape_order() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());
    let alt = db
        .open_typed_map::<SMessageAlt, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("alt");

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SLink {
        chat_id: SChatId,
        msg_id: SMsgId,
    }
    impl CollectionMeta for SLink {
        type SelfId = [u8; 8];
        const NAME: &'static str = "s_links";
    }
    let links = db
        .open_typed_map::<SLink, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("links");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register_entity(&alt);
    reg.register(&links, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let graph = built.graph();
    let to_messages = graph
        .relations
        .iter()
        .find(|r| r.from == "s_links" && r.to == "s_messages")
        .expect("edge to messages");
    let to_alt = graph
        .relations
        .iter()
        .find(|r| r.from == "s_links" && r.to == "s_messages_alt")
        .expect("edge to alt");
    let armour_core::RelationKind::Fk {
        brand: brand_msg, ..
    } = &to_messages.kind
    else {
        panic!("fk");
    };
    let armour_core::RelationKind::Fk {
        brand: brand_alt, ..
    } = &to_alt.kind
    else {
        panic!("fk");
    };
    assert_eq!(brand_msg, "chat+msg");
    assert_eq!(
        brand_alt, "msg+chat",
        "graph brand must preserve each home's SHAPE order"
    );
}

#[test]
fn schema_t7_self_shape_suppress_marks_validated() {
    // Key-root self-shape suppression must still mark brands validated so we
    // do not emit spurious UnvalidatedEdge for a clean entity.
    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());
    let c = fuid_from_native::<s_chat::Hasher>(1);
    let m = fuid_from_native::<s_msg::Hasher>(1);
    chats.insert(&c, SChat { title: "c".into() }).expect("ins");
    messages
        .insert(
            &SMessageKey {
                chat_id: c,
                msg_id: m,
            },
            SMessage {
                reply_to: None,
                text: "m".into(),
            },
        )
        .expect("ins");

    let built = finish_chat_message_registry(&db, &chats, &messages);
    let report = validate(&db, &built);
    assert!(
        !report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::UnvalidatedEdge { .. })),
        "self-shape suppress must mark brands validated: {report}"
    );
    assert_no_errors(&report, "clean chat/message fixture");
}

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

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    enum SUserEvent {
        Mentions { user: SUserId },
        Other,
    }
    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SEventRow {
        event: SUserEvent,
    }
    impl CollectionMeta for SEventRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "s_events";
    }
    let events = db
        .open_typed_map::<SEventRow, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("events");

    let alive = fuid_from_native::<s_user::Hasher>(1);
    let missing = fuid_from_native::<s_user::Hasher>(99);
    users.insert(&alive, suser("a", alive)).expect("ins");
    events
        .insert(
            &[0u8; 8],
            SEventRow {
                event: SUserEvent::Mentions { user: missing },
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register(&events, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);
    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::DanglingFk {
                collection,
                ..
            } if collection == "s_events"
        )),
        "enum Variant enter/leave must gather branded ids: {report}"
    );
}

/// A home whose shape repeats a brand must not be assembled from a scope that
/// carries that brand only once: testing brand *presence* instead of the brand
/// multiset filled two `msg` slots from one field and reported a consistent row
/// as dangling. Also guards the ordering fix — the home must not raise an
/// `AmbiguousRef` about its own repeated-brand key — and the sorted order of
/// `UnvalidatedEdge`, which is collected from a `HashSet`.
#[test]
fn schema_t7_repeated_brand_home_needs_every_component() {
    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SSpaceMsgRow {
        space_id: SSpaceId,
        msg_id: SMsgId,
    }
    impl CollectionMeta for SSpaceMsgRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "s_space_msg_rows";
    }

    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let forum = db
        .open_zero_tree::<SForumMsgA, { std::mem::size_of::<SForumMsgA>() }, _>(
            Config::balanced().build(),
            NoHook,
            &[],
        )
        .expect("forum");
    let rows = db
        .open_typed_map::<SSpaceMsgRow, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("rows");

    let s = fuid_from_native::<s_space::Hasher>(1);
    let m = fuid_from_native::<s_msg::Hasher>(2);
    rows.insert(
        &[7u8; 8],
        SSpaceMsgRow {
            space_id: s,
            msg_id: m,
        },
    )
    .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&forum);
    reg.register(&rows, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);

    assert!(
        !report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::DanglingFk { .. })),
        "shape [space, msg, msg] must not assemble from a single msg: {report}"
    );
    assert!(
        !report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::AmbiguousRef(_))),
        "the forum home must not accuse its own key shape: {report}"
    );

    let unvalidated: Vec<&str> = report
        .findings
        .iter()
        .filter_map(|f| match f {
            Finding::UnvalidatedEdge { from, brand } if from == "s_space_msg_rows" => {
                Some(brand.as_str())
            }
            _ => None,
        })
        .collect();
    assert_eq!(
        unvalidated,
        vec!["msg", "space"],
        "unvalidated brands must be reported in a stable (sorted) order: {report}"
    );
}

/// A reference assembled in a scope *below* a `Vec` element — an enum variant or
/// a nested struct inside the element. The ids are flushed when the element is
/// left, so matching the flush against the edge scope by equality silently
/// dropped every such edge.
#[test]
fn schema_t7_scope_below_vec_element_is_checked() {
    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    enum SMention {
        Msg { chat_id: SChatId, msg_id: SMsgId },
        Nothing,
    }
    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SMentionRow {
        mentions: Vec<SMention>,
    }
    impl CollectionMeta for SMentionRow {
        type SelfId = SUserId;
        const NAME: &'static str = "s_mention_rows";
    }

    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());
    let rows = db
        .open_typed_map::<SMentionRow, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("rows");

    let c = fuid_from_native::<s_chat::Hasher>(1);
    let alive = fuid_from_native::<s_msg::Hasher>(1);
    let missing = fuid_from_native::<s_msg::Hasher>(99);
    chats.insert(&c, SChat { title: "c".into() }).expect("ins");
    messages
        .insert(
            &SMessageKey {
                chat_id: c,
                msg_id: alive,
            },
            SMessage {
                reply_to: None,
                text: "m".into(),
            },
        )
        .expect("ins");
    rows.insert(
        &fuid_from_native::<s_user::Hasher>(1),
        SMentionRow {
            mentions: vec![
                SMention::Msg {
                    chat_id: c,
                    msg_id: alive,
                },
                SMention::Nothing,
                SMention::Msg {
                    chat_id: c,
                    msg_id: missing,
                },
            ],
        },
    )
    .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register(&rows, CollectionKind::Standalone);
    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_mention_rows")
        })
        .collect();
    assert_eq!(dangling.len(), 1, "{report}");
    assert!(
        matches!(
            dangling[0],
            Finding::DanglingFk {
                part: RowPart::Value,
                reference: RefDesc::Assembled { ids, .. },
                ..
            } if ids.as_slice() == [1, 99]
        ),
        "only the mention pointing at a missing message must dangle: {report}"
    );
}

/// A branded id at position 0 of an unnamed struct. `Fuid::visit_refs` used to
/// mark its own root with `PathSeg::Index(0)`, which the registry stripped —
/// indistinguishable from a genuine field 0, so the assembled path never matched
/// the plan and the whole reference went unchecked.
#[test]
fn schema_t7_unnamed_field_zero_is_checked() {
    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct SPairRow(SChatId, SMsgId);
    impl CollectionMeta for SPairRow {
        type SelfId = SUserId;
        const NAME: &'static str = "s_pair_rows";
    }

    let dir = tempfile::tempdir().expect("tmp");
    let (db, chats, messages) = open_chat_message_fixture(dir.path());
    let rows = db
        .open_typed_map::<SPairRow, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("rows");

    let c = fuid_from_native::<s_chat::Hasher>(1);
    let missing = fuid_from_native::<s_msg::Hasher>(99);
    chats.insert(&c, SChat { title: "c".into() }).expect("ins");
    rows.insert(&fuid_from_native::<s_user::Hasher>(1), SPairRow(c, missing))
        .expect("ins");

    assert_eq!(
        SPairRow::refs()
            .into_iter()
            .map(|r| r.path)
            .collect::<Vec<_>>(),
        vec![vec![PathSeg::Index(0)], vec![PathSeg::Index(1)]],
    );

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register(&rows, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);

    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::DanglingFk {
                collection,
                reference: RefDesc::Assembled { ids, .. },
                ..
            } if collection == "s_pair_rows" && ids.as_slice() == [1, 99]
        )),
        "a branded id at unnamed position 0 must still be assembled: {report}"
    );
}

/// `ref_with_key` completing a shape from the row key needs each missing brand
/// exactly once. Two key components of the same brand is an ambiguity the author
/// must resolve — treating it as "no match" dropped the declaration silently,
/// leaving only a warning that an `errors()`-based CI gate never sees.
#[test]
fn schema_t7_ref_with_key_repeated_key_brand_is_ambiguous() {
    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        Eq,
        PartialOrd,
        Ord,
        Hash,
        FromBytes,
        IntoBytes,
        Immutable,
        KnownLayout,
        GetRefs,
        GetType,
        KeyShape,
    )]
    #[repr(C)]
    struct SSpaceMsgKey {
        space_id: SSpaceId,
        msg_id: SMsgId,
    }
    crate::impl_key_zerocopy!(
        SSpaceMsgKey,
        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 SSpaceMsgMarker(pub u8);
    impl CollectionMeta for SSpaceMsgMarker {
        type SelfId = SSpaceMsgKey;
        const NAME: &'static str = "s_space_msgs";
    }

    #[derive(
        Clone,
        Copy,
        Debug,
        PartialEq,
        Eq,
        PartialOrd,
        Ord,
        Hash,
        FromBytes,
        IntoBytes,
        Immutable,
        KnownLayout,
        GetRefs,
        GetType,
    )]
    #[repr(C)]
    struct STwoSpaceKey {
        left: SSpaceId,
        right: SSpaceId,
    }
    crate::impl_key_zerocopy!(
        STwoSpaceKey,
        armour_core::KeyScheme::Typed(&[armour_core::KeyType::Fuid, armour_core::KeyType::Fuid])
    );

    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    struct STwoSpaceRow {
        #[get_type(ref_with_key)]
        msg_ref: Option<SMsgId>,
    }
    impl CollectionMeta for STwoSpaceRow {
        type SelfId = STwoSpaceKey;
        const NAME: &'static str = "s_two_space_rows";
    }

    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let homes = db
        .open_zero_tree::<SSpaceMsgMarker, { std::mem::size_of::<SSpaceMsgMarker>() }, _>(
            Config::balanced().build(),
            NoHook,
            &[],
        )
        .expect("homes");
    let rows = db
        .open_typed_map::<STwoSpaceRow, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("rows");

    rows.insert(
        &STwoSpaceKey {
            left: fuid_from_native::<s_space::Hasher>(1),
            right: fuid_from_native::<s_space::Hasher>(2),
        },
        STwoSpaceRow {
            msg_ref: Some(fuid_from_native::<s_msg::Hasher>(3)),
        },
    )
    .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&homes);
    reg.register(&rows, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    let report = validate(&db, &built);

    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::AmbiguousRef(amb)
                if amb.collection == "s_two_space_rows"
                    && matches!(&amb.reason, AmbiguityReason::RepeatedBrand { brand, .. } if brand == "space")
        )),
        "two key components of one brand must be an ambiguity, not a silent drop: {report}"
    );
}

/// A single-component `ref(...)` names a shape the automatic pass already covers
/// (a len-1 shape is never suppressed by a declaration). Planning it on both
/// paths duplicated every finding and every graph edge.
#[test]
fn schema_t7_len1_declaration_does_not_duplicate() {
    #[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
    #[get_type(ref(owner))]
    struct SOwnerRow {
        owner: SUserId,
    }
    impl CollectionMeta for SOwnerRow {
        type SelfId = [u8; 8];
        const NAME: &'static str = "s_owner_rows";
    }

    let dir = tempfile::tempdir().expect("tmp");
    let (db, users, _groups) = open_fixture(dir.path());
    let rows = db
        .open_typed_map::<SOwnerRow, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("rows");

    rows.insert(
        &[3u8; 8],
        SOwnerRow {
            owner: fuid_from_native::<s_user::Hasher>(99),
        },
    )
    .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&users);
    reg.register(&rows, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");

    let edges = built
        .graph()
        .relations
        .iter()
        .filter(|r| r.from == "s_owner_rows")
        .count();
    assert_eq!(edges, 1, "one edge per (source, reference, home)");

    let report = validate(&db, &built);
    let dangling = report
        .findings
        .iter()
        .filter(
            |f| matches!(f, Finding::DanglingFk { collection, .. } if collection == "s_owner_rows"),
        )
        .count();
    assert_eq!(dangling, 1, "one finding per dangling reference: {report}");
}

// The golden-path example from `docs/schema-metadata.md`, kept compiling and
// rendering here so the contract document cannot drift from the code. Names,
// field order and shapes mirror the doc; the brands are this module's `chat` /
// `msg`.
#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct DocChat {
    title: String,
}
impl CollectionMeta for DocChat {
    type SelfId = SChatId;
    const NAME: &'static str = "chats";
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct DocMessage {
    #[get_type(ref_with_key)]
    reply_to: Option<SMsgId>,
    text: String,
}
impl CollectionMeta for DocMessage {
    type SelfId = SMessageKey;
    const NAME: &'static str = "messages";
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct DocReaction {
    chat_id: SChatId,
    msg_id: SMsgId,
}
impl CollectionMeta for DocReaction {
    type SelfId = [u8; 8];
    const NAME: &'static str = "reactions";
}

#[test]
fn schema_docs_golden_path_mermaid() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("db"));
    let cfg = Config::balanced().build();
    let chats = db
        .open_typed_map::<DocChat, RapiraCodec, _>(cfg.clone(), NoHook, &[])
        .expect("chats");
    let messages = db
        .open_typed_map::<DocMessage, RapiraCodec, _>(cfg.clone(), NoHook, &[])
        .expect("messages");
    let reactions = db
        .open_typed_map::<DocReaction, RapiraCodec, _>(cfg, NoHook, &[])
        .expect("reactions");

    let c = fuid_from_native::<s_chat::Hasher>(1);
    let m = fuid_from_native::<s_msg::Hasher>(1);
    chats
        .insert(&c, DocChat { title: "c".into() })
        .expect("ins");
    messages
        .insert(
            &SMessageKey {
                chat_id: c,
                msg_id: m,
            },
            DocMessage {
                reply_to: None,
                text: "hi".into(),
            },
        )
        .expect("ins");
    reactions
        .insert(
            &[0u8; 8],
            DocReaction {
                chat_id: c,
                msg_id: m,
            },
        )
        .expect("ins");

    let mut reg = SchemaRegistry::new();
    reg.register_entity(&chats);
    reg.register_entity(&messages);
    reg.register(&reactions, CollectionKind::Standalone);
    let built = reg.finish().expect("unique collection names");

    let report = validate(&db, &built);
    assert!(report.errors().next().is_none(), "{report}");

    let mermaid = MermaidEr.format(&built.graph());
    println!("{mermaid}");
    assert_eq!(mermaid, DOC_MERMAID, "docs/schema-metadata.md is stale");
}

/// Verbatim copy of the diagram in `docs/schema-metadata.md`.
const DOC_MERMAID: &str = "\
erDiagram
    chats {
        Str title
    }
    messages {
        Opt_Fuid reply_to
        Str text
    }
    reactions {
        Fuid chat_id
        Fuid msg_id
    }
    messages }o--|| chats : \"fk chat_id (chat)\"
    messages }o--|| messages : \"fk reply_to (chat+msg)\"
    reactions }o--|| chats : \"fk chat_id (chat)\"
    reactions }o--|| messages : \"fk msg_id (chat+msg)\"
";

// --- T9 Task 11 — counter validation and repair over a shared expected pass --

// Reuses the shared T9 counter fixture (`Group`/`Member`/`MemberKey`,
// `group_counter_targets`) from `counter_tests`, but wired through
// `SchemaRegistry::counted` instead of `counter_tests::counted_fixture_at`'s
// raw `Projected::and_counters` — only `SchemaRegistry::counted` records the
// `Decl::Counter` this module's validator/repair read.

/// Db + schema registry over one counted source (`members`) and its target
/// (`groups`), Typed family on both sides.
#[allow(clippy::type_complexity)]
fn counted_registry_fixture(
    dir: &std::path::Path,
) -> (Arc<Db>, BuiltRegistry, Arc<GroupsMap>, Arc<MembersMap>) {
    let db = Arc::new(Db::open_test(dir).expect("open db"));
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");

    let mut reg = SchemaRegistry::new();
    let counter_targets = group_counter_targets(&groups);
    let groups_for_project = Arc::clone(&groups);
    let hook = reg
        .counted::<Member, _>(counter_targets, move |_k: &MemberKey, v: &Member, sink| {
            sink.add(&groups_for_project, v.group, 1)
        });
    let members = db
        .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .expect("open members");

    reg.register(&groups, CollectionKind::Standalone);
    reg.register(&members, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    (db, built, groups, members)
}

#[test]
fn schema_counter_validate_reports_drift_and_repair_fixes_it() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, built, groups, _members) = counted_registry_fixture(dir.path());

    // Drift: write straight into the target, bypassing the engine's own
    // counter maintenance entirely (no `Member` row backs this).
    groups
        .put(&[1u8; 8], Group { member_count: 42 })
        .expect("put");

    let report = validate(&db, &built);
    assert!(
        report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::CounterMismatch { stored: 42, .. })),
        "{report}"
    );

    let fixed = repair(&db, &built, RepairOptions::default()).expect("repair");
    assert_eq!(fixed.per_counter.len(), 1, "{fixed:?}");
    assert_eq!(fixed.per_counter[0].fixed, 1);

    assert!(
        validate(&db, &built).errors().next().is_none(),
        "clean after repair"
    );
}

/// Final whole-branch review, fix 3: a **dangling contribution** must not
/// abort `repair`.
///
/// `docs/schema-metadata.md` documents the leniency as applying "during a
/// recount, **and in `validate`/`repair`**", but `repair` propagated the
/// `CounterTargetMissing` that `resolve_typed(None, …)` raises under
/// `OnMissingTarget::Error` — aborting part-way through, after earlier deltas
/// had already been written, over a row `validate` reports anyway. It now
/// mirrors `counter::recount`: warn and continue.
#[test]
fn schema_counter_repair_skips_a_dangling_contribution_and_still_fixes_the_rest() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, built, groups, members) = counted_registry_fixture(dir.path());

    // A live source row whose target row does not exist. The write path keeps
    // `OnMissingTarget::Error`, so the `put` itself fails — but the source row
    // still lands, which is exactly the state `repair` has to survive.
    let _ = members.put(&MemberKey::new([9u8; 8], 1), Member { group: [9u8; 8] });

    // Plus an ordinary, repairable drift on a target row that does exist.
    groups
        .put(&[1u8; 8], Group { member_count: 42 })
        .expect("put");

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

    let fixed = repair(&db, &built, RepairOptions::default())
        .expect("a dangling contribution must not abort the whole repair");
    assert_eq!(fixed.per_counter.len(), 1, "{fixed:?}");
    assert_eq!(
        fixed.per_counter[0].fixed, 1,
        "the reachable row was repaired despite the dangling one: {fixed:?}"
    );
    assert_eq!(
        groups.get(&[1u8; 8]).expect("row").member_count,
        0,
        "no live contributions, so the drifted counter is driven to zero"
    );
    assert!(
        groups.get(&[9u8; 8]).is_none(),
        "repair must not materialise a target row it was told does not exist"
    );
}

// --- T9 Task 11 review Finding 1 — two counter fields, one target collection

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct DualCounterGroup {
    member_count: u64,
    admin_count: u64,
}
impl CollectionMeta for DualCounterGroup {
    type SelfId = [u8; 8];
    const NAME: &'static str = "dual_groups";
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct DualMember {
    group: [u8; 8],
}
impl CollectionMeta for DualMember {
    type SelfId = MemberKey;
    const NAME: &'static str = "dual_members";
}

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct DualAdmin {
    group: [u8; 8],
}
impl CollectionMeta for DualAdmin {
    type SelfId = MemberKey;
    const NAME: &'static str = "dual_admins";
}

/// `build_counters` used to key its expected map on `(collection, key)`
/// alone. Two *different* fields declared on the same target collection —
/// explicitly supported by `SchemaRegistry::counted` ("several different
/// `(target, field)` pairs may each be fed by their own source") — collided
/// on that key: whichever declaration's sums were folded in last silently
/// overwrote the other's, and `validate`/`repair` would then compare/rewrite
/// one field's stored value against the other field's expected sum.
/// Regression test for the fix (the expected map's key now includes the
/// field).
#[test]
fn schema_counter_two_fields_on_one_target_do_not_collide() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, built, groups, members, admins) = dual_counted_fixture(dir.path());

    groups
        .put(
            &[1u8; 8],
            DualCounterGroup {
                member_count: 0,
                admin_count: 0,
            },
        )
        .expect("seed group");
    members
        .put(&MemberKey::new([1u8; 8], 1), DualMember { group: [1u8; 8] })
        .expect("put member 1");
    members
        .put(&MemberKey::new([1u8; 8], 2), DualMember { group: [1u8; 8] })
        .expect("put member 2");
    admins
        .put(&MemberKey::new([1u8; 8], 1), DualAdmin { group: [1u8; 8] })
        .expect("put admin");

    let (member_count, admin_count) = {
        let stored = groups.get(&[1u8; 8]).expect("group row");
        (stored.member_count, stored.admin_count)
    };
    assert_eq!(member_count, 2, "live write path: two members");
    assert_eq!(admin_count, 1, "live write path: one admin");

    let report = validate(&db, &built);
    assert!(
        report.errors().next().is_none(),
        "two fields on one target must not collide: {report}"
    );

    let fixed = repair(&db, &built, RepairOptions::default()).expect("repair");
    assert!(
        fixed.per_counter.iter().all(|s| s.fixed == 0),
        "both fields already correct — repair must be a no-op: {fixed:?}"
    );
    let after = groups.get(&[1u8; 8]).expect("group row");
    assert_eq!(
        (after.member_count, after.admin_count),
        (member_count, admin_count),
        "repair must not have touched either field"
    );
}

/// Two counted sources, each maintaining a **different field** of the same
/// target collection — the topology `SchemaRegistry::finish`'s
/// `(collection, field)` duplicate rule exists to allow. Shared by the
/// non-transactional test above and the `db.tx()` one below.
#[allow(clippy::type_complexity)]
fn dual_counted_fixture(
    dir: &std::path::Path,
) -> (
    Arc<Db>,
    BuiltRegistry,
    Arc<typed_map!(DualCounterGroup, RapiraCodec, NoHook)>,
    Arc<typed_map!(DualMember, RapiraCodec, crate::Projected<MemberKey, DualMember, NoHook>)>,
    Arc<typed_map!(DualAdmin, RapiraCodec, crate::Projected<MemberKey, DualAdmin, NoHook>)>,
) {
    let db = Arc::new(Db::open_test(dir).expect("open db"));
    let groups = db
        .open_typed_map::<DualCounterGroup, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");

    let mut reg = SchemaRegistry::new();

    let member_targets = crate::CounterTargets::new().with_counter(
        &groups,
        "member_count",
        crate::OnMissingTarget::Error,
        |g: &DualCounterGroup| g.member_count,
        |cur: Option<&DualCounterGroup>, n: u64| match cur {
            Some(g) => crate::Upsert::Put(DualCounterGroup {
                member_count: n,
                ..g.clone()
            }),
            None => crate::Upsert::Keep,
        },
    );
    let groups_for_members = Arc::clone(&groups);
    let members_hook = reg.counted::<DualMember, _>(
        member_targets,
        move |_k: &MemberKey, v: &DualMember, sink| sink.add(&groups_for_members, v.group, 1),
    );
    let members = db
        .open_typed_map::<DualMember, RapiraCodec, _>(Config::test(), members_hook, &[])
        .expect("open dual_members");

    let admin_targets = crate::CounterTargets::new().with_counter(
        &groups,
        "admin_count",
        crate::OnMissingTarget::Error,
        |g: &DualCounterGroup| g.admin_count,
        |cur: Option<&DualCounterGroup>, n: u64| match cur {
            Some(g) => crate::Upsert::Put(DualCounterGroup {
                admin_count: n,
                ..g.clone()
            }),
            None => crate::Upsert::Keep,
        },
    );
    let groups_for_admins = Arc::clone(&groups);
    let admins_hook = reg
        .counted::<DualAdmin, _>(admin_targets, move |_k: &MemberKey, v: &DualAdmin, sink| {
            sink.add(&groups_for_admins, v.group, 1)
        });
    let admins = db
        .open_typed_map::<DualAdmin, RapiraCodec, _>(Config::test(), admins_hook, &[])
        .expect("open dual_admins");

    reg.register(&groups, CollectionKind::Standalone);
    reg.register(&members, CollectionKind::Standalone);
    reg.register(&admins, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    (db, built, groups, members, admins)
}

/// Codex PR #140 review, fix 1: the same two-field topology **inside one
/// transaction**.
///
/// `project_rank0` coalesced a transaction's counter deltas on
/// `(participant, key)`, dropping the field: two sources contributing to two
/// different fields of the same target *row* collapsed into a single
/// `PendingCounter` that kept only the first source's `TypedCounterTarget`.
/// The summed delta then landed on the first field — the wrong value, since
/// the two sums are unrelated — and the second field was never written at all.
/// The validator half of this was fixed on its own (the test above); the
/// transaction path was missed.
#[test]
fn schema_counter_tx_two_fields_on_one_target_row_both_land() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, built, groups, members, admins) = dual_counted_fixture(dir.path());

    let g = [1u8; 8];
    groups
        .put(
            &g,
            DualCounterGroup {
                member_count: 0,
                admin_count: 0,
            },
        )
        .expect("seed group");

    // One transaction, two sources, one target row. Two members and one admin,
    // so a field-blind coalescing produces a single delta of +3 on whichever
    // field it kept — visibly neither (2, 1) nor a plausible near-miss.
    db.tx()
        .put(&members, &MemberKey::new(g, 1), &DualMember { group: g })
        .unwrap()
        .put(&members, &MemberKey::new(g, 2), &DualMember { group: g })
        .unwrap()
        .put(&admins, &MemberKey::new(g, 1), &DualAdmin { group: g })
        .unwrap()
        .commit()
        .expect("commit");

    let after = groups.get(&g).expect("group row");
    assert_eq!(
        (after.member_count, after.admin_count),
        (2, 1),
        "each field takes its own source's sum, in the one logged value for the row"
    );

    // And the two paths agree: the validator recomputes the same pair.
    let report = validate(&db, &built);
    assert!(report.errors().next().is_none(), "{report}");
    let fixed = repair(&db, &built, RepairOptions::default()).expect("repair");
    assert!(
        fixed.per_counter.iter().all(|s| s.fixed == 0),
        "the transaction already left both fields correct: {fixed:?}"
    );
}

/// T9 Task 11 review Finding 2 — a declared counter source that was never
/// registered (forgotten `reg.register`, or classified via
/// `register_external`) has no `CollectionEntry`, so `build_counters`
/// silently skips its whole declaration. That must not be read as "this
/// source contributes nothing, so its targets are confirmed at zero": before
/// the fix, `validate` reported every one of its targets' live rows as
/// `CounterMismatch { expected: 0 }`, and `repair` applied that as a real
/// write — zeroing (or deleting) a counter that was correct.
#[test]
fn schema_counter_repair_never_wipes_an_unregistered_sources_targets() {
    let dir = tempfile::tempdir().expect("tmp");
    let db = Arc::new(Db::open_test(dir.path()).expect("open db"));
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");

    let mut reg = SchemaRegistry::new();
    let counter_targets = group_counter_targets(&groups);
    let groups_for_project = Arc::clone(&groups);
    let hook = reg
        .counted::<Member, _>(counter_targets, move |_k: &MemberKey, v: &Member, sink| {
            sink.add(&groups_for_project, v.group, 1)
        });
    let members = db
        .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .expect("open members");

    // `groups` is registered; `members` — the declared counter source — is
    // deliberately not.
    reg.register(&groups, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");

    groups
        .put(&[1u8; 8], Group { member_count: 0 })
        .expect("seed group");
    members
        .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
        .expect("put member");
    assert_eq!(groups.get(&[1u8; 8]).expect("row").member_count, 1);

    let report = validate(&db, &built);
    assert!(
        report.findings.iter().any(|f| matches!(
            f,
            Finding::RegistryError { message }
                if message.contains("members") && message.contains("not registered")
        )),
        "{report}"
    );
    assert!(
        !report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::CounterMismatch { .. })),
        "an unregistered source's targets must not be compared against an implied zero: {report}"
    );

    let fixed = repair(&db, &built, RepairOptions::default()).expect("repair");
    assert_eq!(
        groups.get(&[1u8; 8]).expect("row").member_count,
        1,
        "repair must not wipe a target whose source it cannot see"
    );
    assert!(fixed.per_counter.is_empty(), "{fixed:?}");
}

#[cfg(feature = "var-collections")]
type VarMembersMap =
    crate::var_typed_map!(Member, RapiraCodec, Projected<MemberKey, Member, NoHook>);

/// `counted_registry_fixture` with the source opened through
/// `open_var_typed_map` instead — the Var families are the ones whose scan
/// can skip a row, which is what makes
/// `schema_counter_repair_declines_a_source_with_undecodable_rows` possible.
#[cfg(feature = "var-collections")]
fn var_counted_registry_fixture(
    dir: &std::path::Path,
) -> (Arc<Db>, BuiltRegistry, Arc<GroupsMap>, Arc<VarMembersMap>) {
    let db = Arc::new(Db::open_test(dir).expect("open db"));
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");

    let mut reg = SchemaRegistry::new();
    let counter_targets = group_counter_targets(&groups);
    let groups_for_project = Arc::clone(&groups);
    let hook = reg
        .counted::<Member, _>(counter_targets, move |_k: &MemberKey, v: &Member, sink| {
            sink.add(&groups_for_project, v.group, 1)
        });
    let members = db
        .open_var_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .expect("open members");

    reg.register(&groups, CollectionKind::Standalone);
    reg.register(&members, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    (db, built, groups, members)
}

#[cfg(feature = "var-collections")]
#[test]
fn schema_counter_repair_declines_a_source_with_undecodable_rows() {
    use crate::{ByteView, MigrateAction};

    let dir = tempfile::tempdir().expect("tmp");
    let (db, built, groups, var_members) = var_counted_registry_fixture(dir.path());

    // `OnMissingTarget::Error` (see `group_counter_targets`) requires the
    // target row to exist before a member can reference it — mirrors every
    // `counted_fixture` test in `counter_tests`.
    groups
        .put(&[1u8; 8], Group { member_count: 0 })
        .expect("seed group");

    let key = MemberKey::new([1u8; 8], 1);
    // A normal, decodable write: the live write path has the typed value in
    // hand (never decodes it back), so this correctly bumps
    // `groups[[1u8;8]].member_count` to 1.
    var_members
        .put(&key, &Member { group: [1u8; 8] })
        .expect("put");
    assert_eq!(
        groups.get(&[1u8; 8]).expect("group row").member_count,
        1,
        "the live write path counted it"
    );

    // Corrupt the stored bytes directly, *after* that successful write —
    // `migrate` never fires `on_write`/projections (see its own rustdoc), so
    // this leaves the counter untouched and simulates bit rot on an
    // already-correctly-counted row, not a bad write.
    var_members
        .as_inner()
        .migrate(|k, _v| {
            if *k == key {
                MigrateAction::Update(ByteView::from_vec(vec![0u8; 2]))
            } else {
                MigrateAction::Keep
            }
        })
        .expect("corrupt row");

    let report = validate(&db, &built);
    assert!(
        report.findings.iter().any(
            |f| matches!(f, Finding::UndecodableRows { collection, .. } if collection == "members")
        ),
        "{report}"
    );
    assert!(
        report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::CounterMismatch { .. })),
        "the short expected sum still drifts from the live stored count: {report}"
    );

    let before = groups.get(&[1u8; 8]).expect("group row").member_count;
    assert_eq!(before, 1, "the live write path still counted it correctly");

    let fixed = repair(&db, &built, RepairOptions::default()).expect("repair");

    assert_eq!(
        groups.get(&[1u8; 8]).expect("group row").member_count,
        before,
        "left alone"
    );
    assert!(
        fixed.per_counter.is_empty(),
        "a declined target was never compared, so it contributes no repair-stats row: {fixed:?}"
    );
    let declined = fixed
        .undecodable_sources
        .iter()
        .find(|s| s.name == "members")
        .expect("the declined source is named in the report");
    assert!(
        declined.protected_targets.contains(&"groups"),
        "the declined target is named too: {declined:?}"
    );
}

// --- Codex PR #140 review, fix 5: `OnMissingTarget::Skip` in the expected set

/// `counted_registry_fixture`'s twin with an `OnMissingTarget::Skip` target:
/// a contribution whose target row is absent is **dropped by declaration**,
/// on the write path and therefore everywhere else too.
#[allow(clippy::type_complexity)]
fn skip_counted_registry_fixture(
    dir: &std::path::Path,
) -> (Arc<Db>, BuiltRegistry, Arc<GroupsMap>, Arc<MembersMap>) {
    use crate::Upsert;
    use crate::counter::{CounterTargets, OnMissingTarget};

    let db = Arc::new(Db::open_test(dir).expect("open db"));
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");

    let mut reg = SchemaRegistry::new();
    let counter_targets = CounterTargets::new().with_counter(
        &groups,
        "member_count",
        OnMissingTarget::Skip,
        |g: &Group| g.member_count,
        |cur: Option<&Group>, n: u64| match cur {
            #[allow(clippy::needless_update)]
            Some(g) => Upsert::Put(Group {
                member_count: n,
                ..g.clone()
            }),
            None => Upsert::Keep,
        },
    );
    let groups_for_project = Arc::clone(&groups);
    let hook = reg
        .counted::<Member, _>(counter_targets, move |_k: &MemberKey, v: &Member, sink| {
            sink.add(&groups_for_project, v.group, 1)
        });
    let members = db
        .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .expect("open members");

    reg.register(&groups, CollectionKind::Standalone);
    reg.register(&members, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    (db, built, groups, members)
}

/// A contribution to an absent row under `OnMissingTarget::Skip` must not
/// enter the expected map.
///
/// `expected::build_counters` returned a non-`one_of` group's member
/// unconditionally, with no probe for the row, so the dropped contribution was
/// still recorded as expected: `validate` reported a `CounterMismatch` forever
/// on a database that was exactly right, and `repair` called `apply_delta`,
/// got the no-op `Skip` back, and **counted it as fixed anyway** — a repair it
/// had not performed, with the same mismatch waiting on the next run.
#[test]
fn schema_counter_skip_policy_drops_a_contribution_to_an_absent_target_row() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, built, groups, members) = skip_counted_registry_fixture(dir.path());

    // A target row that exists — counted normally, so the test cannot pass by
    // the Skip policy simply disabling the counter.
    groups
        .put(&[1u8; 8], Group { member_count: 0 })
        .expect("put");
    members
        .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
        .expect("put");
    assert_eq!(groups.get(&[1u8; 8]).expect("row").member_count, 1);

    // …and one whose target row does not exist. Unlike `OnMissingTarget::Error`
    // this write *succeeds*: the contribution is dropped, by declaration.
    members
        .put(&MemberKey::new([7u8; 8], 1), Member { group: [7u8; 8] })
        .expect("a Skip target must not fail the source write");
    assert!(groups.get(&[7u8; 8]).is_none(), "no row was materialised");

    let report = validate(&db, &built);
    assert!(
        !report
            .findings
            .iter()
            .any(|f| matches!(f, Finding::CounterMismatch { .. })),
        "a deliberately dropped contribution is not drift: {report}"
    );
    assert!(report.errors().next().is_none(), "{report}");

    let fixed = repair(&db, &built, RepairOptions::default()).expect("repair");
    assert_eq!(fixed.per_counter.len(), 1, "{fixed:?}");
    assert_eq!(
        fixed.per_counter[0].fixed, 0,
        "nothing to fix — and nothing may be *reported* fixed that was not written: {fixed:?}"
    );
    assert_eq!(
        groups.get(&[1u8; 8]).expect("row").member_count,
        1,
        "the live row is untouched"
    );
    assert!(groups.get(&[7u8; 8]).is_none(), "and none was created");
}

// --- Codex PR #140 re-review — `apply_delta`'s `DbResult<bool>` return: a
// declined write on a *present* row must not be counted as fixed

/// `ErasedCounterTarget::apply_delta` reports whether it actually wrote
/// (`Ok(true)`) or resolved to `Upsert::Keep` (`Ok(false)`) — see its rustdoc
/// in `counter.rs`. The `OnMissingTarget::Skip` test above already proves
/// `repair` does not *misreport* a dropped contribution, but `Skip` is
/// filtered out of `expected::build_counters` before `repair` ever reaches
/// `apply_delta` (`expected.rs`'s "no live target row holds this key"
/// branch), so that test alone does not exercise the `bool` return: reverting
/// `apply_delta` to `DbResult<()>` and counting `fixed` on every `Ok` (instead
/// of only `Ok(true)`, `repair.rs:262`) leaves it green too.
///
/// This fixture instead declines on a row that **is** present, by
/// application choice — the same "the application chooses not to write"
/// shape `counter.rs`'s own `counter_resolve_keep_erases_to_skip` documents
/// for `resolve`, here reached through `schema::repair` end to end via
/// `apply_delta`: a high-water-mark `member_count` a repair may only ever
/// raise, never lower. `group_a` seeds a drift `repair` could only clear by
/// *decreasing* the stored value (declined, `Keep`, not counted); `group_x`
/// seeds one only fixable by *increasing* it (a real `Put`, counted). A
/// `fixed` count that is not scoped to `Ok(true)` reports `2`; the correct
/// count is `1`.
#[allow(clippy::type_complexity)]
fn ratchet_counted_registry_fixture(
    dir: &std::path::Path,
) -> (Arc<Db>, BuiltRegistry, Arc<GroupsMap>, Arc<MembersMap>) {
    use crate::Upsert;
    use crate::counter::{CounterTargets, OnMissingTarget};

    let db = Arc::new(Db::open_test(dir).expect("open db"));
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");

    let mut reg = SchemaRegistry::new();
    let counter_targets = CounterTargets::new().with_counter(
        &groups,
        "member_count",
        OnMissingTarget::Error,
        |g: &Group| g.member_count,
        // High-water mark: only ever raises the stored count, never lowers
        // it. `n < g.member_count` is exactly the shape `repair` computes for
        // a row whose live contributions sum to less than a stale stored
        // value (`group_a` below) — a legitimate, declared refusal to write,
        // not a usage bug: `read(apply(cur, n)) == n` is never violated,
        // because `Keep` leaves the value exactly as it was rather than
        // producing a wrong one.
        |cur: Option<&Group>, n: u64| match cur {
            Some(g) if n < g.member_count => Upsert::Keep,
            Some(g) => Upsert::Put(Group {
                member_count: n,
                ..g.clone()
            }),
            None => Upsert::Keep, // unreachable: OnMissingTarget::Error
        },
    );
    let groups_for_project = Arc::clone(&groups);
    let hook = reg
        .counted::<Member, _>(counter_targets, move |_k: &MemberKey, v: &Member, sink| {
            sink.add(&groups_for_project, v.group, 1)
        });
    let members = db
        .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .expect("open members");

    reg.register(&groups, CollectionKind::Standalone);
    reg.register(&members, CollectionKind::Standalone);
    let built = reg.finish().expect("finish");
    (db, built, groups, members)
}

#[test]
fn schema_counter_repair_does_not_count_a_declined_write_as_fixed() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, built, groups, members) = ratchet_counted_registry_fixture(dir.path());

    let a = [1u8; 8];
    let x = [2u8; 8];

    // `group_a`: no live contributions (want = 0), a stale high value from a
    // direct write straight into the target. `repair` computes
    // delta = 0 - 5 = -5; the ratchet's `Keep` branch declines it — a real
    // no-op `resolve_typed` decision, not an error, so `repair` must survive
    // it and must not count it.
    groups
        .put(&a, Group { member_count: 5 })
        .expect("seed group_a");

    // `group_x`: three live contributions, correctly reflected by the write
    // path as they land (member_count reaches 3 through ordinary increments,
    // each one an increase the ratchet always allows) …
    groups
        .put(&x, Group { member_count: 0 })
        .expect("seed group_x");
    for user in 1..=3u64 {
        members
            .put(&MemberKey::new(x, user), Member { group: x })
            .expect("put member");
    }
    assert_eq!(
        groups.get(&x).expect("row").member_count,
        3,
        "live write path already caught up"
    );
    // … then corrupted downward by a direct write that bypasses the engine
    // entirely, so `repair` must raise it back to the still-accurate
    // `want = 3` — this time a real `Put`, since the ratchet only refuses
    // decreases, and 3 > 1.
    groups
        .put(&x, Group { member_count: 1 })
        .expect("corrupt group_x");

    let report = validate(&db, &built);
    assert_eq!(
        report
            .findings
            .iter()
            .filter(|f| matches!(f, Finding::CounterMismatch { .. }))
            .count(),
        2,
        "both group_a and group_x are drifted: {report}"
    );

    let fixed = repair(&db, &built, RepairOptions::default()).expect("repair");
    assert_eq!(fixed.per_counter.len(), 1, "{fixed:?}");
    assert_eq!(
        fixed.per_counter[0].fixed, 1,
        "only group_x was actually written — group_a's decline must not be counted: {fixed:?}"
    );

    assert_eq!(
        groups.get(&a).expect("row").member_count,
        5,
        "the declined write left group_a untouched"
    );
    assert_eq!(
        groups.get(&x).expect("row").member_count,
        3,
        "group_x was repaired back up to its real count"
    );
}

// -- T10 Task 3b: unique-namespace ER edge -----------------------------
//
// Renders a claim edge (`Decl::Unique`) end to end through the ER export and
// pins that it is `RelationKind::Unique`, not `RelationKind::Index` — see
// `armour-typ/src/graph.rs` and `armdb/src/schema/mermaid.rs`.

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash, FromBytes, IntoBytes, Immutable, KnownLayout)]
#[repr(C)]
pub struct UUsernameKey([u8; 24]);
crate::impl_key_zerocopy!(
    UUsernameKey,
    armour_core::KeyScheme::Typed(&[armour_core::KeyType::Array(24)])
);

impl UUsernameKey {
    fn new(s: &str) -> Self {
        let mut b = [0u8; 24];
        b[..s.len()].copy_from_slice(s.as_bytes());
        Self(b)
    }
}

crate::unique_index!(UUsernames, key = UUsernameKey, name = "usernames");

#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
struct UUser {
    username: Option<String>,
}
impl CollectionMeta for UUser {
    type SelfId = SUserId;
    const NAME: &'static str = "users";
}

#[allow(clippy::type_complexity)]
fn unique_fixture(
    dir: &std::path::Path,
) -> (
    Arc<Db>,
    Arc<UniqueIndex<UUsernames>>,
    Arc<typed_map!(UUser, RapiraCodec)>,
) {
    let db = Arc::new(Db::open_test(dir).expect("open db"));
    let usernames = db
        .open_unique_index::<UUsernames>(FixedConfig::test())
        .expect("open usernames");
    let users = db
        .open_typed_map::<UUser, RapiraCodec, _>(Config::balanced().build(), NoHook, &[])
        .expect("open users");
    (db, usernames, users)
}

/// The declaration alone is enough to produce the ER edge — nothing in this
/// test actually claims a username, so `users` stays open under `NoHook`.
fn build_unique_registry(
    _db: &Arc<Db>,
    usernames: &Arc<UniqueIndex<UUsernames>>,
    users: &Arc<typed_map!(UUser, RapiraCodec)>,
) -> BuiltRegistry {
    let mut reg = SchemaRegistry::new();
    reg.register_external(usernames.name(), CollectionKind::UniqueIndex);
    reg.register_entity(users);

    let claim_ns = Arc::clone(usernames);
    let _ = reg.unique_index::<UUser, _>(
        UniqueTargets::new().with(usernames),
        move |_k, v: &UUser, sink| {
            if let Some(u) = &v.username {
                sink.claim(&claim_ns, UUsernameKey::new(u))?;
            }
            Ok(())
        },
    );

    reg.finish().expect("registry consistent")
}

#[test]
fn schema_unique_namespace_renders_its_own_er_edge() {
    let dir = tempfile::tempdir().expect("tmp");
    let (db, usernames, users) = unique_fixture(dir.path());
    let built = build_unique_registry(&db, &usernames, &users);
    let mermaid = MermaidEr.format(&built.graph());

    // A claim edge must not be byte-identical to a secondary-index edge.
    assert!(
        mermaid.contains(r#"users ||--o{ usernames : "unique""#),
        "{mermaid}"
    );
    assert!(
        !mermaid.contains(r#"users ||--o{ usernames : "index""#),
        "{mermaid}"
    );
}