eidetica/transaction/

mod.rs

//! Transaction system for atomic database modifications
//!
//! This module provides the transaction API for making atomic changes to an Eidetica database.
//! Transactions ensure that all changes within a transaction are applied atomically and maintain
//! proper parent-child relationships in the Merkle-CRDT DAG structure.
//!
//! # Subtree Parent Management
//!
//! One of the critical responsibilities of the transaction system is establishing proper
//! subtree parent relationships. When a store (subtree) is accessed for the first time
//! in a transaction, the system must determine the correct parent entries for that subtree.
//! This involves:
//!
//! 1. Checking for existing subtree tips (leaf nodes)
//! 2. If no tips exist, traversing the DAG to find reachable subtree entries
//! 3. Setting appropriate parent relationships (empty for first entry, or proper parents)

pub mod errors;

#[cfg(test)]
mod tests;

use std::{
    collections::{BTreeMap, HashMap},
    sync::{
        Arc, Mutex,
        atomic::{AtomicBool, Ordering},
    },
};

pub use errors::TransactionError;
use serde::{Deserialize, Serialize};

use crate::{
    Database, Result, Snapshot, Store,
    auth::{
        AuthSettings,
        crypto::{PrivateKey, sign_entry},
        types::{AuthInfo, SigKey},
        validation::AuthValidator,
    },
    backend::{RecordMutations, RecordRange, RecordView, VerificationStatus},
    constants::{INDEX, ROOT, SETTINGS},
    crdt::{CRDT, Data, Doc, doc::Value},
    entry::{Entry, EntryBuilder, ID},
    height::HeightStrategy,
    instance::WriteSource,
    store::{ProjectionDescriptor, RecordProjection, Registry, SettingsStore, StoreError, state},
};

/// Creates a synthetic entry ID for multi-tip merged CRDT state caching.
///
/// Tips are sorted to ensure deterministic keys regardless of input order.
/// The resulting ID has format `merge:{tip1}:{tip2}:...` which is distinct
/// from real content-addressed entry IDs.
fn create_merge_cache_id(tip_ids: &[ID]) -> ID {
    let mut sorted_tips = tip_ids.to_vec();
    sorted_tips.sort();

    // Create a deterministic cache key by hashing the sorted tip IDs
    let mut key = String::from("merge");
    for tip in &sorted_tips {
        key.push(':');
        key.push_str(&tip.to_string());
    }
    ID::from_bytes(key.as_bytes())
}

/// Trait for encrypting/decrypting subtree data transparently
///
/// Encryptors are registered with a Transaction for specific subtrees, allowing
/// transparent encryption/decryption at the transaction boundary. When an encryptor
/// is registered:
///
/// - `get_full_state()` decrypts each historical entry before CRDT merging
/// - `get_local_data()` returns plaintext (cached in EntryBuilder)
/// - `update_subtree()` stores plaintext in cache, encrypted on commit
///
/// This ensures proper CRDT merge semantics while keeping data encrypted at rest.
///
/// # Wire Format
///
/// The trait operates on raw bytes, allowing implementations to define their own
/// wire format. For example, AES-GCM implementations typically use `nonce || ciphertext`.
/// Entry subtree payloads are opaque bytes, so ciphertext is stored verbatim with no
/// additional encoding.
///
/// # Example
///
/// ```rust,ignore
/// struct PasswordEncryptor { /* ... */ }
///
/// impl Encryptor for PasswordEncryptor {
///     fn decrypt(&self, ciphertext: &[u8]) -> Result<Vec<u8>> {
///         let (nonce, ct) = ciphertext.split_at(12);
///         // decrypt with nonce and ciphertext...
///     }
///
///     fn encrypt(&self, plaintext: &[u8]) -> Result<Vec<u8>> {
///         let nonce = generate_nonce();
///         let ct = encrypt(plaintext, &nonce);
///         // return nonce || ciphertext
///     }
/// }
/// ```
pub(crate) trait Encryptor: Send + Sync {
    /// Decrypt ciphertext bytes to plaintext bytes
    ///
    /// # Arguments
    /// * `ciphertext` - Encrypted data in implementation-defined format
    ///
    /// # Returns
    /// Plaintext bytes in whatever format the wrapped store produces (e.g.
    /// JSON for `DocStore`/`Table`, binary Yrs updates for `YDoc`).
    fn decrypt(&self, ciphertext: &[u8]) -> Result<Vec<u8>>;

    /// Encrypt plaintext bytes to ciphertext bytes
    ///
    /// # Arguments
    /// * `plaintext` - Bytes to encrypt; format is whatever the wrapped store
    ///   produces (JSON, binary CRDT update, etc.). The Encryptor itself does
    ///   not interpret the bytes.
    ///
    /// # Returns
    /// Encrypted data in implementation-defined format
    fn encrypt(&self, plaintext: &[u8]) -> Result<Vec<u8>>;

    fn physical_record_key(&self, logical_key: &[u8]) -> Result<Vec<u8>> {
        Ok(logical_key.to_vec())
    }

    fn encrypt_record(&self, _logical_key: &[u8], plaintext: &[u8]) -> Result<Vec<u8>> {
        self.encrypt(plaintext)
    }

    fn decrypt_record(&self, physical_key: &[u8], ciphertext: &[u8]) -> Result<(Vec<u8>, Vec<u8>)> {
        Ok((physical_key.to_vec(), self.decrypt(ciphertext)?))
    }

    fn projection_descriptor(&self, descriptor: ProjectionDescriptor) -> ProjectionDescriptor {
        descriptor
    }
}

/// Metadata structure for entries
#[derive(Debug, Clone, Serialize, Deserialize)]
pub(crate) struct EntryMetadata {
    /// Snapshot of the `_settings` subtree at the time this entry was created.
    /// This is the entry's **pin**: the exact `_settings` state its
    /// signature must be validated against. Used for sync performance,
    /// sparse-checkout validation, and deferred re-verification.
    ///
    /// Wire name remains `settings_tips` for on-disk stability — `Snapshot`
    /// serializes as a bare ID array, identical to the legacy `Vec<ID>` shape.
    #[serde(rename = "settings_tips")]
    pub(crate) settings_snapshot: Snapshot,
    /// Random entropy for ensuring unique IDs for root entries
    pub(crate) entropy: Option<u64>,
}

/// Represents a single, atomic transaction for modifying a `Database`.
///
/// An `Transaction` encapsulates a mutable `EntryBuilder` being constructed. Users interact with
/// specific `Store` instances obtained via `Transaction::get_store` to stage changes.
/// All staged changes across different subtrees within the transaction are recorded
/// in the internal `EntryBuilder`.
///
/// When `commit()` is called, the transaction:
/// 1. Finalizes the `EntryBuilder` by building an immutable `Entry`
/// 2. Calculates the entry's content-addressable ID
/// 3. Ensures the correct parent links are set based on the tree's state
/// 4. Removes any empty subtrees that didn't have data staged
/// 5. Signs the entry if authentication is configured
/// 6. Persists the resulting immutable `Entry` to the backend
///
/// `Transaction` instances are typically created via `Database::new_transaction()`.
#[derive(Clone)]
pub struct Transaction {
    /// The entry builder being modified, wrapped in Option to support consuming on commit
    entry_builder: Arc<Mutex<Option<EntryBuilder>>>,
    /// The database this transaction belongs to
    db: Database,
    /// Provided signing key paired with its auth identity
    provided_signing_key: Option<(PrivateKey, SigKey)>,
    /// Registered encryptors for transparent encryption/decryption of specific subtrees
    /// Maps subtree name -> encryptor implementation
    /// When an encryptor is registered, the transaction automatically encrypts writes
    /// and decrypts reads for that subtree
    encryptors: Arc<Mutex<HashMap<String, Box<dyn Encryptor>>>>,
    /// When true, `get_store` rejects any `_`-prefixed subtree name. Used by
    /// `Database::create_with_init` to keep its init callback from opening the
    /// system subtrees (`_settings`, `_root`, `_index`) that `create_with_init`
    /// itself manages. Legitimate internal paths — `get_index()` (via
    /// `Registry::new` → `DocStore::load`) and `Store::register`'s own
    /// `_index` updates — bypass `get_store` and remain unaffected.
    system_subtrees_locked: Arc<AtomicBool>,
    record_mutations: Arc<Mutex<HashMap<String, RecordMutations>>>,
    logical_record_mutations: Arc<Mutex<HashMap<String, RecordMutations>>>,
    record_views: Arc<Mutex<HashMap<String, RecordView>>>,
}

/// RAII guard returned by [`Transaction::lock_system_subtrees`]. Releases the
/// lock on drop, covering early-return-via-`?` and panic-unwind alike.
pub(crate) struct SystemSubtreeLockGuard {
    flag: Arc<AtomicBool>,
}

impl Drop for SystemSubtreeLockGuard {
    fn drop(&mut self) {
        self.flag.store(false, Ordering::Release);
    }
}

impl Transaction {
    /// Creates a new atomic transaction for a specific `Database` anchored at a snapshot.
    ///
    /// Initializes an internal `EntryBuilder` with its main parent pointers set to the
    /// snapshot's tips instead of the database's current state. This allows creating
    /// transactions that branch from specific points in the database history (e.g.
    /// diamond patterns).
    ///
    /// # Arguments
    /// * `database` - The `Database` this transaction will modify.
    /// * `snapshot` - The snapshot to anchor the transaction at. Must contain at least one tip,
    ///   unless this transaction is creating the database's root entry.
    pub(crate) async fn new_at(database: &Database, snapshot: &Snapshot) -> Result<Self> {
        let tips = snapshot.tips();
        // Validate that tips are not empty, unless we're creating the root entry
        if tips.is_empty() {
            // Check if this is a root entry creation by seeing if the database root exists in backend
            let root_exists = database.ops().get(database.root_id()).await.is_ok();

            if root_exists {
                return Err(TransactionError::EmptyTipsNotAllowed.into());
            }
            // If root doesn't exist, this is valid (creating the root entry)
        }

        // Validate that all tips belong to the same tree
        let backend = database.ops();
        for tip_id in tips {
            let entry = backend.get(tip_id).await?;
            if !entry.in_tree(database.root_id()) {
                return Err(TransactionError::InvalidTip {
                    tip_id: tip_id.clone(),
                }
                .into());
            }
        }

        // Start with a basic entry linked to the database's root.
        // Data and parents will be filled based on the transaction type.
        let mut builder = Entry::builder(database.root_id().clone());

        // Use the provided tips as parents (only if not empty)
        if !tips.is_empty() {
            builder.set_parents_mut(tips.to_vec());
        }

        Ok(Self {
            entry_builder: Arc::new(Mutex::new(Some(builder))),
            db: database.clone(),
            provided_signing_key: None,
            encryptors: Arc::new(Mutex::new(HashMap::new())),
            system_subtrees_locked: Arc::new(AtomicBool::new(false)),
            record_mutations: Arc::new(Mutex::new(HashMap::new())),
            logical_record_mutations: Arc::new(Mutex::new(HashMap::new())),
            record_views: Arc::new(Mutex::new(HashMap::new())),
        })
    }

    /// Lock the system subtrees (`_settings`, `_root`, `_index`) for the
    /// returned guard's lifetime.
    ///
    /// While the guard is live, [`Self::get_store`] rejects any subtree name
    /// beginning with `_` (the system-subtree prefix). Used by
    /// [`Database::create_with_init`] to guard its init callback against
    /// clobbering `_settings`/`_root`/`_index`, all of which `create_with_init`
    /// manages itself or via a dedicated accessor. The lock releases on the
    /// guard's `Drop`, so it covers both early-return-via-`?` and panic-unwind
    /// paths.
    ///
    /// The lock is process-local state on the (cloned-by-Arc) transaction;
    /// clones share the same flag. Legitimate internal paths — `get_index()`
    /// (via `Registry::new` → `DocStore::load`) and `Store::register`'s own
    /// `_index` writes — bypass `get_store` and are unaffected.
    pub(crate) fn lock_system_subtrees(&self) -> SystemSubtreeLockGuard {
        self.system_subtrees_locked.store(true, Ordering::Release);
        SystemSubtreeLockGuard {
            flag: self.system_subtrees_locked.clone(),
        }
    }

    /// Set signing key directly for user context (internal API).
    ///
    /// This method is used when a Database has a key attached
    /// (via `Database::open().with_key()`). The provided SigningKey is already
    /// decrypted and ready to use, eliminating the need for backend key lookup.
    ///
    /// # Arguments
    /// * `signing_key` - The decrypted signing key from UserKeyManager
    /// * `identity` - The SigKey identity used in database auth settings
    pub(crate) fn set_provided_key(&mut self, signing_key: PrivateKey, identity: SigKey) {
        self.provided_signing_key = Some((signing_key, identity));
    }

    /// Get current time as RFC3339 string.
    ///
    /// Delegates to the underlying instance's clock.
    pub(crate) fn now_rfc3339(&self) -> Result<String> {
        Ok(self.db.instance()?.clock().now_rfc3339())
    }

    /// Register an encryptor for transparent encryption/decryption of a specific subtree.
    ///
    /// Once registered, the transaction will automatically:
    /// - Decrypt each historical entry before CRDT merging in `get_full_state()`
    /// - Return plaintext data from `get_local_data()` (cached in EntryBuilder)
    /// - Encrypt plaintext data before persisting in `commit()`
    ///
    /// This ensures proper CRDT merge semantics while keeping data encrypted at rest.
    ///
    /// # Arguments
    /// * `subtree` - The name of the subtree to encrypt/decrypt
    /// * `encryptor` - The encryptor implementation to use
    ///
    /// # Example
    ///
    /// For password-based encryption, use [`PasswordStore`] which handles
    /// encryptor registration automatically:
    ///
    /// ```rust,ignore
    /// let mut encrypted = tx.get_store::<PasswordStore<DocStore>>("secrets")?;
    /// encrypted.initialize("my_password", Doc::new())?;
    ///
    /// // PasswordStore registers the encryptor internally
    /// let docstore = encrypted.inner()?;
    /// ```
    ///
    /// For custom encryption, implement the [`Encryptor`] trait:
    ///
    /// ```rust,ignore
    /// struct MyEncryptor { /* ... */ }
    /// impl Encryptor for MyEncryptor {
    ///     fn encrypt(&self, plaintext: &[u8]) -> Result<Vec<u8>> { /* ... */ }
    ///     fn decrypt(&self, ciphertext: &[u8]) -> Result<Vec<u8>> { /* ... */ }
    /// }
    ///
    /// transaction.register_encryptor("secrets", Box::new(MyEncryptor::new()))?;
    /// ```
    ///
    /// [`PasswordStore`]: crate::store::PasswordStore
    /// [`Encryptor`]: crate::Encryptor
    pub(crate) fn register_encryptor(
        &self,
        subtree: impl Into<String>,
        encryptor: Box<dyn Encryptor>,
    ) -> Result<()> {
        self.encryptors
            .lock()
            .unwrap()
            .insert(subtree.into(), encryptor);
        Ok(())
    }

    fn physical_record_key(&self, store: &str, logical_key: &[u8]) -> Result<Vec<u8>> {
        self.encryptors.lock().unwrap().get(store).map_or_else(
            || Ok(logical_key.to_vec()),
            |encryptor| encryptor.physical_record_key(logical_key),
        )
    }

    fn encrypt_record(&self, store: &str, logical_key: &[u8], value: &[u8]) -> Result<Vec<u8>> {
        self.encryptors.lock().unwrap().get(store).map_or_else(
            || Ok(value.to_vec()),
            |encryptor| encryptor.encrypt_record(logical_key, value),
        )
    }

    fn decrypt_record(&self, store: &str, key: &[u8], value: &[u8]) -> Result<(Vec<u8>, Vec<u8>)> {
        self.encryptors.lock().unwrap().get(store).map_or_else(
            || Ok((key.to_vec(), value.to_vec())),
            |encryptor| encryptor.decrypt_record(key, value),
        )
    }

    pub(crate) fn database_id(&self) -> &ID {
        self.db.root_id()
    }

    /// Decrypt bytes if an encryptor is registered, otherwise return them unchanged.
    ///
    /// This is used throughout Transaction to transparently decrypt encrypted payloads
    /// before deserializing into CRDT types.
    fn decrypt_if_needed(&self, subtree: &str, data: &[u8]) -> Result<Vec<u8>> {
        if let Some(encryptor) = self.encryptors.lock().unwrap().get(subtree) {
            encryptor.decrypt(data)
        } else {
            Ok(data.to_vec())
        }
    }

    /// Encrypt bytes if an encryptor is registered for the subtree, otherwise return them unchanged.
    fn encrypt_if_needed(&self, subtree: &str, plaintext: &[u8]) -> Result<Vec<u8>> {
        if let Some(encryptor) = self.encryptors.lock().unwrap().get(subtree) {
            encryptor.encrypt(plaintext)
        } else {
            Ok(plaintext.to_vec())
        }
    }

    /// Get a SettingsStore handle for the settings subtree within this transaction.
    ///
    /// This method returns a `SettingsStore` that provides specialized access to the `_settings` subtree,
    /// allowing you to read and modify settings data within this atomic transaction.
    /// The DocStore automatically merges historical settings from the database with any
    /// staged changes in this transaction.
    ///
    /// # Returns
    ///
    /// Returns a `Result<SettingsStore>` that can be used to:
    /// - Read current settings values (including both historical and staged data)
    /// - Stage new settings changes within this transaction
    /// - Access nested settings structures
    ///
    /// # Example
    ///
    /// ```rust,no_run
    /// # use eidetica::Database;
    /// # async fn example(database: Database) -> eidetica::Result<()> {
    /// let txn = database.new_transaction().await?;
    /// let settings = txn.get_settings()?;
    ///
    /// // Read a setting
    /// if let Ok(name) = settings.get_name().await {
    ///     println!("Database name: {}", name);
    /// }
    ///
    /// // Modify a setting
    /// settings.set_name("Updated Database Name").await?;
    /// # Ok(())
    /// # }
    /// ```
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Unable to create the SettingsStore for the settings subtree
    /// - Operation has already been committed
    pub fn get_settings(&self) -> Result<SettingsStore> {
        // Create a SettingsStore for the settings subtree
        SettingsStore::new(self)
    }

    /// Gets a handle to the Index for managing subtree registry and metadata.
    ///
    /// The Index provides access to the `_index` subtree, which stores metadata
    /// about all subtrees in the database including their type identifiers and configurations.
    ///
    /// # Returns
    ///
    /// A `Result<Registry>` containing the handle for managing the index.
    ///
    /// # Errors
    ///
    /// Returns an error if:
    /// - Unable to create the Registry for the _index subtree
    /// - Operation has already been committed
    pub async fn get_index(&self) -> Result<Registry> {
        Registry::new(self, INDEX).await
    }

    /// Set the tree root field for the entry being built.
    ///
    /// Called by `Database::create()` to override the placeholder root with
    /// `ID::default()`, making the entry a proper top-level root.
    ///
    /// # Arguments
    /// * `root` - The tree root ID to set (use `ID::default()` for top-level roots)
    pub(crate) fn set_entry_root(&self, root: ID) -> Result<()> {
        let mut builder_ref = self.entry_builder.lock().unwrap();
        let builder = builder_ref
            .as_mut()
            .ok_or(TransactionError::TransactionAlreadyCommitted)?;
        builder.set_root_mut(root);
        Ok(())
    }

    /// Set entropy in the entry metadata.
    ///
    /// This is used during database creation to ensure unique IDs for databases
    /// even when they have identical settings.
    ///
    /// # Arguments
    /// * `entropy` - Random entropy value
    pub(crate) fn set_metadata_entropy(&self, entropy: u64) -> Result<()> {
        let mut builder_ref = self.entry_builder.lock().unwrap();
        let builder = builder_ref
            .as_mut()
            .ok_or(TransactionError::TransactionAlreadyCommitted)?;

        // Parse existing metadata if present, or create new
        let mut metadata = builder
            .metadata()
            .and_then(|m| serde_json::from_slice::<EntryMetadata>(m).ok())
            .unwrap_or(EntryMetadata {
                settings_snapshot: Snapshot::EMPTY,
                entropy: None,
            });

        // Set entropy
        metadata.entropy = Some(entropy);

        // Serialize and set metadata
        let metadata_json = serde_json::to_vec(&metadata)?;
        builder.set_metadata_mut(metadata_json);

        Ok(())
    }

    /// Stages an update for a specific subtree within this atomic transaction.
    ///
    /// This method is primarily intended for internal use by `Store` implementations
    /// (like `DocStore::set`). It records the serialized `data` for the given `subtree`
    /// name within the transaction's internal `EntryBuilder`.
    ///
    /// If this is the first modification to the named subtree within this transaction,
    /// it also fetches and records the current tips of that subtree from the backend
    /// to set the correct `subtree_parents` for the new entry.
    ///
    /// # Arguments
    /// * `subtree` - The name of the subtree to update.
    /// * `data` - The serialized CRDT data to stage for the subtree.
    ///
    /// # Returns
    /// A `Result<()>` indicating success or an error.
    pub(crate) async fn update_subtree(
        &self,
        subtree: impl AsRef<str>,
        data: impl Into<Vec<u8>>,
    ) -> Result<()> {
        let subtree = subtree.as_ref();
        let data = data.into();

        // Check if we need to fetch tips (check without holding borrow across await)
        let needs_tips = {
            let builder_ref = self.entry_builder.lock().unwrap();
            let builder = builder_ref
                .as_ref()
                .ok_or(TransactionError::TransactionAlreadyCommitted)?;
            !builder.subtrees().contains(&subtree.to_string())
        };

        // Fetch tips if needed (no borrow held across this await)
        let tips = if needs_tips {
            let backend = self.db.ops();
            // FIXME: we should get the subtree snapshot while still using the parent pointers
            Some(
                backend
                    .store_snapshot(self.db.root_id(), subtree)
                    .await?
                    .into_tips(),
            )
        } else {
            None
        };

        // Now update the builder
        let mut builder_ref = self.entry_builder.lock().unwrap();
        let builder = builder_ref
            .as_mut()
            .ok_or(TransactionError::TransactionAlreadyCommitted)?;

        builder.set_subtree_data_mut(subtree.to_string(), data);
        if let Some(tips) = tips {
            builder.set_subtree_parents_mut(subtree, tips);
        }

        Ok(())
    }

    pub(crate) fn stage_record(
        &self,
        store: &str,
        projection: &dyn RecordProjection<Doc>,
        key: Vec<u8>,
        value: Option<Vec<u8>>,
    ) -> Result<()> {
        let Some(key) = projection.normalize_record_key(&key)? else {
            return Ok(());
        };
        let conflicting_keys = {
            let mut staged = self.logical_record_mutations.lock().unwrap();
            let mutations = staged.entry(store.to_string()).or_default();
            let conflicting_keys = mutations
                .keys()
                .filter(|staged_key| projection.staged_keys_conflict(staged_key, &key))
                .cloned()
                .collect::<Vec<_>>();
            mutations.retain(|staged_key, _| !conflicting_keys.contains(staged_key));
            mutations.insert(key.clone(), value.clone());
            conflicting_keys
        };
        let physical_key = self.physical_record_key(store, &key)?;
        let conflicting_physical_keys = conflicting_keys
            .iter()
            .map(|key| self.physical_record_key(store, key))
            .collect::<Result<Vec<_>>>()?;
        let value = value
            .map(|value| self.encrypt_record(store, &key, &value))
            .transpose()?;
        let mut record_mutations = self.record_mutations.lock().unwrap();
        let mutations = record_mutations.entry(store.to_string()).or_default();
        mutations.retain(|staged_key, _| !conflicting_physical_keys.contains(staged_key));
        mutations.insert(physical_key, value);
        Ok(())
    }

    pub(crate) async fn record_get(
        &self,
        store: &str,
        projection: &dyn RecordProjection<Doc>,
        key: &[u8],
    ) -> Result<Option<Vec<u8>>> {
        let Some(key) = projection.normalize_record_key(key)? else {
            return Ok(None);
        };
        if let Some(mutations) = self.logical_record_mutations.lock().unwrap().get(store) {
            if let Some(value) = mutations.get(&key) {
                return Ok(value.clone());
            }
            if mutations
                .keys()
                .any(|staged_key| projection.staged_key_shadows_cached(staged_key, &key))
            {
                return Ok(None);
            }
        }
        let physical_key = self.physical_record_key(store, &key)?;
        let view = match self.record_view(store, projection).await {
            Err(err) if err.is_unsupported_store_state() => {
                return self.record_get_from_history(store, &key).await;
            }
            result => result?,
        };
        match self
            .db
            .ops()
            .store_state_record_get(&view, &physical_key)
            .await
        {
            Err(err) if err.is_unsupported_store_state() => {
                self.record_get_from_history(store, &key).await
            }
            Err(err) if err.is_invalid_store_state_view() => {
                self.record_views.lock().unwrap().remove(store);
                let view = match self.record_view(store, projection).await {
                    Err(err) if err.is_unsupported_store_state() => {
                        return self.record_get_from_history(store, &key).await;
                    }
                    result => result?,
                };
                match self
                    .db
                    .ops()
                    .store_state_record_get(&view, &physical_key)
                    .await
                {
                    Err(err) if err.is_unsupported_store_state() => {
                        self.record_get_from_history(store, &key).await
                    }
                    result => result?.map_or(Ok(None), |value| {
                        self.decrypt_record(store, &physical_key, &value)
                            .map(|(_, value)| Some(value))
                    }),
                }
            }
            result => result?.map_or(Ok(None), |value| {
                self.decrypt_record(store, &physical_key, &value)
                    .map(|(_, value)| Some(value))
            }),
        }
    }

    pub(crate) fn record_has_staged_descendant(
        &self,
        store: &str,
        projection: &dyn RecordProjection<Doc>,
        key: &[u8],
    ) -> Result<bool> {
        let Some(key) = projection.normalize_record_key(key)? else {
            return Ok(false);
        };
        Ok(self
            .logical_record_mutations
            .lock()
            .unwrap()
            .get(store)
            .is_some_and(|mutations| {
                mutations.iter().any(|(staged_key, value)| {
                    value.is_some() && projection.staged_key_descends_from(staged_key, &key)
                })
            }))
    }

    pub(crate) async fn record_scan(
        &self,
        store: &str,
        projection: &dyn RecordProjection<Doc>,
        after: Option<&[u8]>,
        limit: usize,
    ) -> Result<crate::backend::RecordPage> {
        if limit == 0 {
            return Ok(crate::backend::RecordPage::default());
        }
        let view = match self.record_view(store, projection).await {
            Err(err) if err.is_unsupported_store_state() => {
                return self
                    .record_scan_from_history(store, projection, after, limit)
                    .await;
            }
            result => result?,
        };
        let logical_mutations = self
            .logical_record_mutations
            .lock()
            .unwrap()
            .get(store)
            .cloned()
            .unwrap_or_default();
        let mutations = self
            .record_mutations
            .lock()
            .unwrap()
            .get(store)
            .cloned()
            .unwrap_or_default();
        let mut merged = mutations
            .iter()
            .filter(|(key, value)| {
                value.is_some() && after.is_none_or(|after| key.as_slice() > after)
            })
            .map(|(key, value)| (key.clone(), value.clone().unwrap()))
            .collect::<BTreeMap<_, _>>();
        let mut backend_after = after.map(ToOwned::to_owned);
        let mut backend_has_more = true;
        while backend_has_more {
            let page = match self
                .db
                .ops()
                .store_state_record_scan(
                    &view,
                    &RecordRange::default(),
                    backend_after.as_deref(),
                    limit.max(1),
                )
                .await
            {
                Err(err) if err.is_unsupported_store_state() => {
                    return self
                        .record_scan_from_history(store, projection, after, limit)
                        .await;
                }
                Err(err) if err.is_invalid_store_state_view() => {
                    self.record_views.lock().unwrap().remove(store);
                    match self.record_view(store, projection).await {
                        Err(err) if err.is_unsupported_store_state() => {
                            return self
                                .record_scan_from_history(store, projection, after, limit)
                                .await;
                        }
                        Ok(view) => match self
                            .db
                            .ops()
                            .store_state_record_scan(
                                &view,
                                &RecordRange::default(),
                                backend_after.as_deref(),
                                limit.max(1),
                            )
                            .await
                        {
                            Err(err) if err.is_unsupported_store_state() => {
                                return self
                                    .record_scan_from_history(store, projection, after, limit)
                                    .await;
                            }
                            result => result?,
                        },
                        Err(err) => return Err(err),
                    }
                }
                result => result?,
            };
            for (physical_key, value) in page.records {
                let (logical_key, _) = self.decrypt_record(store, &physical_key, &value)?;
                if !logical_mutations.keys().any(|staged_key| {
                    projection.staged_key_shadows_cached(staged_key, &logical_key)
                }) {
                    merged.insert(physical_key, value);
                }
            }
            backend_after = page.next;
            backend_has_more = backend_after.is_some();
            let backend_reached_page_end = merged.keys().nth(limit).is_some_and(|page_end| {
                backend_after
                    .as_ref()
                    .is_some_and(|backend_after| backend_after >= page_end)
            });
            if backend_reached_page_end {
                break;
            }
        }
        let mut records = merged
            .into_iter()
            .take(limit.saturating_add(1))
            .collect::<Vec<_>>();
        let has_more = records.len() > limit || backend_has_more;
        records.truncate(limit);
        let next = has_more.then(|| records.last().unwrap().0.clone());
        for (key, value) in &mut records {
            (*key, *value) = self.decrypt_record(store, key, value)?;
        }
        Ok(crate::backend::RecordPage { records, next })
    }

    async fn record_get_from_history(&self, store: &str, key: &[u8]) -> Result<Option<Vec<u8>>> {
        let state: Doc = self.get_full_state(store).await?;
        Ok(state
            .get(std::str::from_utf8(key).map_err(|error| {
                TransactionError::StoreDeserializationFailed {
                    store: store.to_string(),
                    reason: error.to_string(),
                }
            })?)
            .and_then(Value::as_text)
            .map(|value| value.as_bytes().to_vec()))
    }

    async fn record_scan_from_history(
        &self,
        store: &str,
        projection: &dyn RecordProjection<Doc>,
        after: Option<&[u8]>,
        limit: usize,
    ) -> Result<crate::backend::RecordPage> {
        let state: Doc = self.get_full_state(store).await?;
        let mut projected = RecordMutations::new();
        projection.project_delta(&state, &mut projected)?;
        let mutations = self
            .logical_record_mutations
            .lock()
            .unwrap()
            .get(store)
            .cloned()
            .unwrap_or_default();
        let mut records = projected
            .into_iter()
            .filter_map(|(key, value)| value.map(|value| (key, value)))
            .filter(|(cached_key, _)| {
                !mutations
                    .keys()
                    .any(|staged_key| projection.staged_key_shadows_cached(staged_key, cached_key))
            })
            .collect::<BTreeMap<_, _>>();
        for (key, value) in mutations {
            match value {
                Some(value) => {
                    records.insert(key, value);
                }
                None => {
                    records.remove(&key);
                }
            }
        }
        let mut records = records
            .into_iter()
            .map(|(logical_key, value)| {
                Ok((
                    self.physical_record_key(store, &logical_key)?,
                    (logical_key, value),
                ))
            })
            .collect::<Result<BTreeMap<_, _>>>()?
            .into_iter()
            .filter(|(physical_key, _)| after.is_none_or(|after| physical_key.as_slice() > after))
            .take(limit.saturating_add(1))
            .collect::<Vec<_>>();
        let has_more = records.len() > limit;
        records.truncate(limit);
        let next = has_more.then(|| records.last().unwrap().0.clone());
        let records = records
            .into_iter()
            .map(|(_, (logical_key, value))| (logical_key, value))
            .collect();
        Ok(crate::backend::RecordPage { records, next })
    }

    fn encrypted_projection_descriptor(
        &self,
        store: &str,
        descriptor: ProjectionDescriptor,
    ) -> ProjectionDescriptor {
        self.encryptors
            .lock()
            .unwrap()
            .get(store)
            .map_or(descriptor.clone(), |encryptor| {
                encryptor.projection_descriptor(descriptor)
            })
    }

    async fn publish_record_view(
        &self,
        store: &str,
        projection: &dyn RecordProjection<Doc>,
        request: crate::backend::StoreStateRequest,
        entries: &[Entry],
    ) -> Result<RecordView> {
        if !self.encryptors.lock().unwrap().contains_key(store) {
            return state::publish_records(
                self.db.ops(),
                request,
                entries
                    .iter()
                    .filter_map(|entry| entry.data(store).ok().map(|bytes| bytes.as_slice())),
                projection,
            )
            .await;
        }

        let token = self.db.ops().begin_store_state_staging(request).await?;
        let result = async {
            let mut logical = RecordMutations::new();
            for entry in entries {
                if let Ok(bytes) = entry.data(store) {
                    let plaintext = self.decrypt_if_needed(store, bytes)?;
                    let delta: Doc = serde_json::from_slice(&plaintext)?;
                    projection.project_delta(&delta, &mut logical)?;
                }
            }
            logical.retain(|_, value| value.is_some());
            let mut physical = RecordMutations::new();
            for (key, value) in logical {
                if let Some(value) = value {
                    physical.insert(
                        self.physical_record_key(store, &key)?,
                        Some(self.encrypt_record(store, &key, &value)?),
                    );
                }
            }
            if !physical.is_empty() {
                self.db
                    .ops()
                    .stage_store_state_records(&token, physical)
                    .await?;
            }
            self.db.ops().publish_store_state(token.clone()).await
        }
        .await;
        if result.is_err() {
            let _ = self.db.ops().abort_store_state(token).await;
        }
        result
    }

    async fn record_view(
        &self,
        store: &str,
        projection: &dyn RecordProjection<Doc>,
    ) -> Result<RecordView> {
        if let Some(view) = self.record_views.lock().unwrap().get(store).cloned() {
            return Ok(view);
        }
        self.init_subtree_parents(store).await?;
        let parents = self
            .entry_builder
            .lock()
            .unwrap()
            .as_ref()
            .ok_or(TransactionError::TransactionAlreadyCommitted)?
            .subtree_parents(store)
            .unwrap_or_default();
        let source_key = create_merge_cache_id(&parents).to_string().into_bytes();
        let descriptor = self.encrypted_projection_descriptor(store, projection.descriptor());
        let request = state::records_request(
            self.db.root_id(),
            store,
            descriptor,
            source_key,
            crate::backend::CacheScope::Shared,
        );
        let view = if let Some(view) = self.db.ops().resolve_store_state(&request).await? {
            view
        } else {
            let boundary = Snapshot::from(parents);
            let entries = self
                .db
                .ops()
                .store_at(self.db.root_id(), store, &boundary)
                .await?;
            self.publish_record_view(store, projection, request, &entries)
                .await?
        };
        self.record_views
            .lock()
            .unwrap()
            .insert(store.to_string(), view.clone());
        Ok(view)
    }

    /// Gets a handle to a specific `Store` for modification within this transaction.
    ///
    /// This method creates and returns an instance of the specified `Store` type `T`,
    /// associated with this `Transaction`. The returned `Store` handle can be used to
    /// stage changes (e.g., using `DocStore::set`).
    /// These changes are recorded within this `Transaction`.
    ///
    /// If this is the first time this subtree is accessed within the transaction,
    /// its parent tips will be fetched and stored.
    ///
    /// # Type Parameters
    /// * `T` - The concrete `Store` implementation type to create.
    ///
    /// # Arguments
    /// * `subtree_name` - The name of the subtree to get a modification handle for.
    ///
    /// # Returns
    /// A `Result<T>` containing the `Store` handle.
    pub async fn get_store<T>(&self, subtree_name: impl Into<String> + Send) -> Result<T>
    where
        T: Store + Send,
    {
        let subtree_name = subtree_name.into();

        // Skip special system subtrees to avoid circular dependencies
        let is_system_subtree =
            subtree_name == INDEX || subtree_name == SETTINGS || subtree_name == ROOT;

        if is_system_subtree && self.system_subtrees_locked.load(Ordering::Acquire) {
            return Err(TransactionError::SystemSubtreeLocked { name: subtree_name }.into());
        }

        // Initialize subtree parents before checking _index
        self.init_subtree_parents(&subtree_name).await?;

        if is_system_subtree {
            // System subtrees don't use _index registration
            return T::load(self, subtree_name).await;
        }

        // Check _index to determine if this is a new or existing subtree
        let index_store = self.get_index().await?;
        if index_store.contains(&subtree_name).await {
            // Type validation for existing subtree
            let subtree_info = index_store.get_entry(&subtree_name).await?;

            if !T::supports_type_id(&subtree_info.type_id) {
                return Err(StoreError::TypeMismatch {
                    store: subtree_name,
                    expected: T::type_id().to_string(),
                    actual: subtree_info.type_id,
                }
                .into());
            }

            // Type supported - create the Store
            T::load(self, subtree_name).await
        } else {
            // New subtree - register adds it to _index
            T::register(self, subtree_name).await
        }
    }

    /// Get the subtree tips reachable from the given main tree entries.
    async fn get_subtree_tips(&self, subtree_name: &str, main_parents: &[ID]) -> Result<Vec<ID>> {
        let boundary = Snapshot::from(main_parents.to_vec());
        self.db
            .ops()
            .store_snapshot_at(self.db.root_id(), subtree_name, &boundary)
            .await
            .map(Snapshot::into_tips)
    }

    /// Initialize subtree parents if this is the first time accessing this subtree
    /// in this transaction.
    pub(crate) async fn init_subtree_parents(&self, subtree_name: &str) -> Result<()> {
        let main_parents = {
            let builder_ref = self.entry_builder.lock().unwrap();
            let builder = builder_ref
                .as_ref()
                .ok_or(TransactionError::TransactionAlreadyCommitted)?;

            let subtrees = builder.subtrees();
            if subtrees.contains(&subtree_name.to_string()) {
                return Ok(()); // Already initialized
            }
            builder.parents().unwrap_or_default()
        };

        let tips = self.get_subtree_tips(subtree_name, &main_parents).await?;

        let mut builder_ref = self.entry_builder.lock().unwrap();
        let builder = builder_ref
            .as_mut()
            .ok_or(TransactionError::TransactionAlreadyCommitted)?;

        // Initialize the subtree with proper parent relationships
        // set_subtree_parents_mut creates the subtree with data=None if it doesn't exist
        builder.set_subtree_parents_mut(subtree_name, tips);

        Ok(())
    }

    /// Gets the currently staged data for a specific subtree within this transaction.
    ///
    /// This is intended for use by `Store` implementations to retrieve the data
    /// they have staged locally within the `Transaction` before potentially merging
    /// it with historical data.
    ///
    /// # Type Parameters
    /// * `T` - The data type (expected to be a CRDT) to deserialize the staged data into.
    ///
    /// # Arguments
    /// * `subtree_name` - The name of the subtree whose staged data is needed.
    ///
    /// # Returns
    /// A `Result<Option<T>>`:
    ///
    /// # Behavior
    /// - If the subtree doesn't exist or has no data, returns `Ok(None)`
    /// - If the subtree exists but has empty data (empty string or whitespace), returns `Ok(None)`
    /// - Otherwise deserializes the JSON data to type `T` and returns `Ok(Some(T))`
    ///
    /// # Errors
    /// Returns an error if the transaction has already been committed or if the
    /// subtree data exists but cannot be deserialized to type `T`.
    pub fn get_local_data<T>(&self, subtree_name: impl AsRef<str>) -> Result<Option<T>>
    where
        T: Data,
    {
        let subtree_name = subtree_name.as_ref();
        let builder_ref = self.entry_builder.lock().unwrap();
        let builder = builder_ref
            .as_ref()
            .ok_or(TransactionError::TransactionAlreadyCommitted)?;

        if let Ok(data) = builder.data(subtree_name) {
            if data.is_empty() {
                Ok(None)
            } else {
                serde_json::from_slice(data).map(Some).map_err(|e| {
                    TransactionError::StoreDeserializationFailed {
                        store: subtree_name.to_string(),
                        reason: e.to_string(),
                    }
                    .into()
                })
            }
        } else {
            Ok(None)
        }
    }

    /// Gets the fully merged historical state of a subtree up to the point this transaction began.
    ///
    /// This retrieves all relevant historical entries for the `subtree_name` from the backend,
    /// considering the parent tips recorded when this `Transaction` was created (or when the
    /// subtree was first accessed within the transaction). It deserializes the data from each
    /// relevant entry into the CRDT type `T` and merges them according to `T`'s `CRDT::merge`
    /// implementation.
    ///
    /// This is intended for use by `Store` implementations (e.g., in their `get` or `get_all` methods)
    /// to provide the historical context against which staged changes might be applied or compared.
    ///
    /// # Type Parameters
    /// * `T` - The CRDT type to deserialize and merge the historical subtree data into.
    ///
    /// # Arguments
    /// * `subtree_name` - The name of the subtree.
    ///
    /// # Returns
    /// A `Result<T>` containing the merged historical data of type `T`. Returns `Ok(T::default())`
    /// if the subtree has no history prior to this transaction.
    pub(crate) async fn get_full_state<T>(&self, subtree_name: impl AsRef<str> + Send) -> Result<T>
    where
        T: CRDT + Send,
    {
        self.get_full_state_with_descriptor::<T>(
            subtree_name,
            ProjectionDescriptor {
                name: "eidetica/opaque".to_string(),
                version: 0,
            },
        )
        .await
    }

    async fn get_full_state_with_descriptor<T>(
        &self,
        subtree_name: impl AsRef<str> + Send,
        descriptor: ProjectionDescriptor,
    ) -> Result<T>
    where
        T: CRDT + Send,
    {
        let subtree_name = subtree_name.as_ref();

        // Check if we need to initialize subtree tips (get data from RefCell before await)
        let (needs_init, main_parents) = {
            let builder_ref = self.entry_builder.lock().unwrap();
            let builder = builder_ref
                .as_ref()
                .ok_or(TransactionError::TransactionAlreadyCommitted)?;

            let subtrees = builder.subtrees();
            if subtrees.contains(&subtree_name.to_string()) {
                (false, Vec::new())
            } else {
                (true, builder.parents().unwrap_or_default())
            }
        };

        // Initialize subtree tips if needed (async operations)
        if needs_init {
            let current_database_snapshot = self.db.ops().snapshot(self.db.root_id()).await?;

            // Set-equal comparison via Snapshot canonical form.
            let parents_snapshot = Snapshot::from(main_parents.clone());
            let tips = if parents_snapshot == current_database_snapshot {
                let backend = self.db.ops();
                backend
                    .store_snapshot(self.db.root_id(), subtree_name)
                    .await?
                    .into_tips()
            } else {
                // This transaction uses custom tips - use special handler
                self.db
                    .ops()
                    .store_snapshot_at(self.db.root_id(), subtree_name, &parents_snapshot)
                    .await?
                    .into_tips()
            };

            // Update RefCell after async operations
            let mut builder_ref = self.entry_builder.lock().unwrap();
            let builder = builder_ref
                .as_mut()
                .ok_or(TransactionError::TransactionAlreadyCommitted)?;
            builder.set_subtree_parents_mut(subtree_name, tips);
        }

        // Get the parent pointers for this subtree
        let parents = {
            let builder_ref = self.entry_builder.lock().unwrap();
            let builder = builder_ref
                .as_ref()
                .ok_or(TransactionError::TransactionAlreadyCommitted)?;
            builder.subtree_parents(subtree_name).unwrap_or_default()
        };

        // If there are no parents, return a default
        if parents.is_empty() {
            return Ok(T::default());
        }

        // Compute the CRDT state using merge-base ROOT-to-target computation
        self.compute_subtree_state_merge_based(subtree_name, &parents, &descriptor)
            .await
    }

    /// Computes the CRDT state for a subtree using correct recursive merge-base algorithm.
    ///
    /// Algorithm:
    /// 1. If no entries, return default state
    /// 2. If single entry, compute its state recursively
    /// 3. If multiple entries, find their merge base and compute state from there
    ///
    /// # Type Parameters
    /// * `T` - The CRDT type to compute the state for
    ///
    /// # Arguments
    /// * `subtree_name` - The name of the subtree
    /// * `entry_ids` - The entry IDs to compute the merged state for (tips)
    ///
    /// # Returns
    /// A `Result<T>` containing the computed CRDT state
    async fn compute_subtree_state_merge_based<T>(
        &self,
        subtree_name: impl AsRef<str> + Send,
        entry_ids: &[ID],
        descriptor: &ProjectionDescriptor,
    ) -> Result<T>
    where
        T: CRDT + Send,
    {
        // Base case: no entries
        if entry_ids.is_empty() {
            return Ok(T::default());
        }

        let subtree_name = subtree_name.as_ref();

        // If we have a single entry, compute its state recursively
        if entry_ids.len() == 1 {
            return self
                .compute_single_entry_state_recursive(subtree_name, &entry_ids[0], descriptor)
                .await;
        }

        // Multiple entries: check multi-tip cache first
        let cache_id = create_merge_cache_id(entry_ids);

        let cache_request = state::opaque_request(
            self.db.root_id(),
            subtree_name,
            descriptor.clone(),
            cache_id.to_string().into_bytes(),
            crate::backend::CacheScope::Shared,
        );
        if let Some(bytes) = state::load_cached(self.db.ops(), &cache_request).await? {
            let decrypted = self.decrypt_if_needed(subtree_name, &bytes)?;
            let result: T = serde_json::from_slice(&decrypted)?;
            return Ok(result);
        }

        // Cache miss: resolve the merge base and the path to fold in a
        // single call, so both come from one view of the store.
        let merge = self
            .db
            .ops()
            .compute_merge_state(self.db.root_id(), subtree_name, entry_ids)
            .await?;

        let result = match &merge.merge_base {
            Some(base) => {
                // Compute the base state recursively, then fold the path
                // entries (deduplicated, height/ID sorted) on top of it.
                let state: T = self
                    .compute_single_entry_state_recursive(subtree_name, base, descriptor)
                    .await?;
                self.merge_path_entries(subtree_name, state, &merge.path)
                    .await?
            }
            // With no merge base the histories are disjoint — a store
            // created independently on both sides of a fork — so fold the
            // full ancestry from a default state. Convergence must not
            // depend on which peer created the store first. `store_at`
            // batch-fetches the whole entries in fold order: one query
            // instead of a per-ID fetch of the entire history.
            None => {
                let boundary = Snapshot::from(entry_ids.to_vec());
                let entries = self
                    .db
                    .ops()
                    .store_at(self.db.root_id(), subtree_name, &boundary)
                    .await?;
                self.fold_store_entries(subtree_name, &entries)?
            }
        };

        // Cache the computed merge result
        let bytes = self.encrypt_if_needed(subtree_name, &serde_json::to_vec(&result)?)?;
        state::store_cached(self.db.ops(), cache_request, bytes).await?;

        Ok(result)
    }

    /// Computes the CRDT state for a single entry using batch fetching.
    ///
    /// Algorithm:
    /// 1. Check if entry state is cached → return it
    /// 2. Fetch all ancestors in one batch query (sorted by height)
    /// 3. Merge all entries in order from root to target
    /// 4. Cache only the final result
    ///
    /// # Type Parameters
    /// * `T` - The CRDT type to compute the state for
    ///
    /// # Arguments
    /// * `subtree_name` - The name of the subtree
    /// * `entry_id` - The entry ID to compute the state for
    ///
    /// # Returns
    /// A `Result<T>` containing the computed CRDT state for the entry
    fn compute_single_entry_state_recursive<'a, T>(
        &'a self,
        subtree_name: &'a str,
        entry_id: &'a ID,
        descriptor: &'a ProjectionDescriptor,
    ) -> std::pin::Pin<Box<dyn std::future::Future<Output = Result<T>> + Send + 'a>>
    where
        T: CRDT + Send + 'a,
    {
        Box::pin(async move {
            let request = state::opaque_request(
                self.db.root_id(),
                subtree_name,
                descriptor.clone(),
                entry_id.to_string().into_bytes(),
                crate::backend::CacheScope::Shared,
            );
            if let Some(bytes) = state::load_cached(self.db.ops(), &request).await? {
                let decrypted = self.decrypt_if_needed(subtree_name, &bytes)?;
                let result: T = serde_json::from_slice(&decrypted)?;
                return Ok(result);
            }

            // Step 2: Batch fetch all ancestors sorted by height (root first)
            // This single query replaces N recursive queries
            let boundary = Snapshot::from([entry_id.clone()]);
            let entries = self
                .db
                .ops()
                .store_at(self.db.root_id(), subtree_name, &boundary)
                .await?;

            // Step 3: Merge all entries in order (already sorted by height, root first)
            let result: T = self.fold_store_entries(subtree_name, &entries)?;

            // Step 4: Cache only the final result (encrypted if encryptor is registered)
            let bytes = self.encrypt_if_needed(subtree_name, &serde_json::to_vec(&result)?)?;
            state::store_cached(self.db.ops(), request, bytes).await?;

            Ok(result)
        })
    }

    /// Folds already-fetched entries into a CRDT state from the default.
    ///
    /// The entries must be in fold order (height then ID, root first), as
    /// `store_at` returns them.
    fn fold_store_entries<T>(&self, subtree_name: &str, entries: &[Entry]) -> Result<T>
    where
        T: CRDT,
    {
        let mut result = T::default();
        for entry in entries {
            let local_data = if let Ok(data) = entry.data(subtree_name) {
                // Decrypt before deserializing
                let plaintext = self.decrypt_if_needed(subtree_name, data)?;
                serde_json::from_slice::<T>(&plaintext)?
            } else {
                T::default()
            };
            result = result.merge(&local_data)?;
        }
        Ok(result)
    }

    /// Merges a sequence of entries into a CRDT state.
    ///
    /// # Arguments
    /// * `subtree_name` - The name of the subtree
    /// * `initial_state` - The initial CRDT state to merge into
    /// * `entry_ids` - The entry IDs to merge in order
    ///
    /// # Returns
    /// A `Result<T>` containing the merged CRDT state
    async fn merge_path_entries<T>(
        &self,
        subtree_name: &str,
        mut state: T,
        entry_ids: &[ID],
    ) -> Result<T>
    where
        T: CRDT,
    {
        for entry_id in entry_ids {
            let entry = self.db.ops().get(entry_id).await?;

            // Get local data for this entry in the subtree
            let local_data = if let Ok(data) = entry.data(subtree_name) {
                // Decrypt before deserializing
                let plaintext = self.decrypt_if_needed(subtree_name, data)?;
                serde_json::from_slice::<T>(&plaintext)?
            } else {
                T::default()
            };

            state = state.merge(&local_data)?;
        }

        Ok(state)
    }

    /// Commits the transaction, finalizing and persisting the entry to the backend.
    ///
    /// This method:
    /// 1. Takes ownership of the `EntryBuilder` from the internal `Option`
    /// 2. Removes any empty subtrees
    /// 3. Adds metadata if appropriate
    /// 4. Sets authentication if configured
    /// 5. Builds the immutable `Entry` using `EntryBuilder::build()`
    /// 6. Signs the entry if authentication is configured
    /// 7. Validates authentication if present
    /// 8. Calculates the entry's content-addressable ID
    /// 9. Persists the entry to the backend
    /// 10. Returns the ID of the newly created entry
    ///
    /// After commit, the transaction cannot be used again, as the internal
    /// `EntryBuilder` has been consumed.
    ///
    /// # Returns
    /// A `Result<ID>` containing the ID of the committed entry.
    pub async fn commit(self) -> Result<ID> {
        self.commit_inner(None).await
    }

    /// Commit while the caller holds this database's per-tree write lock.
    ///
    /// This is used by operations whose decision depends on a read made before
    /// the write. Ordinary transactions acquire the same lock when persisting.
    pub(crate) async fn commit_under_tree_lock(
        self,
        guard: tokio::sync::OwnedMutexGuard<()>,
    ) -> Result<ID> {
        self.commit_inner(Some(guard)).await
    }

    async fn commit_inner(self, guard: Option<tokio::sync::OwnedMutexGuard<()>>) -> Result<ID> {
        {
            let staged = self.logical_record_mutations.lock().unwrap().clone();
            for (store, mutations) in staged {
                let delta = crate::store::table::encode_entry_delta(&mutations)?;
                self.update_subtree(store, serde_json::to_vec(&delta)?)
                    .await?;
            }
        }

        // Check if this is a settings subtree update and get the effective settings before any borrowing
        let has_settings_update = {
            let builder_cell = self.entry_builder.lock().unwrap();
            let builder = builder_cell
                .as_ref()
                .ok_or(TransactionError::TransactionAlreadyCommitted)?;
            builder.subtrees().contains(&SETTINGS.to_string())
        };

        // Get settings using full CRDT state computation
        let historical_settings = self.get_full_state::<Doc>(SETTINGS).await?;

        // However, if this is a settings update and there's no historical auth but staged auth exists,
        // use the staged settings for validation (this handles initial database creation with auth)
        let effective_settings_for_validation = if has_settings_update {
            let historical_has_auth = matches!(historical_settings.get("auth"), Some(Value::Doc(auth_map)) if !auth_map.is_empty());
            if !historical_has_auth {
                let staged_settings = self.get_local_data::<Doc>(SETTINGS)?.unwrap_or_default();
                let staged_has_auth = matches!(staged_settings.get("auth"), Some(Value::Doc(auth_map)) if !auth_map.is_empty());
                if staged_has_auth {
                    staged_settings
                } else {
                    historical_settings
                }
            } else {
                historical_settings
            }
        } else {
            historical_settings
        };

        // VALIDATION: Ensure that the new settings state (after this transaction) doesn't corrupt auth
        // This prevents committing entries that would corrupt the database's auth configuration
        if has_settings_update {
            // Compute what the new settings state will be after merging local changes
            let local_settings = self.get_local_data::<Doc>(SETTINGS)?.unwrap_or_default();
            let new_settings = effective_settings_for_validation.merge(&local_settings)?;

            // Check if the new settings would have corrupted auth
            if new_settings.is_tombstone("auth") {
                // Auth was explicitly deleted - this would corrupt the database
                return Err(TransactionError::CorruptedAuthConfiguration.into());
            } else if let Some(auth_value) = new_settings.get("auth") {
                // Auth exists in new settings - check if it's the right type
                if !matches!(auth_value, Value::Doc(_)) {
                    // Auth exists but has wrong type (not a Doc) - this would corrupt the database
                    return Err(TransactionError::CorruptedAuthConfiguration.into());
                }
            }
            // If auth is None (not configured), that's fine - we allow empty auth
        }

        // Ensure _index constraint: subtrees referenced in _index must appear in Entry.
        // This adds subtrees with None data if they're referenced in _index but not yet in builder.
        // First, get the data we need before any async operations
        let (_index_data_opt, main_parents, missing_subtrees) = {
            let builder_ref = self.entry_builder.lock().unwrap();
            let builder = builder_ref
                .as_ref()
                .ok_or(TransactionError::TransactionAlreadyCommitted)?;

            let index_data_opt = builder.data(INDEX).ok().cloned();
            let main_parents = builder.parents().unwrap_or_default();
            let existing_subtrees = builder.subtrees();

            // Find missing subtrees
            let missing = if let Some(ref index_data) = index_data_opt
                && let Ok(index_doc) = serde_json::from_slice::<Doc>(index_data)
            {
                index_doc
                    .keys()
                    .filter(|name| !existing_subtrees.contains(&name.to_string()))
                    .cloned()
                    .collect::<Vec<_>>()
            } else {
                Vec::new()
            };

            (index_data_opt, main_parents, missing)
        };

        // Get tips for missing subtrees (async)
        let mut subtree_tips: Vec<(String, Vec<ID>)> = Vec::new();
        for subtree_name in missing_subtrees {
            let tips = self.get_subtree_tips(&subtree_name, &main_parents).await?;
            subtree_tips.push((subtree_name, tips));
        }

        // Now update the builder with the tips
        {
            let mut builder_ref = self.entry_builder.lock().unwrap();
            let builder = builder_ref
                .as_mut()
                .ok_or(TransactionError::TransactionAlreadyCommitted)?;

            for (subtree_name, tips) in subtree_tips {
                builder.set_subtree_parents_mut(&subtree_name, tips);
            }

            builder.remove_empty_subtrees_mut()?;
        }

        // Add metadata with settings snapshot for all entries
        // Get the backend to access the settings snapshot (do async ops before RefCell borrow)
        let db_snapshot = self.db.snapshot().await?;
        let settings_snapshot = self
            .db
            .ops()
            .store_snapshot_at(self.db.root_id(), SETTINGS, &db_snapshot)
            .await?;

        // Clone the builder from RefCell (limit borrow scope to avoid holding across await)
        let mut builder = {
            let builder_cell = self.entry_builder.lock().unwrap();
            let builder_from_cell = builder_cell
                .as_ref()
                .ok_or(TransactionError::TransactionAlreadyCommitted)?;
            builder_from_cell.clone()
        };

        // Parse existing metadata if present, or create new
        let mut metadata = builder
            .metadata()
            .and_then(|m| serde_json::from_slice::<EntryMetadata>(m).ok())
            .unwrap_or(EntryMetadata {
                settings_snapshot: Snapshot::EMPTY,
                entropy: None,
            });

        // Update settings snapshot
        metadata.settings_snapshot = settings_snapshot;

        // Serialize the metadata
        let metadata_json = serde_json::to_vec(&metadata)?;

        // Add metadata to the entry builder
        builder.set_metadata_mut(metadata_json);

        // Handle authentication configuration before building
        // All entries must now be authenticated - fail if no auth key is configured

        // Use provided signing key
        let signing_key = if let Some((ref provided_key, ref identity)) = self.provided_signing_key
        {
            // Use provided signing key directly (already decrypted from UserKeyManager or device key)
            let key_clone = provided_key.clone();

            // Build AuthInfo from the already-typed SigKey identity
            let sig_builder = AuthInfo::builder().key(identity.clone());

            // Set auth ID on the entry builder (without signature initially)
            builder.set_auth_mut(sig_builder.build());

            Some(key_clone)
        } else {
            // No authentication key configured
            return Err(TransactionError::AuthenticationRequired.into());
        };
        // Encrypt subtree data if encryptors are registered
        // This must happen before building the entry to ensure encrypted data is persisted
        {
            let encryptors = self.encryptors.lock().unwrap();
            for subtree_name in builder.subtrees() {
                if let Some(encryptor) = encryptors.get(&subtree_name)
                    && let Ok(plaintext_data) = builder.data(&subtree_name)
                    && !plaintext_data.is_empty()
                {
                    let ciphertext = encryptor.encrypt(plaintext_data)?;
                    builder.set_subtree_data_mut(subtree_name.clone(), ciphertext);
                }
            }
        }

        // Extract height strategy from settings (defaults to Incremental)
        // If this transaction includes settings updates, merge them to get the effective strategy
        let settings_for_height = if has_settings_update {
            let local_settings = self.get_local_data::<Doc>(SETTINGS)?.unwrap_or_default();
            effective_settings_for_validation.merge(&local_settings)?
        } else {
            effective_settings_for_validation.clone()
        };
        let height_strategy: HeightStrategy = settings_for_height
            .get_json("height_strategy")
            .unwrap_or_default();

        // Compute heights from parent entries using the configured strategy
        {
            let backend = self.db.ops();
            let instance = self.db.instance()?;
            let calculator = height_strategy.into_calculator(instance.clock_arc());

            // Compute main tree height using the height strategy
            let main_parents = builder.parents().unwrap_or_default();
            let max_parent_height = if main_parents.is_empty() {
                None
            } else {
                let mut max_height = 0u64;
                for parent_id in &main_parents {
                    if let Ok(parent) = backend.get(parent_id).await {
                        max_height = max_height.max(parent.height());
                    }
                }
                Some(max_height)
            };
            let tree_height = calculator.calculate_height(max_parent_height);
            builder.set_height_mut(tree_height);

            // Compute subtree heights based on per-subtree settings from _index
            // System subtrees (prefixed with _) always inherit from tree.
            // Regular subtrees check _index for a height_strategy override.
            //
            // If a subtree has no override, its height is left as None, which means
            // Entry.subtree_height() will return the tree height (inheritance).
            let index = self.get_index().await.ok();

            for subtree_name in builder.subtrees() {
                // Determine the effective strategy for this subtree:
                // - System subtrees (_settings, _index, etc.): inherit (None)
                // - User subtrees: look up in _index, default to inherit (None)
                let subtree_strategy: Option<HeightStrategy> = if subtree_name.starts_with('_') {
                    // System subtrees always inherit from tree
                    None
                } else if let Some(ref idx) = index {
                    idx.get_subtree_settings(&subtree_name)
                        .await
                        .ok()
                        .and_then(|s| s.height_strategy)
                } else {
                    None
                };

                match subtree_strategy {
                    None => {
                        // Inherit from tree - height stays None (default)
                        // Entry.subtree_height() will return tree height
                    }
                    Some(strategy) => {
                        // Calculate independent height from subtree parents
                        let subtree_calculator = strategy.into_calculator(instance.clock_arc());
                        let subtree_parents =
                            builder.subtree_parents(&subtree_name).unwrap_or_default();
                        let max_subtree_parent_height = if subtree_parents.is_empty() {
                            None
                        } else {
                            let mut max_height = 0u64;
                            for parent_id in &subtree_parents {
                                if let Ok(parent) = backend.get(parent_id).await
                                    && let Ok(height) = parent.subtree_height(&subtree_name)
                                {
                                    max_height = max_height.max(height);
                                }
                            }
                            Some(max_height)
                        };
                        let subtree_height =
                            subtree_calculator.calculate_height(max_subtree_parent_height);
                        builder.set_subtree_height_mut(&subtree_name, Some(subtree_height));
                    }
                }
            }
        }

        // Build the final immutable Entry
        let mut entry = builder.build()?;

        // CRITICAL VALIDATION: Ensure entry structural integrity before commit
        //
        // This validation is crucial because the transaction layer has already:
        // 1. Discovered proper parent relationships through DAG traversal
        // 2. Set up correct subtree parents via find_subtree_parents_from_main_parents()
        // 3. Ensured all references point to valid entries in the backend
        //
        // The validate() call here ensures that:
        // - Non-root entries have main tree parents (preventing orphaned nodes)
        // - Parent IDs are not empty strings (preventing reference errors)
        // - The entry structure is valid before signing and storage
        //
        // This catches any issues early in the transaction, providing clear error
        // messages before the entry is signed or reaches the backend storage layer.
        entry.validate()?;

        // Sign the entry if we have a signing key
        if let Some(signing_key) = signing_key {
            let signature = sign_entry(&entry, &signing_key)?;
            entry = entry.with_auth(|auth| auth.signature = Some(signature));
        }

        // Validate authentication (all entries must be authenticated)
        let mut validator = AuthValidator::new();

        // Get the final settings state for validation
        // IMPORTANT: For permission checking, we must use the historical auth configuration
        // (before this transaction), not the auth configuration from the current entry.
        // This prevents operations from modifying their own permission requirements.

        // Extract AuthSettings from effective settings for validation
        // IMPORTANT: Distinguish between empty auth vs corrupted/deleted auth:
        // - None: No auth ever configured → Allow unsigned operations (empty AuthSettings)
        // - Some(Doc): Normal auth configuration → Use it for validation
        // - Tombstone (deleted): Auth was configured then deleted → CORRUPTED (fail-safe)
        // - Some(other types): Wrong type in auth field → CORRUPTED (fail-safe)
        //
        // NOTE: Doc::get() hides tombstones (returns None for deleted values), so we need
        // to check for tombstones explicitly using is_tombstone() before using get().
        let auth_settings_for_validation = if effective_settings_for_validation.is_tombstone("auth")
        {
            // Auth was configured then explicitly deleted - this is corrupted
            return Err(TransactionError::CorruptedAuthConfiguration.into());
        } else {
            match effective_settings_for_validation.get("auth") {
                Some(Value::Doc(auth_doc)) => auth_doc.clone().into(),
                None => AuthSettings::new(), // Empty auth - never configured
                Some(_) => {
                    // Auth exists but has wrong type (not a Doc) - this is corrupted
                    return Err(TransactionError::CorruptedAuthConfiguration.into());
                }
            }
        };

        let instance = self.db.instance()?;

        // Validate entry (signature + permissions)
        let is_valid = validator
            .validate_entry(&entry, &auth_settings_for_validation, Some(&instance))
            .await?;

        if !is_valid {
            return Err(TransactionError::EntryValidationFailed.into());
        }

        let verification_status = VerificationStatus::Verified;

        // Get the entry's ID
        let id = entry.id();

        // Write entry through Instance which handles backend storage and callback dispatch
        let instance = self.db.instance()?;
        if let Some(guard) = guard {
            instance
                .put_entry_under_tree_lock(
                    guard,
                    self.db.root_id(),
                    verification_status,
                    entry.clone(),
                    WriteSource::Local,
                )
                .await?;
        } else {
            instance
                .put_entry(
                    self.db.root_id(),
                    verification_status,
                    entry.clone(),
                    WriteSource::Local,
                )
                .await?;
        }

        Ok(id)
    }
}