use crate::catalog::DEFAULT_SCOPE_ID;
use crate::catalog::types::Value;
use crate::commit::tx::{ReadAssertion, ReadSet, TransactionEnvelope, WriteClass, WriteIntent};
use crate::commit::validation::{
    KvU64MissingPolicy, KvU64MutatorOp, KvU64OverflowPolicy, KvU64UnderflowPolicy, KvU256MutatorOp,
    MAX_COUNTER_SHARDS, Mutation,
};
use crate::error::AedbError;
use crate::lib_helpers::next_prefix_bytes;
use crate::permission::{CallerContext, Permission};
use crate::query::error::QueryError;
use crate::query::plan::ConsistencyMode;
use crate::query::{KvCursor, KvScanResult, ScopedKvEntry};
use crate::query_authorization::ensure_external_caller_allowed;
use crate::storage::keyspace::KvEntry;
use crate::{
    AedbInstance, CommitResult, ENVELOPE_OVERHEAD_UPPER_BOUND, MUTATION_OVERHEAD_UPPER_BOUND,
    TableU256MutationRequest,
};
use std::ops::Bound;
use std::time::Instant;

impl AedbInstance {
    pub async fn kv_get(
        &self,
        project_id: &str,
        scope_id: &str,
        key: &[u8],
        consistency: ConsistencyMode,
        caller: &CallerContext,
    ) -> Result<Option<KvEntry>, QueryError> {
        ensure_external_caller_allowed(caller)?;
        let snapshot_started = Instant::now();
        let lease = self
            .acquire_snapshot(consistency)
            .await
            .map_err(QueryError::from)?;
        let snapshot_micros = snapshot_started.elapsed().as_micros() as u64;
        let snapshot = &lease.view.keyspace;
        let catalog = &lease.view.catalog;
        if !catalog.has_kv_read_permission(&caller.caller_id, project_id, scope_id, key) {
            return Err(QueryError::PermissionDenied {
                permission: format!("KvRead({project_id}.{scope_id})"),
                scope: caller.caller_id.clone(),
            });
        }
        let execute_started = Instant::now();
        let result = snapshot
            .try_kv_get(project_id, scope_id, key)
            .map_err(QueryError::from)?;
        let execute_micros = execute_started.elapsed().as_micros() as u64;
        self.maybe_log_read_phase(
            "kv_get",
            consistency,
            snapshot_micros,
            execute_micros,
            usize::from(result.is_some()),
            true,
        );
        Ok(result)
    }

    pub async fn kv_get_default_scope(
        &self,
        project_id: &str,
        key: &[u8],
        consistency: ConsistencyMode,
        caller: &CallerContext,
    ) -> Result<Option<KvEntry>, QueryError> {
        self.kv_get(project_id, DEFAULT_SCOPE_ID, key, consistency, caller)
            .await
    }

    pub(crate) async fn kv_get_unchecked(
        &self,
        project_id: &str,
        scope_id: &str,
        key: &[u8],
        consistency: ConsistencyMode,
    ) -> Result<Option<KvEntry>, QueryError> {
        let snapshot_started = Instant::now();
        let lease = self
            .acquire_snapshot(consistency)
            .await
            .map_err(QueryError::from)?;
        let snapshot_micros = snapshot_started.elapsed().as_micros() as u64;
        let execute_started = Instant::now();
        let result = lease
            .view
            .keyspace
            .try_kv_get(project_id, scope_id, key)
            .map_err(QueryError::from)?;
        let execute_micros = execute_started.elapsed().as_micros() as u64;
        self.maybe_log_read_phase(
            "kv_get",
            consistency,
            snapshot_micros,
            execute_micros,
            usize::from(result.is_some()),
            true,
        );
        Ok(result)
    }

    pub async fn kv_get_no_auth(
        &self,
        project_id: &str,
        scope_id: &str,
        key: &[u8],
        consistency: ConsistencyMode,
    ) -> Result<Option<KvEntry>, QueryError> {
        if self.require_authenticated_calls {
            return Err(QueryError::PermissionDenied {
                permission: "kv_get_no_auth is unavailable in secure mode".into(),
                scope: "anonymous".into(),
            });
        }
        self.kv_get_unchecked(project_id, scope_id, key, consistency)
            .await
    }

    pub async fn kv_get_many_no_auth(
        &self,
        project_id: &str,
        scope_id: &str,
        keys: &[Vec<u8>],
        consistency: ConsistencyMode,
    ) -> Result<Vec<Option<KvEntry>>, QueryError> {
        if self.require_authenticated_calls {
            return Err(QueryError::PermissionDenied {
                permission: "kv_get_many_no_auth is unavailable in secure mode".into(),
                scope: "anonymous".into(),
            });
        }

        if keys.is_empty() {
            return Ok(Vec::new());
        }

        let snapshot_started = Instant::now();
        let lease = self
            .acquire_snapshot(consistency)
            .await
            .map_err(QueryError::from)?;
        let snapshot_micros = snapshot_started.elapsed().as_micros() as u64;
        let snapshot = &lease.view.keyspace;
        let execute_started = Instant::now();
        let mut out = Vec::with_capacity(keys.len());
        for key in keys {
            out.push(
                snapshot
                    .try_kv_get(project_id, scope_id, key)
                    .map_err(QueryError::from)?,
            );
        }
        let execute_micros = execute_started.elapsed().as_micros() as u64;
        self.maybe_log_read_phase(
            "kv_get_many",
            consistency,
            snapshot_micros,
            execute_micros,
            keys.len(),
            true,
        );
        Ok(out)
    }

    pub(crate) async fn kv_scan_prefix_unchecked(
        &self,
        project_id: &str,
        scope_id: &str,
        prefix: &[u8],
        limit: u64,
        consistency: ConsistencyMode,
    ) -> Result<Vec<(Vec<u8>, KvEntry)>, QueryError> {
        let snapshot_started = Instant::now();
        let lease = self
            .acquire_snapshot(consistency)
            .await
            .map_err(QueryError::from)?;
        let snapshot_micros = snapshot_started.elapsed().as_micros() as u64;
        let page_size = limit.min(self._config.max_scan_rows as u64) as usize;
        let start_bound = Bound::Included(prefix.to_vec());
        let end_bound = next_prefix_bytes(prefix).map_or(Bound::Unbounded, Bound::Excluded);
        let execute_started = Instant::now();
        let entries = lease
            .view
            .keyspace
            .try_kv_scan_range(project_id, scope_id, start_bound, end_bound, page_size)
            .map_err(QueryError::from)?;
        let execute_micros = execute_started.elapsed().as_micros() as u64;
        self.maybe_log_read_phase(
            "kv_scan_prefix",
            consistency,
            snapshot_micros,
            execute_micros,
            entries.len(),
            true,
        );
        Ok(entries)
    }

    pub async fn kv_scan_prefix_no_auth(
        &self,
        project_id: &str,
        scope_id: &str,
        prefix: &[u8],
        limit: u64,
        consistency: ConsistencyMode,
    ) -> Result<Vec<(Vec<u8>, KvEntry)>, QueryError> {
        if self.require_authenticated_calls {
            return Err(QueryError::PermissionDenied {
                permission: "kv_scan_prefix_no_auth is unavailable in secure mode".into(),
                scope: "anonymous".into(),
            });
        }
        self.kv_scan_prefix_unchecked(project_id, scope_id, prefix, limit, consistency)
            .await
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn kv_scan_prefix(
        &self,
        project_id: &str,
        scope_id: &str,
        prefix: &[u8],
        limit: u64,
        cursor: Option<KvCursor>,
        consistency: ConsistencyMode,
        caller: &CallerContext,
    ) -> Result<KvScanResult, QueryError> {
        ensure_external_caller_allowed(caller)?;
        let effective_consistency = if let Some(c) = &cursor {
            ConsistencyMode::AtSeq(c.snapshot_seq)
        } else {
            consistency
        };
        let snapshot_started = Instant::now();
        let lease = self
            .acquire_snapshot(effective_consistency)
            .await
            .map_err(QueryError::from)?;
        let snapshot_micros = snapshot_started.elapsed().as_micros() as u64;
        let snapshot = &lease.view.keyspace;
        let catalog = &lease.view.catalog;
        let snapshot_seq = lease.view.seq;
        let allowed_prefixes =
            match catalog.kv_read_prefixes_for_caller(&caller.caller_id, project_id, scope_id) {
                Some(prefixes) => prefixes,
                None => {
                    return Err(QueryError::PermissionDenied {
                        permission: format!("KvRead({project_id}.{scope_id})"),
                        scope: caller.caller_id.clone(),
                    });
                }
            };
        if let Some(c) = &cursor
            && c.snapshot_seq != snapshot_seq
        {
            return Err(QueryError::InvalidQuery {
                reason: "cursor snapshot_seq mismatch".into(),
            });
        }
        let page_size = limit.min(self._config.max_scan_rows as u64) as usize;
        let start_bound = cursor
            .as_ref()
            .map_or(Bound::Included(prefix.to_vec()), |c| {
                Bound::Excluded(c.last_key.clone())
            });
        let end_bound = next_prefix_bytes(prefix).map_or(Bound::Unbounded, Bound::Excluded);

        let execute_started = Instant::now();
        let scan_limit = if allowed_prefixes.is_empty() {
            page_size + 1
        } else {
            self._config.max_scan_rows
        };
        let mut entries = snapshot
            .try_kv_scan_range(project_id, scope_id, start_bound, end_bound, scan_limit)
            .map_err(QueryError::from)?;
        if !allowed_prefixes.is_empty() {
            entries.retain(|(k, _)| {
                allowed_prefixes
                    .iter()
                    .any(|allowed| k.starts_with(allowed))
            });
        }

        let truncated = entries.len() > page_size;
        if truncated {
            entries.truncate(page_size);
        }
        let execute_micros = execute_started.elapsed().as_micros() as u64;
        let next_cursor = if truncated {
            entries.last().map(|(k, _)| KvCursor {
                snapshot_seq,
                last_key: k.clone(),
                page_size: page_size as u64,
            })
        } else {
            None
        };
        Ok(KvScanResult {
            entries,
            cursor: next_cursor,
            snapshot_seq,
            truncated,
        })
        .inspect(|res| {
            self.maybe_log_read_phase(
                "kv_scan_prefix",
                effective_consistency,
                snapshot_micros,
                execute_micros,
                res.entries.len(),
                true,
            );
        })
    }

    pub async fn kv_scan_prefix_default_scope(
        &self,
        project_id: &str,
        prefix: &[u8],
        limit: u64,
        cursor: Option<KvCursor>,
        consistency: ConsistencyMode,
        caller: &CallerContext,
    ) -> Result<KvScanResult, QueryError> {
        self.kv_scan_prefix(
            project_id,
            DEFAULT_SCOPE_ID,
            prefix,
            limit,
            cursor,
            consistency,
            caller,
        )
        .await
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn kv_scan_range(
        &self,
        project_id: &str,
        scope_id: &str,
        start: Bound<Vec<u8>>,
        end: Bound<Vec<u8>>,
        limit: u64,
        cursor: Option<KvCursor>,
        consistency: ConsistencyMode,
        caller: &CallerContext,
    ) -> Result<KvScanResult, QueryError> {
        ensure_external_caller_allowed(caller)?;
        let effective_consistency = if let Some(c) = &cursor {
            ConsistencyMode::AtSeq(c.snapshot_seq)
        } else {
            consistency
        };
        let snapshot_started = Instant::now();
        let lease = self
            .acquire_snapshot(effective_consistency)
            .await
            .map_err(QueryError::from)?;
        let snapshot_micros = snapshot_started.elapsed().as_micros() as u64;
        let snapshot = &lease.view.keyspace;
        let catalog = &lease.view.catalog;
        let snapshot_seq = lease.view.seq;
        let allowed_prefixes =
            match catalog.kv_read_prefixes_for_caller(&caller.caller_id, project_id, scope_id) {
                Some(prefixes) => prefixes,
                None => {
                    return Err(QueryError::PermissionDenied {
                        permission: format!("KvRead({project_id}.{scope_id})"),
                        scope: caller.caller_id.clone(),
                    });
                }
            };
        if let Some(c) = &cursor
            && c.snapshot_seq != snapshot_seq
        {
            return Err(QueryError::InvalidQuery {
                reason: "cursor snapshot_seq mismatch".into(),
            });
        }
        let page_size = limit.min(self._config.max_scan_rows as u64) as usize;
        let adjusted_start = match (&cursor, start) {
            (Some(c), _) => Bound::Excluded(c.last_key.clone()),
            (None, b) => b,
        };

        let execute_started = Instant::now();
        let scan_limit = if allowed_prefixes.is_empty() {
            page_size + 1
        } else {
            self._config.max_scan_rows
        };
        let mut entries = snapshot
            .try_kv_scan_range(project_id, scope_id, adjusted_start, end, scan_limit)
            .map_err(QueryError::from)?;
        if !allowed_prefixes.is_empty() {
            entries.retain(|(k, _)| {
                allowed_prefixes
                    .iter()
                    .any(|allowed| k.starts_with(allowed))
            });
        }

        let truncated = entries.len() > page_size;
        if truncated {
            entries.truncate(page_size);
        }
        let execute_micros = execute_started.elapsed().as_micros() as u64;
        let next_cursor = if truncated {
            entries.last().map(|(k, _)| KvCursor {
                snapshot_seq,
                last_key: k.clone(),
                page_size: page_size as u64,
            })
        } else {
            None
        };
        Ok(KvScanResult {
            entries,
            cursor: next_cursor,
            snapshot_seq,
            truncated,
        })
        .inspect(|res| {
            self.maybe_log_read_phase(
                "kv_scan_range",
                effective_consistency,
                snapshot_micros,
                execute_micros,
                res.entries.len(),
                true,
            );
        })
    }

    pub async fn kv_scan_all_scopes(
        &self,
        project_id: &str,
        prefix: &[u8],
        limit: u64,
        consistency: ConsistencyMode,
        caller: &CallerContext,
    ) -> Result<Vec<ScopedKvEntry>, QueryError> {
        ensure_external_caller_allowed(caller)?;
        let lease = self
            .acquire_snapshot(consistency)
            .await
            .map_err(QueryError::from)?;
        let project_read = lease.view.catalog.has_permission(
            &caller.caller_id,
            &Permission::KvRead {
                project_id: project_id.to_string(),
                scope_id: None,
                prefix: None,
            },
        ) || lease.view.catalog.has_permission(
            &caller.caller_id,
            &Permission::ProjectAdmin {
                project_id: project_id.to_string(),
            },
        );
        if !project_read {
            return Err(QueryError::PermissionDenied {
                permission: format!("KvRead({project_id}.*)"),
                scope: caller.caller_id.clone(),
            });
        }
        let snapshot = &lease.view.keyspace;
        let mut out = Vec::new();
        for (ns_id, _) in snapshot.namespaces.iter() {
            let Some(ns_key) = ns_id.as_project_scope_key() else {
                continue;
            };
            let Some((p, scope)) = ns_key.split_once("::") else {
                continue;
            };
            if p != project_id {
                continue;
            }
            let remaining = (limit as usize).saturating_sub(out.len());
            if remaining == 0 {
                return Ok(out);
            }
            for (k, v) in snapshot
                .try_kv_scan_prefix(project_id, scope, prefix, remaining)
                .map_err(QueryError::from)?
            {
                out.push(ScopedKvEntry {
                    scope_id: scope.to_string(),
                    key: k,
                    value: v.value,
                    version: v.version,
                });
                if out.len() >= limit as usize {
                    return Ok(out);
                }
            }
        }
        Ok(out)
    }

    pub async fn kv_set(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        value: Vec<u8>,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvSet {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            value,
        })
        .await
    }

    /// Set a KV entry without a caller identity.
    ///
    /// Explicit anonymous twin of [`Self::kv_set`], mirroring
    /// [`Self::query_no_auth`]. Unavailable in secure mode (open the instance
    /// with [`Self::open_anonymous`] to use it); prefer [`Self::kv_set_as`] in
    /// services.
    pub async fn kv_set_no_auth(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        value: Vec<u8>,
    ) -> Result<CommitResult, AedbError> {
        self.commit_no_auth(Mutation::KvSet {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            value,
        })
        .await
    }

    pub async fn kv_set_default_scope(
        &self,
        project_id: &str,
        key: Vec<u8>,
        value: Vec<u8>,
    ) -> Result<CommitResult, AedbError> {
        self.kv_set(project_id, DEFAULT_SCOPE_ID, key, value).await
    }

    pub async fn kv_set_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        value: Vec<u8>,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvSet {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                value,
            },
        )
        .await
    }

    pub async fn kv_set_many_atomic(
        &self,
        project_id: &str,
        scope_id: &str,
        entries: Vec<(Vec<u8>, Vec<u8>)>,
    ) -> Result<CommitResult, AedbError> {
        if entries.is_empty() {
            return Err(AedbError::Validation(
                "kv_set_many_atomic requires at least one entry".into(),
            ));
        }
        let project_id_owned = project_id.to_string();
        let scope_id_owned = scope_id.to_string();
        let mut encoded_size_hint = Some(ENVELOPE_OVERHEAD_UPPER_BOUND);
        let mut mutations = Vec::with_capacity(entries.len());
        for (key, value) in entries {
            encoded_size_hint = encoded_size_hint.and_then(|size| {
                size.checked_add(
                    MUTATION_OVERHEAD_UPPER_BOUND
                        .saturating_add(project_id.len())
                        .saturating_add(scope_id.len())
                        .saturating_add(key.len())
                        .saturating_add(value.len()),
                )
            });
            mutations.push(Mutation::KvSet {
                project_id: project_id_owned.clone(),
                scope_id: scope_id_owned.clone(),
                key,
                value,
            });
        }
        self.commit_envelope_with_size_hint(
            TransactionEnvelope {
                caller: None,
                idempotency_key: None,
                write_class: WriteClass::Standard,
                assertions: Vec::new(),
                read_set: ReadSet::default(),
                write_intent: WriteIntent { mutations },
                base_seq: 0,
            },
            encoded_size_hint,
        )
        .await
    }

    pub async fn kv_set_many_atomic_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        entries: Vec<(Vec<u8>, Vec<u8>)>,
    ) -> Result<CommitResult, AedbError> {
        if entries.is_empty() {
            return Err(AedbError::Validation(
                "kv_set_many_atomic_as requires at least one entry".into(),
            ));
        }
        let project_id_owned = project_id.to_string();
        let scope_id_owned = scope_id.to_string();
        let mut encoded_size_hint = Some(ENVELOPE_OVERHEAD_UPPER_BOUND);
        let mut mutations = Vec::with_capacity(entries.len());
        for (key, value) in entries {
            encoded_size_hint = encoded_size_hint.and_then(|size| {
                size.checked_add(
                    MUTATION_OVERHEAD_UPPER_BOUND
                        .saturating_add(project_id.len())
                        .saturating_add(scope_id.len())
                        .saturating_add(key.len())
                        .saturating_add(value.len()),
                )
            });
            mutations.push(Mutation::KvSet {
                project_id: project_id_owned.clone(),
                scope_id: scope_id_owned.clone(),
                key,
                value,
            });
        }
        self.commit_envelope_with_size_hint(
            TransactionEnvelope {
                caller: Some(caller),
                idempotency_key: None,
                write_class: WriteClass::Standard,
                assertions: Vec::new(),
                read_set: ReadSet::default(),
                write_intent: WriteIntent { mutations },
                base_seq: 0,
            },
            encoded_size_hint,
        )
        .await
    }

    pub async fn kv_del(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvDel {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
        })
        .await
    }

    /// Delete a KV entry without a caller identity.
    ///
    /// Explicit anonymous twin of [`Self::kv_del`], mirroring
    /// [`Self::query_no_auth`]. Unavailable in secure mode (open the instance
    /// with [`Self::open_anonymous`] to use it); prefer [`Self::kv_del_as`] in
    /// services.
    pub async fn kv_del_no_auth(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
    ) -> Result<CommitResult, AedbError> {
        self.commit_no_auth(Mutation::KvDel {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
        })
        .await
    }

    pub async fn kv_del_default_scope(
        &self,
        project_id: &str,
        key: Vec<u8>,
    ) -> Result<CommitResult, AedbError> {
        self.kv_del(project_id, DEFAULT_SCOPE_ID, key).await
    }

    pub async fn kv_del_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvDel {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
            },
        )
        .await
    }

    pub async fn kv_inc_u256(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 32],
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvIncU256 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            amount_be,
        })
        .await
    }

    pub async fn kv_inc_u256_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 32],
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvIncU256 {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                amount_be,
            },
        )
        .await
    }

    pub async fn kv_dec_u256(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 32],
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvDecU256 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            amount_be,
        })
        .await
    }

    pub async fn kv_dec_u256_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 32],
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvDecU256 {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                amount_be,
            },
        )
        .await
    }

    pub async fn kv_add_u64_ex(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 8],
        on_missing: KvU64MissingPolicy,
        on_overflow: KvU64OverflowPolicy,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvAddU64Ex {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            amount_be,
            on_missing,
            on_overflow,
        })
        .await
    }

    pub async fn kv_sub_u64_ex(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 8],
        on_missing: KvU64MissingPolicy,
        on_underflow: KvU64UnderflowPolicy,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvSubU64Ex {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            amount_be,
            on_missing,
            on_underflow,
        })
        .await
    }

    pub async fn kv_max_u64(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        candidate_be: [u8; 8],
        on_missing: KvU64MissingPolicy,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvMaxU64 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            candidate_be,
            on_missing,
        })
        .await
    }

    pub async fn kv_min_u64(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        candidate_be: [u8; 8],
        on_missing: KvU64MissingPolicy,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvMinU64 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            candidate_be,
            on_missing,
        })
        .await
    }

    pub async fn kv_mutate_u64(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        op: KvU64MutatorOp,
        operand_be: [u8; 8],
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvMutateU64 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            op,
            operand_be,
            expected_seq: None,
        })
        .await
    }

    pub async fn kv_mutate_u64_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        op: KvU64MutatorOp,
        operand_be: [u8; 8],
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvMutateU64 {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                op,
                operand_be,
                expected_seq: None,
            },
        )
        .await
    }

    pub async fn counter_add_sharded(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 8],
        shard_count: u16,
        shard_hint: u32,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::CounterAdd {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            amount_be,
            shard_count,
            shard_hint,
        })
        .await
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn counter_add_sharded_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 8],
        shard_count: u16,
        shard_hint: u32,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::CounterAdd {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                amount_be,
                shard_count,
                shard_hint,
            },
        )
        .await
    }

    pub async fn counter_read_sharded(
        &self,
        project_id: &str,
        scope_id: &str,
        key: &[u8],
        shard_count: u16,
        consistency: ConsistencyMode,
    ) -> Result<u64, AedbError> {
        if shard_count == 0 {
            return Err(AedbError::Validation(
                "counter shard_count must be > 0".into(),
            ));
        }
        if shard_count > MAX_COUNTER_SHARDS {
            return Err(AedbError::Validation(format!(
                "counter shard_count exceeds maximum {}",
                MAX_COUNTER_SHARDS
            )));
        }
        if self.require_authenticated_calls {
            return Err(AedbError::PermissionDenied(
                "authenticated caller required in secure mode".into(),
            ));
        }
        let lease = self.acquire_snapshot(consistency).await?;
        lease
            .view
            .keyspace
            .counter_read_sharded(project_id, scope_id, key, shard_count)
    }

    pub async fn counter_read_sharded_as(
        &self,
        caller: &CallerContext,
        project_id: &str,
        scope_id: &str,
        key: &[u8],
        shard_count: u16,
        consistency: ConsistencyMode,
    ) -> Result<u64, AedbError> {
        if shard_count == 0 {
            return Err(AedbError::Validation(
                "counter shard_count must be > 0".into(),
            ));
        }
        if shard_count > MAX_COUNTER_SHARDS {
            return Err(AedbError::Validation(format!(
                "counter shard_count exceeds maximum {}",
                MAX_COUNTER_SHARDS
            )));
        }
        ensure_external_caller_allowed(caller)?;
        let lease = self.acquire_snapshot(consistency).await?;
        if !lease
            .view
            .catalog
            .has_kv_read_permission(&caller.caller_id, project_id, scope_id, key)
        {
            return Err(AedbError::PermissionDenied(format!(
                "caller={} missing kv read permission for counter",
                caller.caller_id
            )));
        }
        lease
            .view
            .keyspace
            .counter_read_sharded(project_id, scope_id, key, shard_count)
    }

    pub async fn kv_compare_and_swap(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        value: Vec<u8>,
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit_envelope(TransactionEnvelope {
            caller: None,
            idempotency_key: None,
            write_class: WriteClass::Standard,
            assertions: vec![ReadAssertion::KeyVersion {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key: key.clone(),
                expected_seq,
            }],
            read_set: ReadSet::default(),
            write_intent: WriteIntent {
                mutations: vec![Mutation::KvSet {
                    project_id: project_id.to_string(),
                    scope_id: scope_id.to_string(),
                    key,
                    value,
                }],
            },
            base_seq: self.snapshot_probe(ConsistencyMode::AtLatest).await?,
        })
        .await
    }

    pub async fn kv_compare_and_swap_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        value: Vec<u8>,
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit_envelope(TransactionEnvelope {
            caller: Some(caller),
            idempotency_key: None,
            write_class: WriteClass::Standard,
            assertions: vec![ReadAssertion::KeyVersion {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key: key.clone(),
                expected_seq,
            }],
            read_set: ReadSet::default(),
            write_intent: WriteIntent {
                mutations: vec![Mutation::KvSet {
                    project_id: project_id.to_string(),
                    scope_id: scope_id.to_string(),
                    key,
                    value,
                }],
            },
            base_seq: self.snapshot_probe(ConsistencyMode::AtLatest).await?,
        })
        .await
    }

    pub async fn kv_compare_and_inc_u256(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 32],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvMutateU256 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            op: KvU256MutatorOp::Add,
            operand_be: amount_be,
            expected_seq: Some(expected_seq),
        })
        .await
    }

    pub async fn kv_compare_and_add_u64(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 8],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvMutateU64 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            op: KvU64MutatorOp::Add,
            operand_be: amount_be,
            expected_seq: Some(expected_seq),
        })
        .await
    }

    pub async fn kv_compare_and_add_u64_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 8],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvMutateU64 {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                op: KvU64MutatorOp::Add,
                operand_be: amount_be,
                expected_seq: Some(expected_seq),
            },
        )
        .await
    }

    pub async fn kv_compare_and_set_u64(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        value_be: [u8; 8],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvMutateU64 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            op: KvU64MutatorOp::Set,
            operand_be: value_be,
            expected_seq: Some(expected_seq),
        })
        .await
    }

    pub async fn kv_compare_and_set_u64_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        value_be: [u8; 8],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvMutateU64 {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                op: KvU64MutatorOp::Set,
                operand_be: value_be,
                expected_seq: Some(expected_seq),
            },
        )
        .await
    }

    pub async fn kv_compare_and_inc_u256_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 32],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvMutateU256 {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                op: KvU256MutatorOp::Add,
                operand_be: amount_be,
                expected_seq: Some(expected_seq),
            },
        )
        .await
    }

    pub async fn kv_compare_and_set_u256(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        value_be: [u8; 32],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvMutateU256 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            op: KvU256MutatorOp::Set,
            operand_be: value_be,
            expected_seq: Some(expected_seq),
        })
        .await
    }

    pub async fn kv_compare_and_dec_u256(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 32],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvMutateU256 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            op: KvU256MutatorOp::Sub,
            operand_be: amount_be,
            expected_seq: Some(expected_seq),
        })
        .await
    }

    pub async fn kv_compare_and_sub_u64(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 8],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvMutateU64 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            op: KvU64MutatorOp::Sub,
            operand_be: amount_be,
            expected_seq: Some(expected_seq),
        })
        .await
    }

    pub async fn kv_compare_and_sub_u64_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 8],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvMutateU64 {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                op: KvU64MutatorOp::Sub,
                operand_be: amount_be,
                expected_seq: Some(expected_seq),
            },
        )
        .await
    }

    pub async fn kv_compare_and_set_u256_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        value_be: [u8; 32],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvMutateU256 {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                op: KvU256MutatorOp::Set,
                operand_be: value_be,
                expected_seq: Some(expected_seq),
            },
        )
        .await
    }

    pub async fn kv_mutate_u256(
        &self,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        op: KvU256MutatorOp,
        operand_be: [u8; 32],
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::KvMutateU256 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            key,
            op,
            operand_be,
            expected_seq: None,
        })
        .await
    }

    pub async fn kv_mutate_u256_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        op: KvU256MutatorOp,
        operand_be: [u8; 32],
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvMutateU256 {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                op,
                operand_be,
                expected_seq: None,
            },
        )
        .await
    }

    pub async fn kv_compare_and_dec_u256_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        key: Vec<u8>,
        amount_be: [u8; 32],
        expected_seq: u64,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            caller,
            Mutation::KvMutateU256 {
                project_id: project_id.to_string(),
                scope_id: scope_id.to_string(),
                key,
                op: KvU256MutatorOp::Sub,
                operand_be: amount_be,
                expected_seq: Some(expected_seq),
            },
        )
        .await
    }

    pub async fn table_inc_u256(
        &self,
        project_id: &str,
        scope_id: &str,
        table_name: &str,
        primary_key: Vec<Value>,
        column: &str,
        amount_be: [u8; 32],
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::TableIncU256 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            table_name: table_name.to_string(),
            primary_key,
            column: column.to_string(),
            amount_be,
        })
        .await
    }

    pub async fn table_inc_u256_as_with(
        &self,
        request: TableU256MutationRequest,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            request.caller,
            Mutation::TableIncU256 {
                project_id: request.project_id,
                scope_id: request.scope_id,
                table_name: request.table_name,
                primary_key: request.primary_key,
                column: request.column,
                amount_be: request.amount_be,
            },
        )
        .await
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn table_inc_u256_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        table_name: &str,
        primary_key: Vec<Value>,
        column: &str,
        amount_be: [u8; 32],
    ) -> Result<CommitResult, AedbError> {
        self.table_inc_u256_as_with(TableU256MutationRequest {
            caller,
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            table_name: table_name.to_string(),
            primary_key,
            column: column.to_string(),
            amount_be,
        })
        .await
    }

    pub async fn table_dec_u256(
        &self,
        project_id: &str,
        scope_id: &str,
        table_name: &str,
        primary_key: Vec<Value>,
        column: &str,
        amount_be: [u8; 32],
    ) -> Result<CommitResult, AedbError> {
        self.commit(Mutation::TableDecU256 {
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            table_name: table_name.to_string(),
            primary_key,
            column: column.to_string(),
            amount_be,
        })
        .await
    }

    pub async fn table_dec_u256_as_with(
        &self,
        request: TableU256MutationRequest,
    ) -> Result<CommitResult, AedbError> {
        self.commit_as(
            request.caller,
            Mutation::TableDecU256 {
                project_id: request.project_id,
                scope_id: request.scope_id,
                table_name: request.table_name,
                primary_key: request.primary_key,
                column: request.column,
                amount_be: request.amount_be,
            },
        )
        .await
    }

    #[allow(clippy::too_many_arguments)]
    pub async fn table_dec_u256_as(
        &self,
        caller: CallerContext,
        project_id: &str,
        scope_id: &str,
        table_name: &str,
        primary_key: Vec<Value>,
        column: &str,
        amount_be: [u8; 32],
    ) -> Result<CommitResult, AedbError> {
        self.table_dec_u256_as_with(TableU256MutationRequest {
            caller,
            project_id: project_id.to_string(),
            scope_id: scope_id.to_string(),
            table_name: table_name.to_string(),
            primary_key,
            column: column.to_string(),
            amount_be,
        })
        .await
    }
}
