scosh-core 0.1.0

//! Single-owner portable session state machine.

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

use crate::{
    CoreError,
    events::{
        AcknowledgeResult, CommitReceipt, SessionEvent, SessionStatus, SnapshotRequiredReason,
        TerminalEffect,
    },
    input::{InputId, InputOutcome, InputRequest},
    types::{StateRevision, TerminalDelta, TerminalSnapshot, TerminalState},
};

/// Bounds the host event queue. Lossy state events are never silently
/// overwritten; the caller receives `ConsumerStalled` and can recover with a
/// fresh snapshot.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct SessionOptions {
    pub event_capacity: usize,
}

impl Default for SessionOptions {
    fn default() -> Self {
        Self {
            event_capacity: crate::types::MAX_EVENT_QUEUE,
        }
    }
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ReceiptState {
    Pending,
    Committed,
}

/// Portable state owner. Transport adapters feed snapshots/deltas and drain
/// events; host UI code only sees owned DTOs and opaque receipts.
#[derive(Debug)]
pub struct Session {
    options: SessionOptions,
    status: SessionStatus,
    state: Option<TerminalState>,
    revision: StateRevision,
    events: VecDeque<SessionEvent>,
    receipts: HashMap<CommitReceipt, ReceiptState>,
    next_receipt: u64,
    next_input: u64,
    pending_input: VecDeque<InputRequest>,
    input_generation: u64,
}

impl Session {
    pub fn new(options: SessionOptions) -> Self {
        Self {
            options: SessionOptions {
                event_capacity: options.event_capacity.max(1),
            },
            status: SessionStatus::Connecting,
            state: None,
            revision: StateRevision::new(0),
            events: VecDeque::new(),
            receipts: HashMap::new(),
            next_receipt: 0,
            next_input: 0,
            pending_input: VecDeque::new(),
            input_generation: 0,
        }
    }

    pub fn status(&self) -> SessionStatus {
        self.status
    }

    pub fn revision(&self) -> StateRevision {
        self.revision
    }

    pub fn state(&self) -> Option<&TerminalState> {
        self.state.as_ref()
    }

    pub fn install_snapshot(&mut self, snapshot: TerminalSnapshot) -> Result<(), CoreError> {
        if self.status == SessionStatus::Closed {
            return Err(CoreError::NotLive);
        }
        if !self.can_enqueue() {
            return Err(CoreError::ConsumerStalled);
        }
        self.state = Some(snapshot.state.clone());
        self.revision = snapshot.revision;
        self.status = SessionStatus::Live;
        self.events.push_back(SessionEvent::Snapshot { snapshot });
        if self.can_enqueue() {
            self.events
                .push_back(SessionEvent::StatusChanged(SessionStatus::Live));
        }
        Ok(())
    }

    pub fn apply_delta(&mut self, delta: &TerminalDelta) -> Result<(), CoreError> {
        if self.status == SessionStatus::Closed {
            return Err(CoreError::NotLive);
        }
        let Some(current) = self.state.as_ref() else {
            self.enqueue_snapshot_required(
                StateRevision::new(0),
                delta.revision,
                SnapshotRequiredReason::NoBase,
            )?;
            return Err(CoreError::NoSnapshot);
        };
        if delta.base_revision != self.revision
            || delta.revision.get() != self.revision.get().saturating_add(1)
        {
            self.enqueue_snapshot_required(
                self.revision.next(),
                delta.revision,
                SnapshotRequiredReason::RevisionGap,
            )?;
            return Err(CoreError::RevisionGap {
                expected: self.revision.next(),
                got: delta.revision,
            });
        }
        if delta.state.dimensions != current.dimensions {
            self.enqueue_snapshot_required(
                self.revision.next(),
                delta.revision,
                SnapshotRequiredReason::InvalidDelta,
            )?;
            return Err(CoreError::DimensionsMismatch);
        }
        let mut next = current.clone();
        delta.state.apply_to(&mut next)?;
        let effects = effects_for(&delta.state, current);
        let receipt = (!effects.is_empty()).then(|| self.issue_receipt());
        if let Some(receipt) = receipt {
            self.receipts.insert(receipt, ReceiptState::Pending);
        }
        if !self.can_enqueue() {
            if let Some(receipt) = receipt {
                self.receipts.remove(&receipt);
            }
            return Err(CoreError::ConsumerStalled);
        }
        self.state = Some(next);
        self.revision = delta.revision;
        self.events.push_back(SessionEvent::Delta {
            delta: delta.clone(),
            effects,
            receipt,
        });
        Ok(())
    }

    pub fn acknowledge(&mut self, receipt: CommitReceipt) -> AcknowledgeResult {
        match self.receipts.get_mut(&receipt) {
            Some(state @ ReceiptState::Pending) => {
                *state = ReceiptState::Committed;
                AcknowledgeResult::Committed
            }
            Some(ReceiptState::Committed) => AcknowledgeResult::Duplicate,
            None if receipt.0 <= self.next_receipt => AcknowledgeResult::RejectedStale,
            None => AcknowledgeResult::RejectedUnknown,
        }
    }

    /// A binding-friendly acknowledgement entry point that keeps receipt
    /// construction inside the core.
    pub fn acknowledge_raw(&mut self, receipt: u64) -> AcknowledgeResult {
        self.acknowledge(CommitReceipt(receipt))
    }

    pub fn submit_input(&mut self, bytes: &[u8]) -> Result<InputId, CoreError> {
        if self.status != SessionStatus::Live {
            return Err(CoreError::NotLive);
        }
        let id = InputId(self.next_input);
        self.next_input = self.next_input.saturating_add(1);
        let request = InputRequest::new(id, bytes.to_vec())?;
        if !self.can_enqueue() {
            return Err(CoreError::ConsumerStalled);
        }
        self.pending_input.push_back(request.clone());
        self.events.push_back(SessionEvent::InputQueued { request });
        Ok(id)
    }

    pub fn complete_input(&mut self, id: InputId, outcome: InputOutcome) -> Result<(), CoreError> {
        if self.status == SessionStatus::Closed {
            return Err(CoreError::NotLive);
        }
        self.pending_input.retain(|request| request.id() != id);
        if !self.can_enqueue() {
            return Err(CoreError::ConsumerStalled);
        }
        self.events
            .push_back(SessionEvent::InputCompleted { id, outcome });
        Ok(())
    }

    /// A binding-friendly completion entry point for an input id emitted by
    /// the core.
    pub fn complete_input_raw(&mut self, id: u64, outcome: InputOutcome) -> Result<(), CoreError> {
        self.complete_input(InputId(id), outcome)
    }

    pub fn begin_recovery(&mut self) -> Result<(), CoreError> {
        if self.status == SessionStatus::Closed {
            return Err(CoreError::NotLive);
        }
        self.input_generation = self.input_generation.saturating_add(1);
        let pending: Vec<_> = self.pending_input.drain(..).collect();
        self.status = SessionStatus::Recovering;
        for request in pending {
            if !self.can_enqueue() {
                return Err(CoreError::ConsumerStalled);
            }
            self.events.push_back(SessionEvent::InputCompleted {
                id: request.id(),
                outcome: InputOutcome::Superseded,
            });
        }
        if self.can_enqueue() {
            self.events
                .push_back(SessionEvent::StatusChanged(SessionStatus::Recovering));
            Ok(())
        } else {
            Err(CoreError::ConsumerStalled)
        }
    }

    pub fn suspend(&mut self) -> Result<(), CoreError> {
        if self.status == SessionStatus::Closed {
            return Err(CoreError::NotLive);
        }
        self.input_generation = self.input_generation.saturating_add(1);
        let pending: Vec<_> = self.pending_input.drain(..).collect();
        self.status = SessionStatus::Suspended;
        for request in pending {
            if !self.can_enqueue() {
                return Err(CoreError::ConsumerStalled);
            }
            self.events.push_back(SessionEvent::InputCompleted {
                id: request.id(),
                outcome: InputOutcome::Superseded,
            });
        }
        self.enqueue_status(SessionStatus::Suspended)
    }

    pub fn close(&mut self) {
        self.pending_input.clear();
        self.receipts.clear();
        self.state = None;
        self.status = SessionStatus::Closed;
        self.events.clear();
    }

    pub fn next_event(&mut self) -> Option<SessionEvent> {
        self.events.pop_front()
    }

    pub fn drain_events(&mut self) {
        self.events.clear();
    }

    pub fn pending_input_count(&self) -> usize {
        self.pending_input.len()
    }

    fn issue_receipt(&mut self) -> CommitReceipt {
        self.next_receipt = self.next_receipt.saturating_add(1);
        CommitReceipt(self.next_receipt)
    }

    fn can_enqueue(&self) -> bool {
        self.events.len() < self.options.event_capacity
    }

    fn enqueue_snapshot_required(
        &mut self,
        expected: StateRevision,
        received: StateRevision,
        reason: SnapshotRequiredReason,
    ) -> Result<(), CoreError> {
        if !self.can_enqueue() {
            return Err(CoreError::ConsumerStalled);
        }
        self.events.push_back(SessionEvent::SnapshotRequired {
            expected,
            received,
            reason,
        });
        Ok(())
    }

    fn enqueue_status(&mut self, status: SessionStatus) -> Result<(), CoreError> {
        if !self.can_enqueue() {
            return Err(CoreError::ConsumerStalled);
        }
        self.events.push_back(SessionEvent::StatusChanged(status));
        Ok(())
    }
}

fn effects_for(
    delta: &crate::types::TerminalStateDelta,
    current: &TerminalState,
) -> Vec<TerminalEffect> {
    let mut effects = Vec::new();
    if !delta.primary_scroll.is_empty() {
        effects.push(TerminalEffect::PrimaryScroll {
            rows: delta.primary_scroll.clone(),
        });
    }
    if delta.alternate != current.alternate {
        effects.push(TerminalEffect::Screen {
            alternate: delta.alternate,
        });
    }
    if delta.modes != current.modes {
        effects.push(TerminalEffect::InputModes { modes: delta.modes });
    }
    effects
}