frame-conv 0.2.0

//! Opening, joining, and resuming conversations: enrollment (implicit
//! create — F-0c assertion 1), and restore-then-reattach resume (F-0c §R4).

use std::collections::{HashMap, VecDeque};

use liminal_protocol::wire::{
    AttachAttemptToken, AttachSecret, ClientRequest, CredentialAttachRequest, EnrollmentRequest,
    EnrollmentToken, Generation, ServerValue,
};
use liminal_sdk::ConnectionPoolConfig;
use liminal_sdk::remote::{RemoteConfig, RemoteParticipantError, RemoteParticipantHandle};

use crate::config::BusAttachment;
use crate::error::AttachError;
use crate::id::{ConversationId, ParticipantRef};
use crate::outcome::{JoinCredential, JoinGrant};
use crate::seam::{StoreAdapter, classify_refusal, mint_token16};
use crate::store::ResumeStore;
use crate::track::{CursorTracker, ExchangeRegistry};

use super::pump::SubmitFailure;
use super::state::ConversationHandle;

/// The participant surface still rides the channel-era remote configuration,
/// which requires a channel name and a string conversation name the
/// participant path never uses (recorded upstream ASK-3 — F-0c-FINDINGS
/// §R2 recipe wart, re-measured at published 0.3.0). These fillers configure
/// nothing and are deliberately NOT caller-facing configuration.
const ASK3_UNUSED_CHANNEL: &str = "frame-conv-ask3-unused";
const ASK3_UNUSED_CONVERSATION: &str = "frame-conv-ask3-unused";

pub(super) fn connect(attachment: &BusAttachment) -> Result<RemoteConfig, AttachError> {
    let config = RemoteConfig::new(
        attachment.endpoint.clone(),
        ASK3_UNUSED_CHANNEL,
        ASK3_UNUSED_CONVERSATION,
        ConnectionPoolConfig::new(
            attachment.session_capacity,
            attachment.operation_window_millis,
            attachment.inbound_buffer,
        ),
    )
    .map_err(|error| AttachError::Endpoint {
        detail: format!("bus configuration rejected: {error}"),
    })?;
    config.connect_tcp().map_err(|error| AttachError::Endpoint {
        detail: format!("bus endpoint unreachable: {error}"),
    })
}

pub(super) fn map_restore_error(error: RemoteParticipantError) -> AttachError {
    match error {
        RemoteParticipantError::ResumeDecode(_)
        | RemoteParticipantError::ResumeRestore(_)
        | RemoteParticipantError::ResumeEncode(_) => AttachError::ResumeRejected {
            detail: format!("{error}"),
        },
        RemoteParticipantError::Storage(_) => AttachError::Store {
            detail: format!("{error}"),
        },
        other => AttachError::Protocol {
            detail: format!("{other}"),
        },
    }
}

pub(crate) fn map_submit_failure(failure: SubmitFailure) -> AttachError {
    match failure {
        SubmitFailure::RecordRefused { detail, .. } | SubmitFailure::State { detail } => {
            AttachError::Protocol { detail }
        }
        SubmitFailure::TransportLost { detail } | SubmitFailure::ConnectionFate { detail } => {
            AttachError::ConnectionLost { detail }
        }
        SubmitFailure::AnswerTimeout { waited } => AttachError::AnswerTimeout { waited },
        SubmitFailure::InboundRefused { detail } => AttachError::Protocol { detail },
    }
}

impl<S: ResumeStore> ConversationHandle<S> {
    /// Builds a handle whose identity fields are placeholders; every
    /// constructor replaces them from a decoded binding receipt (or, on
    /// the resumed-leave path, from the caller's persisted grant) before
    /// the handle escapes.
    pub(super) fn bare(
        sdk: RemoteParticipantHandle<StoreAdapter<S>>,
        conversation: ConversationId,
        attachment: &BusAttachment,
    ) -> Self {
        Self {
            sdk,
            conversation,
            me: ParticipantRef::from_wire(0),
            grant: JoinGrant {
                conversation,
                participant: ParticipantRef::from_wire(0),
                credential: JoinCredential::from_bytes([0; 32]),
                generation: Generation::ONE.get(),
            },
            generation: Generation::ONE,
            answer_window: attachment.answer_window,
            exchanges: ExchangeRegistry::default(),
            replies: HashMap::new(),
            requests_inbox: VecDeque::new(),
            events_inbox: VecDeque::new(),
            publications_inbox: VecDeque::new(),
            last_presented_publication: None,
            deaths: Vec::new(),
            anomalies: Vec::new(),
            counters: crate::anomaly::AnomalyCounters::default(),
            tracker: CursorTracker::default(),
            attached: true,
        }
    }

    /// Opens a NEW conversation: mints its identity and enrolls this
    /// participant as its first member. Creation is implicit at first
    /// enrollment (F-0c assertion 1 — the substrate offers no explicit
    /// create and no create-versus-join testimony; the minting policy is
    /// design §2.4's, carried until a server-issued binding authority
    /// exists).
    ///
    /// # Errors
    ///
    /// Returns typed endpoint, refusal, transport, or answer-window
    /// failures.
    pub fn open(attachment: &BusAttachment, store: S) -> Result<(Self, JoinGrant), AttachError> {
        Self::join(attachment, ConversationId::mint(), store)
    }

    /// Joins an existing conversation (or implicitly creates it — the
    /// substrate cannot distinguish; see [`open`](Self::open)). Joining is
    /// membership-forward: history before the join is neither delivered nor
    /// ackable (the F-0c late-joiner finding).
    ///
    /// # Errors
    ///
    /// Returns typed endpoint, refusal, transport, or answer-window
    /// failures.
    pub fn join(
        attachment: &BusAttachment,
        conversation: ConversationId,
        store: S,
    ) -> Result<(Self, JoinGrant), AttachError> {
        let config = connect(attachment)?;
        let sdk = RemoteParticipantHandle::new(&config, StoreAdapter::new(store))
            .map_err(map_restore_error)?;
        let mut handle = Self::bare(sdk, conversation, attachment);
        let answer = handle
            .submit(ClientRequest::Enrollment(EnrollmentRequest {
                conversation_id: conversation.to_wire(),
                enrollment_token: EnrollmentToken::new(mint_token16()),
            }))
            .map_err(map_submit_failure)?;
        match answer {
            ServerValue::EnrollBound(receipt) => {
                let me = ParticipantRef::from_wire(receipt.participant_id());
                let generation = receipt.origin_binding_epoch().capability_generation;
                let grant = JoinGrant {
                    conversation,
                    participant: me,
                    credential: JoinCredential::from_bytes(receipt.attach_secret().into_bytes()),
                    generation: generation.get(),
                };
                handle.me = me;
                handle.generation = generation;
                handle.grant = grant.clone();
                Ok((handle, grant))
            }
            other => {
                let (class, detail) = classify_refusal(&other);
                Err(AttachError::Refused { class, detail })
            }
        }
    }

    /// Resumes a previously joined participant: restores the aggregate from
    /// the caller's persisted canonical state, then re-attaches with the
    /// grant's credential. Resume is restore-then-reattach, never restore
    /// alone; the credential ROTATES and the returned grant carries the new
    /// one — persist it (F-0c §R4, proven at published 0.3.0).
    ///
    /// On success the substrate replays the unacked obligation window from
    /// the committed cursor — including records admitted while this
    /// participant was dead — which arrives through the normal typed
    /// surfaces ([`next_event`](Self::next_event) and friends).
    ///
    /// # Errors
    ///
    /// Returns typed endpoint, resume-state, refusal, transport, or
    /// answer-window failures.
    pub fn resume(
        attachment: &BusAttachment,
        grant: &JoinGrant,
        canonical_state: &[u8],
        store: S,
    ) -> Result<(Self, JoinGrant), AttachError> {
        let generation =
            Generation::new(grant.generation).ok_or_else(|| AttachError::ResumeRejected {
                detail: "grant carries the forbidden zero generation".to_owned(),
            })?;
        let config = connect(attachment)?;
        let sdk =
            RemoteParticipantHandle::restore(&config, StoreAdapter::new(store), canonical_state)
                .map_err(map_restore_error)?;
        let mut handle = Self::bare(sdk, grant.conversation, attachment);
        handle.me = grant.participant;
        handle.generation = generation;
        handle.grant = grant.clone();
        let answer = handle
            .submit(ClientRequest::CredentialAttach(CredentialAttachRequest {
                conversation_id: grant.conversation.to_wire(),
                participant_id: grant.participant.to_wire(),
                capability_generation: generation,
                attach_secret: AttachSecret::new(grant.credential.to_bytes()),
                attach_attempt_token: AttachAttemptToken::new(mint_token16()),
                accept_marker_delivery_seq: None,
            }))
            .map_err(map_submit_failure)?;
        match answer {
            ServerValue::AttachBound(receipt) => {
                let rotated_generation = receipt.origin_binding_epoch().capability_generation;
                let rotated = JoinGrant {
                    conversation: grant.conversation,
                    participant: grant.participant,
                    credential: JoinCredential::from_bytes(receipt.attach_secret().into_bytes()),
                    generation: rotated_generation.get(),
                };
                handle.generation = rotated_generation;
                handle.grant = rotated.clone();
                Ok((handle, rotated))
            }
            other => {
                let (class, detail) = classify_refusal(&other);
                Err(AttachError::Refused { class, detail })
            }
        }
    }
}