coven_protocol/membership/

chain.rs

use super::entry::store_membership_anchor_stream;
use super::*;

impl MembershipChain {
    pub fn from_entries_with_coords_and_heads_and_provider_admin(
        entries: Vec<(MembershipCoord, MembershipEntry)>,
        heads: Vec<(MembershipHeadRef, AuthorHead)>,
        provider_admin: crate::provider::ProviderAdminState,
    ) -> Result<Self, MembershipError> {
        let expected_store = entries
            .first()
            .map(|(_, entry)| entry.store_id.as_str())
            .ok_or(MembershipError::EmptyChain)?;
        if heads.iter().any(|(reference, head)| {
            reference.head_hash != head.head_hash()
                || head.store_id != expected_store
                || entries
                    .iter()
                    .find(|(coord, _)| *coord == head.entry_coord())
                    .is_none()
        }) {
            return Err(MembershipError::MissingExactHeads);
        }
        Self::from_entries_with_coords_and_head_refs(
            entries,
            heads.into_iter().map(|(reference, _)| reference).collect(),
            provider_admin,
        )
    }

    fn from_entries_with_coords_and_head_refs(
        entries: Vec<(MembershipCoord, MembershipEntry)>,
        head_refs: Vec<MembershipHeadRef>,
        provider_admin_genesis: crate::provider::ProviderAdminState,
    ) -> Result<Self, MembershipError> {
        if entries.is_empty() {
            return Err(MembershipError::EmptyChain);
        }
        for (index, (_, entry)) in entries.iter().enumerate() {
            Self::validate_entry_authenticity(index, entry)?;
        }
        let (coords, entries): (Vec<_>, Vec<_>) = entries.into_iter().unzip();
        let (included, resolved) = Self::reduce(&entries, &coords, &provider_admin_genesis)?;
        Ok(Self {
            entries,
            coords,
            included,
            resolved,
            head_refs,
            provider_admin_genesis,
        })
    }

    pub fn entries(&self) -> &[MembershipEntry] {
        &self.entries
    }

    pub fn resolved(&self) -> &ResolvedStoreMembership {
        &self.resolved
    }

    pub fn head_refs(&self) -> &[MembershipHeadRef] {
        &self.head_refs
    }

    pub fn head_ref_for_stream(
        &self,
        author: &str,
        grant: &MembershipGrantId,
        stream_id: AuthorStreamId,
    ) -> Option<&MembershipHeadRef> {
        self.head_refs.iter().find(|reference| {
            reference.coord.author_pubkey == author
                && reference.coord.author_owner_grant == *grant
                && reference.coord.stream_id == stream_id
        })
    }

    /// Whether this chain reaches at least as far as `heads` on every stream
    /// they name.
    ///
    /// A membership stream is a hash-linked list walked from its founder anchor
    /// forward: a walk cannot start in the middle, so two walks of the same
    /// stream verify the same heads up to the shorter one's end. A chain that
    /// covers a set of cursors therefore contains everything a walk anchored at
    /// those cursors would have verified, and can stand in for it.
    pub fn covers_heads(&self, heads: &[MembershipHeadRef]) -> bool {
        heads.iter().all(|covered| {
            self.head_ref_for_stream(
                &covered.coord.author_pubkey,
                &covered.coord.author_owner_grant,
                covered.coord.stream_id,
            )
            .is_some_and(|current| {
                current.coord.seq > covered.coord.seq
                    || (current.coord.seq == covered.coord.seq && current == covered)
            })
        })
    }

    pub fn membership_anchor(&self, grant: &MembershipGrantId) -> Option<&GrantStreamAnchor> {
        self.entries.iter().find_map(|entry| match &entry.change {
            StoreAuthorityChange::Founder {
                owner_grant_id,
                membership,
                ..
            } if owner_grant_id == grant => Some(membership),
            StoreAuthorityChange::SetMember {
                grant_id,
                membership: Some(membership),
                ..
            } if grant_id == grant => Some(membership),
            _ => None,
        })
    }

    pub fn membership_stream_id(&self, grant: &MembershipGrantId) -> Option<AuthorStreamId> {
        let record = self.resolved().grants.get(grant)?.record();
        let founder = self.founder_coord()?;
        if founder.author_owner_grant == *grant {
            // Creation owns the founder's first slot before grant-based paths
            // exist; the verified founder entry owns its stream identity.
            return Some(founder.stream_id);
        }
        store_membership_anchor_stream(&record.member_pubkey, grant, self.membership_anchor(grant)?)
    }

    pub fn activated_membership_streams(&self) -> Vec<(MembershipStreamKey, GrantStreamAnchor)> {
        let mut streams = self
            .resolved()
            .grants
            .iter()
            .filter_map(|(grant, state)| {
                let record = state.record();
                let anchor = self.membership_anchor(grant)?.clone();
                let stream_id = self.membership_stream_id(grant)?;
                Some((
                    MembershipStreamKey {
                        author_pubkey: record.member_pubkey.clone(),
                        author_owner_grant: grant.clone(),
                        stream_id,
                    },
                    anchor,
                ))
            })
            .collect::<BTreeMap<_, _>>();
        let included = &self.included;
        for (coord, entry) in self.entries_with_coords() {
            if !included.contains(coord) {
                continue;
            }
            let (owner_pubkey, grant, anchor, stream_id) = match &entry.change {
                StoreAuthorityChange::Founder {
                    owner_pubkey,
                    owner_grant_id,
                    membership,
                    ..
                } => (owner_pubkey, owner_grant_id, membership, entry.stream_id),
                StoreAuthorityChange::SetMember {
                    user_pubkey,
                    role: StoreMembershipRoleGrant::Owner { .. },
                    grant_id,
                    membership: Some(membership),
                    ..
                } => (
                    user_pubkey,
                    grant_id,
                    membership,
                    store_membership_anchor_stream(user_pubkey, grant_id, membership)
                        .expect("validated Owner grant has a Store membership stream anchor"),
                ),
                _ => continue,
            };
            streams.insert(
                MembershipStreamKey {
                    author_pubkey: owner_pubkey.clone(),
                    author_owner_grant: grant.clone(),
                    stream_id,
                },
                anchor.clone(),
            );
        }
        streams.into_iter().collect()
    }

    pub fn activate_head_ref(
        &mut self,
        reference: MembershipHeadRef,
    ) -> Result<(), MembershipError> {
        if !self.coords.contains(&reference.coord) {
            return Err(MembershipError::MissingExactHeads);
        }
        let stream = reference.coord.stream_key();
        self.head_refs
            .retain(|current| current.coord.stream_key() != stream);
        self.head_refs.push(reference);
        self.head_refs.sort();
        Ok(())
    }

    pub(crate) fn entries_with_coords(
        &self,
    ) -> impl Iterator<Item = (&MembershipCoord, &MembershipEntry)> {
        self.coords.iter().zip(self.entries.iter())
    }

    pub fn store_id(&self) -> Option<&str> {
        self.entries.first().map(|entry| entry.store_id.as_str())
    }

    pub fn founder_coord(&self) -> Option<&MembershipCoord> {
        self.entries_with_coords().find_map(|(coord, entry)| {
            matches!(entry.change, StoreAuthorityChange::Founder { .. }).then_some(coord)
        })
    }

    pub(crate) fn founder_entry(&self) -> Option<&MembershipEntry> {
        self.entries
            .iter()
            .find(|entry| matches!(entry.change, StoreAuthorityChange::Founder { .. }))
    }

    pub fn founder_pubkey(&self) -> Option<&str> {
        self.founder_entry().and_then(|entry| match &entry.change {
            StoreAuthorityChange::Founder { owner_pubkey, .. } => Some(owner_pubkey.as_str()),
            StoreAuthorityChange::SetMember { .. }
            | StoreAuthorityChange::RemoveMember { .. }
            | StoreAuthorityChange::DeviceRegistrationActivation { .. }
            | StoreAuthorityChange::DeviceExclusionProposal { .. }
            | StoreAuthorityChange::DeviceExclusionOutcome { .. }
            | StoreAuthorityChange::ProviderAdmin => None,
        })
    }

    pub fn is_founded_by(&self, owner_pubkey: &str) -> bool {
        self.founder_pubkey() == Some(owner_pubkey)
    }

    pub fn add_entry(&mut self, entry: MembershipEntry) -> Result<(), MembershipError> {
        self.add_entry_at(entry.coord(), entry)
    }

    pub fn add_entry_at(
        &mut self,
        coord: MembershipCoord,
        entry: MembershipEntry,
    ) -> Result<(), MembershipError> {
        Self::validate_entry_authenticity(self.entries.len(), &entry)?;
        self.entries.push(entry);
        self.coords.push(coord);
        match Self::reduce(&self.entries, &self.coords, &self.provider_admin_genesis) {
            Ok((included, resolved)) => {
                self.included = included;
                self.resolved = resolved;
            }
            Err(error) => {
                self.entries.pop();
                self.coords.pop();
                return Err(error);
            }
        }
        Ok(())
    }

    pub fn with_exact_entry(&self, entry: &MembershipEntry) -> Result<Self, MembershipError> {
        let coord = entry.coord();
        if let Some((_, stored)) = self
            .entries_with_coords()
            .find(|(stored_coord, _)| **stored_coord == coord)
        {
            if stored != entry {
                return Err(MembershipError::ExactEntryMismatch {
                    coord: Box::new(coord),
                });
            }
            return Ok(self.clone());
        }
        let mut chain = self.clone();
        chain.add_entry_at(coord, entry.clone())?;
        Ok(chain)
    }

    pub fn contains_coord(&self, expected: &MembershipCoord) -> bool {
        self.coords.iter().any(|coord| coord == expected)
    }

    pub fn effectively_contains_coord(&self, expected: &MembershipCoord) -> bool {
        self.included.contains(expected)
    }

    pub fn device_registration_activations(
        &self,
    ) -> impl Iterator<
        Item = (
            &MembershipCoord,
            &crate::store_commit::ActivatedStoreDeviceRegistrationRef,
        ),
    > {
        self.entries_with_coords().filter_map(|(coord, entry)| {
            let StoreAuthorityChange::DeviceRegistrationActivation { registration } = &entry.change
            else {
                return None;
            };
            self.included
                .contains(coord)
                .then_some((coord, registration))
        })
    }

    pub(crate) fn contains_member_history(&self, pubkey: &str) -> bool {
        self.resolved()
            .grants
            .values()
            .any(|state| state.record().member_pubkey == pubkey)
    }

    pub fn reusable_author_streams(
        &self,
        author_pubkey: &str,
        grant: &MembershipGrantId,
    ) -> BTreeSet<AuthorStreamId> {
        self.effective_frontier()
            .into_iter()
            .filter(|coord| {
                coord.author_pubkey == author_pubkey
                    && coord.author_owner_grant == *grant
                    && self.raw_stream_tip(author_pubkey, grant, coord.stream_id)
                        == Some(coord.clone())
            })
            .map(|coord| coord.stream_id)
            .collect()
    }

    /// Effective authoring frontier after causal pruning.
    pub fn effective_frontier(&self) -> Vec<MembershipCoord> {
        causal_grants::stream_frontier(
            self.coords
                .iter()
                .filter(|coord| self.included.contains(*coord))
                .cloned(),
        )
    }

    pub fn causally_includes(&self, predecessor: &MembershipChain) -> bool {
        predecessor.included.is_subset(&self.included)
    }

    pub(crate) fn stream_tip(
        &self,
        author_pubkey: &str,
        grant: &MembershipGrantId,
        stream_id: AuthorStreamId,
    ) -> Option<MembershipCoord> {
        self.effective_frontier().into_iter().find(|coord| {
            coord.author_pubkey == author_pubkey
                && coord.author_owner_grant == *grant
                && coord.stream_id == stream_id
        })
    }

    pub(crate) fn raw_stream_tip(
        &self,
        author_pubkey: &str,
        grant: &MembershipGrantId,
        stream_id: AuthorStreamId,
    ) -> Option<MembershipCoord> {
        self.coords
            .iter()
            .filter(|coord| {
                coord.author_pubkey == author_pubkey
                    && coord.author_owner_grant == *grant
                    && coord.stream_id == stream_id
            })
            .max_by_key(|coord| coord.seq)
            .cloned()
    }

    pub(crate) fn next_member_grant_id_in_stream(
        &self,
        signer: &UserKeypair,
        stream_id: AuthorStreamId,
        user_pubkey: &str,
    ) -> Result<MembershipGrantId, MembershipError> {
        let author = keys::public_key_hex(signer);
        let author_grant = self
            .active_owner_grant(&author)
            .ok_or_else(|| MembershipError::SignerIsNotOwner(author.clone()))?;
        let (seq, _) = self.next_stream_position(&author, &author_grant, stream_id)?;
        Ok(derive_grant_id(
            self.store_id().expect("validated chain has a store id"),
            &author,
            &author_grant,
            stream_id,
            seq,
            user_pubkey,
        ))
    }

    pub fn next_stream_position(
        &self,
        author: &str,
        grant: &MembershipGrantId,
        stream_id: AuthorStreamId,
    ) -> Result<(u64, Option<ObjectHash>), MembershipError> {
        let raw_tip = self.raw_stream_tip(author, grant, stream_id);
        let effective_tip = self.stream_tip(author, grant, stream_id);
        if raw_tip != effective_tip {
            return Err(MembershipError::PrunedAuthorStream);
        }
        effective_tip.map_or(Ok((1, None)), |tip| {
            tip.seq
                .checked_add(1)
                .map(|seq| (seq, Some(tip.entry_hash)))
                .ok_or(MembershipError::SequenceExhausted)
        })
    }

    fn validate_entry_authenticity(
        index: usize,
        entry: &MembershipEntry,
    ) -> Result<(), MembershipError> {
        if verify_membership_entry(entry) {
            Ok(())
        } else {
            Err(MembershipError::InvalidSignature(index))
        }
    }

    fn reduce(
        entries: &[MembershipEntry],
        coords: &[MembershipCoord],
        provider_admin_genesis: &crate::provider::ProviderAdminState,
    ) -> Result<(BTreeSet<MembershipCoord>, ResolvedStoreMembership), MembershipError> {
        let expected_store = entries
            .first()
            .ok_or(MembershipError::EmptyChain)?
            .store_id
            .clone();
        if expected_store.is_empty() {
            return Err(MembershipError::InvalidFounder);
        }

        for (index, (coord, entry)) in coords.iter().zip(entries).enumerate() {
            if entry.require_version().is_err() {
                return Err(MembershipError::UnsupportedVersion(index));
            }
            if entry.store_id != expected_store {
                return Err(MembershipError::StoreMismatch {
                    index,
                    expected: expected_store.clone(),
                    actual: entry.store_id.clone(),
                });
            }
            let actual = entry.coord();
            if *coord != actual {
                return Err(MembershipError::CoordinateMismatch {
                    index,
                    expected: Box::new(coord.clone()),
                    actual: Box::new(actual),
                });
            }
            if !entry
                .dependencies
                .windows(2)
                .all(|pair| pair[0].stream_key() < pair[1].stream_key())
            {
                return Err(MembershipError::NonCanonicalDependencyFrontier { index });
            }
            let (barriers, retirement_device_state) = match &entry.change {
                StoreAuthorityChange::SetMember {
                    user_pubkey,
                    role,
                    grant_id,
                    replaces,
                    membership,
                    retirement_barriers,
                    retirement_device_state,
                    ..
                } => {
                    let valid_owner_origin = match role {
                        StoreMembershipRoleGrant::Owner {
                            recovery: OwnerRecoveryAnchorRef::Promotion { acceptance },
                        } => {
                            let request = &acceptance.request;
                            let anchors_match =
                                Some(&acceptance.anchors.membership) == membership.as_ref();
                            let finalization_matches = request.finalization.author_stream
                                == entry.stream_id
                                && request.finalization.seq == entry.seq
                                && request.finalization.previous_hash == entry.previous_hash;
                            request.member_pubkey == *user_pubkey
                                && replaces.len() == 1
                                && replaces.contains(&request.member_grant)
                                && request.intended_owner_grant == *grant_id
                                && request.promoter_owner_grant == entry.author_owner_grant
                                && anchors_match
                                && finalization_matches
                        }
                        StoreMembershipRoleGrant::Owner { .. } => false,
                        StoreMembershipRoleGrant::Member | StoreMembershipRoleGrant::Follower => {
                            membership.is_none()
                        }
                    };
                    if role.is_owner()
                        != membership.as_ref().is_some_and(|anchor| {
                            store_membership_anchor_stream(user_pubkey, grant_id, anchor).is_some()
                        })
                        || !valid_owner_origin
                    {
                        return Err(MembershipError::InvalidOwnerMembershipAnchor(index));
                    }
                    (retirement_barriers, retirement_device_state)
                }
                StoreAuthorityChange::RemoveMember {
                    retirement_barriers,
                    retirement_device_state,
                    ..
                } => (retirement_barriers, retirement_device_state),
                StoreAuthorityChange::ProviderAdmin => {
                    let Some(crate::provider::ProviderAdminMembershipChange {
                        owner_barriers, ..
                    }) = &entry.provider_admin
                    else {
                        return Err(MembershipError::InvalidProviderAdminChange(index));
                    };
                    if owner_barriers.values().any(|barrier| {
                        !barrier
                            .observed_streams
                            .windows(2)
                            .all(|pair| pair[0].stream_key() < pair[1].stream_key())
                    }) {
                        return Err(MembershipError::InvalidProviderAdminChange(index));
                    }
                    continue;
                }
                StoreAuthorityChange::DeviceRegistrationActivation { .. }
                | StoreAuthorityChange::DeviceExclusionProposal { .. }
                | StoreAuthorityChange::DeviceExclusionOutcome { .. } => {
                    if entry.provider_admin.is_some() {
                        return Err(MembershipError::InvalidProviderAdminChange(index));
                    }
                    continue;
                }
                StoreAuthorityChange::Founder { .. } => continue,
            };
            if entry.provider_admin.is_some() {
                return Err(MembershipError::InvalidProviderAdminChange(index));
            }
            let owner_recoveries = barriers
                .values()
                .filter_map(|barrier| match barrier {
                    MergeMembershipGrantRetirementBarrier::Owner { barrier } => {
                        Some(&barrier.recovery)
                    }
                    MergeMembershipGrantRetirementBarrier::NonOwner { .. } => None,
                })
                .collect::<Vec<_>>();
            match (owner_recoveries.is_empty(), retirement_device_state) {
                (true, None) => {}
                (false, Some(state))
                    if owner_recoveries
                        .iter()
                        .all(|cursor| state.recovery().binary_search(cursor).is_ok()) => {}
                (true, Some(_)) => return Err(MembershipError::UnexpectedOwnerRecoveryState),
                (false, None | Some(_)) => return Err(MembershipError::MissingOwnerRecoveryState),
            }
            if let Some((grant, _)) = barriers.iter().find(|(_, barrier)| {
                !barrier
                    .author_streams()
                    .observed_streams
                    .windows(2)
                    .all(|pair| pair[0].stream_key() < pair[1].stream_key())
            }) {
                return Err(MembershipError::InvalidOwnerRevocationBarrier {
                    index,
                    grant: grant.clone(),
                });
            }
        }

        let founders = entries
            .iter()
            .filter_map(|entry| {
                let StoreAuthorityChange::Founder {
                    owner_pubkey,
                    owner_grant_id,
                    ..
                } = &entry.change
                else {
                    return None;
                };
                Some((entry, owner_pubkey, owner_grant_id))
            })
            .collect::<Vec<_>>();
        let [(founder, owner_pubkey, owner_grant_id)] = founders.as_slice() else {
            return Err(MembershipError::InvalidFounder);
        };
        if founder.author_pubkey != **owner_pubkey
            || founder.author_owner_grant != **owner_grant_id
            || founder.stream_id != derive_founder_stream_id(&founder.store_id, owner_pubkey)
            || founder.provider_admin.is_some()
        {
            return Err(MembershipError::InvalidFounder);
        }

        validate_provider_admin_controls(entries)?;
        validate_membership_retirement_barriers(entries)?;
        validate_membership_wrapped_keys(entries)?;
        let reduced = reduce_store_membership(entries)?;
        let provider_admin = crate::provider::ProviderAdminState::reduce_merge(
            provider_admin_genesis,
            entries,
            &reduced.included,
        )?;
        let resolved = resolved_store_membership(&reduced, provider_admin, entries)?;
        Ok((reduced.included, resolved))
    }

    #[cfg(any(test, feature = "test-utils"))]
    pub fn from_entries(entries: Vec<MembershipEntry>) -> Result<Self, MembershipError> {
        let provider_admin = test_provider_admin_genesis(&entries)?;
        Self::from_entries_with_coords_and_provider_admin(
            entries
                .into_iter()
                .map(|entry| (entry.coord(), entry))
                .collect(),
            provider_admin,
        )
    }

    #[cfg(test)]
    pub(crate) fn from_entries_with_coords_and_heads(
        entries: Vec<(MembershipCoord, MembershipEntry)>,
        heads: Vec<(MembershipHeadRef, AuthorHead)>,
    ) -> Result<Self, MembershipError> {
        let values = entries
            .iter()
            .map(|(_, entry)| entry.clone())
            .collect::<Vec<_>>();
        let provider_admin = test_provider_admin_genesis(&values)?;
        Self::from_entries_with_coords_and_heads_and_provider_admin(entries, heads, provider_admin)
    }

    #[cfg(any(test, feature = "test-utils"))]
    pub(crate) fn from_entries_with_coords_and_provider_admin(
        entries: Vec<(MembershipCoord, MembershipEntry)>,
        provider_admin: crate::provider::ProviderAdminState,
    ) -> Result<Self, MembershipError> {
        Self::from_entries_with_coords_and_head_refs(entries, Vec::new(), provider_admin)
    }

    /// Raw signed coverage: the greatest loaded coordinate in every stream,
    /// including suffixes removed by causal pruning.
    #[cfg(test)]
    pub fn author_heads(&self) -> Vec<MembershipCoord> {
        causal_grants::stream_frontier(self.coords.iter().cloned())
    }
}