liminal_server/auth_membership/

pages.rs

use super::{
    BTreeSet, CONTRACT_VERSION, ChannelPage, MembershipAction, MembershipGate, MembershipPage,
    MembershipRefusal, MembershipSubject, Mode, PageCursor, PageOutcome, PageParts, PageQuestion,
    RecipientPage, RecipientPageRequest, Resource, ResourcePageRequest, bump, bump_by,
    continuation, encoded_bytes, page_refusal,
};

impl MembershipGate {
    /// One page of the protected channels `holder` may `action` within
    /// `workspace` (ACCESS-006 R4): both bounds the operator's, every row
    /// checked against the question, the continuation bound to it.
    ///
    /// # Errors
    /// The exact [`MembershipRefusal`]: no verified holder, an unmapped
    /// workspace, the authority's state, a foreign or stale continuation,
    /// the provider's failure or refusal, or a page that is not the one
    /// asked (over bound, foreign or duplicate row, inconsistent counts).
    pub fn resource_page(
        &self,
        holder: Option<&MembershipSubject>,
        workspace: &str,
        action: MembershipAction,
        after: Option<&PageCursor>,
    ) -> Result<ChannelPage, MembershipRefusal> {
        let holder = holder.ok_or(MembershipRefusal::HolderUnverified)?;
        if !self.map.workspaces.contains(workspace) {
            return Err(MembershipRefusal::WorkspaceNotMapped);
        }
        let (revision, generation) = self.ready()?;
        let question = PageQuestion::Resources {
            workspace: workspace.to_owned(),
            subject: holder.clone(),
            action,
        };
        let request = ResourcePageRequest {
            contract: CONTRACT_VERSION,
            log: self.served.clone(),
            workspace: Resource {
                kind: self.workspace_kind.clone(),
                id: workspace.to_owned(),
            },
            subject: holder.to_contract(),
            kind: self.channel_kind.clone(),
            action: action.as_str().to_owned(),
            at_least: revision,
            bounds: self.bounds,
            after: continuation(after, &question, generation)?,
        };
        bump(&self.counters.pages_asked);
        let page = self
            .provider
            .resources(&request)
            .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
        let parts = self.page_parts(&page, revision)?;
        let mut channels = Vec::with_capacity(parts.rows.len());
        let mut seen = BTreeSet::new();
        let mut held = 0_u64;
        let mut unmapped = 0_u64;
        for row in parts.rows {
            if !holder.is(&row.subject) || row.resource.kind != self.channel_kind {
                return Err(MembershipRefusal::PageForeignRow);
            }
            if !seen.insert(row.resource.id.as_str()) {
                return Err(MembershipRefusal::PageDuplicateRow);
            }
            let Some(channel) = self.map.by_resource.get(&row.resource.id) else {
                unmapped = unmapped.saturating_add(1);
                continue;
            };
            let placed_here = self
                .map
                .channels
                .get(channel)
                .is_some_and(|mapped| mapped.workspace.id == workspace);
            if !placed_here {
                return Err(MembershipRefusal::PageForeignRow);
            }
            if row.mode == Mode::Outright {
                channels.push(channel.clone());
            } else {
                held = held.saturating_add(1);
            }
        }
        let current = self.stream_fence(generation)?;
        if parts.revision < current {
            return Err(MembershipRefusal::RevisionStale {
                required: current,
                decided: Some(parts.revision),
            });
        }
        Ok(ChannelPage {
            returned: u64::try_from(parts.rows.len())
                .map_err(|_| MembershipRefusal::PageInconsistent)?,
            channels,
            skipped: parts.skipped,
            held,
            unmapped,
            bytes: parts.bytes,
            revision: parts.revision,
            complete: parts.complete,
            next: parts.next.map(|token| PageCursor {
                token,
                question,
                generation,
            }),
        })
    }

    /// One page of the subjects who may `action` on the protected `channel`
    /// (ACCESS-006 R4), checked as [`Self::resource_page`] is.
    ///
    /// # Errors
    /// As for [`Self::resource_page`], with
    /// [`MembershipRefusal::ChannelNotMapped`] for an unmapped channel.
    pub fn recipient_page(
        &self,
        channel: &str,
        action: MembershipAction,
        after: Option<&PageCursor>,
    ) -> Result<RecipientPage, MembershipRefusal> {
        let mapped = self
            .map
            .channels
            .get(channel)
            .ok_or(MembershipRefusal::ChannelNotMapped)?;
        let (revision, generation) = self.ready()?;
        let question = PageQuestion::Recipients {
            channel: channel.to_owned(),
            action,
        };
        let request = RecipientPageRequest {
            contract: CONTRACT_VERSION,
            log: self.served.clone(),
            workspace: mapped.workspace.clone(),
            resource: mapped.resource.clone(),
            action: action.as_str().to_owned(),
            at_least: revision,
            bounds: self.bounds,
            after: continuation(after, &question, generation)?,
        };
        bump(&self.counters.pages_asked);
        let page = self
            .provider
            .recipients(&request)
            .map_err(|error| MembershipRefusal::Unavailable { cause: error.cause })?;
        let parts = self.page_parts(&page, revision)?;
        let mut recipients = Vec::with_capacity(parts.rows.len());
        let mut seen = BTreeSet::new();
        let mut held = 0_u64;
        for row in parts.rows {
            if row.resource != mapped.resource {
                return Err(MembershipRefusal::PageForeignRow);
            }
            if !seen.insert((row.subject.id.as_str(), row.subject.kind.as_str())) {
                return Err(MembershipRefusal::PageDuplicateRow);
            }
            if row.mode == Mode::Outright {
                recipients.push(row.subject.clone());
            } else {
                held = held.saturating_add(1);
            }
        }
        let current = self.stream_fence(generation)?;
        if parts.revision < current {
            return Err(MembershipRefusal::RevisionStale {
                required: current,
                decided: Some(parts.revision),
            });
        }
        Ok(RecipientPage {
            returned: u64::try_from(parts.rows.len())
                .map_err(|_| MembershipRefusal::PageInconsistent)?,
            recipients,
            skipped: parts.skipped,
            held,
            bytes: parts.bytes,
            revision: parts.revision,
            complete: parts.complete,
            next: parts.next.map(|token| PageCursor {
                token,
                question,
                generation,
            }),
        })
    }

    pub(super) fn page_parts<'p>(
        &self,
        page: &'p MembershipPage,
        at_least: u64,
    ) -> Result<PageParts<'p>, MembershipRefusal> {
        if page.contract != CONTRACT_VERSION {
            return Err(MembershipRefusal::ContractUnsupported {
                contract: page.contract,
            });
        }
        if page.log != self.served {
            return Err(MembershipRefusal::LogMismatch);
        }
        let (rows, returned, skipped, complete, next) = match &page.outcome {
            PageOutcome::Refused { refusal, .. } => return Err(page_refusal(refusal)),
            PageOutcome::Page {
                rows,
                returned,
                skipped,
                complete,
                next,
            } => (rows, *returned, *skipped, *complete, next),
        };
        let revision = page.revision.ok_or(MembershipRefusal::RevisionStale {
            required: at_least,
            decided: None,
        })?;
        if revision < at_least {
            return Err(MembershipRefusal::RevisionStale {
                required: at_least,
                decided: Some(revision),
            });
        }
        let count = u64::try_from(rows.len()).map_err(|_| MembershipRefusal::PageInconsistent)?;
        bump_by(&self.counters.page_rows, count);
        if count != returned || complete == next.is_some() {
            return Err(MembershipRefusal::PageInconsistent);
        }
        let bytes = encoded_bytes(rows)?;
        if count > u64::from(self.bounds.rows) || bytes > u64::from(self.bounds.bytes) {
            return Err(MembershipRefusal::PageOverBound { rows: count, bytes });
        }
        Ok(PageParts {
            rows,
            skipped,
            complete,
            next: next.clone(),
            revision,
            bytes,
        })
    }
}