armdb 0.11.2

//! Cross-cutting counter tests (T9). The fixture types below (`Group`,
//! `Member`, `MemberKey`, the collection aliases, `counted_fixture[_at]`) are
//! the **shared test fixture** from the T9 plan
//! (`docs/superpowers/plans/26-08-09-engine-maintained-counters.md`, "Shared
//! test fixture") — later tasks (6-11) reuse them verbatim. Keep names and
//! shapes stable; extend rather than rename.
//!
//! Task 6 note: `counted_fixture_at` opens `members` with the real
//! counter-projecting carrier (`Projected::new(...).and_counters(...)`), and
//! the write path (this task) now diffs and applies the resulting deltas —
//! see `counter_mutation_matrix_typed_map` below and its four sibling tests
//! (`_typed_tree`, `_zero_tree`, `_zero_tree_fixed`, `_var_typed_tree`), one
//! per declared family kind (Typed, Zero on both durability backends,
//! VarTyped).
#![allow(dead_code)]

use std::sync::Arc;

use rapira::Rapira;
use zerocopy::{FromBytes, Immutable, IntoBytes, KnownLayout};

use armour_core::{GetRefs, GetType};

use crate::counter::{CounterCollection, CounterTargets, OnMissingTarget};
use crate::projection::{IndexKeyed, ProjectableIndexTx};
use crate::{
    CollectionMeta, Config, IndexTargets, NoHook, Projected, RapiraCodec, TtlHook, TypedWriteHook,
    Upsert, armour::Db,
};

// ---------------------------------------------------------------------------
// Shared fixture
// ---------------------------------------------------------------------------

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

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

#[derive(
    Clone,
    Copy,
    Debug,
    PartialEq,
    Eq,
    PartialOrd,
    Ord,
    Hash,
    GetType,
    GetRefs,
    FromBytes,
    IntoBytes,
    Immutable,
    KnownLayout,
)]
#[repr(transparent)]
pub(crate) struct MemberKey(pub [u8; 16]);
impl MemberKey {
    pub(crate) fn new(group: [u8; 8], user: u64) -> Self {
        let mut b = [0u8; 16];
        b[..8].copy_from_slice(&group);
        b[8..].copy_from_slice(&user.to_be_bytes());
        Self(b)
    }
}
// `MemberKey` is a plain 16-byte key — use the crate's existing byte-array key
// impl macro rather than hand-writing `Key` (see `grep -n "impl_key"
// armdb/src/key.rs`).
crate::impl_key_zerocopy!(
    MemberKey,
    armour_core::KeyScheme::Typed(&[armour_core::KeyType::Array(16)])
);

/// Generic over the target collection family (`TypedMap`/`TypedTree`/
/// `VarTypedTree`, all Rapira-coded and value type `Group`) — Task 6's
/// mutation-matrix tests reuse this across the Typed/VarTyped families the
/// plan declares as coverage; only the Zero family needs its own
/// `zero_group_counter_targets` below (a distinct, zerocopy `ZeroGroup` value
/// type).
pub(crate) fn group_counter_targets<C>(groups: &Arc<C>) -> CounterTargets
where
    C: CounterCollection + ProjectableIndexTx + IndexKeyed<K = [u8; 8], V = Group>,
{
    CounterTargets::new().with_counter(
        groups,
        "member_count",
        OnMissingTarget::Error,
        |g: &Group| g.member_count,
        |cur: Option<&Group>, n: u64| match cur {
            Some(_) if n == 0 => Upsert::Put(Group { member_count: 0 }),
            // `..g.clone()` is a no-op while `Group` has only `member_count` —
            // kept so this stays a one-line diff when the fixture (shared
            // verbatim with later T9 tasks) grows more fields.
            #[allow(clippy::needless_update)]
            Some(g) => Upsert::Put(Group {
                member_count: n,
                ..g.clone()
            }),
            None => Upsert::Keep,
        },
    )
}

pub(crate) type GroupsMap = crate::typed_map!(Group, RapiraCodec, NoHook);
pub(crate) type MembersMap =
    crate::typed_map!(Member, RapiraCodec, Projected<MemberKey, Member, NoHook>);

/// Open `groups` (target) first, then `members` (counted source) — `members`'
/// carrier holds a clone of `groups`' `Arc`, so `groups` must exist first.
///
/// `members` opens with the real counter-projecting carrier: no secondary
/// index (an empty `IndexTargets` and a no-op index projection), plus a
/// counter projection that contributes `1` to `member.group`'s `member_count`
/// on every row. The non-transactional write path (facade and `atomic()`/TTL
/// replay) diffs and applies that contribution — see
/// `counter_mutation_matrix_typed_map` below.
pub(crate) fn counted_fixture_at(
    path: &std::path::Path,
) -> (Arc<Db>, Arc<MembersMap>, Arc<GroupsMap>) {
    let db = Arc::new(Db::open_test(path).expect("open db"));
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let counter_targets = group_counter_targets(&groups);
    let groups_for_project = Arc::clone(&groups);
    let hook = Projected::<MemberKey, Member, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _sink| Ok(()),
    )
    .and_counters(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");
    (db, members, groups)
}

pub(crate) fn counted_fixture() -> (Arc<Db>, Arc<MembersMap>, Arc<GroupsMap>, tempfile::TempDir) {
    let dir = tempfile::tempdir().expect("tempdir");
    let (db, m, g) = counted_fixture_at(dir.path());
    (db, m, g, dir)
}

/// T9 Task 5 forwarding fix: `open_typed_map` always wraps the caller's hook
/// in `TtlHook` (see `typed_map!`'s expansion of `MembersMap` above and
/// `Db::open_typed_map`'s `TtlHook::<T, H>::new(hook)`), so the hook a real
/// `members` collection actually runs is `TtlHook<Member, Projected<MemberKey,
/// Member, NoHook>>` — not the bare `Projected` that
/// `projection::tests::counter_carrier_reports_both_projections` checks. That
/// test would still pass if `TtlHook` failed to forward `COUNTS` from its
/// inner hook: nothing would fail to compile, and `MembersMap`'s counters
/// would just silently never fire. Assert `COUNTS` through the exact wrapper
/// `counted_fixture_at` hands to `open_typed_map`.
#[test]
fn counted_fixture_hook_reports_counts_through_ttl_wrapper() {
    assert!(
        <TtlHook<Member, Projected<MemberKey, Member, NoHook>> as TypedWriteHook<
            MemberKey,
            Member,
        >>::COUNTS
    );
}

// ---------------------------------------------------------------------------
// Zero-family fixture value (Copy + zerocopy, unlike the Rapira-coded `Group`
// above) — local to this task's roundtrip tests, not part of the shared
// fixture reused by later tasks.
// ---------------------------------------------------------------------------

#[derive(
    Clone, Copy, Debug, Default, PartialEq, GetType, FromBytes, IntoBytes, Immutable, KnownLayout,
)]
#[repr(C)]
pub(crate) struct ZeroGroup {
    pub member_count: u64,
}
impl CollectionMeta for ZeroGroup {
    type SelfId = [u8; 8];
    const NAME: &'static str = "zero_groups";
}

/// Zero-family counted source, mirrors `Member` — reuses `MemberKey` (already
/// a plain zerocopy byte array, so it needs no Zero-specific twin).
#[derive(
    Clone, Copy, Debug, Default, PartialEq, GetType, FromBytes, IntoBytes, Immutable, KnownLayout,
)]
#[repr(C)]
pub(crate) struct ZeroMember {
    pub group: [u8; 8],
}
impl CollectionMeta for ZeroMember {
    type SelfId = MemberKey;
    const NAME: &'static str = "zero_members";
}

/// `zero_group_counter_targets`'s twin of `group_counter_targets` for the
/// zerocopy `ZeroGroup` value type (Zero family, both durability backends).
pub(crate) fn zero_group_counter_targets<C>(groups: &Arc<C>) -> CounterTargets
where
    C: CounterCollection + ProjectableIndexTx + IndexKeyed<K = [u8; 8], V = ZeroGroup>,
{
    CounterTargets::new().with_counter(
        groups,
        "member_count",
        OnMissingTarget::Error,
        |g: &ZeroGroup| g.member_count,
        |cur: Option<&ZeroGroup>, n: u64| match cur {
            Some(_) => Upsert::Put(ZeroGroup { member_count: n }),
            None => Upsert::Keep,
        },
    )
}

// ---------------------------------------------------------------------------
// Step 1 (TDD) and its extension to every family: `CounterCollection`
// roundtrip via bytes — `contains_bytes` sees the row, `upsert_counter` reads
// and writes it back through the erased seam, matching the typed `get`.
// ---------------------------------------------------------------------------

#[test]
fn counter_collection_upsert_roundtrip_typed_map() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open"));
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let key = [7u8; 8];
    groups.put(&key, Group { member_count: 3 }).expect("put");

    // The counter seam sees the same row through bytes.
    let bytes = key.to_vec();
    assert!(groups.contains_bytes(&bytes).expect("contains"));

    // A key that was never written is absent through both methods, not just
    // "always true"/"always false".
    let missing = [42u8; 8].to_vec();
    assert!(!groups.contains_bytes(&missing).expect("contains missing"));
    assert!(groups.read_bytes(&missing).expect("read missing").is_none());

    groups
        .upsert_counter(&bytes, &mut |cur| {
            let n = cur.map_or(0, |g| g.member_count) + 1;
            Upsert::Put(Group { member_count: n })
        })
        .expect("upsert_counter");
    assert_eq!(groups.get(&key).expect("row").member_count, 4);

    // `read_bytes` is the read path validate/repair/open-time recount will use
    // on a full scan — it must see the same value `get` does, via `decode_value`.
    let read = groups
        .read_bytes(&bytes)
        .expect("read_bytes")
        .expect("row present");
    assert_eq!(
        groups
            .decode_value(&read)
            .expect("decode_value")
            .member_count,
        4
    );
}

#[test]
fn counter_collection_upsert_roundtrip_typed_tree() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open"));
    let groups = db
        .open_typed_tree::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let key = [7u8; 8];
    groups.put(&key, Group { member_count: 3 }).expect("put");

    let bytes = key.to_vec();
    assert!(groups.contains_bytes(&bytes).expect("contains"));

    let missing = [42u8; 8].to_vec();
    assert!(!groups.contains_bytes(&missing).expect("contains missing"));
    assert!(groups.read_bytes(&missing).expect("read missing").is_none());

    groups
        .upsert_counter(&bytes, &mut |cur| {
            let n = cur.map_or(0, |g| g.member_count) + 1;
            Upsert::Put(Group { member_count: n })
        })
        .expect("upsert_counter");
    assert_eq!(groups.get(&key).expect("row").member_count, 4);

    let read = groups
        .read_bytes(&bytes)
        .expect("read_bytes")
        .expect("row present");
    assert_eq!(
        groups
            .decode_value(&read)
            .expect("decode_value")
            .member_count,
        4
    );
}

#[test]
fn counter_collection_upsert_roundtrip_zero_tree() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open"));
    let groups = db
        .open_zero_tree::<ZeroGroup, { size_of::<ZeroGroup>() }, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let key = [7u8; 8];
    groups
        .put(&key, &ZeroGroup { member_count: 3 })
        .expect("put");

    let bytes = key.to_vec();
    assert!(groups.contains_bytes(&bytes).expect("contains"));

    let missing = [42u8; 8].to_vec();
    assert!(!groups.contains_bytes(&missing).expect("contains missing"));
    assert!(groups.read_bytes(&missing).expect("read missing").is_none());

    groups
        .upsert_counter(&bytes, &mut |cur| {
            let n = cur.map_or(0, |g| g.member_count) + 1;
            Upsert::Put(ZeroGroup { member_count: n })
        })
        .expect("upsert_counter");
    assert_eq!(groups.get(&key).expect("row").member_count, 4);

    let read = groups
        .read_bytes(&bytes)
        .expect("read_bytes")
        .expect("row present");
    assert_eq!(
        groups
            .decode_value(&read)
            .expect("decode_value")
            .member_count,
        4
    );
}

#[test]
fn counter_collection_upsert_roundtrip_zero_tree_fixed() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open"));
    let groups = db
        .open_zero_tree_fixed::<ZeroGroup, { size_of::<ZeroGroup>() }, _>(
            crate::FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("open groups");
    let key = [7u8; 8];
    groups
        .put(&key, &ZeroGroup { member_count: 3 })
        .expect("put");

    let bytes = key.to_vec();
    assert!(groups.contains_bytes(&bytes).expect("contains"));

    let missing = [42u8; 8].to_vec();
    assert!(!groups.contains_bytes(&missing).expect("contains missing"));
    assert!(groups.read_bytes(&missing).expect("read missing").is_none());

    groups
        .upsert_counter(&bytes, &mut |cur| {
            let n = cur.map_or(0, |g| g.member_count) + 1;
            Upsert::Put(ZeroGroup { member_count: n })
        })
        .expect("upsert_counter");
    assert_eq!(groups.get(&key).expect("row").member_count, 4);

    let read = groups
        .read_bytes(&bytes)
        .expect("read_bytes")
        .expect("row present");
    assert_eq!(
        groups
            .decode_value(&read)
            .expect("decode_value")
            .member_count,
        4
    );
}

#[test]
fn counter_collection_upsert_roundtrip_zero_map() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open"));
    let groups = db
        .open_zero_map::<ZeroGroup, { size_of::<ZeroGroup>() }, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let key = [7u8; 8];
    groups
        .put(&key, &ZeroGroup { member_count: 3 })
        .expect("put");

    let bytes = key.to_vec();
    assert!(groups.contains_bytes(&bytes).expect("contains"));

    let missing = [42u8; 8].to_vec();
    assert!(!groups.contains_bytes(&missing).expect("contains missing"));
    assert!(groups.read_bytes(&missing).expect("read missing").is_none());

    groups
        .upsert_counter(&bytes, &mut |cur| {
            let n = cur.map_or(0, |g| g.member_count) + 1;
            Upsert::Put(ZeroGroup { member_count: n })
        })
        .expect("upsert_counter");
    assert_eq!(groups.get(&key).expect("row").member_count, 4);

    let read = groups
        .read_bytes(&bytes)
        .expect("read_bytes")
        .expect("row present");
    assert_eq!(
        groups
            .decode_value(&read)
            .expect("decode_value")
            .member_count,
        4
    );
}

#[test]
fn counter_collection_upsert_roundtrip_zero_map_fixed() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open"));
    let groups = db
        .open_zero_map_fixed::<ZeroGroup, { size_of::<ZeroGroup>() }, _>(
            crate::FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("open groups");
    let key = [7u8; 8];
    groups
        .put(&key, &ZeroGroup { member_count: 3 })
        .expect("put");

    let bytes = key.to_vec();
    assert!(groups.contains_bytes(&bytes).expect("contains"));

    let missing = [42u8; 8].to_vec();
    assert!(!groups.contains_bytes(&missing).expect("contains missing"));
    assert!(groups.read_bytes(&missing).expect("read missing").is_none());

    groups
        .upsert_counter(&bytes, &mut |cur| {
            let n = cur.map_or(0, |g| g.member_count) + 1;
            Upsert::Put(ZeroGroup { member_count: n })
        })
        .expect("upsert_counter");
    assert_eq!(groups.get(&key).expect("row").member_count, 4);

    let read = groups
        .read_bytes(&bytes)
        .expect("read_bytes")
        .expect("row present");
    assert_eq!(
        groups
            .decode_value(&read)
            .expect("decode_value")
            .member_count,
        4
    );
}

#[test]
fn counter_collection_upsert_roundtrip_var_typed_tree() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open"));
    let groups = db
        .open_var_typed_tree::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let key = [7u8; 8];
    groups.put(&key, &Group { member_count: 3 }).expect("put");

    let bytes = key.to_vec();
    assert!(groups.contains_bytes(&bytes).expect("contains"));

    let missing = [42u8; 8].to_vec();
    assert!(!groups.contains_bytes(&missing).expect("contains missing"));
    assert!(groups.read_bytes(&missing).expect("read missing").is_none());

    groups
        .upsert_counter(&bytes, &mut |cur| {
            let n = cur.map_or(0, |g| g.member_count) + 1;
            Upsert::Put(Group { member_count: n })
        })
        .expect("upsert_counter");
    assert_eq!(groups.get(&key).expect("row").member_count, 4);

    let read = groups
        .read_bytes(&bytes)
        .expect("read_bytes")
        .expect("row present");
    assert_eq!(
        groups
            .decode_value(&read)
            .expect("decode_value")
            .member_count,
        4
    );
}

#[test]
fn counter_collection_upsert_roundtrip_var_typed_map() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open"));
    let groups = db
        .open_var_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let key = [7u8; 8];
    groups.put(&key, &Group { member_count: 3 }).expect("put");

    let bytes = key.to_vec();
    assert!(groups.contains_bytes(&bytes).expect("contains"));

    let missing = [42u8; 8].to_vec();
    assert!(!groups.contains_bytes(&missing).expect("contains missing"));
    assert!(groups.read_bytes(&missing).expect("read missing").is_none());

    groups
        .upsert_counter(&bytes, &mut |cur| {
            let n = cur.map_or(0, |g| g.member_count) + 1;
            Upsert::Put(Group { member_count: n })
        })
        .expect("upsert_counter");
    assert_eq!(groups.get(&key).expect("row").member_count, 4);

    let read = groups
        .read_bytes(&bytes)
        .expect("read_bytes")
        .expect("row present");
    assert_eq!(
        groups
            .decode_value(&read)
            .expect("decode_value")
            .member_count,
        4
    );
}

// ---------------------------------------------------------------------------
// Task 6: mutation matrix — insert / update-same-group / update-changed-group
// / delete, one test per declared family kind (Typed, Zero on both durability
// backends, VarTyped). `counted_fixture` already wires `members`' real
// counter-projecting carrier; these tests are the first to exercise the
// write-path diff + apply this task adds.
// ---------------------------------------------------------------------------

#[test]
fn counter_mutation_matrix_typed_map() {
    let (_db, members, groups, _dir) = counted_fixture();
    let g1 = [1u8; 8];
    let g2 = [2u8; 8];
    groups.put(&g1, Group { member_count: 0 }).unwrap();
    groups.put(&g2, Group { member_count: 0 }).unwrap();

    // insert -> +1
    members
        .put(&MemberKey::new(g1, 10), Member { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    // update without a group change -> no write
    members
        .put(&MemberKey::new(g1, 10), Member { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    // update with a group change -> -1 / +1
    members
        .put(&MemberKey::new(g1, 10), Member { group: g2 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 0);
    assert_eq!(groups.get(&g2).unwrap().member_count, 1);

    // delete -> -1
    members.delete(&MemberKey::new(g1, 10)).unwrap();
    assert_eq!(groups.get(&g2).unwrap().member_count, 0);
}

#[test]
fn counter_mutation_matrix_typed_tree() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open db"));
    let groups = db
        .open_typed_tree::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let counter_targets = group_counter_targets(&groups);
    let groups_for_project = Arc::clone(&groups);
    let hook = Projected::<MemberKey, Member, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _sink| Ok(()),
    )
    .and_counters(counter_targets, move |_k: &MemberKey, v: &Member, sink| {
        sink.add(&groups_for_project, v.group, 1)
    });
    let members = db
        .open_typed_tree::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .expect("open members");

    let g1 = [1u8; 8];
    let g2 = [2u8; 8];
    groups.put(&g1, Group { member_count: 0 }).unwrap();
    groups.put(&g2, Group { member_count: 0 }).unwrap();

    members
        .put(&MemberKey::new(g1, 10), Member { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    members
        .put(&MemberKey::new(g1, 10), Member { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    members
        .put(&MemberKey::new(g1, 10), Member { group: g2 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 0);
    assert_eq!(groups.get(&g2).unwrap().member_count, 1);

    members.delete(&MemberKey::new(g1, 10)).unwrap();
    assert_eq!(groups.get(&g2).unwrap().member_count, 0);
}

#[test]
fn counter_mutation_matrix_zero_tree() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open db"));
    let groups = db
        .open_zero_tree::<ZeroGroup, { size_of::<ZeroGroup>() }, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let counter_targets = zero_group_counter_targets(&groups);
    let groups_for_project = Arc::clone(&groups);
    let hook = Projected::<MemberKey, ZeroMember, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &ZeroMember, _sink| Ok(()),
    )
    .and_counters(
        counter_targets,
        move |_k: &MemberKey, v: &ZeroMember, sink| sink.add(&groups_for_project, v.group, 1),
    );
    let members = db
        .open_zero_tree::<ZeroMember, { size_of::<ZeroMember>() }, _>(Config::test(), hook, &[])
        .expect("open members");

    let g1 = [1u8; 8];
    let g2 = [2u8; 8];
    groups.put(&g1, &ZeroGroup { member_count: 0 }).unwrap();
    groups.put(&g2, &ZeroGroup { member_count: 0 }).unwrap();

    members
        .put(&MemberKey::new(g1, 10), &ZeroMember { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    members
        .put(&MemberKey::new(g1, 10), &ZeroMember { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    members
        .put(&MemberKey::new(g1, 10), &ZeroMember { group: g2 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 0);
    assert_eq!(groups.get(&g2).unwrap().member_count, 1);

    members.delete(&MemberKey::new(g1, 10)).unwrap();
    assert_eq!(groups.get(&g2).unwrap().member_count, 0);
}

#[test]
fn counter_mutation_matrix_zero_tree_fixed() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open db"));
    let groups = db
        .open_zero_tree_fixed::<ZeroGroup, { size_of::<ZeroGroup>() }, _>(
            crate::FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("open groups");
    let counter_targets = zero_group_counter_targets(&groups);
    let groups_for_project = Arc::clone(&groups);
    let hook = Projected::<MemberKey, ZeroMember, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &ZeroMember, _sink| Ok(()),
    )
    .and_counters(
        counter_targets,
        move |_k: &MemberKey, v: &ZeroMember, sink| sink.add(&groups_for_project, v.group, 1),
    );
    let members = db
        .open_zero_tree_fixed::<ZeroMember, { size_of::<ZeroMember>() }, _>(
            crate::FixedConfig::test(),
            hook,
            &[],
        )
        .expect("open members");

    let g1 = [1u8; 8];
    let g2 = [2u8; 8];
    groups.put(&g1, &ZeroGroup { member_count: 0 }).unwrap();
    groups.put(&g2, &ZeroGroup { member_count: 0 }).unwrap();

    members
        .put(&MemberKey::new(g1, 10), &ZeroMember { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    members
        .put(&MemberKey::new(g1, 10), &ZeroMember { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    members
        .put(&MemberKey::new(g1, 10), &ZeroMember { group: g2 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 0);
    assert_eq!(groups.get(&g2).unwrap().member_count, 1);

    members.delete(&MemberKey::new(g1, 10)).unwrap();
    assert_eq!(groups.get(&g2).unwrap().member_count, 0);
}

#[test]
fn counter_mutation_matrix_var_typed_tree() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open db"));
    let groups = db
        .open_var_typed_tree::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let counter_targets = group_counter_targets(&groups);
    let groups_for_project = Arc::clone(&groups);
    let hook = Projected::<MemberKey, Member, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _sink| Ok(()),
    )
    .and_counters(counter_targets, move |_k: &MemberKey, v: &Member, sink| {
        sink.add(&groups_for_project, v.group, 1)
    });
    let members = db
        .open_var_typed_tree::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .expect("open members");

    let g1 = [1u8; 8];
    let g2 = [2u8; 8];
    groups.put(&g1, &Group { member_count: 0 }).unwrap();
    groups.put(&g2, &Group { member_count: 0 }).unwrap();

    members
        .put(&MemberKey::new(g1, 10), &Member { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    members
        .put(&MemberKey::new(g1, 10), &Member { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    members
        .put(&MemberKey::new(g1, 10), &Member { group: g2 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 0);
    assert_eq!(groups.get(&g2).unwrap().member_count, 1);

    members.delete(&MemberKey::new(g1, 10)).unwrap();
    assert_eq!(groups.get(&g2).unwrap().member_count, 0);
}

// ---------------------------------------------------------------------------
// Task 7: the `counters_dirty` marker, taken when a counted source opens.
// ---------------------------------------------------------------------------

#[test]
fn counter_dirty_is_persisted_when_the_source_opens_and_cleared_on_close() {
    let dir = tempfile::tempdir().expect("tempdir");
    let (db, members, groups) = counted_fixture_at(dir.path());
    // Set by opening `members`, before a single write: the write path runs
    // after the shard lock is released, so a mark taken there could no longer
    // abort the write it was supposed to witness (see `counter_dirty.rs`).
    assert!(
        db.debug_counters_dirty("members"),
        "set when the counted source was opened"
    );
    assert!(
        !db.debug_counters_dirty("groups"),
        "a plain counter target is not marked — its flag means something else"
    );
    // The counter target row must pre-exist: the fixture's `OnMissingTarget`
    // is `Error`.
    groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
    members
        .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
        .unwrap();
    db.close().unwrap();
    assert!(
        !db.debug_counters_dirty("members"),
        "cleared by a clean close"
    );
}

/// The flag now stands for a whole **session**, not for "a delta has been
/// applied": it goes up when the counted source opens and comes down only at
/// a successful close. Both halves of that trade are pinned here — an idle
/// session still pays a recount after an unclean exit, and (the accepted
/// cost of moving the mark to open) a write *after* `close()`, which the
/// crate otherwise tolerates, is no longer re-marked. Stop writers before
/// closing.
#[test]
fn counter_dirty_spans_the_session_and_a_post_close_write_is_not_re_marked() {
    let dir = tempfile::tempdir().expect("tempdir");
    let (db, members, groups) = counted_fixture_at(dir.path());
    groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
    assert!(
        db.debug_counters_dirty("members"),
        "set at open, with no counted write yet"
    );

    db.close().unwrap();
    assert!(
        !db.debug_counters_dirty("members"),
        "cleared by a clean close"
    );

    // A different key so this is a genuine insert (nonzero delta) — an update
    // that doesn't change the projected group nets to no delta at all, which
    // would make the assertion vacuous.
    members
        .put(&MemberKey::new([1u8; 8], 2), Member { group: [1u8; 8] })
        .unwrap();
    assert!(
        !db.debug_counters_dirty("members"),
        "the write path does not mark: evidence is taken at open, once per session"
    );
}

#[test]
fn counter_carrier_rejects_a_direct_open() {
    let dir = tempfile::tempdir().expect("tempdir");
    // The target still needs a Db to exist; only the *source* is opened directly.
    let target_dir = tempfile::tempdir().expect("tempdir");
    let tdb = Arc::new(Db::open_test(target_dir.path()).expect("open"));
    let groups = tdb
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let targets = group_counter_targets(&groups);
    let groups_for_hook = Arc::clone(&groups);
    let hook = Projected::<MemberKey, Member, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _s| Ok(()),
    )
    .and_counters(targets, move |_k: &MemberKey, v: &Member, sink| {
        sink.add(&groups_for_hook, v.group, 1)
    });
    let err = crate::TypedMap::<MemberKey, Member, RapiraCodec, _>::open_hooked(
        dir.path(),
        Config::test(),
        RapiraCodec,
        hook,
    )
    .err()
    .expect("a counter carrier has nowhere to persist its dirty marker");
    assert!(matches!(err, crate::DbError::CountersRequireManagedOpen));
}

#[test]
fn counter_dirty_survives_a_failed_close() {
    // A collection whose flush fails must leave counters_dirty set.
    let dir = tempfile::tempdir().expect("tempdir");
    let (db, members, groups) = counted_fixture_at(dir.path());
    assert!(db.debug_counters_dirty("members"), "set at open");
    groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
    members
        .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
        .unwrap();
    crate::projection::FAIL_COLLECTION_FLUSH.store(true, std::sync::atomic::Ordering::SeqCst);
    let _ = db.close();
    crate::projection::FAIL_COLLECTION_FLUSH.store(false, std::sync::atomic::Ordering::SeqCst);
    assert!(db.debug_counters_dirty("members"));
}

/// Global Constraint: `db.info` gains one `serde` field with `#[serde(default)]`
/// — a database written by an older build (no `counters_dirty` key at all in
/// its stored `CollectionInfo`) must still open, reading the field as `false`.
#[test]
fn counters_dirty_defaults_to_false_when_missing_from_a_pre_t9_stored_db_info() {
    let dir = tempfile::tempdir().expect("tempdir");
    std::fs::write(
        dir.path().join("db.info"),
        r#"{"version":0,"collections":{"legacy":{"version":0,"typ_hash":0,"seq":0,"projections_dirty":false}},"clean":true}"#,
    )
    .expect("write pre-T9 db.info");
    let db = Db::open_test(dir.path())
        .expect("a pre-T9 db.info (no counters_dirty key at all) must still load");
    assert!(!db.debug_counters_dirty("legacy"));
}

// ---------------------------------------------------------------------------
// Task 8: open-time recount.
// ---------------------------------------------------------------------------

/// `Group` retyped at a bumped `VERSION` — same field, used only by
/// `counter_recount_triggers_on_a_target_only_migration` below to simulate a
/// schema change on the *target* that loses the maintained value.
#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
pub(crate) struct GroupV2 {
    pub member_count: u64,
}
impl CollectionMeta for GroupV2 {
    type SelfId = [u8; 8];
    const NAME: &'static str = "groups";
    const VERSION: u16 = 1;
}

/// `group_counter_targets` retyped for `GroupV2`.
pub(crate) fn group_v2_counter_targets<C>(groups: &Arc<C>) -> CounterTargets
where
    C: CounterCollection + ProjectableIndexTx + IndexKeyed<K = [u8; 8], V = GroupV2>,
{
    CounterTargets::new().with_counter(
        groups,
        "member_count",
        OnMissingTarget::Error,
        |g: &GroupV2| g.member_count,
        |cur: Option<&GroupV2>, n: u64| match cur {
            Some(_) if n == 0 => Upsert::Put(GroupV2 { member_count: 0 }),
            #[allow(clippy::needless_update)]
            Some(g) => Upsert::Put(GroupV2 {
                member_count: n,
                ..g.clone()
            }),
            None => Upsert::Keep,
        },
    )
}

#[test]
fn counter_recount_fixes_drift_and_zeroes_contributionless_rows() {
    // The recount procedure itself, driven synchronously.
    let (db, members, groups, _dir) = counted_fixture();
    groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
    groups.put(&[2u8; 8], Group { member_count: 0 }).unwrap();
    members
        .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
        .unwrap();
    // Corrupt both rows behind the engine's back.
    groups.put(&[1u8; 8], Group { member_count: 99 }).unwrap();
    groups.put(&[2u8; 8], Group { member_count: 7 }).unwrap(); // no contributions

    let repaired = db.recount_counters("members").unwrap();

    assert_eq!(repaired, 2);
    assert_eq!(groups.get(&[1u8; 8]).unwrap().member_count, 1, "recomputed");
    assert_eq!(
        groups.get(&[2u8; 8]).unwrap().member_count,
        0,
        "driven to zero"
    );
}

#[test]
fn counter_recount_runs_when_the_previous_close_failed() {
    // `Db::drop` is a clean shutdown and the directory holds `db.lock`, so an
    // unclean session is simulated by a *failing* close, not by leaking the Db.
    //
    // The failpoint stays armed across the whole block, not just the explicit
    // `db.close()` call: `Drop for Db` also runs `shutdown()` -> `close()`
    // when `db` goes out of scope, and disarming the failpoint beforehand
    // would let that implicit close succeed and clear the very flag this
    // test means to leave set.
    let dir = tempfile::tempdir().unwrap();
    crate::projection::FAIL_COLLECTION_FLUSH.store(true, std::sync::atomic::Ordering::SeqCst);
    {
        let (db, members, groups) = counted_fixture_at(dir.path());
        groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
        members
            .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
            .unwrap();
        groups.put(&[1u8; 8], Group { member_count: 99 }).unwrap(); // drift
        let _ = db.close(); // fails: the flag stays set
        // `db` drops here — the implicit close must fail too (see above).
    }
    crate::projection::FAIL_COLLECTION_FLUSH.store(false, std::sync::atomic::Ordering::SeqCst);
    let (db, _members, groups) = counted_fixture_at(dir.path());
    assert_eq!(
        groups.get(&[1u8; 8]).unwrap().member_count,
        1,
        "recounted at open"
    );
    assert!(
        db.debug_counters_dirty("members"),
        "the recount cleared the old evidence; this session's open then took its own"
    );
    db.close().unwrap();
    assert!(
        !db.debug_counters_dirty("members"),
        "and a clean close retires it, so the next open does not recount again"
    );
}

#[test]
fn counter_recount_triggers_on_a_target_only_migration() {
    // `groups` migrates (version bump + a migration that rewrites the field);
    // `members` does not change at all. The source must still recount,
    // because the migration moved the counter's own storage.
    let dir = tempfile::tempdir().unwrap();
    {
        let (db, members, groups) = counted_fixture_at(dir.path());
        groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
        members
            .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
            .unwrap();
        db.close().unwrap(); // clean: no flag left behind
    }
    // Reopen `groups` under `GroupV2` (bumped `VERSION`) with a migration
    // that zeroes the field — simulating a schema change that loses the
    // maintained value. `members` is untouched.
    fn zero_member_count(_k: &[u8; 8], _v: &GroupV2) -> crate::MigrateAction<GroupV2> {
        crate::MigrateAction::Update(GroupV2 { member_count: 0 })
    }
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let groups = db
        .open_typed_map::<GroupV2, RapiraCodec, _>(
            Config::test(),
            NoHook,
            &[(0, zero_member_count)],
        )
        .unwrap();
    let targets = group_v2_counter_targets(&groups);
    let groups_for_hook = Arc::clone(&groups);
    let hook = Projected::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _s| Ok(()),
    )
    .and_counters(targets, move |_k: &MemberKey, v: &Member, sink| {
        sink.add(&groups_for_hook, v.group, 1)
    });
    let _members = db
        .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .unwrap();
    assert_eq!(
        groups.get(&[1u8; 8]).unwrap().member_count,
        1,
        "the target's migration marked it dirty, so the source recounted"
    );
    // Makes the "target-only" property self-evident: `members`' own version
    // never changed and the previous session closed cleanly, so the only way
    // a recount could have been triggered here is via the *target*'s flag
    // reaching it — not via `Member::VERSION` or a leftover source flag.
    assert!(
        db.debug_counters_dirty("groups"),
        "the target's flag is still up while the session runs — a second source \
         opening later must see it too"
    );
    db.close().unwrap();
    assert!(
        !db.debug_counters_dirty("groups"),
        "a successful close retires the evidence the recount consumed"
    );
}

/// A target's migration flag must be **retired** once the sources that needed
/// it have recounted. It has no in-memory marker behind it, so `close` used to
/// leave it up unconditionally — and every subsequent open of the source ran
/// another full recount, forever.
#[test]
fn counter_recount_after_a_target_migration_does_not_repeat_on_the_next_open() {
    fn zero_member_count(_k: &[u8; 8], _v: &GroupV2) -> crate::MigrateAction<GroupV2> {
        crate::MigrateAction::Update(GroupV2 { member_count: 0 })
    }

    /// Open `groups` (migrating to `GroupV2`) and then the counted `members`
    /// source on top of it — the shape all three sessions below share.
    fn open_pair(
        dir: &std::path::Path,
    ) -> (
        Arc<Db>,
        Arc<crate::typed_map!(GroupV2, RapiraCodec, NoHook)>,
    ) {
        let db = Arc::new(Db::open_test(dir).unwrap());
        let groups = db
            .open_typed_map::<GroupV2, RapiraCodec, _>(
                Config::test(),
                NoHook,
                &[(0, zero_member_count)],
            )
            .unwrap();
        let groups_for_hook = Arc::clone(&groups);
        let hook = Projected::new(
            IndexTargets::new(),
            |_k: &MemberKey, _v: &Member, _s| Ok(()),
        )
        .and_counters(
            group_v2_counter_targets(&groups),
            move |_k: &MemberKey, v: &Member, sink| sink.add(&groups_for_hook, v.group, 1),
        );
        let _members = db
            .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
            .unwrap();
        (db, groups)
    }

    let dir = tempfile::tempdir().unwrap();
    // Session 1: a correct, engine-maintained counter, closed cleanly.
    {
        let (db, members, groups) = counted_fixture_at(dir.path());
        groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
        members
            .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
            .unwrap();
        db.close().unwrap();
    }
    // Session 2: the target migrates (the migration zeroes the maintained
    // field) and the source recounts it back.
    {
        let (db, groups) = open_pair(dir.path());
        assert_eq!(groups.get(&[1u8; 8]).unwrap().member_count, 1, "recounted");
        db.close().unwrap();
    }
    // Session 3: nothing is dirty any more, so no recount may run. Proven by
    // corrupting the target *before* the source opens: a recount would drive
    // it back to 1, so the corrupted value surviving is the evidence.
    {
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let groups = db
            .open_typed_map::<GroupV2, RapiraCodec, _>(Config::test(), NoHook, &[])
            .unwrap();
        assert!(
            !db.debug_counters_dirty("groups"),
            "the migration flag was retired by session 2's close"
        );
        groups.put(&[1u8; 8], GroupV2 { member_count: 99 }).unwrap();
        let groups_for_hook = Arc::clone(&groups);
        let hook = Projected::new(
            IndexTargets::new(),
            |_k: &MemberKey, _v: &Member, _s| Ok(()),
        )
        .and_counters(
            group_v2_counter_targets(&groups),
            move |_k: &MemberKey, v: &Member, sink| sink.add(&groups_for_hook, v.group, 1),
        );
        let _members = db
            .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
            .unwrap();
        assert_eq!(
            groups.get(&[1u8; 8]).unwrap().member_count,
            99,
            "no recount ran — a repeated one would have driven this back to 1"
        );
    }
}

#[test]
fn counter_recount_after_a_hard_exit_between_source_and_delta() {
    // Checked before creating a tempdir: `abort()` below skips `Drop`, so a
    // tempdir created before this check would leak on every child run.
    if std::env::var("ARMDB_COUNTER_CHILD").is_ok() {
        // Child: the source write lands, the delta is made to fail, and the
        // process aborts before any clean close can clear the dirty flag.
        //
        // `Db::open_test` disables the background flush worker, so a bare
        // `put` only reaches the in-memory write buffer — an `abort()` right
        // after it would lose the row along with the delta, which is not the
        // scenario this test means to cover. `durable_flush` makes the row
        // durable first, so only the delta is missing when the process dies.
        let (_db, members, _groups) = counted_fixture_at(std::path::Path::new(
            &std::env::var("ARMDB_COUNTER_DIR").unwrap(),
        ));
        crate::counter::FAIL_COUNTER_DELTA.store(true, std::sync::atomic::Ordering::SeqCst);
        let _ = members.put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] });
        let _ = members.durable_flush();
        std::process::abort();
    }

    let dir = tempfile::tempdir().unwrap();
    {
        // Parent, phase 1: create the target row and close cleanly.
        let (db, _m, groups) = counted_fixture_at(dir.path());
        groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
        db.close().unwrap();
    }

    // Re-exec this same test binary in "child" mode, filtered down to just
    // this test — mirrors the T5 hard-exit tests' self-relaunch technique
    // (`grep -rn "Command::new(std::env::current_exe" armdb/tests`), inlined
    // here because those live in a separate integration-test binary and
    // cannot see this module's `pub(crate)` fixture.
    let status = std::process::Command::new(std::env::current_exe().unwrap())
        .arg("--exact")
        .arg("counter_tests::counter_recount_after_a_hard_exit_between_source_and_delta")
        .env("ARMDB_COUNTER_CHILD", "1")
        .env("ARMDB_COUNTER_DIR", dir.path())
        .status()
        .unwrap();
    assert!(!status.success(), "the child aborts by design");

    // Parent, phase 2: reopening recounts, because the child never closed.
    let (_db, _m, groups) = counted_fixture_at(dir.path());
    assert_eq!(
        groups.get(&[1u8; 8]).unwrap().member_count,
        1,
        "the source row survived, so the recount restores its contribution"
    );
}

// ---------------------------------------------------------------------------
// Task 9: transaction integration. A counted collection mutated inside
// `db.tx()` resolves its deltas to an **absolute** value under the target's
// own lock during commit, and that absolute value is what the transaction log
// record carries — so replay is exactly-once and no dirty flag / recount is
// involved on this path.
// ---------------------------------------------------------------------------

#[test]
fn counter_tx_sums_two_deltas_into_one_absolute_write() {
    let (db, members, groups, _dir) = counted_fixture();
    groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
    db.tx()
        .put(
            &members,
            &MemberKey::new([1u8; 8], 1),
            &Member { group: [1u8; 8] },
        )
        .unwrap()
        .put(
            &members,
            &MemberKey::new([1u8; 8], 2),
            &Member { group: [1u8; 8] },
        )
        .unwrap()
        .commit()
        .unwrap();
    assert_eq!(groups.get(&[1u8; 8]).unwrap().member_count, 2);
}

#[test]
fn counter_tx_sees_a_user_write_to_the_same_target_row() {
    let (db, members, groups, _dir) = counted_fixture();
    groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
    db.tx()
        .put(&groups, &[1u8; 8], &Group { member_count: 10 }) // base op
        .unwrap()
        .put(
            &members,
            &MemberKey::new([1u8; 8], 1),
            &Member { group: [1u8; 8] },
        )
        .unwrap()
        .commit()
        .unwrap();
    // The derived delta is applied after the base write, so it reads 10, not 0.
    assert_eq!(groups.get(&[1u8; 8]).unwrap().member_count, 11);
}

/// Third tier for the topology tests below: a counter target for `groups`, so
/// `groups` can be made a counter *source* as well as a counter *target*.
#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
pub(crate) struct Org {
    pub group_count: u64,
}
impl CollectionMeta for Org {
    type SelfId = [u8; 8];
    const NAME: &'static str = "orgs";
}

/// `group_counter_targets`' twin one tier up: `orgs.group_count`, counted from
/// `groups`.
pub(crate) fn org_counter_targets<C>(orgs: &Arc<C>) -> CounterTargets
where
    C: CounterCollection + ProjectableIndexTx + IndexKeyed<K = [u8; 8], V = Org>,
{
    CounterTargets::new().with_counter(
        orgs,
        "group_count",
        OnMissingTarget::Error,
        |o: &Org| o.group_count,
        |cur: Option<&Org>, n: u64| match cur {
            Some(_) => Upsert::Put(Org { group_count: n }),
            None => Upsert::Keep,
        },
    )
}

/// A counter target is a **rank-1** participant, and the commit's rank-1 round
/// decides writes without projecting them — so a target that is itself a
/// projecting/counting source would have its own indexes and counters silently
/// stop being maintained for any write made inside a transaction. `plan`
/// refuses that topology up front instead.
///
/// Here `orgs <- groups <- members`: `groups` is both a counter target (of
/// `members`) and a counter source (of `orgs`).
#[test]
fn counter_tx_rejects_writing_a_target_that_is_itself_a_projecting_source() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open db"));

    let orgs = db
        .open_typed_map::<Org, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open orgs");
    let orgs_for_project = Arc::clone(&orgs);
    let groups_hook = Projected::<[u8; 8], Group, NoHook>::new(
        IndexTargets::new(),
        |_k: &[u8; 8], _v: &Group, _s| Ok(()),
    )
    .and_counters(
        org_counter_targets(&orgs),
        move |_k: &[u8; 8], _v: &Group, sink| sink.add(&orgs_for_project, [9u8; 8], 1),
    );
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), groups_hook, &[])
        .expect("open groups");

    let groups_for_project = Arc::clone(&groups);
    let members_hook = Projected::<MemberKey, Member, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _s| Ok(()),
    )
    .and_counters(
        group_counter_targets(&groups),
        move |_k: &MemberKey, v: &Member, sink| sink.add(&groups_for_project, v.group, 1),
    );
    let _members = db
        .open_typed_map::<Member, RapiraCodec, _>(Config::test(), members_hook, &[])
        .expect("open members");

    orgs.put(&[9u8; 8], Org { group_count: 0 }).unwrap();
    // The facade path is unaffected — it projects normally, so `orgs` moves.
    groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
    assert_eq!(orgs.get(&[9u8; 8]).unwrap().group_count, 1);

    let err = db
        .tx()
        .put(&groups, &[1u8; 8], &Group { member_count: 5 })
        .unwrap()
        .commit()
        .expect_err("a target that is itself a source cannot be written in a transaction");
    assert!(
        matches!(
            err,
            crate::DbError::TxTargetIsAlsoASource {
                collection: "groups"
            }
        ),
        "{err}"
    );
    // Refused in `plan`, before any lock and long before the log point.
    assert_eq!(groups.get(&[1u8; 8]).unwrap().member_count, 0);
    assert_eq!(orgs.get(&[9u8; 8]).unwrap().group_count, 1);
}

/// The other direction: a **plain** counter target — rank 1, but with no
/// projections of its own — is still perfectly writable inside a transaction.
/// The check above must be precise, not a blanket ban on writing targets.
#[test]
fn counter_tx_accepts_writing_a_plain_counter_target() {
    let (db, _members, groups, _dir) = counted_fixture();
    groups.put(&[1u8; 8], Group { member_count: 3 }).unwrap();
    db.tx()
        .put(&groups, &[1u8; 8], &Group { member_count: 7 })
        .unwrap()
        .commit()
        .expect("a plain counter target has nothing of its own to project");
    assert_eq!(groups.get(&[1u8; 8]).unwrap().member_count, 7);
}

/// `insert` into a **rank-1** participant — a declared counter target, decided
/// by `eval_rank1` rather than `eval_rank0`. Both outcomes, because this is the
/// only test that executes that arm at all.
#[test]
fn counter_tx_insert_into_a_plain_counter_target() {
    let (db, _members, groups, _dir) = counted_fixture();

    db.tx()
        .insert(&groups, &[1u8; 8], &Group { member_count: 3 })
        .unwrap()
        .commit()
        .expect("a free key on a plain counter target is insertable");
    assert_eq!(groups.get(&[1u8; 8]).unwrap().member_count, 3);
    db.checkpoint_now().unwrap();

    let err = db
        .tx()
        .insert(&groups, &[1u8; 8], &Group { member_count: 9 })
        .unwrap()
        .commit()
        .unwrap_err();
    assert!(matches!(err, crate::DbError::KeyExists), "{err}");
    assert!(
        !matches!(err, crate::DbError::TxFailedAfterLog { .. }),
        "the rank-1 arm must also abort before the log point: {err}"
    );
    assert_eq!(
        groups.get(&[1u8; 8]).unwrap().member_count,
        3,
        "the occupied target row is untouched"
    );
    assert!(
        db.pending_transactions().is_empty(),
        "nothing reached the log"
    );
}

/// The crash-atomicity claim, end to end: kill the process **after** the log
/// record is durable but before the counter's derived write is applied, then
/// reopen and find the counter at its exact value — restored by replaying the
/// logged **absolute** value, with no recount involved.
///
/// This is the property the transactional path has and the write path does
/// not. `counter_recount_after_a_hard_exit_between_source_and_delta` above is
/// the write path's answer to the same crash: a dirty flag and a full O(N)
/// recount at the next open. Here nothing is marked dirty at all — the
/// assertion below reads `db.info` off disk to prove it, so a passing result
/// cannot be a recount in disguise.
///
/// `ApplyFailGuard::at(1)` targets the counter write: index 0 is the source
/// `put` into `members`, and derived writes are appended after the base ops,
/// so the `groups` counter write is index 1. (Verified by probe: at index 1
/// the commit fails with `fenced = [groups]`, `members` applied, `groups`
/// still 0 — exactly the window replay must cover.)
#[test]
fn counter_tx_replay_restores_the_exact_value_after_a_hard_exit() {
    // Checked before creating a tempdir: `abort()` below skips `Drop`, so a
    // tempdir created before this check would leak on every child run.
    if let Ok(path) = std::env::var("ARMDB_COUNTER_DIR")
        && std::env::var("ARMDB_COUNTER_TX_CHILD").is_ok()
    {
        let (db, members, groups) = counted_fixture_at(std::path::Path::new(&path));
        // Fail while applying the derived counter write — after the record is
        // durable, which is exactly the window replay must cover.
        let guard = crate::txn::failpoints::ApplyFailGuard::at(1);
        let outcome = db
            .tx()
            .put(
                &members,
                &MemberKey::new([1u8; 8], 1),
                &Member { group: [1u8; 8] },
            )
            .and_then(crate::Tx::commit);
        drop(guard);
        // Pin the window rather than trusting the index. Were plan index 1
        // something other than the counter write, the counter would apply
        // before the crash and the parent's assertion would pass for the
        // wrong reason. Both checks below fail the child by *panic* (a normal
        // non-zero exit), which the parent tells apart from the `abort()`
        // signal it expects.
        assert!(
            matches!(
                &outcome,
                Err(crate::DbError::TxFailedAfterLog { fenced, .. })
                    if fenced.iter().any(|f| f.collection == "groups")
            ),
            "plan index 1 must be the counter write into `groups`, got {outcome:?}"
        );
        assert_eq!(
            groups.get(&[1u8; 8]).unwrap().member_count,
            0,
            "the counter must still be unapplied when the process dies"
        );
        std::process::abort();
    }

    let dir = tempfile::tempdir().unwrap();
    {
        let (db, _m, groups) = counted_fixture_at(dir.path());
        groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
        db.close().unwrap();
    }

    let status = std::process::Command::new(std::env::current_exe().unwrap())
        .args([
            "--exact",
            "counter_tests::counter_tx_replay_restores_the_exact_value_after_a_hard_exit",
        ])
        .env("ARMDB_COUNTER_TX_CHILD", "1")
        .env("ARMDB_COUNTER_DIR", dir.path())
        .status()
        .unwrap();
    assert!(!status.success(), "the child aborts by design");
    // `abort()` raises SIGABRT, so there is no exit code. A child that instead
    // *panicked* on one of its two invariant checks exits normally with 101 —
    // this is what keeps a wrong failpoint index from passing silently.
    #[cfg(unix)]
    assert!(
        status.code().is_none(),
        "the child must reach abort(), not fail an assertion: {status:?}"
    );

    // Read the persisted flags *before* reopening. The transactional path
    // itself still marks nothing — the counter's absolute value rides the log
    // record, so no dirty evidence is needed for it — which is what the
    // *target*'s flag proves: nothing in the commit path touched `groups`.
    // The *source*'s flag is set, but by its own open (the post-review
    // open-time mark), not by the transaction.
    let info_path = dir.path().join("db.info");
    let raw = std::fs::read_to_string(&info_path).expect("db.info");
    let info: crate::armour::DbInfo = serde_json::from_str(&raw).expect("parse db.info");
    assert!(
        !info.collections["groups"].counters_dirty,
        "the transactional path must mark nothing dirty: {raw}"
    );

    // Neutralise the source's open-time mark so the reopen below cannot
    // recount: with no flag set anywhere, `open_managed` has no trigger, and
    // the value it reads back can only have come from log replay. (Without
    // this the recount would recompute the same `1` from the source row the
    // committed record wrote, and the assertion would pass for the wrong
    // reason.) Edited as raw JSON rather than through a `Db`, because opening
    // one would checkpoint the very log record under test.
    std::fs::write(
        &info_path,
        raw.replace("\"counters_dirty\": true", "\"counters_dirty\": false")
            .replace("\"counters_dirty\":true", "\"counters_dirty\":false"),
    )
    .expect("rewrite db.info");
    let cleared = std::fs::read_to_string(&info_path).expect("db.info");
    let cleared: crate::armour::DbInfo = serde_json::from_str(&cleared).expect("parse db.info");
    assert!(
        cleared.collections.values().all(|c| !c.counters_dirty),
        "no flag may survive, or the reopen could recount instead of replaying"
    );

    // Reopening replays the logged absolute value.
    let (_db, _m, groups) = counted_fixture_at(dir.path());
    assert_eq!(groups.get(&[1u8; 8]).unwrap().member_count, 1);
}

// ---------------------------------------------------------------------------
// Review round: three fixes on top of the initial Task 8 implementation.
// ---------------------------------------------------------------------------

/// Finding 1 (critical): `recount` must durably flush the targets it wrote
/// *before* returning — both callers (`Db::open_managed`, `Db::recount_counters`)
/// clear the durable `counters_dirty` flag right after it returns, so a crash
/// between the recount and the target's own next flush would otherwise lose
/// the repaired rows *and* the flag that would trigger another recount.
#[test]
fn counter_recount_flushes_targets_before_returning_so_a_crash_cannot_lose_the_repair() {
    // Checked before creating a tempdir: `abort()` below skips `Drop`.
    if std::env::var("ARMDB_COUNTER_CHILD").is_ok() {
        // Child: the target is *already* durably drifted (99, written by the
        // parent's own clean close below, before this process even started)
        // — recount it and abort immediately, no close, no later flush of
        // any kind. If `recount` didn't durably flush the repair itself, the
        // drifted value would still be what's on disk after this process
        // dies — the parent's phase-1 write is the only thing that reached
        // disk, so this is the one arrangement that can't pass by accident
        // (unlike repairing a value already durably correct, or drifting and
        // repairing within the same unflushed child session, either of which
        // would still read back as "correct" with no flush at all).
        let (db, _members, _groups) = counted_fixture_at(std::path::Path::new(
            &std::env::var("ARMDB_COUNTER_DIR").unwrap(),
        ));
        let repaired = db.recount_counters("members").unwrap();
        assert_eq!(repaired, 1);
        std::process::abort();
    }

    let dir = tempfile::tempdir().unwrap();
    {
        let (db, members, groups) = counted_fixture_at(dir.path());
        groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
        members
            .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
            .unwrap();
        groups.put(&[1u8; 8], Group { member_count: 99 }).unwrap(); // drift, durable
        db.close().unwrap();
    }

    let status = std::process::Command::new(std::env::current_exe().unwrap())
        .arg("--exact")
        .arg("counter_tests::counter_recount_flushes_targets_before_returning_so_a_crash_cannot_lose_the_repair")
        .env("ARMDB_COUNTER_CHILD", "1")
        .env("ARMDB_COUNTER_DIR", dir.path())
        .status()
        .unwrap();
    assert!(!status.success(), "the child aborts by design");

    let (_db, _m, groups) = counted_fixture_at(dir.path());
    assert_eq!(
        groups.get(&[1u8; 8]).unwrap().member_count,
        1,
        "recount's own durable_flush made the repair survive the crash"
    );
}

/// Finding 2 (important): `Db::recount_counters` must not clear the
/// `counters_dirty` flag of a collection that is not itself a counted
/// source — for a plain counter *target*, that flag is the only evidence a
/// source's own later open reads to know a recount is owed.
///
/// It must not clear a **source's** flag either. This runs on an already-open
/// collection that may be taking concurrent writes: a write landing after its
/// source row was scanned increments the target before the repair reads it,
/// and the repair then subtracts that legitimate increment to match its stale
/// snapshot. Clearing the flag would make that drift permanent; leaving it
/// costs one idempotent recount at the next unclean open.
#[test]
fn counter_recount_counters_leaves_the_dirty_evidence_in_place() {
    let (db, members, groups, _dir) = counted_fixture();
    groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
    members
        .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
        .unwrap();
    assert!(db.debug_counters_dirty("members"), "set when it was opened");

    let repaired = db.recount_counters("members").unwrap();
    assert_eq!(repaired, 0, "nothing had actually drifted");
    assert!(
        db.debug_counters_dirty("members"),
        "an on-demand recount is not proof the counters are settled"
    );

    // Still an ordinary session flag: a clean close — which does imply the
    // writers stopped — retires it as usual.
    db.close().unwrap();
    assert!(!db.debug_counters_dirty("members"));
}

#[test]
fn counter_recount_counters_refuses_a_non_source_and_leaves_its_flag_intact() {
    let dir = tempfile::tempdir().unwrap();
    {
        let (db, members, groups) = counted_fixture_at(dir.path());
        groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
        members
            .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
            .unwrap();
        db.close().unwrap();
    }
    // Migrate `groups` alone, in a session that never opens `members` — its
    // own migration sets its flag exactly like the target-only-migration
    // test above, but here nothing ever consumes it: this session only
    // attempts `recount_counters("groups")` directly.
    fn zero_member_count(_k: &[u8; 8], _v: &GroupV2) -> crate::MigrateAction<GroupV2> {
        crate::MigrateAction::Update(GroupV2 { member_count: 0 })
    }
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let _groups = db
        .open_typed_map::<GroupV2, RapiraCodec, _>(
            Config::test(),
            NoHook,
            &[(0, zero_member_count)],
        )
        .unwrap();
    assert!(
        db.debug_counters_dirty("groups"),
        "the migration set the flag"
    );

    let err = db.recount_counters("groups").unwrap_err();
    assert!(
        matches!(err, crate::DbError::NotACounterSource { .. }),
        "groups declares no counter targets of its own"
    );
    assert!(
        db.debug_counters_dirty("groups"),
        "refused, not silently cleared — a later source's open still needs to see this flag"
    );
}

/// Finding 3 (important, plan-mandated): a live source row whose target row
/// was deleted (a "dangling contribution") must not make the database
/// unopenable. On the write path the same state costs one write; at open, a
/// hard failure here would cost the whole database (every subsequent open
/// hits the same error and never clears the flag that triggers it). The
/// recount degrades this to a `warn` and continues — diagnosis moves to
/// `schema::validate` (Task 11) instead of an unstartable database.
#[test]
fn counter_recount_dangling_contribution_does_not_block_open() {
    let dir = tempfile::tempdir().unwrap();
    crate::projection::FAIL_COLLECTION_FLUSH.store(true, std::sync::atomic::Ordering::SeqCst);
    {
        let (db, members, groups) = counted_fixture_at(dir.path());
        groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
        members
            .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
            .unwrap();
        // A second contribution whose target row never exists. The write
        // itself hard-fails — the write path keeps `OnMissingTarget::Error`
        // — but the source row still lands, same "write lands, delta fails"
        // shape as the hard-exit test above.
        let _ = members.put(&MemberKey::new([2u8; 8], 1), Member { group: [2u8; 8] });
        let _ = db.close();
        // `db` drops here — keep the failpoint armed through the implicit
        // close too (see `counter_recount_runs_when_the_previous_close_failed`).
    }
    crate::projection::FAIL_COLLECTION_FLUSH.store(false, std::sync::atomic::Ordering::SeqCst);

    // Reopening must succeed: a dangling contribution degrades to a warning
    // during the recount, it does not make the database unopenable.
    let (db, _members, groups) = counted_fixture_at(dir.path());
    assert_eq!(
        groups.get(&[1u8; 8]).unwrap().member_count,
        1,
        "the valid contribution is still counted correctly"
    );
    assert!(
        groups.get(&[2u8; 8]).is_none(),
        "the dangling target was never created"
    );
    // That the recount actually ran to completion (rather than the open
    // silently skipping it) is already proven by the value assertions above —
    // a skipped recount could not know the valid contribution is 1 or that
    // the dangling target was never materialised. `!dirty` after `close()`
    // below is *not* further evidence of that: every counted source marks
    // itself dirty at open unconditionally (see `db.rs`'s "the dirty seam"),
    // whether or not a recount ran this session, so a clean close always
    // clears the mark this session's own open set. What it does still rule
    // out is an *aborted* recount — a clean close is only reachable when
    // nothing errored along the way.
    db.close().unwrap();
    assert!(
        !db.debug_counters_dirty("members"),
        "this session's own open-time mark was retired by a clean close"
    );
}

// ---------------------------------------------------------------------------
// Final whole-branch review: the two places the "no delta is lost without
// durable evidence" invariant was silently waived, plus the coverage the
// review found missing (a FixedStore counter *target*, a real `one_of` group,
// and `CounterApply::Delete`/`Skip` through the transaction path).
// ---------------------------------------------------------------------------

/// Fix 1: a `counters_dirty` mark that **fails** must fail the **open** of the
/// counted source, so it never takes a write.
///
/// The mark used to run in the write-path teardown, and "abort the write" was
/// not something it could deliver there: `Ctx::with_shard` releases the shard
/// lock before the teardown runs, so the source row was already committed by
/// the time the mark could return an error — leaving a written row whose delta
/// was skipped, and a retry of the same `put` produced `old == new` and hence
/// no delta at all. Moving the mark to open makes the refusal real: the
/// collection is not published, so there is no handle to write through.
#[test]
fn counter_dirty_mark_failure_fails_the_open_of_a_counted_source() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open db"));
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("a plain target opens fine — it is not a counted source");
    let groups_for_project = Arc::clone(&groups);
    let hook = Projected::<MemberKey, Member, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _s| Ok(()),
    )
    .and_counters(
        group_counter_targets(&groups),
        move |_k: &MemberKey, v: &Member, sink| sink.add(&groups_for_project, v.group, 1),
    );

    crate::counter_dirty::FAIL_COUNTER_MARK.store(true, std::sync::atomic::Ordering::SeqCst);
    let err = db
        .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .err()
        .expect("a source that cannot record evidence must not start taking writes");
    crate::counter_dirty::FAIL_COUNTER_MARK.store(false, std::sync::atomic::Ordering::SeqCst);

    assert!(
        matches!(
            err,
            crate::DbError::Internal("injected counters_dirty mark failure")
        ),
        "the mark's own error is what propagates: {err}"
    );
    assert!(
        !db.debug_counters_dirty("members"),
        "nothing was persisted — which is exactly why the open had to fail"
    );

    // The name was released with the failed open, so a second attempt — with
    // the mark working — succeeds and behaves normally.
    let groups_for_project = Arc::clone(&groups);
    let hook = Projected::<MemberKey, Member, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _s| Ok(()),
    )
    .and_counters(
        group_counter_targets(&groups),
        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 with a working mark");
    assert!(db.debug_counters_dirty("members"));

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

/// Fix 2: a `counters_dirty` flag a **migration** set on a counter *target*
/// must survive a clean close by a session that never opened the source.
///
/// The flag has two readings, and only a source's open consumes the target
/// one. An offline migration tool (or phased adoption before the carrier is
/// attached) migrates `groups` alone; the old `close` cleared every open
/// collection's flag whenever finalisation succeeded, so the next session saw
/// nothing to recount and the counter the migration invalidated stayed wrong
/// forever.
#[test]
fn counter_dirty_on_a_migrated_target_survives_a_close_that_consumed_nothing() {
    fn zero_member_count(_k: &[u8; 8], _v: &GroupV2) -> crate::MigrateAction<GroupV2> {
        crate::MigrateAction::Update(GroupV2 { member_count: 0 })
    }

    let dir = tempfile::tempdir().unwrap();
    // Session 1: a correct, engine-maintained counter, closed cleanly.
    {
        let (db, members, groups) = counted_fixture_at(dir.path());
        groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
        members
            .put(&MemberKey::new([1u8; 8], 1), Member { group: [1u8; 8] })
            .unwrap();
        db.close().unwrap();
        assert!(
            !db.debug_counters_dirty("members"),
            "the source's own flag is cleared"
        );
    }

    // Session 2: an offline migration tool — migrates the TARGET only (the
    // migration zeroes the maintained field), never opens the source, and
    // closes cleanly. Nothing here consumed the flag it set.
    {
        let db = Arc::new(Db::open_test(dir.path()).unwrap());
        let groups = db
            .open_typed_map::<GroupV2, RapiraCodec, _>(
                Config::test(),
                NoHook,
                &[(0, zero_member_count)],
            )
            .unwrap();
        assert!(
            db.debug_counters_dirty("groups"),
            "the migration marks the target"
        );
        assert_eq!(
            groups.get(&[1u8; 8]).unwrap().member_count,
            0,
            "the migration destroyed the maintained value"
        );
        db.close().unwrap();
        assert!(
            db.debug_counters_dirty("groups"),
            "a clean close that consumed nothing must leave the evidence in place"
        );
    }

    // Session 3: the source finally opens, reads the target's surviving flag,
    // and recounts.
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let groups = db
        .open_typed_map::<GroupV2, RapiraCodec, _>(
            Config::test(),
            NoHook,
            &[(0, zero_member_count)],
        )
        .unwrap();
    let groups_for_hook = Arc::clone(&groups);
    let hook = Projected::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _s| Ok(()),
    )
    .and_counters(
        group_v2_counter_targets(&groups),
        move |_k: &MemberKey, v: &Member, sink| sink.add(&groups_for_hook, v.group, 1),
    );
    let _members = db
        .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .unwrap();
    assert_eq!(
        groups.get(&[1u8; 8]).unwrap().member_count,
        1,
        "the surviving target flag drove the recount"
    );
}

/// Fix 5a: a **FixedStore** counter target fed by a **Bitcask** source — the
/// consumer's actual shape (every one of its dedicated counter collections is
/// FixedStore, while the counted sources are ordinary Bitcask collections).
/// `counter_mutation_matrix_zero_tree_fixed` already covers Fixed→Fixed; this
/// is the mixed-backend combination, on the `ZeroMap` side.
#[test]
fn counter_mutation_matrix_fixed_map_target_bitcask_source() {
    let dir = tempfile::tempdir().expect("tempdir");
    let db = Arc::new(Db::open_test(dir.path()).expect("open db"));
    let groups = db
        .open_zero_map_fixed::<ZeroGroup, { size_of::<ZeroGroup>() }, _>(
            crate::FixedConfig::test(),
            NoHook,
            &[],
        )
        .expect("open groups");
    let counter_targets = zero_group_counter_targets(&groups);
    let groups_for_project = Arc::clone(&groups);
    let hook = Projected::<MemberKey, Member, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _sink| Ok(()),
    )
    .and_counters(counter_targets, move |_k: &MemberKey, v: &Member, sink| {
        sink.add(&groups_for_project, v.group, 1)
    });
    // Bitcask source into a FixedStore target.
    let members = db
        .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .expect("open members");

    let g1 = [1u8; 8];
    let g2 = [2u8; 8];
    groups.put(&g1, &ZeroGroup { member_count: 0 }).unwrap();
    groups.put(&g2, &ZeroGroup { member_count: 0 }).unwrap();

    members
        .put(&MemberKey::new(g1, 10), Member { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    members
        .put(&MemberKey::new(g1, 10), Member { group: g1 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 1);

    members
        .put(&MemberKey::new(g1, 10), Member { group: g2 })
        .unwrap();
    assert_eq!(groups.get(&g1).unwrap().member_count, 0);
    assert_eq!(groups.get(&g2).unwrap().member_count, 1);

    members.delete(&MemberKey::new(g1, 10)).unwrap();
    assert_eq!(groups.get(&g2).unwrap().member_count, 0);

    // The FixedStore target also carries the recount: drift it behind the
    // engine's back and let `recount_counters` walk both collections.
    groups.put(&g2, &ZeroGroup { member_count: 77 }).unwrap();
    members
        .put(&MemberKey::new(g2, 11), Member { group: g2 })
        .unwrap();
    assert_eq!(db.recount_counters("members").unwrap(), 1);
    assert_eq!(groups.get(&g2).unwrap().member_count, 1, "recounted");
}

// --- Fix 5b: a real `one_of` group, two live collections ------------------

/// The `one_of` alternative to `Group` — same key space, different collection,
/// so a `chat_id` is held by exactly one of the two (or, when it is corrupt,
/// by both).
#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
pub(crate) struct Channel {
    pub subscriber_count: u64,
}
impl CollectionMeta for Channel {
    type SelfId = [u8; 8];
    const NAME: &'static str = "channels";
}

pub(crate) type ChannelsMap = crate::typed_map!(Channel, RapiraCodec, NoHook);

/// `groups` / `channels` / `members`, with `members` counting into whichever
/// of the two holds its `group` key — the consumer's live `one_of` shape.
fn one_of_fixture_at(
    path: &std::path::Path,
) -> (Arc<Db>, Arc<MembersMap>, Arc<GroupsMap>, Arc<ChannelsMap>) {
    let db = Arc::new(Db::open_test(path).expect("open db"));
    let groups = db
        .open_typed_map::<Group, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open groups");
    let channels = db
        .open_typed_map::<Channel, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open channels");
    let targets = CounterTargets::new()
        .one_of()
        .with(
            &groups,
            "member_count",
            OnMissingTarget::Error,
            |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,
            },
        )
        .with(
            &channels,
            "subscriber_count",
            OnMissingTarget::Error,
            |c: &Channel| c.subscriber_count,
            |cur: Option<&Channel>, n: u64| match cur {
                #[allow(clippy::needless_update)]
                Some(c) => Upsert::Put(Channel {
                    subscriber_count: n,
                    ..c.clone()
                }),
                None => Upsert::Keep,
            },
        )
        .build();
    let hook = Projected::<MemberKey, Member, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _sink| Ok(()),
    )
    .and_counters(targets, |_k: &MemberKey, v: &Member, sink| {
        sink.add_one_of(v.group, 1)
    });
    let members = db
        .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .expect("open members");
    (db, members, groups, channels)
}

/// Fix 5b: `one_of` end to end against two **real** collections. The unit
/// tests in `counter.rs` run against `StubCounterColl`, whose `contains_bytes`
/// ignores the key entirely — it cannot tell "member A holds key X" from
/// "member A holds *something*", so it cannot prove routing at all. Here each
/// alternative genuinely holds only its own keys.
#[test]
fn counter_one_of_routes_to_whichever_collection_holds_the_key() {
    let dir = tempfile::tempdir().expect("tempdir");
    let (_db, members, groups, channels) = one_of_fixture_at(dir.path());
    let g = [1u8; 8];
    let c = [2u8; 8];
    groups.put(&g, Group { member_count: 0 }).unwrap();
    channels
        .put(
            &c,
            Channel {
                subscriber_count: 0,
            },
        )
        .unwrap();

    members
        .put(&MemberKey::new(g, 10), Member { group: g })
        .unwrap();
    members
        .put(&MemberKey::new(c, 11), Member { group: c })
        .unwrap();
    assert_eq!(groups.get(&g).unwrap().member_count, 1, "routed to groups");
    assert_eq!(
        channels.get(&c).unwrap().subscriber_count,
        1,
        "routed to channels"
    );
    // Each alternative counts only its own key — the stub could never show this.
    assert!(channels.get(&g).is_none());
    assert!(groups.get(&c).is_none());

    members.delete(&MemberKey::new(c, 11)).unwrap();
    assert_eq!(channels.get(&c).unwrap().subscriber_count, 0);
    assert_eq!(groups.get(&g).unwrap().member_count, 1, "untouched");
}

/// Fix 5b, failure direction 1 on the write path and in the recount: **zero**
/// holders. The write fails with `CounterTargetMissing`; the recount degrades
/// the same state to a warning so the database still opens.
#[test]
fn counter_one_of_with_no_holder_fails_the_write_and_warns_in_the_recount() {
    let dir = tempfile::tempdir().expect("tempdir");
    let (db, members, groups, channels) = one_of_fixture_at(dir.path());
    let g = [1u8; 8];
    let orphan = [9u8; 8];
    groups.put(&g, Group { member_count: 0 }).unwrap();

    let Err(err) = members.put(&MemberKey::new(orphan, 10), Member { group: orphan }) else {
        panic!("no alternative holds this key");
    };
    assert!(
        matches!(
            err,
            crate::DbError::ProjectionIncomplete(_) | crate::DbError::CounterTargetMissing { .. }
        ),
        "{err}"
    );
    assert!(groups.get(&orphan).is_none());
    assert!(channels.get(&orphan).is_none());

    // The source row landed anyway (the delta is applied after the source's
    // own lock is released), so the recount meets the same dangling
    // contribution — and must warn past it rather than refuse to open.
    members
        .put(&MemberKey::new(g, 11), Member { group: g })
        .unwrap();
    let repaired = db
        .recount_counters("members")
        .expect("a dangling one_of contribution must not abort the recount");
    assert_eq!(groups.get(&g).unwrap().member_count, 1);
    assert_eq!(repaired, 0, "nothing had drifted; the orphan was skipped");
}

/// Fix 5b, failure direction 2: **two** holders. Ambiguity has no safe target
/// to skip to, so it is a hard error on the write path *and* in the recount.
#[test]
fn counter_one_of_with_two_holders_is_ambiguous_on_the_write_path_and_in_the_recount() {
    let dir = tempfile::tempdir().expect("tempdir");
    let (db, members, groups, channels) = one_of_fixture_at(dir.path());
    let both = [3u8; 8];
    // The corruption `one_of` exists to detect: one key tagged into both.
    groups.put(&both, Group { member_count: 0 }).unwrap();
    channels
        .put(
            &both,
            Channel {
                subscriber_count: 0,
            },
        )
        .unwrap();

    let Err(err) = members.put(&MemberKey::new(both, 10), Member { group: both }) else {
        panic!("two holders is ambiguous");
    };
    assert!(
        matches!(
            err,
            crate::DbError::ProjectionIncomplete(_) | crate::DbError::AmbiguousCounterTarget { .. }
        ),
        "{err}"
    );

    let err = db
        .recount_counters("members")
        .expect_err("ambiguity is a hard failure in the recount too");
    assert!(
        matches!(err, crate::DbError::AmbiguousCounterTarget { .. }),
        "{err}"
    );
}

// --- Fix 5c: CounterApply::Delete / Skip through the transaction path ------

/// The `thread_meta` shape — the counter owns the row's whole lifecycle:
/// created on the first contribution (`OnMissingTarget::Create`) and deleted
/// when the count returns to zero. 2 of the consumer's 7 live declarations.
#[derive(Clone, Debug, PartialEq, GetType, GetRefs, Rapira)]
pub(crate) struct ThreadMeta {
    pub reply_count: u64,
}
impl CollectionMeta for ThreadMeta {
    type SelfId = [u8; 8];
    const NAME: &'static str = "thread_meta";
}

pub(crate) type ThreadMetaMap = crate::typed_map!(ThreadMeta, RapiraCodec, NoHook);

fn thread_meta_fixture_at(
    path: &std::path::Path,
    on_missing: OnMissingTarget,
) -> (Arc<Db>, Arc<MembersMap>, Arc<ThreadMetaMap>) {
    let db = Arc::new(Db::open_test(path).expect("open db"));
    let meta = db
        .open_typed_map::<ThreadMeta, RapiraCodec, _>(Config::test(), NoHook, &[])
        .expect("open thread_meta");
    let targets = CounterTargets::new().with_counter(
        &meta,
        "reply_count",
        on_missing,
        |m: &ThreadMeta| m.reply_count,
        |cur: Option<&ThreadMeta>, n: u64| match (cur, n) {
            (_, 0) => Upsert::Delete,
            #[allow(clippy::needless_update)]
            (Some(m), n) => Upsert::Put(ThreadMeta {
                reply_count: n,
                ..m.clone()
            }),
            (None, n) => Upsert::Put(ThreadMeta { reply_count: n }),
        },
    );
    let meta_for_project = Arc::clone(&meta);
    let hook = Projected::<MemberKey, Member, NoHook>::new(
        IndexTargets::new(),
        |_k: &MemberKey, _v: &Member, _sink| Ok(()),
    )
    .and_counters(targets, move |_k: &MemberKey, v: &Member, sink| {
        sink.add(&meta_for_project, v.group, 1)
    });
    let threads = db
        .open_typed_map::<Member, RapiraCodec, _>(Config::test(), hook, &[])
        .expect("open threads");
    (db, threads, meta)
}

/// Fix 5c: `CounterApply::Create` **and** `CounterApply::Delete` resolved
/// inside a `db.tx()` commit. The transaction path previously only ever saw
/// `Put` on an existing row: delete-at-zero (the `thread_meta` shape) was
/// exercised on the facade path alone.
#[test]
fn counter_tx_creates_and_deletes_a_target_row_it_owns() {
    let dir = tempfile::tempdir().expect("tempdir");
    let t = [1u8; 8];
    {
        // Scoped: every collection handle holds its own directory lock, so all
        // three must drop before the reopen below.
        let (db, threads, meta) = thread_meta_fixture_at(dir.path(), OnMissingTarget::Create);
        assert!(meta.get(&t).is_none(), "no row before the first reply");

        // Create: the target row does not exist and the commit materialises it.
        db.tx()
            .put(&threads, &MemberKey::new(t, 1), &Member { group: t })
            .unwrap()
            .put(&threads, &MemberKey::new(t, 2), &Member { group: t })
            .unwrap()
            .commit()
            .unwrap();
        assert_eq!(meta.get(&t).unwrap().reply_count, 2);

        // Delete: the last contribution goes away and the row goes with it.
        db.tx()
            .delete(&threads, &MemberKey::new(t, 1))
            .unwrap()
            .delete(&threads, &MemberKey::new(t, 2))
            .unwrap()
            .commit()
            .unwrap();
        assert!(
            meta.get(&t).is_none(),
            "CounterApply::Delete must reach the log record as a delete"
        );
        db.close().unwrap();
    }
    // Reopening proves the delete really was in the durable record.
    let (_db, _threads, meta) = thread_meta_fixture_at(dir.path(), OnMissingTarget::Create);
    assert!(meta.get(&t).is_none(), "still gone after reopen");
}

/// Fix 5c, the third arm: `CounterApply::Skip`. With `OnMissingTarget::Skip`
/// a contribution to a target row that does not exist is deliberately dropped
/// — the commit succeeds and writes nothing at all for that counter.
#[test]
fn counter_tx_skips_a_contribution_whose_target_row_is_missing() {
    let dir = tempfile::tempdir().expect("tempdir");
    let (db, threads, meta) = thread_meta_fixture_at(dir.path(), OnMissingTarget::Skip);
    let t = [1u8; 8];
    db.tx()
        .put(&threads, &MemberKey::new(t, 1), &Member { group: t })
        .unwrap()
        .commit()
        .expect("a Skip target must not fail the commit");
    assert!(
        meta.get(&t).is_none(),
        "Skip writes nothing — not a zero row"
    );

    // And once the row exists, the same source keeps counting normally.
    meta.put(&t, ThreadMeta { reply_count: 0 }).unwrap();
    db.tx()
        .put(&threads, &MemberKey::new(t, 2), &Member { group: t })
        .unwrap()
        .commit()
        .unwrap();
    assert_eq!(meta.get(&t).unwrap().reply_count, 1);
}

/// Fix 4 (documented v1 boundary, no code change): `one_of` resolves against
/// **storage**, in `project_rank0` — before the transaction's own pending
/// writes exist as rows. So creating the parent row and its first contributing
/// source row in the *same* `db.tx()` fails with `CounterTargetMissing`, while
/// the identical shape with a single (non-`one_of`) target succeeds, because a
/// single target never probes `contains` at all.
///
/// This test pins the boundary rather than the fix; resolving groups against
/// pending writes is a follow-up. The workaround — create the parent in a
/// prior transaction — is asserted below and documented in
/// `docs/schema-metadata.md` and the T9 breaking-changes note.
#[test]
fn counter_one_of_cannot_resolve_a_target_created_in_the_same_transaction() {
    let dir = tempfile::tempdir().expect("tempdir");
    let (db, members, groups, _channels) = one_of_fixture_at(dir.path());
    let g = [1u8; 8];

    let err = db
        .tx()
        .put(&groups, &g, &Group { member_count: 0 })
        .unwrap()
        .put(&members, &MemberKey::new(g, 10), &Member { group: g })
        .unwrap()
        .commit()
        .expect_err("v1: the pending parent row is not visible to `contains`");
    assert!(
        matches!(err, crate::DbError::CounterTargetMissing { .. }),
        "{err}"
    );
    assert!(groups.get(&g).is_none(), "the whole transaction is refused");

    // The documented workaround: the parent lands in its own transaction first.
    db.tx()
        .put(&groups, &g, &Group { member_count: 0 })
        .unwrap()
        .commit()
        .unwrap();
    db.tx()
        .put(&members, &MemberKey::new(g, 10), &Member { group: g })
        .unwrap()
        .commit()
        .unwrap();
    assert_eq!(groups.get(&g).unwrap().member_count, 1);
}

/// The counter half of the same claim: a successful `insert` into a counted
/// source contributes exactly once, and a refused one contributes nothing.
#[test]
fn counter_tx_insert_contributes_once_and_not_at_all_when_refused() {
    let (db, members, groups, _dir) = counted_fixture();
    groups.put(&[1u8; 8], Group { member_count: 0 }).unwrap();
    let k = MemberKey::new([1u8; 8], 1);

    db.tx()
        .insert(&members, &k, &Member { group: [1u8; 8] })
        .unwrap()
        .commit()
        .unwrap();
    assert_eq!(groups.get(&[1u8; 8]).unwrap().member_count, 1);

    let err = db
        .tx()
        .insert(&members, &k, &Member { group: [1u8; 8] })
        .unwrap()
        .commit()
        .unwrap_err();
    assert!(matches!(err, crate::DbError::KeyExists), "{err}");
    assert_eq!(
        groups.get(&[1u8; 8]).unwrap().member_count,
        1,
        "a refused insert must not move the counter"
    );
}