use alloc::boxed::Box;
use crate::wire::{
AttachSecret, ClosureCheckedEnvelope, EnrollmentEnvelope, EnrollmentRequest,
EnrollmentResponse, OrderAllocatingEnvelope, SequenceAllocatingEnvelope,
};
use super::super::{
AllocatedParticipantSlot, AttachedRecordPosition, BindingSlotDecision, BindingSlotOccupancy,
BindingState, ConnectionConversationCapacityCommit, ConnectionConversationTracking,
EnrollmentCapacityCommit, EnrollmentCapacityCounters, EnrollmentCapacityDecision,
EnrollmentCommit, EnrollmentCommitError, EnrollmentCommitParameters, EnrollmentFingerprint,
EnrollmentLookupResult, EnrollmentTokenPhase, InitialEnrollmentClosureError,
InitialEnrollmentClosureInput, InitialEnrollmentClosureProjection, ObserverCheckedOperation,
ObserverFloorDecision, ObserverFloorPermit, OrderAdmissionError, OrderAllocation,
ParticipantSlotAllocationError, ParticipantSlotAllocatorProof, RemainingClosureDecision,
RemainingClosurePermit, SemanticConnectionCapacityDecision, SequenceAdmission,
SequenceAdmissionError, admit_sequence, allocate_order, check_observer_floor,
commit_enrollment, lookup_enrollment, project_initial_enrollment_closure,
select_enrollment_binding_slot, select_enrollment_capacity,
select_semantic_connection_capacity,
};
pub struct InitialEnrollmentOperationInput<'a, EF, V, LF> {
request: &'a EnrollmentRequest,
token_phase: EnrollmentTokenPhase<'a, EF, V, LF>,
lookup_binding: &'a BindingState,
connection_tracking: ConnectionConversationTracking,
connection_capacity: super::super::CapacityCounter,
binding_occupancy: BindingSlotOccupancy,
enrollment_capacity: EnrollmentCapacityCounters,
closure: InitialEnrollmentClosureInput,
}
impl<'a, EF, V, LF> InitialEnrollmentOperationInput<'a, EF, V, LF> {
#[allow(clippy::too_many_arguments)]
#[must_use]
pub const fn new(
request: &'a EnrollmentRequest,
token_phase: EnrollmentTokenPhase<'a, EF, V, LF>,
lookup_binding: &'a BindingState,
connection_tracking: ConnectionConversationTracking,
connection_capacity: super::super::CapacityCounter,
binding_occupancy: BindingSlotOccupancy,
enrollment_capacity: EnrollmentCapacityCounters,
closure: InitialEnrollmentClosureInput,
) -> Self {
Self {
request,
token_phase,
lookup_binding,
connection_tracking,
connection_capacity,
binding_occupancy,
enrollment_capacity,
closure,
}
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub struct ReceiptDeadlines {
receipt_expires_at: u128,
provenance_expires_at: u128,
}
impl ReceiptDeadlines {
pub fn try_from_ttls(
now_ms: u64,
attach_receipt_ttl_ms: u64,
receipt_provenance_ttl_ms: u64,
) -> Result<Self, ReceiptDeadlineError> {
if attach_receipt_ttl_ms == 0 {
return Err(ReceiptDeadlineError::ZeroAttachReceiptTtl);
}
if receipt_provenance_ttl_ms == 0 {
return Err(ReceiptDeadlineError::ZeroReceiptProvenanceTtl);
}
if receipt_provenance_ttl_ms < attach_receipt_ttl_ms {
return Err(ReceiptDeadlineError::ProvenanceTtlShorterThanReceipt {
attach_receipt_ttl_ms,
receipt_provenance_ttl_ms,
});
}
let widened_now = u128::from(now_ms);
Ok(Self {
receipt_expires_at: widened_now + u128::from(attach_receipt_ttl_ms),
provenance_expires_at: widened_now + u128::from(receipt_provenance_ttl_ms),
})
}
pub const fn try_from_absolute(
receipt_expires_at: u128,
provenance_expires_at: u128,
) -> Result<Self, ReceiptDeadlineError> {
if receipt_expires_at == 0 {
return Err(ReceiptDeadlineError::ZeroAbsoluteReceiptDeadline);
}
if provenance_expires_at < receipt_expires_at {
return Err(ReceiptDeadlineError::AbsoluteProvenanceBeforeReceipt);
}
Ok(Self {
receipt_expires_at,
provenance_expires_at,
})
}
#[must_use]
pub const fn receipt_expires_at(self) -> u128 {
self.receipt_expires_at
}
#[must_use]
pub const fn provenance_expires_at(self) -> u128 {
self.provenance_expires_at
}
}
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ReceiptDeadlineError {
ZeroAttachReceiptTtl,
ZeroReceiptProvenanceTtl,
ProvenanceTtlShorterThanReceipt {
attach_receipt_ttl_ms: u64,
receipt_provenance_ttl_ms: u64,
},
ZeroAbsoluteReceiptDeadline,
AbsoluteProvenanceBeforeReceipt,
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InitialEnrollmentCommitValues<F> {
attach_secret: AttachSecret,
deadlines: ReceiptDeadlines,
enrollment_fingerprint: EnrollmentFingerprint<F>,
}
impl<F> InitialEnrollmentCommitValues<F> {
#[must_use]
pub const fn new(
attach_secret: AttachSecret,
deadlines: ReceiptDeadlines,
enrollment_fingerprint: EnrollmentFingerprint<F>,
) -> Self {
Self {
attach_secret,
deadlines,
enrollment_fingerprint,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct InitialEnrollmentOperationCommit<F> {
enrollment: EnrollmentCommit<F>,
connection_capacity: ConnectionConversationCapacityCommit,
enrollment_capacity: EnrollmentCapacityCommit,
order: OrderAllocation,
sequence: SequenceAdmission,
observer_floor: ObserverFloorPermit,
closure_permit: Box<RemainingClosurePermit>,
closure_projection: InitialEnrollmentClosureProjection,
}
impl<F> InitialEnrollmentOperationCommit<F> {
#[must_use]
pub const fn enrollment(&self) -> &EnrollmentCommit<F> {
&self.enrollment
}
#[must_use]
pub const fn connection_capacity(&self) -> ConnectionConversationCapacityCommit {
self.connection_capacity
}
#[must_use]
pub const fn enrollment_capacity(&self) -> EnrollmentCapacityCommit {
self.enrollment_capacity
}
#[must_use]
pub const fn order(&self) -> OrderAllocation {
self.order
}
#[must_use]
pub const fn sequence(&self) -> SequenceAdmission {
self.sequence
}
#[must_use]
pub const fn observer_floor(&self) -> ObserverFloorPermit {
self.observer_floor
}
#[must_use]
pub const fn closure_permit(&self) -> &RemainingClosurePermit {
&self.closure_permit
}
#[must_use]
pub const fn closure_projection(&self) -> &InitialEnrollmentClosureProjection {
&self.closure_projection
}
#[must_use]
pub fn into_enrollment(self) -> EnrollmentCommit<F> {
self.enrollment
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum InitialEnrollmentOperationFault {
Closure(InitialEnrollmentClosureError),
Order(OrderAdmissionError),
Sequence(SequenceAdmissionError),
SlotAllocation(ParticipantSlotAllocationError),
AllocatedParticipantMismatch {
expected: u64,
actual: u64,
},
AllocatedIdentityLimitMismatch {
expected: u64,
actual: u64,
},
Commit(EnrollmentCommitError),
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum InitialEnrollmentOperationDecision<F> {
Respond(EnrollmentResponse),
Commit(Box<InitialEnrollmentOperationCommit<F>>),
Fault(InitialEnrollmentOperationFault),
}
struct InitialEnrollmentCapacityPermits {
connection: ConnectionConversationCapacityCommit,
enrollment: EnrollmentCapacityCommit,
}
struct OrderedInitialEnrollment {
projection: InitialEnrollmentClosureProjection,
order: OrderAllocation,
}
struct ClosedInitialEnrollment {
projection: InitialEnrollmentClosureProjection,
order: OrderAllocation,
sequence: SequenceAdmission,
observer_floor: ObserverFloorPermit,
closure_permit: Box<RemainingClosurePermit>,
}
enum InitialEnrollmentGateFailure {
Respond(Box<EnrollmentResponse>),
Fault(Box<InitialEnrollmentOperationFault>),
}
impl InitialEnrollmentGateFailure {
fn respond(value: EnrollmentResponse) -> Self {
Self::Respond(Box::new(value))
}
fn fault(value: InitialEnrollmentOperationFault) -> Self {
Self::Fault(Box::new(value))
}
fn into_decision<F>(self) -> InitialEnrollmentOperationDecision<F> {
match self {
Self::Respond(value) => InitialEnrollmentOperationDecision::Respond(*value),
Self::Fault(value) => InitialEnrollmentOperationDecision::Fault(*value),
}
}
}
#[must_use]
pub fn apply_initial_enrollment<EF, V, LF, F, P, A, M>(
input: &InitialEnrollmentOperationInput<'_, EF, V, LF>,
mint_commit_values: M,
allocate_participant: A,
) -> InitialEnrollmentOperationDecision<F>
where
P: ParticipantSlotAllocatorProof,
A: FnOnce() -> Result<AllocatedParticipantSlot<P>, ParticipantSlotAllocationError>,
M: FnOnce() -> InitialEnrollmentCommitValues<F>,
{
let envelope = enrollment_envelope(input.request);
if let Some(response) = initial_enrollment_lookup_response(input) {
return InitialEnrollmentOperationDecision::Respond(response);
}
let capacity = match admit_initial_enrollment_capacity(input, &envelope) {
Ok(value) => value,
Err(error) => return error.into_decision(),
};
let ordered = match plan_initial_enrollment_order(&input.closure, &envelope) {
Ok(value) => value,
Err(error) => return error.into_decision(),
};
let closed = match close_initial_enrollment(ordered, &envelope) {
Ok(value) => value,
Err(error) => return error.into_decision(),
};
commit_initial_enrollment(
input.request,
&capacity,
closed,
mint_commit_values,
allocate_participant,
)
}
fn initial_enrollment_lookup_response<EF, V, LF>(
input: &InitialEnrollmentOperationInput<'_, EF, V, LF>,
) -> Option<EnrollmentResponse> {
match lookup_enrollment(input.token_phase, input.lookup_binding, input.request) {
EnrollmentLookupResult::Retired(value) => Some(EnrollmentResponse::from_retired(value)),
EnrollmentLookupResult::Bound(value) => Some(EnrollmentResponse::from_bound(value)),
EnrollmentLookupResult::UnboundReceipt(value) => {
Some(EnrollmentResponse::from_unbound_receipt(value))
}
EnrollmentLookupResult::ReceiptExpired(value) => {
Some(EnrollmentResponse::from_receipt_expired(value))
}
EnrollmentLookupResult::EnrollmentKnown(value) => {
Some(EnrollmentResponse::enrollment_known(value))
}
EnrollmentLookupResult::AuthorizedNew => None,
}
}
fn admit_initial_enrollment_capacity<EF, V, LF>(
input: &InitialEnrollmentOperationInput<'_, EF, V, LF>,
envelope: &EnrollmentEnvelope,
) -> Result<InitialEnrollmentCapacityPermits, InitialEnrollmentGateFailure> {
let connection = match select_semantic_connection_capacity(
input.connection_tracking,
input.connection_capacity,
) {
SemanticConnectionCapacityDecision::Commit(value) => value,
SemanticConnectionCapacityDecision::Respond { limit } => {
return Err(InitialEnrollmentGateFailure::respond(
EnrollmentResponse::connection_conversation_capacity_exceeded(
envelope.clone(),
limit,
),
));
}
};
if let BindingSlotDecision::Respond(value) =
select_enrollment_binding_slot(input.request, input.binding_occupancy)
{
return Err(InitialEnrollmentGateFailure::respond(value));
}
let enrollment = match select_enrollment_capacity(input.request, input.enrollment_capacity) {
EnrollmentCapacityDecision::Commit(value) => value,
EnrollmentCapacityDecision::Respond(value) => {
return Err(InitialEnrollmentGateFailure::respond(value));
}
};
Ok(InitialEnrollmentCapacityPermits {
connection,
enrollment,
})
}
fn plan_initial_enrollment_order(
closure: &InitialEnrollmentClosureInput,
envelope: &EnrollmentEnvelope,
) -> Result<OrderedInitialEnrollment, InitialEnrollmentGateFailure> {
let projection = project_initial_enrollment_closure(*closure).map_err(|error| {
InitialEnrollmentGateFailure::fault(InitialEnrollmentOperationFault::Closure(error))
})?;
let order_plan = projection
.plan_order()
.map_err(initial_enrollment_order_failure)?;
let order = allocate_order(
OrderAllocatingEnvelope::Enrollment(envelope.clone()),
projection.current_order(),
order_plan,
)
.map_err(initial_enrollment_order_failure)?;
Ok(OrderedInitialEnrollment { projection, order })
}
fn initial_enrollment_order_failure(error: OrderAdmissionError) -> InitialEnrollmentGateFailure {
match error {
OrderAdmissionError::Exhausted(value) => InitialEnrollmentGateFailure::respond(
EnrollmentResponse::from_conversation_order_exhausted(value),
),
other => InitialEnrollmentGateFailure::fault(InitialEnrollmentOperationFault::Order(other)),
}
}
fn close_initial_enrollment(
ordered: OrderedInitialEnrollment,
envelope: &EnrollmentEnvelope,
) -> Result<ClosedInitialEnrollment, InitialEnrollmentGateFailure> {
let sequence_plan = ordered
.projection
.plan_sequence()
.map_err(initial_enrollment_sequence_failure)?;
let sequence = admit_sequence(
SequenceAllocatingEnvelope::Enrollment(envelope.clone()),
sequence_plan,
)
.map_err(initial_enrollment_sequence_failure)?;
let observer_floor = match check_observer_floor(
ObserverCheckedOperation::Enrollment(envelope.clone()),
ordered.projection.observer_progress(),
ordered.projection.resulting_floor(),
) {
ObserverFloorDecision::Eligible(value) => value,
ObserverFloorDecision::Respond(value) => {
return Err(InitialEnrollmentGateFailure::respond(
EnrollmentResponse::from_observer_backpressure(value),
));
}
};
let closure_permit = match ordered
.projection
.remaining_closure_decision(&ClosureCheckedEnvelope::Enrollment(envelope.clone()))
{
RemainingClosureDecision::Eligible(value) => value,
RemainingClosureDecision::Respond(value) => {
return Err(InitialEnrollmentGateFailure::respond(
EnrollmentResponse::from_marker_closure_capacity_exceeded(value),
));
}
};
Ok(ClosedInitialEnrollment {
projection: ordered.projection,
order: ordered.order,
sequence,
observer_floor,
closure_permit,
})
}
fn initial_enrollment_sequence_failure(
error: SequenceAdmissionError,
) -> InitialEnrollmentGateFailure {
match error {
SequenceAdmissionError::Exhausted(value) => InitialEnrollmentGateFailure::respond(
EnrollmentResponse::from_conversation_sequence_exhausted(value),
),
other => {
InitialEnrollmentGateFailure::fault(InitialEnrollmentOperationFault::Sequence(other))
}
}
}
fn commit_initial_enrollment<F, P, A, M>(
request: &EnrollmentRequest,
capacity: &InitialEnrollmentCapacityPermits,
closed: ClosedInitialEnrollment,
mint_commit_values: M,
allocate_participant: A,
) -> InitialEnrollmentOperationDecision<F>
where
P: ParticipantSlotAllocatorProof,
A: FnOnce() -> Result<AllocatedParticipantSlot<P>, ParticipantSlotAllocationError>,
M: FnOnce() -> InitialEnrollmentCommitValues<F>,
{
let participant_slot = match allocate_participant() {
Ok(value) => value,
Err(error) => {
return InitialEnrollmentOperationDecision::Fault(
InitialEnrollmentOperationFault::SlotAllocation(error),
);
}
};
if participant_slot.participant_id() != closed.projection.participant_index() {
return InitialEnrollmentOperationDecision::Fault(
InitialEnrollmentOperationFault::AllocatedParticipantMismatch {
expected: closed.projection.participant_index(),
actual: participant_slot.participant_id(),
},
);
}
if participant_slot.identity_limit() != closed.projection.identity_slots() {
return InitialEnrollmentOperationDecision::Fault(
InitialEnrollmentOperationFault::AllocatedIdentityLimitMismatch {
expected: closed.projection.identity_slots(),
actual: participant_slot.identity_limit(),
},
);
}
let commit_values = mint_commit_values();
let enrollment = match commit_enrollment(
request,
EnrollmentCommitParameters {
allocated_slot: participant_slot,
attach_secret: commit_values.attach_secret,
origin_binding_epoch: closed.projection.binding_epoch(),
attached_position: AttachedRecordPosition::new(
closed.order.major(),
closed.sequence.resulting().high_watermark(),
),
receipt_expires_at: commit_values.deadlines.receipt_expires_at(),
provenance_expires_at: commit_values.deadlines.provenance_expires_at(),
enrollment_fingerprint: commit_values.enrollment_fingerprint,
},
) {
Ok(value) => value,
Err(error) => {
return InitialEnrollmentOperationDecision::Fault(
InitialEnrollmentOperationFault::Commit(error),
);
}
};
InitialEnrollmentOperationDecision::Commit(Box::new(InitialEnrollmentOperationCommit {
enrollment,
connection_capacity: capacity.connection,
enrollment_capacity: capacity.enrollment,
order: closed.order,
sequence: closed.sequence,
observer_floor: closed.observer_floor,
closure_permit: closed.closure_permit,
closure_projection: closed.projection,
}))
}
const fn enrollment_envelope(request: &EnrollmentRequest) -> EnrollmentEnvelope {
EnrollmentEnvelope {
conversation_id: request.conversation_id,
enrollment_token: request.enrollment_token,
}
}