armdb 0.13.0

//! T10 Task 6: claim intents reach the commit plan.
//! T10 Task 7: the resolver — a claim actually wins or loses its CAS race.
//!
//! Task 6's tests are unit-scoped: they exercise the seam Task 6 builds
//! (`TxParticipant::tx_claims`, via the `armdb::testing::tx_claims_of` escape
//! hatch, since `ClaimOp`/`TxParticipant` are crate-private; and
//! `tx_preplan_projection_targets`, via `armdb::testing::tx_preplan_names`)
//! without needing `resolve_claims` to exist yet.
//!
//! Fixture mirrors `unique_nontx.rs`'s `typed_map` module (`User`/
//! `UsernameKey`/`Usernames`) — same shape, so a reader who already knows
//! that file's fixture recognizes this one immediately. `fixture()` returns
//! its `TempDir` (unlike Task 6's original version, which dropped it: Task
//! 6's tests never touched storage past `Db`/collection open, but Task 7's
//! do real `db.tx()...commit()` writes, which need the directory to outlive
//! the fixture call — the same reason `unique_nontx.rs`'s fixtures return it).

#![cfg(all(
    feature = "armour",
    feature = "schema",
    feature = "typed-tree",
    feature = "rapira-codec"
))]

use std::sync::Arc;

use armdb::armour::Db;
use armdb::{Config, FixedConfig, NoHook, Projected, RapiraCodec, UniqueIndex, UniqueTargets};
use armour_core::{Fuid, GetType};
use rapira::Rapira;

armour_core::const_hasher_or!(user_id, "USR", "dev-key-user_id", brand = "user");
type UserId = Fuid<user_id::Hasher>;

fn id(n: u64) -> UserId {
    Fuid::from_be_bytes(n.to_be_bytes())
}

fn encode(user: &User) -> Vec<u8> {
    rapira::serialize(user)
}

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

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

armdb::unique_index!(Usernames, key = UsernameKey, name = "unique_tx_usernames");

fn open_usernames(db: &Arc<Db>) -> Arc<UniqueIndex<Usernames>> {
    db.open_unique_index::<Usernames>(FixedConfig::test())
        .expect("open usernames namespace")
}

/// `avatar` is along for the ride — mirrors `unique_nontx.rs`'s `User`: a
/// value that varies while `username` stays fixed proves a helper reports
/// *claim* changes, not whole-row changes.
#[derive(Clone, Debug, PartialEq, GetType, Rapira)]
struct User {
    username: Option<String>,
    avatar: u32,
}
impl armdb::CollectionMeta for User {
    type SelfId = UserId;
    const NAME: &'static str = "unique_tx_users";
}

/// Claims `value.username` (when set) into `usernames`.
type UserHook = Projected<UserId, User, NoHook>;

fn claim_hook_user(usernames: &Arc<UniqueIndex<Usernames>>) -> UserHook {
    let targets = UniqueTargets::new().with(usernames);
    let ns = Arc::clone(usernames);
    Projected::<UserId, User>::new(armdb::IndexTargets::new(), |_, _, _| Ok(())).and_unique(
        targets,
        move |_k, v: &User, sink| {
            if let Some(u) = &v.username {
                sink.claim(&ns, UsernameKey::new(u))?;
            }
            Ok(())
        },
    )
}

type Coll = armdb::typed_map!(User, RapiraCodec, UserHook);

fn fixture() -> (
    tempfile::TempDir,
    Arc<Db>,
    Arc<UniqueIndex<Usernames>>,
    Arc<Coll>,
) {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let usernames = open_usernames(&db);
    let hook = claim_hook_user(&usernames);
    let users = db
        .open_typed_map::<User, RapiraCodec, _>(Config::test(), hook, &[])
        .unwrap();
    (dir, db, usernames, users)
}

/// Commit a single `put` as its own transaction — the shorthand the T10
/// Task 7 matrix uses to set up a claim before the test's own transaction.
fn tx_put(db: &Arc<Db>, users: &Arc<Coll>, key: UserId, username: Option<&str>) {
    db.tx()
        .put(
            users,
            &key,
            &User {
                username: username.map(str::to_string),
                avatar: 1,
            },
        )
        .unwrap()
        .commit()
        .unwrap();
}

armour_core::const_hasher_or!(channel_id, "CHN", "dev-key-channel_id", brand = "channel");
type ChannelId = Fuid<channel_id::Hasher>;

fn cid(n: u64) -> ChannelId {
    Fuid::from_be_bytes(n.to_be_bytes())
}

/// A second source claiming into the *same* `usernames` namespace as `User`
/// — T10 Task 7's "two sources claiming one key" scenario needs two
/// independent collections sharing one namespace, which `fixture()`'s single
/// `users` collection cannot exercise on its own.
#[derive(Clone, Debug, PartialEq, GetType, Rapira)]
struct Channel {
    username: Option<String>,
}
impl armdb::CollectionMeta for Channel {
    type SelfId = ChannelId;
    const NAME: &'static str = "unique_tx_channels";
}

type ChannelHook = Projected<ChannelId, Channel, NoHook>;

fn claim_hook_channel(usernames: &Arc<UniqueIndex<Usernames>>) -> ChannelHook {
    let targets = UniqueTargets::new().with(usernames);
    let ns = Arc::clone(usernames);
    Projected::<ChannelId, Channel>::new(armdb::IndexTargets::new(), |_, _, _| Ok(())).and_unique(
        targets,
        move |_k, v: &Channel, sink| {
            if let Some(u) = &v.username {
                sink.claim(&ns, UsernameKey::new(u))?;
            }
            Ok(())
        },
    )
}

type ChannelColl = armdb::typed_map!(Channel, RapiraCodec, ChannelHook);

fn shared_namespace_fixture() -> (
    tempfile::TempDir,
    Arc<Db>,
    Arc<UniqueIndex<Usernames>>,
    Arc<Coll>,
    Arc<ChannelColl>,
) {
    let dir = tempfile::tempdir().unwrap();
    let db = Arc::new(Db::open_test(dir.path()).unwrap());
    let usernames = open_usernames(&db);
    let users = db
        .open_typed_map::<User, RapiraCodec, _>(Config::test(), claim_hook_user(&usernames), &[])
        .unwrap();
    let channels = db
        .open_typed_map::<Channel, RapiraCodec, _>(
            Config::test(),
            claim_hook_channel(&usernames),
            &[],
        )
        .unwrap();
    (dir, db, usernames, users, channels)
}

// ===========================================================================
// T10 Task 7: the resolver — a claim only ever wins the CAS race it should.
// ===========================================================================

#[test]
fn unique_tx_writes_the_owner_row() {
    let (_dir, db, usernames, users) = fixture();
    db.tx()
        .put(
            &users,
            &id(1),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .commit()
        .unwrap();

    let owner = usernames.owner_of(&UsernameKey::new("rail")).unwrap();
    assert_eq!(owner.belongs_to::<User>(), Some(id(1)));
    assert!(!owner.is_reservation());
}

#[test]
fn unique_tx_refuses_a_key_held_by_another_owner() {
    let (_dir, db, usernames, users) = fixture();
    tx_put(&db, &users, id(1), Some("rail"));

    let err = db
        .tx()
        .put(
            &users,
            &id(2),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .commit()
        .unwrap_err();
    assert!(
        matches!(
            err,
            armdb::DbError::UniqueViolation {
                namespace: "unique_tx_usernames",
                ..
            }
        ),
        "{err}"
    );
    // The whole transaction is refused: the source row never landed.
    assert!(users.get(&id(2)).is_none());
    assert_eq!(
        usernames
            .owner_of(&UsernameKey::new("rail"))
            .unwrap()
            .belongs_to::<User>(),
        Some(id(1))
    );
}

#[test]
fn unique_tx_two_sources_claiming_one_key_conflict() {
    let (_dir, db, usernames, users, channels) = shared_namespace_fixture();
    let err = db
        .tx()
        .put(
            &users,
            &id(1),
            &User {
                username: Some("dup".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .put(
            &channels,
            &cid(1),
            &Channel {
                username: Some("dup".into()),
            },
        )
        .unwrap()
        .commit()
        .unwrap_err();
    assert!(
        matches!(err, armdb::DbError::UniqueViolation { .. }),
        "{err}"
    );

    // The whole transaction is refused: neither source row landed, and
    // exactly one of the two would-be owners could ever have won it.
    assert!(users.get(&id(1)).is_none());
    assert!(channels.get(&cid(1)).is_none());
    assert_eq!(usernames.owner_of(&UsernameKey::new("dup")), None);
}

#[test]
fn unique_tx_hands_a_key_from_one_owner_to_another_in_one_transaction() {
    let (_dir, db, usernames, users) = fixture();
    tx_put(&db, &users, id(1), Some("rail"));

    db.tx()
        .put(
            &users,
            &id(1),
            &User {
                username: None,
                avatar: 1,
            },
        ) // release
        .unwrap()
        .put(
            &users,
            &id(2),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        ) // claim
        .unwrap()
        .commit()
        .unwrap();

    assert_eq!(
        usernames
            .owner_of(&UsernameKey::new("rail"))
            .unwrap()
            .belongs_to::<User>(),
        Some(id(2))
    );
}

/// Review round I3: the same hand-off as the test above, but with the two
/// `put`s in the **opposite** order — claim `id(2)` first, release `id(1)`
/// second. `project_rank0` collapses each source *row* independently (Task
/// 6), so this emits `[Claim("rail", id2), Release("rail", id1)]` through
/// the real stack — the exact legal-hand-off shape `PendingClaim`'s own doc
/// calls out, and the only list order that fails if `resolve_claims` ever
/// resolved in list order instead of partitioning by kind. The test above
/// alone cannot catch that regression: writing the release first already
/// emits `[Release, Claim]`, which a naive list-order resolver would also
/// get right by accident.
#[test]
fn unique_tx_hands_a_key_over_with_the_claim_listed_before_the_release() {
    let (_dir, db, usernames, users) = fixture();
    tx_put(&db, &users, id(1), Some("rail"));

    db.tx()
        .put(
            &users,
            &id(2),
            &User {
                username: Some("rail".into()),
                avatar: 1,
            },
        ) // claim, listed first
        .unwrap()
        .put(
            &users,
            &id(1),
            &User {
                username: None,
                avatar: 1,
            },
        ) // release, listed second
        .unwrap()
        .commit()
        .unwrap();

    assert_eq!(
        usernames
            .owner_of(&UsernameKey::new("rail"))
            .unwrap()
            .belongs_to::<User>(),
        Some(id(2))
    );
}

/// A regression check, not a release/claim exercise — despite `commit()`ing
/// a `put` with the same `username` and a different `avatar`, `claim_delta`
/// emits neither a release nor a claim for a key present in both the old and
/// new snapshot, so `resolve_claims` never even sees this row (renamed from
/// `unique_tx_releases_and_reclaims_the_same_key_for_the_same_owner` in
/// review round M2 — that name implied end-to-end release/claim coverage
/// this test cannot provide; see `unique_tx_rename_via_put_releases_the_old_
/// key_and_lets_it_be_reclaimed` and `unique_tx_delete_releases_the_claim`
/// below, added in the same round (I1), for tests that actually commit a
/// `Delete` decision to a namespace row).
#[test]
fn unique_tx_an_unrelated_field_update_keeps_the_claim() {
    let (_dir, db, usernames, users) = fixture();
    tx_put(&db, &users, id(1), Some("rail"));

    db.tx()
        .put(
            &users,
            &id(1),
            &User {
                username: Some("rail".into()),
                avatar: 7,
            },
        )
        .unwrap()
        .commit()
        .unwrap();

    assert_eq!(
        usernames
            .owner_of(&UsernameKey::new("rail"))
            .unwrap()
            .belongs_to::<User>(),
        Some(id(1))
    );
}

// ===========================================================================
// Review round I1: `ClaimDecision::Delete` must actually reach storage.
//
// Every test above either collapses a release's `None` write back to `Some`
// through `set_write`'s in-place merge (the hand-off tests: `id(1)`'s
// release and `id(2)`'s claim are two different *rows*, so they don't
// collapse against each other, but nothing before this point ever left a
// bare, uncollapsed `Delete` sitting in `writes` and then actually checked
// the namespace row was gone from storage afterward) or never produced a
// release at all. So: a rename via `put` (different keys, one row, one
// `Release` + one `Claim` — not collapsed, since `project_rank0` collapses
// by *row*, not by namespace key) and a `delete` (`Some -> None`, a
// different `Action` from `put` entirely, so a different `claim_delta`
// input) each get their own test, each asserting the *old* key is genuinely
// free again — not merely absent from one reader, but re-claimable by a
// third row.
// ===========================================================================

#[test]
fn unique_tx_rename_via_put_releases_the_old_key_and_lets_it_be_reclaimed() {
    let (_dir, db, usernames, users) = fixture();
    tx_put(&db, &users, id(1), Some("rail"));

    db.tx()
        .put(
            &users,
            &id(1),
            &User {
                username: Some("new".into()),
                avatar: 1,
            },
        )
        .unwrap()
        .commit()
        .unwrap();

    assert_eq!(usernames.owner_of(&UsernameKey::new("rail")), None);
    assert_eq!(
        usernames
            .owner_of(&UsernameKey::new("new"))
            .unwrap()
            .belongs_to::<User>(),
        Some(id(1))
    );

    // The old key must be genuinely free, not merely unread: a third row can
    // claim it.
    tx_put(&db, &users, id(2), Some("rail"));
    assert_eq!(
        usernames
            .owner_of(&UsernameKey::new("rail"))
            .unwrap()
            .belongs_to::<User>(),
        Some(id(2))
    );
}

#[test]
fn unique_tx_delete_releases_the_claim() {
    let (_dir, db, usernames, users) = fixture();
    tx_put(&db, &users, id(1), Some("rail"));

    db.tx().delete(&users, &id(1)).unwrap().commit().unwrap();

    assert_eq!(usernames.owner_of(&UsernameKey::new("rail")), None);
    assert!(users.get(&id(1)).is_none());

    tx_put(&db, &users, id(2), Some("rail"));
    assert_eq!(
        usernames
            .owner_of(&UsernameKey::new("rail"))
            .unwrap()
            .belongs_to::<User>(),
        Some(id(2))
    );
}

#[test]
fn unique_tx_claims_carry_the_source_as_owner() {
    let (_dir, _db, _usernames, users) = fixture();
    let old = encode(&User {
        username: Some("old".into()),
        avatar: 1,
    });
    let new = encode(&User {
        username: Some("new".into()),
        avatar: 1,
    });

    let ops = armdb::testing::tx_claims_of(&users, &id(1), Some(&old), Some(&new)).unwrap();

    // A rename is one release and one claim, both owned by (users, id(1)).
    assert_eq!(ops.len(), 2);
    assert!(
        ops.iter()
            .all(|o| o.owner.belongs_to::<User>() == Some(id(1)))
    );
    assert_eq!(ops.iter().filter(|o| o.is_release()).count(), 1);
    assert_eq!(ops.iter().filter(|o| o.is_claim()).count(), 1);
}

#[test]
fn unique_tx_preplans_every_declared_namespace() {
    let (_dir, _db, usernames, users) = fixture();
    // The plan cannot know whether a write claims before it runs the
    // projection, so a declared namespace is locked either way.
    let preplanned = armdb::testing::tx_preplan_names(&users).unwrap();
    assert!(preplanned.contains(&usernames.name()));
}