pharos_runtime_database/

operations.rs

//! Logical data operations shared by apps and primary-backend implementations.
//! App code supplies validated entity and field names; an adapter owns physical
//! queries, connections, migrations, and durable transaction semantics.

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

use serde::{Deserialize, Serialize};
use serde_json::{Map, Value};
use sha2::{Digest, Sha256};

use crate::DbBackendConfig;

/// Maximum declared text length; stored values also obey the one-MiB byte bound.
pub const MAX_TEXT_LENGTH: u32 = 1_048_576;

/// Maximum atomic write steps, independently bounded by the eight-MiB payload limit.
pub const MAX_TRANSACTION_STEPS: usize = 2048;

/// Stable failure categories; an unavailable backend never appears as empty data.
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub enum DatabaseError {
    InvalidContract,
    InvalidInput,
    Unsupported,
    Unavailable,
    Conflict,
    /// A transient database lock prevented the operation from starting or committing.
    Busy,
    MissingRecord,
    AccessDenied,
    CapacityExceeded,
    StorageFailure,
}

/// An unrevoked browser session resolved against the caller's clock.
#[derive(Debug, Clone, Copy, Eq, PartialEq)]
pub struct ActiveSession {
    pub user_id: i64,
    pub expires_at_epoch: u64,
}

/// Credentials and account status for shared password verification.
pub struct UserCredentials {
    pub user_id: i64,
    pub password_hash: String,
    pub is_active: bool,
}

/// One business visible through an active tenant membership.
#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
pub struct ManagedBusiness {
    pub id: i64,
    pub slug: String,
    pub name: String,
    pub description: String,
}

/// Bounds one managed-business read before any adapter executes a query.
pub fn validate_managed_business_read(user_id: i64, limit: usize) -> Result<(), DatabaseError> {
    if user_id <= 0 || limit == 0 || limit > 256 {
        return Err(DatabaseError::InvalidInput);
    }
    Ok(())
}

/// Accepts a bounded tenant role identifier without interpreting its authority.
pub fn valid_tenant_role(role: &str) -> bool {
    !role.is_empty()
        && role.len() <= 64
        && role
            .bytes()
            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'_' | b'-'))
}

pub fn validate_username_lookup(username: &str) -> Result<(), DatabaseError> {
    if username.is_empty() || username.len() > 256 || username.contains('\0') {
        return Err(DatabaseError::InvalidInput);
    }
    Ok(())
}

/// Bounds a pre-hashed credential update before an adapter opens a connection.
pub fn validate_password_hash_update(
    user_id: i64,
    password_hash: &str,
) -> Result<(), DatabaseError> {
    if user_id <= 0
        || password_hash.is_empty()
        || password_hash.len() > 1024
        || password_hash.contains('\0')
    {
        return Err(DatabaseError::InvalidInput);
    }
    Ok(())
}

/// Reject malformed or unbounded session lookups before reaching a backend.
pub fn validate_session_lookup(token_hash: &str, now: u64) -> Result<i64, DatabaseError> {
    if token_hash.is_empty() || token_hash.len() > 256 || token_hash.contains('\0') {
        return Err(DatabaseError::InvalidInput);
    }
    let epoch = i64::try_from(now).map_err(|_| DatabaseError::InvalidInput)?;
    canonical_timestamp_boundary(epoch)?;
    Ok(epoch)
}

/// Validated UTC expiry used by each backend's native session representation.
pub struct SessionExpiry {
    pub epoch: i64,
    pub rfc3339_utc: String,
}

/// Validates session input and preserves the exact expiry instant.
pub fn validate_session_issue(
    user_id: i64,
    token_hash: &str,
    expires_at: &str,
) -> Result<SessionExpiry, DatabaseError> {
    validate_session_lookup(token_hash, 0)?;
    if user_id <= 0 || expires_at.len() > 128 {
        return Err(DatabaseError::InvalidInput);
    }
    let parsed = chrono::DateTime::parse_from_rfc3339(expires_at)
        .map_err(|_| DatabaseError::InvalidInput)?
        .with_timezone(&chrono::Utc);
    if parsed.timestamp() <= 0 {
        return Err(DatabaseError::InvalidInput);
    }
    let encoded = parsed.to_rfc3339_opts(chrono::SecondsFormat::Nanos, true);
    if encoded.len() != 30 {
        return Err(DatabaseError::InvalidInput);
    }
    let epoch = parsed.timestamp() + i64::from(parsed.timestamp_subsec_nanos() != 0);
    Ok(SessionExpiry {
        epoch,
        rfc3339_utc: encoded,
    })
}

/// Encodes a whole-second comparison boundary using the canonical storage representation.
pub fn canonical_timestamp_boundary(epoch: i64) -> Result<String, DatabaseError> {
    let encoded = chrono::DateTime::<chrono::Utc>::from_timestamp(epoch, 0)
        .map(|time| time.to_rfc3339_opts(chrono::SecondsFormat::Nanos, true))
        .ok_or(DatabaseError::InvalidInput)?;
    if encoded.len() != 30 {
        return Err(DatabaseError::InvalidInput);
    }
    Ok(encoded)
}

/// Returns the first whole second at which an exact stored session expiry has passed.
pub fn stored_session_expiry_epoch(value: &str) -> Result<u64, DatabaseError> {
    let canonical = canonical_stored_timestamp(value)?;
    let time = chrono::DateTime::parse_from_rfc3339(&canonical)
        .map_err(|_| DatabaseError::StorageFailure)?;
    let expiry = time
        .timestamp()
        .checked_add(i64::from(time.timestamp_subsec_nanos() != 0))
        .ok_or(DatabaseError::StorageFailure)?;
    u64::try_from(expiry).map_err(|_| DatabaseError::StorageFailure)
}

/// Declared field types. Decimal values use an exact, two-place decimal string.
#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FieldType {
    Text,
    Integer,
    Decimal,
    Boolean,
    Timestamp,
    Json,
    Binary,
}

/// Parses a bounded two-place decimal without floating-point arithmetic.
/// JSON numbers are accepted at the input boundary; storage and reads use
/// canonical strings so adapters cannot silently round monetary values.
pub fn decimal_cents(value: &Value) -> Result<i64, DatabaseError> {
    use std::borrow::Cow;
    let text: Cow<'_, str> = match value {
        Value::String(text) => Cow::Borrowed(text.as_str()),
        Value::Number(number) => Cow::Owned(number.to_string()),
        _ => return Err(DatabaseError::InvalidInput),
    };
    let (negative, digits) = text
        .strip_prefix('-')
        .map_or((false, text.as_ref()), |rest| (true, rest));
    let mut parts = digits.split('.');
    let whole = parts.next().ok_or(DatabaseError::InvalidInput)?;
    let fraction = parts.next().unwrap_or("");
    if parts.next().is_some()
        || whole.is_empty()
        || whole.len() > 8
        || !whole.bytes().all(|byte| byte.is_ascii_digit())
        || fraction.len() > 2
        || !fraction.bytes().all(|byte| byte.is_ascii_digit())
    {
        return Err(DatabaseError::InvalidInput);
    }
    let units = whole
        .parse::<i64>()
        .map_err(|_| DatabaseError::InvalidInput)?;
    let cents = match fraction.len() {
        0 => 0,
        1 => i64::from(fraction.as_bytes()[0] - b'0') * 10,
        _ => fraction
            .parse::<i64>()
            .map_err(|_| DatabaseError::InvalidInput)?,
    };
    let amount = units * 100 + cents;
    Ok(if negative { -amount } else { amount })
}

/// Encodes scaled integer storage as the public exact decimal representation.
pub fn decimal_from_cents(cents: i64) -> String {
    let magnitude = cents.unsigned_abs();
    format!(
        "{}{}.{:02}",
        if cents < 0 { "-" } else { "" },
        magnitude / 100,
        magnitude % 100
    )
}

/// One logical field, independent of a SQL column or document representation.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Field {
    pub name: String,
    pub kind: FieldType,
    pub nullable: bool,
    pub max_length: Option<u32>,
    pub default: Option<FieldDefault>,
    pub references: Option<(String, String)>,
    pub on_delete: Option<OnDelete>,
    pub constraints: Vec<FieldConstraint>,
}

/// A portable schema default, including database-generated current time.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FieldDefault {
    Literal(Value),
    CurrentTimestamp,
}

/// Referential action applied when a referenced record is deleted.
#[derive(Debug, Clone, Copy, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum OnDelete {
    Restrict,
    Cascade,
    SetNull,
}

/// Portable field checks retained when app SQL migrations become declarative.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum FieldConstraint {
    Allowed(Vec<Value>),
    LengthBetween { minimum: u32, maximum: u32 },
    GreaterThan(Value),
    AtMost(Value),
    Between { minimum: Value, maximum: Value },
}

/// A physical index key derived from a declared field.
#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum IndexKey {
    Field(String),
    Lowercase(String),
}

/// Indexes may be unique and conditional without embedding vendor SQL.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Index {
    pub name: String,
    pub keys: Vec<IndexKey>,
    pub unique: bool,
    pub filter: Option<Predicate>,
}

/// One logical entity with declared uniqueness and lookup requirements.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Entity {
    pub name: String,
    pub identity: Identity,
    pub fields: Vec<Field>,
    pub unique: Vec<Vec<String>>,
    pub indexes: Vec<Vec<String>>,
    pub conditional_indexes: Vec<Index>,
}

impl Entity {
    /// Accepts a declared unique key or primary identity as an upsert conflict target.
    pub fn accepts_upsert_key(&self, key: &[String]) -> bool {
        if key.is_empty() {
            return false;
        }
        self.unique.iter().any(|candidate| candidate == key)
            || match &self.identity {
                Identity::AutoInteger(field) => key.len() == 1 && key[0] == *field,
                Identity::Composite(fields) => fields == key,
            }
    }
}

/// A generated numeric key or a declared compound primary key.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Identity {
    AutoInteger(String),
    Composite(Vec<String>),
}

impl Identity {
    /// Returns the generated numeric field when an entity uses one.
    pub fn generated_field(&self) -> Option<&str> {
        match self {
            Self::AutoInteger(name) => Some(name),
            Self::Composite(_) => None,
        }
    }

    /// Returns the ordered fields of the primary key.
    pub fn fields(&self) -> Vec<&str> {
        match self {
            Self::AutoInteger(name) => vec![name],
            Self::Composite(names) => names.iter().map(String::as_str).collect(),
        }
    }
}

/// The app-owned logical schema that every configured backend must implement.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AppSchema {
    pub app_id: String,
    pub entities: Vec<Entity>,
}

/// Additive schema changes with a backend-independent migration identity.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum SchemaChange {
    CreateEntity(Entity),
    /// Retires exactly one obsolete physical constraint absent from the logical field.
    /// Other historical constraints, records, identities and indexes remain intact.
    RetireFieldConstraint {
        entity: String,
        field: String,
        constraint: FieldConstraint,
    },
    /// Converts stored values and the physical field using one shared lossless preflight.
    ConvertField {
        entity: String,
        previous: Field,
        field: Field,
    },
    /// Widens an existing integer value to an exact two-place decimal without
    /// changing its name, identity, or historical value.
    WidenIntegerToDecimal {
        entity: String,
        field: String,
    },
    /// Adopts declared records from a historical JSON array into typed storage.
    /// Source filtering and the explicit field list prevent importing unrelated
    /// app data or accepting unrecognized columns.
    CopyJsonRecords {
        source_entity: String,
        source_field: String,
        source_filter: Predicate,
        section: String,
        target_entity: String,
        fields: Vec<String>,
    },
    /// Convert an existing single-column integer primary key into a generated key.
    PromoteIdentity {
        entity: String,
        field: String,
    },
    AddField {
        entity: String,
        field: Field,
    },
    AddUnique {
        entity: String,
        fields: Vec<String>,
    },
    AddIndex {
        entity: String,
        fields: Vec<String>,
    },
    AddConditionalIndex {
        entity: String,
        index: Index,
    },
    Seed {
        entity: String,
        values: Map<String, Value>,
    },
    /// Creates a missing logical record from exactly one qualifying source row.
    /// Existing target records are preserved; absent or ambiguous sources fail closed.
    SeedFrom {
        entity: String,
        absent_filter: Predicate,
        source_entity: String,
        source_filter: Predicate,
        fields: BTreeMap<String, String>,
        values: Map<String, Value>,
    },
    Backfill {
        entity: String,
        filter: Predicate,
        values: Map<String, Value>,
    },
    CopyField {
        entity: String,
        source: String,
        target: String,
        filter: Predicate,
    },
    DeleteWhere {
        entity: String,
        filter: Predicate,
    },
}

/// One ordered migration whose checksum must match the committed ledger entry.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Migration {
    pub app_id: String,
    pub id: String,
    pub checksum: String,
    pub changes: Vec<SchemaChange>,
}

/// A bounded, parameterized predicate over declared fields.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Predicate {
    Equal(String, Value),
    GeneratedEqual(String, usize),
    NotEqual(String, Value),
    LessThan(String, Value),
    AtMost(String, Value),
    GreaterThan(String, Value),
    AtLeast(String, Value),
    Contains(String, String),
    IsNull(String),
    In(String, Vec<Value>),
    All(Vec<Predicate>),
    Any(Vec<Predicate>),
}

/// One sort key for a bounded read; ascending puts nulls first and descending last.
/// Include a unique identity key to make equal-value paging deterministic.
#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
pub struct SortKey {
    pub field: String,
    pub descending: bool,
}

/// A logical read; explicit fields and a bounded page prevent accidental scans.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Query {
    pub entity: String,
    pub fields: Vec<String>,
    pub filter: Option<Predicate>,
    pub order: Vec<SortKey>,
    pub limit: u32,
    pub offset: u32,
}

/// Replaces transaction-local identity references with identities produced by earlier steps.
pub fn materialize_generated_predicate(
    predicate: &Predicate,
    generated_ids: &[Option<i64>],
) -> Result<Predicate, DatabaseError> {
    match predicate {
        Predicate::GeneratedEqual(field, step) => {
            let id = generated_ids
                .get(*step)
                .copied()
                .flatten()
                .filter(|id| *id > 0)
                .ok_or(DatabaseError::InvalidInput)?;
            Ok(Predicate::Equal(field.clone(), Value::from(id)))
        }
        Predicate::All(parts) => Ok(Predicate::All(
            parts
                .iter()
                .map(|part| materialize_generated_predicate(part, generated_ids))
                .collect::<Result<Vec<_>, _>>()?,
        )),
        Predicate::Any(parts) => Ok(Predicate::Any(
            parts
                .iter()
                .map(|part| materialize_generated_predicate(part, generated_ids))
                .collect::<Result<Vec<_>, _>>()?,
        )),
        other => Ok(other.clone()),
    }
}

/// Rejects an identity reference unless an earlier transaction step produces it.
fn validate_generated_predicate_steps(
    predicate: &Predicate,
    steps: &[Mutation],
    current_step: usize,
) -> Result<(), DatabaseError> {
    match predicate {
        Predicate::GeneratedEqual(_, index) => {
            if *index >= current_step
                || !matches!(
                    steps.get(*index),
                    Some(Mutation::Insert { .. } | Mutation::InsertIfAbsent { .. })
                )
            {
                return Err(DatabaseError::InvalidInput);
            }
        }
        Predicate::All(parts) | Predicate::Any(parts) => {
            for part in parts {
                validate_generated_predicate_steps(part, steps, current_step)?;
            }
        }
        _ => {}
    }
    Ok(())
}

/// One bounded, backend-neutral join of two declared entities.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct JoinedQuery {
    pub primary: Query,
    pub secondary_entity: String,
    pub secondary_fields: Vec<String>,
    pub secondary_order: Vec<SortKey>,
    pub primary_key: String,
    pub secondary_key: String,
    pub secondary_filter: Option<Predicate>,
    pub include_unmatched: bool,
}

/// One join result retains the two entity namespaces to avoid field collisions.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct JoinedRecord {
    pub primary: Map<String, Value>,
    pub secondary: Option<Map<String, Value>>,
}

/// Atomically selects one matching record and applies its lease or status fields.
/// The adapter returns the selected numeric identity after the update commits.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Claim {
    pub entity: String,
    pub filter: Predicate,
    pub order: Vec<SortKey>,
    pub values: BTreeMap<String, WriteValue>,
}

/// Aggregates required by declared summaries and guarded writes.
#[derive(Debug, Clone, Eq, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Aggregate {
    Count,
    Sum(String),
    Minimum(String),
    Maximum(String),
}

/// One aggregate against an entity and optional filter.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct AggregateQuery {
    pub entity: String,
    pub filter: Option<Predicate>,
    pub aggregate: Aggregate,
}

/// Identifies the logical type of an aggregate result for exact decoding.
pub fn aggregate_result_kind(
    schema: &AppSchema,
    query: &AggregateQuery,
) -> Result<FieldType, DatabaseError> {
    match &query.aggregate {
        Aggregate::Count => Ok(FieldType::Integer),
        Aggregate::Sum(name) | Aggregate::Minimum(name) | Aggregate::Maximum(name) => {
            let kind = entity(schema, &query.entity)?
                .fields
                .iter()
                .find(|field| field.name == *name)
                .map(|field| field.kind)
                .ok_or(DatabaseError::InvalidInput)?;
            if matches!(query.aggregate, Aggregate::Sum(_))
                && !matches!(kind, FieldType::Integer | FieldType::Decimal)
            {
                return Err(DatabaseError::InvalidInput);
            }
            Ok(kind)
        }
    }
}

/// Decodes a relational aggregate without guessing a text field's logical type.
pub fn decode_aggregate_text(kind: FieldType, text: &str) -> Result<Value, DatabaseError> {
    match kind {
        FieldType::Text | FieldType::Decimal => Ok(Value::String(text.to_owned())),
        FieldType::Timestamp => canonical_stored_timestamp(text).map(Value::String),
        FieldType::Integer => text
            .parse::<i64>()
            .map(Value::from)
            .map_err(|_| DatabaseError::StorageFailure),
        FieldType::Boolean => match text {
            "true" | "1" => Ok(Value::Bool(true)),
            "false" | "0" => Ok(Value::Bool(false)),
            _ => Err(DatabaseError::StorageFailure),
        },
        FieldType::Json => serde_json::from_str::<Value>(text)
            .map_err(|_| DatabaseError::StorageFailure)
            .and_then(|value| canonical_json_value(&value)),
        FieldType::Binary => {
            if !text.len().is_multiple_of(2) || text.len() > 2_097_152 {
                return Err(DatabaseError::StorageFailure);
            }
            text.as_bytes()
                .chunks_exact(2)
                .map(|pair| {
                    let digit = |byte: u8| match byte {
                        b'0'..=b'9' => Some(byte - b'0'),
                        b'a'..=b'f' => Some(byte - b'a' + 10),
                        b'A'..=b'F' => Some(byte - b'A' + 10),
                        _ => None,
                    };
                    Ok(Value::from(
                        digit(pair[0]).ok_or(DatabaseError::StorageFailure)? * 16
                            + digit(pair[1]).ok_or(DatabaseError::StorageFailure)?,
                    ))
                })
                .collect::<Result<Vec<_>, DatabaseError>>()
                .map(Value::Array)
        }
    }
}

/// Expiry transition applied before reservation capacity is measured.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct ReservationExpiry {
    /// Additional declared eligibility, such as an unbound checkout reference.
    #[serde(default)]
    pub filter: Option<Predicate>,
    pub timestamp_field: String,
    pub cutoff: Value,
    pub statuses: Vec<Value>,
    pub to_status: Value,
    pub released_at: Option<(String, Value)>,
    pub reason: Option<(String, Value)>,
}

/// Combines reservation scope and expiry eligibility identically for every adapter.
pub fn reservation_expiry_filter(
    scope: Predicate,
    status_field: &str,
    rule: &ReservationExpiry,
    generated_ids: &[Option<i64>],
) -> Result<Predicate, DatabaseError> {
    let mut filters = vec![
        scope,
        Predicate::In(status_field.to_owned(), rule.statuses.clone()),
        Predicate::AtMost(rule.timestamp_field.clone(), rule.cutoff.clone()),
    ];
    if let Some(filter) = &rule.filter {
        filters.push(materialize_generated_predicate(filter, generated_ids)?);
    }
    Ok(Predicate::All(filters))
}

/// A mutation evaluated as one part of a durable transaction.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum Mutation {
    Insert {
        entity: String,
        values: BTreeMap<String, WriteValue>,
    },
    Upsert {
        entity: String,
        unique_key: Vec<String>,
        values: BTreeMap<String, WriteValue>,
    },
    InsertIfAbsent {
        entity: String,
        unique_key: Vec<String>,
        values: BTreeMap<String, WriteValue>,
    },
    Update {
        entity: String,
        filter: Predicate,
        values: BTreeMap<String, WriteValue>,
        expected_rows: Option<u64>,
    },
    Delete {
        entity: String,
        filter: Predicate,
        expected_rows: Option<u64>,
    },
    RequireCount {
        entity: String,
        filter: Predicate,
        minimum: u64,
        maximum: Option<u64>,
    },
    Reserve {
        entity: String,
        status_field: String,
        active_statuses: Vec<Value>,
        owner_filter: Box<Predicate>,
        scope_filter: Box<Predicate>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        lock_scope_filter: Option<Box<Predicate>>,
        capacity: u64,
        expiry: Option<Box<ReservationExpiry>>,
        values: BTreeMap<String, WriteValue>,
    },
}

/// A write value can refer to an earlier inserted or ensured record.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case")]
pub enum WriteValue {
    Literal(Value),
    GeneratedId(usize),
    GeneratedText {
        index: usize,
        prefix: String,
    },
    GeneratedJson {
        index: usize,
        marker: String,
        template: Value,
    },
}

impl WriteValue {
    /// Identifies the prior transaction result used by a typed generated value.
    pub fn generated_index(&self) -> Option<usize> {
        match self {
            Self::Literal(_) => None,
            Self::GeneratedId(index)
            | Self::GeneratedText { index, .. }
            | Self::GeneratedJson { index, .. } => Some(*index),
        }
    }

    /// Relocates references when a transaction-local plan is appended.
    pub fn relocate(&mut self, offset: usize) -> Result<(), DatabaseError> {
        match self {
            Self::Literal(_) => {}
            Self::GeneratedId(index)
            | Self::GeneratedText { index, .. }
            | Self::GeneratedJson { index, .. } => {
                *index = index
                    .checked_add(offset)
                    .ok_or(DatabaseError::CapacityExceeded)?;
            }
        }
        Ok(())
    }
}

/// Resolves generated references uniformly before any adapter binds parameters.
/// JSON replacement matches the full declared resource marker, never substrings.
pub fn resolve_write_value(
    value: &WriteValue,
    ids: &[Option<i64>],
) -> Result<Value, DatabaseError> {
    let id = |index: usize| {
        ids.get(index)
            .copied()
            .flatten()
            .filter(|id| *id > 0)
            .ok_or(DatabaseError::InvalidInput)
    };
    match value {
        WriteValue::Literal(value) => Ok(value.clone()),
        WriteValue::GeneratedId(index) => Ok(Value::from(id(*index)?)),
        WriteValue::GeneratedText { index, prefix } => {
            Ok(Value::String(format!("{prefix}{}", id(*index)?)))
        }
        WriteValue::GeneratedJson {
            index,
            marker,
            template,
        } => {
            let prefix = marker
                .strip_suffix('0')
                .ok_or(DatabaseError::InvalidInput)?;
            let replacement = format!("{prefix}{}", id(*index)?);
            let mut result = template.clone();
            replace_resource_marker(&mut result, marker, &replacement, 0)?;
            Ok(result)
        }
    }
}

/// Resolves transaction-local expressions and validates the resulting field value
/// before a physical adapter binds it. Template admission alone cannot certify
/// the size or constraints of the generated result.
pub fn resolve_field_write(
    field: &Field,
    value: &WriteValue,
    ids: &[Option<i64>],
) -> Result<Value, DatabaseError> {
    let resolved = resolve_write_value(value, ids)?;
    validate_field_value(field, &resolved)?;
    Ok(resolved)
}

/// Replaces only declared resource keys within bounded JSON nesting.
fn replace_resource_marker(
    value: &mut Value,
    marker: &str,
    replacement: &str,
    depth: usize,
) -> Result<(), DatabaseError> {
    if depth > 32 {
        return Err(DatabaseError::CapacityExceeded);
    }
    match value {
        Value::String(text) if depth == 0 && text == marker => *text = replacement.to_owned(),
        Value::Array(items) => {
            for value in items {
                replace_resource_marker(value, marker, replacement, depth + 1)?;
            }
        }
        Value::Object(items) => {
            for (name, value) in items {
                if name == "resource_id" && value.as_str() == Some(marker) {
                    *value = Value::String(replacement.to_owned());
                } else {
                    replace_resource_marker(value, marker, replacement, depth + 1)?;
                }
            }
        }

        _ => {}
    }
    Ok(())
}

/// A transaction must commit every step or none, including guard decisions.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Transaction {
    pub steps: Vec<Mutation>,
    pub idempotency_key: Option<String>,
}

/// Per-step affected row counts and numeric identities of inserted or ensured records.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct TransactionResult {
    pub affected: Vec<u64>,
    pub generated_ids: Vec<Option<i64>>,
}

/// One record in a bounded export page or import batch.
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub struct Record {
    pub entity: String,
    pub fields: Map<String, Value>,
}

/// Bounds one returned page without creating a second serialized copy of its records.
#[derive(Default)]
pub struct QueryResultBudget {
    bytes: usize,
}

impl QueryResultBudget {
    /// Accounts for a decoded record before retaining it in the returned page.
    pub fn accept(&mut self, record: &Map<String, Value>) -> Result<(), DatabaseError> {
        struct Counter {
            bytes: usize,
            maximum: usize,
        }
        impl std::io::Write for Counter {
            /// Rejects encoded output before the page exceeds its fixed memory budget.
            fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
                self.bytes = self
                    .bytes
                    .checked_add(bytes.len())
                    .ok_or_else(|| std::io::Error::other("page limit"))?;
                if self.bytes > self.maximum {
                    return Err(std::io::Error::other("page limit"));
                }
                Ok(bytes.len())
            }
            /// Completes a counter-only write without an external output buffer.
            fn flush(&mut self) -> std::io::Result<()> {
                Ok(())
            }
        }
        let mut counter = Counter {
            bytes: self.bytes,
            maximum: 8 * 1024 * 1024,
        };
        serde_json::to_writer(&mut counter, record).map_err(|_| DatabaseError::CapacityExceeded)?;
        self.bytes = counter.bytes;
        Ok(())
    }
}

/// Explicit resource bounds for a consistent operator snapshot.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub struct SnapshotLimits {
    pub max_records: u32,
    pub max_bytes: u64,
}

/// Shared keyset pagination and accounting. Adapters must keep every page in
/// the same database read transaction; separate calls to query are insufficient.
pub struct SnapshotCollector {
    schema: AppSchema,
    limits: SnapshotLimits,
    entity_index: usize,
    cursor: Option<Value>,
    records: Vec<Record>,
    bytes: u64,
}

impl SnapshotCollector {
    pub fn new(schema: &AppSchema, limits: SnapshotLimits) -> Result<Self, DatabaseError> {
        validate_schema(schema)?;
        if limits.max_records == 0
            || limits.max_records > 1_000_000
            || limits.max_bytes < 2
            || limits.max_bytes > 256 * 1024 * 1024
        {
            return Err(DatabaseError::InvalidInput);
        }
        Ok(Self {
            schema: schema.clone(),
            limits,
            entity_index: 0,
            cursor: None,
            records: Vec::new(),
            bytes: 2,
        })
    }

    pub fn next_query(&self) -> Result<Option<Query>, DatabaseError> {
        let Some(declaration) = self.schema.entities.get(self.entity_index) else {
            return Ok(None);
        };
        let (filter, order) = identity_page(declaration, self.cursor.clone())?;
        Ok(Some(Query {
            entity: declaration.name.clone(),
            fields: declaration
                .fields
                .iter()
                .map(|field| field.name.clone())
                .collect(),
            filter,
            order,
            limit: 256,
            offset: 0,
        }))
    }

    pub fn accept(&mut self, rows: Vec<Map<String, Value>>) -> Result<(), DatabaseError> {
        let declaration = self
            .schema
            .entities
            .get(self.entity_index)
            .ok_or(DatabaseError::InvalidInput)?;
        if rows.len() > 256 {
            return Err(DatabaseError::StorageFailure);
        }
        let complete = rows.len() < 256;
        for fields in rows {
            if fields.len() != declaration.fields.len() {
                return Err(DatabaseError::StorageFailure);
            }
            for field in &declaration.fields {
                validate_field_value(
                    field,
                    fields
                        .get(&field.name)
                        .ok_or(DatabaseError::StorageFailure)?,
                )?;
            }
            let cursor = match &declaration.identity {
                Identity::AutoInteger(name) => fields
                    .get(name)
                    .cloned()
                    .ok_or(DatabaseError::StorageFailure)?,
                Identity::Composite(names) => Value::Object(
                    names
                        .iter()
                        .map(|name| {
                            fields
                                .get(name)
                                .cloned()
                                .map(|value| (name.clone(), value))
                                .ok_or(DatabaseError::StorageFailure)
                        })
                        .collect::<Result<Map<_, _>, _>>()?,
                ),
            };
            if self.cursor.as_ref() == Some(&cursor) {
                return Err(DatabaseError::StorageFailure);
            }
            let record = Record {
                entity: declaration.name.clone(),
                fields,
            };
            self.bytes = self
                .bytes
                .checked_add(
                    serde_json::to_vec(&record)
                        .map_err(|_| DatabaseError::StorageFailure)?
                        .len() as u64
                        + 1,
                )
                .ok_or(DatabaseError::InvalidInput)?;
            if self.records.len() >= self.limits.max_records as usize
                || self.bytes > self.limits.max_bytes
            {
                return Err(DatabaseError::CapacityExceeded);
            }
            self.cursor = Some(cursor);
            self.records.push(record);
        }
        if complete {
            self.entity_index += 1;
            self.cursor = None;
        }
        Ok(())
    }

    pub fn finish(self) -> Result<Vec<Record>, DatabaseError> {
        if self.entity_index != self.schema.entities.len() {
            return Err(DatabaseError::InvalidInput);
        }
        Ok(self.records)
    }
}

/// Computes a stable SHA-256 fingerprint for a serialized logical operation.
pub fn operation_fingerprint<T: Serialize>(value: &T) -> Result<String, DatabaseError> {
    let value = serde_json::to_value(value).map_err(|_| DatabaseError::InvalidInput)?;
    let mut bytes = Vec::new();
    canonical_json(&value, &mut bytes, 0)?;
    Ok(bytes_fingerprint(&bytes))
}

/// Writes JSON with recursively sorted object keys so migration and replay
/// hashes cannot depend on map insertion order or Serde feature unification.
fn canonical_json(value: &Value, output: &mut Vec<u8>, depth: usize) -> Result<(), DatabaseError> {
    if depth > 64 || output.len() > 8 * 1024 * 1024 {
        return Err(DatabaseError::InvalidInput);
    }
    match value {
        Value::Null => output.extend_from_slice(b"null"),
        Value::Bool(true) => output.extend_from_slice(b"true"),
        Value::Bool(false) => output.extend_from_slice(b"false"),
        Value::Number(number) => output.extend_from_slice(number.to_string().as_bytes()),
        Value::String(string) => output.extend_from_slice(
            serde_json::to_string(string)
                .map_err(|_| DatabaseError::InvalidInput)?
                .as_bytes(),
        ),
        Value::Array(values) => {
            output.push(b'[');
            for (index, item) in values.iter().enumerate() {
                if index > 0 {
                    output.push(b',');
                }
                canonical_json(item, output, depth + 1)?;
            }
            output.push(b']');
        }
        Value::Object(values) => {
            output.push(b'{');
            let mut keys = values.keys().collect::<Vec<_>>();
            keys.sort_unstable();
            for (index, key) in keys.into_iter().enumerate() {
                if index > 0 {
                    output.push(b',');
                }
                output.extend_from_slice(
                    serde_json::to_string(key)
                        .map_err(|_| DatabaseError::InvalidInput)?
                        .as_bytes(),
                );
                output.push(b':');
                canonical_json(&values[key], output, depth + 1)?;
            }
            output.push(b'}');
        }
    }
    if output.len() > 8 * 1024 * 1024 {
        return Err(DatabaseError::InvalidInput);
    }
    Ok(())
}

/// Computes a lowercase SHA-256 fingerprint without retaining source bytes.
pub fn bytes_fingerprint(bytes: &[u8]) -> String {
    let hash = Sha256::digest(bytes);
    let mut result = String::with_capacity(64);
    for byte in hash {
        result.push(char::from(b"0123456789abcdef"[(byte >> 4) as usize]));
        result.push(char::from(b"0123456789abcdef"[(byte & 0x0f) as usize]));
    }
    result
}

/// Decodes a logical byte array without interpreting ordinary JSON arrays as binary.
pub fn binary_bytes(value: &Value) -> Result<Vec<u8>, DatabaseError> {
    let bytes = value.as_array().ok_or(DatabaseError::InvalidInput)?;
    if bytes.len() > 1_048_576 {
        return Err(DatabaseError::CapacityExceeded);
    }
    bytes
        .iter()
        .map(|byte| {
            byte.as_u64()
                .and_then(|byte| u8::try_from(byte).ok())
                .ok_or(DatabaseError::InvalidInput)
        })
        .collect()
}

/// Decodes persisted UTC timestamp text without losing fractional seconds.
/// RFC 3339 represents logical timestamps; native UTC date-time storage may
/// omit its zone. Adapters normalize that physical representation at adoption.
pub fn canonical_stored_timestamp(value: &str) -> Result<String, DatabaseError> {
    let timestamp = chrono::DateTime::parse_from_rfc3339(value)
        .map(|value| value.with_timezone(&chrono::Utc))
        .or_else(|_| {
            chrono::NaiveDateTime::parse_from_str(value, "%Y-%m-%d %H:%M:%S%.f")
                .map(|value| value.and_utc())
        })
        .ok()
        .or_else(|| {
            // Historical physical timestamp columns also stored Unix seconds.
            // Logical input validation continues to require RFC 3339.
            if value.is_empty()
                || value.len() > 20
                || !value.bytes().all(|byte| byte.is_ascii_digit())
            {
                return None;
            }
            chrono::DateTime::<chrono::Utc>::from_timestamp(value.parse().ok()?, 0)
        })
        .ok_or(DatabaseError::StorageFailure)?;
    // Fixed precision makes textual ordering agree with chronological ordering,
    // including zero fractions and instants sharing a fractional prefix.
    let encoded = timestamp.to_rfc3339_opts(chrono::SecondsFormat::Nanos, true);
    if encoded.len() != 30 {
        return Err(DatabaseError::StorageFailure);
    }
    Ok(encoded)
}

/// Validates JSON depth, node count and encoded size without copying the value tree.
pub fn validate_json_value(value: &Value) -> Result<(), DatabaseError> {
    let mut pending = vec![(value, 0usize)];
    let mut nodes = 0usize;
    while let Some((value, depth)) = pending.pop() {
        nodes += 1;
        if nodes > 65536 || depth > 32 {
            return Err(DatabaseError::CapacityExceeded);
        }
        let children = match value {
            Value::Array(values) => values.len(),
            Value::Object(values) => values.len(),
            _ => 0,
        };
        if nodes + pending.len() + children > 65536 {
            return Err(DatabaseError::CapacityExceeded);
        }
        match value {
            Value::String(text) if text.contains('\0') => {
                return Err(DatabaseError::InvalidInput);
            }
            Value::Array(values) => pending.extend(values.iter().map(|value| (value, depth + 1))),
            Value::Object(values) => {
                if values.keys().any(|key| key.contains('\0')) {
                    return Err(DatabaseError::InvalidInput);
                }
                pending.extend(values.values().map(|value| (value, depth + 1)))
            }
            _ => {}
        }
    }
    struct SizeLimit(usize);
    impl std::io::Write for SizeLimit {
        /// Counts encoded bytes and stops serialization at the shared field limit.
        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
            if bytes.len() > self.0 {
                return Err(std::io::Error::other("logical JSON capacity exceeded"));
            }
            self.0 -= bytes.len();
            Ok(bytes.len())
        }
        /// The counter has no buffered output to flush.
        fn flush(&mut self) -> std::io::Result<()> {
            Ok(())
        }
    }
    serde_json::to_writer(SizeLimit(1_048_576), value).map_err(|_| DatabaseError::CapacityExceeded)
}

/// Canonicalizes bounded JSON values so equivalent integral numeric representations
/// and object-key order do not acquire backend-specific equality semantics.
/// Arrays retain their order, and strings never undergo numeric coercion.
pub fn canonical_json_value(value: &Value) -> Result<Value, DatabaseError> {
    validate_json_value(value)?;
    /// Walks a depth- and node-bounded tree before any serialization or adapter conversion.
    fn visit(value: &Value, depth: usize, remaining: &mut usize) -> Result<Value, DatabaseError> {
        if depth > 32 || *remaining == 0 {
            return Err(DatabaseError::CapacityExceeded);
        }
        *remaining -= 1;
        Ok(match value {
            Value::Number(number) if number.as_i64().is_none() && number.as_u64().is_none() => {
                let numeric = number.as_f64().ok_or(DatabaseError::InvalidInput)?;
                if numeric.fract() == 0.0
                    && (-9_223_372_036_854_775_808.0..9_223_372_036_854_775_808.0)
                        .contains(&numeric)
                {
                    Value::from(numeric as i64)
                } else if numeric.fract() == 0.0
                    && (0.0..18_446_744_073_709_551_616.0).contains(&numeric)
                {
                    Value::from(numeric as u64)
                } else {
                    value.clone()
                }
            }
            Value::Array(items) => Value::Array(
                items
                    .iter()
                    .map(|item| visit(item, depth + 1, remaining))
                    .collect::<Result<_, _>>()?,
            ),
            Value::Object(fields) => {
                let mut canonical = fields
                    .iter()
                    .map(|(name, item)| Ok((name.clone(), visit(item, depth + 1, remaining)?)))
                    .collect::<Result<Map<String, Value>, DatabaseError>>()?;
                canonical.sort_keys();
                Value::Object(canonical)
            }
            _ => value.clone(),
        })
    }
    visit(value, 0, &mut 65536)
}

/// Validates a projected stored record against the same typed contract as writes.
/// Malformed persisted values indicate storage corruption, not invalid user input.
/// This operates on already-loaded rows and performs no additional database reads.
pub fn validate_stored_record(
    entity: &Entity,
    record: &Map<String, Value>,
) -> Result<(), DatabaseError> {
    for (name, value) in record {
        let field = entity
            .fields
            .iter()
            .find(|field| field.name == *name)
            .ok_or(DatabaseError::InvalidContract)?;
        validate_field_value(field, value).map_err(|error| match error {
            DatabaseError::CapacityExceeded => DatabaseError::CapacityExceeded,
            _ => DatabaseError::StorageFailure,
        })?;
    }
    Ok(())
}

/// Returns canonical typed values from an already-loaded projection without I/O.
pub fn normalize_stored_record(
    entity: &Entity,
    mut record: Map<String, Value>,
) -> Result<Map<String, Value>, DatabaseError> {
    validate_stored_record(entity, &record)?;
    for field in &entity.fields {
        if let Some(value) = record.get_mut(&field.name)
            && !value.is_null()
        {
            if field.kind == FieldType::Json {
                *value = canonical_json_value(value).map_err(|_| DatabaseError::StorageFailure)?;
            } else if field.kind == FieldType::Timestamp {
                *value = Value::String(canonical_stored_timestamp(
                    value.as_str().ok_or(DatabaseError::StorageFailure)?,
                )?);
            }
        }
    }
    Ok(record)
}

/// Executable logical operations implemented by every registered adapter.
/// App-facing services resolve this operation contract through the configured registry.
pub trait DatabaseOperations: Send + Sync {
    fn backend_name(&self) -> &'static str;
    /// Identifies an adapter-owned local database artifact format for safe historical adoption.
    /// Remote storage returns None; callers must never infer a format from a backend name.
    fn local_artifact_format(&self) -> Option<&'static str>;
    /// Ensures one app-owned database namespace exists before applying its schema.
    fn ensure_database(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
    ) -> Result<(), DatabaseError>;
    /// Loads credentials for one username without deciding password policy.
    fn user_credentials(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        username: &str,
    ) -> Result<Option<UserCredentials>, DatabaseError> {
        validate_username_lookup(username)?;
        let rows = self.query(
            config,
            schema,
            &Query {
                entity: "user_account".to_owned(),
                fields: vec![
                    "id".to_owned(),
                    "password_hash".to_owned(),
                    "is_active".to_owned(),
                ],
                filter: Some(Predicate::Equal(
                    "username".to_owned(),
                    Value::from(username),
                )),
                order: Vec::new(),
                limit: 1,
                offset: 0,
            },
        )?;
        let Some(row) = rows.first() else {
            return Ok(None);
        };
        let user_id = row
            .get("id")
            .and_then(Value::as_i64)
            .ok_or(DatabaseError::StorageFailure)?;
        let password_hash = row
            .get("password_hash")
            .and_then(Value::as_str)
            .ok_or(DatabaseError::StorageFailure)?
            .to_owned();
        let is_active = row
            .get("is_active")
            .and_then(Value::as_bool)
            .ok_or(DatabaseError::StorageFailure)?;
        Ok((user_id > 0).then_some(UserCredentials {
            user_id,
            password_hash,
            is_active,
        }))
    }
    /// Replaces one user's pre-hashed password. Missing users fail explicitly.
    fn update_user_password(
        &self,
        config: &DbBackendConfig,
        user_id: i64,
        password_hash: &str,
    ) -> Result<(), DatabaseError>;
    /// Creates or replaces one browser session in authoritative storage.
    fn issue_session(
        &self,
        config: &DbBackendConfig,
        user_id: i64,
        session_token_hash: &str,
        expires_at: &str,
    ) -> Result<(), DatabaseError>;
    /// Revokes one browser session by token hash.
    fn revoke_session(
        &self,
        config: &DbBackendConfig,
        session_token_hash: &str,
    ) -> Result<(), DatabaseError>;
    /// Revokes every browser session owned by one user.
    fn revoke_user_sessions(
        &self,
        config: &DbBackendConfig,
        user_id: i64,
    ) -> Result<(), DatabaseError>;
    /// Resolves one unrevoked browser session against the supplied clock.
    fn active_session(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        session_token_hash: &str,
        now_epoch_seconds: u64,
    ) -> Result<Option<ActiveSession>, DatabaseError> {
        validate_session_lookup(session_token_hash, now_epoch_seconds)?;
        let rows = self.query(
            config,
            schema,
            &Query {
                entity: "auth_session".to_owned(),
                fields: vec!["user_id".to_owned(), "expires_at".to_owned()],
                filter: Some(Predicate::All(vec![
                    Predicate::Equal(
                        "session_token_hash".to_owned(),
                        Value::from(session_token_hash),
                    ),
                    Predicate::IsNull("revoked_at".to_owned()),
                ])),
                order: Vec::new(),
                limit: 1,
                offset: 0,
            },
        )?;
        let Some(row) = rows.first() else {
            return Ok(None);
        };
        let user_id = row
            .get("user_id")
            .and_then(Value::as_i64)
            .ok_or(DatabaseError::StorageFailure)?;
        let expires_at_epoch = stored_session_expiry_epoch(
            row.get("expires_at")
                .and_then(Value::as_str)
                .ok_or(DatabaseError::StorageFailure)?,
        )?;
        Ok(
            (user_id > 0 && expires_at_epoch > now_epoch_seconds).then_some(ActiveSession {
                user_id,
                expires_at_epoch,
            }),
        )
    }
    /// Records a bounded security admission outcome in authoritative storage.
    fn append_rate_limit_event(
        &self,
        config: &DbBackendConfig,
        policy_key: &str,
        bucket_key: &str,
        outcome: &str,
    ) -> Result<(), DatabaseError>;
    /// Atomically counts and records one request in a moving security window.
    fn admit_rate_limited_request(
        &self,
        config: &DbBackendConfig,
        policy_key: &str,
        bucket_key: &str,
        maximum: usize,
        window_seconds: u64,
    ) -> Result<bool, DatabaseError>;
    /// Applies an app-owned logical migration sequence through this backend.
    /// `CreateEntity` creates a table or collection as appropriate for the
    /// configured adapter; callers never select a vendor migration file.
    fn apply_contract(
        &self,
        config: &DbBackendConfig,
        contract: &crate::contract::AppDataContract,
    ) -> Result<(), DatabaseError> {
        crate::contract::validate_contract(contract)?;
        let already_applied = self.migration_ledger(config, &contract.schema.app_id)?;
        let adoption_schema = crate::contract::legacy_adoption_schema(contract)?;
        for (index, migration) in contract.migrations.iter().enumerate() {
            if let Some(checksum) = already_applied.get(&migration.id) {
                if checksum != &migration.checksum {
                    return Err(DatabaseError::Conflict);
                }
                continue;
            }
            if !already_applied.contains_key(&migration.id)
                && let Some(adoption) = contract
                    .legacy_adoption
                    .as_ref()
                    .filter(|adoption| adoption.logical_migrations.contains(&migration.id))
            {
                let historical_schema = adoption_schema
                    .as_ref()
                    .ok_or(DatabaseError::InvalidContract)?;
                let mut adopted = false;
                let mut conflicting_legacy = false;
                for expected in &adoption.ledgers {
                    match self.adopt_legacy_migration(
                        config,
                        historical_schema,
                        migration,
                        expected,
                    ) {
                        Ok(()) => {
                            adopted = true;
                            break;
                        }
                        Err(DatabaseError::MissingRecord) => {}
                        Err(DatabaseError::Conflict) => conflicting_legacy = true,
                        Err(error) => return Err(error),
                    }
                }
                if adopted {
                    continue;
                }
                if conflicting_legacy {
                    return Err(DatabaseError::Conflict);
                }
            }
            let staged = crate::contract::schema_through_migrations(contract, index + 1)?;
            self.migrate(config, &staged, migration)?;
        }
        if let Some(last) = contract.migrations.last() {
            self.migrate(config, &contract.schema, last)?;
        }
        self.verify_contract(config, contract)
    }
    fn migration_ledger(
        &self,
        config: &DbBackendConfig,
        app_id: &str,
    ) -> Result<BTreeMap<String, String>, DatabaseError>;
    fn applied_migration_checksum(
        &self,
        config: &DbBackendConfig,
        app_id: &str,
        migration_id: &str,
    ) -> Result<Option<String>, DatabaseError> {
        if !valid_identifier(migration_id) {
            return Err(DatabaseError::InvalidInput);
        }
        Ok(self.migration_ledger(config, app_id)?.remove(migration_id))
    }
    fn verify_contract(
        &self,
        config: &DbBackendConfig,
        contract: &crate::contract::AppDataContract,
    ) -> Result<(), DatabaseError> {
        crate::contract::validate_contract(contract)?;
        self.ready(config, &contract.schema)?;
        let ledger = self.migration_ledger(config, &contract.schema.app_id)?;
        for migration in &contract.migrations {
            match ledger.get(&migration.id) {
                Some(checksum) if checksum == &migration.checksum => {}
                Some(_) => return Err(DatabaseError::Conflict),
                None => return Err(DatabaseError::MissingRecord),
            }
        }
        Ok(())
    }
    fn ready(&self, config: &DbBackendConfig, schema: &AppSchema) -> Result<(), DatabaseError>;
    /// Reports whether a declared field exists in authoritative storage.
    /// Document backends report declared fields because records have no fixed columns.
    fn field_exists(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        entity_name: &str,
        field_name: &str,
    ) -> Result<bool, DatabaseError>;
    fn migrate(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        migration: &Migration,
    ) -> Result<(), DatabaseError>;
    /// Stamps one logical migration over a fully verified populated database.
    /// Adapters must compare the complete historical ledger, inspect the
    /// physical schema, and write the logical marker atomically. A mismatch
    /// must never trigger the migration body against populated data.
    fn adopt_legacy_migration(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        migration: &Migration,
        expected_legacy: &BTreeMap<String, String>,
    ) -> Result<(), DatabaseError>;
    fn query(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        query: &Query,
    ) -> Result<Vec<Map<String, Value>>, DatabaseError>;
    /// Executes a bounded two-entity join through the same logical reads on every backend.
    fn join_query(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        request: &JoinedQuery,
    ) -> Result<Vec<JoinedRecord>, DatabaseError> {
        validate_query(schema, &request.primary)?;
        let primary_entity = entity(schema, &request.primary.entity)?;
        let secondary_entity = entity(schema, &request.secondary_entity)?;
        let primary_field = primary_entity
            .fields
            .iter()
            .find(|field| field.name == request.primary_key)
            .ok_or(DatabaseError::InvalidInput)?;
        let secondary_field = secondary_entity
            .fields
            .iter()
            .find(|field| field.name == request.secondary_key)
            .ok_or(DatabaseError::InvalidInput)?;
        if primary_field.kind != secondary_field.kind
            || !matches!(primary_field.kind, FieldType::Integer | FieldType::Text)
            || !request.primary.fields.contains(&request.primary_key)
            || request.primary.limit > 64
            || request.secondary_fields.is_empty()
            || request.secondary_fields.len() > 64
            || request.secondary_fields.iter().any(|name| {
                !secondary_entity
                    .fields
                    .iter()
                    .any(|field| field.name == *name)
            })
        {
            return Err(DatabaseError::InvalidInput);
        }
        if let Some(filter) = &request.secondary_filter {
            validate_predicate(secondary_entity, filter)?;
        }
        let primary = self.query(config, schema, &request.primary)?;
        if primary.is_empty() {
            return Ok(Vec::new());
        }
        let keys = primary
            .iter()
            .filter_map(|record| {
                record
                    .get(&request.primary_key)
                    .filter(|value| !value.is_null())
            })
            .map(|value| (value.to_string(), value.clone()))
            .collect::<BTreeMap<_, _>>();
        if keys.is_empty() {
            return Ok(if request.include_unmatched {
                primary
                    .into_iter()
                    .map(|record| JoinedRecord {
                        primary: record,
                        secondary: None,
                    })
                    .collect()
            } else {
                Vec::new()
            });
        }
        let mut secondary_fields = request.secondary_fields.clone();
        if !secondary_fields.contains(&request.secondary_key) {
            secondary_fields.push(request.secondary_key.clone());
        }
        let key_filter = Predicate::In(request.secondary_key.clone(), keys.into_values().collect());
        let filter = request
            .secondary_filter
            .clone()
            .map_or(key_filter.clone(), |other| {
                Predicate::All(vec![key_filter, other])
            });
        let secondary = self.query(
            config,
            schema,
            &Query {
                entity: request.secondary_entity.clone(),
                fields: secondary_fields,
                filter: Some(filter),
                order: request.secondary_order.clone(),
                limit: 1000,
                offset: 0,
            },
        )?;
        // A full page could hide more matching rows, so fail closed instead of truncating a view.
        if secondary.len() == 1000 {
            return Err(DatabaseError::InvalidInput);
        }
        let mut indexed = BTreeMap::<String, Vec<Map<String, Value>>>::new();
        for record in secondary {
            let key = record
                .get(&request.secondary_key)
                .ok_or(DatabaseError::StorageFailure)?;
            indexed.entry(key.to_string()).or_default().push(record);
        }
        let mut result = Vec::new();
        let mut budget = QueryResultBudget::default();
        for record in primary {
            let matches = record
                .get(&request.primary_key)
                .and_then(|key| indexed.get(&key.to_string()));
            if let Some(matches) = matches {
                for secondary in matches {
                    budget.accept(&record)?;
                    budget.accept(secondary)?;
                    result.push(JoinedRecord {
                        primary: record.clone(),
                        secondary: Some(secondary.clone()),
                    });
                    if result.len() > 1000 {
                        return Err(DatabaseError::InvalidInput);
                    }
                }
            } else if request.include_unmatched {
                budget.accept(&record)?;
                result.push(JoinedRecord {
                    primary: record,
                    secondary: None,
                });
            }
        }
        Ok(result)
    }
    fn aggregate(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        query: &AggregateQuery,
    ) -> Result<Value, DatabaseError>;
    fn transact(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        transaction: &Transaction,
    ) -> Result<TransactionResult, DatabaseError>;
    fn claim_one(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        claim: &Claim,
    ) -> Result<Option<i64>, DatabaseError>;
    fn export_page(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        entity: &str,
        after_identity: Option<Value>,
        limit: u32,
    ) -> Result<Vec<Record>, DatabaseError>;
    /// Exports all declared entities from one consistent read transaction.
    fn export_snapshot(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        limits: SnapshotLimits,
    ) -> Result<Vec<Record>, DatabaseError>;
    /// Replaces declared app records atomically after validating schema and
    /// preconditions. No partial import batch may become visible on failure.
    fn restore_snapshot(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        records: &[Record],
        limits: SnapshotLimits,
        counts: &[crate::snapshot::RestoreCount],
    ) -> Result<(), DatabaseError>;
    fn import_batch(
        &self,
        config: &DbBackendConfig,
        schema: &AppSchema,
        records: &[Record],
    ) -> Result<(), DatabaseError>;
}

/// Enforces the same bounded security keys before any adapter touches storage.
pub fn validate_rate_limit_request(
    policy_key: &str,
    bucket_key: &str,
    maximum: usize,
    window_seconds: u64,
) -> Result<(), DatabaseError> {
    if policy_key.is_empty()
        || policy_key.len() > 256
        || policy_key.contains('\0')
        || bucket_key.len() > 512
        || bucket_key.contains('\0')
        || maximum == 0
        || maximum > 10_000
        || window_seconds == 0
        || window_seconds > 86_400
    {
        return Err(DatabaseError::InvalidInput);
    }
    Ok(())
}

/// Enforces audit row bounds without interpreting user input as query syntax.
pub fn validate_rate_limit_event(
    policy_key: &str,
    bucket_key: &str,
    outcome: &str,
) -> Result<(), DatabaseError> {
    if policy_key.is_empty()
        || policy_key.len() > 256
        || policy_key.contains('\0')
        || bucket_key.len() > 512
        || bucket_key.contains('\0')
        || outcome.is_empty()
        || outcome.len() > 128
        || outcome.contains('\0')
    {
        return Err(DatabaseError::InvalidInput);
    }
    Ok(())
}

/// Validates a bounded claim and its deterministic selection order.
pub fn validate_claim(schema: &AppSchema, claim: &Claim) -> Result<(), DatabaseError> {
    validate_schema(schema)?;
    let declaration = entity(schema, &claim.entity)?;
    validate_predicate(declaration, &claim.filter)?;
    let Some(identity) = declaration.identity.generated_field() else {
        return Err(DatabaseError::InvalidInput);
    };
    if claim.order.is_empty()
        || claim.order.len() > 8
        || !claim.order.iter().any(|key| key.field == identity)
        || claim.order.iter().any(|key| {
            !declaration
                .fields
                .iter()
                .any(|field| field.name == key.field)
        })
        || claim.values.is_empty()
        || claim.values.len() > 32
        || claim.values.contains_key(identity)
        || claim.values.iter().any(|(name, value)| {
            !declaration.fields.iter().any(|field| field.name == *name)
                || !matches!(value, WriteValue::Literal(_))
        })
    {
        return Err(DatabaseError::InvalidInput);
    }
    for (name, value) in &claim.values {
        let field = declaration
            .fields
            .iter()
            .find(|field| field.name == *name)
            .ok_or(DatabaseError::InvalidInput)?;
        let WriteValue::Literal(value) = value else {
            return Err(DatabaseError::InvalidInput);
        };
        validate_field_value(field, value)?;
    }
    Ok(())
}

/// Converts a primary-key cursor to a lexicographic page filter and ordering.
/// Composite cursors carry every key field in one JSON object.
pub fn identity_page(
    declaration: &Entity,
    after: Option<Value>,
) -> Result<(Option<Predicate>, Vec<SortKey>), DatabaseError> {
    let fields = declaration.identity.fields();
    let order = fields
        .iter()
        .map(|field| SortKey {
            field: (*field).to_owned(),
            descending: false,
        })
        .collect();
    let Some(after) = after else {
        return Ok((None, order));
    };
    let values = match &declaration.identity {
        Identity::AutoInteger(_) => vec![after],
        Identity::Composite(_) => {
            let map = after.as_object().ok_or(DatabaseError::InvalidInput)?;
            if map.len() != fields.len() {
                return Err(DatabaseError::InvalidInput);
            }
            fields
                .iter()
                .map(|field| map.get(*field).cloned().ok_or(DatabaseError::InvalidInput))
                .collect::<Result<Vec<_>, _>>()?
        }
    };
    if values.iter().any(Value::is_null) {
        return Err(DatabaseError::InvalidInput);
    }
    let branches = (0..fields.len())
        .map(|index| {
            let mut prefix = (0..index)
                .map(|earlier| {
                    Predicate::Equal(fields[earlier].to_owned(), values[earlier].clone())
                })
                .collect::<Vec<_>>();
            prefix.push(Predicate::GreaterThan(
                fields[index].to_owned(),
                values[index].clone(),
            ));
            Predicate::All(prefix)
        })
        .collect();
    Ok((Some(Predicate::Any(branches)), order))
}

/// Accepts a name only if it is a bounded, portable identifier.
pub fn valid_identifier(value: &str) -> bool {
    let mut bytes = value.bytes();
    value.len() <= 63
        && matches!(bytes.next(), Some(first) if first.is_ascii_alphabetic() || first == b'_')
        && bytes.all(|byte| byte.is_ascii_alphanumeric() || byte == b'_')
}

/// Keeps app declarations separate from adapter metadata and derived storage keys.
fn valid_app_identifier(value: &str) -> bool {
    valid_identifier(value)
        && !value.eq_ignore_ascii_case("_id")
        && !value
            .get(..7)
            .is_some_and(|prefix| prefix.eq_ignore_ascii_case("pharos_"))
        && !value
            .get(..8)
            .is_some_and(|prefix| prefix.eq_ignore_ascii_case("_pharos_"))
}

/// Derives one portable physical index name from a declared logical key.
pub fn generated_index_name(
    prefix: &str,
    entity: &str,
    fields: &[String],
) -> Result<String, DatabaseError> {
    if !matches!(prefix, "pk" | "uq" | "ix" | "fk")
        || !valid_identifier(entity)
        || fields.is_empty()
        || fields.iter().any(|field| !valid_identifier(field))
    {
        return Err(DatabaseError::InvalidContract);
    }
    let full_name = format!("{}_{}_{}", prefix, entity, fields.join("_"));
    if full_name.len() <= 63 {
        return Ok(full_name);
    }
    let digest = operation_fingerprint(&full_name)?;
    Ok(format!("{}_{}", &full_name[..46], &digest[..16]))
}

/// Checks a logical value before any adapter can map it to a driver parameter.
pub fn validate_field_value(field: &Field, value: &Value) -> Result<(), DatabaseError> {
    validate_value(field, value, true)
}

/// Validates a historical field copy before any backend updates existing records.
pub fn validate_copy_field(
    declaration: &Entity,
    source: &str,
    target: &str,
    filter: &Predicate,
) -> Result<(), DatabaseError> {
    let source_field = declaration
        .fields
        .iter()
        .find(|field| field.name == source)
        .ok_or(DatabaseError::InvalidContract)?;
    let target_field = declaration
        .fields
        .iter()
        .find(|field| field.name == target)
        .ok_or(DatabaseError::InvalidContract)?;
    if source == target
        || source_field.kind != target_field.kind
        || (source_field.nullable && !target_field.nullable)
        || target_field.max_length.is_some_and(|limit| {
            source_field
                .max_length
                .is_none_or(|source_limit| source_limit > limit)
        })
        || (!target_field.constraints.is_empty()
            && source_field.constraints != target_field.constraints)
    {
        return Err(DatabaseError::InvalidContract);
    }
    validate_predicate(declaration, filter)
}

fn validate_value(
    field: &Field,
    value: &Value,
    check_constraints: bool,
) -> Result<(), DatabaseError> {
    if value.is_null() {
        return if field.nullable {
            Ok(())
        } else {
            Err(DatabaseError::InvalidInput)
        };
    }
    match field.kind {
        FieldType::Text => {
            let text = value.as_str().ok_or(DatabaseError::InvalidInput)?;
            if field
                .max_length
                .is_some_and(|maximum| text.chars().count() > maximum as usize)
                || text.len() > 1_048_576
                || text.contains('\0')
            {
                return Err(DatabaseError::InvalidInput);
            }
        }
        FieldType::Integer => {
            value.as_i64().ok_or(DatabaseError::InvalidInput)?;
        }
        FieldType::Decimal => {
            decimal_cents(value)?;
        }
        FieldType::Boolean => {
            value.as_bool().ok_or(DatabaseError::InvalidInput)?;
        }
        FieldType::Timestamp => {
            let text = value
                .as_str()
                .filter(|text| text.len() <= 64)
                .ok_or(DatabaseError::InvalidInput)?;
            chrono::DateTime::parse_from_rfc3339(text).map_err(|_| DatabaseError::InvalidInput)?;
            canonical_stored_timestamp(text).map_err(|_| DatabaseError::InvalidInput)?;
        }
        FieldType::Json => {
            validate_json_value(value)?;
            if let Some(maximum) = field.max_length {
                let canonical = canonical_json_value(value)?;
                if serde_json::to_vec(&canonical)
                    .map_err(|_| DatabaseError::InvalidInput)?
                    .len()
                    > maximum as usize
                {
                    return Err(DatabaseError::InvalidInput);
                }
            }
        }
        FieldType::Binary => {
            let bytes = value
                .as_array()
                .filter(|bytes| bytes.len() <= 1_048_576)
                .ok_or(DatabaseError::InvalidInput)?;
            if bytes
                .iter()
                .any(|byte| byte.as_u64().is_none_or(|number| number > 255))
            {
                return Err(DatabaseError::InvalidInput);
            }
            if field
                .max_length
                .is_some_and(|maximum| bytes.len() > maximum as usize)
            {
                return Err(DatabaseError::InvalidInput);
            }
        }
    }
    if !check_constraints {
        return Ok(());
    }
    for constraint in &field.constraints {
        let valid = match constraint {
            FieldConstraint::Allowed(allowed) if field.kind == FieldType::Json => {
                let canonical = canonical_json_value(value)?;
                allowed
                    .iter()
                    .map(canonical_json_value)
                    .collect::<Result<Vec<_>, _>>()?
                    .contains(&canonical)
            }
            FieldConstraint::Allowed(allowed) => allowed.contains(value),
            FieldConstraint::LengthBetween { minimum, maximum } => {
                value.as_str().is_some_and(|text| {
                    (*minimum as usize..=*maximum as usize).contains(&text.chars().count())
                })
            }
            FieldConstraint::GreaterThan(minimum) => numeric_constraint_value(field.kind, value)
                .zip(numeric_constraint_value(field.kind, minimum))
                .is_some_and(|(value, minimum)| value > minimum),
            FieldConstraint::AtMost(maximum) => numeric_constraint_value(field.kind, value)
                .zip(numeric_constraint_value(field.kind, maximum))
                .is_some_and(|(value, maximum)| value <= maximum),
            FieldConstraint::Between { minimum, maximum } => {
                numeric_constraint_value(field.kind, value)
                    .zip(numeric_constraint_value(field.kind, minimum))
                    .zip(numeric_constraint_value(field.kind, maximum))
                    .is_some_and(|((value, minimum), maximum)| (minimum..=maximum).contains(&value))
            }
        };
        if !valid {
            return Err(DatabaseError::InvalidInput);
        }
    }
    Ok(())
}

/// Converts a typed numeric constraint operand to its exact comparison unit.
fn numeric_constraint_value(kind: FieldType, value: &Value) -> Option<i64> {
    match kind {
        FieldType::Integer => value.as_i64(),
        FieldType::Decimal => decimal_cents(value).ok(),
        _ => None,
    }
}

fn field_value_type_matches(kind: FieldType, value: &Value) -> bool {
    match kind {
        FieldType::Text | FieldType::Timestamp => value.is_string(),
        FieldType::Integer => value.as_i64().is_some(),
        FieldType::Boolean => value.is_boolean(),
        FieldType::Json => true,
        FieldType::Binary => value.is_array(),
        FieldType::Decimal => decimal_cents(value).is_ok(),
    }
}

/// Validates the logical schema before an adapter can execute it.
pub fn validate_schema(schema: &AppSchema) -> Result<(), DatabaseError> {
    if !valid_identifier(&schema.app_id)
        || schema.entities.is_empty()
        || schema.entities.len() > 256
    {
        return Err(DatabaseError::InvalidContract);
    }
    let mut names = BTreeSet::new();
    let mut index_names = schema
        .entities
        .iter()
        .map(|entity| entity.name.to_ascii_lowercase())
        .collect::<BTreeSet<_>>();
    for entity in &schema.entities {
        if !valid_app_identifier(&entity.name) || !names.insert(entity.name.to_ascii_lowercase()) {
            return Err(DatabaseError::InvalidContract);
        }
        if entity.fields.is_empty() || entity.fields.len() > 256 {
            return Err(DatabaseError::InvalidContract);
        }
        let mut fields = BTreeSet::new();
        let mut folded_fields = BTreeSet::new();
        for field in &entity.fields {
            if !valid_app_identifier(&field.name)
                || !fields.insert(field.name.as_str())
                || !folded_fields.insert(field.name.to_ascii_lowercase())
            {
                return Err(DatabaseError::InvalidContract);
            }
            if field
                .max_length
                .is_some_and(|length| length == 0 || length > MAX_TEXT_LENGTH)
            {
                return Err(DatabaseError::InvalidContract);
            }
            if field.on_delete.is_some() != field.references.is_some() {
                return Err(DatabaseError::InvalidContract);
            }
            if field.on_delete == Some(OnDelete::SetNull) && !field.nullable {
                return Err(DatabaseError::InvalidContract);
            }
            match &field.default {
                Some(FieldDefault::Literal(value)) => {
                    validate_field_value(field, value)
                        .map_err(|_| DatabaseError::InvalidContract)?;
                }
                Some(FieldDefault::CurrentTimestamp) if field.kind != FieldType::Timestamp => {
                    return Err(DatabaseError::InvalidContract);
                }
                _ => {}
            }
            if field.constraints.len() > 8 {
                return Err(DatabaseError::InvalidContract);
            }
            for constraint in &field.constraints {
                match constraint {
                    FieldConstraint::Allowed(values) if values.is_empty() || values.len() > 64 => {
                        return Err(DatabaseError::InvalidContract);
                    }
                    FieldConstraint::Allowed(values)
                        if values.iter().any(|value| {
                            value.is_null() || !field_value_type_matches(field.kind, value)
                        }) =>
                    {
                        return Err(DatabaseError::InvalidContract);
                    }
                    FieldConstraint::LengthBetween { minimum, maximum }
                        if minimum > maximum || *maximum > MAX_TEXT_LENGTH =>
                    {
                        return Err(DatabaseError::InvalidContract);
                    }
                    FieldConstraint::LengthBetween { .. } if field.kind != FieldType::Text => {
                        return Err(DatabaseError::InvalidContract);
                    }
                    FieldConstraint::GreaterThan(value) | FieldConstraint::AtMost(value)
                        if !matches!(field.kind, FieldType::Integer | FieldType::Decimal)
                            || (field.kind == FieldType::Integer && value.as_i64().is_none())
                            || (field.kind == FieldType::Decimal
                                && decimal_cents(value).is_err()) =>
                    {
                        return Err(DatabaseError::InvalidContract);
                    }
                    FieldConstraint::Between { minimum, maximum }
                        if !matches!(field.kind, FieldType::Integer | FieldType::Decimal)
                            || match field.kind {
                                FieldType::Integer => minimum.as_i64().zip(maximum.as_i64()),
                                FieldType::Decimal => {
                                    decimal_cents(minimum).ok().zip(decimal_cents(maximum).ok())
                                }
                                _ => None,
                            }
                            .is_none_or(|(minimum, maximum)| minimum > maximum) =>
                    {
                        return Err(DatabaseError::InvalidContract);
                    }
                    _ => {}
                }
            }
        }
        match &entity.identity {
            Identity::AutoInteger(name) => {
                if !entity.fields.iter().any(|field| {
                    field.name == *name && field.kind == FieldType::Integer && !field.nullable
                }) {
                    return Err(DatabaseError::InvalidContract);
                }
            }
            Identity::Composite(names) => {
                let mut seen = BTreeSet::new();
                if names.is_empty()
                    || names.len() > 8
                    || names.iter().any(|name| {
                        !seen.insert(name.as_str())
                            || !entity
                                .fields
                                .iter()
                                .any(|field| field.name == *name && !field.nullable)
                    })
                {
                    return Err(DatabaseError::InvalidContract);
                }
            }
        }
        let identity_keys = entity
            .identity
            .fields()
            .into_iter()
            .map(|name| IndexKey::Field(name.to_owned()))
            .collect::<Vec<_>>();
        crate::index_values::validate_index_budget(entity, &identity_keys)?;
        if entity.unique.iter().chain(&entity.indexes).any(|key| {
            key.is_empty()
                || key.len() > 16
                || key.iter().any(|field| !fields.contains(field.as_str()))
        }) {
            return Err(DatabaseError::InvalidContract);
        }
        if entity
            .unique
            .iter()
            .chain(&entity.indexes)
            .flatten()
            .any(|name| {
                entity.fields.iter().any(|field| {
                    field.name == *name
                        && field.kind == FieldType::Text
                        && field.max_length.is_none()
                })
            })
        {
            return Err(DatabaseError::InvalidContract);
        }
        if entity.conditional_indexes.len() > 128 {
            return Err(DatabaseError::InvalidContract);
        }
        crate::index_values::validate_lowercase_sources(entity)?;
        for (prefix, keys) in [("uq", &entity.unique), ("ix", &entity.indexes)] {
            for key in keys {
                crate::index_values::validate_index_budget(
                    entity,
                    &key.iter().cloned().map(IndexKey::Field).collect::<Vec<_>>(),
                )?;
                let name = generated_index_name(prefix, &entity.name, key)?;
                if !index_names.insert(name.to_ascii_lowercase()) {
                    return Err(DatabaseError::InvalidContract);
                }
            }
        }
        for index in &entity.conditional_indexes {
            crate::index_values::validate_index_budget(entity, &index.keys)?;
            if !valid_app_identifier(&index.name)
                || index.name.eq_ignore_ascii_case("primary")
                || index.name.eq_ignore_ascii_case("_id_")
                || !index_names.insert(index.name.to_ascii_lowercase())
                || index.keys.is_empty()
                || index.keys.len() > 16
            {
                return Err(DatabaseError::InvalidContract);
            }
            for key in &index.keys {
                let field = match key {
                    IndexKey::Field(name) | IndexKey::Lowercase(name) => name,
                };
                if !fields.contains(field.as_str()) {
                    return Err(DatabaseError::InvalidContract);
                }
            }
            if let Some(filter) = &index.filter {
                validate_predicate(entity, filter)?;
            }
        }
        crate::index_values::validate_entity_budget(entity)?;
    }
    for entity in &schema.entities {
        for field in &entity.fields {
            if let Some((other_entity, other_field)) = &field.references {
                let Some(target) = schema
                    .entities
                    .iter()
                    .find(|candidate| candidate.name == *other_entity)
                else {
                    return Err(DatabaseError::InvalidContract);
                };
                let Some(target_field) = target
                    .fields
                    .iter()
                    .find(|candidate| candidate.name == *other_field)
                else {
                    return Err(DatabaseError::InvalidContract);
                };
                if field.kind != target_field.kind
                    || !(target.identity.fields().contains(&other_field.as_str())
                        && target.identity.fields().len() == 1
                        || target
                            .unique
                            .iter()
                            .any(|key| key.len() == 1 && key[0] == *other_field))
                {
                    return Err(DatabaseError::InvalidContract);
                }
            }
        }
    }
    Ok(())
}

/// Finds a declared entity without accepting a physical table name from a request.
pub fn entity<'a>(schema: &'a AppSchema, name: &str) -> Result<&'a Entity, DatabaseError> {
    schema
        .entities
        .iter()
        .find(|entity| entity.name == name)
        .ok_or(DatabaseError::InvalidContract)
}

/// Validates a nested predicate against one declared entity and a fixed depth bound.
pub fn validate_predicate(entity: &Entity, predicate: &Predicate) -> Result<(), DatabaseError> {
    let mut pending = vec![(predicate, 0usize)];
    let fields = entity
        .fields
        .iter()
        .map(|field| field.name.as_str())
        .collect::<BTreeSet<_>>();
    let mut nodes = 0usize;
    while let Some((part, depth)) = pending.pop() {
        nodes += 1;
        if nodes > 256 || depth > 8 {
            return Err(DatabaseError::InvalidInput);
        }
        match part {
            Predicate::Equal(field, _)
            | Predicate::GeneratedEqual(field, _)
            | Predicate::NotEqual(field, _)
            | Predicate::LessThan(field, _)
            | Predicate::AtMost(field, _)
            | Predicate::GreaterThan(field, _)
            | Predicate::AtLeast(field, _)
            | Predicate::Contains(field, _)
            | Predicate::IsNull(field)
            | Predicate::In(field, _) => {
                if !fields.contains(field.as_str()) {
                    return Err(DatabaseError::InvalidInput);
                }
                let declaration = entity
                    .fields
                    .iter()
                    .find(|item| item.name == *field)
                    .ok_or(DatabaseError::InvalidInput)?;
                match part {
                    Predicate::Equal(_, value) | Predicate::NotEqual(_, value) => {
                        if !value.is_null() {
                            validate_value(declaration, value, false)?;
                        }
                    }
                    Predicate::GeneratedEqual(_, _) => {
                        if declaration.kind != FieldType::Integer {
                            return Err(DatabaseError::InvalidInput);
                        }
                    }
                    Predicate::LessThan(_, value)
                    | Predicate::AtMost(_, value)
                    | Predicate::GreaterThan(_, value)
                    | Predicate::AtLeast(_, value) => {
                        if value.is_null() {
                            return Err(DatabaseError::InvalidInput);
                        }
                        validate_value(declaration, value, false)?;
                    }
                    Predicate::In(_, values) => {
                        if values.len() > 64 {
                            return Err(DatabaseError::InvalidInput);
                        }
                        for value in values {
                            if value.is_null() {
                                return Err(DatabaseError::InvalidInput);
                            }
                            validate_value(declaration, value, false)?;
                        }
                    }
                    Predicate::IsNull(_) => {}
                    Predicate::Contains(_, needle) => {
                        if declaration.kind != FieldType::Text
                            || needle.is_empty()
                            || needle.len() > 256
                            || needle.contains('\0')
                        {
                            return Err(DatabaseError::InvalidInput);
                        }
                    }
                    Predicate::All(_) | Predicate::Any(_) => unreachable!(),
                }
            }
            Predicate::All(parts) | Predicate::Any(parts) => {
                if parts.is_empty() || parts.len() > 32 {
                    return Err(DatabaseError::InvalidInput);
                }
                pending.extend(parts.iter().map(|part| (part, depth + 1)));
            }
        }
    }
    Ok(())
}

/// Declares one bounded relationship read independently of its configured adapter.
pub struct RelatedRowsQuery<'a> {
    pub entity: &'a str,
    pub foreign_key: &'a str,
    pub keys: &'a [Value],
    pub fields: &'a [String],
    pub maximum: usize,
}

/// Reads a bounded relationship set using the contract's operand and result
/// limits. Duplicate keys are removed before paging; no vendor query escapes
/// the operation interface.
pub fn query_related_rows(
    adapter: &dyn DatabaseOperations,
    config: &DbBackendConfig,
    schema: &AppSchema,
    query: RelatedRowsQuery<'_>,
) -> Result<Vec<serde_json::Map<String, Value>>, DatabaseError> {
    let RelatedRowsQuery {
        entity: entity_name,
        foreign_key,
        keys,
        fields,
        maximum,
    } = query;
    if keys.len() > 1000 || maximum == 0 || maximum > 10_000 {
        return Err(DatabaseError::InvalidInput);
    }
    let definition = entity(schema, entity_name)?;
    let selected = if fields.is_empty() {
        definition
            .fields
            .iter()
            .map(|field| field.name.clone())
            .collect()
    } else {
        fields.to_vec()
    };
    let mut unique = BTreeMap::new();
    for key in keys {
        if key.is_null() || key.is_array() || key.is_object() {
            return Err(DatabaseError::InvalidInput);
        }
        unique.insert(
            serde_json::to_string(key).map_err(|_| DatabaseError::InvalidInput)?,
            key.clone(),
        );
    }
    let keys = unique.into_values().collect::<Vec<_>>();
    let mut result = Vec::new();
    for keys in keys.chunks(64) {
        let mut offset = 0;
        loop {
            let limit = (maximum - result.len() + 1).min(1000) as u32;
            let rows = adapter.query(
                config,
                schema,
                &Query {
                    entity: entity_name.to_owned(),
                    fields: selected.clone(),
                    filter: Some(Predicate::In(foreign_key.to_owned(), keys.to_vec())),
                    order: match &definition.identity {
                        Identity::AutoInteger(field) => vec![SortKey {
                            field: field.clone(),
                            descending: false,
                        }],
                        Identity::Composite(fields) => fields
                            .iter()
                            .map(|field| SortKey {
                                field: field.clone(),
                                descending: false,
                            })
                            .collect(),
                    },
                    limit,
                    offset,
                },
            )?;
            let received = rows.len();
            result.extend(rows);
            if result.len() > maximum {
                return Err(DatabaseError::CapacityExceeded);
            }
            if received < limit as usize {
                break;
            }
            offset += limit;
        }
    }
    Ok(result)
}

/// Validates one bounded query before lowering it to a backend operation.
pub fn validate_query(schema: &AppSchema, query: &Query) -> Result<(), DatabaseError> {
    validate_schema(schema)?;
    let entity = entity(schema, &query.entity)?;
    let fields = entity
        .fields
        .iter()
        .map(|field| field.name.as_str())
        .collect::<BTreeSet<_>>();
    if query.fields.is_empty()
        || query.fields.len() > 256
        || query
            .fields
            .iter()
            .any(|field| !fields.contains(field.as_str()))
        || query.order.len() > 8
        || query
            .order
            .iter()
            .any(|key| !fields.contains(key.field.as_str()))
        || query.limit == 0
        || query.limit > 1000
        || query.offset > 1_000_000
    {
        return Err(DatabaseError::InvalidInput);
    }
    if serde_json::to_vec(query)
        .map_err(|_| DatabaseError::InvalidInput)?
        .len()
        > 1_048_576
    {
        return Err(DatabaseError::InvalidInput);
    }
    if let Some(filter) = &query.filter {
        validate_predicate(entity, filter)?;
        materialize_generated_predicate(filter, &[])?;
    }
    Ok(())
}

/// Validates a write transaction without exposing a backend statement surface.
pub fn validate_transaction(
    schema: &AppSchema,
    transaction: &Transaction,
) -> Result<(), DatabaseError> {
    validate_schema(schema)?;
    if transaction.steps.is_empty()
        || transaction.steps.len() > MAX_TRANSACTION_STEPS
        || transaction
            .idempotency_key
            .as_ref()
            .is_some_and(|key| key.is_empty() || key.len() > 128)
    {
        return Err(DatabaseError::InvalidInput);
    }
    if serde_json::to_vec(transaction)
        .map_err(|_| DatabaseError::InvalidInput)?
        .len()
        > 8 * 1024 * 1024
    {
        return Err(DatabaseError::InvalidInput);
    }
    for (step_index, step) in transaction.steps.iter().enumerate() {
        let (name, values, filter) = match step {
            Mutation::Insert { entity, values }
            | Mutation::Upsert { entity, values, .. }
            | Mutation::InsertIfAbsent { entity, values, .. }
            | Mutation::Reserve { entity, values, .. } => (entity, Some(values), None),
            Mutation::Update {
                entity,
                values,
                filter,
                ..
            } => (entity, Some(values), Some(filter)),
            Mutation::Delete { entity, filter, .. }
            | Mutation::RequireCount { entity, filter, .. } => (entity, None, Some(filter)),
        };
        let declaration = entity(schema, name)?;
        if let Mutation::Reserve {
            status_field,
            active_statuses,
            owner_filter,
            scope_filter,
            lock_scope_filter,
            capacity,
            expiry,
            ..
        } = step
        {
            let status = declaration
                .fields
                .iter()
                .find(|field| field.name == *status_field)
                .ok_or(DatabaseError::InvalidContract)?;
            if active_statuses.is_empty() || active_statuses.len() > 16 || *capacity > 100_000 {
                return Err(DatabaseError::InvalidInput);
            }
            for value in active_statuses {
                validate_field_value(status, value)?;
            }
            validate_predicate(declaration, owner_filter)?;
            validate_predicate(declaration, scope_filter)?;
            validate_generated_predicate_steps(owner_filter, &transaction.steps, step_index)?;
            validate_generated_predicate_steps(scope_filter, &transaction.steps, step_index)?;
            if let Some(lock_scope_filter) = lock_scope_filter {
                validate_predicate(declaration, lock_scope_filter)?;
                validate_generated_predicate_steps(
                    lock_scope_filter,
                    &transaction.steps,
                    step_index,
                )?;
            }
            if let Some(expiry) = expiry {
                if let Some(filter) = &expiry.filter {
                    validate_predicate(declaration, filter)?;
                    validate_generated_predicate_steps(filter, &transaction.steps, step_index)?;
                }
                let timestamp = declaration
                    .fields
                    .iter()
                    .find(|field| field.name == expiry.timestamp_field)
                    .ok_or(DatabaseError::InvalidContract)?;
                validate_field_value(timestamp, &expiry.cutoff)?;
                validate_field_value(status, &expiry.to_status)?;
                if expiry.statuses.is_empty() || expiry.statuses.len() > 16 {
                    return Err(DatabaseError::InvalidInput);
                }
                for value in &expiry.statuses {
                    validate_field_value(status, value)?;
                }
                for extra in [&expiry.released_at, &expiry.reason].into_iter().flatten() {
                    let field = declaration
                        .fields
                        .iter()
                        .find(|field| field.name == extra.0)
                        .ok_or(DatabaseError::InvalidContract)?;
                    validate_field_value(field, &extra.1)?;
                }
            }
        }
        let fields = declaration
            .fields
            .iter()
            .map(|field| field.name.as_str())
            .collect::<BTreeSet<_>>();
        if let Some(values) = values
            && (values.is_empty()
                || values.len() > 64
                || values.keys().any(|name| !fields.contains(name.as_str())))
        {
            return Err(DatabaseError::InvalidInput);
        }
        if let Some(values) = values
            && values.values().any(|value| {
                value.generated_index().is_some_and(|index| {
                    index >= step_index
                        || !matches!(
                            transaction.steps[index],
                            Mutation::Insert { .. } | Mutation::InsertIfAbsent { .. }
                        )
                })
            })
        {
            return Err(DatabaseError::InvalidInput);
        }
        if let Some(values) = values {
            for (name, value) in values {
                let field = declaration
                    .fields
                    .iter()
                    .find(|field| field.name == *name)
                    .ok_or(DatabaseError::InvalidInput)?;
                match value {
                    WriteValue::Literal(value) => validate_field_value(field, value)?,
                    WriteValue::GeneratedId(index)
                    | WriteValue::GeneratedText { index, .. }
                    | WriteValue::GeneratedJson { index, .. } => {
                        let source = match transaction.steps.get(*index) {
                            Some(
                                Mutation::Insert { entity, .. }
                                | Mutation::InsertIfAbsent { entity, .. },
                            ) => entity,
                            _ => return Err(DatabaseError::InvalidInput),
                        };
                        if entity(schema, source)?.identity.generated_field().is_none() {
                            return Err(DatabaseError::InvalidInput);
                        }
                        match value {
                            WriteValue::GeneratedId(_) if field.kind == FieldType::Integer => {}
                            WriteValue::GeneratedText { prefix, .. }
                                if field.kind == FieldType::Text && prefix.len() <= 128 =>
                            {
                                if field.max_length.is_some_and(|maximum| {
                                    prefix.chars().count() + 20 > maximum as usize
                                }) {
                                    return Err(DatabaseError::InvalidInput);
                                }
                            }
                            WriteValue::GeneratedJson {
                                marker, template, ..
                            } if field.kind == FieldType::Json
                                && marker.len() <= 128
                                && marker.ends_with(":0") =>
                            {
                                validate_field_value(field, template)?;
                            }
                            _ => return Err(DatabaseError::InvalidInput),
                        }
                    }
                }
            }
            if matches!(
                step,
                Mutation::Insert { .. }
                    | Mutation::Upsert { .. }
                    | Mutation::InsertIfAbsent { .. }
                    | Mutation::Reserve { .. }
            ) && declaration.fields.iter().any(|field| {
                !field.nullable
                    && field.default.is_none()
                    && declaration.identity.generated_field() != Some(field.name.as_str())
                    && !values.contains_key(&field.name)
            }) {
                return Err(DatabaseError::InvalidInput);
            }
            if matches!(step, Mutation::Update { .. })
                && declaration
                    .identity
                    .fields()
                    .iter()
                    .any(|name| values.contains_key(*name))
            {
                return Err(DatabaseError::InvalidInput);
            }
        }
        if let Some(filter) = filter {
            validate_predicate(declaration, filter)?;
            validate_generated_predicate_steps(filter, &transaction.steps, step_index)?;
        }
        if let Mutation::Upsert {
            unique_key, values, ..
        }
        | Mutation::InsertIfAbsent {
            unique_key, values, ..
        } = step
            && (!declaration.accepts_upsert_key(unique_key)
                || unique_key.iter().any(|field| !values.contains_key(field)))
        {
            return Err(DatabaseError::InvalidContract);
        }
        if let Mutation::RequireCount {
            minimum,
            maximum: Some(maximum),
            ..
        } = step
            && minimum > maximum
        {
            return Err(DatabaseError::InvalidInput);
        }
    }
    Ok(())
}

/// Guards a bounded read set while preserving transaction capacity for its writes.
/// Complete identities permit grouped checks; projections without identities use single-row checks.
pub fn guard_read_set(
    schema: &AppSchema,
    entity_name: &str,
    rows: &[&serde_json::Map<String, Value>],
) -> Result<Vec<Mutation>, DatabaseError> {
    let declaration = entity(schema, entity_name)?;
    if rows.len() > 100 {
        return Err(DatabaseError::CapacityExceeded);
    }
    if rows.is_empty() {
        return Ok(Vec::new());
    }
    let identity = declaration.identity.fields();
    let complete = rows
        .iter()
        .all(|row| identity.iter().all(|field| row.contains_key(*field)));
    let fields = rows
        .iter()
        .map(|row| row.len())
        .max()
        .ok_or(DatabaseError::InvalidInput)?;
    if fields == 0 || fields > 64 {
        return Err(DatabaseError::InvalidInput);
    }
    let group_size = if complete {
        (255 / (fields + 3)).clamp(1, 32)
    } else {
        1
    };
    let mut seen = BTreeSet::new();
    let mut guards = Vec::new();
    for chunk in rows.chunks(group_size) {
        let mut alternatives = Vec::with_capacity(chunk.len());
        for row in chunk {
            if complete {
                let key = identity
                    .iter()
                    .map(|field| &row[*field])
                    .collect::<Vec<_>>();
                if !seen.insert(operation_fingerprint(&key)?) {
                    return Err(DatabaseError::InvalidInput);
                }
            }
            let predicates = row
                .iter()
                .map(|(field, value)| {
                    if value.is_null() {
                        Predicate::IsNull(field.clone())
                    } else {
                        Predicate::Equal(field.clone(), value.clone())
                    }
                })
                .collect::<Vec<_>>();
            let mut groups = predicates
                .chunks(32)
                .map(|chunk| Predicate::All(chunk.to_vec()))
                .collect::<Vec<_>>();
            alternatives.push(if groups.len() == 1 {
                groups.remove(0)
            } else {
                Predicate::All(groups)
            });
        }
        let filter = if alternatives.len() == 1 {
            alternatives.remove(0)
        } else {
            Predicate::Any(alternatives)
        };
        validate_predicate(declaration, &filter)?;
        guards.push(Mutation::RequireCount {
            entity: entity_name.to_owned(),
            filter,
            minimum: chunk.len() as u64,
            maximum: Some(chunk.len() as u64),
        });
    }
    Ok(guards)
}

#[cfg(test)]
mod fingerprint_tests {
    use super::{bytes_fingerprint, operation_fingerprint};

    #[test]
    fn nested_object_order_does_not_change_contract_checksum() {
        let mut nested = serde_json::Map::new();
        nested.insert("c".to_owned(), serde_json::json!(3));
        nested.insert("b".to_owned(), serde_json::json!(2));
        let mut object = serde_json::Map::new();
        object.insert("z".to_owned(), serde_json::Value::Object(nested));
        object.insert("a".to_owned(), serde_json::json!(1));
        let checksum = operation_fingerprint(&serde_json::Value::Object(object))
            .expect("bounded contract checksum");
        assert_eq!(checksum, bytes_fingerprint(br#"{"a":1,"z":{"b":2,"c":3}}"#));
    }
}

#[cfg(test)]
mod generated_value_tests {
    use super::*;
    use serde_json::json;

    /// Rejects native-storage NUL incompatibilities before any adapter receives JSON.
    #[test]
    fn json_nul_is_rejected_at_the_shared_boundary() {
        for value in [json!("a\0b"), json!({"a\0b": 1}), json!([{"nested":"\0"}])] {
            assert_eq!(
                validate_json_value(&value),
                Err(DatabaseError::InvalidInput)
            );
        }
        assert_eq!(validate_json_value(&json!([0, 1, 255])), Ok(()));
    }

    /// Rejects a page before retaining another record beyond the encoded result budget.
    #[test]
    fn query_budget_rejects_the_first_overflowing_record() {
        let mut budget = QueryResultBudget {
            bytes: 8 * 1024 * 1024 - 2,
        };
        assert_eq!(budget.accept(&Map::new()), Ok(()));
        assert_eq!(
            budget.accept(&Map::new()),
            Err(DatabaseError::CapacityExceeded)
        );
    }

    /// Numeric aliases normalize recursively without coercing strings or reordering arrays.
    #[test]
    fn json_numeric_identity_is_backend_neutral() {
        assert_eq!(
            canonical_json_value(&json!({"n": [1.0, -0.0, 1.25], "text": "1.0", "wide": u64::MAX})),
            Ok(json!({"n": [1, 0, 1.25], "text": "1.0", "wide": u64::MAX}))
        );
        let mut nested = Value::Null;
        for _ in 0..34 {
            nested = Value::Array(vec![nested]);
        }
        assert_eq!(
            canonical_json_value(&nested),
            Err(DatabaseError::CapacityExceeded)
        );
        assert_eq!(
            serde_json::to_string(
                &canonical_json_value(&json!({"z": {"b": 1, "a": 2}, "a": 0})).unwrap()
            )
            .unwrap(),
            r#"{"a":0,"z":{"a":2,"b":1}}"#
        );
    }

    /// Generated references must obey final field constraints before binding.
    #[test]
    fn generated_text_validates_the_resolved_value() {
        let field = Field {
            name: "resource_id".to_owned(),
            kind: FieldType::Text,
            nullable: false,
            max_length: Some(8),
            default: None,
            references: None,
            on_delete: None,
            constraints: Vec::new(),
        };
        let expression = WriteValue::GeneratedText {
            index: 0,
            prefix: "metric:".to_owned(),
        };
        assert_eq!(
            resolve_field_write(&field, &expression, &[Some(1)]),
            Ok(json!("metric:1"))
        );
        assert_eq!(
            resolve_field_write(&field, &expression, &[Some(100)]),
            Err(DatabaseError::InvalidInput)
        );
        assert_eq!(
            resolve_field_write(&field, &expression, &[None]),
            Err(DatabaseError::InvalidInput)
        );
    }

    /// Resource replacement does not rewrite unrelated user text or nested scalars.
    #[test]
    fn generated_json_changes_only_resource_keys() {
        let expression = WriteValue::GeneratedJson {
            index: 0,
            marker: "metric:0".to_owned(),
            template: json!({"resource_id": "metric:0", "note": "metric:0", "nested": ["metric:0", {"resource_id": "metric:0"}]}),
        };
        assert_eq!(
            resolve_write_value(&expression, &[Some(42)]),
            Ok(json!({
                "resource_id": "metric:42", "note": "metric:0", "nested": ["metric:0", {"resource_id": "metric:42"}]
            }))
        );
    }
}

#[cfg(test)]
mod timestamp_storage_tests {
    use super::canonical_stored_timestamp;

    /// Nanoseconds survive storage normalization and fractional expiry rounds forward.
    #[test]
    fn timestamp_storage_preserves_fractional_instants() {
        assert_eq!(
            super::stored_session_expiry_epoch("2033-05-18T03:33:20.000000001Z").unwrap(),
            2_000_000_001
        );
        assert_eq!(
            super::stored_session_expiry_epoch("2033-05-18T03:33:20Z").unwrap(),
            2_000_000_000
        );
        assert_eq!(
            canonical_stored_timestamp("2026-10-06 01:02:03.123456").unwrap(),
            "2026-10-06T01:02:03.123456000Z"
        );
        assert_eq!(
            canonical_stored_timestamp("2026-10-06T10:02:03.123456+09:00").unwrap(),
            "2026-10-06T01:02:03.123456000Z"
        );
        assert_eq!(
            canonical_stored_timestamp("2026-10-06T01:02:03Z").unwrap(),
            "2026-10-06T01:02:03.000000000Z"
        );
        assert_eq!(
            canonical_stored_timestamp("1").unwrap(),
            "1970-01-01T00:00:01.000000000Z"
        );
        for invalid in ["", "2026-13-01 00:00:00", "2030-01-01"] {
            assert!(canonical_stored_timestamp(invalid).is_err());
        }
    }
}

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

    /// Supplies the smallest declared entity for portable namespace checks.
    fn simple_schema() -> AppSchema {
        AppSchema {
            app_id: "namespace_test".to_owned(),
            entities: vec![Entity {
                name: "record".to_owned(),
                identity: Identity::AutoInteger("id".to_owned()),
                fields: vec![Field {
                    name: "id".to_owned(),
                    kind: FieldType::Integer,
                    nullable: false,
                    max_length: None,
                    default: None,
                    references: None,
                    on_delete: None,
                    constraints: Vec::new(),
                }],
                unique: Vec::new(),
                indexes: Vec::new(),
                conditional_indexes: Vec::new(),
            }],
        }
    }

    /// Allows complete source indexes atomically while rejecting oversized transactions.
    #[test]
    fn transaction_step_limit_supports_complete_app_indexes_and_remains_bounded() {
        let schema = simple_schema();
        let step = Mutation::RequireCount {
            entity: "record".to_owned(),
            filter: Predicate::Equal("id".to_owned(), serde_json::json!(1)),
            minimum: 0,
            maximum: None,
        };
        let mut transaction = Transaction {
            steps: vec![step.clone(); MAX_TRANSACTION_STEPS],
            idempotency_key: None,
        };
        assert_eq!(validate_transaction(&schema, &transaction), Ok(()));
        transaction.steps.push(step);
        assert_eq!(
            validate_transaction(&schema, &transaction),
            Err(DatabaseError::InvalidInput)
        );
        transaction.steps = vec![
            Mutation::RequireCount {
                entity: "record".to_owned(),
                filter: Predicate::Equal("id".to_owned(), serde_json::json!("a".repeat(64 * 1024))),
                minimum: 0,
                maximum: None,
            };
            128
        ];
        assert!(serde_json::to_vec(&transaction).unwrap().len() > 8 * 1024 * 1024);
        assert_eq!(
            validate_transaction(&schema, &transaction),
            Err(DatabaseError::InvalidInput)
        );
    }

    /// Rejects declarations that alias another physical name on a primary adapter.
    #[test]
    fn ambiguous_names_fail_before_adapter_dispatch() {
        let mut schema = simple_schema();
        let mut duplicate = schema.entities[0].clone();
        duplicate.name = "Record".to_owned();
        schema.entities.push(duplicate);
        assert_eq!(
            validate_schema(&schema),
            Err(DatabaseError::InvalidContract)
        );

        let mut schema = simple_schema();
        let mut duplicate = schema.entities[0].fields[0].clone();
        duplicate.name = "ID".to_owned();
        schema.entities[0].fields.push(duplicate);
        assert_eq!(
            validate_schema(&schema),
            Err(DatabaseError::InvalidContract)
        );

        for name in ["record", "PRIMARY", "_id_", "ix_record_id"] {
            let mut schema = simple_schema();
            schema.entities[0].indexes.push(vec!["id".to_owned()]);
            schema.entities[0].conditional_indexes.push(Index {
                name: name.to_owned(),
                keys: vec![IndexKey::Field("id".to_owned())],
                unique: false,
                filter: None,
            });
            assert_eq!(
                validate_schema(&schema),
                Err(DatabaseError::InvalidContract),
                "{name}"
            );
        }

        let mut schema = simple_schema();
        schema.entities[0].indexes = vec![vec!["id".to_owned()], vec!["id".to_owned()]];
        assert_eq!(
            validate_schema(&schema),
            Err(DatabaseError::InvalidContract)
        );

        let mut schema = simple_schema();
        schema.entities[0].conditional_indexes = vec![Index {
            name: "record_custom".to_owned(),
            keys: vec![IndexKey::Field("id".to_owned())],
            unique: false,
            filter: None,
        }];
        let mut other = schema.entities[0].clone();
        other.name = "other".to_owned();
        other.conditional_indexes[0].name = "Record_Custom".to_owned();
        schema.entities.push(other);
        assert_eq!(
            validate_schema(&schema),
            Err(DatabaseError::InvalidContract)
        );
    }

    /// Keeps short historical names unchanged and bounds long foreign-key names.
    #[test]
    fn foreign_key_names_are_portable_and_stable() {
        assert_eq!(
            generated_index_name("fk", "child", &["parent_id".to_owned()]).unwrap(),
            "fk_child_parent_id"
        );
        let table = "a".repeat(63);
        let key = "b".repeat(63);
        let name = generated_index_name("fk", &table, std::slice::from_ref(&key)).unwrap();
        assert_eq!(name.len(), 63);
        assert!(valid_identifier(&name));
        assert_eq!(
            generated_index_name("fk", &table, std::slice::from_ref(&key)).unwrap(),
            name
        );
        assert_ne!(
            generated_index_name("fk", &table, &["c".repeat(63)]).unwrap(),
            name
        );
    }

    /// App schemas cannot overwrite metadata or document identities on any backend.
    #[test]
    fn internal_storage_identifiers_are_reserved() {
        let schema = AppSchema {
            app_id: "example".to_owned(),
            entities: vec![Entity {
                name: "record".to_owned(),
                identity: Identity::AutoInteger("id".to_owned()),
                fields: vec![Field {
                    name: "id".to_owned(),
                    kind: FieldType::Integer,
                    nullable: false,
                    max_length: None,
                    default: None,
                    references: None,
                    on_delete: None,
                    constraints: Vec::new(),
                }],
                unique: Vec::new(),
                indexes: Vec::new(),
                conditional_indexes: Vec::new(),
            }],
        };
        validate_schema(&schema).unwrap();
        for name in [
            "pharos_migration_ledger",
            "PHAROS_transaction_ledger",
            "_pharos_lower_name",
            "_id",
            "_ID",
        ] {
            let mut entity_collision = schema.clone();
            entity_collision.entities[0].name = name.to_owned();
            assert_eq!(
                validate_schema(&entity_collision),
                Err(DatabaseError::InvalidContract)
            );
            let mut field_collision = schema.clone();
            field_collision.entities[0].fields[0].name = name.to_owned();
            field_collision.entities[0].identity = Identity::AutoInteger(name.to_owned());
            assert_eq!(
                validate_schema(&field_collision),
                Err(DatabaseError::InvalidContract)
            );
        }
        for name in ["user_account", "business", "_custom_value"] {
            let mut ordinary = schema.clone();
            ordinary.entities[0].name = name.to_owned();
            validate_schema(&ordinary).unwrap();
        }
    }
}

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

    /// Native text and timestamp limitations never become backend-dependent errors.
    #[test]
    fn security_keys_and_session_clocks_use_portable_input_bounds() {
        assert_eq!(
            validate_session_lookup("bad\0token", 0),
            Err(DatabaseError::InvalidInput)
        );
        assert_eq!(
            validate_session_lookup("hash", 253_402_300_800),
            Err(DatabaseError::InvalidInput)
        );
        assert_eq!(
            validate_session_lookup("hash", 253_402_300_799),
            Ok(253_402_300_799)
        );
        assert!(matches!(
            validate_session_issue(1, "bad\0token", "2030-01-01T00:00:00Z"),
            Err(DatabaseError::InvalidInput)
        ));
        assert_eq!(
            validate_rate_limit_request("bad\0policy", "bucket", 1, 1),
            Err(DatabaseError::InvalidInput)
        );
        assert_eq!(
            validate_rate_limit_request("policy", "bad\0bucket", 1, 1),
            Err(DatabaseError::InvalidInput)
        );
        assert_eq!(
            validate_rate_limit_event("policy", "bucket", "bad\0outcome"),
            Err(DatabaseError::InvalidInput)
        );
        assert_eq!(
            decode_aggregate_text(FieldType::Binary, "abc"),
            Err(DatabaseError::StorageFailure)
        );
        assert_eq!(
            decode_aggregate_text(FieldType::Binary, "0080ff"),
            Ok(serde_json::json!([0, 128, 255]))
        );
    }
}