softchat/

storage.rs

//! Rust-owned account persistence and bounded use-case queries.
//!
//! One [`AccountDatabase`] owns one SQLite connection for one opaque account.
//! Protocol truth, authenticated rumors, projections, outgoing operations,
//! delivery leases, relay state, and synchronization checkpoints are committed
//! through Rust transactions. Platforms provide only the database path and
//! execute network, file, lifecycle, and UI work.

#![allow(
    clippy::too_many_lines,
    reason = "the schema and its transactional invariants are reviewed together"
)]

use std::collections::BTreeSet;
use std::fmt;
use std::ops::Deref;
use std::path::Path;
use std::time::Duration;

use rusqlite::{
    Connection, OptionalExtension, Savepoint, Transaction, TransactionBehavior, params,
};
use serde::Serialize;
use url::Url;

use crate::account_diagnostics::{
    AccountLogCategory, AccountLogCounterKind as Counter, AccountLogLevel, AccountLogSource,
    AccountLogSubject, CommitDiagnostics, Diagnostic,
};
use crate::chat::subject_icon_metadata;
use crate::event::{RumorEvent, SignedEvent};
use crate::runtime::stable_intent_id;
use crate::{
    AccountEngine, DeliveryClaim, DeliveryIntentSnapshot, DeliveryIntentState, DeliveryReducer,
    DnsRelayRecord, IngestionReceipt, LocalIdentity, NegentropyItem, Nip59EnvelopeKind,
    OperationSnapshot, PreparedIncomingBatch, PreparedOutgoingOperation, ProjectionKind,
    RelayCatalogEntry, RelayCatalogPlan, RelayCatalogReducer, RelayCatalogSource,
    RelayDeliveryResult, SignedNostrEvent, SoftchatError,
};

const ACCOUNT_SCHEMA_VERSION: i64 = 16;

// The single SQLite definition of effective edit eligibility and ordering.
// Page/context/preview queries and the search index all consume this view.
const EFFECTIVE_MESSAGE_EDITS_SCHEMA: &str = "
    CREATE VIEW IF NOT EXISTS effective_message_edits AS
    SELECT original.logical_event_id AS message_id,
           edit.logical_event_id AS edit_id,
           edit.content AS content
    FROM projections original
    JOIN projections edit ON edit.logical_event_id = (
        SELECT candidate.logical_event_id
        FROM projection_targets target
        JOIN projections candidate
          ON candidate.logical_event_id = target.logical_event_id
        WHERE target.target_event_id = original.logical_event_id
          AND candidate.kind = 'edit'
          AND candidate.author_public_key = original.author_public_key
          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 = candidate.logical_event_id
                AND deletion.kind = 'deletion'
                AND deletion.author_public_key = candidate.author_public_key
          )
        ORDER BY candidate.created_at DESC, candidate.logical_event_id ASC
        LIMIT 1
    )
    WHERE original.kind = 'chat_message' AND original.ephemeral = 0;
";
const MAX_DATABASE_PATH_BYTES: usize = 4_096;
const MAX_QUERY_PAGE: u32 = 200;
const MAX_LOCAL_CONTACT_NAME_BYTES: usize = 256;
const MAX_CACHE_URL_BYTES: usize = 4_096;
const MAX_STICKER_NAME_BYTES: usize = 128;
const MAX_STICKER_EMOJIS: usize = 64;
const MAX_STICKER_EMOJI_BYTES: usize = 64;
const MAX_LINK_PREVIEW_TITLE_BYTES: usize = 1_024;
const MAX_LINK_PREVIEW_DESCRIPTION_BYTES: usize = 8_192;
const DEFAULT_BUSY_TIMEOUT: Duration = Duration::from_secs(5);
const DEFAULT_RELAY_URL: &str = "wss://it2.softchat.c84ad60.xyz/";
// The account transport emits at most 128 actions. Delivery payloads are
// bounded to 512 KiB in aggregate, sync frames to 60 KiB, typing state to
// 1,024 entries, and relay URLs to 2 KiB. The factor of six is serde_json's
// maximum string-byte expansion; the remaining term bounds action metadata.
const MAX_STORED_TRANSPORT_OUTCOME_BYTES: usize = 6
    * (crate::MAX_DELIVERY_CLAIM_BYTES
        + crate::MAX_RELAY_FRAME_BYTES
        + crate::account_transport::MAX_TRANSPORT_ACTIONS
            * crate::account_transport::SYNC_FRAME_SIZE_LIMIT as usize)
    + crate::account_transport::MAX_TRANSPORT_ACTIONS * 8 * 1_024;

/// Authoritative child-first inventory of account-owned rows removed by reset.
///
/// `softchat_meta` is deliberately absent because it binds the file to one
/// account and owns the monotonic schema/revision metadata. `relay_catalog`
/// and `account_settings` are reset to their safe defaults after this list is
/// emptied.
const ACCOUNT_CLEAR_TABLES: &[&str] = &[
    "pending_message_attachments",
    "pending_asset_replacements",
    "pending_media_messages",
    "media_operations",
    "message_local_extras",
    "message_fts",
    "projection_targets",
    "projection_participants",
    "conversation_members",
    "conversation_subject_icons",
    "conversation_read_state",
    "conversation_state",
    "app_data_effective",
    "projections",
    "authenticated_rumor_wrappers",
    "authenticated_rumors",
    "event_fingerprints",
    "signed_events",
    "relay_delivery_intents",
    "outgoing_event_copies",
    "outgoing_operations",
    "sync_checkpoints",
    "local_contacts",
    "sticker_cache",
    "link_preview_cache",
    "quarantine",
    "processed_transport_results",
];

enum AccountWriteScope<'connection> {
    Transaction(Transaction<'connection>),
    Savepoint(Savepoint<'connection>),
}

impl Deref for AccountWriteScope<'_> {
    type Target = Connection;

    fn deref(&self) -> &Self::Target {
        match self {
            Self::Transaction(transaction) => transaction,
            Self::Savepoint(savepoint) => savepoint,
        }
    }
}

impl AccountWriteScope<'_> {
    fn commit(self) -> Result<(), SoftchatError> {
        match self {
            Self::Transaction(transaction) => transaction.commit(),
            Self::Savepoint(savepoint) => savepoint.commit(),
        }
        .map_err(map_sqlite_error)
    }
}

fn begin_write_scope(connection: &mut Connection) -> Result<AccountWriteScope<'_>, SoftchatError> {
    if connection.is_autocommit() {
        connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map(AccountWriteScope::Transaction)
            .map_err(map_sqlite_error)
    } else {
        connection
            .savepoint()
            .map(AccountWriteScope::Savepoint)
            .map_err(map_sqlite_error)
    }
}

/// Static database information returned after opening one account.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountDatabaseInfo {
    /// Opaque non-secret account partition bound to the file.
    pub account_id: String,
    /// Rust-owned schema version.
    pub schema_version: i64,
    /// Monotonic revision incremented by every successful state mutation.
    pub revision: i64,
    /// Whether SQLite accepted WAL journal mode.
    pub write_ahead_logging: bool,
}

/// Redacted bounded account health information for support and recovery UI.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountDiagnostics {
    /// Rust-owned schema version.
    pub schema_version: i64,
    /// Monotonic durable account revision.
    pub revision: i64,
    /// Canonical active relay, empty when the catalog is not configured.
    pub active_relay_url: String,
    /// Number of retained relay candidates.
    pub relay_count: u32,
    /// Number of unfinished recipient delivery intents.
    pub unfinished_delivery_count: u32,
    /// Number of logical media operations that may still need platform work.
    pub recoverable_media_count: u32,
    /// Number of authenticated durable outer events.
    pub stored_event_count: u32,
    /// Number of redacted quarantined inputs.
    pub quarantine_count: u32,
    /// Main SQLite database allocation, excluding WAL and SHM sidecars.
    pub database_bytes: u64,
}

/// Result of one atomic authenticated ingestion.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct StoredIngestion {
    /// Total classification of the supplied batch.
    pub receipt: IngestionReceipt,
    /// Database revision after commit.
    pub revision: i64,
    /// Every authenticated projection selected for the caller.
    ///
    /// Durable values have committed to SQLite before this result is returned.
    /// Ephemeral values are authenticated but intentionally remain memory-only.
    /// The bounded JSON push boundary also returns an authenticated projection
    /// when a fresh wrapper proves a logical rumor that was already projected.
    pub projections: Vec<AccountProjection>,
    /// Authenticated non-durable values for immediate Android use cases.
    pub ephemeral_projections: Vec<AccountProjection>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum ReplayProjectionPolicy {
    SuppressKnownRumors,
    IncludeKnownRumors,
}

/// Generic result for a transaction that changed account state.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountMutationResult {
    /// Database revision after commit.
    pub revision: i64,
}

/// Result of atomically selecting a failover relay and rebinding unfinished work.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct RelayFailoverMutation {
    /// Complete catalog after rotation.
    pub entries: Vec<RelayCatalogEntry>,
    /// Newly selected canonical relay URL.
    pub active_relay_url: String,
    /// Number of unfinished delivery intents rebound to the new relay.
    pub rebound_intent_count: u32,
    /// Durable account revision after the atomic transaction.
    pub revision: i64,
}

/// Exact stored event payload selected by an authenticated ID.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct StoredEventJson {
    /// Canonical event ID.
    pub event_id: String,
    /// Exact canonical signed event JSON.
    pub event_json: String,
}

/// Persisted relay catalog mutation.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct RelayCatalogMutation {
    /// Complete persisted relay rows.
    pub entries: Vec<RelayCatalogEntry>,
    /// Database revision after commit.
    pub revision: i64,
}

/// Persisted authenticated relay-discovery reconciliation.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct StoredRelayCatalogPlan {
    /// Deterministic catalog and refresh plan.
    pub plan: RelayCatalogPlan,
    /// Database revision after commit.
    pub revision: i64,
}

/// Result of an idempotent outgoing operation transaction.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountOperationResult {
    /// Current aggregate operation state.
    pub operation: OperationSnapshot,
    /// Database revision after commit, or the current revision for a replay.
    pub revision: i64,
    /// Whether this call inserted the operation for the first time.
    pub inserted: bool,
}

pub(crate) struct ProductAccountOperationResult {
    pub operation: AccountOperationResult,
    pub result_message_id: String,
}

/// Exact payload attached to one claimed relay-delivery intent.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct DeliveryPayload {
    /// Stable intent ID.
    pub intent_id: String,
    /// Canonical configured relay URL.
    pub relay_url: String,
    /// Exact canonical signed event JSON.
    pub event_json: String,
}

/// One committed delivery lease and the exact bytes Android may send.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct ClaimedDelivery {
    /// Rust-validated lease snapshot.
    pub claim: DeliveryClaim,
    /// Payloads in the same order as `claim.intents`.
    pub payloads: Vec<DeliveryPayload>,
    /// Database revision after the lease transaction.
    pub revision: i64,
}

/// Bounded conversation list item derived from authenticated projections.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct ConversationSummary {
    /// Stable conversation hash derived from the complete participant set.
    pub conversation_id: String,
    /// Sorted canonical participant public keys.
    pub participants: Vec<String>,
    /// Effective last-message text after valid edits.
    pub last_message: String,
    /// Timestamp of the newest durable message projection.
    pub last_created_at: i64,
    /// Number of non-deleted durable chat messages.
    pub message_count: u32,
}

/// One complete conversation-list use case returned in a single native call.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct ConversationView {
    /// Aggregate ordering, participants, effective text, and count.
    pub summary: ConversationSummary,
    /// Newest non-deleted message with exact authenticated protocol truth.
    pub last_message: AccountEventNode,
    /// Newest authenticated subject update, when one exists.
    pub subject: Option<AccountProjection>,
}

/// One bounded message view derived without replacing protocol truth.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountMessage {
    /// Authenticated rumor ID.
    pub event_id: String,
    /// Stable conversation hash.
    pub conversation_id: String,
    /// Authenticated logical author.
    pub author_public_key: String,
    /// Original event timestamp.
    pub created_at: i64,
    /// Effective text after deterministic valid-edit ordering.
    pub content: String,
    /// Whether an authorized deletion takes precedence.
    pub deleted: bool,
    /// Whether `content` came from a later authorized edit.
    pub edited: bool,
    /// Marked reply target, or an empty string.
    pub reply_to_event_id: String,
}

/// One authenticated projection with exact protocol truth attached.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountProjection {
    /// Semantic projection family.
    pub kind: ProjectionKind,
    /// Authenticated logical rumor or event ID.
    pub logical_event_id: String,
    /// Exact verified outer signed event.
    pub outer_event: SignedEvent,
    /// Exact authenticated rumor for NIP-59 data, when present.
    pub rumor: Option<RumorEvent>,
    /// Authenticated logical author.
    pub author_public_key: String,
    /// Portable logical event timestamp.
    pub created_at: i64,
    /// Conversation hash, or an empty string for account-wide data.
    pub conversation_id: String,
    /// Sorted complete participant set.
    pub participants: Vec<String>,
    /// Ordered reply, edit, reaction, or deletion targets.
    pub target_event_ids: Vec<String>,
    /// Validated primary text or JSON payload.
    pub content: String,
}

/// One use-case node derived from authenticated protocol truth.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct AccountEventNode {
    /// Exact authenticated projection and protocol values.
    pub projection: AccountProjection,
    /// Effective text after authorized deterministic edits.
    pub effective_content: String,
    /// Whether an authorized deletion takes precedence.
    pub deleted: bool,
    /// Whether `effective_content` came from an authorized edit.
    pub edited: bool,
    /// Number of non-deleted authenticated message children.
    pub reply_count: u32,
}

/// One cached custom emoji or sticker descriptor.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct StickerRecord {
    /// Stable shortcode without surrounding colons.
    pub name: String,
    /// Validated HTTPS asset URL.
    pub url: String,
    /// Bounded emoji strings used for search and suggestions.
    pub associated_emojis: Vec<String>,
}

/// One cached HTTP link-preview result.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct LinkPreviewRecord {
    /// Exact validated HTTP(S) request URL and cache key.
    pub url: String,
    /// Optional bounded page title.
    pub title: Option<String>,
    /// Optional bounded page description.
    pub description: Option<String>,
    /// Optional validated HTTP(S) preview-image URL.
    pub image_url: Option<String>,
    /// Android-supplied wall-clock observation time in milliseconds.
    pub updated_at_millis: i64,
}

/// One Android-local address-book entry.
///
/// Local contacts are use-case state, not authenticated Nostr metadata. They
/// are deliberately stored beside the account in Rust SQLite so Android does
/// not need a second database or an in-memory persistence exception.
#[derive(Clone, Debug, Eq, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
#[cfg_attr(feature = "native-bindings", derive(uniffi::Record))]
pub struct LocalContactRecord {
    /// Canonical remote public key.
    pub public_key: String,
    /// Optional bounded local display name.
    pub name: Option<String>,
    /// Android-supplied wall-clock update time in milliseconds.
    pub updated_at_millis: i64,
}

/// One account-scoped SQLite owner.
///
/// The type is intentionally synchronous. Native language facades serialize a
/// database handle and Android invokes it on its I/O dispatcher. Rust never
/// calls a platform database callback and never owns a background thread.
pub struct AccountDatabase {
    pub(crate) connection: Connection,
    pub(crate) account_id: String,
    write_ahead_logging: bool,
    pub(crate) diagnostics: CommitDiagnostics,
}

struct RawConversationView {
    conversation_id: String,
    message_count: u32,
    logical_event_id: String,
    author_public_key: String,
    created_at: i64,
    content: String,
    outer_json: String,
    rumor_json: Option<String>,
    participants_csv: Option<String>,
    targets_csv: Option<String>,
    edited_content: Option<String>,
    reply_count: u32,
    subject: Option<RawProjection>,
}

struct RawProjection {
    stored_kind: String,
    logical_event_id: String,
    author_public_key: String,
    created_at: i64,
    conversation_id: String,
    content: String,
    outer_json: String,
    rumor_json: Option<String>,
    participants_csv: Option<String>,
    targets_csv: Option<String>,
}

struct RawEventNode {
    stored_kind: String,
    logical_event_id: String,
    author_public_key: String,
    created_at: i64,
    conversation_id: String,
    content: String,
    outer_json: String,
    rumor_json: Option<String>,
    participants_csv: Option<String>,
    targets_csv: Option<String>,
    deleted: bool,
    edited_content: Option<String>,
    reply_count: u32,
}

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

impl AccountDatabase {
    pub(crate) fn next_app_data_event_timestamp(
        &self,
        context: &str,
        now: i64,
    ) -> Result<i64, SoftchatError> {
        validate_timestamp(now)?;
        let current = self
            .connection
            .query_row(
                "SELECT event_created_at FROM app_data_effective WHERE context = ?1",
                [context],
                |row| row.get::<_, i64>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?;
        current
            .filter(|created_at| *created_at >= now)
            .map_or(Ok(now), |created_at| {
                created_at
                    .checked_add(1)
                    .ok_or(SoftchatError::InvalidAccountOperation)
            })
    }

    /// Open or create one account database and apply Rust-owned migrations.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for an invalid path,
    /// account mismatch, foreign schema, unsupported schema, or SQLite
    /// failure.
    pub fn open(path: impl AsRef<Path>, account_id: String) -> Result<Self, SoftchatError> {
        AccountEngine::validate_account_id(&account_id)?;
        let path = path.as_ref();
        let path_text = path.to_string_lossy();
        if path_text.is_empty()
            || path_text.len() > MAX_DATABASE_PATH_BYTES
            || path_text.as_bytes().contains(&0)
        {
            return Err(SoftchatError::InvalidPersistenceResult);
        }

        let mut connection = Connection::open(path).map_err(map_sqlite_error)?;
        connection
            .busy_timeout(DEFAULT_BUSY_TIMEOUT)
            .map_err(map_sqlite_error)?;
        connection
            .pragma_update(None, "foreign_keys", "ON")
            .map_err(map_sqlite_error)?;
        connection
            .pragma_update(None, "synchronous", "NORMAL")
            .map_err(map_sqlite_error)?;
        connection
            .pragma_update(None, "temp_store", "MEMORY")
            .map_err(map_sqlite_error)?;
        let journal_mode: String = connection
            .pragma_query_value(None, "journal_mode", |row| row.get(0))
            .map_err(map_sqlite_error)?;
        let journal_mode = if journal_mode.eq_ignore_ascii_case("wal") || path_text == ":memory:" {
            journal_mode
        } else {
            connection
                .pragma_update_and_check(None, "journal_mode", "WAL", |row| row.get(0))
                .map_err(map_sqlite_error)?
        };
        let write_ahead_logging = journal_mode.eq_ignore_ascii_case("wal");
        if path_text != ":memory:" && !write_ahead_logging {
            return Err(SoftchatError::InvalidPersistenceResult);
        }

        match account_schema_ownership(&connection)? {
            AccountSchemaOwnership::Foreign => {
                return Err(SoftchatError::InvalidPersistenceResult);
            }
            AccountSchemaOwnership::Owned if current_schema_matches(&connection, &account_id)? => {}
            AccountSchemaOwnership::Empty | AccountSchemaOwnership::Owned => {
                migrate(&mut connection, &account_id)?;
            }
        }
        Ok(Self {
            connection,
            account_id,
            write_ahead_logging,
            diagnostics: CommitDiagnostics::default(),
        })
    }

    pub(crate) fn record_diagnostic(&self, diagnostic: Diagnostic) {
        self.diagnostics
            .record(self.connection.is_autocommit(), diagnostic);
    }

    // Optional diagnostic reads neither allocate payload copies nor affect writes.
    pub(crate) fn local_state_differs<P: rusqlite::Params>(&self, sql: &str, params: P) -> bool {
        self.diagnostics.debug_enabled()
            && self
                .connection
                .query_row(sql, params, |row| row.get::<_, bool>(0))
                .unwrap_or(false)
    }

    pub(crate) fn record_local_change(&self, changed: bool, revision: i64) {
        if changed {
            let mut event = Diagnostic::new(
                AccountLogCategory::Storage,
                "storage.local.changed",
                "Local account state changed",
            )
            .debug();
            event.direction = Some(crate::account_diagnostics::AccountLogDirection::Local);
            event.source = Some(AccountLogSource::LocalCommand);
            event.revision = Some(revision);
            self.record_diagnostic(event);
        }
    }

    /// Return schema, revision, and journal information.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence error if metadata cannot be read.
    pub fn info(&self) -> Result<AccountDatabaseInfo, SoftchatError> {
        Ok(AccountDatabaseInfo {
            account_id: self.account_id.clone(),
            schema_version: meta_integer(&self.connection, "schema_version")?,
            revision: self.revision()?,
            write_ahead_logging: self.write_ahead_logging,
        })
    }

    /// Return the current monotonic database revision.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence error if metadata cannot be read.
    pub fn revision(&self) -> Result<i64, SoftchatError> {
        meta_integer(&self.connection, "revision")
    }

    /// Return redacted bounded account health without exposing event or key data.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence error if SQLite metadata is inconsistent.
    pub fn diagnostics(&self) -> Result<AccountDiagnostics, SoftchatError> {
        let info = self.info()?;
        let catalog = self.relay_catalog()?;
        let active_relay_url = catalog
            .iter()
            .find(|entry| entry.is_active)
            .map(|entry| entry.url.clone())
            .unwrap_or_default();
        let relay_count =
            u32::try_from(catalog.len()).map_err(|_| SoftchatError::InvalidPersistenceResult)?;
        let unfinished_delivery_count = unfinished_delivery_count(&self.connection)?;
        let recoverable_media_count = bounded_query_count(
            &self.connection,
            "SELECT COUNT(*) FROM media_operations
             WHERE state IN ('prepared', 'running', 'retryable')",
        )?;
        let stored_event_count =
            bounded_query_count(&self.connection, "SELECT COUNT(*) FROM signed_events")?;
        let quarantine_count =
            bounded_query_count(&self.connection, "SELECT COUNT(*) FROM quarantine")?;
        let page_count: i64 = self
            .connection
            .pragma_query_value(None, "page_count", |row| row.get(0))
            .map_err(map_sqlite_error)?;
        let page_size: i64 = self
            .connection
            .pragma_query_value(None, "page_size", |row| row.get(0))
            .map_err(map_sqlite_error)?;
        let page_count =
            u64::try_from(page_count).map_err(|_| SoftchatError::InvalidPersistenceResult)?;
        let page_size =
            u64::try_from(page_size).map_err(|_| SoftchatError::InvalidPersistenceResult)?;
        Ok(AccountDiagnostics {
            schema_version: info.schema_version,
            revision: info.revision,
            active_relay_url,
            relay_count,
            unfinished_delivery_count,
            recoverable_media_count,
            stored_event_count,
            quarantine_count,
            database_bytes: page_count.saturating_mul(page_size),
        })
    }

    /// Authenticate, unwrap, classify, and atomically persist incoming events.
    ///
    /// Ephemeral projections are authenticated and reported in the receipt but
    /// never written to durable history.
    ///
    /// # Errors
    ///
    /// Returns stable event, envelope, account, projection, or persistence
    /// errors. No partial transaction is committed.
    pub fn ingest(
        &mut self,
        identity: &LocalIdentity,
        events: Vec<SignedEvent>,
        received_at: i64,
    ) -> Result<StoredIngestion, SoftchatError> {
        self.ingest_with_projection_policy(
            identity,
            events,
            received_at,
            ReplayProjectionPolicy::SuppressKnownRumors,
            AccountLogSource::IncomingApi,
        )
    }

    pub(crate) fn ingest_with_authenticated_replays(
        &mut self,
        identity: &LocalIdentity,
        events: Vec<SignedEvent>,
        received_at: i64,
    ) -> Result<StoredIngestion, SoftchatError> {
        self.ingest_with_projection_policy(
            identity,
            events,
            received_at,
            ReplayProjectionPolicy::IncludeKnownRumors,
            AccountLogSource::IncomingApi,
        )
    }

    pub(crate) fn ingest_from_relay(
        &mut self,
        identity: &LocalIdentity,
        events: Vec<SignedEvent>,
        received_at: i64,
        historical: bool,
    ) -> Result<StoredIngestion, SoftchatError> {
        self.ingest_with_projection_policy(
            identity,
            events,
            received_at,
            ReplayProjectionPolicy::SuppressKnownRumors,
            if historical {
                AccountLogSource::RelaySync
            } else {
                AccountLogSource::RelayLive
            },
        )
    }

    fn ingest_with_projection_policy(
        &mut self,
        identity: &LocalIdentity,
        events: Vec<SignedEvent>,
        received_at: i64,
        replay_policy: ReplayProjectionPolicy,
        source: AccountLogSource,
    ) -> Result<StoredIngestion, SoftchatError> {
        validate_timestamp(received_at)?;
        let batch = AccountEngine::prepare_incoming(identity, self.account_id.clone(), events)?;
        let ephemeral_projections = batch
            .events
            .iter()
            .filter(|event| event.projection.ephemeral)
            .map(account_projection_from_prepared)
            .collect();
        let transaction = begin_write_scope(&mut self.connection)?;
        let committed = ingest_prepared(
            &transaction,
            &batch,
            received_at,
            self.diagnostics.enabled(),
        )?;
        AccountEngine::validate_ingestion_receipt(&batch, &committed.receipt)?;
        let quarantined = committed
            .receipt
            .quarantined_event_ids
            .iter()
            .cloned()
            .collect::<BTreeSet<_>>();
        let mut returned_logical_ids = BTreeSet::new();
        let projections = batch
            .events
            .iter()
            .filter(|event| {
                !quarantined.contains(&event.outer_event.id)
                    && (event.projection.ephemeral
                        || replay_policy == ReplayProjectionPolicy::IncludeKnownRumors
                        || !committed
                            .suppressed_projection_outer_event_ids
                            .contains(&event.outer_event.id))
            })
            .filter(|event| {
                replay_policy == ReplayProjectionPolicy::SuppressKnownRumors
                    || returned_logical_ids.insert(event.projection.logical_event_id.clone())
            })
            .map(account_projection_from_prepared)
            .collect();
        let revision = bump_revision(&transaction)?;
        transaction.commit()?;
        for mut diagnostic in committed.diagnostics {
            diagnostic.revision = Some(revision);
            diagnostic.source = Some(source);
            diagnostic.direction = Some(crate::account_diagnostics::AccountLogDirection::Rx);
            if source == AccountLogSource::RelaySync {
                diagnostic.summarize = true;
                if diagnostic.level == AccountLogLevel::Info {
                    diagnostic.category = AccountLogCategory::Synchronization;
                    diagnostic.operation = "sync.progress";
                    diagnostic.summary = "Historical events stored";
                }
            }
            self.record_diagnostic(diagnostic);
        }
        Ok(StoredIngestion {
            receipt: committed.receipt,
            revision,
            projections,
            ephemeral_projections,
        })
    }

    /// Create and atomically persist one idempotent NIP-59 operation.
    ///
    /// The local sender copy is authenticated and projected in the same
    /// transaction as event copies and delivery intents. Reusing an existing
    /// operation ID returns its current state without generating new wrappers.
    ///
    /// # Errors
    ///
    /// Returns stable operation, identity, envelope, relay, or persistence
    /// errors. No socket may send until this method succeeds.
    #[allow(clippy::too_many_arguments)]
    pub fn enqueue_rumor(
        &mut self,
        identity: &LocalIdentity,
        operation_id: String,
        rumor: RumorEvent,
        recipients: Vec<String>,
        notification_recipients: Vec<String>,
        relay_urls: Vec<String>,
        kind: Nip59EnvelopeKind,
        now: i64,
    ) -> Result<AccountOperationResult, SoftchatError> {
        if let Some(operation) = self.operation_optional(&operation_id)? {
            self.record_diagnostic(Diagnostic::replay());
            return Ok(AccountOperationResult {
                operation,
                revision: self.revision()?,
                inserted: false,
            });
        }
        let prepared = AccountEngine::prepare_rumor_operation(
            identity,
            self.account_id.clone(),
            operation_id,
            rumor,
            recipients,
            notification_recipients,
            relay_urls,
            kind,
            now,
        )?;
        let sender_public_key = identity.public_key().to_hex();
        self.persist_operation(identity, prepared, &sender_public_key, now, "")
    }

    /// Atomically persist already signed direct events and delivery intents.
    ///
    /// This is intended for authenticated direct protocol events such as
    /// NIP-42 and NIP-98, not for bypassing private-chat wrapping.
    ///
    /// # Errors
    ///
    /// Returns stable event, operation, relay, or persistence errors.
    pub fn enqueue_signed(
        &mut self,
        identity: &LocalIdentity,
        operation_id: String,
        events: Vec<SignedEvent>,
        relay_urls: Vec<String>,
        now: i64,
    ) -> Result<AccountOperationResult, SoftchatError> {
        if let Some(operation) = self.operation_optional(&operation_id)? {
            self.record_diagnostic(Diagnostic::replay());
            return Ok(AccountOperationResult {
                operation,
                revision: self.revision()?,
                inserted: false,
            });
        }
        let prepared = AccountEngine::prepare_signed_operation(
            self.account_id.clone(),
            operation_id,
            events,
            relay_urls,
        )?;
        self.persist_operation(identity, prepared, "", now, "")
    }

    /// Create one high-level command after checking its canonical payload hash.
    ///
    /// This is the product idempotency boundary. Repeating the same command and
    /// payload returns the existing operation without creating fresh wrappers;
    /// reusing the identifier for different intent fails deterministically.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn enqueue_product_rumor(
        &mut self,
        identity: &LocalIdentity,
        command_id: String,
        command_hash: String,
        result_message_id: String,
        rumor: RumorEvent,
        recipients: Vec<String>,
        notification_recipients: Vec<String>,
        kind: Nip59EnvelopeKind,
        now: i64,
        draft_to_consume: Option<(&str, &crate::account_product::ProductDraftMatch)>,
    ) -> Result<ProductAccountOperationResult, SoftchatError> {
        if command_hash.len() != 64 || !command_hash.bytes().all(|byte| byte.is_ascii_hexdigit()) {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        if let Some(existing) = self.product_operation_if_committed(&command_id, &command_hash)? {
            return Ok(existing);
        }
        let relay_url = self.active_relay_url()?;
        let prepared = AccountEngine::prepare_rumor_operation(
            identity,
            self.account_id.clone(),
            command_id,
            rumor,
            recipients,
            notification_recipients,
            vec![relay_url],
            kind,
            now,
        )?;
        let sender_public_key = identity.public_key().to_hex();
        self.persist_product_operation(
            identity,
            prepared,
            &sender_public_key,
            now,
            &command_hash,
            &result_message_id,
            draft_to_consume,
        )
    }

    pub(crate) fn product_operation_if_committed(
        &self,
        command_id: &str,
        command_hash: &str,
    ) -> Result<Option<ProductAccountOperationResult>, SoftchatError> {
        let Some(existing_hash) = self.operation_command_hash(command_id)? else {
            return Ok(None);
        };
        if existing_hash != command_hash {
            return Err(SoftchatError::CommandConflict);
        }
        self.record_diagnostic(Diagnostic::replay());
        Ok(Some(ProductAccountOperationResult {
            operation: AccountOperationResult {
                operation: self.operation(command_id)?,
                revision: self.revision()?,
                inserted: false,
            },
            result_message_id: self.operation_result_message_id(command_id)?,
        }))
    }

    /// Create one high-level direct signed command with payload idempotency.
    pub(crate) fn enqueue_product_signed(
        &mut self,
        command_id: String,
        command_hash: String,
        result_message_id: String,
        events: Vec<SignedEvent>,
        now: i64,
        diagnostic_subject: AccountLogSubject,
    ) -> Result<ProductAccountOperationResult, SoftchatError> {
        if command_hash.len() != 64 || !command_hash.bytes().all(|byte| byte.is_ascii_hexdigit()) {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        if let Some(existing_hash) = self.operation_command_hash(&command_id)? {
            if existing_hash != command_hash {
                return Err(SoftchatError::CommandConflict);
            }
            self.record_diagnostic(Diagnostic::replay());
            return Ok(ProductAccountOperationResult {
                operation: AccountOperationResult {
                    operation: self.operation(&command_id)?,
                    revision: self.revision()?,
                    inserted: false,
                },
                result_message_id: self.operation_result_message_id(&command_id)?,
            });
        }
        let relay_url = self.active_relay_url()?;
        let prepared = AccountEngine::prepare_signed_operation(
            self.account_id.clone(),
            command_id,
            events,
            vec![relay_url],
        )?;
        self.persist_product_operation_without_identity(
            prepared,
            now,
            &command_hash,
            &result_message_id,
            diagnostic_subject,
        )
    }

    /// Atomically commit one complete settings value and its signed sync event.
    #[allow(clippy::too_many_arguments)]
    pub(crate) fn enqueue_product_signed_with_settings(
        &mut self,
        command_id: String,
        command_hash: String,
        events: Vec<SignedEvent>,
        settings_schema_version: u32,
        canonical_settings_json: String,
        wire_settings_json: String,
        now: i64,
    ) -> Result<ProductAccountOperationResult, SoftchatError> {
        validate_timestamp(now)?;
        if command_hash.len() != 64
            || !command_hash.bytes().all(|byte| byte.is_ascii_hexdigit())
            || canonical_settings_json.is_empty()
            || canonical_settings_json.len() > crate::MAX_ACCOUNT_SETTINGS_BYTES
            || wire_settings_json.is_empty()
            || wire_settings_json.len() > crate::MAX_ACCOUNT_SETTINGS_BYTES
        {
            return Err(SoftchatError::InvalidAccountSettings);
        }
        if let Some(existing_hash) = self.operation_command_hash(&command_id)? {
            if existing_hash != command_hash {
                return Err(SoftchatError::CommandConflict);
            }
            self.record_diagnostic(Diagnostic::replay());
            return Ok(ProductAccountOperationResult {
                operation: AccountOperationResult {
                    operation: self.operation(&command_id)?,
                    revision: self.revision()?,
                    inserted: false,
                },
                result_message_id: self.operation_result_message_id(&command_id)?,
            });
        }
        let effective_event = events
            .first()
            .filter(|_| events.len() == 1)
            .ok_or(SoftchatError::InvalidAccountSettings)?;
        let effective_event_id = effective_event.id.clone();
        let effective_created_at = effective_event.created_at;
        let relay_url = self.active_relay_url()?;
        let prepared = AccountEngine::prepare_signed_operation(
            self.account_id.clone(),
            command_id,
            events,
            vec![relay_url],
        )?;
        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![settings_schema_version, canonical_settings_json, now],
            )
            .map_err(map_sqlite_error)?;
        store_effective_app_data(
            &transaction,
            "app-settings",
            &wire_settings_json,
            effective_created_at,
            &effective_event_id,
        )?;
        persist_prepared_operation(&transaction, &prepared, now, 0, &command_hash, "")?;
        let intents = load_operation_intents(&transaction, &prepared.operation_id)?;
        let operation = DeliveryReducer::operation_snapshot(intents)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        let mut diagnostic = Diagnostic::for_subject(AccountLogSubject::AppSettings, true)
            .count(Counter::TotalIntents, u64::from(operation.total_intents));
        diagnostic.revision = Some(revision);
        self.record_diagnostic(diagnostic);
        Ok(ProductAccountOperationResult {
            operation: AccountOperationResult {
                operation,
                revision,
                inserted: true,
            },
            result_message_id: String::new(),
        })
    }

    /// Atomically commit one read-state entry, complete snapshot, and signed sync event.
    pub(crate) fn enqueue_product_signed_with_read_state(
        &mut self,
        command_id: String,
        command_hash: String,
        events: Vec<SignedEvent>,
        state: crate::AccountReadState,
        canonical_read_state_json: String,
        now: i64,
    ) -> Result<ProductAccountOperationResult, SoftchatError> {
        validate_timestamp(now)?;
        if command_hash.len() != 64
            || !command_hash.bytes().all(|byte| byte.is_ascii_hexdigit())
            || canonical_read_state_json.is_empty()
            || canonical_read_state_json.len() > crate::MAX_ACCOUNT_READ_STATE_BYTES
        {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        if let Some(existing_hash) = self.operation_command_hash(&command_id)? {
            if existing_hash != command_hash {
                return Err(SoftchatError::CommandConflict);
            }
            self.record_diagnostic(Diagnostic::replay());
            return Ok(ProductAccountOperationResult {
                operation: AccountOperationResult {
                    operation: self.operation(&command_id)?,
                    revision: self.revision()?,
                    inserted: false,
                },
                result_message_id: self.operation_result_message_id(&command_id)?,
            });
        }
        let effective_event = events
            .first()
            .filter(|_| events.len() == 1)
            .ok_or(SoftchatError::InvalidAccountOperation)?;
        let effective_event_id = effective_event.id.clone();
        let effective_created_at = effective_event.created_at;
        let parsed = crate::account_product::parse_read_state_snapshot(
            &canonical_read_state_json,
            effective_created_at,
            &effective_event_id,
        )?;
        let parsed_entry = parsed
            .entries
            .into_iter()
            .find(|entry| entry.state.conversation_id == state.conversation_id)
            .ok_or(SoftchatError::InvalidAccountOperation)?;
        if parsed_entry.state != state {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        let relay_url = self.active_relay_url()?;
        let prepared = AccountEngine::prepare_signed_operation(
            self.account_id.clone(),
            command_id,
            events,
            vec![relay_url],
        )?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let (read_at, read_message_id) = state
            .boundary
            .as_ref()
            .map_or((None, String::new()), |boundary| {
                (Some(boundary.created_at), boundary.message_id.clone())
            });
        transaction
            .execute(
                "INSERT INTO conversation_read_state
                 (conversation_id, read_at, read_message_id, forced_unread,
                  updated_at, update_id, unknown_json)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
                 ON CONFLICT(conversation_id) DO UPDATE SET
                   read_at = excluded.read_at,
                   read_message_id = excluded.read_message_id,
                   forced_unread = excluded.forced_unread,
                   updated_at = excluded.updated_at,
                   update_id = excluded.update_id,
                   unknown_json = excluded.unknown_json",
                params![
                    state.conversation_id,
                    read_at,
                    read_message_id,
                    i64::from(state.forced_unread),
                    state.updated_at,
                    state.update_id,
                    parsed_entry.unknown_json,
                ],
            )
            .map_err(map_sqlite_error)?;
        store_effective_app_data(
            &transaction,
            "read-state",
            &canonical_read_state_json,
            effective_created_at,
            &effective_event_id,
        )?;
        persist_prepared_operation(&transaction, &prepared, now, 0, &command_hash, "")?;
        let intents = load_operation_intents(&transaction, &prepared.operation_id)?;
        let operation = DeliveryReducer::operation_snapshot(intents)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        let mut diagnostic = Diagnostic::for_subject(AccountLogSubject::ReadState, true)
            .count(Counter::TotalIntents, u64::from(operation.total_intents));
        diagnostic.revision = Some(revision);
        self.record_diagnostic(diagnostic);
        Ok(ProductAccountOperationResult {
            operation: AccountOperationResult {
                operation,
                revision,
                inserted: true,
            },
            result_message_id: String::new(),
        })
    }

    /// Claim pending work, recover expired leases, and return exact payloads.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery or persistence error for a malformed lease,
    /// state conflict, invalid timestamp, or SQLite failure.
    pub fn claim_deliveries(
        &mut self,
        lease_id: String,
        relay_url: String,
        now: i64,
        lease_duration_seconds: u32,
    ) -> Result<Option<ClaimedDelivery>, SoftchatError> {
        validate_timestamp(now)?;
        let relay_url = if relay_url.is_empty() {
            String::new()
        } else {
            crate::parse_relay_endpoint(&relay_url)
                .map_err(|_| SoftchatError::InvalidDeliveryState)?
                .configured_url
        };
        if lease_duration_seconds == 0 || lease_duration_seconds > 3_600 {
            return Err(SoftchatError::InvalidDeliveryState);
        }
        let expires_at = now
            .checked_add(i64::from(lease_duration_seconds))
            .ok_or(SoftchatError::InvalidDeliveryState)?;
        let transaction = begin_write_scope(&mut self.connection)?;
        let expired_leases = transaction
            .execute(
                "UPDATE relay_delivery_intents
                 SET state = 'retryable', lease_id = '', lease_expires_at = 0,
                     last_category = 'lease_expired'
                 WHERE state IN ('claimed', 'socket_written')
                   AND lease_expires_at > 0 AND lease_expires_at <= ?1",
                [now],
            )
            .map_err(map_sqlite_error)?;

        // Older account profiles admitted event JSON up to the whole relay
        // frame bound, without reserving the EVENT envelope. Preserve those
        // copies but fail their unfinished local intents before claiming so
        // one unpublishable legacy row cannot starve otherwise valid work.
        let rejected_oversized = transaction
            .execute(
                "UPDATE relay_delivery_intents
                 SET state = 'rejected', lease_id = '', lease_expires_at = 0,
                     last_category = 'event_too_large'
                 WHERE state IN ('pending', 'retryable') AND payload_bytes > ?1",
                [i64::try_from(crate::session::MAX_ACCOUNT_EVENT_JSON_BYTES)
                    .map_err(|_| SoftchatError::InternalFailure)?],
            )
            .map_err(map_sqlite_error)?;

        let candidates = load_claimable(&transaction, &relay_url)?;
        if candidates.is_empty() {
            if expired_leases > 0 || rejected_oversized > 0 {
                bump_revision(&transaction)?;
            }
            transaction.commit()?;
            return Ok(None);
        }
        let claim = DeliveryReducer::claim(candidates, lease_id)?;
        for mutation in &claim.mutations {
            let changed = transaction
                .execute(
                    "UPDATE relay_delivery_intents
                     SET state = 'claimed', lease_id = ?1, lease_expires_at = ?2,
                         attempt_count = ?3, last_category = ?4
                     WHERE intent_id = ?5 AND state IN ('pending', 'retryable')
                       AND lease_id = ''",
                    params![
                        mutation.lease_id,
                        expires_at,
                        mutation.attempt_count,
                        mutation.category,
                        mutation.intent_id,
                    ],
                )
                .map_err(map_sqlite_error)?;
            if changed != 1 {
                return Err(SoftchatError::InvalidDeliveryState);
            }
        }
        let payloads = load_claim_payloads(&transaction, &claim)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit()?;
        Ok(Some(ClaimedDelivery {
            claim,
            payloads,
            revision,
        }))
    }

    /// Commit that selected leased payloads were written to a socket.
    ///
    /// Socket write is not relay acceptance.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery or persistence error for stale lease state.
    pub fn mark_socket_written(
        &mut self,
        claim: &DeliveryClaim,
        intent_ids: Vec<String>,
    ) -> Result<AccountMutationResult, SoftchatError> {
        let mutations = DeliveryReducer::mark_socket_written(claim, intent_ids)?;
        self.apply_delivery_mutations(claim, &mutations, "claimed")
    }

    /// Commit correlated terminal or retryable relay results.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery or persistence error for stale or
    /// contradictory state.
    pub fn apply_relay_results(
        &mut self,
        claim: &DeliveryClaim,
        results: Vec<RelayDeliveryResult>,
    ) -> Result<AccountMutationResult, SoftchatError> {
        let mutations = DeliveryReducer::apply_relay_results(claim, results)?;
        self.apply_delivery_mutations(claim, &mutations, "result")
    }

    /// Release unfinished leased work after cancellation or transport loss.
    ///
    /// # Errors
    ///
    /// Returns a stable delivery or persistence error for stale lease state.
    pub fn release_claim(
        &mut self,
        claim: &DeliveryClaim,
    ) -> Result<AccountMutationResult, SoftchatError> {
        let mutations = DeliveryReducer::release_claim(claim)?;
        self.apply_delivery_mutations(claim, &mutations, "release")
    }

    /// Return one operation's current durable aggregate state.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error when the operation is
    /// unknown or its durable rows are inconsistent.
    pub fn operation(&self, operation_id: &str) -> Result<OperationSnapshot, SoftchatError> {
        self.operation_optional(operation_id)?
            .ok_or(SoftchatError::InvalidAccountOperation)
    }

    /// Return a bounded newest-first page of durable outgoing operations.
    ///
    /// # Errors
    ///
    /// Returns a stable bound, operation, or persistence error.
    pub fn operations(&self, limit: u32) -> Result<Vec<OperationSnapshot>, SoftchatError> {
        validate_page(limit)?;
        let mut statement = self
            .connection
            .prepare(
                "SELECT operation_id FROM outgoing_operations
                 ORDER BY created_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 operations = Vec::new();
        for row in rows {
            operations.push(self.operation(&row.map_err(map_sqlite_error)?)?);
        }
        Ok(operations)
    }

    pub(crate) fn stored_transport_result(
        &self,
        run_id: &str,
        action_id: &str,
    ) -> Result<Option<(String, String)>, SoftchatError> {
        self.connection
            .query_row(
                "SELECT result_hash, outcome_json
                 FROM processed_transport_results
                 WHERE run_id = ?1 AND action_id = ?2",
                params![run_id, action_id],
                |row| Ok((row.get(0)?, row.get(1)?)),
            )
            .optional()
            .map_err(map_sqlite_error)
    }

    pub(crate) fn apply_transport_result_atomically<T>(
        &mut self,
        operation: impl FnOnce(&mut Self) -> Result<T, SoftchatError>,
    ) -> Result<T, SoftchatError> {
        if !self.connection.is_autocommit() {
            return Err(SoftchatError::InvalidPersistenceResult);
        }
        self.connection
            .execute_batch("BEGIN IMMEDIATE")
            .map_err(map_sqlite_error)?;
        self.diagnostics.finish(false);
        let result = operation(self);
        let outcome = match result {
            Ok(value) => match self.connection.execute_batch("COMMIT") {
                Ok(()) => Ok(value),
                Err(error) => {
                    let _ = self.connection.execute_batch("ROLLBACK");
                    Err(map_sqlite_error(error))
                }
            },
            Err(error) => {
                self.connection
                    .execute_batch("ROLLBACK")
                    .map_err(map_sqlite_error)?;
                Err(error)
            }
        };
        self.diagnostics.finish(outcome.is_ok());
        outcome
    }

    pub(crate) fn record_transport_result(
        &mut self,
        run_id: &str,
        action_id: &str,
        generation: u64,
        result_hash: &str,
        outcome_json: &str,
    ) -> Result<i64, SoftchatError> {
        let generation =
            i64::try_from(generation).map_err(|_| SoftchatError::InvalidRelaySession)?;
        if result_hash.len() != 64 || outcome_json.len() > MAX_STORED_TRANSPORT_OUTCOME_BYTES {
            return Err(SoftchatError::InvalidRelaySession);
        }
        let transaction = begin_write_scope(&mut self.connection)?;
        transaction
            .execute(
                "INSERT INTO processed_transport_results
                 (run_id, action_id, generation, result_hash, outcome_json)
                 VALUES (?1, ?2, ?3, ?4, ?5)",
                params![run_id, action_id, generation, result_hash, outcome_json],
            )
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "DELETE FROM processed_transport_results
                 WHERE rowid NOT IN (
                   SELECT rowid FROM processed_transport_results
                   ORDER BY rowid DESC LIMIT 4096
                 )",
                [],
            )
            .map_err(map_sqlite_error)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit()?;
        Ok(revision)
    }

    /// Requeue terminally rejected intents for an explicit user retry.
    ///
    /// Accepted copies are never duplicated. Repeating retry while work is
    /// already queued is idempotent.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for an unknown operation
    /// or an operation with active leased work.
    pub fn retry_operation(
        &mut self,
        operation_id: &str,
    ) -> Result<AccountOperationResult, SoftchatError> {
        let current = self.operation(operation_id)?;
        if current.active_intents != 0 {
            return Err(SoftchatError::InvalidDeliveryState);
        }
        let prior_revision = self.revision()?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let changed = transaction
            .execute(
                "UPDATE relay_delivery_intents
                 SET state = 'retryable', lease_id = '', lease_expires_at = 0,
                     last_category = 'manual_retry'
                 WHERE operation_id = ?1 AND state = 'rejected'",
                [operation_id],
            )
            .map_err(map_sqlite_error)?;
        let revision = if changed == 0 {
            prior_revision
        } else {
            bump_revision(&transaction)?
        };
        let intents = load_operation_intents(&transaction, operation_id)?;
        let operation = DeliveryReducer::operation_snapshot(intents)?;
        let diagnostic = (changed > 0 && self.diagnostics.enabled())
            .then(|| delivery_diagnostic(&transaction, &operation, "retry", true));
        transaction.commit().map_err(map_sqlite_error)?;
        if let Some(mut diagnostic) = diagnostic {
            diagnostic.revision = Some(revision);
            self.record_diagnostic(diagnostic);
        }
        Ok(AccountOperationResult {
            operation,
            revision,
            inserted: false,
        })
    }

    /// Cancel every unclaimed delivery intent that is still safe to cancel.
    ///
    /// Accepted or rejected terminal rows remain exact history. Active leases
    /// must first be released by the transport run.
    ///
    /// # Errors
    ///
    /// Returns a stable account or delivery-state error for unknown or active
    /// work.
    pub fn cancel_operation(
        &mut self,
        operation_id: &str,
    ) -> Result<AccountOperationResult, SoftchatError> {
        let current = self.operation(operation_id)?;
        if current.active_intents != 0 {
            return Err(SoftchatError::InvalidDeliveryState);
        }
        let prior_revision = self.revision()?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let changed = transaction
            .execute(
                "UPDATE relay_delivery_intents
                 SET state = 'cancelled', lease_id = '', lease_expires_at = 0,
                     last_category = 'user_cancelled'
                 WHERE operation_id = ?1 AND state IN ('pending', 'retryable')",
                [operation_id],
            )
            .map_err(map_sqlite_error)?;
        let revision = if changed == 0 {
            prior_revision
        } else {
            bump_revision(&transaction)?
        };
        let intents = load_operation_intents(&transaction, operation_id)?;
        let operation = DeliveryReducer::operation_snapshot(intents)?;
        let diagnostic = (changed > 0 && self.diagnostics.enabled())
            .then(|| delivery_diagnostic(&transaction, &operation, "cancel", false));
        transaction.commit().map_err(map_sqlite_error)?;
        if let Some(mut diagnostic) = diagnostic {
            diagnostic.revision = Some(revision);
            self.record_diagnostic(diagnostic);
        }
        Ok(AccountOperationResult {
            operation,
            revision,
            inserted: false,
        })
    }

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

    /// Reconcile authenticated discovery and persist the complete result.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn reconcile_relay_catalog(
        &mut self,
        discovered: Vec<DnsRelayRecord>,
        refreshed_at: i64,
        fallback_url: String,
        dns_authenticated: bool,
    ) -> Result<StoredRelayCatalogPlan, SoftchatError> {
        let plan = RelayCatalogReducer::reconcile(
            self.relay_catalog()?,
            discovered,
            refreshed_at,
            fallback_url,
            dns_authenticated,
        )?;
        let revision = self.persist_relay_catalog(&plan.entries, refreshed_at)?;
        Ok(StoredRelayCatalogPlan { plan, revision })
    }

    /// Add or reactivate one custom relay and persist the catalog.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn add_custom_relay(
        &mut self,
        relay_url: String,
        created_at: i64,
    ) -> Result<RelayCatalogMutation, SoftchatError> {
        let entries =
            RelayCatalogReducer::add_custom(self.relay_catalog()?, relay_url, created_at)?;
        let revision = self.persist_relay_catalog(&entries, created_at)?;
        Ok(RelayCatalogMutation { entries, revision })
    }

    /// Exclude or remove one relay and persist the catalog.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn exclude_relay(
        &mut self,
        relay_url: String,
        now: i64,
    ) -> Result<RelayCatalogMutation, SoftchatError> {
        validate_timestamp(now).map_err(|_| SoftchatError::InvalidRelayCatalog)?;
        let entries = RelayCatalogReducer::exclude(self.relay_catalog()?, relay_url)?;
        let revision = self.persist_relay_catalog(&entries, now)?;
        Ok(RelayCatalogMutation { entries, revision })
    }

    /// Restore automatic relays and persist the catalog.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn restore_automatic_relays(
        &mut self,
        now: i64,
    ) -> Result<RelayCatalogMutation, SoftchatError> {
        validate_timestamp(now).map_err(|_| SoftchatError::InvalidRelayCatalog)?;
        let entries = RelayCatalogReducer::restore_automatic(self.relay_catalog()?)?;
        let revision = self.persist_relay_catalog(&entries, now)?;
        Ok(RelayCatalogMutation { entries, revision })
    }

    /// Select and persist one active relay.
    ///
    /// # Errors
    ///
    /// Returns a stable relay-catalog or persistence error.
    pub fn set_active_relay(
        &mut self,
        relay_url: String,
        now: i64,
    ) -> Result<RelayCatalogMutation, SoftchatError> {
        validate_timestamp(now).map_err(|_| SoftchatError::InvalidRelayCatalog)?;
        let entries = RelayCatalogReducer::set_active(self.relay_catalog()?, relay_url)?;
        let revision = self.persist_relay_catalog(&entries, now)?;
        Ok(RelayCatalogMutation { entries, revision })
    }

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

    /// Atomically rotate away from a failed relay and rebind unfinished intent.
    ///
    /// # Errors
    ///
    /// Returns a stable catalog, delivery-state, or persistence error.
    pub fn rotate_failed_relay(
        &mut self,
        failed_url: String,
        now: i64,
    ) -> Result<RelayFailoverMutation, SoftchatError> {
        validate_timestamp(now).map_err(|_| SoftchatError::InvalidRelayCatalog)?;
        let failed = crate::parse_relay_endpoint(&failed_url)
            .map_err(|_| SoftchatError::InvalidRelayCatalog)?
            .configured_url;
        let current = self.relay_catalog()?;
        let replacement =
            RelayCatalogReducer::failover_candidates(current.clone(), failed.clone())?
                .into_iter()
                .next()
                .ok_or(SoftchatError::InvalidRelayCatalog)?;
        let entries = RelayCatalogReducer::set_active(current, replacement.clone())?;
        let transaction = begin_write_scope(&mut self.connection)?;
        replace_relay_catalog(&transaction, &entries, now)?;

        let mut statement = transaction
            .prepare(
                "SELECT intent_id, operation_id, event_copy_id
                 FROM relay_delivery_intents
                 WHERE relay_url = ?1
                   AND state IN ('pending', 'retryable', 'rejected')
                   AND lease_id = ''
                 ORDER BY intent_id ASC",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([failed], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                ))
            })
            .map_err(map_sqlite_error)?
            .collect::<Result<Vec<_>, _>>()
            .map_err(map_sqlite_error)?;
        drop(statement);
        for (old_id, operation_id, event_copy_id) in &rows {
            let new_id = stable_intent_id(operation_id, event_copy_id, &replacement);
            transaction
                .execute(
                    "UPDATE relay_delivery_intents
                     SET intent_id = ?1, relay_url = ?2, state = 'retryable',
                         lease_id = '', lease_expires_at = 0,
                         last_category = 'relay_rotated'
                     WHERE intent_id = ?3",
                    params![new_id, replacement, old_id],
                )
                .map_err(map_sqlite_error)?;
        }
        let revision = bump_revision(&transaction)?;
        transaction.commit()?;
        Ok(RelayFailoverMutation {
            entries,
            active_relay_url: replacement,
            rebound_intent_count: u32::try_from(rows.len())
                .map_err(|_| SoftchatError::InvalidDeliveryState)?,
            revision,
        })
    }

    /// Return a newest-first bounded conversation page.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for an invalid limit or
    /// inconsistent projection data.
    pub fn list_conversations(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<ConversationSummary>, SoftchatError> {
        self.list_conversation_views(limit, offset).map(|views| {
            views
                .into_iter()
                .map(|view| view.summary)
                .collect::<Vec<_>>()
        })
    }

    /// Return complete newest-first conversation-list rows in one bounded call.
    ///
    /// This avoids one Android-to-Rust transition per conversation while
    /// keeping exact event values and effective edit/deletion state in Rust.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for invalid bounds or
    /// inconsistent projection data.
    pub fn list_conversation_views(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<ConversationView>, SoftchatError> {
        validate_page(limit)?;
        let mut statement = self
            .connection
            .prepare(
                "WITH live_messages AS (
                   SELECT p.*,
                          ROW_NUMBER() OVER (
                            PARTITION BY p.conversation_id
                            ORDER BY p.created_at DESC, p.logical_event_id ASC
                          ) AS row_number,
                          COUNT(*) OVER (
                            PARTITION BY p.conversation_id
                          ) AS message_count
                   FROM projections p
                   WHERE 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
                     )
                 ),
                 page AS (
                   SELECT * FROM live_messages
                   WHERE row_number = 1
                   ORDER BY created_at DESC, conversation_id ASC
                   LIMIT ?1 OFFSET ?2
                 ),
                 latest_subjects AS (
                   SELECT subject.*,
                          ROW_NUMBER() OVER (
                            PARTITION BY subject.conversation_id
                            ORDER BY subject.created_at DESC,
                                     subject.logical_event_id ASC
                          ) AS row_number
                   FROM projections subject
                   JOIN page
                     ON page.conversation_id = subject.conversation_id
                   WHERE subject.kind = 'subject' AND subject.ephemeral = 0
                 )
                 SELECT
                   p.conversation_id,
                   p.message_count,
                   p.logical_event_id,
                   p.author_public_key,
                   p.created_at,
                   p.content,
                   se.canonical_json,
                   ar.canonical_json,
                   (
                     SELECT group_concat(public_key, ',')
                     FROM (
                       SELECT public_key
                       FROM projection_participants
                       WHERE logical_event_id = p.logical_event_id
                       ORDER BY public_key ASC
                     )
                   ),
                   (
                     SELECT group_concat(target_event_id, ',')
                     FROM (
                       SELECT target_event_id
                       FROM projection_targets
                       WHERE logical_event_id = p.logical_event_id
                       ORDER BY position ASC
                     )
                   ),
                   (
                     SELECT effective.content
                     FROM effective_message_edits effective
                     WHERE effective.message_id = p.logical_event_id
                   ),
                   (
                     SELECT COUNT(*)
                     FROM projection_targets rt
                     JOIN projections reply
                       ON reply.logical_event_id = rt.logical_event_id
                     WHERE rt.target_event_id = p.logical_event_id
                       AND reply.kind = 'chat_message'
                       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 = reply.logical_event_id
                           AND deletion.kind = 'deletion'
                           AND deletion.author_public_key = reply.author_public_key
                     )
                   ),
                   subject.kind,
                   subject.logical_event_id,
                   subject.author_public_key,
                   subject.created_at,
                   subject.conversation_id,
                   subject.content,
                   subject_event.canonical_json,
                   subject_rumor.canonical_json,
                   (
                     SELECT group_concat(
                       spp.public_key,
                       ',' ORDER BY spp.public_key ASC
                     )
                     FROM projection_participants spp
                     WHERE spp.logical_event_id = subject.logical_event_id
                   ),
                   (
                     SELECT group_concat(
                       spt.target_event_id,
                       ',' ORDER BY spt.position ASC
                     )
                     FROM projection_targets spt
                     WHERE spt.logical_event_id = subject.logical_event_id
                   )
                 FROM page p
                 JOIN signed_events se ON se.event_id = p.outer_event_id
                 LEFT JOIN authenticated_rumors ar
                   ON ar.rumor_id = p.logical_event_id
                 LEFT JOIN latest_subjects subject
                   ON subject.conversation_id = p.conversation_id
                  AND subject.row_number = 1
                 LEFT JOIN signed_events subject_event
                   ON subject_event.event_id = subject.outer_event_id
                 LEFT JOIN authenticated_rumors subject_rumor
                   ON subject_rumor.rumor_id = subject.logical_event_id
                 ORDER BY p.created_at DESC, p.conversation_id ASC",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params![limit, offset], |row| {
                let subject_id = row.get::<_, Option<String>>(13)?;
                let subject = if let Some(logical_event_id) = subject_id {
                    Some(RawProjection {
                        stored_kind: row.get(12)?,
                        logical_event_id,
                        author_public_key: row.get(14)?,
                        created_at: row.get(15)?,
                        conversation_id: row.get(16)?,
                        content: row.get(17)?,
                        outer_json: row.get(18)?,
                        rumor_json: row.get(19)?,
                        participants_csv: row.get(20)?,
                        targets_csv: row.get(21)?,
                    })
                } else {
                    None
                };
                Ok(RawConversationView {
                    conversation_id: row.get(0)?,
                    message_count: row.get(1)?,
                    logical_event_id: row.get(2)?,
                    author_public_key: row.get(3)?,
                    created_at: row.get(4)?,
                    content: row.get(5)?,
                    outer_json: row.get(6)?,
                    rumor_json: row.get(7)?,
                    participants_csv: row.get(8)?,
                    targets_csv: row.get(9)?,
                    edited_content: row.get(10)?,
                    reply_count: row.get(11)?,
                    subject,
                })
            })
            .map_err(map_sqlite_error)?;
        let mut views = Vec::new();
        for row in rows {
            let row = row.map_err(map_sqlite_error)?;
            let participants = parse_hex_list(row.participants_csv, true)?;
            let target_event_ids = parse_hex_list(row.targets_csv, false)?;
            let outer = SignedNostrEvent::from_json(&row.outer_json)?;
            let rumor = row
                .rumor_json
                .map(|json| {
                    serde_json::from_str::<RumorEvent>(&json)
                        .map_err(|_| SoftchatError::InvalidPersistenceResult)
                })
                .transpose()?;
            let projection = AccountProjection {
                kind: ProjectionKind::ChatMessage,
                logical_event_id: row.logical_event_id,
                outer_event: SignedEvent::from(&outer),
                rumor,
                author_public_key: row.author_public_key,
                created_at: row.created_at,
                conversation_id: row.conversation_id.clone(),
                participants: participants.clone(),
                target_event_ids,
                content: row.content.clone(),
            };
            let edited = row.edited_content.is_some();
            let effective_content = row.edited_content.unwrap_or(row.content);
            let summary = ConversationSummary {
                conversation_id: row.conversation_id,
                participants,
                last_message: effective_content.clone(),
                last_created_at: row.created_at,
                message_count: row.message_count,
            };
            let subject = row.subject.map(projection_from_raw).transpose()?;
            views.push(ConversationView {
                summary,
                last_message: AccountEventNode {
                    projection,
                    effective_content,
                    deleted: false,
                    edited,
                    reply_count: row.reply_count,
                },
                subject,
            });
        }
        Ok(views)
    }

    /// Return a newest-first bounded message page for one conversation.
    ///
    /// Passing `before_created_at` excludes messages at or after that timestamp.
    /// Callers that need lossless same-timestamp paging should use the returned
    /// event ID as an application cursor in a later SDK revision.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for malformed input.
    pub fn list_messages(
        &self,
        conversation_id: &str,
        before_created_at: Option<i64>,
        limit: u32,
    ) -> Result<Vec<AccountMessage>, SoftchatError> {
        if conversation_id.len() != 64
            || !conversation_id.bytes().all(|byte| byte.is_ascii_hexdigit())
        {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        validate_page(limit)?;
        if let Some(timestamp) = before_created_at {
            validate_timestamp(timestamp)?;
        }
        let mut statement = self
            .connection
            .prepare(
                "SELECT
                   p.logical_event_id,
                   p.author_public_key,
                   p.created_at,
                   p.content,
                   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
                   ),
                   (
                     SELECT effective.content
                     FROM effective_message_edits effective
                     WHERE effective.message_id = p.logical_event_id
                   ),
                   COALESCE(
                     (
                       SELECT target_event_id
                       FROM projection_targets
                       WHERE logical_event_id = p.logical_event_id
                       ORDER BY position ASC
                       LIMIT 1
                     ),
                     ''
                   )
                 FROM projections p
                 WHERE p.kind = 'chat_message' AND p.ephemeral = 0
                   AND p.conversation_id = ?1
                   AND (?2 IS NULL OR p.created_at < ?2)
                 ORDER BY p.created_at DESC, p.logical_event_id ASC
                 LIMIT ?3",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params![conversation_id, before_created_at, limit], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, i64>(2)?,
                    row.get::<_, String>(3)?,
                    row.get::<_, i64>(4)? != 0,
                    row.get::<_, Option<String>>(5)?,
                    row.get::<_, String>(6)?,
                ))
            })
            .map_err(map_sqlite_error)?;
        let mut messages = Vec::new();
        for row in rows {
            let (
                event_id,
                author_public_key,
                created_at,
                original_content,
                deleted,
                edited_content,
                reply_to_event_id,
            ) = row.map_err(map_sqlite_error)?;
            messages.push(AccountMessage {
                event_id,
                conversation_id: conversation_id.to_owned(),
                author_public_key,
                created_at,
                content: edited_content.clone().unwrap_or(original_content),
                deleted,
                edited: edited_content.is_some(),
                reply_to_event_id,
            });
        }
        Ok(messages)
    }

    /// Return a bounded newest-first authenticated projection page.
    ///
    /// Empty string filters are ignored. This one query surface supports
    /// subjects, reactions, deletions, edits, profiles, contacts, application
    /// data, and detail screens without duplicating protocol interpretation in
    /// Android.
    ///
    /// # Errors
    ///
    /// Returns a stable account, event, or persistence error for invalid
    /// filters, limits, or stored truth.
    #[allow(clippy::too_many_arguments)]
    pub fn list_projections(
        &self,
        kind: Option<ProjectionKind>,
        logical_event_id: String,
        conversation_id: String,
        target_event_id: String,
        author_public_key: String,
        before_created_at: Option<i64>,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<AccountProjection>, SoftchatError> {
        self.query_event_nodes(
            kind,
            logical_event_id,
            conversation_id,
            target_event_id,
            author_public_key,
            before_created_at,
            limit,
            offset,
        )
        .map(|nodes| {
            nodes
                .into_iter()
                .map(|node| node.projection)
                .collect::<Vec<_>>()
        })
    }

    /// Return a bounded page of authenticated projections with use-case state.
    ///
    /// Filtering and edit/deletion/reply semantics execute in the same Rust
    /// database owner as ingestion. Android receives complete immutable values
    /// and never issues SQL or reinterprets protocol tags.
    ///
    /// # Errors
    ///
    /// Returns a stable account, event, or persistence error for invalid
    /// filters, limits, or stored truth.
    #[allow(clippy::too_many_arguments)]
    pub fn list_event_nodes(
        &self,
        kind: Option<ProjectionKind>,
        logical_event_id: String,
        conversation_id: String,
        target_event_id: String,
        author_public_key: String,
        before_created_at: Option<i64>,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<AccountEventNode>, SoftchatError> {
        self.query_event_nodes(
            kind,
            logical_event_id,
            conversation_id,
            target_event_id,
            author_public_key,
            before_created_at,
            limit,
            offset,
        )
    }

    #[allow(clippy::too_many_arguments)]
    fn query_event_nodes(
        &self,
        kind: Option<ProjectionKind>,
        logical_event_id: String,
        conversation_id: String,
        target_event_id: String,
        author_public_key: String,
        before_created_at: Option<i64>,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<AccountEventNode>, SoftchatError> {
        validate_page(limit)?;
        if !logical_event_id.is_empty() {
            crate::NostrEventId::from_hex(&logical_event_id)?;
        }
        if !conversation_id.is_empty()
            && (conversation_id.len() != 64
                || !conversation_id.bytes().all(|byte| byte.is_ascii_hexdigit()))
        {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        if !target_event_id.is_empty() {
            crate::NostrEventId::from_hex(&target_event_id)?;
        }
        if !author_public_key.is_empty() {
            crate::NostrPublicKey::from_hex(&author_public_key)?;
        }
        if let Some(value) = before_created_at {
            validate_timestamp(value)?;
        }
        let kind_name = kind.map(projection_kind_name);
        let mut statement = self
            .connection
            .prepare(
                "SELECT
                   p.kind,
                   p.logical_event_id,
                   p.author_public_key,
                   p.created_at,
                   p.conversation_id,
                   p.content,
                   se.canonical_json,
                   ar.canonical_json,
                   (
                     SELECT group_concat(
                       pp.public_key,
                       ',' ORDER BY pp.public_key ASC
                     )
                     FROM projection_participants pp
                     WHERE pp.logical_event_id = p.logical_event_id
                   ),
                   (
                     SELECT group_concat(
                       pt.target_event_id,
                       ',' ORDER BY pt.position ASC
                     )
                     FROM projection_targets pt
                     WHERE pt.logical_event_id = p.logical_event_id
                   ),
                   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
                   ),
                   (
                     SELECT effective.content
                     FROM effective_message_edits effective
                     WHERE effective.message_id = p.logical_event_id
                   ),
                   (
                     SELECT COUNT(*)
                     FROM projection_targets rt
                     JOIN projections reply
                       ON reply.logical_event_id = rt.logical_event_id
                     WHERE rt.target_event_id = p.logical_event_id
                       AND reply.kind = 'chat_message'
                       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 = reply.logical_event_id
                           AND deletion.kind = 'deletion'
                           AND deletion.author_public_key = reply.author_public_key
                       )
                   )
                 FROM projections p
                 JOIN signed_events se ON se.event_id = p.outer_event_id
                 LEFT JOIN authenticated_rumors ar
                   ON ar.rumor_id = p.logical_event_id
                 WHERE (?1 IS NULL OR p.kind = ?1)
                   AND (?2 = '' OR p.logical_event_id = ?2)
                   AND (?3 = '' OR p.conversation_id = ?3)
                   AND (?4 = '' OR EXISTS (
                     SELECT 1 FROM projection_targets pt
                     WHERE pt.logical_event_id = p.logical_event_id
                       AND pt.target_event_id = ?4
                   ))
                   AND (?5 = '' OR p.author_public_key = ?5)
                   AND (?6 IS NULL OR p.created_at <= ?6)
                 ORDER BY p.created_at DESC, p.logical_event_id ASC
                 LIMIT ?7 OFFSET ?8",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(
                params![
                    kind_name,
                    logical_event_id,
                    conversation_id,
                    target_event_id,
                    author_public_key,
                    before_created_at,
                    limit,
                    offset,
                ],
                |row| {
                    Ok(RawEventNode {
                        stored_kind: row.get(0)?,
                        logical_event_id: row.get(1)?,
                        author_public_key: row.get(2)?,
                        created_at: row.get(3)?,
                        conversation_id: row.get(4)?,
                        content: row.get(5)?,
                        outer_json: row.get(6)?,
                        rumor_json: row.get(7)?,
                        participants_csv: row.get(8)?,
                        targets_csv: row.get(9)?,
                        deleted: row.get::<_, i64>(10)? != 0,
                        edited_content: row.get(11)?,
                        reply_count: row.get(12)?,
                    })
                },
            )
            .map_err(map_sqlite_error)?;
        let mut nodes = Vec::new();
        for row in rows {
            nodes.push(event_node_from_raw(row.map_err(map_sqlite_error)?)?);
        }
        Ok(nodes)
    }

    /// Return bounded canonical public keys known to account state.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for an invalid limit or
    /// inconsistent stored truth.
    pub fn list_known_public_keys(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<String>, SoftchatError> {
        validate_page(limit)?;
        let mut statement = self
            .connection
            .prepare(
                "SELECT public_key FROM (
                   SELECT author_public_key AS public_key FROM projections
                   UNION
                   SELECT public_key FROM projection_participants
                   UNION
                   SELECT public_key FROM local_contacts
                 )
                 ORDER BY public_key ASC
                 LIMIT ?1 OFFSET ?2",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params![limit, offset], |row| row.get::<_, String>(0))
            .map_err(map_sqlite_error)?;
        let mut public_keys = Vec::new();
        for row in rows {
            let public_key = row.map_err(map_sqlite_error)?;
            crate::NostrPublicKey::from_hex(&public_key)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            public_keys.push(public_key);
        }
        Ok(public_keys)
    }

    /// Insert or replace one bounded Android-local contact.
    ///
    /// This does not create or impersonate authenticated kind-0 metadata.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for an invalid key, name,
    /// timestamp, or SQLite failure.
    pub fn put_local_contact(
        &mut self,
        contact: LocalContactRecord,
    ) -> Result<AccountMutationResult, SoftchatError> {
        validate_local_contact(&contact)?;
        let diagnostic_changed = self.local_state_differs(
            "SELECT NOT EXISTS(SELECT 1 FROM local_contacts
             WHERE public_key = ?1 AND name IS ?2 AND updated_at_millis = ?3)",
            params![contact.public_key, contact.name, contact.updated_at_millis],
        );
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO local_contacts(public_key, name, updated_at_millis)
                 VALUES (?1, ?2, ?3)
                 ON CONFLICT(public_key) DO UPDATE SET
                   name = excluded.name,
                   updated_at_millis = excluded.updated_at_millis",
                params![contact.public_key, contact.name, contact.updated_at_millis],
            )
            .map_err(map_sqlite_error)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        self.record_local_change(diagnostic_changed, revision);
        Ok(AccountMutationResult { revision })
    }

    /// Remove one Android-local contact idempotently.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for an invalid key or
    /// SQLite failure.
    pub fn remove_local_contact(
        &mut self,
        public_key: String,
    ) -> Result<AccountMutationResult, SoftchatError> {
        crate::NostrPublicKey::from_hex(&public_key)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        let changed = transaction
            .execute(
                "DELETE FROM local_contacts WHERE public_key = ?1",
                [public_key],
            )
            .map_err(map_sqlite_error)?;
        let revision = if changed == 0 {
            meta_integer(&transaction, "revision")?
        } else {
            bump_revision(&transaction)?
        };
        transaction.commit().map_err(map_sqlite_error)?;
        self.record_local_change(changed > 0, revision);
        Ok(AccountMutationResult { revision })
    }

    /// Return a bounded stable page of Android-local contacts.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for invalid bounds or
    /// malformed stored state.
    pub fn local_contacts(
        &self,
        limit: u32,
        offset: u32,
    ) -> Result<Vec<LocalContactRecord>, SoftchatError> {
        validate_page(limit)?;
        let mut statement = self
            .connection
            .prepare(
                "SELECT public_key, name, updated_at_millis
                 FROM local_contacts
                 ORDER BY public_key ASC
                 LIMIT ?1 OFFSET ?2",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(params![limit, offset], |row| {
                Ok(LocalContactRecord {
                    public_key: row.get(0)?,
                    name: row.get(1)?,
                    updated_at_millis: row.get(2)?,
                })
            })
            .map_err(map_sqlite_error)?;
        let mut contacts = Vec::new();
        for row in rows {
            let contact = row.map_err(map_sqlite_error)?;
            validate_local_contact(&contact)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            contacts.push(contact);
        }
        Ok(contacts)
    }

    /// Atomically replace the complete bounded custom-emoji/sticker cache.
    ///
    /// The cache is use-case state rather than protocol truth, but remains in
    /// the Rust-owned account database so Android never owns a second SQLite
    /// schema.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for invalid names, URLs,
    /// emoji bounds, duplicates, or SQLite failure.
    pub fn replace_stickers(
        &mut self,
        stickers: Vec<StickerRecord>,
    ) -> Result<AccountMutationResult, SoftchatError> {
        if stickers.len() > MAX_QUERY_PAGE as usize {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        let mut names = BTreeSet::new();
        for sticker in &stickers {
            validate_sticker(sticker)?;
            if !names.insert(sticker.name.as_str()) {
                return Err(SoftchatError::InvalidAccountOperation);
            }
        }
        let diagnostic_changed = self.diagnostics.debug_enabled()
            && self.stickers().is_ok_and(|stored| {
                stored.len() != stickers.len() || stored.iter().any(|item| !stickers.contains(item))
            });
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        transaction
            .execute("DELETE FROM sticker_cache", [])
            .map_err(map_sqlite_error)?;
        for sticker in stickers {
            let emojis = serde_json::to_string(&sticker.associated_emojis)
                .map_err(|_| SoftchatError::InternalFailure)?;
            transaction
                .execute(
                    "INSERT INTO sticker_cache(name, url, associated_emojis_json)
                     VALUES (?1, ?2, ?3)",
                    params![sticker.name, sticker.url, emojis],
                )
                .map_err(map_sqlite_error)?;
        }
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        self.record_local_change(diagnostic_changed, revision);
        Ok(AccountMutationResult { revision })
    }

    /// Return the complete bounded sticker cache in stable name order.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence error for malformed stored state.
    pub fn stickers(&self) -> Result<Vec<StickerRecord>, SoftchatError> {
        let mut statement = self
            .connection
            .prepare(
                "SELECT name, url, associated_emojis_json
                 FROM sticker_cache ORDER BY name ASC LIMIT 200",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                ))
            })
            .map_err(map_sqlite_error)?;
        let mut stickers = Vec::new();
        for row in rows {
            let (name, url, emojis_json) = row.map_err(map_sqlite_error)?;
            let associated_emojis = serde_json::from_str::<Vec<String>>(&emojis_json)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            let sticker = StickerRecord {
                name,
                url,
                associated_emojis,
            };
            validate_sticker(&sticker).map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            stickers.push(sticker);
        }
        Ok(stickers)
    }

    /// Insert or replace one bounded link-preview cache result.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for invalid URLs, text
    /// bounds, timestamps, or SQLite failure.
    pub fn put_link_preview(
        &mut self,
        preview: LinkPreviewRecord,
    ) -> Result<AccountMutationResult, SoftchatError> {
        validate_link_preview(&preview)?;
        let diagnostic_changed = self.local_state_differs(
            "SELECT NOT EXISTS(SELECT 1 FROM link_preview_cache
             WHERE url = ?1 AND title IS ?2 AND description IS ?3
               AND image_url IS ?4 AND updated_at_millis = ?5)",
            params![
                preview.url,
                preview.title,
                preview.description,
                preview.image_url,
                preview.updated_at_millis
            ],
        );
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO link_preview_cache
                 (url, title, description, image_url, updated_at_millis)
                 VALUES (?1, ?2, ?3, ?4, ?5)
                 ON CONFLICT(url) DO UPDATE SET
                   title = excluded.title,
                   description = excluded.description,
                   image_url = excluded.image_url,
                   updated_at_millis = excluded.updated_at_millis",
                params![
                    preview.url,
                    preview.title,
                    preview.description,
                    preview.image_url,
                    preview.updated_at_millis,
                ],
            )
            .map_err(map_sqlite_error)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        self.record_local_change(diagnostic_changed, revision);
        Ok(AccountMutationResult { revision })
    }

    /// Return one cached link preview by its exact validated URL.
    ///
    /// # Errors
    ///
    /// Returns a stable account or persistence error for an invalid URL or
    /// malformed stored result.
    pub fn link_preview(&self, url: String) -> Result<Option<LinkPreviewRecord>, SoftchatError> {
        validate_http_url(&url, false)?;
        let preview = self
            .connection
            .query_row(
                "SELECT url, title, description, image_url, updated_at_millis
                 FROM link_preview_cache WHERE url = ?1",
                [url],
                |row| {
                    Ok(LinkPreviewRecord {
                        url: row.get(0)?,
                        title: row.get(1)?,
                        description: row.get(2)?,
                        image_url: row.get(3)?,
                        updated_at_millis: row.get(4)?,
                    })
                },
            )
            .optional()
            .map_err(map_sqlite_error)?;
        if let Some(value) = &preview {
            validate_link_preview(value).map_err(|_| SoftchatError::InvalidPersistenceResult)?;
        }
        Ok(preview)
    }

    /// Clear every account-owned row while retaining binding and revision.
    ///
    /// Relay catalog and account settings are recreated at safe defaults in
    /// the same transaction. The authoritative table inventory includes the
    /// standalone FTS table and transport-result journal.
    ///
    /// # Errors
    ///
    /// Returns a stable persistence error if the atomic clear fails.
    pub fn clear(&mut self) -> Result<AccountMutationResult, SoftchatError> {
        let transaction = begin_write_scope(&mut self.connection)?;
        for table in ACCOUNT_CLEAR_TABLES {
            transaction
                .execute(&format!("DELETE FROM {table}"), [])
                .map_err(map_sqlite_error)?;
        }
        transaction
            .execute("DELETE FROM relay_catalog", [])
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO relay_catalog
                 (relay_url, created_at, source, active, is_dns_discovered,
                  excluded, priority, weight, expires_at, updated_at)
                 VALUES (?1, 0, 'automatic', 1, 1, 0, 0, 0, 3600, 0)",
                [DEFAULT_RELAY_URL],
            )
            .map_err(map_sqlite_error)?;
        transaction
            .execute("DELETE FROM account_settings", [])
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO account_settings
                 (singleton, schema_version, canonical_json, updated_at)
                 VALUES (1, 1, '{}', 0)",
                [],
            )
            .map_err(map_sqlite_error)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit()?;
        Ok(AccountMutationResult { revision })
    }

    /// Return exact canonical signed-event JSON by outer event ID.
    ///
    /// # Errors
    ///
    /// Returns a stable event or persistence error when the ID is malformed,
    /// unknown, or unreadable.
    pub fn event_json(&self, event_id: &str) -> Result<String, SoftchatError> {
        crate::NostrEventId::from_hex(event_id)?;
        self.connection
            .query_row(
                "SELECT canonical_json FROM signed_events WHERE event_id = ?1",
                [event_id],
                |row| row.get(0),
            )
            .optional()
            .map_err(map_sqlite_error)?
            .ok_or(SoftchatError::InvalidEventId)
    }

    /// Return exact event JSON for a bounded unique ID set.
    ///
    /// This executes synchronization resend actions without exposing a
    /// database adapter to Android.
    ///
    /// # Errors
    ///
    /// Returns a stable event, account, or persistence error for malformed,
    /// duplicate, excessive, or unknown IDs.
    pub fn event_jsons(
        &self,
        event_ids: Vec<String>,
    ) -> Result<Vec<StoredEventJson>, SoftchatError> {
        if event_ids.is_empty() || event_ids.len() > crate::SYNC_EVENT_REQUEST_CHUNK {
            return Err(SoftchatError::InvalidAccountOperation);
        }
        let mut seen = BTreeSet::new();
        let mut events = Vec::with_capacity(event_ids.len());
        for event_id in event_ids {
            crate::NostrEventId::from_hex(&event_id)?;
            if !seen.insert(event_id.clone()) {
                return Err(SoftchatError::InvalidAccountOperation);
            }
            events.push(StoredEventJson {
                event_json: self.event_json(&event_id)?,
                event_id,
            });
        }
        Ok(events)
    }

    /// Load one bounded authenticated fingerprint snapshot for synchronization.
    ///
    /// # Errors
    ///
    /// Returns a stable sync or persistence error for invalid ranges, limits,
    /// or database state.
    pub fn sync_fingerprints(
        &self,
        since_timestamp: Option<i64>,
        until_timestamp: i64,
        limit: u32,
    ) -> Result<Vec<NegentropyItem>, SoftchatError> {
        validate_timestamp(until_timestamp).map_err(|_| SoftchatError::InvalidSyncState)?;
        if limit == 0
            || limit > crate::SYNC_MAX_PAGE_LIMIT
            || since_timestamp.is_some_and(|since| since < 0 || since > until_timestamp)
        {
            return Err(SoftchatError::InvalidSyncState);
        }
        let mut statement = self
            .connection
            .prepare(
                "SELECT fingerprints.event_id, fingerprints.created_at
                 FROM event_fingerprints AS fingerprints
                 JOIN signed_events AS events ON events.event_id = fingerprints.event_id
                 WHERE (?1 IS NULL OR fingerprints.created_at >= ?1)
                   AND fingerprints.created_at <= ?2
                   AND length(CAST(events.canonical_json AS BLOB)) <= ?4
                 ORDER BY fingerprints.created_at DESC, fingerprints.event_id ASC
                 LIMIT ?3",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map(
                params![
                    since_timestamp,
                    until_timestamp,
                    limit,
                    i64::try_from(crate::session::MAX_ACCOUNT_EVENT_JSON_BYTES)
                        .map_err(|_| SoftchatError::InternalFailure)?
                ],
                |row| {
                    Ok(NegentropyItem {
                        event_id: row.get(0)?,
                        created_at: row.get(1)?,
                    })
                },
            )
            .map_err(map_sqlite_error)?;
        let mut items = Vec::new();
        for row in rows {
            items.push(row.map_err(map_sqlite_error)?);
        }
        Ok(items)
    }

    /// Return a previously committed synchronization checkpoint.
    ///
    /// # Errors
    ///
    /// Returns a stable sync or persistence error for an invalid ID or
    /// database state.
    pub fn sync_checkpoint(&self, sync_id: &str) -> Result<Option<i64>, SoftchatError> {
        validate_sync_id(sync_id)?;
        self.connection
            .query_row(
                "SELECT checkpoint_timestamp FROM sync_checkpoints WHERE sync_id = ?1",
                [sync_id],
                |row| row.get(0),
            )
            .optional()
            .map_err(map_sqlite_error)
    }

    /// Atomically persist one synchronization checkpoint.
    ///
    /// # Errors
    ///
    /// Returns a stable sync or persistence error for invalid input or a failed
    /// transaction.
    pub fn save_sync_checkpoint(
        &mut self,
        sync_id: &str,
        checkpoint_timestamp: i64,
        now: i64,
    ) -> Result<AccountMutationResult, SoftchatError> {
        validate_sync_id(sync_id)?;
        validate_timestamp(checkpoint_timestamp).map_err(|_| SoftchatError::InvalidSyncState)?;
        validate_timestamp(now).map_err(|_| SoftchatError::InvalidSyncState)?;
        let transaction = begin_write_scope(&mut self.connection)?;
        transaction
            .execute(
                "INSERT INTO sync_checkpoints(sync_id, checkpoint_timestamp, updated_at)
                 VALUES (?1, ?2, ?3)
                 ON CONFLICT(sync_id) DO UPDATE SET
                   checkpoint_timestamp = excluded.checkpoint_timestamp,
                   updated_at = excluded.updated_at",
                params![sync_id, checkpoint_timestamp, now],
            )
            .map_err(map_sqlite_error)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit()?;
        Ok(AccountMutationResult { revision })
    }

    /// Remove every durable synchronization checkpoint and wake observers.
    ///
    /// Authenticated event truth is retained. The next synchronization run
    /// therefore reconciles the complete retained history instead of
    /// downloading an unverified replacement database.
    ///
    /// # Errors
    ///
    /// Returns a stable timestamp or persistence error.
    pub fn reset_sync(&mut self, now: i64) -> Result<AccountMutationResult, SoftchatError> {
        validate_timestamp(now).map_err(|_| SoftchatError::InvalidSyncState)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        transaction
            .execute("DELETE FROM sync_checkpoints", [])
            .map_err(map_sqlite_error)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        Ok(AccountMutationResult { revision })
    }

    fn persist_operation(
        &mut self,
        identity: &LocalIdentity,
        prepared: PreparedOutgoingOperation,
        sender_public_key: &str,
        now: i64,
        command_hash: &str,
    ) -> Result<AccountOperationResult, SoftchatError> {
        validate_timestamp(now)?;
        let mut queued = self
            .diagnostics
            .enabled()
            .then(|| Diagnostic::for_subject(AccountLogSubject::Unknown, true));
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        persist_prepared_operation(&transaction, &prepared, now, 0, command_hash, "")?;
        if !sender_public_key.is_empty() {
            let sender_copy = prepared
                .event_copies
                .iter()
                .find(|copy| copy.recipient_public_key == sender_public_key)
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let local_batch = AccountEngine::prepare_incoming(
                identity,
                self.account_id.clone(),
                vec![sender_copy.event.clone()],
            )?;
            if queued.is_some()
                && let Some(event) = local_batch.events.first()
            {
                queued = Some(Diagnostic::queued_from(event));
            }
            let committed = ingest_prepared(&transaction, &local_batch, now, false)?;
            AccountEngine::validate_ingestion_receipt(&local_batch, &committed.receipt)?;
        }
        let intents = load_operation_intents(&transaction, &prepared.operation_id)?;
        let operation = DeliveryReducer::operation_snapshot(intents)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        if let Some(mut queued) = queued {
            queued.revision = Some(revision);
            self.record_diagnostic(
                queued.count(Counter::TotalIntents, u64::from(operation.total_intents)),
            );
        }
        Ok(AccountOperationResult {
            operation,
            revision,
            inserted: true,
        })
    }

    #[allow(clippy::too_many_arguments)]
    fn persist_product_operation(
        &mut self,
        identity: &LocalIdentity,
        prepared: PreparedOutgoingOperation,
        sender_public_key: &str,
        now: i64,
        command_hash: &str,
        result_message_id: &str,
        draft_to_consume: Option<(&str, &crate::account_product::ProductDraftMatch)>,
    ) -> Result<ProductAccountOperationResult, SoftchatError> {
        validate_timestamp(now)?;
        let mut queued = self
            .diagnostics
            .enabled()
            .then(|| Diagnostic::for_subject(AccountLogSubject::Unknown, true));
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        persist_prepared_operation(
            &transaction,
            &prepared,
            now,
            0,
            command_hash,
            result_message_id,
        )?;
        if !sender_public_key.is_empty() {
            let sender_copy = prepared
                .event_copies
                .iter()
                .find(|copy| copy.recipient_public_key == sender_public_key)
                .ok_or(SoftchatError::InvalidAccountOperation)?;
            let local_batch = AccountEngine::prepare_incoming(
                identity,
                self.account_id.clone(),
                vec![sender_copy.event.clone()],
            )?;
            if queued.is_some()
                && let Some(event) = local_batch.events.first()
            {
                queued = Some(Diagnostic::queued_from(event));
            }
            let committed = ingest_prepared(&transaction, &local_batch, now, false)?;
            AccountEngine::validate_ingestion_receipt(&local_batch, &committed.receipt)?;
        }
        if let Some((conversation_id, draft_match)) = draft_to_consume {
            crate::account_product::validate_conversation_id(conversation_id)?;
            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)?;
        }
        let intents = load_operation_intents(&transaction, &prepared.operation_id)?;
        let operation = DeliveryReducer::operation_snapshot(intents)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        if let Some(mut queued) = queued {
            queued.revision = Some(revision);
            self.record_diagnostic(
                queued.count(Counter::TotalIntents, u64::from(operation.total_intents)),
            );
        }
        Ok(ProductAccountOperationResult {
            operation: AccountOperationResult {
                operation,
                revision,
                inserted: true,
            },
            result_message_id: result_message_id.to_owned(),
        })
    }

    fn persist_product_operation_without_identity(
        &mut self,
        prepared: PreparedOutgoingOperation,
        now: i64,
        command_hash: &str,
        result_message_id: &str,
        diagnostic_subject: AccountLogSubject,
    ) -> Result<ProductAccountOperationResult, SoftchatError> {
        validate_timestamp(now)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        persist_prepared_operation(
            &transaction,
            &prepared,
            now,
            0,
            command_hash,
            result_message_id,
        )?;
        let intents = load_operation_intents(&transaction, &prepared.operation_id)?;
        let operation = DeliveryReducer::operation_snapshot(intents)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        let mut diagnostic = Diagnostic::for_subject(diagnostic_subject, true)
            .count(Counter::TotalIntents, u64::from(operation.total_intents));
        if diagnostic_subject == AccountLogSubject::Unknown {
            diagnostic.category = AccountLogCategory::Recovery;
            diagnostic.operation = "recovery.completed";
            diagnostic.summary = "Signed outbox batch recovered";
            diagnostic.source = Some(AccountLogSource::Recovery);
            diagnostic.summarize = true;
            diagnostic = diagnostic.count(Counter::RecoveredItems, 1);
        }
        diagnostic.revision = Some(revision);
        self.record_diagnostic(diagnostic);
        Ok(ProductAccountOperationResult {
            operation: AccountOperationResult {
                operation,
                revision,
                inserted: true,
            },
            result_message_id: result_message_id.to_owned(),
        })
    }

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

    fn operation_result_message_id(&self, operation_id: &str) -> Result<String, SoftchatError> {
        self.connection
            .query_row(
                "SELECT result_message_id FROM outgoing_operations WHERE operation_id = ?1",
                [operation_id],
                |row| row.get(0),
            )
            .map_err(map_sqlite_error)
    }

    pub(crate) fn active_relay_url(&self) -> Result<String, SoftchatError> {
        self.connection
            .query_row(
                "SELECT relay_url FROM relay_catalog
                 WHERE active = 1 AND excluded = 0
                 ORDER BY relay_url ASC LIMIT 1",
                [],
                |row| row.get(0),
            )
            .optional()
            .map_err(map_sqlite_error)?
            .ok_or(SoftchatError::InvalidRelayCatalog)
    }

    fn operation_optional(
        &self,
        operation_id: &str,
    ) -> Result<Option<OperationSnapshot>, SoftchatError> {
        let exists = self
            .connection
            .query_row(
                "SELECT 1 FROM outgoing_operations WHERE operation_id = ?1",
                [operation_id],
                |_| Ok(()),
            )
            .optional()
            .map_err(map_sqlite_error)?
            .is_some();
        if !exists {
            return Ok(None);
        }
        let intents = load_operation_intents(&self.connection, operation_id)?;
        DeliveryReducer::operation_snapshot(intents).map(Some)
    }

    fn apply_delivery_mutations(
        &mut self,
        claim: &DeliveryClaim,
        mutations: &[crate::DeliveryStateMutation],
        transition: &str,
    ) -> Result<AccountMutationResult, SoftchatError> {
        let transaction = begin_write_scope(&mut self.connection)?;
        for mutation in mutations {
            let expected_state = match transition {
                "claimed" => "claimed",
                "result" => "claimed_or_written",
                "release" => "claimed_or_written",
                _ => return Err(SoftchatError::InvalidDeliveryState),
            };
            let changed = if expected_state == "claimed" {
                transaction.execute(
                    "UPDATE relay_delivery_intents
                     SET state = ?1, lease_id = ?2, lease_expires_at = 0,
                         attempt_count = ?3, last_category = ?4
                     WHERE intent_id = ?5 AND lease_id = ?6 AND state = 'claimed'",
                    params![
                        intent_state_name(mutation.state),
                        mutation.lease_id,
                        mutation.attempt_count,
                        mutation.category,
                        mutation.intent_id,
                        claim.lease_id,
                    ],
                )
            } else {
                transaction.execute(
                    "UPDATE relay_delivery_intents
                     SET state = ?1, lease_id = ?2, lease_expires_at = 0,
                         attempt_count = ?3, last_category = ?4
                     WHERE intent_id = ?5 AND lease_id = ?6
                       AND state IN ('claimed', 'socket_written')",
                    params![
                        intent_state_name(mutation.state),
                        mutation.lease_id,
                        mutation.attempt_count,
                        mutation.category,
                        mutation.intent_id,
                        claim.lease_id,
                    ],
                )
            }
            .map_err(map_sqlite_error)?;
            if changed != 1 {
                return Err(SoftchatError::InvalidDeliveryState);
            }
        }
        let mut diagnostics = Vec::new();
        if self.diagnostics.enabled() && !mutations.is_empty() {
            let changed_ids = mutations
                .iter()
                .map(|value| &value.intent_id)
                .collect::<BTreeSet<_>>();
            let operation_ids = claim
                .intents
                .iter()
                .filter(|intent| changed_ids.contains(&intent.intent_id))
                .map(|intent| &intent.operation_id)
                .collect::<BTreeSet<_>>();
            for operation_id in operation_ids {
                // Diagnostic-only reads must never make a valid state transition fail.
                if let Ok(snapshot) = load_operation_intents(&transaction, operation_id)
                    .and_then(DeliveryReducer::operation_snapshot)
                {
                    let retryable = mutations.iter().any(|mutation| {
                        mutation.state == DeliveryIntentState::Retryable
                            && claim.intents.iter().any(|intent| {
                                &intent.operation_id == operation_id
                                    && intent.intent_id == mutation.intent_id
                            })
                    });
                    diagnostics.push(delivery_diagnostic(
                        &transaction,
                        &snapshot,
                        transition,
                        retryable,
                    ));
                }
            }
        }
        let revision = bump_revision(&transaction)?;
        transaction.commit()?;
        for mut diagnostic in diagnostics {
            diagnostic.revision = Some(revision);
            self.record_diagnostic(diagnostic);
        }
        Ok(AccountMutationResult { revision })
    }

    fn persist_relay_catalog(
        &mut self,
        entries: &[RelayCatalogEntry],
        updated_at: i64,
    ) -> Result<i64, SoftchatError> {
        validate_timestamp(updated_at).map_err(|_| SoftchatError::InvalidRelayCatalog)?;
        let transaction = self
            .connection
            .transaction_with_behavior(TransactionBehavior::Immediate)
            .map_err(map_sqlite_error)?;
        replace_relay_catalog(&transaction, entries, updated_at)?;
        let revision = bump_revision(&transaction)?;
        transaction.commit().map_err(map_sqlite_error)?;
        Ok(revision)
    }
}

// Optional, indexed diagnostic reads never change the success of a delivery write.
fn delivery_diagnostic(
    connection: &Connection,
    snapshot: &OperationSnapshot,
    transition: &str,
    retryable: bool,
) -> Diagnostic {
    let mut diagnostic = crate::account_diagnostics::delivery(snapshot, transition, retryable);
    let kind = connection
        .query_row(
            "SELECT projection.kind FROM outgoing_event_copies copy
         JOIN authenticated_rumor_wrappers wrapper ON wrapper.outer_event_id = copy.event_copy_id
         JOIN projections projection ON projection.logical_event_id = wrapper.rumor_id
         WHERE copy.operation_id = ?1 LIMIT 1",
            [&snapshot.operation_id],
            |row| row.get::<_, String>(0),
        )
        .ok()
        .and_then(|value| projection_kind_from_name(&value));
    if let Some(kind) = kind {
        let subject = crate::account_diagnostics::projection_subject(kind, "");
        diagnostic.subject = Some(subject);
        diagnostic.category = Diagnostic::for_subject(subject, true).category;
    }
    // Ephemeral sender copies deliberately have no durable product projection.
    // Their authenticated outer kind still proves that routine ACKs are DEBUG.
    let ephemeral = connection.query_row(
        "SELECT COUNT(*) > 0 AND SUM(COALESCE(json_extract(event_json, '$.kind') = ?2, 0)) = COUNT(*)
         FROM outgoing_event_copies WHERE operation_id = ?1",
        params![snapshot.operation_id, crate::NostrEventKind::EPHEMERAL_GIFT_WRAP.as_u16()],
        |row| row.get::<_, bool>(0),
    ).unwrap_or(false);
    if ephemeral && diagnostic.level == AccountLogLevel::Info {
        diagnostic = diagnostic.debug();
    }
    diagnostic
}

fn replace_relay_catalog(
    transaction: &Connection,
    entries: &[RelayCatalogEntry],
    updated_at: i64,
) -> Result<(), SoftchatError> {
    transaction
        .execute("DELETE FROM relay_catalog", [])
        .map_err(map_sqlite_error)?;
    for entry in entries {
        transaction
            .execute(
                "INSERT INTO relay_catalog
                 (relay_url, created_at, source, active, is_dns_discovered,
                  excluded, priority, weight, expires_at, updated_at)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10)",
                params![
                    entry.url,
                    entry.created_at,
                    relay_source_name(entry.source),
                    i64::from(entry.is_active),
                    i64::from(entry.is_dns_discovered),
                    i64::from(entry.is_user_excluded),
                    entry.dns_priority,
                    entry.dns_weight,
                    entry.dns_expires_at,
                    updated_at,
                ],
            )
            .map_err(map_sqlite_error)?;
    }
    Ok(())
}

fn migrate(connection: &mut Connection, account_id: &str) -> Result<(), SoftchatError> {
    let transaction = connection
        .transaction_with_behavior(TransactionBehavior::Immediate)
        .map_err(map_sqlite_error)?;
    transaction
        .execute_batch(
            "
            CREATE TABLE IF NOT EXISTS softchat_meta (
                key TEXT PRIMARY KEY NOT NULL,
                integer_value INTEGER,
                text_value TEXT
            );
            CREATE TABLE IF NOT EXISTS signed_events (
                event_id TEXT PRIMARY KEY NOT NULL,
                canonical_json TEXT NOT NULL,
                author_public_key TEXT NOT NULL,
                kind INTEGER NOT NULL,
                created_at INTEGER NOT NULL,
                received_at INTEGER NOT NULL
            );
            CREATE INDEX IF NOT EXISTS signed_events_created
                ON signed_events(created_at DESC, event_id ASC);
            CREATE TABLE IF NOT EXISTS authenticated_rumors (
                rumor_id TEXT PRIMARY KEY NOT NULL,
                outer_event_id TEXT NOT NULL REFERENCES signed_events(event_id),
                canonical_json TEXT NOT NULL,
                author_public_key TEXT NOT NULL,
                kind INTEGER NOT NULL,
                created_at INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS authenticated_rumor_wrappers (
                outer_event_id TEXT PRIMARY KEY NOT NULL
                    REFERENCES signed_events(event_id) ON DELETE CASCADE,
                rumor_id TEXT NOT NULL
                    REFERENCES authenticated_rumors(rumor_id) ON DELETE CASCADE
            );
            CREATE INDEX IF NOT EXISTS rumor_wrappers_rumor
                ON authenticated_rumor_wrappers(rumor_id, outer_event_id);
            CREATE TABLE IF NOT EXISTS projections (
                logical_event_id TEXT PRIMARY KEY NOT NULL,
                outer_event_id TEXT NOT NULL REFERENCES signed_events(event_id),
                kind TEXT NOT NULL,
                author_public_key TEXT NOT NULL,
                created_at INTEGER NOT NULL,
                conversation_id TEXT NOT NULL,
                content TEXT NOT NULL,
                ephemeral INTEGER NOT NULL CHECK(ephemeral IN (0, 1))
            );
            CREATE INDEX IF NOT EXISTS projections_conversation
                ON projections(conversation_id, created_at DESC, logical_event_id ASC);
            CREATE INDEX IF NOT EXISTS projections_author_created
                ON projections(author_public_key, ephemeral, created_at ASC);
            CREATE INDEX IF NOT EXISTS projections_kind
                ON projections(kind, created_at DESC, logical_event_id ASC);
            CREATE INDEX IF NOT EXISTS projections_kind_conversation
                ON projections(
                    kind, conversation_id, created_at DESC, logical_event_id ASC
                );
            CREATE INDEX IF NOT EXISTS projections_kind_author
                ON projections(
                    kind, author_public_key, created_at DESC, logical_event_id ASC
                );
            CREATE TABLE IF NOT EXISTS projection_participants (
                logical_event_id TEXT NOT NULL REFERENCES projections(logical_event_id)
                    ON DELETE CASCADE,
                public_key TEXT NOT NULL,
                PRIMARY KEY(logical_event_id, public_key)
            );
            CREATE INDEX IF NOT EXISTS participants_public_key
                ON projection_participants(public_key, logical_event_id);
            CREATE TABLE IF NOT EXISTS projection_targets (
                logical_event_id TEXT NOT NULL REFERENCES projections(logical_event_id)
                    ON DELETE CASCADE,
                target_event_id TEXT NOT NULL,
                position INTEGER NOT NULL,
                PRIMARY KEY(logical_event_id, position)
            );
            CREATE INDEX IF NOT EXISTS projection_target
                ON projection_targets(target_event_id, logical_event_id);
            CREATE TABLE IF NOT EXISTS event_fingerprints (
                event_id TEXT PRIMARY KEY NOT NULL REFERENCES signed_events(event_id)
                    ON DELETE CASCADE,
                created_at INTEGER NOT NULL
            );
            CREATE INDEX IF NOT EXISTS fingerprints_created
                ON event_fingerprints(created_at ASC, event_id ASC);
            CREATE TABLE IF NOT EXISTS outgoing_operations (
                operation_id TEXT PRIMARY KEY NOT NULL,
                account_id TEXT NOT NULL,
                created_at INTEGER NOT NULL,
                command_hash TEXT NOT NULL DEFAULT '',
                result_message_id TEXT NOT NULL DEFAULT ''
            );
            CREATE TABLE IF NOT EXISTS outgoing_event_copies (
                event_copy_id TEXT PRIMARY KEY NOT NULL,
                operation_id TEXT NOT NULL REFERENCES outgoing_operations(operation_id)
                    ON DELETE CASCADE,
                recipient_public_key TEXT NOT NULL,
                event_json TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS event_copies_operation
                ON outgoing_event_copies(operation_id, event_copy_id);
            CREATE TABLE IF NOT EXISTS relay_delivery_intents (
                intent_id TEXT PRIMARY KEY NOT NULL,
                operation_id TEXT NOT NULL REFERENCES outgoing_operations(operation_id)
                    ON DELETE CASCADE,
                event_copy_id TEXT NOT NULL REFERENCES outgoing_event_copies(event_copy_id)
                    ON DELETE CASCADE,
                relay_url TEXT NOT NULL,
                payload_bytes INTEGER NOT NULL,
                state TEXT NOT NULL,
                lease_id TEXT NOT NULL DEFAULT '',
                lease_expires_at INTEGER NOT NULL DEFAULT 0,
                attempt_count INTEGER NOT NULL DEFAULT 0,
                last_category TEXT NOT NULL DEFAULT 'queued',
                created_at INTEGER NOT NULL,
                UNIQUE(event_copy_id, relay_url)
            );
            CREATE INDEX IF NOT EXISTS delivery_claimable
                ON relay_delivery_intents(state, created_at, intent_id);
            CREATE INDEX IF NOT EXISTS delivery_operation
                ON relay_delivery_intents(operation_id, intent_id);
            CREATE TABLE IF NOT EXISTS relay_catalog (
                relay_url TEXT PRIMARY KEY NOT NULL,
                created_at INTEGER NOT NULL,
                source TEXT NOT NULL,
                active INTEGER NOT NULL,
                is_dns_discovered INTEGER NOT NULL,
                excluded INTEGER NOT NULL,
                priority INTEGER NOT NULL,
                weight INTEGER NOT NULL,
                expires_at INTEGER NOT NULL,
                updated_at INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS sync_checkpoints (
                sync_id TEXT PRIMARY KEY NOT NULL,
                checkpoint_timestamp INTEGER NOT NULL,
                updated_at INTEGER NOT NULL
            );
            DROP TABLE IF EXISTS attachment_jobs;
            CREATE TABLE IF NOT EXISTS local_contacts (
                public_key TEXT PRIMARY KEY NOT NULL,
                name TEXT,
                updated_at_millis INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS sticker_cache (
                name TEXT PRIMARY KEY NOT NULL,
                url TEXT NOT NULL,
                associated_emojis_json TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS link_preview_cache (
                url TEXT PRIMARY KEY NOT NULL,
                title TEXT,
                description TEXT,
                image_url TEXT,
                updated_at_millis INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS quarantine (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                source TEXT NOT NULL,
                source_id TEXT NOT NULL,
                category TEXT NOT NULL,
                observed_at INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS conversation_state (
                conversation_id TEXT PRIMARY KEY NOT NULL,
                archived INTEGER NOT NULL DEFAULT 0 CHECK(archived IN (0, 1)),
                pinned INTEGER NOT NULL DEFAULT 0 CHECK(pinned IN (0, 1)),
                read_at INTEGER,
                unread_override INTEGER NOT NULL DEFAULT 0
                    CHECK(unread_override IN (0, 1)),
                draft_json TEXT,
                draft_reply_to TEXT,
                draft_updated_at INTEGER,
                created_at INTEGER NOT NULL,
                updated_at INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS conversation_members (
                conversation_id TEXT NOT NULL
                    REFERENCES conversation_state(conversation_id) ON DELETE CASCADE,
                public_key TEXT NOT NULL,
                PRIMARY KEY(conversation_id, public_key)
            );
            CREATE TABLE IF NOT EXISTS conversation_subject_icons (
                conversation_id TEXT PRIMARY KEY NOT NULL
                    REFERENCES conversation_state(conversation_id) ON DELETE CASCADE,
                subject_event_id TEXT NOT NULL,
                subject_created_at INTEGER NOT NULL,
                attachment_json TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS conversation_state_order
                ON conversation_state(pinned DESC, updated_at DESC, conversation_id ASC);
            CREATE TABLE IF NOT EXISTS conversation_read_state (
                conversation_id TEXT PRIMARY KEY NOT NULL,
                read_at INTEGER,
                read_message_id TEXT NOT NULL DEFAULT '',
                forced_unread INTEGER NOT NULL DEFAULT 0
                    CHECK(forced_unread IN (0, 1)),
                updated_at INTEGER NOT NULL,
                update_id TEXT NOT NULL,
                unknown_json TEXT NOT NULL DEFAULT '{}'
            );
            CREATE TABLE IF NOT EXISTS account_settings (
                singleton INTEGER PRIMARY KEY NOT NULL CHECK(singleton = 1),
                schema_version INTEGER NOT NULL,
                canonical_json TEXT NOT NULL,
                updated_at INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS app_data_effective (
                context TEXT PRIMARY KEY NOT NULL,
                canonical_json TEXT NOT NULL,
                event_created_at INTEGER NOT NULL,
                event_id TEXT NOT NULL
            );
            CREATE TABLE IF NOT EXISTS media_operations (
                operation_id TEXT PRIMARY KEY NOT NULL,
                command_id TEXT NOT NULL UNIQUE,
                conversation_id TEXT,
                source_message_id TEXT,
                operation_kind TEXT NOT NULL DEFAULT 'transfer'
                    CHECK(operation_kind IN ('transfer', 'conversation_icon')),
                direction TEXT NOT NULL,
                protection TEXT NOT NULL DEFAULT 'encrypted'
                    CHECK(protection IN ('encrypted', 'public')),
                state TEXT NOT NULL,
                source_fingerprint TEXT NOT NULL,
                attachment_json TEXT,
                byte_count INTEGER,
                attempt_count INTEGER NOT NULL DEFAULT 0,
                last_error_category TEXT NOT NULL DEFAULT '',
                lease_id TEXT NOT NULL DEFAULT '',
                lease_expires_at INTEGER NOT NULL DEFAULT 0,
                created_at INTEGER NOT NULL,
                updated_at INTEGER NOT NULL
            );
            CREATE INDEX IF NOT EXISTS media_operations_state
                ON media_operations(state, updated_at, operation_id);
            CREATE TABLE IF NOT EXISTS pending_media_messages (
                command_id TEXT PRIMARY KEY NOT NULL,
                conversation_id TEXT NOT NULL,
                text_content TEXT NOT NULL,
                emoji_tags_json TEXT NOT NULL,
                reply_to_message_id TEXT NOT NULL DEFAULT '',
                command_hash TEXT NOT NULL,
                state TEXT NOT NULL,
                publication_claimed INTEGER NOT NULL DEFAULT 0
                    CHECK(publication_claimed IN (0, 1)),
                result_message_id TEXT NOT NULL DEFAULT '',
                created_at INTEGER NOT NULL,
                updated_at INTEGER NOT NULL
            );
            CREATE INDEX IF NOT EXISTS pending_media_messages_state
                ON pending_media_messages(state, updated_at, command_id);
            CREATE TABLE IF NOT EXISTS pending_message_attachments (
                command_id TEXT NOT NULL
                    REFERENCES pending_media_messages(command_id) ON DELETE CASCADE,
                attachment_index INTEGER NOT NULL,
                operation_id TEXT NOT NULL UNIQUE
                    REFERENCES media_operations(operation_id) ON DELETE CASCADE,
                PRIMARY KEY(command_id, attachment_index)
            );
            CREATE TABLE IF NOT EXISTS pending_asset_replacements (
                command_id TEXT PRIMARY KEY NOT NULL,
                target TEXT NOT NULL
                    CHECK(target IN (
                        'profile_picture', 'profile_banner', 'conversation_icon'
                    )),
                conversation_id TEXT NOT NULL DEFAULT '',
                operation_id TEXT NOT NULL UNIQUE
                    REFERENCES media_operations(operation_id) ON DELETE CASCADE,
                subject TEXT NOT NULL DEFAULT '',
                emoji_tags_json TEXT NOT NULL DEFAULT '[]',
                command_hash TEXT NOT NULL,
                state TEXT NOT NULL,
                result_operation_id TEXT NOT NULL DEFAULT '',
                created_at INTEGER NOT NULL,
                updated_at INTEGER NOT NULL
            );
            CREATE INDEX IF NOT EXISTS pending_asset_replacements_state
                ON pending_asset_replacements(state, updated_at, command_id);
            CREATE TABLE IF NOT EXISTS message_local_extras (
                message_id TEXT PRIMARY KEY NOT NULL,
                transcript TEXT,
                waveform BLOB,
                layout_width INTEGER,
                layout_height INTEGER,
                updated_at INTEGER NOT NULL
            );
            CREATE TABLE IF NOT EXISTS processed_transport_results (
                run_id TEXT NOT NULL,
                action_id TEXT NOT NULL,
                generation INTEGER NOT NULL,
                result_hash TEXT NOT NULL,
                outcome_json TEXT NOT NULL DEFAULT '',
                PRIMARY KEY(run_id, action_id)
            );
            CREATE VIRTUAL TABLE IF NOT EXISTS message_fts USING fts5(
                message_id UNINDEXED,
                conversation_id UNINDEXED,
                content,
                tokenize = 'unicode61 remove_diacritics 2'
            );
            ",
        )
        .map_err(map_sqlite_error)?;

    let version = meta_integer_optional(&transaction, "schema_version")?.unwrap_or(0);
    if version > ACCOUNT_SCHEMA_VERSION {
        return Err(SoftchatError::InvalidPersistenceResult);
    }
    if version == 1 {
        transaction
            .execute_batch(
                "
                ALTER TABLE relay_catalog
                    ADD COLUMN created_at INTEGER NOT NULL DEFAULT 0;
                ALTER TABLE relay_catalog
                    ADD COLUMN is_dns_discovered INTEGER NOT NULL DEFAULT 0;
                ",
            )
            .map_err(map_sqlite_error)?;
    }
    if (1..5).contains(&version) {
        transaction
            .execute_batch(
                "ALTER TABLE outgoing_operations
                   ADD COLUMN command_hash TEXT NOT NULL DEFAULT '';
                 ALTER TABLE outgoing_operations
                   ADD COLUMN result_message_id TEXT NOT NULL DEFAULT '';",
            )
            .map_err(map_sqlite_error)?;
    }
    if (1..6).contains(&version) {
        transaction
            .execute_batch(
                "ALTER TABLE processed_transport_results
                   ADD COLUMN outcome_json TEXT NOT NULL DEFAULT '';",
            )
            .map_err(map_sqlite_error)?;
    }
    if (1..9).contains(&version) {
        transaction
            .execute_batch(
                "ALTER TABLE media_operations
                   ADD COLUMN source_message_id TEXT;",
            )
            .map_err(map_sqlite_error)?;
    }
    if (1..10).contains(&version) {
        transaction
            .execute_batch(
                "DROP TABLE IF EXISTS legacy_import_journal;
                 DELETE FROM softchat_meta
                 WHERE key = 'legacy_local_contacts_imported';",
            )
            .map_err(map_sqlite_error)?;
    }
    if (1..11).contains(&version) {
        transaction
            .execute_batch(
                "ALTER TABLE media_operations
                   ADD COLUMN protection TEXT NOT NULL DEFAULT 'encrypted'
                   CHECK(protection IN ('encrypted', 'public'));",
            )
            .map_err(map_sqlite_error)?;
    }
    if (1..12).contains(&version) {
        transaction
            .execute(
                "INSERT OR IGNORE INTO authenticated_rumor_wrappers
                 (outer_event_id, rumor_id)
                 SELECT outer_event_id, rumor_id FROM authenticated_rumors",
                [],
            )
            .map_err(map_sqlite_error)?;
    }
    if (1..14).contains(&version) {
        transaction
            .execute_batch(
                "ALTER TABLE media_operations
                   ADD COLUMN operation_kind TEXT NOT NULL DEFAULT 'transfer'
                   CHECK(operation_kind IN ('transfer', 'conversation_icon'));",
            )
            .map_err(map_sqlite_error)?;
    }
    if version < 15 {
        let has_publication_claim = transaction
            .query_row(
                "SELECT EXISTS(
                   SELECT 1 FROM pragma_table_info('pending_media_messages')
                   WHERE name = 'publication_claimed'
                 )",
                [],
                |row| row.get::<_, bool>(0),
            )
            .map_err(map_sqlite_error)?;
        if !has_publication_claim {
            transaction
                .execute_batch(
                    "ALTER TABLE pending_media_messages
                       ADD COLUMN publication_claimed INTEGER NOT NULL DEFAULT 0
                       CHECK(publication_claimed IN (0, 1));",
                )
                .map_err(map_sqlite_error)?;
        }
    }
    if version < 7 {
        transaction
            .execute(
                "INSERT OR IGNORE INTO conversation_read_state
                 (conversation_id, read_at, read_message_id, forced_unread,
                  updated_at, update_id, unknown_json)
                 SELECT
                   cs.conversation_id,
                   cs.read_at,
                   COALESCE(
                     (
                       SELECT p.logical_event_id
                       FROM projections p
                       WHERE p.conversation_id = cs.conversation_id
                         AND p.kind = 'chat_message'
                         AND p.created_at = cs.read_at
                       ORDER BY p.logical_event_id ASC LIMIT 1
                     ),
                     ''
                   ),
                   cs.unread_override,
                   cs.updated_at,
                   'legacy-local-' || cs.conversation_id,
                   '{}'
                 FROM conversation_state cs
                 WHERE cs.read_at IS NOT NULL OR cs.unread_override = 1",
                [],
            )
            .map_err(map_sqlite_error)?;
    }
    transaction
        .execute(
            "INSERT INTO relay_catalog
             (relay_url, created_at, source, active, is_dns_discovered,
              excluded, priority, weight, expires_at, updated_at)
             SELECT
               ?1, 0, 'automatic', 1, 1,
               0, 0, 0, 3600, 0
             WHERE NOT EXISTS (SELECT 1 FROM relay_catalog)",
            [DEFAULT_RELAY_URL],
        )
        .map_err(map_sqlite_error)?;
    transaction
        .execute(
            "INSERT OR IGNORE INTO account_settings
             (singleton, schema_version, canonical_json, updated_at)
             VALUES (1, 1, '{}', 0)",
            [],
        )
        .map_err(map_sqlite_error)?;
    if (1..13).contains(&version) {
        reconcile_effective_app_data(&transaction)?;
    }
    transaction
        .execute(
            "INSERT OR IGNORE INTO conversation_state
             (conversation_id, created_at, updated_at)
             SELECT DISTINCT conversation_id, MIN(created_at), MAX(created_at)
             FROM projections
             WHERE conversation_id <> ''
             GROUP BY conversation_id",
            [],
        )
        .map_err(map_sqlite_error)?;
    transaction
        .execute(
            "INSERT OR IGNORE INTO conversation_members(conversation_id, public_key)
             SELECT DISTINCT p.conversation_id, pp.public_key
             FROM projections p
             JOIN projection_participants pp
               ON pp.logical_event_id = p.logical_event_id
             WHERE p.conversation_id <> ''",
            [],
        )
        .map_err(map_sqlite_error)?;
    if version < 14 {
        reconcile_effective_subject_icons(&transaction)?;
        reconcile_media_operation_kinds(&transaction)?;
    }
    if version < 16 {
        transaction
            .execute_batch(EFFECTIVE_MESSAGE_EDITS_SCHEMA)
            .map_err(map_sqlite_error)?;
        transaction
            .execute("DELETE FROM message_fts", [])
            .map_err(map_sqlite_error)?;
        insert_message_fts(&transaction, None)?;
    }
    transaction
        .execute(
            "INSERT INTO softchat_meta(key, integer_value)
             VALUES ('schema_version', ?1)
             ON CONFLICT(key) DO UPDATE SET integer_value = excluded.integer_value",
            [ACCOUNT_SCHEMA_VERSION],
        )
        .map_err(map_sqlite_error)?;
    transaction
        .execute(
            "INSERT OR IGNORE INTO softchat_meta(key, integer_value)
             VALUES ('revision', 0)",
            [],
        )
        .map_err(map_sqlite_error)?;
    let existing_account = transaction
        .query_row(
            "SELECT text_value FROM softchat_meta WHERE key = 'account_id'",
            [],
            |row| row.get::<_, Option<String>>(0),
        )
        .optional()
        .map_err(map_sqlite_error)?
        .flatten();
    if existing_account
        .as_deref()
        .is_some_and(|value| value != account_id)
    {
        return Err(SoftchatError::InvalidPersistenceResult);
    }
    transaction
        .execute(
            "INSERT OR IGNORE INTO softchat_meta(key, text_value)
             VALUES ('account_id', ?1)",
            [account_id],
        )
        .map_err(map_sqlite_error)?;
    transaction.commit().map_err(map_sqlite_error)
}

struct PreparedIngestionCommit {
    receipt: IngestionReceipt,
    suppressed_projection_outer_event_ids: BTreeSet<String>,
    diagnostics: Vec<Diagnostic>,
}

fn ingest_prepared(
    transaction: &Connection,
    batch: &PreparedIncomingBatch,
    received_at: i64,
    diagnostics_enabled: bool,
) -> Result<PreparedIngestionCommit, SoftchatError> {
    let mut inserted_event_ids = Vec::new();
    let mut duplicate_event_ids = Vec::new();
    let mut quarantined_event_ids = Vec::new();
    let mut suppressed_projection_outer_event_ids = BTreeSet::new();
    let mut diagnostics = Vec::new();
    for prepared in &batch.events {
        let event = SignedNostrEvent::try_from(prepared.outer_event.clone())?;
        let event_id = event.id().to_hex();
        if prepared.projection.ephemeral {
            if diagnostics_enabled {
                crate::account_diagnostics::accumulate(
                    &mut diagnostics,
                    Diagnostic::received(prepared).count(Counter::EphemeralEvents, 1),
                );
            }
            inserted_event_ids.push(event_id);
            continue;
        }
        let canonical_json = event.to_json()?;
        let existing = transaction
            .query_row(
                "SELECT canonical_json FROM signed_events WHERE event_id = ?1",
                [&event_id],
                |row| row.get::<_, String>(0),
            )
            .optional()
            .map_err(map_sqlite_error)?;
        if let Some(existing) = existing {
            if existing == canonical_json {
                duplicate_event_ids.push(event_id);
                if diagnostics_enabled {
                    crate::account_diagnostics::accumulate(
                        &mut diagnostics,
                        Diagnostic::new(
                            AccountLogCategory::Storage,
                            "ingestion.duplicate",
                            "Repeated event ignored",
                        )
                        .debug()
                        .count(Counter::Duplicates, 1),
                    );
                }
            } else {
                transaction
                    .execute(
                        "INSERT INTO quarantine(source, source_id, category, observed_at)
                         VALUES ('ingestion', ?1, 'event_id_conflict', ?2)",
                        params![event_id, received_at],
                    )
                    .map_err(map_sqlite_error)?;
                quarantined_event_ids.push(event_id);
                if diagnostics_enabled {
                    let mut diagnostic = Diagnostic::new(
                        AccountLogCategory::Storage,
                        "ingestion.quarantined",
                        "Conflicting event quarantined",
                    )
                    .count(Counter::QuarantinedEvents, 1);
                    diagnostic.level = AccountLogLevel::Warn;
                    diagnostic.reason = "event_id_conflict";
                    diagnostic.summarize = true;
                    crate::account_diagnostics::accumulate(&mut diagnostics, diagnostic);
                }
            }
            continue;
        }
        transaction
            .execute(
                "INSERT INTO signed_events
                 (event_id, canonical_json, author_public_key, kind, created_at, received_at)
                 VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                params![
                    event_id,
                    canonical_json,
                    event.public_key().to_hex(),
                    i64::from(event.kind().as_u16()),
                    i64::try_from(event.created_at())
                        .map_err(|_| SoftchatError::InvalidEventTimestamp)?,
                    received_at,
                ],
            )
            .map_err(map_sqlite_error)?;
        transaction
            .execute(
                "INSERT INTO event_fingerprints(event_id, created_at) VALUES (?1, ?2)",
                params![
                    event_id,
                    i64::try_from(event.created_at())
                        .map_err(|_| SoftchatError::InvalidEventTimestamp)?,
                ],
            )
            .map_err(map_sqlite_error)?;
        let mut insert_logical_projection = true;
        if let Some(rumor) = &prepared.rumor {
            let canonical_rumor =
                serde_json::to_string(rumor).map_err(|_| SoftchatError::InternalFailure)?;
            let existing = transaction
                .query_row(
                    "SELECT canonical_json, author_public_key, kind, created_at
                     FROM authenticated_rumors WHERE rumor_id = ?1",
                    [&rumor.id],
                    |row| {
                        Ok((
                            row.get::<_, String>(0)?,
                            row.get::<_, String>(1)?,
                            row.get::<_, i64>(2)?,
                            row.get::<_, i64>(3)?,
                        ))
                    },
                )
                .optional()
                .map_err(map_sqlite_error)?;
            if let Some((stored_json, stored_author, stored_kind, stored_created_at)) = existing {
                if stored_json != canonical_rumor
                    || stored_author != rumor.public_key
                    || stored_kind != i64::from(rumor.kind)
                    || stored_created_at != rumor.created_at
                {
                    return Err(SoftchatError::InvalidPersistenceResult);
                }
                insert_logical_projection = false;
                suppressed_projection_outer_event_ids.insert(event_id.clone());
            } else {
                transaction
                    .execute(
                        "INSERT INTO authenticated_rumors
                         (rumor_id, outer_event_id, canonical_json, author_public_key, kind,
                          created_at)
                         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                        params![
                            rumor.id,
                            event_id,
                            canonical_rumor,
                            rumor.public_key,
                            i64::from(rumor.kind),
                            rumor.created_at,
                        ],
                    )
                    .map_err(map_sqlite_error)?;
            }
            transaction
                .execute(
                    "INSERT INTO authenticated_rumor_wrappers
                     (outer_event_id, rumor_id) VALUES (?1, ?2)",
                    params![event_id, rumor.id],
                )
                .map_err(map_sqlite_error)?;
        }
        if insert_logical_projection {
            insert_projection(
                transaction,
                &event_id,
                &prepared.projection,
                prepared.rumor.as_ref(),
            )?;
        }
        if diagnostics_enabled {
            let diagnostic = if insert_logical_projection {
                Diagnostic::received(prepared).count(Counter::LogicalEvents, 1)
            } else {
                Diagnostic::new(
                    AccountLogCategory::Storage,
                    "ingestion.known_copy",
                    "Known logical event copy stored",
                )
                .debug()
                .count(Counter::KnownCopies, 1)
            }
            .count(Counter::DurableEvents, 1);
            crate::account_diagnostics::accumulate(&mut diagnostics, diagnostic);
        }
        inserted_event_ids.push(event_id);
    }
    Ok(PreparedIngestionCommit {
        receipt: IngestionReceipt {
            account_id: batch.account_id.clone(),
            inserted_event_ids,
            duplicate_event_ids,
            quarantined_event_ids,
        },
        suppressed_projection_outer_event_ids,
        diagnostics,
    })
}

fn account_projection_from_prepared(prepared: &crate::PreparedIncomingEvent) -> AccountProjection {
    AccountProjection {
        kind: prepared.projection.kind,
        logical_event_id: prepared.projection.logical_event_id.clone(),
        outer_event: prepared.outer_event.clone(),
        rumor: prepared.rumor.clone(),
        author_public_key: prepared.projection.author_public_key.clone(),
        created_at: prepared.projection.created_at,
        conversation_id: prepared.projection.conversation_id.clone(),
        participants: prepared.projection.participants.clone(),
        target_event_ids: prepared.projection.target_event_ids.clone(),
        content: prepared.projection.content.clone(),
    }
}

fn insert_projection(
    transaction: &Connection,
    outer_event_id: &str,
    projection: &crate::ProjectionMutation,
    rumor: Option<&RumorEvent>,
) -> Result<(), SoftchatError> {
    transaction
        .execute(
            "INSERT INTO projections
             (logical_event_id, outer_event_id, kind, author_public_key, created_at,
              conversation_id, content, ephemeral)
             VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
            params![
                projection.logical_event_id,
                outer_event_id,
                projection_kind_name(projection.kind),
                projection.author_public_key,
                projection.created_at,
                projection.conversation_id,
                projection.content,
                i64::from(projection.ephemeral),
            ],
        )
        .map_err(map_sqlite_error)?;
    for participant in &projection.participants {
        transaction
            .execute(
                "INSERT INTO projection_participants(logical_event_id, public_key)
                 VALUES (?1, ?2)",
                params![projection.logical_event_id, participant],
            )
            .map_err(map_sqlite_error)?;
    }
    for (position, target) in projection.target_event_ids.iter().enumerate() {
        transaction
            .execute(
                "INSERT INTO projection_targets(logical_event_id, target_event_id, position)
                 VALUES (?1, ?2, ?3)",
                params![
                    projection.logical_event_id,
                    target,
                    i64::try_from(position).map_err(|_| SoftchatError::InternalFailure)?,
                ],
            )
            .map_err(map_sqlite_error)?;
    }
    if !projection.conversation_id.is_empty() {
        transaction
            .execute(
                "INSERT INTO conversation_state
                 (conversation_id, created_at, updated_at)
                 VALUES (?1, ?2, ?2)
                 ON CONFLICT(conversation_id) DO UPDATE SET
                   updated_at = MAX(updated_at, excluded.updated_at)",
                params![projection.conversation_id, projection.created_at],
            )
            .map_err(map_sqlite_error)?;
        for participant in &projection.participants {
            transaction
                .execute(
                    "INSERT OR IGNORE INTO conversation_members
                     (conversation_id, public_key)
                     VALUES (?1, ?2)",
                    params![projection.conversation_id, participant],
                )
                .map_err(map_sqlite_error)?;
        }
    }
    match projection.kind {
        ProjectionKind::ChatMessage => {
            refresh_message_fts(transaction, &projection.logical_event_id)?;
        }
        ProjectionKind::Edit | ProjectionKind::Deletion => {
            let mut affected_messages = projection
                .target_event_ids
                .iter()
                .cloned()
                .collect::<BTreeSet<_>>();
            if projection.kind == ProjectionKind::Deletion {
                let mut statement = transaction
                    .prepare(
                        "SELECT target.target_event_id
                         FROM projection_targets target
                         JOIN projections edit
                           ON edit.logical_event_id = target.logical_event_id
                         WHERE edit.logical_event_id = ?1 AND edit.kind = 'edit'",
                    )
                    .map_err(map_sqlite_error)?;
                for target in &projection.target_event_ids {
                    let rows = statement
                        .query_map([target], |row| row.get::<_, String>(0))
                        .map_err(map_sqlite_error)?;
                    for row in rows {
                        affected_messages.insert(row.map_err(map_sqlite_error)?);
                    }
                }
            }
            for message_id in affected_messages {
                refresh_message_fts(transaction, &message_id)?;
            }
        }
        ProjectionKind::AppDataSync => {
            apply_app_data_projection(transaction, projection)?;
        }
        ProjectionKind::Subject => {
            let rumor = rumor.ok_or(SoftchatError::InvalidPersistenceResult)?;
            apply_subject_icon_projection(
                transaction,
                &projection.conversation_id,
                &projection.logical_event_id,
                projection.created_at,
                rumor,
            )?;
        }
        ProjectionKind::Reaction
        | ProjectionKind::Typing
        | ProjectionKind::UserMetadata
        | ProjectionKind::FollowList
        | ProjectionKind::ApplicationData
        | ProjectionKind::GenericRepost
        | ProjectionKind::Unknown => {}
    }
    Ok(())
}

fn apply_subject_icon_projection(
    connection: &Connection,
    conversation_id: &str,
    subject_event_id: &str,
    subject_created_at: i64,
    rumor: &RumorEvent,
) -> Result<(), SoftchatError> {
    let Some(attachment) =
        subject_icon_metadata(rumor).map_err(|_| SoftchatError::InvalidPersistenceResult)?
    else {
        // Profile version 1 inherits the last explicit icon.
        return Ok(());
    };
    let attachment_json =
        serde_json::to_string(&attachment).map_err(|_| SoftchatError::InternalFailure)?;
    connection
        .execute(
            "INSERT INTO conversation_subject_icons
             (conversation_id, subject_event_id, subject_created_at, attachment_json)
             VALUES (?1, ?2, ?3, ?4)
             ON CONFLICT(conversation_id) DO UPDATE SET
               subject_event_id = excluded.subject_event_id,
               subject_created_at = excluded.subject_created_at,
               attachment_json = excluded.attachment_json
             WHERE excluded.subject_created_at > conversation_subject_icons.subject_created_at
                OR (
                  excluded.subject_created_at = conversation_subject_icons.subject_created_at
                  AND excluded.subject_event_id < conversation_subject_icons.subject_event_id
                )",
            params![
                conversation_id,
                subject_event_id,
                subject_created_at,
                attachment_json,
            ],
        )
        .map_err(map_sqlite_error)?;
    Ok(())
}

fn reconcile_effective_subject_icons(connection: &Connection) -> Result<(), SoftchatError> {
    connection
        .execute("DELETE FROM conversation_subject_icons", [])
        .map_err(map_sqlite_error)?;
    let candidates = {
        let mut statement = connection
            .prepare(
                "SELECT p.conversation_id, p.logical_event_id, p.created_at,
                        ar.canonical_json
                 FROM projections p
                 LEFT JOIN authenticated_rumors ar
                   ON ar.rumor_id = p.logical_event_id
                 WHERE p.kind = 'subject' AND p.ephemeral = 0",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, i64>(2)?,
                    row.get::<_, Option<String>>(3)?,
                ))
            })
            .map_err(map_sqlite_error)?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(map_sqlite_error)?
    };
    for (conversation_id, subject_event_id, subject_created_at, rumor_json) in candidates {
        let rumor_json = rumor_json.ok_or(SoftchatError::InvalidPersistenceResult)?;
        let rumor = serde_json::from_str::<RumorEvent>(&rumor_json)
            .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
        apply_subject_icon_projection(
            connection,
            &conversation_id,
            &subject_event_id,
            subject_created_at,
            &rumor,
        )?;
    }
    Ok(())
}

fn reconcile_media_operation_kinds(connection: &Connection) -> Result<(), SoftchatError> {
    let known_icons = {
        let mut statement = connection
            .prepare(
                "SELECT p.conversation_id, p.logical_event_id, ar.canonical_json
                 FROM projections p
                 JOIN authenticated_rumors ar
                   ON ar.rumor_id = p.logical_event_id
                 WHERE p.kind = 'subject' AND p.ephemeral = 0",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                ))
            })
            .map_err(map_sqlite_error)?;
        let mut known_icons = BTreeSet::new();
        for row in rows {
            let (conversation_id, subject_event_id, rumor_json) = row.map_err(map_sqlite_error)?;
            let rumor = serde_json::from_str::<RumorEvent>(&rumor_json)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
            let Some(attachment) = subject_icon_metadata(&rumor)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)?
            else {
                continue;
            };
            let source_fingerprint = crate::account_product::conversation_icon_source_fingerprint(
                &conversation_id,
                &subject_event_id,
                &attachment,
            )?;
            let attachment_json =
                serde_json::to_string(&attachment).map_err(|_| SoftchatError::InternalFailure)?;
            known_icons.insert((conversation_id, source_fingerprint, attachment_json));
        }
        known_icons
    };
    let candidates = {
        let mut statement = connection
            .prepare(
                "SELECT operation_id, conversation_id, source_fingerprint, attachment_json
                 FROM media_operations
                 WHERE direction = 'download'
                   AND COALESCE(source_message_id, '') = ''
                   AND conversation_id IS NOT NULL
                   AND attachment_json IS NOT NULL",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([], |row| {
                Ok((
                    row.get::<_, String>(0)?,
                    row.get::<_, String>(1)?,
                    row.get::<_, String>(2)?,
                    row.get::<_, String>(3)?,
                ))
            })
            .map_err(map_sqlite_error)?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(map_sqlite_error)?
    };
    for (operation_id, conversation_id, source_fingerprint, attachment_json) in candidates {
        let attachment = serde_json::from_str::<crate::AttachmentMetadata>(&attachment_json)
            .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
        attachment
            .to_tag()
            .map_err(|_| SoftchatError::InvalidPersistenceResult)?;
        let canonical_attachment =
            serde_json::to_string(&attachment).map_err(|_| SoftchatError::InternalFailure)?;
        if known_icons.contains(&(conversation_id, source_fingerprint, canonical_attachment)) {
            connection
                .execute(
                    "UPDATE media_operations SET operation_kind = 'conversation_icon'
                     WHERE operation_id = ?1",
                    [&operation_id],
                )
                .map_err(map_sqlite_error)?;
        }
    }
    Ok(())
}

fn reconcile_effective_app_data(transaction: &Connection) -> Result<(), SoftchatError> {
    let candidates = {
        let mut statement = transaction
            .prepare(
                "SELECT logical_event_id, author_public_key, created_at, content
                 FROM projections
                 WHERE kind = 'app_data_sync'
                 ORDER BY created_at ASC, logical_event_id DESC",
            )
            .map_err(map_sqlite_error)?;
        let rows = statement
            .query_map([], |row| {
                Ok(crate::ProjectionMutation {
                    kind: ProjectionKind::AppDataSync,
                    logical_event_id: row.get(0)?,
                    author_public_key: row.get(1)?,
                    created_at: row.get(2)?,
                    conversation_id: String::new(),
                    participants: Vec::new(),
                    target_event_ids: Vec::new(),
                    content: row.get(3)?,
                    ephemeral: false,
                })
            })
            .map_err(map_sqlite_error)?;
        rows.collect::<Result<Vec<_>, _>>()
            .map_err(map_sqlite_error)?
    };

    for candidate in candidates {
        let context = candidate
            .content
            .split_once(':')
            .map(|(context, _)| context)
            .ok_or(SoftchatError::InvalidApplicationData)?;
        if replaceable_is_newer(
            transaction,
            context,
            candidate.created_at,
            &candidate.logical_event_id,
        )? {
            apply_app_data_projection(transaction, &candidate)?;
        }
    }
    Ok(())
}

fn apply_app_data_projection(
    transaction: &Connection,
    projection: &crate::ProjectionMutation,
) -> Result<(), SoftchatError> {
    let (context, json) = projection
        .content
        .split_once(':')
        .ok_or(SoftchatError::InvalidApplicationData)?;
    match context {
        "app-settings" => {
            let canonical = crate::account_product::canonical_settings_json(json)?;
            let settings = crate::account_product::settings_from_json(&canonical)?;
            if replaceable_is_newer(
                transaction,
                context,
                projection.created_at,
                &projection.logical_event_id,
            )? {
                transaction
                    .execute(
                        "UPDATE account_settings
                         SET schema_version = ?1, canonical_json = ?2, updated_at = ?3
                         WHERE singleton = 1",
                        params![settings.schema_version, canonical, projection.created_at],
                    )
                    .map_err(map_sqlite_error)?;
                store_effective_app_data(
                    transaction,
                    context,
                    json,
                    projection.created_at,
                    &projection.logical_event_id,
                )?;
            }
        }
        "read-state" => {
            let snapshot = crate::account_product::parse_read_state_snapshot(
                json,
                projection.created_at,
                &projection.logical_event_id,
            )?;
            for entry in snapshot.entries {
                let current = transaction
                    .query_row(
                        "SELECT updated_at, update_id
                         FROM conversation_read_state WHERE conversation_id = ?1",
                        [&entry.state.conversation_id],
                        |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
                    )
                    .optional()
                    .map_err(map_sqlite_error)?;
                if current.as_ref().is_some_and(|(updated_at, update_id)| {
                    (*updated_at, update_id.as_str())
                        >= (entry.state.updated_at, entry.state.update_id.as_str())
                }) {
                    continue;
                }
                let (read_at, read_message_id) = entry
                    .state
                    .boundary
                    .as_ref()
                    .map_or((None, String::new()), |boundary| {
                        (Some(boundary.created_at), boundary.message_id.clone())
                    });
                transaction
                    .execute(
                        "INSERT INTO conversation_read_state
                         (conversation_id, read_at, read_message_id, forced_unread,
                          updated_at, update_id, unknown_json)
                         VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
                         ON CONFLICT(conversation_id) DO UPDATE SET
                           read_at = excluded.read_at,
                           read_message_id = excluded.read_message_id,
                           forced_unread = excluded.forced_unread,
                           updated_at = excluded.updated_at,
                           update_id = excluded.update_id,
                           unknown_json = excluded.unknown_json",
                        params![
                            entry.state.conversation_id,
                            read_at,
                            read_message_id,
                            i64::from(entry.state.forced_unread),
                            entry.state.updated_at,
                            entry.state.update_id,
                            entry.unknown_json,
                        ],
                    )
                    .map_err(map_sqlite_error)?;
            }
            if replaceable_is_newer(
                transaction,
                context,
                projection.created_at,
                &projection.logical_event_id,
            )? {
                let mut root: serde_json::Map<String, serde_json::Value> =
                    serde_json::from_str(&snapshot.unknown_json)
                        .map_err(|_| SoftchatError::InvalidApplicationData)?;
                root.insert(
                    "schemaVersion".to_owned(),
                    serde_json::json!(crate::ACCOUNT_READ_STATE_SCHEMA_VERSION),
                );
                root.insert(
                    "chats".to_owned(),
                    serde_json::from_str::<serde_json::Value>(json)
                        .ok()
                        .and_then(|value| value.get("chats").cloned())
                        .unwrap_or_else(|| serde_json::Value::Array(Vec::new())),
                );
                let canonical = serde_json::to_string(&root)
                    .map_err(|_| SoftchatError::InvalidApplicationData)?;
                store_effective_app_data(
                    transaction,
                    context,
                    &canonical,
                    projection.created_at,
                    &projection.logical_event_id,
                )?;
            }
        }
        _ => return Err(SoftchatError::InvalidApplicationData),
    }
    Ok(())
}

fn replaceable_is_newer(
    transaction: &Connection,
    context: &str,
    created_at: i64,
    event_id: &str,
) -> Result<bool, SoftchatError> {
    let current = transaction
        .query_row(
            "SELECT event_created_at, event_id
             FROM app_data_effective WHERE context = ?1",
            [context],
            |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?)),
        )
        .optional()
        .map_err(map_sqlite_error)?;
    Ok(current.as_ref().is_none_or(|(current_at, current_id)| {
        crate::replacement::is_newer(&created_at, event_id, current_at, current_id.as_str())
    }))
}

fn store_effective_app_data(
    transaction: &Connection,
    context: &str,
    canonical_json: &str,
    event_created_at: i64,
    event_id: &str,
) -> Result<(), SoftchatError> {
    transaction
        .execute(
            "INSERT INTO app_data_effective
             (context, canonical_json, event_created_at, event_id)
             VALUES (?1, ?2, ?3, ?4)
             ON CONFLICT(context) DO UPDATE SET
               canonical_json = excluded.canonical_json,
               event_created_at = excluded.event_created_at,
               event_id = excluded.event_id",
            params![context, canonical_json, event_created_at, event_id],
        )
        .map_err(map_sqlite_error)?;
    Ok(())
}

fn refresh_message_fts(transaction: &Connection, message_id: &str) -> Result<(), SoftchatError> {
    transaction
        .execute(
            "DELETE FROM message_fts WHERE message_id = ?1",
            [message_id],
        )
        .map_err(map_sqlite_error)?;
    insert_message_fts(transaction, Some(message_id))
}

fn insert_message_fts(
    transaction: &Connection,
    message_id: Option<&str>,
) -> Result<(), SoftchatError> {
    // A fixed single-message predicate lets SQLite use its primary-key index;
    // an optional OR predicate would scan the history on every ingestion.
    let predicate = if message_id.is_some() {
        "p.logical_event_id = ?1"
    } else {
        "?1 IS NULL"
    };
    transaction
        .execute(
            &format!(
                "INSERT INTO message_fts(message_id, conversation_id, content)
             SELECT p.logical_event_id, p.conversation_id,
                    COALESCE(json_extract(edit.canonical_json, '$.content'),
                             json_extract(original.canonical_json, '$.content'))
             FROM projections p
             JOIN authenticated_rumors original
               ON original.rumor_id = p.logical_event_id
             LEFT JOIN authenticated_rumors edit
               ON edit.rumor_id = (
                 SELECT effective.edit_id FROM effective_message_edits effective
                 WHERE effective.message_id = p.logical_event_id
               )
             WHERE {predicate}
               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
               )"
            ),
            [message_id],
        )
        .map_err(map_sqlite_error)?;
    Ok(())
}

fn persist_prepared_operation(
    transaction: &Connection,
    operation: &PreparedOutgoingOperation,
    now: i64,
    initial_attempt_count: u32,
    command_hash: &str,
    result_message_id: &str,
) -> Result<(), SoftchatError> {
    transaction
        .execute(
            "INSERT INTO outgoing_operations
             (operation_id, account_id, created_at, command_hash, result_message_id)
             VALUES (?1, ?2, ?3, ?4, ?5)",
            params![
                operation.operation_id,
                operation.account_id,
                now,
                command_hash,
                result_message_id
            ],
        )
        .map_err(map_sqlite_error)?;
    for copy in &operation.event_copies {
        let event = SignedNostrEvent::try_from(copy.event.clone())?;
        transaction
            .execute(
                "INSERT INTO outgoing_event_copies
                 (event_copy_id, operation_id, recipient_public_key, event_json)
                 VALUES (?1, ?2, ?3, ?4)",
                params![
                    copy.event_copy_id,
                    operation.operation_id,
                    copy.recipient_public_key,
                    event.to_json()?,
                ],
            )
            .map_err(map_sqlite_error)?;
    }
    for intent in &operation.relay_intents {
        let payload_bytes = i64::try_from(intent.payload_bytes)
            .map_err(|_| SoftchatError::InvalidAccountOperation)?;
        transaction
            .execute(
                "INSERT INTO relay_delivery_intents
                 (intent_id, operation_id, event_copy_id, relay_url, payload_bytes,
                  state, lease_id, lease_expires_at, attempt_count, last_category, created_at)
                 VALUES (?1, ?2, ?3, ?4, ?5, 'pending', '', 0, ?6, 'queued', ?7)",
                params![
                    intent.intent_id,
                    intent.operation_id,
                    intent.event_copy_id,
                    intent.relay_url,
                    payload_bytes,
                    initial_attempt_count,
                    now,
                ],
            )
            .map_err(map_sqlite_error)?;
    }
    Ok(())
}

fn load_relay_catalog(connection: &Connection) -> Result<Vec<RelayCatalogEntry>, SoftchatError> {
    let mut statement = connection
        .prepare(
            "SELECT relay_url, created_at, source, active, is_dns_discovered,
                    excluded, priority, weight, expires_at
             FROM relay_catalog
             ORDER BY relay_url ASC",
        )
        .map_err(map_sqlite_error)?;
    let rows = statement
        .query_map([], |row| {
            let source = match row.get::<_, String>(2)?.as_str() {
                "automatic" => RelayCatalogSource::Automatic,
                "custom" => RelayCatalogSource::Custom,
                _ => {
                    return Err(rusqlite::Error::FromSqlConversionFailure(
                        2,
                        rusqlite::types::Type::Text,
                        "invalid relay source".into(),
                    ));
                }
            };
            Ok(RelayCatalogEntry {
                url: row.get(0)?,
                created_at: row.get(1)?,
                source,
                is_active: row.get::<_, i64>(3)? != 0,
                is_dns_discovered: row.get::<_, i64>(4)? != 0,
                is_user_excluded: row.get::<_, i64>(5)? != 0,
                dns_priority: row.get(6)?,
                dns_weight: row.get(7)?,
                dns_expires_at: row.get(8)?,
            })
        })
        .map_err(map_sqlite_error)?;
    let mut entries = Vec::new();
    for row in rows {
        entries.push(row.map_err(map_sqlite_error)?);
    }
    if entries.len() > crate::MAX_RELAY_CATALOG_ENTRIES {
        return Err(SoftchatError::InvalidRelayCatalog);
    }
    Ok(entries)
}

fn load_claimable(
    connection: &Connection,
    relay_url: &str,
) -> Result<Vec<DeliveryIntentSnapshot>, SoftchatError> {
    let mut statement = connection
        .prepare(
            "SELECT intent_id, operation_id, event_copy_id, relay_url, payload_bytes,
                    state, lease_id, attempt_count
             FROM relay_delivery_intents
             WHERE state IN ('pending', 'retryable') AND lease_id = ''
               AND (?1 = '' OR relay_url = ?1)
             ORDER BY created_at ASC, intent_id ASC
             LIMIT 100",
        )
        .map_err(map_sqlite_error)?;
    let rows = statement
        .query_map([relay_url], row_to_intent)
        .map_err(map_sqlite_error)?;
    let mut intents = Vec::new();
    for row in rows {
        intents.push(row.map_err(map_sqlite_error)?);
    }
    Ok(intents)
}

fn load_operation_intents(
    connection: &Connection,
    operation_id: &str,
) -> Result<Vec<DeliveryIntentSnapshot>, SoftchatError> {
    let mut statement = connection
        .prepare(
            "SELECT intent_id, operation_id, event_copy_id, relay_url, payload_bytes,
                    state, lease_id, attempt_count
             FROM relay_delivery_intents
             WHERE operation_id = ?1
             ORDER BY intent_id ASC",
        )
        .map_err(map_sqlite_error)?;
    let rows = statement
        .query_map([operation_id], row_to_intent)
        .map_err(map_sqlite_error)?;
    let mut intents = Vec::new();
    for row in rows {
        intents.push(row.map_err(map_sqlite_error)?);
    }
    Ok(intents)
}

fn row_to_intent(row: &rusqlite::Row<'_>) -> rusqlite::Result<DeliveryIntentSnapshot> {
    let state_text = row.get::<_, String>(5)?;
    let state = intent_state_from_name(&state_text).ok_or_else(|| {
        rusqlite::Error::FromSqlConversionFailure(
            5,
            rusqlite::types::Type::Text,
            "invalid durable delivery state".into(),
        )
    })?;
    let raw_payload_bytes = row.get::<_, i64>(4)?;
    let payload_bytes = u64::try_from(raw_payload_bytes).map_err(|_| {
        rusqlite::Error::FromSqlConversionFailure(
            4,
            rusqlite::types::Type::Integer,
            "negative durable payload size".into(),
        )
    })?;
    Ok(DeliveryIntentSnapshot {
        intent_id: row.get(0)?,
        operation_id: row.get(1)?,
        event_copy_id: row.get(2)?,
        relay_url: row.get(3)?,
        payload_bytes,
        state,
        lease_id: row.get(6)?,
        attempt_count: row.get(7)?,
    })
}

fn load_claim_payloads(
    connection: &Connection,
    claim: &DeliveryClaim,
) -> Result<Vec<DeliveryPayload>, SoftchatError> {
    let mut payloads = Vec::with_capacity(claim.intents.len());
    for intent in &claim.intents {
        let event_json = connection
            .query_row(
                "SELECT event_json FROM outgoing_event_copies WHERE event_copy_id = ?1",
                [&intent.event_copy_id],
                |row| row.get::<_, String>(0),
            )
            .map_err(map_sqlite_error)?;
        payloads.push(DeliveryPayload {
            intent_id: intent.intent_id.clone(),
            relay_url: intent.relay_url.clone(),
            event_json,
        });
    }
    Ok(payloads)
}

fn validate_sticker(sticker: &StickerRecord) -> Result<(), SoftchatError> {
    if sticker.name.is_empty()
        || sticker.name.len() > MAX_STICKER_NAME_BYTES
        || sticker
            .name
            .chars()
            .any(|character| character.is_control() || character == ':')
        || sticker.associated_emojis.len() > MAX_STICKER_EMOJIS
        || sticker.associated_emojis.iter().any(|emoji| {
            emoji.is_empty()
                || emoji.len() > MAX_STICKER_EMOJI_BYTES
                || emoji.chars().any(char::is_control)
        })
    {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    validate_http_url(&sticker.url, true)
}

fn validate_local_contact(contact: &LocalContactRecord) -> Result<(), SoftchatError> {
    crate::NostrPublicKey::from_hex(&contact.public_key)?;
    if contact.updated_at_millis < 0
        || contact.name.as_ref().is_some_and(|name| {
            name.is_empty()
                || name.len() > MAX_LOCAL_CONTACT_NAME_BYTES
                || name.chars().any(char::is_control)
        })
    {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    Ok(())
}

fn validate_link_preview(preview: &LinkPreviewRecord) -> Result<(), SoftchatError> {
    validate_http_url(&preview.url, false)?;
    if preview.updated_at_millis < 0
        || preview
            .title
            .as_ref()
            .is_some_and(|value| value.len() > MAX_LINK_PREVIEW_TITLE_BYTES)
        || preview
            .description
            .as_ref()
            .is_some_and(|value| value.len() > MAX_LINK_PREVIEW_DESCRIPTION_BYTES)
    {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    if let Some(image_url) = &preview.image_url {
        validate_http_url(image_url, false)?;
    }
    Ok(())
}

fn validate_http_url(value: &str, https_only: bool) -> Result<(), SoftchatError> {
    if value.is_empty() || value.len() > MAX_CACHE_URL_BYTES {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    let parsed = Url::parse(value).map_err(|_| SoftchatError::InvalidAccountOperation)?;
    if parsed.host_str().is_none()
        || parsed.username() != ""
        || parsed.password().is_some()
        || (https_only && parsed.scheme() != "https")
        || (!https_only && !matches!(parsed.scheme(), "http" | "https"))
    {
        return Err(SoftchatError::InvalidAccountOperation);
    }
    Ok(())
}

fn projection_from_raw(row: RawProjection) -> Result<AccountProjection, SoftchatError> {
    let outer = SignedNostrEvent::from_json(&row.outer_json)?;
    let rumor = row
        .rumor_json
        .map(|json| {
            serde_json::from_str::<RumorEvent>(&json)
                .map_err(|_| SoftchatError::InvalidPersistenceResult)
        })
        .transpose()?;
    Ok(AccountProjection {
        kind: projection_kind_from_name(&row.stored_kind)
            .ok_or(SoftchatError::InvalidPersistenceResult)?,
        logical_event_id: row.logical_event_id,
        outer_event: SignedEvent::from(&outer),
        rumor,
        author_public_key: row.author_public_key,
        created_at: row.created_at,
        conversation_id: row.conversation_id,
        participants: parse_hex_list(row.participants_csv, true)?,
        target_event_ids: parse_hex_list(row.targets_csv, false)?,
        content: row.content,
    })
}

fn event_node_from_raw(row: RawEventNode) -> Result<AccountEventNode, SoftchatError> {
    let edited = row.edited_content.is_some();
    let effective_content = row
        .edited_content
        .clone()
        .unwrap_or_else(|| row.content.clone());
    let projection = projection_from_raw(RawProjection {
        stored_kind: row.stored_kind,
        logical_event_id: row.logical_event_id,
        author_public_key: row.author_public_key,
        created_at: row.created_at,
        conversation_id: row.conversation_id,
        content: row.content,
        outer_json: row.outer_json,
        rumor_json: row.rumor_json,
        participants_csv: row.participants_csv,
        targets_csv: row.targets_csv,
    })?;
    Ok(AccountEventNode {
        projection,
        effective_content,
        deleted: row.deleted,
        edited,
        reply_count: row.reply_count,
    })
}

fn parse_hex_list(
    value: Option<String>,
    contains_public_keys: bool,
) -> Result<Vec<String>, SoftchatError> {
    let Some(value) = value else {
        return Ok(Vec::new());
    };
    let mut values = value.split(',').map(str::to_owned).collect::<Vec<String>>();
    for item in &values {
        if contains_public_keys {
            crate::NostrPublicKey::from_hex(item)?;
        } else {
            crate::NostrEventId::from_hex(item)?;
        }
    }
    if contains_public_keys {
        values.sort_unstable();
        values.dedup();
    }
    Ok(values)
}

fn table_exists(connection: &Connection, name: &str) -> Result<bool, SoftchatError> {
    connection
        .query_row(
            "SELECT EXISTS(
               SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ?1
             )",
            [name],
            |row| row.get(0),
        )
        .map_err(map_sqlite_error)
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum AccountSchemaOwnership {
    Empty,
    Owned,
    Foreign,
}

fn account_schema_ownership(
    connection: &Connection,
) -> Result<AccountSchemaOwnership, SoftchatError> {
    if table_exists(connection, "softchat_meta")? {
        return Ok(AccountSchemaOwnership::Owned);
    }
    let has_application_tables = connection
        .query_row(
            "SELECT EXISTS(
               SELECT 1 FROM sqlite_master
               WHERE type = 'table' AND name NOT LIKE 'sqlite_%'
             )",
            [],
            |row| row.get(0),
        )
        .map_err(map_sqlite_error)?;
    Ok(if has_application_tables {
        AccountSchemaOwnership::Foreign
    } else {
        AccountSchemaOwnership::Empty
    })
}

fn current_schema_matches(
    connection: &Connection,
    account_id: &str,
) -> Result<bool, SoftchatError> {
    let (schema_version, stored_account_id) = connection
        .query_row(
            "SELECT
               MAX(CASE WHEN key = 'schema_version' THEN integer_value END),
               MAX(CASE WHEN key = 'account_id' THEN text_value END)
             FROM softchat_meta
             WHERE key IN ('schema_version', 'account_id')",
            [],
            |row| {
                Ok((
                    row.get::<_, Option<i64>>(0)?,
                    row.get::<_, Option<String>>(1)?,
                ))
            },
        )
        .map_err(map_sqlite_error)?;
    Ok(schema_version == Some(ACCOUNT_SCHEMA_VERSION)
        && stored_account_id.as_deref() == Some(account_id))
}

pub(crate) fn bump_revision(transaction: &Connection) -> Result<i64, SoftchatError> {
    transaction
        .query_row(
            "UPDATE softchat_meta SET integer_value = integer_value + 1
             WHERE key = 'revision'
             RETURNING integer_value",
            [],
            |row| row.get(0),
        )
        .map_err(map_sqlite_error)
}

fn meta_integer(connection: &Connection, key: &str) -> Result<i64, SoftchatError> {
    meta_integer_optional(connection, key)?.ok_or(SoftchatError::InvalidPersistenceResult)
}

fn bounded_query_count(connection: &Connection, query: &str) -> Result<u32, SoftchatError> {
    let value = connection
        .query_row(query, [], |row| row.get::<_, i64>(0))
        .map_err(map_sqlite_error)?;
    u32::try_from(value).map_err(|_| SoftchatError::InvalidPersistenceResult)
}

fn meta_integer_optional(connection: &Connection, key: &str) -> Result<Option<i64>, SoftchatError> {
    connection
        .query_row(
            "SELECT integer_value FROM softchat_meta WHERE key = ?1",
            [key],
            |row| row.get(0),
        )
        .optional()
        .map_err(map_sqlite_error)
}

fn validate_timestamp(value: i64) -> Result<(), SoftchatError> {
    let unsigned = u64::try_from(value).map_err(|_| SoftchatError::InvalidEventTimestamp)?;
    if unsigned > crate::MAX_PORTABLE_TIMESTAMP_SECONDS {
        return Err(SoftchatError::InvalidEventTimestamp);
    }
    Ok(())
}

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

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

const fn projection_kind_name(kind: ProjectionKind) -> &'static str {
    match kind {
        ProjectionKind::ChatMessage => "chat_message",
        ProjectionKind::Subject => "subject",
        ProjectionKind::Reaction => "reaction",
        ProjectionKind::Deletion => "deletion",
        ProjectionKind::Edit => "edit",
        ProjectionKind::Typing => "typing",
        ProjectionKind::UserMetadata => "user_metadata",
        ProjectionKind::FollowList => "follow_list",
        ProjectionKind::ApplicationData => "application_data",
        ProjectionKind::AppDataSync => "app_data_sync",
        ProjectionKind::GenericRepost => "generic_repost",
        ProjectionKind::Unknown => "unknown",
    }
}

fn projection_kind_from_name(value: &str) -> Option<ProjectionKind> {
    match value {
        "chat_message" => Some(ProjectionKind::ChatMessage),
        "subject" => Some(ProjectionKind::Subject),
        "reaction" => Some(ProjectionKind::Reaction),
        "deletion" => Some(ProjectionKind::Deletion),
        "edit" => Some(ProjectionKind::Edit),
        "typing" => Some(ProjectionKind::Typing),
        "user_metadata" => Some(ProjectionKind::UserMetadata),
        "follow_list" => Some(ProjectionKind::FollowList),
        "application_data" => Some(ProjectionKind::ApplicationData),
        "app_data_sync" => Some(ProjectionKind::AppDataSync),
        "generic_repost" => Some(ProjectionKind::GenericRepost),
        "unknown" => Some(ProjectionKind::Unknown),
        _ => None,
    }
}

const fn intent_state_name(state: DeliveryIntentState) -> &'static str {
    match state {
        DeliveryIntentState::Pending => "pending",
        DeliveryIntentState::Claimed => "claimed",
        DeliveryIntentState::SocketWritten => "socket_written",
        DeliveryIntentState::Accepted => "accepted",
        DeliveryIntentState::Rejected => "rejected",
        DeliveryIntentState::Retryable => "retryable",
        DeliveryIntentState::Cancelled => "cancelled",
    }
}

const UNFINISHED_DELIVERY_STATES: [DeliveryIntentState; 4] = [
    DeliveryIntentState::Pending,
    DeliveryIntentState::Claimed,
    DeliveryIntentState::SocketWritten,
    DeliveryIntentState::Retryable,
];

fn unfinished_delivery_count(connection: &Connection) -> Result<u32, SoftchatError> {
    let [pending, claimed, socket_written, retryable] =
        UNFINISHED_DELIVERY_STATES.map(intent_state_name);
    let value = connection
        .query_row(
            "SELECT COUNT(*) FROM relay_delivery_intents
             WHERE state IN (?1, ?2, ?3, ?4)",
            params![pending, claimed, socket_written, retryable],
            |row| row.get::<_, i64>(0),
        )
        .map_err(map_sqlite_error)?;
    u32::try_from(value).map_err(|_| SoftchatError::InvalidPersistenceResult)
}

fn intent_state_from_name(value: &str) -> Option<DeliveryIntentState> {
    match value {
        "pending" => Some(DeliveryIntentState::Pending),
        "claimed" => Some(DeliveryIntentState::Claimed),
        "socket_written" => Some(DeliveryIntentState::SocketWritten),
        "accepted" => Some(DeliveryIntentState::Accepted),
        "rejected" => Some(DeliveryIntentState::Rejected),
        "retryable" => Some(DeliveryIntentState::Retryable),
        "cancelled" => Some(DeliveryIntentState::Cancelled),
        _ => None,
    }
}

const fn relay_source_name(source: RelayCatalogSource) -> &'static str {
    match source {
        RelayCatalogSource::Automatic => "automatic",
        RelayCatalogSource::Custom => "custom",
    }
}

pub(crate) fn map_sqlite_error(error: rusqlite::Error) -> SoftchatError {
    let _ = error;
    SoftchatError::InvalidPersistenceResult
}

#[cfg(test)]
mod tests {
    #![allow(clippy::expect_used)]

    use super::*;
    use crate::{AttachmentMetadata, ChatMessageDraft, ChatRelation, SubjectDraft};

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

    fn database(account_id: &str) -> AccountDatabase {
        AccountDatabase::open(":memory:", account_id.to_owned()).expect("open database")
    }

    fn legacy_oversized_event(identity: &LocalIdentity) -> Result<SignedNostrEvent, SoftchatError> {
        let sign = |content: String| {
            identity.sign_event(crate::NostrEventDraft::new(
                1_700_000_002,
                crate::NostrEventKind::SHORT_TEXT_NOTE,
                Vec::new(),
                content,
            )?)
        };
        let content_bytes = crate::MAX_RELAY_FRAME_BYTES - "[\"EVENT\",]".len() + 1
            - sign(String::new())?.to_json()?.len();
        sign(format!(
            "{}{}",
            "🦀".repeat(content_bytes / 4),
            "x".repeat(content_bytes % 4)
        ))
    }

    #[test]
    fn legacy_wire_oversized_events_do_not_hide_sync_fingerprints()
    -> Result<(), Box<dyn std::error::Error>> {
        let identity = LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let mut database = database("legacy-sync-frame");
        let valid = identity.sign_event(crate::NostrEventDraft::new(
            1_700_000_001,
            crate::NostrEventKind::SHORT_TEXT_NOTE,
            Vec::new(),
            "valid older event",
        )?)?;
        database.ingest(&identity, vec![SignedEvent::from(&valid)], 1_700_000_003)?;
        let oversized = legacy_oversized_event(&identity)?;
        let json = oversized.to_json()?;
        // Insert the canonical shape accepted by the previous account profile.
        database.connection.execute(
            "INSERT INTO signed_events (event_id, canonical_json, author_public_key, kind, created_at, received_at)
             VALUES (?1, ?2, ?3, 1, 1700000002, 1700000003)",
            params![oversized.id().to_hex(), json, identity.public_key().to_hex()],
        )?;
        database.connection.execute(
            "INSERT INTO event_fingerprints(event_id, created_at) VALUES (?1, 1700000002)",
            [oversized.id().to_hex()],
        )?;
        let page = database.sync_fingerprints(None, 1_700_000_003, 1)?;
        assert_eq!(page.len(), 1);
        assert_eq!(page[0].event_id, valid.id().to_hex());
        assert_eq!(database.event_json(&oversized.id().to_hex())?, json);
        assert!(SignedNostrEvent::from_json(&json).is_ok());
        Ok(())
    }

    #[test]
    fn legacy_wire_oversized_deliveries_fail_locally_without_starving_valid_work()
    -> Result<(), Box<dyn std::error::Error>> {
        let identity = LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let mut database = database("legacy-delivery-frame");
        let relay = database.active_relay_url()?;
        let oversized = legacy_oversized_event(&identity)?;
        let json = oversized.to_json()?;
        database.connection.execute("INSERT INTO outgoing_operations(operation_id, account_id, created_at) VALUES ('legacy-large', 'legacy-delivery-frame', 1)", [])?;
        database.connection.execute("INSERT INTO outgoing_event_copies(event_copy_id, operation_id, recipient_public_key, event_json) VALUES (?1, 'legacy-large', '', ?2)", params![oversized.id().to_hex(), json])?;
        for (index, state, expires) in [
            (0, "pending", 0),
            (1, "claimed", 1),
            (2, "accepted", 0),
            (3, "rejected", 0),
        ] {
            database.connection.execute(
                "INSERT INTO relay_delivery_intents(intent_id, operation_id, event_copy_id, relay_url, payload_bytes, state, lease_id, lease_expires_at, attempt_count, last_category, created_at)
                 VALUES (?1, 'legacy-large', ?2, ?3, ?4, ?5, ?6, ?7, 1, 'previous', 1)",
                params![format!("{index:064x}"), oversized.id().to_hex(), if index == 0 { relay.clone() } else { format!("wss://legacy-{index}.example/") }, i64::try_from(json.len())?, state, if expires == 0 { "" } else { "old-lease" }, expires],
            )?;
        }
        let valid = identity.sign_event(crate::NostrEventDraft::new(
            1_700_000_001,
            crate::NostrEventKind::SHORT_TEXT_NOTE,
            Vec::new(),
            "valid work",
        )?)?;
        database.enqueue_signed(
            &identity,
            "valid-work".to_owned(),
            vec![SignedEvent::from(&valid)],
            vec![relay.clone()],
            1_700_000_003,
        )?;
        let before = database.info()?.revision;
        let claim = database
            .claim_deliveries("valid-lease".to_owned(), relay.clone(), 1_700_000_004, 30)?
            .ok_or("valid work was starved")?;
        assert_eq!(claim.payloads.len(), 1);
        assert_eq!(
            SignedNostrEvent::from_json(&claim.payloads[0].event_json)?.id(),
            valid.id()
        );
        assert_eq!(database.info()?.revision, before + 1);
        for (index, expected, category) in [
            (0, "rejected", "event_too_large"),
            (1, "rejected", "event_too_large"),
            (2, "accepted", "previous"),
            (3, "rejected", "previous"),
        ] {
            let actual = database.connection.query_row("SELECT state, last_category, lease_id, lease_expires_at FROM relay_delivery_intents WHERE intent_id = ?1", [format!("{index:064x}")], |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?, row.get::<_, String>(2)?, row.get::<_, i64>(3)?)))?;
            assert_eq!(
                actual,
                (expected.to_owned(), category.to_owned(), String::new(), 0)
            );
        }
        assert_eq!(
            database.connection.query_row(
                "SELECT event_json FROM outgoing_event_copies WHERE event_copy_id = ?1",
                [oversized.id().to_hex()],
                |row| row.get::<_, String>(0)
            )?,
            json
        );
        // A cleanup-only claim must also publish a new durable revision.
        database.connection.execute(
            "UPDATE relay_delivery_intents SET state = 'retryable' WHERE intent_id = ?1",
            [format!("{:064x}", 0)],
        )?;
        let before_cleanup = database.info()?.revision;
        assert!(
            database
                .claim_deliveries("cleanup-lease".to_owned(), relay, 1_700_000_005, 30)?
                .is_none()
        );
        assert_eq!(database.info()?.revision, before_cleanup + 1);
        Ok(())
    }

    #[test]
    fn delivery_expiration_on_another_relay_advances_revision_without_a_new_claim()
    -> Result<(), Box<dyn std::error::Error>> {
        let identity = LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let mut database = database("expiration-revision");
        let other_relay = "wss://other-relay.example/".to_owned();
        let event = identity.sign_event(crate::NostrEventDraft::new(
            1_700_000_000,
            crate::NostrEventKind::SHORT_TEXT_NOTE,
            Vec::new(),
            "leased elsewhere",
        )?)?;
        database.enqueue_signed(
            &identity,
            "expired-elsewhere".to_owned(),
            vec![SignedEvent::from(&event)],
            vec![other_relay.clone()],
            1_700_000_000,
        )?;
        assert!(
            database
                .claim_deliveries("other-lease".to_owned(), other_relay, 1_700_000_000, 1)?
                .is_some()
        );
        let before = database.info()?.revision;
        let active_relay = database.active_relay_url()?;
        assert!(
            database
                .claim_deliveries(
                    "active-lease".to_owned(),
                    active_relay.clone(),
                    1_700_000_002,
                    30
                )?
                .is_none()
        );
        assert_eq!(database.info()?.revision, before + 1);
        let operation = database.operation("expired-elsewhere")?;
        assert_eq!((operation.queued_intents, operation.active_intents), (1, 0));
        assert!(
            database
                .claim_deliveries(
                    "active-next-lease".to_owned(),
                    active_relay,
                    1_700_000_002,
                    30
                )?
                .is_none()
        );
        assert_eq!(database.info()?.revision, before + 1);
        Ok(())
    }

    #[test]
    fn ingest_is_atomic_idempotent_and_queryable() {
        let alice = LocalIdentity::from_secret_hex(ALICE_SECRET).expect("alice");
        let bob = LocalIdentity::from_secret_hex(BOB_SECRET).expect("bob");
        let message = alice
            .create_chat_message(ChatMessageDraft {
                created_at: 1_700_000_000,
                participants: vec![bob.public_key()],
                content: "hello".to_owned(),
                relation: ChatRelation::Reply {
                    event_id: crate::NostrEventId::from_hex(&hex::encode([9_u8; 32]))
                        .expect("reply target"),
                    relay_hint: None,
                    legacy_unmarked: false,
                },
                attachments: Vec::new(),
                emoji_tags: Vec::new(),
                extension_tags: Vec::new(),
            })
            .expect("message");
        let rumor = crate::NostrRumor::try_from(message.rumor).expect("rumor");
        let wrap = alice
            .gift_wrap(
                &alice.public_key(),
                &rumor,
                Nip59EnvelopeKind::Durable,
                1_700_000_100,
            )
            .expect("wrap");

        let mut database = database("alice");
        let logs = std::sync::Arc::new(crate::account_diagnostics::AccountLogBuffer::new(
            AccountLogLevel::Debug,
        ));
        database.diagnostics.buffer = Some(logs.clone());
        let failed: Result<(), SoftchatError> =
            database.apply_transport_result_atomically(|database| {
                database.ingest(&alice, vec![SignedEvent::from(&wrap)], 1_700_000_199)?;
                assert!(
                    logs.drain(64)?.events.is_empty(),
                    "A released savepoint is not a commit"
                );
                Err(SoftchatError::InvalidPersistenceResult)
            });
        assert!(failed.is_err());
        assert!(
            logs.drain(64)
                .expect("rollback diagnostics")
                .events
                .is_empty()
        );
        let first = database
            .ingest(&alice, vec![SignedEvent::from(&wrap)], 1_700_000_200)
            .expect("ingest");
        assert_eq!(first.revision, 1);
        assert_eq!(first.receipt.inserted_event_ids.len(), 1);
        assert_eq!(first.projections.len(), 1);
        assert_eq!(first.projections[0].content, "hello");
        assert!(first.ephemeral_projections.is_empty());
        let duplicate = database
            .ingest(&alice, vec![SignedEvent::from(&wrap)], 1_700_000_201)
            .expect("duplicate");
        assert_eq!(duplicate.receipt.duplicate_event_ids.len(), 1);
        assert_eq!(duplicate.projections.len(), 1);
        let conversations = database.list_conversations(20, 0).expect("conversations");
        assert_eq!(conversations.len(), 1);
        assert_eq!(conversations[0].last_message, "hello");
        let conversation_views = database
            .list_conversation_views(20, 0)
            .expect("conversation views");
        assert_eq!(conversation_views.len(), 1);
        assert_eq!(
            conversation_views[0].last_message.projection.content,
            "hello"
        );
        assert!(conversation_views[0].subject.is_none());
        let messages = database
            .list_messages(&conversations[0].conversation_id, None, 20)
            .expect("messages");
        assert_eq!(messages.len(), 1);
        assert_eq!(messages[0].content, "hello");
        let projections = database
            .list_projections(
                Some(ProjectionKind::ChatMessage),
                String::new(),
                conversations[0].conversation_id.clone(),
                String::new(),
                alice.public_key().to_hex(),
                None,
                20,
                0,
            )
            .expect("projections");
        assert_eq!(projections.len(), 1);
        assert_eq!(projections[0].logical_event_id, rumor.id().to_hex());
        assert_eq!(projections[0].content, "hello");
        assert_eq!(
            projections[0].participants,
            vec![alice.public_key().to_hex(), bob.public_key().to_hex()]
        );
        assert_eq!(
            projections[0]
                .rumor
                .as_ref()
                .expect("authenticated rumor")
                .id,
            rumor.id().to_hex()
        );
        let events = logs.drain(64).expect("diagnostics").events;
        assert_eq!(
            events
                .iter()
                .filter(|event| event.operation == "message.received")
                .count(),
            1
        );
        let received = &events[0];
        assert_eq!(received.source, Some(AccountLogSource::IncomingApi));
        assert_eq!(received.subject, Some(AccountLogSubject::ChatMessage));
        assert_eq!(
            received.relation,
            Some(crate::account_diagnostics::AccountLogRelation::Reply)
        );
        assert!(
            received
                .counters
                .iter()
                .any(|counter| counter.kind == Counter::LogicalEvents && counter.value == 1)
        );
        assert!(
            events
                .iter()
                .any(|event| event.operation == "ingestion.duplicate")
        );
        assert!(
            events
                .iter()
                .all(|event| !format!("{event:?}").contains("hello"))
        );
    }

    #[test]
    fn diagnostics_count_every_and_only_unfinished_delivery_state() {
        let database = database("diagnostics");
        database
            .connection
            .execute(
                "INSERT INTO outgoing_operations
                 (operation_id, account_id, created_at, command_hash, result_message_id)
                 VALUES ('diagnostics-operation', 'diagnostics', 1, '', '')",
                [],
            )
            .expect("operation");
        let states = [
            DeliveryIntentState::Pending,
            DeliveryIntentState::Claimed,
            DeliveryIntentState::SocketWritten,
            DeliveryIntentState::Accepted,
            DeliveryIntentState::Rejected,
            DeliveryIntentState::Retryable,
            DeliveryIntentState::Cancelled,
        ];
        for (index, state) in states.into_iter().enumerate() {
            let event_copy_id = format!("diagnostics-copy-{index}");
            database
                .connection
                .execute(
                    "INSERT INTO outgoing_event_copies
                     (event_copy_id, operation_id, recipient_public_key, event_json)
                     VALUES (?1, 'diagnostics-operation', ?2, '{}')",
                    params![event_copy_id, format!("recipient-{index}")],
                )
                .expect("event copy");
            database
                .connection
                .execute(
                    "INSERT INTO relay_delivery_intents
                     (intent_id, operation_id, event_copy_id, relay_url, payload_bytes,
                      state, lease_id, lease_expires_at, attempt_count, last_category, created_at)
                     VALUES (?1, 'diagnostics-operation', ?2, ?3, 2, ?4, '', 0, 0, '', 1)",
                    params![
                        format!("diagnostics-intent-{index}"),
                        event_copy_id,
                        format!("wss://relay-{index}.example"),
                        intent_state_name(state),
                    ],
                )
                .expect("delivery intent");
            assert_eq!(
                intent_state_from_name(intent_state_name(state)),
                Some(state),
            );
        }

        assert_eq!(
            database
                .diagnostics()
                .expect("diagnostics")
                .unfinished_delivery_count,
            u32::try_from(UNFINISHED_DELIVERY_STATES.len()).expect("bounded states"),
        );
    }

    #[test]
    fn fresh_wrappers_share_one_logical_rumor_and_retain_every_provenance_edge() {
        let alice = LocalIdentity::from_secret_hex(ALICE_SECRET).expect("alice");
        let bob = LocalIdentity::from_secret_hex(BOB_SECRET).expect("bob");
        let message = alice
            .create_chat_message(ChatMessageDraft {
                created_at: 1_700_000_000,
                participants: vec![bob.public_key()],
                content: "one rumor, three wrappers".to_owned(),
                relation: ChatRelation::None,
                attachments: Vec::new(),
                emoji_tags: Vec::new(),
                extension_tags: Vec::new(),
            })
            .expect("message");
        let rumor = crate::NostrRumor::try_from(message.rumor).expect("rumor");
        let first_wrap = alice
            .gift_wrap(
                &alice.public_key(),
                &rumor,
                Nip59EnvelopeKind::Durable,
                1_700_000_100,
            )
            .expect("first wrap");
        let second_wrap = alice
            .gift_wrap(
                &alice.public_key(),
                &rumor,
                Nip59EnvelopeKind::Durable,
                1_700_000_101,
            )
            .expect("second wrap");
        let notification_wrap = alice
            .gift_wrap(
                &alice.public_key(),
                &rumor,
                Nip59EnvelopeKind::Durable,
                1_700_000_102,
            )
            .expect("notification wrap");
        assert_ne!(first_wrap.id(), second_wrap.id());
        assert_ne!(second_wrap.id(), notification_wrap.id());

        let mut database = database("alice");
        let logs = std::sync::Arc::new(crate::account_diagnostics::AccountLogBuffer::new(
            AccountLogLevel::Debug,
        ));
        database.diagnostics.buffer = Some(logs.clone());
        let first = database
            .ingest(&alice, vec![SignedEvent::from(&first_wrap)], 1_700_000_200)
            .expect("first ingest");
        let second = database
            .ingest(&alice, vec![SignedEvent::from(&second_wrap)], 1_700_000_201)
            .expect("second ingest");
        let notification = database
            .ingest_with_authenticated_replays(
                &alice,
                vec![SignedEvent::from(&notification_wrap)],
                1_700_000_202,
            )
            .expect("notification ingest");

        assert_eq!(first.projections.len(), 1);
        assert!(second.projections.is_empty());
        assert_eq!(notification.projections.len(), 1);
        assert_eq!(
            notification.projections[0].logical_event_id,
            rumor.id().to_hex()
        );
        assert_eq!(
            second.receipt.inserted_event_ids,
            vec![second_wrap.id().to_hex()]
        );
        assert_eq!(
            notification.receipt.inserted_event_ids,
            vec![notification_wrap.id().to_hex()]
        );
        assert_eq!(
            database
                .connection
                .query_row("SELECT COUNT(*) FROM signed_events", [], |row| {
                    row.get::<_, i64>(0)
                })
                .expect("signed events"),
            3
        );
        assert_eq!(
            database
                .connection
                .query_row("SELECT COUNT(*) FROM authenticated_rumors", [], |row| {
                    row.get::<_, i64>(0)
                })
                .expect("rumors"),
            1
        );
        assert_eq!(
            database
                .connection
                .query_row(
                    "SELECT COUNT(*) FROM authenticated_rumor_wrappers WHERE rumor_id = ?1",
                    [rumor.id().to_hex()],
                    |row| row.get::<_, i64>(0),
                )
                .expect("wrapper provenance"),
            3
        );
        assert_eq!(
            database
                .connection
                .query_row("SELECT COUNT(*) FROM projections", [], |row| {
                    row.get::<_, i64>(0)
                })
                .expect("projections"),
            1
        );
        logs.flush();
        let events = logs.drain(64).expect("diagnostics").events;
        assert_eq!(
            events
                .iter()
                .filter(|event| event.operation == "message.received")
                .count(),
            1
        );
        let copies: u64 = events
            .iter()
            .flat_map(|event| &event.counters)
            .filter(|counter| counter.kind == Counter::KnownCopies)
            .map(|counter| counter.value)
            .sum();
        assert_eq!(
            copies, 2,
            "Notification replays must not announce a new logical message"
        );
    }

    #[test]
    fn version_eleven_rumors_gain_wrapper_provenance() {
        let alice = LocalIdentity::from_secret_hex(ALICE_SECRET).expect("alice");
        let bob = LocalIdentity::from_secret_hex(BOB_SECRET).expect("bob");
        let message = alice
            .create_chat_message(ChatMessageDraft {
                created_at: 1_700_000_000,
                participants: vec![bob.public_key()],
                content: "migrate provenance".to_owned(),
                relation: ChatRelation::None,
                attachments: Vec::new(),
                emoji_tags: Vec::new(),
                extension_tags: Vec::new(),
            })
            .expect("message");
        let rumor = crate::NostrRumor::try_from(message.rumor).expect("rumor");
        let wrap = alice
            .gift_wrap(
                &alice.public_key(),
                &rumor,
                Nip59EnvelopeKind::Durable,
                1_700_000_100,
            )
            .expect("wrap");
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("clock")
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "softchat-provenance-migration-{}-{nonce}.sqlite",
            std::process::id()
        ));
        {
            let mut database = AccountDatabase::open(&path, "alice".to_owned()).expect("create");
            database
                .ingest(&alice, vec![SignedEvent::from(&wrap)], 1_700_000_200)
                .expect("ingest");
        }
        {
            let legacy = Connection::open(&path).expect("legacy database");
            legacy
                .execute_batch(
                    "DROP TABLE authenticated_rumor_wrappers;
                     DROP TABLE conversation_subject_icons;
                     ALTER TABLE media_operations DROP COLUMN operation_kind;
                     UPDATE softchat_meta SET integer_value = 11
                     WHERE key = 'schema_version';",
                )
                .expect("downgrade fixture");
        }

        let migrated = AccountDatabase::open(&path, "alice".to_owned()).expect("migrate");
        assert_eq!(migrated.info().expect("info").schema_version, 16);
        assert_eq!(
            migrated
                .connection
                .query_row(
                    "SELECT outer_event_id FROM authenticated_rumor_wrappers
                     WHERE rumor_id = ?1",
                    [rumor.id().to_hex()],
                    |row| row.get::<_, String>(0),
                )
                .expect("backfilled wrapper"),
            wrap.id().to_hex()
        );
        drop(migrated);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn version_twelve_reconciles_equal_time_app_data_to_the_lowest_event_id() {
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("clock")
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "softchat-app-data-order-migration-{}-{nonce}.sqlite",
            std::process::id()
        ));
        let low_id = "00".repeat(32);
        let high_id = "ff".repeat(32);
        {
            let database = AccountDatabase::open(&path, "alice".to_owned()).expect("create");
            for (event_id, value) in [(&low_id, "low"), (&high_id, "high")] {
                database
                    .connection
                    .execute(
                        "INSERT INTO signed_events
                         (event_id, canonical_json, author_public_key, kind, created_at,
                          received_at)
                         VALUES (?1, '{}', 'author', 30079, 1700000000, 1700000001)",
                        [event_id],
                    )
                    .expect("signed source");
                database
                    .connection
                    .execute(
                        "INSERT INTO projections
                         (logical_event_id, outer_event_id, kind, author_public_key, created_at,
                          conversation_id, content, ephemeral)
                         VALUES (?1, ?1, 'app_data_sync', 'author', 1700000000, '', ?2, 0)",
                        params![
                            event_id,
                            format!("app-settings:{{\"quickReaction\":\"{value}\"}}")
                        ],
                    )
                    .expect("app-data projection");
            }
            database
                .connection
                .execute(
                    "INSERT INTO app_data_effective
                     (context, canonical_json, event_created_at, event_id)
                     VALUES ('app-settings', '{\"quickReaction\":\"high\"}', 1700000000, ?1)",
                    [&high_id],
                )
                .expect("legacy high-id winner");
            database
                .connection
                .execute(
                    "UPDATE account_settings
                     SET canonical_json = '{\"quickReaction\":\"high\"}',
                         updated_at = 1700000000
                     WHERE singleton = 1",
                    [],
                )
                .expect("legacy effective settings");
            database
                .connection
                .execute_batch(
                    "DROP TABLE conversation_subject_icons;
                     ALTER TABLE media_operations DROP COLUMN operation_kind;
                     UPDATE softchat_meta SET integer_value = 12
                     WHERE key = 'schema_version';",
                )
                .expect("downgrade fixture");
        }

        let migrated = AccountDatabase::open(&path, "alice".to_owned()).expect("migrate");
        assert_eq!(migrated.info().expect("info").schema_version, 16);
        assert_eq!(
            migrated
                .connection
                .query_row(
                    "SELECT event_id FROM app_data_effective WHERE context = 'app-settings'",
                    [],
                    |row| row.get::<_, String>(0),
                )
                .expect("effective event"),
            low_id
        );
        assert_eq!(
            migrated
                .product_settings()
                .expect("effective settings")
                .quick_reaction
                .as_deref(),
            Some("low")
        );
        drop(migrated);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn version_thirteen_rebuilds_the_effective_inherited_subject_icon() {
        let alice = LocalIdentity::from_secret_hex(ALICE_SECRET).expect("alice");
        let bob = LocalIdentity::from_secret_hex(BOB_SECRET).expect("bob");
        let icon = AttachmentMetadata {
            url: "https://media.example/inherited.png".to_owned(),
            sha256: Some(hex::encode([7_u8; 32])),
            dimensions: Some("256x256".to_owned()),
            ..AttachmentMetadata::default()
        };
        let icon_subject = alice
            .create_subject(SubjectDraft {
                created_at: 1_700_000_000,
                participants: vec![bob.public_key()],
                subject: "Original".to_owned(),
                icon: Some(icon.clone()),
                emoji_tags: Vec::new(),
                extension_tags: Vec::new(),
            })
            .expect("icon subject");
        let text_subject = alice
            .create_subject(SubjectDraft {
                created_at: 1_700_000_001,
                participants: vec![bob.public_key()],
                subject: "Renamed".to_owned(),
                icon: None,
                emoji_tags: Vec::new(),
                extension_tags: Vec::new(),
            })
            .expect("text subject");
        let wrap = |rumor: RumorEvent, now| {
            let rumor = crate::NostrRumor::try_from(rumor).expect("rumor");
            SignedEvent::from(
                &alice
                    .gift_wrap(&alice.public_key(), &rumor, Nip59EnvelopeKind::Durable, now)
                    .expect("wrap"),
            )
        };
        let icon_subject_id = icon_subject.rumor.id.clone();
        let events = vec![
            wrap(icon_subject.rumor, 1_700_172_800),
            wrap(text_subject.rumor, 1_700_172_801),
        ];
        let conversation_id =
            crate::conversation_id(vec![alice.public_key().to_hex(), bob.public_key().to_hex()])
                .expect("conversation id");
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("clock")
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "softchat-subject-icon-migration-{}-{nonce}.sqlite",
            std::process::id()
        ));
        let icon_operation_id;
        let caller_prefixed_operation_id;
        {
            let mut database = AccountDatabase::open(&path, "alice".to_owned()).expect("create");
            database
                .ingest(&alice, events, 1_700_172_802)
                .expect("ingest subjects");
            assert_eq!(
                database
                    .product_conversation(&conversation_id, &alice.public_key().to_hex())
                    .expect("conversation")
                    .expect("stored conversation")
                    .subject_icon,
                Some(icon.clone())
            );
            icon_operation_id = database
                .prepare_product_conversation_icon_download(&conversation_id, 1_700_172_803)
                .expect("prepare icon download")
                .id;
            caller_prefixed_operation_id = database
                .prepare_product_media(
                    "conversation-icon-download-caller-owned",
                    Some(conversation_id.clone()),
                    "download",
                    "caller-owned-fingerprint",
                    1_700_172_804,
                )
                .expect("prepare caller-named transfer")
                .id;
        }
        {
            let legacy = Connection::open(&path).expect("legacy database");
            legacy
                .execute_batch(
                    "DROP TABLE conversation_subject_icons;
                     ALTER TABLE media_operations DROP COLUMN operation_kind;
                     ALTER TABLE pending_media_messages DROP COLUMN publication_claimed;
                     UPDATE softchat_meta SET integer_value = 13
                     WHERE key = 'schema_version';",
                )
                .expect("downgrade fixture");
        }

        let migrated = AccountDatabase::open(&path, "alice".to_owned()).expect("migrate");
        assert_eq!(migrated.info().expect("info").schema_version, 16);
        assert_eq!(
            migrated
                .connection
                .query_row(
                    "SELECT subject_event_id FROM conversation_subject_icons
                     WHERE conversation_id = ?1",
                    [&conversation_id],
                    |row| row.get::<_, String>(0),
                )
                .expect("effective icon source"),
            icon_subject_id
        );
        let conversation = migrated
            .product_conversation(&conversation_id, &alice.public_key().to_hex())
            .expect("conversation")
            .expect("stored conversation");
        assert_eq!(conversation.subject, "Renamed");
        assert_eq!(conversation.subject_icon, Some(icon));
        assert_eq!(
            migrated
                .connection
                .query_row(
                    "SELECT operation_kind FROM media_operations WHERE operation_id = ?1",
                    [&icon_operation_id],
                    |row| row.get::<_, String>(0),
                )
                .expect("migrated icon operation kind"),
            "conversation_icon"
        );
        assert_eq!(
            migrated
                .connection
                .query_row(
                    "SELECT operation_kind FROM media_operations WHERE operation_id = ?1",
                    [&caller_prefixed_operation_id],
                    |row| row.get::<_, String>(0),
                )
                .expect("migrated caller operation kind"),
            "transfer"
        );
        assert_eq!(
            migrated
                .connection
                .query_row(
                    "SELECT publication_claimed FROM pending_media_messages LIMIT 1",
                    [],
                    |row| row.get::<_, bool>(0),
                )
                .optional()
                .expect("publication claim column"),
            None
        );
        drop(migrated);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn sql_edit_projection_uses_nip01_equal_time_order_in_every_arrival_order() {
        let alice = LocalIdentity::from_secret_hex(ALICE_SECRET).expect("alice");
        let bob = LocalIdentity::from_secret_hex(BOB_SECRET).expect("bob");
        let original = alice
            .create_chat_message(ChatMessageDraft {
                created_at: 1_700_000_000,
                participants: vec![bob.public_key()],
                content: "original".to_owned(),
                relation: ChatRelation::None,
                attachments: Vec::new(),
                emoji_tags: Vec::new(),
                extension_tags: Vec::new(),
            })
            .expect("original");
        let original_id = crate::NostrEventId::from_hex(&original.rumor.id).expect("original id");
        let first = alice
            .create_edit(
                vec![bob.public_key()],
                original_id,
                "alpha".to_owned(),
                Vec::new(),
                1_700_000_001,
            )
            .expect("first edit");
        let second = alice
            .create_edit(
                vec![bob.public_key()],
                original_id,
                "beta".to_owned(),
                Vec::new(),
                1_700_000_001,
            )
            .expect("second edit");
        let expected = if first.rumor.id < second.rumor.id {
            "alpha"
        } else {
            "beta"
        };

        let wrap = |rumor: RumorEvent, now| {
            let rumor = crate::NostrRumor::try_from(rumor).expect("rumor");
            SignedEvent::from(
                &alice
                    .gift_wrap(&alice.public_key(), &rumor, Nip59EnvelopeKind::Durable, now)
                    .expect("wrap"),
            )
        };
        let original_wrap = wrap(original.rumor, 1_700_172_800);
        let first_wrap = wrap(first.rumor, 1_700_172_801);
        let second_wrap = wrap(second.rumor, 1_700_172_802);
        for (index, edits) in [
            vec![first_wrap.clone(), second_wrap.clone()],
            vec![second_wrap.clone(), first_wrap.clone()],
        ]
        .into_iter()
        .enumerate()
        {
            let mut database = database(&format!("alice-{index}"));
            database
                .ingest(&alice, vec![original_wrap.clone()], 1_700_200_000)
                .expect("ingest original");
            for edit in edits {
                database
                    .ingest(&alice, vec![edit], 1_700_200_001)
                    .expect("ingest edit");
            }
            let conversation = database
                .list_conversations(1, 0)
                .expect("conversation")
                .remove(0);
            let message = database
                .list_messages(&conversation.conversation_id, None, 1)
                .expect("message")
                .remove(0);
            assert_eq!(message.content, expected);
        }
    }

    #[test]
    fn enqueue_claim_and_accept_are_durable_transitions() {
        let alice = LocalIdentity::from_secret_hex(ALICE_SECRET).expect("alice");
        let bob = LocalIdentity::from_secret_hex(BOB_SECRET).expect("bob");
        let message = alice
            .create_chat_message(ChatMessageDraft {
                created_at: 1_700_000_000,
                participants: vec![bob.public_key()],
                content: "hello".to_owned(),
                relation: ChatRelation::None,
                attachments: Vec::new(),
                emoji_tags: Vec::new(),
                extension_tags: Vec::new(),
            })
            .expect("message");
        let mut database = database("alice");
        let logs = std::sync::Arc::new(crate::account_diagnostics::AccountLogBuffer::new(
            AccountLogLevel::Debug,
        ));
        database.diagnostics.buffer = Some(logs.clone());
        let queued = database
            .enqueue_rumor(
                &alice,
                "operation-1".to_owned(),
                message.rumor,
                vec![bob.public_key().to_hex()],
                vec![bob.public_key().to_hex()],
                vec!["wss://relay.example".to_owned()],
                Nip59EnvelopeKind::Durable,
                1_700_000_100,
            )
            .expect("enqueue");
        assert_eq!(queued.operation.total_intents, 2);
        assert_eq!(queued.operation.state, crate::OperationState::Queued);

        let claimed = database
            .claim_deliveries(
                "lease-1".to_owned(),
                "wss://relay.example".to_owned(),
                1_700_000_200,
                30,
            )
            .expect("claim")
            .expect("work");
        assert_eq!(claimed.payloads.len(), 2);
        database
            .mark_socket_written(
                &claimed.claim,
                claimed
                    .claim
                    .intents
                    .iter()
                    .map(|intent| intent.intent_id.clone())
                    .collect(),
            )
            .expect("socket written");
        let before_ack = logs.drain(64).expect("diagnostics").events;
        assert_eq!(
            before_ack
                .iter()
                .filter(|event| event.operation == "message.queued")
                .count(),
            1
        );
        assert!(
            !before_ack
                .iter()
                .any(|event| event.operation == "message.received"
                    || event.operation == "delivery.accepted")
        );
        let first_ack = RelayDeliveryResult {
            intent_id: claimed.claim.intents[0].intent_id.clone(),
            kind: crate::RelayDeliveryResultKind::Accepted,
            category: "accepted".to_owned(),
        };
        database
            .apply_relay_results(&claimed.claim, vec![first_ack.clone()])
            .expect("partial ACK");
        let partial = logs.drain(64).expect("partial diagnostic").events;
        assert_eq!(partial.len(), 1);
        assert_eq!(partial[0].operation, "delivery.partial");
        assert!(
            partial[0]
                .counters
                .iter()
                .any(|value| value.kind == Counter::AcceptedIntents && value.value == 1)
        );
        assert_eq!(
            database.apply_relay_results(&claimed.claim, vec![first_ack]),
            Err(SoftchatError::InvalidDeliveryState)
        );
        assert!(
            logs.drain(64)
                .expect("duplicate diagnostic")
                .events
                .is_empty()
        );
        database
            .apply_relay_results(
                &claimed.claim,
                claimed
                    .claim
                    .intents
                    .iter()
                    .skip(1)
                    .map(|intent| RelayDeliveryResult {
                        intent_id: intent.intent_id.clone(),
                        kind: crate::RelayDeliveryResultKind::Accepted,
                        category: "accepted".to_owned(),
                    })
                    .collect(),
            )
            .expect("accepted");
        assert_eq!(
            database.operation("operation-1").expect("operation").state,
            crate::OperationState::Sent
        );
        let accepted = logs.drain(64).expect("diagnostics").events;
        assert_eq!(accepted.len(), 1);
        assert_eq!(accepted[0].operation, "delivery.accepted");
        assert!(
            accepted[0]
                .counters
                .iter()
                .any(|counter| counter.kind == Counter::AcceptedIntents && counter.value == 2)
        );
    }

    #[test]
    fn typing_delivery_does_not_emit_info_even_without_a_durable_projection()
    -> Result<(), Box<dyn std::error::Error>> {
        let alice = LocalIdentity::from_secret_hex(ALICE_SECRET)?;
        let bob = LocalIdentity::from_secret_hex(BOB_SECRET)?;
        let typing = alice.create_typing(vec![bob.public_key()], 1_700_000_000)?;
        let mut database = database("typing-diagnostics");
        let logs = std::sync::Arc::new(crate::account_diagnostics::AccountLogBuffer::new(
            AccountLogLevel::Info,
        ));
        database.diagnostics.buffer = Some(logs.clone());
        database.enqueue_rumor(
            &alice,
            "typing-operation".to_owned(),
            typing.rumor,
            vec![bob.public_key().to_hex()],
            vec![bob.public_key().to_hex()],
            vec!["wss://relay.example".to_owned()],
            Nip59EnvelopeKind::Ephemeral,
            1_700_000_100,
        )?;
        let claimed = database
            .claim_deliveries(
                "typing-lease".to_owned(),
                "wss://relay.example".to_owned(),
                1_700_000_101,
                30,
            )?
            .expect("typing publication");
        database.mark_socket_written(
            &claimed.claim,
            claimed
                .claim
                .intents
                .iter()
                .map(|intent| intent.intent_id.clone())
                .collect(),
        )?;
        database.apply_relay_results(
            &claimed.claim,
            claimed
                .claim
                .intents
                .iter()
                .map(|intent| RelayDeliveryResult {
                    intent_id: intent.intent_id.clone(),
                    kind: crate::RelayDeliveryResultKind::Accepted,
                    category: "accepted".to_owned(),
                })
                .collect(),
        )?;
        assert_eq!(
            database.operation("typing-operation")?.state,
            crate::OperationState::Sent
        );
        assert!(
            logs.drain(64)?.events.is_empty(),
            "Routine typing must stay quiet at INFO"
        );
        Ok(())
    }

    #[test]
    fn boundary_frame_outcomes_and_correlated_mutations_commit_as_one_transition() {
        let mut database = database("alice");
        for (index, (contact_marker, frame_bytes)) in [
            (0x42, crate::MAX_RELAY_FRAME_BYTES - 1),
            (0x43, crate::MAX_RELAY_FRAME_BYTES),
        ]
        .into_iter()
        .enumerate()
        {
            let run_id = format!("boundary-run-{index}");
            let action_id = format!("boundary-action-{index}");
            let outcome = crate::account_transport::StoredAccountTransportOutcome {
                actions: vec![crate::AccountTransportAction {
                    run_id: run_id.clone(),
                    action_id: action_id.clone(),
                    generation: 1,
                    kind: crate::AccountTransportActionKind::SendText,
                    configured_url: String::new(),
                    network_url: String::new(),
                    noise_remote_static_key: Vec::new(),
                    text_frame: "x".repeat(frame_bytes),
                    delay_ms: 0,
                }],
                connection_state: crate::RelayConnectionState::Ready,
                idle: false,
                revision: i64::try_from(index + 1).expect("bounded revision"),
                typing_indicators: Vec::new(),
                sync: None,
            };
            let outcome_json = serde_json::to_string(&outcome).expect("outcome json");
            assert!(outcome_json.len() > frame_bytes);
            if frame_bytes == crate::MAX_RELAY_FRAME_BYTES {
                assert!(outcome_json.len() > crate::MAX_RELAY_FRAME_BYTES);
            }

            let revision = database
                .apply_transport_result_atomically(|database| {
                    database
                        .connection
                        .execute(
                            "INSERT INTO local_contacts
                             (public_key, name, updated_at_millis) VALUES (?1, ?2, 1)",
                            params![hex::encode([contact_marker; 32]), &run_id],
                        )
                        .map_err(map_sqlite_error)?;
                    database.record_transport_result(
                        &run_id,
                        &action_id,
                        1,
                        &"ab".repeat(32),
                        &outcome_json,
                    )
                })
                .expect("atomic transition");
            assert_eq!(
                revision,
                i64::try_from(index + 1).expect("bounded revision")
            );
            assert!(
                database
                    .stored_transport_result(&run_id, &action_id)
                    .expect("stored outcome")
                    .is_some()
            );
        }
        assert_eq!(
            database
                .connection
                .query_row("SELECT COUNT(*) FROM local_contacts", [], |row| {
                    row.get::<_, i64>(0)
                })
                .expect("correlated mutation"),
            2
        );

        let before_failure = database.revision().expect("revision before failure");
        assert_eq!(
            database.apply_transport_result_atomically(|database| {
                database
                    .connection
                    .execute(
                        "INSERT INTO local_contacts
                         (public_key, name, updated_at_millis) VALUES (?1, 'rollback', 2)",
                        [hex::encode([0x44; 32])],
                    )
                    .map_err(map_sqlite_error)?;
                database.record_transport_result(
                    "rollback-run",
                    "rollback-action",
                    1,
                    "invalid-hash",
                    "{}",
                )
            }),
            Err(SoftchatError::InvalidRelaySession)
        );
        assert_eq!(
            database.revision().expect("revision after rollback"),
            before_failure
        );
        assert_eq!(
            database
                .connection
                .query_row(
                    "SELECT COUNT(*) FROM local_contacts WHERE name = 'rollback'",
                    [],
                    |row| row.get::<_, i64>(0),
                )
                .expect("rolled back mutation"),
            0
        );
    }

    #[test]
    fn account_reopen_recovers_views_and_expired_delivery_leases() {
        let alice = LocalIdentity::from_secret_hex(ALICE_SECRET).expect("alice");
        let bob = LocalIdentity::from_secret_hex(BOB_SECRET).expect("bob");
        let message = alice
            .create_chat_message(ChatMessageDraft {
                created_at: 1_700_000_000,
                participants: vec![bob.public_key()],
                content: "durable restart".to_owned(),
                relation: ChatRelation::None,
                attachments: Vec::new(),
                emoji_tags: Vec::new(),
                extension_tags: Vec::new(),
            })
            .expect("message");
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("clock")
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "softchat-reopen-{}-{nonce}.sqlite",
            std::process::id()
        ));

        let committed_revision = {
            let mut database =
                AccountDatabase::open(&path, "alice".to_owned()).expect("open first");
            database
                .enqueue_rumor(
                    &alice,
                    "operation-reopen".to_owned(),
                    message.rumor,
                    vec![bob.public_key().to_hex()],
                    Vec::new(),
                    vec!["wss://relay.example".to_owned()],
                    Nip59EnvelopeKind::Durable,
                    1_700_172_800,
                )
                .expect("enqueue");
            let claimed = database
                .claim_deliveries(
                    "lease-before-reopen".to_owned(),
                    "wss://relay.example".to_owned(),
                    1_700_172_801,
                    5,
                )
                .expect("claim")
                .expect("work");
            assert!(
                claimed
                    .claim
                    .intents
                    .iter()
                    .all(|intent| intent.attempt_count == 1)
            );
            database.revision().expect("revision")
        };

        let mut reopened =
            AccountDatabase::open(&path, "alice".to_owned()).expect("reopen account");
        assert_eq!(
            reopened.revision().expect("reopened revision"),
            committed_revision
        );
        let views = reopened.list_conversation_views(20, 0).expect("views");
        assert_eq!(
            views
                .first()
                .expect("one view")
                .last_message
                .effective_content,
            "durable restart"
        );
        let recovered = reopened
            .claim_deliveries(
                "lease-after-reopen".to_owned(),
                "wss://relay.example".to_owned(),
                1_700_172_807,
                30,
            )
            .expect("recover")
            .expect("expired work");
        assert!(
            recovered
                .claim
                .intents
                .iter()
                .all(|intent| intent.attempt_count == 2)
        );
        drop(reopened);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn an_account_file_cannot_be_reopened_for_another_account() {
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("clock")
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "softchat-storage-{}-{nonce}.sqlite",
            std::process::id()
        ));
        let first = AccountDatabase::open(&path, "first".to_owned()).expect("first");
        drop(first);
        assert!(matches!(
            AccountDatabase::open(&path, "second".to_owned()),
            Err(SoftchatError::InvalidPersistenceResult)
        ));
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn version_eight_media_operations_gain_exact_source_message_correlation() {
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("clock")
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "softchat-media-migration-{}-{nonce}.sqlite",
            std::process::id()
        ));
        drop(AccountDatabase::open(&path, "alice".to_owned()).expect("create"));
        {
            let legacy = Connection::open(&path).expect("legacy database");
            legacy
                .execute_batch(
                    "DROP TABLE pending_asset_replacements;
                     DROP TABLE conversation_subject_icons;
                     ALTER TABLE media_operations DROP COLUMN operation_kind;
                     ALTER TABLE media_operations DROP COLUMN protection;
                     ALTER TABLE media_operations DROP COLUMN source_message_id;
                     UPDATE softchat_meta
                     SET integer_value = 8
                     WHERE key = 'schema_version';",
                )
                .expect("downgrade fixture");
        }

        let migrated = AccountDatabase::open(&path, "alice".to_owned()).expect("migrate");
        assert_eq!(migrated.info().expect("info").schema_version, 16);
        let columns = migrated
            .connection
            .prepare("PRAGMA table_info(media_operations)")
            .expect("table info")
            .query_map([], |row| row.get::<_, String>(1))
            .expect("columns")
            .filter_map(Result::ok)
            .collect::<Vec<_>>();
        assert!(columns.iter().any(|column| column == "source_message_id"));
        assert!(columns.iter().any(|column| column == "protection"));
        assert!(columns.iter().any(|column| column == "operation_kind"));
        drop(migrated);
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn rejects_an_existing_database_not_owned_by_softchat() {
        let nonce = std::time::SystemTime::now()
            .duration_since(std::time::UNIX_EPOCH)
            .expect("clock")
            .as_nanos();
        let path = std::env::temp_dir().join(format!(
            "softchat-foreign-{}-{nonce}.sqlite",
            std::process::id()
        ));
        {
            let foreign = Connection::open(&path).expect("foreign database");
            foreign
                .execute_batch(
                    "CREATE TABLE application_data (
                       id TEXT PRIMARY KEY NOT NULL,
                       value TEXT NOT NULL
                     );",
                )
                .expect("foreign schema");
        }

        assert!(matches!(
            AccountDatabase::open(&path, "alice".to_owned()),
            Err(SoftchatError::InvalidPersistenceResult)
        ));
        let _ = std::fs::remove_file(path);
    }

    #[test]
    fn relay_catalog_is_committed_as_one_revisioned_state() {
        let mut database = database("alice");
        let reconciled = database
            .reconcile_relay_catalog(
                vec![
                    DnsRelayRecord {
                        target: "wss://relay-two.example".to_owned(),
                        priority: 20,
                        weight: 1,
                        ttl_seconds: 300,
                    },
                    DnsRelayRecord {
                        target: "wss://relay-one.example".to_owned(),
                        priority: 10,
                        weight: 2,
                        ttl_seconds: 600,
                    },
                ],
                1_700_000_000,
                "wss://fallback.example".to_owned(),
                true,
            )
            .expect("reconcile");
        assert_eq!(reconciled.revision, 1);
        assert_eq!(reconciled.plan.active_relay_url, "wss://relay-one.example/");
        assert_eq!(database.relay_catalog().expect("catalog").len(), 2);

        let custom = database
            .add_custom_relay("wss://custom.example".to_owned(), 1_700_000_001)
            .expect("custom");
        assert_eq!(custom.revision, 2);
        let selected = database
            .set_active_relay("wss://custom.example".to_owned(), 1_700_000_002)
            .expect("active");
        assert_eq!(selected.revision, 3);
        assert_eq!(
            selected
                .entries
                .iter()
                .find(|entry| entry.is_active)
                .expect("one active relay")
                .url,
            "wss://custom.example/"
        );
        assert_eq!(
            database
                .relay_failover_candidates("wss://custom.example".to_owned())
                .expect("failover")
                .first()
                .map(String::as_str),
            Some("wss://relay-one.example/")
        );
    }

    #[test]
    fn local_diagnostics_ignore_unchanged_writes_and_keep_values_private()
    -> Result<(), Box<dyn std::error::Error>> {
        let mut database = database("local-diagnostics");
        let logs = std::sync::Arc::new(crate::account_diagnostics::AccountLogBuffer::new(
            AccountLogLevel::Debug,
        ));
        database.diagnostics.buffer = Some(logs.clone());
        let contact = LocalContactRecord {
            public_key: LocalIdentity::from_secret_hex(BOB_SECRET)?
                .public_key()
                .to_hex(),
            name: Some("private-contact-name".to_owned()),
            updated_at_millis: 1_700_000_000_000,
        };
        database.put_local_contact(contact.clone())?;
        let first = logs.drain(64)?.events;
        assert_eq!(first.len(), 1);
        assert_eq!(first[0].operation, "storage.local.changed");
        assert_eq!(
            first[0].direction,
            Some(crate::account_diagnostics::AccountLogDirection::Local)
        );
        assert!(!format!("{first:?}").contains("private-contact-name"));
        database.put_local_contact(contact)?;
        logs.flush();
        assert!(logs.drain(64)?.events.is_empty());
        Ok(())
    }

    #[test]
    fn use_case_caches_are_bounded_revisioned_and_clearable() {
        let mut database = database("alice");
        let bob = LocalIdentity::from_secret_hex(BOB_SECRET).expect("bob");
        let contact = LocalContactRecord {
            public_key: bob.public_key().to_hex(),
            name: Some("Bob".to_owned()),
            updated_at_millis: 1_700_000_000_000,
        };
        assert_eq!(
            database
                .put_local_contact(contact.clone())
                .expect("put contact")
                .revision,
            1
        );
        assert_eq!(
            database.local_contacts(20, 0).expect("contacts"),
            vec![contact.clone()]
        );
        assert!(
            database
                .list_known_public_keys(20, 0)
                .expect("known keys")
                .contains(&contact.public_key)
        );
        let sticker = StickerRecord {
            name: "wave".to_owned(),
            url: "https://cdn.example/wave.webp".to_owned(),
            associated_emojis: vec!["👋".to_owned()],
        };
        assert_eq!(
            database
                .replace_stickers(vec![sticker.clone()])
                .expect("replace stickers")
                .revision,
            2
        );
        assert_eq!(database.stickers().expect("stickers"), vec![sticker]);

        let preview = LinkPreviewRecord {
            url: "https://example.com/article".to_owned(),
            title: Some("Article".to_owned()),
            description: Some("Description".to_owned()),
            image_url: Some("https://example.com/image.webp".to_owned()),
            updated_at_millis: 1_700_000_000_000,
        };
        assert_eq!(
            database
                .put_link_preview(preview.clone())
                .expect("put preview")
                .revision,
            3
        );
        assert_eq!(
            database
                .link_preview(preview.url.clone())
                .expect("link preview"),
            Some(preview)
        );
        assert!(database.clear().expect("clear").revision > 3);
        assert!(
            database
                .local_contacts(20, 0)
                .expect("cleared contacts")
                .is_empty()
        );
        assert!(database.stickers().expect("cleared stickers").is_empty());
        assert!(
            database
                .link_preview("https://example.com/article".to_owned())
                .expect("cleared preview")
                .is_none()
        );
        assert_eq!(database.info().expect("info").account_id, "alice");
    }

    #[test]
    fn clear_inventory_covers_every_account_table_and_fts_index() {
        let mut database = database("alice");
        database
            .connection
            .execute_batch(
                "
                INSERT INTO signed_events VALUES
                  ('outer', '{}', 'author', 1059, 1, 1);
                INSERT INTO authenticated_rumors VALUES
                  ('rumor', 'outer', '{}', 'author', 14, 1);
                INSERT INTO authenticated_rumor_wrappers VALUES ('outer', 'rumor');
                INSERT INTO projections
                  (logical_event_id, outer_event_id, kind, author_public_key, created_at,
                   conversation_id, content, ephemeral)
                  VALUES ('rumor', 'outer', 'chat_message', 'author', 1,
                          'conversation', 'searchable retained text', 0);
                INSERT INTO projection_participants VALUES ('rumor', 'author');
                INSERT INTO projection_targets VALUES ('rumor', 'target', 0);
                INSERT INTO event_fingerprints VALUES ('outer', 1);
                INSERT INTO outgoing_operations VALUES ('operation', 'alice', 1, '', '');
                INSERT INTO outgoing_event_copies VALUES
                  ('copy', 'operation', 'recipient', '{}');
                INSERT INTO relay_delivery_intents
                  (intent_id, operation_id, event_copy_id, relay_url, payload_bytes,
                   state, created_at)
                  VALUES ('intent', 'operation', 'copy', 'wss://custom.example/', 2,
                          'pending', 1);
                DELETE FROM relay_catalog;
                INSERT INTO relay_catalog VALUES
                  ('wss://custom.example/', 1, 'custom', 1, 0, 0, -1, -1, -1, 1);
                INSERT INTO sync_checkpoints VALUES ('sync', 1, 1);
                INSERT INTO local_contacts VALUES ('contact', 'Contact', 1);
                INSERT INTO sticker_cache VALUES ('sticker', 'https://example.com/s', '[]');
                INSERT INTO link_preview_cache VALUES
                  ('https://example.com', 'title', 'description', NULL, 1);
                INSERT INTO quarantine(source, source_id, category, observed_at)
                  VALUES ('test', 'source', 'category', 1);
                INSERT INTO conversation_state
                  (conversation_id, created_at, updated_at)
                  VALUES ('conversation', 1, 1);
                INSERT INTO conversation_members VALUES ('conversation', 'author');
                INSERT INTO conversation_read_state
                  (conversation_id, read_at, read_message_id, forced_unread,
                   updated_at, update_id, unknown_json)
                  VALUES ('conversation', 1, 'rumor', 0, 1, 'read', '{}');
                UPDATE account_settings
                  SET schema_version = 9, canonical_json = '{\"retained\":true}',
                      updated_at = 1 WHERE singleton = 1;
                INSERT INTO app_data_effective VALUES
                  ('app-settings', '{\"retained\":true}', 1, 'rumor');
                INSERT INTO media_operations
                  (operation_id, command_id, direction, state, source_fingerprint,
                   created_at, updated_at)
                  VALUES ('media-message', 'media-command', 'upload', 'prepared',
                          'fingerprint-message', 1, 1);
                INSERT INTO media_operations
                  (operation_id, command_id, direction, state, source_fingerprint,
                   created_at, updated_at)
                  VALUES ('media-asset', 'asset-command', 'upload', 'prepared',
                          'fingerprint-asset', 1, 1);
                INSERT INTO pending_media_messages
                  (command_id, conversation_id, text_content, emoji_tags_json,
                   command_hash, state, created_at, updated_at)
                  VALUES ('media-command', 'conversation', 'pending', '[]',
                          'hash', 'waiting_for_uploads', 1, 1);
                INSERT INTO pending_message_attachments VALUES
                  ('media-command', 0, 'media-message');
                INSERT INTO pending_asset_replacements
                  (command_id, target, operation_id, command_hash, state,
                   created_at, updated_at)
                  VALUES ('asset-command', 'profile_picture', 'media-asset', 'hash',
                          'waiting_for_upload', 1, 1);
                INSERT INTO message_local_extras
                  (message_id, transcript, waveform, layout_width, layout_height,
                   updated_at)
                  VALUES ('rumor', 'transcript', X'0102', 100, 50, 1);
                INSERT INTO processed_transport_results VALUES
                  ('run', 'action', 1,
                   'aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa', '{}');
                INSERT INTO message_fts(message_id, conversation_id, content)
                  VALUES ('rumor', 'conversation', 'searchable retained text');
                ",
            )
            .expect("seed every account table");

        let schema_tables = database
            .connection
            .prepare("SELECT name FROM sqlite_schema WHERE type = 'table' ORDER BY name")
            .expect("schema query")
            .query_map([], |row| row.get::<_, String>(0))
            .expect("schema rows")
            .filter_map(Result::ok)
            .filter(|name| !name.starts_with("sqlite_") && !name.starts_with("message_fts_"))
            .collect::<BTreeSet<_>>();
        let expected_tables = ACCOUNT_CLEAR_TABLES
            .iter()
            .copied()
            .chain(["softchat_meta", "relay_catalog", "account_settings"])
            .map(str::to_owned)
            .collect::<BTreeSet<_>>();
        assert_eq!(schema_tables, expected_tables);

        let cleared = database.clear().expect("clear complete account");
        assert!(cleared.revision > 0);
        for table in ACCOUNT_CLEAR_TABLES {
            let count = database
                .connection
                .query_row(&format!("SELECT COUNT(*) FROM {table}"), [], |row| {
                    row.get::<_, i64>(0)
                })
                .expect("cleared row count");
            assert_eq!(count, 0, "{table} retained account data");
        }
        assert_eq!(
            database
                .connection
                .query_row(
                    "SELECT relay_url FROM relay_catalog WHERE active = 1",
                    [],
                    |row| row.get::<_, String>(0),
                )
                .expect("default relay"),
            DEFAULT_RELAY_URL
        );
        assert_eq!(
            database
                .connection
                .query_row(
                    "SELECT schema_version, canonical_json, updated_at
                     FROM account_settings WHERE singleton = 1",
                    [],
                    |row| {
                        Ok((
                            row.get::<_, i64>(0)?,
                            row.get::<_, String>(1)?,
                            row.get::<_, i64>(2)?,
                        ))
                    },
                )
                .expect("default settings"),
            (1, "{}".to_owned(), 0)
        );
        assert_eq!(
            database.info().expect("account binding").account_id,
            "alice"
        );
    }
}