armdb 0.14.2

//! Claim-set diff (T10 Task 4): compute which claims a write's old and new
//! values had, and the delta between them.
//!
//! This is the claim-side twin of [`crate::projection::project_snapshot`] /
//! `diff_projection` (`projection.rs:315-335`) — read those first, this
//! module mirrors their shape closely — with one deliberate difference: the
//! set here is a [`BTreeSet`], not a `HashMap`. `project_snapshot` only ever
//! compares two sets for equality, so hash-iteration order does not matter;
//! a claim delta is *ordered work* — Task 7's transaction resolver must
//! apply releases before claims (handing a key from one owner to another in
//! one transaction resolves the release first, or the claim rejects itself
//! against a key its own source still appears to hold) — so iteration order
//! must be deterministic, not hash-seed-dependent.
//!
//! Task 5 (the non-transactional rule) uses `first_changed` — which names the
//! first namespace whose claim subset differs, for the refusal's error — to
//! refuse a write outside `db.tx()` whose claim set would change; Task 7 (the
//! transaction resolver) uses `claim_delta` for its ordered release-then-claim
//! resolution. `claim_set`/`claim_set_bytes` back both. (`claims_changed`, an
//! earlier boolean-only twin of `first_changed`, was deleted once
//! `first_changed` — which also names the namespace — became the guard's
//! actual caller and left it with no production caller of its own; the test
//! that exercised it now exercises `first_changed`'s "no change" case
//! instead.)
//!
//! Everything above is bound on [`TypedWriteHook<K, T>`] — the two families
//! that own their value in memory (`TypedTree`/`TypedMap`). The four
//! byte-level families (`ConstTree`/`ConstMap`, backing `ZeroTree`/`ZeroMap`;
//! `VarTree`/`VarMap`, backing `VarTypedTree`/`VarTypedMap`) run their locked
//! write bodies over [`WriteHook<K>`] and `Option<&[u8]>` snapshots instead —
//! `claim_set_bytes`/`first_changed_bytes` are their twins, mirroring
//! `crate::projection::project_snapshot_bytes` /
//! `crate::projection::diff_projection_bytes` exactly the way the typed
//! functions above mirror `project_snapshot`/`diff_projection`.

use std::collections::BTreeSet;

use crate::Key;
use crate::error::DbResult;
use crate::hook::{TypedWriteHook, WriteHook};
use crate::unique::{ClaimIntent, ClaimSink, UniqueTargets};

/// The set of `(namespace, key bytes)` claims `hook`'s claim projection makes
/// for `snapshot` at `key` — empty when `snapshot` is `None` (an absent row,
/// e.g. before an insert or after a delete, claims nothing).
///
/// `max_rows`/`max_bytes` are the caller's already-resolved budget
/// (`min(carrier, Config::max_projected_*)`, exactly as for the index and
/// counter halves) — this function does not read `Config` itself.
pub(crate) fn claim_set<K: Key, T, H: TypedWriteHook<K, T>>(
    hook: &H,
    targets: &UniqueTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    snapshot: Option<&T>,
) -> DbResult<BTreeSet<(&'static str, Vec<u8>)>> {
    let mut set = BTreeSet::new();
    if let Some(value) = snapshot {
        let mut sink = ClaimSink::new(targets, max_rows, max_bytes);
        hook.claim_project(key, value, &mut sink)?;
        for intent in sink.claims() {
            set.insert((intent.namespace, intent.key.clone()));
        }
    }
    Ok(set)
}

/// The first `(namespace, key)` claim that differs between `old`'s and
/// `new`'s claim sets, or `None` when the two sets are equal — the namespace
/// half names [`crate::error::DbError::UniqueIndexRequiresTx`]'s `namespace`
/// field. `BTreeSet::symmetric_difference` walks both sets in sorted
/// `(namespace, key)` order, so which entry is "first" is deterministic, not
/// an artifact of claim-emission order.
pub(crate) fn first_changed<K: Key, T, H: TypedWriteHook<K, T>>(
    hook: &H,
    targets: &UniqueTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    old: Option<&T>,
    new: Option<&T>,
) -> DbResult<Option<&'static str>> {
    let before = claim_set(hook, targets, max_rows, max_bytes, key, old)?;
    let after = claim_set(hook, targets, max_rows, max_bytes, key, new)?;
    Ok(before
        .symmetric_difference(&after)
        .next()
        .map(|(namespace, _)| *namespace))
}

/// The ordered claim delta between `old` and `new`: `releases` are keys
/// `old` claimed that `new` does not, `claims` are keys `new` claims that
/// `old` did not. Task 7's transaction resolver applies `releases` before
/// `claims`.
#[derive(Debug, Default, PartialEq, Eq)]
pub(crate) struct ClaimDelta {
    pub claims: Vec<ClaimIntent>,
    pub releases: Vec<ClaimIntent>,
}

/// Compute [`ClaimDelta`] between `old` and `new` — see the module docs for
/// why the underlying sets are `BTreeSet`, not `HashMap`.
pub(crate) fn claim_delta<K: Key, T, H: TypedWriteHook<K, T>>(
    hook: &H,
    targets: &UniqueTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    old: Option<&T>,
    new: Option<&T>,
) -> DbResult<ClaimDelta> {
    let before = claim_set(hook, targets, max_rows, max_bytes, key, old)?;
    let after = claim_set(hook, targets, max_rows, max_bytes, key, new)?;
    Ok(ClaimDelta {
        releases: before
            .difference(&after)
            .cloned()
            .map(into_intent)
            .collect(),
        claims: after
            .difference(&before)
            .cloned()
            .map(into_intent)
            .collect(),
    })
}

fn into_intent((namespace, key): (&'static str, Vec<u8>)) -> ClaimIntent {
    ClaimIntent { namespace, key }
}

/// Byte-snapshot twin of [`claim_set`] for the `WriteHook` (Const/Var)
/// families — mirrors [`crate::projection::project_snapshot_bytes`]. The
/// byte-level [`WriteHook::claim_project`] takes `(_old, new)` together (like
/// [`WriteHook::project`]/`counter_project`) and only ever reads `new` — the
/// adapter decodes it and forwards to the inner [`TypedWriteHook`]'s
/// single-snapshot `claim_project`. So this function, exactly like
/// `project_snapshot_bytes`, calls it once per snapshot with that snapshot
/// passed as `new`.
pub(crate) fn claim_set_bytes<K: Key, H: WriteHook<K>>(
    hook: &H,
    targets: &UniqueTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    snapshot: Option<&[u8]>,
) -> DbResult<BTreeSet<(&'static str, Vec<u8>)>> {
    let mut set = BTreeSet::new();
    if let Some(value) = snapshot {
        let mut sink = ClaimSink::new(targets, max_rows, max_bytes);
        hook.claim_project(key, None, Some(value), &mut sink)?;
        for intent in sink.claims() {
            set.insert((intent.namespace, intent.key.clone()));
        }
    }
    Ok(set)
}

/// Byte-snapshot twin of [`first_changed`] for the `WriteHook` (Const/Var)
/// families — used by [`crate::unique::guard_nontx_write_bytes`] for
/// `ZeroTree`/`ZeroMap`/`VarTypedTree`/`VarTypedMap`'s locked write bodies
/// (`ConstTree`/`ConstMap`/`VarTree`/`VarMap`), which own their hook as
/// `H: WriteHook<K>` and never materialize a decoded `T` at the write path.
pub(crate) fn first_changed_bytes<K: Key, H: WriteHook<K>>(
    hook: &H,
    targets: &UniqueTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    old: Option<&[u8]>,
    new: Option<&[u8]>,
) -> DbResult<Option<&'static str>> {
    let before = claim_set_bytes(hook, targets, max_rows, max_bytes, key, old)?;
    let after = claim_set_bytes(hook, targets, max_rows, max_bytes, key, new)?;
    Ok(before
        .symmetric_difference(&after)
        .next()
        .map(|(namespace, _)| *namespace))
}

/// Byte-snapshot twin of [`claim_delta`] for the `WriteHook` (Const/Var)
/// families — mirrors [`crate::counter::diff_counters_bytes`] /
/// [`crate::projection::diff_projection_bytes`]. T10 Task 6: consumed by
/// `ConstTree`/`ConstMap`/`VarTree`/`VarMap`'s `tx_claims_for`.
pub(crate) fn claim_delta_bytes<K: Key, H: WriteHook<K>>(
    hook: &H,
    targets: &UniqueTargets,
    max_rows: usize,
    max_bytes: usize,
    key: &K,
    old: Option<&[u8]>,
    new: Option<&[u8]>,
) -> DbResult<ClaimDelta> {
    let before = claim_set_bytes(hook, targets, max_rows, max_bytes, key, old)?;
    let after = claim_set_bytes(hook, targets, max_rows, max_bytes, key, new)?;
    Ok(ClaimDelta {
        releases: before
            .difference(&after)
            .cloned()
            .map(into_intent)
            .collect(),
        claims: after
            .difference(&before)
            .cloned()
            .map(into_intent)
            .collect(),
    })
}

#[cfg(all(test, feature = "armour"))]
mod tests {
    use std::sync::Arc;

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

    use super::*;
    use crate::armour::Db;
    use crate::unique::UniqueIndex;
    use crate::{FixedConfig, unique_index};

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

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

    unique_index!(
        DiffTestUsernames,
        key = UsernameKey,
        name = "unique_diff_test_usernames"
    );

    /// Source value: `username` drives the claim, `avatar` is along for the
    /// ride to prove the diff is over *claims*, not the whole value — two
    /// values with the same `username` but a different `avatar` must not
    /// register as a claim-set change.
    struct User {
        username: Option<String>,
        avatar: u32,
    }

    /// Claims `value.username` (when set) into the `usernames` namespace.
    /// Holds its own `Db`/`TempDir` alive for as long as the fixture is —
    /// leaking the tempdir here would be the same bug T10 Task 3's own
    /// review fix F7 called out for the integration tests.
    struct ClaimingHook {
        usernames: Arc<UniqueIndex<DiffTestUsernames>>,
        _db: Arc<Db>,
        _dir: tempfile::TempDir,
    }

    impl TypedWriteHook<[u8; 8], User> for ClaimingHook {
        const CLAIMS: bool = true;

        fn on_write(&self, _key: &[u8; 8], _old: Option<&User>, _new: Option<&User>) {}

        fn claim_project(
            &self,
            _key: &[u8; 8],
            value: &User,
            sink: &mut ClaimSink<'_>,
        ) -> DbResult<()> {
            if let Some(username) = &value.username {
                sink.claim(&self.usernames, UsernameKey::new(username))?;
            }
            Ok(())
        }

        fn claim_targets(&self) -> Option<&UniqueTargets> {
            None
        }
    }

    fn key() -> [u8; 8] {
        [7u8; 8]
    }

    fn fixture() -> (ClaimingHook, UniqueTargets) {
        let dir = tempfile::tempdir().expect("tempdir");
        let db = Arc::new(Db::open_test(dir.path()).expect("open db"));
        let usernames = db
            .open_unique_index::<DiffTestUsernames>(FixedConfig::test())
            .expect("open namespace");
        let targets = UniqueTargets::new().with(&usernames);
        (
            ClaimingHook {
                usernames,
                _db: db,
                _dir: dir,
            },
            targets,
        )
    }

    /// The bytes an owned `UsernameKey` would encode to — lets a test assert
    /// *which* key landed in a `ClaimIntent`, not just how many did.
    fn username_key_bytes(s: &str) -> Vec<u8> {
        crate::Key::as_bytes(&UsernameKey::new(s)).to_vec()
    }

    #[test]
    fn unique_diff_reports_no_change_when_the_claim_set_is_equal() {
        let (hook, targets) = fixture();
        let a = User {
            username: Some("rail".into()),
            avatar: 1,
        };
        let b = User {
            username: Some("rail".into()),
            avatar: 2,
        };
        assert_ne!(
            a.avatar, b.avatar,
            "fixture must vary a non-claimed field, or this test cannot tell \
             'diff ignores non-claim fields' from 'diff ignores everything'"
        );

        // Pin that the set genuinely holds the "rail" claim — otherwise
        // "unchanged" would pass just as well for a hook that claims
        // nothing at all, which is not what this test's name asserts.
        let set = claim_set(&hook, &targets, 4096, 1 << 20, &key(), Some(&a)).unwrap();
        assert_eq!(
            set,
            BTreeSet::from([(hook.usernames.name(), username_key_bytes("rail"))])
        );

        // T10 Task 5 review (F7): `claims_changed` was deleted (superseded
        // by `first_changed`, its only production caller since Task 5) —
        // this now pins the same "unchanged" behavior on `first_changed`.
        assert_eq!(
            first_changed(&hook, &targets, 4096, 1 << 20, &key(), Some(&a), Some(&b)).unwrap(),
            None
        );
    }

    /// T10 Task 5 review (F7): `first_changed` names the *namespace* of the
    /// first differing claim (not just "changed or not") — pin that it
    /// reports the real namespace, not e.g. an empty string or a panic, on
    /// a genuine change. `unique_diff_splits_a_rename_into_one_release_and_one_claim`
    /// above already proves `claim_delta` gets the direction right for the
    /// same rename; this is `first_changed`'s own twin of that.
    #[test]
    fn unique_diff_first_changed_names_the_namespace_on_a_real_change() {
        let (hook, targets) = fixture();
        let a = User {
            username: Some("old".into()),
            avatar: 1,
        };
        let b = User {
            username: Some("new".into()),
            avatar: 1,
        };
        let ns = hook.usernames.name();
        assert_eq!(
            first_changed(&hook, &targets, 4096, 1 << 20, &key(), Some(&a), Some(&b)).unwrap(),
            Some(ns)
        );
        // An insert (∅ -> {claim}) and a delete ({claim} -> ∅) are changes too.
        assert_eq!(
            first_changed(&hook, &targets, 4096, 1 << 20, &key(), None, Some(&a)).unwrap(),
            Some(ns)
        );
        assert_eq!(
            first_changed(&hook, &targets, 4096, 1 << 20, &key(), Some(&a), None).unwrap(),
            Some(ns)
        );
    }

    #[test]
    fn unique_diff_splits_a_rename_into_one_release_and_one_claim() {
        let (hook, targets) = fixture();
        let a = User {
            username: Some("old".into()),
            avatar: 1,
        };
        let b = User {
            username: Some("new".into()),
            avatar: 1,
        };
        let d = claim_delta(&hook, &targets, 4096, 1 << 20, &key(), Some(&a), Some(&b)).unwrap();

        // Not just the counts: pin *which* key is released and *which* is
        // claimed. `releases`/`claims` swapped would still leave both
        // lengths at 1 — and Task 7 applies releases before claims
        // specifically so a key can move from one owner to another inside
        // one transaction, so a swap here would silently invert that.
        let ns = hook.usernames.name();
        assert_eq!(d.releases.len(), 1);
        assert_eq!(d.releases[0].namespace, ns);
        assert_eq!(d.releases[0].key, username_key_bytes("old"));

        assert_eq!(d.claims.len(), 1);
        assert_eq!(d.claims[0].namespace, ns);
        assert_eq!(d.claims[0].key, username_key_bytes("new"));
    }

    #[test]
    fn unique_diff_treats_insert_and_delete_as_claim_set_changes() {
        let (hook, targets) = fixture();
        let v = User {
            username: Some("rail".into()),
            avatar: 1,
        };
        let ns = hook.usernames.name();
        let rail = username_key_bytes("rail");

        // T10 Task 5 review (F7): `claims_changed` deleted — see this
        // file's module doc — so these now pin `first_changed` instead.
        assert_eq!(
            first_changed(&hook, &targets, 4096, 1 << 20, &key(), None, Some(&v)).unwrap(),
            Some(ns)
        );
        // An insert must be a pure claim — no release — not just "changed".
        let insert = claim_delta(&hook, &targets, 4096, 1 << 20, &key(), None, Some(&v)).unwrap();
        assert_eq!(insert.releases.len(), 0);
        assert_eq!(insert.claims.len(), 1);
        assert_eq!(insert.claims[0].namespace, ns);
        assert_eq!(insert.claims[0].key, rail);

        assert_eq!(
            first_changed(&hook, &targets, 4096, 1 << 20, &key(), Some(&v), None).unwrap(),
            Some(ns)
        );
        // A delete must be a pure release — no claim.
        let delete = claim_delta(&hook, &targets, 4096, 1 << 20, &key(), Some(&v), None).unwrap();
        assert_eq!(delete.releases.len(), 1);
        assert_eq!(delete.claims.len(), 0);
        assert_eq!(delete.releases[0].namespace, ns);
        assert_eq!(delete.releases[0].key, rail);
    }
}