softchat/

session.rs

//! Serialized relay session, delivery decisions, and platform persistence contract.

#![allow(
    unreachable_pub,
    reason = "UniFFI handle methods stay inside this private core module"
)]

use std::collections::{BTreeMap, BTreeSet, VecDeque};
#[cfg(feature = "native-bindings")]
use std::fmt;
#[cfg(feature = "native-bindings")]
use std::sync::{Mutex, MutexGuard};

use crate::event::SignedEvent;
use crate::{
    BatchAcknowledgement, BatchEventState, ClientRelayFrame, MAX_RELAY_BATCH_EVENTS, RelayFilter,
    RelayResponseFrame, SignedNostrEvent, SoftchatError, parse_relay_response_frame,
};
use serde::{Deserialize, Serialize};

/// Maximum queued outbound frames in one session.
pub const MAX_OUTBOUND_RELAY_FRAMES: usize = 1_024;
/// Maximum simultaneous subscription identifiers.
pub const MAX_RELAY_SUBSCRIPTIONS: usize = 128;
/// Maximum received events awaiting authoritative platform persistence.
pub const MAX_INFLIGHT_INGESTION: usize = 1_024;
/// Maximum authored events retained for durable delivery intent.
pub const MAX_PENDING_PUBLISHES: usize = 1_024;
const MAX_TERMINAL_DELIVERY_RECORDS: usize = 1_024;

/// Transport-independent relay connection state.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum RelayConnectionState {
    /// No transport exists.
    Disconnected,
    /// The platform is opening a WebSocket.
    Connecting,
    /// A transport exists and authentication is pending.
    Authenticating,
    /// The session can send subscriptions and events.
    Ready,
    /// The host should delay before another connection attempt.
    Backoff,
    /// The caller permanently cancelled this session.
    Cancelled,
}

/// Durable delivery state for one authored event.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum DeliveryState {
    /// Persisted locally and not currently in flight.
    Pending,
    /// Sent on the current connection and awaiting `OK`.
    InFlight,
    /// Relay accepted the event, including an idempotent duplicate.
    Accepted,
    /// Relay rejected the event permanently.
    Rejected,
    /// The event can be retried according to host policy.
    Retryable,
}

/// Stable host-facing delivery decision.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct DeliveryDecision {
    /// Canonical event ID.
    pub event_id: String,
    /// New durable state.
    pub state: DeliveryState,
    /// Stable category independent of arbitrary relay prose.
    pub category: String,
    /// Bounded original relay message for redacted diagnostics.
    pub relay_message: String,
}

/// One ordered text frame removed from the session's bounded outbound queue.
///
/// The host sends [`Self::frame`] through its platform WebSocket. Only after
/// that write completes successfully may it pass [`Self::delivery_event_ids`]
/// to [`RelaySession::confirm_sent`]. An empty ID list means the frame carries
/// no authored delivery intent.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct OutboundRelayFrame {
    /// Complete text frame for the host-owned WebSocket.
    pub frame: String,
    /// Authored event IDs whose socket-write handoff this frame represents.
    pub delivery_event_ids: Vec<String>,
}

const EVENT_FRAME_OVERHEAD_BYTES: usize = "[\"EVENT\",]".len();
pub(crate) const MAX_ACCOUNT_EVENT_JSON_BYTES: usize =
    crate::MAX_RELAY_FRAME_BYTES - EVENT_FRAME_OVERHEAD_BYTES;

/// Account-profile events must also fit their minimum publish envelope.
pub(crate) fn validate_event_frame_json(event_json: &str) -> Result<(), SoftchatError> {
    if event_json.len() > MAX_ACCOUNT_EVENT_JSON_BYTES {
        Err(SoftchatError::InvalidRelayFrame)
    } else {
        Ok(())
    }
}

/// Frame already authenticated, canonical event JSON from the typed event API
/// or the account store. Keep event order and exact JSON while enforcing both
/// wire limits; framing does not authenticate or publish the events again.
pub(crate) fn batch_event_frames(
    events: impl IntoIterator<Item = (String, String)>,
) -> Result<Vec<OutboundRelayFrame>, SoftchatError> {
    let mut frames = Vec::new();
    let mut ids = Vec::new();
    let mut jsons = Vec::new();
    let mut event_bytes = 0_usize;
    for (event_id, event_json) in events {
        // A single EVENT has ten framing bytes. EVENTS has ten fixed bytes
        // plus one comma per event, so adding a second event changes overhead.
        validate_event_frame_json(&event_json)?;
        let next_bytes = event_bytes
            .saturating_add(event_json.len())
            .saturating_add(ids.len())
            .saturating_add(EVENT_FRAME_OVERHEAD_BYTES + 1);
        if !ids.is_empty()
            && (ids.len() == MAX_RELAY_BATCH_EVENTS || next_bytes > crate::MAX_RELAY_FRAME_BYTES)
        {
            frames.push(finish_event_frame(
                std::mem::take(&mut ids),
                std::mem::take(&mut jsons),
            ));
            event_bytes = 0;
        }
        event_bytes += event_json.len();
        ids.push(event_id);
        jsons.push(event_json);
    }
    if !ids.is_empty() {
        frames.push(finish_event_frame(ids, jsons));
    }
    Ok(frames)
}

fn finish_event_frame(ids: Vec<String>, jsons: Vec<String>) -> OutboundRelayFrame {
    let prefix = if ids.len() == 1 {
        "[\"EVENT\","
    } else {
        "[\"EVENTS\","
    };
    OutboundRelayFrame {
        frame: format!("{prefix}{}]", jsons.join(",")),
        delivery_event_ids: ids,
    }
}

/// Action kind emitted by the pure session.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum RelaySessionActionKind {
    /// Platform should open a WebSocket.
    OpenTransport,
    /// Platform should send `frame`.
    SendFrame,
    /// Host should build and enqueue a NIP-42 AUTH frame for `value`.
    Authenticate,
    /// Platform must transactionally ingest `event`.
    PersistEvent,
    /// Subscription reached EOSE.
    SubscriptionReady,
    /// Relay ended the subscription named by `value`.
    SubscriptionClosed,
    /// One delivery reached a terminal or retry state.
    DeliveryChanged,
    /// Platform should close the current transport.
    CloseTransport,
    /// Host should schedule another connection attempt after `delay_ms`.
    ScheduleReconnect,
}

/// One transport/platform action emitted by [`RelaySession`].
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct RelaySessionAction {
    /// Action type.
    pub kind: RelaySessionActionKind,
    /// Wire frame for `SendFrame`, otherwise empty.
    pub frame: String,
    /// Subscription ID, challenge, event ID, or reason depending on `kind`.
    pub value: String,
    /// Bounded diagnostic reason for `SubscriptionClosed`, otherwise empty.
    ///
    /// This relay-provided text must never be parsed for control flow.
    pub reason: String,
    /// Signed event for `PersistEvent`, otherwise absent.
    pub event: Option<SignedEvent>,
    /// Delay for `ScheduleReconnect`, otherwise zero.
    pub delay_ms: u64,
    /// Complete delivery decision for `DeliveryChanged`, otherwise absent.
    pub delivery_decision: Option<DeliveryDecision>,
}

/// Serializable non-secret session state for diagnostics only.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct RelaySessionSnapshot {
    /// Current connection state.
    pub state: RelayConnectionState,
    /// Active subscription IDs.
    pub subscriptions: Vec<String>,
    /// Subscriptions that observed EOSE on this connection.
    pub ready_subscriptions: Vec<String>,
    /// Event IDs and delivery states at matching indices.
    pub delivery_event_ids: Vec<String>,
    /// Event states at matching indices.
    pub delivery_states: Vec<DeliveryState>,
    /// Verified inbound IDs waiting for a platform transaction.
    pub pending_ingestion_ids: Vec<String>,
    /// Current reconnect attempt.
    pub reconnect_attempt: u32,
}

/// One bounded atomic ingestion request to a platform-owned store.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct IngestionBatch {
    /// Stable account partition identifier; never a secret key.
    pub account_id: String,
    /// Verified events to insert idempotently in one transaction.
    pub events: Vec<SignedEvent>,
}

/// Result returned by an authoritative Room, GRDB, IndexedDB, or host adapter.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct IngestionResult {
    /// IDs inserted for the first time.
    pub inserted_event_ids: Vec<String>,
    /// IDs already present with identical canonical data.
    pub duplicate_event_ids: Vec<String>,
    /// IDs quarantined instead of projected.
    pub quarantined_event_ids: Vec<String>,
}

/// One deterministic transport-independent relay session.
#[derive(Debug)]
pub struct RelaySession {
    state: RelayConnectionState,
    subscriptions: BTreeMap<String, Vec<RelayFilter>>,
    ready_subscriptions: BTreeSet<String>,
    delivery: BTreeMap<String, DeliveryState>,
    terminal_delivery_order: VecDeque<String>,
    pending_events: BTreeMap<String, SignedNostrEvent>,
    pending_authentication_event_id: Option<String>,
    pending_ingestion: BTreeSet<String>,
    outbound: VecDeque<OutboundRelayFrame>,
    reconnect_attempt: u32,
}

/// Thread-safe generated-binding owner for one serialized relay session.
#[cfg(feature = "native-bindings")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Object))]
pub struct RelaySessionHandle {
    session: Mutex<RelaySession>,
}

#[cfg(feature = "native-bindings")]
impl fmt::Debug for RelaySessionHandle {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("RelaySessionHandle")
            .field("snapshot", &lock(&self.session).snapshot())
            .finish()
    }
}

#[cfg(feature = "native-bindings")]
impl Default for RelaySessionHandle {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(feature = "native-bindings")]
#[cfg_attr(feature = "native-bindings", uniffi::export)]
impl RelaySessionHandle {
    /// Construct an empty disconnected session.
    #[cfg_attr(feature = "native-bindings", uniffi::constructor)]
    #[must_use]
    pub fn new() -> Self {
        Self {
            session: Mutex::new(RelaySession::new()),
        }
    }

    /// Begin a platform-owned WebSocket connection attempt.
    ///
    /// # Errors
    ///
    /// Returns a stable invalid-transition error.
    pub fn connect(&self) -> Result<RelaySessionAction, SoftchatError> {
        lock(&self.session).connect()
    }

    /// Notify the session that the native transport is connected.
    ///
    /// # Errors
    ///
    /// Returns a stable invalid-transition error.
    pub fn transport_connected(&self) -> Result<(), SoftchatError> {
        lock(&self.session).transport_connected()
    }

    /// Mark relay authentication complete and replay retained intent.
    ///
    /// # Errors
    ///
    /// Returns a stable transition or bounded-queue error.
    pub fn authenticated(&self) -> Result<(), SoftchatError> {
        lock(&self.session).authenticated()
    }

    /// Queue one validated NIP-42 response while authentication is pending.
    ///
    /// # Errors
    ///
    /// Returns a stable authentication, state, or queue error.
    pub fn authenticate(&self, event: SignedEvent) -> Result<(), SoftchatError> {
        lock(&self.session).authenticate(SignedNostrEvent::try_from(event)?)
    }

    /// Add or replace a subscription using strict filter JSON objects.
    ///
    /// # Errors
    ///
    /// Returns a stable filter, transition, or bounded-queue error.
    pub fn subscribe(
        &self,
        subscription_id: String,
        filter_json: Vec<String>,
    ) -> Result<(), SoftchatError> {
        let filters = filter_json
            .iter()
            .map(|json| RelayFilter::from_json(json))
            .collect::<Result<Vec<_>, _>>()?;
        lock(&self.session).subscribe(subscription_id, filters)
    }

    /// Close one retained subscription.
    ///
    /// # Errors
    ///
    /// Returns a stable state or bounded-queue error.
    pub fn close_subscription(&self, subscription_id: String) -> Result<(), SoftchatError> {
        lock(&self.session).close_subscription(&subscription_id)
    }

    /// Queue already-persisted authored events for delivery.
    ///
    /// # Errors
    ///
    /// Returns a stable validation, state, or queue error.
    pub fn publish(&self, events: Vec<SignedEvent>) -> Result<(), SoftchatError> {
        let events = events
            .into_iter()
            .map(SignedNostrEvent::try_from)
            .collect::<Result<Vec<_>, _>>()?;
        lock(&self.session).publish(events)
    }

    /// Confirm that one drained authored-event frame was written to the socket.
    ///
    /// An empty list is accepted so a host may call this uniformly for every
    /// [`OutboundRelayFrame`].
    ///
    /// # Errors
    ///
    /// Returns a stable delivery-state error for duplicate, unknown, or
    /// already in-flight IDs. A terminal ID is an idempotent no-op because a
    /// relay acknowledgement may race the local socket-write completion.
    pub fn confirm_sent(&self, event_ids: Vec<String>) -> Result<(), SoftchatError> {
        lock(&self.session).confirm_sent(&event_ids)
    }

    /// Consume one complete native WebSocket text frame.
    ///
    /// # Errors
    ///
    /// Returns a stable frame, correlation, state, or backpressure error.
    pub fn receive(&self, frame_json: String) -> Result<Vec<RelaySessionAction>, SoftchatError> {
        lock(&self.session).receive(&frame_json)
    }

    /// Confirm an authoritative platform transaction handled an event.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence-result error for unknown IDs.
    pub fn confirm_ingested(&self, event_id: String) -> Result<(), SoftchatError> {
        lock(&self.session).confirm_ingested(&event_id)
    }

    /// Handle transport loss and return the reconnect plan.
    ///
    /// # Errors
    ///
    /// Returns a stable invalid-transition error.
    pub fn transport_lost(&self) -> Result<Vec<RelaySessionAction>, SoftchatError> {
        lock(&self.session).transport_lost()
    }

    /// Drain at most `limit` ordered outbound text frames.
    ///
    /// # Errors
    ///
    /// Returns a stable error for a zero limit.
    pub fn drain_outbound(&self, limit: u32) -> Result<Vec<OutboundRelayFrame>, SoftchatError> {
        let limit = usize::try_from(limit).map_err(|_| SoftchatError::InvalidRelaySession)?;
        lock(&self.session).drain_outbound(limit)
    }

    /// Return one redacted serializable snapshot.
    #[must_use]
    pub fn snapshot(&self) -> RelaySessionSnapshot {
        lock(&self.session).snapshot()
    }

    /// Permanently cancel the session and clear process-local queues.
    #[must_use]
    pub fn cancel(&self) -> RelaySessionAction {
        lock(&self.session).cancel()
    }
}

impl Default for RelaySession {
    fn default() -> Self {
        Self::new()
    }
}

impl RelaySession {
    /// Create one disconnected session with empty bounded queues.
    #[must_use]
    pub const fn new() -> Self {
        Self {
            state: RelayConnectionState::Disconnected,
            subscriptions: BTreeMap::new(),
            ready_subscriptions: BTreeSet::new(),
            delivery: BTreeMap::new(),
            terminal_delivery_order: VecDeque::new(),
            pending_events: BTreeMap::new(),
            pending_authentication_event_id: None,
            pending_ingestion: BTreeSet::new(),
            outbound: VecDeque::new(),
            reconnect_attempt: 0,
        }
    }

    /// Begin a platform-owned connection attempt.
    ///
    /// # Errors
    ///
    /// Returns [`SoftchatError::InvalidRelaySession`] unless disconnected or
    /// waiting in backoff.
    pub fn connect(&mut self) -> Result<RelaySessionAction, SoftchatError> {
        if !matches!(
            self.state,
            RelayConnectionState::Disconnected | RelayConnectionState::Backoff
        ) {
            return Err(SoftchatError::InvalidRelaySession);
        }
        self.state = RelayConnectionState::Connecting;
        Ok(action(RelaySessionActionKind::OpenTransport))
    }

    /// Notify the session that the native WebSocket is open.
    ///
    /// Existing subscriptions are replayed only after the host indicates
    /// authentication is complete.
    ///
    /// # Errors
    ///
    /// Returns a stable transition error from any state except `Connecting`.
    pub fn transport_connected(&mut self) -> Result<(), SoftchatError> {
        if self.state != RelayConnectionState::Connecting {
            return Err(SoftchatError::InvalidRelaySession);
        }
        self.state = RelayConnectionState::Authenticating;
        self.ready_subscriptions.clear();
        Ok(())
    }

    /// Mark NIP-42 authentication complete and replay session intent.
    ///
    /// # Errors
    ///
    /// Returns a transition or bounded-queue failure.
    pub fn authenticated(&mut self) -> Result<(), SoftchatError> {
        if self.state != RelayConnectionState::Authenticating {
            return Err(SoftchatError::InvalidRelaySession);
        }
        self.pending_authentication_event_id = None;
        let mut replay = self
            .subscriptions
            .iter()
            .map(|(subscription_id, filters)| {
                Ok(OutboundRelayFrame {
                    frame: ClientRelayFrame::Req {
                        subscription_id: subscription_id.clone(),
                        filters: filters.clone(),
                    }
                    .to_json()?,
                    delivery_event_ids: Vec::new(),
                })
            })
            .collect::<Result<Vec<_>, SoftchatError>>()?;
        let pending = self
            .pending_events
            .iter()
            .filter(|(event_id, _)| {
                matches!(
                    self.delivery.get(*event_id),
                    Some(DeliveryState::Pending | DeliveryState::Retryable)
                )
            })
            .map(|(_, event)| event.clone())
            .collect::<Vec<_>>();
        replay.extend(batch_event_frames(
            pending
                .iter()
                .map(|event| Ok((event.id().to_hex(), event.to_json()?)))
                .collect::<Result<Vec<_>, SoftchatError>>()?,
        )?);
        self.ensure_outbound_capacity(replay.len())?;
        self.state = RelayConnectionState::Ready;
        self.reconnect_attempt = 0;
        self.ready_subscriptions.clear();
        self.outbound.extend(replay);
        Ok(())
    }

    /// Queue one NIP-42 `AUTH` response for the platform WebSocket.
    ///
    /// The event must already be verified, kind 22242, and bound to the
    /// challenge/relay by the identity writer. Authentication becomes ready
    /// only after the host writes this frame and calls [`Self::authenticated`].
    ///
    /// # Errors
    ///
    /// Returns a stable authentication, state, or bounded-queue error.
    pub fn authenticate(&mut self, event: SignedNostrEvent) -> Result<(), SoftchatError> {
        if self.state != RelayConnectionState::Authenticating {
            return Err(SoftchatError::InvalidRelaySession);
        }
        let event_id = event.id().to_hex();
        let frame = ClientRelayFrame::Auth(event).to_json()?;
        self.ensure_outbound_capacity(1)?;
        self.enqueue(frame, Vec::new())?;
        self.pending_authentication_event_id = Some(event_id);
        Ok(())
    }

    /// Add or replace one subscription intent.
    ///
    /// The request is queued immediately when ready and retained across
    /// reconnects.
    ///
    /// # Errors
    ///
    /// Returns a validation, state, or bounded-queue failure.
    pub fn subscribe(
        &mut self,
        subscription_id: String,
        filters: Vec<RelayFilter>,
    ) -> Result<(), SoftchatError> {
        self.subscribe_many(vec![(subscription_id, filters)])
    }

    /// Atomically add or replace multiple subscription intents.
    ///
    /// This internal operation validates every identifier, filter, encoded
    /// frame, subscription slot, and outbound queue slot before changing the
    /// session. It lets the account synchronizer shard a large exact-ID fetch
    /// without leaving a partially installed request set.
    pub(crate) fn subscribe_many(
        &mut self,
        subscriptions: Vec<(String, Vec<RelayFilter>)>,
    ) -> Result<(), SoftchatError> {
        if self.state == RelayConnectionState::Cancelled || subscriptions.is_empty() {
            return Err(SoftchatError::InvalidRelaySession);
        }
        let unique_ids = subscriptions
            .iter()
            .map(|(subscription_id, _)| subscription_id)
            .collect::<BTreeSet<_>>();
        if unique_ids.len() != subscriptions.len() {
            return Err(SoftchatError::InvalidRelaySession);
        }
        let additional_subscriptions = unique_ids
            .iter()
            .filter(|subscription_id| !self.subscriptions.contains_key(subscription_id.as_str()))
            .count();
        if self
            .subscriptions
            .len()
            .saturating_add(additional_subscriptions)
            > MAX_RELAY_SUBSCRIPTIONS
        {
            return Err(SoftchatError::InvalidRelaySession);
        }
        let frames = subscriptions
            .iter()
            .map(|(subscription_id, filters)| {
                ClientRelayFrame::Req {
                    subscription_id: subscription_id.clone(),
                    filters: filters.clone(),
                }
                .to_json()
            })
            .collect::<Result<Vec<_>, _>>()?;
        if self.state == RelayConnectionState::Ready {
            self.ensure_outbound_capacity(frames.len())?;
        }
        for (subscription_id, filters) in subscriptions {
            self.subscriptions.insert(subscription_id.clone(), filters);
            self.ready_subscriptions.remove(&subscription_id);
        }
        if self.state == RelayConnectionState::Ready {
            self.outbound
                .extend(frames.into_iter().map(|frame| OutboundRelayFrame {
                    frame,
                    delivery_event_ids: Vec::new(),
                }));
        }
        Ok(())
    }

    /// Close one retained subscription.
    ///
    /// # Errors
    ///
    /// Returns a state or bounded-queue failure.
    pub fn close_subscription(&mut self, subscription_id: &str) -> Result<(), SoftchatError> {
        if !self.subscriptions.contains_key(subscription_id) {
            return Err(SoftchatError::InvalidRelaySession);
        }
        let frame = if self.state == RelayConnectionState::Ready {
            self.ensure_outbound_capacity(1)?;
            Some(
                ClientRelayFrame::Close {
                    subscription_id: subscription_id.to_owned(),
                }
                .to_json()?,
            )
        } else {
            None
        };
        self.subscriptions.remove(subscription_id);
        self.ready_subscriptions.remove(subscription_id);
        if let Some(frame) = frame {
            self.enqueue(frame, Vec::new())?;
        }
        Ok(())
    }

    /// Queue one or more already persisted authored events for publication.
    ///
    /// # Errors
    ///
    /// Returns a state, duplicate-ID, event-validation, or queue failure.
    pub fn publish(&mut self, events: Vec<SignedNostrEvent>) -> Result<(), SoftchatError> {
        if self.state == RelayConnectionState::Cancelled
            || events.is_empty()
            || events.len() > MAX_RELAY_BATCH_EVENTS
            || self.pending_events.len().saturating_add(events.len()) > MAX_PENDING_PUBLISHES
        {
            return Err(SoftchatError::InvalidRelaySession);
        }
        let event_ids = events
            .iter()
            .map(|event| event.id().to_hex())
            .collect::<Vec<_>>();
        if event_ids.iter().collect::<BTreeSet<_>>().len() != event_ids.len() {
            return Err(SoftchatError::InvalidDeliveryState);
        }
        for event_id in &event_ids {
            if self.delivery.contains_key(event_id) {
                return Err(SoftchatError::InvalidDeliveryState);
            }
        }
        let frames = batch_event_frames(
            events
                .iter()
                .map(|event| Ok((event.id().to_hex(), event.to_json()?)))
                .collect::<Result<Vec<_>, SoftchatError>>()?,
        )?;
        if self.state == RelayConnectionState::Ready {
            self.ensure_outbound_capacity(frames.len())?;
        }
        for (event, event_id) in events.iter().zip(&event_ids) {
            self.delivery
                .insert(event_id.clone(), DeliveryState::Pending);
            self.pending_events.insert(event_id.clone(), event.clone());
        }
        if self.state == RelayConnectionState::Ready {
            self.outbound.extend(frames);
        }
        Ok(())
    }

    /// Confirm that the host successfully wrote one authored-event frame.
    ///
    /// This is deliberately separate from [`Self::drain_outbound`]: removal
    /// from a process-local queue is not evidence that a platform socket
    /// accepted the frame. An empty list is a no-op for subscription and
    /// control frames.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery-state error for duplicate, unknown, or
    /// already in-flight IDs. A terminal ID is an idempotent no-op because a
    /// relay acknowledgement may race the local socket-write completion.
    pub fn confirm_sent(&mut self, event_ids: &[String]) -> Result<(), SoftchatError> {
        if event_ids.is_empty() {
            return Ok(());
        }
        if self.state != RelayConnectionState::Ready || event_ids.len() > MAX_RELAY_BATCH_EVENTS {
            return Err(SoftchatError::InvalidDeliveryState);
        }
        let unique = event_ids.iter().collect::<BTreeSet<_>>();
        if unique.len() != event_ids.len()
            || event_ids.iter().any(|event_id| {
                !matches!(
                    self.delivery.get(event_id),
                    Some(
                        DeliveryState::Pending
                            | DeliveryState::Retryable
                            | DeliveryState::Accepted
                            | DeliveryState::Rejected
                    )
                )
            })
        {
            return Err(SoftchatError::InvalidDeliveryState);
        }
        for event_id in event_ids {
            if matches!(
                self.delivery.get(event_id),
                Some(DeliveryState::Pending | DeliveryState::Retryable)
            ) {
                self.delivery
                    .insert(event_id.clone(), DeliveryState::InFlight);
            }
        }
        Ok(())
    }

    /// Consume one complete relay WebSocket text frame.
    ///
    /// # Errors
    ///
    /// Returns a frame, state, correlation, or backpressure failure.
    pub fn receive(&mut self, frame_json: &str) -> Result<Vec<RelaySessionAction>, SoftchatError> {
        if !matches!(
            self.state,
            RelayConnectionState::Authenticating | RelayConnectionState::Ready
        ) {
            return Err(SoftchatError::InvalidRelaySession);
        }
        let frame = parse_relay_response_frame(frame_json)?;
        if self.state == RelayConnectionState::Authenticating
            && matches!(
                &frame,
                RelayResponseFrame::Event { .. }
                    | RelayResponseFrame::Eose { .. }
                    | RelayResponseFrame::Closed { .. }
            )
        {
            return Ok(Vec::new());
        }
        match frame {
            RelayResponseFrame::Auth(challenge) => {
                self.state = RelayConnectionState::Authenticating;
                self.outbound.clear();
                self.pending_authentication_event_id = None;
                self.ready_subscriptions.clear();
                for state in self.delivery.values_mut() {
                    if *state == DeliveryState::InFlight {
                        *state = DeliveryState::Retryable;
                    }
                }
                Ok(vec![RelaySessionAction {
                    kind: RelaySessionActionKind::Authenticate,
                    value: challenge,
                    ..action(RelaySessionActionKind::Authenticate)
                }])
            }
            RelayResponseFrame::Event {
                subscription_id,
                event,
            } => {
                if !self.subscriptions.contains_key(&subscription_id) {
                    return Err(SoftchatError::InvalidRelaySession);
                }
                let event_id = event.id().to_hex();
                if self.pending_ingestion.contains(&event_id) {
                    return Ok(Vec::new());
                }
                if self.pending_ingestion.len() >= MAX_INFLIGHT_INGESTION {
                    return Err(SoftchatError::RelaySessionQueueFull);
                }
                self.pending_ingestion.insert(event_id.clone());
                Ok(vec![RelaySessionAction {
                    kind: RelaySessionActionKind::PersistEvent,
                    value: subscription_id,
                    event: Some(SignedEvent::from(&event)),
                    ..action(RelaySessionActionKind::PersistEvent)
                }])
            }
            RelayResponseFrame::Eose { subscription_id } => {
                if !self.subscriptions.contains_key(&subscription_id) {
                    return Err(SoftchatError::InvalidRelaySession);
                }
                self.ready_subscriptions.insert(subscription_id.clone());
                Ok(vec![RelaySessionAction {
                    kind: RelaySessionActionKind::SubscriptionReady,
                    value: subscription_id,
                    ..action(RelaySessionActionKind::SubscriptionReady)
                }])
            }
            RelayResponseFrame::Ok(acknowledgement) => {
                if self.pending_authentication_event_id.as_deref()
                    == Some(acknowledgement.event_id.as_str())
                {
                    if !acknowledgement.accepted {
                        self.pending_authentication_event_id = None;
                        return Err(SoftchatError::InvalidRelayAuthentication);
                    }
                    self.pending_authentication_event_id = None;
                    self.authenticated()?;
                    return Ok(Vec::new());
                }
                let decision = classify_delivery_acknowledgement(acknowledgement)?;
                let current = self
                    .delivery
                    .get(&decision.event_id)
                    .copied()
                    .ok_or(SoftchatError::InvalidDeliveryState)?;
                let is_idempotent_terminal =
                    matches!(current, DeliveryState::Accepted | DeliveryState::Rejected)
                        && current == decision.state;
                let is_ack_before_write =
                    matches!(current, DeliveryState::Pending | DeliveryState::Retryable);
                if current != DeliveryState::InFlight
                    && !is_ack_before_write
                    && !is_idempotent_terminal
                {
                    return Err(SoftchatError::InvalidDeliveryState);
                }
                self.delivery
                    .insert(decision.event_id.clone(), decision.state);
                if matches!(
                    decision.state,
                    DeliveryState::Accepted | DeliveryState::Rejected
                ) {
                    self.pending_events.remove(&decision.event_id);
                    if !is_idempotent_terminal {
                        self.remember_terminal_delivery(decision.event_id.clone());
                    }
                }
                Ok(vec![RelaySessionAction {
                    kind: RelaySessionActionKind::DeliveryChanged,
                    value: decision.event_id.clone(),
                    delivery_decision: Some(decision),
                    ..action(RelaySessionActionKind::DeliveryChanged)
                }])
            }
            RelayResponseFrame::Closed {
                subscription_id,
                message,
            } => {
                if !self.subscriptions.contains_key(&subscription_id) {
                    return Ok(Vec::new());
                }
                self.subscriptions.remove(&subscription_id);
                self.ready_subscriptions.remove(&subscription_id);
                Ok(vec![RelaySessionAction {
                    kind: RelaySessionActionKind::SubscriptionClosed,
                    value: subscription_id,
                    reason: message,
                    ..action(RelaySessionActionKind::SubscriptionClosed)
                }])
            }
            RelayResponseFrame::Notice(_) | RelayResponseFrame::Count { .. } => Ok(Vec::new()),
        }
    }

    /// Confirm that an authoritative platform transaction handled an event.
    ///
    /// # Errors
    ///
    /// Returns a stable state error for an unknown or repeated confirmation.
    pub fn confirm_ingested(&mut self, event_id: &str) -> Result<(), SoftchatError> {
        if self.pending_ingestion.remove(event_id) {
            Ok(())
        } else {
            Err(SoftchatError::InvalidPersistenceResult)
        }
    }

    /// Retire a rejected input without claiming that it was durably ingested.
    #[cfg(feature = "sqlite-storage")]
    pub(crate) fn discard_ingestion(&mut self, event_id: &str) -> Result<(), SoftchatError> {
        if self.pending_ingestion.remove(event_id) {
            Ok(())
        } else {
            Err(SoftchatError::InvalidPersistenceResult)
        }
    }

    /// Handle transport loss and classify only nonterminal in-flight events for retry.
    ///
    /// # Errors
    ///
    /// Returns a transition error when no live transport existed.
    pub fn transport_lost(&mut self) -> Result<Vec<RelaySessionAction>, SoftchatError> {
        if !matches!(
            self.state,
            RelayConnectionState::Connecting
                | RelayConnectionState::Authenticating
                | RelayConnectionState::Ready
        ) {
            return Err(SoftchatError::InvalidRelaySession);
        }
        self.outbound.clear();
        self.pending_authentication_event_id = None;
        self.ready_subscriptions.clear();
        for state in self.delivery.values_mut() {
            if *state == DeliveryState::InFlight {
                *state = DeliveryState::Retryable;
            }
        }
        self.reconnect_attempt = self.reconnect_attempt.saturating_add(1);
        self.state = RelayConnectionState::Backoff;
        let exponent = self.reconnect_attempt.saturating_sub(1).min(4);
        let delay_ms = 500_u64.saturating_mul(1_u64 << exponent);
        Ok(vec![RelaySessionAction {
            kind: RelaySessionActionKind::ScheduleReconnect,
            delay_ms,
            ..action(RelaySessionActionKind::ScheduleReconnect)
        }])
    }

    /// Permanently cancel and clear process-local queues.
    #[must_use]
    pub fn cancel(&mut self) -> RelaySessionAction {
        self.state = RelayConnectionState::Cancelled;
        self.subscriptions.clear();
        self.ready_subscriptions.clear();
        self.delivery.clear();
        self.terminal_delivery_order.clear();
        self.outbound.clear();
        self.pending_events.clear();
        self.pending_authentication_event_id = None;
        self.pending_ingestion.clear();
        action(RelaySessionActionKind::CloseTransport)
    }

    /// Drain at most `limit` queued frames for the platform WebSocket.
    ///
    /// # Errors
    ///
    /// Returns a stable state error for zero limits.
    pub fn drain_outbound(
        &mut self,
        limit: usize,
    ) -> Result<Vec<OutboundRelayFrame>, SoftchatError> {
        if limit == 0 {
            return Err(SoftchatError::InvalidRelaySession);
        }
        let count = limit.min(self.outbound.len());
        Ok(self.outbound.drain(..count).collect())
    }

    /// Return a stable non-secret snapshot.
    #[must_use]
    pub fn snapshot(&self) -> RelaySessionSnapshot {
        RelaySessionSnapshot {
            state: self.state,
            subscriptions: self.subscriptions.keys().cloned().collect(),
            ready_subscriptions: self.ready_subscriptions.iter().cloned().collect(),
            delivery_event_ids: self.delivery.keys().cloned().collect(),
            delivery_states: self.delivery.values().copied().collect(),
            pending_ingestion_ids: self.pending_ingestion.iter().cloned().collect(),
            reconnect_attempt: self.reconnect_attempt,
        }
    }

    fn enqueue(
        &mut self,
        frame: String,
        delivery_event_ids: Vec<String>,
    ) -> Result<(), SoftchatError> {
        self.ensure_outbound_capacity(1)?;
        self.outbound.push_back(OutboundRelayFrame {
            frame,
            delivery_event_ids,
        });
        Ok(())
    }

    fn ensure_outbound_capacity(&self, additional: usize) -> Result<(), SoftchatError> {
        if self.outbound.len().saturating_add(additional) > MAX_OUTBOUND_RELAY_FRAMES {
            Err(SoftchatError::RelaySessionQueueFull)
        } else {
            Ok(())
        }
    }

    fn remember_terminal_delivery(&mut self, event_id: String) {
        self.terminal_delivery_order.push_back(event_id);
        while self.terminal_delivery_order.len() > MAX_TERMINAL_DELIVERY_RECORDS {
            if let Some(expired) = self.terminal_delivery_order.pop_front()
                && matches!(
                    self.delivery.get(&expired),
                    Some(DeliveryState::Accepted | DeliveryState::Rejected)
                )
            {
                self.delivery.remove(&expired);
            }
        }
    }
}

/// Classify one standard `OK` without trusting its human-readable message as identity.
///
/// # Errors
///
/// Returns [`SoftchatError::InvalidDeliveryState`] for malformed values.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn classify_delivery_acknowledgement(
    acknowledgement: BatchAcknowledgement,
) -> Result<DeliveryDecision, SoftchatError> {
    let parsed = crate::track_batch_acknowledgements(
        vec![acknowledgement.event_id.clone()],
        vec![acknowledgement.clone()],
    )
    .map_err(|_| SoftchatError::InvalidDeliveryState)?;
    let state = match parsed.states.first() {
        Some(BatchEventState::Accepted) => DeliveryState::Accepted,
        Some(BatchEventState::Rejected) => {
            if is_retryable_message(&acknowledgement.message) {
                DeliveryState::Retryable
            } else {
                DeliveryState::Rejected
            }
        }
        _ => return Err(SoftchatError::InvalidDeliveryState),
    };
    let category = if acknowledgement.accepted {
        if acknowledgement.message.starts_with("duplicate:") {
            "duplicate"
        } else {
            "accepted"
        }
    } else if is_retryable_message(&acknowledgement.message) {
        "retryable"
    } else if acknowledgement.message.starts_with("auth-required:") {
        "authentication"
    } else if acknowledgement.message.starts_with("blocked:") {
        "blocked"
    } else if acknowledgement.message.starts_with("invalid:") {
        "invalid"
    } else {
        "rejected"
    };
    Ok(DeliveryDecision {
        event_id: acknowledgement.event_id,
        state,
        category: category.to_owned(),
        relay_message: acknowledgement.message,
    })
}

/// Verify that a platform transaction classified every requested event exactly once.
///
/// # Errors
///
/// Returns [`SoftchatError::InvalidPersistenceResult`] for missing, duplicate,
/// or unknown IDs and accountless batches.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn validate_ingestion_result(
    batch: &IngestionBatch,
    result: &IngestionResult,
) -> Result<(), SoftchatError> {
    if batch.account_id.is_empty() || batch.events.is_empty() {
        return Err(SoftchatError::InvalidPersistenceResult);
    }
    let requested = batch
        .events
        .iter()
        .map(|event| event.id.clone())
        .collect::<BTreeSet<_>>();
    if requested.len() != batch.events.len() {
        return Err(SoftchatError::InvalidPersistenceResult);
    }
    let classified = result
        .inserted_event_ids
        .iter()
        .chain(&result.duplicate_event_ids)
        .chain(&result.quarantined_event_ids)
        .cloned()
        .collect::<Vec<_>>();
    let unique = classified.iter().cloned().collect::<BTreeSet<_>>();
    if unique.len() != classified.len() || unique != requested {
        return Err(SoftchatError::InvalidPersistenceResult);
    }
    Ok(())
}

fn is_retryable_message(message: &str) -> bool {
    ["rate-limited:", "error:", "unavailable:", "timeout:"]
        .iter()
        .any(|prefix| message.starts_with(prefix))
}

fn action(kind: RelaySessionActionKind) -> RelaySessionAction {
    RelaySessionAction {
        kind,
        frame: String::new(),
        value: String::new(),
        reason: String::new(),
        event: None,
        delay_ms: 0,
        delivery_decision: None,
    }
}

#[cfg(feature = "native-bindings")]
fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
    match mutex.lock() {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    }
}

#[cfg(test)]
mod tests {
    use crate::{LocalIdentity, NostrEventDraft, NostrEventKind};

    use super::*;

    const ALICE_SECRET: &str = "5c0c523f52a5b6fad39ed2403092df8cebc36318b39383bca6c00808626fab3a";

    fn event(created_at: u64) -> Result<SignedNostrEvent, SoftchatError> {
        LocalIdentity::from_secret_hex(ALICE_SECRET)?.sign_event(NostrEventDraft::new(
            created_at,
            NostrEventKind::SHORT_TEXT_NOTE,
            Vec::new(),
            "test",
        )?)
    }

    #[test]
    fn session_publish_and_authentication_replay_split_large_frames() -> Result<(), SoftchatError> {
        let identity = LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let events = (0..3)
            .map(|timestamp| {
                identity.sign_event(NostrEventDraft::new(
                    timestamp,
                    NostrEventKind::SHORT_TEXT_NOTE,
                    Vec::new(),
                    "large 🦀".repeat(28_000),
                )?)
            })
            .collect::<Result<Vec<_>, SoftchatError>>()?;
        let expected = events
            .iter()
            .map(|event| event.id().to_hex())
            .collect::<Vec<_>>();
        for authenticated in [false, true] {
            let mut session = RelaySession::new();
            session.connect()?;
            session.transport_connected()?;
            if authenticated {
                session.authenticated()?;
            }
            session.publish(events.clone())?;
            if !authenticated {
                session.authenticated()?;
            }
            let frames = session.drain_outbound(8)?;
            assert_eq!(frames.len(), 3);
            let actual = frames
                .iter()
                .flat_map(|frame| frame.delivery_event_ids.clone())
                .collect::<Vec<_>>();
            let expected_order = if authenticated {
                expected.clone()
            } else {
                expected
                    .iter()
                    .cloned()
                    .collect::<BTreeSet<_>>()
                    .into_iter()
                    .collect()
            };
            assert_eq!(actual, expected_order);
            for frame in &frames {
                let event = events
                    .iter()
                    .find(|event| frame.delivery_event_ids.contains(&event.id().to_hex()))
                    .ok_or(SoftchatError::InvalidRelaySession)?;
                assert!(frame.frame.len() <= crate::MAX_RELAY_FRAME_BYTES);
                assert_eq!(
                    crate::parse_client_relay_frame(&frame.frame)?,
                    ClientRelayFrame::Event(event.clone())
                );
                session.confirm_sent(&frame.delivery_event_ids)?;
            }
            assert!(
                session
                    .snapshot()
                    .delivery_states
                    .iter()
                    .all(|state| *state == DeliveryState::InFlight)
            );
        }
        Ok(())
    }

    #[test]
    fn session_authentication_replay_preserves_the_event_count_limit() -> Result<(), SoftchatError>
    {
        let events = (0..=MAX_RELAY_BATCH_EVENTS)
            .map(|timestamp| event(timestamp as u64))
            .collect::<Result<Vec<_>, _>>()?;
        let mut session = RelaySession::new();
        for chunk in events.chunks(MAX_RELAY_BATCH_EVENTS) {
            session.publish(chunk.to_vec())?;
        }
        session.connect()?;
        session.transport_connected()?;
        session.authenticated()?;
        let frames = session.drain_outbound(8)?;
        assert_eq!(frames.len(), 2);
        assert_eq!(frames[0].delivery_event_ids.len(), MAX_RELAY_BATCH_EVENTS);
        assert_eq!(frames[1].delivery_event_ids.len(), 1);
        assert_eq!(
            frames
                .iter()
                .flat_map(|frame| frame.delivery_event_ids.clone())
                .collect::<Vec<_>>(),
            events
                .iter()
                .map(|event| event.id().to_hex())
                .collect::<BTreeSet<_>>()
                .into_iter()
                .collect::<Vec<_>>()
        );
        Ok(())
    }

    #[test]
    fn session_replays_subscriptions_and_tracks_eose_ingestion_and_delivery()
    -> Result<(), SoftchatError> {
        let mut session = RelaySession::new();
        session.subscribe("messages".to_owned(), vec![RelayFilter::default()])?;
        assert_eq!(
            session.connect()?.kind,
            RelaySessionActionKind::OpenTransport
        );
        session.transport_connected()?;
        session.authenticated()?;
        assert_eq!(session.drain_outbound(8)?.len(), 1);

        let authored = event(1)?;
        session.publish(vec![authored.clone()])?;
        let outbound = session.drain_outbound(8)?;
        assert_eq!(outbound.len(), 1);
        assert_eq!(outbound[0].delivery_event_ids, vec![authored.id().to_hex()]);
        session.confirm_sent(&outbound[0].delivery_event_ids)?;
        let ok = RelayResponseFrame::Ok(BatchAcknowledgement {
            event_id: authored.id().to_hex(),
            accepted: true,
            message: String::new(),
        })
        .to_json()?;
        assert_eq!(
            session.receive(&ok)?[0]
                .delivery_decision
                .as_ref()
                .map(|decision| decision.state),
            Some(DeliveryState::Accepted),
        );

        let received = event(2)?;
        let frame = RelayResponseFrame::Event {
            subscription_id: "messages".to_owned(),
            event: received.clone(),
        }
        .to_json()?;
        let action = session.receive(&frame)?;
        assert_eq!(action[0].kind, RelaySessionActionKind::PersistEvent);
        session.confirm_ingested(&received.id().to_hex())?;
        assert!(session.snapshot().pending_ingestion_ids.is_empty());

        let eose = RelayResponseFrame::Eose {
            subscription_id: "messages".to_owned(),
        }
        .to_json()?;
        assert_eq!(
            session.receive(&eose)?[0].kind,
            RelaySessionActionKind::SubscriptionReady
        );
        Ok(())
    }

    #[test]
    fn publish_rejects_duplicates_in_one_call_without_mutating_state() -> Result<(), SoftchatError>
    {
        let authored = event(1)?;
        for stage in 0..4 {
            let mut session = RelaySession::new();
            if stage >= 1 {
                session.connect()?;
            }
            if stage >= 2 {
                session.transport_connected()?;
            }
            if stage >= 3 {
                session.authenticated()?;
            }
            assert_eq!(
                session.publish(vec![authored.clone(), authored.clone()]),
                Err(SoftchatError::InvalidDeliveryState),
            );
            assert!(session.snapshot().delivery_event_ids.is_empty());
            assert!(session.drain_outbound(8)?.is_empty());
        }
        Ok(())
    }

    #[test]
    fn duplicate_inflight_event_emits_one_persistence_action() -> Result<(), SoftchatError> {
        let mut session = RelaySession::new();
        session.subscribe("messages".to_owned(), vec![RelayFilter::default()])?;
        session.connect()?;
        session.transport_connected()?;
        session.authenticated()?;
        session.drain_outbound(8)?;
        let received = event(2)?;
        let frame = RelayResponseFrame::Event {
            subscription_id: "messages".to_owned(),
            event: received,
        }
        .to_json()?;

        assert_eq!(session.receive(&frame)?.len(), 1);
        assert!(session.receive(&frame)?.is_empty());
        assert_eq!(session.snapshot().pending_ingestion_ids.len(), 1);
        Ok(())
    }

    #[test]
    fn closed_ends_subscription_and_reports_bounded_reason() -> Result<(), SoftchatError> {
        let mut session = RelaySession::new();
        session.subscribe("messages".to_owned(), vec![RelayFilter::default()])?;
        session.connect()?;
        session.transport_connected()?;
        session.authenticated()?;
        session.drain_outbound(8)?;
        let frame = RelayResponseFrame::Closed {
            subscription_id: "messages".to_owned(),
            message: "rate-limited".to_owned(),
        }
        .to_json()?;

        let actions = session.receive(&frame)?;
        assert_eq!(actions.len(), 1);
        assert_eq!(actions[0].kind, RelaySessionActionKind::SubscriptionClosed);
        assert_eq!(actions[0].value, "messages");
        assert_eq!(actions[0].reason, "rate-limited");
        assert!(session.snapshot().subscriptions.is_empty());
        assert!(session.receive(&frame)?.is_empty());
        let unknown = RelayResponseFrame::Closed {
            subscription_id: "unknown".to_owned(),
            message: "not ours".to_owned(),
        }
        .to_json()?;
        assert!(session.receive(&unknown)?.is_empty());

        let mut authenticating = RelaySession::new();
        authenticating.subscribe("retained".to_owned(), vec![RelayFilter::default()])?;
        authenticating.connect()?;
        authenticating.transport_connected()?;
        let stale = RelayResponseFrame::Closed {
            subscription_id: "retained".to_owned(),
            message: "authentication required".to_owned(),
        }
        .to_json()?;
        assert!(authenticating.receive(&stale)?.is_empty());
        assert_eq!(
            authenticating.snapshot().subscriptions,
            vec!["retained".to_owned()],
        );
        Ok(())
    }

    #[test]
    fn disconnect_preserves_terminal_state_and_retries_only_inflight() -> Result<(), SoftchatError>
    {
        let mut session = RelaySession::new();
        session.connect()?;
        session.transport_connected()?;
        session.authenticated()?;
        let first = event(1)?;
        let second = event(2)?;
        session.publish(vec![first.clone(), second.clone()])?;
        let outbound = session.drain_outbound(8)?;
        session.confirm_sent(&outbound[0].delivery_event_ids)?;
        let accepted = RelayResponseFrame::Ok(BatchAcknowledgement {
            event_id: first.id().to_hex(),
            accepted: true,
            message: String::new(),
        })
        .to_json()?;
        session.receive(&accepted)?;
        let reconnect = session.transport_lost()?;
        assert_eq!(reconnect[0].kind, RelaySessionActionKind::ScheduleReconnect);
        let snapshot = session.snapshot();
        let states = snapshot
            .delivery_event_ids
            .iter()
            .cloned()
            .zip(snapshot.delivery_states)
            .collect::<BTreeMap<_, _>>();
        assert_eq!(
            states.get(&first.id().to_hex()),
            Some(&DeliveryState::Accepted)
        );
        assert_eq!(
            states.get(&second.id().to_hex()),
            Some(&DeliveryState::Retryable)
        );
        Ok(())
    }

    #[test]
    fn persisted_publish_intent_queues_after_authentication() -> Result<(), SoftchatError> {
        let mut session = RelaySession::new();
        let authored = event(1)?;
        session.publish(vec![authored.clone()])?;
        assert!(session.drain_outbound(8)?.is_empty());
        session.connect()?;
        session.transport_connected()?;
        session.authenticated()?;
        let outbound = session.drain_outbound(8)?;
        assert_eq!(outbound.len(), 1);
        assert!(outbound[0].frame.starts_with("[\"EVENT\","));
        let snapshot = session.snapshot();
        let state = snapshot
            .delivery_event_ids
            .iter()
            .position(|event_id| event_id == &authored.id().to_hex())
            .and_then(|index| snapshot.delivery_states.get(index).copied());
        assert_eq!(state, Some(DeliveryState::Pending));
        session.confirm_sent(&outbound[0].delivery_event_ids)?;
        let snapshot = session.snapshot();
        let state = snapshot
            .delivery_event_ids
            .iter()
            .position(|event_id| event_id == &authored.id().to_hex())
            .and_then(|index| snapshot.delivery_states.get(index).copied());
        assert_eq!(state, Some(DeliveryState::InFlight));
        Ok(())
    }

    #[test]
    fn acknowledgement_may_race_socket_handoff_and_confirmation_is_idempotent()
    -> Result<(), SoftchatError> {
        let mut session = RelaySession::new();
        session.connect()?;
        session.transport_connected()?;
        session.authenticated()?;
        let authored = event(1)?;
        session.publish(vec![authored.clone()])?;
        let outbound = session.drain_outbound(8)?;
        assert_eq!(
            session.snapshot().delivery_states,
            vec![DeliveryState::Pending]
        );
        let acknowledgement = RelayResponseFrame::Ok(BatchAcknowledgement {
            event_id: authored.id().to_hex(),
            accepted: true,
            message: String::new(),
        })
        .to_json()?;
        let decisions = session.receive(&acknowledgement)?;
        assert_eq!(
            decisions[0]
                .delivery_decision
                .as_ref()
                .map(|decision| decision.state),
            Some(DeliveryState::Accepted)
        );
        session.confirm_sent(&outbound[0].delivery_event_ids)?;
        assert_eq!(
            session.snapshot().delivery_states,
            vec![DeliveryState::Accepted]
        );
        let duplicate = session.receive(&acknowledgement)?;
        assert_eq!(
            duplicate[0]
                .delivery_decision
                .as_ref()
                .map(|decision| decision.state),
            Some(DeliveryState::Accepted)
        );
        Ok(())
    }

    #[test]
    fn queue_backpressure_does_not_partially_mutate_intent() -> Result<(), SoftchatError> {
        let mut session = RelaySession::new();
        session.connect()?;
        session.transport_connected()?;
        session.authenticated()?;
        for _ in 0..MAX_OUTBOUND_RELAY_FRAMES {
            session.enqueue("[]".to_owned(), Vec::new())?;
        }

        let authored = event(1)?;
        assert_eq!(
            session.publish(vec![authored.clone()]),
            Err(SoftchatError::RelaySessionQueueFull),
        );
        assert!(
            !session
                .snapshot()
                .delivery_event_ids
                .contains(&authored.id().to_hex())
        );
        assert_eq!(
            session.subscribe("blocked".to_owned(), vec![RelayFilter::default()]),
            Err(SoftchatError::RelaySessionQueueFull),
        );
        assert!(
            !session
                .snapshot()
                .subscriptions
                .contains(&"blocked".to_owned())
        );

        let mut closing = RelaySession::new();
        closing.subscribe("retained".to_owned(), vec![RelayFilter::default()])?;
        closing.connect()?;
        closing.transport_connected()?;
        closing.authenticated()?;
        closing.drain_outbound(MAX_OUTBOUND_RELAY_FRAMES)?;
        for _ in 0..MAX_OUTBOUND_RELAY_FRAMES {
            closing.enqueue("[]".to_owned(), Vec::new())?;
        }
        assert_eq!(
            closing.close_subscription("retained"),
            Err(SoftchatError::RelaySessionQueueFull),
        );
        assert_eq!(
            closing.snapshot().subscriptions,
            vec!["retained".to_owned()]
        );

        let mut replay = RelaySession::new();
        replay.subscribe("retained".to_owned(), vec![RelayFilter::default()])?;
        replay.publish(vec![event(2)?])?;
        replay.connect()?;
        replay.transport_connected()?;
        for _ in 0..MAX_OUTBOUND_RELAY_FRAMES {
            replay.enqueue("[]".to_owned(), Vec::new())?;
        }
        assert_eq!(
            replay.authenticated(),
            Err(SoftchatError::RelaySessionQueueFull),
        );
        assert_eq!(
            replay.snapshot().state,
            RelayConnectionState::Authenticating,
        );
        assert_eq!(replay.outbound.len(), MAX_OUTBOUND_RELAY_FRAMES);
        Ok(())
    }

    #[test]
    fn terminal_delivery_history_is_bounded_and_does_not_consume_pending_capacity()
    -> Result<(), SoftchatError> {
        let mut session = RelaySession::new();
        for index in 0..=MAX_TERMINAL_DELIVERY_RECORDS {
            let event_id = format!("{index:064x}");
            session
                .delivery
                .insert(event_id.clone(), DeliveryState::Accepted);
            session.remember_terminal_delivery(event_id);
        }
        assert_eq!(session.delivery.len(), MAX_TERMINAL_DELIVERY_RECORDS);
        assert_eq!(
            session.terminal_delivery_order.len(),
            MAX_TERMINAL_DELIVERY_RECORDS
        );
        assert!(!session.delivery.contains_key(&format!("{:064x}", 0)));
        session.publish(vec![event(1)?])?;
        assert_eq!(session.pending_events.len(), 1);
        Ok(())
    }

    #[test]
    fn authentication_challenge_discards_stale_frames_and_replays_pending_intent()
    -> Result<(), SoftchatError> {
        let mut session = RelaySession::new();
        session.connect()?;
        session.transport_connected()?;
        session.authenticated()?;
        let authored = event(1)?;
        session.publish(vec![authored.clone()])?;
        assert_eq!(session.outbound.len(), 1);

        let actions =
            session.receive(&RelayResponseFrame::Auth("challenge".to_owned()).to_json()?)?;
        assert_eq!(actions[0].kind, RelaySessionActionKind::Authenticate);
        assert!(session.outbound.is_empty());
        let identity = LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let authentication =
            crate::plan_nip42_authentication(&identity, "challenge", "wss://relay.example", 1_000)?;
        let authentication_id = authentication.id().to_hex();
        session.authenticate(authentication)?;
        let auth_frame = session.drain_outbound(1)?;
        assert!(auth_frame[0].delivery_event_ids.is_empty());
        assert!(matches!(
            crate::parse_client_relay_frame(&auth_frame[0].frame)?,
            ClientRelayFrame::Auth(_)
        ));
        session.receive(
            &RelayResponseFrame::Ok(BatchAcknowledgement {
                event_id: authentication_id,
                accepted: true,
                message: String::new(),
            })
            .to_json()?,
        )?;
        let replay = session.drain_outbound(1)?;
        assert_eq!(replay[0].delivery_event_ids, vec![authored.id().to_hex()]);
        Ok(())
    }

    #[test]
    fn persistence_result_is_total_unique_and_account_scoped() -> Result<(), SoftchatError> {
        let first = SignedEvent::from(&event(1)?);
        let second = SignedEvent::from(&event(2)?);
        let batch = IngestionBatch {
            account_id: "account-1".to_owned(),
            events: vec![first.clone(), second.clone()],
        };
        validate_ingestion_result(
            &batch,
            &IngestionResult {
                inserted_event_ids: vec![first.id],
                duplicate_event_ids: vec![second.id],
                quarantined_event_ids: Vec::new(),
            },
        )?;
        assert!(
            validate_ingestion_result(
                &batch,
                &IngestionResult {
                    inserted_event_ids: Vec::new(),
                    duplicate_event_ids: Vec::new(),
                    quarantined_event_ids: Vec::new(),
                },
            )
            .is_err()
        );
        Ok(())
    }

    #[test]
    fn delivery_classifier_separates_retryable_and_permanent_rejection() -> Result<(), SoftchatError>
    {
        let event_id = event(1)?.id().to_hex();
        assert_eq!(
            classify_delivery_acknowledgement(BatchAcknowledgement {
                event_id: event_id.clone(),
                accepted: false,
                message: "rate-limited: slow down".to_owned(),
            })?
            .state,
            DeliveryState::Retryable
        );
        assert_eq!(
            classify_delivery_acknowledgement(BatchAcknowledgement {
                event_id,
                accepted: false,
                message: "blocked: policy".to_owned(),
            })?
            .state,
            DeliveryState::Rejected
        );
        Ok(())
    }
}