softchat/
session.rs

1//! Serialized relay session, delivery decisions, and platform persistence contract.
2
3use std::collections::{BTreeMap, BTreeSet, VecDeque};
4use std::fmt;
5use std::sync::{Mutex, MutexGuard};
6
7use crate::{
8    BatchAcknowledgement, BatchEventState, ClientRelayFrame, MAX_RELAY_BATCH_EVENTS, RelayFilter,
9    RelayResponseFrame, SignedEvent, SignedNostrEvent, SoftchatError, parse_relay_response_frame,
10};
11use serde::Serialize;
12
13/// Maximum queued outbound frames in one session.
14pub const MAX_OUTBOUND_RELAY_FRAMES: usize = 1_024;
15/// Maximum simultaneous subscription identifiers.
16pub const MAX_RELAY_SUBSCRIPTIONS: usize = 128;
17/// Maximum received events awaiting authoritative platform persistence.
18pub const MAX_INFLIGHT_INGESTION: usize = 1_024;
19/// Maximum authored events retained for durable delivery intent.
20pub const MAX_PENDING_PUBLISHES: usize = 1_024;
21
22/// Transport-independent relay connection state.
23#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
24#[serde(rename_all = "camelCase")]
25#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
26pub enum RelayConnectionState {
27    /// No transport exists.
28    Disconnected,
29    /// The platform is opening a WebSocket.
30    Connecting,
31    /// A transport exists and authentication is pending.
32    Authenticating,
33    /// The session can send subscriptions and events.
34    Ready,
35    /// The host should delay before another connection attempt.
36    Backoff,
37    /// The caller permanently cancelled this session.
38    Cancelled,
39}
40
41/// Durable delivery state for one authored event.
42#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
43#[serde(rename_all = "camelCase")]
44#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
45pub enum DeliveryState {
46    /// Persisted locally and not currently in flight.
47    Pending,
48    /// Sent on the current connection and awaiting `OK`.
49    InFlight,
50    /// Relay accepted the event, including an idempotent duplicate.
51    Accepted,
52    /// Relay rejected the event permanently.
53    Rejected,
54    /// The event can be retried according to host policy.
55    Retryable,
56}
57
58/// Stable host-facing delivery decision.
59#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
60#[serde(rename_all = "camelCase")]
61#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
62pub struct DeliveryDecision {
63    /// Canonical event ID.
64    pub event_id: String,
65    /// New durable state.
66    pub state: DeliveryState,
67    /// Stable category independent of arbitrary relay prose.
68    pub category: String,
69    /// Bounded original relay message for redacted diagnostics.
70    pub relay_message: String,
71}
72
73/// Action kind emitted by the pure session.
74#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
75#[serde(rename_all = "camelCase")]
76#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
77pub enum RelaySessionActionKind {
78    /// Platform should open a WebSocket.
79    OpenTransport,
80    /// Platform should send `frame`.
81    SendFrame,
82    /// Host should build and enqueue a NIP-42 AUTH frame for `value`.
83    Authenticate,
84    /// Platform must transactionally ingest `event`.
85    PersistEvent,
86    /// Subscription reached EOSE.
87    SubscriptionReady,
88    /// One delivery reached a terminal or retry state.
89    DeliveryChanged,
90    /// Platform should close the current transport.
91    CloseTransport,
92    /// Host should schedule another connection attempt after `delay_ms`.
93    ScheduleReconnect,
94}
95
96/// One transport/platform action emitted by [`RelaySession`].
97#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
98#[serde(rename_all = "camelCase")]
99#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
100pub struct RelaySessionAction {
101    /// Action type.
102    pub kind: RelaySessionActionKind,
103    /// Wire frame for `SendFrame`, otherwise empty.
104    pub frame: String,
105    /// Subscription ID, challenge, event ID, or reason depending on `kind`.
106    pub value: String,
107    /// Signed event for `PersistEvent`, otherwise absent.
108    pub event: Option<SignedEvent>,
109    /// Delay for `ScheduleReconnect`, otherwise zero.
110    pub delay_ms: u64,
111    /// Delivery state for `DeliveryChanged`, otherwise absent.
112    pub delivery_state: Option<DeliveryState>,
113}
114
115/// Serializable non-secret session state for diagnostics and restoration policy.
116#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
117#[serde(rename_all = "camelCase")]
118#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
119pub struct RelaySessionSnapshot {
120    /// Current connection state.
121    pub state: RelayConnectionState,
122    /// Active subscription IDs.
123    pub subscriptions: Vec<String>,
124    /// Subscriptions that observed EOSE on this connection.
125    pub ready_subscriptions: Vec<String>,
126    /// Event IDs and delivery states at matching indices.
127    pub delivery_event_ids: Vec<String>,
128    /// Event states at matching indices.
129    pub delivery_states: Vec<DeliveryState>,
130    /// Verified inbound IDs waiting for a platform transaction.
131    pub pending_ingestion_ids: Vec<String>,
132    /// Current reconnect attempt.
133    pub reconnect_attempt: u32,
134}
135
136/// One bounded atomic ingestion request to a platform-owned store.
137#[derive(Clone, Debug, Eq, PartialEq)]
138#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
139pub struct IngestionBatch {
140    /// Stable account partition identifier; never a secret key.
141    pub account_id: String,
142    /// Verified events to insert idempotently in one transaction.
143    pub events: Vec<SignedEvent>,
144}
145
146/// Result returned by an authoritative Room, GRDB, IndexedDB, or host adapter.
147#[derive(Clone, Debug, Eq, PartialEq)]
148#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
149pub struct IngestionResult {
150    /// IDs inserted for the first time.
151    pub inserted_event_ids: Vec<String>,
152    /// IDs already present with identical canonical data.
153    pub duplicate_event_ids: Vec<String>,
154    /// IDs quarantined instead of projected.
155    pub quarantined_event_ids: Vec<String>,
156}
157
158/// Synchronous semantic contract for platform persistence adapters.
159///
160/// Platform-specific implementations normally bridge this operation to a
161/// transaction on a platform executor. The shared layer never owns Room,
162/// GRDB, IndexedDB, files, schemas, or account lifecycle.
163pub trait PersistenceAdapter {
164    /// Atomically ingest or quarantine every event in `batch`.
165    ///
166    /// # Errors
167    ///
168    /// The adapter must return a stable redacted failure and must not report
169    /// partial success outside one committed transaction.
170    fn ingest(&self, batch: &IngestionBatch) -> Result<IngestionResult, SoftchatError>;
171}
172
173/// One deterministic transport-independent relay session.
174#[derive(Debug)]
175pub struct RelaySession {
176    state: RelayConnectionState,
177    subscriptions: BTreeMap<String, Vec<RelayFilter>>,
178    ready_subscriptions: BTreeSet<String>,
179    delivery: BTreeMap<String, DeliveryState>,
180    pending_events: BTreeMap<String, SignedNostrEvent>,
181    pending_ingestion: BTreeSet<String>,
182    outbound: VecDeque<String>,
183    reconnect_attempt: u32,
184}
185
186/// Thread-safe generated-binding owner for one serialized relay session.
187#[cfg_attr(feature = "native-bindings", derive(uniffi::Object))]
188pub struct RelaySessionHandle {
189    session: Mutex<RelaySession>,
190}
191
192impl fmt::Debug for RelaySessionHandle {
193    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
194        formatter
195            .debug_struct("RelaySessionHandle")
196            .field("snapshot", &lock(&self.session).snapshot())
197            .finish()
198    }
199}
200
201impl Default for RelaySessionHandle {
202    fn default() -> Self {
203        Self::new()
204    }
205}
206
207#[cfg_attr(feature = "native-bindings", uniffi::export)]
208impl RelaySessionHandle {
209    /// Construct an empty disconnected session.
210    #[cfg_attr(feature = "native-bindings", uniffi::constructor)]
211    #[must_use]
212    pub fn new() -> Self {
213        Self {
214            session: Mutex::new(RelaySession::new()),
215        }
216    }
217
218    /// Begin a platform-owned WebSocket connection attempt.
219    ///
220    /// # Errors
221    ///
222    /// Returns a stable invalid-transition error.
223    pub fn connect(&self) -> Result<RelaySessionAction, SoftchatError> {
224        lock(&self.session).connect()
225    }
226
227    /// Notify the session that the native transport is connected.
228    ///
229    /// # Errors
230    ///
231    /// Returns a stable invalid-transition error.
232    pub fn transport_connected(&self) -> Result<(), SoftchatError> {
233        lock(&self.session).transport_connected()
234    }
235
236    /// Mark relay authentication complete and replay retained intent.
237    ///
238    /// # Errors
239    ///
240    /// Returns a stable transition or bounded-queue error.
241    pub fn authenticated(&self) -> Result<(), SoftchatError> {
242        lock(&self.session).authenticated()
243    }
244
245    /// Add or replace a subscription using strict filter JSON objects.
246    ///
247    /// # Errors
248    ///
249    /// Returns a stable filter, transition, or bounded-queue error.
250    pub fn subscribe(
251        &self,
252        subscription_id: String,
253        filter_json: Vec<String>,
254    ) -> Result<(), SoftchatError> {
255        let filters = filter_json
256            .iter()
257            .map(|json| RelayFilter::from_json(json))
258            .collect::<Result<Vec<_>, _>>()?;
259        lock(&self.session).subscribe(subscription_id, filters)
260    }
261
262    /// Close one retained subscription.
263    ///
264    /// # Errors
265    ///
266    /// Returns a stable state or bounded-queue error.
267    pub fn close_subscription(&self, subscription_id: String) -> Result<(), SoftchatError> {
268        lock(&self.session).close_subscription(&subscription_id)
269    }
270
271    /// Queue already-persisted authored events for delivery.
272    ///
273    /// # Errors
274    ///
275    /// Returns a stable validation, state, or queue error.
276    pub fn publish(&self, events: Vec<SignedEvent>) -> Result<(), SoftchatError> {
277        let events = events
278            .into_iter()
279            .map(SignedNostrEvent::try_from)
280            .collect::<Result<Vec<_>, _>>()?;
281        lock(&self.session).publish(events)
282    }
283
284    /// Consume one complete native WebSocket text frame.
285    ///
286    /// # Errors
287    ///
288    /// Returns a stable frame, correlation, state, or backpressure error.
289    pub fn receive(&self, frame_json: String) -> Result<Vec<RelaySessionAction>, SoftchatError> {
290        lock(&self.session).receive(&frame_json)
291    }
292
293    /// Confirm an authoritative platform transaction handled an event.
294    ///
295    /// # Errors
296    ///
297    /// Returns a stable persistence-result error for unknown IDs.
298    pub fn confirm_ingested(&self, event_id: String) -> Result<(), SoftchatError> {
299        lock(&self.session).confirm_ingested(&event_id)
300    }
301
302    /// Handle transport loss and return the reconnect plan.
303    ///
304    /// # Errors
305    ///
306    /// Returns a stable invalid-transition error.
307    pub fn transport_lost(&self) -> Result<Vec<RelaySessionAction>, SoftchatError> {
308        lock(&self.session).transport_lost()
309    }
310
311    /// Drain at most `limit` ordered outbound text frames.
312    ///
313    /// # Errors
314    ///
315    /// Returns a stable error for a zero limit.
316    pub fn drain_outbound(&self, limit: u32) -> Result<Vec<String>, SoftchatError> {
317        let limit = usize::try_from(limit).map_err(|_| SoftchatError::InvalidRelaySession)?;
318        lock(&self.session).drain_outbound(limit)
319    }
320
321    /// Return one redacted serializable snapshot.
322    #[must_use]
323    pub fn snapshot(&self) -> RelaySessionSnapshot {
324        lock(&self.session).snapshot()
325    }
326
327    /// Permanently cancel the session and clear process-local queues.
328    #[must_use]
329    pub fn cancel(&self) -> RelaySessionAction {
330        lock(&self.session).cancel()
331    }
332}
333
334impl Default for RelaySession {
335    fn default() -> Self {
336        Self::new()
337    }
338}
339
340impl RelaySession {
341    /// Create one disconnected session with empty bounded queues.
342    #[must_use]
343    pub const fn new() -> Self {
344        Self {
345            state: RelayConnectionState::Disconnected,
346            subscriptions: BTreeMap::new(),
347            ready_subscriptions: BTreeSet::new(),
348            delivery: BTreeMap::new(),
349            pending_events: BTreeMap::new(),
350            pending_ingestion: BTreeSet::new(),
351            outbound: VecDeque::new(),
352            reconnect_attempt: 0,
353        }
354    }
355
356    /// Begin a platform-owned connection attempt.
357    ///
358    /// # Errors
359    ///
360    /// Returns [`SoftchatError::InvalidRelaySession`] unless disconnected or
361    /// waiting in backoff.
362    pub fn connect(&mut self) -> Result<RelaySessionAction, SoftchatError> {
363        if !matches!(
364            self.state,
365            RelayConnectionState::Disconnected | RelayConnectionState::Backoff
366        ) {
367            return Err(SoftchatError::InvalidRelaySession);
368        }
369        self.state = RelayConnectionState::Connecting;
370        Ok(action(RelaySessionActionKind::OpenTransport))
371    }
372
373    /// Notify the session that the native WebSocket is open.
374    ///
375    /// Existing subscriptions are replayed only after the host indicates
376    /// authentication is complete.
377    ///
378    /// # Errors
379    ///
380    /// Returns a stable transition error from any state except `Connecting`.
381    pub fn transport_connected(&mut self) -> Result<(), SoftchatError> {
382        if self.state != RelayConnectionState::Connecting {
383            return Err(SoftchatError::InvalidRelaySession);
384        }
385        self.state = RelayConnectionState::Authenticating;
386        self.ready_subscriptions.clear();
387        Ok(())
388    }
389
390    /// Mark NIP-42 authentication complete and replay session intent.
391    ///
392    /// # Errors
393    ///
394    /// Returns a transition or bounded-queue failure.
395    pub fn authenticated(&mut self) -> Result<(), SoftchatError> {
396        if self.state != RelayConnectionState::Authenticating {
397            return Err(SoftchatError::InvalidRelaySession);
398        }
399        self.state = RelayConnectionState::Ready;
400        self.reconnect_attempt = 0;
401        let subscriptions = self
402            .subscriptions
403            .iter()
404            .map(|(id, filters)| (id.clone(), filters.clone()))
405            .collect::<Vec<_>>();
406        for (subscription_id, filters) in subscriptions {
407            self.enqueue(
408                ClientRelayFrame::Req {
409                    subscription_id,
410                    filters,
411                }
412                .to_json()?,
413            )?;
414        }
415        let pending = self
416            .pending_events
417            .iter()
418            .filter(|(event_id, _)| {
419                matches!(
420                    self.delivery.get(*event_id),
421                    Some(DeliveryState::Pending | DeliveryState::Retryable)
422                )
423            })
424            .map(|(_, event)| event.clone())
425            .collect::<Vec<_>>();
426        for chunk in pending.chunks(MAX_RELAY_BATCH_EVENTS) {
427            let frame = if chunk.len() == 1 {
428                ClientRelayFrame::Event(
429                    chunk
430                        .first()
431                        .cloned()
432                        .ok_or(SoftchatError::InvalidRelaySession)?,
433                )
434            } else {
435                ClientRelayFrame::Events(chunk.to_vec())
436            }
437            .to_json()?;
438            self.enqueue(frame)?;
439            for event in chunk {
440                self.delivery
441                    .insert(event.id().to_hex(), DeliveryState::InFlight);
442            }
443        }
444        Ok(())
445    }
446
447    /// Add or replace one subscription intent.
448    ///
449    /// The request is queued immediately when ready and retained across
450    /// reconnects.
451    ///
452    /// # Errors
453    ///
454    /// Returns a validation, state, or bounded-queue failure.
455    pub fn subscribe(
456        &mut self,
457        subscription_id: String,
458        filters: Vec<RelayFilter>,
459    ) -> Result<(), SoftchatError> {
460        if self.state == RelayConnectionState::Cancelled
461            || (!self.subscriptions.contains_key(&subscription_id)
462                && self.subscriptions.len() >= MAX_RELAY_SUBSCRIPTIONS)
463        {
464            return Err(SoftchatError::InvalidRelaySession);
465        }
466        let frame = ClientRelayFrame::Req {
467            subscription_id: subscription_id.clone(),
468            filters: filters.clone(),
469        }
470        .to_json()?;
471        self.subscriptions.insert(subscription_id.clone(), filters);
472        self.ready_subscriptions.remove(&subscription_id);
473        if self.state == RelayConnectionState::Ready {
474            self.enqueue(frame)?;
475        }
476        Ok(())
477    }
478
479    /// Close one retained subscription.
480    ///
481    /// # Errors
482    ///
483    /// Returns a state or bounded-queue failure.
484    pub fn close_subscription(&mut self, subscription_id: &str) -> Result<(), SoftchatError> {
485        if self.subscriptions.remove(subscription_id).is_none() {
486            return Err(SoftchatError::InvalidRelaySession);
487        }
488        self.ready_subscriptions.remove(subscription_id);
489        if self.state == RelayConnectionState::Ready {
490            self.enqueue(
491                ClientRelayFrame::Close {
492                    subscription_id: subscription_id.to_owned(),
493                }
494                .to_json()?,
495            )?;
496        }
497        Ok(())
498    }
499
500    /// Queue one or more already persisted authored events for publication.
501    ///
502    /// # Errors
503    ///
504    /// Returns a state, duplicate-ID, event-validation, or queue failure.
505    pub fn publish(&mut self, events: Vec<SignedNostrEvent>) -> Result<(), SoftchatError> {
506        if self.state == RelayConnectionState::Cancelled
507            || events.is_empty()
508            || events.len() > MAX_RELAY_BATCH_EVENTS
509            || self.delivery.len().saturating_add(events.len()) > MAX_PENDING_PUBLISHES
510        {
511            return Err(SoftchatError::InvalidRelaySession);
512        }
513        for event in &events {
514            if self.delivery.contains_key(&event.id().to_hex()) {
515                return Err(SoftchatError::InvalidDeliveryState);
516            }
517        }
518        for event in &events {
519            self.delivery
520                .insert(event.id().to_hex(), DeliveryState::Pending);
521            self.pending_events
522                .insert(event.id().to_hex(), event.clone());
523        }
524        if self.state == RelayConnectionState::Ready {
525            let frame = if events.len() == 1 {
526                ClientRelayFrame::Event(
527                    events
528                        .first()
529                        .cloned()
530                        .ok_or(SoftchatError::InvalidRelaySession)?,
531                )
532            } else {
533                ClientRelayFrame::Events(events.clone())
534            }
535            .to_json()?;
536            self.enqueue(frame)?;
537            for event in events {
538                self.delivery
539                    .insert(event.id().to_hex(), DeliveryState::InFlight);
540            }
541        }
542        Ok(())
543    }
544
545    /// Consume one complete relay WebSocket text frame.
546    ///
547    /// # Errors
548    ///
549    /// Returns a frame, state, correlation, or backpressure failure.
550    pub fn receive(&mut self, frame_json: &str) -> Result<Vec<RelaySessionAction>, SoftchatError> {
551        if !matches!(
552            self.state,
553            RelayConnectionState::Authenticating | RelayConnectionState::Ready
554        ) {
555            return Err(SoftchatError::InvalidRelaySession);
556        }
557        let frame = parse_relay_response_frame(frame_json)?;
558        match frame {
559            RelayResponseFrame::Auth(challenge) => {
560                self.state = RelayConnectionState::Authenticating;
561                Ok(vec![RelaySessionAction {
562                    kind: RelaySessionActionKind::Authenticate,
563                    value: challenge,
564                    ..action(RelaySessionActionKind::Authenticate)
565                }])
566            }
567            RelayResponseFrame::Event {
568                subscription_id,
569                event,
570            } => {
571                if !self.subscriptions.contains_key(&subscription_id) {
572                    return Err(SoftchatError::InvalidRelaySession);
573                }
574                if self.pending_ingestion.len() >= MAX_INFLIGHT_INGESTION {
575                    return Err(SoftchatError::RelaySessionQueueFull);
576                }
577                let event_id = event.id().to_hex();
578                self.pending_ingestion.insert(event_id.clone());
579                Ok(vec![RelaySessionAction {
580                    kind: RelaySessionActionKind::PersistEvent,
581                    value: subscription_id,
582                    event: Some(SignedEvent::from(&event)),
583                    ..action(RelaySessionActionKind::PersistEvent)
584                }])
585            }
586            RelayResponseFrame::Eose { subscription_id } => {
587                if !self.subscriptions.contains_key(&subscription_id) {
588                    return Err(SoftchatError::InvalidRelaySession);
589                }
590                self.ready_subscriptions.insert(subscription_id.clone());
591                Ok(vec![RelaySessionAction {
592                    kind: RelaySessionActionKind::SubscriptionReady,
593                    value: subscription_id,
594                    ..action(RelaySessionActionKind::SubscriptionReady)
595                }])
596            }
597            RelayResponseFrame::Ok(acknowledgement) => {
598                let decision = classify_delivery_acknowledgement(acknowledgement)?;
599                let current = self
600                    .delivery
601                    .get(&decision.event_id)
602                    .copied()
603                    .ok_or(SoftchatError::InvalidDeliveryState)?;
604                if matches!(current, DeliveryState::Accepted | DeliveryState::Rejected)
605                    && current != decision.state
606                {
607                    return Err(SoftchatError::InvalidDeliveryState);
608                }
609                self.delivery
610                    .insert(decision.event_id.clone(), decision.state);
611                if matches!(
612                    decision.state,
613                    DeliveryState::Accepted | DeliveryState::Rejected
614                ) {
615                    self.pending_events.remove(&decision.event_id);
616                }
617                Ok(vec![RelaySessionAction {
618                    kind: RelaySessionActionKind::DeliveryChanged,
619                    value: decision.event_id,
620                    delivery_state: Some(decision.state),
621                    ..action(RelaySessionActionKind::DeliveryChanged)
622                }])
623            }
624            RelayResponseFrame::Closed {
625                subscription_id,
626                message,
627            } => {
628                self.ready_subscriptions.remove(&subscription_id);
629                Ok(vec![RelaySessionAction {
630                    kind: RelaySessionActionKind::SubscriptionReady,
631                    value: format!("{subscription_id}:{message}"),
632                    ..action(RelaySessionActionKind::SubscriptionReady)
633                }])
634            }
635            RelayResponseFrame::Notice(_) | RelayResponseFrame::Count { .. } => Ok(Vec::new()),
636        }
637    }
638
639    /// Confirm that an authoritative platform transaction handled an event.
640    ///
641    /// # Errors
642    ///
643    /// Returns a stable state error for an unknown or repeated confirmation.
644    pub fn confirm_ingested(&mut self, event_id: &str) -> Result<(), SoftchatError> {
645        if self.pending_ingestion.remove(event_id) {
646            Ok(())
647        } else {
648            Err(SoftchatError::InvalidPersistenceResult)
649        }
650    }
651
652    /// Handle transport loss and classify only nonterminal in-flight events for retry.
653    ///
654    /// # Errors
655    ///
656    /// Returns a transition error when no live transport existed.
657    pub fn transport_lost(&mut self) -> Result<Vec<RelaySessionAction>, SoftchatError> {
658        if !matches!(
659            self.state,
660            RelayConnectionState::Connecting
661                | RelayConnectionState::Authenticating
662                | RelayConnectionState::Ready
663        ) {
664            return Err(SoftchatError::InvalidRelaySession);
665        }
666        self.outbound.clear();
667        self.ready_subscriptions.clear();
668        for state in self.delivery.values_mut() {
669            if *state == DeliveryState::InFlight {
670                *state = DeliveryState::Retryable;
671            }
672        }
673        self.reconnect_attempt = self.reconnect_attempt.saturating_add(1);
674        self.state = RelayConnectionState::Backoff;
675        let exponent = self.reconnect_attempt.saturating_sub(1).min(6);
676        let delay_ms = 1_000_u64.saturating_mul(1_u64 << exponent);
677        Ok(vec![RelaySessionAction {
678            kind: RelaySessionActionKind::ScheduleReconnect,
679            delay_ms,
680            ..action(RelaySessionActionKind::ScheduleReconnect)
681        }])
682    }
683
684    /// Permanently cancel and clear process-local queues.
685    #[must_use]
686    pub fn cancel(&mut self) -> RelaySessionAction {
687        self.state = RelayConnectionState::Cancelled;
688        self.outbound.clear();
689        self.pending_events.clear();
690        self.pending_ingestion.clear();
691        action(RelaySessionActionKind::CloseTransport)
692    }
693
694    /// Drain at most `limit` queued frames for the platform WebSocket.
695    ///
696    /// # Errors
697    ///
698    /// Returns a stable state error for zero limits.
699    pub fn drain_outbound(&mut self, limit: usize) -> Result<Vec<String>, SoftchatError> {
700        if limit == 0 {
701            return Err(SoftchatError::InvalidRelaySession);
702        }
703        let count = limit.min(self.outbound.len());
704        Ok(self.outbound.drain(..count).collect())
705    }
706
707    /// Return a stable non-secret snapshot.
708    #[must_use]
709    pub fn snapshot(&self) -> RelaySessionSnapshot {
710        RelaySessionSnapshot {
711            state: self.state,
712            subscriptions: self.subscriptions.keys().cloned().collect(),
713            ready_subscriptions: self.ready_subscriptions.iter().cloned().collect(),
714            delivery_event_ids: self.delivery.keys().cloned().collect(),
715            delivery_states: self.delivery.values().copied().collect(),
716            pending_ingestion_ids: self.pending_ingestion.iter().cloned().collect(),
717            reconnect_attempt: self.reconnect_attempt,
718        }
719    }
720
721    fn enqueue(&mut self, frame: String) -> Result<(), SoftchatError> {
722        if self.outbound.len() >= MAX_OUTBOUND_RELAY_FRAMES {
723            return Err(SoftchatError::RelaySessionQueueFull);
724        }
725        self.outbound.push_back(frame);
726        Ok(())
727    }
728}
729
730/// Classify one standard `OK` without trusting its human-readable message as identity.
731///
732/// # Errors
733///
734/// Returns [`SoftchatError::InvalidDeliveryState`] for malformed values.
735#[cfg_attr(feature = "native-bindings", uniffi::export)]
736pub fn classify_delivery_acknowledgement(
737    acknowledgement: BatchAcknowledgement,
738) -> Result<DeliveryDecision, SoftchatError> {
739    let parsed = crate::track_batch_acknowledgements(
740        vec![acknowledgement.event_id.clone()],
741        vec![acknowledgement.clone()],
742    )
743    .map_err(|_| SoftchatError::InvalidDeliveryState)?;
744    let state = match parsed.states.first() {
745        Some(BatchEventState::Accepted) => DeliveryState::Accepted,
746        Some(BatchEventState::Rejected) => {
747            if is_retryable_message(&acknowledgement.message) {
748                DeliveryState::Retryable
749            } else {
750                DeliveryState::Rejected
751            }
752        }
753        _ => return Err(SoftchatError::InvalidDeliveryState),
754    };
755    let category = if acknowledgement.accepted {
756        if acknowledgement.message.starts_with("duplicate:") {
757            "duplicate"
758        } else {
759            "accepted"
760        }
761    } else if is_retryable_message(&acknowledgement.message) {
762        "retryable"
763    } else if acknowledgement.message.starts_with("auth-required:") {
764        "authentication"
765    } else if acknowledgement.message.starts_with("blocked:") {
766        "blocked"
767    } else if acknowledgement.message.starts_with("invalid:") {
768        "invalid"
769    } else {
770        "rejected"
771    };
772    Ok(DeliveryDecision {
773        event_id: acknowledgement.event_id,
774        state,
775        category: category.to_owned(),
776        relay_message: acknowledgement.message,
777    })
778}
779
780/// Verify that a platform transaction classified every requested event exactly once.
781///
782/// # Errors
783///
784/// Returns [`SoftchatError::InvalidPersistenceResult`] for missing, duplicate,
785/// or unknown IDs and accountless batches.
786#[cfg_attr(feature = "native-bindings", uniffi::export)]
787pub fn validate_ingestion_result(
788    batch: &IngestionBatch,
789    result: &IngestionResult,
790) -> Result<(), SoftchatError> {
791    if batch.account_id.is_empty() || batch.events.is_empty() {
792        return Err(SoftchatError::InvalidPersistenceResult);
793    }
794    let requested = batch
795        .events
796        .iter()
797        .map(|event| event.id.clone())
798        .collect::<BTreeSet<_>>();
799    if requested.len() != batch.events.len() {
800        return Err(SoftchatError::InvalidPersistenceResult);
801    }
802    let classified = result
803        .inserted_event_ids
804        .iter()
805        .chain(&result.duplicate_event_ids)
806        .chain(&result.quarantined_event_ids)
807        .cloned()
808        .collect::<Vec<_>>();
809    let unique = classified.iter().cloned().collect::<BTreeSet<_>>();
810    if unique.len() != classified.len() || unique != requested {
811        return Err(SoftchatError::InvalidPersistenceResult);
812    }
813    Ok(())
814}
815
816fn is_retryable_message(message: &str) -> bool {
817    ["rate-limited:", "error:", "unavailable:", "timeout:"]
818        .iter()
819        .any(|prefix| message.starts_with(prefix))
820}
821
822fn action(kind: RelaySessionActionKind) -> RelaySessionAction {
823    RelaySessionAction {
824        kind,
825        frame: String::new(),
826        value: String::new(),
827        event: None,
828        delay_ms: 0,
829        delivery_state: None,
830    }
831}
832
833fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
834    match mutex.lock() {
835        Ok(guard) => guard,
836        Err(poisoned) => poisoned.into_inner(),
837    }
838}
839
840#[cfg(test)]
841mod tests {
842    use crate::{LocalIdentity, NostrEventDraft, NostrEventKind};
843
844    use super::*;
845
846    const ALICE_SECRET: &str = "5c0c523f52a5b6fad39ed2403092df8cebc36318b39383bca6c00808626fab3a";
847
848    fn event(created_at: u64) -> Result<SignedNostrEvent, SoftchatError> {
849        LocalIdentity::from_secret_hex(ALICE_SECRET)?.sign_event(NostrEventDraft::new(
850            created_at,
851            NostrEventKind::SHORT_TEXT_NOTE,
852            Vec::new(),
853            "test",
854        )?)
855    }
856
857    #[test]
858    fn session_replays_subscriptions_and_tracks_eose_ingestion_and_delivery()
859    -> Result<(), SoftchatError> {
860        let mut session = RelaySession::new();
861        session.subscribe("messages".to_owned(), vec![RelayFilter::default()])?;
862        assert_eq!(
863            session.connect()?.kind,
864            RelaySessionActionKind::OpenTransport
865        );
866        session.transport_connected()?;
867        session.authenticated()?;
868        assert_eq!(session.drain_outbound(8)?.len(), 1);
869
870        let authored = event(1)?;
871        session.publish(vec![authored.clone()])?;
872        assert_eq!(session.drain_outbound(8)?.len(), 1);
873        let ok = RelayResponseFrame::Ok(BatchAcknowledgement {
874            event_id: authored.id().to_hex(),
875            accepted: true,
876            message: String::new(),
877        })
878        .to_json()?;
879        assert_eq!(
880            session.receive(&ok)?[0].delivery_state,
881            Some(DeliveryState::Accepted)
882        );
883
884        let received = event(2)?;
885        let frame = RelayResponseFrame::Event {
886            subscription_id: "messages".to_owned(),
887            event: received.clone(),
888        }
889        .to_json()?;
890        let action = session.receive(&frame)?;
891        assert_eq!(action[0].kind, RelaySessionActionKind::PersistEvent);
892        session.confirm_ingested(&received.id().to_hex())?;
893        assert!(session.snapshot().pending_ingestion_ids.is_empty());
894
895        let eose = RelayResponseFrame::Eose {
896            subscription_id: "messages".to_owned(),
897        }
898        .to_json()?;
899        assert_eq!(
900            session.receive(&eose)?[0].kind,
901            RelaySessionActionKind::SubscriptionReady
902        );
903        Ok(())
904    }
905
906    #[test]
907    fn disconnect_preserves_terminal_state_and_retries_only_inflight() -> Result<(), SoftchatError>
908    {
909        let mut session = RelaySession::new();
910        session.connect()?;
911        session.transport_connected()?;
912        session.authenticated()?;
913        let first = event(1)?;
914        let second = event(2)?;
915        session.publish(vec![first.clone(), second.clone()])?;
916        let accepted = RelayResponseFrame::Ok(BatchAcknowledgement {
917            event_id: first.id().to_hex(),
918            accepted: true,
919            message: String::new(),
920        })
921        .to_json()?;
922        session.receive(&accepted)?;
923        let reconnect = session.transport_lost()?;
924        assert_eq!(reconnect[0].kind, RelaySessionActionKind::ScheduleReconnect);
925        let snapshot = session.snapshot();
926        let states = snapshot
927            .delivery_event_ids
928            .iter()
929            .cloned()
930            .zip(snapshot.delivery_states)
931            .collect::<BTreeMap<_, _>>();
932        assert_eq!(
933            states.get(&first.id().to_hex()),
934            Some(&DeliveryState::Accepted)
935        );
936        assert_eq!(
937            states.get(&second.id().to_hex()),
938            Some(&DeliveryState::Retryable)
939        );
940        Ok(())
941    }
942
943    #[test]
944    fn persisted_publish_intent_queues_after_authentication() -> Result<(), SoftchatError> {
945        let mut session = RelaySession::new();
946        let authored = event(1)?;
947        session.publish(vec![authored.clone()])?;
948        assert!(session.drain_outbound(8)?.is_empty());
949        session.connect()?;
950        session.transport_connected()?;
951        session.authenticated()?;
952        let outbound = session.drain_outbound(8)?;
953        assert_eq!(outbound.len(), 1);
954        assert!(outbound[0].starts_with("[\"EVENT\","));
955        let snapshot = session.snapshot();
956        let state = snapshot
957            .delivery_event_ids
958            .iter()
959            .position(|event_id| event_id == &authored.id().to_hex())
960            .and_then(|index| snapshot.delivery_states.get(index).copied());
961        assert_eq!(state, Some(DeliveryState::InFlight));
962        Ok(())
963    }
964
965    #[test]
966    fn persistence_result_is_total_unique_and_account_scoped() -> Result<(), SoftchatError> {
967        let first = SignedEvent::from(&event(1)?);
968        let second = SignedEvent::from(&event(2)?);
969        let batch = IngestionBatch {
970            account_id: "account-1".to_owned(),
971            events: vec![first.clone(), second.clone()],
972        };
973        validate_ingestion_result(
974            &batch,
975            &IngestionResult {
976                inserted_event_ids: vec![first.id],
977                duplicate_event_ids: vec![second.id],
978                quarantined_event_ids: Vec::new(),
979            },
980        )?;
981        assert!(
982            validate_ingestion_result(
983                &batch,
984                &IngestionResult {
985                    inserted_event_ids: Vec::new(),
986                    duplicate_event_ids: Vec::new(),
987                    quarantined_event_ids: Vec::new(),
988                },
989            )
990            .is_err()
991        );
992        Ok(())
993    }
994
995    #[test]
996    fn delivery_classifier_separates_retryable_and_permanent_rejection() -> Result<(), SoftchatError>
997    {
998        let event_id = event(1)?.id().to_hex();
999        assert_eq!(
1000            classify_delivery_acknowledgement(BatchAcknowledgement {
1001                event_id: event_id.clone(),
1002                accepted: false,
1003                message: "rate-limited: slow down".to_owned(),
1004            })?
1005            .state,
1006            DeliveryState::Retryable
1007        );
1008        assert_eq!(
1009            classify_delivery_acknowledgement(BatchAcknowledgement {
1010                event_id,
1011                accepted: false,
1012                message: "blocked: policy".to_owned(),
1013            })?
1014            .state,
1015            DeliveryState::Rejected
1016        );
1017        Ok(())
1018    }
1019}
softchat/session.rs

softchat/
session.rs
