softchat/

runtime_facade.rs

//! Generated-binding handles for the account, delivery, relay, and sync plans.

#![allow(
    unreachable_pub,
    reason = "UniFFI exports public handles from this private adapter module"
)]

use std::collections::{BTreeMap, BTreeSet};
use std::fmt;
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Mutex, MutexGuard, OnceLock};

use bech32::{Bech32, Hrp};
use sha2::{Digest, Sha256};
use url::Url;

use crate::account_diagnostics::{
    AccountLogBatch, AccountLogBuffer, AccountLogCategory, AccountLogLevel, AccountLogSource,
    AccountLogSubject, Diagnostic,
};
#[cfg(test)]
use crate::account_diagnostics::{AccountLogMetadata, MAX_ACCOUNT_LOG_BATCH};
use crate::account_product::{
    command_hash, product_draft_match, profile_update_from_patch, validate_command_id,
    validate_conversation_id,
};
use crate::account_transport::{AccountTransportRun, StoredAccountTransportOutcome, result_hash};
use crate::event::{RumorEvent, SignedEvent, SignedNostrEvent};
use crate::media::FfiHttpAuthorizationPlan;
use crate::message::LocalIdentityHandle;
use crate::storage::ProductAccountOperationResult;
use crate::{
    AccountDatabase, AccountDatabaseInfo, AccountDiagnostics, AccountDraft, AccountEngine,
    AccountEventNode, AccountMediaItem, AccountMediaLease, AccountMediaOperation,
    AccountMediaOperationPage, AccountMessage, AccountMessageView, AccountMutationResult,
    AccountOperationResult, AccountProfile, AccountProjection, AccountSettings,
    AccountSettingsMutation, AccountTransportBatch, AccountTransportResult, AppDataContext,
    AssetReplacementTarget, AttachmentMetadata, ChatMessageDraft, ChatRelation, ClaimedDelivery,
    Contact, ConversationCursor, ConversationPage, ConversationRecord, ConversationSummary,
    ConversationView, DeliveryClaim, DeliveryIntentSnapshot, DeliveryReducer,
    DeliveryStateMutation, DnsRelayRecord, IngestionReceipt, LinkPreviewRecord, LocalContactRecord,
    MessageContentInput, MessageCursor, MessagePage, MessageReactionInput, NegentropyItem,
    Nip59EnvelopeKind, NostrEventId, NostrEventKind, NostrPublicKey, NostrTag, OperationSnapshot,
    PendingAssetReplacement, PreparedIncomingBatch, PreparedOutgoingOperation,
    ProductOperationResult, PushPlatform, ReactionDraft, RelayCatalogEntry, RelayCatalogMutation,
    RelayCatalogPlan, RelayCatalogReducer, RelayConnectionState, RelayDeliveryResult,
    RelayEndpointPlan, RelayFailoverMutation, RelayFailureKind, RelayFilter, RelayRetryPlan,
    SoftchatError, StickerRecord, StoredEventJson, StoredIngestion, StoredRelayCatalogPlan,
    SubjectDraft, SyncAction, SyncActionKind, SyncEngine, SyncEngineSnapshot, conversation_id,
    parse_relay_endpoint, plan_relay_retry, route_nip59_envelope,
};

/// Android-facing owner of one identity and one Rust-managed account database.
#[cfg_attr(feature = "native-bindings", derive(uniffi::Object))]
pub struct AccountRuntimeHandle {
    account_id: String,
    identity: Arc<LocalIdentityHandle>,
    database: Mutex<AccountDatabase>,
    database_path: PathBuf,
    closed: AtomicBool,
    sync_engines: Mutex<BTreeMap<String, SyncEngine>>,
    transport_run: Mutex<Option<AccountTransportRun>>,
    typing_authored: Mutex<BTreeMap<String, (i64, String)>>,
    pub(crate) logs: Arc<AccountLogBuffer>,
    logging_configured: AtomicBool,
}

/// Non-secret identity-derived account storage descriptor.
#[derive(Clone, Debug)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountDescriptor {
    /// Canonical local account public key.
    pub public_key: String,
    /// Deterministic database filename without a directory.
    pub database_filename: String,
}

/// Verified account-routing hint from one signed NIP-59 outer wrapper.
#[derive(Clone, Debug)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct IncomingAccountRoute {
    /// Verified outer wrapper event ID.
    pub event_id: String,
    /// Canonical recipient public key from the one `p` tag.
    pub recipient_public_key: String,
    /// Durable or ephemeral wrapper profile.
    pub envelope_kind: Nip59EnvelopeKind,
}

/// Verify one outer wrapper and extract its account-routing hint.
///
/// The selected account must still ingest and authenticate every encrypted
/// layer. No plaintext is exposed by this operation.
///
/// # Errors
///
/// Returns a stable event or envelope failure for malformed, forged,
/// unsupported, or ambiguously routed input.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn route_incoming_account_event(
    event_json: String,
) -> Result<IncomingAccountRoute, SoftchatError> {
    let route = route_nip59_envelope(&event_json)?;
    Ok(IncomingAccountRoute {
        event_id: route.outer_event_id().to_hex(),
        recipient_public_key: route.recipient().to_hex(),
        envelope_kind: route.kind(),
    })
}

/// Generate one cryptographically secure local account secret.
///
/// The returned 32-byte mutable buffer is the only exported secret copy.
/// Platforms must protect it immediately and clear it after opening an
/// account. Validation and public-key derivation occur again at account open.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
#[must_use]
pub fn generate_account_secret_key() -> Vec<u8> {
    nostr::SecretKey::generate().to_secret_bytes().to_vec()
}

/// Parse one hexadecimal or `nsec` account authority into a mutable buffer.
///
/// # Errors
///
/// Returns [`SoftchatError::InvalidSecretKey`] for malformed input.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn parse_account_secret_key(encoded: String) -> Result<Vec<u8>, SoftchatError> {
    nostr::SecretKey::parse(&encoded)
        .map(|secret| secret.to_secret_bytes().to_vec())
        .map_err(|_| SoftchatError::InvalidSecretKey)
}

/// Encode one temporary 32-byte local authority as a canonical `nsec`.
///
/// The mutable binding buffer is cleared before return. Platforms must expose
/// the returned string only inside an authenticated, explicitly user-requested
/// export flow.
///
/// # Errors
///
/// Returns [`SoftchatError::InvalidSecretKey`] for a malformed secret.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn encode_account_secret_key(mut secret_key_bytes: Vec<u8>) -> Result<String, SoftchatError> {
    let result = (|| {
        let secret = nostr::SecretKey::from_slice(&secret_key_bytes)
            .map_err(|_| SoftchatError::InvalidSecretKey)?;
        bech32::encode::<Bech32>(
            Hrp::parse("nsec").map_err(|_| SoftchatError::InternalFailure)?,
            &secret.to_secret_bytes(),
        )
        .map_err(|_| SoftchatError::InternalFailure)
    })();
    secret_key_bytes.fill(0);
    result
}

/// Parse one stable product conversation identifier.
///
/// # Errors
///
/// Returns a stable account-operation error for malformed input.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn parse_account_conversation_id(value: String) -> Result<String, SoftchatError> {
    if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    Ok(value.to_ascii_lowercase())
}

/// Parse one stable product message identifier.
///
/// # Errors
///
/// Returns a stable event-ID error for malformed input.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn parse_account_message_id(value: String) -> Result<String, SoftchatError> {
    Ok(NostrEventId::from_hex(&value)?.to_hex())
}

/// Validate a secret and derive its non-secret account storage descriptor.
///
/// The supplied mutable binding buffer is cleared inside Rust before return.
///
/// # Errors
///
/// Returns a stable invalid-secret failure.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn describe_account_secret(
    secret_key_bytes: Vec<u8>,
) -> Result<AccountDescriptor, SoftchatError> {
    let identity = LocalIdentityHandle::new(secret_key_bytes)?;
    let public_key = identity.public_key()?;
    identity.erase();
    Ok(AccountDescriptor {
        database_filename: account_database_filename(&public_key)?,
        public_key,
    })
}

impl fmt::Debug for AccountRuntimeHandle {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("AccountRuntimeHandle")
            .field("account_id", &self.account_id)
            .field("database", &"<account-private sqlite>")
            .field("closed", &self.closed.load(Ordering::Acquire))
            .field("active_syncs", &lock(&self.sync_engines).len())
            .finish_non_exhaustive()
    }
}

#[cfg_attr(feature = "native-bindings", uniffi::export)]
impl AccountRuntimeHandle {
    /// Open one Rust-owned SQLite account runtime.
    ///
    /// Android supplies a private per-account file path and retains lifecycle,
    /// HTTP, files, background scheduling, and UI ownership. Relay transport
    /// can be host-driven or delegated to the optional managed resource.
    ///
    /// # Errors
    ///
    /// Returns a stable identity, account, schema, path, or persistence error.
    #[cfg_attr(feature = "native-bindings", uniffi::constructor)]
    pub fn open(
        secret_key_bytes: Vec<u8>,
        database_directory: String,
    ) -> Result<Self, SoftchatError> {
        let identity = Arc::new(LocalIdentityHandle::new(secret_key_bytes)?);
        let account_id = identity.public_key()?;
        let database_path = derived_account_path(&database_directory, &account_id)?;
        {
            let mut registry = lock(open_account_paths());
            if !registry.insert(database_path.clone()) {
                identity.erase();
                return Err(SoftchatError::AccountAlreadyOpen);
            }
        }
        let mut database = match AccountDatabase::open(&database_path, account_id.clone()) {
            Ok(database) => database,
            Err(error) => {
                lock(open_account_paths()).remove(&database_path);
                identity.erase();
                return Err(error);
            }
        };
        let logs = Arc::new(AccountLogBuffer::new(AccountLogLevel::Off));
        database.diagnostics.buffer = Some(logs.clone());
        Ok(Self {
            account_id,
            identity,
            database: Mutex::new(database),
            database_path,
            closed: AtomicBool::new(false),
            sync_engines: Mutex::new(BTreeMap::new()),
            transport_run: Mutex::new(None),
            typing_authored: Mutex::new(BTreeMap::new()),
            logs,
            logging_configured: AtomicBool::new(false),
        })
    }

    /// Return the non-secret account partition.
    #[must_use]
    pub fn account_id(&self) -> String {
        self.account_id.clone()
    }

    /// Return the canonical public key that owns this account.
    ///
    /// # Errors
    ///
    /// Returns [`SoftchatError::AccountClosed`] after explicit close.
    pub fn public_key(&self) -> Result<String, SoftchatError> {
        self.ensure_open()?;
        Ok(self.account_id.clone())
    }

    /// Return the derived account database filename without exposing a path.
    #[must_use]
    pub fn database_filename(&self) -> String {
        self.database_path
            .file_name()
            .and_then(|value| value.to_str())
            .unwrap_or_default()
            .to_owned()
    }

    /// Return the validated non-secret account identity and storage filename.
    ///
    /// This compact snapshot lets authored facades initialize without
    /// repeating native calls or querying SQLite metadata.
    ///
    /// # Errors
    ///
    /// Returns [`SoftchatError::AccountClosed`] after explicit close.
    pub fn descriptor(&self) -> Result<AccountDescriptor, SoftchatError> {
        self.ensure_open()?;
        Ok(AccountDescriptor {
            public_key: self.account_id.clone(),
            database_filename: self.database_filename(),
        })
    }

    /// Irreversibly close this account authority.
    ///
    /// The authored facade immediately releases the generated object after
    /// this call, which closes SQLite and removes the same-path registry entry.
    /// Repeated close calls are harmless.
    pub fn close_account(&self) {
        if !self.closed.swap(true, Ordering::AcqRel) {
            lock(&self.sync_engines).clear();
            lock(&self.transport_run).take();
            lock(&self.typing_authored).clear();
            self.identity.erase();
            self.logs.flush();
            self.logs.emit(Diagnostic::new(
                AccountLogCategory::Lifecycle,
                "account.close",
                "Account closed",
            ));
        }
    }

    /// Change the bounded account diagnostic threshold.
    ///
    /// `Off` keeps the disabled hot path to one relaxed atomic load. Enabling
    /// diagnostics records one lifecycle event; no secret, event identifier,
    /// URL, frame, or caller content is retained.
    ///
    /// # Errors
    ///
    /// Returns [`SoftchatError::AccountClosed`] after explicit close.
    pub fn set_log_level(&self, level: AccountLogLevel) -> Result<(), SoftchatError> {
        self.ensure_open()?;
        let previous = self.logs.level();
        self.logs.set_level(level);
        if level != AccountLogLevel::Off && previous != level {
            if self.logging_configured.swap(true, Ordering::Relaxed) {
                self.logs.emit(
                    Diagnostic::new(
                        AccountLogCategory::Lifecycle,
                        "account.logging.configure",
                        "Logging configured",
                    )
                    .debug(),
                );
            } else {
                self.logs.emit(Diagnostic::new(
                    AccountLogCategory::Lifecycle,
                    "account.opened",
                    "Account opened",
                ));
            }
        }
        Ok(())
    }

    /// Drain at most 64 structured account diagnostics in sequence order.
    ///
    /// The ring is process-memory-only and retains at most 256 events. The
    /// returned drop count reports overwritten events since the previous drain.
    ///
    /// # Errors
    ///
    /// Returns a stable bound error. A final drain remains valid immediately
    /// after `close_account` so the authored facade can deliver `account.close`
    /// before releasing the generated handle.
    pub fn drain_log_events(&self, limit: u32) -> Result<AccountLogBatch, SoftchatError> {
        self.logs.drain(limit)
    }

    /// Create or return one conversation from remote member keys.
    ///
    /// The local account key is derived internally and included exactly once.
    ///
    /// # Errors
    ///
    /// Returns a stable account, key, bound, or persistence error.
    pub fn get_or_create_conversation(
        &self,
        member_public_keys: Vec<String>,
        now: i64,
    ) -> Result<ConversationRecord, SoftchatError> {
        let own_public_key = self.public_key()?;
        self.database()?.get_or_create_product_conversation(
            &own_public_key,
            member_public_keys,
            now,
        )
    }

    /// Return one complete conversation by stable ID.
    ///
    /// # Errors
    ///
    /// Returns a stable account, key, or persistence error.
    pub fn conversation(
        &self,
        conversation_id: String,
    ) -> Result<Option<ConversationRecord>, SoftchatError> {
        let own_public_key = self.public_key()?;
        self.database()?
            .product_conversation(&conversation_id, &own_public_key)
    }

    /// Return one bounded stable page of complete conversations.
    ///
    /// # Errors
    ///
    /// Returns a stable cursor, bound, account, or persistence error.
    pub fn conversation_page(
        &self,
        archived: bool,
        cursor: Option<ConversationCursor>,
        limit: u32,
    ) -> Result<ConversationPage, SoftchatError> {
        let own_public_key = self.public_key()?;
        self.database()?
            .product_conversations(&own_public_key, archived, cursor, limit)
    }

    /// Return one complete effective message by ID.
    ///
    /// # Errors
    ///
    /// Returns a stable ID, account, or persistence error.
    pub fn message(&self, message_id: String) -> Result<Option<AccountMessageView>, SoftchatError> {
        self.database()?.product_message(&message_id)
    }

    /// Return up to 512 unique exact messages in caller order, omitting unknown IDs.
    ///
    /// Includes effective edits, reactions, and deletion tombstones from one database snapshot.
    /// Empty input returns an empty result.
    ///
    /// # Errors
    ///
    /// Returns a stable ID, duplicate, bound, account, or persistence error.
    pub fn messages_by_ids(
        &self,
        message_ids: Vec<String>,
    ) -> Result<Vec<AccountMessageView>, SoftchatError> {
        self.database()?.product_messages_by_ids(&message_ids)
    }

    /// Return one bounded lossless tuple-cursor message page.
    ///
    /// # Errors
    ///
    /// Returns a stable ID, cursor, bound, account, or persistence error.
    pub fn message_page(
        &self,
        conversation_id: String,
        cursor: Option<MessageCursor>,
        limit: u32,
    ) -> Result<MessagePage, SoftchatError> {
        self.database()?
            .product_messages(&conversation_id, cursor, limit)
    }

    /// Return a newest-first tuple-cursor page of direct replies in the root conversation.
    ///
    /// Unrelated messages and forwards do not consume the page. Includes deletion tombstones;
    /// an unknown root returns an empty page. Limits and cursors match `message_page`.
    ///
    /// # Errors
    ///
    /// Returns a stable ID, cursor, bound, account, or persistence error.
    pub fn message_replies(
        &self,
        message_id: String,
        cursor: Option<MessageCursor>,
        limit: u32,
    ) -> Result<MessagePage, SoftchatError> {
        self.database()?
            .product_message_replies(&message_id, cursor, limit)
    }

    /// Return a bounded context centered on one stored message.
    ///
    /// # Errors
    ///
    /// Returns a stable ID, bound, account, or persistence error.
    pub fn message_context(
        &self,
        message_id: String,
        before: u32,
        after: u32,
    ) -> Result<Vec<AccountMessageView>, SoftchatError> {
        self.database()?
            .product_message_context(&message_id, before, after)
    }

    /// Search effective live message text through the Rust-owned FTS index.
    ///
    /// Empty [conversation_id] searches the whole account.
    ///
    /// # Errors
    ///
    /// Returns a stable query, bound, account, or persistence error.
    pub fn search_messages(
        &self,
        conversation_id: String,
        query: String,
        limit: u32,
    ) -> Result<Vec<AccountMessageView>, SoftchatError> {
        self.database()?.search_product_messages(
            (!conversation_id.is_empty()).then_some(conversation_id),
            &query,
            limit,
        )
    }

    /// Return sparse account-local transcript and waveform data for one message.
    ///
    /// # Errors
    ///
    /// Returns a stable message, account, or persistence error.
    pub fn message_local_extras(
        &self,
        message_id: String,
    ) -> Result<Option<crate::MessageLocalExtras>, SoftchatError> {
        self.database()?.product_message_local_extras(&message_id)
    }

    /// Replace sparse account-local transcript and normalized waveform data.
    ///
    /// # Errors
    ///
    /// Returns a stable message, bound, account, or persistence error.
    pub fn set_message_local_extras(
        &self,
        message_id: String,
        transcript: Option<String>,
        waveform: Vec<u8>,
        now: i64,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?
            .set_product_message_local_extras(&message_id, transcript, waveform, now)
    }

    /// Return the exact number of archived or active conversations.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence error.
    pub fn conversation_count(&self, archived: bool) -> Result<u32, SoftchatError> {
        self.database()?.product_conversation_count(archived)
    }

    /// Return whether another participant has an unread live message.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn has_unread(&self, include_archived: bool) -> Result<bool, SoftchatError> {
        let own_public_key = self.public_key()?;
        self.database()?
            .product_has_unread(&own_public_key, include_archived)
    }

    /// Search complete conversation views by subject, member key, or known name.
    /// Matching ignores ASCII case; other Unicode characters remain literal.
    ///
    /// An empty archived filter searches both active and archived conversations.
    ///
    /// # Errors
    ///
    /// Returns a stable search, bound, account, or persistence error.
    pub fn search_conversations(
        &self,
        query: String,
        archived: Option<bool>,
        limit: u32,
    ) -> Result<Vec<ConversationRecord>, SoftchatError> {
        let own_public_key = self.public_key()?;
        self.database()?
            .search_product_conversations(&own_public_key, &query, archived, limit)
    }

    /// Send one atomic idempotent chat command.
    ///
    /// Rust resolves participants, relay selection, notification recipients,
    /// local echo, wrappers, and delivery intents from durable account state.
    /// The host persists its exact portable composer snapshot before passing it
    /// in `draft_to_consume`. When present,
    /// the same SQLite transaction clears it only if the stored semantic draft
    /// still matches. A newer draft is preserved, including when its autosave
    /// wins the race before this command acquires SQLite. An idempotent replay
    /// returns before draft matching.
    ///
    /// # Errors
    ///
    /// Returns a stable command, conversation, content, relay, or persistence error.
    pub fn send_message(
        &self,
        command_id: String,
        conversation_id: String,
        content: MessageContentInput,
        reply_to_message_id: String,
        draft_to_consume: Option<AccountDraft>,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            let draft_match = draft_to_consume.map(product_draft_match).transpose()?;
            let payload_hash = send_message_command_hash(
                &conversation_id,
                &content,
                &reply_to_message_id,
                draft_match.as_ref(),
            )?;
            let (rumor, recipients) = author_product_chat_message(
                self,
                &conversation_id,
                content,
                &reply_to_message_id,
                now,
            )?;
            let message_id = rumor.id.clone();
            let operation = {
                let mut database = self.database()?;
                self.identity.with_identity(|identity| {
                    database.enqueue_product_rumor(
                        identity,
                        command_id.clone(),
                        payload_hash,
                        message_id.clone(),
                        rumor,
                        recipients.clone(),
                        recipients,
                        Nip59EnvelopeKind::Durable,
                        now,
                        draft_match
                            .as_ref()
                            .map(|value| (conversation_id.as_str(), value)),
                    )
                })?
            };
            stored_product_result(self, operation)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Messaging,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::ChatMessage);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Forward one stored authenticated message to another conversation.
    ///
    /// Replaying a committed command returns its original operation even if
    /// the source message was subsequently deleted.
    ///
    /// # Errors
    ///
    /// Returns a stable command, message, conversation, relay, or persistence error.
    pub fn forward_message(
        &self,
        command_id: String,
        message_id: String,
        destination_conversation_id: String,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            NostrEventId::from_hex(&message_id)?;
            let payload_hash =
                command_hash(&("forward", &message_id, &destination_conversation_id))?;
            let existing = self
                .database()?
                .product_operation_if_committed(&command_id, &payload_hash)?;
            if let Some(operation) = existing {
                return stored_product_result(self, operation);
            }
            let original = self
                .database()?
                .product_message(&message_id)?
                .filter(|message| !message.deleted)
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let own_public_key = self.public_key()?;
            let destination = self
                .database()?
                .product_conversation(&destination_conversation_id, &own_public_key)?
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let recipients = remote_members(&destination, &own_public_key);
            let created_at =
                u64::try_from(now).map_err(|_| SoftchatError::InvalidAccountOperation)?;
            let relay_hint = self.database()?.active_relay_url()?;
            let rumor = self.identity.with_identity(|identity| {
                identity
                    .create_chat_message(ChatMessageDraft {
                        created_at,
                        participants: public_keys(&recipients)?,
                        content: original.text,
                        relation: ChatRelation::Forward {
                            event_id: NostrEventId::from_hex(&original.id)?,
                            relay_hint: Some(relay_hint),
                            original_author: NostrPublicKey::from_hex(&original.author_public_key)?,
                        },
                        attachments: original.attachments,
                        emoji_tags: nostr_tags(original.emoji_tags)?,
                        extension_tags: Vec::new(),
                    })
                    .map(|view| view.rumor)
            })?;
            let forwarded_message_id = rumor.id.clone();
            let operation = {
                let mut database = self.database()?;
                self.identity.with_identity(|identity| {
                    database.enqueue_product_rumor(
                        identity,
                        command_id,
                        payload_hash,
                        forwarded_message_id.clone(),
                        rumor,
                        recipients.clone(),
                        recipients,
                        Nip59EnvelopeKind::Durable,
                        now,
                        None,
                    )
                })?
            };
            stored_product_result(self, operation)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Messaging,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::ChatMessage);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Publish one conversation subject update.
    ///
    /// # Errors
    ///
    /// Returns a stable command, conversation, subject, relay, or persistence error.
    pub fn set_conversation_subject(
        &self,
        command_id: String,
        conversation_id: String,
        subject: String,
        icon: Option<AttachmentMetadata>,
        emoji_tags: Vec<Vec<String>>,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            if subject.len() > crate::MAX_PRODUCT_MESSAGE_BYTES {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            let own_public_key = self.public_key()?;
            let conversation = self
                .database()?
                .product_conversation(&conversation_id, &own_public_key)?
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let recipients = remote_members(&conversation, &own_public_key);
            let payload_hash =
                command_hash(&("subject", &conversation_id, &subject, &icon, &emoji_tags))?;
            let created_at =
                u64::try_from(now).map_err(|_| SoftchatError::InvalidAccountOperation)?;
            let rumor = self.identity.with_identity(|identity| {
                identity
                    .create_subject(SubjectDraft {
                        created_at,
                        participants: public_keys(&recipients)?,
                        subject,
                        emoji_tags: nostr_tags(emoji_tags)?,
                        icon,
                        extension_tags: Vec::new(),
                    })
                    .map(|view| view.rumor)
            })?;
            let operation = {
                let mut database = self.database()?;
                self.identity.with_identity(|identity| {
                    database.enqueue_product_rumor(
                        identity,
                        command_id,
                        payload_hash,
                        String::new(),
                        rumor,
                        recipients.clone(),
                        recipients,
                        Nip59EnvelopeKind::Durable,
                        now,
                        None,
                    )
                })?
            };
            stored_product_result(self, operation)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Messaging,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::Subject);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Toggle one reaction by the local account on a stored message.
    ///
    /// A committed command replays its original result without toggling again,
    /// including after later reactions or deletion of the parent message.
    /// Removal addresses all active local copies of the value, independently
    /// of the bounded message view. More than 256 matching copies returns
    /// `InvalidAccountOperation` without a partial deletion or revision change.
    ///
    /// # Errors
    ///
    /// Returns a stable command, message, reaction, authorization, or persistence error.
    pub fn toggle_message_reaction(
        &self,
        command_id: String,
        message_id: String,
        reaction: MessageReactionInput,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            NostrEventId::from_hex(&message_id)?;
            // Released commands hash the state before the toggle. Either historic
            // hash identifies the same caller intent, regardless of current state.
            let adding_hash = command_hash(&("reaction", &message_id, &reaction, false))?;
            let removing_hash = command_hash(&("reaction", &message_id, &reaction, true))?;
            let mut database = self.database()?;
            let committed = match database.product_operation_if_committed(&command_id, &adding_hash)
            {
                Err(SoftchatError::CommandConflict) => {
                    database.product_operation_if_committed(&command_id, &removing_hash)?
                }
                result => result?,
            };
            if let Some(operation) = committed {
                drop(database);
                return stored_product_result(self, operation);
            }
            let own_public_key = self.public_key()?;
            let parent = database
                .product_message(&message_id)?
                .filter(|message| !message.deleted)
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let conversation = database
                .product_conversation(&parent.conversation_id, &own_public_key)?
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let recipients = remote_members(&conversation, &own_public_key);
            let existing_ids = database.active_product_reaction_ids(
                &message_id,
                &own_public_key,
                &reaction.value,
            )?;
            let payload_hash = if !existing_ids.is_empty() {
                removing_hash
            } else {
                adding_hash
            };
            let created_at =
                u64::try_from(now).map_err(|_| SoftchatError::InvalidAccountOperation)?;
            let rumor = if !existing_ids.is_empty() {
                self.identity.with_identity(|identity| {
                    identity
                        .create_deletion(
                            public_keys(&recipients)?,
                            existing_ids
                                .iter()
                                .map(|id| NostrEventId::from_hex(id))
                                .collect::<Result<Vec<_>, _>>()?,
                            String::new(),
                            created_at,
                        )
                        .map(|view| view.rumor)
                })?
            } else {
                self.identity.with_identity(|identity| {
                    identity
                        .create_reaction(ReactionDraft {
                            participants: public_keys(&recipients)?,
                            parent_id: NostrEventId::from_hex(&parent.id)?,
                            parent_author: NostrPublicKey::from_hex(&parent.author_public_key)?,
                            parent_kind: NostrEventKind::PRIVATE_DIRECT_MESSAGE,
                            reaction: reaction.value,
                            custom_emoji_url: reaction.custom_emoji_url,
                            created_at,
                        })
                        .map(|view| view.rumor)
                })?
            };
            let operation = self.identity.with_identity(|identity| {
                database.enqueue_product_rumor(
                    identity,
                    command_id,
                    payload_hash,
                    message_id.clone(),
                    rumor,
                    recipients.clone(),
                    Vec::new(),
                    Nip59EnvelopeKind::Durable,
                    now,
                    None,
                )
            })?;
            drop(database);
            stored_product_result(self, operation)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Messaging,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::Reaction);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Replace text and emoji on one locally authored stored message.
    ///
    /// Replaying a committed command with the same payload returns its existing
    /// operation even if the original message has subsequently been deleted.
    /// A different payload for that command remains a command conflict.
    ///
    /// # Errors
    ///
    /// Returns a stable command, authorization, content, or persistence error.
    pub fn edit_message(
        &self,
        command_id: String,
        message_id: String,
        replacement: MessageContentInput,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            replacement.validate_edit()?;
            let target = NostrEventId::from_hex(&message_id)?;
            let payload_hash = command_hash(&("edit", &message_id, &replacement))?;
            let existing = {
                let database = self.database()?;
                database.product_operation_if_committed(&command_id, &payload_hash)?
            };
            if let Some(operation) = existing {
                return stored_product_result(self, operation);
            }
            let own_public_key = self.public_key()?;
            let original = self
                .database()?
                .product_message(&message_id)?
                .filter(|message| message.author_public_key == own_public_key && !message.deleted)
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            if replacement.text.is_empty() && original.attachments.is_empty() {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            let conversation = self
                .database()?
                .product_conversation(&original.conversation_id, &own_public_key)?
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let recipients = remote_members(&conversation, &own_public_key);
            let created_at =
                u64::try_from(now).map_err(|_| SoftchatError::InvalidAccountOperation)?;
            let rumor = self.identity.with_identity(|identity| {
                identity
                    .create_edit(
                        public_keys(&recipients)?,
                        target,
                        replacement.text,
                        nostr_tags(replacement.emoji_tags)?,
                        created_at,
                    )
                    .map(|view| view.rumor)
            })?;
            let operation = {
                let mut database = self.database()?;
                self.identity.with_identity(|identity| {
                    database.enqueue_product_rumor(
                        identity,
                        command_id,
                        payload_hash,
                        message_id.clone(),
                        rumor,
                        recipients,
                        Vec::new(),
                        Nip59EnvelopeKind::Durable,
                        now,
                        None,
                    )
                })?
            };
            stored_product_result(self, operation)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Messaging,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::Edit);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Delete one locally authored stored message.
    ///
    /// # Errors
    ///
    /// Returns a stable command, authorization, message, or persistence error.
    pub fn delete_message(
        &self,
        command_id: String,
        message_id: String,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            if message_id.len() != 64 {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            let target = NostrEventId::from_hex(&message_id)?;
            let payload_hash = command_hash(&("delete", &message_id))?;
            let existing = {
                let database = self.database()?;
                database.product_operation_if_committed(&command_id, &payload_hash)?
            };
            if let Some(operation) = existing {
                return stored_product_result(self, operation);
            }
            let own_public_key = self.public_key()?;
            let original = self
                .database()?
                .product_message(&message_id)?
                .filter(|message| message.author_public_key == own_public_key && !message.deleted)
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let conversation = self
                .database()?
                .product_conversation(&original.conversation_id, &own_public_key)?
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let recipients = remote_members(&conversation, &own_public_key);
            let created_at =
                u64::try_from(now).map_err(|_| SoftchatError::InvalidAccountOperation)?;
            let rumor = self.identity.with_identity(|identity| {
                identity
                    .create_deletion(
                        public_keys(&recipients)?,
                        vec![target],
                        String::new(),
                        created_at,
                    )
                    .map(|view| view.rumor)
            })?;
            let operation = {
                let mut database = self.database()?;
                self.identity.with_identity(|identity| {
                    database.enqueue_product_rumor(
                        identity,
                        command_id,
                        payload_hash,
                        message_id.clone(),
                        rumor,
                        recipients,
                        Vec::new(),
                        Nip59EnvelopeKind::Durable,
                        now,
                        None,
                    )
                })?
            };
            stored_product_result(self, operation)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Messaging,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::Deletion);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Delete a bounded set of locally authored messages from one conversation atomically.
    ///
    /// One idempotent command authors one deletion rumor containing every target. Either the
    /// complete command is committed and projected or no target is changed.
    ///
    /// # Errors
    ///
    /// Returns a stable command, bound, authorization, message, conversation, or persistence
    /// error.
    pub fn delete_messages(
        &self,
        command_id: String,
        message_ids: Vec<String>,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            if message_ids.is_empty() || message_ids.len() > crate::MAX_PRODUCT_QUERY_PAGE as usize
            {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            if message_ids.iter().any(|message_id| message_id.len() != 64) {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            let targets = message_ids
                .iter()
                .map(|message_id| NostrEventId::from_hex(message_id))
                .collect::<Result<Vec<_>, _>>()?;
            let mut distinct_ids = BTreeSet::new();
            if message_ids
                .iter()
                .any(|message_id| !distinct_ids.insert(message_id.as_str()))
            {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            let payload_hash = command_hash(&("delete-many", &message_ids))?;
            let existing = {
                let database = self.database()?;
                database.product_operation_if_committed(&command_id, &payload_hash)?
            };
            if let Some(operation) = existing {
                return stored_product_result(self, operation);
            }

            let own_public_key = self.public_key()?;
            let messages = message_ids
                .iter()
                .map(|message_id| {
                    self.database()?
                        .product_message(message_id)?
                        .filter(|message| {
                            message.author_public_key == own_public_key && !message.deleted
                        })
                        .ok_or(SoftchatError::InvalidAccountOperation)
                })
                .collect::<Result<Vec<_>, _>>()?;
            let conversation_id = messages
                .first()
                .map(|message| message.conversation_id.clone())
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            if messages
                .iter()
                .any(|message| message.conversation_id != conversation_id)
            {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            let conversation = self
                .database()?
                .product_conversation(&conversation_id, &own_public_key)?
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let recipients = remote_members(&conversation, &own_public_key);
            let created_at =
                u64::try_from(now).map_err(|_| SoftchatError::InvalidAccountOperation)?;
            let rumor = self.identity.with_identity(|identity| {
                identity
                    .create_deletion(
                        public_keys(&recipients)?,
                        targets,
                        String::new(),
                        created_at,
                    )
                    .map(|view| view.rumor)
            })?;
            let result_message_id = message_ids
                .first()
                .cloned()
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let operation = {
                let mut database = self.database()?;
                self.identity.with_identity(|identity| {
                    database.enqueue_product_rumor(
                        identity,
                        command_id,
                        payload_hash,
                        result_message_id,
                        rumor,
                        recipients,
                        Vec::new(),
                        Nip59EnvelopeKind::Durable,
                        now,
                        None,
                    )
                })?
            };
            stored_product_result(self, operation)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Messaging,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::Deletion);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Publish an ephemeral typing event when enabled by account settings.
    ///
    /// Returning `None` means no wire event was required.
    ///
    /// # Errors
    ///
    /// Returns a stable command, conversation, settings, or persistence error.
    pub fn typing_changed(
        &self,
        command_id: String,
        conversation_id: String,
        is_typing: bool,
        now: i64,
    ) -> Result<Option<ProductOperationResult>, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            if !is_typing {
                lock(&self.typing_authored).remove(&conversation_id);
                return Ok(None);
            }
            if !self
                .database()?
                .product_settings()?
                .typing_indicators_enabled
            {
                return Ok(None);
            }
            {
                let authored = lock(&self.typing_authored);
                if authored
                    .get(&conversation_id)
                    .is_some_and(|(last, prior_id)| {
                        prior_id != &command_id && now >= *last && now.saturating_sub(*last) < 3
                    })
                {
                    return Ok(None);
                }
            }
            let own_public_key = self.public_key()?;
            let conversation = self
                .database()?
                .product_conversation(&conversation_id, &own_public_key)?
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let recipients = remote_members(&conversation, &own_public_key);
            let payload_hash = command_hash(&("typing", &conversation_id, is_typing))?;
            let created_at =
                u64::try_from(now).map_err(|_| SoftchatError::InvalidAccountOperation)?;
            let rumor = self.identity.with_identity(|identity| {
                identity
                    .create_typing(public_keys(&recipients)?, created_at)
                    .map(|view| view.rumor)
            })?;
            let operation = {
                let mut database = self.database()?;
                self.identity.with_identity(|identity| {
                    database.enqueue_product_rumor(
                        identity,
                        command_id.clone(),
                        payload_hash,
                        String::new(),
                        rumor,
                        recipients,
                        Vec::new(),
                        Nip59EnvelopeKind::Ephemeral,
                        now,
                        None,
                    )
                })?
            };
            let result = stored_product_result(self, operation)?;
            lock(&self.typing_authored).insert(conversation_id, (now, command_id));
            Ok(Some(result))
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Messaging,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::Typing);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Set the local archived state for one conversation.
    ///
    /// # Errors
    ///
    /// Returns a stable conversation or persistence error.
    pub fn set_conversation_archived(
        &self,
        conversation_id: String,
        archived: bool,
        now: i64,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?
            .set_product_conversation_flag(&conversation_id, "archived", archived, now)
    }

    /// Set the local pinned state for one conversation.
    ///
    /// # Errors
    ///
    /// Returns a stable conversation or persistence error.
    pub fn set_conversation_pinned(
        &self,
        conversation_id: String,
        pinned: bool,
        now: i64,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?
            .set_product_conversation_flag(&conversation_id, "pinned", pinned, now)
    }

    /// Mark one conversation read through an exact stable message cursor.
    ///
    /// The cursor timestamp must match its stored message. State preview,
    /// signing and commit share one account lock so concurrent changes retain
    /// a complete cross-device snapshot.
    /// Committed commands replay before preparing another snapshot, preserving
    /// later read-state updates and the current account revision.
    ///
    /// # Errors
    ///
    /// Returns a stable conversation, timestamp, or persistence error.
    pub fn mark_conversation_read(
        &self,
        command_id: String,
        conversation_id: String,
        boundary: MessageCursor,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            validate_conversation_id(&conversation_id)?;
            NostrEventId::from_hex(&boundary.message_id)?;
            let boundary_created_at = u64::try_from(boundary.created_at)
                .map_err(|_| SoftchatError::InvalidAccountOperation)?;
            if boundary_created_at > crate::MAX_PORTABLE_TIMESTAMP_SECONDS {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            let payload_hash = command_hash(&("mark-read", &conversation_id, &boundary))?;
            let mut database = self.database()?;
            if let Some(operation) =
                database.product_operation_if_committed(&command_id, &payload_hash)?
            {
                drop(database);
                return stored_product_result(self, operation);
            }
            let preview = database.preview_product_read_state_update(
                &conversation_id,
                Some(boundary.clone()),
                false,
                now,
                &command_id,
            )?;
            let authored_at = database.next_app_data_event_timestamp("read-state", now)?;
            let created_at =
                u64::try_from(authored_at).map_err(|_| SoftchatError::InvalidAccountOperation)?;
            let event = self.identity.with_identity(|identity| {
                identity
                    .create_app_data_sync(
                        created_at,
                        AppDataContext::ReadState,
                        &preview.canonical_json,
                    )
                    .map(|view| view.event)
            })?;
            let operation = database.enqueue_product_signed_with_read_state(
                command_id,
                payload_hash,
                vec![event],
                preview.state,
                preview.canonical_json,
                now,
            )?;
            drop(database);
            stored_product_result(self, operation)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Synchronization,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::ReadState);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Mark one conversation unread without destroying its prior read boundary.
    ///
    /// State preview, signing and commit share one account lock.
    /// Committed commands replay without preparing or changing current state.
    ///
    /// # Errors
    ///
    /// Returns a stable conversation or persistence error.
    pub fn mark_conversation_unread(
        &self,
        command_id: String,
        conversation_id: String,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            validate_conversation_id(&conversation_id)?;
            let payload_hash = command_hash(&("mark-unread", &conversation_id))?;
            let mut database = self.database()?;
            if let Some(operation) =
                database.product_operation_if_committed(&command_id, &payload_hash)?
            {
                drop(database);
                return stored_product_result(self, operation);
            }
            let preview = database.preview_product_read_state_update(
                &conversation_id,
                None,
                true,
                now,
                &command_id,
            )?;
            let authored_at = database.next_app_data_event_timestamp("read-state", now)?;
            let created_at =
                u64::try_from(authored_at).map_err(|_| SoftchatError::InvalidAccountOperation)?;
            let event = self.identity.with_identity(|identity| {
                identity
                    .create_app_data_sync(
                        created_at,
                        AppDataContext::ReadState,
                        &preview.canonical_json,
                    )
                    .map(|view| view.event)
            })?;
            let operation = database.enqueue_product_signed_with_read_state(
                command_id,
                payload_hash,
                vec![event],
                preview.state,
                preview.canonical_json,
                now,
            )?;
            drop(database);
            stored_product_result(self, operation)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Synchronization,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::ReadState);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Return the complete effective cross-device read-state snapshot.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn account_read_state(&self) -> Result<Vec<crate::AccountReadState>, SoftchatError> {
        self.database()?.product_read_states()
    }

    /// Save or clear one portable conversation draft.
    ///
    /// # Errors
    ///
    /// Returns a stable draft, conversation, or persistence error.
    pub fn save_conversation_draft(
        &self,
        conversation_id: String,
        draft: Option<AccountDraft>,
        now: i64,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?
            .save_product_draft(&conversation_id, draft, now)
    }

    /// Return the complete versioned account settings.
    ///
    /// # Errors
    ///
    /// Returns a stable settings or persistence error.
    pub fn account_settings(&self) -> Result<AccountSettings, SoftchatError> {
        self.database()?.product_settings()
    }

    /// Apply one JSON merge patch while retaining unknown settings fields.
    ///
    /// The account lock spans preview, signing and commit so concurrent patches
    /// preserve each other's independent fields.
    ///
    /// Replaying a committed bounded patch returns current settings without
    /// merging it again, creating another operation or advancing the revision.
    ///
    /// The authored Android facade exposes typed patch values; JSON remains a
    /// private FFI representation.
    ///
    /// # Errors
    ///
    /// Returns a stable settings, timestamp, or persistence error.
    pub fn apply_account_settings_patch(
        &self,
        command_id: String,
        patch_json: String,
        now: i64,
    ) -> Result<AccountSettingsMutation, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            if patch_json.len() > crate::MAX_ACCOUNT_SETTINGS_BYTES {
                return Err(SoftchatError::InvalidAccountSettings);
            }
            let canonical_patch: serde_json::Value = serde_json::from_str(&patch_json)
                .map_err(|_| SoftchatError::InvalidAccountSettings)?;
            if !canonical_patch.is_object() {
                return Err(SoftchatError::InvalidAccountSettings);
            }
            let payload_hash = command_hash(&("settings", canonical_patch))?;
            let mut database = self.database()?;
            let operation = if let Some(committed) =
                database.product_operation_if_committed(&command_id, &payload_hash)?
            {
                committed
            } else {
                let mutation = database.preview_product_settings_patch(&patch_json)?;
                let authored_at = database.next_app_data_event_timestamp("app-settings", now)?;
                let wire_json =
                    crate::account_product::settings_wire_json(&mutation.canonical_json)?;
                let created_at = u64::try_from(authored_at)
                    .map_err(|_| SoftchatError::InvalidAccountOperation)?;
                let event = self.identity.with_identity(|identity| {
                    identity
                        .create_app_data_sync(created_at, AppDataContext::AppSettings, &wire_json)
                        .map(|view| view.event)
                })?;
                database.enqueue_product_signed_with_settings(
                    command_id,
                    payload_hash,
                    vec![event],
                    mutation.settings.schema_version,
                    mutation.canonical_json.clone(),
                    wire_json,
                    now,
                )?
            };
            let settings = database.product_settings()?;
            let canonical_json = database.product_settings_canonical_json()?;
            Ok(AccountSettingsMutation {
                settings,
                canonical_json,
                revision: operation.operation.revision,
            })
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Synchronization,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::AppSettings);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Return one effective authenticated/local profile.
    ///
    /// # Errors
    ///
    /// Returns a stable key, account, or persistence error.
    pub fn profile(&self, public_key: String) -> Result<AccountProfile, SoftchatError> {
        self.database()?.product_profile(&public_key)
    }

    /// Resolve effective authenticated/local profiles for exact keys in input order.
    ///
    /// At most 512 keys are accepted. Unknown keys return empty profile values
    /// carrying their canonical public key, so consumers never need per-row
    /// fallback queries or a second profile cache.
    ///
    /// # Errors
    ///
    /// Returns a stable key, bound, account, or persistence error.
    pub fn profiles(&self, public_keys: Vec<String>) -> Result<Vec<AccountProfile>, SoftchatError> {
        self.database()?.product_profiles_for_keys(&public_keys)
    }

    /// Return one bounded stable profile page.
    ///
    /// # Errors
    ///
    /// Returns a stable bound, account, or persistence error.
    pub fn profile_page(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<AccountProfile>, SoftchatError> {
        self.database()?.product_profiles(limit, offset)
    }

    /// Return the effective private follow list authored by this account.
    ///
    /// # Errors
    ///
    /// Returns a stable account, contact-list, or persistence error.
    pub fn follows(&self) -> Result<Vec<Contact>, SoftchatError> {
        let own_public_key = self.public_key()?;
        self.database()?.product_follows(&own_public_key)
    }

    /// Search effective profiles by public key, authenticated metadata, or local label.
    /// Matching ignores ASCII case; other Unicode characters remain literal.
    ///
    /// # Errors
    ///
    /// Returns a stable search, bound, account, or persistence error.
    pub fn search_profiles(
        &self,
        query: String,
        limit: u32,
    ) -> Result<Vec<AccountProfile>, SoftchatError> {
        self.database()?.search_product_profiles(&query, limit)
    }

    /// Merge and publish a private profile patch.
    ///
    /// Missing fields remain unchanged, JSON `null` clears a field, and a
    /// concrete value replaces it. Authenticated fields unknown to this SDK
    /// version are retained automatically.
    ///
    /// # Errors
    ///
    /// Returns a stable command, profile, recipient, relay, or persistence error.
    pub fn update_profile(
        &self,
        command_id: String,
        patch_json: String,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            let own_public_key = self.public_key()?;
            let mut database = self.database()?;
            let current = database.authenticated_product_profile(&own_public_key)?;
            let authored_at = if current.updated_at > 0 && current.updated_at >= now {
                current
                    .updated_at
                    .checked_add(1)
                    .ok_or(SoftchatError::InvalidAccountOperation)?
            } else {
                now
            };
            let created_at =
                u64::try_from(authored_at).map_err(|_| SoftchatError::InvalidAccountOperation)?;
            let (profile, canonical_patch) = profile_update_from_patch(&current, &patch_json)?;
            let payload_hash = command_hash(&("profile_patch", canonical_patch))?;
            let rumor = self.identity.with_identity(|identity| {
                identity
                    .create_user_metadata(
                        created_at,
                        profile.name,
                        profile.display_name,
                        profile.about,
                        profile.picture,
                        profile.website,
                        profile.banner,
                        profile.bot,
                        &profile.unknown_json,
                    )
                    .map(|view| view.rumor)
            })?;
            let recipients = all_known_public_keys(&database)?
                .into_iter()
                .filter(|key| key != &own_public_key)
                .collect::<Vec<_>>();
            let operation = self.identity.with_identity(|identity| {
                database.enqueue_product_rumor(
                    identity,
                    command_id,
                    payload_hash,
                    String::new(),
                    rumor,
                    recipients,
                    Vec::new(),
                    Nip59EnvelopeKind::Durable,
                    now,
                    None,
                )
            })?;
            stored_product_result(self, operation)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Profile,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::UserMetadata);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Publish a complete private follow-list replacement.
    ///
    /// # Errors
    ///
    /// Returns a stable command, contact, relay, or persistence error.
    pub fn replace_follows(
        &self,
        command_id: String,
        contacts: Vec<Contact>,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            let mut database = self.database()?;
            let operation =
                self.enqueue_follow_replacement(&mut database, command_id, contacts, now)?;
            drop(database);
            stored_product_result(self, operation)
        })();
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Profile,
                "profile.follows.replace",
                "Follow list command failed",
            ),
            &result,
        );
        result
    }

    /// Add or replace one entry in the private follow list atomically.
    ///
    /// Rust reads the current authenticated list, applies the keyed update,
    /// authors its self-copy, and commits the outgoing operation while holding
    /// one account-database lock. Concurrent callers cannot lose updates.
    ///
    /// # Errors
    ///
    /// Returns a stable command, contact, relay, or persistence error.
    pub fn follow(
        &self,
        command_id: String,
        contact: Contact,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            let own_public_key = self.public_key()?;
            let mut database = self.database()?;
            let mut contacts = database.product_follows(&own_public_key)?;
            contacts.retain(|value| value.public_key != contact.public_key);
            contacts.push(contact);
            contacts.sort_by(|left, right| left.public_key.cmp(&right.public_key));
            let operation =
                self.enqueue_follow_replacement(&mut database, command_id, contacts, now)?;
            drop(database);
            stored_product_result(self, operation)
        })();
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Profile,
                "profile.follow",
                "Follow list command failed",
            ),
            &result,
        );
        result
    }

    /// Remove one public key from the private follow list atomically.
    ///
    /// Removing an absent key is a valid idempotent replacement. Reusing the
    /// same command ID with a different resulting list is still rejected.
    ///
    /// # Errors
    ///
    /// Returns a stable command, key, relay, or persistence error.
    pub fn unfollow(
        &self,
        command_id: String,
        public_key: String,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            NostrPublicKey::from_hex(&public_key)?;
            let own_public_key = self.public_key()?;
            let mut database = self.database()?;
            let mut contacts = database.product_follows(&own_public_key)?;
            contacts.retain(|value| value.public_key != public_key);
            let operation =
                self.enqueue_follow_replacement(&mut database, command_id, contacts, now)?;
            drop(database);
            stored_product_result(self, operation)
        })();
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Profile,
                "profile.unfollow",
                "Follow list command failed",
            ),
            &result,
        );
        result
    }

    /// Register an FCM or APNs token through its released NIP-78 wire profile.
    ///
    /// # Errors
    ///
    /// Returns a stable command, token, relay, signing, or persistence error.
    pub fn register_push_token(
        &self,
        command_id: String,
        platform: PushPlatform,
        token: String,
        installation_id: String,
        now: i64,
    ) -> Result<ProductOperationResult, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            if token.is_empty()
                || token.len() > 16 * 1024
                || installation_id.is_empty()
                || installation_id.len() > 512
            {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            let (identifier, token_key) = match platform {
                PushPlatform::Fcm => ("fcmToken-v2", "fcmToken"),
                PushPlatform::Apns => ("apnsToken-v2", "apnsToken"),
            };
            let content = serde_json::to_string(&serde_json::json!({
                (token_key): token,
                "deviceIdentifier": installation_id,
            }))
            .map_err(|_| SoftchatError::InternalFailure)?;
            let payload_hash = command_hash(&("push", platform, &content))?;
            let created_at =
                u64::try_from(now).map_err(|_| SoftchatError::InvalidAccountOperation)?;
            let event = self.identity.with_identity(|identity| {
                identity
                    .create_application_data(created_at, identifier.to_owned(), content)
                    .map(|view| view.event)
            })?;
            let operation = self.database()?.enqueue_product_signed(
                command_id,
                payload_hash,
                String::new(),
                vec![event],
                now,
                AccountLogSubject::PushRegistration,
            )?;
            stored_product_result(self, operation)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Lifecycle,
            "message.command.failed",
            "Account command failed",
        );
        diagnostic.subject = Some(AccountLogSubject::PushRegistration);
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Return a bounded account or conversation media gallery.
    ///
    /// `limit` is a soft attachment count from 1 through 100. The final
    /// message's attachment group is returned in full so a message cursor
    /// cannot skip its remainder. At most `limit + 31` items are returned.
    /// Deleted messages are excluded.
    ///
    /// # Errors
    ///
    /// Returns a stable cursor, bound, account, or persistence error.
    pub fn media_gallery(
        &self,
        conversation_id: String,
        before: Option<MessageCursor>,
        limit: u32,
    ) -> Result<Vec<AccountMediaItem>, SoftchatError> {
        self.database()?.product_media_gallery(
            (!conversation_id.is_empty()).then_some(conversation_id),
            before,
            limit,
        )
    }

    /// Persist one attachment-dependent message before Android starts file or HTTP work.
    ///
    /// The host persists its exact portable composer snapshot before passing it
    /// in `draft_to_consume`. When present,
    /// preparation clears it in the same transaction only while the stored
    /// semantic draft still matches. A newer autosave and every idempotent
    /// replay are preserved.
    ///
    /// # Errors
    ///
    /// Returns a stable command, conversation, content, media, or persistence error.
    #[allow(clippy::too_many_arguments)]
    pub fn prepare_media_message(
        &self,
        command_id: String,
        conversation_id: String,
        content: MessageContentInput,
        reply_to_message_id: String,
        sources: Vec<crate::MediaPreparationInput>,
        draft_to_consume: Option<AccountDraft>,
        now: i64,
    ) -> Result<crate::PendingMediaMessage, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            if sources.is_empty() || sources.len() > crate::MAX_PRODUCT_DRAFT_ATTACHMENTS {
                return Err(SoftchatError::InvalidMediaOperation);
            }
            content
                .validate_bounds()
                .map_err(|_| SoftchatError::InvalidMediaOperation)?;
            if !content.attachments.is_empty() {
                return Err(SoftchatError::InvalidMediaOperation);
            }
            let probe_content = MessageContentInput {
                text: content.text.clone(),
                attachments: crate::account_product::reserved_product_attachments(sources.len())?,
                emoji_tags: content.emoji_tags.clone(),
            };
            author_product_chat_message(
                self,
                &conversation_id,
                probe_content,
                &reply_to_message_id,
                now,
            )
            .map_err(|_| SoftchatError::InvalidMediaOperation)?;
            let draft_match = draft_to_consume
                .map(product_draft_match)
                .transpose()
                .map_err(|_| SoftchatError::InvalidMediaOperation)?;
            let payload_hash = media_message_command_hash(
                &conversation_id,
                &content,
                &reply_to_message_id,
                &sources,
                draft_match.as_ref(),
            )?;
            self.database()?.prepare_product_media_message(
                crate::account_product::ProductMediaMessagePreparation {
                    command_id: &command_id,
                    conversation_id: &conversation_id,
                    content: &content,
                    reply_to_message_id: &reply_to_message_id,
                    sources,
                    command_hash: &payload_hash,
                    draft_match,
                    now,
                },
            )
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Media,
            "media.command.failed",
            "Media command failed",
        );
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Return one durable attachment-dependent message intent.
    ///
    /// # Errors
    ///
    /// Returns a stable command, media, or persistence error.
    pub fn pending_media_message(
        &self,
        command_id: String,
    ) -> Result<Option<crate::PendingMediaMessage>, SoftchatError> {
        self.database()?.product_pending_media_message(&command_id)
    }

    /// Cancel one unpublished attachment-dependent message and its unfinished transfers.
    ///
    /// Cancellation is idempotent and may retire a permanently failed message.
    /// Completed attachment operations remain immutable, while every unfinished
    /// dependency loses its lease and becomes cancelled in the same transaction.
    ///
    /// # Errors
    ///
    /// Returns a stable command, media, timestamp, account, or persistence error.
    pub fn cancel_pending_media_message(
        &self,
        command_id: String,
        now: i64,
    ) -> Result<crate::PendingMediaMessage, SoftchatError> {
        self.database()?
            .cancel_product_pending_media_message(&command_id, now)
    }

    /// Return a bounded newest-update-first page of attachment-dependent messages.
    ///
    /// # Errors
    ///
    /// Returns a stable bound, media, or persistence error.
    pub fn pending_media_messages(
        &self,
        limit: u32,
    ) -> Result<Vec<crate::PendingMediaMessage>, SoftchatError> {
        self.database()?.product_pending_media_messages(limit)
    }

    /// Persist one public asset replacement before Android starts file or HTTP work.
    ///
    /// The upload is intentionally public because profile and subject icon
    /// readers receive only a URL. Rust retains the dependent profile or
    /// subject command and completes it idempotently after the upload.
    ///
    /// [Guide](https://softchat-sdk.jackl.dev/android-account-runtime/#durable-public-asset-replacement)
    ///
    /// # Errors
    ///
    /// Returns a stable command, target, conversation, media, or persistence error.
    pub fn prepare_asset_replacement(
        &self,
        command_id: String,
        target: AssetReplacementTarget,
        conversation_id: String,
        source_fingerprint: String,
        now: i64,
    ) -> Result<PendingAssetReplacement, SoftchatError> {
        let result = (|| {
            validate_command_id(&command_id)?;
            let payload_hash = command_hash(&(
                "asset-replacement",
                target,
                &conversation_id,
                &source_fingerprint,
            ))?;
            self.database()?.prepare_product_asset_replacement(
                crate::account_product::ProductAssetReplacementPreparation {
                    command_id: &command_id,
                    target,
                    conversation_id: &conversation_id,
                    source_fingerprint: &source_fingerprint,
                    command_hash: &payload_hash,
                    now,
                },
            )
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Media,
            "media.command.failed",
            "Media command failed",
        );
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Return one durable upload-dependent asset replacement.
    ///
    /// [Guide](https://softchat-sdk.jackl.dev/android-account-runtime/#durable-public-asset-replacement)
    ///
    /// # Errors
    ///
    /// Returns a stable command, media, or persistence error.
    pub fn pending_asset_replacement(
        &self,
        command_id: String,
    ) -> Result<Option<PendingAssetReplacement>, SoftchatError> {
        self.database()?
            .product_pending_asset_replacement(&command_id)
    }

    /// Return a bounded newest-update-first page of asset replacements.
    ///
    /// [Guide](https://softchat-sdk.jackl.dev/android-account-runtime/#durable-public-asset-replacement)
    ///
    /// # Errors
    ///
    /// Returns a stable bound, media, or persistence error.
    pub fn pending_asset_replacements(
        &self,
        limit: u32,
    ) -> Result<Vec<PendingAssetReplacement>, SoftchatError> {
        self.database()?.product_pending_asset_replacements(limit)
    }

    /// Prepare one idempotent logical media operation.
    ///
    /// # Errors
    ///
    /// Returns a stable command, media, conversation, or persistence error.
    pub fn prepare_media_operation(
        &self,
        command_id: String,
        conversation_id: String,
        direction: String,
        source_fingerprint: String,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        let result = (|| {
            self.database()?.prepare_product_media(
                &command_id,
                (!conversation_id.is_empty()).then_some(conversation_id),
                &direction,
                &source_fingerprint,
                now,
            )
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Media,
            "media.command.failed",
            "Media command failed",
        );
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Prepare or return the stable download for one authenticated attachment.
    ///
    /// Rust resolves the message, validates the attachment URL, and derives
    /// the idempotency key and source fingerprint. Android receives only the
    /// logical operation and performs the file/HTTP effect.
    ///
    /// # Errors
    ///
    /// Returns a stable message, media, or persistence error.
    pub fn prepare_media_download(
        &self,
        message_id: String,
        attachment_url: String,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        let result = (|| {
            self.database()?
                .prepare_product_media_download(&message_id, &attachment_url, now)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Media,
            "media.command.failed",
            "Media command failed",
        );
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Prepare or return the stable download for the current authenticated conversation icon.
    ///
    /// Rust resolves the effective subject event and retains its complete `imeta` integrity
    /// metadata. Public icons require the downloaded-object SHA-256 (`x`); encrypted icons may
    /// rely on final AEAD authentication. The operation is bound to that exact subject version,
    /// so completion is rejected only when a later authenticated subject
    /// explicitly changes the icon. Text- or emoji-only subjects inherit it.
    ///
    /// # Errors
    ///
    /// Returns a stable conversation, media, or persistence error.
    pub fn prepare_conversation_icon_download(
        &self,
        conversation_id: String,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        let result = (|| {
            self.database()?
                .prepare_product_conversation_icon_download(&conversation_id, now)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Media,
            "media.command.failed",
            "Media command failed",
        );
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Restart an authenticated download after cache eviction or user retry.
    ///
    /// Completed uploads and non-download operations are rejected. Running,
    /// prepared, and retryable downloads are returned without modification.
    /// A terminal conversation-icon download whose authenticated source was
    /// superseded remains terminal; callers prepare the current icon instead.
    ///
    /// # Errors
    ///
    /// Returns a stable media or persistence error.
    pub fn restart_media_download(
        &self,
        operation_id: String,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        let result = (|| {
            self.database()?
                .restart_product_media_download(&operation_id, now)
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Media,
            "media.command.failed",
            "Media command failed",
        );
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Commit final authenticated attachment metadata.
    ///
    /// An exact retry of an already committed completion returns the persisted
    /// result before freshness checks. Freshness gates only the first running
    /// conversation-icon transition, so a lost success response remains safe
    /// to retry after a later subject changes the icon.
    ///
    /// # Errors
    ///
    /// Returns [`SoftchatError::InvalidAttachmentMetadata`] when `attachment`
    /// is malformed, [`SoftchatError::InvalidMediaOperation`] when the
    /// operation or lease is stale, or a stable persistence error.
    pub fn complete_media_operation(
        &self,
        operation_id: String,
        lease_id: String,
        attachment: AttachmentMetadata,
        byte_count: u64,
        now: i64,
    ) -> Result<crate::MediaCompletionResult, SoftchatError> {
        let completion = self.database()?.complete_product_media(
            &operation_id,
            &lease_id,
            attachment,
            byte_count,
            now,
        );
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Media,
                "media.completion.failed",
                "Media completion rejected",
            ),
            &completion,
        );
        let operation = completion?;
        let publication = (|| {
            let pending_message_command = self
                .database()?
                .pending_media_command_for_operation(&operation_id)?;
            let pending_asset_command = self
                .database()?
                .pending_asset_command_for_operation(&operation_id)?;
            let completed_operation = pending_message_command
                .map(|command_id| self.finalize_pending_media_message(&command_id, now))
                .transpose()?
                .flatten()
                .or(pending_asset_command
                    .map(|command_id| self.finalize_pending_asset_replacement(&command_id, now))
                    .transpose()?
                    .flatten());
            Ok(completed_operation)
        })();
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Media,
                "media.publication.deferred",
                "Media committed; dependent publication deferred",
            ),
            &publication,
        );
        let completed_operation = publication?;
        Ok(crate::MediaCompletionResult {
            operation,
            completed_operation,
        })
    }

    /// Recover every ready attachment-dependent message after process death.
    ///
    /// # Errors
    ///
    /// Returns a stable media, messaging, relay, or persistence error.
    pub fn recover_pending_media_messages(
        &self,
        limit: u32,
        now: i64,
    ) -> Result<Vec<ProductOperationResult>, SoftchatError> {
        let result = (|| {
            let commands = self.database()?.ready_pending_media_commands(limit)?;
            if !commands.is_empty() {
                let mut event = Diagnostic::new(
                    AccountLogCategory::Recovery,
                    "recovery.started",
                    "Pending publication recovery started",
                );
                event.source = Some(AccountLogSource::Recovery);
                event.summarize = true;
                self.logs.emit(event);
            }
            let mut published = Vec::new();
            for command_id in commands {
                if let Some(result) = self.finalize_pending_media_message(&command_id, now)? {
                    let mut diagnostic = Diagnostic::new(
                        AccountLogCategory::Recovery,
                        "recovery.completed",
                        "Pending publication recovered",
                    )
                    .count(
                        crate::account_diagnostics::AccountLogCounterKind::RecoveredItems,
                        1,
                    );
                    diagnostic.source = Some(AccountLogSource::Recovery);
                    diagnostic.summarize = true;
                    self.logs.emit(diagnostic);
                    published.push(result);
                }
            }
            Ok(published)
        })();
        let mut event = Diagnostic::new(
            AccountLogCategory::Recovery,
            "recovery.failed",
            "Pending publication recovery failed",
        );
        event.source = Some(AccountLogSource::Recovery);
        self.logs.failed(event, &result);
        result
    }

    /// Recover every ready upload-dependent asset replacement after process death.
    ///
    /// [Guide](https://softchat-sdk.jackl.dev/android-account-runtime/#durable-public-asset-replacement)
    ///
    /// # Errors
    ///
    /// Returns a stable media, profile, subject, relay, or persistence error.
    pub fn recover_pending_asset_replacements(
        &self,
        limit: u32,
        now: i64,
    ) -> Result<Vec<ProductOperationResult>, SoftchatError> {
        let result = (|| {
            let commands = self.database()?.ready_pending_asset_commands(limit)?;
            if !commands.is_empty() {
                let mut event = Diagnostic::new(
                    AccountLogCategory::Recovery,
                    "recovery.started",
                    "Pending publication recovery started",
                );
                event.source = Some(AccountLogSource::Recovery);
                event.summarize = true;
                self.logs.emit(event);
            }
            let mut published = Vec::new();
            for command_id in commands {
                if let Some(result) = self.finalize_pending_asset_replacement(&command_id, now)? {
                    let mut diagnostic = Diagnostic::new(
                        AccountLogCategory::Recovery,
                        "recovery.completed",
                        "Pending publication recovered",
                    )
                    .count(
                        crate::account_diagnostics::AccountLogCounterKind::RecoveredItems,
                        1,
                    );
                    diagnostic.source = Some(AccountLogSource::Recovery);
                    diagnostic.summarize = true;
                    self.logs.emit(diagnostic);
                    published.push(result);
                }
            }
            Ok(published)
        })();
        let mut event = Diagnostic::new(
            AccountLogCategory::Recovery,
            "recovery.failed",
            "Pending publication recovery failed",
        );
        event.source = Some(AccountLogSource::Recovery);
        self.logs.failed(event, &result);
        result
    }

    /// Claim one exact recoverable media operation with a bounded lease.
    ///
    /// Expired leases are recovered transactionally before selection.
    ///
    /// # Errors
    ///
    /// Returns a stable media-operation or persistence error.
    pub fn claim_media_operation(
        &self,
        operation_id: String,
        lease_id: String,
        now: i64,
        lease_duration_seconds: u32,
    ) -> Result<Option<AccountMediaLease>, SoftchatError> {
        let result = (|| {
            self.database()?.claim_product_media(
                &operation_id,
                &lease_id,
                now,
                lease_duration_seconds,
            )
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Media,
            "media.command.failed",
            "Media command failed",
        );
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Finish one matching media lease with a redacted failure category.
    ///
    /// # Errors
    ///
    /// Returns a stable media-operation or persistence error.
    pub fn finish_media_operation_lease(
        &self,
        operation_id: String,
        lease_id: String,
        retryable: bool,
        error_category: String,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        let result = (|| {
            self.database()?.finish_product_media_lease(
                &operation_id,
                &lease_id,
                retryable,
                &error_category,
                now,
            )
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Media,
            "media.command.failed",
            "Media command failed",
        );
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Create a signed NIP-98 upload plan from a streaming SHA-256.
    ///
    /// Android executes the exact HTTP request and must mint a new plan after
    /// any redirect.
    ///
    /// # Errors
    ///
    /// Returns a stable HTTP, hash, timestamp, or account lifecycle error.
    pub fn create_media_upload_authorization(
        &self,
        method: String,
        url: String,
        payload_sha256: String,
        now: i64,
    ) -> Result<FfiHttpAuthorizationPlan, SoftchatError> {
        let result = (|| {
            let created_at =
                u64::try_from(now).map_err(|_| SoftchatError::InvalidHttpAuthorization)?;
            self.identity.with_identity(|identity| {
                crate::create_nip98_authorization_for_payload_hash(
                    identity,
                    &method,
                    &url,
                    &payload_sha256,
                    created_at,
                )
                .map(FfiHttpAuthorizationPlan::from)
            })
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Media,
            "media.authorization.failed",
            "Upload authorization failed",
        );
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        if result.is_ok() {
            self.logs.emit(
                Diagnostic::new(
                    AccountLogCategory::Media,
                    "media.authorization.prepared",
                    "Upload authorization prepared",
                )
                .debug(),
            );
        }
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Apply one retry/failure/cancel transition to logical media state.
    ///
    /// # Errors
    ///
    /// Returns a stable media transition or persistence error.
    pub fn transition_media_operation(
        &self,
        operation_id: String,
        transition: String,
        error_category: String,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        let result = (|| {
            self.database()?.transition_product_media(
                &operation_id,
                &transition,
                &error_category,
                now,
            )
        })();
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Media,
            "media.command.failed",
            "Media command failed",
        );
        diagnostic.source = Some(AccountLogSource::LocalCommand);
        self.logs.failed(diagnostic, &result);
        result
    }

    /// Return existing downloads for the current attachments of one live message, in attachment order.
    ///
    /// This read creates no operation, changes no revision and schedules no work. Unknown or deleted
    /// messages return an empty result; unrelated recent transfers cannot hide a cached attachment.
    ///
    /// # Errors
    ///
    /// Returns a stable ID, account, or persistence error.
    pub fn message_downloads(
        &self,
        message_id: String,
    ) -> Result<Vec<AccountMediaOperation>, SoftchatError> {
        self.database()?.product_message_downloads(&message_id)
    }

    /// Return a bounded newest-update-first media-operation page.
    ///
    /// # Errors
    ///
    /// Returns a stable bound or persistence error.
    pub fn media_operations(
        &self,
        limit: u32,
    ) -> Result<Vec<AccountMediaOperation>, SoftchatError> {
        self.database()?.product_media_operations(limit)
    }

    /// Return existing logical media operations for one bounded exact ID set.
    ///
    /// Results preserve caller order. Missing IDs are omitted, while duplicate,
    /// malformed, or oversized input fails before any row is returned.
    ///
    /// # Errors
    ///
    /// Returns a stable bound, media-operation, or persistence error.
    pub fn media_operations_by_ids(
        &self,
        operation_ids: Vec<String>,
    ) -> Result<Vec<AccountMediaOperation>, SoftchatError> {
        self.database()?
            .product_media_operations_by_ids(&operation_ids)
    }

    /// Return one stable ID-ordered page of non-terminal media operations.
    ///
    /// Platform executors must follow `next_cursor` to exhaustion during startup recovery instead
    /// of assuming that a newest-first UI page contains every durable operation.
    ///
    /// # Errors
    ///
    /// Returns a stable cursor, bound, media-operation, or persistence error.
    pub fn recoverable_media_operations(
        &self,
        cursor: Option<String>,
        limit: u32,
    ) -> Result<AccountMediaOperationPage, SoftchatError> {
        self.database()?
            .product_recoverable_media_operations(cursor.as_deref(), limit)
    }

    /// Return one exact logical media operation.
    ///
    /// # Errors
    ///
    /// Returns a stable media-operation or persistence error.
    pub fn media_operation(
        &self,
        operation_id: String,
    ) -> Result<Option<AccountMediaOperation>, SoftchatError> {
        self.database()?.product_media_operation(&operation_id)
    }

    fn finalize_pending_media_message(
        &self,
        command_id: &str,
        now: i64,
    ) -> Result<Option<ProductOperationResult>, SoftchatError> {
        if !self
            .database()?
            .claim_pending_media_publication(command_id)?
        {
            return Ok(None);
        }
        let pending = self
            .database()?
            .product_pending_media_message(command_id)?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        match pending.state {
            crate::PendingMediaMessageState::Published
            | crate::PendingMediaMessageState::WaitingForUploads
            | crate::PendingMediaMessageState::Failed
            | crate::PendingMediaMessageState::Cancelled => return Ok(None),
            crate::PendingMediaMessageState::ReadyToPublish => {}
        }
        let attachments = pending
            .attachments
            .iter()
            .map(|operation| {
                if operation.state != crate::AccountMediaOperationState::Completed {
                    return Err(SoftchatError::InvalidMediaOperation);
                }
                operation
                    .attachment
                    .clone()
                    .ok_or(SoftchatError::InvalidMediaOperation)
            })
            .collect::<Result<Vec<_>, _>>()?;
        let result = self.send_message(
            pending.command_id.clone(),
            pending.conversation_id,
            MessageContentInput {
                text: pending.text,
                attachments,
                emoji_tags: pending.emoji_tags,
            },
            pending.reply_to_message_id,
            None,
            now,
        )?;
        let message_id = result
            .message
            .as_ref()
            .map(|message| message.id.clone())
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        self.database()?
            .mark_pending_media_published(command_id, &message_id, now)?;
        Ok(Some(result))
    }

    fn finalize_pending_asset_replacement(
        &self,
        command_id: &str,
        now: i64,
    ) -> Result<Option<ProductOperationResult>, SoftchatError> {
        let pending = self
            .database()?
            .product_pending_asset_replacement(command_id)?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        match pending.state {
            crate::PendingAssetReplacementState::Published
            | crate::PendingAssetReplacementState::WaitingForUpload
            | crate::PendingAssetReplacementState::Failed
            | crate::PendingAssetReplacementState::Cancelled => return Ok(None),
            crate::PendingAssetReplacementState::ReadyToPublish => {}
        }
        let attachment = pending
            .operation
            .attachment
            .clone()
            .filter(|_| {
                pending.operation.state == crate::AccountMediaOperationState::Completed
                    && pending.operation.protection == crate::AccountMediaProtection::Public
            })
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        if attachment.encryption.is_some() {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let result = match pending.target {
            AssetReplacementTarget::ProfilePicture => self.update_profile(
                pending.command_id.clone(),
                serde_json::json!({ "picture": attachment.url }).to_string(),
                now,
            )?,
            AssetReplacementTarget::ProfileBanner => self.update_profile(
                pending.command_id.clone(),
                serde_json::json!({ "banner": attachment.url }).to_string(),
                now,
            )?,
            AssetReplacementTarget::ConversationIcon => {
                let (subject, emoji_tags) = self
                    .database()?
                    .pending_asset_subject(&pending.command_id)?;
                self.set_conversation_subject(
                    pending.command_id.clone(),
                    pending.conversation_id,
                    subject,
                    Some(attachment),
                    emoji_tags,
                    now,
                )?
            }
        };
        self.database()?.mark_pending_asset_published(
            command_id,
            &result.operation.operation_id,
            now,
        )?;
        Ok(Some(result))
    }

    /// Return schema, revision, and WAL information.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence error if metadata cannot be read.
    pub fn database_info(&self) -> Result<AccountDatabaseInfo, SoftchatError> {
        self.database()?.info()
    }

    /// Return redacted bounded account health for support and recovery UI.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence error.
    pub fn account_diagnostics(&self) -> Result<AccountDiagnostics, SoftchatError> {
        self.database()?.diagnostics()
    }

    /// Forget durable synchronization checkpoints while retaining event truth.
    ///
    /// The active platform transport must be restarted after this call so its
    /// next Rust-owned synchronization begins from complete retained history.
    ///
    /// # Errors
    ///
    /// Returns a stable sync or persistence error.
    pub fn reset_account_sync(&self, now: i64) -> Result<AccountTransportBatch, SoftchatError> {
        let result = (|| {
            lock(&self.sync_engines).clear();
            let mut database = self.database()?;
            let reset = database.reset_sync(now)?;
            let mut diagnostic = Diagnostic::new(
                AccountLogCategory::Synchronization,
                "sync.reset",
                "Synchronization checkpoint reset",
            );
            diagnostic.revision = Some(reset.revision);
            self.logs.emit(diagnostic);
            let mut transport = lock(&self.transport_run);
            if let Some(run) = transport.as_mut() {
                run.restart_sync(&mut database, &self.identity, now)
            } else {
                Ok(AccountTransportBatch {
                    actions: Vec::new(),
                    connection_state: RelayConnectionState::Disconnected,
                    idle: true,
                    revision: database.info()?.revision,
                    typing_indicators: Vec::new(),
                    sync: None,
                })
            }
        })();
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Synchronization,
                "sync.reset",
                "Synchronization reset failed",
            ),
            &result,
        );
        result
    }

    /// Start one correlated Rust-owned account transport run.
    ///
    /// A host using the low-level boundary executes only the returned WebSocket
    /// actions. A second active run must be explicitly cancelled first.
    ///
    /// # Errors
    ///
    /// Returns a stable account, relay, filter, or lifecycle error.
    pub fn start_account_transport(
        &self,
        run_id: String,
    ) -> Result<AccountTransportBatch, SoftchatError> {
        let result = (|| {
            let recipient = self.public_key()?;
            let filters = vec![inbox_filter_json(&recipient, None, None)?];
            let database = self.database()?;
            let mut slot = lock(&self.transport_run);
            if slot.is_some() {
                return Err(SoftchatError::InvalidRelaySession);
            }
            let (run, batch) = AccountTransportRun::start(run_id, filters, &database)?;
            *slot = Some(run);
            Ok(batch)
        })();
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Relay,
                "relay.operation.failed",
                "Relay operation failed",
            ),
            &result,
        );
        result
    }

    /// Apply one exact correlated system result idempotently.
    ///
    /// Repeating the same result returns the originally committed action
    /// batch. A conflicting reuse or unknown action fails closed without
    /// consuming the active run. If an otherwise valid transition cannot
    /// commit, its database work is rolled back and the process-local run is
    /// retired; the host must start a fresh run.
    ///
    /// # Errors
    ///
    /// Returns a stable system-action, relay, delivery, ingestion, or
    /// persistence error.
    pub fn apply_account_transport_result(
        &self,
        transport_result: AccountTransportResult,
        now: i64,
    ) -> Result<AccountTransportBatch, SoftchatError> {
        let result = (|| {
            let hash = result_hash(&transport_result)?;
            let mut database = self.database()?;
            if let Some((stored_hash, outcome_json)) = database
                .stored_transport_result(&transport_result.run_id, &transport_result.action_id)?
            {
                if stored_hash != hash || outcome_json.is_empty() {
                    return Err(SoftchatError::InvalidRelaySession);
                }
                let outcome: StoredAccountTransportOutcome = serde_json::from_str(&outcome_json)
                    .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
                return Ok(outcome.into_batch());
            }
            let mut slot = lock(&self.transport_run);
            slot.as_ref()
                .ok_or(SoftchatError::InvalidRelaySession)?
                .validate_result_preflight(&transport_result)?;
            let mut candidate = slot.take().ok_or(SoftchatError::InvalidRelaySession)?;
            let transition = database.apply_transport_result_atomically(|database| {
                let mut batch =
                    candidate.handle_result(&transport_result, database, &self.identity, now)?;
                let expected_revision = database
                    .info()?
                    .revision
                    .checked_add(1)
                    .ok_or(SoftchatError::InvalidPersistenceResult)?;
                batch.revision = expected_revision;
                let outcome_json =
                    serde_json::to_string(&StoredAccountTransportOutcome::from(&batch))
                        .map_err(|_| SoftchatError::InternalFailure)?;
                let committed_revision = database.record_transport_result(
                    &transport_result.run_id,
                    &transport_result.action_id,
                    transport_result.generation,
                    &hash,
                    &outcome_json,
                )?;
                if committed_revision != expected_revision {
                    return Err(SoftchatError::InvalidPersistenceResult);
                }
                Ok(batch)
            });
            if transition.is_ok() {
                *slot = Some(candidate);
            }
            transition
        })();
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Relay,
                "relay.operation.failed",
                "Relay operation failed",
            ),
            &result,
        );
        result
    }

    /// Recheck durable outgoing work without waiting for the current receive deadline.
    ///
    /// Android calls this after observing a newer durable account revision.
    /// Existing receive actions remain correlated and active; only newly
    /// available ordered effects are returned.
    ///
    /// # Errors
    ///
    /// Returns a stable lifecycle, delivery, relay, or persistence error.
    pub fn wake_account_transport(&self, now: i64) -> Result<AccountTransportBatch, SoftchatError> {
        let result = (|| {
            let mut database = self.database()?;
            let mut slot = lock(&self.transport_run);
            let run = slot.as_mut().ok_or(SoftchatError::InvalidRelaySession)?;
            run.wake(&mut database, now)
        })();
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Relay,
                "relay.operation.failed",
                "Relay operation failed",
            ),
            &result,
        );
        result
    }

    /// Cancel the active account transport and release unfinished leases.
    ///
    /// # Errors
    ///
    /// Returns a stable lifecycle, delivery, or persistence error.
    pub fn cancel_account_transport(&self) -> Result<AccountTransportBatch, SoftchatError> {
        let result = (|| {
            let mut database = self.database()?;
            let mut slot = lock(&self.transport_run);
            let run = slot.as_mut().ok_or(SoftchatError::InvalidRelaySession)?;
            let batch = run.cancel(&mut database)?;
            slot.take();
            Ok(batch)
        })();
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Relay,
                "relay.operation.failed",
                "Relay operation failed",
            ),
            &result,
        );
        result
    }

    /// Authenticate and atomically persist one bounded incoming batch.
    ///
    /// # Errors
    ///
    /// Returns stable event, envelope, projection, account, or persistence
    /// errors. No partial transaction is committed.
    pub fn ingest(
        &self,
        events: Vec<SignedEvent>,
        received_at: i64,
    ) -> Result<StoredIngestion, SoftchatError> {
        let result = self.ingest_unlogged(events, received_at);
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Storage,
                "storage.ingest",
                "Incoming batch rejected",
            ),
            &result,
        );
        result
    }

    /// Parse, authenticate, unwrap, and commit a bounded system event batch.
    ///
    /// This is the Android push/notification boundary. Product code receives
    /// only authored message and conversation values, never raw Nostr types.
    ///
    /// # Errors
    ///
    /// Returns a stable event, envelope, account, projection, or persistence
    /// error with no partial commit.
    pub fn ingest_event_jsons(
        &self,
        event_jsons: Vec<String>,
        received_at: i64,
    ) -> Result<StoredIngestion, SoftchatError> {
        let preflight = preflight_ingestion_jsons(&event_jsons);
        let result = (|| {
            preflight?;
            let events = event_jsons
                .into_iter()
                .map(|event_json| {
                    SignedNostrEvent::from_json(&event_json).map(|event| SignedEvent::from(&event))
                })
                .collect::<Result<Vec<_>, _>>()?;
            let mut database = self.database()?;
            self.identity.with_identity(|identity| {
                database.ingest_with_authenticated_replays(identity, events, received_at)
            })
        })();
        self.logs.failed(
            Diagnostic::new(
                AccountLogCategory::Storage,
                "storage.ingest_json",
                "Incoming batch rejected",
            ),
            &result,
        );
        result
    }

    /// Create and atomically enqueue one private rumor operation.
    ///
    /// # Errors
    ///
    /// Returns stable identity, event, envelope, account, relay, delivery, or
    /// persistence errors.
    #[allow(clippy::too_many_arguments)]
    pub fn enqueue_rumor(
        &self,
        operation_id: String,
        rumor: RumorEvent,
        recipients: Vec<String>,
        notification_recipients: Vec<String>,
        relay_urls: Vec<String>,
        kind: Nip59EnvelopeKind,
        now: i64,
    ) -> Result<AccountOperationResult, SoftchatError> {
        let mut database = self.database()?;
        self.identity.with_identity(|identity| {
            database.enqueue_rumor(
                identity,
                operation_id,
                rumor,
                recipients,
                notification_recipients,
                relay_urls,
                kind,
                now,
            )
        })
    }

    /// Return the effective HTTPS upload endpoint for this account.
    ///
    /// A validated explicit account setting wins. Otherwise the endpoint is
    /// derived from the selected relay host using the released
    /// `/api/v2/files` profile. Android does not duplicate this policy.
    ///
    /// # Errors
    ///
    /// Returns a stable settings, relay-catalog, or HTTP-authorization error.
    pub fn media_upload_url(&self) -> Result<String, SoftchatError> {
        let (configured, relay_url) = {
            let database = self.database()?;
            (
                database.product_settings()?.media_service,
                database.active_relay_url()?,
            )
        };
        if let Some(configured) = configured {
            return Ok(configured);
        }
        let endpoint = parse_relay_endpoint(&relay_url)?;
        let mut url = Url::parse(&endpoint.network_url)
            .map_err(|_| SoftchatError::InvalidHttpAuthorization)?;
        url.set_scheme("https")
            .map_err(|_| SoftchatError::InvalidHttpAuthorization)?;
        url.set_path("/api/v2/files");
        url.set_query(None);
        url.set_fragment(None);
        Ok(url.to_string())
    }

    /// Atomically enqueue already signed direct protocol events.
    ///
    /// # Errors
    ///
    /// Returns stable identity, event, account, relay, or persistence errors.
    pub fn enqueue_signed(
        &self,
        operation_id: String,
        events: Vec<SignedEvent>,
        relay_urls: Vec<String>,
        now: i64,
    ) -> Result<AccountOperationResult, SoftchatError> {
        let mut database = self.database()?;
        self.identity.with_identity(|identity| {
            database.enqueue_signed(identity, operation_id, events, relay_urls, now)
        })
    }

    /// Verify and recover one bounded JSON array of already signed events.
    ///
    /// This narrow migration boundary lets a platform extract an opaque batch
    /// from retired storage without decoding protocol fields. Every event ID
    /// and signature is verified before one idempotent outgoing transaction is
    /// committed. The caller retains and owns the retired storage.
    ///
    /// # Errors
    ///
    /// Returns stable event, operation, relay, timestamp, or persistence
    /// errors. Malformed, empty, oversized, duplicate, or forged batches commit
    /// nothing.
    pub fn recover_signed_outbox_batch(
        &self,
        operation_id: String,
        event_batch_json: String,
        now: i64,
    ) -> Result<AccountOperationResult, SoftchatError> {
        if event_batch_json.len() > crate::MAX_NOSTR_EVENT_JSON_BYTES {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        let serde_json::Value::Array(raw_events) = crate::json::parse_value(&event_batch_json)
            .map_err(|_| SoftchatError::InvalidAccountOperation)?
        else {
            return Err(SoftchatError::InvalidAccountOperation);
        };
        if raw_events.is_empty() || raw_events.len() > crate::MAX_OPERATION_EVENT_COPIES {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        let events = raw_events
            .into_iter()
            .map(|event| {
                SignedNostrEvent::from_json(&event.to_string())
                    .map(|verified| SignedEvent::from(&verified))
            })
            .collect::<Result<Vec<_>, _>>()?;
        let payload_hash = command_hash(&("signed_outbox_recovery", &events))?;
        Ok(self
            .database()?
            .enqueue_product_signed(
                operation_id,
                payload_hash,
                String::new(),
                events,
                now,
                AccountLogSubject::Unknown,
            )?
            .operation)
    }

    /// Atomically claim bounded relay work and return exact payloads.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery or persistence error.
    pub fn claim_deliveries(
        &self,
        lease_id: String,
        relay_url: String,
        now: i64,
        lease_duration_seconds: u32,
    ) -> Result<Option<ClaimedDelivery>, SoftchatError> {
        self.database()?
            .claim_deliveries(lease_id, relay_url, now, lease_duration_seconds)
    }

    /// Persist that selected lease payloads were written to a socket.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery or persistence error for stale state.
    pub fn mark_socket_written(
        &self,
        claim: DeliveryClaim,
        intent_ids: Vec<String>,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?.mark_socket_written(&claim, intent_ids)
    }

    /// Persist correlated relay outcomes.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery or persistence error for stale state.
    pub fn apply_relay_results(
        &self,
        claim: DeliveryClaim,
        results: Vec<RelayDeliveryResult>,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?.apply_relay_results(&claim, results)
    }

    /// Release unfinished work after transport loss or cancellation.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery or persistence error for stale state.
    pub fn release_claim(
        &self,
        claim: DeliveryClaim,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?.release_claim(&claim)
    }

    /// Return one durable operation's aggregate state.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for unknown state.
    pub fn operation(&self, operation_id: String) -> Result<OperationSnapshot, SoftchatError> {
        self.database()?.operation(&operation_id)
    }

    /// Return a bounded newest-first page of durable outgoing operations.
    ///
    /// # Errors
    ///
    /// Returns a stable bound, account, or persistence error.
    pub fn operation_page(&self, limit: u32) -> Result<Vec<OperationSnapshot>, SoftchatError> {
        self.database()?.operations(limit)
    }

    /// Explicitly retry the rejected copies of one durable operation.
    ///
    /// # Errors
    ///
    /// Returns a stable account or delivery-state error.
    pub fn retry_operation(
        &self,
        operation_id: String,
    ) -> Result<AccountOperationResult, SoftchatError> {
        self.database()?.retry_operation(&operation_id)
    }

    /// Cancel every unclaimed copy that is still safe to cancel.
    ///
    /// # Errors
    ///
    /// Returns a stable account or delivery-state error.
    pub fn cancel_operation(
        &self,
        operation_id: String,
    ) -> Result<AccountOperationResult, SoftchatError> {
        self.database()?.cancel_operation(&operation_id)
    }

    /// Return the complete durable relay catalog.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn relay_catalog(&self) -> Result<Vec<RelayCatalogEntry>, SoftchatError> {
        self.database()?.relay_catalog()
    }

    /// Return the canonical authenticated inbox filter for this account.
    ///
    /// Android passes the returned JSON directly to [`crate::RelaySession`].
    /// The filter accepts durable and ephemeral NIP-59 gift wraps addressed to
    /// the account. An optional lower bound supports live subscriptions without
    /// making Android construct or validate protocol JSON.
    ///
    /// # Errors
    ///
    /// Returns a stable filter or timestamp error.
    pub fn inbox_filters(
        &self,
        since_timestamp: Option<i64>,
    ) -> Result<Vec<String>, SoftchatError> {
        self.ensure_open()?;
        let recipient_public_key = self
            .identity
            .with_identity(|identity| Ok(identity.public_key().to_hex()))?;
        Ok(vec![inbox_filter_json(
            &recipient_public_key,
            since_timestamp,
            None,
        )?])
    }

    /// Reconcile and atomically persist authenticated relay discovery.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn reconcile_account_relays(
        &self,
        discovered: Vec<DnsRelayRecord>,
        refreshed_at: i64,
        fallback_url: String,
        dns_authenticated: bool,
    ) -> Result<StoredRelayCatalogPlan, SoftchatError> {
        self.database()?.reconcile_relay_catalog(
            discovered,
            refreshed_at,
            fallback_url,
            dns_authenticated,
        )
    }

    /// Add and atomically persist one custom relay.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn add_account_relay(
        &self,
        relay_url: String,
        created_at: i64,
    ) -> Result<RelayCatalogMutation, SoftchatError> {
        self.database()?.add_custom_relay(relay_url, created_at)
    }

    /// Exclude and atomically persist one relay.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn exclude_account_relay(
        &self,
        relay_url: String,
        now: i64,
    ) -> Result<RelayCatalogMutation, SoftchatError> {
        self.database()?.exclude_relay(relay_url, now)
    }

    /// Restore and atomically persist automatic relays.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn restore_account_relays(&self, now: i64) -> Result<RelayCatalogMutation, SoftchatError> {
        self.database()?.restore_automatic_relays(now)
    }

    /// Select and atomically persist one active relay.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn set_account_active_relay(
        &self,
        relay_url: String,
        now: i64,
    ) -> Result<RelayCatalogMutation, SoftchatError> {
        self.database()?.set_active_relay(relay_url, now)
    }

    /// Return failover candidates from durable relay state.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn account_relay_failover_candidates(
        &self,
        failed_url: String,
    ) -> Result<Vec<String>, SoftchatError> {
        self.database()?.relay_failover_candidates(failed_url)
    }

    /// Atomically select the next relay and rebind unfinished work.
    ///
    /// # Errors
    ///
    /// Returns a stable catalog, delivery-state, or persistence error.
    pub fn rotate_failed_account_relay(
        &self,
        failed_url: String,
        now: i64,
    ) -> Result<RelayFailoverMutation, SoftchatError> {
        self.database()?.rotate_failed_relay(failed_url, now)
    }

    /// Return a bounded newest-first conversation page.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn list_conversations(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<ConversationSummary>, SoftchatError> {
        self.database()?.list_conversations(limit, offset)
    }

    /// Return complete bounded conversation-list use cases in one native call.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn list_conversation_views(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<ConversationView>, SoftchatError> {
        self.database()?.list_conversation_views(limit, offset)
    }

    /// Return a bounded newest-first message page.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn list_messages(
        &self,
        conversation_id: String,
        before_created_at: Option<i64>,
        limit: u32,
    ) -> Result<Vec<AccountMessage>, SoftchatError> {
        self.database()?
            .list_messages(&conversation_id, before_created_at, limit)
    }

    /// Return a bounded newest-first authenticated projection page.
    ///
    /// Empty string filters are ignored.
    ///
    /// # Errors
    ///
    /// Returns a stable account, event, or persistence error.
    #[allow(clippy::too_many_arguments)]
    pub fn list_projections(
        &self,
        kind: Option<crate::ProjectionKind>,
        logical_event_id: String,
        conversation_id: String,
        target_event_id: String,
        author_public_key: String,
        before_created_at: Option<i64>,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<AccountProjection>, SoftchatError> {
        self.database()?.list_projections(
            kind,
            logical_event_id,
            conversation_id,
            target_event_id,
            author_public_key,
            before_created_at,
            limit,
            offset,
        )
    }

    /// Return bounded authenticated projections with use-case state.
    ///
    /// # Errors
    ///
    /// Returns a stable account, event, or persistence error.
    #[allow(clippy::too_many_arguments)]
    pub fn list_event_nodes(
        &self,
        kind: Option<crate::ProjectionKind>,
        logical_event_id: String,
        conversation_id: String,
        target_event_id: String,
        author_public_key: String,
        before_created_at: Option<i64>,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<AccountEventNode>, SoftchatError> {
        self.database()?.list_event_nodes(
            kind,
            logical_event_id,
            conversation_id,
            target_event_id,
            author_public_key,
            before_created_at,
            limit,
            offset,
        )
    }

    /// Return bounded canonical public keys known to authenticated state.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn list_known_public_keys(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<String>, SoftchatError> {
        self.database()?.list_known_public_keys(limit, offset)
    }

    /// Insert or replace one bounded Android-local contact.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn put_local_contact(
        &self,
        contact: LocalContactRecord,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?.put_local_contact(contact)
    }

    /// Remove one Android-local contact idempotently.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn remove_local_contact(
        &self,
        public_key: String,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?.remove_local_contact(public_key)
    }

    /// Return a bounded stable page of Android-local contacts.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn local_contacts(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<LocalContactRecord>, SoftchatError> {
        self.database()?.local_contacts(limit, offset)
    }

    /// Replace the complete bounded account sticker cache.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn replace_stickers(
        &self,
        stickers: Vec<StickerRecord>,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?.replace_stickers(stickers)
    }

    /// Return the complete bounded account sticker cache.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence error.
    pub fn stickers(&self) -> Result<Vec<StickerRecord>, SoftchatError> {
        self.database()?.stickers()
    }

    /// Insert or replace one account link-preview cache result.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn put_link_preview(
        &self,
        preview: LinkPreviewRecord,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?.put_link_preview(preview)
    }

    /// Return one account link-preview cache result.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error.
    pub fn link_preview(&self, url: String) -> Result<Option<LinkPreviewRecord>, SoftchatError> {
        self.database()?.link_preview(url)
    }

    /// Clear every account-owned durable row and process-local coordinator.
    ///
    /// The account binding and revision remain valid. Settings and relay
    /// catalog return to safe defaults. Active sync engines, transport state,
    /// and authored typing state are discarded only after SQLite commits.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence error.
    pub fn clear_account(&self) -> Result<AccountMutationResult, SoftchatError> {
        let mut sync_engines = lock(&self.sync_engines);
        let mut database = self.database()?;
        let mut transport_run = lock(&self.transport_run);
        let result = database.clear()?;
        let mut diagnostic = Diagnostic::new(
            AccountLogCategory::Recovery,
            "account.cleared",
            "Account data cleared",
        );
        diagnostic.revision = Some(result.revision);
        self.logs.emit(diagnostic);
        sync_engines.clear();
        transport_run.take();
        lock(&self.typing_authored).clear();
        Ok(result)
    }

    /// Return exact canonical signed-event JSON by ID.
    ///
    /// # Errors
    ///
    /// Returns a stable event or persistence error.
    pub fn event_json(&self, event_id: String) -> Result<String, SoftchatError> {
        self.database()?.event_json(&event_id)
    }

    /// Return exact canonical JSON for bounded synchronization resend IDs.
    ///
    /// # Errors
    ///
    /// Returns a stable event, account, or persistence error.
    pub fn event_jsons(
        &self,
        event_ids: Vec<String>,
    ) -> Result<Vec<StoredEventJson>, SoftchatError> {
        self.database()?.event_jsons(event_ids)
    }

    /// Load a bounded Rust-owned synchronization fingerprint snapshot.
    ///
    /// # Errors
    ///
    /// Returns a stable sync or persistence error.
    pub fn sync_fingerprints(
        &self,
        since_timestamp: Option<i64>,
        until_timestamp: i64,
        limit: u32,
    ) -> Result<Vec<NegentropyItem>, SoftchatError> {
        self.database()?
            .sync_fingerprints(since_timestamp, until_timestamp, limit)
    }

    /// Return a durable synchronization checkpoint, when present.
    ///
    /// # Errors
    ///
    /// Returns a stable sync or persistence error.
    pub fn sync_checkpoint(&self, sync_id: String) -> Result<Option<i64>, SoftchatError> {
        self.database()?.sync_checkpoint(&sync_id)
    }

    /// Commit one synchronization checkpoint and increment account revision.
    ///
    /// # Errors
    ///
    /// Returns a stable sync or persistence error.
    pub fn save_sync_checkpoint(
        &self,
        sync_id: String,
        checkpoint_timestamp: i64,
        now: i64,
    ) -> Result<AccountMutationResult, SoftchatError> {
        self.database()?
            .save_sync_checkpoint(&sync_id, checkpoint_timestamp, now)
    }

    /// Start one sync and return its first transport action.
    ///
    /// Fingerprint snapshot loading is completed inside Rust before this
    /// method returns, so Android never receives a database action.
    ///
    /// # Errors
    ///
    /// Returns a stable sync, Negentropy, or persistence error.
    #[allow(clippy::too_many_arguments)]
    pub fn start_sync(
        &self,
        sync_id: String,
        neg_subscription_id: String,
        event_subscription_id: String,
        now: i64,
        overlap_seconds: u32,
        frame_size_limit: u64,
    ) -> Result<SyncAction, SoftchatError> {
        self.ensure_open()?;
        if overlap_seconds > 604_800 {
            return Err(SoftchatError::InvalidSyncState);
        }
        let mut engines = lock(&self.sync_engines);
        if engines.contains_key(&sync_id) {
            return Err(SoftchatError::InvalidSyncState);
        }
        let since_timestamp = self
            .database()?
            .sync_checkpoint(&sync_id)?
            .map(|checkpoint| checkpoint.saturating_sub(i64::from(overlap_seconds)));
        let mut engine = SyncEngine::new(
            sync_id.clone(),
            neg_subscription_id,
            event_subscription_id,
            since_timestamp,
            now,
        )?;
        let action = engine.begin()?;
        let action = {
            let mut database = self.database()?;
            resolve_sync_action(&mut database, &mut engine, action, frame_size_limit)?
        };
        let recipient_public_key = self
            .identity
            .with_identity(|identity| Ok(identity.public_key().to_hex()))?;
        let action = plan_sync_transport(action, &recipient_public_key)?;
        engines.insert(sync_id, engine);
        Ok(action)
    }

    /// Consume one correlated `NEG-MSG` and return transport-only actions.
    ///
    /// # Errors
    ///
    /// Returns a stable sync, Negentropy, or persistence error.
    pub fn reconcile_sync(
        &self,
        sync_id: String,
        subscription_id: String,
        message: Vec<u8>,
        frame_size_limit: u64,
    ) -> Result<Vec<SyncAction>, SoftchatError> {
        self.ensure_open()?;
        let mut engines = lock(&self.sync_engines);
        let engine = engines
            .get_mut(&sync_id)
            .ok_or(SoftchatError::InvalidSyncState)?;
        let actions = engine.reconcile(subscription_id, message)?;
        let actions = {
            let mut database = self.database()?;
            resolve_sync_actions(&mut database, engine, actions, frame_size_limit)?
        };
        let recipient_public_key = self
            .identity
            .with_identity(|identity| Ok(identity.public_key().to_hex()))?;
        actions
            .into_iter()
            .map(|action| plan_sync_transport(action, &recipient_public_key))
            .collect()
    }

    /// Route one complete relay text frame to Negentropy when applicable.
    ///
    /// Ordinary Nostr relay frames return `None` and must be passed to
    /// `RelaySession.receive`. Correlated `NEG-MSG` is parsed, bounded, and
    /// reconciled entirely in Rust.
    ///
    /// # Errors
    ///
    /// Returns a stable sync error for malformed, stale, or relay-error frames.
    pub fn receive_sync_frame(
        &self,
        sync_id: String,
        frame_json: String,
        frame_size_limit: u64,
    ) -> Result<Option<Vec<SyncAction>>, SoftchatError> {
        let Some(frame) = parse_sync_message_frame(&frame_json)? else {
            return Ok(None);
        };
        let SyncTransportFrame::Message {
            subscription_id,
            message,
        } = frame
        else {
            return Err(SoftchatError::InvalidSyncState);
        };
        self.reconcile_sync(sync_id, subscription_id, message, frame_size_limit)
            .map(Some)
    }

    /// Authenticate and commit requested sync events before marking progress.
    ///
    /// # Errors
    ///
    /// Returns a stable sync, event, envelope, projection, or persistence
    /// error. Sync state does not advance unless the database commits.
    pub fn ingest_sync_events(
        &self,
        sync_id: String,
        subscription_id: String,
        events: Vec<SignedEvent>,
        received_at: i64,
    ) -> Result<StoredIngestion, SoftchatError> {
        let event_ids = events
            .iter()
            .map(|event| event.id.clone())
            .collect::<Vec<_>>();
        let mut engines = lock(&self.sync_engines);
        let engine = engines
            .get_mut(&sync_id)
            .ok_or(SoftchatError::InvalidSyncState)?;
        engine.validate_fetched_events(&subscription_id, &event_ids)?;
        let mut database = self.database()?;
        let stored = self
            .identity
            .with_identity(|identity| database.ingest(identity, events, received_at))?;
        for event_id in &event_ids {
            engine.record_fetched_event(subscription_id.clone(), event_id.clone())?;
        }
        engine.confirm_committed(event_ids)?;
        Ok(stored)
    }

    /// Finish a requested-event subscription after every event committed.
    ///
    /// # Errors
    ///
    /// Returns a stable sync, Negentropy, or persistence error.
    pub fn finish_sync_fetch(
        &self,
        sync_id: String,
        subscription_id: String,
        frame_size_limit: u64,
    ) -> Result<Vec<SyncAction>, SoftchatError> {
        let mut engines = lock(&self.sync_engines);
        let engine = engines
            .get_mut(&sync_id)
            .ok_or(SoftchatError::InvalidSyncState)?;
        let actions = engine.finish_fetch(subscription_id)?;
        let actions = {
            let mut database = self.database()?;
            resolve_sync_actions(&mut database, engine, actions, frame_size_limit)?
        };
        let recipient_public_key = self
            .identity
            .with_identity(|identity| Ok(identity.public_key().to_hex()))?;
        actions
            .into_iter()
            .map(|action| plan_sync_transport(action, &recipient_public_key))
            .collect()
    }

    /// Return exact locally stored payloads requested by a resend action.
    ///
    /// # Errors
    ///
    /// Returns a stable event, account, or persistence error.
    pub fn sync_resend_payloads(
        &self,
        event_ids: Vec<String>,
    ) -> Result<Vec<StoredEventJson>, SoftchatError> {
        self.database()?.event_jsons(event_ids)
    }

    /// Return one active synchronization snapshot.
    ///
    /// # Errors
    ///
    /// Returns a stable sync error for an unknown ID.
    pub fn sync_snapshot(&self, sync_id: String) -> Result<SyncEngineSnapshot, SoftchatError> {
        self.ensure_open()?;
        lock(&self.sync_engines)
            .get(&sync_id)
            .map(SyncEngine::snapshot)
            .ok_or(SoftchatError::InvalidSyncState)
    }

    /// Cancel and remove one active synchronization coordinator.
    ///
    /// # Errors
    ///
    /// Returns a stable sync error for an unknown ID.
    pub fn cancel_sync(&self, sync_id: String) -> Result<SyncEngineSnapshot, SoftchatError> {
        self.ensure_open()?;
        let mut engine = lock(&self.sync_engines)
            .remove(&sync_id)
            .ok_or(SoftchatError::InvalidSyncState)?;
        Ok(engine.cancel())
    }
}

impl AccountRuntimeHandle {
    fn ingest_unlogged(
        &self,
        events: Vec<SignedEvent>,
        received_at: i64,
    ) -> Result<StoredIngestion, SoftchatError> {
        let mut database = self.database()?;
        self.identity
            .with_identity(|identity| database.ingest(identity, events, received_at))
    }

    fn enqueue_follow_replacement(
        &self,
        database: &mut AccountDatabase,
        command_id: String,
        contacts: Vec<Contact>,
        now: i64,
    ) -> Result<ProductAccountOperationResult, SoftchatError> {
        validate_command_id(&command_id)?;
        let payload_hash = command_hash(&("follows", &contacts))?;
        let own_public_key = self.public_key()?;
        let latest = database.authenticated_product_follow_updated_at(&own_public_key)?;
        let authored_at = if latest > 0 && latest >= now {
            latest
                .checked_add(1)
                .ok_or(SoftchatError::InvalidAccountOperation)?
        } else {
            now
        };
        let created_at =
            u64::try_from(authored_at).map_err(|_| SoftchatError::InvalidAccountOperation)?;
        let rumor = self.identity.with_identity(|identity| {
            identity
                .create_follow_list(created_at, contacts.clone())
                .map(|view| view.rumor)
        })?;
        let recipients = contacts
            .into_iter()
            .map(|contact| contact.public_key)
            .filter(|key| key != &own_public_key)
            .collect::<Vec<_>>();
        self.identity.with_identity(|identity| {
            database.enqueue_product_rumor(
                identity,
                command_id,
                payload_hash,
                String::new(),
                rumor,
                recipients,
                Vec::new(),
                Nip59EnvelopeKind::Durable,
                now,
                None,
            )
        })
    }

    fn ensure_open(&self) -> Result<(), SoftchatError> {
        if self.closed.load(Ordering::Acquire) {
            Err(SoftchatError::AccountClosed)
        } else {
            Ok(())
        }
    }

    fn database(&self) -> Result<MutexGuard<'_, AccountDatabase>, SoftchatError> {
        self.ensure_open()?;
        Ok(lock(&self.database))
    }
}

impl Drop for AccountRuntimeHandle {
    fn drop(&mut self) {
        self.closed.store(true, Ordering::Release);
        self.identity.erase();
        lock(open_account_paths()).remove(&self.database_path);
    }
}

fn remote_members(conversation: &ConversationRecord, own_public_key: &str) -> Vec<String> {
    conversation
        .member_public_keys
        .iter()
        .filter(|key| key.as_str() != own_public_key)
        .cloned()
        .collect()
}

fn all_known_public_keys(database: &AccountDatabase) -> Result<Vec<String>, SoftchatError> {
    let page_limit = crate::MAX_PRODUCT_QUERY_PAGE;
    let mut offset = 0;
    let mut result = Vec::new();
    while result.len() < crate::MAX_CONTACTS {
        let page = database.list_known_public_keys(page_limit, offset)?;
        let page_len = page.len();
        result.extend(page);
        if page_len < page_limit as usize {
            break;
        }
        offset = offset
            .checked_add(page_limit)
            .ok_or(SoftchatError::InvalidAccountOperation)?;
    }
    if result.len() > crate::MAX_CONTACTS {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    Ok(result)
}

fn send_message_command_hash(
    conversation_id: &str,
    content: &MessageContentInput,
    reply_to_message_id: &str,
    draft_match: Option<&crate::account_product::ProductDraftMatch>,
) -> Result<String, SoftchatError> {
    if let Some(draft_match) = draft_match {
        command_hash(&(
            "send",
            conversation_id,
            content,
            reply_to_message_id,
            draft_match,
        ))
    } else {
        command_hash(&("send", conversation_id, content, reply_to_message_id))
    }
}

fn media_message_command_hash(
    conversation_id: &str,
    content: &MessageContentInput,
    reply_to_message_id: &str,
    sources: &[crate::MediaPreparationInput],
    draft_match: Option<&crate::account_product::ProductDraftMatch>,
) -> Result<String, SoftchatError> {
    if let Some(draft_match) = draft_match {
        command_hash(&(
            "media-message",
            conversation_id,
            content,
            reply_to_message_id,
            sources,
            draft_match,
        ))
    } else {
        command_hash(&(
            "media-message",
            conversation_id,
            content,
            reply_to_message_id,
            sources,
        ))
    }
}

fn public_keys(values: &[String]) -> Result<Vec<NostrPublicKey>, SoftchatError> {
    values
        .iter()
        .map(|value| NostrPublicKey::from_hex(value))
        .collect()
}

fn author_product_chat_message(
    account: &AccountRuntimeHandle,
    conversation_id: &str,
    content: MessageContentInput,
    reply_to_message_id: &str,
    now: i64,
) -> Result<(RumorEvent, Vec<String>), SoftchatError> {
    content.validate()?;
    let own_public_key = account.public_key()?;
    let conversation = account
        .database()?
        .product_conversation(conversation_id, &own_public_key)?
        .ok_or(SoftchatError::InvalidAccountOperation)?;
    let recipients = remote_members(&conversation, &own_public_key);
    let relation = if reply_to_message_id.is_empty() {
        ChatRelation::None
    } else {
        let parent = account
            .database()?
            .product_message(reply_to_message_id)?
            .filter(|message| message.conversation_id == conversation_id)
            .ok_or(SoftchatError::InvalidAccountOperation)?;
        ChatRelation::Reply {
            event_id: NostrEventId::from_hex(&parent.id)?,
            relay_hint: Some(account.database()?.active_relay_url()?),
            legacy_unmarked: false,
        }
    };
    let created_at = u64::try_from(now).map_err(|_| SoftchatError::InvalidAccountOperation)?;
    let draft = crate::chat::prepare_chat_message_draft(
        NostrPublicKey::from_hex(&own_public_key)?,
        ChatMessageDraft {
            created_at,
            participants: public_keys(&recipients)?,
            content: content.text,
            relation,
            attachments: content.attachments,
            emoji_tags: nostr_tags(content.emoji_tags)?,
            extension_tags: Vec::new(),
        },
    )?;
    crate::envelope::validate_nip59_draft_wire_size(
        &draft,
        Nip59EnvelopeKind::Durable,
        created_at,
        !recipients.is_empty(),
    )
    .map_err(|_| SoftchatError::InvalidAccountOperation)?;
    let rumor = account.identity.with_identity(|identity| {
        identity
            .create_rumor(draft)
            .map(|rumor| RumorEvent::from(&rumor))
            .map_err(|_| SoftchatError::SoftchatEventCreationFailed)
    })?;
    Ok((rumor, recipients))
}

fn preflight_ingestion_jsons(event_jsons: &[String]) -> Result<u64, SoftchatError> {
    if event_jsons.is_empty() || event_jsons.len() > crate::MAX_ACCOUNT_INGESTION_EVENTS {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    bounded_ingestion_json_bytes(event_jsons.iter().map(String::len))
}

fn bounded_ingestion_json_bytes(
    lengths: impl IntoIterator<Item = usize>,
) -> Result<u64, SoftchatError> {
    let total = lengths.into_iter().try_fold(0_usize, |total, length| {
        total
            .checked_add(length)
            .filter(|next| *next <= crate::MAX_ACCOUNT_INGESTION_BYTES)
            .ok_or(SoftchatError::InvalidAccountOperation)
    })?;
    u64::try_from(total).map_err(|_| SoftchatError::InvalidAccountOperation)
}

fn nostr_tags(values: Vec<Vec<String>>) -> Result<Vec<NostrTag>, SoftchatError> {
    values.into_iter().map(NostrTag::new).collect()
}

fn stored_product_result(
    account: &AccountRuntimeHandle,
    operation: ProductAccountOperationResult,
) -> Result<ProductOperationResult, SoftchatError> {
    let message = if operation.result_message_id.is_empty() {
        None
    } else {
        account
            .database()?
            .product_message(&operation.result_message_id)?
    };
    Ok(ProductOperationResult {
        operation: operation.operation.operation,
        revision: operation.operation.revision,
        inserted: operation.operation.inserted,
        message,
    })
}

/// Session-scoped account planner backed by one opaque identity.
#[cfg_attr(feature = "native-bindings", derive(uniffi::Object))]
pub struct AccountEngineHandle {
    account_id: String,
    identity: Arc<LocalIdentityHandle>,
}

impl fmt::Debug for AccountEngineHandle {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("AccountEngineHandle")
            .field("account_id", &self.account_id)
            .finish_non_exhaustive()
    }
}

#[cfg_attr(feature = "native-bindings", uniffi::export)]
impl AccountEngineHandle {
    /// Bind one opaque identity to one non-secret account partition.
    ///
    /// # Errors
    ///
    /// Returns a stable account error for an invalid partition identifier.
    #[cfg_attr(feature = "native-bindings", uniffi::constructor)]
    pub fn new(
        account_id: String,
        identity: Arc<LocalIdentityHandle>,
    ) -> Result<Self, SoftchatError> {
        AccountEngine::validate_account_id(&account_id)?;
        identity.with_identity(|_| Ok(()))?;
        Ok(Self {
            account_id,
            identity,
        })
    }

    /// Return the non-secret account partition.
    #[must_use]
    pub fn account_id(&self) -> String {
        self.account_id.clone()
    }

    /// Authenticate a bounded incoming batch before a host transaction.
    ///
    /// # Errors
    ///
    /// Returns stable event, envelope, projection, or account errors.
    pub fn prepare_incoming(
        &self,
        events: Vec<SignedEvent>,
    ) -> Result<PreparedIncomingBatch, SoftchatError> {
        self.identity.with_identity(|identity| {
            AccountEngine::prepare_incoming(identity, self.account_id.clone(), events)
        })
    }

    /// Validate a host store's total transaction receipt.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence-result error for any mismatch.
    pub fn validate_ingestion_receipt(
        &self,
        batch: PreparedIncomingBatch,
        receipt: IngestionReceipt,
    ) -> Result<(), SoftchatError> {
        AccountEngine::validate_ingestion_receipt(&batch, &receipt)
    }

    /// Create independently wrapped copies and per-relay durable intents.
    ///
    /// # Errors
    ///
    /// Returns stable identity, rumor, account, recipient, or relay errors.
    #[allow(clippy::too_many_arguments)]
    pub fn prepare_rumor_operation(
        &self,
        operation_id: String,
        rumor: RumorEvent,
        recipients: Vec<String>,
        notification_recipients: Vec<String>,
        relay_urls: Vec<String>,
        kind: Nip59EnvelopeKind,
        now: i64,
    ) -> Result<PreparedOutgoingOperation, SoftchatError> {
        self.identity.with_identity(|identity| {
            AccountEngine::prepare_rumor_operation(
                identity,
                self.account_id.clone(),
                operation_id,
                rumor,
                recipients,
                notification_recipients,
                relay_urls,
                kind,
                now,
            )
        })
    }

    /// Create durable intents for already signed direct events.
    ///
    /// # Errors
    ///
    /// Returns stable event, account, operation, or relay errors.
    pub fn prepare_signed_operation(
        &self,
        operation_id: String,
        events: Vec<SignedEvent>,
        relay_urls: Vec<String>,
    ) -> Result<PreparedOutgoingOperation, SoftchatError> {
        self.identity.with_identity(|_| {
            AccountEngine::prepare_signed_operation(
                self.account_id.clone(),
                operation_id,
                events,
                relay_urls,
            )
        })
    }
}

/// Claim one bounded durable-delivery batch.
///
/// # Errors
///
/// Returns a stable delivery-state error for malformed or unclaimable rows.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn claim_deliveries(
    intents: Vec<DeliveryIntentSnapshot>,
    lease_id: String,
) -> Result<DeliveryClaim, SoftchatError> {
    DeliveryReducer::claim(intents, lease_id)
}

/// Mark an exact leased subset as socket-written.
///
/// # Errors
///
/// Returns a stable delivery-state error for stale or contradictory state.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn mark_deliveries_socket_written(
    claim: DeliveryClaim,
    intent_ids: Vec<String>,
) -> Result<Vec<DeliveryStateMutation>, SoftchatError> {
    DeliveryReducer::mark_socket_written(&claim, intent_ids)
}

/// Apply stable relay results to one durable lease.
///
/// # Errors
///
/// Returns a stable delivery-state error for stale or contradictory results.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn apply_relay_delivery_results(
    claim: DeliveryClaim,
    results: Vec<RelayDeliveryResult>,
) -> Result<Vec<DeliveryStateMutation>, SoftchatError> {
    DeliveryReducer::apply_relay_results(&claim, results)
}

/// Release every unfinished row from one durable lease.
///
/// # Errors
///
/// Returns a stable delivery-state error for malformed claim state.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn release_delivery_claim(
    claim: DeliveryClaim,
) -> Result<Vec<DeliveryStateMutation>, SoftchatError> {
    DeliveryReducer::release_claim(&claim)
}

/// Derive one aggregate user-visible operation state.
///
/// # Errors
///
/// Returns a stable operation error for empty or mixed-operation rows.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn derive_operation_snapshot(
    intents: Vec<DeliveryIntentSnapshot>,
) -> Result<OperationSnapshot, SoftchatError> {
    DeliveryReducer::operation_snapshot(intents)
}

/// Compute the released-Android conversation hash from complete participants.
///
/// # Errors
///
/// Returns a stable account error for malformed or duplicate public keys.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn derive_conversation_id(public_keys: Vec<String>) -> Result<String, SoftchatError> {
    conversation_id(public_keys)
}

/// Reconcile authenticated DNS records with persisted user catalog state.
///
/// # Errors
///
/// Returns a stable relay-catalog error for malformed or excessive input.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn reconcile_relay_catalog(
    current: Vec<RelayCatalogEntry>,
    discovered: Vec<DnsRelayRecord>,
    refreshed_at: i64,
    fallback_url: String,
    dns_authenticated: bool,
) -> Result<RelayCatalogPlan, SoftchatError> {
    RelayCatalogReducer::reconcile(
        current,
        discovered,
        refreshed_at,
        fallback_url,
        dns_authenticated,
    )
}

/// Add one custom relay.
///
/// # Errors
///
/// Returns a stable relay-catalog error for malformed state or URL.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn add_custom_relay(
    current: Vec<RelayCatalogEntry>,
    relay_url: String,
    created_at: i64,
) -> Result<Vec<RelayCatalogEntry>, SoftchatError> {
    RelayCatalogReducer::add_custom(current, relay_url, created_at)
}

/// Exclude or remove one relay.
///
/// # Errors
///
/// Returns a stable relay-catalog error for malformed state or URL.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn exclude_relay(
    current: Vec<RelayCatalogEntry>,
    relay_url: String,
) -> Result<Vec<RelayCatalogEntry>, SoftchatError> {
    RelayCatalogReducer::exclude(current, relay_url)
}

/// Restore all currently discovered automatic relays.
///
/// # Errors
///
/// Returns a stable relay-catalog error for malformed persisted state.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn restore_automatic_relays(
    current: Vec<RelayCatalogEntry>,
) -> Result<Vec<RelayCatalogEntry>, SoftchatError> {
    RelayCatalogReducer::restore_automatic(current)
}

/// Select one effective active relay.
///
/// # Errors
///
/// Returns a stable relay-catalog error for malformed or excluded targets.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn set_active_relay(
    current: Vec<RelayCatalogEntry>,
    relay_url: String,
) -> Result<Vec<RelayCatalogEntry>, SoftchatError> {
    RelayCatalogReducer::set_active(current, relay_url)
}

/// Return effective failover candidates.
///
/// # Errors
///
/// Returns a stable relay-catalog error for malformed persisted state.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn relay_failover_candidates(
    current: Vec<RelayCatalogEntry>,
    failed_url: String,
) -> Result<Vec<String>, SoftchatError> {
    RelayCatalogReducer::failover_candidates(current, failed_url)
}

/// Return the bounded retry/failover plan for one failure.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
#[must_use]
pub fn relay_retry_plan(
    attempt: u32,
    failure: RelayFailureKind,
    bypass_delay_once: bool,
) -> RelayRetryPlan {
    plan_relay_retry(attempt, failure, bypass_delay_once)
}

/// Validate and split one configured relay endpoint.
///
/// # Errors
///
/// Returns a stable relay-catalog error for insecure or malformed endpoints.
#[cfg_attr(feature = "native-bindings", uniffi::export)]
pub fn relay_endpoint_plan(configured_url: String) -> Result<RelayEndpointPlan, SoftchatError> {
    parse_relay_endpoint(&configured_url)
}

/// Thread-safe generated-binding owner for one synchronization coordinator.
#[cfg_attr(feature = "native-bindings", derive(uniffi::Object))]
pub struct SyncEngineHandle {
    engine: Mutex<SyncEngine>,
}

impl fmt::Debug for SyncEngineHandle {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        formatter
            .debug_struct("SyncEngineHandle")
            .field("snapshot", &lock(&self.engine).snapshot())
            .finish()
    }
}

#[cfg_attr(feature = "native-bindings", uniffi::export)]
impl SyncEngineHandle {
    /// Construct one bounded synchronization coordinator.
    ///
    /// # Errors
    ///
    /// Returns a stable sync-state error for invalid IDs or timestamps.
    #[cfg_attr(feature = "native-bindings", uniffi::constructor)]
    pub fn new(
        sync_id: String,
        neg_subscription_id: String,
        event_subscription_id: String,
        since_timestamp: Option<i64>,
        checkpoint_timestamp: i64,
    ) -> Result<Self, SoftchatError> {
        Ok(Self {
            engine: Mutex::new(SyncEngine::new(
                sync_id,
                neg_subscription_id,
                event_subscription_id,
                since_timestamp,
                checkpoint_timestamp,
            )?),
        })
    }

    /// Begin and request the first durable fingerprint snapshot.
    ///
    /// # Errors
    ///
    /// Returns a stable sync-state error outside the initial phase.
    pub fn begin(&self) -> Result<SyncAction, SoftchatError> {
        lock(&self.engine).begin()
    }

    /// Consume a durable fingerprint snapshot and produce NEG-OPEN.
    ///
    /// # Errors
    ///
    /// Returns stable sync/Negentropy errors for malformed or stale input.
    pub fn accept_snapshot(
        &self,
        items: Vec<NegentropyItem>,
        frame_size_limit: u64,
    ) -> Result<SyncAction, SoftchatError> {
        lock(&self.engine).accept_snapshot(items, frame_size_limit)
    }

    /// Consume one correlated NEG-MSG.
    ///
    /// # Errors
    ///
    /// Returns stable sync/Negentropy errors for malformed or stale input.
    pub fn reconcile(
        &self,
        subscription_id: String,
        message: Vec<u8>,
    ) -> Result<Vec<SyncAction>, SoftchatError> {
        lock(&self.engine).reconcile(subscription_id, message)
    }

    /// Record one correlated requested event.
    ///
    /// # Errors
    ///
    /// Returns a stable sync-state error for stale or unrequested IDs.
    pub fn record_fetched_event(
        &self,
        subscription_id: String,
        event_id: String,
    ) -> Result<(), SoftchatError> {
        lock(&self.engine).record_fetched_event(subscription_id, event_id)
    }

    /// Confirm a durable transaction committed requested events.
    ///
    /// # Errors
    ///
    /// Returns a stable sync-state error for stale or unknown IDs.
    pub fn confirm_committed(&self, event_ids: Vec<String>) -> Result<(), SoftchatError> {
        lock(&self.engine).confirm_committed(event_ids)
    }

    /// Finish one requested-event subscription.
    ///
    /// # Errors
    ///
    /// Returns a stable sync-state error until all requested IDs committed.
    pub fn finish_fetch(&self, subscription_id: String) -> Result<Vec<SyncAction>, SoftchatError> {
        lock(&self.engine).finish_fetch(subscription_id)
    }

    /// Confirm the final checkpoint transaction.
    ///
    /// # Errors
    ///
    /// Returns a stable sync-state error for stale or mismatched state.
    pub fn checkpoint_saved(&self, checkpoint_timestamp: i64) -> Result<SyncAction, SoftchatError> {
        lock(&self.engine).checkpoint_saved(checkpoint_timestamp)
    }

    /// Permanently cancel and clear process-local reconciliation state.
    #[must_use]
    pub fn cancel(&self) -> SyncEngineSnapshot {
        lock(&self.engine).cancel()
    }

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

fn lock<T>(mutex: &Mutex<T>) -> MutexGuard<'_, T> {
    match mutex.lock() {
        Ok(guard) => guard,
        Err(poisoned) => poisoned.into_inner(),
    }
}

fn open_account_paths() -> &'static Mutex<BTreeSet<PathBuf>> {
    static OPEN_ACCOUNT_PATHS: OnceLock<Mutex<BTreeSet<PathBuf>>> = OnceLock::new();
    OPEN_ACCOUNT_PATHS.get_or_init(|| Mutex::new(BTreeSet::new()))
}

fn derived_account_path(
    database_directory: &str,
    public_key: &str,
) -> Result<PathBuf, SoftchatError> {
    if database_directory.is_empty() || database_directory.len() > 4_096 {
        return Err(SoftchatError::InvalidPersistenceResult);
    }
    let directory = Path::new(database_directory)
        .canonicalize()
        .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
    if !directory.is_dir() {
        return Err(SoftchatError::InvalidPersistenceResult);
    }
    Ok(directory.join(account_database_filename(public_key)?))
}

fn account_database_filename(public_key: &str) -> Result<String, SoftchatError> {
    let public_key_bytes =
        hex::decode(public_key).map_err(|_| SoftchatError::InvalidPersistenceResult)?;
    let digest = hex::encode(Sha256::digest(public_key_bytes));
    Ok(format!("softchat-account-{digest}.sqlite3"))
}

pub(crate) fn resolve_sync_actions(
    database: &mut AccountDatabase,
    engine: &mut SyncEngine,
    actions: Vec<SyncAction>,
    frame_size_limit: u64,
) -> Result<Vec<SyncAction>, SoftchatError> {
    actions
        .into_iter()
        .map(|action| resolve_sync_action(database, engine, action, frame_size_limit))
        .collect()
}

pub(crate) fn resolve_sync_action(
    database: &mut AccountDatabase,
    engine: &mut SyncEngine,
    action: SyncAction,
    frame_size_limit: u64,
) -> Result<SyncAction, SoftchatError> {
    match action.kind {
        SyncActionKind::LoadSnapshot => {
            let since = (action.since_timestamp >= 0).then_some(action.since_timestamp);
            let items = database.sync_fingerprints(since, action.until_timestamp, action.limit)?;
            engine.accept_snapshot(items, frame_size_limit)
        }
        SyncActionKind::SaveCheckpoint => {
            database.save_sync_checkpoint(
                &engine.snapshot().sync_id,
                action.checkpoint_timestamp,
                action.checkpoint_timestamp,
            )?;
            engine.checkpoint_saved(action.checkpoint_timestamp)
        }
        _ => Ok(action),
    }
}

pub(crate) fn plan_sync_transport(
    mut action: SyncAction,
    recipient_public_key: &str,
) -> Result<SyncAction, SoftchatError> {
    NostrPublicKey::from_hex(recipient_public_key).map_err(|_| SoftchatError::InvalidSyncState)?;
    match action.kind {
        SyncActionKind::SendNegOpen => {
            if action.subscription_id.is_empty()
                || action.until_timestamp < 0
                || action.message.is_empty()
            {
                return Err(SoftchatError::InvalidSyncState);
            }
            let mut filter = serde_json::Map::new();
            filter.insert(
                "kinds".to_owned(),
                serde_json::json!([
                    NostrEventKind::GIFT_WRAP.as_u16(),
                    NostrEventKind::EPHEMERAL_GIFT_WRAP.as_u16()
                ]),
            );
            filter.insert("#p".to_owned(), serde_json::json!([recipient_public_key]));
            if action.since_timestamp >= 0 {
                filter.insert(
                    "since".to_owned(),
                    serde_json::json!(action.since_timestamp),
                );
            }
            filter.insert(
                "until".to_owned(),
                serde_json::json!(action.until_timestamp),
            );
            action.frame_json = serde_json::to_string(&serde_json::json!([
                "NEG-OPEN",
                action.subscription_id,
                serde_json::Value::Object(filter),
                hex::encode(&action.message)
            ]))
            .map_err(|_| SoftchatError::InternalFailure)?;
        }
        SyncActionKind::SendNegMessage => {
            if action.subscription_id.is_empty() || action.message.is_empty() {
                return Err(SoftchatError::InvalidSyncState);
            }
            action.frame_json = serde_json::to_string(&serde_json::json!([
                "NEG-MSG",
                action.subscription_id,
                hex::encode(&action.message)
            ]))
            .map_err(|_| SoftchatError::InternalFailure)?;
        }
        SyncActionKind::SendNegClose => {
            if action.subscription_id.is_empty() {
                return Err(SoftchatError::InvalidSyncState);
            }
            action.frame_json =
                serde_json::to_string(&serde_json::json!(["NEG-CLOSE", action.subscription_id]))
                    .map_err(|_| SoftchatError::InternalFailure)?;
        }
        SyncActionKind::RequestEvents => {
            if action.subscription_id.is_empty() || action.event_ids.is_empty() {
                return Err(SoftchatError::InvalidSyncState);
            }
            action.filters_json.push(
                serde_json::to_string(&serde_json::json!({ "ids": action.event_ids }))
                    .map_err(|_| SoftchatError::InternalFailure)?,
            );
        }
        SyncActionKind::LoadSnapshot
        | SyncActionKind::SaveCheckpoint
        | SyncActionKind::ResendEvents
        | SyncActionKind::Complete => {}
    }
    Ok(action)
}

pub(crate) fn inbox_filter_json(
    recipient_public_key: &str,
    since_timestamp: Option<i64>,
    until_timestamp: Option<i64>,
) -> Result<String, SoftchatError> {
    NostrPublicKey::from_hex(recipient_public_key)
        .map_err(|_| SoftchatError::InvalidRelayFilter)?;
    let since = since_timestamp
        .map(|value| u64::try_from(value).map_err(|_| SoftchatError::InvalidEventTimestamp))
        .transpose()?;
    let until = until_timestamp
        .map(|value| u64::try_from(value).map_err(|_| SoftchatError::InvalidEventTimestamp))
        .transpose()?;
    if since
        .into_iter()
        .chain(until)
        .any(|value| value > crate::MAX_PORTABLE_TIMESTAMP_SECONDS)
    {
        return Err(SoftchatError::InvalidEventTimestamp);
    }
    let mut generic_tags = BTreeMap::new();
    generic_tags.insert("#p".to_owned(), vec![recipient_public_key.to_owned()]);
    RelayFilter {
        kinds: vec![
            NostrEventKind::GIFT_WRAP.as_u16(),
            NostrEventKind::EPHEMERAL_GIFT_WRAP.as_u16(),
        ],
        since,
        until,
        generic_tags,
        ..RelayFilter::default()
    }
    .to_json()
}

#[derive(Clone, Debug, Eq, PartialEq)]
pub(crate) enum SyncTransportFrame {
    Message {
        subscription_id: String,
        message: Vec<u8>,
    },
    Error {
        subscription_id: String,
        max_records: Option<u64>,
    },
    Control {
        subscription_id: String,
    },
}

pub(crate) fn parse_sync_message_frame(
    frame_json: &str,
) -> Result<Option<SyncTransportFrame>, SoftchatError> {
    if frame_json.len() > crate::MAX_RELAY_FRAME_BYTES {
        return Err(SoftchatError::InvalidSyncState);
    }
    let Ok(serde_json::Value::Array(values)) =
        serde_json::from_str::<serde_json::Value>(frame_json)
    else {
        return Ok(None);
    };
    let Some(kind) = values.first().and_then(serde_json::Value::as_str) else {
        return Ok(None);
    };
    match kind {
        "NEG-MSG" => {
            if values.len() != 3 {
                return Err(SoftchatError::InvalidSyncState);
            }
            let subscription_id = values
                .get(1)
                .and_then(serde_json::Value::as_str)
                .ok_or(SoftchatError::InvalidSyncState)?
                .to_owned();
            let message_hex = values
                .get(2)
                .and_then(serde_json::Value::as_str)
                .ok_or(SoftchatError::InvalidSyncState)?;
            if message_hex.is_empty() || message_hex.len() > crate::MAX_RELAY_FRAME_BYTES {
                return Err(SoftchatError::InvalidSyncState);
            }
            let message = hex::decode(message_hex).map_err(|_| SoftchatError::InvalidSyncState)?;
            Ok(Some(SyncTransportFrame::Message {
                subscription_id,
                message,
            }))
        }
        "NEG-ERR" => {
            if !matches!(values.len(), 3 | 4) {
                return Err(SoftchatError::InvalidSyncState);
            }
            let subscription_id = values
                .get(1)
                .and_then(serde_json::Value::as_str)
                .filter(|value| !value.is_empty() && value.len() <= crate::MAX_RELAY_FRAME_BYTES)
                .ok_or(SoftchatError::InvalidSyncState)?
                .to_owned();
            values
                .get(2)
                .and_then(serde_json::Value::as_str)
                .filter(|value| value.len() <= crate::MAX_RELAY_FRAME_BYTES)
                .ok_or(SoftchatError::InvalidSyncState)?;
            let max_records = values
                .get(3)
                .map(|value| value.as_u64().ok_or(SoftchatError::InvalidSyncState))
                .transpose()?;
            Ok(Some(SyncTransportFrame::Error {
                subscription_id,
                max_records,
            }))
        }
        "NEG-OPEN" | "NEG-CLOSE" => {
            let expected_len = if kind == "NEG-OPEN" { 4 } else { 2 };
            if values.len() != expected_len {
                return Err(SoftchatError::InvalidSyncState);
            }
            let subscription_id = values
                .get(1)
                .and_then(serde_json::Value::as_str)
                .filter(|value| !value.is_empty() && value.len() <= crate::MAX_RELAY_FRAME_BYTES)
                .ok_or(SoftchatError::InvalidSyncState)?
                .to_owned();
            Ok(Some(SyncTransportFrame::Control { subscription_id }))
        }
        _ => Ok(None),
    }
}

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

    #[test]
    fn ingestion_json_preflight_checks_count_bytes_and_overflow_before_parsing() {
        assert!(matches!(
            preflight_ingestion_jsons(&[]),
            Err(SoftchatError::InvalidAccountOperation)
        ));
        assert!(matches!(
            preflight_ingestion_jsons(&vec![
                "not-json".to_owned();
                crate::MAX_ACCOUNT_INGESTION_EVENTS + 1
            ]),
            Err(SoftchatError::InvalidAccountOperation)
        ));
        assert_eq!(
            bounded_ingestion_json_bytes([crate::MAX_ACCOUNT_INGESTION_BYTES]),
            Ok(crate::MAX_ACCOUNT_INGESTION_BYTES as u64)
        );
        assert!(matches!(
            bounded_ingestion_json_bytes([crate::MAX_ACCOUNT_INGESTION_BYTES, 1]),
            Err(SoftchatError::InvalidAccountOperation)
        ));
        assert!(matches!(
            bounded_ingestion_json_bytes([crate::MAX_ACCOUNT_INGESTION_BYTES, usize::MAX]),
            Err(SoftchatError::InvalidAccountOperation)
        ));
    }

    #[test]
    fn diagnostic_ring_is_bounded_ordered_redacted_and_thresholded() -> Result<(), SoftchatError> {
        let logs = AccountLogBuffer::new(AccountLogLevel::Warn);
        logs.record(
            AccountLogLevel::Info,
            AccountLogCategory::Messaging,
            "message.send",
            None,
            None,
            AccountLogMetadata::EMPTY,
        );
        for index in 0..260 {
            logs.record(
                AccountLogLevel::Error,
                AccountLogCategory::Storage,
                "storage.ingest",
                None,
                Some(SoftchatError::InvalidEventJson),
                AccountLogMetadata {
                    revision: Some(i64::from(index)),
                    item_count: Some(1),
                    byte_count: Some(128),
                },
            );
        }

        let first = logs.drain(MAX_ACCOUNT_LOG_BATCH)?;
        assert_eq!(first.dropped_count, 4);
        assert_eq!(first.events.len(), MAX_ACCOUNT_LOG_BATCH as usize);
        assert_eq!(first.events.first().map(|event| event.sequence), Some(5));
        assert_eq!(first.events.last().map(|event| event.sequence), Some(68));
        assert!(first.events.iter().all(|event| {
            event.operation == "storage.ingest"
                && event.error_code == "invalid_event_json"
                && event.item_count == Some(1)
                && event.byte_count == Some(128)
        }));

        let second = logs.drain(1)?;
        assert_eq!(second.dropped_count, 0);
        assert_eq!(second.events[0].sequence, 69);
        assert_eq!(logs.drain(0), Err(SoftchatError::InvalidAccountOperation));
        assert_eq!(
            logs.drain(MAX_ACCOUNT_LOG_BATCH + 1),
            Err(SoftchatError::InvalidAccountOperation)
        );
        Ok(())
    }

    #[test]
    fn consecutive_equivalent_diagnostics_are_counted_once() -> Result<(), SoftchatError> {
        let logs = AccountLogBuffer::new(AccountLogLevel::Info);
        for _ in 0..3 {
            logs.record(
                AccountLogLevel::Info,
                AccountLogCategory::Profile,
                "profile.follow",
                None,
                None,
                AccountLogMetadata {
                    revision: Some(7),
                    item_count: Some(1),
                    byte_count: None,
                },
            );
        }

        let batch = logs.drain(1)?;
        assert_eq!(batch.events.len(), 1);
        assert_eq!(batch.events[0].sequence, 3);
        assert_eq!(batch.events[0].occurrence_count, 3);
        assert_eq!(batch.events[0].operation, "profile.follow");
        Ok(())
    }

    #[test]
    fn disabled_diagnostics_do_not_retain_events() -> Result<(), SoftchatError> {
        let logs = AccountLogBuffer::new(AccountLogLevel::Off);
        logs.record(
            AccountLogLevel::Error,
            AccountLogCategory::Lifecycle,
            "account.open",
            None,
            Some(SoftchatError::InternalFailure),
            AccountLogMetadata::EMPTY,
        );
        assert_eq!(
            logs.drain(1)?,
            AccountLogBatch {
                events: Vec::new(),
                dropped_count: 0,
            }
        );
        Ok(())
    }
}

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

    const RECIPIENT: &str = "4b22aa260e4acb7021e32f38a6cdf4b673c6a277755bfce287e370c924dc936d";

    fn action(kind: SyncActionKind) -> SyncAction {
        SyncAction {
            kind,
            subscription_id: "sync-1".to_owned(),
            since_timestamp: 10,
            until_timestamp: 20,
            limit: 500,
            message: vec![1, 2, 3],
            frame_json: String::new(),
            filters_json: Vec::new(),
            event_ids: Vec::new(),
            checkpoint_timestamp: -1,
        }
    }

    #[test]
    fn sync_transport_plans_complete_frames_and_filters() -> Result<(), Box<dyn std::error::Error>>
    {
        let open = plan_sync_transport(action(SyncActionKind::SendNegOpen), RECIPIENT)?;
        let open_json = serde_json::from_str::<serde_json::Value>(&open.frame_json)?;
        assert_eq!(open_json[0], "NEG-OPEN");
        assert_eq!(open_json[2]["kinds"], serde_json::json!([1059, 21059]));
        assert_eq!(open_json[2]["#p"], serde_json::json!([RECIPIENT]));
        assert_eq!(open_json[2]["since"], 10);
        assert_eq!(open_json[2]["until"], 20);

        let mut request = action(SyncActionKind::RequestEvents);
        request.message.clear();
        request.event_ids = vec![hex::encode([7_u8; 32])];
        let request = plan_sync_transport(request, RECIPIENT)?;
        assert_eq!(request.filters_json.len(), 1);
        let filter_json = serde_json::from_str::<serde_json::Value>(&request.filters_json[0])?;
        assert_eq!(filter_json["ids"][0], hex::encode([7_u8; 32]));
        Ok(())
    }

    #[test]
    fn inbox_filter_is_canonical_bounded_and_account_addressed()
    -> Result<(), Box<dyn std::error::Error>> {
        let filter = inbox_filter_json(RECIPIENT, Some(10), Some(20))?;
        assert_eq!(
            filter,
            format!(r##"{{"kinds":[1059,21059],"since":10,"until":20,"#p":["{RECIPIENT}"]}}"##)
        );
        assert!(inbox_filter_json(RECIPIENT, Some(-1), None).is_err());
        let maximum_timestamp = i64::try_from(crate::MAX_PORTABLE_TIMESTAMP_SECONDS)?;
        assert!(inbox_filter_json(RECIPIENT, Some(maximum_timestamp), None).is_ok());
        Ok(())
    }

    #[test]
    fn sync_frame_router_accepts_only_bounded_correlated_shapes()
    -> Result<(), Box<dyn std::error::Error>> {
        assert_eq!(
            parse_sync_message_frame(r#"["NEG-MSG","sync-1","010203"]"#)?,
            Some(SyncTransportFrame::Message {
                subscription_id: "sync-1".to_owned(),
                message: vec![1, 2, 3],
            })
        );
        assert!(parse_sync_message_frame(r#"["EVENT","sync-1",{}]"#)?.is_none());
        assert!(parse_sync_message_frame(r#"["NEG-MSG","sync-1","zz"]"#).is_err());
        assert_eq!(
            parse_sync_message_frame(r#"["NEG-ERR","sync-1","failure",500]"#)?,
            Some(SyncTransportFrame::Error {
                subscription_id: "sync-1".to_owned(),
                max_records: Some(500),
            })
        );
        assert_eq!(
            parse_sync_message_frame(r#"["NEG-CLOSE","sync-1"]"#)?,
            Some(SyncTransportFrame::Control {
                subscription_id: "sync-1".to_owned(),
            })
        );
        Ok(())
    }
}

#[cfg(test)]
mod product_tests {
    use super::*;
    use crate::{
        AccountTransportAction, AccountTransportActionKind, AccountTransportResultKind,
        RelayConnectionState, RelayResponseFrame, SyncPhase,
    };

    const ALICE_SECRET: &str = "5c0c523f52a5b6fad39ed2403092df8cebc36318b39383bca6c00808626fab3a";
    const BOB_SECRET: &str = "4b22aa260e4acb7021e32f38a6cdf4b673c6a277755bfce287e370c924dc936d";

    fn secret(value: &str) -> Result<Vec<u8>, SoftchatError> {
        hex::decode(value).map_err(|_| SoftchatError::InvalidSecretKey)
    }

    fn temporary_directory(label: &str) -> Result<PathBuf, Box<dyn std::error::Error>> {
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)?
            .as_nanos();
        let path =
            std::env::temp_dir().join(format!("softchat-{label}-{}-{nonce}", std::process::id()));
        std::fs::create_dir_all(&path)?;
        Ok(path)
    }

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

    fn neg_subscription_id(
        batch: &AccountTransportBatch,
    ) -> Result<String, Box<dyn std::error::Error>> {
        for action in &batch.actions {
            if action.kind != AccountTransportActionKind::SendText {
                continue;
            }
            let value = serde_json::from_str::<serde_json::Value>(&action.text_frame)?;
            if value.get(0).and_then(serde_json::Value::as_str) == Some("NEG-OPEN") {
                return value
                    .get(1)
                    .and_then(serde_json::Value::as_str)
                    .map(ToOwned::to_owned)
                    .ok_or_else(|| SoftchatError::InvalidSyncState.into());
            }
        }
        Err(SoftchatError::InvalidSyncState.into())
    }

    fn delivery_send_actions(
        batch: &AccountTransportBatch,
    ) -> Result<Vec<(AccountTransportAction, Vec<String>)>, SoftchatError> {
        let mut deliveries = Vec::new();
        for action in &batch.actions {
            if action.kind != AccountTransportActionKind::SendText {
                continue;
            }
            let event_ids = match crate::parse_client_relay_frame(&action.text_frame) {
                Ok(crate::ClientRelayFrame::Event(event)) => vec![event.id().to_hex()],
                Ok(crate::ClientRelayFrame::Events(events)) => {
                    events.iter().map(|event| event.id().to_hex()).collect()
                }
                Ok(_) | Err(_) => Vec::new(),
            };
            if !event_ids.is_empty() {
                deliveries.push((action.clone(), event_ids));
            }
        }
        Ok(deliveries)
    }

    fn content(text: &str) -> MessageContentInput {
        MessageContentInput {
            text: text.to_owned(),
            attachments: Vec::new(),
            emoji_tags: Vec::new(),
        }
    }

    #[test]
    fn draftless_command_hashes_preserve_the_released_shapes()
    -> Result<(), Box<dyn std::error::Error>> {
        let conversation_id = "ab".repeat(32);
        let message = content("legacy command");
        let reply_to_message_id = String::new();
        let sources = vec![crate::MediaPreparationInput {
            source_fingerprint: "legacy-source".to_owned(),
        }];

        let legacy_send =
            command_hash(&("send", &conversation_id, &message, &reply_to_message_id))?;
        assert_eq!(
            send_message_command_hash(&conversation_id, &message, &reply_to_message_id, None,)?,
            legacy_send,
        );

        let legacy_media = command_hash(&(
            "media-message",
            &conversation_id,
            &message,
            &reply_to_message_id,
            &sources,
        ))?;
        assert_eq!(
            media_message_command_hash(
                &conversation_id,
                &message,
                &reply_to_message_id,
                &sources,
                None,
            )?,
            legacy_media,
        );

        let draft_match = product_draft_match(AccountDraft {
            text: "legacy command".to_owned(),
            emoji_tags: Vec::new(),
            spans: Vec::new(),
            reply_to_message_id: String::new(),
            attachment_operation_ids: Vec::new(),
            updated_at: 1_700_000_000,
        })?;
        let draft_send = send_message_command_hash(
            &conversation_id,
            &message,
            &reply_to_message_id,
            Some(&draft_match),
        );
        assert_eq!(
            draft_send?,
            command_hash(&(
                "send",
                &conversation_id,
                &message,
                &reply_to_message_id,
                &Some(draft_match.clone()),
            ))?,
        );
        assert_ne!(
            send_message_command_hash(
                &conversation_id,
                &message,
                &reply_to_message_id,
                Some(&draft_match),
            )?,
            legacy_send,
        );
        assert_eq!(
            media_message_command_hash(
                &conversation_id,
                &message,
                &reply_to_message_id,
                &sources,
                Some(&draft_match),
            )?,
            command_hash(&(
                "media-message",
                &conversation_id,
                &message,
                &reply_to_message_id,
                &sources,
                &Some(draft_match.clone()),
            ))?,
        );
        assert_ne!(
            media_message_command_hash(
                &conversation_id,
                &message,
                &reply_to_message_id,
                &sources,
                Some(&draft_match),
            )?,
            legacy_media,
        );
        Ok(())
    }

    #[test]
    fn delete_message_rejects_an_oversized_id_before_command_work()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("delete-message-bound")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;

        assert!(matches!(
            account.delete_message(
                "delete-message-oversized-id".to_owned(),
                "a".repeat(1_000_000),
                1_700_000_000,
            ),
            Err(SoftchatError::InvalidAccountOperation)
        ));

        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn product_message_preserves_a_sticker_name_with_spaces()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("sticker-message")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
        let emoji_tags = vec![vec![
            "emoji".to_owned(),
            "Device sticker".to_owned(),
            "https://cdn.example/device-sticker.webp".to_owned(),
        ]];

        let sent = account.send_message(
            "sticker-message".to_owned(),
            conversation.id,
            MessageContentInput {
                text: ":Device sticker:".to_owned(),
                attachments: Vec::new(),
                emoji_tags: emoji_tags.clone(),
            },
            String::new(),
            None,
            1_700_000_001,
        )?;

        let sender_copy = sent
            .message
            .ok_or_else(|| std::io::Error::other("missing stored sender copy"))?;
        assert_eq!(sender_copy.emoji_tags, emoji_tags);
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn signed_outbox_recovery_is_bounded_verified_and_idempotent()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("signed-outbox-recovery")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let identity = crate::LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let event = identity.sign_event(crate::NostrEventDraft::new(
            1_700_000_000,
            NostrEventKind::SHORT_TEXT_NOTE,
            Vec::new(),
            "recover me",
        )?)?;
        let event_value = serde_json::from_str::<serde_json::Value>(&event.to_json()?)?;
        let batch = serde_json::to_string(&vec![event_value.clone()])?;

        let inserted = account.recover_signed_outbox_batch(
            "legacy-batch-1".to_owned(),
            batch.clone(),
            1_700_000_001,
        )?;
        assert!(inserted.inserted);
        let repeated = account.recover_signed_outbox_batch(
            "legacy-batch-1".to_owned(),
            batch,
            1_700_000_001,
        )?;
        assert!(!repeated.inserted);
        assert_eq!(inserted.operation, repeated.operation);

        let mut forged = event_value;
        forged["content"] = serde_json::Value::String("forged".to_owned());
        assert!(matches!(
            account.recover_signed_outbox_batch(
                "legacy-batch-forged".to_owned(),
                serde_json::to_string(&vec![forged])?,
                1_700_000_001,
            ),
            Err(SoftchatError::InvalidEventId)
        ));
        let second = identity.sign_event(crate::NostrEventDraft::new(
            1_700_000_001,
            NostrEventKind::SHORT_TEXT_NOTE,
            Vec::new(),
            "different recovery",
        )?)?;
        assert!(matches!(
            account.recover_signed_outbox_batch(
                "legacy-batch-1".to_owned(),
                serde_json::to_string(&vec![serde_json::from_str::<serde_json::Value>(
                    &second.to_json()?
                )?])?,
                1_700_000_002,
            ),
            Err(SoftchatError::CommandConflict)
        ));
        assert!(matches!(
            account.recover_signed_outbox_batch(
                "legacy-batch-oversized".to_owned(),
                " ".repeat(crate::MAX_NOSTR_EVENT_JSON_BYTES + 1),
                1_700_000_001,
            ),
            Err(SoftchatError::InvalidAccountOperation)
        ));

        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn account_derives_path_rejects_duplicate_open_and_erases_on_close()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("account-open")?;
        let first =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        assert_eq!(first.account_id(), first.public_key()?);
        assert!(first.database_filename().starts_with("softchat-account-"));
        assert!(first.database_filename().ends_with(".sqlite3"));
        assert!(matches!(
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string()),
            Err(SoftchatError::AccountAlreadyOpen)
        ));
        first.close_account();
        assert!(matches!(
            first.public_key(),
            Err(SoftchatError::AccountClosed)
        ));
        drop(first);
        let reopened =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        drop(reopened);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn send_is_payload_idempotent_and_returns_the_original_logical_message()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("command-idempotency")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;

        let first = account.send_message(
            "command-send-1".to_owned(),
            conversation.id.clone(),
            content("one durable message"),
            String::new(),
            None,
            1_700_000_001,
        )?;
        let retried = account.send_message(
            "command-send-1".to_owned(),
            conversation.id.clone(),
            content("one durable message"),
            String::new(),
            None,
            1_700_000_999,
        )?;
        assert!(first.inserted);
        assert!(!retried.inserted);
        assert_eq!(first.operation, retried.operation);
        assert_eq!(first.message, retried.message);
        assert_eq!(
            first.message.as_ref().map(|message| message.text.as_str()),
            Some("one durable message")
        );
        assert!(matches!(
            account.send_message(
                "command-send-1".to_owned(),
                conversation.id,
                content("different payload"),
                String::new(),
                None,
                1_700_001_000,
            ),
            Err(SoftchatError::CommandConflict)
        ));
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn sending_can_consume_only_the_draft_owned_by_the_new_command()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("send-consumes-draft")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
        let first_draft = AccountDraft {
            text: "send this".to_owned(),
            emoji_tags: Vec::new(),
            spans: Vec::new(),
            reply_to_message_id: String::new(),
            attachment_operation_ids: Vec::new(),
            updated_at: 1_700_000_001,
        };
        account.save_conversation_draft(
            conversation.id.clone(),
            Some(first_draft.clone()),
            1_700_000_001,
        )?;

        let sent = account.send_message(
            "send-consumes-draft-command".to_owned(),
            conversation.id.clone(),
            content("send this"),
            String::new(),
            Some(first_draft.clone()),
            1_700_000_002,
        )?;
        assert!(sent.inserted);
        assert!(
            account
                .conversation(conversation.id.clone())?
                .ok_or(SoftchatError::InvalidPersistenceResult)?
                .draft
                .is_none()
        );

        let newer_draft = AccountDraft {
            text: "typed after send".to_owned(),
            emoji_tags: Vec::new(),
            spans: Vec::new(),
            reply_to_message_id: String::new(),
            attachment_operation_ids: Vec::new(),
            updated_at: 1_700_000_003,
        };
        account.save_conversation_draft(
            conversation.id.clone(),
            Some(newer_draft.clone()),
            1_700_000_003,
        )?;
        let raced = account.send_message(
            "send-preserves-raced-draft-command".to_owned(),
            conversation.id.clone(),
            content("send this"),
            String::new(),
            Some(first_draft.clone()),
            1_700_000_004,
        )?;
        assert!(raced.inserted);
        assert_eq!(
            account
                .conversation(conversation.id.clone())?
                .ok_or(SoftchatError::InvalidPersistenceResult)?
                .draft,
            Some(newer_draft.clone())
        );
        let replay = account.send_message(
            "send-consumes-draft-command".to_owned(),
            conversation.id.clone(),
            content("send this"),
            String::new(),
            Some(first_draft),
            1_700_000_005,
        )?;
        assert!(!replay.inserted);
        assert_eq!(
            account
                .conversation(conversation.id.clone())?
                .ok_or(SoftchatError::InvalidPersistenceResult)?
                .draft,
            Some(newer_draft)
        );
        assert!(matches!(
            account.send_message(
                "send-consumes-draft-command".to_owned(),
                conversation.id,
                content("send this"),
                String::new(),
                None,
                1_700_000_006,
            ),
            Err(SoftchatError::CommandConflict)
        ));

        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn media_preparation_clears_only_the_matching_draft_snapshot()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("media-message-draft-race")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
        let expected = AccountDraft {
            text: "attachment".to_owned(),
            emoji_tags: Vec::new(),
            spans: Vec::new(),
            reply_to_message_id: String::new(),
            attachment_operation_ids: Vec::new(),
            updated_at: 1_700_000_001,
        };
        let newer = AccountDraft {
            text: "typed while staging".to_owned(),
            updated_at: 1_700_000_002,
            ..expected.clone()
        };
        account.save_conversation_draft(
            conversation.id.clone(),
            Some(newer.clone()),
            1_700_000_002,
        )?;
        let prepared = account.prepare_media_message(
            "media-message-draft-race-command".to_owned(),
            conversation.id.clone(),
            content("attachment"),
            String::new(),
            vec![crate::MediaPreparationInput {
                source_fingerprint: "media-message-draft-race-source".to_owned(),
            }],
            Some(expected),
            1_700_000_003,
        )?;
        assert_eq!(
            prepared.state,
            crate::PendingMediaMessageState::WaitingForUploads
        );
        assert_eq!(
            account
                .conversation(conversation.id)?
                .ok_or(SoftchatError::InvalidPersistenceResult)?
                .draft,
            Some(newer)
        );

        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn delete_many_is_atomic_bounded_and_retry_idempotent() -> Result<(), Box<dyn std::error::Error>>
    {
        let directory = temporary_directory("delete-many")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
        let first = account
            .send_message(
                "delete-many-first".to_owned(),
                conversation.id.clone(),
                content("first"),
                String::new(),
                None,
                1_700_000_001,
            )?
            .message
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        let second = account
            .send_message(
                "delete-many-second".to_owned(),
                conversation.id.clone(),
                content("second"),
                String::new(),
                None,
                1_700_000_002,
            )?
            .message
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        let targets = vec![first.id.clone(), second.id.clone()];

        assert!(matches!(
            account.delete_messages(
                "delete-many-oversized-id".to_owned(),
                vec!["a".repeat(1024 * 1024)],
                1_700_000_003,
            ),
            Err(SoftchatError::InvalidAccountOperation)
        ));
        assert!(matches!(
            account.delete_messages(
                "delete-many-malformed-id".to_owned(),
                vec!["z".repeat(64)],
                1_700_000_003,
            ),
            Err(SoftchatError::InvalidEventId)
        ));
        assert!(matches!(
            account.delete_messages(
                "delete-many-duplicate".to_owned(),
                vec![first.id.clone(), first.id.clone()],
                1_700_000_003,
            ),
            Err(SoftchatError::InvalidAccountOperation)
        ));
        let deleted = account.delete_messages(
            "delete-many-command".to_owned(),
            targets.clone(),
            1_700_000_004,
        )?;
        assert!(deleted.inserted);
        let retried = account.delete_messages(
            "delete-many-command".to_owned(),
            targets.clone(),
            1_700_000_999,
        )?;
        assert!(!retried.inserted);
        assert_eq!(retried.operation, deleted.operation);
        assert!(matches!(
            account.delete_messages(
                "delete-many-command".to_owned(),
                targets.into_iter().rev().collect(),
                1_700_001_000,
            ),
            Err(SoftchatError::CommandConflict)
        ));
        let page = account.message_page(conversation.id, None, 50)?;
        assert!(
            page.messages
                .iter()
                .filter(|message| message.id == first.id || message.id == second.id)
                .all(|message| message.deleted)
        );

        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn profile_patch_is_lossless_and_uses_the_account_owned_self_copy()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("profile-patch")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        account.update_profile(
            "profile-first".to_owned(),
            r#"{"name":"alice","picture":"https://cdn.example/avatar.png","website":"https://example.com"}"#.to_owned(),
            1_700_000_001,
        )?;
        account.update_profile(
            "profile-second".to_owned(),
            r#"{"about":"updated"}"#.to_owned(),
            1_700_000_002,
        )?;
        let profile = account.profile(account.public_key()?)?;
        assert_eq!(profile.name.as_deref(), Some("alice"));
        assert_eq!(profile.about.as_deref(), Some("updated"));
        assert_eq!(
            profile.picture.as_deref(),
            Some("https://cdn.example/avatar.png")
        );
        assert_eq!(profile.website.as_deref(), Some("https://example.com"));

        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        account.replace_follows(
            "follow-first".to_owned(),
            vec![Contact {
                public_key: bob.public_key().to_hex(),
                relay_hint: String::new(),
                local_name: "Bob".to_owned(),
            }],
            1_700_000_003,
        )?;
        assert!(
            account
                .conversation_page(false, None, 100)?
                .conversations
                .is_empty(),
            "account-data p tags must not create product conversations"
        );
        account.replace_follows("follow-second".to_owned(), Vec::new(), 1_700_000_003)?;
        assert!(account.follows()?.is_empty());
        account.follow(
            "follow-atomic".to_owned(),
            Contact {
                public_key: bob.public_key().to_hex(),
                relay_hint: "wss://relay.example".to_owned(),
                local_name: "Robert".to_owned(),
            },
            1_700_000_003,
        )?;
        assert_eq!(account.follows()?.len(), 1);
        assert_eq!(account.follows()?[0].local_name, "Robert");
        account.unfollow(
            "unfollow-atomic".to_owned(),
            bob.public_key().to_hex(),
            1_700_000_003,
        )?;
        assert!(account.follows()?.is_empty());

        account.apply_account_settings_patch(
            "settings-first".to_owned(),
            r#"{"quickReaction":"one"}"#.to_owned(),
            1_700_000_004,
        )?;
        let first_settings_at = account.database()?.connection.query_row(
            "SELECT event_created_at FROM app_data_effective WHERE context = 'app-settings'",
            [],
            |row| row.get::<_, i64>(0),
        )?;
        account.apply_account_settings_patch(
            "settings-second".to_owned(),
            r#"{"quickReaction":"two"}"#.to_owned(),
            1_700_000_004,
        )?;
        let second_settings_at = account.database()?.connection.query_row(
            "SELECT event_created_at FROM app_data_effective WHERE context = 'app-settings'",
            [],
            |row| row.get::<_, i64>(0),
        )?;
        assert!(second_settings_at > first_settings_at);
        assert_eq!(
            account.account_settings()?.quick_reaction.as_deref(),
            Some("two")
        );

        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_005)?;
        let message = account
            .send_message(
                "read-state-message".to_owned(),
                conversation.id.clone(),
                content("read state"),
                String::new(),
                None,
                1_700_000_006,
            )?
            .message
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        account.mark_conversation_read(
            "mark-read-first".to_owned(),
            conversation.id.clone(),
            MessageCursor {
                created_at: message.created_at,
                message_id: message.id,
            },
            1_700_000_007,
        )?;
        let read_at = account.account_read_state()?[0].updated_at;
        account.mark_conversation_unread(
            "mark-unread-second".to_owned(),
            conversation.id,
            1_700_000_007,
        )?;
        let unread = account.account_read_state()?.remove(0);
        assert!(unread.forced_unread);
        assert!(unread.updated_at > read_at);
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn message_tuple_cursor_is_lossless_for_equal_timestamps()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("message-cursor")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
        for (command, text) in [("equal-time-1", "alpha"), ("equal-time-2", "beta")] {
            account.send_message(
                command.to_owned(),
                conversation.id.clone(),
                content(text),
                String::new(),
                None,
                1_700_000_001,
            )?;
        }
        let first = account.message_page(conversation.id.clone(), None, 1)?;
        let second = account.message_page(conversation.id, first.next_cursor.clone(), 1)?;
        assert_eq!(first.messages.len(), 1);
        assert_eq!(second.messages.len(), 1);
        assert_ne!(first.messages[0].id, second.messages[0].id);
        assert_eq!(first.messages[0].created_at, second.messages[0].created_at);
        assert!(second.next_cursor.is_none());
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn product_pages_batch_complete_effective_state() -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("batched-product-pages")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let alice = crate::LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let carol = crate::LocalIdentity::from_secret_hex(
            "0000000000000000000000000000000000000000000000000000000000000002",
        )?;
        let conversation = account.get_or_create_conversation(
            vec![bob.public_key().to_hex(), carol.public_key().to_hex()],
            1_700_000_000,
        )?;
        let first = account
            .send_message(
                "batch-first".to_owned(),
                conversation.id.clone(),
                content("first"),
                String::new(),
                None,
                1_700_000_001,
            )?
            .message
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        let second = account
            .send_message(
                "batch-second".to_owned(),
                conversation.id.clone(),
                content("second"),
                String::new(),
                None,
                1_700_000_002,
            )?
            .message
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        account.edit_message(
            "batch-edit".to_owned(),
            first.id.clone(),
            content("edited"),
            1_700_000_003,
        )?;
        let reaction = bob.create_reaction(ReactionDraft {
            participants: vec![alice.public_key(), carol.public_key()],
            parent_id: NostrEventId::from_hex(&first.id)?,
            parent_author: alice.public_key(),
            parent_kind: NostrEventKind::PRIVATE_DIRECT_MESSAGE,
            reaction: "👍".to_owned(),
            custom_emoji_url: None,
            created_at: 1_700_000_004,
        })?;
        let reaction_rumor = crate::NostrRumor::try_from(reaction.rumor)?;
        let reaction_wrap = bob.gift_wrap(
            &alice.public_key(),
            &reaction_rumor,
            Nip59EnvelopeKind::Durable,
            1_700_000_004,
        )?;
        account.ingest(vec![SignedEvent::from(&reaction_wrap)], 1_700_000_004)?;
        account.delete_message("batch-delete".to_owned(), second.id.clone(), 1_700_000_005)?;
        account.set_conversation_subject(
            "batch-subject".to_owned(),
            conversation.id.clone(),
            "Batched".to_owned(),
            None,
            Vec::new(),
            1_700_000_006,
        )?;
        account.update_profile(
            "batch-profile".to_owned(),
            r#"{"name":"alice"}"#.to_owned(),
            1_700_000_007,
        )?;

        let messages = account.message_page(conversation.id.clone(), None, 50)?;
        let effective_first = messages
            .messages
            .iter()
            .find(|message| message.id == first.id)
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        assert_eq!(effective_first.text, "edited");
        assert!(effective_first.edited);
        assert_eq!(effective_first.reactions.len(), 1);
        assert_eq!(effective_first.reactions[0].value, "👍");
        assert!(
            messages
                .messages
                .iter()
                .any(|message| message.id == second.id && message.deleted)
        );

        // Exact lookup shares the effective projection, independently of the page window.
        let exact = account.messages_by_ids(vec![
            first.id.to_uppercase(),
            "00".repeat(32),
            second.id.clone(),
        ])?;
        assert_eq!(exact.len(), 2);
        assert_eq!(&exact[0], effective_first);
        assert_eq!(exact[1].id, second.id);
        assert!(exact[1].deleted);
        assert!(account.messages_by_ids(Vec::new())?.is_empty());
        assert!(matches!(
            account.messages_by_ids(vec![first.id.clone(), first.id.to_uppercase()]),
            Err(SoftchatError::InvalidAccountOperation)
        ));
        let maximum_ids = (0..crate::MAX_PRODUCT_MESSAGE_LOOKUPS)
            .map(|index| format!("{index:064x}"))
            .collect::<Vec<_>>();
        assert!(account.messages_by_ids(maximum_ids)?.is_empty());
        assert!(matches!(
            account.messages_by_ids(vec![String::new(); crate::MAX_PRODUCT_MESSAGE_LOOKUPS + 1]),
            Err(SoftchatError::InvalidAccountOperation)
        ));
        assert!(account.messages_by_ids(vec!["invalid".to_owned()]).is_err());

        let conversations = account.conversation_page(false, None, 50)?;
        assert_eq!(conversations.conversations.len(), 1);
        assert_eq!(conversations.conversations[0].subject, "Batched");
        assert_eq!(conversations.conversations[0].message_count, 1);
        assert_eq!(
            conversations.conversations[0]
                .last_message
                .as_ref()
                .map(|message| message.id.as_str()),
            Some(first.id.as_str())
        );
        let own_public_key = account.public_key()?;
        assert_eq!(
            account
                .profile_page(50, 0)?
                .into_iter()
                .find(|profile| profile.public_key == own_public_key)
                .and_then(|profile| profile.name),
            Some("alice".to_owned())
        );
        let bob_public_key = bob.public_key().to_hex();
        let exact_profiles =
            account.profiles(vec![bob_public_key.clone(), own_public_key.clone()])?;
        assert_eq!(
            exact_profiles
                .iter()
                .map(|profile| profile.public_key.as_str())
                .collect::<Vec<_>>(),
            vec![bob_public_key.as_str(), own_public_key.as_str()]
        );
        assert!(exact_profiles[0].name.is_none());
        assert_eq!(exact_profiles[1].name.as_deref(), Some("alice"));
        assert!(matches!(
            account.profiles(vec![own_public_key; crate::MAX_PRODUCT_PROFILE_LOOKUPS + 1]),
            Err(SoftchatError::InvalidAccountOperation)
        ));

        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn message_local_extras_are_bounded_revisioned_and_account_local()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("message-extras")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
        let message = account
            .send_message(
                "extras-message".to_owned(),
                conversation.id,
                content("voice note"),
                String::new(),
                None,
                1_700_000_001,
            )?
            .message
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        let before = account.database_info()?.revision;
        let mutation = account.set_message_local_extras(
            message.id.clone(),
            Some("local transcript".to_owned()),
            vec![0, 64, 255],
            1_700_000_002,
        )?;
        assert_eq!(mutation.revision, before + 1);
        let extras = account
            .message_local_extras(message.id.clone())?
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        assert_eq!(extras.transcript.as_deref(), Some("local transcript"));
        assert_eq!(extras.waveform, vec![0, 64, 255]);
        assert!(matches!(
            account.set_message_local_extras(
                message.id,
                None,
                vec![0; crate::MAX_PRODUCT_WAVEFORM_SAMPLES + 1],
                1_700_000_003,
            ),
            Err(SoftchatError::InvalidAccountOperation)
        ));
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn settings_and_media_are_revisioned_and_survive_reopen()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("product-state")?;
        let filename;
        {
            let account =
                AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
            filename = account.database_filename();
            assert_eq!(
                account.media_upload_url()?,
                "https://it2.softchat.c84ad60.xyz/api/v2/files"
            );
            let revision_before_settings = account.database_info()?.revision;
            let settings = account.apply_account_settings_patch(
                "settings-1".to_owned(),
                r#"{"future":{"preserved":true},"mediaService":"https://media.example/upload","quickReaction":"🔥"}"#.to_owned(),
                1_700_000_001,
            )?;
            assert_eq!(settings.settings.quick_reaction.as_deref(), Some("🔥"));
            assert_eq!(account.media_upload_url()?, "https://media.example/upload");
            assert_eq!(settings.revision, revision_before_settings + 1);
            assert!(matches!(
                account.apply_account_settings_patch(
                    "settings-1".to_owned(),
                    r#"{"quickReaction":"different"}"#.to_owned(),
                    1_700_000_002,
                ),
                Err(SoftchatError::CommandConflict)
            ));
            assert_eq!(
                account.account_settings()?.quick_reaction.as_deref(),
                Some("🔥")
            );
            let diagnostics = account.account_diagnostics()?;
            assert_eq!(diagnostics.revision, settings.revision);
            assert!(diagnostics.unfinished_delivery_count > 0);
            let media = account.prepare_media_operation(
                "media-1".to_owned(),
                String::new(),
                "upload".to_owned(),
                "sha256:fixture".to_owned(),
                1_700_000_003,
            )?;
            let other_media = account.prepare_media_operation(
                "media-2".to_owned(),
                String::new(),
                "upload".to_owned(),
                "sha256:other-fixture".to_owned(),
                1_700_000_003,
            )?;
            let caller_prefixed_media = account.prepare_media_operation(
                "conversation-icon-download-caller-owned".to_owned(),
                String::new(),
                "upload".to_owned(),
                "caller-owned-prefix".to_owned(),
                1_700_000_003,
            )?;
            let caller_prefixed_lease = account
                .claim_media_operation(
                    caller_prefixed_media.id,
                    "caller-owned-prefix-lease".to_owned(),
                    1_700_000_004,
                    300,
                )?
                .ok_or(SoftchatError::InvalidMediaOperation)?;
            assert_eq!(
                account
                    .complete_media_operation(
                        caller_prefixed_lease.operation.id,
                        caller_prefixed_lease.lease_id,
                        AttachmentMetadata {
                            url: "https://media.example/caller-owned-prefix.bin".to_owned(),
                            sha256: Some(hex::encode([3_u8; 32])),
                            byte_size: Some(4),
                            ..AttachmentMetadata::default()
                        },
                        4,
                        1_700_000_005,
                    )?
                    .operation
                    .state,
                crate::AccountMediaOperationState::Completed
            );
            assert_eq!(media.state, crate::AccountMediaOperationState::Prepared);
            let lease = account
                .claim_media_operation(
                    media.id.clone(),
                    "media-lease-1".to_owned(),
                    1_700_000_004,
                    300,
                )?
                .ok_or(SoftchatError::InvalidMediaOperation)?;
            assert_eq!(lease.operation.id, media.id);
            assert_eq!(
                account
                    .claim_media_operation(
                        other_media.id.clone(),
                        "media-lease-2".to_owned(),
                        1_700_000_004,
                        300,
                    )?
                    .ok_or(SoftchatError::InvalidMediaOperation)?
                    .operation
                    .id,
                other_media.id
            );
            assert_eq!(
                lease.operation.state,
                crate::AccountMediaOperationState::Running
            );
            assert!(matches!(
                account.complete_media_operation(
                    lease.operation.id.clone(),
                    lease.lease_id.clone(),
                    AttachmentMetadata {
                        url: "http://media.example/insecure".to_owned(),
                        ..AttachmentMetadata::default()
                    },
                    42,
                    1_700_000_005,
                ),
                Err(SoftchatError::InvalidAttachmentMetadata)
            ));
            assert_eq!(
                account
                    .media_operation(lease.operation.id.clone())?
                    .ok_or(SoftchatError::InvalidPersistenceResult)?
                    .state,
                crate::AccountMediaOperationState::Running
            );
            let retryable = account.finish_media_operation_lease(
                lease.operation.id,
                lease.lease_id,
                true,
                "network_unavailable".to_owned(),
                1_700_000_006,
            )?;
            assert_eq!(
                retryable.state,
                crate::AccountMediaOperationState::Retryable
            );
        }
        let reopened =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        assert_eq!(reopened.database_filename(), filename);
        assert_eq!(
            reopened.account_settings()?.quick_reaction.as_deref(),
            Some("🔥")
        );
        assert_eq!(reopened.media_operations(10)?.len(), 3);
        drop(reopened);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn recoverable_media_pagination_does_not_drop_work_beyond_the_recent_limit()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("recoverable-media-pages")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let mut expected_ids = Vec::new();
        for index in 0..(crate::MAX_PRODUCT_QUERY_PAGE + 5) {
            let operation = account.prepare_media_operation(
                format!("recoverable-command-{index}"),
                String::new(),
                "upload".to_owned(),
                format!("recoverable-source-{index}"),
                1_700_000_000 + i64::from(index),
            )?;
            expected_ids.push(operation.id);
        }
        let cancelled = expected_ids.remove(2);
        account.transition_media_operation(
            cancelled.clone(),
            "cancel".to_owned(),
            String::new(),
            1_700_001_000,
        )?;
        expected_ids.sort();

        let first = account.recoverable_media_operations(None, crate::MAX_PRODUCT_QUERY_PAGE)?;
        assert_eq!(
            first.operations.len(),
            crate::MAX_PRODUCT_QUERY_PAGE as usize
        );
        let cursor = first
            .next_cursor
            .clone()
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        let second =
            account.recoverable_media_operations(Some(cursor), crate::MAX_PRODUCT_QUERY_PAGE)?;
        assert_eq!(second.operations.len(), 4);
        assert!(second.next_cursor.is_none());
        let actual_ids = first
            .operations
            .into_iter()
            .chain(second.operations)
            .map(|operation| operation.id)
            .collect::<Vec<_>>();
        assert_eq!(actual_ids, expected_ids);
        let exact_ids = vec![
            expected_ids[expected_ids.len() - 1].clone(),
            cancelled.clone(),
            "0".repeat(64),
            expected_ids[0].clone(),
        ];
        let exact = account.media_operations_by_ids(exact_ids.clone())?;
        assert_eq!(
            exact
                .into_iter()
                .map(|operation| operation.id)
                .collect::<Vec<_>>(),
            vec![exact_ids[0].clone(), cancelled, exact_ids[3].clone()],
        );
        assert!(matches!(
            account.media_operations_by_ids(vec![expected_ids[0].clone(); 2]),
            Err(SoftchatError::InvalidMediaOperation)
        ));
        assert!(matches!(
            account.recoverable_media_operations(Some("not-an-operation".to_owned()), 10),
            Err(SoftchatError::InvalidMediaOperation)
        ));

        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn media_diagnostics_preserve_completion_when_dependent_publication_fails()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("media-commit-diagnostics")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        account.set_log_level(AccountLogLevel::Info)?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
        let pending = account.prepare_media_message(
            "diagnostic-media".to_owned(),
            conversation.id,
            content("private caption"),
            String::new(),
            vec![crate::MediaPreparationInput {
                source_fingerprint: "private-source".to_owned(),
            }],
            None,
            1_700_000_001,
        )?;
        let lease = account
            .claim_media_operation(
                pending.attachments[0].id.clone(),
                "diagnostic-lease".to_owned(),
                1_700_000_002,
                30,
            )?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        account.drain_log_events(64)?;
        account.database()?.connection.execute_batch(
            "CREATE TEMP TRIGGER reject_test_publication BEFORE INSERT ON outgoing_operations
             BEGIN SELECT RAISE(FAIL, 'test publication failure'); END;",
        )?;
        let result = account.complete_media_operation(
            lease.operation.id.clone(),
            lease.lease_id.clone(),
            AttachmentMetadata {
                url: "https://media.example/test".to_owned(),
                sha256: Some(hex::encode([7_u8; 32])),
                byte_size: Some(42),
                ..AttachmentMetadata::default()
            },
            42,
            1_700_000_003,
        );
        assert!(result.is_err());
        assert_eq!(
            account
                .database()?
                .product_media_operation(&lease.operation.id)?
                .ok_or(SoftchatError::InvalidMediaOperation)?
                .state,
            crate::AccountMediaOperationState::Completed
        );
        let events = account.drain_log_events(64)?.events;
        assert_eq!(
            events
                .iter()
                .filter(|event| event.operation == "media.upload.completed")
                .count(),
            1
        );
        assert!(
            events
                .iter()
                .any(|event| event.operation == "media.publication.deferred")
        );
        assert!(
            !events
                .iter()
                .any(|event| event.operation == "message.queued"
                    || event.operation == "media.completion.failed")
        );
        assert!(!format!("{events:?}").contains("private"));
        account
            .database()?
            .connection
            .execute_batch("DROP TRIGGER reject_test_publication")?;
        assert_eq!(
            account
                .recover_pending_media_messages(10, 1_700_000_004)?
                .len(),
            1
        );
        let recovered = account.drain_log_events(64)?.events;
        assert!(
            recovered
                .iter()
                .any(|event| event.operation == "recovery.completed")
        );
        assert!(
            !recovered
                .iter()
                .any(|event| event.operation == "media.upload.completed")
        );
        assert!(
            account
                .recover_pending_media_messages(10, 1_700_000_005)?
                .is_empty()
        );
        assert!(account.drain_log_events(64)?.events.is_empty());
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn media_message_dependencies_claim_publication_and_recover_after_reopen()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("media-message-recovery")?;
        let command_id = "media-message-command".to_owned();
        {
            let account =
                AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
            let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
            let conversation = account
                .get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
            let revision_before_rejection = account.database_info()?.revision;
            assert!(matches!(
                account.prepare_media_message(
                    "media-message-too-large".to_owned(),
                    conversation.id.clone(),
                    content(&"x".repeat(crate::MAX_PRODUCT_MESSAGE_BYTES)),
                    String::new(),
                    vec![crate::MediaPreparationInput {
                        source_fingerprint: "must-not-start-upload".to_owned(),
                    }],
                    None,
                    1_700_000_001,
                ),
                Err(SoftchatError::InvalidMediaOperation)
            ));
            assert_eq!(account.database_info()?.revision, revision_before_rejection);
            assert!(
                account
                    .pending_media_message("media-message-too-large".to_owned())?
                    .is_none()
            );
            let draft = AccountDraft {
                text: "attachment".to_owned(),
                emoji_tags: Vec::new(),
                spans: Vec::new(),
                reply_to_message_id: String::new(),
                attachment_operation_ids: Vec::new(),
                updated_at: 1_700_000_001,
            };
            account.save_conversation_draft(
                conversation.id.clone(),
                Some(draft.clone()),
                1_700_000_001,
            )?;
            let pending = account.prepare_media_message(
                command_id.clone(),
                conversation.id.clone(),
                content("attachment"),
                String::new(),
                vec![crate::MediaPreparationInput {
                    source_fingerprint: "staged-sha256-transform-v1".to_owned(),
                }],
                Some(draft),
                1_700_000_001,
            )?;
            assert!(
                account
                    .conversation(conversation.id)?
                    .ok_or(SoftchatError::InvalidPersistenceResult)?
                    .draft
                    .is_none()
            );
            assert_eq!(
                pending.state,
                crate::PendingMediaMessageState::WaitingForUploads
            );
            let failed_command = "media-message-failed".to_owned();
            let failed_pending = account.prepare_media_message(
                failed_command.clone(),
                pending.conversation_id.clone(),
                content("failed attachment"),
                String::new(),
                vec![crate::MediaPreparationInput {
                    source_fingerprint: "failed-staged-source-v1".to_owned(),
                }],
                None,
                1_700_000_002,
            )?;
            let failed_lease = account
                .claim_media_operation(
                    failed_pending.attachments[0].id.clone(),
                    "failed-media-message-lease".to_owned(),
                    1_700_000_003,
                    300,
                )?
                .ok_or(SoftchatError::InvalidMediaOperation)?;
            account.finish_media_operation_lease(
                failed_lease.operation.id,
                failed_lease.lease_id,
                false,
                "upload_rejected".to_owned(),
                1_700_000_004,
            )?;
            assert_eq!(
                account
                    .pending_media_message(failed_command.clone())?
                    .ok_or(SoftchatError::InvalidPersistenceResult)?
                    .state,
                crate::PendingMediaMessageState::Failed
            );
            let cancelled =
                account.cancel_pending_media_message(failed_command.clone(), 1_700_000_005)?;
            assert_eq!(cancelled.state, crate::PendingMediaMessageState::Cancelled);
            assert_eq!(
                cancelled.attachments[0].state,
                crate::AccountMediaOperationState::Cancelled
            );
            assert_eq!(
                account.cancel_pending_media_message(failed_command, 1_700_000_006)?,
                cancelled
            );
            let partially_cancelled_command = "media-message-partially-cancelled".to_owned();
            let partially_cancelled = account.prepare_media_message(
                partially_cancelled_command.clone(),
                pending.conversation_id.clone(),
                content("two attachments"),
                String::new(),
                vec![
                    crate::MediaPreparationInput {
                        source_fingerprint: "partially-cancelled-source-a".to_owned(),
                    },
                    crate::MediaPreparationInput {
                        source_fingerprint: "partially-cancelled-source-b".to_owned(),
                    },
                ],
                None,
                1_700_000_006,
            )?;
            account.transition_media_operation(
                partially_cancelled.attachments[0].id.clone(),
                "cancel".to_owned(),
                String::new(),
                1_700_000_007,
            )?;
            let partially_cancelled = account
                .pending_media_message(partially_cancelled_command.clone())?
                .ok_or(SoftchatError::InvalidPersistenceResult)?;
            assert_eq!(
                partially_cancelled.state,
                crate::PendingMediaMessageState::Cancelled
            );
            assert_eq!(
                partially_cancelled.attachments[0].state,
                crate::AccountMediaOperationState::Cancelled
            );
            assert_eq!(
                partially_cancelled.attachments[1].state,
                crate::AccountMediaOperationState::Prepared
            );
            let fully_cancelled = account
                .cancel_pending_media_message(partially_cancelled_command.clone(), 1_700_000_008)?;
            assert!(
                fully_cancelled
                    .attachments
                    .iter()
                    .all(|attachment| attachment.state
                        == crate::AccountMediaOperationState::Cancelled)
            );
            assert_eq!(
                account.cancel_pending_media_message(partially_cancelled_command, 1_700_000_009,)?,
                fully_cancelled
            );
            let lease = account
                .claim_media_operation(
                    pending.attachments[0].id.clone(),
                    "media-message-lease".to_owned(),
                    1_700_000_007,
                    300,
                )?
                .ok_or(SoftchatError::InvalidMediaOperation)?;
            account.database()?.complete_product_media(
                &lease.operation.id,
                &lease.lease_id,
                AttachmentMetadata {
                    url: "https://media.example/file".to_owned(),
                    mime_type: Some("image/jpeg".to_owned()),
                    sha256: Some(hex::encode([7_u8; 32])),
                    byte_size: Some(42),
                    ..AttachmentMetadata::default()
                },
                42,
                1_700_000_008,
            )?;
            assert_eq!(
                account
                    .pending_media_message(command_id.clone())?
                    .ok_or(SoftchatError::InvalidPersistenceResult)?
                    .state,
                crate::PendingMediaMessageState::ReadyToPublish
            );
            assert!(
                account
                    .database()?
                    .claim_pending_media_publication(&command_id)?
            );
            assert!(matches!(
                account.cancel_pending_media_message(command_id.clone(), 1_700_000_009),
                Err(SoftchatError::InvalidMediaOperation)
            ));
            let ready = account
                .pending_media_message(command_id.clone())?
                .ok_or(SoftchatError::InvalidPersistenceResult)?;
            let attachments = ready
                .attachments
                .iter()
                .map(|operation| {
                    operation
                        .attachment
                        .clone()
                        .ok_or(SoftchatError::InvalidPersistenceResult)
                })
                .collect::<Result<Vec<_>, _>>()?;
            let sent = account.send_message(
                ready.command_id,
                ready.conversation_id,
                MessageContentInput {
                    text: ready.text,
                    attachments,
                    emoji_tags: ready.emoji_tags,
                },
                ready.reply_to_message_id,
                None,
                1_700_000_010,
            )?;
            assert!(sent.inserted);
        }

        let reopened =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let published = reopened.recover_pending_media_messages(10, 1_700_000_004)?;
        assert_eq!(published.len(), 1);
        assert!(!published[0].inserted);
        assert_eq!(
            published[0]
                .message
                .as_ref()
                .map(|message| message.attachments.len()),
            Some(1)
        );
        let published_message = published[0]
            .message
            .as_ref()
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        let cleared_caption = reopened.edit_message(
            "media-message-clear-caption".to_owned(),
            published_message.id.clone(),
            MessageContentInput {
                text: String::new(),
                attachments: Vec::new(),
                emoji_tags: Vec::new(),
            },
            1_700_000_005,
        )?;
        let cleared_message = cleared_caption
            .message
            .as_ref()
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        assert!(cleared_message.text.is_empty());
        assert_eq!(cleared_message.attachments, published_message.attachments);
        assert!(cleared_message.edited);
        let download = reopened.prepare_media_download(
            published_message.id.clone(),
            "https://media.example/file".to_owned(),
            1_700_000_006,
        )?;
        let repeated = reopened.prepare_media_download(
            published_message.id.clone(),
            "https://media.example/file".to_owned(),
            1_700_000_999,
        )?;
        assert_eq!(download, repeated);
        assert_eq!(download.source_message_id, published_message.id);
        assert_eq!(
            download
                .attachment
                .as_ref()
                .map(|attachment| attachment.url.as_str()),
            Some("https://media.example/file")
        );
        let download_lease = reopened
            .claim_media_operation(
                download.id.clone(),
                "media-download-lease".to_owned(),
                1_700_001_000,
                300,
            )?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        let completed_download = reopened.complete_media_operation(
            download.id.clone(),
            download_lease.lease_id,
            download
                .attachment
                .clone()
                .ok_or(SoftchatError::InvalidMediaOperation)?,
            42,
            1_700_001_001,
        )?;
        assert_eq!(
            completed_download.operation.state,
            crate::AccountMediaOperationState::Completed
        );
        let restarted = reopened.restart_media_download(download.id.clone(), 1_700_001_002)?;
        assert_eq!(restarted.id, download.id);
        assert_eq!(restarted.state, crate::AccountMediaOperationState::Prepared);
        assert_eq!(restarted.attachment, download.attachment);
        let cancelled = reopened.transition_media_operation(
            download.id.clone(),
            "cancel".to_owned(),
            String::new(),
            1_700_001_003,
        )?;
        assert_eq!(
            cancelled.state,
            crate::AccountMediaOperationState::Cancelled
        );
        let restarted_cancelled =
            reopened.restart_media_download(download.id.clone(), 1_700_001_004)?;
        assert_eq!(
            restarted_cancelled.state,
            crate::AccountMediaOperationState::Prepared
        );
        let failed_lease = reopened
            .claim_media_operation(
                download.id.clone(),
                "media-download-failed-lease".to_owned(),
                1_700_001_005,
                300,
            )?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        let failed = reopened.finish_media_operation_lease(
            download.id.clone(),
            failed_lease.lease_id,
            false,
            "download_rejected".to_owned(),
            1_700_001_006,
        )?;
        assert_eq!(failed.state, crate::AccountMediaOperationState::Failed);
        assert_eq!(
            reopened
                .restart_media_download(download.id.clone(), 1_700_001_007)?
                .state,
            crate::AccountMediaOperationState::Prepared
        );
        assert!(matches!(
            reopened.prepare_media_download(
                published_message.id.clone(),
                "https://media.example/other".to_owned(),
                1_700_000_005,
            ),
            Err(SoftchatError::InvalidMediaOperation)
        ));
        assert_eq!(
            reopened
                .pending_media_message(command_id)?
                .ok_or(SoftchatError::InvalidPersistenceResult)?
                .state,
            crate::PendingMediaMessageState::Published
        );
        drop(reopened);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn public_asset_replacements_are_durable_idempotent_and_recoverable()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("asset-replacement-recovery")?;
        let profile_command = "replace-profile-picture".to_owned();
        let conversation_id;
        {
            let account =
                AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
            let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
            let conversation = account
                .get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
            conversation_id = conversation.id.clone();
            account.set_conversation_subject(
                "initial-subject".to_owned(),
                conversation.id,
                "Family".to_owned(),
                None,
                vec![vec![
                    "emoji".to_owned(),
                    "party".to_owned(),
                    "https://media.example/party.png".to_owned(),
                ]],
                1_700_000_001,
            )?;

            let pending = account.prepare_asset_replacement(
                profile_command.clone(),
                AssetReplacementTarget::ProfilePicture,
                String::new(),
                "profile-picture-source-v1".to_owned(),
                1_700_000_002,
            )?;
            assert_eq!(
                pending.operation.protection,
                crate::AccountMediaProtection::Public
            );
            assert_eq!(
                pending.state,
                crate::PendingAssetReplacementState::WaitingForUpload
            );
            assert_eq!(
                account.prepare_asset_replacement(
                    profile_command.clone(),
                    AssetReplacementTarget::ProfilePicture,
                    String::new(),
                    "profile-picture-source-v1".to_owned(),
                    1_700_000_999,
                )?,
                pending
            );
            assert!(matches!(
                account.prepare_asset_replacement(
                    profile_command.clone(),
                    AssetReplacementTarget::ProfilePicture,
                    String::new(),
                    "different-source".to_owned(),
                    1_700_000_003,
                ),
                Err(SoftchatError::CommandConflict)
            ));
            let lease = account
                .claim_media_operation(
                    pending.operation.id,
                    "profile-picture-lease".to_owned(),
                    1_700_000_003,
                    300,
                )?
                .ok_or(SoftchatError::InvalidMediaOperation)?;
            account.database()?.complete_product_media(
                &lease.operation.id,
                &lease.lease_id,
                AttachmentMetadata {
                    url: "https://media.example/avatar.jpg".to_owned(),
                    mime_type: Some("image/jpeg".to_owned()),
                    sha256: Some(hex::encode([8_u8; 32])),
                    byte_size: Some(42),
                    ..AttachmentMetadata::default()
                },
                42,
                1_700_000_004,
            )?;
            assert_eq!(
                account
                    .pending_asset_replacement(profile_command.clone())?
                    .ok_or(SoftchatError::InvalidPersistenceResult)?
                    .state,
                crate::PendingAssetReplacementState::ReadyToPublish
            );
        }

        let reopened =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        assert_eq!(
            reopened
                .recover_pending_asset_replacements(10, 1_700_000_005)?
                .len(),
            1
        );
        let own_public_key = reopened.public_key()?;
        assert_eq!(
            reopened.profile(own_public_key)?.picture.as_deref(),
            Some("https://media.example/avatar.jpg")
        );
        assert_eq!(
            reopened
                .pending_asset_replacement(profile_command)?
                .ok_or(SoftchatError::InvalidPersistenceResult)?
                .state,
            crate::PendingAssetReplacementState::Published
        );

        let icon = reopened.prepare_asset_replacement(
            "replace-conversation-icon".to_owned(),
            AssetReplacementTarget::ConversationIcon,
            conversation_id.clone(),
            "conversation-icon-source-v1".to_owned(),
            1_700_000_006,
        )?;
        let icon_lease = reopened
            .claim_media_operation(
                icon.operation.id,
                "conversation-icon-lease".to_owned(),
                1_700_000_007,
                300,
            )?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        let icon_completion = reopened.complete_media_operation(
            icon_lease.operation.id,
            icon_lease.lease_id,
            AttachmentMetadata {
                url: "https://media.example/conversation.png".to_owned(),
                mime_type: Some("image/png".to_owned()),
                sha256: Some(hex::encode([9_u8; 32])),
                dimensions: Some("256x256".to_owned()),
                byte_size: Some(64),
                ..AttachmentMetadata::default()
            },
            64,
            1_700_000_008,
        )?;
        assert!(icon_completion.completed_operation.is_some());
        let conversation = reopened
            .conversation(conversation_id.clone())?
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        assert_eq!(conversation.subject, "Family");
        assert_eq!(
            conversation
                .subject_icon
                .as_ref()
                .map(|value| value.url.as_str()),
            Some("https://media.example/conversation.png")
        );
        assert_eq!(
            conversation
                .subject_icon
                .as_ref()
                .and_then(|value| value.sha256.clone()),
            Some(hex::encode([9_u8; 32]))
        );
        let icon_download =
            reopened.prepare_conversation_icon_download(conversation_id.clone(), 1_700_000_009)?;
        assert_eq!(icon_download.direction, "download");
        assert_eq!(icon_download.conversation_id, conversation_id);
        assert!(icon_download.source_message_id.is_empty());
        assert_eq!(icon_download.attachment, conversation.subject_icon);
        let original_icon = conversation
            .subject_icon
            .clone()
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        reopened.set_conversation_subject(
            "rename-conversation-without-icon".to_owned(),
            conversation_id.clone(),
            "Family renamed".to_owned(),
            None,
            Vec::new(),
            1_700_000_010,
        )?;
        let renamed = reopened
            .conversation(conversation_id.clone())?
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        assert_eq!(renamed.subject, "Family renamed");
        assert_eq!(renamed.subject_icon, Some(original_icon.clone()));
        assert_eq!(
            reopened.prepare_conversation_icon_download(
                icon_download.conversation_id.clone(),
                1_700_000_010,
            )?,
            icon_download
        );
        let inherited_icon_lease = reopened
            .claim_media_operation(
                icon_download.id.clone(),
                "inherited-conversation-icon-lease".to_owned(),
                1_700_000_011,
                300,
            )?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        let inherited_operation_id = inherited_icon_lease.operation.id.clone();
        let inherited_lease_id = inherited_icon_lease.lease_id.clone();
        assert_eq!(
            reopened
                .complete_media_operation(
                    inherited_operation_id.clone(),
                    inherited_lease_id.clone(),
                    original_icon.clone(),
                    64,
                    1_700_000_012,
                )?
                .operation
                .state,
            crate::AccountMediaOperationState::Completed
        );
        let replacement_icon = AttachmentMetadata {
            url: "https://media.example/conversation-v2.png".to_owned(),
            mime_type: Some("image/png".to_owned()),
            sha256: Some(hex::encode([10_u8; 32])),
            dimensions: Some("512x512".to_owned()),
            byte_size: Some(128),
            ..AttachmentMetadata::default()
        };
        reopened.set_conversation_subject(
            "replace-conversation-icon-subject".to_owned(),
            conversation_id.clone(),
            "Family".to_owned(),
            Some(replacement_icon.clone()),
            Vec::new(),
            1_700_000_013,
        )?;
        assert_eq!(
            reopened
                .complete_media_operation(
                    inherited_operation_id,
                    inherited_lease_id,
                    original_icon,
                    64,
                    1_700_000_014,
                )?
                .operation
                .state,
            crate::AccountMediaOperationState::Completed
        );
        let replacement_download =
            reopened.prepare_conversation_icon_download(conversation_id.clone(), 1_700_000_014)?;
        assert_ne!(replacement_download.id, icon_download.id);
        assert_eq!(
            replacement_download.attachment,
            Some(replacement_icon.clone())
        );
        let stale_icon_lease = reopened
            .claim_media_operation(
                replacement_download.id,
                "stale-conversation-icon-lease".to_owned(),
                1_700_000_015,
                300,
            )?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        let stale_operation_id = stale_icon_lease.operation.id.clone();
        let stale_lease_id = stale_icon_lease.lease_id.clone();
        reopened.set_conversation_subject(
            "replace-conversation-icon-again".to_owned(),
            conversation_id.clone(),
            "Family".to_owned(),
            Some(AttachmentMetadata {
                url: "https://media.example/conversation-v3.png".to_owned(),
                mime_type: Some("image/png".to_owned()),
                sha256: Some(hex::encode([11_u8; 32])),
                dimensions: Some("768x768".to_owned()),
                byte_size: Some(256),
                ..AttachmentMetadata::default()
            }),
            Vec::new(),
            1_700_000_016,
        )?;
        assert!(matches!(
            reopened.complete_media_operation(
                stale_operation_id.clone(),
                stale_lease_id.clone(),
                replacement_icon,
                128,
                1_700_000_017,
            ),
            Err(SoftchatError::InvalidMediaOperation)
        ));
        assert_eq!(
            reopened
                .finish_media_operation_lease(
                    stale_operation_id.clone(),
                    stale_lease_id,
                    false,
                    "stale_icon".to_owned(),
                    1_700_000_018,
                )?
                .state,
            crate::AccountMediaOperationState::Failed
        );
        assert!(matches!(
            reopened.restart_media_download(stale_operation_id.clone(), 1_700_000_019),
            Err(SoftchatError::InvalidMediaOperation)
        ));
        assert_eq!(
            reopened
                .media_operation(stale_operation_id)?
                .ok_or(SoftchatError::InvalidPersistenceResult)?
                .state,
            crate::AccountMediaOperationState::Failed
        );
        reopened.set_conversation_subject(
            "replace-conversation-icon-ox-only".to_owned(),
            conversation_id.clone(),
            "Family".to_owned(),
            Some(AttachmentMetadata {
                url: "https://media.example/conversation-ox-only.png".to_owned(),
                original_sha256: Some(hex::encode([12_u8; 32])),
                ..AttachmentMetadata::default()
            }),
            Vec::new(),
            1_700_000_020,
        )?;
        assert!(matches!(
            reopened.prepare_conversation_icon_download(conversation_id.clone(), 1_700_000_021),
            Err(SoftchatError::InvalidMediaOperation)
        ));

        let banner = reopened.prepare_asset_replacement(
            "replace-profile-banner".to_owned(),
            AssetReplacementTarget::ProfileBanner,
            String::new(),
            "profile-banner-source-v1".to_owned(),
            1_700_000_009,
        )?;
        let banner_lease = reopened
            .claim_media_operation(
                banner.operation.id,
                "profile-banner-lease".to_owned(),
                1_700_000_010,
                300,
            )?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        assert!(matches!(
            reopened.complete_media_operation(
                banner_lease.operation.id.clone(),
                banner_lease.lease_id.clone(),
                AttachmentMetadata {
                    url: "https://media.example/banner.jpg".to_owned(),
                    encryption: Some("encrypted-value".to_owned()),
                    ..AttachmentMetadata::default()
                },
                64,
                1_700_000_011,
            ),
            Err(SoftchatError::InvalidMediaOperation)
        ));
        let failed = reopened.finish_media_operation_lease(
            banner_lease.operation.id,
            banner_lease.lease_id,
            false,
            "public_upload_rejected".to_owned(),
            1_700_000_012,
        )?;
        assert_eq!(failed.state, crate::AccountMediaOperationState::Failed);
        assert_eq!(
            reopened
                .pending_asset_replacement("replace-profile-banner".to_owned())?
                .ok_or(SoftchatError::InvalidPersistenceResult)?
                .state,
            crate::PendingAssetReplacementState::Failed
        );

        drop(reopened);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn settings_and_read_state_merge_incoming_ios_compatible_snapshots()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("app-data-merge")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let alice = crate::LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
        let sent = account.send_message(
            "read-message".to_owned(),
            conversation.id.clone(),
            content("read boundary"),
            String::new(),
            None,
            1_700_000_001,
        )?;
        let message = sent
            .message
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        let marked = account.mark_conversation_read(
            "read-command".to_owned(),
            conversation.id.clone(),
            MessageCursor {
                created_at: message.created_at,
                message_id: message.id,
            },
            1_700_000_002,
        )?;
        assert!(marked.inserted);
        assert!(!account.account_read_state()?[0].forced_unread);

        let legacy_chat_key = [alice.public_key().to_hex(), bob.public_key().to_hex()]
            .into_iter()
            .collect::<BTreeSet<_>>()
            .into_iter()
            .collect::<Vec<_>>()
            .join(",");
        let unread = alice.create_app_data_sync(
            1_700_000_004,
            AppDataContext::ReadState,
            &serde_json::json!({
                "chats": [{
                    "chatKey": legacy_chat_key,
                    "readAt": null
                }]
            })
            .to_string(),
        )?;
        account.ingest(vec![unread.event], 1_700_000_005)?;
        let read_state = account.account_read_state()?;
        assert_eq!(read_state.len(), 1);
        assert!(read_state[0].forced_unread);

        let settings = alice.create_app_data_sync(
            1_700_000_006,
            AppDataContext::AppSettings,
            r#"{
                "videoUploadPreset":"AVAssetExportPresetHighestQuality",
                "preserveHDROnUpload":false,
                "christmasTheme":true,
                "typingIndicatorsEnabled":false,
                "hideNotificationContent":true,
                "domainReplacementRules":[],
                "savedStickerURLs":["https://example.com/sticker.webp"]
            }"#,
        )?;
        account.ingest(vec![settings.event], 1_700_000_007)?;
        let effective = account.account_settings()?;
        assert_eq!(effective.video_upload_quality, "high");
        assert!(!effective.preserve_hdr);
        assert!(!effective.typing_indicators_enabled);
        assert_eq!(effective.notification_content_privacy, "hidden");
        assert_eq!(effective.seasonal_appearance.as_deref(), Some("christmas"));
        assert_eq!(
            effective.saved_sticker_references,
            vec!["https://example.com/sticker.webp"]
        );

        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn operation_retry_cancel_and_relay_rotation_are_durable()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("operation-recovery")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;

        account.add_account_relay("wss://relay-a.example".to_owned(), 1_700_000_001)?;
        account.add_account_relay("wss://relay-b.example".to_owned(), 1_700_000_002)?;
        account.set_account_active_relay("wss://relay-a.example/".to_owned(), 1_700_000_003)?;
        let first = account.send_message(
            "recover-first".to_owned(),
            conversation.id.clone(),
            content("retry me"),
            String::new(),
            None,
            1_700_000_004,
        )?;
        let claim = account
            .claim_deliveries(
                "lease-recovery".to_owned(),
                "wss://relay-a.example/".to_owned(),
                1_700_000_005,
                30,
            )?
            .ok_or(SoftchatError::InvalidDeliveryState)?;
        let intent_ids = claim
            .claim
            .intents
            .iter()
            .map(|intent| intent.intent_id.clone())
            .collect::<Vec<_>>();
        account.mark_socket_written(claim.claim.clone(), intent_ids.clone())?;
        account.apply_relay_results(
            claim.claim,
            intent_ids
                .into_iter()
                .map(|intent_id| RelayDeliveryResult {
                    intent_id,
                    kind: crate::RelayDeliveryResultKind::Rejected,
                    category: "relay_rejected".to_owned(),
                })
                .collect(),
        )?;
        assert_eq!(
            account
                .operation(first.operation.operation_id.clone())?
                .state,
            crate::OperationState::Failed
        );
        assert_eq!(
            account
                .retry_operation(first.operation.operation_id.clone())?
                .operation
                .state,
            crate::OperationState::Queued
        );

        let second = account.send_message(
            "recover-second".to_owned(),
            conversation.id,
            content("cancel me"),
            String::new(),
            None,
            1_700_000_006,
        )?;
        assert_eq!(
            account
                .cancel_operation(second.operation.operation_id)?
                .operation
                .state,
            crate::OperationState::Cancelled
        );

        let rotated = account
            .rotate_failed_account_relay("wss://relay-a.example/".to_owned(), 1_700_000_007)?;
        assert_ne!(rotated.active_relay_url, "wss://relay-a.example/");
        assert!(rotated.rebound_intent_count > 0);
        assert_eq!(
            account
                .relay_catalog()?
                .into_iter()
                .filter(|entry| entry.is_active)
                .count(),
            1
        );
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn account_clear_resets_durable_and_process_local_state_without_crossing_accounts()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("complete-account-clear")?;
        let alice =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let bob = AccountRuntimeHandle::open(secret(BOB_SECRET)?, directory.display().to_string())?;
        assert_ne!(alice.account_id(), bob.account_id());

        {
            let database = lock(&alice.database);
            database.connection.execute(
                "INSERT INTO app_data_effective
                 (context, canonical_json, event_created_at, event_id)
                 VALUES ('app-settings', '{\"quickReaction\":\"retained\"}', 1, 'event')",
                [],
            )?;
        }
        bob.put_local_contact(LocalContactRecord {
            public_key: alice.account_id(),
            name: Some("Alice survives Bob isolation".to_owned()),
            updated_at_millis: 1,
        })?;
        alice.start_sync(
            "manual-reset-sync".to_owned(),
            "manual-reset-neg".to_owned(),
            "manual-reset-events".to_owned(),
            1_700_000_000,
            0,
            crate::MIN_NEGENTROPY_FRAME_BYTES,
        )?;
        let first_transport = alice.start_account_transport("clear-run-before".to_owned())?;
        assert_eq!(first_transport.actions.len(), 1);
        lock(&alice.typing_authored).insert(
            hex::encode([0x44; 32]),
            (1_700_000_000, "typing-command".to_owned()),
        );

        let cleared = alice.clear_account()?;
        assert!(cleared.revision > 0);
        assert!(lock(&alice.sync_engines).is_empty());
        assert!(lock(&alice.transport_run).is_none());
        assert!(lock(&alice.typing_authored).is_empty());
        assert_eq!(
            alice.sync_snapshot("manual-reset-sync".to_owned()),
            Err(SoftchatError::InvalidSyncState)
        );
        assert_eq!(
            lock(&alice.database).connection.query_row(
                "SELECT COUNT(*) FROM app_data_effective",
                [],
                |row| row.get::<_, i64>(0),
            )?,
            0
        );
        assert_eq!(bob.local_contacts(20, 0)?.len(), 1);

        alice.start_sync(
            "manual-reset-sync".to_owned(),
            "manual-reset-neg-after".to_owned(),
            "manual-reset-events-after".to_owned(),
            1_700_000_001,
            0,
            crate::MIN_NEGENTROPY_FRAME_BYTES,
        )?;
        let restarted = alice.start_account_transport("clear-run-after".to_owned())?;
        assert_eq!(
            restarted.actions[0].kind,
            AccountTransportActionKind::Connect
        );
        assert_eq!(
            restarted.actions[0].configured_url,
            "wss://it2.softchat.c84ad60.xyz/"
        );
        alice.cancel_account_transport()?;
        alice.cancel_sync("manual-reset-sync".to_owned())?;
        drop(alice);
        drop(bob);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn correlated_transport_results_are_idempotent_and_cancel_releases_work()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("transport-actions")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        account.set_log_level(AccountLogLevel::Debug)?;
        let started = account.start_account_transport("android-run-1".to_owned())?;
        assert_eq!(started.actions.len(), 1);
        assert_eq!(started.actions[0].kind, AccountTransportActionKind::Connect);
        let connected = AccountTransportResult {
            run_id: started.actions[0].run_id.clone(),
            action_id: started.actions[0].action_id.clone(),
            generation: started.actions[0].generation,
            kind: AccountTransportResultKind::Connected,
            text_frame: String::new(),
        };
        let first = account.apply_account_transport_result(connected.clone(), 1_700_000_000)?;
        account.logs.flush();
        account.drain_log_events(64)?;
        let duplicate = account.apply_account_transport_result(connected, 1_700_000_001)?;
        assert!(
            account.drain_log_events(64)?.events.is_empty(),
            "Replayed transport results are silent"
        );
        for _ in 0..30 {
            account.wake_account_transport(1_700_000_001)?;
        }
        let receive = transport_action(&first, AccountTransportActionKind::ReceiveText)?;
        account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::TimedOut,
                String::new(),
            ),
            1_700_000_001,
        )?;
        assert!(
            account.drain_log_events(64)?.events.is_empty(),
            "Idle wake and receive timeout are silent at DEBUG"
        );
        assert_eq!(first.actions, duplicate.actions);
        assert_eq!(first.revision, duplicate.revision);
        assert_eq!(
            first.sync.as_ref().map(|sync| sync.phase),
            Some(SyncPhase::Reconciling)
        );
        assert_eq!(first.sync, duplicate.sync);
        assert!(
            first
                .actions
                .iter()
                .any(|action| action.kind == AccountTransportActionKind::ReceiveText)
        );
        let unknown = AccountTransportResult {
            run_id: started.actions[0].run_id.clone(),
            action_id: "unknown-stale-action".to_owned(),
            generation: started.actions[0].generation,
            kind: AccountTransportResultKind::Connected,
            text_frame: String::new(),
        };
        assert_eq!(
            account.apply_account_transport_result(unknown, 1_700_000_002),
            Err(SoftchatError::UnsupportedSystemAction)
        );
        assert!(!account.wake_account_transport(1_700_000_002)?.idle);
        let reset = account.reset_account_sync(1_700_000_002)?;
        assert!(reset.sync.is_none());
        assert_eq!(reset.actions.len(), 1);
        assert_eq!(reset.actions[0].kind, AccountTransportActionKind::Connect);
        assert!(reset.actions[0].generation > started.actions[0].generation);

        let cancelled = account.cancel_account_transport()?;
        assert_eq!(cancelled.connection_state, RelayConnectionState::Cancelled);
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn account_transport_replays_a_socket_written_profile_after_authentication()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("transport-profile-during-connect")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let started = account.start_account_transport("profile-connect-run".to_owned())?;
        account.update_profile(
            "profile-during-connect".to_owned(),
            serde_json::json!({ "displayName": "Alice" }).to_string(),
            1_700_000_000,
        )?;

        let connected = account.apply_account_transport_result(
            transport_result(
                &started.actions[0],
                AccountTransportResultKind::Connected,
                String::new(),
            ),
            1_700_000_001,
        )?;
        for action in connected
            .actions
            .iter()
            .filter(|action| action.kind == AccountTransportActionKind::SendText)
        {
            account.apply_account_transport_result(
                transport_result(
                    action,
                    AccountTransportResultKind::FrameWritten,
                    String::new(),
                ),
                1_700_000_002,
            )?;
        }

        let receive = transport_action(&connected, AccountTransportActionKind::ReceiveText)?;
        let authenticating = account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::FrameReceived,
                RelayResponseFrame::Auth("profile-challenge".to_owned()).to_json()?,
            ),
            1_700_000_003,
        )?;
        let authentication_send =
            transport_action(&authenticating, AccountTransportActionKind::SendText)?;
        let authentication_event_id =
            match crate::parse_client_relay_frame(&authentication_send.text_frame)? {
                crate::ClientRelayFrame::Auth(event) => event.id().to_hex(),
                _ => return Err(SoftchatError::InvalidRelayAuthentication.into()),
            };
        account.apply_account_transport_result(
            transport_result(
                &authentication_send,
                AccountTransportResultKind::FrameWritten,
                String::new(),
            ),
            1_700_000_004,
        )?;
        let authentication_receive =
            transport_action(&authenticating, AccountTransportActionKind::ReceiveText)?;
        let ready = account.apply_account_transport_result(
            transport_result(
                &authentication_receive,
                AccountTransportResultKind::FrameReceived,
                RelayResponseFrame::Ok(crate::BatchAcknowledgement {
                    event_id: authentication_event_id,
                    accepted: true,
                    message: String::new(),
                })
                .to_json()?,
            ),
            1_700_000_005,
        )?;
        for action in ready
            .actions
            .iter()
            .filter(|action| action.kind == AccountTransportActionKind::SendText)
        {
            account.apply_account_transport_result(
                transport_result(
                    action,
                    AccountTransportResultKind::FrameWritten,
                    String::new(),
                ),
                1_700_000_006,
            )?;
        }

        account.cancel_account_transport()?;
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn account_transport_authentication_fence_drains_stale_sync_and_restarts_fresh()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("transport-authentication-fence")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let started = account.start_account_transport("android-auth-run".to_owned())?;
        let connected = account.apply_account_transport_result(
            transport_result(
                &started.actions[0],
                AccountTransportResultKind::Connected,
                String::new(),
            ),
            1_700_000_000,
        )?;
        let initial_neg_subscription = neg_subscription_id(&connected)?;
        let stale_send = transport_action(&connected, AccountTransportActionKind::SendText)?;
        let initial_receive =
            transport_action(&connected, AccountTransportActionKind::ReceiveText)?;

        let authenticating = account.apply_account_transport_result(
            transport_result(
                &initial_receive,
                AccountTransportResultKind::FrameReceived,
                RelayResponseFrame::Auth("device-challenge".to_owned()).to_json()?,
            ),
            1_700_000_001,
        )?;
        assert_eq!(
            authenticating.connection_state,
            RelayConnectionState::Authenticating
        );
        assert!(authenticating.sync.is_none());
        let authentication_send =
            transport_action(&authenticating, AccountTransportActionKind::SendText)?;
        let mut receive =
            transport_action(&authenticating, AccountTransportActionKind::ReceiveText)?;
        let authentication_event_id =
            match crate::parse_client_relay_frame(&authentication_send.text_frame)? {
                crate::ClientRelayFrame::Auth(event) => event.id().to_hex(),
                _ => return Err(SoftchatError::InvalidRelayAuthentication.into()),
            };

        account.apply_account_transport_result(
            transport_result(
                &stale_send,
                AccountTransportResultKind::FrameWritten,
                String::new(),
            ),
            1_700_000_002,
        )?;
        let after_notice = account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::FrameReceived,
                RelayResponseFrame::Notice("authentication required".to_owned()).to_json()?,
            ),
            1_700_000_003,
        )?;
        receive = transport_action(&after_notice, AccountTransportActionKind::ReceiveText)?;
        let after_closed = account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::FrameReceived,
                RelayResponseFrame::Closed {
                    subscription_id: initial_neg_subscription.clone(),
                    message: "authentication required".to_owned(),
                }
                .to_json()?,
            ),
            1_700_000_004,
        )?;
        receive = transport_action(&after_closed, AccountTransportActionKind::ReceiveText)?;
        let after_neg_error = account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::FrameReceived,
                serde_json::json!([
                    "NEG-ERR",
                    initial_neg_subscription,
                    "authentication required",
                    500_u64
                ])
                .to_string(),
            ),
            1_700_000_005,
        )?;
        receive = transport_action(&after_neg_error, AccountTransportActionKind::ReceiveText)?;
        account.apply_account_transport_result(
            transport_result(
                &authentication_send,
                AccountTransportResultKind::FrameWritten,
                String::new(),
            ),
            1_700_000_006,
        )?;

        let ready = account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::FrameReceived,
                RelayResponseFrame::Ok(crate::BatchAcknowledgement {
                    event_id: authentication_event_id,
                    accepted: true,
                    message: String::new(),
                })
                .to_json()?,
            ),
            1_700_000_007,
        )?;
        assert_eq!(ready.connection_state, RelayConnectionState::Ready);
        assert_eq!(
            ready.sync.as_ref().map(|snapshot| snapshot.phase),
            Some(SyncPhase::Reconciling)
        );
        let fresh_neg_subscription = neg_subscription_id(&ready)?;
        assert_ne!(fresh_neg_subscription, initial_neg_subscription);

        receive = transport_action(&ready, AccountTransportActionKind::ReceiveText)?;
        let stale_close_after_auth = account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::FrameReceived,
                serde_json::json!(["NEG-CLOSE", initial_neg_subscription.clone()]).to_string(),
            ),
            1_700_000_008,
        )?;
        assert_eq!(
            stale_close_after_auth.connection_state,
            RelayConnectionState::Ready
        );
        receive = transport_action(
            &stale_close_after_auth,
            AccountTransportActionKind::ReceiveText,
        )?;
        let stale_after_auth = account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::FrameReceived,
                serde_json::json!(["NEG-ERR", initial_neg_subscription, "stale", 500_u64])
                    .to_string(),
            ),
            1_700_000_009,
        )?;
        assert_eq!(
            stale_after_auth.connection_state,
            RelayConnectionState::Ready
        );
        receive = transport_action(&stale_after_auth, AccountTransportActionKind::ReceiveText)?;
        let reconnecting = account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::FrameReceived,
                serde_json::json!(["NEG-ERR", fresh_neg_subscription, "current", 500_u64])
                    .to_string(),
            ),
            1_700_000_010,
        )?;
        assert_eq!(
            reconnecting.actions[0].kind,
            AccountTransportActionKind::Connect
        );
        assert!(reconnecting.actions[0].generation > started.actions[0].generation);

        account.cancel_account_transport()?;
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn account_transport_quarantines_bad_frames_and_keeps_the_same_connection()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("transport-bad-frame-recovery")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let started = account.start_account_transport("bad-frame-run".to_owned())?;
        let connected = account.apply_account_transport_result(
            transport_result(
                &started.actions[0],
                AccountTransportResultKind::Connected,
                String::new(),
            ),
            1_700_000_000,
        )?;
        let generation = started.actions[0].generation;
        let mut receive = transport_action(&connected, AccountTransportActionKind::ReceiveText)?;
        let bad_frames = [
            "{not-json".to_owned(),
            serde_json::json!(["FUTURE", { "extension": true }]).to_string(),
            serde_json::json!(["EVENT", "softchat-live-0", { "id": "00" }]).to_string(),
            RelayResponseFrame::Eose {
                subscription_id: "unknown-subscription".to_owned(),
            }
            .to_json()?,
            RelayResponseFrame::Ok(crate::BatchAcknowledgement {
                event_id: "00".repeat(32),
                accepted: true,
                message: String::new(),
            })
            .to_json()?,
        ];

        for (index, bad_frame) in bad_frames.into_iter().enumerate() {
            let after_bad = account.apply_account_transport_result(
                transport_result(
                    &receive,
                    AccountTransportResultKind::FrameReceived,
                    bad_frame,
                ),
                1_700_000_001 + i64::try_from(index)? * 2,
            )?;
            assert_eq!(after_bad.connection_state, RelayConnectionState::Ready);
            assert!(
                after_bad
                    .actions
                    .iter()
                    .all(|action| action.generation == generation)
            );
            receive = transport_action(&after_bad, AccountTransportActionKind::ReceiveText)?;

            let after_valid = account.apply_account_transport_result(
                transport_result(
                    &receive,
                    AccountTransportResultKind::FrameReceived,
                    RelayResponseFrame::Notice("valid-next-frame".to_owned()).to_json()?,
                ),
                1_700_000_002 + i64::try_from(index)? * 2,
            )?;
            assert_eq!(after_valid.connection_state, RelayConnectionState::Ready);
            receive = transport_action(&after_valid, AccountTransportActionKind::ReceiveText)?;
            assert_eq!(receive.generation, generation);
        }

        account.cancel_account_transport()?;
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn account_transport_reconnects_when_relay_closes_a_live_subscription()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("transport-live-closed")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let started = account.start_account_transport("closed-live-run".to_owned())?;
        let connected = account.apply_account_transport_result(
            transport_result(
                &started.actions[0],
                AccountTransportResultKind::Connected,
                String::new(),
            ),
            1_700_000_000,
        )?;
        let receive = transport_action(&connected, AccountTransportActionKind::ReceiveText)?;

        let reconnecting = account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::FrameReceived,
                RelayResponseFrame::Closed {
                    subscription_id: "softchat-live-0".to_owned(),
                    message: "rate-limited".to_owned(),
                }
                .to_json()?,
            ),
            1_700_000_001,
        )?;
        let connect = transport_action(&reconnecting, AccountTransportActionKind::Connect)?;
        assert!(connect.generation > started.actions[0].generation);
        assert_eq!(
            reconnecting.connection_state,
            RelayConnectionState::Connecting
        );

        account.cancel_account_transport()?;
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn account_transport_accepts_delivery_ack_before_socket_write()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("transport-ack-before-write")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
        let sent = account.send_message(
            "ack-before-write".to_owned(),
            conversation.id,
            content("delivery ordering"),
            String::new(),
            None,
            1_700_000_001,
        )?;
        let operation_id = sent.operation.operation_id;

        let started = account.start_account_transport("ack-order-run".to_owned())?;
        let connected = account.apply_account_transport_result(
            transport_result(
                &started.actions[0],
                AccountTransportResultKind::Connected,
                String::new(),
            ),
            1_700_000_002,
        )?;
        let delivery_sends = delivery_send_actions(&connected)?;
        assert!(!delivery_sends.is_empty());
        let first_event_id = delivery_sends[0].1[0].clone();
        let mut receive = transport_action(&connected, AccountTransportActionKind::ReceiveText)?;

        for (_, event_ids) in &delivery_sends {
            for event_id in event_ids {
                let acknowledged = account.apply_account_transport_result(
                    transport_result(
                        &receive,
                        AccountTransportResultKind::FrameReceived,
                        RelayResponseFrame::Ok(crate::BatchAcknowledgement {
                            event_id: event_id.clone(),
                            accepted: true,
                            message: String::new(),
                        })
                        .to_json()?,
                    ),
                    1_700_000_003,
                )?;
                receive = transport_action(&acknowledged, AccountTransportActionKind::ReceiveText)?;
            }
        }
        for (action, _) in &delivery_sends {
            account.apply_account_transport_result(
                transport_result(
                    action,
                    AccountTransportResultKind::FrameWritten,
                    String::new(),
                ),
                1_700_000_004,
            )?;
        }
        assert_eq!(
            account.operation(operation_id.clone())?.state,
            crate::OperationState::Sent
        );

        let duplicate = account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::FrameReceived,
                RelayResponseFrame::Ok(crate::BatchAcknowledgement {
                    event_id: first_event_id.clone(),
                    accepted: true,
                    message: String::new(),
                })
                .to_json()?,
            ),
            1_700_000_005,
        )?;
        let conflicting_receive =
            transport_action(&duplicate, AccountTransportActionKind::ReceiveText)?;
        let after_conflict = account.apply_account_transport_result(
            transport_result(
                &conflicting_receive,
                AccountTransportResultKind::FrameReceived,
                RelayResponseFrame::Ok(crate::BatchAcknowledgement {
                    event_id: first_event_id,
                    accepted: false,
                    message: "blocked".to_owned(),
                })
                .to_json()?,
            ),
            1_700_000_006,
        )?;
        assert_eq!(after_conflict.connection_state, RelayConnectionState::Ready);
        transport_action(&after_conflict, AccountTransportActionKind::ReceiveText)?;
        assert_eq!(
            account.operation(operation_id)?.state,
            crate::OperationState::Sent
        );

        assert_eq!(
            account.cancel_account_transport()?.connection_state,
            RelayConnectionState::Cancelled
        );
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }

    #[test]
    fn account_transport_reclaims_a_delivery_once_after_reconnect()
    -> Result<(), Box<dyn std::error::Error>> {
        let directory = temporary_directory("transport-delivery-reconnect")?;
        let account =
            AccountRuntimeHandle::open(secret(ALICE_SECRET)?, directory.display().to_string())?;
        let bob = crate::LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let conversation =
            account.get_or_create_conversation(vec![bob.public_key().to_hex()], 1_700_000_000)?;
        let sent = account.send_message(
            "delivery-reconnect".to_owned(),
            conversation.id,
            content("retry after disconnect"),
            String::new(),
            None,
            1_700_000_001,
        )?;

        let started = account.start_account_transport("reconnect-run".to_owned())?;
        let connected = account.apply_account_transport_result(
            transport_result(
                &started.actions[0],
                AccountTransportResultKind::Connected,
                String::new(),
            ),
            1_700_000_002,
        )?;
        let delivery_sends = delivery_send_actions(&connected)?;
        assert!(!delivery_sends.is_empty());
        let expected_event_ids = delivery_sends
            .iter()
            .flat_map(|(_, event_ids)| event_ids.iter().cloned())
            .collect::<BTreeSet<_>>();
        for (action, _) in &delivery_sends {
            account.apply_account_transport_result(
                transport_result(
                    action,
                    AccountTransportResultKind::FrameWritten,
                    String::new(),
                ),
                1_700_000_003,
            )?;
        }
        let receive = transport_action(&connected, AccountTransportActionKind::ReceiveText)?;
        let reconnecting = account.apply_account_transport_result(
            transport_result(
                &receive,
                AccountTransportResultKind::Disconnected,
                String::new(),
            ),
            1_700_000_004,
        )?;
        let reconnect = transport_action(&reconnecting, AccountTransportActionKind::Connect)?;

        let reconnected = account.apply_account_transport_result(
            transport_result(
                &reconnect,
                AccountTransportResultKind::Connected,
                String::new(),
            ),
            1_700_000_005,
        )?;
        assert_eq!(reconnected.connection_state, RelayConnectionState::Ready);
        let reclaimed_event_ids = delivery_send_actions(&reconnected)?
            .into_iter()
            .flat_map(|(_, event_ids)| event_ids)
            .collect::<Vec<_>>();
        assert_eq!(
            reclaimed_event_ids.iter().collect::<BTreeSet<_>>(),
            expected_event_ids.iter().collect::<BTreeSet<_>>(),
        );
        assert_eq!(reclaimed_event_ids.len(), expected_event_ids.len());
        assert_eq!(
            account.operation(sent.operation.operation_id)?.state,
            crate::OperationState::Sending,
        );

        account.cancel_account_transport()?;
        drop(account);
        std::fs::remove_dir_all(directory)?;
        Ok(())
    }
}