cipherstash-client 0.41.1

//! Types for representing EQL payloads, and encryption/decryption functions.
//!
//! Two wire envelopes are produced here:
//! - **v2** ([`encrypt_eql`] → [`EqlCiphertext`]): matches the EQL v2.3 schema
//!   at `cipherstash/encrypt-query-language/docs/reference/schema/eql-payload-v2.3.schema.json`.
//! - **v3** ([`encrypt_eql_v3`] → [`EqlCiphertextV3`]): the `eql_v3` envelope —
//!   scalar payloads drop the `k` discriminator, bloom bit positions are signed
//!   `smallint[]`, and SteVec entries order by the CLLW-OPE `op` term only.
//!
//! The two share one encryption pipeline (see `encrypt_eql_with`) and differ
//! only in the wire types they assemble.

#![cfg(feature = "tokio")]

mod formats;

use std::borrow::Cow;
use std::sync::Arc;

use crate::{
    encryption::{
        self, Encrypted, EncryptedEntry, EncryptedSteVecTerm, EncryptedSteVecTermCompat,
        EncryptedSteVecTermStandard, IndexTerm, QueryOp, SteQueryVec,
    },
    zerokms::{self, RecordDecryptError},
};
use stack_auth::AuthStrategyBounds;

use crate::{
    encryption::StorageBuilder,
    zerokms::{GenerateKeyPayload, IndexKey},
};

use super::zerokms::EncryptedRecord;
use cipherstash_config::{column::IndexType, ColumnConfig};
use serde::{Deserialize, Serialize};
use thiserror::Error;

use uuid::Uuid;

use crate::encryption::{PlaintextTarget, Queryable, ScopedCipher};

use cipherstash_config::column::Index;

use zerokms_protocol::{Context, DecryptionPolicy, UnverifiedContext};

/// The current version of the EQL schema format.
///
/// Encoded as `v` on every storage and query payload; the v2.x format hard-codes
/// this to `2`.
pub const EQL_SCHEMA_VERSION: u16 = 2;

/// The EQL **v3** schema version — encoded as `v` on every payload produced by
/// [`encrypt_eql_v3`].
///
/// v3 is a distinct wire envelope from v2 (see [`EqlCiphertextV3`]): scalar
/// storage payloads drop the `k` discriminator, bloom-filter bit positions are
/// carried as signed `smallint[]`, and SteVec entries order by the CLLW-OPE
/// `op` term only (CLLW-ORE `oc` has no v3 representation). The v2 constant and
/// wire types above are left untouched; v3 is an additive, parallel surface.
pub const EQL_SCHEMA_VERSION_V3: u16 = 3;

/// Encrypts multiple plaintexts into EQL format.
///
/// This is the main encryption entry point for the EQL system. It takes prepared plaintext values
/// and produces EQL payloads suitable for database storage or for use as query inputs. Each
/// `PreparedPlaintext` is processed independently, preserving input order in the result.
///
/// # Arguments
///
/// * `cipher` - The scoped cipher for performing cryptographic operations
/// * `plaintexts` - A vector of prepared plaintext values to encrypt
/// * `opts` - Encryption options including keyset ID, lock context, and service token
///
/// # Returns
///
/// A vector of [`EqlOutput`] values, one per input plaintext, in the same order:
/// - [`EqlOperation::Store`] inputs yield [`EqlOutput::Store`] carrying an [`EqlCiphertext`]
///   storage payload (one of [`EqlCiphertext::Encrypted`] for scalars or
///   [`EqlCiphertext::SteVec`] for structured values).
/// - [`EqlOperation::Query`] inputs yield [`EqlOutput::Query`] carrying an [`EqlQueryPayload`]
///   (no `c` — query payloads are matched against stored ciphertexts in the database).
///
/// # Errors
///
/// Returns `EqlError` if:
/// - Data key generation fails
/// - Encryption of any plaintext fails
/// - Index generation fails
/// - The ZeroKMS service is unavailable
/// - A query-mode plaintext produces an [`IndexTerm`] with no v2.3 representation
///
/// # Examples
///
/// ```no_run
/// # use std::sync::Arc;
/// # use cipherstash_client::eql::{encrypt_eql, PreparedPlaintext, EqlEncryptOpts};
/// # use cipherstash_client::encryption::ScopedCipher;
/// # use cipherstash_client::AutoStrategy;
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// # let cipher: Arc<ScopedCipher<AutoStrategy>> = unimplemented!();
/// # let plaintexts: Vec<PreparedPlaintext> = vec![];
/// let opts = EqlEncryptOpts::default();
/// let outputs = encrypt_eql(cipher, plaintexts, &opts).await?;
/// # Ok(())
/// # }
/// # fn main() {}
/// ```
pub async fn encrypt_eql<'a, C>(
    cipher: Arc<ScopedCipher<C>>,
    plaintexts: Vec<PreparedPlaintext<'a>>,
    opts: &EqlEncryptOpts<'a>,
) -> Result<Vec<EqlOutput>, EqlError>
where
    C: AuthStrategyBounds,
    for<'b> &'b C: stack_auth::AuthStrategy,
{
    encrypt_eql_with(
        cipher,
        plaintexts,
        opts,
        |encrypted, identifier| Ok(EqlOutput::Store(to_eql_ciphertext(encrypted, identifier)?)),
        |index_term, identifier| {
            Ok(EqlOutput::Query(to_eql_query_payload(
                index_term, identifier,
            )?))
        },
    )
    .await
}

/// Encrypts multiple plaintexts into **EQL v3** payloads.
///
/// The v3 counterpart of [`encrypt_eql`]: identical inputs and the same
/// underlying encryption pipeline (data-key generation, indexing, ZeroKMS round
/// trip), but assembles the [v3 wire envelope](EqlCiphertextV3) instead of v2.
/// Callers opt into v3 purely by calling this function — the v2 path and its
/// types are left untouched.
///
/// Each input yields one [`EqlOutputV3`], in the same order:
/// - [`EqlOperation::Store`] → [`EqlOutputV3::Store`] ([`EqlCiphertextV3`]): a
///   bare scalar `{v: 3, i, c, <terms>}` (no `k`; `bf` a signed `smallint[]`)
///   or a SteVec document `{v: 3, k: "sv", i, sv}`.
/// - [`EqlOperation::Query`] → [`EqlOutputV3::Query`] ([`EqlQueryPayloadV3`]): a
///   scalar operand `{v: 3, i, <term>}` or a jsonb containment needle
///   `{sv: [{s, hm|op}]}`.
///
/// # Errors
///
/// As [`encrypt_eql`], plus:
/// - [`EqlError::UnsupportedSteVecOreInV3`] if a SteVec column is configured in
///   Standard (CLLW-ORE) mode — v3 has no `oc` representation.
/// - [`EqlError::UnsupportedV3QueryTerm`] for a query term with no v3
///   query-operand shape.
pub async fn encrypt_eql_v3<'a, C>(
    cipher: Arc<ScopedCipher<C>>,
    plaintexts: Vec<PreparedPlaintext<'a>>,
    opts: &EqlEncryptOpts<'a>,
) -> Result<Vec<EqlOutputV3>, EqlError>
where
    C: AuthStrategyBounds,
    for<'b> &'b C: stack_auth::AuthStrategy,
{
    encrypt_eql_with(
        cipher,
        plaintexts,
        opts,
        |encrypted, identifier| {
            Ok(EqlOutputV3::Store(to_eql_ciphertext_v3(
                encrypted, identifier,
            )?))
        },
        |index_term, identifier| {
            Ok(EqlOutputV3::Query(to_eql_query_payload_v3(
                index_term, identifier,
            )?))
        },
    )
    .await
}

/// The shared encryption pipeline behind [`encrypt_eql`] / [`encrypt_eql_v3`].
///
/// Runs the full pipeline — target resolution, data-key generation, per-value
/// encryption / query indexing — and defers only the final payload assembly to
/// the caller-supplied closures, so the v2 and v3 entry points share one code
/// path and differ solely in the wire types they produce.
///
/// - `store_fn` turns a storage [`Encrypted`] result + its [`Identifier`] into
///   the output type (v2 [`EqlCiphertext`] or v3 [`EqlCiphertextV3`]).
/// - `query_fn` turns a query-mode [`IndexTerm`] + its [`Identifier`] into the
///   output type.
async fn encrypt_eql_with<'a, C, O>(
    cipher: Arc<ScopedCipher<C>>,
    plaintexts: Vec<PreparedPlaintext<'a>>,
    opts: &EqlEncryptOpts<'a>,
    store_fn: impl Fn(Encrypted, &Identifier) -> Result<O, EqlError>,
    query_fn: impl Fn(IndexTerm, Identifier) -> Result<O, EqlError>,
) -> Result<Vec<O>, EqlError>
where
    C: AuthStrategyBounds,
    for<'b> &'b C: stack_auth::AuthStrategy,
{
    use std::collections::VecDeque;

    let effective_keyset_id = opts.keyset_id.unwrap_or(cipher.keyset_id());

    let targets: Vec<EncryptionTarget> =
        to_encryption_targets(cipher.index_key(), plaintexts, effective_keyset_id)?;

    let mut data_keys = VecDeque::from(
        cipher
            .generate_data_keys(
                generate_data_key_payloads(opts, &targets),
                opts.unverified_context.clone(),
            )
            .await?,
    );

    targets
        .into_iter()
        .map(|target| -> Result<O, EqlError> {
            match target {
                EncryptionTarget::ForStorage(identifier, builder) => {
                    let encrypted = builder.build_for_encryption().encrypt(
                        // PANIC SAFETY: a successful result from
                        // `generate_data_keys` returns exactly the requested
                        // number of keys, so `remove(0)` cannot fail.
                        data_keys
                            .remove(0)
                            .expect("insufficient data keys to encrypt all plaintexts"),
                    )?;

                    store_fn(encrypted, &identifier)
                }
                EncryptionTarget::ForQuery(identifier, plaintext, index_type, query_op) => {
                    let index = Index::new(index_type.clone());
                    let index_term =
                        (index, plaintext).build_queryable(cipher.clone(), query_op)?;

                    query_fn(index_term, identifier)
                }
            }
        })
        .collect()
}

/// Decrypts multiple EQL storage payloads back to plaintext.
///
/// This is the main decryption entry point for the EQL system. It takes encrypted EQL payloads
/// (as retrieved from the database) and decrypts them using ZeroKMS, returning the original
/// plaintext values. Accepts both [`EqlCiphertext::Encrypted`] (root scalar) and
/// [`EqlCiphertext::SteVec`] (structured) payloads — for the SteVec variant the root entry's
/// ciphertext (`sv[0].c`) is used.
///
/// # Arguments
///
/// * `cipher` - The scoped cipher for performing cryptographic operations
/// * `ciphertexts` - An iterator of encrypted EQL storage payloads to decrypt
/// * `opts` - Decryption options including keyset ID, lock context, and service token
///
/// # Returns
///
/// A vector of decrypted [`encryption::Plaintext`] values, one for each input ciphertext, in the
/// same order.
///
/// # Errors
///
/// Returns `EqlError` if:
/// - An [`EqlCiphertext::SteVec`] payload has an empty `sv` array (no root entry)
/// - Decryption fails (e.g., wrong keyset, tampered data)
/// - The ZeroKMS service is unavailable
/// - The decrypted data cannot be parsed as plaintext
///
/// # Examples
///
/// ```no_run
/// # use std::sync::Arc;
/// # use cipherstash_client::eql::{decrypt_eql, EqlCiphertext, EqlDecryptOpts};
/// # use cipherstash_client::encryption::ScopedCipher;
/// # use cipherstash_client::AutoStrategy;
/// # async fn example() -> Result<(), Box<dyn std::error::Error>> {
/// # let cipher: Arc<ScopedCipher<AutoStrategy>> = unimplemented!();
/// # let ciphertexts: Vec<EqlCiphertext> = vec![];
/// let opts = EqlDecryptOpts::default();
/// let plaintexts = decrypt_eql(cipher, ciphertexts, &opts).await?;
/// # Ok(())
/// # }
/// # fn main() {}
/// ```
pub async fn decrypt_eql<'a, C>(
    cipher: Arc<ScopedCipher<C>>,
    ciphertexts: impl IntoIterator<Item = EqlCiphertext>,
    opts: &EqlDecryptOpts<'a>,
) -> Result<Vec<encryption::Plaintext>, EqlError>
where
    C: AuthStrategyBounds,
    for<'b> &'b C: stack_auth::AuthStrategy,
{
    use crate::{encryption::DecryptOptions, zerokms::WithContext};

    let decrypt_opts = DecryptOptions {
        keyset_id: opts.keyset_id,
        unverified_context: opts.unverified_context.clone(),
    };

    let ciphertexts = ciphertexts
        .into_iter()
        .map(|eql| {
            let (_, ciphertext) = extract_root_ciphertext(eql)?;
            Ok(WithContext {
                record: ciphertext,
                context: opts.lock_context.clone(),
            })
        })
        .collect::<Result<Vec<_>, EqlError>>()?;

    Ok(cipher
        .decrypt(ciphertexts, &decrypt_opts)
        .await
        .map_err(|err| convert_zerokms_error(err, cipher.keyset_id(), opts.keyset_id))?
        .into_iter()
        .map(|decrypted| encryption::Plaintext::from_slice(&decrypted))
        .collect::<Result<Vec<_>, _>>()?)
}

/// Like [`decrypt_eql`] but decryption failure of one or more ciphertexts does not fail the entire operation.
/// Instead, a decryption result (success or failure) is returned for every ciphertext.
///
/// # Per-Item Errors
///
/// The following conditions produce per-item `Err` results without aborting:
/// - Empty STE vector on a [`EqlCiphertext::SteVec`] payload: Returns `Err(EqlError::MissingCiphertext)`.
/// - Decryption failure: Returns `Err` with the underlying decryption error.
///
/// # Operation-Level Errors
///
/// Only a failure to retrieve the access token will result in the entire operation failing.
pub async fn decrypt_eql_fallible<'a, C>(
    cipher: Arc<ScopedCipher<C>>,
    ciphertexts: impl IntoIterator<Item = EqlCiphertext>,
    opts: &EqlDecryptOpts<'a>,
) -> Result<Vec<Result<encryption::Plaintext, EqlError>>, EqlError>
where
    C: AuthStrategyBounds,
    for<'b> &'b C: stack_auth::AuthStrategy,
{
    use crate::{encryption::DecryptOptions, zerokms::WithContext};

    let decrypt_opts = DecryptOptions {
        keyset_id: opts.keyset_id,
        unverified_context: opts.unverified_context.clone(),
    };

    let inputs: Vec<_> = ciphertexts.into_iter().collect();
    let input_count = inputs.len();
    let mut results: Vec<Option<Result<encryption::Plaintext, EqlError>>> =
        (0..input_count).map(|_| None).collect();
    let mut valid_payloads: Vec<(usize, WithContext<EncryptedRecord>)> =
        Vec::with_capacity(input_count);

    for (index, eql) in inputs.into_iter().enumerate() {
        match extract_root_ciphertext(eql) {
            Ok((_, ciphertext)) => valid_payloads.push((
                index,
                WithContext {
                    record: ciphertext,
                    context: opts.lock_context.clone(),
                },
            )),
            Err(err) => {
                results[index] = Some(Err(err));
            }
        }
    }

    let (indices, payloads): (Vec<usize>, Vec<_>) = valid_payloads.into_iter().unzip();

    let decrypt_results = cipher
        .decrypt_fallible(payloads, &decrypt_opts)
        .await
        .map_err(|err| convert_zerokms_error(err, cipher.keyset_id(), opts.keyset_id))?;

    for (index, decrypt_result) in indices.into_iter().zip(decrypt_results) {
        results[index] = Some(match decrypt_result {
            Ok(bytes) => encryption::Plaintext::from_slice(&bytes).map_err(Into::into),
            Err(err) => Err(EqlError::from(err)),
        });
    }

    Ok(results
        .into_iter()
        .map(|r| r.expect("all result slots filled"))
        .collect())
}

/// Identifies a specific database table and column pair.
///
/// Serialized as `{"t": <table>, "c": <column>}` per the v2.3 schema `Ident`
/// definition. Implements `Hash` and `Eq` to allow use as a map key.
#[derive(Clone, Debug, Deserialize, Eq, Hash, PartialEq, Serialize)]
pub struct Identifier {
    /// The database table name (serialized as `"t"`).
    #[serde(rename = "t")]
    pub table: String,

    /// The column name within the table (serialized as `"c"`).
    #[serde(rename = "c")]
    pub column: String,
}

impl Identifier {
    /// Creates a new `Identifier` from table and column names.
    ///
    /// # Arguments
    ///
    /// * `table` - The table name (any type convertible to `String`)
    /// * `column` - The column name (any type convertible to `String`)
    ///
    /// # Examples
    ///
    /// ```
    /// use cipherstash_client::eql::Identifier;
    ///
    /// let id = Identifier::new("users", "email");
    /// ```
    pub fn new(table: impl Into<String>, column: impl Into<String>) -> Self {
        Self {
            table: table.into(),
            column: column.into(),
        }
    }

    /// Returns a reference to the table name.
    pub fn table(&self) -> &str {
        &self.table
    }

    /// Returns a reference to the column name.
    pub fn column(&self) -> &str {
        &self.column
    }
}

/// EQL v2.3 storage payload. One of two mutually exclusive root shapes
/// discriminated by `k`.
///
/// The `Encrypted` variant is larger than `SteVec` (it inlines the scalar
/// `EncryptedRecord` plus the root index terms), so clippy's
/// `large_enum_variant` fires. Boxing a variant would change this published
/// SDK type's public shape and ripple into downstream consumers (the proxy)
/// for no real benefit: these payloads are produced once per value and moved,
/// not stored in large collections. Keeping the enum unboxed keeps the public
/// API stable.
#[allow(clippy::large_enum_variant)]
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(tag = "k")]
pub enum EqlCiphertext {
    /// Scalar ciphertext payload (`k = "ct"`).
    #[serde(rename = "ct")]
    Encrypted(EncryptedPayload),
    /// STE-vector payload (`k = "sv"`) for jsonb / structured queries.
    #[serde(rename = "sv")]
    SteVec(SteVecPayload),
}

impl EqlCiphertext {
    /// Returns the table/column [`Identifier`] this payload belongs to.
    pub fn identifier(&self) -> &Identifier {
        match self {
            EqlCiphertext::Encrypted(p) => &p.identifier,
            EqlCiphertext::SteVec(p) => &p.identifier,
        }
    }

    /// Returns the EQL schema version recorded on this payload.
    ///
    /// Always [`EQL_SCHEMA_VERSION`] for v2.x payloads; preserved on the type
    /// to allow downstream consumers to detect version mismatches.
    pub fn version(&self) -> u16 {
        match self {
            EqlCiphertext::Encrypted(p) => p.version,
            EqlCiphertext::SteVec(p) => p.version,
        }
    }
}

/// Scalar EQL storage payload (`k = "ct"`).
///
/// Carries the encrypted record (`c`) plus any configured root-scope index
/// terms (`hm`, `bf`, `ob`, `op`). Per the v2.3 schema, `c` is required.
///
/// Two order-preserving families can live at this scope:
/// - Block ORE (`ob`) — emitted by the `ore` index; an array of hex-encoded
///   ORE blocks.
/// - CLLW OPE (`op`) — emitted by the `ope` index; a single hex-encoded
///   order-preserving ciphertext (the same `op` key used at sv-element scope,
///   see [`EncryptedSteVecTermCompat::Ope`]). Both numeric (`OpeFixed`) and
///   string (`OpeVariable`) plaintexts use this one key; their ciphertexts
///   sort into disjoint ranges via plaintext-level domain separation.
///
/// CLLW ORE (`oc`) is distinct from both and lives only on STE-vec elements.
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct EncryptedPayload {
    /// Payload version — always [`EQL_SCHEMA_VERSION`].
    #[serde(rename = "v")]
    pub version: u16,

    /// Table and column this payload belongs to.
    #[serde(rename = "i")]
    pub identifier: Identifier,

    /// Opaque encrypted record (MessagePack Base85). Required.
    #[serde(rename = "c", with = "formats::mp_base85")]
    pub ciphertext: EncryptedRecord,

    /// HMAC-SHA-256 hex string for exact-match equality (`hm`).
    #[serde(rename = "hm", default, skip_serializing_if = "Option::is_none")]
    pub hmac_256: Option<String>,

    /// Bloom filter (set bit positions) for `LIKE` / `ILIKE` (`bf`).
    #[serde(rename = "bf", default, skip_serializing_if = "Option::is_none")]
    pub bloom_filter: Option<Vec<u16>>,

    /// Block ORE u64_8_256 term for ordered comparisons (`ob`).
    ///
    /// Stored as an array of hex-encoded ORE blocks. Emitted by the `ore`
    /// index. Mutually exclusive with [`SteVecEntryTerm::OreCllw`] (`oc`),
    /// which lives only on STE-vec elements.
    #[serde(rename = "ob", default, skip_serializing_if = "Option::is_none")]
    pub ore_block_u64_8_256: Option<Vec<String>>,

    /// CLLW OPE ciphertext (hex) for ordered comparisons (`op`).
    ///
    /// Emitted by the `ope` index for scalar columns. A single hex string —
    /// the order-preserving ciphertext is directly bytewise-comparable. The
    /// same `op` key is used at sv-element scope.
    #[serde(rename = "op", default, skip_serializing_if = "Option::is_none")]
    pub ope_cllw: Option<String>,
}

/// STE-vector EQL storage payload (`k = "sv"`).
///
/// Mutually exclusive with [`EncryptedPayload`]: carries only the
/// discriminator metadata (`v`, `i`) and the per-selector entries in `sv`.
/// The root entry's ciphertext lives in `ste_vec[0].ciphertext`.
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct SteVecPayload {
    /// Payload version — always [`EQL_SCHEMA_VERSION`].
    #[serde(rename = "v")]
    pub version: u16,

    /// Table and column this payload belongs to.
    #[serde(rename = "i")]
    pub identifier: Identifier,

    /// Per-selector encrypted entries.
    #[serde(rename = "sv")]
    pub ste_vec: Vec<SteVecEntry>,
}

/// One entry inside an [`SteVecPayload`].
///
/// Every element carries `s` (selector), `c` (ciphertext), and exactly one
/// of [`SteVecEntryTerm::Hmac`], [`SteVecEntryTerm::OreCllw`], or
/// [`SteVecEntryTerm::Ope`]. The v2.3 schema file predates sv-level `op`
/// (it defines only `hm` / `oc` at element scope); the client emits it
/// ahead of the v3 envelope, the same precedent as the scalar-root `op`
/// field.
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct SteVecEntry {
    /// Hex-encoded tokenized selector — deterministic per (path, key).
    #[serde(rename = "s")]
    pub selector: String,

    /// Per-entry encrypted record (MessagePack Base85). Required.
    #[serde(rename = "c", with = "formats::mp_base85")]
    pub ciphertext: EncryptedRecord,

    /// Array marker — true when the selector points at a JSON array context.
    #[serde(rename = "a", default, skip_serializing_if = "Option::is_none")]
    pub is_array: Option<bool>,

    /// Per-element equality / ordering term. Exactly one of `hm`, `oc`, or
    /// `op`.
    #[serde(flatten)]
    pub term: SteVecEntryTerm,
}

/// Equality / ordering term carried by an [`SteVecEntry`].
///
/// `Hmac` (`hm`) covers boolean leaves and the placeholder entries for
/// array / object roots. String and number leaves carry the ordering term
/// selected by the column's `SteVecMode`: `Ope` (`op`, Compat mode — CLLW
/// OPE) or `OreCllw` (`oc`, Standard mode — CLLW ORE, the legacy v2
/// protocol). Domain separation between numeric and string ordering
/// ciphertexts is enforced on the plaintext bit-stream before CLLW
/// encryption — numeric and string ciphertexts therefore sort into disjoint
/// ranges, but the domain tag is not visible by hex-decoding the `oc` / `op`
/// ciphertext alone.
///
/// `#[non_exhaustive]`, matching the sibling [`EncryptedSteVecTerm`] family:
/// a future ordering primitive must not be a breaking change for downstream
/// consumers that match on this enum.
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(untagged)]
#[non_exhaustive]
pub enum SteVecEntryTerm {
    Hmac {
        #[serde(rename = "hm")]
        hmac_256: String,
    },
    OreCllw {
        #[serde(rename = "oc")]
        ore_cllw_8: String,
    },
    /// CLLW OPE ordering term (`op`) — a single hex-encoded order-preserving
    /// ciphertext, directly bytewise-comparable. The same `op` key (and hex
    /// encoding) the scalar root uses; the v2.3 schema file predates sv-level
    /// `op` and the client emits it ahead of the v3 envelope.
    Ope {
        #[serde(rename = "op")]
        ope_cllw: String,
    },
}

/// Output of [`encrypt_eql`] for a single [`PreparedPlaintext`].
///
/// Storage encryption produces an [`EqlOutput::Store`] carrying an
/// [`EqlCiphertext`]; query-mode encryption produces an [`EqlOutput::Query`]
/// carrying an [`EqlQueryPayload`].
///
/// Not `Clone` because [`EqlQueryPayload::SteVec`] can carry a
/// [`SteQueryVec`], which is intentionally non-`Clone` (see the FIXME on
/// `SteQueryVec` in the encryption crate). Consumers should pass these by
/// value or behind a shared pointer.
#[derive(Debug, Deserialize, Serialize)]
#[serde(untagged)]
pub enum EqlOutput {
    Store(EqlCiphertext),
    Query(EqlQueryPayload),
}

/// EQL query payload — partial encrypted payload used to match against stored
/// [`EqlCiphertext`] values.
///
/// Mirrors the storage payload's root shape (discriminated by `k`) but omits
/// the `c` ciphertext: queries do not encrypt for storage. Not described by
/// the v2.3 storage schema; this is the cipherstash-suite query-side shape.
#[derive(Debug, Deserialize, Serialize)]
#[serde(tag = "k")]
pub enum EqlQueryPayload {
    /// Root-scalar query payload (`k = "ct"`) — matches an
    /// [`EncryptedPayload`] on one of `hm`, `bf`, or `ob`.
    #[serde(rename = "ct")]
    Encrypted(EncryptedQueryPayload),
    /// STE-vector query payload (`k = "sv"`) — matches an [`SteVecPayload`]
    /// on a selector path (`s`) or per-element term (`hm` / `oc` / `op`).
    #[serde(rename = "sv")]
    SteVec(SteVecQueryPayload),
}

impl EqlQueryPayload {
    /// Returns the table/column [`Identifier`] this query payload targets.
    pub fn identifier(&self) -> &Identifier {
        match self {
            EqlQueryPayload::Encrypted(p) => &p.identifier,
            EqlQueryPayload::SteVec(p) => &p.identifier,
        }
    }
}

/// Root-scalar query payload — carries exactly one root-scope index term.
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct EncryptedQueryPayload {
    #[serde(rename = "v")]
    pub version: u16,

    #[serde(rename = "i")]
    pub identifier: Identifier,

    #[serde(flatten)]
    pub term: RootQueryTerm,
}

/// Per-root-scalar query term. Exactly one of `hm`, `bf`, `ob`, or `op`.
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(untagged)]
pub enum RootQueryTerm {
    Hmac {
        #[serde(rename = "hm")]
        hmac_256: String,
    },
    BloomFilter {
        #[serde(rename = "bf")]
        bloom_filter: Vec<u16>,
    },
    OreBlock {
        #[serde(rename = "ob")]
        ore_block_u64_8_256: Vec<String>,
    },
    /// CLLW OPE query term (`op`) — a single hex-encoded order-preserving
    /// ciphertext, matched bytewise against a stored scalar `op` term.
    Ope {
        #[serde(rename = "op")]
        ope_cllw: String,
    },
}

/// STE-vector query payload — carries either a selector lookup or a single
/// per-element term used to match against an [`SteVecEntry`].
#[derive(Debug, Deserialize, Serialize)]
pub struct SteVecQueryPayload {
    #[serde(rename = "v")]
    pub version: u16,

    #[serde(rename = "i")]
    pub identifier: Identifier,

    #[serde(flatten)]
    pub term: SteVecQueryTerm,
}

/// Per-STE-vector query term. Exactly one of `s`, `hm`, `oc`, `op`, or `q`.
///
/// `Containment` carries the full STE query vector emitted for JSON
/// containment (`@>`) queries (one `(selector, term)` per path in the query
/// JSON). Other variants carry a single per-element lookup term.
///
/// `#[non_exhaustive]` for the same reason as [`SteVecEntryTerm`].
#[derive(Debug, Deserialize, Serialize)]
#[serde(untagged)]
#[non_exhaustive]
pub enum SteVecQueryTerm {
    Selector {
        #[serde(rename = "s")]
        selector: String,
    },
    Hmac {
        #[serde(rename = "hm")]
        hmac_256: String,
    },
    OreCllw {
        #[serde(rename = "oc")]
        ore_cllw_8: String,
    },
    /// CLLW OPE query term (`op`) — a single hex-encoded order-preserving
    /// ciphertext, matched bytewise against a stored sv-element `op` term
    /// (Compat mode).
    Ope {
        #[serde(rename = "op")]
        ope_cllw: String,
    },
    /// Full STE query vector for JSON containment queries.
    ///
    /// Serialized under `q` to avoid collision with [`SteVecPayload::ste_vec`]
    /// (`sv`). The wire shape of [`SteQueryVec`] is opaque to this layer; the
    /// proxy / database decodes it via the encryption crate's serde impls.
    Containment {
        #[serde(rename = "q")]
        query_vec: SteQueryVec<16>,
    },
}

// ============================================================================
// EQL v3 wire types
//
// The v3 envelope diverges from v2 (the types above) in three ways: scalar
// storage payloads carry NO `k` discriminator; bloom-filter bit positions are
// signed (`smallint[]`); and SteVec entries order by the CLLW-OPE `op` term
// only — CLLW-ORE (`oc`) has no v3 representation. These types are produced by
// [`encrypt_eql_v3`]; the v2 surface above is left unchanged.
// ============================================================================

/// EQL v3 storage payload — the v3 analogue of [`EqlCiphertext`].
///
/// Unlike v2, the two root shapes are NOT both `k`-tagged: a scalar payload
/// carries no `k` at all, while a SteVec document keeps `k: "sv"`. The enum is
/// therefore `untagged` and distinguished structurally (a SteVec doc has an
/// `sv` array; a scalar has `c`).
#[allow(clippy::large_enum_variant)]
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(untagged)]
pub enum EqlCiphertextV3 {
    /// SteVec document payload (`{v, k: "sv", i, sv}`). Listed first so untagged
    /// deserialization matches the discriminated shape before the bare scalar.
    SteVec(SteVecPayloadV3),
    /// Scalar storage payload — a bare `{v, i, c, <terms>}` object, no `k`.
    Encrypted(EncryptedPayloadV3),
}

impl EqlCiphertextV3 {
    /// The table/column [`Identifier`] this payload belongs to.
    pub fn identifier(&self) -> &Identifier {
        match self {
            EqlCiphertextV3::Encrypted(p) => &p.identifier,
            EqlCiphertextV3::SteVec(p) => &p.identifier,
        }
    }

    /// The EQL schema version recorded on this payload (always
    /// [`EQL_SCHEMA_VERSION_V3`]).
    pub fn version(&self) -> u16 {
        match self {
            EqlCiphertextV3::Encrypted(p) => p.version,
            EqlCiphertextV3::SteVec(p) => p.version,
        }
    }

    /// Project this stored payload into its query operand by dropping the record
    /// ciphertext `c` (and, for SteVec, the envelope + per-entry `c`/`a`).
    ///
    /// The v3→v3 counterpart of casting a stored payload to its `query_<name>`
    /// operand: a query operand carries exactly the column's terms but never a
    /// decryptable ciphertext. This is how a **multi-term** scalar operand (e.g.
    /// `{v, i, hm, ob}` for `query_text_ord`) is built — a single
    /// `EqlOperation::Query` produces only one term, so callers encrypt in Store
    /// mode and project the result here. SteVec documents project to the bare
    /// jsonb containment needle `{sv: [{s, hm|op}]}`.
    pub fn into_query_operand(self) -> EqlQueryPayloadV3 {
        match self {
            EqlCiphertextV3::Encrypted(p) => {
                EqlQueryPayloadV3::Encrypted(EncryptedQueryPayloadV3 {
                    version: p.version,
                    identifier: p.identifier,
                    hmac_256: p.hmac_256,
                    bloom_filter: p.bloom_filter,
                    ore_block_u64_8_256: p.ore_block_u64_8_256,
                    ope_cllw: p.ope_cllw,
                })
            }
            EqlCiphertextV3::SteVec(doc) => EqlQueryPayloadV3::SteVec(SteVecQueryPayloadV3 {
                ste_vec: doc
                    .ste_vec
                    .into_iter()
                    .map(|e| SteVecQueryEntryV3 {
                        selector: e.selector,
                        term: e.term,
                    })
                    .collect(),
            }),
        }
    }
}

/// EQL v3 scalar storage payload — a flat `{v, i, c, hm?, ob?, op?, bf?}`
/// object with no `k` discriminator (the shape the `eql_v3` scalar domain
/// CHECKs accept).
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct EncryptedPayloadV3 {
    /// Payload version — always [`EQL_SCHEMA_VERSION_V3`].
    #[serde(rename = "v")]
    pub version: u16,

    /// Table and column this payload belongs to.
    #[serde(rename = "i")]
    pub identifier: Identifier,

    /// Opaque encrypted record (MessagePack Base85). Required.
    #[serde(rename = "c", with = "formats::mp_base85")]
    pub ciphertext: EncryptedRecord,

    /// HMAC-SHA-256 hex string for exact-match equality (`hm`).
    #[serde(rename = "hm", default, skip_serializing_if = "Option::is_none")]
    pub hmac_256: Option<String>,

    /// Bloom-filter bit positions as **signed** `smallint[]` (`bf`) — the v3
    /// representation. v2 emits unsigned `u16`; the upper half (`32768..=65535`)
    /// reinterprets two's-complement negative here.
    #[serde(rename = "bf", default, skip_serializing_if = "Option::is_none")]
    pub bloom_filter: Option<Vec<i16>>,

    /// Block-ORE term for ordered comparisons (`ob`), array of hex blocks.
    #[serde(rename = "ob", default, skip_serializing_if = "Option::is_none")]
    pub ore_block_u64_8_256: Option<Vec<String>>,

    /// CLLW-OPE ciphertext (hex) for ordered comparisons (`op`).
    #[serde(rename = "op", default, skip_serializing_if = "Option::is_none")]
    pub ope_cllw: Option<String>,
}

/// EQL v3 SteVec document payload — `{v, k: "sv", i, sv}`. The root `k`
/// discriminator IS kept (the v3 jsonb document models it).
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct SteVecPayloadV3 {
    /// Payload version — always [`EQL_SCHEMA_VERSION_V3`].
    #[serde(rename = "v")]
    pub version: u16,

    /// The `k: "sv"` form discriminator.
    #[serde(rename = "k")]
    pub kind: SteVecKind,

    /// Table and column this payload belongs to.
    #[serde(rename = "i")]
    pub identifier: Identifier,

    /// Per-selector encrypted entries. `sv[0]` is the decryption root.
    #[serde(rename = "sv")]
    pub ste_vec: Vec<SteVecEntryV3>,
}

/// The `k: "sv"` form discriminator carried by [`SteVecPayloadV3`]. A
/// single-variant enum so serialization emits exactly `"sv"` and
/// deserialization rejects anything else.
#[derive(Clone, Copy, Debug, Default, Deserialize, Serialize)]
pub enum SteVecKind {
    /// The only kind — serializes as `"sv"`.
    #[default]
    #[serde(rename = "sv")]
    SteVec,
}

/// One entry in a v3 SteVec document — selector `s`, per-entry ciphertext `c`,
/// optional array marker `a`, and exactly one of `hm`/`op`. Per-entry
/// `v`/`i`/`k` are NOT carried; the envelope lives only at the root.
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct SteVecEntryV3 {
    /// Hex-encoded tokenized selector.
    #[serde(rename = "s")]
    pub selector: String,

    /// Per-entry encrypted record (MessagePack Base85). Required.
    #[serde(rename = "c", with = "formats::mp_base85")]
    pub ciphertext: EncryptedRecord,

    /// Array marker — true when the selector points at a JSON array context.
    #[serde(rename = "a", default, skip_serializing_if = "Option::is_none")]
    pub is_array: Option<bool>,

    /// Per-entry equality / ordering term — exactly one of `hm`/`op`.
    #[serde(flatten)]
    pub term: SteVecEntryTermV3,
}

/// Equality / ordering term on a v3 SteVec entry — exactly one of `hm`
/// (equality) or `op` (CLLW-OPE ordering).
///
/// Unlike the v2 [`SteVecEntryTerm`], there is NO `oc` (CLLW-ORE) variant: v3
/// orders SteVec entries by the byte-comparable CLLW-OPE `op` term, so an
/// ORE-mode column has no v3 representation and is rejected at assembly time
/// with [`EqlError::UnsupportedSteVecOreInV3`].
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(untagged)]
#[non_exhaustive]
pub enum SteVecEntryTermV3 {
    Hmac {
        #[serde(rename = "hm")]
        hmac_256: String,
    },
    /// CLLW-OPE ordering term (`op`) — a single hex-encoded order-preserving
    /// ciphertext, directly bytewise-comparable.
    Ope {
        #[serde(rename = "op")]
        ope_cllw: String,
    },
}

/// Output of [`encrypt_eql_v3`] for a single [`PreparedPlaintext`] — the v3
/// analogue of [`EqlOutput`]. `Store` inputs yield an [`EqlCiphertextV3`];
/// `Query` inputs an [`EqlQueryPayloadV3`].
#[derive(Debug, Deserialize, Serialize)]
#[serde(untagged)]
pub enum EqlOutputV3 {
    Store(EqlCiphertextV3),
    Query(EqlQueryPayloadV3),
}

/// EQL v3 query operand — the query-side counterpart of [`EqlCiphertextV3`].
///
/// Two shapes: a scalar operand `{v, i, <term>}` (enveloped, no `k`/`c`) for
/// the `eql_v3.query_<name>` twins, or the bare jsonb containment needle
/// `{sv: [{s, hm|op}]}` for `eql_v3.query_jsonb` (a containment operand is
/// never enveloped).
#[derive(Debug, Deserialize, Serialize)]
#[serde(untagged)]
pub enum EqlQueryPayloadV3 {
    /// Bare jsonb containment needle `{sv: [{s, hm|op}]}`. Listed first: it is
    /// the only shape carrying `sv`, so untagged matching is unambiguous.
    SteVec(SteVecQueryPayloadV3),
    /// Scalar query operand `{v, i, hm?, bf?, ob?, op?}` (an object).
    Encrypted(EncryptedQueryPayloadV3),
    /// A bare tokenized-selector hash (hex) for a jsonb path query
    /// (`->`/`->>`). Unlike the other variants it is not an encrypted operand —
    /// v3 has no encrypted-selector envelope; the selector is used directly as
    /// `text`. Serializes as a bare JSON string, so untagged matching is
    /// unambiguous against the two object variants above.
    Selector(String),
}

/// v3 scalar query operand — `{v, i, hm?, bf?, ob?, op?}`: the stored
/// [`EncryptedPayloadV3`] minus the record ciphertext `c` (no `k`, no `c`).
///
/// A `query_<name>` operand must carry EVERY term its domain requires — e.g.
/// `hm` + `ob` for `query_text_ord` — so this is a **multi-term** shape, not a
/// single term. A single `EqlOperation::Query` produces only one term (see
/// [`to_eql_query_payload_v3`]); the full multi-term operand is produced by
/// projecting a Store-mode payload with [`EqlCiphertextV3::into_query_operand`].
#[derive(Clone, Debug, Deserialize, Serialize)]
pub struct EncryptedQueryPayloadV3 {
    #[serde(rename = "v")]
    pub version: u16,
    #[serde(rename = "i")]
    pub identifier: Identifier,
    #[serde(rename = "hm", default, skip_serializing_if = "Option::is_none")]
    pub hmac_256: Option<String>,
    #[serde(rename = "bf", default, skip_serializing_if = "Option::is_none")]
    pub bloom_filter: Option<Vec<i16>>,
    #[serde(rename = "ob", default, skip_serializing_if = "Option::is_none")]
    pub ore_block_u64_8_256: Option<Vec<String>>,
    #[serde(rename = "op", default, skip_serializing_if = "Option::is_none")]
    pub ope_cllw: Option<String>,
}

/// v3 jsonb containment needle — `{sv: [{s, hm|op}]}`. Each entry is a selector
/// plus exactly one term (mirrors `eql_v3.to_ste_vec_query`). No envelope.
#[derive(Debug, Deserialize, Serialize)]
pub struct SteVecQueryPayloadV3 {
    #[serde(rename = "sv")]
    pub ste_vec: Vec<SteVecQueryEntryV3>,
}

/// One entry of a v3 jsonb containment needle — selector `s` plus exactly one
/// of `hm`/`op`.
#[derive(Debug, Deserialize, Serialize)]
pub struct SteVecQueryEntryV3 {
    #[serde(rename = "s")]
    pub selector: String,
    #[serde(flatten)]
    pub term: SteVecEntryTermV3,
}

/// Lifts a stored per-element term onto the query term that matches it.
///
/// [`SteVecQueryTerm`] is a strict superset of [`SteVecEntryTerm`] — it adds
/// `Selector` and `Containment`, which have no stored-entry counterpart. Every
/// shared variant keeps its wire key, so a query term always matches the
/// entries the same mode produced.
impl From<SteVecEntryTerm> for SteVecQueryTerm {
    fn from(term: SteVecEntryTerm) -> Self {
        match term {
            SteVecEntryTerm::Hmac { hmac_256 } => Self::Hmac { hmac_256 },
            SteVecEntryTerm::OreCllw { ore_cllw_8 } => Self::OreCllw { ore_cllw_8 },
            SteVecEntryTerm::Ope { ope_cllw } => Self::Ope { ope_cllw },
        }
    }
}

/// Extracts the root ciphertext from an [`EqlCiphertext`] for decryption.
///
/// For [`EqlCiphertext::Encrypted`] this is the payload's `c` field directly.
/// For [`EqlCiphertext::SteVec`] this is the first entry's `c`, which holds
/// the root document ciphertext.
fn extract_root_ciphertext(eql: EqlCiphertext) -> Result<(Identifier, EncryptedRecord), EqlError> {
    match eql {
        EqlCiphertext::Encrypted(p) => Ok((p.identifier, p.ciphertext)),
        EqlCiphertext::SteVec(p) => {
            let SteVecPayload {
                identifier,
                ste_vec,
                ..
            } = p;
            let root = ste_vec
                .into_iter()
                .next()
                .ok_or_else(|| EqlError::MissingCiphertext(identifier.clone()))?;
            Ok((identifier, root.ciphertext))
        }
    }
}

/// Builds an [`EqlCiphertext`] from an [`Encrypted`] storage result.
fn to_eql_ciphertext(
    encrypted: Encrypted,
    identifier: &Identifier,
) -> Result<EqlCiphertext, EqlError> {
    match encrypted {
        Encrypted::Record(ciphertext, terms) => {
            let mut payload = EncryptedPayload {
                version: EQL_SCHEMA_VERSION,
                identifier: identifier.clone(),
                ciphertext,
                hmac_256: None,
                bloom_filter: None,
                ore_block_u64_8_256: None,
                ope_cllw: None,
            };

            for term in terms {
                apply_root_term(&mut payload, term);
            }

            Ok(EqlCiphertext::Encrypted(payload))
        }
        Encrypted::SteVec(ste_vec) => {
            let elements: Vec<SteVecEntry> = ste_vec
                .into_iter()
                .map(
                    |EncryptedEntry {
                         tokenized_selector,
                         term,
                         record,
                         parent_is_array,
                     }| {
                        SteVecEntry {
                            selector: hex::encode(tokenized_selector.as_bytes()),
                            ciphertext: record,
                            is_array: Some(parent_is_array),
                            term: ste_vec_term_from_encrypted(term),
                        }
                    },
                )
                .collect();

            Ok(EqlCiphertext::SteVec(SteVecPayload {
                version: EQL_SCHEMA_VERSION,
                identifier: identifier.clone(),
                ste_vec: elements,
            }))
        }
    }
}

/// Maps a root-scope [`IndexTerm`] onto the matching field of an
/// [`EncryptedPayload`].
///
/// CLLW OPE terms (`OpeFixed` for numerics, `OpeVariable` for strings) map onto
/// the scalar-root `op` field as a single hex string — both share the one `op`
/// key because their order-preserving ciphertexts are directly bytewise
/// comparable and sort into disjoint ranges via plaintext-level domain
/// separation. Block ORE maps onto `ob`.
///
/// Terms with no v2.3 root-scope representation (`BinaryVec`, STE-vec terms,
/// `Null`) are skipped silently — STE-vec terms belong on `sv` elements and the
/// rest are unused by EQL at this scope.
fn apply_root_term(payload: &mut EncryptedPayload, term: IndexTerm) {
    match term {
        IndexTerm::Binary(bytes) => {
            payload.hmac_256 = Some(hex::encode(bytes));
        }
        IndexTerm::BitMap(bf) => {
            payload.bloom_filter = Some(bf);
        }
        IndexTerm::OreFull(bytes) | IndexTerm::OreLeft(bytes) => {
            payload.ore_block_u64_8_256 = Some(vec![hex::encode(bytes)]);
        }
        IndexTerm::OreArray(arr) => {
            payload.ore_block_u64_8_256 = Some(arr.iter().map(hex::encode).collect());
        }
        IndexTerm::OpeFixed(bytes) | IndexTerm::OpeVariable(bytes) => {
            payload.ope_cllw = Some(hex::encode(bytes));
        }
        // No v2.3 root-scope field for these. STE-vec terms belong on `sv`
        // elements; the rest are unused by EQL at this scope.
        IndexTerm::BinaryVec(_)
        | IndexTerm::SteVecSelector(_)
        | IndexTerm::SteVecTerm(_)
        | IndexTerm::SteQueryVec(_)
        | IndexTerm::Null => {}
    }
}

/// Maps an [`EncryptedSteVecTerm`] onto an [`SteVecEntryTerm`].
///
/// Compat-mode OPE bytes are emitted under the sv-element `op` key — the same
/// key (and hex encoding) the scalar root uses for CLLW OPE. The v2.3 schema
/// file predates sv-level `op` (it defines only `hm` / `oc` at element
/// scope); the client emits it ahead of the v3 envelope, the same precedent
/// as the scalar-root `op` field. Standard-mode ORE bytes stay on `oc` — the
/// legacy v2 protocol.
fn ste_vec_term_from_encrypted(term: EncryptedSteVecTerm) -> SteVecEntryTerm {
    match term {
        EncryptedSteVecTerm::Compat(EncryptedSteVecTermCompat::Mac(bytes))
        | EncryptedSteVecTerm::Standard(EncryptedSteVecTermStandard::Mac(bytes)) => {
            SteVecEntryTerm::Hmac {
                hmac_256: hex::encode(bytes),
            }
        }
        EncryptedSteVecTerm::Standard(EncryptedSteVecTermStandard::Ore(ore)) => {
            SteVecEntryTerm::OreCllw {
                ore_cllw_8: hex::encode(ore.as_ref()),
            }
        }
        EncryptedSteVecTerm::Compat(EncryptedSteVecTermCompat::Ope(ope)) => SteVecEntryTerm::Ope {
            ope_cllw: hex::encode(ope.as_ref()),
        },
    }
}

/// Builds an [`EqlQueryPayload`] from a single query-mode [`IndexTerm`].
///
/// Returns [`EqlError::InvalidIndexTerm`] for terms with no v2.3
/// representation (accumulator-style `BinaryVec` terms or `Null`).
fn to_eql_query_payload(
    index_term: IndexTerm,
    identifier: Identifier,
) -> Result<EqlQueryPayload, EqlError> {
    match index_term {
        IndexTerm::Binary(bytes) => Ok(EqlQueryPayload::Encrypted(EncryptedQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier,
            term: RootQueryTerm::Hmac {
                hmac_256: hex::encode(bytes),
            },
        })),
        IndexTerm::BitMap(bf) => Ok(EqlQueryPayload::Encrypted(EncryptedQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier,
            term: RootQueryTerm::BloomFilter { bloom_filter: bf },
        })),
        IndexTerm::OreFull(bytes) | IndexTerm::OreLeft(bytes) => {
            Ok(EqlQueryPayload::Encrypted(EncryptedQueryPayload {
                version: EQL_SCHEMA_VERSION,
                identifier,
                term: RootQueryTerm::OreBlock {
                    ore_block_u64_8_256: vec![hex::encode(bytes)],
                },
            }))
        }
        IndexTerm::OreArray(arr) => Ok(EqlQueryPayload::Encrypted(EncryptedQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier,
            term: RootQueryTerm::OreBlock {
                ore_block_u64_8_256: arr.iter().map(hex::encode).collect(),
            },
        })),
        IndexTerm::SteVecSelector(selector) => Ok(EqlQueryPayload::SteVec(SteVecQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier,
            term: SteVecQueryTerm::Selector {
                selector: hex::encode(selector.as_bytes()),
            },
        })),
        IndexTerm::SteVecTerm(term) => Ok(EqlQueryPayload::SteVec(SteVecQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier,
            term: ste_vec_term_from_encrypted(term).into(),
        })),
        IndexTerm::SteQueryVec(query_vec) => Ok(EqlQueryPayload::SteVec(SteVecQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier,
            term: SteVecQueryTerm::Containment { query_vec },
        })),
        IndexTerm::OpeFixed(bytes) | IndexTerm::OpeVariable(bytes) => {
            Ok(EqlQueryPayload::Encrypted(EncryptedQueryPayload {
                version: EQL_SCHEMA_VERSION,
                identifier,
                term: RootQueryTerm::Ope {
                    ope_cllw: hex::encode(bytes),
                },
            }))
        }
        IndexTerm::BinaryVec(_) | IndexTerm::Null => Err(EqlError::InvalidIndexTerm),
    }
}

/// Builds an [`EqlCiphertextV3`] from an [`Encrypted`] storage result — the v3
/// analogue of [`to_eql_ciphertext`].
///
/// Scalar records become a bare `{v, i, c, <terms>}` payload (no `k`); SteVec
/// documents keep `k: "sv"` and their per-entry `hm`/`op` terms. A CLLW-ORE
/// SteVec entry (Standard mode) is rejected — v3 has no `oc` representation.
fn to_eql_ciphertext_v3(
    encrypted: Encrypted,
    identifier: &Identifier,
) -> Result<EqlCiphertextV3, EqlError> {
    match encrypted {
        Encrypted::Record(ciphertext, terms) => {
            let mut payload = EncryptedPayloadV3 {
                version: EQL_SCHEMA_VERSION_V3,
                identifier: identifier.clone(),
                ciphertext,
                hmac_256: None,
                bloom_filter: None,
                ore_block_u64_8_256: None,
                ope_cllw: None,
            };

            for term in terms {
                apply_root_term_v3(&mut payload, term);
            }

            Ok(EqlCiphertextV3::Encrypted(payload))
        }
        Encrypted::SteVec(ste_vec) => {
            let elements = ste_vec
                .into_iter()
                .map(
                    |EncryptedEntry {
                         tokenized_selector,
                         term,
                         record,
                         parent_is_array,
                     }| {
                        Ok(SteVecEntryV3 {
                            selector: hex::encode(tokenized_selector.as_bytes()),
                            ciphertext: record,
                            is_array: Some(parent_is_array),
                            term: ste_vec_entry_term_v3(term)?,
                        })
                    },
                )
                .collect::<Result<Vec<_>, EqlError>>()?;

            Ok(EqlCiphertextV3::SteVec(SteVecPayloadV3 {
                version: EQL_SCHEMA_VERSION_V3,
                kind: SteVecKind::SteVec,
                identifier: identifier.clone(),
                ste_vec: elements,
            }))
        }
    }
}

/// Maps a root-scope [`IndexTerm`] onto an [`EncryptedPayloadV3`] — the v3
/// analogue of [`apply_root_term`]. The only wire difference is `bf`: v3 carries
/// bloom bit positions as signed `smallint[]`, so the unsigned `u16` positions
/// are reinterpreted here (see [`bloom_filter_to_signed`]).
fn apply_root_term_v3(payload: &mut EncryptedPayloadV3, term: IndexTerm) {
    match term {
        IndexTerm::Binary(bytes) => {
            payload.hmac_256 = Some(hex::encode(bytes));
        }
        IndexTerm::BitMap(bf) => {
            payload.bloom_filter = Some(bloom_filter_to_signed(bf));
        }
        IndexTerm::OreFull(bytes) | IndexTerm::OreLeft(bytes) => {
            payload.ore_block_u64_8_256 = Some(vec![hex::encode(bytes)]);
        }
        IndexTerm::OreArray(arr) => {
            payload.ore_block_u64_8_256 = Some(arr.iter().map(hex::encode).collect());
        }
        IndexTerm::OpeFixed(bytes) | IndexTerm::OpeVariable(bytes) => {
            payload.ope_cllw = Some(hex::encode(bytes));
        }
        // No v3 root-scope field for these (as v2).
        IndexTerm::BinaryVec(_)
        | IndexTerm::SteVecSelector(_)
        | IndexTerm::SteVecTerm(_)
        | IndexTerm::SteQueryVec(_)
        | IndexTerm::Null => {}
    }
}

/// Reinterprets v2 unsigned bloom-filter bit positions into the signed
/// `smallint[]` representation v3 stores. Each `u16` maps to an `i16` by
/// two's-complement reinterpretation (`0..=32767` unchanged; `32768..=65535`
/// wrap negative). Lossless and total — a `u16` is always a valid `smallint`
/// bit position.
fn bloom_filter_to_signed(bf: Vec<u16>) -> Vec<i16> {
    bf.into_iter().map(|bit| bit as i16).collect()
}

/// Maps a v2 [`EncryptedSteVecTerm`] onto a v3 [`SteVecEntryTermV3`], rejecting
/// CLLW-ORE (Standard mode) — v3 has no `oc` representation.
fn ste_vec_entry_term_v3(term: EncryptedSteVecTerm) -> Result<SteVecEntryTermV3, EqlError> {
    match term {
        EncryptedSteVecTerm::Compat(EncryptedSteVecTermCompat::Mac(bytes))
        | EncryptedSteVecTerm::Standard(EncryptedSteVecTermStandard::Mac(bytes)) => {
            Ok(SteVecEntryTermV3::Hmac {
                hmac_256: hex::encode(bytes),
            })
        }
        EncryptedSteVecTerm::Compat(EncryptedSteVecTermCompat::Ope(ope)) => {
            Ok(SteVecEntryTermV3::Ope {
                ope_cllw: hex::encode(ope.as_ref()),
            })
        }
        EncryptedSteVecTerm::Standard(EncryptedSteVecTermStandard::Ore(_)) => {
            Err(EqlError::UnsupportedSteVecOreInV3)
        }
    }
}

/// Builds an [`EqlQueryPayloadV3`] from a single query-mode [`IndexTerm`] — the
/// v3 analogue of [`to_eql_query_payload`].
///
/// A single scalar term (`hm`/`bf`/`ob`/`op`) becomes a **one-term** enveloped
/// operand `{v, i, <term>}` (a `query_<name>` domain that requires several terms
/// needs the multi-term projection [`EqlCiphertextV3::into_query_operand`]
/// instead). A `SteQueryVec` (jsonb containment) becomes the bare needle
/// `{sv: [{s, hm|op}]}`. A `SteVecSelector` (jsonb path query) becomes the bare
/// selector hash. A single sv term has no v3 query-operand shape
/// ([`EqlError::UnsupportedV3QueryTerm`]).
fn to_eql_query_payload_v3(
    index_term: IndexTerm,
    identifier: Identifier,
) -> Result<EqlQueryPayloadV3, EqlError> {
    let mut operand = EncryptedQueryPayloadV3 {
        version: EQL_SCHEMA_VERSION_V3,
        identifier,
        hmac_256: None,
        bloom_filter: None,
        ore_block_u64_8_256: None,
        ope_cllw: None,
    };
    match index_term {
        IndexTerm::Binary(bytes) => operand.hmac_256 = Some(hex::encode(bytes)),
        IndexTerm::BitMap(bf) => operand.bloom_filter = Some(bloom_filter_to_signed(bf)),
        IndexTerm::OreFull(bytes) | IndexTerm::OreLeft(bytes) => {
            operand.ore_block_u64_8_256 = Some(vec![hex::encode(bytes)]);
        }
        IndexTerm::OreArray(arr) => {
            operand.ore_block_u64_8_256 = Some(arr.iter().map(hex::encode).collect());
        }
        IndexTerm::OpeFixed(bytes) | IndexTerm::OpeVariable(bytes) => {
            operand.ope_cllw = Some(hex::encode(bytes));
        }
        IndexTerm::SteQueryVec(query_vec) => {
            let entries = query_vec
                .into_iter()
                .map(|entry| {
                    let (selector, term) = entry.into_parts();
                    Ok(SteVecQueryEntryV3 {
                        selector: hex::encode(selector.as_bytes()),
                        term: ste_vec_entry_term_v3(term)?,
                    })
                })
                .collect::<Result<Vec<_>, EqlError>>()?;
            return Ok(EqlQueryPayloadV3::SteVec(SteVecQueryPayloadV3 {
                ste_vec: entries,
            }));
        }
        IndexTerm::SteVecSelector(selector) => {
            // A jsonb path query (`->`/`->>`): the bare tokenized-selector hash,
            // used directly as `text`. Not an encrypted operand.
            return Ok(EqlQueryPayloadV3::Selector(hex::encode(
                selector.as_bytes(),
            )));
        }
        // A bare single sv term has no v3 query-operand shape (v3 jsonb queries
        // are the containment needle or the selector).
        IndexTerm::SteVecTerm(_) => return Err(EqlError::UnsupportedV3QueryTerm),
        IndexTerm::BinaryVec(_) | IndexTerm::Null => return Err(EqlError::InvalidIndexTerm),
    }
    Ok(EqlQueryPayloadV3::Encrypted(operand))
}

/// Errors that can occur during EQL encryption, decryption, and index operations.
#[derive(Error, Debug)]
pub enum EqlError {
    /// Failed to serialize ciphertext to JSON format.
    #[error(transparent)]
    CiphertextCouldNotBeSerialised(#[from] serde_json::Error),

    /// An encrypted column value could not be parsed from the database format.
    #[error("Encrypted column could not be parsed")]
    ColumnCouldNotBeParsed,

    /// An encrypted column value was NULL when a value was expected.
    #[error("Encrypted column is null")]
    ColumnIsNull,

    /// Failed to deserialize an encrypted column from the database.
    #[error("Column '{column}' in table '{table}' could not be deserialised")]
    ColumnCouldNotBeDeserialised { table: String, column: String },

    /// Failed to encrypt a column value.
    #[error("Column '{column}' in table '{table}' could not be encrypted")]
    ColumnCouldNotBeEncrypted { table: String, column: String },

    /// The encryption configuration for a column doesn't match the encrypted data.
    #[error("Column configuration for column '{column}' in table '{table}' does not match the encrypted column")]
    ColumnConfigurationMismatch { table: String, column: String },

    /// Decryption failed using the specified keyset.
    ///
    /// The underlying `zerokms::Error` is preserved through `#[source]` —
    /// Display stays opaque (keyset_id only, no dynamic data), but anyhow
    /// chain printing (`{:?}` / `{:#}`) and `tracing` can walk through to
    /// the actual `RetrieveKeyError`.
    #[error("Could not decrypt data using keyset '{keyset_id}'")]
    CouldNotDecryptDataForKeyset {
        keyset_id: String,
        #[source]
        source: zerokms::Error,
    },

    /// An index term was not of the expected type for the operation.
    #[error("InvalidIndexTerm")]
    InvalidIndexTerm,

    /// A SteVec entry carried a CLLW-ORE ordering term while producing a v3
    /// payload. v3 orders SteVec entries by the byte-comparable CLLW-OPE `op`
    /// term; ORE ciphertext bytes have no v3 representation. Re-encrypt the
    /// column in Compat (OPE) mode to produce v3 SteVec payloads.
    #[error(
        "SteVec CLLW-ORE (`oc`) ordering terms have no EQL v3 representation; \
         encrypt the column in Compat (OPE) mode for v3"
    )]
    UnsupportedSteVecOreInV3,

    /// A query-mode [`IndexTerm`] has no EQL v3 query-operand representation
    /// (e.g. a bare SteVec selector or single per-entry term, which v3 exposes
    /// only through the jsonb containment needle).
    #[error("index term has no EQL v3 query-operand representation")]
    UnsupportedV3QueryTerm,

    /// The EQL payload is missing a decryptable ciphertext.
    ///
    /// Raised when an [`EqlCiphertext::SteVec`] payload has an empty `sv`
    /// array (no root entry to decrypt).
    #[error("EQL payload for column '{}' in table '{}' is missing ciphertext", _0.column(), _0.table())]
    MissingCiphertext(Identifier),

    /// The keyset ID provided via `SET CIPHERSTASH.KEYSET_ID` is not a valid UUID.
    #[error("KeysetId `{id}` could not be parsed using `SET CIPHERSTASH.KEYSET_ID`. KeysetId should be a valid UUID")]
    KeysetIdCouldNotBeParsed { id: String },

    /// Failed to set the keyset ID using `SET CIPHERSTASH.KEYSET_ID`.
    #[error("Keyset Id could not be set using `SET CIPHERSTASH.KEYSET_ID`")]
    KeysetIdCouldNotBeSet,

    /// Failed to set the keyset name using `SET CIPHERSTASH.KEYSET_NAME`.
    #[error("Keyset Name could not be set using `SET CIPHERSTASH.KEYSET_NAME`")]
    KeysetNameCouldNotBeSet,

    /// Missing encryption configuration for a specific column type.
    ///
    /// This should be unreachable in practice.
    #[error("Missing encrypt configuration for column type `{plaintext_type}`")]
    MissingEncryptConfiguration { plaintext_type: &'static str },

    /// Failed to encode decrypted plaintext as the expected data type.
    #[error("Decrypted column could not be encoded as the expected type")]
    PlaintextCouldNotBeEncoded,

    /// An error occurred in the encryption pipeline.
    #[error(transparent)]
    Pipeline(#[from] encryption::EncryptionError),

    /// Failed to decode plaintext from the expected type format.
    #[error(transparent)]
    PlaintextCouldNotBeDecoded(#[from] encryption::TypeParseError),

    /// No keyset identifier was provided when one was required.
    #[error("Missing keyset identifer")]
    MissingKeysetIdentifier,

    /// Attempted to set a keyset when a default keyset is already configured.
    #[error("Cannot SET CIPHERSTASH.KEYSET if a default keyset has been configured")]
    UnexpectedSetKeyset,

    /// The specified column has no encryption configuration.
    #[error("Column '{column}' in table '{table}' has no Encrypt configuration")]
    UnknownColumn { table: String, column: String },

    /// The keyset name or ID is not found in the configured credentials.
    #[error("Unknown keyset name or id '{keyset}'. Check the configured credentials")]
    UnknownKeysetIdentifier { keyset: String },

    /// The specified table has no encryption configuration.
    #[error("Table '{table}' has no Encrypt configuration")]
    UnknownTable { table: String },

    /// An unknown or unsupported index term type was encountered for the column.
    #[error("Unknown Index Term for column '{}' in table '{}'", _0.column(), _0.table())]
    UnknownIndexTerm(Identifier),

    /// A ZeroKMS error
    #[error("ZeroKMS error '{}'", _0)]
    ZeroKMS(#[from] zerokms::Error),

    /// A record decryption error
    #[error("Record decryption error '{}'", _0)]
    RecordDecrypt(#[from] RecordDecryptError),
}

/// Specifies what to encrypt when encrypting [`encryption::Plaintext`] values.
///
/// `EqlOperation` determines whether to perform full encryption for storage or
/// to generate a query-mode payload carrying a single encrypted search term.
#[derive(Debug)]
pub enum EqlOperation<'a> {
    /// Encrypt both the plaintext and its associated encrypted search terms.
    ///
    /// Use this operation when persisting encrypted data while keeping it
    /// searchable. The result is an [`EqlCiphertext`].
    Store,

    /// Generate an encrypted search term for this specific index type.
    ///
    /// The tuple carries the [`IndexType`] and [`QueryOp`] to use. The result
    /// is an [`EqlQueryPayload`].
    Query(&'a IndexType, QueryOp),
}

/// A prepared plaintext value ready for EQL encryption.
///
/// `PreparedPlaintext` bundles together all the information needed to encrypt a plaintext value
/// into an EQL payload: the value itself, its location identifier (table/column), the operation
/// to perform (storage or query), and the column's encryption configuration.
///
/// # Fields
///
/// * `identifier` - The table and column this plaintext belongs to
/// * `plaintext` - The actual plaintext value to encrypt
/// * `eql_op` - Whether to encrypt for storage or generate a query payload
/// * `column_config` - The encryption configuration for this column
///
/// # Examples
///
/// ```no_run
/// use cipherstash_client::eql::{PreparedPlaintext, Identifier, EqlOperation};
/// use cipherstash_client::encryption::Plaintext;
/// use std::borrow::Cow;
/// # use cipherstash_config::ColumnConfig;
///
/// let identifier = Identifier::new("users", "email");
/// let plaintext = Plaintext::Text(Some("user@example.com".into()));
/// # let column_config: ColumnConfig = unimplemented!();
///
/// let prepared = PreparedPlaintext::new(
///     Cow::Owned(column_config),
///     identifier,
///     plaintext,
///     EqlOperation::Store,
/// );
/// ```
pub struct PreparedPlaintext<'a> {
    /// The table and column this plaintext belongs to.
    identifier: Identifier,

    /// The actual plaintext value to encrypt.
    plaintext: encryption::Plaintext,

    /// Whether to encrypt for storage or generate a query payload.
    eql_op: EqlOperation<'a>,

    /// The encryption configuration for this column.
    column_config: Cow<'a, ColumnConfig>,
}

impl<'a> PreparedPlaintext<'a> {
    /// Constructs a [`PreparedPlaintext`] from its component parts. See the
    /// type-level docs for an example.
    pub fn new(
        column_config: Cow<'a, ColumnConfig>,
        identifier: Identifier,
        plaintext: encryption::Plaintext,
        eql_op: EqlOperation<'a>,
    ) -> Self {
        Self {
            identifier,
            plaintext,
            eql_op,
            column_config,
        }
    }
}

/// Internal representation of an encryption target during EQL encryption.
///
/// This enum distinguishes between encrypting data for storage (which generates both ciphertext
/// and all configured SEM terms) and generating a single query SEM term for a query payload.
enum EncryptionTarget<'a> {
    /// Encrypt plaintext for storage with all configured SEM term types.
    ///
    /// Contains the identifier and a configured storage builder.
    ForStorage(Identifier, StorageBuilder<'a, encryption::Plaintext>),

    /// Generate a single SEM term for a specific SEM term type.
    ///
    /// Contains the identifier, plaintext, SEM type, and query operation.
    ForQuery(Identifier, encryption::Plaintext, &'a IndexType, QueryOp),
}

/// Generates data key payloads for encryption targets that require storage.
///
/// Filters the encryption targets to those that need data key generation (i.e., `ForStorage`
/// targets) and creates the corresponding key-generation payloads for the ZeroKMS service.
///
/// # Arguments
///
/// * `opts` - Encryption options containing the lock context (and optional decryption policy)
/// * `targets` - The encryption targets to generate keys for
///
/// # Returns
///
/// A vector of [`GenerateKeyPayload`] values, one for each storage target. Query-only targets
/// are excluded as they don't require new data keys.
fn generate_data_key_payloads<'a>(
    opts: &EqlEncryptOpts<'a>,
    targets: &'a Vec<EncryptionTarget<'a>>,
) -> Vec<GenerateKeyPayload<'a>> {
    targets
        .iter()
        .filter_map(|target| match target {
            EncryptionTarget::ForStorage(_, builder) => {
                let payload =
                    GenerateKeyPayload::new(builder.descriptor(), opts.lock_context.clone());
                Some(match opts.decryption_policy.clone() {
                    Some(p) => payload.with_decryption_policy(p),
                    None => payload,
                })
            }
            EncryptionTarget::ForQuery(_, _, _, _) => None,
        })
        .collect()
}

/// Converts prepared plaintexts into encryption targets.
///
/// Transforms high-level [`PreparedPlaintext`] values into low-level [`EncryptionTarget`]
/// representations, preparing them for encryption. Distinguishes between storage encryption
/// (which uses [`StorageBuilder`]) and query payload generation.
///
/// # Arguments
///
/// * `index_key` - The index key for generating searchable indexes
/// * `plaintexts` - The prepared plaintext values to convert
/// * `effective_keyset_id` - The keyset ID to use for encryption
///
/// # Returns
///
/// A vector of [`EncryptionTarget`] values ready for encryption.
///
/// # Errors
///
/// Returns `EncryptionError` if the plaintext values cannot be properly configured for
/// encryption (e.g., invalid column configuration).
fn to_encryption_targets<'a>(
    index_key: &'a IndexKey,
    plaintexts: Vec<PreparedPlaintext<'a>>,
    effective_keyset_id: Uuid,
) -> Result<Vec<EncryptionTarget<'a>>, encryption::EncryptionError> {
    plaintexts
        .into_iter()
        .map(
            move |prepared_plaintext| -> Result<EncryptionTarget, encryption::EncryptionError> {
                use crate::encryption::Encryptable;

                let PreparedPlaintext {
                    identifier,
                    plaintext,
                    eql_op,
                    column_config,
                } = prepared_plaintext;

                match eql_op {
                    EqlOperation::Store => Ok(EncryptionTarget::ForStorage(
                        identifier,
                        PlaintextTarget::new(plaintext, (*column_config).clone())
                            .build_encryptable(index_key, effective_keyset_id)?,
                    )),
                    EqlOperation::Query(index_type, query_op) => Ok(EncryptionTarget::ForQuery(
                        identifier, plaintext, index_type, query_op,
                    )),
                }
            },
        )
        .collect::<Result<Vec<_>, _>>()
}

/// Options for EQL decryption operations.
///
/// Configuration for [`decrypt_eql`] and [`decrypt_eql_fallible`], controlling which keyset to
/// use and the lock context for key unwrapping.
///
/// # Fields
///
/// * `keyset_id` - Optional keyset UUID; if `None`, uses the cipher's default keyset
/// * `lock_context` - Context information for unwrapping encrypted keys
/// * `unverified_context` - Optional unverified context for additional metadata
///
/// # Examples
///
/// ```
/// use cipherstash_client::eql::EqlDecryptOpts;
/// use std::borrow::Cow;
///
/// let opts = EqlDecryptOpts {
///     keyset_id: None, // Use default keyset
///     lock_context: Cow::Borrowed(&[]),
///     unverified_context: None,
/// };
/// ```
#[derive(Debug, Default)]
pub struct EqlDecryptOpts<'a> {
    /// Keyset to use for all decryption operations.
    ///
    /// When set to `None`, decryption uses the keyset associated with the ZeroKMS client.
    pub keyset_id: Option<Uuid>,

    /// The lock context to use for decryption operations.
    pub lock_context: Cow<'a, [Context]>,

    /// Optional unverified context for additional metadata.
    pub unverified_context: Option<Cow<'a, UnverifiedContext>>,
}

/// Options for EQL encryption operations.
///
/// Configuration for [`encrypt_eql`], controlling which keyset to use, the lock context for key
/// wrapping, and which indexes to generate.
///
/// # Fields
///
/// * `keyset_id` - Optional keyset UUID; if `None`, uses the cipher's default keyset
/// * `lock_context` - Context information for wrapping data encryption keys
/// * `unverified_context` - Optional unverified context for additional metadata
/// * `index_types` - Optional filter to generate only specific index types instead of all
///   configured indexes
/// * `decryption_policy` - Optional decryption policy for OR-style lock context
///
/// # Examples
///
/// ```
/// use cipherstash_client::eql::EqlEncryptOpts;
/// use std::borrow::Cow;
///
/// // Use default options (all indexes, default keyset)
/// let opts = EqlEncryptOpts::default();
///
/// // Or customize
/// # use cipherstash_config::column::IndexType;
/// let opts = EqlEncryptOpts {
///     keyset_id: None,
///     lock_context: Cow::Owned(vec![]),
///     unverified_context: None,
///     index_types: Some(Cow::Owned(vec![IndexType::Ore])), // Generate only ORE indexes
///     decryption_policy: None,
/// };
/// ```
#[derive(Debug, Default)]
pub struct EqlEncryptOpts<'a> {
    /// Keyset to use for all encryption operations.
    ///
    /// When set to `None`, encryption uses the keyset associated with the ZeroKMS client.
    pub keyset_id: Option<Uuid>,

    /// The lock context to use for encryption operations.
    pub lock_context: Cow<'a, [Context]>,

    /// Optional unverified context for additional metadata.
    pub unverified_context: Option<Cow<'a, UnverifiedContext>>,

    /// Optional filter to generate only specific index types.
    ///
    /// When present, generate only the specified index types instead of all index types
    /// specified in the column configuration.
    pub index_types: Option<Cow<'a, [IndexType]>>,

    /// Optional decryption policy for OR-style lock context.
    ///
    /// When set, `tag_version = 1` is used instead of the `lock_context` field.
    pub decryption_policy: Option<DecryptionPolicy>,
}

/// Converts ZeroKMS errors into more user-friendly EQL errors.
///
/// Inspects ZeroKMS errors and provides more specific error messages, particularly for the
/// common case of a missing or incorrect keyset.
///
/// # Arguments
///
/// * `err` - The ZeroKMS error to convert
/// * `cipher_keyset_id` - The keyset ID configured on the cipher
/// * `keyset_id_override` - The optional per-call keyset ID override (from `EqlEncryptOpts` /
///   `EqlDecryptOpts`)
///
/// # Returns
///
/// An [`EqlError`] with a more specific error message when possible, or the original ZeroKMS
/// error wrapped in [`EqlError::ZeroKMS`].
fn convert_zerokms_error(
    err: zerokms::Error,
    cipher_keyset_id: Uuid,
    keyset_id_override: Option<Uuid>,
) -> EqlError {
    match err {
        zerokms::Error::Decrypt(_) => {
            let error_msg = err.to_string();
            if error_msg.contains("Failed to retrieve key") {
                EqlError::CouldNotDecryptDataForKeyset {
                    keyset_id: keyset_id_override
                        .map(|id| id.to_string())
                        .unwrap_or(cipher_keyset_id.to_string()),
                    source: err,
                }
            } else {
                EqlError::ZeroKMS(err)
            }
        }
        _ => EqlError::ZeroKMS(err),
    }
}

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

    #[test]
    fn convert_zerokms_error_preserves_source_chain() {
        use crate::zerokms::{DecryptError, RetrieveKeyError};
        use std::error::Error as StdError;

        let cipher_keyset_id = Uuid::new_v4();
        let override_id = Uuid::new_v4();

        let inner = zerokms::Error::from(DecryptError::from(RetrieveKeyError::FailedRetrieval(
            "no such key".into(),
        )));

        let converted = convert_zerokms_error(inner, cipher_keyset_id, Some(override_id));

        match converted {
            EqlError::CouldNotDecryptDataForKeyset { ref keyset_id, .. } => {
                assert_eq!(keyset_id, &override_id.to_string());
            }
            ref other => panic!("expected CouldNotDecryptDataForKeyset, got {other:?}"),
        }

        let source = StdError::source(&converted).expect("source() should be Some");
        assert!(
            source.downcast_ref::<zerokms::Error>().is_some(),
            "source should downcast to &zerokms::Error"
        );
    }

    #[test]
    fn empty_ste_vec_payload_yields_missing_ciphertext() {
        // A SteVec payload with no entries has no root ciphertext to decrypt.
        let identifier = Identifier::new("test_table", "test_column");
        let eql = EqlCiphertext::SteVec(SteVecPayload {
            version: EQL_SCHEMA_VERSION,
            identifier: identifier.clone(),
            ste_vec: Vec::new(),
        });

        let result = extract_root_ciphertext(eql);
        assert!(matches!(result, Err(EqlError::MissingCiphertext(_))));
    }

    #[test]
    fn mp_base85_deserialize_invalid_input_returns_error() {
        use serde::de::value::{Error as ValueError, StrDeserializer};
        use serde::de::IntoDeserializer;

        let invalid: StrDeserializer<ValueError> = "not-valid-base85!!!".into_deserializer();

        let result: Result<EncryptedRecord, ValueError> = formats::mp_base85::deserialize(invalid);

        assert!(result.is_err(), "Invalid base85 input should return error");
    }

    #[test]
    fn encrypted_payload_serializes_with_k_ct_discriminator() {
        // Build a fake EncryptedRecord then serialize through the enum.
        let record = EncryptedRecord {
            iv: Default::default(),
            ciphertext: vec![1; 16],
            tag: vec![2; 16],
            descriptor: "users/email".to_string(),
            keyset_id: Some(Uuid::nil()),
            decryption_policy: None,
        };
        let payload = EqlCiphertext::Encrypted(EncryptedPayload {
            version: EQL_SCHEMA_VERSION,
            identifier: Identifier::new("users", "email"),
            ciphertext: record,
            hmac_256: Some("deadbeef".into()),
            bloom_filter: None,
            ore_block_u64_8_256: None,
            ope_cllw: None,
        });

        let value = serde_json::to_value(&payload).unwrap();
        assert_eq!(value["k"], "ct");
        assert_eq!(value["v"], EQL_SCHEMA_VERSION);
        assert_eq!(value["i"]["t"], "users");
        assert_eq!(value["i"]["c"], "email");
        assert_eq!(value["hm"], "deadbeef");
        assert!(value.get("c").is_some());
        assert!(value.get("sv").is_none());
        assert!(value.get("ob").is_none());
        assert!(value.get("op").is_none());
    }

    fn fake_record(descriptor: &str) -> EncryptedRecord {
        EncryptedRecord {
            iv: Default::default(),
            ciphertext: vec![1; 16],
            tag: vec![2; 16],
            descriptor: descriptor.to_string(),
            keyset_id: Some(Uuid::nil()),
            decryption_policy: None,
        }
    }

    /// EQL v3 wire-shape tests. These are pure (no ZeroKMS): they drive the v3
    /// assemblers directly, or serialize the v3 structs, and assert the JSON
    /// shape the `eql_v3` domain CHECKs expect. The v2 tests above must stay
    /// green — v3 is additive.
    mod v3 {
        use super::*;

        /// Build an [`EncryptedSteVecTerm`] from its wire JSON. The term's
        /// inner types (`Mac`, the CLLW ciphertexts) are not publicly
        /// re-exported, so tests construct terms through the `Deserialize` impl
        /// rather than naming those types. `{hm}` → Mac, `{op}` → Compat OPE,
        /// `{oc}` → Standard ORE (keys are mode-unambiguous).
        fn term(json: serde_json::Value) -> EncryptedSteVecTerm {
            serde_json::from_value(json).expect("valid EncryptedSteVecTerm wire shape")
        }

        #[test]
        fn store_scalar_drops_k_and_bumps_version() {
            let encrypted = Encrypted::Record(
                fake_record("users/email"),
                vec![
                    IndexTerm::Binary(vec![0xAB; 32]),
                    IndexTerm::OreArray(vec![vec![0x01; 8], vec![0x02; 8]]),
                ],
            );
            let eql = to_eql_ciphertext_v3(encrypted, &Identifier::new("users", "email")).unwrap();
            let value = serde_json::to_value(&eql).unwrap();

            assert_eq!(value["v"], EQL_SCHEMA_VERSION_V3);
            assert!(value.get("k").is_none(), "v3 scalar payloads carry no `k`");
            assert_eq!(value["i"]["t"], "users");
            assert_eq!(value["hm"], hex::encode([0xAB; 32]));
            assert!(value.get("c").is_some());
            assert!(value["ob"].is_array());
            assert!(value.get("op").is_none());
            assert!(value.get("bf").is_none());
        }

        #[test]
        fn store_scalar_bloom_positions_are_signed_smallint() {
            // v2 emits unsigned u16 bit positions; v3 reinterprets the upper
            // half two's-complement negative (40000 - 65536 = -25536).
            let encrypted = Encrypted::Record(
                fake_record("t/c"),
                vec![IndexTerm::BitMap(vec![0, 32767, 40000])],
            );
            let eql = to_eql_ciphertext_v3(encrypted, &Identifier::new("t", "c")).unwrap();
            let value = serde_json::to_value(&eql).unwrap();
            assert_eq!(value["bf"], serde_json::json!([0, 32767, -25536]));
        }

        #[test]
        fn store_scalar_ope_term_on_op() {
            let encrypted = Encrypted::Record(
                fake_record("orders/total"),
                vec![IndexTerm::OpeFixed(vec![0xAB; 65])],
            );
            let eql = to_eql_ciphertext_v3(encrypted, &Identifier::new("orders", "total")).unwrap();
            let value = serde_json::to_value(&eql).unwrap();
            assert_eq!(value["op"], hex::encode([0xAB; 65]));
            assert!(value.get("k").is_none());
        }

        #[test]
        fn ste_vec_entry_term_maps_mac_and_ope_rejects_ore() {
            let hm =
                ste_vec_entry_term_v3(term(serde_json::json!({ "hm": "11".repeat(16) }))).unwrap();
            assert!(matches!(hm, SteVecEntryTermV3::Hmac { .. }));

            let op = ste_vec_entry_term_v3(term(serde_json::json!({ "op": "010203" }))).unwrap();
            assert!(matches!(op, SteVecEntryTermV3::Ope { .. }));

            let ore = ste_vec_entry_term_v3(term(serde_json::json!({ "oc": "4243" })));
            assert!(matches!(ore, Err(EqlError::UnsupportedSteVecOreInV3)));
        }

        #[test]
        fn ste_vec_document_keeps_k_and_omits_per_entry_envelope() {
            let doc = SteVecPayloadV3 {
                version: EQL_SCHEMA_VERSION_V3,
                kind: SteVecKind::SteVec,
                identifier: Identifier::new("docs", "attrs"),
                ste_vec: vec![
                    SteVecEntryV3 {
                        selector: "aa".into(),
                        ciphertext: fake_record("docs/attrs"),
                        is_array: None,
                        term: SteVecEntryTermV3::Hmac {
                            hmac_256: "beef".into(),
                        },
                    },
                    SteVecEntryV3 {
                        selector: "bb".into(),
                        ciphertext: fake_record("docs/attrs"),
                        is_array: Some(false),
                        term: SteVecEntryTermV3::Ope {
                            ope_cllw: "00ff".into(),
                        },
                    },
                ],
            };
            let v = serde_json::to_value(EqlCiphertextV3::SteVec(doc)).unwrap();
            assert_eq!(v["v"], EQL_SCHEMA_VERSION_V3);
            assert_eq!(v["k"], "sv");
            let sv = v["sv"].as_array().unwrap();
            assert_eq!(sv.len(), 2);
            assert_eq!(sv[0]["s"], "aa");
            assert_eq!(sv[0]["hm"], "beef");
            assert!(sv[0].get("a").is_none(), "None array marker is omitted");
            assert!(sv[0].get("v").is_none(), "entries carry no per-entry `v`");
            assert!(sv[0].get("i").is_none());
            assert!(sv[0].get("k").is_none());
            assert_eq!(sv[1]["op"], "00ff");
            assert_eq!(sv[1]["a"], false);
            assert!(sv[1].get("hm").is_none());
        }

        #[test]
        fn query_scalar_operand_is_enveloped_without_k_or_c() {
            let out = to_eql_query_payload_v3(
                IndexTerm::Binary(vec![0xCD; 32]),
                Identifier::new("users", "email"),
            )
            .unwrap();
            let v = serde_json::to_value(&out).unwrap();
            assert_eq!(v["v"], EQL_SCHEMA_VERSION_V3);
            assert_eq!(v["i"]["t"], "users");
            assert_eq!(v["hm"], hex::encode([0xCD; 32]));
            assert!(v.get("k").is_none());
            assert!(v.get("c").is_none());
        }

        #[test]
        fn query_scalar_bloom_operand_is_signed() {
            let out = to_eql_query_payload_v3(
                IndexTerm::BitMap(vec![1, 40000]),
                Identifier::new("t", "c"),
            )
            .unwrap();
            let v = serde_json::to_value(&out).unwrap();
            assert_eq!(v["bf"], serde_json::json!([1, -25536]));
        }

        #[test]
        fn into_query_operand_scalar_keeps_all_terms_and_drops_c() {
            // A multi-term store payload (hm + ob) projects to a multi-term
            // operand `{v, i, hm, ob}` — the shape a `query_<name>` domain needs
            // (a single EqlOperation::Query yields only ONE term) — with `c` gone.
            let stored = to_eql_ciphertext_v3(
                Encrypted::Record(
                    fake_record("users/email"),
                    vec![
                        IndexTerm::Binary(vec![0xAB; 32]),
                        IndexTerm::OreArray(vec![vec![1; 8], vec![2; 8]]),
                    ],
                ),
                &Identifier::new("users", "email"),
            )
            .unwrap();
            let v = serde_json::to_value(stored.into_query_operand()).unwrap();
            assert_eq!(v["v"], EQL_SCHEMA_VERSION_V3);
            assert_eq!(v["i"]["t"], "users");
            assert_eq!(v["hm"], hex::encode([0xAB; 32]));
            assert!(v["ob"].is_array());
            assert!(v.get("c").is_none(), "a query operand must not carry `c`");
            assert!(v.get("k").is_none());
        }

        #[test]
        fn into_query_operand_ste_vec_doc_becomes_the_bare_needle() {
            let doc = EqlCiphertextV3::SteVec(SteVecPayloadV3 {
                version: EQL_SCHEMA_VERSION_V3,
                kind: SteVecKind::SteVec,
                identifier: Identifier::new("docs", "attrs"),
                ste_vec: vec![SteVecEntryV3 {
                    selector: "aa".into(),
                    ciphertext: fake_record("docs/attrs"),
                    is_array: Some(false),
                    term: SteVecEntryTermV3::Ope {
                        ope_cllw: "00ff".into(),
                    },
                }],
            });
            let v = serde_json::to_value(doc.into_query_operand()).unwrap();
            assert!(v.get("v").is_none(), "the needle is not enveloped");
            assert!(v.get("i").is_none());
            let sv = v["sv"].as_array().unwrap();
            assert_eq!(sv[0]["s"], "aa");
            assert_eq!(sv[0]["op"], "00ff");
            assert!(sv[0].get("c").is_none(), "per-entry `c` is dropped");
            assert!(sv[0].get("a").is_none(), "per-entry `a` is dropped");
        }

        #[test]
        fn query_ste_vec_selector_is_the_bare_hash() {
            use crate::encryption::TokenizedSelector;
            let out = to_eql_query_payload_v3(
                IndexTerm::SteVecSelector(TokenizedSelector([7; 16])),
                Identifier::new("docs", "attrs"),
            )
            .unwrap();
            assert!(matches!(out, EqlQueryPayloadV3::Selector(_)));
            // Serializes as a bare string (the selector used directly as `text`).
            assert_eq!(
                serde_json::to_value(&out).unwrap(),
                serde_json::json!(hex::encode([7u8; 16]))
            );
        }

        #[test]
        fn jsonb_containment_needle_is_bare_sv_array() {
            let needle = EqlQueryPayloadV3::SteVec(SteVecQueryPayloadV3 {
                ste_vec: vec![SteVecQueryEntryV3 {
                    selector: "abcd".into(),
                    term: SteVecEntryTermV3::Hmac {
                        hmac_256: "beef".into(),
                    },
                }],
            });
            let v = serde_json::to_value(&needle).unwrap();
            assert!(v.get("v").is_none(), "the needle is not enveloped");
            assert!(v.get("i").is_none());
            let sv = v["sv"].as_array().unwrap();
            assert_eq!(sv[0]["s"], "abcd");
            assert_eq!(sv[0]["hm"], "beef");
        }

        #[test]
        fn output_store_serializes_transparently() {
            let encrypted =
                Encrypted::Record(fake_record("t/c"), vec![IndexTerm::Binary(vec![9; 32])]);
            let store = EqlOutputV3::Store(
                to_eql_ciphertext_v3(encrypted, &Identifier::new("t", "c")).unwrap(),
            );
            let v = serde_json::to_value(&store).unwrap();
            assert_eq!(v["v"], EQL_SCHEMA_VERSION_V3);
            assert_eq!(v["hm"], hex::encode([9u8; 32]));
        }

        // ---- Deserialize / round-trip coverage ----
        //
        // Every test above only *serializes* the v3 structs. The `Deserialize`
        // derives — and specifically the `#[serde(flatten)]` of the `untagged`
        // `SteVecEntryTermV3` inside `SteVecEntryV3` / `SteVecQueryEntryV3` — are
        // never exercised. flatten-over-untagged is the exact serde combination
        // that has historically misbehaved, so these round-trip the structs and
        // assert the `hm`/`op` discriminator survives a decode.

        #[test]
        fn ste_vec_document_round_trips_through_serde() {
            // A full stored document (with the required per-entry `c`) survives
            // serialize -> deserialize -> serialize unchanged, and the flattened
            // untagged term decodes back to the right variant on each entry.
            let doc = SteVecPayloadV3 {
                version: EQL_SCHEMA_VERSION_V3,
                kind: SteVecKind::SteVec,
                identifier: Identifier::new("docs", "attrs"),
                ste_vec: vec![
                    SteVecEntryV3 {
                        selector: "aa".into(),
                        ciphertext: fake_record("docs/attrs"),
                        is_array: None,
                        term: SteVecEntryTermV3::Hmac {
                            hmac_256: "beef".into(),
                        },
                    },
                    SteVecEntryV3 {
                        selector: "bb".into(),
                        ciphertext: fake_record("docs/attrs"),
                        is_array: Some(true),
                        term: SteVecEntryTermV3::Ope {
                            ope_cllw: "00ff".into(),
                        },
                    },
                ],
            };

            let wire = serde_json::to_value(&doc).unwrap();
            let parsed: SteVecPayloadV3 = serde_json::from_value(wire.clone()).unwrap();

            assert_eq!(
                serde_json::to_value(&parsed).unwrap(),
                wire,
                "document must round-trip byte-for-byte"
            );
            assert!(matches!(
                parsed.ste_vec[0].term,
                SteVecEntryTermV3::Hmac { ref hmac_256 } if hmac_256 == "beef"
            ));
            assert_eq!(parsed.ste_vec[0].is_array, None);
            assert!(matches!(
                parsed.ste_vec[1].term,
                SteVecEntryTermV3::Ope { ref ope_cllw } if ope_cllw == "00ff"
            ));
            assert_eq!(parsed.ste_vec[1].is_array, Some(true));
        }

        #[test]
        fn ste_vec_query_needle_deserializes_from_wire() {
            // The containment needle is what an EQL SQL function actually emits,
            // so decode it straight from hand-written wire JSON (no `c` here) and
            // check flatten-over-untagged picks the term by key: `hm` -> Hmac,
            // `op` -> Ope.
            let hm: SteVecQueryEntryV3 =
                serde_json::from_value(serde_json::json!({ "s": "abcd", "hm": "beef" })).unwrap();
            assert_eq!(hm.selector, "abcd");
            assert!(matches!(
                hm.term,
                SteVecEntryTermV3::Hmac { ref hmac_256 } if hmac_256 == "beef"
            ));

            let op: SteVecQueryEntryV3 =
                serde_json::from_value(serde_json::json!({ "s": "ef01", "op": "00ff" })).unwrap();
            assert!(matches!(
                op.term,
                SteVecEntryTermV3::Ope { ref ope_cllw } if ope_cllw == "00ff"
            ));

            let needle_wire = serde_json::json!({
                "sv": [
                    { "s": "abcd", "hm": "beef" },
                    { "s": "ef01", "op": "00ff" },
                ],
            });
            let needle: SteVecQueryPayloadV3 = serde_json::from_value(needle_wire.clone()).unwrap();
            assert_eq!(
                serde_json::to_value(&needle).unwrap(),
                needle_wire,
                "needle must round-trip byte-for-byte"
            );
        }

        // ---- `to_eql_query_payload_v3` SteQueryVec / error branches ----
        //
        // `jsonb_containment_needle_is_bare_sv_array` builds the needle by hand,
        // so the assembler's `SteQueryVec` branch — and the
        // `SteQueryVec::into_iter` / `QueryEntry::into_parts` plumbing added in
        // this PR — is never driven end-to-end. These tests build a query vec (as
        // the ste_vec indexer would, via its wire form) and run it through the
        // assembler, then cover the three error branches.
        //
        // A `QueryEntry<16>` is `(TokenizedSelector<16>, EncryptedSteVecTerm)`,
        // whose fields are `pub(super)`; tests construct it through `Deserialize`
        // (selector = 16-byte hex, term = `{hm}`/`{op}`/`{oc}`) rather than
        // naming those types.

        #[test]
        fn query_containment_needle_built_from_ste_query_vec() {
            let query_vec: SteQueryVec<16> = serde_json::from_value(serde_json::json!([
                ["aa".repeat(16), { "hm": "11".repeat(16) }],
                ["bb".repeat(16), { "op": "00ff" }],
            ]))
            .unwrap();

            let out = to_eql_query_payload_v3(
                IndexTerm::SteQueryVec(query_vec),
                Identifier::new("docs", "attrs"),
            )
            .unwrap();
            assert!(matches!(out, EqlQueryPayloadV3::SteVec(_)));

            let v = serde_json::to_value(&out).unwrap();
            assert!(v.get("v").is_none(), "the needle is not enveloped");
            assert!(v.get("i").is_none());
            let sv = v["sv"].as_array().unwrap();
            assert_eq!(sv.len(), 2);
            assert_eq!(sv[0]["s"], "aa".repeat(16));
            assert_eq!(sv[0]["hm"], "11".repeat(16));
            assert_eq!(sv[1]["s"], "bb".repeat(16));
            assert_eq!(sv[1]["op"], "00ff");
        }

        #[test]
        fn query_containment_needle_rejects_ore_entry() {
            // An ORE (`oc`, Standard-mode) term has no v3 representation; the
            // error must propagate out of the per-entry `map` (through
            // `into_parts`) rather than panic or be dropped.
            let query_vec: SteQueryVec<16> = serde_json::from_value(serde_json::json!([
                ["aa".repeat(16), { "oc": "4243" }],
            ]))
            .unwrap();

            let out = to_eql_query_payload_v3(
                IndexTerm::SteQueryVec(query_vec),
                Identifier::new("docs", "attrs"),
            );
            assert!(matches!(out, Err(EqlError::UnsupportedSteVecOreInV3)));
        }

        #[test]
        fn query_bare_ste_vec_term_is_unsupported() {
            // A single stored sv term has no v3 query-operand shape.
            let out = to_eql_query_payload_v3(
                IndexTerm::SteVecTerm(term(serde_json::json!({ "hm": "11".repeat(16) }))),
                Identifier::new("docs", "attrs"),
            );
            assert!(matches!(out, Err(EqlError::UnsupportedV3QueryTerm)));
        }

        #[test]
        fn query_binary_vec_and_null_are_invalid_index_terms() {
            let binary_vec = to_eql_query_payload_v3(
                IndexTerm::BinaryVec(vec![vec![1, 2], vec![3, 4]]),
                Identifier::new("t", "c"),
            );
            assert!(matches!(binary_vec, Err(EqlError::InvalidIndexTerm)));

            let null = to_eql_query_payload_v3(IndexTerm::Null, Identifier::new("t", "c"));
            assert!(matches!(null, Err(EqlError::InvalidIndexTerm)));
        }
    }

    #[test]
    fn store_scalar_ope_fixed_term_is_carried_on_op() {
        // A scalar column with an `ope` index produces an OpeFixed term for
        // numeric plaintexts. EQL v2.3 carries an order-preserving CLLW OPE
        // ciphertext under the scalar-root `op` key (the same key used at
        // sv-element scope), as a single hex string. Storing it lets a
        // bytewise-OPE compare branch order the column correctly.
        let identifier = Identifier::new("orders", "total");
        let encrypted = Encrypted::Record(
            fake_record("orders/total"),
            vec![IndexTerm::OpeFixed(vec![0xAB; 65])],
        );

        let eql = to_eql_ciphertext(encrypted, &identifier).unwrap();
        let payload = match eql {
            EqlCiphertext::Encrypted(p) => p,
            other => panic!("expected Encrypted payload, got {other:?}"),
        };
        assert_eq!(
            payload.ope_cllw.as_deref(),
            Some(hex::encode([0xAB; 65]).as_str())
        );

        let value = serde_json::to_value(&payload).unwrap();
        assert_eq!(value["op"], hex::encode([0xAB; 65]));
    }

    #[test]
    fn store_scalar_ope_variable_term_is_carried_on_op() {
        // A scalar TEXT column with an `ope` index produces an OpeVariable term,
        // also carried under `op` as a single hex string.
        let identifier = Identifier::new("orders", "name");
        let encrypted = Encrypted::Record(
            fake_record("orders/name"),
            vec![IndexTerm::OpeVariable(vec![0xCD; 40])],
        );

        let eql = to_eql_ciphertext(encrypted, &identifier).unwrap();
        let payload = match eql {
            EqlCiphertext::Encrypted(p) => p,
            other => panic!("expected Encrypted payload, got {other:?}"),
        };
        assert_eq!(
            payload.ope_cllw.as_deref(),
            Some(hex::encode([0xCD; 40]).as_str())
        );

        let value = serde_json::to_value(&payload).unwrap();
        assert_eq!(value["op"], hex::encode([0xCD; 40]));
    }

    #[test]
    fn store_scalar_ore_term_still_succeeds() {
        // Regression guard: Block ORE (`ob`) remains the supported scalar
        // orderable term and must keep round-tripping into the payload.
        let identifier = Identifier::new("orders", "total");
        let encrypted = Encrypted::Record(
            fake_record("orders/total"),
            vec![IndexTerm::OreFull(vec![0xEF; 16])],
        );

        let eql = to_eql_ciphertext(encrypted, &identifier).unwrap();
        match eql {
            EqlCiphertext::Encrypted(p) => {
                assert_eq!(p.ore_block_u64_8_256, Some(vec![hex::encode([0xEF; 16])]));
                assert_eq!(p.ope_cllw, None);
            }
            other => panic!("expected Encrypted payload, got {other:?}"),
        }
    }

    #[test]
    fn query_scalar_ope_fixed_term_is_carried_on_op() {
        // Range / ORDER BY query-mode encryption on a scalar `ope` column must
        // carry the OPE query operand under `op` so EQL can compare it against
        // the stored `op` ciphertext.
        let identifier = Identifier::new("orders", "total");

        let payload =
            to_eql_query_payload(IndexTerm::OpeFixed(vec![0xAB; 65]), identifier.clone()).unwrap();

        match payload {
            EqlQueryPayload::Encrypted(EncryptedQueryPayload {
                term: RootQueryTerm::Ope { ope_cllw },
                ..
            }) => assert_eq!(ope_cllw, hex::encode([0xAB; 65])),
            other => panic!("expected Encrypted Ope query term, got {other:?}"),
        }
    }

    #[test]
    fn query_scalar_ope_variable_term_is_carried_on_op() {
        let identifier = Identifier::new("orders", "name");

        let payload =
            to_eql_query_payload(IndexTerm::OpeVariable(vec![0xCD; 40]), identifier).unwrap();

        match payload {
            EqlQueryPayload::Encrypted(EncryptedQueryPayload {
                term: RootQueryTerm::Ope { ope_cllw },
                ..
            }) => assert_eq!(ope_cllw, hex::encode([0xCD; 40])),
            other => panic!("expected Encrypted Ope query term, got {other:?}"),
        }
    }

    #[test]
    fn root_query_term_ope_round_trips_under_untagged() {
        // The untagged RootQueryTerm enum must deserialize `op` back to the Ope
        // variant without colliding with `hm` / `bf` / `ob`.
        let term: RootQueryTerm =
            serde_json::from_value(serde_json::json!({ "op": "abcd" })).unwrap();
        match term {
            RootQueryTerm::Ope { ope_cllw } => assert_eq!(ope_cllw, "abcd"),
            other => panic!("expected Ope, got {other:?}"),
        }
    }

    #[test]
    fn ste_vec_entry_term_round_trips_under_flatten() {
        let term: SteVecEntryTerm =
            serde_json::from_value(serde_json::json!({ "hm": "deadbeef" })).unwrap();
        match term {
            SteVecEntryTerm::Hmac { hmac_256 } => assert_eq!(hmac_256, "deadbeef"),
            other => panic!("expected Hmac, got {other:?}"),
        }

        let term: SteVecEntryTerm =
            serde_json::from_value(serde_json::json!({ "oc": "cafebabe" })).unwrap();
        match term {
            SteVecEntryTerm::OreCllw { ore_cllw_8 } => assert_eq!(ore_cllw_8, "cafebabe"),
            other => panic!("expected OreCllw, got {other:?}"),
        }

        let term: SteVecEntryTerm =
            serde_json::from_value(serde_json::json!({ "op": "feedface" })).unwrap();
        match term {
            SteVecEntryTerm::Ope { ope_cllw } => assert_eq!(ope_cllw, "feedface"),
            other => panic!("expected Ope, got {other:?}"),
        }
    }

    /// Indexes `json` under a ste_vec config that omits `mode`, then maps the
    /// encrypted entries onto their EQL wire shape exactly as `to_eql_ciphertext`
    /// does. Needs no ZeroKMS: the index key and data key are both fixtures.
    fn sv_wire_entries_for_defaulted_mode(json: serde_json::Value) -> Vec<serde_json::Value> {
        use crate::encryption::{JsonIndexer, JsonIndexerOptions};
        use crate::zerokms::DataKeyWithTag;

        // The shape a ste_vec config written before `mode` existed has on disk.
        let index_type: IndexType = serde_json::from_value(serde_json::json!({
            "kind": "ste-vec",
            "prefix": "cs_ste_vec_v1",
        }))
        .expect("mode-less ste_vec config parses");

        let opts = JsonIndexerOptions::try_from(&index_type).expect("SteVec config -> options");
        let indexer = JsonIndexer::new(opts);

        indexer
            .index(json, &IndexKey::from([0; 32]))
            .expect("indexing succeeds")
            .encrypt(DataKeyWithTag::default())
            .expect("encryption succeeds")
            .into_iter()
            .map(
                |EncryptedEntry {
                     tokenized_selector,
                     term,
                     record,
                     parent_is_array,
                 }| {
                    serde_json::to_value(SteVecEntry {
                        selector: hex::encode(tokenized_selector.as_bytes()),
                        ciphertext: record,
                        is_array: Some(parent_is_array),
                        term: ste_vec_term_from_encrypted(term),
                    })
                    .expect("sv entry serializes")
                },
            )
            .collect()
    }

    #[test]
    fn defaulted_ste_vec_config_puts_every_ordering_term_on_op() {
        // Joins the two halves the other tests cover separately: a config that
        // omits `mode` resolves to Compat, and a Compat index puts each entry's
        // ordering term on `op`. Without this, `SteVecMode::default() == Compat`
        // and "Compat emits `op`" are both pinned but nothing proves the default
        // path actually reaches the `op` wire key.
        let entries = sv_wire_entries_for_defaulted_mode(serde_json::json!({
            "email": "alice@example.com",
            "metrics": { "login_count": 42, "score": 95.5 },
            "tags": ["premium", "active"],
        }));

        assert!(
            entries.iter().any(|e| e.get("op").is_some()),
            "string / number leaves carry an `op` ordering term, got {entries:?}"
        );
        assert!(
            entries.iter().any(|e| e.get("hm").is_some()),
            "array / object root placeholders still ride `hm`, got {entries:?}"
        );
        assert!(
            entries
                .iter()
                .all(|e| e.get("hm").is_some() ^ e.get("op").is_some()),
            "every Compat sv entry carries exactly one of `hm` / `op`, got {entries:?}"
        );
        assert!(
            entries.iter().all(|e| e.get("oc").is_none()),
            "no Compat sv entry may carry the legacy `oc` key, got {entries:?}"
        );
    }

    #[test]
    fn ste_vec_entry_term_untagged_precedence_is_hm_then_oc_then_op() {
        // `SteVecEntryTerm` is `#[serde(untagged)]`, so a malformed entry
        // carrying more than one term key resolves to the first matching
        // variant in declaration order and silently drops the rest. A
        // well-formed entry never carries two terms; this pins the resolution
        // order so a variant reorder can't silently change which term wins.
        let term: SteVecEntryTerm =
            serde_json::from_value(serde_json::json!({ "hm": "aa", "op": "bb" })).unwrap();
        assert!(
            matches!(term, SteVecEntryTerm::Hmac { .. }),
            "`hm` precedes `op`, got {term:?}"
        );

        let term: SteVecEntryTerm =
            serde_json::from_value(serde_json::json!({ "oc": "cc", "op": "bb" })).unwrap();
        assert!(
            matches!(term, SteVecEntryTerm::OreCllw { .. }),
            "`oc` precedes `op`, got {term:?}"
        );
    }

    #[test]
    fn ste_vec_entry_with_ope_term_round_trips_through_json() {
        // The read path stacks `#[serde(flatten)]` on the term, `with =
        // "formats::mp_base85"` on `c`, and `#[serde(untagged)]` on
        // `SteVecEntryTerm` — the shape `decrypt_eql` parses out of a stored
        // `sv` payload. Serializing is not enough: a flattened `with` adapter
        // sees a buffered `Value` rather than the original stream, so decode
        // needs its own guard.
        use cllw_ore::OpeCllw8VariableV1;

        let entry = SteVecEntry {
            selector: "00".repeat(16),
            ciphertext: fake_record("users/profile"),
            is_array: Some(false),
            term: ste_vec_term_from_encrypted(EncryptedSteVecTerm::Compat(
                EncryptedSteVecTermCompat::Ope(OpeCllw8VariableV1::from_bytes(vec![0xAB; 10])),
            )),
        };

        let back: SteVecEntry =
            serde_json::from_value(serde_json::to_value(&entry).unwrap()).unwrap();

        assert_eq!(back.selector, "00".repeat(16));
        assert_eq!(back.is_array, Some(false));
        match back.term {
            SteVecEntryTerm::Ope { ope_cllw } => assert_eq!(ope_cllw, hex::encode([0xAB; 10])),
            other => panic!("expected Ope, got {other:?}"),
        }
    }

    #[test]
    fn ste_vec_query_term_ope_round_trips_under_untagged() {
        // Mirror of `root_query_term_ope_round_trips_under_untagged` for the
        // sv-level term: `op` must not be captured by `s` / `hm` / `oc` / `q`.
        use cllw_ore::OpeCllw8VariableV1;

        let term: SteVecQueryTerm =
            serde_json::from_value(serde_json::json!({ "op": "feedface" })).unwrap();
        match term {
            SteVecQueryTerm::Ope { ope_cllw } => assert_eq!(ope_cllw, "feedface"),
            other => panic!("expected Ope, got {other:?}"),
        }

        // And through the `k`-tagged envelope, which is how the proxy reads it.
        let payload = to_eql_query_payload(
            IndexTerm::SteVecTerm(EncryptedSteVecTerm::Compat(EncryptedSteVecTermCompat::Ope(
                OpeCllw8VariableV1::from_bytes(vec![0xEF; 12]),
            ))),
            Identifier::new("users", "profile"),
        )
        .unwrap();
        let back: EqlQueryPayload =
            serde_json::from_value(serde_json::to_value(&payload).unwrap()).unwrap();
        assert!(matches!(
            back,
            EqlQueryPayload::SteVec(SteVecQueryPayload {
                term: SteVecQueryTerm::Ope { .. },
                ..
            })
        ));
    }

    #[test]
    fn store_ste_vec_compat_ope_term_is_carried_on_op() {
        // A Compat-mode SteVec entry's ordering term must serialize under the
        // sv-element `op` key (a single hex string, same encoding as the
        // scalar-root `op`), never under the legacy `oc` key. EQL v3 accepts
        // `hm` XOR `op` on sv entries and rejects `oc`.
        use cllw_ore::OpeCllw8VariableV1;

        let term = ste_vec_term_from_encrypted(EncryptedSteVecTerm::Compat(
            EncryptedSteVecTermCompat::Ope(OpeCllw8VariableV1::from_bytes(vec![0xAB; 10])),
        ));
        match &term {
            SteVecEntryTerm::Ope { ope_cllw } => assert_eq!(ope_cllw, &hex::encode([0xAB; 10])),
            other => panic!("expected Ope, got {other:?}"),
        }

        let entry = SteVecEntry {
            selector: "00".repeat(16),
            ciphertext: fake_record("users/profile"),
            is_array: Some(false),
            term,
        };
        let value = serde_json::to_value(&entry).unwrap();
        assert_eq!(value["op"], hex::encode([0xAB; 10]));
        assert!(
            value.get("oc").is_none(),
            "Compat OPE must not ride the legacy `oc` key"
        );
    }

    #[test]
    fn store_ste_vec_standard_ore_term_still_carried_on_oc() {
        // Regression guard: Standard-mode (legacy v2 protocol) entries keep
        // emitting CLLW ORE under `oc`.
        use cllw_ore::OreCllw8VariableV1;

        let term = ste_vec_term_from_encrypted(EncryptedSteVecTerm::Standard(
            EncryptedSteVecTermStandard::Ore(OreCllw8VariableV1::from(vec![0xCD; 19])),
        ));
        match &term {
            SteVecEntryTerm::OreCllw { ore_cllw_8 } => {
                assert_eq!(ore_cllw_8, &hex::encode([0xCD; 19]))
            }
            other => panic!("expected OreCllw, got {other:?}"),
        }

        let entry = SteVecEntry {
            selector: "00".repeat(16),
            ciphertext: fake_record("users/profile"),
            is_array: Some(false),
            term,
        };
        let value = serde_json::to_value(&entry).unwrap();
        assert_eq!(value["oc"], hex::encode([0xCD; 19]));
        assert!(value.get("op").is_none());
    }

    #[test]
    fn query_ste_vec_compat_ope_term_is_carried_on_op() {
        // A Compat-mode SteVec query term (e.g. a `->` / range operand) must
        // serialize under `op` on the sv query payload so EQL can compare it
        // bytewise against stored sv-element `op` terms.
        use cllw_ore::OpeCllw8VariableV1;

        let identifier = Identifier::new("users", "profile");
        let payload = to_eql_query_payload(
            IndexTerm::SteVecTerm(EncryptedSteVecTerm::Compat(EncryptedSteVecTermCompat::Ope(
                OpeCllw8VariableV1::from_bytes(vec![0xEF; 12]),
            ))),
            identifier,
        )
        .unwrap();

        match &payload {
            EqlQueryPayload::SteVec(SteVecQueryPayload {
                term: SteVecQueryTerm::Ope { ope_cllw },
                ..
            }) => assert_eq!(ope_cllw, &hex::encode([0xEF; 12])),
            other => panic!("expected SteVec Ope query term, got {other:?}"),
        }

        let value = serde_json::to_value(&payload).unwrap();
        assert_eq!(value["k"], "sv");
        assert_eq!(value["op"], hex::encode([0xEF; 12]));
        assert!(value.get("oc").is_none());
    }

    #[test]
    fn query_ste_vec_standard_ore_term_still_carried_on_oc() {
        use cllw_ore::OreCllw8VariableV1;

        let identifier = Identifier::new("users", "profile");
        let payload = to_eql_query_payload(
            IndexTerm::SteVecTerm(EncryptedSteVecTerm::Standard(
                EncryptedSteVecTermStandard::Ore(OreCllw8VariableV1::from(vec![0x42; 19])),
            )),
            identifier,
        )
        .unwrap();

        match &payload {
            EqlQueryPayload::SteVec(SteVecQueryPayload {
                term: SteVecQueryTerm::OreCllw { ore_cllw_8 },
                ..
            }) => assert_eq!(ore_cllw_8, &hex::encode([0x42; 19])),
            other => panic!("expected SteVec OreCllw query term, got {other:?}"),
        }

        let value = serde_json::to_value(&payload).unwrap();
        assert_eq!(value["k"], "sv");
        assert_eq!(value["oc"], hex::encode([0x42; 19]));
        assert!(value.get("op").is_none());
    }

    #[test]
    fn query_payload_root_serializes_with_k_ct() {
        let payload = EqlQueryPayload::Encrypted(EncryptedQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier: Identifier::new("users", "name"),
            term: RootQueryTerm::BloomFilter {
                bloom_filter: vec![1, 2, 3],
            },
        });

        let value = serde_json::to_value(&payload).unwrap();
        assert_eq!(value["k"], "ct");
        assert_eq!(value["bf"], serde_json::json!([1, 2, 3]));
        assert!(value.get("c").is_none(), "query payloads omit ciphertext");
    }

    #[test]
    fn eql_output_round_trips_untagged() {
        let query = EqlOutput::Query(EqlQueryPayload::Encrypted(EncryptedQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier: Identifier::new("users", "name"),
            term: RootQueryTerm::Hmac {
                hmac_256: "deadbeef".into(),
            },
        }));

        let value = serde_json::to_value(&query).unwrap();
        assert_eq!(value["k"], "ct");
        assert_eq!(value["hm"], "deadbeef");
        assert!(
            value.get("c").is_none(),
            "Query payload must not serialize a ciphertext — if this fires, Store won disambiguation"
        );

        let back: EqlOutput = serde_json::from_value(value).unwrap();
        match back {
            EqlOutput::Query(EqlQueryPayload::Encrypted(p)) => {
                assert_eq!(p.identifier, Identifier::new("users", "name"));
                match p.term {
                    RootQueryTerm::Hmac { hmac_256 } => assert_eq!(hmac_256, "deadbeef"),
                    other => panic!("expected Hmac term, got {other:?}"),
                }
            }
            other => panic!("expected EqlOutput::Query(Encrypted), got {other:?}"),
        }

        let record = EncryptedRecord {
            iv: Default::default(),
            ciphertext: vec![1; 16],
            tag: vec![2; 16],
            descriptor: "users/email".to_string(),
            keyset_id: Some(Uuid::nil()),
            decryption_policy: None,
        };
        let store = EqlOutput::Store(EqlCiphertext::Encrypted(EncryptedPayload {
            version: EQL_SCHEMA_VERSION,
            identifier: Identifier::new("users", "email"),
            ciphertext: record,
            hmac_256: Some("cafebabe".into()),
            bloom_filter: None,
            ore_block_u64_8_256: None,
            ope_cllw: None,
        }));

        let value = serde_json::to_value(&store).unwrap();
        assert_eq!(value["k"], "ct");
        assert!(value.get("c").is_some());

        let back: EqlOutput = serde_json::from_value(value).unwrap();
        match back {
            EqlOutput::Store(EqlCiphertext::Encrypted(p)) => {
                assert_eq!(p.identifier, Identifier::new("users", "email"));
                assert_eq!(p.hmac_256.as_deref(), Some("cafebabe"));
            }
            other => panic!("expected EqlOutput::Store(Encrypted), got {other:?}"),
        }
    }

    #[test]
    fn query_payload_ste_vec_selector_serializes_with_k_sv() {
        let payload = EqlQueryPayload::SteVec(SteVecQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier: Identifier::new("users", "profile"),
            term: SteVecQueryTerm::Selector {
                selector: "abcd".into(),
            },
        });

        let value = serde_json::to_value(&payload).unwrap();
        assert_eq!(value["k"], "sv");
        assert_eq!(value["s"], "abcd");
    }

    #[test]
    fn eql_output_query_hmac_renders_exact_json() {
        let query = EqlOutput::Query(EqlQueryPayload::Encrypted(EncryptedQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier: Identifier::new("users", "name"),
            term: RootQueryTerm::Hmac {
                hmac_256: "deadbeef".into(),
            },
        }));

        assert_eq!(
            serde_json::to_value(&query).unwrap(),
            serde_json::json!({
                "k": "ct",
                "v": EQL_SCHEMA_VERSION,
                "i": { "t": "users", "c": "name" },
                "hm": "deadbeef",
            })
        );
    }

    #[test]
    fn eql_output_query_bloom_filter_renders_exact_json() {
        let query = EqlOutput::Query(EqlQueryPayload::Encrypted(EncryptedQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier: Identifier::new("users", "name"),
            term: RootQueryTerm::BloomFilter {
                bloom_filter: vec![1, 2, 3],
            },
        }));

        assert_eq!(
            serde_json::to_value(&query).unwrap(),
            serde_json::json!({
                "k": "ct",
                "v": EQL_SCHEMA_VERSION,
                "i": { "t": "users", "c": "name" },
                "bf": [1, 2, 3],
            })
        );
    }

    #[test]
    fn eql_output_query_ste_vec_selector_renders_exact_json() {
        let query = EqlOutput::Query(EqlQueryPayload::SteVec(SteVecQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier: Identifier::new("users", "profile"),
            term: SteVecQueryTerm::Selector {
                selector: "abcd".into(),
            },
        }));

        assert_eq!(
            serde_json::to_value(&query).unwrap(),
            serde_json::json!({
                "k": "sv",
                "v": EQL_SCHEMA_VERSION,
                "i": { "t": "users", "c": "profile" },
                "s": "abcd",
            })
        );
    }

    #[test]
    fn eql_output_store_encrypted_renders_exact_json() {
        let record = EncryptedRecord {
            iv: Default::default(),
            ciphertext: vec![1; 16],
            tag: vec![2; 16],
            descriptor: "users/email".to_string(),
            keyset_id: Some(Uuid::nil()),
            decryption_policy: None,
        };
        let store = EqlOutput::Store(EqlCiphertext::Encrypted(EncryptedPayload {
            version: EQL_SCHEMA_VERSION,
            identifier: Identifier::new("users", "email"),
            ciphertext: record.clone(),
            hmac_256: Some("cafebabe".into()),
            bloom_filter: None,
            ore_block_u64_8_256: None,
            ope_cllw: None,
        }));

        // `c` is opaque MessagePack+Base85; render it once via the canonical encoder
        // so the assertion below is a single literal of the full payload shape.
        let encoded_c = record.to_mp_base85().unwrap();

        assert_eq!(
            serde_json::to_value(&store).unwrap(),
            serde_json::json!({
                "k": "ct",
                "v": EQL_SCHEMA_VERSION,
                "i": { "t": "users", "c": "email" },
                "c": encoded_c,
                "hm": "cafebabe",
            })
        );
    }
}