liminal-server 0.25.4

//! The durable outbox delivering an approved agent draft's outcome to its holder.

use super::AgentDelivery;
use crate::auth_lys::live::drafts::{DraftError, same_request};
use crate::services::lifecycle::site::executor::OutcomeDelivery;
use liminal::durability::AppendResolution;
use liminal::envelope::PublisherId;
use lys_pass::drafts::{ApprovedDraft, Execution};

impl OutcomeDelivery for AgentDelivery {
    fn deliver(
        &self,
        holder: &str,
        draft: &ApprovedDraft,
        receipt: &Execution,
    ) -> Result<(), DraftError> {
        let guard = self.serial.lock().map_err(|_| DraftError::Store)?;
        let prepared = self
            .journal
            .read(|snapshot| {
                snapshot
                    .prepared
                    .values()
                    .find(|prepared| prepared.draft.as_deref() == Some(&draft.id))
                    .cloned()
            })?
            .ok_or(DraftError::Contract)?;
        if prepared.holder != holder
            || (matches!(receipt, Execution::Executed { .. })
                && !same_request(&prepared.request, draft))
        {
            return Err(DraftError::ReceiptMismatch);
        }
        let Some(agent) = &prepared.agent else {
            return Ok(());
        };
        let inbox = self
            .inboxes
            .lock()
            .map_err(|_| DraftError::Store)?
            .get(agent)
            .cloned()
            .ok_or(DraftError::Contract)?;
        let from = inbox.durable_positions().map_err(|_| DraftError::Store)?;
        let delivery = self.message(agent, draft, receipt, from)?;
        if delivery.agent != *agent {
            return Err(DraftError::Contract);
        }
        if delivery.delivered {
            return Ok(());
        }
        let key = format!("draft:{}:outcome", draft.id);
        match inbox
            .resolve_idempotency_key(&key, &delivery.from)
            .map_err(|_| DraftError::Store)?
        {
            AppendResolution::Committed { .. } => {}
            AppendResolution::Absent => {
                let bytes = serde_json::to_vec(&delivery.message).map_err(|_| DraftError::Store)?;
                if bytes.len() > self.agents.bounds().max_event_bytes {
                    return Err(DraftError::Bound);
                }
                if !inbox
                    .publish_with_idempotency_key(
                        bytes,
                        PublisherId::new(self.exchange.clone()),
                        None,
                        &key,
                    )
                    .map_err(|_| DraftError::Store)?
                    .is_admitted()
                {
                    return Err(DraftError::Store);
                }
            }
        }
        let subject = format!("agents.{agent}.state");
        let mut current = self
            .journal
            .read(|snapshot| snapshot.waiting.get(agent).cloned())?
            .ok_or(DraftError::Contract)?;
        current.gone = self
            .agents
            .presence()
            .connection(agent)
            .map_err(|_| DraftError::Store)?
            .is_none();
        let state = current.snapshot().map_err(|_| DraftError::Store)?;
        if self
            .agents
            .retained()
            .value(&subject)
            .map_err(|_| DraftError::Store)?
            .as_ref()
            != Some(&state)
        {
            self.agents
                .retained()
                .publish(&[(subject, state)])
                .map_err(|_| DraftError::Store)?;
        }
        self.journal.update(|snapshot| {
            snapshot
                .deliveries
                .get_mut(&draft.id)
                .ok_or(DraftError::Contract)?
                .delivered = true;
            Ok(())
        })?;
        drop(guard);
        Ok(())
    }
}