softchat/

account_product.rs

//! High-level account values, commands, and bounded product queries.
//!
//! This module is deliberately above protocol events and below platform
//! effects. Android and, later, Apple call these operations with user intent;
//! Rust resolves authenticated state and persists the complete result.

use std::collections::{BTreeMap, BTreeSet};

use base64::Engine;
use base64::engine::general_purpose::STANDARD;
use rusqlite::{
    OptionalExtension, TransactionBehavior, params, params_from_iter, types::Value as SqlValue,
};
use serde::{Deserialize, Serialize};
use serde_json::{Map, Value, json};
use sha2::{Digest, Sha256};
use url::Url;

use crate::account_diagnostics::{AccountLogCategory, Diagnostic};
use crate::chat::subject_icon_metadata_from_validated;
use crate::event::RumorEvent;
use crate::storage::{bump_revision, map_sqlite_error};
use crate::{
    AccountDatabase, AccountMutationResult, AttachmentMetadata, ChatRelationKind, NostrEventId,
    NostrPublicKey, OperationSnapshot, ProjectionKind, SoftchatError, classify_chat_message,
    classify_edit, classify_follow_list, classify_reaction, classify_subject,
    classify_user_metadata, conversation_id,
};

/// Maximum requested rows for one product query.
/// Media gallery pages finish the last message's attachment group and may
/// return at most `limit + MAX_CHAT_ATTACHMENTS - 1` items.
pub const MAX_PRODUCT_QUERY_PAGE: u32 = 100;
/// Maximum exact profiles resolved by one ordered product lookup.
pub const MAX_PRODUCT_PROFILE_LOOKUPS: usize = 512;
/// Maximum exact messages resolved by one ordered product lookup.
pub const MAX_PRODUCT_MESSAGE_LOOKUPS: usize = 512;
/// Maximum exact media operations resolved by one ordered lookup.
pub const MAX_PRODUCT_MEDIA_OPERATION_LOOKUPS: usize = 512;
/// Maximum UTF-8 bytes in a portable message draft.
///
/// This is only the cheap single-field ceiling. Before any durable or platform
/// effect, the SDK measures the complete rumor, seal, and wrapper and may
/// reject a smaller draft whose tags consume the remaining wire budget.
pub const MAX_PRODUCT_MESSAGE_BYTES: usize = crate::MAX_NIP44_WRITER_PLAINTEXT_BYTES;
/// Maximum UTF-8 bytes in one search query.
pub const MAX_PRODUCT_SEARCH_BYTES: usize = 512;
/// Maximum portable rich-text spans in one draft.
pub const MAX_PRODUCT_DRAFT_SPANS: usize = 256;
/// Maximum logical attachment preparation references in one draft.
pub const MAX_PRODUCT_DRAFT_ATTACHMENTS: usize = crate::MAX_CHAT_ATTACHMENTS;
/// Maximum compact JSON bytes reserved for one outgoing `imeta` tag.
///
/// Media preparation reserves this complete amount before upload; completion
/// accepts only metadata that fits the same reservation.
pub const MAX_PRODUCT_ATTACHMENT_TAG_BYTES: usize = 2 * 1024;
/// Maximum UTF-8 bytes in one local transcript.
pub const MAX_PRODUCT_TRANSCRIPT_BYTES: usize = 64 * 1024;
/// Maximum normalized waveform samples stored for one message.
pub const MAX_PRODUCT_WAVEFORM_SAMPLES: usize = 8_192;
/// Maximum UTF-8 bytes in retained settings JSON.
pub const MAX_ACCOUNT_SETTINGS_BYTES: usize = 64 * 1024;
/// Current portable settings schema.
pub const ACCOUNT_SETTINGS_SCHEMA_VERSION: u32 = 1;
/// Current portable read-state schema.
pub const ACCOUNT_READ_STATE_SCHEMA_VERSION: u32 = 1;
/// Maximum UTF-8 bytes in one complete read-state snapshot.
pub const MAX_ACCOUNT_READ_STATE_BYTES: usize = 256 * 1024;
/// Maximum conversations retained in one cross-device read-state snapshot.
pub const MAX_ACCOUNT_READ_STATE_ENTRIES: usize = 4_096;

const SWIFT_REFERENCE_DATE_UNIX_SECONDS: i64 = 978_307_200;

/// Stable newest-first message paging cursor.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct MessageCursor {
    /// Timestamp of the last row from the preceding page.
    pub created_at: i64,
    /// Event ID of the last row from the preceding page.
    pub message_id: String,
}

/// Effective cross-device read state for one conversation.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountReadState {
    /// Stable participant-set conversation ID.
    pub conversation_id: String,
    /// Stable read boundary, absent before any message was marked read.
    pub boundary: Option<MessageCursor>,
    /// Explicit manual unread state, cleared by the next mark-read action.
    pub forced_unread: bool,
    /// Portable timestamp of the state-changing action.
    pub updated_at: i64,
    /// Stable tie-breaker and idempotency identifier for that action.
    pub update_id: String,
}

/// Stable conversation-list paging cursor.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct ConversationCursor {
    /// Whether the last row was pinned.
    pub pinned: bool,
    /// Activity timestamp of the last row.
    pub last_activity_at: i64,
    /// Conversation ID of the last row.
    pub conversation_id: String,
}

/// Complete user-authored message input.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct MessageContentInput {
    /// Plain UTF-8 message text.
    pub text: String,
    /// Validated uploaded attachment metadata.
    pub attachments: Vec<AttachmentMetadata>,
    /// Exact canonical `emoji` tag arrays.
    pub emoji_tags: Vec<Vec<String>>,
}

/// Complete profile fields after an account-owned merge update.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
pub(crate) struct ProfileUpdateInput {
    /// Short account name.
    pub name: Option<String>,
    /// Human-readable display name.
    pub display_name: Option<String>,
    /// Profile description.
    pub about: Option<String>,
    /// HTTPS avatar URL.
    pub picture: Option<String>,
    /// HTTPS website URL.
    pub website: Option<String>,
    /// HTTPS banner URL.
    pub banner: Option<String>,
    /// Optional automation marker.
    pub bot: Option<bool>,
    /// Canonical object of unrecognized fields retained by a prior read.
    pub unknown_json: String,
}

/// User-selected reaction input.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct MessageReactionInput {
    /// Unicode reaction or `:shortcode:`.
    pub value: String,
    /// HTTPS custom-emoji URL, absent for Unicode.
    pub custom_emoji_url: Option<String>,
}

/// Supported released push-registration wire profiles.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum PushPlatform {
    /// Android Firebase Cloud Messaging.
    Fcm,
    /// Apple Push Notification service.
    Apns,
}

impl MessageContentInput {
    pub(crate) fn validate(&self) -> Result<(), SoftchatError> {
        self.validate_bounds()?;
        if self.text.is_empty() && self.attachments.is_empty() {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        Ok(())
    }

    pub(crate) fn validate_edit(&self) -> Result<(), SoftchatError> {
        self.validate_bounds()?;
        if !self.attachments.is_empty() {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        Ok(())
    }

    pub(crate) fn validate_bounds(&self) -> Result<(), SoftchatError> {
        if self.text.len() > MAX_PRODUCT_MESSAGE_BYTES {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        if self.attachments.len() > MAX_PRODUCT_DRAFT_ATTACHMENTS
            || self.emoji_tags.len() > crate::MAX_CHAT_EMOJI_TAGS
        {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        for attachment in &self.attachments {
            validate_product_attachment_wire_size(attachment)?;
        }
        Ok(())
    }
}

pub(crate) fn reserved_product_attachments(
    count: usize,
) -> Result<Vec<AttachmentMetadata>, SoftchatError> {
    if count > MAX_PRODUCT_DRAFT_ATTACHMENTS {
        return Err(SoftchatError::InvalidMediaOperation);
    }
    let mut reservation = AttachmentMetadata {
        url: "https://attachment.invalid/".to_owned(),
        unknown_fields: vec!["reserved x".to_owned()],
        ..AttachmentMetadata::default()
    };
    let current = attachment_tag_json_bytes(&reservation)?;
    let padding = MAX_PRODUCT_ATTACHMENT_TAG_BYTES
        .checked_sub(current)
        .ok_or(SoftchatError::InternalFailure)?;
    reservation.unknown_fields[0].push_str(&"x".repeat(padding));
    if attachment_tag_json_bytes(&reservation)? != MAX_PRODUCT_ATTACHMENT_TAG_BYTES {
        return Err(SoftchatError::InternalFailure);
    }
    Ok(vec![reservation; count])
}

fn validate_product_attachment_wire_size(
    attachment: &AttachmentMetadata,
) -> Result<(), SoftchatError> {
    if attachment_tag_json_bytes(attachment)? > MAX_PRODUCT_ATTACHMENT_TAG_BYTES {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    Ok(())
}

fn attachment_tag_json_bytes(attachment: &AttachmentMetadata) -> Result<usize, SoftchatError> {
    let tag = attachment
        .to_tag()
        .map_err(|_| SoftchatError::InvalidAccountOperation)?;
    serde_json::to_string(tag.values())
        .map(|json| json.len())
        .map_err(|_| SoftchatError::InternalFailure)
}

/// One effective reaction attached to a message.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct MessageReactionView {
    /// Reaction event ID.
    pub id: String,
    /// Authenticated reaction author.
    pub author_public_key: String,
    /// Unicode reaction or `:shortcode:`.
    pub value: String,
    /// HTTPS custom-emoji URL, empty for Unicode.
    pub custom_emoji_url: String,
    /// Portable event timestamp.
    pub created_at: i64,
}

/// Complete effective message consumed by product UI.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountMessageView {
    /// Canonical logical message ID.
    pub id: String,
    /// Stable participant-set conversation ID.
    pub conversation_id: String,
    /// Authenticated logical author.
    pub author_public_key: String,
    /// Portable event timestamp.
    pub created_at: i64,
    /// Effective text after an authorized edit.
    pub text: String,
    /// Original authenticated attachment metadata.
    pub attachments: Vec<AttachmentMetadata>,
    /// Effective custom-emoji tags.
    pub emoji_tags: Vec<Vec<String>>,
    /// Reply/forward relation.
    pub relation: ChatRelationKind,
    /// Related message ID, empty when unrelated.
    pub related_message_id: String,
    /// Original author for a forward, empty otherwise.
    pub forwarded_author_public_key: String,
    /// Whether an authorized edit changed the visible text.
    pub edited: bool,
    /// Whether an authorized deletion dominates this message.
    pub deleted: bool,
    /// Number of live replies.
    pub reply_count: u32,
    /// Bounded effective reactions in deterministic order; older reactions may
    /// be omitted. Toggle commands query authoritative local state separately.
    pub reactions: Vec<MessageReactionView>,
}

/// One bounded page of messages.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct MessagePage {
    /// Newest-first rows.
    pub messages: Vec<AccountMessageView>,
    /// Cursor for the next page, absent when this page is terminal.
    pub next_cursor: Option<MessageCursor>,
}

/// Portable semantic span kind for one draft.
#[derive(Clone, Copy, Debug, Deserialize, Eq, Ord, PartialEq, PartialOrd, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum DraftSpanKind {
    /// Bold presentation.
    Bold,
    /// Italic presentation.
    Italic,
    /// Inline code presentation.
    Code,
    /// HTTPS link whose URL is stored in the span value.
    Link,
    /// Mention whose canonical public key is stored in the span value.
    Mention,
    /// Custom emoji whose HTTPS URL is stored in the span value.
    CustomEmoji,
}

/// One bounded half-open UTF-16 range in a portable draft.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct DraftSpan {
    /// Inclusive UTF-16 code-unit offset.
    pub start_utf16: u32,
    /// Exclusive UTF-16 code-unit offset.
    pub end_utf16: u32,
    /// Portable semantic span kind.
    pub kind: DraftSpanKind,
    /// Kind-specific canonical value, empty for presentation-only spans.
    pub value: String,
}

/// One portable draft stored with its conversation.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountDraft {
    /// Plain portable text. Platform attributed-text state is reconstructed.
    pub text: String,
    /// Exact custom-emoji tags referenced by the text.
    pub emoji_tags: Vec<Vec<String>>,
    /// Portable validated rich-text ranges.
    pub spans: Vec<DraftSpan>,
    /// Replied message ID, empty when the draft is not a reply.
    pub reply_to_message_id: String,
    /// Stable logical media-operation IDs referenced by this draft.
    pub attachment_operation_ids: Vec<String>,
    /// Last durable update time.
    pub updated_at: i64,
}

/// Sparse account-local non-protocol data attached to one message.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct MessageLocalExtras {
    /// Stable authenticated message ID.
    pub message_id: String,
    /// Optional local transcript, never sent as protocol content.
    pub transcript: Option<String>,
    /// Normalized unsigned amplitude samples in the range 0 through 255.
    pub waveform: Vec<u8>,
    /// Last durable local update time.
    pub updated_at: i64,
}

/// Complete conversation state consumed by product UI.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct ConversationRecord {
    /// Stable participant-set identifier.
    pub id: String,
    /// Sorted canonical participant public keys including the local account.
    pub member_public_keys: Vec<String>,
    /// Latest authenticated subject text, empty when unset or cleared.
    pub subject: String,
    /// Latest authenticated subject icon, absent when unset.
    pub subject_icon: Option<AttachmentMetadata>,
    /// Newest live message, absent for a newly created empty conversation.
    pub last_message: Option<AccountMessageView>,
    /// Number of live messages.
    pub message_count: u32,
    /// Number of unread messages authored by another participant.
    pub unread_count: u32,
    /// Whether the local account archived this conversation.
    pub archived: bool,
    /// Whether the local account pinned this conversation.
    pub pinned: bool,
    /// Durable tuple read boundary, absent before the first explicit read action.
    pub read_cursor: Option<MessageCursor>,
    /// Whether the conversation was explicitly marked unread.
    pub forced_unread: bool,
    /// Portable draft, absent when empty.
    pub draft: Option<AccountDraft>,
    /// Timestamp used for stable list ordering.
    pub last_activity_at: i64,
}

/// One bounded page of complete conversations.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct ConversationPage {
    /// Ordered complete rows.
    pub conversations: Vec<ConversationRecord>,
    /// Cursor for the next page, absent when terminal.
    pub next_cursor: Option<ConversationCursor>,
}

/// Effective authenticated or explicitly local profile.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountProfile {
    /// Canonical public key.
    pub public_key: String,
    /// Short account name.
    pub name: Option<String>,
    /// Human-readable display name.
    pub display_name: Option<String>,
    /// Profile description.
    pub about: Option<String>,
    /// Avatar URL.
    pub picture: Option<String>,
    /// Website URL.
    pub website: Option<String>,
    /// Banner URL.
    pub banner: Option<String>,
    /// Optional automation marker.
    pub bot: Option<bool>,
    /// Canonical object containing authenticated fields unknown to this SDK version.
    pub unknown_json: String,
    /// Whether the visible name comes from Android-local address-book state.
    pub local_override: bool,
    /// Timestamp of the selected value.
    pub updated_at: i64,
    /// Earliest stored authenticated non-ephemeral event by this author, if known.
    pub first_seen_at: Option<i64>,
}

/// Versioned platform-neutral account settings.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct CustomEmojiReference {
    /// Case-insensitive reaction shortcode without surrounding colons.
    pub shortcode: String,
    /// HTTPS image URL.
    pub url: String,
}

/// Versioned platform-neutral account settings.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountSettings {
    /// Settings schema version.
    pub schema_version: u32,
    /// Default reaction text or shortcode.
    pub quick_reaction: Option<String>,
    /// Whether typing indicators are authored and rendered.
    pub typing_indicators_enabled: bool,
    /// `full`, `sender`, or `hidden`.
    pub notification_content_privacy: String,
    /// Selected HTTPS media service, absent for deployment default.
    pub media_service: Option<String>,
    /// Semantic image quality (`low`, `balanced`, or `high`).
    pub image_upload_quality: String,
    /// Semantic video quality (`low`, `balanced`, or `high`).
    pub video_upload_quality: String,
    /// Whether supported platforms should retain HDR.
    pub preserve_hdr: bool,
    /// BCP-47 transcription locale, absent for system selection.
    pub transcription_locale: Option<String>,
    /// Canonical JSON array of portable domain replacement rules.
    pub domain_replacement_rules_json: String,
    /// Portable wallpaper identifier shared by clients, absent for the default.
    pub chat_wallpaper: Option<String>,
    /// Validated portable custom emoji references.
    pub custom_emoji_references: Vec<CustomEmojiReference>,
    /// Stable saved sticker references.
    pub saved_sticker_references: Vec<String>,
    /// Shared seasonal appearance choice, absent for default.
    pub seasonal_appearance: Option<String>,
    /// Canonical object containing fields unknown to this SDK version.
    pub unknown_json: String,
}

/// Result of a committed settings patch.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountSettingsMutation {
    /// Complete effective settings after the patch.
    pub settings: AccountSettings,
    /// Canonical complete JSON used by kind-30079 synchronization.
    pub canonical_json: String,
    /// Durable account revision after commit.
    pub revision: i64,
}

/// Logical media-operation state.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum AccountMediaOperationState {
    /// Durable intent exists but no platform transfer has started.
    Prepared,
    /// A platform executor owns a bounded lease.
    Running,
    /// Authenticated attachment metadata was committed.
    Completed,
    /// Retryable failure awaits policy or user action.
    Retryable,
    /// Permanent failure cannot be retried without new input.
    Failed,
    /// Cancellation completed before message publication.
    Cancelled,
}

/// Confidentiality policy for one platform-executed media transfer.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum AccountMediaProtection {
    /// Encrypt upload bytes before they leave private staging.
    Encrypted,
    /// Upload the selected bytes without content encryption for a public asset.
    Public,
}

/// One durable logical media operation.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountMediaOperation {
    /// SDK-generated stable operation ID.
    pub id: String,
    /// User-action idempotency key.
    pub command_id: String,
    /// Owning conversation, empty for profile media.
    pub conversation_id: String,
    /// Authenticated source message for downloads, empty for uploads.
    pub source_message_id: String,
    /// `upload` or `download`.
    pub direction: String,
    /// Whether the platform encrypts the transferred object.
    pub protection: AccountMediaProtection,
    /// Current durable state.
    pub state: AccountMediaOperationState,
    /// Platform-computed stable source fingerprint.
    pub source_fingerprint: String,
    /// Final validated metadata, absent before completion.
    pub attachment: Option<AttachmentMetadata>,
    /// Processed byte count when known.
    pub byte_count: Option<u64>,
    /// Number of started platform attempts.
    pub attempt_count: u32,
    /// Stable redacted failure category.
    pub last_error_category: String,
    /// Last durable update time.
    pub updated_at: i64,
}

/// One stable page of non-terminal media operations for a platform executor.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountMediaOperationPage {
    /// Recoverable operations ordered by stable operation ID.
    pub operations: Vec<AccountMediaOperation>,
    /// Cursor for the next page, absent when this page reaches the end.
    pub next_cursor: Option<String>,
}

/// Exclusive bounded lease for one platform-owned media transfer.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountMediaLease {
    /// Host-generated idempotent lease identifier.
    pub lease_id: String,
    /// Absolute wall-clock expiry persisted by Rust.
    pub expires_at: i64,
    /// Complete logical operation after the claim commits.
    pub operation: AccountMediaOperation,
}

/// Logical state of one attachment-dependent outgoing message.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum PendingMediaMessageState {
    /// Durable message intent is waiting for one or more platform uploads.
    WaitingForUploads,
    /// Every attachment completed and the message can be authored.
    ReadyToPublish,
    /// The complete outgoing message operation was committed.
    Published,
    /// A required attachment failed permanently.
    Failed,
    /// The user cancelled a required attachment before publication.
    Cancelled,
}

/// One platform source selected for an attachment-dependent message.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct MediaPreparationInput {
    /// Stable fingerprint of the private staged source bytes and transformation intent.
    pub source_fingerprint: String,
}

pub(crate) struct ProductMediaMessagePreparation<'a> {
    pub(crate) command_id: &'a str,
    pub(crate) conversation_id: &'a str,
    pub(crate) content: &'a MessageContentInput,
    pub(crate) reply_to_message_id: &'a str,
    pub(crate) sources: Vec<MediaPreparationInput>,
    pub(crate) command_hash: &'a str,
    pub(crate) draft_match: Option<ProductDraftMatch>,
    pub(crate) now: i64,
}

#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
pub(crate) struct ProductDraftMatch {
    pub(crate) draft_json: String,
    pub(crate) reply_to_message_id: String,
}

struct ProductMediaOperationPreparation<'a> {
    command_id: &'a str,
    conversation_id: Option<String>,
    source_message_id: Option<&'a str>,
    kind: ProductMediaOperationKind,
    direction: &'a str,
    protection: AccountMediaProtection,
    source_fingerprint: &'a str,
    attachment: Option<AttachmentMetadata>,
    now: i64,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ProductMediaOperationKind {
    Transfer,
    ConversationIcon,
}

impl ProductMediaOperationKind {
    const fn as_str(self) -> &'static str {
        match self {
            Self::Transfer => "transfer",
            Self::ConversationIcon => "conversation_icon",
        }
    }

    fn from_str(value: &str) -> Result<Self, SoftchatError> {
        match value {
            "transfer" => Ok(Self::Transfer),
            "conversation_icon" => Ok(Self::ConversationIcon),
            _ => Err(SoftchatError::InvalidPersistenceResult),
        }
    }
}

pub(crate) struct ProductAssetReplacementPreparation<'a> {
    pub(crate) command_id: &'a str,
    pub(crate) target: AssetReplacementTarget,
    pub(crate) conversation_id: &'a str,
    pub(crate) source_fingerprint: &'a str,
    pub(crate) command_hash: &'a str,
    pub(crate) now: i64,
}

/// Durable message intent with its ordered logical attachment operations.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct PendingMediaMessage {
    /// Stable user-action idempotency key, reused by the eventual send operation.
    pub command_id: String,
    /// Owning conversation.
    pub conversation_id: String,
    /// Plain message text retained before publication.
    pub text: String,
    /// Exact canonical custom-emoji tags.
    pub emoji_tags: Vec<Vec<String>>,
    /// Optional replied message ID, empty when unrelated.
    pub reply_to_message_id: String,
    /// Current logical dependency state.
    pub state: PendingMediaMessageState,
    /// Ordered required upload operations.
    pub attachments: Vec<AccountMediaOperation>,
    /// Published message ID, empty before publication.
    pub result_message_id: String,
    /// Last durable update time.
    pub updated_at: i64,
}

/// Public asset changed after one durable upload dependency completes.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum AssetReplacementTarget {
    /// Replace the local account's public profile picture URL.
    ProfilePicture,
    /// Replace the local account's public profile banner URL.
    ProfileBanner,
    /// Replace one conversation's public subject icon.
    ConversationIcon,
}

/// Logical state of one upload-dependent public asset replacement.
#[derive(Clone, Copy, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Enum))]
pub enum PendingAssetReplacementState {
    /// Durable intent exists and waits for the platform upload.
    WaitingForUpload,
    /// The public upload completed and the dependent event can be authored.
    ReadyToPublish,
    /// The dependent profile or subject event was durably queued.
    Published,
    /// The required upload failed permanently.
    Failed,
    /// The user cancelled the upload before publication.
    Cancelled,
}

/// One durable public asset replacement and its platform upload dependency.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct PendingAssetReplacement {
    /// Stable user-action idempotency key reused by the dependent event.
    pub command_id: String,
    /// Profile or conversation field replaced by the completed upload.
    pub target: AssetReplacementTarget,
    /// Owning conversation for a subject icon, empty for profile assets.
    pub conversation_id: String,
    /// Current dependency and publication state.
    pub state: PendingAssetReplacementState,
    /// Single public upload operation.
    pub operation: AccountMediaOperation,
    /// Durable outgoing operation ID, empty before publication.
    pub result_operation_id: String,
    /// Last durable update time.
    pub updated_at: i64,
}

/// Result of acknowledging one completed attachment lease.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct MediaCompletionResult {
    /// Completed logical attachment operation.
    pub operation: AccountMediaOperation,
    /// Newly published message, profile, or subject operation when ready.
    pub completed_operation: Option<ProductOperationResult>,
}

/// One attachment projected into an account media gallery.
#[derive(Clone, Debug, Deserialize, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountMediaItem {
    /// Owning message ID.
    pub message_id: String,
    /// Owning conversation ID.
    pub conversation_id: String,
    /// Authenticated message author.
    pub author_public_key: String,
    /// Message timestamp.
    pub created_at: i64,
    /// Validated attachment metadata.
    pub attachment: AttachmentMetadata,
}

/// High-level command result with the complete effective message when relevant.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct ProductOperationResult {
    /// Durable delivery aggregate.
    pub operation: OperationSnapshot,
    /// Durable account revision after the command.
    pub revision: i64,
    /// False when an identical command was already committed.
    pub inserted: bool,
    /// Created or affected message, absent for non-message commands.
    pub message: Option<AccountMessageView>,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase", deny_unknown_fields)]
struct StoredDraft {
    text: String,
    emoji_tags: Vec<Vec<String>>,
    #[serde(default)]
    spans: Vec<DraftSpan>,
    #[serde(default)]
    attachment_operation_ids: Vec<String>,
}

#[derive(Clone, Debug)]
struct ProductMessageRow {
    id: String,
    conversation_id: String,
    author_public_key: String,
    created_at: i64,
    rumor_json: String,
    deleted: bool,
    edit_id: Option<String>,
    edit_rumor_json: Option<String>,
    reply_count: u32,
}

enum ProductMessageQuery<'a> {
    ConversationPage {
        conversation_id: &'a str,
        reply_to: Option<&'a str>,
        cursor: Option<&'a MessageCursor>,
        limit: u32,
    },
    Ids(&'a [String]),
}

#[derive(Clone, Debug)]
struct ProductConversationRow {
    id: String,
    archived: bool,
    pinned: bool,
    read_at: Option<i64>,
    read_message_id: String,
    forced_unread: bool,
    draft_json: Option<String>,
    draft_reply_to: Option<String>,
    draft_updated_at: Option<i64>,
    member_public_keys: Vec<String>,
    subject_rumor_json: Option<String>,
    subject_icon_event_id: Option<String>,
    subject_icon_json: Option<String>,
    last_message_id: Option<String>,
    message_count: u32,
    unread_count: u32,
    last_activity_at: i64,
}

#[derive(Clone, Debug)]
struct RawProductConversationRow {
    id: String,
    archived: bool,
    pinned: bool,
    read_at: Option<i64>,
    read_message_id: String,
    forced_unread: bool,
    draft_json: Option<String>,
    draft_reply_to: Option<String>,
    draft_updated_at: Option<i64>,
    member_public_keys_csv: Option<String>,
    subject_rumor_json: Option<String>,
    subject_icon_event_id: Option<String>,
    subject_icon_json: Option<String>,
    last_message_id: Option<String>,
    message_count: u32,
    unread_count: u32,
    last_activity_at: i64,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct StoredReadStateSnapshot {
    #[serde(default = "default_read_state_schema")]
    schema_version: u32,
    chats: Vec<StoredReadStateEntry>,
    #[serde(flatten)]
    unknown: BTreeMap<String, Value>,
}

#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct StoredReadStateEntry {
    chat_key: String,
    #[serde(default)]
    conversation_id: String,
    #[serde(default)]
    read_at: Option<Option<f64>>,
    #[serde(default)]
    read_at_unix: Option<i64>,
    #[serde(default)]
    read_message_id: String,
    #[serde(default)]
    forced_unread: Option<bool>,
    #[serde(default)]
    updated_at: Option<i64>,
    #[serde(default)]
    update_id: String,
    #[serde(flatten)]
    unknown: BTreeMap<String, Value>,
}

#[derive(Clone, Debug)]
pub(crate) struct ReadStateMutationPreview {
    pub(crate) state: AccountReadState,
    pub(crate) canonical_json: String,
}

#[derive(Clone, Debug)]
pub(crate) struct ParsedReadStateSnapshot {
    pub(crate) entries: Vec<ParsedReadStateEntry>,
    pub(crate) unknown_json: String,
}

#[derive(Clone, Debug)]
pub(crate) struct ParsedReadStateEntry {
    pub(crate) state: AccountReadState,
    pub(crate) unknown_json: String,
}

#[derive(Clone, Debug)]
struct StoredReadStateRow {
    state: AccountReadState,
    unknown_json: String,
}

struct ProductSubject {
    icon_event_id: String,
    text: String,
    icon: Option<AttachmentMetadata>,
    emoji_tags: Vec<Vec<String>>,
}

fn classify_product_subject(rumor: RumorEvent) -> Result<ProductSubject, SoftchatError> {
    let event_id = rumor.id.clone();
    let parsed = classify_subject(rumor.clone())?;
    let icon = subject_icon_metadata_from_validated(&rumor, &parsed)?;
    Ok(ProductSubject {
        icon_event_id: if icon.is_some() {
            event_id
        } else {
            String::new()
        },
        text: parsed.subject,
        icon,
        emoji_tags: parsed.emoji_tags,
    })
}

pub(crate) fn conversation_icon_source_fingerprint(
    conversation_id: &str,
    subject_event_id: &str,
    attachment: &AttachmentMetadata,
) -> Result<String, SoftchatError> {
    command_hash(&(
        "conversation_icon_download",
        conversation_id,
        subject_event_id,
        attachment,
    ))
}

fn has_conversation_icon_integrity(metadata: &AttachmentMetadata) -> bool {
    metadata.encryption.is_some() || metadata.sha256.is_some()
}

fn parse_stored_subject_icon(
    event_id: Option<String>,
    attachment_json: Option<String>,
) -> Result<Option<(String, AttachmentMetadata)>, SoftchatError> {
    match (event_id, attachment_json) {
        (None, None) => Ok(None),
        (Some(event_id), Some(attachment_json)) => {
            NostrEventId::from_hex(&event_id)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            let attachment = serde_json::from_str::<AttachmentMetadata>(&attachment_json)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            attachment
                .to_tag()
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            Ok(Some((event_id, attachment)))
        }
        _ => Err(SoftchatError::InvalidPersistenceResult),
    }
}

impl AccountDatabase {
    pub(crate) fn get_or_create_product_conversation(
        &mut self,
        own_public_key: &str,
        member_public_keys: Vec<String>,
        now: i64,
    ) -> Result<ConversationRecord, SoftchatError> {
        validate_timestamp(now)?;
        let members = canonical_members(own_public_key, member_public_keys)?;
        let id = conversation_id(members.clone())?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let inserted = transaction
            .execute(
                "INSERT OR IGNORE INTO conversation_state
                 (conversation_id, created_at, updated_at)
                 VALUES (?1, ?2, ?2)",
                params![id, now],
            )
            .map_err(map_sqlite_error)?
            != 0;
        for member in &members {
            transaction
                .execute(
                    "INSERT OR IGNORE INTO conversation_members
                     (conversation_id, public_key) VALUES (?1, ?2)",
                    params![id, member],
                )
                .map_err(map_sqlite_error)?;
        }
        if inserted {
            bump_revision(&transaction)?;
        }
        transaction.commit().map_err(map_sqlite_error)?;
        self.product_conversation(&id, own_public_key)?
            .ok_or(SoftchatError::InvalidPersistenceResult)
    }

    pub(crate) fn product_conversation(
        &self,
        conversation_id: &str,
        own_public_key: &str,
    ) -> Result<Option<ConversationRecord>, SoftchatError> {
        validate_conversation_id(conversation_id)?;
        NostrPublicKey::from_hex(own_public_key)?;
        let state = self
            .connection
            .query_row(
                "SELECT cs.archived, cs.pinned, rs.read_at,
                        COALESCE(rs.read_message_id, ''),
                        COALESCE(rs.forced_unread, 0),
                        cs.draft_json, cs.draft_reply_to, cs.draft_updated_at,
                        cs.created_at, cs.updated_at
                 FROM conversation_state cs
                 LEFT JOIN conversation_read_state rs
                   ON rs.conversation_id = cs.conversation_id
                 WHERE cs.conversation_id = ?1",
                [conversation_id],
                |row| {
                    Ok((
                        row.get::<_, i64>(0)? != 0,
                        row.get::<_, i64>(1)? != 0,
                        row.get::<_, Option<i64>>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, i64>(4)? != 0,
                        row.get::<_, Option<String>>(5)?,
                        row.get::<_, Option<String>>(6)?,
                        row.get::<_, Option<i64>>(7)?,
                        row.get::<_, i64>(8)?,
                        row.get::<_, i64>(9)?,
                    ))
                },
            )
            .optional()
            .map_err(map_sqlite_error)?;
        let Some((
            archived,
            pinned,
            read_at,
            read_message_id,
            forced_unread,
            draft_json,
            draft_reply_to,
            draft_updated_at,
            created_at,
            updated_at,
        )) = state
        else {
            return Ok(None);
        };
        let members = self.product_conversation_members(conversation_id)?;
        let last_message_id = self
            .connection
            .query_row(
                "SELECT p.logical_event_id
                 FROM projections p
                 WHERE p.conversation_id = ?1 AND p.kind = 'chat_message'
                   AND p.ephemeral = 0
                   AND NOT EXISTS (
                     SELECT 1 FROM projection_targets dt
                     JOIN projections d ON d.logical_event_id = dt.logical_event_id
                     WHERE dt.target_event_id = p.logical_event_id
                       AND d.kind = 'deletion'
                       AND d.author_public_key = p.author_public_key
                   )
                 ORDER BY p.created_at DESC, p.logical_event_id ASC LIMIT 1",
                [conversation_id],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?;
        let last_message = last_message_id
            .as_deref()
            .map(|id| self.product_message(id))
            .transpose()?
            .flatten();
        let message_count = self
            .connection
            .query_row(
                "SELECT COUNT(*)
                 FROM projections p
                 WHERE p.conversation_id = ?1 AND p.kind = 'chat_message'
                   AND p.ephemeral = 0
                   AND NOT EXISTS (
                     SELECT 1 FROM projection_targets dt
                     JOIN projections d ON d.logical_event_id = dt.logical_event_id
                     WHERE dt.target_event_id = p.logical_event_id
                       AND d.kind = 'deletion'
                       AND d.author_public_key = p.author_public_key
                   )",
                [conversation_id],
                |row| row.get::<_, u32>(0),
            )
            .map_err(map_sqlite_error)?;
        let unread_count = self
            .connection
            .query_row(
                "SELECT COUNT(*)
                 FROM projections p
                 WHERE p.conversation_id = ?1 AND p.kind = 'chat_message'
                   AND p.ephemeral = 0 AND p.author_public_key <> ?2
                   AND (
                     ?3 = 1
                     OR ?4 IS NULL
                     OR p.created_at > ?4
                     OR (
                       p.created_at = ?4 AND ?5 <> ''
                       AND p.logical_event_id < ?5
                     )
                   )
                   AND NOT EXISTS (
                     SELECT 1 FROM projection_targets dt
                     JOIN projections d ON d.logical_event_id = dt.logical_event_id
                     WHERE dt.target_event_id = p.logical_event_id
                       AND d.kind = 'deletion'
                       AND d.author_public_key = p.author_public_key
                   )",
                params![
                    conversation_id,
                    own_public_key,
                    i64::from(forced_unread),
                    read_at,
                    read_message_id
                ],
                |row| row.get::<_, u32>(0),
            )
            .map_err(map_sqlite_error)?;
        let subject = self.product_subject(conversation_id)?;
        let draft = match (draft_json, draft_updated_at) {
            (Some(value), Some(updated_at)) => {
                let stored: StoredDraft = serde_json::from_str(&value)
                    .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
                Some(AccountDraft {
                    text: stored.text,
                    emoji_tags: stored.emoji_tags,
                    spans: stored.spans,
                    reply_to_message_id: draft_reply_to.unwrap_or_default(),
                    attachment_operation_ids: stored.attachment_operation_ids,
                    updated_at,
                })
            }
            (None, None) => None,
            _ => return Err(SoftchatError::InvalidPersistenceResult),
        };
        let last_activity_at = last_message
            .as_ref()
            .map_or(updated_at.max(created_at), |message| message.created_at);
        let read_cursor = match read_at {
            Some(created_at) if !read_message_id.is_empty() => Some(MessageCursor {
                created_at,
                message_id: read_message_id,
            }),
            Some(created_at) => self
                .connection
                .query_row(
                    "SELECT logical_event_id
                     FROM projections
                     WHERE conversation_id = ?1 AND kind = 'chat_message'
                       AND created_at = ?2
                     ORDER BY logical_event_id ASC LIMIT 1",
                    params![conversation_id, created_at],
                    |row| row.get::<_, String>(0),
                )
                .optional()
                .map_err(map_sqlite_error)?
                .map(|message_id| MessageCursor {
                    created_at,
                    message_id,
                }),
            None => None,
        };
        Ok(Some(ConversationRecord {
            id: conversation_id.to_owned(),
            member_public_keys: members,
            subject: subject.text,
            subject_icon: subject.icon,
            last_message,
            message_count,
            unread_count,
            archived,
            pinned,
            read_cursor,
            forced_unread,
            draft,
            last_activity_at,
        }))
    }

    pub(crate) fn product_conversations(
        &self,
        own_public_key: &str,
        archived: bool,
        cursor: Option<ConversationCursor>,
        limit: u32,
    ) -> Result<ConversationPage, SoftchatError> {
        validate_limit(limit)?;
        NostrPublicKey::from_hex(own_public_key)?;
        if let Some(value) = &cursor {
            validate_conversation_id(&value.conversation_id)?;
            validate_timestamp(value.last_activity_at)?;
        }
        let fetch_limit = limit
            .checked_add(1)
            .ok_or(SoftchatError::InvalidAccountOperation)?;
        let (cursor_pinned, cursor_activity, cursor_id, has_cursor) = cursor
            .map(|value| {
                (
                    i64::from(value.pinned),
                    value.last_activity_at,
                    value.conversation_id,
                    1_i64,
                )
            })
            .unwrap_or((0, 0, String::new(), 0));
        let mut statement = self
            .connection
            .prepare(
                "WITH live_messages AS (
                   SELECT p.*
                   FROM projections p
                   WHERE p.kind = 'chat_message' AND p.ephemeral = 0
                     AND NOT EXISTS (
                       SELECT 1 FROM projection_targets dt
                       JOIN projections deletion
                         ON deletion.logical_event_id = dt.logical_event_id
                       WHERE dt.target_event_id = p.logical_event_id
                         AND deletion.kind = 'deletion'
                         AND deletion.author_public_key = p.author_public_key
                     )
                 )
                 , message_stats AS (
                   SELECT
                     message.conversation_id,
                     MAX(message.created_at) AS last_activity,
                     COUNT(*) AS message_count,
                     SUM(
                       CASE
                         WHEN message.author_public_key <> ?1
                           AND (
                             COALESCE(read_state.forced_unread, 0) = 1
                             OR read_state.read_at IS NULL
                             OR message.created_at > read_state.read_at
                             OR (
                               message.created_at = read_state.read_at
                               AND COALESCE(read_state.read_message_id, '') <> ''
                               AND message.logical_event_id
                                 < read_state.read_message_id
                             )
                           )
                         THEN 1 ELSE 0
                       END
                     ) AS unread_count
                   FROM live_messages message
                   LEFT JOIN conversation_read_state read_state
                     ON read_state.conversation_id = message.conversation_id
                   GROUP BY message.conversation_id
                 )
                 , ranked_messages AS (
                   SELECT
                     message.conversation_id,
                     message.logical_event_id,
                     ROW_NUMBER() OVER (
                       PARTITION BY message.conversation_id
                       ORDER BY message.created_at DESC,
                                message.logical_event_id ASC
                     ) AS row_number
                   FROM live_messages message
                 )
                 SELECT
                   cs.conversation_id,
                   cs.archived,
                   cs.pinned,
                   read_state.read_at,
                   CASE
                     WHEN read_state.read_at IS NOT NULL
                       AND COALESCE(read_state.read_message_id, '') = ''
                     THEN COALESCE(
                       (
                         SELECT boundary.logical_event_id
                         FROM projections boundary
                         WHERE boundary.conversation_id = cs.conversation_id
                           AND boundary.kind = 'chat_message'
                           AND boundary.created_at = read_state.read_at
                         ORDER BY boundary.logical_event_id ASC
                         LIMIT 1
                       ),
                       ''
                     )
                     ELSE COALESCE(read_state.read_message_id, '')
                   END,
                   COALESCE(read_state.forced_unread, 0),
                   cs.draft_json,
                   cs.draft_reply_to,
                   cs.draft_updated_at,
                   (
                     SELECT group_concat(
                       member.public_key,
                       ',' ORDER BY member.public_key ASC
                     )
                     FROM conversation_members member
                     WHERE member.conversation_id = cs.conversation_id
                   ),
                   (
                     SELECT COALESCE(subject_rumor.canonical_json, '')
                     FROM projections subject
                     LEFT JOIN authenticated_rumors subject_rumor
                       ON subject_rumor.rumor_id = subject.logical_event_id
                     WHERE subject.conversation_id = cs.conversation_id
                       AND subject.kind = 'subject'
                     ORDER BY subject.created_at DESC,
                              subject.logical_event_id ASC
                     LIMIT 1
                   ),
                   subject_icon.subject_event_id,
                   subject_icon.attachment_json,
                   latest_message.logical_event_id,
                   COALESCE(message_stats.message_count, 0),
                   COALESCE(message_stats.unread_count, 0),
                   COALESCE(
                     message_stats.last_activity,
                     MAX(cs.updated_at, cs.created_at)
                   ) AS activity
                 FROM conversation_state cs
                 LEFT JOIN conversation_read_state read_state
                   ON read_state.conversation_id = cs.conversation_id
                 LEFT JOIN message_stats
                   ON message_stats.conversation_id = cs.conversation_id
                 LEFT JOIN ranked_messages latest_message
                   ON latest_message.conversation_id = cs.conversation_id
                  AND latest_message.row_number = 1
                 LEFT JOIN conversation_subject_icons subject_icon
                   ON subject_icon.conversation_id = cs.conversation_id
                 WHERE cs.archived = ?2
                   AND (
                     ?3 = 0
                     OR cs.pinned < ?4
                     OR (
                       cs.pinned = ?4
                       AND COALESCE(
                         message_stats.last_activity,
                         MAX(cs.updated_at, cs.created_at)
                       ) < ?5
                     )
                     OR (
                       cs.pinned = ?4
                       AND COALESCE(
                         message_stats.last_activity,
                         MAX(cs.updated_at, cs.created_at)
                       ) = ?5
                       AND cs.conversation_id > ?6
                     )
                   )
                 ORDER BY cs.pinned DESC, activity DESC, cs.conversation_id ASC
                 LIMIT ?7",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(
                params![
                    own_public_key,
                    i64::from(archived),
                    has_cursor,
                    cursor_pinned,
                    cursor_activity,
                    cursor_id,
                    fetch_limit
                ],
                |row| {
                    Ok(RawProductConversationRow {
                        id: row.get(0)?,
                        archived: row.get::<_, i64>(1)? != 0,
                        pinned: row.get::<_, i64>(2)? != 0,
                        read_at: row.get(3)?,
                        read_message_id: row.get(4)?,
                        forced_unread: row.get::<_, i64>(5)? != 0,
                        draft_json: row.get(6)?,
                        draft_reply_to: row.get(7)?,
                        draft_updated_at: row.get(8)?,
                        member_public_keys_csv: row.get(9)?,
                        subject_rumor_json: row.get(10)?,
                        subject_icon_event_id: row.get(11)?,
                        subject_icon_json: row.get(12)?,
                        last_message_id: row.get(13)?,
                        message_count: row.get(14)?,
                        unread_count: row.get(15)?,
                        last_activity_at: row.get(16)?,
                    })
                },
            )
            .map_err(map_sqlite_error)?;
        let mut page_rows = Vec::new();
        for row in rows {
            let row = row.map_err(map_sqlite_error)?;
            page_rows.push(ProductConversationRow {
                id: row.id,
                archived: row.archived,
                pinned: row.pinned,
                read_at: row.read_at,
                read_message_id: row.read_message_id,
                forced_unread: row.forced_unread,
                draft_json: row.draft_json,
                draft_reply_to: row.draft_reply_to,
                draft_updated_at: row.draft_updated_at,
                member_public_keys: parse_public_key_csv(row.member_public_keys_csv)?,
                subject_rumor_json: row.subject_rumor_json,
                subject_icon_event_id: row.subject_icon_event_id,
                subject_icon_json: row.subject_icon_json,
                last_message_id: row.last_message_id,
                message_count: row.message_count,
                unread_count: row.unread_count,
                last_activity_at: row.last_activity_at,
            });
        }
        let has_more = page_rows.len()
            > usize::try_from(limit).map_err(|_| SoftchatError::InvalidAccountOperation)?;
        page_rows
            .truncate(usize::try_from(limit).map_err(|_| SoftchatError::InvalidAccountOperation)?);
        let message_ids = page_rows
            .iter()
            .filter_map(|row| row.last_message_id.clone())
            .collect::<Vec<_>>();
        let messages = self
            .product_message_views(ProductMessageQuery::Ids(&message_ids))?
            .into_iter()
            .map(|message| (message.id.clone(), message))
            .collect::<BTreeMap<_, _>>();
        let mut conversations = Vec::with_capacity(page_rows.len());
        for row in page_rows {
            let subject = row
                .subject_rumor_json
                .map(|json| {
                    serde_json::from_str::<RumorEvent>(&json)
                        .map_err(|_| SoftchatError::InvalidPersistenceResult)
                        .and_then(classify_product_subject)
                })
                .transpose()?;
            let subject_icon =
                parse_stored_subject_icon(row.subject_icon_event_id, row.subject_icon_json)?
                    .map(|(_, attachment)| attachment);
            let draft = product_draft(row.draft_json, row.draft_reply_to, row.draft_updated_at)?;
            let read_cursor = row.read_at.map(|created_at| MessageCursor {
                created_at,
                message_id: row.read_message_id,
            });
            let last_message = row
                .last_message_id
                .as_ref()
                .map(|id| {
                    messages
                        .get(id)
                        .cloned()
                        .ok_or(SoftchatError::InvalidPersistenceResult)
                })
                .transpose()?;
            conversations.push(ConversationRecord {
                id: row.id,
                member_public_keys: row.member_public_keys,
                subject: subject.map_or_else(String::new, |value| value.text),
                subject_icon,
                last_message,
                message_count: row.message_count,
                unread_count: row.unread_count,
                archived: row.archived,
                pinned: row.pinned,
                read_cursor,
                forced_unread: row.forced_unread,
                draft,
                last_activity_at: row.last_activity_at,
            });
        }
        let next_cursor = if has_more {
            conversations.last().map(|value| ConversationCursor {
                pinned: value.pinned,
                last_activity_at: value.last_activity_at,
                conversation_id: value.id.clone(),
            })
        } else {
            None
        };
        Ok(ConversationPage {
            conversations,
            next_cursor,
        })
    }

    pub(crate) fn product_conversation_count(&self, archived: bool) -> Result<u32, SoftchatError> {
        self.connection
            .query_row(
                "SELECT COUNT(*) FROM conversation_state WHERE archived = ?1",
                [i64::from(archived)],
                |row| row.get(0),
            )
            .map_err(map_sqlite_error)
    }

    pub(crate) fn product_has_unread(
        &self,
        own_public_key: &str,
        include_archived: bool,
    ) -> Result<bool, SoftchatError> {
        NostrPublicKey::from_hex(own_public_key)?;
        self.connection
            .query_row(
                "SELECT EXISTS (
                   SELECT 1
                   FROM conversation_state cs
                   LEFT JOIN conversation_read_state rs
                     ON rs.conversation_id = cs.conversation_id
                   JOIN projections p ON p.conversation_id = cs.conversation_id
                   WHERE (?1 = 1 OR cs.archived = 0)
                     AND p.kind = 'chat_message'
                     AND p.ephemeral = 0
                     AND p.author_public_key <> ?2
                     AND (
                       COALESCE(rs.forced_unread, 0) = 1
                       OR rs.read_at IS NULL
                       OR p.created_at > rs.read_at
                       OR (
                         p.created_at = rs.read_at
                         AND rs.read_message_id <> ''
                         AND p.logical_event_id < rs.read_message_id
                       )
                     )
                     AND NOT EXISTS (
                       SELECT 1
                       FROM projection_targets dt
                       JOIN projections d ON d.logical_event_id = dt.logical_event_id
                       WHERE dt.target_event_id = p.logical_event_id
                         AND d.kind = 'deletion'
                         AND d.author_public_key = p.author_public_key
                     )
                 )",
                params![i64::from(include_archived), own_public_key],
                |row| row.get::<_, i64>(0).map(|value| value != 0),
            )
            .map_err(map_sqlite_error)
    }

    pub(crate) fn search_product_conversations(
        &self,
        own_public_key: &str,
        query: &str,
        archived: Option<bool>,
        limit: u32,
    ) -> Result<Vec<ConversationRecord>, SoftchatError> {
        NostrPublicKey::from_hex(own_public_key)?;
        validate_limit(limit)?;
        let pattern = escaped_like_pattern(query)?;
        let archived = archived.map(i64::from);
        let mut statement = self
            .connection
            .prepare(
                r"SELECT cs.conversation_id
                 FROM conversation_state cs
                 WHERE (?1 IS NULL OR cs.archived = ?1)
                   AND (
                     lower(cs.conversation_id) LIKE ?2 ESCAPE '\'
                     OR lower((
                       SELECT subject.content
                       FROM projections subject
                       WHERE subject.conversation_id = cs.conversation_id
                         AND subject.kind = 'subject'
                       ORDER BY subject.created_at DESC, subject.logical_event_id ASC
                       LIMIT 1
                     )) LIKE ?2 ESCAPE '\'
                     OR EXISTS (
                       SELECT 1
                       FROM conversation_members member
                       LEFT JOIN local_contacts local
                         ON local.public_key = member.public_key
                       WHERE member.conversation_id = cs.conversation_id
                         AND (
                           lower(member.public_key) LIKE ?2 ESCAPE '\'
                           OR lower(COALESCE(local.name, '')) LIKE ?2 ESCAPE '\'
                           OR EXISTS (
                             SELECT 1 FROM json_tree((
                               SELECT profile.content
                               FROM projections profile
                               WHERE profile.kind = 'user_metadata'
                                 AND profile.author_public_key = member.public_key
                               ORDER BY profile.created_at DESC, profile.logical_event_id ASC
                               LIMIT 1
                             )) metadata
                             WHERE metadata.type = 'text'
                               AND lower(metadata.atom) LIKE ?2 ESCAPE '\'
                           )
                         )
                     )
                   )
                 ORDER BY cs.pinned DESC, cs.updated_at DESC, cs.conversation_id ASC
                 LIMIT ?3",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params![archived, pattern, limit], |row| {
                row.get::<_, String>(0)
            })
            .map_err(map_sqlite_error)?;
        let mut conversations = Vec::new();
        for row in rows {
            let id = row.map_err(map_sqlite_error)?;
            conversations.push(
                self.product_conversation(&id, own_public_key)?
                    .ok_or(SoftchatError::InvalidPersistenceResult)?,
            );
        }
        Ok(conversations)
    }

    pub(crate) fn product_message(
        &self,
        message_id: &str,
    ) -> Result<Option<AccountMessageView>, SoftchatError> {
        NostrEventId::from_hex(message_id)?;
        let ids = vec![message_id.to_owned()];
        Ok(self
            .product_message_views(ProductMessageQuery::Ids(&ids))?
            .into_iter()
            .next())
    }

    pub(crate) fn product_messages(
        &self,
        conversation_id: &str,
        cursor: Option<MessageCursor>,
        limit: u32,
    ) -> Result<MessagePage, SoftchatError> {
        self.product_messages_matching(conversation_id, None, cursor, limit)
    }

    pub(crate) fn product_message_replies(
        &self,
        message_id: &str,
        cursor: Option<MessageCursor>,
        limit: u32,
    ) -> Result<MessagePage, SoftchatError> {
        validate_limit(limit)?;
        if let Some(value) = &cursor {
            validate_timestamp(value.created_at)?;
            NostrEventId::from_hex(&value.message_id)?;
        }
        let Some(root) = self.product_message(message_id)? else {
            return Ok(MessagePage {
                messages: Vec::new(),
                next_cursor: None,
            });
        };
        self.product_messages_matching(&root.conversation_id, Some(message_id), cursor, limit)
    }

    fn product_messages_matching(
        &self,
        conversation_id: &str,
        reply_to: Option<&str>,
        cursor: Option<MessageCursor>,
        limit: u32,
    ) -> Result<MessagePage, SoftchatError> {
        validate_conversation_id(conversation_id)?;
        validate_limit(limit)?;
        if let Some(value) = &cursor {
            validate_timestamp(value.created_at)?;
            NostrEventId::from_hex(&value.message_id)?;
        }
        let fetch_limit = limit
            .checked_add(1)
            .ok_or(SoftchatError::InvalidAccountOperation)?;
        let mut messages = self.product_message_views(ProductMessageQuery::ConversationPage {
            conversation_id,
            reply_to,
            cursor: cursor.as_ref(),
            limit: fetch_limit,
        })?;
        let requested =
            usize::try_from(limit).map_err(|_| SoftchatError::InvalidAccountOperation)?;
        let has_more = messages.len() > requested;
        messages.truncate(requested);
        let next_cursor = if has_more {
            messages.last().map(|message| MessageCursor {
                created_at: message.created_at,
                message_id: message.id.clone(),
            })
        } else {
            None
        };
        Ok(MessagePage {
            messages,
            next_cursor,
        })
    }

    pub(crate) fn product_messages_by_ids(
        &self,
        message_ids: &[String],
    ) -> Result<Vec<AccountMessageView>, SoftchatError> {
        if message_ids.len() > MAX_PRODUCT_MESSAGE_LOOKUPS {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        let mut distinct = BTreeSet::new();
        let mut canonical_ids = Vec::with_capacity(message_ids.len());
        for message_id in message_ids {
            let parsed = NostrEventId::from_hex(message_id)?;
            if !distinct.insert(parsed) {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            canonical_ids.push(parsed.to_hex());
        }
        let mut messages: BTreeMap<_, _> = self
            .product_message_views(ProductMessageQuery::Ids(&canonical_ids))?
            .into_iter()
            .map(|message| (message.id.clone(), message))
            .collect();
        Ok(canonical_ids
            .iter()
            .filter_map(|id| messages.remove(id))
            .collect())
    }

    pub(crate) fn product_message_context(
        &self,
        message_id: &str,
        before: u32,
        after: u32,
    ) -> Result<Vec<AccountMessageView>, SoftchatError> {
        if before > MAX_PRODUCT_QUERY_PAGE || after > MAX_PRODUCT_QUERY_PAGE {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        let target = self
            .product_message(message_id)?
            .ok_or(SoftchatError::InvalidAccountOperation)?;
        let mut before_statement = self
            .connection
            .prepare(
                "SELECT logical_event_id FROM projections
                 WHERE kind = 'chat_message' AND ephemeral = 0
                   AND conversation_id = ?1
                   AND (
                     created_at > ?2
                     OR (created_at = ?2 AND logical_event_id < ?3)
                   )
                 ORDER BY created_at ASC, logical_event_id DESC LIMIT ?4",
            )
            .map_err(map_sqlite_error)?;
        let before_rows = before_statement
            .query_map(
                params![target.conversation_id, target.created_at, target.id, before],
                |row| row.get::<_, String>(0),
            )
            .map_err(map_sqlite_error)?;
        let mut ids = Vec::new();
        for row in before_rows {
            ids.push(row.map_err(map_sqlite_error)?);
        }
        ids.reverse();
        ids.push(target.id.clone());
        let mut after_statement = self
            .connection
            .prepare(
                "SELECT logical_event_id FROM projections
                 WHERE kind = 'chat_message' AND ephemeral = 0
                   AND conversation_id = ?1
                   AND (
                     created_at < ?2
                     OR (created_at = ?2 AND logical_event_id > ?3)
                   )
                 ORDER BY created_at DESC, logical_event_id ASC LIMIT ?4",
            )
            .map_err(map_sqlite_error)?;
        let after_rows = after_statement
            .query_map(
                params![target.conversation_id, target.created_at, target.id, after],
                |row| row.get::<_, String>(0),
            )
            .map_err(map_sqlite_error)?;
        for row in after_rows {
            ids.push(row.map_err(map_sqlite_error)?);
        }
        self.product_message_views(ProductMessageQuery::Ids(&ids))
    }

    pub(crate) fn search_product_messages(
        &self,
        conversation_id: Option<String>,
        query: &str,
        limit: u32,
    ) -> Result<Vec<AccountMessageView>, SoftchatError> {
        validate_limit(limit)?;
        if query.is_empty() || query.len() > MAX_PRODUCT_SEARCH_BYTES {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        if let Some(value) = &conversation_id {
            validate_conversation_id(value)?;
        }
        let fts_query = escaped_fts_query(query)?;
        let mut statement = self
            .connection
            .prepare(
                "SELECT f.message_id
                 FROM message_fts f
                 JOIN projections p ON p.logical_event_id = f.message_id
                 WHERE message_fts MATCH ?1
                   AND (?2 IS NULL OR f.conversation_id = ?2)
                   AND NOT EXISTS (
                     SELECT 1 FROM projection_targets dt
                     JOIN projections d ON d.logical_event_id = dt.logical_event_id
                     WHERE dt.target_event_id = p.logical_event_id
                       AND d.kind = 'deletion'
                       AND d.author_public_key = p.author_public_key
                   )
                 ORDER BY p.created_at DESC, p.logical_event_id ASC LIMIT ?3",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params![fts_query, conversation_id, limit], |row| {
                row.get::<_, String>(0)
            })
            .map_err(map_sqlite_error)?;
        let mut ids = Vec::new();
        for row in rows {
            ids.push(row.map_err(map_sqlite_error)?);
        }
        self.product_message_views(ProductMessageQuery::Ids(&ids))
    }

    pub(crate) fn set_product_conversation_flag(
        &mut self,
        conversation_id: &str,
        column: &str,
        value: bool,
        now: i64,
    ) -> Result<AccountMutationResult, SoftchatError> {
        validate_conversation_id(conversation_id)?;
        validate_timestamp(now)?;
        let sql = match column {
            "archived" => {
                "UPDATE conversation_state
                 SET archived = ?1, updated_at = MAX(updated_at, ?2)
                 WHERE conversation_id = ?3"
            }
            "pinned" => {
                "UPDATE conversation_state
                 SET pinned = ?1, updated_at = MAX(updated_at, ?2)
                 WHERE conversation_id = ?3"
            }
            _ => return Err(SoftchatError::InvalidAccountOperation),
        };
        let diagnostic_changed = self.local_state_differs(
            "SELECT CASE ?2 WHEN 'archived' THEN archived ELSE pinned END != ?3
             FROM conversation_state WHERE conversation_id = ?1",
            params![conversation_id, column, i64::from(value)],
        );
        let result = self
            .mutate_existing_conversation(sql, params![i64::from(value), now, conversation_id])?;
        self.record_local_change(diagnostic_changed, result.revision);
        Ok(result)
    }

    pub(crate) fn product_read_states(&self) -> Result<Vec<AccountReadState>, SoftchatError> {
        Ok(self
            .load_product_read_state_rows()?
            .into_iter()
            .map(|row| row.state)
            .collect())
    }

    pub(crate) fn preview_product_read_state_update(
        &self,
        conversation_id: &str,
        boundary: Option<MessageCursor>,
        forced_unread: bool,
        now: i64,
        update_id: &str,
    ) -> Result<ReadStateMutationPreview, SoftchatError> {
        validate_conversation_id(conversation_id)?;
        validate_timestamp(now)?;
        validate_command_id(update_id)?;
        if self
            .connection
            .query_row(
                "SELECT 1 FROM conversation_state WHERE conversation_id = ?1",
                [conversation_id],
                |row| row.get::<_, i64>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?
            .is_none()
        {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        if let Some(value) = &boundary {
            validate_timestamp(value.created_at)?;
            NostrEventId::from_hex(&value.message_id)?;
            let belongs = self
                .connection
                .query_row(
                    "SELECT 1 FROM projections
                     WHERE logical_event_id = ?1 AND conversation_id = ?2
                       AND kind = 'chat_message' AND created_at = ?3",
                    params![value.message_id, conversation_id, value.created_at],
                    |row| row.get::<_, i64>(0),
                )
                .optional()
                .map_err(map_sqlite_error)?
                .is_some();
            if !belongs {
                return Err(SoftchatError::InvalidAccountOperation);
            }
        }

        let mut rows = self.load_product_read_state_rows()?;
        let existing = rows
            .iter()
            .find(|row| row.state.conversation_id == conversation_id);
        let updated_at = existing
            .map(|row| row.state.updated_at)
            .filter(|updated_at| *updated_at >= now)
            .map_or(Ok(now), |updated_at| {
                updated_at
                    .checked_add(1)
                    .ok_or(SoftchatError::InvalidAccountOperation)
            })?;
        let boundary = if forced_unread {
            existing.and_then(|row| row.state.boundary.clone())
        } else {
            Some(advance_read_boundary(
                existing.and_then(|row| row.state.boundary.clone()),
                boundary.ok_or(SoftchatError::InvalidAccountOperation)?,
            ))
        };
        let state = AccountReadState {
            conversation_id: conversation_id.to_owned(),
            boundary,
            forced_unread,
            updated_at,
            update_id: update_id.to_owned(),
        };
        if let Some(row) = rows
            .iter_mut()
            .find(|row| row.state.conversation_id == conversation_id)
        {
            row.state = state.clone();
        } else {
            rows.push(StoredReadStateRow {
                state: state.clone(),
                unknown_json: "{}".to_owned(),
            });
        }
        rows.sort_by(|left, right| left.state.conversation_id.cmp(&right.state.conversation_id));
        let canonical_json = self.product_read_state_wire_json(&rows)?;
        Ok(ReadStateMutationPreview {
            state,
            canonical_json,
        })
    }

    pub(crate) fn save_product_draft(
        &mut self,
        conversation_id: &str,
        draft: Option<AccountDraft>,
        now: i64,
    ) -> Result<AccountMutationResult, SoftchatError> {
        validate_conversation_id(conversation_id)?;
        validate_timestamp(now)?;
        let (draft_json, reply_to, updated_at) = if let Some(value) = draft {
            validate_product_draft(&value)?;
            if !value.reply_to_message_id.is_empty() {
                NostrEventId::from_hex(&value.reply_to_message_id)?;
            }
            for operation_id in &value.attachment_operation_ids {
                let belongs = self
                    .connection
                    .query_row(
                        "SELECT 1 FROM media_operations
                         WHERE operation_id = ?1 AND conversation_id = ?2",
                        params![operation_id, conversation_id],
                        |row| row.get::<_, i64>(0),
                    )
                    .optional()
                    .map_err(map_sqlite_error)?
                    .is_some();
                if !belongs {
                    return Err(SoftchatError::InvalidAccountOperation);
                }
            }
            let json = serde_json::to_string(&StoredDraft {
                text: value.text,
                emoji_tags: value.emoji_tags,
                spans: value.spans,
                attachment_operation_ids: value.attachment_operation_ids,
            })
            .map_err(|_| SoftchatError::InternalFailure)?;
            (Some(json), Some(value.reply_to_message_id), Some(now))
        } else {
            (None, None, None)
        };
        let diagnostic_changed = self.local_state_differs(
            "SELECT draft_json IS NOT ?2 OR draft_reply_to IS NOT ?3
             FROM conversation_state WHERE conversation_id = ?1",
            params![conversation_id, draft_json, reply_to],
        );
        let result = self.mutate_existing_conversation(
            "UPDATE conversation_state
             SET draft_json = ?1, draft_reply_to = ?2, draft_updated_at = ?3,
                 updated_at = MAX(updated_at, ?4)
             WHERE conversation_id = ?5",
            params![draft_json, reply_to, updated_at, now, conversation_id],
        )?;
        self.record_local_change(diagnostic_changed, result.revision);
        Ok(result)
    }

    pub(crate) fn product_message_local_extras(
        &self,
        message_id: &str,
    ) -> Result<Option<MessageLocalExtras>, SoftchatError> {
        NostrEventId::from_hex(message_id)?;
        self.connection
            .query_row(
                "SELECT transcript, waveform, updated_at
                 FROM message_local_extras WHERE message_id = ?1",
                [message_id],
                |row| {
                    Ok(MessageLocalExtras {
                        message_id: message_id.to_owned(),
                        transcript: row.get(0)?,
                        waveform: row.get::<_, Option<Vec<u8>>>(1)?.unwrap_or_default(),
                        updated_at: row.get(2)?,
                    })
                },
            )
            .optional()
            .map_err(map_sqlite_error)
    }

    pub(crate) fn set_product_message_local_extras(
        &mut self,
        message_id: &str,
        transcript: Option<String>,
        waveform: Vec<u8>,
        now: i64,
    ) -> Result<AccountMutationResult, SoftchatError> {
        NostrEventId::from_hex(message_id)?;
        validate_timestamp(now)?;
        if transcript
            .as_ref()
            .is_some_and(|value| value.len() > MAX_PRODUCT_TRANSCRIPT_BYTES)
            || waveform.len() > MAX_PRODUCT_WAVEFORM_SAMPLES
        {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let exists = transaction
            .query_row(
                "SELECT 1 FROM projections
                 WHERE logical_event_id = ?1 AND kind = 'chat_message'",
                [message_id],
                |row| row.get::<_, i64>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?
            .is_some();
        if !exists {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        transaction
            .execute(
                "INSERT INTO message_local_extras
                 (message_id, transcript, waveform, updated_at)
                 VALUES (?1, ?2, ?3, ?4)
                 ON CONFLICT(message_id) DO UPDATE SET
                   transcript = excluded.transcript,
                   waveform = excluded.waveform,
                   updated_at = excluded.updated_at",
                params![message_id, transcript, waveform, now],
            )
            .map_err(map_sqlite_error)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        Ok(AccountMutationResult { revision })
    }

    fn load_product_read_state_rows(&self) -> Result<Vec<StoredReadStateRow>, SoftchatError> {
        let mut statement = self
            .connection
            .prepare(
                "SELECT conversation_id, read_at, read_message_id, forced_unread,
                        updated_at, update_id, unknown_json
                 FROM conversation_read_state
                 ORDER BY conversation_id ASC",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, Option<i64>>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, i64>(3)? != 0,
                    row.get::<_, i64>(4)?,
                    row.get::<_, String>(5)?,
                    row.get::<_, String>(6)?,
                ))
            })
            .map_err(map_sqlite_error)?;
        let mut states = Vec::new();
        for row in rows {
            let (
                conversation_id,
                read_at,
                read_message_id,
                forced_unread,
                updated_at,
                update_id,
                unknown_json,
            ) = row.map_err(map_sqlite_error)?;
            let boundary = read_at.map(|created_at| MessageCursor {
                created_at,
                message_id: if read_message_id.is_empty() {
                    minimum_event_id()
                } else {
                    read_message_id
                },
            });
            states.push(StoredReadStateRow {
                state: AccountReadState {
                    conversation_id,
                    boundary,
                    forced_unread,
                    updated_at,
                    update_id,
                },
                unknown_json,
            });
        }
        Ok(states)
    }

    fn product_read_state_wire_json(
        &self,
        rows: &[StoredReadStateRow],
    ) -> Result<String, SoftchatError> {
        if rows.len() > MAX_ACCOUNT_READ_STATE_ENTRIES {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        let mut root = self
            .connection
            .query_row(
                "SELECT canonical_json FROM app_data_effective
                 WHERE context = 'read-state'",
                [],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?
            .and_then(|value| serde_json::from_str::<Map<String, Value>>(&value).ok())
            .unwrap_or_default();
        root.remove("schemaVersion");
        root.remove("chats");
        let mut chats = Vec::with_capacity(rows.len());
        for row in rows {
            let mut entry: Map<String, Value> = serde_json::from_str(&row.unknown_json)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            let chat_key = self
                .product_conversation_members(&row.state.conversation_id)
                .ok()
                .filter(|members| !members.is_empty())
                .map(|members| members.join(","))
                .unwrap_or_else(|| row.state.conversation_id.clone());
            entry.insert("chatKey".to_owned(), json!(chat_key));
            entry.insert(
                "conversationId".to_owned(),
                json!(row.state.conversation_id),
            );
            entry.insert(
                "readAt".to_owned(),
                row.state.boundary.as_ref().map_or(Value::Null, |boundary| {
                    json!(boundary.created_at as f64 - SWIFT_REFERENCE_DATE_UNIX_SECONDS as f64)
                }),
            );
            entry.insert(
                "readAtUnix".to_owned(),
                row.state
                    .boundary
                    .as_ref()
                    .map_or(Value::Null, |boundary| json!(boundary.created_at)),
            );
            entry.insert(
                "readMessageId".to_owned(),
                json!(
                    row.state
                        .boundary
                        .as_ref()
                        .map_or("", |boundary| boundary.message_id.as_str())
                ),
            );
            entry.insert("forcedUnread".to_owned(), json!(row.state.forced_unread));
            entry.insert("updatedAt".to_owned(), json!(row.state.updated_at));
            entry.insert("updateId".to_owned(), json!(row.state.update_id));
            chats.push(Value::Object(entry));
        }
        root.insert(
            "schemaVersion".to_owned(),
            json!(ACCOUNT_READ_STATE_SCHEMA_VERSION),
        );
        root.insert("chats".to_owned(), Value::Array(chats));
        let json =
            serde_json::to_string(&root).map_err(|_| SoftchatError::InvalidAccountOperation)?;
        if json.len() > MAX_ACCOUNT_READ_STATE_BYTES {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        Ok(json)
    }

    pub(crate) fn product_settings(&self) -> Result<AccountSettings, SoftchatError> {
        let json = self.product_settings_canonical_json()?;
        settings_from_json(&json)
    }

    pub(crate) fn product_settings_canonical_json(&self) -> Result<String, SoftchatError> {
        self.connection
            .query_row(
                "SELECT canonical_json FROM account_settings WHERE singleton = 1",
                [],
                |row| row.get::<_, String>(0),
            )
            .map_err(map_sqlite_error)
    }

    #[cfg(test)]
    pub(crate) fn apply_product_settings_patch(
        &mut self,
        patch_json: &str,
        now: i64,
    ) -> Result<AccountSettingsMutation, SoftchatError> {
        validate_timestamp(now)?;
        let preview = self.preview_product_settings_patch(patch_json)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "UPDATE account_settings
                 SET schema_version = ?1, canonical_json = ?2, updated_at = ?3
                 WHERE singleton = 1",
                params![ACCOUNT_SETTINGS_SCHEMA_VERSION, preview.canonical_json, now],
            )
            .map_err(map_sqlite_error)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        Ok(AccountSettingsMutation {
            revision,
            ..preview
        })
    }

    pub(crate) fn preview_product_settings_patch(
        &self,
        patch_json: &str,
    ) -> Result<AccountSettingsMutation, SoftchatError> {
        if patch_json.len() > MAX_ACCOUNT_SETTINGS_BYTES {
            return Err(SoftchatError::InvalidAccountSettings);
        }
        let patch: Map<String, Value> =
            serde_json::from_str(patch_json).map_err(|_| SoftchatError::InvalidAccountSettings)?;
        let current = self.product_settings()?;
        let mut value = settings_to_object(&current)?;
        for (key, patch_value) in patch {
            if key == "schemaVersion" {
                return Err(SoftchatError::InvalidAccountSettings);
            }
            if patch_value.is_null() {
                value.remove(&key);
            } else {
                value.insert(key, patch_value);
            }
        }
        let canonical =
            serde_json::to_string(&value).map_err(|_| SoftchatError::InvalidAccountSettings)?;
        let settings = settings_from_json(&canonical)?;
        let canonical = serde_json::to_string(&settings_to_object(&settings)?)
            .map_err(|_| SoftchatError::InvalidAccountSettings)?;
        Ok(AccountSettingsMutation {
            settings,
            canonical_json: canonical,
            revision: self.revision()?,
        })
    }

    pub(crate) fn product_profile(
        &self,
        public_key: &str,
    ) -> Result<AccountProfile, SoftchatError> {
        let public_key = NostrPublicKey::from_hex(public_key)?.to_hex();
        let mut profile = self.authenticated_product_profile(&public_key)?;
        let local = self
            .connection
            .query_row(
                "SELECT name, updated_at_millis FROM local_contacts WHERE public_key = ?1",
                [&public_key],
                |row| Ok((row.get::<_, Option<String>>(0)?, row.get::<_, i64>(1)?)),
            )
            .optional()
            .map_err(map_sqlite_error)?;
        let use_local = local.as_ref().is_some_and(|(local_name, local_time)| {
            local_name.is_some() && *local_time / 1_000 >= profile.updated_at
        });
        if use_local {
            profile.name = local.as_ref().and_then(|value| value.0.clone());
            profile.local_override = true;
            profile.updated_at = local.map_or(profile.updated_at, |value| value.1 / 1_000);
        }
        Ok(profile)
    }

    pub(crate) fn authenticated_product_profile(
        &self,
        public_key: &str,
    ) -> Result<AccountProfile, SoftchatError> {
        let public_key = NostrPublicKey::from_hex(public_key)?.to_hex();
        let authenticated = self
            .list_event_nodes(
                Some(ProjectionKind::UserMetadata),
                String::new(),
                String::new(),
                String::new(),
                public_key.clone(),
                None,
                1,
                0,
            )?
            .into_iter()
            .next()
            .map(|node| {
                let rumor = node
                    .projection
                    .rumor
                    .ok_or(SoftchatError::InvalidPersistenceResult)?;
                let metadata = classify_user_metadata(rumor)?;
                Ok::<_, SoftchatError>((
                    metadata.name,
                    metadata.display_name,
                    metadata.about,
                    metadata.picture,
                    metadata.website,
                    metadata.banner,
                    metadata.bot,
                    metadata.unknown_json,
                    node.projection.created_at,
                ))
            })
            .transpose()?;
        let (name, display_name, about, picture, website, banner, bot, unknown_json, updated_at) =
            authenticated.unwrap_or((None, None, None, None, None, None, None, "{}".to_owned(), 0));
        let first_seen_at = self.connection.query_row(
            "SELECT MIN(created_at) FROM projections WHERE author_public_key = ?1 AND ephemeral = 0",
            [&public_key], |row| row.get::<_, Option<i64>>(0),
        ).map_err(map_sqlite_error)?;
        Ok(AccountProfile {
            public_key,
            name,
            display_name,
            about,
            picture,
            website,
            banner,
            bot,
            unknown_json,
            local_override: false,
            updated_at,
            first_seen_at,
        })
    }

    pub(crate) fn authenticated_product_follow_updated_at(
        &self,
        own_public_key: &str,
    ) -> Result<i64, SoftchatError> {
        let own_public_key = NostrPublicKey::from_hex(own_public_key)?.to_hex();
        Ok(self
            .list_event_nodes(
                Some(ProjectionKind::FollowList),
                String::new(),
                String::new(),
                String::new(),
                own_public_key,
                None,
                1,
                0,
            )?
            .first()
            .map_or(0, |node| node.projection.created_at))
    }

    pub(crate) fn product_profiles(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<AccountProfile>, SoftchatError> {
        validate_limit(limit)?;
        let keys = self.list_known_public_keys(limit, offset)?;
        self.product_profiles_for_keys(&keys)
    }

    pub(crate) fn product_profiles_for_keys(
        &self,
        public_keys: &[String],
    ) -> Result<Vec<AccountProfile>, SoftchatError> {
        if public_keys.len() > MAX_PRODUCT_PROFILE_LOOKUPS {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        if public_keys.is_empty() {
            return Ok(Vec::new());
        }
        for public_key in public_keys {
            NostrPublicKey::from_hex(public_key)?;
        }
        let values = public_keys
            .iter()
            .enumerate()
            .map(|(position, _)| format!("(?, {position})"))
            .collect::<Vec<_>>()
            .join(",");
        let sql = format!(
            "WITH requested(public_key, position) AS (
               VALUES {values}
             )
             , ranked_metadata AS (
               SELECT
                 profile.author_public_key,
                 profile.logical_event_id,
                 rumor.canonical_json,
                 profile.created_at,
                 ROW_NUMBER() OVER (
                   PARTITION BY profile.author_public_key
                   ORDER BY profile.created_at DESC,
                            profile.logical_event_id ASC
                 ) AS row_number
               FROM projections profile
               JOIN requested
                 ON requested.public_key = profile.author_public_key
               LEFT JOIN authenticated_rumors rumor
                 ON rumor.rumor_id = profile.logical_event_id
               WHERE profile.kind = 'user_metadata'
             )
             SELECT
               requested.public_key,
               metadata.logical_event_id,
               metadata.canonical_json,
               metadata.created_at,
               local.name,
               local.updated_at_millis,
               (SELECT MIN(first.created_at) FROM projections first
                WHERE first.author_public_key = requested.public_key AND first.ephemeral = 0)
             FROM requested
             LEFT JOIN ranked_metadata metadata
               ON metadata.author_public_key = requested.public_key
              AND metadata.row_number = 1
             LEFT JOIN local_contacts local
               ON local.public_key = requested.public_key
             ORDER BY requested.position ASC"
        );
        let parameters = public_keys
            .iter()
            .cloned()
            .map(SqlValue::Text)
            .collect::<Vec<_>>();
        let mut statement = self.connection.prepare(&sql).map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params_from_iter(parameters), |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, Option<String>>(1)?,
                    row.get::<_, Option<String>>(2)?,
                    row.get::<_, Option<i64>>(3)?,
                    row.get::<_, Option<String>>(4)?,
                    row.get::<_, Option<i64>>(5)?,
                    row.get::<_, Option<i64>>(6)?,
                ))
            })
            .map_err(map_sqlite_error)?;
        let mut profiles = Vec::with_capacity(public_keys.len());
        for row in rows {
            let (
                public_key,
                metadata_id,
                metadata_json,
                metadata_updated_at,
                local_name,
                local_updated_at_millis,
                first_seen_at,
            ) = row.map_err(map_sqlite_error)?;
            let metadata = match (metadata_id, metadata_json, metadata_updated_at) {
                (Some(_), Some(json), Some(updated_at)) => {
                    let rumor = serde_json::from_str::<RumorEvent>(&json)
                        .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
                    Some((classify_user_metadata(rumor)?, updated_at))
                }
                (None, None, None) => None,
                _ => return Err(SoftchatError::InvalidPersistenceResult),
            };
            let (mut profile, authenticated_updated_at) = metadata.map_or_else(
                || {
                    (
                        AccountProfile {
                            public_key: public_key.clone(),
                            name: None,
                            display_name: None,
                            about: None,
                            picture: None,
                            website: None,
                            banner: None,
                            bot: None,
                            unknown_json: "{}".to_owned(),
                            local_override: false,
                            updated_at: 0,
                            first_seen_at,
                        },
                        0,
                    )
                },
                |(value, updated_at)| {
                    (
                        AccountProfile {
                            public_key: public_key.clone(),
                            name: value.name,
                            display_name: value.display_name,
                            about: value.about,
                            picture: value.picture,
                            website: value.website,
                            banner: value.banner,
                            bot: value.bot,
                            unknown_json: value.unknown_json,
                            local_override: false,
                            updated_at,
                            first_seen_at,
                        },
                        updated_at,
                    )
                },
            );
            if let (Some(name), Some(updated_at_millis)) = (local_name, local_updated_at_millis)
                && updated_at_millis / 1_000 >= authenticated_updated_at
            {
                profile.name = Some(name);
                profile.local_override = true;
                profile.updated_at = updated_at_millis / 1_000;
            }
            profiles.push(profile);
        }
        Ok(profiles)
    }

    pub(crate) fn product_follows(
        &self,
        own_public_key: &str,
    ) -> Result<Vec<crate::Contact>, SoftchatError> {
        let own_public_key = NostrPublicKey::from_hex(own_public_key)?.to_hex();
        let node = self
            .list_event_nodes(
                Some(ProjectionKind::FollowList),
                String::new(),
                String::new(),
                String::new(),
                own_public_key,
                None,
                1,
                0,
            )?
            .into_iter()
            .next();
        let Some(node) = node else {
            return Ok(Vec::new());
        };
        let rumor = node
            .projection
            .rumor
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        Ok(classify_follow_list(rumor)?.contacts)
    }

    pub(crate) fn search_product_profiles(
        &self,
        query: &str,
        limit: u32,
    ) -> Result<Vec<AccountProfile>, SoftchatError> {
        validate_limit(limit)?;
        let pattern = escaped_like_pattern(query)?;
        let mut statement = self
            .connection
            .prepare(
                r"WITH known(public_key) AS (
                   SELECT author_public_key FROM projections
                   UNION
                   SELECT public_key FROM projection_participants
                   UNION
                   SELECT public_key FROM local_contacts
                 )
                 SELECT known.public_key
                 FROM known
                 LEFT JOIN local_contacts local
                   ON local.public_key = known.public_key
                 WHERE lower(known.public_key) LIKE ?1 ESCAPE '\'
                    OR lower(COALESCE(local.name, '')) LIKE ?1 ESCAPE '\'
                    OR EXISTS (
                      SELECT 1 FROM json_tree((
                        SELECT profile.content
                        FROM projections profile
                        WHERE profile.kind = 'user_metadata'
                          AND profile.author_public_key = known.public_key
                        ORDER BY profile.created_at DESC, profile.logical_event_id ASC
                        LIMIT 1
                      )) metadata
                      WHERE metadata.type = 'text'
                        AND lower(metadata.atom) LIKE ?1 ESCAPE '\'
                    )
                 ORDER BY
                   CASE WHEN local.name IS NULL THEN 1 ELSE 0 END,
                   lower(COALESCE(local.name, '')),
                   known.public_key
                 LIMIT ?2",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params![pattern, limit], |row| row.get::<_, String>(0))
            .map_err(map_sqlite_error)?;
        let mut keys = Vec::new();
        for row in rows {
            keys.push(row.map_err(map_sqlite_error)?);
        }
        self.product_profiles_for_keys(&keys)
    }

    pub(crate) fn product_media_gallery(
        &self,
        conversation_id: Option<String>,
        before: Option<MessageCursor>,
        limit: u32,
    ) -> Result<Vec<AccountMediaItem>, SoftchatError> {
        validate_limit(limit)?;
        if let Some(value) = &conversation_id {
            validate_conversation_id(value)?;
        }
        if let Some(value) = &before {
            validate_timestamp(value.created_at)?;
            NostrEventId::from_hex(&value.message_id)?;
        }
        let (before_at, before_id, has_cursor) = before
            .map(|value| (value.created_at, value.message_id, 1_i64))
            .unwrap_or((0, String::new(), 0));
        // Filter before applying the bound: arbitrarily many intervening text
        // messages must not make an older attachment unreachable. Authenticated
        // chat parsing guarantees each selected imeta tag is an attachment.
        let mut statement = self
            .connection
            .prepare(
                "SELECT p.logical_event_id FROM projections p
                 JOIN authenticated_rumors rumor
                   ON rumor.rumor_id = p.logical_event_id
                 WHERE p.kind = 'chat_message' AND p.ephemeral = 0
                   AND (?1 IS NULL OR p.conversation_id = ?1)
                   AND (
                     ?2 = 0 OR p.created_at < ?3
                     OR (p.created_at = ?3 AND p.logical_event_id > ?4)
                   )
                   AND NOT EXISTS (
                     SELECT 1 FROM projection_targets target
                     JOIN projections deletion
                       ON deletion.logical_event_id = target.logical_event_id
                     WHERE target.target_event_id = p.logical_event_id
                       AND deletion.kind = 'deletion'
                       AND deletion.author_public_key = p.author_public_key
                   )
                   AND EXISTS (
                     SELECT 1 FROM json_each(rumor.canonical_json, '$.tags') tag
                     WHERE json_extract(tag.value, '$[0]') = 'imeta'
                   )
                 ORDER BY p.created_at DESC, p.logical_event_id ASC LIMIT ?5",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(
                params![conversation_id, has_cursor, before_at, before_id, limit],
                |row| row.get::<_, String>(0),
            )
            .map_err(map_sqlite_error)?;
        let mut ids = Vec::new();
        for row in rows {
            ids.push(row.map_err(map_sqlite_error)?);
        }
        let mut result = Vec::new();
        let requested =
            usize::try_from(limit).map_err(|_| SoftchatError::InvalidAccountOperation)?;
        for message in self.product_message_views(ProductMessageQuery::Ids(&ids))? {
            for attachment in message.attachments {
                result.push(AccountMediaItem {
                    message_id: message.id.clone(),
                    conversation_id: message.conversation_id.clone(),
                    author_public_key: message.author_public_key.clone(),
                    created_at: message.created_at,
                    attachment,
                });
            }
            // The public cursor addresses messages, so splitting this group
            // would permanently skip its remaining attachments on the next page.
            if result.len() >= requested {
                return Ok(result);
            }
        }
        Ok(result)
    }

    pub(crate) fn prepare_product_media_message(
        &mut self,
        preparation: ProductMediaMessagePreparation<'_>,
    ) -> Result<PendingMediaMessage, SoftchatError> {
        let ProductMediaMessagePreparation {
            command_id,
            conversation_id,
            content,
            reply_to_message_id,
            sources,
            command_hash,
            draft_match,
            now,
        } = preparation;
        validate_command_id(command_id)?;
        validate_conversation_id(conversation_id)?;
        validate_timestamp(now)?;
        if sources.is_empty()
            || sources.len() > MAX_PRODUCT_DRAFT_ATTACHMENTS
            || content.text.len() > MAX_PRODUCT_MESSAGE_BYTES
            || !content.attachments.is_empty()
            || content.emoji_tags.len() > crate::MAX_CHAT_EMOJI_TAGS
            || command_hash.len() != 64
            || !command_hash.bytes().all(|byte| byte.is_ascii_hexdigit())
        {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        for tag in &content.emoji_tags {
            crate::NostrTag::new(tag.clone())?;
        }
        if !reply_to_message_id.is_empty() {
            NostrEventId::from_hex(reply_to_message_id)?;
        }
        let existing_hash = self
            .connection
            .query_row(
                "SELECT command_hash FROM pending_media_messages WHERE command_id = ?1",
                [command_id],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?;
        if let Some(existing_hash) = existing_hash {
            if existing_hash != command_hash {
                return Err(SoftchatError::CommandConflict);
            }
            return self
                .product_pending_media_message(command_id)?
                .ok_or(SoftchatError::InvalidPersistenceResult);
        }
        let conversation_exists = self
            .connection
            .query_row(
                "SELECT 1 FROM conversation_state WHERE conversation_id = ?1",
                [conversation_id],
                |row| row.get::<_, i64>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?
            .is_some();
        if !conversation_exists {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        if !reply_to_message_id.is_empty() {
            let reply_exists = self
                .connection
                .query_row(
                    "SELECT 1 FROM projections
                     WHERE logical_event_id = ?1 AND conversation_id = ?2
                       AND kind = 'chat_message'",
                    params![reply_to_message_id, conversation_id],
                    |row| row.get::<_, i64>(0),
                )
                .optional()
                .map_err(map_sqlite_error)?
                .is_some();
            if !reply_exists {
                return Err(SoftchatError::InvalidMediaOperation);
            }
        }
        for source in &sources {
            if source.source_fingerprint.is_empty() || source.source_fingerprint.len() > 512 {
                return Err(SoftchatError::InvalidMediaOperation);
            }
        }
        let emoji_tags_json = serde_json::to_string(&content.emoji_tags)
            .map_err(|_| SoftchatError::InternalFailure)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO pending_media_messages
                 (command_id, conversation_id, text_content, emoji_tags_json,
                  reply_to_message_id, command_hash, state, created_at, updated_at)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, 'waiting_for_uploads', ?7, ?7)",
                params![
                    command_id,
                    conversation_id,
                    content.text,
                    emoji_tags_json,
                    reply_to_message_id,
                    command_hash,
                    now,
                ],
            )
            .map_err(map_sqlite_error)?;
        for (index, source) in sources.into_iter().enumerate() {
            let attachment_command_id =
                stable_pending_attachment_command_id(command_id, index, &source.source_fingerprint);
            let operation_id = stable_media_operation_id(
                &self.account_id,
                &attachment_command_id,
                "upload",
                &source.source_fingerprint,
            );
            transaction
                .execute(
                    "INSERT INTO media_operations
                     (operation_id, command_id, conversation_id, direction, state,
                      source_fingerprint, created_at, updated_at)
                     VALUES (?1, ?2, ?3, 'upload', 'prepared', ?4, ?5, ?5)",
                    params![
                        operation_id,
                        attachment_command_id,
                        conversation_id,
                        source.source_fingerprint,
                        now,
                    ],
                )
                .map_err(map_sqlite_error)?;
            transaction
                .execute(
                    "INSERT INTO pending_message_attachments
                     (command_id, attachment_index, operation_id)
                     VALUES (?1, ?2, ?3)",
                    params![
                        command_id,
                        i64::try_from(index).map_err(|_| SoftchatError::InvalidMediaOperation)?,
                        operation_id,
                    ],
                )
                .map_err(map_sqlite_error)?;
        }
        if let Some(draft_match) = draft_match {
            transaction
                .execute(
                    "UPDATE conversation_state
                     SET draft_json = NULL, draft_reply_to = '', draft_updated_at = NULL
                     WHERE conversation_id = ?1 AND draft_json = ?2
                       AND COALESCE(draft_reply_to, '') = ?3",
                    params![
                        conversation_id,
                        draft_match.draft_json,
                        draft_match.reply_to_message_id,
                    ],
                )
                .map_err(map_sqlite_error)?;
        }
        bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        self.record_diagnostic(Diagnostic::new(
            AccountLogCategory::Media,
            "media.dependency.waiting",
            "Publication waiting for uploads",
        ));
        self.product_pending_media_message(command_id)?
            .ok_or(SoftchatError::InvalidPersistenceResult)
    }

    pub(crate) fn product_pending_media_message(
        &self,
        command_id: &str,
    ) -> Result<Option<PendingMediaMessage>, SoftchatError> {
        validate_command_id(command_id)?;
        let row = self
            .connection
            .query_row(
                "SELECT conversation_id, text_content, emoji_tags_json,
                        reply_to_message_id, state, result_message_id, updated_at
                 FROM pending_media_messages WHERE command_id = ?1",
                [command_id],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, String>(4)?,
                        row.get::<_, String>(5)?,
                        row.get::<_, i64>(6)?,
                    ))
                },
            )
            .optional()
            .map_err(map_sqlite_error)?;
        let Some((
            conversation_id,
            text,
            emoji_tags_json,
            reply_to_message_id,
            state,
            result_message_id,
            updated_at,
        )) = row
        else {
            return Ok(None);
        };
        let emoji_tags = serde_json::from_str(&emoji_tags_json)
            .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
        let mut statement = self
            .connection
            .prepare(
                "SELECT operation_id FROM pending_message_attachments
                 WHERE command_id = ?1
                 ORDER BY attachment_index ASC",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([command_id], |row| row.get::<_, String>(0))
            .map_err(map_sqlite_error)?;
        let mut attachments = Vec::new();
        for row in rows {
            attachments.push(
                self.product_media_operation(&row.map_err(map_sqlite_error)?)?
                    .ok_or(SoftchatError::InvalidPersistenceResult)?,
            );
        }
        Ok(Some(PendingMediaMessage {
            command_id: command_id.to_owned(),
            conversation_id,
            text,
            emoji_tags,
            reply_to_message_id,
            state: pending_media_message_state_from_name(&state)?,
            attachments,
            result_message_id,
            updated_at,
        }))
    }

    pub(crate) fn cancel_product_pending_media_message(
        &mut self,
        command_id: &str,
        now: i64,
    ) -> Result<PendingMediaMessage, SoftchatError> {
        validate_command_id(command_id)?;
        validate_timestamp(now)?;
        let current = self
            .product_pending_media_message(command_id)?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        match current.state {
            PendingMediaMessageState::Cancelled => {}
            PendingMediaMessageState::Published => {
                return Err(SoftchatError::InvalidMediaOperation);
            }
            PendingMediaMessageState::WaitingForUploads
            | PendingMediaMessageState::ReadyToPublish
            | PendingMediaMessageState::Failed => {}
        }

        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let cancelled_attachments = transaction
            .execute(
                "UPDATE media_operations
                 SET state = 'cancelled', last_error_category = '',
                     lease_id = '', lease_expires_at = 0, updated_at = ?1
                 WHERE operation_id IN (
                   SELECT operation_id FROM pending_message_attachments
                   WHERE command_id = ?2
                 ) AND state IN ('prepared', 'retryable', 'running', 'failed')",
                params![now, command_id],
            )
            .map_err(map_sqlite_error)?;
        let cancelled_message = if current.state == PendingMediaMessageState::Cancelled {
            0
        } else {
            transaction
                .execute(
                    "UPDATE pending_media_messages
                     SET state = 'cancelled', updated_at = ?1
                     WHERE command_id = ?2
                       AND publication_claimed = 0
                       AND state IN ('waiting_for_uploads', 'ready_to_publish', 'failed')",
                    params![now, command_id],
                )
                .map_err(map_sqlite_error)?
        };
        if current.state != PendingMediaMessageState::Cancelled && cancelled_message != 1 {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        if cancelled_message == 0 && cancelled_attachments == 0 {
            transaction.commit().map_err(map_sqlite_error)?;
            return Ok(current);
        }
        bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        let mut event = Diagnostic::new(
            AccountLogCategory::Media,
            "media.dependency.cancelled",
            "Pending message uploads cancelled",
        )
        .count(
            crate::account_diagnostics::AccountLogCounterKind::Attachments,
            cancelled_attachments as u64,
        );
        event.direction = Some(crate::account_diagnostics::AccountLogDirection::Tx);
        self.record_diagnostic(event);
        self.product_pending_media_message(command_id)?
            .ok_or(SoftchatError::InvalidPersistenceResult)
    }

    /// Durably choose publication over cancellation before authoring the message.
    ///
    /// The claim is intentionally not exposed as another product state. A
    /// process that stops after this commit leaves the message ready for the
    /// normal recovery path, while cancellation can no longer report success
    /// for an operation that may already have been enqueued.
    pub(crate) fn claim_pending_media_publication(
        &mut self,
        command_id: &str,
    ) -> Result<bool, SoftchatError> {
        validate_command_id(command_id)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let current = transaction
            .query_row(
                "SELECT state, publication_claimed
                 FROM pending_media_messages WHERE command_id = ?1",
                [command_id],
                |row| Ok((row.get::<_, String>(0)?, row.get::<_, bool>(1)?)),
            )
            .optional()
            .map_err(map_sqlite_error)?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        if current.0 != "ready_to_publish" {
            transaction.commit().map_err(map_sqlite_error)?;
            return Ok(false);
        }
        if !current.1 {
            let changed = transaction
                .execute(
                    "UPDATE pending_media_messages
                     SET publication_claimed = 1
                     WHERE command_id = ?1 AND state = 'ready_to_publish'
                       AND publication_claimed = 0",
                    [command_id],
                )
                .map_err(map_sqlite_error)?;
            if changed != 1 {
                return Err(SoftchatError::InvalidMediaOperation);
            }
        }
        transaction.commit().map_err(map_sqlite_error)?;
        Ok(true)
    }

    pub(crate) fn product_pending_media_messages(
        &self,
        limit: u32,
    ) -> Result<Vec<PendingMediaMessage>, SoftchatError> {
        validate_limit(limit)?;
        let mut statement = self
            .connection
            .prepare(
                "SELECT command_id FROM pending_media_messages
                 ORDER BY updated_at DESC, command_id ASC LIMIT ?1",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([limit], |row| row.get::<_, String>(0))
            .map_err(map_sqlite_error)?;
        let mut messages = Vec::new();
        for row in rows {
            messages.push(
                self.product_pending_media_message(&row.map_err(map_sqlite_error)?)?
                    .ok_or(SoftchatError::InvalidPersistenceResult)?,
            );
        }
        Ok(messages)
    }

    pub(crate) fn ready_pending_media_commands(
        &self,
        limit: u32,
    ) -> Result<Vec<String>, SoftchatError> {
        validate_limit(limit)?;
        let mut statement = self
            .connection
            .prepare(
                "SELECT command_id FROM pending_media_messages
                 WHERE state = 'ready_to_publish'
                 ORDER BY updated_at ASC, command_id ASC LIMIT ?1",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([limit], |row| row.get::<_, String>(0))
            .map_err(map_sqlite_error)?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(map_sqlite_error)
    }

    pub(crate) fn pending_media_command_for_operation(
        &self,
        operation_id: &str,
    ) -> Result<Option<String>, SoftchatError> {
        self.connection
            .query_row(
                "SELECT command_id FROM pending_message_attachments
                 WHERE operation_id = ?1",
                [operation_id],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(map_sqlite_error)
    }

    pub(crate) fn mark_pending_media_published(
        &mut self,
        command_id: &str,
        result_message_id: &str,
        now: i64,
    ) -> Result<AccountMutationResult, SoftchatError> {
        validate_command_id(command_id)?;
        NostrEventId::from_hex(result_message_id)?;
        validate_timestamp(now)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let changed = transaction
            .execute(
                "UPDATE pending_media_messages
                 SET state = 'published', result_message_id = ?1, updated_at = ?2
                 WHERE command_id = ?3
                   AND (
                     (state = 'ready_to_publish' AND publication_claimed = 1)
                     OR state = 'published'
                   )
                   AND (result_message_id = '' OR result_message_id = ?1)",
                params![result_message_id, now, command_id],
            )
            .map_err(map_sqlite_error)?;
        if changed != 1 {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        Ok(AccountMutationResult { revision })
    }

    pub(crate) fn prepare_product_asset_replacement(
        &mut self,
        preparation: ProductAssetReplacementPreparation<'_>,
    ) -> Result<PendingAssetReplacement, SoftchatError> {
        let ProductAssetReplacementPreparation {
            command_id,
            target,
            conversation_id,
            source_fingerprint,
            command_hash,
            now,
        } = preparation;
        validate_command_id(command_id)?;
        validate_timestamp(now)?;
        if source_fingerprint.is_empty()
            || source_fingerprint.len() > 512
            || command_hash.len() != 64
            || !command_hash
                .bytes()
                .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
        {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let existing_hash = self
            .connection
            .query_row(
                "SELECT command_hash FROM pending_asset_replacements WHERE command_id = ?1",
                [command_id],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?;
        if let Some(existing_hash) = existing_hash {
            if existing_hash != command_hash {
                return Err(SoftchatError::CommandConflict);
            }
            return self
                .product_pending_asset_replacement(command_id)?
                .ok_or(SoftchatError::InvalidPersistenceResult);
        }
        let (subject, emoji_tags) = match target {
            AssetReplacementTarget::ProfilePicture | AssetReplacementTarget::ProfileBanner => {
                if !conversation_id.is_empty() {
                    return Err(SoftchatError::InvalidMediaOperation);
                }
                (String::new(), Vec::new())
            }
            AssetReplacementTarget::ConversationIcon => {
                validate_conversation_id(conversation_id)?;
                let conversation_exists = self
                    .connection
                    .query_row(
                        "SELECT 1 FROM conversation_state WHERE conversation_id = ?1",
                        [conversation_id],
                        |row| row.get::<_, i64>(0),
                    )
                    .optional()
                    .map_err(map_sqlite_error)?
                    .is_some();
                if !conversation_exists {
                    return Err(SoftchatError::InvalidMediaOperation);
                }
                let subject = self.product_subject(conversation_id)?;
                (subject.text, subject.emoji_tags)
            }
        };
        let emoji_tags_json =
            serde_json::to_string(&emoji_tags).map_err(|_| SoftchatError::InternalFailure)?;
        let upload_command_id =
            stable_asset_upload_command_id(command_id, target, conversation_id, source_fingerprint);
        let operation_id = stable_media_operation_id(
            &self.account_id,
            &upload_command_id,
            "upload",
            source_fingerprint,
        );
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO media_operations
                 (operation_id, command_id, conversation_id, direction,
                  protection, state, source_fingerprint, created_at, updated_at)
                 VALUES (?1, ?2, ?3, 'upload', 'public', 'prepared', ?4, ?5, ?5)",
                params![
                    operation_id,
                    upload_command_id,
                    (!conversation_id.is_empty()).then_some(conversation_id),
                    source_fingerprint,
                    now,
                ],
            )
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO pending_asset_replacements
                 (command_id, target, conversation_id, operation_id, subject,
                  emoji_tags_json, command_hash, state, created_at, updated_at)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'waiting_for_upload', ?8, ?8)",
                params![
                    command_id,
                    asset_target_name(target),
                    conversation_id,
                    operation_id,
                    subject,
                    emoji_tags_json,
                    command_hash,
                    now,
                ],
            )
            .map_err(map_sqlite_error)?;
        bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        self.record_diagnostic(Diagnostic::new(
            AccountLogCategory::Media,
            "media.dependency.waiting",
            "Publication waiting for uploads",
        ));
        self.product_pending_asset_replacement(command_id)?
            .ok_or(SoftchatError::InvalidPersistenceResult)
    }

    pub(crate) fn product_pending_asset_replacement(
        &self,
        command_id: &str,
    ) -> Result<Option<PendingAssetReplacement>, SoftchatError> {
        validate_command_id(command_id)?;
        let row = self
            .connection
            .query_row(
                "SELECT target, conversation_id, operation_id, state,
                        result_operation_id, updated_at
                 FROM pending_asset_replacements WHERE command_id = ?1",
                [command_id],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, String>(4)?,
                        row.get::<_, i64>(5)?,
                    ))
                },
            )
            .optional()
            .map_err(map_sqlite_error)?;
        let Some((target, conversation_id, operation_id, state, result_operation_id, updated_at)) =
            row
        else {
            return Ok(None);
        };
        let operation = self
            .product_media_operation(&operation_id)?
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        Ok(Some(PendingAssetReplacement {
            command_id: command_id.to_owned(),
            target: asset_target_from_name(&target)?,
            conversation_id,
            state: pending_asset_replacement_state_from_name(&state)?,
            operation,
            result_operation_id,
            updated_at,
        }))
    }

    pub(crate) fn product_pending_asset_replacements(
        &self,
        limit: u32,
    ) -> Result<Vec<PendingAssetReplacement>, SoftchatError> {
        validate_limit(limit)?;
        let mut statement = self
            .connection
            .prepare(
                "SELECT command_id FROM pending_asset_replacements
                 ORDER BY updated_at DESC, command_id ASC LIMIT ?1",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([limit], |row| row.get::<_, String>(0))
            .map_err(map_sqlite_error)?;
        let mut replacements = Vec::new();
        for row in rows {
            replacements.push(
                self.product_pending_asset_replacement(&row.map_err(map_sqlite_error)?)?
                    .ok_or(SoftchatError::InvalidPersistenceResult)?,
            );
        }
        Ok(replacements)
    }

    pub(crate) fn ready_pending_asset_commands(
        &self,
        limit: u32,
    ) -> Result<Vec<String>, SoftchatError> {
        validate_limit(limit)?;
        let mut statement = self
            .connection
            .prepare(
                "SELECT command_id FROM pending_asset_replacements
                 WHERE state = 'ready_to_publish'
                 ORDER BY updated_at ASC, command_id ASC LIMIT ?1",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([limit], |row| row.get::<_, String>(0))
            .map_err(map_sqlite_error)?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(map_sqlite_error)
    }

    pub(crate) fn pending_asset_command_for_operation(
        &self,
        operation_id: &str,
    ) -> Result<Option<String>, SoftchatError> {
        self.connection
            .query_row(
                "SELECT command_id FROM pending_asset_replacements WHERE operation_id = ?1",
                [operation_id],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(map_sqlite_error)
    }

    pub(crate) fn pending_asset_subject(
        &self,
        command_id: &str,
    ) -> Result<(String, Vec<Vec<String>>), SoftchatError> {
        let (subject, emoji_tags_json) = self
            .connection
            .query_row(
                "SELECT subject, emoji_tags_json
                 FROM pending_asset_replacements WHERE command_id = ?1",
                [command_id],
                |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
            )
            .map_err(map_sqlite_error)?;
        let emoji_tags = serde_json::from_str(&emoji_tags_json)
            .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
        Ok((subject, emoji_tags))
    }

    fn pending_asset_subject_for_operation(
        &self,
        operation_id: &str,
    ) -> Result<Option<(String, String)>, SoftchatError> {
        let target = self
            .connection
            .query_row(
                "SELECT target, conversation_id
                 FROM pending_asset_replacements WHERE operation_id = ?1",
                [operation_id],
                |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
            )
            .optional()
            .map_err(map_sqlite_error)?;
        let Some((target, conversation_id)) = target else {
            return Ok(None);
        };
        if asset_target_from_name(&target)? != AssetReplacementTarget::ConversationIcon {
            return Ok(Some((String::new(), "[]".to_owned())));
        }
        let subject = self.product_subject(&conversation_id)?;
        let emoji_tags_json = serde_json::to_string(&subject.emoji_tags)
            .map_err(|_| SoftchatError::InternalFailure)?;
        Ok(Some((subject.text, emoji_tags_json)))
    }

    pub(crate) fn mark_pending_asset_published(
        &mut self,
        command_id: &str,
        result_operation_id: &str,
        now: i64,
    ) -> Result<AccountMutationResult, SoftchatError> {
        validate_command_id(command_id)?;
        validate_command_id(result_operation_id)?;
        validate_timestamp(now)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let changed = transaction
            .execute(
                "UPDATE pending_asset_replacements
                 SET state = 'published', result_operation_id = ?1, updated_at = ?2
                 WHERE command_id = ?3
                   AND state IN ('ready_to_publish', 'published')
                   AND (result_operation_id = '' OR result_operation_id = ?1)",
                params![result_operation_id, now, command_id],
            )
            .map_err(map_sqlite_error)?;
        if changed != 1 {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        Ok(AccountMutationResult { revision })
    }

    pub(crate) fn prepare_product_media(
        &mut self,
        command_id: &str,
        conversation_id: Option<String>,
        direction: &str,
        source_fingerprint: &str,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        self.prepare_product_media_with_attachment(ProductMediaOperationPreparation {
            command_id,
            conversation_id,
            kind: ProductMediaOperationKind::Transfer,
            direction,
            protection: AccountMediaProtection::Encrypted,
            source_fingerprint,
            source_message_id: None,
            attachment: None,
            now,
        })
    }

    pub(crate) fn prepare_product_media_download(
        &mut self,
        message_id: &str,
        attachment_url: &str,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        let message = self
            .product_message(message_id)?
            .filter(|message| !message.deleted)
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        let attachment = message
            .attachments
            .into_iter()
            .find(|attachment| attachment.url == attachment_url)
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        let source_fingerprint = command_hash(&("download", message_id, &attachment))?;
        let command_id = format!("media-download-{source_fingerprint}");
        self.prepare_product_media_with_attachment(ProductMediaOperationPreparation {
            command_id: &command_id,
            conversation_id: Some(message.conversation_id),
            source_message_id: Some(message_id),
            kind: ProductMediaOperationKind::Transfer,
            direction: "download",
            protection: if attachment.encryption.is_some() {
                AccountMediaProtection::Encrypted
            } else {
                AccountMediaProtection::Public
            },
            source_fingerprint: &source_fingerprint,
            attachment: Some(attachment),
            now,
        })
    }

    pub(crate) fn prepare_product_conversation_icon_download(
        &mut self,
        conversation_id: &str,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        validate_conversation_id(conversation_id)?;
        let exists = self
            .connection
            .query_row(
                "SELECT EXISTS(
                    SELECT 1 FROM conversation_state WHERE conversation_id = ?1
                 )",
                [conversation_id],
                |row| row.get::<_, bool>(0),
            )
            .map_err(map_sqlite_error)?;
        if !exists {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let subject = self.product_subject(conversation_id)?;
        let attachment = subject
            .icon
            .filter(has_conversation_icon_integrity)
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        let source_fingerprint = conversation_icon_source_fingerprint(
            conversation_id,
            &subject.icon_event_id,
            &attachment,
        )?;
        let command_id = format!("conversation-icon-download-{source_fingerprint}");
        self.prepare_product_media_with_attachment(ProductMediaOperationPreparation {
            command_id: &command_id,
            conversation_id: Some(conversation_id.to_owned()),
            source_message_id: None,
            kind: ProductMediaOperationKind::ConversationIcon,
            direction: "download",
            protection: if attachment.encryption.is_some() {
                AccountMediaProtection::Encrypted
            } else {
                AccountMediaProtection::Public
            },
            source_fingerprint: &source_fingerprint,
            attachment: Some(attachment),
            now,
        })
    }

    fn prepare_product_media_with_attachment(
        &mut self,
        preparation: ProductMediaOperationPreparation<'_>,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        let ProductMediaOperationPreparation {
            command_id,
            conversation_id,
            source_message_id,
            kind,
            direction,
            protection,
            source_fingerprint,
            attachment,
            now,
        } = preparation;
        validate_command_id(command_id)?;
        validate_timestamp(now)?;
        if let Some(value) = &conversation_id {
            validate_conversation_id(value)?;
        }
        if let Some(value) = source_message_id {
            NostrEventId::from_hex(value)?;
        }
        if !matches!(direction, "upload" | "download")
            || source_fingerprint.is_empty()
            || source_fingerprint.len() > 512
        {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let id =
            stable_media_operation_id(&self.account_id, command_id, direction, source_fingerprint);
        let existing = self.product_media_by_command(command_id)?;
        if let Some(value) = existing {
            if value.id != id
                || value.conversation_id != conversation_id.clone().unwrap_or_default()
                || value.source_message_id != source_message_id.unwrap_or_default()
                || self.product_media_operation_kind(&value.id)? != kind
                || value.protection != protection
                || value.attachment != attachment
            {
                return Err(SoftchatError::CommandConflict);
            }
            return Ok(value);
        }
        let attachment_json = attachment
            .as_ref()
            .map(serde_json::to_string)
            .transpose()
            .map_err(|_| SoftchatError::InvalidMediaOperation)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO media_operations
                 (operation_id, command_id, conversation_id, source_message_id,
                  operation_kind, direction, protection, state, source_fingerprint,
                  attachment_json, created_at, updated_at)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, 'prepared', ?8, ?9, ?10, ?10)",
                params![
                    id,
                    command_id,
                    conversation_id,
                    source_message_id,
                    kind.as_str(),
                    direction,
                    media_protection_name(protection),
                    source_fingerprint,
                    attachment_json,
                    now,
                ],
            )
            .map_err(map_sqlite_error)?;
        bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        let operation = self
            .product_media_operation(&id)?
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        self.record_diagnostic(crate::account_diagnostics::media(&operation));
        Ok(operation)
    }

    pub(crate) fn complete_product_media(
        &mut self,
        operation_id: &str,
        lease_id: &str,
        attachment: AttachmentMetadata,
        byte_count: u64,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        validate_timestamp(now)?;
        attachment.to_tag()?;
        if attachment_tag_json_bytes(&attachment)? > MAX_PRODUCT_ATTACHMENT_TAG_BYTES {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let existing = self
            .product_media_operation(operation_id)?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        if existing.state == AccountMediaOperationState::Completed {
            if existing.attachment.as_ref() == Some(&attachment)
                && existing.byte_count == Some(byte_count)
            {
                return Ok(existing);
            }
            return Err(SoftchatError::InvalidMediaOperation);
        }
        if self.product_media_operation_kind(operation_id)?
            == ProductMediaOperationKind::ConversationIcon
            && attachment != self.validate_current_conversation_icon_operation(&existing)?
        {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        if existing.protection == AccountMediaProtection::Public && attachment.encryption.is_some()
        {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let pending_asset_subject = self.pending_asset_subject_for_operation(operation_id)?;
        let byte_count =
            i64::try_from(byte_count).map_err(|_| SoftchatError::InvalidMediaOperation)?;
        let attachment_json =
            serde_json::to_string(&attachment).map_err(|_| SoftchatError::InternalFailure)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let changed = transaction
            .execute(
                "UPDATE media_operations
                 SET state = 'completed', attachment_json = ?1, byte_count = ?2,
                     lease_id = '', lease_expires_at = 0,
                     last_error_category = '', updated_at = ?3
                 WHERE operation_id = ?4 AND state = 'running' AND lease_id = ?5",
                params![attachment_json, byte_count, now, operation_id, lease_id],
            )
            .map_err(map_sqlite_error)?;
        if changed != 1 {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let message_ready = transaction
            .execute(
                "UPDATE pending_media_messages
                 SET state = 'ready_to_publish', updated_at = ?1
                 WHERE command_id = (
                   SELECT command_id FROM pending_message_attachments
                   WHERE operation_id = ?2
                 )
                   AND state = 'waiting_for_uploads'
                   AND NOT EXISTS (
                     SELECT 1
                     FROM pending_message_attachments dependency
                     JOIN media_operations media
                       ON media.operation_id = dependency.operation_id
                     WHERE dependency.command_id = pending_media_messages.command_id
                       AND media.state <> 'completed'
                   )",
                params![now, operation_id],
            )
            .map_err(map_sqlite_error)?;
        let mut asset_ready = 0;
        if let Some((subject, emoji_tags_json)) = pending_asset_subject {
            asset_ready = transaction
                .execute(
                    "UPDATE pending_asset_replacements
                     SET subject = ?1, emoji_tags_json = ?2,
                         state = 'ready_to_publish', updated_at = ?3
                     WHERE operation_id = ?4 AND state = 'waiting_for_upload'",
                    params![subject, emoji_tags_json, now, operation_id],
                )
                .map_err(map_sqlite_error)?;
        }
        bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        let operation = self
            .product_media_operation(operation_id)?
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        self.record_diagnostic(crate::account_diagnostics::media(&operation));
        if message_ready > 0 || asset_ready > 0 {
            self.record_diagnostic(Diagnostic::new(
                AccountLogCategory::Media,
                "media.dependency.ready",
                "Upload dependencies completed",
            ));
        }
        Ok(operation)
    }

    pub(crate) fn restart_product_media_download(
        &mut self,
        operation_id: &str,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        validate_timestamp(now)?;
        let operation = self
            .product_media_operation(operation_id)?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        if operation.direction != "download" {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        if !matches!(
            operation.state,
            AccountMediaOperationState::Completed
                | AccountMediaOperationState::Failed
                | AccountMediaOperationState::Cancelled
        ) {
            return Ok(operation);
        }
        if self.product_media_operation_kind(operation_id)?
            == ProductMediaOperationKind::ConversationIcon
        {
            self.validate_current_conversation_icon_operation(&operation)?;
        }
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let changed = transaction
            .execute(
                "UPDATE media_operations
                 SET state = 'prepared', byte_count = NULL, lease_id = '',
                     lease_expires_at = 0, last_error_category = '',
                     updated_at = ?1
                 WHERE operation_id = ?2 AND direction = 'download'
                   AND state IN ('completed', 'failed', 'cancelled')",
                params![now, operation_id],
            )
            .map_err(map_sqlite_error)?;
        if changed != 1 {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        let operation = self
            .product_media_operation(operation_id)?
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        self.record_diagnostic(crate::account_diagnostics::media(&operation));
        Ok(operation)
    }

    fn validate_current_conversation_icon_operation(
        &self,
        operation: &AccountMediaOperation,
    ) -> Result<AttachmentMetadata, SoftchatError> {
        let current_subject = self.product_subject(&operation.conversation_id)?;
        let current_attachment = current_subject
            .icon
            .filter(has_conversation_icon_integrity)
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        let current_fingerprint = conversation_icon_source_fingerprint(
            &operation.conversation_id,
            &current_subject.icon_event_id,
            &current_attachment,
        )?;
        if operation.source_fingerprint != current_fingerprint
            || operation.attachment.as_ref() != Some(&current_attachment)
        {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        Ok(current_attachment)
    }

    pub(crate) fn claim_product_media(
        &mut self,
        operation_id: &str,
        lease_id: &str,
        now: i64,
        lease_duration_seconds: u32,
    ) -> Result<Option<AccountMediaLease>, SoftchatError> {
        validate_timestamp(now)?;
        validate_media_operation_id(operation_id)?;
        validate_media_lease_id(lease_id)?;
        if lease_duration_seconds == 0 || lease_duration_seconds > 3_600 {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let expires_at = now
            .checked_add(i64::from(lease_duration_seconds))
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let recovered = transaction
            .execute(
                "UPDATE media_operations
                 SET state = 'retryable', lease_id = '', lease_expires_at = 0,
                     last_error_category = 'lease_expired', updated_at = ?1
                 WHERE state = 'running' AND lease_expires_at > 0
                   AND lease_expires_at <= ?1",
                [now],
            )
            .map_err(map_sqlite_error)?;
        let claimable = transaction
            .query_row(
                "SELECT operation_id FROM media_operations
                 WHERE operation_id = ?1
                   AND state IN ('prepared', 'retryable') AND lease_id = ''",
                [operation_id],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?;
        let Some(operation_id) = claimable else {
            transaction.commit().map_err(map_sqlite_error)?;
            self.record_media_lease_recovery(recovered);
            return Ok(None);
        };
        let changed = transaction
            .execute(
                "UPDATE media_operations
                 SET state = 'running', lease_id = ?1, lease_expires_at = ?2,
                     attempt_count = attempt_count + 1,
                     last_error_category = '', updated_at = ?3
                 WHERE operation_id = ?4
                   AND state IN ('prepared', 'retryable') AND lease_id = ''",
                params![lease_id, expires_at, now, operation_id],
            )
            .map_err(map_sqlite_error)?;
        if changed != 1 {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        self.record_media_lease_recovery(recovered);
        let operation = self
            .product_media_operation(&operation_id)?
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        self.record_diagnostic(crate::account_diagnostics::media(&operation));
        Ok(Some(AccountMediaLease {
            lease_id: lease_id.to_owned(),
            expires_at,
            operation,
        }))
    }

    fn record_media_lease_recovery(&self, recovered: usize) {
        if recovered > 0 {
            let mut event = Diagnostic::new(
                AccountLogCategory::Recovery,
                "recovery.completed",
                "Expired media leases recovered",
            )
            .count(
                crate::account_diagnostics::AccountLogCounterKind::RecoveredItems,
                recovered as u64,
            );
            event.source = Some(crate::account_diagnostics::AccountLogSource::Recovery);
            event.reason = "lease_expired";
            event.summarize = true;
            self.record_diagnostic(event);
        }
    }

    pub(crate) fn finish_product_media_lease(
        &mut self,
        operation_id: &str,
        lease_id: &str,
        retryable: bool,
        error_category: &str,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        validate_timestamp(now)?;
        validate_media_lease_id(lease_id)?;
        validate_media_error_category(error_category)?;
        let state = if retryable { "retryable" } else { "failed" };
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let changed = transaction
            .execute(
                "UPDATE media_operations
                 SET state = ?1, lease_id = '', lease_expires_at = 0,
                     last_error_category = ?2, updated_at = ?3
                 WHERE operation_id = ?4 AND state = 'running' AND lease_id = ?5",
                params![state, error_category, now, operation_id, lease_id],
            )
            .map_err(map_sqlite_error)?;
        if changed != 1 {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        if !retryable {
            transaction
                .execute(
                    "UPDATE pending_media_messages
                     SET state = 'failed', updated_at = ?1
                     WHERE command_id = (
                       SELECT command_id FROM pending_message_attachments
                       WHERE operation_id = ?2
                     ) AND state = 'waiting_for_uploads'",
                    params![now, operation_id],
                )
                .map_err(map_sqlite_error)?;
            transaction
                .execute(
                    "UPDATE pending_asset_replacements
                     SET state = 'failed', updated_at = ?1
                     WHERE operation_id = ?2 AND state = 'waiting_for_upload'",
                    params![now, operation_id],
                )
                .map_err(map_sqlite_error)?;
        }
        bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        let operation = self
            .product_media_operation(operation_id)?
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        self.record_diagnostic(crate::account_diagnostics::media(&operation));
        Ok(operation)
    }

    pub(crate) fn transition_product_media(
        &mut self,
        operation_id: &str,
        transition: &str,
        error_category: &str,
        now: i64,
    ) -> Result<AccountMediaOperation, SoftchatError> {
        validate_timestamp(now)?;
        if error_category.len() > 128 {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let (state, allowed) = match transition {
            "retry" => ("prepared", "retryable"),
            "cancel" => ("cancelled", "prepared,retryable"),
            _ => return Err(SoftchatError::InvalidMediaOperation),
        };
        let current = self
            .product_media_operation(operation_id)?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        let current_name = media_state_name(current.state);
        if !allowed.split(',').any(|value| value == current_name) {
            if current_name == state {
                return Ok(current);
            }
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "UPDATE media_operations
                 SET state = ?1, last_error_category = ?2,
                     lease_id = '', lease_expires_at = 0, updated_at = ?3
                 WHERE operation_id = ?4",
                params![state, error_category, now, operation_id],
            )
            .map_err(map_sqlite_error)?;
        if state == "cancelled" {
            transaction
                .execute(
                    "UPDATE pending_media_messages
                     SET state = 'cancelled', updated_at = ?1
                     WHERE command_id = (
                       SELECT command_id FROM pending_message_attachments
                       WHERE operation_id = ?2
                     ) AND state = 'waiting_for_uploads'",
                    params![now, operation_id],
                )
                .map_err(map_sqlite_error)?;
            transaction
                .execute(
                    "UPDATE pending_asset_replacements
                     SET state = 'cancelled', updated_at = ?1
                     WHERE operation_id = ?2 AND state = 'waiting_for_upload'",
                    params![now, operation_id],
                )
                .map_err(map_sqlite_error)?;
        }
        bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        let operation = self
            .product_media_operation(operation_id)?
            .ok_or(SoftchatError::InvalidPersistenceResult)?;
        self.record_diagnostic(crate::account_diagnostics::media(&operation));
        Ok(operation)
    }

    pub(crate) fn product_message_downloads(
        &self,
        message_id: &str,
    ) -> Result<Vec<AccountMediaOperation>, SoftchatError> {
        let Some(message) = self
            .product_message(message_id)?
            .filter(|value| !value.deleted)
        else {
            return Ok(Vec::new());
        };
        let mut downloads = Vec::new();
        for attachment in message.attachments {
            let fingerprint = command_hash(&("download", message_id, &attachment))?;
            if let Some(operation) =
                self.product_media_by_command(&format!("media-download-{fingerprint}"))?
            {
                if operation.direction != "download"
                    || operation.conversation_id != message.conversation_id
                    || operation.source_message_id != message_id
                    || operation.source_fingerprint != fingerprint
                    || operation.attachment.as_ref() != Some(&attachment)
                {
                    return Err(SoftchatError::CommandConflict);
                }
                downloads.push(operation);
            }
        }
        Ok(downloads)
    }

    pub(crate) fn product_media_operations(
        &self,
        limit: u32,
    ) -> Result<Vec<AccountMediaOperation>, SoftchatError> {
        validate_limit(limit)?;
        let mut statement = self
            .connection
            .prepare(
                "SELECT operation_id FROM media_operations
                 ORDER BY updated_at DESC, operation_id ASC LIMIT ?1",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([limit], |row| row.get::<_, String>(0))
            .map_err(map_sqlite_error)?;
        let mut result = Vec::new();
        for row in rows {
            result.push(
                self.product_media_operation(&row.map_err(map_sqlite_error)?)?
                    .ok_or(SoftchatError::InvalidPersistenceResult)?,
            );
        }
        Ok(result)
    }

    pub(crate) fn product_media_operations_by_ids(
        &self,
        operation_ids: &[String],
    ) -> Result<Vec<AccountMediaOperation>, SoftchatError> {
        if operation_ids.len() > MAX_PRODUCT_MEDIA_OPERATION_LOOKUPS {
            return Err(SoftchatError::InvalidMediaOperation);
        }
        let mut distinct = BTreeSet::new();
        for operation_id in operation_ids {
            validate_media_operation_id(operation_id)?;
            if !distinct.insert(operation_id) {
                return Err(SoftchatError::InvalidMediaOperation);
            }
        }
        operation_ids
            .iter()
            .filter_map(
                |operation_id| match self.product_media_operation(operation_id) {
                    Ok(Some(operation)) => Some(Ok(operation)),
                    Ok(None) => None,
                    Err(error) => Some(Err(error)),
                },
            )
            .collect()
    }

    pub(crate) fn product_recoverable_media_operations(
        &self,
        cursor: Option<&str>,
        limit: u32,
    ) -> Result<AccountMediaOperationPage, SoftchatError> {
        validate_limit(limit)?;
        if let Some(cursor) = cursor {
            validate_media_operation_id(cursor)?;
        }
        let cursor = cursor.unwrap_or_default();
        let query_limit = i64::from(limit) + 1;
        let mut statement = self
            .connection
            .prepare(
                "SELECT operation_id FROM media_operations
                 WHERE state IN ('prepared', 'running', 'retryable')
                   AND (?1 = '' OR operation_id > ?1)
                 ORDER BY operation_id ASC LIMIT ?2",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params![cursor, query_limit], |row| row.get::<_, String>(0))
            .map_err(map_sqlite_error)?;
        let mut operation_ids = rows
            .collect::<Result<Vec<_>, _>>()
            .map_err(map_sqlite_error)?;
        let has_more = operation_ids.len() > limit as usize;
        operation_ids.truncate(limit as usize);
        let next_cursor = has_more.then(|| operation_ids.last().cloned()).flatten();
        let operations = operation_ids
            .into_iter()
            .map(|operation_id| {
                self.product_media_operation(&operation_id)?
                    .ok_or(SoftchatError::InvalidPersistenceResult)
            })
            .collect::<Result<Vec<_>, _>>()?;
        Ok(AccountMediaOperationPage {
            operations,
            next_cursor,
        })
    }

    fn product_message_views(
        &self,
        query: ProductMessageQuery<'_>,
    ) -> Result<Vec<AccountMessageView>, SoftchatError> {
        let (predicate, ordering, parameters) = match query {
            ProductMessageQuery::ConversationPage {
                conversation_id,
                reply_to,
                cursor,
                limit,
            } => {
                let (has_cursor, cursor_timestamp, cursor_id) = cursor.map_or_else(
                    || (0_i64, 0_i64, String::new()),
                    |value| (1_i64, value.created_at, value.message_id.clone()),
                );
                (
                    "p.conversation_id = ?
                     AND (? = '' OR (
                       EXISTS (SELECT 1 FROM projection_targets target
                               WHERE target.logical_event_id = p.logical_event_id
                                 AND target.target_event_id = ?)
                       AND NOT EXISTS (SELECT 1 FROM json_each(rumor.canonical_json, '$.tags') tag
                                       WHERE json_extract(tag.value, '$[0]') = 'q')
                     ))
                     AND (
                       ? = 0
                       OR p.created_at < ?
                       OR (p.created_at = ? AND p.logical_event_id > ?)
                     )"
                    .to_owned(),
                    "ORDER BY p.created_at DESC, p.logical_event_id ASC LIMIT ?".to_owned(),
                    vec![
                        SqlValue::Text(conversation_id.to_owned()),
                        SqlValue::Text(reply_to.unwrap_or_default().to_owned()),
                        SqlValue::Text(reply_to.unwrap_or_default().to_owned()),
                        SqlValue::Integer(has_cursor),
                        SqlValue::Integer(cursor_timestamp),
                        SqlValue::Integer(cursor_timestamp),
                        SqlValue::Text(cursor_id),
                        SqlValue::Integer(i64::from(limit)),
                    ],
                )
            }
            ProductMessageQuery::Ids(message_ids) => {
                if message_ids.is_empty() {
                    return Ok(Vec::new());
                }
                for message_id in message_ids {
                    NostrEventId::from_hex(message_id)?;
                }
                let placeholders = vec!["?"; message_ids.len()].join(",");
                (
                    format!("p.logical_event_id IN ({placeholders})"),
                    "ORDER BY p.created_at DESC, p.logical_event_id ASC".to_owned(),
                    message_ids.iter().cloned().map(SqlValue::Text).collect(),
                )
            }
        };
        let sql = format!(
            "SELECT
               p.logical_event_id,
               p.conversation_id,
               p.author_public_key,
               p.created_at,
               rumor.canonical_json,
               EXISTS(
                 SELECT 1 FROM projection_targets deletion_target
                 JOIN projections deletion
                   ON deletion.logical_event_id = deletion_target.logical_event_id
                 WHERE deletion_target.target_event_id = p.logical_event_id
                   AND deletion.kind = 'deletion'
                   AND deletion.author_public_key = p.author_public_key
               ),
               latest_edit.logical_event_id,
               latest_edit_rumor.canonical_json,
               (
                 SELECT COUNT(*)
                 FROM projection_targets reply_target
                 JOIN projections reply
                   ON reply.logical_event_id = reply_target.logical_event_id
                 JOIN authenticated_rumors reply_rumor
                   ON reply_rumor.rumor_id = reply.logical_event_id
                 WHERE reply_target.target_event_id = p.logical_event_id
                   AND reply.kind = 'chat_message'
                   AND reply.conversation_id = p.conversation_id
                   AND NOT EXISTS (
                     SELECT 1 FROM json_each(reply_rumor.canonical_json, '$.tags') tag
                     WHERE json_extract(tag.value, '$[0]') = 'q'
                   )
                   AND NOT EXISTS (
                     SELECT 1 FROM projection_targets deletion_target
                     JOIN projections deletion
                       ON deletion.logical_event_id =
                          deletion_target.logical_event_id
                     WHERE deletion_target.target_event_id =
                           reply.logical_event_id
                       AND deletion.kind = 'deletion'
                       AND deletion.author_public_key =
                           reply.author_public_key
                   )
               )
             FROM projections p
             JOIN authenticated_rumors rumor
               ON rumor.rumor_id = p.logical_event_id
             LEFT JOIN projections latest_edit
               ON latest_edit.logical_event_id = (
                 SELECT effective.edit_id FROM effective_message_edits effective
                 WHERE effective.message_id = p.logical_event_id
               )
             LEFT JOIN authenticated_rumors latest_edit_rumor
               ON latest_edit_rumor.rumor_id = latest_edit.logical_event_id
             WHERE p.kind = 'chat_message' AND p.ephemeral = 0
               AND {predicate}
             {ordering}"
        );
        let mut statement = self.connection.prepare(&sql).map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params_from_iter(parameters), |row| {
                Ok(ProductMessageRow {
                    id: row.get(0)?,
                    conversation_id: row.get(1)?,
                    author_public_key: row.get(2)?,
                    created_at: row.get(3)?,
                    rumor_json: row.get(4)?,
                    deleted: row.get::<_, i64>(5)? != 0,
                    edit_id: row.get(6)?,
                    edit_rumor_json: row.get(7)?,
                    reply_count: row.get(8)?,
                })
            })
            .map_err(map_sqlite_error)?;
        let mut message_rows = Vec::new();
        for row in rows {
            message_rows.push(row.map_err(map_sqlite_error)?);
        }
        let message_ids = message_rows
            .iter()
            .map(|row| row.id.clone())
            .collect::<Vec<_>>();
        let mut reactions = self.product_reactions_for_messages(&message_ids)?;
        message_rows
            .into_iter()
            .map(|row| {
                let row_reactions = reactions.remove(&row.id).unwrap_or_default();
                product_message_from_row(row, row_reactions)
            })
            .collect()
    }

    /// Load authoritative local toggle state independently of display paging.
    /// One deletion rumor supports at most 256 targets; reject larger groups
    /// before authoring so a successful toggle cannot leave active copies.
    pub(crate) fn active_product_reaction_ids(
        &self,
        message_id: &str,
        author_public_key: &str,
        value: &str,
    ) -> Result<Vec<String>, SoftchatError> {
        NostrEventId::from_hex(message_id)?;
        NostrPublicKey::from_hex(author_public_key)?;
        let mut statement = self
            .connection
            .prepare(
                "SELECT reaction.logical_event_id
             FROM projection_targets target
             JOIN projections reaction ON reaction.logical_event_id = target.logical_event_id
             WHERE target.target_event_id = ?1
               AND reaction.kind = 'reaction'
               AND reaction.author_public_key = ?2
               AND reaction.content = ?3
               AND NOT EXISTS (
                 SELECT 1 FROM projection_targets deletion_target
                 JOIN projections deletion
                   ON deletion.logical_event_id = deletion_target.logical_event_id
                 WHERE deletion_target.target_event_id = reaction.logical_event_id
                   AND deletion.kind = 'deletion'
                   AND deletion.author_public_key = reaction.author_public_key
               )
             ORDER BY reaction.created_at DESC, reaction.logical_event_id ASC
             LIMIT 257",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params![message_id, author_public_key, value], |row| {
                row.get::<_, String>(0)
            })
            .map_err(map_sqlite_error)?;
        let ids = rows
            .collect::<Result<Vec<_>, _>>()
            .map_err(map_sqlite_error)?;
        if ids.len() > 256 {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        Ok(ids)
    }

    fn product_reactions_for_messages(
        &self,
        message_ids: &[String],
    ) -> Result<BTreeMap<String, Vec<MessageReactionView>>, SoftchatError> {
        if message_ids.is_empty() {
            return Ok(BTreeMap::new());
        }
        for message_id in message_ids {
            NostrEventId::from_hex(message_id)?;
        }
        let placeholders = vec!["?"; message_ids.len()].join(",");
        let sql = format!(
            "WITH ranked_reactions AS (
               SELECT
                 reaction_target.target_event_id,
                 reaction.logical_event_id,
                 reaction.author_public_key,
                 reaction.created_at,
                 rumor.canonical_json,
                 ROW_NUMBER() OVER (
                   PARTITION BY reaction_target.target_event_id
                   ORDER BY reaction.created_at DESC,
                            reaction.logical_event_id ASC
                 ) AS reaction_rank
               FROM projection_targets reaction_target
               JOIN projections reaction
                 ON reaction.logical_event_id =
                    reaction_target.logical_event_id
               JOIN authenticated_rumors rumor
                 ON rumor.rumor_id = reaction.logical_event_id
               WHERE reaction.kind = 'reaction'
                 AND reaction_target.target_event_id IN ({placeholders})
                 AND NOT EXISTS (
                   SELECT 1 FROM projection_targets deletion_target
                   JOIN projections deletion
                     ON deletion.logical_event_id =
                        deletion_target.logical_event_id
                   WHERE deletion_target.target_event_id =
                         reaction.logical_event_id
                     AND deletion.kind = 'deletion'
                     AND deletion.author_public_key =
                         reaction.author_public_key
                 )
             )
             SELECT
               target_event_id,
               logical_event_id,
               author_public_key,
               created_at,
               canonical_json
             FROM ranked_reactions
             WHERE reaction_rank <= ?
             ORDER BY target_event_id ASC, created_at DESC,
                      logical_event_id ASC"
        );
        let mut parameters = message_ids
            .iter()
            .cloned()
            .map(SqlValue::Text)
            .collect::<Vec<_>>();
        parameters.push(SqlValue::Integer(i64::from(MAX_PRODUCT_QUERY_PAGE)));
        let mut statement = self.connection.prepare(&sql).map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params_from_iter(parameters), |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, i64>(3)?,
                    row.get::<_, String>(4)?,
                ))
            })
            .map_err(map_sqlite_error)?;
        let mut latest = BTreeMap::<String, BTreeMap<(String, String), MessageReactionView>>::new();
        for row in rows {
            let (target_id, id, author_public_key, created_at, rumor_json) =
                row.map_err(map_sqlite_error)?;
            let rumor = serde_json::from_str::<RumorEvent>(&rumor_json)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            let parsed = classify_reaction(rumor)?;
            latest
                .entry(target_id)
                .or_default()
                .entry((author_public_key.clone(), parsed.reaction.clone()))
                .or_insert(MessageReactionView {
                    id,
                    author_public_key,
                    value: parsed.reaction,
                    custom_emoji_url: parsed.custom_emoji_url,
                    created_at,
                });
        }
        Ok(latest
            .into_iter()
            .map(|(message_id, values)| (message_id, values.into_values().collect()))
            .collect())
    }

    fn product_subject(&self, conversation_id: &str) -> Result<ProductSubject, SoftchatError> {
        let latest = self
            .list_event_nodes(
                Some(ProjectionKind::Subject),
                String::new(),
                conversation_id.to_owned(),
                String::new(),
                String::new(),
                None,
                1,
                0,
            )?
            .into_iter()
            .next();
        let mut subject = if let Some(latest) = latest {
            let rumor = latest
                .projection
                .rumor
                .ok_or(SoftchatError::InvalidPersistenceResult)?;
            classify_product_subject(rumor)?
        } else {
            ProductSubject {
                icon_event_id: String::new(),
                text: String::new(),
                icon: None,
                emoji_tags: Vec::new(),
            }
        };
        let stored_icon = self
            .connection
            .query_row(
                "SELECT subject_event_id, attachment_json
                 FROM conversation_subject_icons WHERE conversation_id = ?1",
                [conversation_id],
                |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)),
            )
            .optional()
            .map_err(map_sqlite_error)?;
        if let Some((event_id, attachment_json)) = stored_icon {
            let (event_id, icon) =
                parse_stored_subject_icon(Some(event_id), Some(attachment_json))?
                    .ok_or(SoftchatError::InvalidPersistenceResult)?;
            subject.icon_event_id = event_id;
            subject.icon = Some(icon);
        } else {
            subject.icon_event_id.clear();
            subject.icon = None;
        }
        Ok(subject)
    }

    fn product_conversation_members(
        &self,
        conversation_id: &str,
    ) -> Result<Vec<String>, SoftchatError> {
        let mut statement = self
            .connection
            .prepare(
                "SELECT public_key FROM conversation_members
                 WHERE conversation_id = ?1 ORDER BY public_key ASC",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([conversation_id], |row| row.get::<_, String>(0))
            .map_err(map_sqlite_error)?;
        let mut result = Vec::new();
        for row in rows {
            let key = row.map_err(map_sqlite_error)?;
            NostrPublicKey::from_hex(&key).map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            result.push(key);
        }
        if result.len() < 2 {
            return Err(SoftchatError::InvalidPersistenceResult);
        }
        Ok(result)
    }

    fn mutate_existing_conversation<P>(
        &mut self,
        sql: &str,
        params: P,
    ) -> Result<AccountMutationResult, SoftchatError>
    where
        P: rusqlite::Params,
    {
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let changed = transaction.execute(sql, params).map_err(map_sqlite_error)?;
        if changed != 1 {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        Ok(AccountMutationResult { revision })
    }

    fn product_media_by_command(
        &self,
        command_id: &str,
    ) -> Result<Option<AccountMediaOperation>, SoftchatError> {
        let id = self
            .connection
            .query_row(
                "SELECT operation_id FROM media_operations WHERE command_id = ?1",
                [command_id],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?;
        id.as_deref()
            .map(|value| self.product_media_operation(value))
            .transpose()
            .map(Option::flatten)
    }

    fn product_media_operation_kind(
        &self,
        operation_id: &str,
    ) -> Result<ProductMediaOperationKind, SoftchatError> {
        let value = self
            .connection
            .query_row(
                "SELECT operation_kind FROM media_operations WHERE operation_id = ?1",
                [operation_id],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?
            .ok_or(SoftchatError::InvalidMediaOperation)?;
        ProductMediaOperationKind::from_str(&value)
    }

    pub(crate) fn product_media_operation(
        &self,
        operation_id: &str,
    ) -> Result<Option<AccountMediaOperation>, SoftchatError> {
        self.connection
            .query_row(
                "SELECT command_id, COALESCE(conversation_id, ''),
                        COALESCE(source_message_id, ''), direction, protection,
                        state, source_fingerprint, attachment_json, byte_count,
                        attempt_count, last_error_category, updated_at
                 FROM media_operations WHERE operation_id = ?1",
                [operation_id],
                |row| {
                    Ok((
                        row.get::<_, String>(0)?,
                        row.get::<_, String>(1)?,
                        row.get::<_, String>(2)?,
                        row.get::<_, String>(3)?,
                        row.get::<_, String>(4)?,
                        row.get::<_, String>(5)?,
                        row.get::<_, String>(6)?,
                        row.get::<_, Option<String>>(7)?,
                        row.get::<_, Option<i64>>(8)?,
                        row.get::<_, u32>(9)?,
                        row.get::<_, String>(10)?,
                        row.get::<_, i64>(11)?,
                    ))
                },
            )
            .optional()
            .map_err(map_sqlite_error)?
            .map(
                |(
                    command_id,
                    conversation_id,
                    source_message_id,
                    direction,
                    protection,
                    state,
                    source_fingerprint,
                    attachment_json,
                    byte_count,
                    attempt_count,
                    last_error_category,
                    updated_at,
                )| {
                    let attachment = attachment_json
                        .map(|value| {
                            serde_json::from_str(&value)
                                .map_err(|_| SoftchatError::InvalidPersistenceResult)
                        })
                        .transpose()?;
                    Ok(AccountMediaOperation {
                        id: operation_id.to_owned(),
                        command_id,
                        conversation_id,
                        source_message_id,
                        direction,
                        protection: media_protection_from_name(&protection)?,
                        state: media_state_from_name(&state)?,
                        source_fingerprint,
                        attachment,
                        byte_count: byte_count
                            .map(u64::try_from)
                            .transpose()
                            .map_err(|_| SoftchatError::InvalidPersistenceResult)?,
                        attempt_count,
                        last_error_category,
                        updated_at,
                    })
                },
            )
            .transpose()
    }
}

fn product_message_from_row(
    row: ProductMessageRow,
    reactions: Vec<MessageReactionView>,
) -> Result<AccountMessageView, SoftchatError> {
    let rumor = serde_json::from_str::<RumorEvent>(&row.rumor_json)
        .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
    let parsed = classify_chat_message(rumor)?;
    let edit = match (row.edit_id, row.edit_rumor_json) {
        (Some(_), Some(json)) => {
            let rumor = serde_json::from_str::<RumorEvent>(&json)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            Some(classify_edit(rumor)?)
        }
        (None, None) => None,
        _ => return Err(SoftchatError::InvalidPersistenceResult),
    };
    let (text, emoji_tags, edited) = edit.map_or_else(
        || (parsed.rumor.content.clone(), parsed.emoji_tags, false),
        |value| (value.content, value.emoji_tags, true),
    );
    Ok(AccountMessageView {
        id: row.id,
        conversation_id: row.conversation_id,
        author_public_key: row.author_public_key,
        created_at: row.created_at,
        text,
        attachments: parsed.attachments,
        emoji_tags,
        relation: parsed.relation_kind,
        related_message_id: parsed.related_event_id,
        forwarded_author_public_key: parsed.original_author,
        edited,
        deleted: row.deleted,
        reply_count: row.reply_count,
        reactions,
    })
}

fn product_draft(
    draft_json: Option<String>,
    draft_reply_to: Option<String>,
    draft_updated_at: Option<i64>,
) -> Result<Option<AccountDraft>, SoftchatError> {
    match (draft_json, draft_updated_at) {
        (Some(value), Some(updated_at)) => {
            let stored: StoredDraft = serde_json::from_str(&value)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            Ok(Some(AccountDraft {
                text: stored.text,
                emoji_tags: stored.emoji_tags,
                spans: stored.spans,
                reply_to_message_id: draft_reply_to.unwrap_or_default(),
                attachment_operation_ids: stored.attachment_operation_ids,
                updated_at,
            }))
        }
        (None, None) => Ok(None),
        _ => Err(SoftchatError::InvalidPersistenceResult),
    }
}

fn parse_public_key_csv(value: Option<String>) -> Result<Vec<String>, SoftchatError> {
    let keys = value
        .filter(|csv| !csv.is_empty())
        .ok_or(SoftchatError::InvalidPersistenceResult)?
        .split(',')
        .map(|key| {
            NostrPublicKey::from_hex(key)
                .map(|value| value.to_hex())
                .map_err(|_| SoftchatError::InvalidPersistenceResult)
        })
        .collect::<Result<Vec<_>, _>>()?;
    if keys.len() < 2 {
        return Err(SoftchatError::InvalidPersistenceResult);
    }
    Ok(keys)
}

fn canonical_members(
    own_public_key: &str,
    member_public_keys: Vec<String>,
) -> Result<Vec<String>, SoftchatError> {
    let own = NostrPublicKey::from_hex(own_public_key)?.to_hex();
    let mut members = member_public_keys
        .into_iter()
        .map(|value| NostrPublicKey::from_hex(&value).map(|key| key.to_hex()))
        .collect::<Result<Vec<_>, _>>()?;
    members.push(own);
    members.sort();
    members.dedup();
    if members.len() < 2 || members.len() > crate::MAX_CHAT_PARTICIPANTS + 1 {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    Ok(members)
}

fn validate_limit(limit: u32) -> Result<(), SoftchatError> {
    if limit == 0 || limit > MAX_PRODUCT_QUERY_PAGE {
        Err(SoftchatError::InvalidAccountOperation)
    } else {
        Ok(())
    }
}

fn validate_timestamp(timestamp: i64) -> Result<(), SoftchatError> {
    if timestamp < 0
        || u64::try_from(timestamp).map_err(|_| SoftchatError::InvalidAccountOperation)?
            > crate::MAX_PORTABLE_TIMESTAMP_SECONDS
    {
        Err(SoftchatError::InvalidAccountOperation)
    } else {
        Ok(())
    }
}

pub(crate) fn validate_conversation_id(value: &str) -> Result<(), SoftchatError> {
    if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
        Err(SoftchatError::InvalidAccountOperation)
    } else {
        Ok(())
    }
}

pub(crate) fn validate_command_id(value: &str) -> Result<(), SoftchatError> {
    if value.is_empty() || value.len() > 256 {
        Err(SoftchatError::InvalidAccountOperation)
    } else {
        Ok(())
    }
}

pub(crate) fn command_hash<T: Serialize>(value: &T) -> Result<String, SoftchatError> {
    let encoded = serde_json::to_vec(value).map_err(|_| SoftchatError::InternalFailure)?;
    Ok(hex::encode(Sha256::digest(encoded)))
}

fn escaped_fts_query(query: &str) -> Result<String, SoftchatError> {
    let terms = query
        .split_whitespace()
        .filter(|term| !term.is_empty())
        .map(|term| format!("\"{}\"", term.replace('"', "\"\"")))
        .collect::<Vec<_>>();
    if terms.is_empty() || terms.len() > 32 {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    Ok(terms.join(" AND "))
}

fn validate_product_draft(draft: &AccountDraft) -> Result<(), SoftchatError> {
    if draft.text.len() > MAX_PRODUCT_MESSAGE_BYTES
        || draft.emoji_tags.len() > crate::MAX_CHAT_EMOJI_TAGS
        || draft.spans.len() > MAX_PRODUCT_DRAFT_SPANS
        || draft.attachment_operation_ids.len() > MAX_PRODUCT_DRAFT_ATTACHMENTS
    {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    let mut boundaries = BTreeSet::new();
    let mut offset = 0_u32;
    boundaries.insert(offset);
    for character in draft.text.chars() {
        offset = offset
            .checked_add(character.len_utf16() as u32)
            .ok_or(SoftchatError::InvalidAccountOperation)?;
        boundaries.insert(offset);
    }
    let mut semantic_ranges = Vec::new();
    let mut unique = BTreeSet::new();
    for span in &draft.spans {
        if span.start_utf16 >= span.end_utf16
            || !boundaries.contains(&span.start_utf16)
            || !boundaries.contains(&span.end_utf16)
            || !unique.insert((
                span.start_utf16,
                span.end_utf16,
                span.kind,
                span.value.clone(),
            ))
        {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        match span.kind {
            DraftSpanKind::Bold | DraftSpanKind::Italic | DraftSpanKind::Code => {
                if !span.value.is_empty() {
                    return Err(SoftchatError::InvalidAccountOperation);
                }
            }
            DraftSpanKind::Mention => {
                NostrPublicKey::from_hex(&span.value)?;
                semantic_ranges.push((span.start_utf16, span.end_utf16));
            }
            DraftSpanKind::Link => {
                let url =
                    Url::parse(&span.value).map_err(|_| SoftchatError::InvalidAccountOperation)?;
                if url.scheme() != "https" || url.host_str().is_none() {
                    return Err(SoftchatError::InvalidAccountOperation);
                }
                semantic_ranges.push((span.start_utf16, span.end_utf16));
            }
            DraftSpanKind::CustomEmoji => {
                if span.value.is_empty() || span.value.len() > 128 || span.value.contains(':') {
                    return Err(SoftchatError::InvalidAccountOperation);
                }
                semantic_ranges.push((span.start_utf16, span.end_utf16));
            }
        }
    }
    semantic_ranges.sort_unstable();
    if semantic_ranges
        .windows(2)
        .any(|ranges| ranges[1].0 < ranges[0].1)
    {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    let mut attachment_ids = BTreeSet::new();
    for operation_id in &draft.attachment_operation_ids {
        if operation_id.is_empty()
            || operation_id.len() > 256
            || operation_id.chars().any(char::is_control)
            || !attachment_ids.insert(operation_id)
        {
            return Err(SoftchatError::InvalidAccountOperation);
        }
    }
    Ok(())
}

pub(crate) fn product_draft_match(draft: AccountDraft) -> Result<ProductDraftMatch, SoftchatError> {
    validate_product_draft(&draft)?;
    if !draft.reply_to_message_id.is_empty() {
        NostrEventId::from_hex(&draft.reply_to_message_id)?;
    }
    let draft_json = serde_json::to_string(&StoredDraft {
        text: draft.text,
        emoji_tags: draft.emoji_tags,
        spans: draft.spans,
        attachment_operation_ids: draft.attachment_operation_ids,
    })
    .map_err(|_| SoftchatError::InternalFailure)?;
    Ok(ProductDraftMatch {
        draft_json,
        reply_to_message_id: draft.reply_to_message_id,
    })
}

pub(crate) fn parse_read_state_snapshot(
    value: &str,
    event_created_at: i64,
    event_id: &str,
) -> Result<ParsedReadStateSnapshot, SoftchatError> {
    if value.is_empty() || value.len() > MAX_ACCOUNT_READ_STATE_BYTES {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    validate_timestamp(event_created_at)?;
    NostrEventId::from_hex(event_id)?;
    let snapshot: StoredReadStateSnapshot =
        serde_json::from_str(value).map_err(|_| SoftchatError::InvalidAccountOperation)?;
    if snapshot.schema_version == 0
        || snapshot.schema_version > ACCOUNT_READ_STATE_SCHEMA_VERSION
        || snapshot.chats.len() > MAX_ACCOUNT_READ_STATE_ENTRIES
    {
        return Err(SoftchatError::InvalidAccountOperation);
    }

    let mut entries: BTreeMap<String, ParsedReadStateEntry> = BTreeMap::new();
    for entry in snapshot.chats {
        let conversation_id = if entry.conversation_id.is_empty() {
            let members = entry
                .chat_key
                .split(',')
                .filter(|value| !value.is_empty())
                .map(NostrPublicKey::from_hex)
                .collect::<Result<Vec<_>, _>>()?;
            if members.is_empty() {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            conversation_id(members.into_iter().map(|member| member.to_hex()).collect())?
        } else {
            validate_conversation_id(&entry.conversation_id)?;
            entry.conversation_id.clone()
        };
        let read_at = match (entry.read_at_unix, entry.read_at) {
            (Some(value), _) => Some(value),
            (None, Some(Some(value))) if value.is_finite() => {
                let unix = value + SWIFT_REFERENCE_DATE_UNIX_SECONDS as f64;
                if unix < 0.0 || unix > crate::MAX_PORTABLE_TIMESTAMP_SECONDS as f64 {
                    return Err(SoftchatError::InvalidAccountOperation);
                }
                Some(unix.floor() as i64)
            }
            (None, Some(Some(_))) => return Err(SoftchatError::InvalidAccountOperation),
            (None, Some(None) | None) => None,
        };
        if let Some(value) = read_at {
            validate_timestamp(value)?;
        }
        let forced_unread = entry
            .forced_unread
            .unwrap_or_else(|| entry.read_at_unix.is_none() && entry.read_at.flatten().is_none());
        let updated_at = entry.updated_at.unwrap_or(event_created_at);
        validate_timestamp(updated_at)?;
        let update_id = if entry.update_id.is_empty() {
            event_id.to_owned()
        } else {
            validate_command_id(&entry.update_id)?;
            entry.update_id
        };
        let boundary = read_at.map(|created_at| {
            let message_id = if entry.read_message_id.is_empty() {
                minimum_event_id()
            } else {
                entry.read_message_id
            };
            NostrEventId::from_hex(&message_id)?;
            Ok(MessageCursor {
                created_at,
                message_id,
            })
        });
        let boundary = boundary.transpose()?;
        let unknown_json = serde_json::to_string(&entry.unknown)
            .map_err(|_| SoftchatError::InvalidAccountOperation)?;
        let parsed = ParsedReadStateEntry {
            state: AccountReadState {
                conversation_id: conversation_id.clone(),
                boundary,
                forced_unread,
                updated_at,
                update_id,
            },
            unknown_json,
        };
        match entries.get_mut(&conversation_id) {
            Some(existing)
                if read_state_order(&parsed.state) > read_state_order(&existing.state) =>
            {
                *existing = parsed;
            }
            Some(existing)
                if read_state_order(&parsed.state) == read_state_order(&existing.state) =>
            {
                if !parsed.state.forced_unread {
                    existing.state.boundary = match (
                        existing.state.boundary.clone(),
                        parsed.state.boundary.clone(),
                    ) {
                        (Some(left), Some(right)) => Some(advance_read_boundary(Some(left), right)),
                        (None, right) => right,
                        (left, None) => left,
                    };
                    existing.state.forced_unread = false;
                }
            }
            Some(_) => {}
            None => {
                entries.insert(conversation_id, parsed);
            }
        }
    }
    let unknown_json = serde_json::to_string(&snapshot.unknown)
        .map_err(|_| SoftchatError::InvalidAccountOperation)?;
    Ok(ParsedReadStateSnapshot {
        entries: entries.into_values().collect(),
        unknown_json,
    })
}

fn default_read_state_schema() -> u32 {
    ACCOUNT_READ_STATE_SCHEMA_VERSION
}

fn minimum_event_id() -> String {
    "0".repeat(64)
}

fn read_state_order(state: &AccountReadState) -> (i64, &str) {
    (state.updated_at, state.update_id.as_str())
}

fn advance_read_boundary(
    current: Option<MessageCursor>,
    candidate: MessageCursor,
) -> MessageCursor {
    let Some(current) = current else {
        return candidate;
    };
    if candidate.created_at > current.created_at
        || (candidate.created_at == current.created_at && candidate.message_id < current.message_id)
    {
        candidate
    } else {
        current
    }
}

pub(crate) fn settings_from_json(value: &str) -> Result<AccountSettings, SoftchatError> {
    if value.len() > MAX_ACCOUNT_SETTINGS_BYTES {
        return Err(SoftchatError::InvalidAccountSettings);
    }
    let mut object: Map<String, Value> =
        serde_json::from_str(value).map_err(|_| SoftchatError::InvalidAccountSettings)?;
    apply_legacy_ios_settings_aliases(&mut object)?;
    let schema_version =
        take_u32(&mut object, "schemaVersion")?.unwrap_or(ACCOUNT_SETTINGS_SCHEMA_VERSION);
    if schema_version == 0 || schema_version > ACCOUNT_SETTINGS_SCHEMA_VERSION {
        return Err(SoftchatError::InvalidAccountSettings);
    }
    let quick_reaction = take_string(&mut object, "quickReaction")?;
    if quick_reaction.as_ref().is_some_and(|value| {
        let custom = value.starts_with(':') && value.ends_with(':');
        if custom {
            value.len() > 130
        } else {
            value.chars().count() > 64
        }
    }) {
        return Err(SoftchatError::InvalidAccountSettings);
    }
    let typing_indicators_enabled =
        take_bool(&mut object, "typingIndicatorsEnabled")?.unwrap_or(true);
    let notification_content_privacy = take_string(&mut object, "notificationContentPrivacy")?
        .unwrap_or_else(|| "full".to_owned());
    if !matches!(
        notification_content_privacy.as_str(),
        "full" | "sender" | "hidden"
    ) {
        return Err(SoftchatError::InvalidAccountSettings);
    }
    let media_service = take_string(&mut object, "mediaService")?;
    if media_service.as_ref().is_some_and(|value| {
        Url::parse(value)
            .ok()
            .is_none_or(|url| url.scheme() != "https")
    }) {
        return Err(SoftchatError::InvalidAccountSettings);
    }
    let image_upload_quality = quality(&mut object, "imageUploadQuality")?;
    let video_upload_quality = quality(&mut object, "videoUploadQuality")?;
    let preserve_hdr = take_bool(&mut object, "preserveHdr")?.unwrap_or(true);
    let transcription_locale = take_string(&mut object, "transcriptionLocale")?;
    if transcription_locale
        .as_ref()
        .is_some_and(|value| value.len() > 128)
    {
        return Err(SoftchatError::InvalidAccountSettings);
    }
    let domain_rules = object
        .remove("domainReplacementRules")
        .unwrap_or_else(|| Value::Array(Vec::new()));
    let domain_rules_array = domain_rules
        .as_array()
        .ok_or(SoftchatError::InvalidAccountSettings)?;
    if domain_rules_array.len() > 128 {
        return Err(SoftchatError::InvalidAccountSettings);
    }
    let domain_replacement_rules_json =
        serde_json::to_string(&domain_rules).map_err(|_| SoftchatError::InvalidAccountSettings)?;
    let chat_wallpaper = take_string(&mut object, "chatWallpaper")?;
    if chat_wallpaper
        .as_ref()
        .is_some_and(|value| value.is_empty() || value.len() > 256)
    {
        return Err(SoftchatError::InvalidAccountSettings);
    }
    let custom_emoji_references =
        take_custom_emoji_references(&mut object, "customEmojiReferences")?;
    let saved_sticker_references = take_string_array(&mut object, "savedStickerReferences", 512)?;
    let seasonal_appearance = take_string(&mut object, "seasonalAppearance")?;
    let unknown_json =
        serde_json::to_string(&object).map_err(|_| SoftchatError::InvalidAccountSettings)?;
    Ok(AccountSettings {
        schema_version,
        quick_reaction,
        typing_indicators_enabled,
        notification_content_privacy,
        media_service,
        image_upload_quality,
        video_upload_quality,
        preserve_hdr,
        transcription_locale,
        domain_replacement_rules_json,
        chat_wallpaper,
        custom_emoji_references,
        saved_sticker_references,
        seasonal_appearance,
        unknown_json,
    })
}

pub(crate) fn profile_update_from_patch(
    current: &AccountProfile,
    patch_json: &str,
) -> Result<(ProfileUpdateInput, Value), SoftchatError> {
    if patch_json.len() > crate::MAX_APP_DATA_JSON_BYTES {
        return Err(SoftchatError::InvalidUserMetadata);
    }
    let mut patch: Map<String, Value> =
        serde_json::from_str(patch_json).map_err(|_| SoftchatError::InvalidUserMetadata)?;
    let name = apply_optional_string_patch(&mut patch, "name", current.name.clone())?;
    let display_name =
        apply_optional_string_patch(&mut patch, "displayName", current.display_name.clone())?;
    let about = apply_optional_string_patch(&mut patch, "about", current.about.clone())?;
    let picture = apply_optional_string_patch(&mut patch, "picture", current.picture.clone())?;
    let website = apply_optional_string_patch(&mut patch, "website", current.website.clone())?;
    let banner = apply_optional_string_patch(&mut patch, "banner", current.banner.clone())?;
    let bot = apply_optional_bool_patch(&mut patch, "bot", current.bot)?;
    if !patch.is_empty() {
        return Err(SoftchatError::InvalidUserMetadata);
    }
    let canonical_patch: Value =
        serde_json::from_str(patch_json).map_err(|_| SoftchatError::InvalidUserMetadata)?;
    Ok((
        ProfileUpdateInput {
            name,
            display_name,
            about,
            picture,
            website,
            banner,
            bot,
            unknown_json: current.unknown_json.clone(),
        },
        canonical_patch,
    ))
}

pub(crate) fn canonical_settings_json(value: &str) -> Result<String, SoftchatError> {
    let settings = settings_from_json(value)?;
    serde_json::to_string(&settings_to_object(&settings)?)
        .map_err(|_| SoftchatError::InvalidAccountSettings)
}

pub(crate) fn settings_wire_json(value: &str) -> Result<String, SoftchatError> {
    let settings = settings_from_json(value)?;
    let mut object = settings_to_object(&settings)?;
    let video_preset = match settings.video_upload_quality.as_str() {
        "low" => "AVAssetExportPresetLowQuality",
        "balanced" => "AVAssetExportPreset960x540",
        "high" => "AVAssetExportPresetHighestQuality",
        _ => return Err(SoftchatError::InvalidAccountSettings),
    };
    object.insert("videoUploadPreset".to_owned(), json!(video_preset));
    object.insert(
        "preserveHDROnUpload".to_owned(),
        json!(settings.preserve_hdr),
    );
    object.insert(
        "christmasTheme".to_owned(),
        json!(settings.seasonal_appearance.as_deref() == Some("christmas")),
    );
    object.insert(
        "hideNotificationContent".to_owned(),
        json!(settings.notification_content_privacy == "hidden"),
    );
    if let Some(reaction) = &settings.quick_reaction {
        let swift_emoji = swift_quick_reaction_json(reaction, &settings.custom_emoji_references)?;
        object.insert(
            "quickReactionEmoji".to_owned(),
            Value::String(STANDARD.encode(swift_emoji.as_bytes())),
        );
    } else {
        object.remove("quickReactionEmoji");
    }
    insert_optional(
        &mut object,
        "chatWallpaperImage",
        settings.chat_wallpaper.clone(),
    );
    object.insert(
        "customEmojis".to_owned(),
        Value::Array(
            settings
                .custom_emoji_references
                .iter()
                .map(|emoji| json!({"id": emoji.shortcode, "url": emoji.url}))
                .collect(),
        ),
    );
    let sticker_urls = settings
        .saved_sticker_references
        .iter()
        .filter(|value| {
            Url::parse(value)
                .ok()
                .is_some_and(|url| matches!(url.scheme(), "https" | "http"))
        })
        .cloned()
        .map(Value::String)
        .collect();
    object.insert("savedStickerURLs".to_owned(), Value::Array(sticker_urls));
    let json = serde_json::to_string(&object).map_err(|_| SoftchatError::InvalidAccountSettings)?;
    if json.len() > MAX_ACCOUNT_SETTINGS_BYTES {
        return Err(SoftchatError::InvalidAccountSettings);
    }
    Ok(json)
}

fn settings_to_object(settings: &AccountSettings) -> Result<Map<String, Value>, SoftchatError> {
    let mut object: Map<String, Value> = serde_json::from_str(&settings.unknown_json)
        .map_err(|_| SoftchatError::InvalidAccountSettings)?;
    object.insert("schemaVersion".to_owned(), json!(settings.schema_version));
    insert_optional(
        &mut object,
        "quickReaction",
        settings.quick_reaction.clone(),
    );
    object.insert(
        "typingIndicatorsEnabled".to_owned(),
        json!(settings.typing_indicators_enabled),
    );
    object.insert(
        "notificationContentPrivacy".to_owned(),
        json!(settings.notification_content_privacy),
    );
    insert_optional(&mut object, "mediaService", settings.media_service.clone());
    object.insert(
        "imageUploadQuality".to_owned(),
        json!(settings.image_upload_quality),
    );
    object.insert(
        "videoUploadQuality".to_owned(),
        json!(settings.video_upload_quality),
    );
    object.insert("preserveHdr".to_owned(), json!(settings.preserve_hdr));
    insert_optional(
        &mut object,
        "transcriptionLocale",
        settings.transcription_locale.clone(),
    );
    object.insert(
        "domainReplacementRules".to_owned(),
        serde_json::from_str(&settings.domain_replacement_rules_json)
            .map_err(|_| SoftchatError::InvalidAccountSettings)?,
    );
    insert_optional(
        &mut object,
        "chatWallpaper",
        settings.chat_wallpaper.clone(),
    );
    object.insert(
        "customEmojiReferences".to_owned(),
        serde_json::to_value(&settings.custom_emoji_references)
            .map_err(|_| SoftchatError::InvalidAccountSettings)?,
    );
    object.insert(
        "savedStickerReferences".to_owned(),
        json!(settings.saved_sticker_references),
    );
    insert_optional(
        &mut object,
        "seasonalAppearance",
        settings.seasonal_appearance.clone(),
    );
    Ok(object)
}

fn apply_legacy_ios_settings_aliases(object: &mut Map<String, Value>) -> Result<(), SoftchatError> {
    if !object.contains_key("quickReaction") {
        if let Some(value) = object.remove("quickReactionEmoji")
            && !value.is_null()
        {
            let encoded = value
                .as_str()
                .ok_or(SoftchatError::InvalidAccountSettings)?;
            let decoded = STANDARD
                .decode(encoded)
                .map_err(|_| SoftchatError::InvalidAccountSettings)?;
            if decoded.len() > 16 * 1024 {
                return Err(SoftchatError::InvalidAccountSettings);
            }
            let reaction = parse_swift_quick_reaction(&decoded)?;
            object.insert("quickReaction".to_owned(), Value::String(reaction));
        }
    } else {
        object.remove("quickReactionEmoji");
    }
    if !object.contains_key("chatWallpaper") {
        if let Some(value) = object.remove("chatWallpaperImage") {
            object.insert("chatWallpaper".to_owned(), value);
        }
    } else {
        object.remove("chatWallpaperImage");
    }
    if !object.contains_key("customEmojiReferences") {
        if let Some(value) = object.remove("customEmojis") {
            let values = value
                .as_array()
                .ok_or(SoftchatError::InvalidAccountSettings)?;
            let mut converted = Vec::with_capacity(values.len());
            for value in values {
                let value = value
                    .as_object()
                    .ok_or(SoftchatError::InvalidAccountSettings)?;
                let shortcode = value
                    .get("id")
                    .and_then(Value::as_str)
                    .ok_or(SoftchatError::InvalidAccountSettings)?;
                let url = value
                    .get("url")
                    .and_then(Value::as_str)
                    .ok_or(SoftchatError::InvalidAccountSettings)?;
                converted.push(json!({"shortcode": shortcode, "url": url}));
            }
            object.insert("customEmojiReferences".to_owned(), Value::Array(converted));
        }
    } else {
        object.remove("customEmojis");
    }
    if !object.contains_key("preserveHdr") {
        if let Some(value) = object.remove("preserveHDROnUpload") {
            object.insert("preserveHdr".to_owned(), value);
        }
    } else {
        object.remove("preserveHDROnUpload");
    }
    if !object.contains_key("notificationContentPrivacy") {
        if let Some(value) = object.remove("hideNotificationContent") {
            let hidden = value
                .as_bool()
                .ok_or(SoftchatError::InvalidAccountSettings)?;
            object.insert(
                "notificationContentPrivacy".to_owned(),
                json!(if hidden { "hidden" } else { "full" }),
            );
        }
    } else {
        object.remove("hideNotificationContent");
    }
    if !object.contains_key("seasonalAppearance") {
        if let Some(value) = object.remove("christmasTheme") {
            let enabled = value
                .as_bool()
                .ok_or(SoftchatError::InvalidAccountSettings)?;
            if enabled {
                object.insert("seasonalAppearance".to_owned(), json!("christmas"));
            }
        }
    } else {
        object.remove("christmasTheme");
    }
    if !object.contains_key("videoUploadQuality") {
        if let Some(value) = object.remove("videoUploadPreset") {
            let preset = value
                .as_str()
                .ok_or(SoftchatError::InvalidAccountSettings)?;
            let quality = if preset.contains("LowQuality") {
                "low"
            } else if preset.contains("HighestQuality") || preset.contains("HEVC1920") {
                "high"
            } else {
                "balanced"
            };
            object.insert("videoUploadQuality".to_owned(), json!(quality));
        }
    } else {
        object.remove("videoUploadPreset");
    }
    if !object.contains_key("savedStickerReferences") {
        if let Some(value) = object.remove("savedStickerURLs") {
            object.insert("savedStickerReferences".to_owned(), value);
        }
    } else {
        object.remove("savedStickerURLs");
    }
    Ok(())
}

fn parse_swift_quick_reaction(value: &[u8]) -> Result<String, SoftchatError> {
    if let Ok(text) = std::str::from_utf8(value)
        && !text.is_empty()
        && !text.trim_start().starts_with('{')
    {
        return Ok(text.to_owned());
    }
    let value: Value =
        serde_json::from_slice(value).map_err(|_| SoftchatError::InvalidAccountSettings)?;
    if let Some(emoji) = value
        .get("unicode")
        .and_then(|value| value.get("_0"))
        .and_then(|value| value.get("emoji"))
        .and_then(Value::as_str)
    {
        if emoji.is_empty() || emoji.chars().count() > 64 {
            return Err(SoftchatError::InvalidAccountSettings);
        }
        return Ok(emoji.to_owned());
    }
    let shortcode = value
        .get("custom")
        .and_then(|value| value.get("_0"))
        .and_then(|value| value.get("id"))
        .and_then(Value::as_str)
        .ok_or(SoftchatError::InvalidAccountSettings)?;
    if shortcode.is_empty() || shortcode.len() > 128 || shortcode.contains(':') {
        return Err(SoftchatError::InvalidAccountSettings);
    }
    Ok(format!(":{shortcode}:"))
}

fn swift_quick_reaction_json(
    reaction: &str,
    custom_emojis: &[CustomEmojiReference],
) -> Result<String, SoftchatError> {
    let value = if let Some(shortcode) = reaction
        .strip_prefix(':')
        .and_then(|value| value.strip_suffix(':'))
        .filter(|value| !value.is_empty() && !value.contains(':'))
    {
        let custom = custom_emojis
            .iter()
            .find(|emoji| emoji.shortcode.eq_ignore_ascii_case(shortcode))
            .ok_or(SoftchatError::InvalidAccountSettings)?;
        json!({"custom":{"_0":{"id": custom.shortcode, "url": custom.url}}})
    } else {
        json!({"unicode":{"_0":{"emoji":reaction}}})
    };
    serde_json::to_string(&value).map_err(|_| SoftchatError::InvalidAccountSettings)
}

fn insert_optional(object: &mut Map<String, Value>, key: &str, value: Option<String>) {
    if let Some(value) = value {
        object.insert(key.to_owned(), Value::String(value));
    } else {
        object.remove(key);
    }
}

fn apply_optional_string_patch(
    patch: &mut Map<String, Value>,
    key: &str,
    current: Option<String>,
) -> Result<Option<String>, SoftchatError> {
    match patch.remove(key) {
        None => Ok(current),
        Some(Value::Null) => Ok(None),
        Some(Value::String(value)) => Ok(Some(value)),
        Some(_) => Err(SoftchatError::InvalidUserMetadata),
    }
}

fn apply_optional_bool_patch(
    patch: &mut Map<String, Value>,
    key: &str,
    current: Option<bool>,
) -> Result<Option<bool>, SoftchatError> {
    match patch.remove(key) {
        None => Ok(current),
        Some(Value::Null) => Ok(None),
        Some(Value::Bool(value)) => Ok(Some(value)),
        Some(_) => Err(SoftchatError::InvalidUserMetadata),
    }
}

fn take_string(
    object: &mut Map<String, Value>,
    key: &str,
) -> Result<Option<String>, SoftchatError> {
    match object.remove(key) {
        Some(Value::String(value)) => Ok(Some(value)),
        Some(Value::Null) | None => Ok(None),
        Some(_) => Err(SoftchatError::InvalidAccountSettings),
    }
}

fn take_custom_emoji_references(
    object: &mut Map<String, Value>,
    key: &str,
) -> Result<Vec<CustomEmojiReference>, SoftchatError> {
    let value = object
        .remove(key)
        .unwrap_or_else(|| Value::Array(Vec::new()));
    let values = value
        .as_array()
        .ok_or(SoftchatError::InvalidAccountSettings)?;
    if values.len() > 512 {
        return Err(SoftchatError::InvalidAccountSettings);
    }
    let mut result = Vec::with_capacity(values.len());
    let mut shortcodes = BTreeSet::new();
    for value in values {
        let emoji: CustomEmojiReference = serde_json::from_value(value.clone())
            .map_err(|_| SoftchatError::InvalidAccountSettings)?;
        if emoji.shortcode.is_empty()
            || emoji.shortcode.len() > 128
            || emoji.shortcode.contains(':')
            || emoji.url.len() > 8_192
            || Url::parse(&emoji.url)
                .ok()
                .is_none_or(|url| url.scheme() != "https")
            || !shortcodes.insert(emoji.shortcode.to_lowercase())
        {
            return Err(SoftchatError::InvalidAccountSettings);
        }
        result.push(emoji);
    }
    result.sort_by(|left, right| {
        left.shortcode
            .to_lowercase()
            .cmp(&right.shortcode.to_lowercase())
            .then_with(|| left.url.cmp(&right.url))
    });
    Ok(result)
}

fn take_bool(object: &mut Map<String, Value>, key: &str) -> Result<Option<bool>, SoftchatError> {
    match object.remove(key) {
        Some(Value::Bool(value)) => Ok(Some(value)),
        Some(Value::Null) | None => Ok(None),
        Some(_) => Err(SoftchatError::InvalidAccountSettings),
    }
}

fn take_u32(object: &mut Map<String, Value>, key: &str) -> Result<Option<u32>, SoftchatError> {
    match object.remove(key) {
        Some(Value::Number(value)) => value
            .as_u64()
            .and_then(|value| u32::try_from(value).ok())
            .map(Some)
            .ok_or(SoftchatError::InvalidAccountSettings),
        Some(Value::Null) | None => Ok(None),
        Some(_) => Err(SoftchatError::InvalidAccountSettings),
    }
}

fn take_string_array(
    object: &mut Map<String, Value>,
    key: &str,
    max: usize,
) -> Result<Vec<String>, SoftchatError> {
    match object.remove(key) {
        Some(Value::Array(values)) => {
            if values.len() > max {
                return Err(SoftchatError::InvalidAccountSettings);
            }
            let mut result = Vec::with_capacity(values.len());
            let mut unique = BTreeSet::new();
            for value in values {
                let Value::String(value) = value else {
                    return Err(SoftchatError::InvalidAccountSettings);
                };
                if value.len() > 512 || !unique.insert(value.clone()) {
                    return Err(SoftchatError::InvalidAccountSettings);
                }
                result.push(value);
            }
            Ok(result)
        }
        Some(Value::Null) | None => Ok(Vec::new()),
        Some(_) => Err(SoftchatError::InvalidAccountSettings),
    }
}

fn quality(object: &mut Map<String, Value>, key: &str) -> Result<String, SoftchatError> {
    let value = take_string(object, key)?.unwrap_or_else(|| "balanced".to_owned());
    if matches!(value.as_str(), "low" | "balanced" | "high") {
        Ok(value)
    } else {
        Err(SoftchatError::InvalidAccountSettings)
    }
}

fn stable_media_operation_id(
    account_id: &str,
    command_id: &str,
    direction: &str,
    source_fingerprint: &str,
) -> String {
    let mut digest = Sha256::new();
    digest.update(account_id.as_bytes());
    digest.update([0]);
    digest.update(command_id.as_bytes());
    digest.update([0]);
    digest.update(direction.as_bytes());
    digest.update([0]);
    digest.update(source_fingerprint.as_bytes());
    hex::encode(digest.finalize())
}

fn stable_pending_attachment_command_id(
    command_id: &str,
    index: usize,
    source_fingerprint: &str,
) -> String {
    let mut digest = Sha256::new();
    digest.update(b"pending-media-attachment");
    digest.update([0]);
    digest.update(command_id.as_bytes());
    digest.update([0]);
    digest.update(index.to_le_bytes());
    digest.update([0]);
    digest.update(source_fingerprint.as_bytes());
    hex::encode(digest.finalize())
}

fn stable_asset_upload_command_id(
    command_id: &str,
    target: AssetReplacementTarget,
    conversation_id: &str,
    source_fingerprint: &str,
) -> String {
    let mut digest = Sha256::new();
    digest.update(b"pending-asset-upload");
    digest.update([0]);
    digest.update(command_id.as_bytes());
    digest.update([0]);
    digest.update(asset_target_name(target).as_bytes());
    digest.update([0]);
    digest.update(conversation_id.as_bytes());
    digest.update([0]);
    digest.update(source_fingerprint.as_bytes());
    hex::encode(digest.finalize())
}

fn validate_media_operation_id(value: &str) -> Result<(), SoftchatError> {
    if value.len() != 64
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_digit() || matches!(byte, b'a'..=b'f'))
    {
        return Err(SoftchatError::InvalidMediaOperation);
    }
    Ok(())
}

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

fn validate_media_error_category(value: &str) -> Result<(), SoftchatError> {
    if value.is_empty()
        || value.len() > 128
        || !value
            .bytes()
            .all(|byte| byte.is_ascii_lowercase() || byte.is_ascii_digit() || byte == b'_')
    {
        return Err(SoftchatError::InvalidMediaOperation);
    }
    Ok(())
}

fn escaped_like_pattern(query: &str) -> Result<String, SoftchatError> {
    if query.is_empty() || query.len() > MAX_PRODUCT_SEARCH_BYTES {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    let mut pattern = String::with_capacity(query.len() + 2);
    pattern.push('%');
    // SQLite lower()/LIKE fold ASCII only. Preserve every other code point so
    // an exact Unicode query cannot be changed into a pattern that never matches.
    for character in query
        .chars()
        .map(|character| character.to_ascii_lowercase())
    {
        if matches!(character, '\\' | '%' | '_') {
            pattern.push('\\');
        }
        pattern.push(character);
    }
    pattern.push('%');
    Ok(pattern)
}

fn media_state_name(value: AccountMediaOperationState) -> &'static str {
    match value {
        AccountMediaOperationState::Prepared => "prepared",
        AccountMediaOperationState::Running => "running",
        AccountMediaOperationState::Completed => "completed",
        AccountMediaOperationState::Retryable => "retryable",
        AccountMediaOperationState::Failed => "failed",
        AccountMediaOperationState::Cancelled => "cancelled",
    }
}

fn media_protection_name(value: AccountMediaProtection) -> &'static str {
    match value {
        AccountMediaProtection::Encrypted => "encrypted",
        AccountMediaProtection::Public => "public",
    }
}

fn media_protection_from_name(value: &str) -> Result<AccountMediaProtection, SoftchatError> {
    match value {
        "encrypted" => Ok(AccountMediaProtection::Encrypted),
        "public" => Ok(AccountMediaProtection::Public),
        _ => Err(SoftchatError::InvalidPersistenceResult),
    }
}

fn media_state_from_name(value: &str) -> Result<AccountMediaOperationState, SoftchatError> {
    match value {
        "prepared" => Ok(AccountMediaOperationState::Prepared),
        "running" => Ok(AccountMediaOperationState::Running),
        "completed" => Ok(AccountMediaOperationState::Completed),
        "retryable" => Ok(AccountMediaOperationState::Retryable),
        "failed" => Ok(AccountMediaOperationState::Failed),
        "cancelled" => Ok(AccountMediaOperationState::Cancelled),
        _ => Err(SoftchatError::InvalidPersistenceResult),
    }
}

fn asset_target_name(value: AssetReplacementTarget) -> &'static str {
    match value {
        AssetReplacementTarget::ProfilePicture => "profile_picture",
        AssetReplacementTarget::ProfileBanner => "profile_banner",
        AssetReplacementTarget::ConversationIcon => "conversation_icon",
    }
}

fn asset_target_from_name(value: &str) -> Result<AssetReplacementTarget, SoftchatError> {
    match value {
        "profile_picture" => Ok(AssetReplacementTarget::ProfilePicture),
        "profile_banner" => Ok(AssetReplacementTarget::ProfileBanner),
        "conversation_icon" => Ok(AssetReplacementTarget::ConversationIcon),
        _ => Err(SoftchatError::InvalidPersistenceResult),
    }
}

fn pending_media_message_state_from_name(
    value: &str,
) -> Result<PendingMediaMessageState, SoftchatError> {
    match value {
        "waiting_for_uploads" => Ok(PendingMediaMessageState::WaitingForUploads),
        "ready_to_publish" => Ok(PendingMediaMessageState::ReadyToPublish),
        "published" => Ok(PendingMediaMessageState::Published),
        "failed" => Ok(PendingMediaMessageState::Failed),
        "cancelled" => Ok(PendingMediaMessageState::Cancelled),
        _ => Err(SoftchatError::InvalidPersistenceResult),
    }
}

fn pending_asset_replacement_state_from_name(
    value: &str,
) -> Result<PendingAssetReplacementState, SoftchatError> {
    match value {
        "waiting_for_upload" => Ok(PendingAssetReplacementState::WaitingForUpload),
        "ready_to_publish" => Ok(PendingAssetReplacementState::ReadyToPublish),
        "published" => Ok(PendingAssetReplacementState::Published),
        "failed" => Ok(PendingAssetReplacementState::Failed),
        "cancelled" => Ok(PendingAssetReplacementState::Cancelled),
        _ => Err(SoftchatError::InvalidPersistenceResult),
    }
}

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

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

    #[test]
    fn equal_timestamp_pages_are_lossless() -> Result<(), SoftchatError> {
        let alice = LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let bob = LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let mut database = AccountDatabase::open(":memory:", alice.public_key().to_hex())?;
        let conversation = database.get_or_create_product_conversation(
            &alice.public_key().to_hex(),
            vec![bob.public_key().to_hex()],
            1,
        )?;
        assert_eq!(conversation.member_public_keys.len(), 2);
        let first = database.product_conversations(&alice.public_key().to_hex(), false, None, 1)?;
        assert_eq!(first.conversations.len(), 1);
        Ok(())
    }

    #[test]
    fn settings_patch_retains_unknown_fields() -> Result<(), SoftchatError> {
        let alice = LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let mut database = AccountDatabase::open(":memory:", alice.public_key().to_hex())?;
        let first = database
            .apply_product_settings_patch(r#"{"future":{"kept":true},"quickReaction":"❤️"}"#, 1)?;
        assert_eq!(first.settings.quick_reaction.as_deref(), Some("❤️"));
        let second = database.apply_product_settings_patch(r#"{"quickReaction":"👍"}"#, 2)?;
        assert_eq!(second.settings.unknown_json, r#"{"future":{"kept":true}}"#);
        Ok(())
    }

    #[test]
    fn ios_wallpaper_and_custom_emoji_settings_round_trip_through_canonical_form()
    -> Result<(), SoftchatError> {
        let input = r#"{
            "chatWallpaperImage":"ocean",
            "customEmojis":[
                {"id":"party","url":"https://cdn.example/party.png"},
                {"id":"Wave","url":"https://cdn.example/wave.png"}
            ],
            "future":{"retained":true}
        }"#;
        let canonical = canonical_settings_json(input)?;
        let settings = settings_from_json(&canonical)?;
        assert_eq!(settings.chat_wallpaper.as_deref(), Some("ocean"));
        assert_eq!(
            settings.custom_emoji_references,
            vec![
                CustomEmojiReference {
                    shortcode: "party".to_owned(),
                    url: "https://cdn.example/party.png".to_owned(),
                },
                CustomEmojiReference {
                    shortcode: "Wave".to_owned(),
                    url: "https://cdn.example/wave.png".to_owned(),
                },
            ]
        );
        assert_eq!(settings.unknown_json, r#"{"future":{"retained":true}}"#);
        let wire = settings_wire_json(&canonical)?;
        let wire: Value =
            serde_json::from_str(&wire).map_err(|_| SoftchatError::InvalidAccountSettings)?;
        assert_eq!(wire["chatWallpaperImage"], "ocean");
        assert_eq!(wire["customEmojis"][0]["id"], "party");
        Ok(())
    }

    #[test]
    fn ios_quick_reaction_data_maps_to_the_portable_setting() -> Result<(), SoftchatError> {
        let unicode = STANDARD.encode(r#"{"unicode":{"_0":{"emoji":"🔥"}}}"#.as_bytes());
        let input = json!({
            "quickReactionEmoji": unicode,
            "customEmojis": [
                {"id":"party","url":"https://cdn.example/party.png"}
            ]
        });
        let settings = settings_from_json(&input.to_string())?;
        assert_eq!(settings.quick_reaction.as_deref(), Some("🔥"));
        let canonical = serde_json::to_string(&settings_to_object(&settings)?)
            .map_err(|_| SoftchatError::InvalidAccountSettings)?;
        let wire = settings_wire_json(&canonical)?;
        let wire: Value =
            serde_json::from_str(&wire).map_err(|_| SoftchatError::InvalidAccountSettings)?;
        let encoded = wire["quickReactionEmoji"]
            .as_str()
            .ok_or(SoftchatError::InvalidAccountSettings)?;
        let decoded = STANDARD
            .decode(encoded)
            .map_err(|_| SoftchatError::InvalidAccountSettings)?;
        let decoded: Value =
            serde_json::from_slice(&decoded).map_err(|_| SoftchatError::InvalidAccountSettings)?;
        assert_eq!(decoded["unicode"]["_0"]["emoji"], "🔥");

        let custom = settings_from_json(
            r#"{"quickReaction":":party:","customEmojiReferences":[{"shortcode":"party","url":"https://cdn.example/party.png"}]}"#,
        )?;
        let canonical = serde_json::to_string(&settings_to_object(&custom)?)
            .map_err(|_| SoftchatError::InvalidAccountSettings)?;
        let wire = settings_wire_json(&canonical)?;
        let wire: Value =
            serde_json::from_str(&wire).map_err(|_| SoftchatError::InvalidAccountSettings)?;
        let encoded = wire["quickReactionEmoji"]
            .as_str()
            .ok_or(SoftchatError::InvalidAccountSettings)?;
        let decoded = STANDARD
            .decode(encoded)
            .map_err(|_| SoftchatError::InvalidAccountSettings)?;
        let decoded: Value =
            serde_json::from_slice(&decoded).map_err(|_| SoftchatError::InvalidAccountSettings)?;
        assert_eq!(decoded["custom"]["_0"]["id"], "party");
        Ok(())
    }

    #[test]
    fn profile_patch_retains_authenticated_fields_not_named_by_android() -> Result<(), SoftchatError>
    {
        let current = AccountProfile {
            public_key: "00".repeat(32),
            name: Some("alice".to_owned()),
            display_name: Some("Alice".to_owned()),
            about: Some("before".to_owned()),
            picture: Some("https://cdn.example/avatar.png".to_owned()),
            website: Some("https://example.com".to_owned()),
            banner: Some("https://cdn.example/banner.png".to_owned()),
            bot: Some(false),
            unknown_json: r#"{"future":{"retained":true}}"#.to_owned(),
            local_override: false,
            updated_at: 1,
            first_seen_at: Some(1),
        };
        let (updated, _) =
            profile_update_from_patch(&current, r#"{"about":"after","displayName":null}"#)?;
        assert_eq!(updated.name.as_deref(), Some("alice"));
        assert_eq!(updated.display_name, None);
        assert_eq!(updated.about.as_deref(), Some("after"));
        assert_eq!(
            updated.picture.as_deref(),
            Some("https://cdn.example/avatar.png")
        );
        assert_eq!(updated.website.as_deref(), Some("https://example.com"));
        assert_eq!(
            updated.banner.as_deref(),
            Some("https://cdn.example/banner.png")
        );
        assert_eq!(updated.bot, Some(false));
        assert_eq!(updated.unknown_json, r#"{"future":{"retained":true}}"#);
        Ok(())
    }

    #[test]
    fn portable_drafts_validate_utf16_boundaries() -> Result<(), SoftchatError> {
        let alice = LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let bob = LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let mut database = AccountDatabase::open(":memory:", alice.public_key().to_hex())?;
        let conversation = database.get_or_create_product_conversation(
            &alice.public_key().to_hex(),
            vec![bob.public_key().to_hex()],
            1,
        )?;
        database.save_product_draft(
            &conversation.id,
            Some(AccountDraft {
                text: "a😀b".to_owned(),
                emoji_tags: Vec::new(),
                spans: vec![DraftSpan {
                    start_utf16: 1,
                    end_utf16: 3,
                    kind: DraftSpanKind::Bold,
                    value: String::new(),
                }],
                reply_to_message_id: String::new(),
                attachment_operation_ids: Vec::new(),
                updated_at: 2,
            }),
            2,
        )?;
        assert_eq!(
            database
                .product_conversation(&conversation.id, &alice.public_key().to_hex())?
                .and_then(|value| value.draft)
                .map(|value| value.spans.len()),
            Some(1)
        );
        database.save_product_draft(
            &conversation.id,
            Some(AccountDraft {
                text: ":Device sticker:".to_owned(),
                emoji_tags: vec![vec![
                    "emoji".to_owned(),
                    "Device sticker".to_owned(),
                    "https://cdn.example/device-sticker.webp".to_owned(),
                ]],
                spans: vec![DraftSpan {
                    start_utf16: 0,
                    end_utf16: 16,
                    kind: DraftSpanKind::CustomEmoji,
                    value: "Device sticker".to_owned(),
                }],
                reply_to_message_id: String::new(),
                attachment_operation_ids: Vec::new(),
                updated_at: 3,
            }),
            3,
        )?;
        assert_eq!(
            database
                .product_conversation(&conversation.id, &alice.public_key().to_hex())?
                .and_then(|value| value.draft)
                .and_then(|value| value.spans.into_iter().next())
                .map(|value| value.value),
            Some("Device sticker".to_owned())
        );
        assert!(matches!(
            database.save_product_draft(
                &conversation.id,
                Some(AccountDraft {
                    text: "a😀b".to_owned(),
                    emoji_tags: Vec::new(),
                    spans: vec![DraftSpan {
                        start_utf16: 2,
                        end_utf16: 3,
                        kind: DraftSpanKind::Bold,
                        value: String::new(),
                    }],
                    reply_to_message_id: String::new(),
                    attachment_operation_ids: Vec::new(),
                    updated_at: 4,
                }),
                4,
            ),
            Err(SoftchatError::InvalidAccountOperation)
        ));
        Ok(())
    }
}