frame-conv 0.2.0

//! The subscription pattern (`frame:conv-subscription@v1`, F-3a R3): typed
//! events at the proven delivery strength, plus the caller-owned cursor
//! commitment lever.

use std::time::{Duration, Instant};

use liminal_protocol::wire::{ClientRequest, ParticipantAck, ServerValue};
use serde::Serialize;
use serde::de::DeserializeOwned;

use crate::envelope::Envelope;
use crate::error::{CallError, CursorCommit, PublishError};
use crate::id::{ConversationSeq, CorrelationId, MessageKind, PatternId};
use crate::outcome::{PublishReceipt, SubscriptionItem};
use crate::seam::classify_refusal;
use crate::store::ResumeStore;

use super::pump::{PumpStep, submit_to_publish};
use super::state::{ConversationHandle, QueuedItem};

impl<S: ResumeStore> ConversationHandle<S> {
    /// Publishes a typed subscription event into the conversation.
    /// Schema-invalid values are refused typed BEFORE the wire (F-3a R1).
    /// The receipt is the correlated commit answer — delivery evidence for
    /// the record's admission, NOT evidence any subscriber consumed it
    /// (finding G1's discipline; the stream carries news, never
    /// obligations).
    ///
    /// # Errors
    ///
    /// Returns typed publish failures.
    pub fn publish_event<E: Serialize>(
        &mut self,
        body: &E,
    ) -> Result<PublishReceipt, PublishError> {
        let envelope = Envelope::from_typed(
            PatternId::Subscription,
            CorrelationId::mint(),
            MessageKind::Event,
            body,
        )
        .map_err(|refusal| PublishError::SchemaInvalid {
            detail: refusal.detail,
        })?;
        let bytes = envelope
            .encode()
            .map_err(|refusal| PublishError::Protocol {
                detail: refusal.detail,
            })?;
        self.admit(bytes)
    }

    /// Waits up to `wait` for the next subscription item. `Ok(None)` is a
    /// benign quiet wait — never an error, never a protocol outcome (the
    /// assertion-8 pin at this crate's surface). Items carry the delivery
    /// guarantee documented on [`SubscriptionItem`]: ordered, contiguous,
    /// sender-excluded within a live session; membership-forward at join;
    /// typed gap surface with resume as the documented resync path.
    ///
    /// The effective wait granularity is the substrate's IO quantum
    /// (hardcoded 5 s at published liminal 0.3.0 — upstream ASK-2): a wait
    /// smaller than one quantum may still block for one quantum.
    ///
    /// # Errors
    ///
    /// Returns a typed connection fate.
    pub fn next_event<E: DeserializeOwned>(
        &mut self,
        wait: Duration,
    ) -> Result<Option<SubscriptionItem<E>>, CallError> {
        let started = Instant::now();
        loop {
            if let Some(item) = self.events_inbox.pop_front() {
                return Ok(Some(convert_item(item)));
            }
            if started.elapsed() >= wait {
                return Ok(None);
            }
            match self.pump_step() {
                Ok(PumpStep::Classified | PumpStep::Quiet) => {}
                Err(detail) => return Err(CallError::ConnectionLost { detail }),
            }
        }
    }

    /// Commits the caller-owned consumption cursor through `through`,
    /// cumulatively. Acking is explicit and caller-owned — this crate never
    /// acks on the caller's behalf, because the committed cursor is the
    /// caller's lever on resume replay volume (F-0c §R4).
    ///
    /// # Errors
    ///
    /// Returns typed publish-path failures; the ack answers themselves
    /// (committed / no-op / gap / regression) are the typed
    /// [`CursorCommit`] outcome, not errors.
    pub fn commit_cursor(
        &mut self,
        through: ConversationSeq,
    ) -> Result<CursorCommit, PublishError> {
        let answer = self
            .submit(ClientRequest::ParticipantAck(ParticipantAck {
                conversation_id: self.conversation.to_wire(),
                participant_id: self.me.to_wire(),
                capability_generation: self.generation,
                through_seq: through.value(),
            }))
            .map_err(submit_to_publish)?;
        match answer {
            ServerValue::AckCommitted(committed) => Ok(CursorCommit::Committed {
                through: ConversationSeq::new(committed.current_cursor()),
            }),
            ServerValue::AckNoOp(noop) => Ok(CursorCommit::NoOp {
                current: ConversationSeq::new(noop.current_cursor()),
            }),
            ServerValue::AckGap(gap) => Ok(CursorCommit::Gap {
                requested: ConversationSeq::new(gap.request().through_seq),
                current: ConversationSeq::new(gap.current_cursor()),
            }),
            ServerValue::AckRegression(regression) => Ok(CursorCommit::Regression {
                requested: ConversationSeq::new(regression.request().through_seq),
            }),
            other => {
                let (class, detail) = classify_refusal(&other);
                Err(PublishError::Refused { class, detail })
            }
        }
    }
}

/// Converts a queued item into the typed subscription surface, decoding
/// event payloads against the caller's typed message (schema-invalid
/// surfaces typed — F-3a R1).
fn convert_item<E: DeserializeOwned>(item: QueuedItem) -> SubscriptionItem<E> {
    match item {
        QueuedItem::Event {
            envelope,
            publisher,
            seq,
        } => match envelope.decode_payload::<E>() {
            Ok(body) => SubscriptionItem::Event {
                body,
                publisher,
                seq,
            },
            Err(refusal) => SubscriptionItem::SchemaInvalid {
                publisher,
                seq,
                detail: refusal.detail,
            },
        },
        QueuedItem::Joined { peer, seq } => SubscriptionItem::PeerJoined { peer, seq },
        QueuedItem::Departed { peer, seq, reason } => {
            SubscriptionItem::PeerDeparted { peer, seq, reason }
        }
        QueuedItem::Failed { peer, seq, failure } => {
            SubscriptionItem::PeerFailed { peer, seq, failure }
        }
        QueuedItem::Compacted { seq } => SubscriptionItem::HistoryCompacted { seq },
        QueuedItem::Gap { expected, observed } => SubscriptionItem::Gap { expected, observed },
    }
}