vta-service 0.35.0

//! A member's MLS group for each room they belong to.
//!
//! The custody half of the room oracle. [`crate::operations::room_oracle`] mints
//! presentations from credentials; this holds the *keys*, which is what
//! `rooms/keys/open` needs and what nothing before this stored.
//!
//! # Three steps, and the third is the one people forget
//!
//! A group arrives once and then has to be kept current:
//!
//! 1. [`mint_key_package`] — the joining side produces something the owner can add.
//! 2. [`join`] — the owner's Welcome arrives and the group exists.
//! 3. [`apply_commit`] — every membership change after that. **This is not optional.** A
//!    member who misses one is stuck at their last epoch and can open nothing sealed after
//!    it, and the symptom is "this record does not open", which reads like corruption
//!    rather than a missed message.
//!
//! # What authorizes each
//!
//! Deliberately three different answers, because they are three different acts:
//!
//! - **Minting** is gated on holding an invitation, because minting retains a private key
//!   against a Welcome that may never come. A key-holder that minted for any room on any
//!   request is one anyone can fill.
//! - **Joining** is gated on that same invitation, *consumed*. This is where the design's
//!   "joining is consent" stops being ceremonial: a Welcome carries a group's secrets, and
//!   a recipient that accepts an uninvited one has made the invitation decorative.
//! - **Committing** is gated *inside the group* — MLS authenticates the committer as a
//!   member of the group we already hold. Never from an ACL of our own, which is the same
//!   rule the rest of the room family follows.

use base64::Engine as _;
use base64::engine::general_purpose::URL_SAFE_NO_PAD as B64;
use vti_common::error::AppError;
use vti_common::store::KeyspaceHandle;
use vti_rooms::mls::{GroupSnapshot, IdentitySnapshot, RoomGroup};
use vti_rooms::sealed::SealedRoom;
use vti_rooms::wire::EpochLink;

/// Prefix for a room's group state, and the scan `list_rooms` walks.
const GROUP_PREFIX: &str = "room-group:";

/// Storage key for a room's group state.
fn group_key(room_id: &str) -> String {
    format!("{GROUP_PREFIX}{room_id}")
}

/// Storage key for a consumed invitation.
pub(super) fn invitation_key(credential_id: &str) -> String {
    format!("room-vic:{credential_id}")
}

/// A minted key package, awaiting the Welcome that consumes it.
///
/// The private half lives in the snapshot; this row *is* the retained key material, which is
/// why it is bounded rather than kept forever.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PendingKeyPackage {
    /// The identity holding the package's private half.
    pub snapshot: IdentitySnapshot,
    /// The KeyPackage itself, base64url — public, and what travels to the owner.
    pub key_package: String,
    /// Unix seconds after which this is discarded unused.
    pub expires_at: u64,
}

/// What this VTA holds for one room.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RoomGroupRecord {
    /// The group, at whatever epoch it last reached.
    pub snapshot: GroupSnapshot,
    /// The member this group is for.
    pub member_did: String,
    /// The room's epoch key chain, as far as this VTA holds it.
    ///
    /// Retained here rather than fetched per open because this is the custody point: a link
    /// is key material, wrapped, and the VTA is where the room's key material lives. Every
    /// commit this VTA applies appends one, so a member who has kept up holds the chain back
    /// to the epoch they joined at without asking anyone.
    ///
    /// **Empty for a room joined before the chain existed**, and for one whose owner has not
    /// sent the history. `open` then reads only from the joining epoch forward and says so
    /// — see `RoomKeyError::EpochUnreachable`.
    #[serde(default, skip_serializing_if = "Vec::is_empty")]
    pub links: Vec<EpochLink>,
    /// Unix seconds of the last change.
    pub updated_at: u64,
}

/// Mint an MLS KeyPackage for `room_id` and retain its private half.
///
/// **Per room, never reused across rooms.** A KeyPackage is a stable public identifier, so
/// the same one offered to two rooms tells anyone who sees both that one party is in both —
/// the correlation a `private` room exists to deny, arriving through the door rather than
/// the wall.
pub async fn mint_key_package(
    groups: &KeyspaceHandle,
    room_id: &str,
    member_did: &str,
    lifetime_secs: u64,
    now: u64,
) -> Result<PendingKeyPackage, AppError> {
    let (snapshot, package) = IdentitySnapshot::mint(member_did)
        .map_err(|e| AppError::Internal(format!("mint a key package: {e}")))?;

    let pending = PendingKeyPackage {
        snapshot,
        key_package: B64.encode(&package),
        expires_at: now + lifetime_secs,
    };
    groups
        .insert(pending_key(room_id), &pending)
        .await
        .map_err(|e| AppError::Internal(format!("store the pending key package: {e}")))?;
    Ok(pending)
}

/// Storage key for a pending key package.
fn pending_key(room_id: &str) -> String {
    format!("room-kp:{room_id}")
}

/// Join `room_id`'s group from a Welcome.
///
/// Refuses a second join rather than merging: two group states for one room is a condition
/// nothing downstream can resolve — [`open`] has no way to choose, and choosing wrong
/// returns "did not open" for a record the member can plainly see.
pub async fn join(
    groups: &KeyspaceHandle,
    room_id: &str,
    member_did: &str,
    welcome: &[u8],
    now: u64,
) -> Result<u64, AppError> {
    if load(groups, room_id).await?.is_some() {
        return Err(AppError::Conflict(format!(
            "this VTA already holds group state for room `{room_id}`; leaving and rejoining \
             is a removal and a fresh invitation, not a second welcome"
        )));
    }

    // The KeyPackage the owner added is the one we minted, so the private half is in the
    // pending record — joining with a fresh identity would produce a group whose leaf
    // nobody added.
    let pending: Option<PendingKeyPackage> = groups
        .get(pending_key(room_id))
        .await
        .map_err(|e| AppError::Internal(format!("read the pending key package: {e}")))?;
    let pending = pending.ok_or_else(|| {
        AppError::Validation(format!(
            "no key package was minted for room `{room_id}`; a welcome can only be accepted \
             against the package the owner was given"
        ))
    })?;

    let group = RoomGroup::join_from_identity(&pending.snapshot, welcome)
        .map_err(|e| AppError::Validation(format!("the welcome did not process: {e}")))?;
    let epoch = group.epoch();

    store(groups, room_id, member_did, &group, Vec::new(), now).await?;
    // The package is consumed. MLS consumes it on add, and a retained private half for a
    // used package is key material kept for nothing.
    groups
        .remove(pending_key(room_id))
        .await
        .map_err(|e| AppError::Internal(format!("clear the pending key package: {e}")))?;
    Ok(epoch)
}

/// Apply a commit, advancing one epoch.
///
/// Ordering is the whole contract. A replay is a no-op success — a retry that failed would
/// make every unreliable transport a liveness problem — and a gap is refused with the epoch
/// we actually hold, so the sender resumes rather than guesses.
pub async fn apply_commit(
    groups: &KeyspaceHandle,
    room_id: &str,
    commit: &[u8],
    claimed_epoch: u64,
    now: u64,
) -> Result<u64, AppError> {
    let record = load(groups, room_id).await?.ok_or_else(|| {
        AppError::NotFound(format!(
            "this VTA holds no group state for room `{room_id}`"
        ))
    })?;
    // A `SealedRoom` rather than a bare group: the epoch link this commit produces is bound
    // to the room, so minting one needs the room's identifier and the group does not carry
    // it. Restoring through the room type is what makes the link mintable at all.
    let mut room = SealedRoom::new(
        room_id,
        RoomGroup::restore(&record.snapshot)
            .map_err(|e| AppError::Internal(format!("restore the group: {e}")))?,
    );
    let current = room.group().epoch();

    if claimed_epoch == current {
        // Already applied. Reporting success with the unchanged epoch is what makes
        // delivery retryable.
        return Ok(current);
    }
    if claimed_epoch != current + 1 {
        return Err(AppError::Conflict(format!(
            "commit produces epoch {claimed_epoch} but room `{room_id}` is at {current}; \
             resume from {}",
            current + 1
        )));
    }

    // The link is what keeps everything already in the room readable across this commit.
    // Dropping it here would advance the epoch and silently sever the history — the defect
    // the chain exists to fix, so it is retained in the same write that advances the group.
    let (_, link) = room
        .apply_commit(commit)
        .map_err(|e| AppError::Validation(format!("the commit did not process: {e}")))?;
    let epoch = room.group().epoch();

    let mut links = record.links;
    if let Some(link) = link {
        links.push(link);
    }
    store(
        groups,
        room_id,
        &record.member_did,
        room.group(),
        links,
        now,
    )
    .await?;
    Ok(epoch)
}

/// Seal a record body with the room's current epoch key.
///
/// The mirror of [`open_record`], and the reason a client can write to a sealed room at all:
/// the key never leaves, so the caller sends plaintext and receives ciphertext.
///
/// # It does not write
///
/// The caller takes the result to a host and presents its own authority there. Sealing and
/// being allowed to store are different questions asked of different parties — this VTA
/// knows the key and nothing about the room's ACL; the host knows the credentials and cannot
/// read a byte. Doing both here would make this VTA the party that decides what goes into a
/// room, which is the one thing the design keeps it out of.
pub async fn seal_record(
    groups: &KeyspaceHandle,
    room_id: &str,
    key: &str,
    version: u64,
    plaintext: &[u8],
) -> Result<vti_rooms::wire::SealedContent, AppError> {
    let record = load(groups, room_id).await?.ok_or_else(|| {
        AppError::NotFound(format!(
            "this VTA holds no group state for room `{room_id}`"
        ))
    })?;
    let group = RoomGroup::restore(&record.snapshot)
        .map_err(|e| AppError::Internal(format!("restore the group: {e}")))?;

    SealedRoom::new(room_id, group)
        .seal_record(key, version, plaintext)
        .map_err(|e| AppError::Validation(e.to_string()))
}

/// Every room this VTA holds group state for, with how far each one reads.
///
/// # Custody, not membership
///
/// This answers "what can I open", never "what am I a member of". A principal may hold a
/// perfectly good membership credential for a room whose Welcome never arrived — absent
/// here, correctly. And a VTA not yet told of a removal still holds keys for a room its
/// principal has left: it cannot write there, because the host checks credentials that no
/// longer verify, and it can still open what it already had. That is what removal has always
/// meant, and a caller MUST NOT read this list as authority to act.
pub async fn list_rooms(groups: &KeyspaceHandle) -> Result<Vec<HeldRoom>, AppError> {
    let pairs = groups.prefix_iter_raw(GROUP_PREFIX.to_string()).await?;
    let mut out = Vec::with_capacity(pairs.len());

    for (k, v) in pairs {
        let record: RoomGroupRecord = serde_json::from_slice(&v).map_err(|e| {
            let which = String::from_utf8_lossy(&k).to_string();
            AppError::Internal(format!("decode group state at `{which}`: {e}"))
        })?;
        let room_id = String::from_utf8_lossy(&k)
            .strip_prefix(GROUP_PREFIX)
            .unwrap_or_default()
            .to_string();

        // Restoring to answer is deliberate. The epoch could be read from the snapshot, but
        // `earliestReadableEpoch` is only knowable by *walking* the chain — and a number
        // derived two different ways is a number that will eventually disagree with itself.
        let group = RoomGroup::restore(&record.snapshot)
            .map_err(|e| AppError::Internal(format!("restore the group for `{room_id}`: {e}")))?;
        let mut room = SealedRoom::new(&room_id, group);
        room.add_links(record.links);

        let epoch = room.room_epoch();
        let earliest = room
            .earliest_readable_epoch()
            .map_err(|e| AppError::Internal(format!("walk the chain for `{room_id}`: {e}")))?;

        out.push(HeldRoom {
            room_id,
            epoch,
            earliest_readable_epoch: earliest,
        });
    }
    out.sort_by(|a, b| a.room_id.cmp(&b.room_id));
    Ok(out)
}

/// One room this VTA can open, as `rooms/keys/list` reports it.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct HeldRoom {
    pub room_id: String,
    pub epoch: u32,
    pub earliest_readable_epoch: u32,
}

/// Retain epoch links this VTA's principal fetched from the room's host.
///
/// The last leg of a joining member's backfill. A key holder accrues a rung for every
/// membership change it lives through — [`apply_commit`] keeps one from each — so what
/// arrives here is the history it did **not** live through.
///
/// Returns `(earliest_readable_epoch, stored)`.
///
/// # Why the answer is not "how many arrived"
///
/// A rung extends reach only if every rung above it is present too, so counting what was
/// delivered says nothing about what can now be read. The number worth returning is the one
/// this VTA can only get by *walking* what it holds — which is why the specification puts it
/// in the response and why a host serving the same rungs could not have answered it.
///
/// # A rung already held is never replaced
///
/// Same rule the hosts follow. A second rung for an epoch is either a replay — which must be
/// a no-op rather than an error, or a retried delivery becomes a failure — or an attempt to
/// re-point this member's history at key material of somebody else's choosing. Keeping the
/// first is the only reading that is safe under both.
pub async fn store_links(
    groups: &KeyspaceHandle,
    room_id: &str,
    incoming: Vec<EpochLink>,
    now: u64,
) -> Result<(u32, usize), AppError> {
    let record = load(groups, room_id).await?.ok_or_else(|| {
        AppError::NotFound(format!(
            "this VTA holds no group state for room `{room_id}`"
        ))
    })?;

    let mut links = record.links;
    let mut stored = 0;
    for link in incoming {
        if !links.iter().any(|held| held.epoch == link.epoch) {
            links.push(link);
            stored += 1;
        }
    }
    links.sort_by_key(|l| l.epoch);

    let group = RoomGroup::restore(&record.snapshot)
        .map_err(|e| AppError::Internal(format!("restore the group: {e}")))?;
    let mut room = SealedRoom::new(room_id, group);
    room.add_links(links.clone());
    let earliest = room
        .earliest_readable_epoch()
        .map_err(|e| AppError::Internal(format!("walk the chain: {e}")))?;

    store(
        groups,
        room_id,
        &record.member_did,
        room.group(),
        links,
        now,
    )
    .await?;
    Ok((earliest, stored))
}

/// Open a sealed record with the room's group key.
///
/// The key never leaves. That is the whole design: the caller sends ciphertext and gets
/// plaintext, and an oracle that returned the key would be a key-release call wearing a
/// different name.
pub async fn open_record(
    groups: &KeyspaceHandle,
    room_id: &str,
    key: &str,
    version: u64,
    ciphertext: &str,
    nonce: &str,
    epoch: u32,
) -> Result<Vec<u8>, AppError> {
    let record = load(groups, room_id).await?.ok_or_else(|| {
        AppError::NotFound(format!(
            "this VTA holds no group state for room `{room_id}`"
        ))
    })?;
    let group = RoomGroup::restore(&record.snapshot)
        .map_err(|e| AppError::Internal(format!("restore the group: {e}")))?;

    let held = group.epoch() + 1;
    if u64::from(epoch) > held {
        // Saying which epoch we hold turns "it does not open" — which reads like corruption
        // — into "you are behind", which an operator can act on.
        return Err(AppError::Validation(format!(
            "record is sealed under epoch {epoch} and this VTA holds room `{room_id}` at \
             epoch {held}; a commit has not been delivered"
        )));
    }

    // A record older than the current epoch is opened with *its own* epoch's key, walked out
    // of the retained chain. Without the links this VTA reaches only the epoch it is at, and
    // the error says which — an unreachable epoch is history that was severed or never
    // delivered, not a record that failed to decrypt.
    let mut room = SealedRoom::new(room_id, group);
    room.add_links(record.links);

    room.open_record(
        key,
        version,
        &vti_rooms::wire::SealedContent {
            ciphertext: ciphertext.to_string(),
            nonce: nonce.to_string(),
            epoch,
        },
    )
    .map_err(|e| AppError::Validation(e.to_string()))
}

/// Record an invitation as consumed, refusing a second use.
///
/// Single use means single use, and this record is the only thing that remembers. It
/// deliberately outlives the group it admitted: a member who leaves a room discards the
/// group, and without this row the same invitation would let them be re-added with no fresh
/// consent from the owner.
pub async fn consume_invitation(
    invitations: &KeyspaceHandle,
    credential_id: &str,
    room_id: &str,
    now: u64,
) -> Result<(), AppError> {
    let key = invitation_key(credential_id);
    if invitations
        .get_raw(key.clone())
        .await
        .map_err(|e| AppError::Internal(format!("read the invitation record: {e}")))?
        .is_some()
    {
        return Err(AppError::Conflict(format!(
            "invitation `{credential_id}` has already been used"
        )));
    }
    invitations
        .insert(
            key,
            &serde_json::json!({ "roomId": room_id, "consumedAt": now }),
        )
        .await
        .map_err(|e| AppError::Internal(format!("record the invitation: {e}")))
}

/// The group this VTA holds for `room_id`, if any.
pub async fn load(
    groups: &KeyspaceHandle,
    room_id: &str,
) -> Result<Option<RoomGroupRecord>, AppError> {
    groups
        .get(group_key(room_id))
        .await
        .map_err(|e| AppError::Internal(format!("read group state for `{room_id}`: {e}")))
}

async fn store(
    groups: &KeyspaceHandle,
    room_id: &str,
    member_did: &str,
    group: &RoomGroup,
    links: Vec<EpochLink>,
    now: u64,
) -> Result<(), AppError> {
    let record = RoomGroupRecord {
        snapshot: group
            .snapshot()
            .map_err(|e| AppError::Internal(format!("snapshot the group: {e}")))?,
        member_did: member_did.to_string(),
        links,
        updated_at: now,
    };
    groups
        .insert(group_key(room_id), &record)
        .await
        .map_err(|e| AppError::Internal(format!("store group state for `{room_id}`: {e}")))
}

/// Prefix for what a host has told this agent about a room's tree.
const ROOTS_PREFIX: &str = "room-roots:";

/// Storage key for a room's observed roots.
fn roots_key(room_id: &str) -> String {
    format!("{ROOTS_PREFIX}{room_id}")
}

/// How many `(headVersion, root)` observations to keep per room.
///
/// One is weaker than it looks in a specific way, and stronger than it looks in
/// another. It *does* catch a host that alternates at one version — `(V, A)`
/// then `(V, B)` differ, and differ again on the way back. What it misses is a
/// head that **advances and then goes backwards**: `(V, A)`, `(V+1, X)`,
/// `(V, B)` leaves the agent holding `V+1` with nothing to compare `V` against.
/// That is the mirror case and the rollback case, which are the two this is for.
///
/// Sixteen versions is more history than a member reads across in a session, and
/// a few hundred bytes.
const KEEP_ROOTS: usize = 16;

/// What a host has said about a room's tree, over time.
///
/// **Keyed by room, not by host**, and that is the decision worth not undoing. A
/// room's tree at version `V` is a fact about the *room*; who served it is not
/// part of that fact. Keying by `(room, host)` would file two hosts of one room
/// in two drawers and never compare them — and a room may deliberately have
/// several, a mirror serving reads while its primary takes writes.
///
/// Keyed this way, a member who reads the same room from two of its hosts gets
/// the comparison for free: two hosts reporting **different roots at one
/// `headVersion`** is exactly as damning as one host disagreeing with itself.
/// That is a comparison the specification's own list does not include, and it
/// needs no gossip channel, no anchor and no second member.
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RootHistory {
    /// Observations, most recent `headVersion` last. At most [`KEEP_ROOTS`].
    pub seen: Vec<RootObservation>,
    /// The highest `headVersion` ever seen for this room.
    ///
    /// Separate from `seen` because it answers a question no single pair can:
    /// **has this room gone backwards?** A pruned observation still leaves this
    /// behind.
    pub highest: u64,
    /// The version at which this room was caught, if it ever was.
    ///
    /// **Remembered rather than recomputed.** A conflict is a fact about the
    /// past, and the pair that revealed it is prunable — an agent that inferred
    /// "caught" from what it still holds would exonerate a host by doing enough
    /// reading. Once set it is never cleared here: it is cleared by
    /// [`forget`], with the membership.
    #[serde(default, skip_serializing_if = "Option::is_none")]
    pub caught_at: Option<u64>,
}

/// One thing a host said about a room's tree.
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RootObservation {
    pub head_version: u64,
    /// The root, as the wire spells it — a `DigestMultibase`.
    pub root: String,
}

/// What comparing a root against what is held came to.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RootVerdict {
    /// Seen at this head before, same root.
    Agree,
    /// Seen at this head before, **different root**. A host caught: there is no
    /// write to attribute the difference to, because a write would have moved
    /// the head.
    Conflict,
    /// First observation at this head. Not evidence of anything — a memory of
    /// one is not a comparison.
    NoneHeld,
}

/// Record what a host asserted, and say what it came to.
///
/// The write happens either way, including on a conflict: an agent that stopped
/// recording once it caught a host would forget the evidence at the moment it
/// acquired it.
pub async fn observe_head(
    groups: &KeyspaceHandle,
    room_id: &str,
    head_version: u64,
    root: &str,
) -> Result<RootVerdict, AppError> {
    let key = roots_key(room_id);
    let mut history: RootHistory = groups
        .get(key.clone())
        .await
        .map_err(|e| AppError::Internal(format!("read the root history of `{room_id}`: {e}")))?
        .unwrap_or_default();

    let verdict = match history.seen.iter().find(|o| o.head_version == head_version) {
        Some(prior) if prior.root == root => RootVerdict::Agree,
        Some(_) => RootVerdict::Conflict,
        None => RootVerdict::NoneHeld,
    };

    if matches!(verdict, RootVerdict::Conflict) {
        // Earliest wins: the first version at which this host was caught is the
        // one worth naming, and later conflicts do not make the first less true.
        history.caught_at = Some(
            history
                .caught_at
                .map_or(head_version, |v| v.min(head_version)),
        );
    }
    if matches!(verdict, RootVerdict::NoneHeld) {
        history.seen.push(RootObservation {
            head_version,
            root: root.to_string(),
        });
        history.seen.sort_by_key(|o| o.head_version);
        // Drop the OLDEST versions, not the oldest writes: what a member is
        // likely to read again is what the room is at now.
        if history.seen.len() > KEEP_ROOTS {
            let excess = history.seen.len() - KEEP_ROOTS;
            history.seen.drain(0..excess);
        }
    }
    history.highest = history.highest.max(head_version);

    groups
        .insert(key, &history)
        .await
        .map_err(|e| AppError::Internal(format!("record the root history of `{room_id}`: {e}")))?;
    Ok(verdict)
}

/// Everything this agent has been told about a room's tree, for a caller that
/// wants to compare out of band.
pub async fn root_history(groups: &KeyspaceHandle, room_id: &str) -> Result<RootHistory, AppError> {
    groups
        .get(roots_key(room_id))
        .await
        .map_err(|e| AppError::Internal(format!("read the root history of `{room_id}`: {e}")))
        .map(Option::unwrap_or_default)
}

/// Discard everything this VTA holds for a room.
///
/// Called on removal from the group. A key-holder that kept its state would retain the
/// ability to open everything sealed up to the epoch it was removed at — which is exactly
/// what the removal was for.
pub async fn forget(groups: &KeyspaceHandle, room_id: &str) -> Result<(), AppError> {
    groups
        .remove(group_key(room_id))
        .await
        .map_err(|e| AppError::Internal(format!("discard group state for `{room_id}`: {e}")))?;
    groups
        .remove(pending_key(room_id))
        .await
        .map_err(|e| AppError::Internal(format!("discard a pending key package: {e}")))?;
    // The root history goes with the membership. It carries no record content,
    // but it is a record of when this member read this room, and an agent that
    // kept it afterwards would be keeping a diary of a room its principal can no
    // longer open.
    groups
        .remove(roots_key(room_id))
        .await
        .map_err(|e| AppError::Internal(format!("discard the root history of `{room_id}`: {e}")))
}

/// The MLS epoch authenticator for a room this agent holds.
///
/// Every member of the group derives this independently and **no host can
/// compute it**, which is the whole reason it is worth anchoring: a member whose
/// own authenticator differs from the anchored one is in a *forked group*.
pub async fn epoch_authenticator(
    groups: &KeyspaceHandle,
    room_id: &str,
) -> Result<(u64, Vec<u8>), AppError> {
    let record = load(groups, room_id).await?.ok_or_else(|| {
        AppError::Validation(format!(
            "this agent holds no group state for `{room_id}`, so it cannot say what epoch the \
             room is at"
        ))
    })?;
    let group = RoomGroup::restore(&record.snapshot)
        .map_err(|e| AppError::Internal(format!("restore the group for `{room_id}`: {e}")))?;
    // The epoch travels with it: an authenticator names the epoch it was
    // derived at, and one published beside a different number is a statement
    // about a group nobody is in.
    Ok((group.epoch(), group.epoch_authenticator()))
}

#[cfg(test)]
mod root_memory_tests {
    use super::*;
    use vti_common::config::StoreConfig;
    use vti_common::store::Store;

    const ROOM: &str = "did:webvh:example.com:rooms:northwind";
    const A: &str = "zQmbWqxBEKC3P8tqsKc98xmWNzrzDtRLMiMPL8wBuTGsMnR";
    const B: &str = "zQmXo1sV5aJ7bT2kQdF9wRnPzYcH4uMgLtEjV6NrBqWsDpK";

    async fn open() -> (tempfile::TempDir, KeyspaceHandle) {
        let dir = tempfile::tempdir().unwrap();
        let store = Store::open(&StoreConfig {
            data_dir: dir.path().to_path_buf(),
        })
        .unwrap();
        let ks = store.keyspace(crate::keyspaces::ROOM_GROUPS).unwrap();
        (dir, ks)
    }

    /// A memory of one is not a comparison, and the first read must say so.
    #[tokio::test]
    async fn a_first_reading_holds_nothing_to_compare() {
        let (_d, ks) = open().await;
        assert_eq!(
            observe_head(&ks, ROOM, 412, A).await.unwrap(),
            RootVerdict::NoneHeld
        );
    }

    /// The whole mechanism: same state, two roots, no write to blame.
    #[tokio::test]
    async fn two_roots_at_one_version_is_a_host_caught() {
        let (_d, ks) = open().await;
        observe_head(&ks, ROOM, 412, A).await.unwrap();
        assert_eq!(
            observe_head(&ks, ROOM, 412, A).await.unwrap(),
            RootVerdict::Agree
        );
        assert_eq!(
            observe_head(&ks, ROOM, 412, B).await.unwrap(),
            RootVerdict::Conflict
        );
    }

    /// A room that moved is not a host that lied, and the version is what tells
    /// them apart. Without it every second read would look like equivocation.
    #[tokio::test]
    async fn a_different_version_is_a_different_moment() {
        let (_d, ks) = open().await;
        observe_head(&ks, ROOM, 412, A).await.unwrap();
        assert_eq!(
            observe_head(&ks, ROOM, 413, B).await.unwrap(),
            RootVerdict::NoneHeld,
            "a write moved the head, so a different root explains itself"
        );
    }

    /// One slot would miss this, which is why the memory is a map.
    ///
    /// The head advances and then a host serves the older version again — the
    /// mirror case and the rollback case. An agent keeping only the latest pair
    /// is holding `V+1` and has nothing to compare `V` against.
    #[tokio::test]
    async fn a_head_that_goes_backwards_is_still_compared() {
        let (_d, ks) = open().await;
        observe_head(&ks, ROOM, 412, A).await.unwrap();
        observe_head(&ks, ROOM, 500, B).await.unwrap();
        assert_eq!(
            observe_head(&ks, ROOM, 412, B).await.unwrap(),
            RootVerdict::Conflict,
            "the older version is still held, and its root still disagrees"
        );
    }

    /// The bound drops the OLDEST versions and keeps the highest-ever anyway,
    /// because "has this room gone backwards" is a question no single pair can
    /// answer.
    #[tokio::test]
    async fn the_history_is_bounded_and_the_high_water_mark_survives_it() {
        let (_d, ks) = open().await;
        for v in 1..=(KEEP_ROOTS as u64 + 5) {
            observe_head(&ks, ROOM, v, A).await.unwrap();
        }
        let history = root_history(&ks, ROOM).await.unwrap();
        assert_eq!(history.seen.len(), KEEP_ROOTS);
        assert_eq!(history.highest, KEEP_ROOTS as u64 + 5);
        assert_eq!(
            history.seen.first().unwrap().head_version,
            6,
            "the oldest versions are what gets dropped"
        );
    }

    /// Bounded by membership. An agent that kept this after a removal would keep
    /// a record of when its principal read a room it can no longer open.
    #[tokio::test]
    async fn forgetting_a_room_forgets_what_its_host_said() {
        let (_d, ks) = open().await;
        observe_head(&ks, ROOM, 412, A).await.unwrap();
        forget(&ks, ROOM).await.unwrap();
        assert_eq!(
            observe_head(&ks, ROOM, 412, B).await.unwrap(),
            RootVerdict::NoneHeld,
            "nothing survived the removal to compare against"
        );
    }
}