softchat/

runtime.rs

//! Account-scoped ingestion, outgoing-operation, and delivery plans.
//!
//! These values deliberately contain no database, socket, HTTP, or lifecycle
//! callbacks. A platform prepares or receives bounded values in Rust, commits
//! them through its native store, and returns explicit receipts.

use std::collections::BTreeSet;

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

use crate::event::{RumorEvent, SignedEvent};
use crate::{
    AppDataContext, LocalIdentity, Nip59EnvelopeKind, NostrEventKind, NostrPublicKey, NostrRumor,
    SignedNostrEvent, SoftchatError, classify_application_data, classify_chat_message,
    classify_deletion, classify_edit, classify_follow_list, classify_generic_repost,
    classify_reaction, classify_subject, classify_typing, classify_user_metadata,
    parse_relay_endpoint,
};

/// Maximum events prepared in one incoming transaction.
pub const MAX_ACCOUNT_INGESTION_EVENTS: usize = 150;
/// Maximum canonical event bytes prepared in one incoming transaction.
pub const MAX_ACCOUNT_INGESTION_BYTES: usize = 512 * 1_024;
/// Maximum relay destinations attached to one outgoing operation.
pub const MAX_OPERATION_RELAYS: usize = 64;
/// Maximum event copies retained by one outgoing operation.
pub const MAX_OPERATION_EVENT_COPIES: usize = 512;
/// Maximum event/relay intents retained by one outgoing operation.
pub const MAX_OPERATION_DELIVERY_INTENTS: usize = 4_096;
/// Maximum delivery rows returned in one platform lease.
pub const MAX_DELIVERY_CLAIM_INTENTS: usize = 100;
/// Maximum canonical payload bytes returned in one platform lease.
pub const MAX_DELIVERY_CLAIM_BYTES: usize = 512 * 1_024;

const MAX_ACCOUNT_ID_BYTES: usize = 128;
const MAX_OPERATION_ID_BYTES: usize = 128;
const MAX_LEASE_ID_BYTES: usize = 128;
const MAX_DELIVERY_CATEGORY_BYTES: usize = 64;

/// Semantic projection family selected after authentication.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum ProjectionKind {
    /// Authenticated kind-14 message.
    ChatMessage,
    /// Authenticated kind-14 subject update.
    Subject,
    /// Authenticated kind-7 reaction.
    Reaction,
    /// Authenticated kind-5 deletion request.
    Deletion,
    /// Authenticated Softchat edit.
    Edit,
    /// Authenticated ephemeral typing state.
    Typing,
    /// Authenticated private kind-0 metadata.
    UserMetadata,
    /// Authenticated private author-copy kind-3 contacts.
    FollowList,
    /// Verified direct NIP-78 application data.
    ApplicationData,
    /// Verified and self-decrypted Softchat app-data sync.
    AppDataSync,
    /// Verified receive-only generic repost.
    GenericRepost,
    /// Authenticated event retained as truth without a typed projection.
    Unknown,
}

/// One normalized immutable projection mutation.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct ProjectionMutation {
    /// Selected projection family.
    pub kind: ProjectionKind,
    /// Authenticated logical event or rumor ID.
    pub logical_event_id: String,
    /// Authenticated author public key.
    pub author_public_key: String,
    /// Portable event timestamp.
    pub created_at: i64,
    /// Android-compatible SHA-256 conversation key, or empty for account data.
    pub conversation_id: String,
    /// Sorted complete conversation participants, including the author.
    pub participants: Vec<String>,
    /// Referenced event IDs for replies, edits, reactions, or deletions.
    pub target_event_ids: Vec<String>,
    /// Validated primary text or JSON payload for the selected projection.
    pub content: String,
    /// Whether this projection must never become durable message history.
    pub ephemeral: bool,
}

/// One fully authenticated event and its optional inner rumor.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct PreparedIncomingEvent {
    /// Exact verified outer event.
    pub outer_event: SignedEvent,
    /// Authenticated inner rumor for NIP-59 input.
    pub rumor: Option<RumorEvent>,
    /// Deterministic projection mutation.
    pub projection: ProjectionMutation,
}

/// One bounded atomic ingestion request for a platform-native database.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct PreparedIncomingBatch {
    /// Opaque non-secret account partition.
    pub account_id: String,
    /// Total canonical outer-event bytes used for batching.
    pub canonical_bytes: u64,
    /// Verified events and deterministic projections.
    pub events: Vec<PreparedIncomingEvent>,
}

/// Platform result for one complete incoming transaction.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct IngestionReceipt {
    /// Opaque account partition committed by the store.
    pub account_id: String,
    /// IDs inserted for the first time.
    pub inserted_event_ids: Vec<String>,
    /// IDs already present with identical canonical bytes.
    pub duplicate_event_ids: Vec<String>,
    /// IDs deliberately quarantined by a conflict or storage policy.
    pub quarantined_event_ids: Vec<String>,
}

/// One independently randomized NIP-59 recipient copy.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct PreparedEventCopy {
    /// Stable outer event ID, also used as the event-copy ID.
    pub event_copy_id: String,
    /// Intended recipient of this independently wrapped copy.
    pub recipient_public_key: String,
    /// Exact signed relay-publishable event.
    pub event: SignedEvent,
}

/// One durable event-copy/relay delivery intent.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct PreparedRelayIntent {
    /// Deterministic ID derived from operation, event copy, and relay.
    pub intent_id: String,
    /// Owning operation.
    pub operation_id: String,
    /// Event copy sent by this intent.
    pub event_copy_id: String,
    /// Canonical relay URL.
    pub relay_url: String,
    /// Canonical event JSON byte count used for bounded claims.
    pub payload_bytes: u64,
}

/// One outgoing operation that must be persisted before any socket write.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct PreparedOutgoingOperation {
    /// Opaque non-secret account partition.
    pub account_id: String,
    /// Caller-supplied idempotency/operation ID.
    pub operation_id: String,
    /// Independently wrapped event copies.
    pub event_copies: Vec<PreparedEventCopy>,
    /// Required per-copy, per-relay delivery intents.
    pub relay_intents: Vec<PreparedRelayIntent>,
}

/// Durable delivery-intent state.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum DeliveryIntentState {
    /// Persisted and available to claim.
    Pending,
    /// Exclusively leased by one drain worker.
    Claimed,
    /// Bytes were written to a socket; relay acceptance is still unknown.
    SocketWritten,
    /// Relay returned a terminal acceptance.
    Accepted,
    /// Relay returned a permanent rejection.
    Rejected,
    /// A retryable result returned the intent to the queue.
    Retryable,
    /// User or account lifecycle cancelled the intent.
    Cancelled,
}

/// Store snapshot used by the pure delivery reducer.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct DeliveryIntentSnapshot {
    /// Stable intent ID.
    pub intent_id: String,
    /// Owning operation ID.
    pub operation_id: String,
    /// Event-copy ID.
    pub event_copy_id: String,
    /// Canonical relay URL.
    pub relay_url: String,
    /// Canonical payload byte count.
    pub payload_bytes: u64,
    /// Current durable state.
    pub state: DeliveryIntentState,
    /// Existing lease ID, or empty.
    pub lease_id: String,
    /// Number of completed claim attempts.
    pub attempt_count: u32,
}

/// State mutation returned for an atomic platform transaction.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct DeliveryStateMutation {
    /// Stable intent ID.
    pub intent_id: String,
    /// Required new state.
    pub state: DeliveryIntentState,
    /// New lease ID, or empty when the lease must be cleared.
    pub lease_id: String,
    /// Stable redacted result category.
    pub category: String,
    /// New attempt count.
    pub attempt_count: u32,
}

/// One bounded exclusive delivery lease.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct DeliveryClaim {
    /// Host-generated lease ID.
    pub lease_id: String,
    /// Claimed rows in stable input order.
    pub intents: Vec<DeliveryIntentSnapshot>,
    /// Sum of canonical payload bytes.
    pub payload_bytes: u64,
    /// Atomic state mutations the store must commit before sending.
    pub mutations: Vec<DeliveryStateMutation>,
}

/// Relay result class accepted by the delivery reducer.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum RelayDeliveryResultKind {
    /// Terminal relay acceptance.
    Accepted,
    /// Terminal permanent rejection.
    Rejected,
    /// Retryable network, rate, or temporary relay failure.
    Retryable,
}

/// Stable redacted relay result.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct RelayDeliveryResult {
    /// Intent receiving the result.
    pub intent_id: String,
    /// Stable result class.
    pub kind: RelayDeliveryResultKind,
    /// Stable bounded category; never arbitrary relay prose.
    pub category: String,
}

/// User-visible aggregate operation state.
#[derive(Clone, Copy, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum OperationState {
    /// Local preparation has not produced durable relay intent yet.
    Preparing,
    /// Durable media prerequisites are still running.
    Uploading,
    /// All remaining work is durably queued.
    Queued,
    /// At least one delivery is leased or socket-written.
    Sending,
    /// Some required intents accepted while others remain or failed.
    PartiallySent,
    /// Every required intent was accepted.
    Sent,
    /// No intent was accepted and a permanent rejection exists.
    Failed,
    /// Every remaining intent was explicitly cancelled.
    Cancelled,
}

/// Aggregate operation state derived entirely from durable intent rows.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct OperationSnapshot {
    /// Owning operation ID.
    pub operation_id: String,
    /// Derived UI state.
    pub state: OperationState,
    /// Required intent count.
    pub total_intents: u32,
    /// Terminal accepted count.
    pub accepted_intents: u32,
    /// Terminal rejected count.
    pub rejected_intents: u32,
    /// Retryable/pending count.
    pub queued_intents: u32,
    /// Claimed/socket-written count.
    pub active_intents: u32,
}

/// Stateless semantic account planner.
#[derive(Clone, Copy, Debug, Default)]
pub struct AccountEngine;

impl AccountEngine {
    /// Validate one opaque non-secret account partition identifier.
    ///
    /// # Errors
    ///
    /// Returns [`SoftchatError::InvalidAccountOperation`] for an empty,
    /// oversized, or non-portable identifier.
    pub fn validate_account_id(account_id: &str) -> Result<(), SoftchatError> {
        validate_identifier(account_id, MAX_ACCOUNT_ID_BYTES)
    }

    /// Authenticate a bounded incoming batch and produce immutable store values.
    ///
    /// # Errors
    ///
    /// Returns a stable account, event, envelope, or projection error before
    /// the platform begins a transaction.
    pub fn prepare_incoming(
        identity: &LocalIdentity,
        account_id: String,
        events: Vec<SignedEvent>,
    ) -> Result<PreparedIncomingBatch, SoftchatError> {
        Self::validate_account_id(&account_id)?;
        if events.is_empty() || events.len() > MAX_ACCOUNT_INGESTION_EVENTS {
            return Err(SoftchatError::InvalidAccountOperation);
        }

        let mut canonical_bytes = 0_u64;
        let mut unique_ids = BTreeSet::new();
        let mut prepared = Vec::with_capacity(events.len());
        for event in events {
            let outer = SignedNostrEvent::try_from(event)?;
            let outer_json = outer.to_json()?;
            crate::session::validate_event_frame_json(&outer_json)?;
            canonical_bytes = canonical_bytes
                .checked_add(
                    u64::try_from(outer_json.len())
                        .map_err(|_| SoftchatError::InvalidAccountOperation)?,
                )
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            if canonical_bytes > MAX_ACCOUNT_INGESTION_BYTES as u64
                || !unique_ids.insert(outer.id().to_hex())
            {
                return Err(SoftchatError::InvalidAccountOperation);
            }

            let (rumor, projection) = if matches!(
                outer.kind(),
                NostrEventKind::GIFT_WRAP | NostrEventKind::EPHEMERAL_GIFT_WRAP
            ) {
                let unwrapped = identity.unwrap_gift_wrap(&outer)?;
                let rumor = unwrapped.into_rumor();
                let projection = projection_for_rumor(
                    &rumor,
                    &identity.public_key(),
                    outer.kind() == NostrEventKind::EPHEMERAL_GIFT_WRAP,
                )?;
                (Some(RumorEvent::from(&rumor)), projection)
            } else {
                (None, projection_for_signed(identity, &outer)?)
            };
            prepared.push(PreparedIncomingEvent {
                outer_event: SignedEvent::from(&outer),
                rumor,
                projection,
            });
        }

        Ok(PreparedIncomingBatch {
            account_id,
            canonical_bytes,
            events: prepared,
        })
    }

    /// Verify that a platform transaction handled every prepared event once.
    ///
    /// # Errors
    ///
    /// Returns [`SoftchatError::InvalidPersistenceResult`] for the wrong
    /// account, unknown IDs, duplicated classifications, or incomplete output.
    pub fn validate_ingestion_receipt(
        batch: &PreparedIncomingBatch,
        receipt: &IngestionReceipt,
    ) -> Result<(), SoftchatError> {
        if batch.account_id != receipt.account_id {
            return Err(SoftchatError::InvalidPersistenceResult);
        }
        let expected = batch
            .events
            .iter()
            .map(|event| event.outer_event.id.clone())
            .collect::<BTreeSet<_>>();
        let mut actual = BTreeSet::new();
        for event_id in receipt
            .inserted_event_ids
            .iter()
            .chain(&receipt.duplicate_event_ids)
            .chain(&receipt.quarantined_event_ids)
        {
            if !actual.insert(event_id.clone()) {
                return Err(SoftchatError::InvalidPersistenceResult);
            }
        }
        if expected == actual {
            Ok(())
        } else {
            Err(SoftchatError::InvalidPersistenceResult)
        }
    }

    /// Create independently randomized NIP-59 copies and required relay intents.
    ///
    /// The returned operation must be committed before any frame is sent.
    ///
    /// # Errors
    ///
    /// Returns a stable operation, relay URL, recipient, or envelope error.
    #[allow(clippy::too_many_arguments)]
    pub fn prepare_rumor_operation(
        identity: &LocalIdentity,
        account_id: String,
        operation_id: String,
        rumor: RumorEvent,
        recipients: Vec<String>,
        notification_recipients: Vec<String>,
        relay_urls: Vec<String>,
        kind: Nip59EnvelopeKind,
        now: i64,
    ) -> Result<PreparedOutgoingOperation, SoftchatError> {
        Self::validate_account_id(&account_id)?;
        validate_identifier(&operation_id, MAX_OPERATION_ID_BYTES)?;
        let now = u64::try_from(now).map_err(|_| SoftchatError::InvalidAccountOperation)?;
        let rumor =
            NostrRumor::try_from(rumor).map_err(|_| SoftchatError::InvalidAccountOperation)?;
        if rumor.public_key() != identity.public_key() {
            return Err(SoftchatError::InvalidAccountOperation);
        }

        let mut recipients = recipients
            .iter()
            .map(|value| {
                NostrPublicKey::from_hex(value).map_err(|_| SoftchatError::InvalidAccountOperation)
            })
            .collect::<Result<Vec<_>, _>>()?;
        let recipient_ids = recipients
            .iter()
            .map(NostrPublicKey::to_hex)
            .collect::<BTreeSet<_>>();
        let notification_recipients = notification_recipients
            .into_iter()
            .map(|value| {
                NostrPublicKey::from_hex(&value)
                    .map(|key| key.to_hex())
                    .map_err(|_| SoftchatError::InvalidAccountOperation)
            })
            .collect::<Result<BTreeSet<_>, _>>()?;
        if !notification_recipients.is_subset(&recipient_ids) {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        crate::envelope::validate_nip59_wire_size(
            &rumor,
            kind,
            now,
            !notification_recipients.is_empty(),
        )?;
        recipients.push(identity.public_key());
        recipients.sort_by_key(NostrPublicKey::to_hex);
        recipients.dedup();
        if recipients.is_empty() || recipients.len() > MAX_OPERATION_EVENT_COPIES {
            return Err(SoftchatError::InvalidAccountOperation);
        }

        let relays = canonical_relay_urls(relay_urls)?;
        let mut event_copies = Vec::with_capacity(recipients.len());
        for recipient in recipients {
            let notify_recipient = notification_recipients.contains(&recipient.to_hex());
            let event = identity.gift_wrap_for_operation_prevalidated(
                &recipient,
                &rumor,
                kind,
                now,
                notify_recipient,
            )?;
            event_copies.push(PreparedEventCopy {
                event_copy_id: event.id().to_hex(),
                recipient_public_key: recipient.to_hex(),
                event: SignedEvent::from(&event),
            });
        }
        prepare_operation(account_id, operation_id, event_copies, relays)
    }

    /// Persist one or more already signed direct events as an outgoing operation.
    ///
    /// # Errors
    ///
    /// Returns a stable operation or relay URL error.
    pub fn prepare_signed_operation(
        account_id: String,
        operation_id: String,
        events: Vec<SignedEvent>,
        relay_urls: Vec<String>,
    ) -> Result<PreparedOutgoingOperation, SoftchatError> {
        Self::validate_account_id(&account_id)?;
        validate_identifier(&operation_id, MAX_OPERATION_ID_BYTES)?;
        if events.is_empty() || events.len() > MAX_OPERATION_EVENT_COPIES {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        let mut unique = BTreeSet::new();
        let mut event_copies = Vec::with_capacity(events.len());
        for event in events {
            let event = SignedNostrEvent::try_from(event)?;
            let event_copy_id = event.id().to_hex();
            if !unique.insert(event_copy_id.clone()) {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            event_copies.push(PreparedEventCopy {
                event_copy_id,
                recipient_public_key: String::new(),
                event: SignedEvent::from(&event),
            });
        }
        prepare_operation(
            account_id,
            operation_id,
            event_copies,
            canonical_relay_urls(relay_urls)?,
        )
    }
}

/// Stateless durable-delivery reducer.
#[derive(Clone, Copy, Debug, Default)]
pub struct DeliveryReducer;

impl DeliveryReducer {
    /// Claim a bounded prefix of pending or retryable intents.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery error for malformed rows, a reused lease, or
    /// a first payload that cannot fit the claim profile.
    pub fn claim(
        intents: Vec<DeliveryIntentSnapshot>,
        lease_id: String,
    ) -> Result<DeliveryClaim, SoftchatError> {
        validate_identifier(&lease_id, MAX_LEASE_ID_BYTES)
            .map_err(|_| SoftchatError::InvalidDeliveryState)?;
        let mut claimed = Vec::new();
        let mut mutations = Vec::new();
        let mut payload_bytes = 0_u64;
        let mut unique = BTreeSet::new();
        for mut intent in intents {
            validate_intent_snapshot(&intent)?;
            if !unique.insert(intent.intent_id.clone()) {
                return Err(SoftchatError::InvalidDeliveryState);
            }
            if !matches!(
                intent.state,
                DeliveryIntentState::Pending | DeliveryIntentState::Retryable
            ) {
                continue;
            }
            if claimed.len() >= MAX_DELIVERY_CLAIM_INTENTS {
                break;
            }
            let next_bytes = payload_bytes
                .checked_add(intent.payload_bytes)
                .ok_or(SoftchatError::InvalidDeliveryState)?;
            if next_bytes > MAX_DELIVERY_CLAIM_BYTES as u64 {
                if claimed.is_empty() {
                    return Err(SoftchatError::InvalidDeliveryState);
                }
                break;
            }
            payload_bytes = next_bytes;
            let attempt_count = intent.attempt_count.saturating_add(1);
            mutations.push(DeliveryStateMutation {
                intent_id: intent.intent_id.clone(),
                state: DeliveryIntentState::Claimed,
                lease_id: lease_id.clone(),
                category: "claimed".to_owned(),
                attempt_count,
            });
            intent.state = DeliveryIntentState::Claimed;
            intent.lease_id.clone_from(&lease_id);
            intent.attempt_count = attempt_count;
            claimed.push(intent);
        }
        if claimed.is_empty() {
            return Err(SoftchatError::InvalidDeliveryState);
        }
        Ok(DeliveryClaim {
            lease_id,
            intents: claimed,
            payload_bytes,
            mutations,
        })
    }

    /// Mark an exact subset of one lease as socket-written.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery error for stale leases, duplicate IDs, or
    /// rows that were not in the claimed state.
    pub fn mark_socket_written(
        claim: &DeliveryClaim,
        intent_ids: Vec<String>,
    ) -> Result<Vec<DeliveryStateMutation>, SoftchatError> {
        transition_claim_subset(
            claim,
            intent_ids,
            DeliveryIntentState::Claimed,
            DeliveryIntentState::SocketWritten,
            "socket_written",
            true,
        )
    }

    /// Apply stable per-intent relay outcomes after a socket write.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery error for stale leases, duplicate/unknown
    /// IDs, invalid categories, or contradictory terminal transitions.
    pub fn apply_relay_results(
        claim: &DeliveryClaim,
        results: Vec<RelayDeliveryResult>,
    ) -> Result<Vec<DeliveryStateMutation>, SoftchatError> {
        if results.is_empty() {
            return Err(SoftchatError::InvalidDeliveryState);
        }
        let claimed = claim
            .intents
            .iter()
            .map(|intent| (intent.intent_id.as_str(), intent))
            .collect::<std::collections::BTreeMap<_, _>>();
        let mut seen = BTreeSet::new();
        let mut mutations = Vec::with_capacity(results.len());
        for result in results {
            if !seen.insert(result.intent_id.clone())
                || result.category.is_empty()
                || result.category.len() > MAX_DELIVERY_CATEGORY_BYTES
                || !result
                    .category
                    .bytes()
                    .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
            {
                return Err(SoftchatError::InvalidDeliveryState);
            }
            let intent = claimed
                .get(result.intent_id.as_str())
                .ok_or(SoftchatError::InvalidDeliveryState)?;
            if intent.state != DeliveryIntentState::SocketWritten
                && intent.state != DeliveryIntentState::Claimed
            {
                return Err(SoftchatError::InvalidDeliveryState);
            }
            let state = match result.kind {
                RelayDeliveryResultKind::Accepted => DeliveryIntentState::Accepted,
                RelayDeliveryResultKind::Rejected => DeliveryIntentState::Rejected,
                RelayDeliveryResultKind::Retryable => DeliveryIntentState::Retryable,
            };
            mutations.push(DeliveryStateMutation {
                intent_id: result.intent_id,
                state,
                lease_id: String::new(),
                category: result.category,
                attempt_count: intent.attempt_count,
            });
        }
        Ok(mutations)
    }

    /// Return every unfinished row in a lease to the retry queue.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery error for an invalid claim.
    pub fn release_claim(
        claim: &DeliveryClaim,
    ) -> Result<Vec<DeliveryStateMutation>, SoftchatError> {
        if claim.intents.is_empty() {
            return Err(SoftchatError::InvalidDeliveryState);
        }
        claim
            .intents
            .iter()
            .map(|intent| {
                validate_intent_snapshot(intent)?;
                if !matches!(
                    intent.state,
                    DeliveryIntentState::Claimed | DeliveryIntentState::SocketWritten
                ) {
                    return Err(SoftchatError::InvalidDeliveryState);
                }
                Ok(DeliveryStateMutation {
                    intent_id: intent.intent_id.clone(),
                    state: DeliveryIntentState::Retryable,
                    lease_id: String::new(),
                    category: "lease_released".to_owned(),
                    attempt_count: intent.attempt_count,
                })
            })
            .collect()
    }

    /// Derive one user-visible operation snapshot from durable intent rows.
    ///
    /// # Errors
    ///
    /// Returns a stable operation error for empty or mixed-operation input.
    pub fn operation_snapshot(
        intents: Vec<DeliveryIntentSnapshot>,
    ) -> Result<OperationSnapshot, SoftchatError> {
        let operation_id = intents
            .first()
            .map(|intent| intent.operation_id.clone())
            .ok_or(SoftchatError::InvalidAccountOperation)?;
        validate_identifier(&operation_id, MAX_OPERATION_ID_BYTES)?;
        let mut accepted = 0_u32;
        let mut rejected = 0_u32;
        let mut queued = 0_u32;
        let mut active = 0_u32;
        let mut cancelled = 0_u32;
        for intent in &intents {
            validate_intent_snapshot(intent)?;
            if intent.operation_id != operation_id {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            match intent.state {
                DeliveryIntentState::Accepted => accepted = accepted.saturating_add(1),
                DeliveryIntentState::Rejected => rejected = rejected.saturating_add(1),
                DeliveryIntentState::Pending | DeliveryIntentState::Retryable => {
                    queued = queued.saturating_add(1);
                }
                DeliveryIntentState::Claimed | DeliveryIntentState::SocketWritten => {
                    active = active.saturating_add(1);
                }
                DeliveryIntentState::Cancelled => cancelled = cancelled.saturating_add(1),
            }
        }
        let total =
            u32::try_from(intents.len()).map_err(|_| SoftchatError::InvalidAccountOperation)?;
        let state = if accepted == total {
            OperationState::Sent
        } else if cancelled == total {
            OperationState::Cancelled
        } else if accepted > 0 {
            OperationState::PartiallySent
        } else if rejected > 0 {
            OperationState::Failed
        } else if active > 0 {
            OperationState::Sending
        } else {
            OperationState::Queued
        };
        Ok(OperationSnapshot {
            operation_id,
            state,
            total_intents: total,
            accepted_intents: accepted,
            rejected_intents: rejected,
            queued_intents: queued,
            active_intents: active,
        })
    }
}

/// Compute the released-Android conversation ID from complete participants.
///
/// # Errors
///
/// Returns a stable account-operation error unless there are at least two
/// unique canonical public keys.
pub fn conversation_id(public_keys: Vec<String>) -> Result<String, SoftchatError> {
    if public_keys.len() < 2 || public_keys.len() > crate::MAX_CHAT_PARTICIPANTS + 1 {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    let mut keys = public_keys
        .iter()
        .map(|value| {
            NostrPublicKey::from_hex(value).map_err(|_| SoftchatError::InvalidAccountOperation)
        })
        .collect::<Result<Vec<_>, _>>()?;
    keys.sort_by_key(NostrPublicKey::to_hex);
    let original_len = keys.len();
    keys.dedup();
    if keys.len() != original_len {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    let mut digest = Sha256::new();
    for key in keys {
        digest.update(key.as_inner().as_bytes());
    }
    Ok(hex::encode(digest.finalize()))
}

fn prepare_operation(
    account_id: String,
    operation_id: String,
    event_copies: Vec<PreparedEventCopy>,
    relay_urls: Vec<String>,
) -> Result<PreparedOutgoingOperation, SoftchatError> {
    let intent_count = event_copies
        .len()
        .checked_mul(relay_urls.len())
        .ok_or(SoftchatError::InvalidAccountOperation)?;
    if intent_count == 0 || intent_count > MAX_OPERATION_DELIVERY_INTENTS {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    let mut relay_intents = Vec::with_capacity(intent_count);
    for event_copy in &event_copies {
        let event = SignedNostrEvent::try_from(event_copy.event.clone())?;
        let event_json = event.to_json()?;
        crate::session::validate_event_frame_json(&event_json)?;
        let payload_bytes =
            u64::try_from(event_json.len()).map_err(|_| SoftchatError::InvalidAccountOperation)?;
        for relay_url in &relay_urls {
            relay_intents.push(PreparedRelayIntent {
                intent_id: stable_intent_id(&operation_id, &event_copy.event_copy_id, relay_url),
                operation_id: operation_id.clone(),
                event_copy_id: event_copy.event_copy_id.clone(),
                relay_url: relay_url.clone(),
                payload_bytes,
            });
        }
    }
    Ok(PreparedOutgoingOperation {
        account_id,
        operation_id,
        event_copies,
        relay_intents,
    })
}

fn projection_for_rumor(
    rumor: &NostrRumor,
    authenticated_recipient: &NostrPublicKey,
    ephemeral: bool,
) -> Result<ProjectionMutation, SoftchatError> {
    let record = RumorEvent::from(rumor);
    let (kind, target_event_ids, content) = match rumor.kind() {
        NostrEventKind::PRIVATE_DIRECT_MESSAGE => {
            if let Ok(view) = classify_chat_message(record.clone()) {
                let targets = (!view.related_event_id.is_empty())
                    .then_some(view.related_event_id)
                    .into_iter()
                    .collect();
                (ProjectionKind::ChatMessage, targets, view.rumor.content)
            } else {
                let view = classify_subject(record.clone())?;
                (ProjectionKind::Subject, Vec::new(), view.subject)
            }
        }
        NostrEventKind::REACTION => {
            let view = classify_reaction(record.clone())?;
            (ProjectionKind::Reaction, vec![view.event_id], view.reaction)
        }
        NostrEventKind::EVENT_DELETION_REQUEST => {
            let view = classify_deletion(record.clone())?;
            (ProjectionKind::Deletion, view.event_ids, view.reason)
        }
        NostrEventKind::UPDATED_CONTENT => {
            let view = classify_edit(record.clone())?;
            (
                ProjectionKind::Edit,
                vec![view.original_event_id],
                view.content,
            )
        }
        NostrEventKind::TYPING => {
            classify_typing(record.clone())?;
            (ProjectionKind::Typing, Vec::new(), String::new())
        }
        NostrEventKind::USER_METADATA => {
            let view = classify_user_metadata(record.clone())?;
            (ProjectionKind::UserMetadata, Vec::new(), view.rumor.content)
        }
        NostrEventKind::FOLLOW_LIST => {
            classify_follow_list(record.clone())?;
            (ProjectionKind::FollowList, Vec::new(), String::new())
        }
        _ => (
            ProjectionKind::Unknown,
            Vec::new(),
            rumor.content().to_owned(),
        ),
    };
    // Only chat-domain rumors are allowed to create product conversations.
    // Account-domain rumors such as private follow lists also contain `p`
    // tags, but those tags are contacts rather than conversation members.
    let (conversation_id, participants) = if matches!(
        kind,
        ProjectionKind::ChatMessage
            | ProjectionKind::Subject
            | ProjectionKind::Reaction
            | ProjectionKind::Deletion
            | ProjectionKind::Edit
            | ProjectionKind::Typing
    ) {
        conversation_for_rumor(rumor, authenticated_recipient)?
    } else {
        (String::new(), Vec::new())
    };
    Ok(ProjectionMutation {
        kind,
        logical_event_id: rumor.id().to_hex(),
        author_public_key: rumor.public_key().to_hex(),
        created_at: i64::try_from(rumor.created_at())
            .map_err(|_| SoftchatError::InvalidAccountOperation)?,
        conversation_id,
        participants,
        target_event_ids,
        content,
        ephemeral,
    })
}

fn projection_for_signed(
    identity: &LocalIdentity,
    event: &SignedNostrEvent,
) -> Result<ProjectionMutation, SoftchatError> {
    let record = SignedEvent::from(event);
    let (kind, content) = match event.kind() {
        NostrEventKind::APPLICATION_DATA => {
            let view = classify_application_data(record.clone())?;
            (ProjectionKind::ApplicationData, view.content)
        }
        NostrEventKind::APPLICATION_DATA_SYNC => {
            let view = identity.decrypt_app_data_sync(event)?;
            let context = match view.context {
                AppDataContext::ReadState => "read-state",
                AppDataContext::AppSettings => "app-settings",
            };
            (
                ProjectionKind::AppDataSync,
                format!("{context}:{}", view.json),
            )
        }
        NostrEventKind::GENERIC_REPOST => {
            classify_generic_repost(record)?;
            (ProjectionKind::GenericRepost, event.content().to_owned())
        }
        _ => (ProjectionKind::Unknown, event.content().to_owned()),
    };
    Ok(ProjectionMutation {
        kind,
        logical_event_id: event.id().to_hex(),
        author_public_key: event.public_key().to_hex(),
        created_at: i64::try_from(event.created_at())
            .map_err(|_| SoftchatError::InvalidAccountOperation)?,
        conversation_id: String::new(),
        participants: Vec::new(),
        target_event_ids: Vec::new(),
        content,
        ephemeral: false,
    })
}

fn conversation_for_rumor(
    rumor: &NostrRumor,
    authenticated_recipient: &NostrPublicKey,
) -> Result<(String, Vec<String>), SoftchatError> {
    let participant_count = rumor
        .tags()
        .filter(|tag| tag.first().map(String::as_str) == Some("p"))
        .count();
    if participant_count == 0 || participant_count > crate::MAX_CHAT_PARTICIPANTS {
        return Err(SoftchatError::InvalidSoftchatEvent);
    }

    let author = rumor.public_key();
    let mut participants = Vec::with_capacity(participant_count.saturating_add(1));
    for tag in rumor
        .tags()
        .filter(|tag| tag.first().map(String::as_str) == Some("p"))
    {
        let value = tag.get(1).ok_or(SoftchatError::InvalidSoftchatEvent)?;
        let participant =
            NostrPublicKey::from_hex(value).map_err(|_| SoftchatError::InvalidSoftchatEvent)?;
        if participant == author {
            return Err(SoftchatError::InvalidSoftchatEvent);
        }
        participants.push(participant);
    }
    participants.sort_by_key(NostrPublicKey::to_hex);
    let original_len = participants.len();
    participants.dedup();
    if participants.len() != original_len {
        return Err(SoftchatError::InvalidSoftchatEvent);
    }
    participants.push(author);
    participants.sort_by_key(NostrPublicKey::to_hex);
    if !participants.contains(authenticated_recipient) || participants.len() < 2 {
        return Err(SoftchatError::InvalidSoftchatEvent);
    }
    let participants = participants
        .into_iter()
        .map(|key| key.to_hex())
        .collect::<Vec<_>>();
    Ok((conversation_id(participants.clone())?, participants))
}

fn canonical_relay_urls(values: Vec<String>) -> Result<Vec<String>, SoftchatError> {
    if values.is_empty() || values.len() > MAX_OPERATION_RELAYS {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    let mut relays = values
        .iter()
        .map(|value| {
            parse_relay_endpoint(value)
                .map(|endpoint| endpoint.configured_url)
                .map_err(|_| SoftchatError::InvalidAccountOperation)
        })
        .collect::<Result<Vec<_>, _>>()?;
    relays.sort();
    let original_len = relays.len();
    relays.dedup();
    if relays.len() != original_len {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    Ok(relays)
}

pub(crate) fn stable_intent_id(operation_id: &str, event_copy_id: &str, relay_url: &str) -> String {
    let mut digest = Sha256::new();
    digest.update(operation_id.as_bytes());
    digest.update([0]);
    digest.update(event_copy_id.as_bytes());
    digest.update([0]);
    digest.update(relay_url.as_bytes());
    hex::encode(digest.finalize())
}

fn validate_identifier(value: &str, max_bytes: usize) -> Result<(), SoftchatError> {
    if value.is_empty()
        || value.len() > max_bytes
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.' | b':'))
    {
        Err(SoftchatError::InvalidAccountOperation)
    } else {
        Ok(())
    }
}

fn validate_intent_snapshot(intent: &DeliveryIntentSnapshot) -> Result<(), SoftchatError> {
    validate_identifier(&intent.intent_id, MAX_OPERATION_ID_BYTES)
        .map_err(|_| SoftchatError::InvalidDeliveryState)?;
    validate_identifier(&intent.operation_id, MAX_OPERATION_ID_BYTES)
        .map_err(|_| SoftchatError::InvalidDeliveryState)?;
    if intent.event_copy_id.len() != 64
        || hex::decode(&intent.event_copy_id)
            .ok()
            .is_none_or(|bytes| bytes.len() != 32)
        || intent.payload_bytes == 0
        || intent.payload_bytes > crate::MAX_NOSTR_EVENT_JSON_BYTES as u64
    {
        return Err(SoftchatError::InvalidDeliveryState);
    }
    canonical_relay_urls(vec![intent.relay_url.clone()])
        .map_err(|_| SoftchatError::InvalidDeliveryState)?;
    if matches!(
        intent.state,
        DeliveryIntentState::Claimed | DeliveryIntentState::SocketWritten
    ) {
        validate_identifier(&intent.lease_id, MAX_LEASE_ID_BYTES)
            .map_err(|_| SoftchatError::InvalidDeliveryState)?;
    } else if !intent.lease_id.is_empty() {
        return Err(SoftchatError::InvalidDeliveryState);
    }
    Ok(())
}

fn transition_claim_subset(
    claim: &DeliveryClaim,
    intent_ids: Vec<String>,
    expected: DeliveryIntentState,
    next: DeliveryIntentState,
    category: &str,
    retain_lease: bool,
) -> Result<Vec<DeliveryStateMutation>, SoftchatError> {
    if intent_ids.is_empty() {
        return Err(SoftchatError::InvalidDeliveryState);
    }
    validate_identifier(&claim.lease_id, MAX_LEASE_ID_BYTES)
        .map_err(|_| SoftchatError::InvalidDeliveryState)?;
    let claimed = claim
        .intents
        .iter()
        .map(|intent| (intent.intent_id.as_str(), intent))
        .collect::<std::collections::BTreeMap<_, _>>();
    let mut seen = BTreeSet::new();
    intent_ids
        .into_iter()
        .map(|intent_id| {
            if !seen.insert(intent_id.clone()) {
                return Err(SoftchatError::InvalidDeliveryState);
            }
            let intent = claimed
                .get(intent_id.as_str())
                .ok_or(SoftchatError::InvalidDeliveryState)?;
            if intent.state != expected
                || (!intent.lease_id.is_empty() && intent.lease_id != claim.lease_id)
            {
                return Err(SoftchatError::InvalidDeliveryState);
            }
            Ok(DeliveryStateMutation {
                intent_id,
                state: next,
                lease_id: if retain_lease {
                    claim.lease_id.clone()
                } else {
                    String::new()
                },
                category: category.to_owned(),
                attempt_count: intent.attempt_count,
            })
        })
        .collect()
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::*;
    use crate::{ChatMessageDraft, ChatRelation, NostrEventDraft};

    const ALICE: &str = "5c0c523f52a5b6fad39ed2403092df8cebc36318b39383bca6c00808626fab3a";
    const BOB: &str = "4b22aa260e4acb7021e32f38a6cdf4b673c6a277755bfce287e370c924dc936d";

    fn intent(id: u8, state: DeliveryIntentState, bytes: u64) -> DeliveryIntentSnapshot {
        DeliveryIntentSnapshot {
            intent_id: format!("intent-{id}"),
            operation_id: "operation-1".to_owned(),
            event_copy_id: hex::encode([id; 32]),
            relay_url: "wss://relay.example/".to_owned(),
            payload_bytes: bytes,
            state,
            lease_id: if matches!(
                state,
                DeliveryIntentState::Claimed | DeliveryIntentState::SocketWritten
            ) {
                "lease-1".to_owned()
            } else {
                String::new()
            },
            attempt_count: 0,
        }
    }

    #[test]
    fn account_events_reserve_their_publish_frame_before_acceptance() -> Result<(), SoftchatError> {
        let identity = LocalIdentity::from_secret_hex(ALICE)?;
        let sign = |content: String| {
            identity.sign_event(NostrEventDraft::new(
                1_700_000_000,
                NostrEventKind::SHORT_TEXT_NOTE,
                Vec::new(),
                content,
            )?)
        };
        let base_bytes = sign(String::new())?.to_json()?.len();
        let envelope_bytes = "[\"EVENT\",]".len();
        for event_bytes in [
            crate::MAX_RELAY_FRAME_BYTES - envelope_bytes,
            crate::MAX_RELAY_FRAME_BYTES - envelope_bytes + 1,
        ] {
            let event = sign("x".repeat(event_bytes - base_bytes))?;
            let json = event.to_json()?;
            assert_eq!(json.len(), event_bytes);
            assert!(SignedNostrEvent::from_json(&json).is_ok());
            let incoming = AccountEngine::prepare_incoming(
                &identity,
                "frame-account".to_owned(),
                vec![SignedEvent::from(&event)],
            );
            let outgoing = AccountEngine::prepare_signed_operation(
                "frame-account".to_owned(),
                "frame-command".to_owned(),
                vec![SignedEvent::from(&event)],
                vec!["wss://relay.example".to_owned()],
            );
            if event_bytes + envelope_bytes <= crate::MAX_RELAY_FRAME_BYTES {
                assert!(incoming.is_ok());
                assert!(outgoing.is_ok());
                assert_eq!(
                    crate::ClientRelayFrame::Event(event).to_json()?.len(),
                    crate::MAX_RELAY_FRAME_BYTES
                );
            } else {
                assert!(matches!(incoming, Err(SoftchatError::InvalidRelayFrame)));
                assert!(matches!(outgoing, Err(SoftchatError::InvalidRelayFrame)));
            }
        }
        Ok(())
    }

    #[test]
    fn prepares_and_authenticates_account_batches_and_operations() {
        let alice = LocalIdentity::from_secret_hex(ALICE).unwrap();
        let bob = LocalIdentity::from_secret_hex(BOB).unwrap();
        let message = alice
            .create_chat_message(ChatMessageDraft {
                created_at: 1_700_000_000,
                participants: vec![bob.public_key()],
                content: "hello".to_owned(),
                relation: ChatRelation::None,
                attachments: Vec::new(),
                emoji_tags: Vec::new(),
                extension_tags: Vec::new(),
            })
            .unwrap();
        let operation = AccountEngine::prepare_rumor_operation(
            &alice,
            "account-1".to_owned(),
            "operation-1".to_owned(),
            message.rumor.clone(),
            vec![bob.public_key().to_hex()],
            vec![bob.public_key().to_hex()],
            vec!["wss://relay.example".to_owned()],
            Nip59EnvelopeKind::Durable,
            1_700_172_800,
        )
        .unwrap();
        assert_eq!(operation.event_copies.len(), 2);
        assert_eq!(operation.relay_intents.len(), 2);
        for copy in &operation.event_copies {
            let event = SignedNostrEvent::try_from(copy.event.clone()).unwrap();
            let has_alert = event
                .tags()
                .any(|tag| tag.first().is_some_and(|value| value == "alert"));
            assert_eq!(
                has_alert,
                copy.recipient_public_key == bob.public_key().to_hex()
            );
        }
        assert_ne!(
            operation.event_copies[0].event.public_key,
            operation.event_copies[1].event.public_key
        );

        let bob_copy = operation
            .event_copies
            .iter()
            .find(|copy| copy.recipient_public_key == bob.public_key().to_hex())
            .unwrap();
        let batch = AccountEngine::prepare_incoming(
            &bob,
            "account-2".to_owned(),
            vec![bob_copy.event.clone()],
        )
        .unwrap();
        assert_eq!(batch.events[0].projection.kind, ProjectionKind::ChatMessage);
        assert_eq!(batch.events[0].projection.content, "hello");
        assert_eq!(batch.events[0].projection.participants.len(), 2);

        let receipt = IngestionReceipt {
            account_id: "account-2".to_owned(),
            inserted_event_ids: vec![bob_copy.event.id.clone()],
            duplicate_event_ids: Vec::new(),
            quarantined_event_ids: Vec::new(),
        };
        AccountEngine::validate_ingestion_receipt(&batch, &receipt).unwrap();
    }

    #[test]
    fn account_data_p_tags_do_not_create_chat_conversations() {
        let alice = LocalIdentity::from_secret_hex(ALICE).unwrap();
        let bob = LocalIdentity::from_secret_hex(BOB).unwrap();
        let follow_list = alice
            .create_follow_list(
                1_700_000_000,
                vec![crate::Contact {
                    public_key: bob.public_key().to_hex(),
                    relay_hint: String::new(),
                    local_name: "Bob".to_owned(),
                }],
            )
            .unwrap();
        let rumor = NostrRumor::try_from(follow_list.rumor).unwrap();

        let projection = projection_for_rumor(&rumor, &alice.public_key(), false).unwrap();

        assert_eq!(projection.kind, ProjectionKind::FollowList);
        assert!(projection.conversation_id.is_empty());
        assert!(projection.participants.is_empty());
    }

    #[test]
    fn conversation_id_matches_the_released_android_hash() {
        let alice = LocalIdentity::from_secret_hex(ALICE).unwrap();
        let bob = LocalIdentity::from_secret_hex(BOB).unwrap();
        let result =
            conversation_id(vec![bob.public_key().to_hex(), alice.public_key().to_hex()]).unwrap();
        let mut digest = Sha256::new();
        let mut keys = [alice.public_key(), bob.public_key()];
        keys.sort_by_key(NostrPublicKey::to_hex);
        for key in keys {
            digest.update(key.as_inner().as_bytes());
        }
        assert_eq!(result, hex::encode(digest.finalize()));
    }

    #[test]
    fn delivery_claim_and_results_never_treat_socket_write_as_acceptance() {
        let claim = DeliveryReducer::claim(
            vec![
                intent(1, DeliveryIntentState::Pending, 100),
                intent(2, DeliveryIntentState::Retryable, 200),
            ],
            "lease-1".to_owned(),
        )
        .unwrap();
        assert_eq!(claim.payload_bytes, 300);
        assert!(
            claim
                .mutations
                .iter()
                .all(|mutation| mutation.state == DeliveryIntentState::Claimed)
        );

        let mut persisted_claim = claim.clone();
        for intent in &mut persisted_claim.intents {
            intent.state = DeliveryIntentState::Claimed;
            intent.lease_id = "lease-1".to_owned();
        }
        let written = DeliveryReducer::mark_socket_written(
            &persisted_claim,
            vec!["intent-1".to_owned(), "intent-2".to_owned()],
        )
        .unwrap();
        assert!(
            written
                .iter()
                .all(|mutation| mutation.state == DeliveryIntentState::SocketWritten)
        );

        for intent in &mut persisted_claim.intents {
            intent.state = DeliveryIntentState::SocketWritten;
        }
        let outcomes = DeliveryReducer::apply_relay_results(
            &persisted_claim,
            vec![
                RelayDeliveryResult {
                    intent_id: "intent-1".to_owned(),
                    kind: RelayDeliveryResultKind::Accepted,
                    category: "accepted".to_owned(),
                },
                RelayDeliveryResult {
                    intent_id: "intent-2".to_owned(),
                    kind: RelayDeliveryResultKind::Retryable,
                    category: "rate_limited".to_owned(),
                },
            ],
        )
        .unwrap();
        assert_eq!(outcomes[0].state, DeliveryIntentState::Accepted);
        assert_eq!(outcomes[1].state, DeliveryIntentState::Retryable);
    }

    #[test]
    fn operation_state_requires_every_relay_intent_to_accept() {
        let queued = DeliveryReducer::operation_snapshot(vec![
            intent(1, DeliveryIntentState::Pending, 1),
            intent(2, DeliveryIntentState::Retryable, 1),
        ])
        .unwrap();
        assert_eq!(queued.state, OperationState::Queued);

        let partial = DeliveryReducer::operation_snapshot(vec![
            intent(1, DeliveryIntentState::Accepted, 1),
            intent(2, DeliveryIntentState::Retryable, 1),
        ])
        .unwrap();
        assert_eq!(partial.state, OperationState::PartiallySent);

        let sent = DeliveryReducer::operation_snapshot(vec![
            intent(1, DeliveryIntentState::Accepted, 1),
            intent(2, DeliveryIntentState::Accepted, 1),
        ])
        .unwrap();
        assert_eq!(sent.state, OperationState::Sent);
    }

    #[test]
    fn incoming_rejects_duplicates_and_receipt_gaps() {
        let alice = LocalIdentity::from_secret_hex(ALICE).unwrap();
        let event = alice
            .sign_event(
                NostrEventDraft::new(
                    1,
                    NostrEventKind::APPLICATION_DATA,
                    vec![crate::NostrTag::new(vec!["d".to_owned(), "fixture".to_owned()]).unwrap()],
                    "content",
                )
                .unwrap(),
            )
            .unwrap();
        let value = SignedEvent::from(&event);
        assert!(matches!(
            AccountEngine::prepare_incoming(
                &alice,
                "account-1".to_owned(),
                vec![value.clone(), value]
            ),
            Err(SoftchatError::InvalidAccountOperation)
        ));
    }

    #[test]
    fn incoming_chat_binds_outer_recipient_to_inner_participants() {
        let alice = LocalIdentity::from_secret_hex(ALICE).unwrap();
        let bob = LocalIdentity::from_secret_hex(BOB).unwrap();
        let carol = LocalIdentity::from_secret_hex(
            "8f40e50a84a7462e2b8d24c28898ef1f23359fff50d8c509e6fb7ce06e142f9c",
        )
        .unwrap();

        let prepare = |recipient: &LocalIdentity, tags: Vec<crate::NostrTag>| {
            let rumor = alice
                .create_rumor(
                    NostrEventDraft::new(
                        1_700_000_000,
                        NostrEventKind::PRIVATE_DIRECT_MESSAGE,
                        tags,
                        "bounded participant fixture",
                    )
                    .unwrap(),
                )
                .unwrap();
            let outer = alice
                .gift_wrap(
                    &recipient.public_key(),
                    &rumor,
                    Nip59EnvelopeKind::Durable,
                    1_700_172_800,
                )
                .unwrap();
            AccountEngine::prepare_incoming(
                recipient,
                "recipient-account".to_owned(),
                vec![SignedEvent::from(&outer)],
            )
        };
        let participant_tag = |identity: &LocalIdentity| {
            crate::NostrTag::new(vec!["p".to_owned(), identity.public_key().to_hex()]).unwrap()
        };

        assert!(prepare(&bob, vec![participant_tag(&bob)]).is_ok());
        // The author copy is valid because the authenticated recipient is the
        // rumor author and the inner room still contains another participant.
        assert!(prepare(&alice, vec![participant_tag(&bob)]).is_ok());
        assert!(matches!(
            prepare(&bob, vec![participant_tag(&carol)]),
            Err(SoftchatError::InvalidSoftchatEvent)
        ));
        assert!(matches!(
            prepare(&bob, vec![participant_tag(&bob), participant_tag(&bob)]),
            Err(SoftchatError::InvalidSoftchatEvent)
        ));
        assert!(matches!(
            prepare(&alice, Vec::new()),
            Err(SoftchatError::InvalidSoftchatEvent)
        ));

        let mut oversized = vec![participant_tag(&bob)];
        for value in 1_u16..=crate::MAX_CHAT_PARTICIPANTS as u16 {
            let identity = LocalIdentity::from_secret_hex(&format!("{value:064x}")).unwrap();
            oversized.push(participant_tag(&identity));
        }
        assert!(oversized.len() > crate::MAX_CHAT_PARTICIPANTS);
        assert!(matches!(
            prepare(&bob, oversized),
            Err(SoftchatError::InvalidSoftchatEvent)
        ));
    }
}