liminal-server 0.24.5

use super::handler::ProductionParticipantHandler;
use crate::server::participant::{
    ParticipantConnectionContext, ParticipantConnectionConversations, ParticipantSemanticError,
};
use liminal_protocol::wire::{
    BindingEpoch, CredentialDigest, RecordAdmission, RecordAdmissionResponse, ServerValue,
    SettlementAnswer, SettlementOutcome, SettlementQuery,
};

const fn answer(request: &SettlementQuery, outcome: SettlementOutcome) -> ServerValue {
    ServerValue::SettlementAnswer(SettlementAnswer {
        conversation_id: request.conversation_id,
        participant_id: request.participant_id,
        record_admission_attempt_token: request.record_admission_attempt_token,
        outcome,
    })
}

impl ProductionParticipantHandler {
    pub(super) fn answer_settlement_query(
        &self,
        context: ParticipantConnectionContext,
        conversations: &ParticipantConnectionConversations,
        request: &SettlementQuery,
    ) -> Result<ServerValue, ParticipantSemanticError> {
        let Some(cell) = self.existing_cell(request.conversation_id)? else {
            return Ok(answer(request, SettlementOutcome::Unknown));
        };
        let owner = cell
            .lock()
            .map_err(|_| ParticipantSemanticError::Internal {
                message: "participant settlement owner lock is poisoned".to_owned(),
            })?;
        let Some(authority) = owner.as_ref() else {
            drop(owner);
            return Ok(answer(request, SettlementOutcome::Unknown));
        };
        let facts = self.operation_facts(context, request.conversation_id, conversations)?;
        let admission = RecordAdmission {
            conversation_id: request.conversation_id,
            participant_id: request.participant_id,
            capability_generation: request.capability_generation,
            record_admission_attempt_token: request.record_admission_attempt_token,
            payload: Vec::new(),
        };
        let receiving_epoch =
            BindingEpoch::new(facts.receiving_incarnation, request.capability_generation);
        if let Some(outcome) =
            authority.classify_record_admission_authority(&admission, &facts, receiving_epoch)
        {
            drop(owner);
            return Ok(outcome.value);
        }
        let token = request.record_admission_attempt_token.into_bytes();
        let key = (
            token,
            request.participant_id,
            CredentialDigest::new(request.payload_fingerprint),
        );
        if let Some(delivery_seq) = authority.committed_admissions.get(&key) {
            let value = answer(
                request,
                SettlementOutcome::Committed {
                    delivery_seq: *delivery_seq,
                },
            );
            drop(owner);
            return Ok(value);
        }
        if authority
            .committed_admissions
            .range(
                (
                    token,
                    request.participant_id,
                    CredentialDigest::new([0; 32]),
                )
                    ..=(
                        token,
                        request.participant_id,
                        CredentialDigest::new([0xff; 32]),
                    ),
            )
            .next()
            .is_some()
        {
            drop(owner);
            return Ok(RecordAdmissionResponse::attempt_token_body_conflict(
                request.record_admission_attempt_token,
                request.conversation_id,
                request.participant_id,
                request.capability_generation,
            )
            .into_server_value());
        }
        drop(owner);
        Ok(answer(request, SettlementOutcome::NotCommitted))
    }
}