softchat/

account_transport.rs

//! Correlated account transport actions over the Rust-owned relay state.
//!
//! At this low-level boundary the host executes WebSocket effects only. This
//! module retains protocol, ingestion, delivery, authentication, retry, and
//! relay-selection decisions.

use std::cell::Cell;
use std::collections::{BTreeMap, BTreeSet, VecDeque};

use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};

use crate::account_diagnostics::{
    AccountLogCategory, AccountLogCounterKind as Counter, AccountLogDirection, AccountLogLevel,
    Diagnostic,
};
use crate::message::LocalIdentityHandle;
use crate::runtime_facade::{
    SyncTransportFrame, parse_sync_message_frame, plan_sync_transport, resolve_sync_action,
    resolve_sync_actions,
};
use crate::{
    AccountDatabase, ClaimedDelivery, DeliveryClaim, DeliveryDecision, DeliveryState,
    ProjectionKind, RelayConnectionState, RelayDeliveryResult, RelayDeliveryResultKind,
    RelayEndpointPlan, RelayFailureKind, RelayResponseFrame, RelayRetryAction, RelaySession,
    RelaySessionAction, RelaySessionActionKind, SignedNostrEvent, SoftchatError, SyncAction,
    SyncActionKind, SyncEngine, SyncEngineSnapshot, SyncPhase, parse_relay_endpoint,
    parse_relay_response_frame, plan_nip42_authentication, plan_relay_retry,
};

const MAX_TRANSPORT_ID_BYTES: usize = 128;
pub(crate) const MAX_TRANSPORT_ACTIONS: usize = 128;
const MAX_TYPING_INDICATORS: usize = 1_024;
const DELIVERY_LEASE_SECONDS: u32 = 30;
const SYNC_ID: &str = "account-sync";
const SYNC_OVERLAP_SECONDS: i64 = 86_400;
pub(crate) const SYNC_FRAME_SIZE_LIMIT: u64 = 60_000;
const MAX_SYNC_RESEND_EVENTS: usize = crate::SYNC_MAX_PAGE_LIMIT as usize;
const TYPING_LIFETIME_SECONDS: i64 = 5;
const MAX_RETIRED_TRANSPORT_ACTIONS: usize = 1_024;

/// One system effect Android executes for the active account transport.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum AccountTransportActionKind {
    /// Open the selected WebSocket, optionally after the bounded delay.
    Connect,
    /// Send one complete ordered text frame.
    SendText,
    /// Wait for one complete incoming text frame or the bounded deadline.
    ReceiveText,
    /// Close the current WebSocket.
    Close,
}

/// One correlated, bounded platform action.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountTransportAction {
    /// Caller-supplied run identifier.
    pub run_id: String,
    /// Unique idempotency key for this action.
    pub action_id: String,
    /// Connection generation; increases after transport loss.
    pub generation: u64,
    /// System effect to execute.
    pub kind: AccountTransportActionKind,
    /// Canonical persisted endpoint for `Connect`, otherwise empty.
    pub configured_url: String,
    /// Ordinary WSS URL for `Connect`, otherwise empty.
    pub network_url: String,
    /// Pinned Noise responder key for `Connect`, otherwise empty.
    pub noise_remote_static_key: Vec<u8>,
    /// Complete relay frame for `SendText`, otherwise empty.
    pub text_frame: String,
    /// Delay before `Connect` or receive deadline for `ReceiveText`.
    pub delay_ms: u64,
}

/// Typed outcome of exactly one transport action.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum AccountTransportResultKind {
    /// The requested WebSocket is open.
    Connected,
    /// The complete text frame was written in order.
    FrameWritten,
    /// One complete text frame was received.
    FrameReceived,
    /// A receive deadline elapsed without a frame.
    TimedOut,
    /// The socket disconnected or failed temporarily.
    Disconnected,
    /// Android reports no usable network.
    NetworkUnavailable,
    /// Relay authentication failed permanently.
    AuthenticationFailed,
    /// Relay or network requested backoff.
    RateLimited,
    /// The selected relay violated the protocol.
    ProtocolFailed,
    /// Android cancelled the run.
    Cancelled,
}

/// Correlated result returned by the Android system executor.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountTransportResult {
    /// Run copied from the action.
    pub run_id: String,
    /// Action copied from the action.
    pub action_id: String,
    /// Generation copied from the action.
    pub generation: u64,
    /// Typed outcome.
    pub kind: AccountTransportResultKind,
    /// Complete received frame for `FrameReceived`, otherwise empty.
    pub text_frame: String,
}

/// One authenticated non-durable typing indicator retained until expiry.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountTypingIndicator {
    /// Stable conversation identifier.
    pub conversation_id: String,
    /// Authenticated remote author.
    pub author_public_key: String,
    /// Portable wall-clock expiry selected by Rust.
    pub expires_at: i64,
}

/// Complete result of one correlated state-machine transition.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountTransportBatch {
    /// Newly committed system actions in execution order.
    pub actions: Vec<AccountTransportAction>,
    /// Rust-owned connection state after the transition.
    pub connection_state: RelayConnectionState,
    /// Whether no delivery is currently leased or claimable.
    pub idle: bool,
    /// Durable account revision observed after the transition.
    pub revision: i64,
    /// Current authenticated non-durable typing indicators.
    pub typing_indicators: Vec<AccountTypingIndicator>,
    /// Rust-owned synchronization state, absent before start or after reset.
    pub sync: Option<SyncEngineSnapshot>,
}

#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub(crate) struct StoredAccountTransportOutcome {
    pub actions: Vec<AccountTransportAction>,
    pub connection_state: RelayConnectionState,
    pub idle: bool,
    pub revision: i64,
    #[serde(default)]
    pub typing_indicators: Vec<AccountTypingIndicator>,
    #[serde(default)]
    pub sync: Option<SyncEngineSnapshot>,
}

impl StoredAccountTransportOutcome {
    pub(crate) fn into_batch(self) -> AccountTransportBatch {
        AccountTransportBatch {
            actions: self.actions,
            connection_state: self.connection_state,
            idle: self.idle,
            revision: self.revision,
            typing_indicators: self.typing_indicators,
            sync: self.sync,
        }
    }
}

impl From<&AccountTransportBatch> for StoredAccountTransportOutcome {
    fn from(value: &AccountTransportBatch) -> Self {
        Self {
            actions: value.actions.clone(),
            connection_state: value.connection_state,
            idle: value.idle,
            revision: value.revision,
            typing_indicators: value.typing_indicators.clone(),
            sync: value.sync.clone(),
        }
    }
}

#[derive(Debug)]
struct PendingAction {
    frame_class: &'static str,
    kind: AccountTransportActionKind,
    delivery_event_ids: Vec<String>,
}

#[derive(Debug)]
struct ActiveDelivery {
    claim: DeliveryClaim,
    event_to_intent: BTreeMap<String, String>,
    remaining_event_ids: BTreeSet<String>,
    socket_written_event_ids: BTreeSet<String>,
    expires_at: i64,
}

#[derive(Debug)]
struct ActiveSync {
    engine: SyncEngine,
    neg_subscription_id: String,
    fetch_subscription_base_id: Option<String>,
    fetch_subscription_ids: BTreeSet<String>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum SyncFrameDisposition {
    NotSync,
    Handled,
    Retry,
}

/// One process-local serialized transport run.
#[derive(Debug)]
pub(crate) struct AccountTransportRun {
    run_id: String,
    generation: u64,
    next_action_sequence: u64,
    relay_url: String,
    endpoint: RelayEndpointPlan,
    inbox_filters: Vec<String>,
    session: RelaySession,
    pending: BTreeMap<String, PendingAction>,
    retired_pending: BTreeMap<String, PendingAction>,
    active_delivery: Option<ActiveDelivery>,
    raw_outbound: VecDeque<String>,
    sync_resend_backlog: VecDeque<String>,
    sync_resend_event_ids: BTreeSet<String>,
    sync: Option<ActiveSync>,
    authentication_fence: bool,
    authentication_event_id: Option<String>,
    typing_indicators: BTreeMap<(String, String), i64>,
    retry_attempt: u32,
    cancelled: bool,
    idle: bool,
    diagnostic_state: Cell<RelayConnectionState>,
}

impl AccountTransportRun {
    pub(crate) fn start(
        run_id: String,
        inbox_filters: Vec<String>,
        database: &AccountDatabase,
    ) -> Result<(Self, AccountTransportBatch), SoftchatError> {
        validate_transport_id(&run_id)?;
        let relay_url = database.active_relay_url()?;
        let endpoint = parse_relay_endpoint(&relay_url)?;
        let revision = database.info()?.revision;
        let mut run = Self {
            run_id,
            generation: 1,
            next_action_sequence: 0,
            relay_url,
            endpoint,
            inbox_filters,
            session: RelaySession::new(),
            pending: BTreeMap::new(),
            retired_pending: BTreeMap::new(),
            active_delivery: None,
            raw_outbound: VecDeque::new(),
            sync_resend_backlog: VecDeque::new(),
            sync_resend_event_ids: BTreeSet::new(),
            sync: None,
            authentication_fence: false,
            authentication_event_id: None,
            typing_indicators: BTreeMap::new(),
            retry_attempt: 0,
            cancelled: false,
            idle: false,
            diagnostic_state: Cell::new(RelayConnectionState::Connecting),
        };
        run.session.connect()?;
        let action = run.connect_action(0);
        database.record_diagnostic(crate::account_diagnostics::connection(
            RelayConnectionState::Connecting,
        ));
        Ok((
            run,
            AccountTransportBatch {
                actions: vec![action],
                connection_state: RelayConnectionState::Connecting,
                idle: false,
                revision,
                typing_indicators: Vec::new(),
                sync: None,
            },
        ))
    }

    pub(crate) fn validate_result_preflight(
        &self,
        result: &AccountTransportResult,
    ) -> Result<(), SoftchatError> {
        if result.run_id != self.run_id {
            return Err(SoftchatError::InvalidRelaySession);
        }
        if result.generation < self.generation {
            return Ok(());
        }
        if result.generation != self.generation || self.cancelled {
            return Err(SoftchatError::InvalidRelaySession);
        }
        let pending = self
            .retired_pending
            .get(&result.action_id)
            .or_else(|| self.pending.get(&result.action_id))
            .ok_or(SoftchatError::UnsupportedSystemAction)?;
        validate_result_for_action(result, pending.kind)
    }

    pub(crate) fn handle_result(
        &mut self,
        result: &AccountTransportResult,
        database: &mut AccountDatabase,
        identity: &LocalIdentityHandle,
        now: i64,
    ) -> Result<AccountTransportBatch, SoftchatError> {
        self.prune_typing(now);
        if result.run_id != self.run_id {
            return Err(SoftchatError::InvalidRelaySession);
        }
        if result.generation < self.generation {
            return self.batch(Vec::new(), database, false);
        }
        if result.generation != self.generation || self.cancelled {
            return Err(SoftchatError::InvalidRelaySession);
        }
        if let Some(retired) = self.retired_pending.remove(&result.action_id) {
            validate_result_for_action(result, retired.kind)?;
            return self.batch(Vec::new(), database, false);
        }
        let pending = self
            .pending
            .remove(&result.action_id)
            .ok_or(SoftchatError::UnsupportedSystemAction)?;
        validate_result_for_action(result, pending.kind)?;

        let mut actions = Vec::new();
        match result.kind {
            AccountTransportResultKind::Connected => {
                self.session.transport_connected()?;
                for (index, filter) in self.inbox_filters.iter().enumerate() {
                    self.session.subscribe(
                        format!("softchat-live-{index}"),
                        vec![crate::RelayFilter::from_json(filter)?],
                    )?;
                }
                self.session.authenticated()?;
                self.authentication_fence = false;
                self.authentication_event_id = None;
                self.retry_attempt = 0;
                self.start_sync(database, identity, now)?;
            }
            AccountTransportResultKind::FrameWritten => {
                database.record_diagnostic(crate::account_diagnostics::frame(
                    pending.frame_class,
                    AccountLogDirection::Tx,
                ));
                let intent_ids =
                    self.unresolved_intent_ids_for_events(&pending.delivery_event_ids)?;
                self.session.confirm_sent(&pending.delivery_event_ids)?;
                if !intent_ids.is_empty() {
                    let claim = self
                        .active_delivery
                        .as_ref()
                        .ok_or(SoftchatError::InvalidDeliveryState)?
                        .claim
                        .clone();
                    database.mark_socket_written(&claim, intent_ids.clone())?;
                    let written_intents = intent_ids.into_iter().collect::<BTreeSet<_>>();
                    if let Some(active) = self.active_delivery.as_mut() {
                        for event_id in &pending.delivery_event_ids {
                            if active
                                .event_to_intent
                                .get(event_id)
                                .is_some_and(|intent_id| written_intents.contains(intent_id))
                            {
                                active.socket_written_event_ids.insert(event_id.clone());
                            }
                        }
                    }
                }
            }
            AccountTransportResultKind::FrameReceived => {
                match self.process_inbound_frame(&result.text_frame, database, identity, now) {
                    Ok(None) => {}
                    Ok(Some(failure)) => {
                        return self.recover_transport(database, failure, now);
                    }
                    Err(error) => {
                        database.record_diagnostic(
                            Diagnostic::new(
                                AccountLogCategory::Relay,
                                "relay.input.rejected",
                                "Relay input rejected",
                            )
                            .failure(error),
                        );
                        match inbound_failure_disposition(error) {
                            Some(InboundFailureDisposition::Ignore) => {}
                            Some(InboundFailureDisposition::Recover(failure)) => {
                                return self.recover_transport(database, failure, now);
                            }
                            None => return Err(error),
                        }
                    }
                }
            }
            AccountTransportResultKind::TimedOut => {}
            AccountTransportResultKind::Disconnected => {
                return self.recover_transport(database, RelayFailureKind::Temporary, now);
            }
            AccountTransportResultKind::NetworkUnavailable => {
                return self.recover_transport(database, RelayFailureKind::NetworkUnavailable, now);
            }
            AccountTransportResultKind::AuthenticationFailed => {
                return self.recover_transport(database, RelayFailureKind::Authentication, now);
            }
            AccountTransportResultKind::RateLimited => {
                return self.recover_transport(database, RelayFailureKind::RateLimited, now);
            }
            AccountTransportResultKind::ProtocolFailed => {
                return self.recover_transport(database, RelayFailureKind::ProtocolPermanent, now);
            }
            AccountTransportResultKind::Cancelled => {
                self.release_active(database)?;
                let _ = self.session.cancel();
                self.pending.clear();
                self.retired_pending.clear();
                self.raw_outbound.clear();
                self.sync_resend_backlog.clear();
                self.sync_resend_event_ids.clear();
                self.typing_indicators.clear();
                if let Some(active) = self.sync.as_mut() {
                    let _ = active.engine.cancel();
                }
                self.sync = None;
                self.authentication_fence = false;
                self.authentication_event_id = None;
                self.cancelled = true;
                self.idle = true;
                return self.batch(Vec::new(), database, true);
            }
        }
        actions.append(&mut self.pump(database, now)?);
        self.batch(actions, database, false)
    }

    pub(crate) fn cancel(
        &mut self,
        database: &mut AccountDatabase,
    ) -> Result<AccountTransportBatch, SoftchatError> {
        if !self.cancelled {
            self.release_active(database)?;
            let _ = self.session.cancel();
            self.pending.clear();
            self.retired_pending.clear();
            self.raw_outbound.clear();
            self.sync_resend_backlog.clear();
            self.sync_resend_event_ids.clear();
            self.typing_indicators.clear();
            if let Some(active) = self.sync.as_mut() {
                let _ = active.engine.cancel();
            }
            self.sync = None;
            self.authentication_fence = false;
            self.authentication_event_id = None;
            self.cancelled = true;
            self.idle = true;
        }
        self.batch(Vec::new(), database, true)
    }

    pub(crate) fn wake(
        &mut self,
        database: &mut AccountDatabase,
        now: i64,
    ) -> Result<AccountTransportBatch, SoftchatError> {
        self.prune_typing(now);
        if database.active_relay_url()? != self.relay_url {
            return self.reconnect_selected(database);
        }
        let actions = self.pump(database, now)?;
        self.batch(actions, database, false)
    }

    pub(crate) fn restart_sync(
        &mut self,
        database: &mut AccountDatabase,
        _identity: &LocalIdentityHandle,
        _now: i64,
    ) -> Result<AccountTransportBatch, SoftchatError> {
        self.reconnect_selected(database)
    }

    fn reconnect_selected(
        &mut self,
        database: &mut AccountDatabase,
    ) -> Result<AccountTransportBatch, SoftchatError> {
        if let Some(mut active) = self.sync.take() {
            for subscription_id in std::mem::take(&mut active.fetch_subscription_ids) {
                let _ = self.session.close_subscription(&subscription_id);
            }
            let _ = active.engine.cancel();
        }
        self.raw_outbound.clear();
        self.sync_resend_backlog.clear();
        self.sync_resend_event_ids.clear();
        self.typing_indicators.clear();
        self.release_active(database)?;
        self.pending.clear();
        self.retired_pending.clear();
        self.authentication_fence = false;
        self.authentication_event_id = None;
        self.session.transport_lost()?;
        // Durable delivery intent is released above and becomes the source of
        // truth for the replacement connection. Do not retain the same event
        // in RelaySession as retryable as well, or authentication replay and
        // the next durable claim would enqueue it twice.
        self.session = RelaySession::new();
        self.relay_url = database.active_relay_url()?;
        self.endpoint = parse_relay_endpoint(&self.relay_url)?;
        self.generation = self
            .generation
            .checked_add(1)
            .ok_or(SoftchatError::InvalidRelaySession)?;
        self.session.connect()?;
        let action = self.connect_action(0);
        self.batch(vec![action], database, false)
    }

    fn apply_relay_actions(
        &mut self,
        actions: Vec<RelaySessionAction>,
        database: &mut AccountDatabase,
        identity: &LocalIdentityHandle,
        now: i64,
    ) -> Result<(), SoftchatError> {
        for action in actions {
            match action.kind {
                RelaySessionActionKind::Authenticate => {
                    self.enter_authentication_fence()?;
                    let created_at = u64::try_from(now)
                        .map_err(|_| SoftchatError::InvalidRelayAuthentication)?;
                    let event = identity.with_identity(|authority| {
                        plan_nip42_authentication(
                            authority,
                            &action.value,
                            &self.relay_url,
                            created_at,
                        )
                    })?;
                    self.authentication_event_id = Some(event.id().to_hex());
                    self.session.authenticate(event)?;
                }
                RelaySessionActionKind::PersistEvent => {
                    let event = action.event.ok_or(SoftchatError::InvalidRelaySession)?;
                    let event_id = event.id.clone();
                    let is_sync_fetch = self.sync.as_ref().is_some_and(|active| {
                        active.fetch_subscription_ids.contains(&action.value)
                    });
                    if is_sync_fetch {
                        let active = self.sync.as_ref().ok_or(SoftchatError::InvalidSyncState)?;
                        let base_id = active
                            .fetch_subscription_base_id
                            .as_deref()
                            .ok_or(SoftchatError::InvalidSyncState)?;
                        active
                            .engine
                            .validate_fetched_events(base_id, std::slice::from_ref(&event_id))?;
                    }
                    let stored = match identity.with_identity(|authority| {
                        database.ingest_from_relay(authority, vec![event], now, is_sync_fetch)
                    }) {
                        Ok(stored) => stored,
                        Err(error) if invalid_incoming_event(error) => {
                            database.record_diagnostic(
                                Diagnostic::new(
                                    AccountLogCategory::Relay,
                                    "relay.input.rejected",
                                    "Relay input rejected",
                                )
                                .failure(error),
                            );
                            self.session.discard_ingestion(&event_id)?;
                            if is_sync_fetch {
                                // An uncommittable requested ID cannot advance the
                                // checkpoint. Recover against another relay from
                                // durable history instead of retiring the account.
                                return Err(SoftchatError::InvalidSyncState);
                            }
                            continue;
                        }
                        Err(error) => return Err(error),
                    };
                    if is_sync_fetch {
                        let active = self.sync.as_mut().ok_or(SoftchatError::InvalidSyncState)?;
                        let base_id = active
                            .fetch_subscription_base_id
                            .as_deref()
                            .ok_or(SoftchatError::InvalidSyncState)?;
                        active
                            .engine
                            .record_fetched_event(base_id.to_owned(), event_id.clone())?;
                        active.engine.confirm_committed(vec![event_id.clone()])?;
                    }
                    let own_public_key =
                        identity.with_identity(|authority| Ok(authority.public_key().to_hex()))?;
                    for projection in stored.ephemeral_projections {
                        if projection.kind == ProjectionKind::Typing
                            && !projection.conversation_id.is_empty()
                            && projection.author_public_key != own_public_key
                        {
                            self.record_typing_indicator(
                                projection.conversation_id,
                                projection.author_public_key,
                                now.saturating_add(TYPING_LIFETIME_SECONDS),
                            );
                        }
                    }
                    self.session.confirm_ingested(&event_id)?;
                }
                RelaySessionActionKind::DeliveryChanged => {
                    let decision = action
                        .delivery_decision
                        .ok_or(SoftchatError::InvalidDeliveryState)?;
                    self.commit_delivery_decision(database, decision)?;
                }
                RelaySessionActionKind::SubscriptionReady => {
                    let sync_actions = {
                        let Some(active) = self.sync.as_mut() else {
                            continue;
                        };
                        if !active.fetch_subscription_ids.remove(&action.value) {
                            continue;
                        }
                        self.session.close_subscription(&action.value)?;
                        if !active.fetch_subscription_ids.is_empty() {
                            continue;
                        }
                        let base_id = active
                            .fetch_subscription_base_id
                            .take()
                            .ok_or(SoftchatError::InvalidSyncState)?;
                        let actions = active.engine.finish_fetch(base_id)?;
                        resolve_sync_actions(
                            database,
                            &mut active.engine,
                            actions,
                            SYNC_FRAME_SIZE_LIMIT,
                        )?
                    };
                    self.apply_sync_actions(sync_actions, database, identity)?;
                }
                RelaySessionActionKind::SubscriptionClosed => {}
                RelaySessionActionKind::OpenTransport
                | RelaySessionActionKind::SendFrame
                | RelaySessionActionKind::CloseTransport
                | RelaySessionActionKind::ScheduleReconnect => {
                    return Err(SoftchatError::UnsupportedSystemAction);
                }
            }
        }
        Ok(())
    }

    fn enter_authentication_fence(&mut self) -> Result<(), SoftchatError> {
        if self
            .retired_pending
            .len()
            .saturating_add(self.pending.len())
            > MAX_RETIRED_TRANSPORT_ACTIONS
        {
            return Err(SoftchatError::RelaySessionQueueFull);
        }
        self.retired_pending.append(&mut self.pending);
        if let Some(mut active) = self.sync.take() {
            for subscription_id in std::mem::take(&mut active.fetch_subscription_ids) {
                self.session.close_subscription(&subscription_id)?;
            }
            let _ = active.engine.cancel();
        }
        self.raw_outbound.clear();
        self.sync_resend_backlog.clear();
        self.sync_resend_event_ids.clear();
        self.authentication_fence = true;
        self.authentication_event_id = None;
        self.idle = false;
        Ok(())
    }

    fn should_ignore_during_authentication(&self, frame_json: &str) -> Result<bool, SoftchatError> {
        if !self.authentication_fence {
            return Ok(false);
        }
        let frame = parse_relay_response_frame(frame_json)?;
        Ok(match frame {
            RelayResponseFrame::Auth(_) => false,
            RelayResponseFrame::Ok(acknowledgement) => {
                self.authentication_event_id.as_deref() != Some(acknowledgement.event_id.as_str())
            }
            RelayResponseFrame::Event { .. }
            | RelayResponseFrame::Eose { .. }
            | RelayResponseFrame::Closed { .. }
            | RelayResponseFrame::Notice(_)
            | RelayResponseFrame::Count { .. } => true,
        })
    }

    fn start_sync(
        &mut self,
        database: &mut AccountDatabase,
        identity: &LocalIdentityHandle,
        now: i64,
    ) -> Result<(), SoftchatError> {
        if self.sync.is_some() {
            return Ok(());
        }
        let since_timestamp = database
            .sync_checkpoint(SYNC_ID)?
            .map(|checkpoint| checkpoint.saturating_sub(SYNC_OVERLAP_SECONDS));
        let suffix = format!("{}-{}", self.generation, self.next_action_sequence);
        let neg_subscription_id = format!("softchat-neg-{suffix}");
        let mut engine = SyncEngine::new(
            SYNC_ID.to_owned(),
            neg_subscription_id.clone(),
            format!("softchat-events-{suffix}"),
            since_timestamp,
            now,
        )?;
        let action = engine.begin()?;
        let action = resolve_sync_action(database, &mut engine, action, SYNC_FRAME_SIZE_LIMIT)?;
        self.sync = Some(ActiveSync {
            engine,
            neg_subscription_id,
            fetch_subscription_base_id: None,
            fetch_subscription_ids: BTreeSet::new(),
        });
        self.apply_sync_actions(vec![action], database, identity)?;
        database.record_diagnostic(Diagnostic::new(
            AccountLogCategory::Synchronization,
            "sync.started",
            "Synchronization started",
        ));
        Ok(())
    }

    fn handle_sync_frame(
        &mut self,
        frame_json: &str,
        database: &mut AccountDatabase,
        identity: &LocalIdentityHandle,
        _now: i64,
    ) -> Result<SyncFrameDisposition, SoftchatError> {
        let parsed = parse_relay_response_frame(frame_json);
        if let Ok(frame) = &parsed {
            database.record_diagnostic(crate::account_diagnostics::frame(
                crate::account_diagnostics::received_frame_class(frame),
                AccountLogDirection::Rx,
            ));
        }
        if let Ok(RelayResponseFrame::Ok(acknowledgement)) = parsed
            && self.sync_resend_event_ids.remove(&acknowledgement.event_id)
        {
            if acknowledgement.accepted {
                return Ok(SyncFrameDisposition::Handled);
            }
            return Ok(SyncFrameDisposition::Retry);
        }
        let Some(frame) = parse_sync_message_frame(frame_json)? else {
            return Ok(SyncFrameDisposition::NotSync);
        };
        let frame_class = match &frame {
            SyncTransportFrame::Message { .. } => "NEG_MSG",
            SyncTransportFrame::Error { .. } => "NEG_ERR",
            SyncTransportFrame::Control { .. } => "NEG_CONTROL",
        };
        database.record_diagnostic(crate::account_diagnostics::frame(
            frame_class,
            AccountLogDirection::Rx,
        ));
        if self.authentication_fence {
            return Ok(SyncFrameDisposition::Handled);
        }
        let (subscription_id, message) = match frame {
            SyncTransportFrame::Message {
                subscription_id,
                message,
            } => (subscription_id, message),
            SyncTransportFrame::Error {
                subscription_id, ..
            } => {
                return Ok(
                    if self
                        .sync
                        .as_ref()
                        .is_some_and(|active| active.neg_subscription_id == subscription_id)
                    {
                        SyncFrameDisposition::Retry
                    } else {
                        SyncFrameDisposition::Handled
                    },
                );
            }
            SyncTransportFrame::Control { subscription_id } => {
                return Ok(
                    if self
                        .sync
                        .as_ref()
                        .is_some_and(|active| active.neg_subscription_id == subscription_id)
                    {
                        SyncFrameDisposition::Retry
                    } else {
                        SyncFrameDisposition::Handled
                    },
                );
            }
        };
        if self
            .sync
            .as_ref()
            .is_some_and(|active| active.neg_subscription_id != subscription_id)
        {
            return Ok(SyncFrameDisposition::Handled);
        }
        let actions = {
            let active = self.sync.as_mut().ok_or(SoftchatError::InvalidSyncState)?;
            let actions = active.engine.reconcile(subscription_id, message)?;
            resolve_sync_actions(database, &mut active.engine, actions, SYNC_FRAME_SIZE_LIMIT)?
        };
        self.apply_sync_actions(actions, database, identity)?;
        Ok(SyncFrameDisposition::Handled)
    }

    fn process_inbound_frame(
        &mut self,
        frame_json: &str,
        database: &mut AccountDatabase,
        identity: &LocalIdentityHandle,
        now: i64,
    ) -> Result<Option<RelayFailureKind>, SoftchatError> {
        match self.handle_sync_frame(frame_json, database, identity, now)? {
            SyncFrameDisposition::Handled => return Ok(None),
            SyncFrameDisposition::Retry => return Ok(Some(RelayFailureKind::Temporary)),
            SyncFrameDisposition::NotSync => {}
        }
        if self.should_ignore_during_authentication(frame_json)? {
            return Ok(None);
        }
        let state_before = self.session.snapshot().state;
        let relay_actions = self.session.receive(frame_json)?;
        let subscription_closed = relay_actions
            .iter()
            .any(|action| action.kind == RelaySessionActionKind::SubscriptionClosed);
        let retry_failure = relay_actions
            .iter()
            .filter_map(|action| action.delivery_decision.as_ref())
            .find(|decision| decision.state == DeliveryState::Retryable)
            .map(|decision| {
                if decision.relay_message.starts_with("rate-limited:") {
                    RelayFailureKind::RateLimited
                } else {
                    RelayFailureKind::Temporary
                }
            });
        self.apply_relay_actions(relay_actions, database, identity, now)?;
        // Commit the retryable outcome before releasing the remaining claim.
        // A fresh session must reclaim it from SQLite, not publish the same ID
        // into a session that still remembers its previous delivery state.
        if retry_failure.is_some() {
            return Ok(retry_failure);
        }
        if subscription_closed {
            return Ok(Some(RelayFailureKind::Temporary));
        }
        if state_before == RelayConnectionState::Authenticating
            && self.session.snapshot().state == RelayConnectionState::Ready
        {
            self.authentication_fence = false;
            self.authentication_event_id = None;
            self.start_sync(database, identity, now)?;
        }
        Ok(None)
    }

    fn apply_sync_actions(
        &mut self,
        actions: Vec<SyncAction>,
        database: &mut AccountDatabase,
        identity: &LocalIdentityHandle,
    ) -> Result<(), SoftchatError> {
        let recipient_public_key =
            identity.with_identity(|authority| Ok(authority.public_key().to_hex()))?;
        let mut request_filters = BTreeMap::<String, Vec<String>>::new();
        for action in actions {
            let action = plan_sync_transport(action, &recipient_public_key)?;
            match action.kind {
                SyncActionKind::SendNegOpen
                | SyncActionKind::SendNegMessage
                | SyncActionKind::SendNegClose => {
                    if action.frame_json.is_empty() {
                        return Err(SoftchatError::InvalidSyncState);
                    }
                    self.raw_outbound.push_back(action.frame_json);
                }
                SyncActionKind::RequestEvents => {
                    if action.filters_json.is_empty() {
                        return Err(SoftchatError::InvalidSyncState);
                    }
                    request_filters
                        .entry(action.subscription_id)
                        .or_default()
                        .extend(action.filters_json);
                }
                SyncActionKind::ResendEvents => {
                    if self
                        .sync_resend_backlog
                        .len()
                        .saturating_add(action.event_ids.len())
                        > MAX_SYNC_RESEND_EVENTS
                    {
                        return Err(SoftchatError::RelaySessionQueueFull);
                    }
                    self.sync_resend_backlog.extend(action.event_ids);
                }
                SyncActionKind::Complete => {
                    if let Some(active) = &self.sync {
                        let snapshot = active.engine.snapshot();
                        database.record_diagnostic(
                            Diagnostic::new(
                                AccountLogCategory::Synchronization,
                                "sync.completed",
                                "Synchronization checkpoint committed",
                            )
                            .count(
                                Counter::RequestedEvents,
                                u64::from(snapshot.requested_count),
                            )
                            .count(
                                Counter::CommittedEvents,
                                u64::from(snapshot.committed_count),
                            ),
                        );
                    }
                }
                SyncActionKind::LoadSnapshot | SyncActionKind::SaveCheckpoint => {
                    return Err(SoftchatError::UnsupportedSystemAction);
                }
            }
        }
        if request_filters.len() > 1 {
            return Err(SoftchatError::InvalidSyncState);
        }
        for (subscription_id, filters) in request_filters {
            let parsed = filters
                .iter()
                .map(|filter| crate::RelayFilter::from_json(filter))
                .collect::<Result<Vec<_>, _>>()?;
            let subscriptions = shard_sync_event_request(&subscription_id, parsed)?;
            let active = self.sync.as_ref().ok_or(SoftchatError::InvalidSyncState)?;
            if active.fetch_subscription_base_id.is_some()
                || !active.fetch_subscription_ids.is_empty()
            {
                return Err(SoftchatError::InvalidSyncState);
            }
            self.session.subscribe_many(subscriptions.clone())?;
            let active = self.sync.as_mut().ok_or(SoftchatError::InvalidSyncState)?;
            active.fetch_subscription_base_id = Some(subscription_id);
            active.fetch_subscription_ids = subscriptions
                .into_iter()
                .map(|(subscription_id, _)| subscription_id)
                .collect();
        }
        if self.raw_outbound.len() > MAX_TRANSPORT_ACTIONS {
            return Err(SoftchatError::RelaySessionQueueFull);
        }
        Ok(())
    }

    fn commit_delivery_decision(
        &mut self,
        database: &mut AccountDatabase,
        decision: DeliveryDecision,
    ) -> Result<(), SoftchatError> {
        let Some(active) = self.active_delivery.as_mut() else {
            return Ok(());
        };
        if !active.remaining_event_ids.contains(&decision.event_id) {
            return Ok(());
        }
        let intent_id = active
            .event_to_intent
            .get(&decision.event_id)
            .cloned()
            .ok_or(SoftchatError::InvalidDeliveryState)?;
        let kind = match decision.state {
            DeliveryState::Accepted => RelayDeliveryResultKind::Accepted,
            DeliveryState::Rejected => RelayDeliveryResultKind::Rejected,
            DeliveryState::Retryable => RelayDeliveryResultKind::Retryable,
            DeliveryState::Pending | DeliveryState::InFlight => {
                return Err(SoftchatError::InvalidDeliveryState);
            }
        };
        database.apply_relay_results(
            &active.claim,
            vec![RelayDeliveryResult {
                intent_id,
                kind,
                category: decision.category,
            }],
        )?;
        active.remaining_event_ids.remove(&decision.event_id);
        if active.remaining_event_ids.is_empty() {
            self.active_delivery = None;
        }
        Ok(())
    }

    fn prune_typing(&mut self, now: i64) {
        self.typing_indicators
            .retain(|_, expires_at| *expires_at > now);
    }

    fn record_typing_indicator(
        &mut self,
        conversation_id: String,
        author_public_key: String,
        expires_at: i64,
    ) {
        let key = (conversation_id, author_public_key);
        if !self.typing_indicators.contains_key(&key)
            && self.typing_indicators.len() == MAX_TYPING_INDICATORS
            && let Some(oldest) = self
                .typing_indicators
                .iter()
                .min_by(|(left_key, left_expiry), (right_key, right_expiry)| {
                    left_expiry
                        .cmp(right_expiry)
                        .then_with(|| left_key.cmp(right_key))
                })
                .map(|(entry, _)| entry.clone())
        {
            self.typing_indicators.remove(&oldest);
        }
        self.typing_indicators.insert(key, expires_at);
    }

    fn pump(
        &mut self,
        database: &mut AccountDatabase,
        now: i64,
    ) -> Result<Vec<AccountTransportAction>, SoftchatError> {
        // Receive activity must not extend the durable delivery deadline.
        // The current result is applied before pumping, so a terminal ACK
        // arriving at the deadline still wins over timeout recovery.
        if self
            .active_delivery
            .as_ref()
            .is_some_and(|delivery| now >= delivery.expires_at)
        {
            return self
                .recover_transport(database, RelayFailureKind::Temporary, now)
                .map(|batch| batch.actions);
        }
        let mut actions = Vec::new();
        if self.session.snapshot().state == RelayConnectionState::Ready
            && self.active_delivery.is_none()
        {
            let lease_id = self.next_lease_id();
            if let Some(claimed) = database.claim_deliveries(
                lease_id,
                self.relay_url.clone(),
                now,
                DELIVERY_LEASE_SECONDS,
            )? {
                self.publish_claim(claimed, now)?;
                self.idle = false;
            } else {
                self.idle = true;
            }
        }
        let send_limit = MAX_TRANSPORT_ACTIONS.saturating_sub(1);
        for frame in self.session.drain_outbound(send_limit)? {
            actions.push(self.new_action(
                AccountTransportActionKind::SendText,
                String::new(),
                frame.frame,
                0,
                frame.delivery_event_ids,
            ));
        }
        while actions.len() < send_limit {
            let Some(frame) = self.raw_outbound.pop_front() else {
                break;
            };
            actions.push(self.new_action(
                AccountTransportActionKind::SendText,
                String::new(),
                frame,
                0,
                Vec::new(),
            ));
        }
        while actions.len() < send_limit {
            if self.sync_resend_backlog.is_empty() {
                break;
            }
            let count = self
                .sync_resend_backlog
                .len()
                .min(crate::MAX_RELAY_BATCH_EVENTS);
            let event_ids = self.sync_resend_backlog.drain(..count).collect();
            let events = database.event_jsons(event_ids)?;
            let mut frames = crate::session::batch_event_frames(
                events
                    .into_iter()
                    // Old account versions admitted standalone Nostr events
                    // that cannot fit even one EVENT frame. Preserve that
                    // history; it is also excluded from sync fingerprints.
                    .filter(|event| {
                        crate::session::validate_event_frame_json(&event.event_json).is_ok()
                    })
                    .map(|event| (event.event_id, event.event_json)),
            )?
            .into_iter();
            while actions.len() < send_limit {
                let Some(frame) = frames.next() else {
                    break;
                };
                self.sync_resend_event_ids.extend(frame.delivery_event_ids);
                actions.push(self.new_action(
                    AccountTransportActionKind::SendText,
                    String::new(),
                    frame.frame,
                    0,
                    Vec::new(),
                ));
            }
            // Byte splitting can fill this action batch before all loaded IDs
            // fit. Keep the rest in original order for the next pump.
            for event_id in frames.flat_map(|frame| frame.delivery_event_ids).rev() {
                self.sync_resend_backlog.push_front(event_id);
            }
        }
        let receive_pending = self
            .pending
            .values()
            .any(|pending| pending.kind == AccountTransportActionKind::ReceiveText);
        if matches!(
            self.session.snapshot().state,
            RelayConnectionState::Authenticating | RelayConnectionState::Ready
        ) && !receive_pending
        {
            let typing_delay_ms = self
                .typing_indicators
                .values()
                .min()
                .map(|expires_at| {
                    u64::try_from(expires_at.saturating_sub(now))
                        .unwrap_or_default()
                        .saturating_mul(1_000)
                        .clamp(1, 30_000)
                })
                .unwrap_or(30_000);
            actions.push(self.new_action(
                AccountTransportActionKind::ReceiveText,
                String::new(),
                String::new(),
                typing_delay_ms,
                Vec::new(),
            ));
        }
        if actions.len() > MAX_TRANSPORT_ACTIONS {
            return Err(SoftchatError::RelaySessionQueueFull);
        }
        let sync_complete = self
            .sync
            .as_ref()
            .is_none_or(|active| active.engine.snapshot().phase == SyncPhase::Complete);
        self.idle = self.idle
            && sync_complete
            && self.raw_outbound.is_empty()
            && self.sync_resend_backlog.is_empty()
            && self.sync_resend_event_ids.is_empty();
        Ok(actions)
    }

    fn publish_claim(&mut self, claimed: ClaimedDelivery, now: i64) -> Result<(), SoftchatError> {
        let mut events = Vec::with_capacity(claimed.payloads.len());
        let mut event_to_intent = BTreeMap::new();
        let mut remaining_event_ids = BTreeSet::new();
        for payload in &claimed.payloads {
            let event = SignedNostrEvent::from_json(&payload.event_json)?;
            let event_id = event.id().to_hex();
            if event_to_intent
                .insert(event_id.clone(), payload.intent_id.clone())
                .is_some()
            {
                return Err(SoftchatError::InvalidDeliveryState);
            }
            remaining_event_ids.insert(event_id);
            events.push(event);
        }
        self.session.publish(events)?;
        self.active_delivery = Some(ActiveDelivery {
            claim: claimed.claim,
            event_to_intent,
            remaining_event_ids,
            socket_written_event_ids: BTreeSet::new(),
            expires_at: now
                .checked_add(i64::from(DELIVERY_LEASE_SECONDS))
                .ok_or(SoftchatError::InvalidDeliveryState)?,
        });
        self.idle = false;
        Ok(())
    }

    fn recover_transport(
        &mut self,
        database: &mut AccountDatabase,
        failure: RelayFailureKind,
        now: i64,
    ) -> Result<AccountTransportBatch, SoftchatError> {
        self.release_active(database)?;
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Relay,
            "relay.retry_scheduled",
            "Relay reconnect scheduled",
        )
        .count(Counter::Attempts, u64::from(self.retry_attempt) + 1);
        diagnostic.level = AccountLogLevel::Warn;
        diagnostic.summarize = true;
        diagnostic.reason = crate::account_diagnostics::relay_failure_reason(failure);
        database.record_diagnostic(diagnostic);
        if self
            .sync
            .as_ref()
            .is_some_and(|active| active.engine.snapshot().phase != SyncPhase::Complete)
        {
            database.record_diagnostic(Diagnostic::new(
                AccountLogCategory::Synchronization,
                "sync.cancelled",
                "Synchronization cancelled before completion",
            ));
        }
        self.pending.clear();
        self.retired_pending.clear();
        self.raw_outbound.clear();
        self.sync_resend_backlog.clear();
        self.sync_resend_event_ids.clear();
        self.typing_indicators.clear();
        if let Some(active) = self.sync.as_mut() {
            let _ = active.engine.cancel();
        }
        self.sync = None;
        self.authentication_fence = false;
        self.authentication_event_id = None;
        self.session.transport_lost()?;
        // The active claim was released to durable retryable state. A fresh
        // relay session must recover it from SQLite exactly once instead of
        // also replaying the process-local pending event.
        self.session = RelaySession::new();
        let retry = plan_relay_retry(self.retry_attempt, failure, false);
        self.retry_attempt = self.retry_attempt.saturating_add(1);
        if retry.action == RelayRetryAction::RotateRelay {
            let mutation = database.rotate_failed_relay(self.relay_url.clone(), now)?;
            self.relay_url = mutation.active_relay_url;
            self.endpoint = parse_relay_endpoint(&self.relay_url)?;
        }
        self.generation = self
            .generation
            .checked_add(1)
            .ok_or(SoftchatError::InvalidRelaySession)?;
        self.session.connect()?;
        let action = self.connect_action(retry.delay_ms);
        self.batch(vec![action], database, false)
    }

    fn release_active(&mut self, database: &mut AccountDatabase) -> Result<(), SoftchatError> {
        let Some(active) = self.active_delivery.take() else {
            return Ok(());
        };
        if active.remaining_event_ids.is_empty() {
            return Ok(());
        }
        let remaining_intents = active
            .remaining_event_ids
            .iter()
            .filter_map(|event_id| active.event_to_intent.get(event_id))
            .cloned()
            .collect::<BTreeSet<_>>();
        let mut claim = active.claim;
        claim
            .intents
            .retain(|intent| remaining_intents.contains(&intent.intent_id));
        claim
            .mutations
            .retain(|mutation| remaining_intents.contains(&mutation.intent_id));
        if !claim.intents.is_empty() {
            database.release_claim(&claim)?;
        }
        Ok(())
    }

    fn unresolved_intent_ids_for_events(
        &self,
        event_ids: &[String],
    ) -> Result<Vec<String>, SoftchatError> {
        if event_ids.is_empty() {
            return Ok(Vec::new());
        }
        let snapshot = self.session.snapshot();
        let states = snapshot
            .delivery_event_ids
            .into_iter()
            .zip(snapshot.delivery_states)
            .collect::<BTreeMap<_, _>>();
        let mut intent_ids = Vec::new();
        for event_id in event_ids {
            match states.get(event_id) {
                Some(DeliveryState::Pending) => {
                    let active = self
                        .active_delivery
                        .as_ref()
                        .ok_or(SoftchatError::InvalidDeliveryState)?;
                    if !active.remaining_event_ids.contains(event_id)
                        || active.socket_written_event_ids.contains(event_id)
                    {
                        return Err(SoftchatError::InvalidDeliveryState);
                    }
                    intent_ids.push(
                        active
                            .event_to_intent
                            .get(event_id)
                            .cloned()
                            .ok_or(SoftchatError::InvalidDeliveryState)?,
                    );
                }
                Some(DeliveryState::Retryable) => {
                    let active = self
                        .active_delivery
                        .as_ref()
                        .ok_or(SoftchatError::InvalidDeliveryState)?;
                    if !active.remaining_event_ids.contains(event_id)
                        || !active.socket_written_event_ids.contains(event_id)
                    {
                        return Err(SoftchatError::InvalidDeliveryState);
                    }
                }
                Some(DeliveryState::Accepted | DeliveryState::Rejected) => {}
                Some(DeliveryState::InFlight) | None => {
                    return Err(SoftchatError::InvalidDeliveryState);
                }
            }
        }
        Ok(intent_ids)
    }

    fn connect_action(&mut self, delay_ms: u64) -> AccountTransportAction {
        self.new_action(
            AccountTransportActionKind::Connect,
            self.endpoint.network_url.clone(),
            String::new(),
            delay_ms,
            Vec::new(),
        )
    }

    fn new_action(
        &mut self,
        kind: AccountTransportActionKind,
        network_url: String,
        text_frame: String,
        delay_ms: u64,
        delivery_event_ids: Vec<String>,
    ) -> AccountTransportAction {
        self.next_action_sequence = self.next_action_sequence.saturating_add(1);
        let action_id = stable_action_id(
            &self.run_id,
            self.generation,
            self.next_action_sequence,
            kind,
        );
        self.pending.insert(
            action_id.clone(),
            PendingAction {
                frame_class: crate::account_diagnostics::planned_frame_class(&text_frame),
                kind,
                delivery_event_ids,
            },
        );
        AccountTransportAction {
            run_id: self.run_id.clone(),
            action_id,
            generation: self.generation,
            kind,
            configured_url: if kind == AccountTransportActionKind::Connect {
                self.endpoint.configured_url.clone()
            } else {
                String::new()
            },
            network_url,
            noise_remote_static_key: if kind == AccountTransportActionKind::Connect {
                self.endpoint.noise_remote_static_key.clone()
            } else {
                Vec::new()
            },
            text_frame,
            delay_ms,
        }
    }

    fn next_lease_id(&mut self) -> String {
        self.next_action_sequence = self.next_action_sequence.saturating_add(1);
        format!(
            "transport-{}-{}",
            self.generation, self.next_action_sequence
        )
    }

    fn batch(
        &self,
        actions: Vec<AccountTransportAction>,
        database: &AccountDatabase,
        idle: bool,
    ) -> Result<AccountTransportBatch, SoftchatError> {
        let state = self.session.snapshot().state;
        if self.diagnostic_state.replace(state) != state {
            database.record_diagnostic(crate::account_diagnostics::connection(state));
        }
        let sync = self.sync.as_ref().map(|active| active.engine.snapshot());
        let sync_complete = sync
            .as_ref()
            .is_none_or(|snapshot| snapshot.phase == SyncPhase::Complete);
        Ok(AccountTransportBatch {
            actions,
            connection_state: state,
            idle: (idle || self.idle) && sync_complete,
            revision: database.info()?.revision,
            typing_indicators: self
                .typing_indicators
                .iter()
                .map(
                    |((conversation_id, author_public_key), expires_at)| AccountTypingIndicator {
                        conversation_id: conversation_id.clone(),
                        author_public_key: author_public_key.clone(),
                        expires_at: *expires_at,
                    },
                )
                .collect(),
            sync,
        })
    }
}

fn shard_sync_event_request(
    base_subscription_id: &str,
    filters: Vec<crate::RelayFilter>,
) -> Result<Vec<(String, Vec<crate::RelayFilter>)>, SoftchatError> {
    if filters.is_empty() {
        return Err(SoftchatError::InvalidSyncState);
    }
    let mut subscriptions = Vec::new();
    let mut current = Vec::new();
    for filter in filters {
        current.push(filter);
        let subscription_id =
            sync_fetch_subscription_id(base_subscription_id, subscriptions.len())?;
        let fits = current.len() <= crate::MAX_RELAY_FILTERS
            && crate::ClientRelayFrame::Req {
                subscription_id: subscription_id.clone(),
                filters: current.clone(),
            }
            .to_json()
            .is_ok();
        if fits {
            continue;
        }
        let last = current.pop().ok_or(SoftchatError::InvalidSyncState)?;
        if current.is_empty() {
            return Err(SoftchatError::InvalidRelayFrame);
        }
        subscriptions.push((subscription_id, std::mem::take(&mut current)));
        current.push(last);
        let next_id = sync_fetch_subscription_id(base_subscription_id, subscriptions.len())?;
        crate::ClientRelayFrame::Req {
            subscription_id: next_id,
            filters: current.clone(),
        }
        .to_json()?;
    }
    if !current.is_empty() {
        let subscription_id =
            sync_fetch_subscription_id(base_subscription_id, subscriptions.len())?;
        subscriptions.push((subscription_id, current));
    }
    if subscriptions.len() > crate::MAX_RELAY_SUBSCRIPTIONS {
        return Err(SoftchatError::RelaySessionQueueFull);
    }
    Ok(subscriptions)
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum InboundFailureDisposition {
    Ignore,
    Recover(RelayFailureKind),
}

/// Untrusted event validation failures are distinct from storage, erased
/// identity, and account-lifecycle failures, which must remain fail-closed.
const fn invalid_incoming_event(error: SoftchatError) -> bool {
    matches!(
        error,
        SoftchatError::InvalidPublicKey
            | SoftchatError::InvalidRelayFrame
            | SoftchatError::InvalidEventId
            | SoftchatError::InvalidEventSignature
            | SoftchatError::InvalidEventKind
            | SoftchatError::InvalidEventTimestamp
            | SoftchatError::InvalidEventTag
            | SoftchatError::InvalidEventJson
            | SoftchatError::EventTooLarge
            | SoftchatError::InvalidNip59Envelope
            | SoftchatError::InvalidNip17Message
            | SoftchatError::InvalidAttachmentMetadata
            | SoftchatError::InvalidSoftchatEvent
            | SoftchatError::InvalidSoftchatProjection
            | SoftchatError::InvalidUserMetadata
            | SoftchatError::InvalidContactList
            | SoftchatError::InvalidApplicationData
            | SoftchatError::InvalidAccountOperation
            | SoftchatError::InvalidAccountSettings
            | SoftchatError::DecryptionFailed
    )
}

const fn inbound_failure_disposition(error: SoftchatError) -> Option<InboundFailureDisposition> {
    match error {
        SoftchatError::InvalidRelayFrame
        | SoftchatError::InvalidRelayBatch
        | SoftchatError::InvalidRelaySession
        | SoftchatError::InvalidDeliveryState => Some(InboundFailureDisposition::Ignore),
        SoftchatError::InvalidRelayAuthentication => Some(InboundFailureDisposition::Recover(
            RelayFailureKind::Authentication,
        )),
        SoftchatError::RelaySessionQueueFull => Some(InboundFailureDisposition::Recover(
            RelayFailureKind::Temporary,
        )),
        SoftchatError::InvalidSyncState | SoftchatError::InvalidNegentropy => Some(
            InboundFailureDisposition::Recover(RelayFailureKind::ProtocolPermanent),
        ),
        _ => None,
    }
}

fn sync_fetch_subscription_id(
    base_subscription_id: &str,
    shard_index: usize,
) -> Result<String, SoftchatError> {
    let subscription_id = format!("{base_subscription_id}-{shard_index}");
    if subscription_id.chars().count() > crate::MAX_SUBSCRIPTION_ID_CHARS {
        return Err(SoftchatError::InvalidSyncState);
    }
    Ok(subscription_id)
}

pub(crate) fn result_hash(result: &AccountTransportResult) -> Result<String, SoftchatError> {
    validate_transport_id(&result.run_id)?;
    validate_transport_id(&result.action_id)?;
    if result.text_frame.len() > crate::MAX_RELAY_FRAME_BYTES {
        return Err(SoftchatError::InvalidRelayFrame);
    }
    let encoded = serde_json::to_vec(result).map_err(|_| SoftchatError::InternalFailure)?;
    Ok(hex::encode(Sha256::digest(encoded)))
}

fn validate_result_for_action(
    result: &AccountTransportResult,
    action: AccountTransportActionKind,
) -> Result<(), SoftchatError> {
    let valid = match action {
        AccountTransportActionKind::Connect => matches!(
            result.kind,
            AccountTransportResultKind::Connected
                | AccountTransportResultKind::Disconnected
                | AccountTransportResultKind::NetworkUnavailable
                | AccountTransportResultKind::AuthenticationFailed
                | AccountTransportResultKind::RateLimited
                | AccountTransportResultKind::ProtocolFailed
                | AccountTransportResultKind::Cancelled
        ),
        AccountTransportActionKind::SendText => matches!(
            result.kind,
            AccountTransportResultKind::FrameWritten
                | AccountTransportResultKind::Disconnected
                | AccountTransportResultKind::NetworkUnavailable
                | AccountTransportResultKind::RateLimited
                | AccountTransportResultKind::ProtocolFailed
                | AccountTransportResultKind::Cancelled
        ),
        AccountTransportActionKind::ReceiveText => matches!(
            result.kind,
            AccountTransportResultKind::FrameReceived
                | AccountTransportResultKind::TimedOut
                | AccountTransportResultKind::Disconnected
                | AccountTransportResultKind::NetworkUnavailable
                | AccountTransportResultKind::AuthenticationFailed
                | AccountTransportResultKind::RateLimited
                | AccountTransportResultKind::ProtocolFailed
                | AccountTransportResultKind::Cancelled
        ),
        AccountTransportActionKind::Close => matches!(
            result.kind,
            AccountTransportResultKind::Disconnected | AccountTransportResultKind::Cancelled
        ),
    };
    if !valid
        || (result.kind == AccountTransportResultKind::FrameReceived
            && result.text_frame.is_empty())
        || (result.kind != AccountTransportResultKind::FrameReceived
            && !result.text_frame.is_empty())
    {
        return Err(SoftchatError::UnsupportedSystemAction);
    }
    Ok(())
}

fn stable_action_id(
    run_id: &str,
    generation: u64,
    sequence: u64,
    kind: AccountTransportActionKind,
) -> String {
    let mut digest = Sha256::new();
    digest.update(run_id.as_bytes());
    digest.update(generation.to_be_bytes());
    digest.update(sequence.to_be_bytes());
    digest.update([kind as u8]);
    hex::encode(digest.finalize())
}

fn validate_transport_id(value: &str) -> Result<(), SoftchatError> {
    if value.is_empty()
        || value.len() > MAX_TRANSPORT_ID_BYTES
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
    {
        return Err(SoftchatError::InvalidRelaySession);
    }
    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;

    const NOW: i64 = 1_700_000_000;

    fn authority() -> Result<(crate::LocalIdentity, LocalIdentityHandle), SoftchatError> {
        let mut secret = vec![0; 32];
        secret[31] = 1;
        Ok((
            crate::LocalIdentity::from_secret_bytes(&secret)?,
            LocalIdentityHandle::new(secret)?,
        ))
    }

    fn result(
        action: &AccountTransportAction,
        kind: AccountTransportResultKind,
        text_frame: String,
    ) -> AccountTransportResult {
        AccountTransportResult {
            run_id: action.run_id.clone(),
            action_id: action.action_id.clone(),
            generation: action.generation,
            kind,
            text_frame,
        }
    }

    fn action(
        batch: &AccountTransportBatch,
        kind: AccountTransportActionKind,
    ) -> Result<AccountTransportAction, SoftchatError> {
        batch
            .actions
            .iter()
            .find(|action| action.kind == kind)
            .cloned()
            .ok_or(SoftchatError::UnsupportedSystemAction)
    }

    fn connected_run(
        database: &mut AccountDatabase,
        identity: &LocalIdentityHandle,
    ) -> Result<(AccountTransportRun, AccountTransportBatch), SoftchatError> {
        let (mut run, started) = AccountTransportRun::start(
            "transport-regression".to_owned(),
            vec!["{}".to_owned()],
            database,
        )?;
        let batch = run.handle_result(
            &result(
                &started.actions[0],
                AccountTransportResultKind::Connected,
                String::new(),
            ),
            database,
            identity,
            NOW,
        )?;
        Ok((run, batch))
    }

    fn signed_note(
        identity: &crate::LocalIdentity,
        timestamp: u64,
        text: String,
    ) -> Result<SignedNostrEvent, SoftchatError> {
        identity.sign_event(crate::NostrEventDraft::new(
            timestamp,
            crate::NostrEventKind::SHORT_TEXT_NOTE,
            Vec::new(),
            text,
        )?)
    }

    fn invalid_wrapper(identity: &crate::LocalIdentity) -> Result<SignedNostrEvent, SoftchatError> {
        identity.sign_event(crate::NostrEventDraft::new(
            NOW as u64,
            crate::NostrEventKind::GIFT_WRAP,
            vec![crate::NostrTag::new(vec![
                "p".to_owned(),
                identity.public_key().to_hex(),
            ])?],
            "invalid ciphertext",
        )?)
    }

    fn reconcile_remote(
        run: &mut AccountTransportRun,
        connected: &AccountTransportBatch,
        database: &mut AccountDatabase,
        identity: &LocalIdentityHandle,
        events: &[SignedNostrEvent],
    ) -> Result<AccountTransportBatch, Box<dyn std::error::Error>> {
        let open = connected
            .actions
            .iter()
            .find_map(|action| {
                let values =
                    serde_json::from_str::<Vec<serde_json::Value>>(&action.text_frame).ok()?;
                (values.first()?.as_str()? == "NEG-OPEN").then_some(values)
            })
            .ok_or("missing NEG-OPEN")?;
        let subscription = open[1].as_str().ok_or("missing subscription")?;
        let initial = hex::decode(
            open.last()
                .and_then(serde_json::Value::as_str)
                .ok_or("missing message")?,
        )?;
        let mut storage = softrelay_negentropy::NegentropyStorageVector::new();
        for event in events {
            storage.insert(
                event.created_at(),
                softrelay_negentropy::Id::from_byte_array(*event.id().as_inner().as_bytes()),
            )?;
        }
        storage.seal()?;
        let mut relay = softrelay_negentropy::Negentropy::owned(storage, SYNC_FRAME_SIZE_LIMIT)?;
        let response = relay.reconcile(&initial)?;
        Ok(run.handle_result(
            &result(
                &action(connected, AccountTransportActionKind::ReceiveText)?,
                AccountTransportResultKind::FrameReceived,
                serde_json::json!(["NEG-MSG", subscription, hex::encode(response)]).to_string(),
            ),
            database,
            identity,
            NOW + 1,
        )?)
    }

    #[test]
    fn transport_invalid_inner_event_does_not_poison_live_ingestion()
    -> Result<(), Box<dyn std::error::Error>> {
        let (identity, handle) = authority()?;
        let mut database = AccountDatabase::open(":memory:", "live-invalid".to_owned())?;
        let (mut run, connected) = connected_run(&mut database, &handle)?;
        let malformed_rumor = identity.create_rumor(crate::NostrEventDraft::new(
            NOW as u64,
            crate::NostrEventKind::REACTION,
            Vec::new(),
            "+",
        )?)?;
        let bad_events = [
            invalid_wrapper(&identity)?,
            identity.gift_wrap(
                &identity.public_key(),
                &malformed_rumor,
                crate::Nip59EnvelopeKind::Durable,
                NOW as u64,
            )?,
        ];
        let before = database.info()?.revision;
        let mut after = connected;
        for bad in bad_events {
            after = run.handle_result(
                &result(
                    &action(&after, AccountTransportActionKind::ReceiveText)?,
                    AccountTransportResultKind::FrameReceived,
                    RelayResponseFrame::Event {
                        subscription_id: "softchat-live-0".to_owned(),
                        event: bad.clone(),
                    }
                    .to_json()?,
                ),
                &mut database,
                &handle,
                NOW + 1,
            )?;
            assert_eq!(after.connection_state, RelayConnectionState::Ready);
            assert!(run.session.snapshot().pending_ingestion_ids.is_empty());
            assert_eq!(database.info()?.revision, before);
            assert!(database.event_json(&bad.id().to_hex()).is_err());
        }
        let valid = signed_note(&identity, NOW as u64, "valid next event".to_owned())?;
        run.handle_result(
            &result(
                &action(&after, AccountTransportActionKind::ReceiveText)?,
                AccountTransportResultKind::FrameReceived,
                RelayResponseFrame::Event {
                    subscription_id: "softchat-live-0".to_owned(),
                    event: valid.clone(),
                }
                .to_json()?,
            ),
            &mut database,
            &handle,
            NOW + 2,
        )?;
        assert_eq!(database.event_json(&valid.id().to_hex())?, valid.to_json()?);
        assert!(run.session.snapshot().pending_ingestion_ids.is_empty());
        // Authority/lifecycle errors still stop the transition; only rejected
        // input is safe to discard without committing an event.
        handle.erase();
        assert_eq!(
            run.process_inbound_frame(
                &RelayResponseFrame::Event {
                    subscription_id: "softchat-live-0".to_owned(),
                    event: valid,
                }
                .to_json()?,
                &mut database,
                &handle,
                NOW + 3
            ),
            Err(SoftchatError::IdentityErased)
        );
        Ok(())
    }

    #[test]
    fn transport_invalid_sync_event_recovers_without_advancing_checkpoint()
    -> Result<(), Box<dyn std::error::Error>> {
        let (identity, handle) = authority()?;
        let mut database = AccountDatabase::open(":memory:", "sync-invalid".to_owned())?;
        let (mut run, connected) = connected_run(&mut database, &handle)?;
        let bad = invalid_wrapper(&identity)?;
        let fetching = reconcile_remote(
            &mut run,
            &connected,
            &mut database,
            &handle,
            std::slice::from_ref(&bad),
        )?;
        let subscription = run
            .sync
            .as_ref()
            .and_then(|sync| sync.fetch_subscription_ids.first())
            .cloned()
            .ok_or("missing fetch")?;
        let recovered = run.handle_result(
            &result(
                &action(&fetching, AccountTransportActionKind::ReceiveText)?,
                AccountTransportResultKind::FrameReceived,
                RelayResponseFrame::Event {
                    subscription_id: subscription,
                    event: bad.clone(),
                }
                .to_json()?,
            ),
            &mut database,
            &handle,
            NOW + 2,
        )?;
        assert_eq!(recovered.connection_state, RelayConnectionState::Connecting);
        assert!(
            action(&recovered, AccountTransportActionKind::Connect)?.generation
                > connected.actions[0].generation
        );
        assert_eq!(database.sync_checkpoint(SYNC_ID)?, None);
        assert!(database.event_json(&bad.id().to_hex()).is_err());
        assert!(run.session.snapshot().pending_ingestion_ids.is_empty());
        Ok(())
    }

    #[test]
    fn transport_retryable_ack_preserves_terminal_outcomes_and_reclaims_once()
    -> Result<(), Box<dyn std::error::Error>> {
        // Cover the final outstanding ACK after a socket write and an ACK
        // racing the write while another delivery still needs release.
        for (count, written) in [(3, true), (4, false)] {
            let (identity, handle) = authority()?;
            let mut database = AccountDatabase::open(":memory:", "retryable-ack".to_owned())?;
            let events = (0..count)
                .map(|offset| signed_note(&identity, NOW as u64 + offset, "outgoing".to_owned()))
                .collect::<Result<Vec<_>, _>>()?;
            database.enqueue_signed(
                &identity,
                "retry-operation".to_owned(),
                events.iter().map(crate::event::SignedEvent::from).collect(),
                vec![database.active_relay_url()?],
                NOW,
            )?;
            let (mut run, connected) = connected_run(&mut database, &handle)?;
            let send = connected
                .actions
                .iter()
                .find(|action| {
                    run.pending
                        .get(&action.action_id)
                        .is_some_and(|pending| !pending.delivery_event_ids.is_empty())
                })
                .cloned()
                .ok_or("missing delivery send")?;
            if written {
                run.handle_result(
                    &result(
                        &send,
                        AccountTransportResultKind::FrameWritten,
                        String::new(),
                    ),
                    &mut database,
                    &handle,
                    NOW + 1,
                )?;
            }
            let mut receive = action(&connected, AccountTransportActionKind::ReceiveText)?;
            for (index, accepted, message) in [
                (0, true, ""),
                (1, false, "blocked: policy"),
                (2, false, "rate-limited: slow down"),
            ] {
                let batch = run.handle_result(
                    &result(
                        &receive,
                        AccountTransportResultKind::FrameReceived,
                        RelayResponseFrame::Ok(crate::BatchAcknowledgement {
                            event_id: events[index].id().to_hex(),
                            accepted,
                            message: message.to_owned(),
                        })
                        .to_json()?,
                    ),
                    &mut database,
                    &handle,
                    NOW + 2 + index as i64,
                )?;
                if index < 2 {
                    receive = action(&batch, AccountTransportActionKind::ReceiveText)?;
                } else {
                    let reconnect = action(&batch, AccountTransportActionKind::Connect)?;
                    assert!(
                        reconnect.delay_ms > 0
                            && reconnect.delay_ms <= crate::MAX_RELAY_RETRY_DELAY_MS
                    );
                    let snapshot = database.operation("retry-operation")?;
                    assert_eq!(
                        (
                            snapshot.accepted_intents,
                            snapshot.rejected_intents,
                            snapshot.queued_intents
                        ),
                        (1, 1, u32::try_from(count - 2)?)
                    );
                    let ready = run.handle_result(
                        &result(
                            &reconnect,
                            AccountTransportResultKind::Connected,
                            String::new(),
                        ),
                        &mut database,
                        &handle,
                        NOW + 5,
                    )?;
                    let resent = ready
                        .actions
                        .iter()
                        .filter_map(|action| run.pending.get(&action.action_id))
                        .flat_map(|pending| pending.delivery_event_ids.clone())
                        .collect::<Vec<_>>();
                    assert_eq!(resent.len(), usize::try_from(count - 2)?);
                    assert_eq!(
                        resent.into_iter().collect::<BTreeSet<_>>(),
                        events[2..]
                            .iter()
                            .map(|event| event.id().to_hex())
                            .collect()
                    );
                    run.handle_result(
                        &result(
                            &send,
                            AccountTransportResultKind::FrameWritten,
                            String::new(),
                        ),
                        &mut database,
                        &handle,
                        NOW + 6,
                    )?;
                }
            }
        }
        Ok(())
    }

    #[test]
    fn transport_delivery_expiry_is_not_postponed_by_inbound_traffic()
    -> Result<(), Box<dyn std::error::Error>> {
        enum Progress {
            Notice,
            Wake,
            Accepted,
        }
        for progress in [Progress::Notice, Progress::Wake, Progress::Accepted] {
            let (identity, handle) = authority()?;
            let mut database = AccountDatabase::open(":memory:", "busy-expiry".to_owned())?;
            let event = signed_note(&identity, NOW as u64, "awaiting acknowledgement".to_owned())?;
            database.enqueue_signed(
                &identity,
                "busy-operation".to_owned(),
                vec![crate::event::SignedEvent::from(&event)],
                vec![database.active_relay_url()?],
                NOW,
            )?;
            let (mut run, connected) = connected_run(&mut database, &handle)?;
            let send = connected
                .actions
                .iter()
                .find(|action| {
                    run.pending
                        .get(&action.action_id)
                        .is_some_and(|pending| !pending.delivery_event_ids.is_empty())
                })
                .ok_or("missing send")?;
            run.handle_result(
                &result(
                    send,
                    AccountTransportResultKind::FrameWritten,
                    String::new(),
                ),
                &mut database,
                &handle,
                NOW + 1,
            )?;
            let busy = run.handle_result(
                &result(
                    &action(&connected, AccountTransportActionKind::ReceiveText)?,
                    AccountTransportResultKind::FrameReceived,
                    RelayResponseFrame::Notice("still connected".to_owned()).to_json()?,
                ),
                &mut database,
                &handle,
                NOW + 20,
            )?;
            assert_eq!(busy.connection_state, RelayConnectionState::Ready);
            let expired = match progress {
                Progress::Wake => run.wake(&mut database, NOW + 31)?,
                Progress::Notice | Progress::Accepted => {
                    let frame = if matches!(progress, Progress::Accepted) {
                        RelayResponseFrame::Ok(crate::BatchAcknowledgement {
                            event_id: event.id().to_hex(),
                            accepted: true,
                            message: String::new(),
                        })
                    } else {
                        RelayResponseFrame::Notice("still busy".to_owned())
                    };
                    run.handle_result(
                        &result(
                            &action(&busy, AccountTransportActionKind::ReceiveText)?,
                            AccountTransportResultKind::FrameReceived,
                            frame.to_json()?,
                        ),
                        &mut database,
                        &handle,
                        NOW + 31,
                    )?
                }
            };
            let operation = database.operation("busy-operation")?;
            if matches!(progress, Progress::Accepted) {
                assert_eq!(operation.state, crate::OperationState::Sent);
                assert_eq!(expired.connection_state, RelayConnectionState::Ready);
            } else {
                let reconnect = action(&expired, AccountTransportActionKind::Connect)?;
                assert!(reconnect.generation > connected.actions[0].generation);
                assert!(reconnect.delay_ms > 0);
                assert_eq!((operation.queued_intents, operation.active_intents), (1, 0));
            }
        }
        Ok(())
    }

    #[test]
    fn transport_sync_accepts_a_maximum_cohort_independent_of_chunk_boundaries()
    -> Result<(), Box<dyn std::error::Error>> {
        let (_, handle) = authority()?;
        let mut database = AccountDatabase::open(":memory:", "sync-max".to_owned())?;
        let (mut run, _) = connected_run(&mut database, &handle)?;
        let ids = (0..crate::SYNC_MAX_PAGE_LIMIT)
            .map(|index| format!("{index:064x}"))
            .collect::<Vec<_>>();
        let resend = |ids: Vec<String>| SyncAction {
            kind: SyncActionKind::ResendEvents,
            event_ids: ids,
            subscription_id: String::new(),
            since_timestamp: -1,
            until_timestamp: -1,
            limit: 0,
            message: Vec::new(),
            frame_json: String::new(),
            filters_json: Vec::new(),
            checkpoint_timestamp: -1,
        };
        run.apply_sync_actions(
            ids.chunks(crate::SYNC_EVENT_REQUEST_CHUNK)
                .map(|chunk| resend(chunk.to_vec()))
                .collect(),
            &mut database,
            &handle,
        )?;
        assert_eq!(
            run.sync_resend_backlog.iter().cloned().collect::<Vec<_>>(),
            ids
        );
        assert_eq!(
            run.apply_sync_actions(vec![resend(vec!["ff".repeat(32)])], &mut database, &handle),
            Err(SoftchatError::RelaySessionQueueFull)
        );
        assert_eq!(run.sync_resend_backlog.len(), ids.len());
        Ok(())
    }

    #[test]
    fn transport_sync_resends_fit_frame_bytes_and_keep_every_event()
    -> Result<(), Box<dyn std::error::Error>> {
        let (identity, handle) = authority()?;
        let mut database = AccountDatabase::open(":memory:", "sync-large".to_owned())?;
        let events = (0..3)
            .map(|offset| signed_note(&identity, NOW as u64 - offset, "x".repeat(280 * 1024)))
            .collect::<Result<Vec<_>, _>>()?;
        for event in &events {
            database.ingest(&identity, vec![crate::event::SignedEvent::from(event)], NOW)?;
        }
        let (mut run, connected) = connected_run(&mut database, &handle)?;
        // Leave one send slot after NEG-CLOSE so byte splitting must retain
        // the remaining events for a later action batch.
        for index in 0..MAX_TRANSPORT_ACTIONS - 3 {
            run.raw_outbound.push_back(
                crate::ClientRelayFrame::Close {
                    subscription_id: format!("finished-{index}"),
                }
                .to_json()?,
            );
        }
        let resent = reconcile_remote(&mut run, &connected, &mut database, &handle, &[])?;
        assert!(!run.sync_resend_backlog.is_empty());
        let next = run.wake(&mut database, NOW + 2)?;
        assert!(run.sync_resend_backlog.is_empty());
        let mut actual = Vec::new();
        for action in resent
            .actions
            .iter()
            .chain(&next.actions)
            .filter(|action| action.kind == AccountTransportActionKind::SendText)
        {
            assert!(
                action.text_frame.len() <= crate::MAX_RELAY_FRAME_BYTES,
                "oversized resend: {} bytes",
                action.text_frame.len()
            );
            let values: Vec<serde_json::Value> = serde_json::from_str(&action.text_frame)?;
            if matches!(values[0].as_str(), Some("EVENT" | "EVENTS")) {
                actual.extend(
                    values[1..]
                        .iter()
                        .filter_map(|event| event["id"].as_str().map(str::to_owned)),
                );
            }
        }
        assert_eq!(actual.len(), events.len());
        assert_eq!(
            actual.into_iter().collect::<BTreeSet<_>>(),
            events.iter().map(|event| event.id().to_hex()).collect()
        );
        Ok(())
    }

    #[test]
    fn maximum_sync_request_is_sharded_into_valid_relay_frames() -> Result<(), SoftchatError> {
        let filters = (0..200_u64)
            .map(|chunk| crate::RelayFilter {
                ids: (0..crate::SYNC_EVENT_REQUEST_CHUNK)
                    .map(|offset| {
                        let value = chunk
                            .saturating_mul(crate::SYNC_EVENT_REQUEST_CHUNK as u64)
                            .saturating_add(offset as u64);
                        format!("{value:064x}")
                    })
                    .collect(),
                ..crate::RelayFilter::default()
            })
            .collect::<Vec<_>>();

        let subscriptions = shard_sync_event_request("softchat-events-1", filters)?;
        assert!(subscriptions.len() > 1);
        assert_eq!(
            subscriptions
                .iter()
                .map(|(_, filters)| filters.len())
                .sum::<usize>(),
            200,
        );
        assert_eq!(
            subscriptions
                .iter()
                .map(|(subscription_id, _)| subscription_id)
                .collect::<BTreeSet<_>>()
                .len(),
            subscriptions.len(),
        );
        for adjacent in subscriptions.windows(2) {
            let mut over_boundary = adjacent[0].1.clone();
            over_boundary.push(
                adjacent[1]
                    .1
                    .first()
                    .cloned()
                    .ok_or(SoftchatError::InvalidSyncState)?,
            );
            let exceeds_count = over_boundary.len() > crate::MAX_RELAY_FILTERS;
            let exceeds_bytes = crate::ClientRelayFrame::Req {
                subscription_id: adjacent[0].0.clone(),
                filters: over_boundary,
            }
            .to_json()
            .is_err();
            assert!(exceeds_count || exceeds_bytes);
        }
        for (subscription_id, filters) in subscriptions {
            assert!(filters.len() <= crate::MAX_RELAY_FILTERS);
            let frame = crate::ClientRelayFrame::Req {
                subscription_id,
                filters,
            }
            .to_json()?;
            assert!(frame.len() <= crate::MAX_RELAY_FRAME_BYTES);
        }
        Ok(())
    }
}