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, MoveOutcome};
use vti_rooms::error::RoomKeyError;
use vti_rooms::mls::{GroupSnapshot, IdentitySnapshot, RoomGroup};
use vti_rooms::sealed::SealedRoom;
use vti_rooms::wire::EpochLink;
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Member {
room_id: String,
persona: String,
}
impl Member {
pub fn new(room_id: &str, persona: &str) -> Result<Self, AppError> {
for (what, v) in [("room id", room_id), ("member DID", persona)] {
if v.is_empty() || v.contains('/') {
return Err(AppError::Validation(format!(
"`{v}` is not a usable {what}: a DID is non-empty and has no `/`"
)));
}
}
Ok(Self {
room_id: room_id.to_string(),
persona: persona.to_string(),
})
}
pub fn room_id(&self) -> &str {
&self.room_id
}
pub fn persona(&self) -> &str {
&self.persona
}
fn key(&self, kind: &str) -> String {
format!("{kind}/{}/{}", self.room_id, self.persona)
}
}
const GROUP: &str = "room-group";
const KEY_PACKAGE: &str = "room-kp";
const ADVANCE: &str = "room-advance";
const ROOTS: &str = "room-roots";
const REMOVED: &str = "room-removed";
const ADMITTED: &str = "room-admitted";
pub const ADMISSION_RETENTION_SECS: u64 = 24 * 60 * 60;
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RetainedAdmission {
pub member: String,
pub key_package_ref: String,
pub epoch: u32,
pub welcome: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub ratchet_tree: Option<String>,
pub applied_at: u64,
}
impl RetainedAdmission {
pub fn kept_at(&self, now: u64) -> bool {
now < self.applied_at.saturating_add(ADMISSION_RETENTION_SECS)
}
pub fn repeated_by(&self, member: &str, key_package_ref: &[u8]) -> bool {
self.member == member && self.key_package_ref == B64.encode(key_package_ref)
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
struct RemovalRecord {
epoch: u32,
removed_at: u64,
}
mod legacy {
pub(super) const GROUP_PREFIX: &str = "room-group:";
pub(super) fn group(room_id: &str) -> String {
format!("{GROUP_PREFIX}{room_id}")
}
pub(super) fn key_package(room_id: &str) -> String {
format!("room-kp:{room_id}")
}
pub(super) fn advance(room_id: &str) -> String {
format!("room-advance:{room_id}")
}
pub(super) fn add(room_id: &str) -> String {
format!("room-add:{room_id}")
}
pub(super) fn roots(room_id: &str) -> String {
format!("room-roots:{room_id}")
}
}
pub(super) fn invitation_key(credential_id: &str) -> String {
format!("room-vic:{credential_id}")
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PendingKeyPackage {
pub snapshot: IdentitySnapshot,
pub key_package: String,
pub expires_at: u64,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RoomGroupRecord {
pub snapshot: GroupSnapshot,
pub member_did: String,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub links: Vec<EpochLink>,
pub updated_at: u64,
}
pub async fn mint_key_package(
groups: &KeyspaceHandle,
member: &Member,
lifetime_secs: u64,
now: u64,
) -> Result<PendingKeyPackage, AppError> {
let (snapshot, package) = IdentitySnapshot::mint(member.persona())
.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(member.key(KEY_PACKAGE), &pending)
.await
.map_err(|e| AppError::Internal(format!("store the pending key package: {e}")))?;
Ok(pending)
}
pub async fn holds_key_package(
groups: &KeyspaceHandle,
member: &Member,
key_package: &[u8],
now: u64,
) -> Result<bool, AppError> {
let Ok(asked) = vti_rooms::mls::key_package_ref(key_package) else {
return Ok(false);
};
let pending: Option<PendingKeyPackage> = groups
.get(member.key(KEY_PACKAGE))
.await
.map_err(|e| AppError::Internal(format!("read the pending key package: {e}")))?;
let Some(pending) = pending else {
return Ok(false);
};
if pending.expires_at <= now || pending.snapshot.member_did() != member.persona() {
return Ok(false);
}
let Ok(minted) = B64.decode(pending.key_package.as_bytes()) else {
return Ok(false);
};
Ok(vti_rooms::mls::key_package_ref(&minted).is_ok_and(|r| r == asked))
}
async fn legacy_key_package(
groups: &KeyspaceHandle,
member: &Member,
) -> Result<Option<PendingKeyPackage>, AppError> {
let old = legacy::key_package(member.room_id());
let Some(raw) = groups.get_raw(old.clone()).await? else {
return Ok(None);
};
let Ok(kp) = serde_json::from_slice::<PendingKeyPackage>(&raw) else {
return Ok(None);
};
if kp.snapshot.member_did() != member.persona() {
return Ok(None);
}
move_row(groups, old, raw, member.key(KEY_PACKAGE), &kp).await?;
Ok(Some(kp))
}
#[derive(Debug)]
pub enum JoinError {
AlreadyJoined,
App(AppError),
}
impl From<AppError> for JoinError {
fn from(e: AppError) -> Self {
Self::App(e)
}
}
pub async fn join(
groups: &KeyspaceHandle,
member: &Member,
welcome: &[u8],
now: u64,
) -> Result<u64, JoinError> {
let room_id = member.room_id();
if load(groups, member).await?.is_some() {
return Err(JoinError::AlreadyJoined);
}
let pending: Option<PendingKeyPackage> = groups
.get(member.key(KEY_PACKAGE))
.await
.map_err(|e| AppError::Internal(format!("read the pending key package: {e}")))?;
let pending = match pending {
Some(p) => Some(p),
None => legacy_key_package(groups, member).await?,
};
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, member, &group, Vec::new(), now).await?;
groups
.remove(member.key(REMOVED))
.await
.map_err(|e| AppError::Internal(format!("clear an earlier removal record: {e}")))?;
groups
.remove(member.key(KEY_PACKAGE))
.await
.map_err(|e| AppError::Internal(format!("clear the pending key package: {e}")))?;
Ok(epoch)
}
pub async fn apply_commit(
groups: &KeyspaceHandle,
member: &Member,
commit: &[u8],
claimed_epoch: u64,
now: u64,
) -> Result<u64, AppError> {
let room_id = member.room_id();
let record = load(groups, member)
.await?
.ok_or_else(|| no_group(member))?;
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 {
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
)));
}
let (_, link) = match room.apply_commit(commit) {
Ok(applied) => applied,
Err(e) => {
if let RoomKeyError::Removed { epoch } = e {
record_removal(groups, member, epoch, now).await?;
}
return Err(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, member, room.group(), links, now).await?;
Ok(epoch)
}
pub async fn seal_record(
groups: &KeyspaceHandle,
member: &Member,
key: &str,
version: u64,
plaintext: &[u8],
) -> Result<vti_rooms::wire::SealedContent, AppError> {
let room_id = member.room_id();
let record = load(groups, member)
.await?
.ok_or_else(|| no_group(member))?;
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()))
}
#[derive(Debug)]
#[non_exhaustive]
pub enum FileKeyError {
NoGroup,
App(AppError),
NotDelivered { epoch: u32, held: u32 },
BeyondChain { epoch: u32, earliest: u32 },
Removed { epoch: u32, removed_at: u32 },
}
pub async fn file_key(
groups: &KeyspaceHandle,
member: &Member,
file_id: &vti_rooms::files::FileId,
open_epoch: Option<u32>,
) -> Result<(vti_rooms::files::FileKey, u32), FileKeyError> {
let room_id = member.room_id();
let record = load(groups, member)
.await
.map_err(FileKeyError::App)?
.ok_or(FileKeyError::NoGroup)?;
let group = RoomGroup::restore(&record.snapshot)
.map_err(|e| FileKeyError::App(AppError::Internal(format!("restore the group: {e}"))))?;
let mut room = SealedRoom::new(room_id, group);
let Some(epoch) = open_epoch else {
return room
.file_key_for_seal(file_id)
.map_err(|e| FileKeyError::App(AppError::Internal(e.to_string())));
};
room.add_links(record.links);
match room.file_key_for_open(file_id, epoch) {
Ok(key) => Ok((key, epoch)),
Err(RoomKeyError::EpochAhead { sealed, held }) => {
if let Some(removed_at) = removed_at(groups, member)
.await
.map_err(FileKeyError::App)?
&& sealed >= removed_at
{
return Err(FileKeyError::Removed {
epoch: sealed,
removed_at,
});
}
Err(FileKeyError::NotDelivered {
epoch: sealed,
held,
})
}
Err(RoomKeyError::EpochUnreachable { sealed, earliest }) => {
Err(FileKeyError::BeyondChain {
epoch: sealed,
earliest,
})
}
Err(e) => Err(FileKeyError::App(AppError::Validation(e.to_string()))),
}
}
pub async fn held_members(groups: &KeyspaceHandle) -> Result<Vec<Member>, AppError> {
members_under(groups, &format!("{GROUP}/")).await
}
pub async fn holders(groups: &KeyspaceHandle, room_id: &str) -> Result<Vec<Member>, AppError> {
members_under(groups, &format!("{GROUP}/{room_id}/")).await
}
async fn members_under(groups: &KeyspaceHandle, prefix: &str) -> Result<Vec<Member>, AppError> {
let keys = groups.prefix_keys(prefix.as_bytes().to_vec()).await?;
let mut out = Vec::with_capacity(keys.len());
for k in keys {
let k = String::from_utf8_lossy(&k);
let Some(rest) = k.strip_prefix(&format!("{GROUP}/")) else {
continue;
};
if let Some((room, persona)) = rest.split_once('/')
&& let Ok(m) = Member::new(room, persona)
{
out.push(m);
}
}
Ok(out)
}
pub async fn held_room(groups: &KeyspaceHandle, member: &Member) -> Result<HeldRoom, AppError> {
let room_id = member.room_id();
let record = load(groups, member)
.await?
.ok_or_else(|| no_group(member))?;
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}")))?;
Ok(HeldRoom {
room_id: room_id.to_string(),
epoch,
earliest_readable_epoch: earliest,
})
}
#[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,
}
pub async fn store_links(
groups: &KeyspaceHandle,
member: &Member,
incoming: Vec<EpochLink>,
now: u64,
) -> Result<(u32, usize), AppError> {
let room_id = member.room_id();
let record = load(groups, member)
.await?
.ok_or_else(|| no_group(member))?;
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, member, room.group(), links, now).await?;
Ok((earliest, stored))
}
pub async fn open_record(
groups: &KeyspaceHandle,
member: &Member,
key: &str,
version: u64,
ciphertext: &str,
nonce: &str,
epoch: u32,
) -> Result<Vec<u8>, AppError> {
open_record_detailed(groups, member, key, version, ciphertext, nonce, epoch)
.await
.map_err(AppError::from)
}
#[derive(Debug)]
#[non_exhaustive]
pub enum OpenError {
NotDelivered { epoch: u32, held: u64 },
Removed { epoch: u32, removed_at: u32 },
BeyondChain { epoch: u32, earliest: u32 },
App(AppError),
}
impl From<AppError> for OpenError {
fn from(e: AppError) -> Self {
Self::App(e)
}
}
impl std::fmt::Display for OpenError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::NotDelivered { epoch, held } => write!(
f,
"record is sealed under epoch {epoch} and this VTA holds the room at epoch \
{held}; a commit has not been delivered"
),
Self::Removed { epoch, removed_at } => write!(
f,
"record is sealed under epoch {epoch}, and this member was removed from the \
room's group at epoch {removed_at}: nothing sealed from then on opens here"
),
Self::BeyondChain { epoch, earliest } => write!(
f,
"{}",
RoomKeyError::EpochUnreachable {
sealed: *epoch,
earliest: *earliest,
}
),
Self::App(e) => write!(f, "{e}"),
}
}
}
impl From<OpenError> for AppError {
fn from(e: OpenError) -> Self {
match e {
OpenError::App(e) => e,
other => AppError::Validation(other.to_string()),
}
}
}
pub async fn open_record_detailed(
groups: &KeyspaceHandle,
member: &Member,
key: &str,
version: u64,
ciphertext: &str,
nonce: &str,
epoch: u32,
) -> Result<Vec<u8>, OpenError> {
let room_id = member.room_id();
let record = load(groups, member)
.await?
.ok_or_else(|| no_group(member))?;
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 {
if let Some(removed_at) = removed_at(groups, member).await?
&& epoch >= removed_at
{
return Err(OpenError::Removed { epoch, removed_at });
}
return Err(OpenError::NotDelivered { epoch, held });
}
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| match e {
RoomKeyError::EpochUnreachable { sealed, earliest } => OpenError::BeyondChain {
epoch: sealed,
earliest,
},
e => OpenError::App(AppError::Validation(e.to_string())),
})
}
pub async fn removed_at(groups: &KeyspaceHandle, member: &Member) -> Result<Option<u32>, AppError> {
let row: Option<RemovalRecord> = groups.get(member.key(REMOVED)).await.map_err(|e| {
AppError::Internal(format!(
"read the removal record for `{}`: {e}",
member.room_id()
))
})?;
Ok(row.map(|r| r.epoch))
}
async fn record_removal(
groups: &KeyspaceHandle,
member: &Member,
epoch: u32,
now: u64,
) -> Result<(), AppError> {
groups
.insert(
member.key(REMOVED),
&RemovalRecord {
epoch,
removed_at: now,
},
)
.await
.map_err(|e| {
AppError::Internal(format!(
"record the removal from `{}`: {e}",
member.room_id()
))
})
}
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}")))
}
pub async fn load(
groups: &KeyspaceHandle,
member: &Member,
) -> Result<Option<RoomGroupRecord>, AppError> {
groups.get(member.key(GROUP)).await.map_err(|e| {
AppError::Internal(format!("read group state for `{}`: {e}", member.room_id()))
})
}
fn no_group(member: &Member) -> AppError {
AppError::NotFound(format!(
"this VTA holds no group state for room `{}`",
member.room_id()
))
}
async fn store(
groups: &KeyspaceHandle,
member: &Member,
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.persona().to_string(),
links,
updated_at: now,
};
groups
.insert(member.key(GROUP), &record)
.await
.map_err(|e| {
AppError::Internal(format!("store group state for `{}`: {e}", member.room_id()))
})
}
#[derive(Debug)]
pub enum CreateGroupError {
AlreadyExists {
epoch: u32,
},
App(AppError),
}
pub async fn create_group(
groups: &KeyspaceHandle,
member: &Member,
now: u64,
) -> Result<u32, CreateGroupError> {
let room_id = member.room_id();
if let Some(existing) = load(groups, member).await.map_err(CreateGroupError::App)? {
let group = RoomGroup::restore(&existing.snapshot).map_err(|e| {
CreateGroupError::App(AppError::Internal(format!("restore the group: {e}")))
})?;
return Err(CreateGroupError::AlreadyExists {
epoch: SealedRoom::new(room_id, group).room_epoch(),
});
}
let group = RoomGroup::create(member.persona())
.map_err(|e| CreateGroupError::App(AppError::Internal(format!("found the group: {e}"))))?;
let epoch = u32::try_from(group.epoch() + 1).unwrap_or(u32::MAX);
store(groups, member, &group, Vec::new(), now)
.await
.map_err(CreateGroupError::App)?;
Ok(epoch)
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub enum AdvanceKind {
#[default]
Admission,
Removal,
Renewal,
}
impl AdvanceKind {
pub fn noun(self) -> &'static str {
match self {
Self::Admission => "admission",
Self::Removal => "removal",
Self::Renewal => "renewal",
}
}
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct PendingAdvance {
#[serde(default)]
pub kind: AdvanceKind,
pub snapshot: GroupSnapshot,
pub links: Vec<EpochLink>,
pub member: String,
pub host: String,
pub epoch: u32,
pub link: Option<EpochLink>,
pub commit: String,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub welcome: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub key_package_ref: Option<String>,
pub created_at: u64,
}
#[derive(Debug)]
pub enum PrepareAddError {
NoGroup,
AlreadyMember,
KeyPackageInvalid(String),
App(AppError),
}
async fn retain(
groups: &KeyspaceHandle,
owner: &Member,
pending: &PendingAdvance,
) -> Result<(), AppError> {
let room_id = owner.room_id();
let taken = |found: &PendingAdvance| {
AppError::Conflict(format!(
"room `{room_id}` already has a pending {} to epoch {} waiting on host `{}`; \
it is settled before another advance is built",
found.kind.noun(),
found.epoch,
found.host
))
};
let claimed = groups
.insert_if_absent(owner.key(ADVANCE), pending)
.await
.map_err(|e| {
AppError::Internal(format!("store the pending {}: {e}", pending.kind.noun()))
})?;
if !claimed {
let found = pending_advance(groups, owner)
.await?
.unwrap_or_else(|| pending.clone());
return Err(taken(&found));
}
discard_retained_admission(groups, owner).await
}
pub async fn prepare_add(
groups: &KeyspaceHandle,
owner: &Member,
host: &str,
member: &str,
key_package: &[u8],
now: u64,
) -> Result<PendingAdvance, PrepareAddError> {
let room_id = owner.room_id();
let record = load(groups, owner)
.await
.map_err(PrepareAddError::App)?
.ok_or(PrepareAddError::NoGroup)?;
let group = RoomGroup::restore(&record.snapshot)
.map_err(|e| PrepareAddError::App(AppError::Internal(format!("restore the group: {e}"))))?;
if group.leaf_of(member).is_some() {
return Err(PrepareAddError::AlreadyMember);
}
let mut room = SealedRoom::new(room_id, group);
room.add_links(record.links);
let (change, link) = room
.add_member(key_package)
.map_err(|e| PrepareAddError::KeyPackageInvalid(e.to_string()))?;
if room.group().leaf_of(member).is_none() {
return Err(PrepareAddError::KeyPackageInvalid(format!(
"the KeyPackage's leaf does not identify `{member}`"
)));
}
let welcome = change.welcome.ok_or_else(|| {
PrepareAddError::App(AppError::Internal("an add produced no Welcome".into()))
})?;
let key_package_ref = vti_rooms::mls::key_package_ref(key_package)
.map_err(|e| PrepareAddError::KeyPackageInvalid(e.to_string()))?;
let pending = PendingAdvance {
kind: AdvanceKind::Admission,
snapshot: room.group().snapshot().map_err(|e| {
PrepareAddError::App(AppError::Internal(format!("snapshot the group: {e}")))
})?,
links: room.links(),
member: member.to_string(),
host: host.to_string(),
epoch: room.room_epoch(),
link,
commit: B64.encode(&change.commit),
welcome: Some(B64.encode(&welcome)),
key_package_ref: Some(B64.encode(&key_package_ref)),
created_at: now,
};
retain(groups, owner, &pending)
.await
.map_err(PrepareAddError::App)?;
Ok(pending)
}
#[derive(Debug)]
pub enum PrepareRemoveError {
NoGroup,
NotMember,
OwnLeaf,
App(AppError),
}
pub async fn prepare_remove(
groups: &KeyspaceHandle,
owner: &Member,
host: &str,
member: &str,
now: u64,
) -> Result<PendingAdvance, PrepareRemoveError> {
let room_id = owner.room_id();
if owner.persona() == member {
return Err(PrepareRemoveError::OwnLeaf);
}
let record = load(groups, owner)
.await
.map_err(PrepareRemoveError::App)?
.ok_or(PrepareRemoveError::NoGroup)?;
if record.member_did == member {
return Err(PrepareRemoveError::OwnLeaf);
}
let group = RoomGroup::restore(&record.snapshot).map_err(|e| {
PrepareRemoveError::App(AppError::Internal(format!("restore the group: {e}")))
})?;
let leaf = group.leaf_of(member).ok_or(PrepareRemoveError::NotMember)?;
let mut room = SealedRoom::new(room_id, group);
room.add_links(record.links);
let (change, link) = room.remove_member(leaf).map_err(|e| {
PrepareRemoveError::App(AppError::Internal(format!("build the removal: {e}")))
})?;
let pending = PendingAdvance {
kind: AdvanceKind::Removal,
snapshot: room.group().snapshot().map_err(|e| {
PrepareRemoveError::App(AppError::Internal(format!("snapshot the group: {e}")))
})?,
links: room.links(),
member: member.to_string(),
host: host.to_string(),
epoch: room.room_epoch(),
link,
commit: B64.encode(&change.commit),
welcome: None,
key_package_ref: None,
created_at: now,
};
retain(groups, owner, &pending)
.await
.map_err(PrepareRemoveError::App)?;
Ok(pending)
}
pub async fn prepare_renew(
groups: &KeyspaceHandle,
owner: &Member,
host: &str,
now: u64,
) -> Result<Option<PendingAdvance>, AppError> {
let Some(record) = load(groups, owner).await? else {
return Ok(None);
};
let group = RoomGroup::restore(&record.snapshot)
.map_err(|e| AppError::Internal(format!("restore the group: {e}")))?;
let mut room = SealedRoom::new(owner.room_id(), group);
room.add_links(record.links);
let (change, link) = room
.renew()
.map_err(|e| AppError::Internal(format!("build the renewal: {e}")))?;
let pending = PendingAdvance {
kind: AdvanceKind::Renewal,
snapshot: room
.group()
.snapshot()
.map_err(|e| AppError::Internal(format!("snapshot the group: {e}")))?,
links: room.links(),
member: owner.persona().to_string(),
host: host.to_string(),
epoch: room.room_epoch(),
link,
commit: B64.encode(&change.commit),
welcome: None,
key_package_ref: None,
created_at: now,
};
retain(groups, owner, &pending).await?;
Ok(Some(pending))
}
pub async fn pending_advance(
groups: &KeyspaceHandle,
owner: &Member,
) -> Result<Option<PendingAdvance>, AppError> {
groups
.get::<PendingAdvance>(owner.key(ADVANCE))
.await
.map_err(|e| AppError::Internal(format!("read the pending advance: {e}")))
}
pub async fn promote_pending_advance(
groups: &KeyspaceHandle,
owner: &Member,
pending: &PendingAdvance,
now: u64,
) -> Result<(), AppError> {
load(groups, owner).await?.ok_or_else(|| no_group(owner))?;
let group = RoomGroup::restore(&pending.snapshot)
.map_err(|e| AppError::Internal(format!("restore the pending group: {e}")))?;
store(groups, owner, &group, pending.links.clone(), now).await?;
retain_admission(groups, owner, pending, now).await?;
discard_pending_advance(groups, owner).await
}
async fn retain_admission(
groups: &KeyspaceHandle,
owner: &Member,
pending: &PendingAdvance,
now: u64,
) -> Result<(), AppError> {
let (AdvanceKind::Admission, Some(welcome), Some(key_package_ref)) =
(pending.kind, &pending.welcome, &pending.key_package_ref)
else {
return Ok(());
};
groups
.insert(
owner.key(ADMITTED),
&RetainedAdmission {
member: pending.member.clone(),
key_package_ref: key_package_ref.clone(),
epoch: pending.epoch,
welcome: welcome.clone(),
ratchet_tree: None,
applied_at: now,
},
)
.await
.map_err(|e| AppError::Internal(format!("retain the applied admission: {e}")))
}
pub async fn retained_admission(
groups: &KeyspaceHandle,
owner: &Member,
now: u64,
) -> Result<Option<RetainedAdmission>, AppError> {
let row: Option<RetainedAdmission> = groups
.get(owner.key(ADMITTED))
.await
.map_err(|e| AppError::Internal(format!("read the retained admission: {e}")))?;
match row {
Some(r) if r.kept_at(now) => Ok(Some(r)),
Some(_) => {
discard_retained_admission(groups, owner).await?;
Ok(None)
}
None => Ok(None),
}
}
async fn discard_retained_admission(
groups: &KeyspaceHandle,
owner: &Member,
) -> Result<(), AppError> {
groups
.remove(owner.key(ADMITTED))
.await
.map_err(|e| AppError::Internal(format!("discard the retained admission: {e}")))
}
pub async fn sweep_retained_admissions(
groups: &KeyspaceHandle,
now: u64,
) -> Result<usize, AppError> {
let rows = groups
.prefix_iter_raw(format!("{ADMITTED}/").into_bytes())
.await
.map_err(|e| AppError::Internal(format!("scan retained admissions: {e}")))?;
let mut discarded = 0;
for (key, raw) in rows {
let keep = serde_json::from_slice::<RetainedAdmission>(&raw).is_ok_and(|r| r.kept_at(now));
if !keep {
groups
.remove(key)
.await
.map_err(|e| AppError::Internal(format!("discard a retained admission: {e}")))?;
discarded += 1;
}
}
Ok(discarded)
}
pub async fn discard_pending_advance(
groups: &KeyspaceHandle,
owner: &Member,
) -> Result<(), AppError> {
groups
.remove(owner.key(ADVANCE))
.await
.map_err(|e| AppError::Internal(format!("clear the pending advance: {e}")))
}
#[derive(Debug, PartialEq, Eq)]
pub enum Reconciled {
Applied,
Discarded,
Conflict,
}
pub async fn reconcile_pending_advance(
groups: &KeyspaceHandle,
owner: &Member,
pending: &PendingAdvance,
host_epoch: u32,
now: u64,
) -> Result<Reconciled, AppError> {
let live_epoch = match load(groups, owner).await? {
Some(r) => SealedRoom::new(
owner.room_id(),
RoomGroup::restore(&r.snapshot)
.map_err(|e| AppError::Internal(format!("restore the group: {e}")))?,
)
.room_epoch(),
None => 0,
};
if live_epoch == pending.epoch {
retain_admission(groups, owner, pending, now).await?;
discard_pending_advance(groups, owner).await?;
return Ok(Reconciled::Applied);
}
if host_epoch == pending.epoch {
promote_pending_advance(groups, owner, pending, now).await?;
return Ok(Reconciled::Applied);
}
discard_pending_advance(groups, owner).await?;
Ok(if host_epoch == live_epoch {
Reconciled::Discarded
} else {
Reconciled::Conflict
})
}
pub async fn legacy_rooms(groups: &KeyspaceHandle) -> Result<Vec<String>, AppError> {
let keys = groups
.prefix_keys(legacy::GROUP_PREFIX.as_bytes().to_vec())
.await?;
Ok(keys
.into_iter()
.filter_map(|k| {
String::from_utf8_lossy(&k)
.strip_prefix(legacy::GROUP_PREFIX)
.map(str::to_string)
})
.collect())
}
pub async fn legacy_group(
groups: &KeyspaceHandle,
room_id: &str,
) -> Result<Option<RoomGroupRecord>, AppError> {
groups
.get(legacy::group(room_id))
.await
.map_err(|e| AppError::Internal(format!("read legacy group state for `{room_id}`: {e}")))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Migrated {
Nothing,
Moved,
Kept,
}
pub async fn migrate_legacy(
groups: &KeyspaceHandle,
member: &Member,
) -> Result<Migrated, AppError> {
let room_id = member.room_id();
let Some(group_raw) = groups.get_raw(legacy::group(room_id)).await? else {
return Ok(Migrated::Nothing);
};
if let Some(raw) = groups.get_raw(legacy::key_package(room_id)).await?
&& let Ok(kp) = serde_json::from_slice::<PendingKeyPackage>(&raw)
&& kp.snapshot.member_did() == member.persona()
{
move_row(
groups,
legacy::key_package(room_id),
raw,
member.key(KEY_PACKAGE),
&kp,
)
.await?;
}
for old in [legacy::advance(room_id), legacy::add(room_id)] {
if let Some(raw) = groups.get_raw(old.clone()).await?
&& let Ok(advance) = serde_json::from_slice::<PendingAdvance>(&raw)
{
move_row(groups, old, raw, member.key(ADVANCE), &advance).await?;
}
}
if let Some(raw) = groups.get_raw(legacy::roots(room_id)).await?
&& let Ok(roots) = serde_json::from_slice::<RootHistory>(&raw)
{
move_row(
groups,
legacy::roots(room_id),
raw,
member.key(ROOTS),
&roots,
)
.await?;
}
let record: RoomGroupRecord = serde_json::from_slice(&group_raw).map_err(|e| {
AppError::Internal(format!("decode legacy group state for `{room_id}`: {e}"))
})?;
Ok(
match move_row(
groups,
legacy::group(room_id),
group_raw,
member.key(GROUP),
&record,
)
.await?
{
MoveOutcome::Moved | MoveOutcome::SourceMissing | MoveOutcome::SourceChanged => {
Migrated::Moved
}
MoveOutcome::TargetExists => {
tracing::warn!(
room = %room_id,
member = %member.persona(),
"a group row from before per-member keying was left in place: this \
member already holds group state of its own for the room"
);
Migrated::Kept
}
},
)
}
async fn move_row<V: serde::Serialize>(
groups: &KeyspaceHandle,
old: String,
expected: Vec<u8>,
new: String,
value: &V,
) -> Result<MoveOutcome, AppError> {
groups
.move_if_unchanged(old, expected, new, value)
.await
.map_err(|e| AppError::Internal(format!("migrate room state: {e}")))
}
#[cfg(test)]
pub(crate) mod legacy_for_test {
use super::*;
pub(crate) async fn store_group(
groups: &KeyspaceHandle,
room_id: &str,
record: &RoomGroupRecord,
) {
groups.insert(legacy::group(room_id), record).await.unwrap();
}
pub(crate) async fn store_pending_add(
groups: &KeyspaceHandle,
room_id: &str,
pending: &serde_json::Value,
) {
groups.insert(legacy::add(room_id), pending).await.unwrap();
}
}
#[cfg(test)]
pub(crate) async fn store_for_test(
groups: &KeyspaceHandle,
member: &Member,
group: &RoomGroup,
links: Vec<EpochLink>,
) -> Result<(), AppError> {
store(groups, member, group, links, 0).await
}
#[cfg(test)]
pub(crate) async fn age_retained_admission_for_test(
groups: &KeyspaceHandle,
owner: &Member,
secs: u64,
) {
let mut row: RetainedAdmission = groups.get(owner.key(ADMITTED)).await.unwrap().unwrap();
row.applied_at = row.applied_at.saturating_sub(secs);
groups.insert(owner.key(ADMITTED), &row).await.unwrap();
}
#[cfg(test)]
pub(crate) async fn has_retained_admission_row(groups: &KeyspaceHandle, owner: &Member) -> bool {
groups.get_raw(owner.key(ADMITTED)).await.unwrap().is_some()
}
#[cfg(test)]
pub(crate) async fn record_removal_for_test(
groups: &KeyspaceHandle,
member: &Member,
epoch: u32,
) -> Result<(), AppError> {
record_removal(groups, member, epoch, 0).await
}
const KEEP_ROOTS: usize = 16;
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RootHistory {
pub seen: Vec<RootObservation>,
pub highest: u64,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub caught_at: Option<u64>,
}
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct RootObservation {
pub head_version: u64,
pub root: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RootVerdict {
Agree,
Conflict,
NoneHeld,
}
pub async fn observe_head(
groups: &KeyspaceHandle,
member: &Member,
head_version: u64,
root: &str,
) -> Result<RootVerdict, AppError> {
let room_id = member.room_id();
let key = member.key(ROOTS);
let mut history = root_history(groups, member).await?;
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) {
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);
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)
}
pub async fn root_history(
groups: &KeyspaceHandle,
member: &Member,
) -> Result<RootHistory, AppError> {
let room_id = member.room_id();
let read =
|e: AppError| AppError::Internal(format!("read the root history of `{room_id}`: {e}"));
if let Some(h) = groups.get(member.key(ROOTS)).await.map_err(read)? {
return Ok(h);
}
Ok(groups
.get(legacy::roots(room_id))
.await
.map_err(read)?
.unwrap_or_default())
}
pub async fn forget(groups: &KeyspaceHandle, member: &Member) -> Result<(), AppError> {
let room_id = member.room_id();
groups
.remove(member.key(GROUP))
.await
.map_err(|e| AppError::Internal(format!("discard group state for `{room_id}`: {e}")))?;
groups
.remove(member.key(KEY_PACKAGE))
.await
.map_err(|e| AppError::Internal(format!("discard a pending key package: {e}")))?;
discard_pending_advance(groups, member).await?;
discard_retained_admission(groups, member).await?;
groups
.remove(member.key(ROOTS))
.await
.map_err(|e| AppError::Internal(format!("discard the root history of `{room_id}`: {e}")))
}
pub async fn epoch_authenticator(
groups: &KeyspaceHandle,
member: &Member,
) -> Result<(u64, Vec<u8>), AppError> {
let room_id = member.room_id();
let record = load(groups, member).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}")))?;
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";
fn member() -> Member {
Member::new(ROOM, "did:key:zAlice").unwrap()
}
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)
}
#[tokio::test]
async fn a_first_reading_holds_nothing_to_compare() {
let (_d, ks) = open().await;
assert_eq!(
observe_head(&ks, &member(), 412, A).await.unwrap(),
RootVerdict::NoneHeld
);
}
#[tokio::test]
async fn two_roots_at_one_version_is_a_host_caught() {
let (_d, ks) = open().await;
observe_head(&ks, &member(), 412, A).await.unwrap();
assert_eq!(
observe_head(&ks, &member(), 412, A).await.unwrap(),
RootVerdict::Agree
);
assert_eq!(
observe_head(&ks, &member(), 412, B).await.unwrap(),
RootVerdict::Conflict
);
}
#[tokio::test]
async fn a_different_version_is_a_different_moment() {
let (_d, ks) = open().await;
observe_head(&ks, &member(), 412, A).await.unwrap();
assert_eq!(
observe_head(&ks, &member(), 413, B).await.unwrap(),
RootVerdict::NoneHeld,
"a write moved the head, so a different root explains itself"
);
}
#[tokio::test]
async fn a_head_that_goes_backwards_is_still_compared() {
let (_d, ks) = open().await;
observe_head(&ks, &member(), 412, A).await.unwrap();
observe_head(&ks, &member(), 500, B).await.unwrap();
assert_eq!(
observe_head(&ks, &member(), 412, B).await.unwrap(),
RootVerdict::Conflict,
"the older version is still held, and its root still disagrees"
);
}
#[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, &member(), v, A).await.unwrap();
}
let history = root_history(&ks, &member()).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"
);
}
#[tokio::test]
async fn forgetting_a_room_forgets_what_its_host_said() {
let (_d, ks) = open().await;
observe_head(&ks, &member(), 412, A).await.unwrap();
forget(&ks, &member()).await.unwrap();
assert_eq!(
observe_head(&ks, &member(), 412, B).await.unwrap(),
RootVerdict::NoneHeld,
"nothing survived the removal to compare against"
);
}
}
#[cfg(test)]
mod file_key_tests {
use super::*;
use vti_common::config::StoreConfig;
use vti_common::store::Store;
const ROOM: &str = "did:webvh:example.com:rooms:northwind";
fn member() -> Member {
Member::new(ROOM, "did:key:zAlice").unwrap()
}
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)
}
async fn held(ks: &KeyspaceHandle, chained: bool) -> SealedRoom {
let mut room = SealedRoom::new(ROOM, RoomGroup::create("did:key:zAlice").unwrap());
let (_bob, kp) = IdentitySnapshot::mint("did:key:zBob").unwrap();
room.add_member(&kp).unwrap();
let links = if chained { room.links() } else { Vec::new() };
store(ks, &member(), room.group(), links, 0).await.unwrap();
room
}
#[tokio::test]
async fn a_sealing_key_is_the_current_epoch_and_opens_later() {
let (_d, ks) = open().await;
let room = held(&ks, true).await;
let id = [4u8; 32];
let (seal_key, epoch) = file_key(&ks, &member(), &id, None).await.unwrap();
assert_eq!(epoch, 2, "a sealing key is always the current epoch");
assert_eq!(seal_key, room.file_key_for_seal(&id).unwrap().0);
let (open_key, e) = file_key(&ks, &member(), &id, Some(2)).await.unwrap();
assert_eq!((open_key, e), (seal_key, 2));
}
#[tokio::test]
async fn an_earlier_epoch_is_reached_through_the_chain() {
let (_d, ks) = open().await;
held(&ks, true).await;
let (k1, e) = file_key(&ks, &member(), &[4u8; 32], Some(1)).await.unwrap();
assert_eq!(e, 1);
assert_ne!(
k1,
file_key(&ks, &member(), &[4u8; 32], Some(2))
.await
.unwrap()
.0
);
}
#[tokio::test]
async fn the_two_unreachable_cases_say_which_they_are() {
let (_d, ks) = open().await;
held(&ks, false).await;
assert!(matches!(
file_key(&ks, &member(), &[4u8; 32], Some(3)).await,
Err(FileKeyError::NotDelivered { epoch: 3, held: 2 })
));
assert!(matches!(
file_key(&ks, &member(), &[4u8; 32], Some(1)).await,
Err(FileKeyError::BeyondChain {
epoch: 1,
earliest: 2
})
));
}
#[tokio::test]
async fn no_group_is_not_found() {
let (_d, ks) = open().await;
assert!(matches!(
file_key(&ks, &member(), &[4u8; 32], None).await,
Err(FileKeyError::NoGroup)
));
}
}
#[cfg(test)]
mod per_member_storage_tests {
use super::*;
use vti_common::config::StoreConfig;
use vti_common::store::Store;
const ROOM: &str = "did:webvh:example.com:rooms:northwind";
const ALICE: &str = "did:key:zAlice";
const BOB: &str = "did:key:zBob";
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)
}
fn m(persona: &str) -> Member {
Member::new(ROOM, persona).unwrap()
}
#[test]
fn identifiers_that_would_make_the_key_ambiguous_are_refused() {
assert!(Member::new("did:key:zA/b", ALICE).is_err());
assert!(Member::new(ROOM, "did:key:zA/b").is_err());
assert!(Member::new("", ALICE).is_err());
}
#[tokio::test]
async fn two_members_of_one_room_are_two_groups() {
let (_d, ks) = open().await;
for who in [ALICE, BOB] {
create_group(&ks, &m(who), 0).await.unwrap();
}
let mut held: Vec<String> = holders(&ks, ROOM)
.await
.unwrap()
.into_iter()
.map(|m| m.persona().to_string())
.collect();
held.sort();
assert_eq!(held, vec![ALICE.to_string(), BOB.to_string()]);
assert!(matches!(
create_group(&ks, &m(ALICE), 0).await,
Err(CreateGroupError::AlreadyExists { .. })
));
}
#[tokio::test]
async fn root_memory_is_the_members_own() {
let (_d, ks) = open().await;
observe_head(&ks, &m(ALICE), 412, "zA").await.unwrap();
assert_eq!(
observe_head(&ks, &m(ALICE), 412, "zB").await.unwrap(),
RootVerdict::Conflict
);
assert_eq!(
observe_head(&ks, &m(BOB), 412, "zB").await.unwrap(),
RootVerdict::NoneHeld,
"bob holds no memory of alice's reading"
);
assert!(
root_history(&ks, &m(BOB))
.await
.unwrap()
.caught_at
.is_none()
);
}
#[tokio::test]
async fn a_legacy_room_moves_whole_under_its_member() {
let (_d, ks) = open().await;
let group = RoomGroup::create(ALICE).unwrap();
let record = RoomGroupRecord {
snapshot: group.snapshot().unwrap(),
member_did: ALICE.into(),
links: Vec::new(),
updated_at: 7,
};
ks.insert(legacy::group(ROOM), &record).await.unwrap();
let (snapshot, kp) = IdentitySnapshot::mint(BOB).unwrap();
let bobs_package = PendingKeyPackage {
snapshot,
key_package: B64.encode(kp),
expires_at: 9,
};
ks.insert(legacy::key_package(ROOM), &bobs_package)
.await
.unwrap();
ks.insert(
legacy::roots(ROOM),
&RootHistory {
caught_at: Some(3),
..Default::default()
},
)
.await
.unwrap();
assert_eq!(
migrate_legacy(&ks, &m(ALICE)).await.unwrap(),
Migrated::Moved
);
assert_eq!(load(&ks, &m(ALICE)).await.unwrap().unwrap().updated_at, 7);
assert_eq!(
root_history(&ks, &m(ALICE)).await.unwrap().caught_at,
Some(3),
"a host caught before the upgrade stays caught"
);
assert!(ks.get_raw(legacy::group(ROOM)).await.unwrap().is_none());
assert!(ks.get_raw(legacy::roots(ROOM)).await.unwrap().is_none());
assert!(
ks.get_raw(legacy::key_package(ROOM))
.await
.unwrap()
.is_some(),
"bob's package is not alice's to take"
);
assert!(
ks.get_raw(m(ALICE).key(KEY_PACKAGE))
.await
.unwrap()
.is_none()
);
assert_eq!(
migrate_legacy(&ks, &m(ALICE)).await.unwrap(),
Migrated::Nothing
);
assert_eq!(
legacy_key_package(&ks, &m(BOB))
.await
.unwrap()
.map(|p| p.expires_at),
Some(9)
);
assert!(
ks.get_raw(legacy::key_package(ROOM))
.await
.unwrap()
.is_none()
);
}
#[tokio::test]
async fn a_legacy_row_never_overwrites_a_members_own_group() {
let (_d, ks) = open().await;
create_group(&ks, &m(ALICE), 42).await.unwrap();
let record = RoomGroupRecord {
snapshot: RoomGroup::create(ALICE).unwrap().snapshot().unwrap(),
member_did: ALICE.into(),
links: Vec::new(),
updated_at: 7,
};
ks.insert(legacy::group(ROOM), &record).await.unwrap();
assert_eq!(
migrate_legacy(&ks, &m(ALICE)).await.unwrap(),
Migrated::Kept
);
assert_eq!(load(&ks, &m(ALICE)).await.unwrap().unwrap().updated_at, 42);
assert!(ks.get_raw(legacy::group(ROOM)).await.unwrap().is_some());
}
}
#[cfg(test)]
mod removal_tests {
use super::*;
use vti_common::config::StoreConfig;
use vti_common::store::Store;
const ROOM: &str = "did:webvh:example.com:rooms:northwind";
const BOB: &str = "did:key:zBob";
#[tokio::test]
async fn a_record_sealed_after_the_members_removal_says_removed() {
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();
let bob = Member::new(ROOM, BOB).unwrap();
let pending = mint_key_package(&ks, &bob, 3600, 0).await.unwrap();
let mut alice = SealedRoom::new(ROOM, RoomGroup::create("did:key:zAlice").unwrap());
let (added, _) = alice
.add_member(&B64.decode(&pending.key_package).unwrap())
.unwrap();
join(&ks, &bob, &added.welcome.unwrap(), 0).await.unwrap();
let before = alice.seal_record("k", 1, b"before").unwrap();
let leaf = alice.group().leaf_of(BOB).unwrap();
let (removal, _) = alice.remove_member(leaf).unwrap();
let after = alice.seal_record("k", 2, b"after").unwrap();
assert!(matches!(
open_record_detailed(
&ks,
&bob,
"k",
2,
&after.ciphertext,
&after.nonce,
after.epoch
)
.await,
Err(OpenError::NotDelivered { .. })
));
let file = [7u8; 32];
assert!(matches!(
file_key(&ks, &bob, &file, Some(after.epoch)).await.err(),
Some(FileKeyError::NotDelivered { .. })
));
let err = apply_commit(&ks, &bob, &removal.commit, removal.epoch, 5)
.await
.unwrap_err();
assert!(err.to_string().contains("removed this holder"), "{err}");
assert_eq!(removed_at(&ks, &bob).await.unwrap(), Some(after.epoch));
match open_record_detailed(
&ks,
&bob,
"k",
2,
&after.ciphertext,
&after.nonce,
after.epoch,
)
.await
{
Err(OpenError::Removed { epoch, removed_at }) => {
assert_eq!((epoch, removed_at), (after.epoch, after.epoch))
}
other => panic!("expected Removed, got {other:?}"),
}
match file_key(&ks, &bob, &file, Some(after.epoch)).await.err() {
Some(FileKeyError::Removed { epoch, removed_at }) => {
assert_eq!((epoch, removed_at), (after.epoch, after.epoch))
}
other => panic!("expected FileKeyError::Removed, got {other:?}"),
}
assert!(
file_key(&ks, &bob, &file, Some(before.epoch)).await.is_ok(),
"a file sealed before the removal still derives"
);
assert_eq!(
open_record(
&ks,
&bob,
"k",
1,
&before.ciphertext,
&before.nonce,
before.epoch
)
.await
.unwrap(),
b"before"
);
}
}