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 {
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,
}),
})
}
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,
})
}
}