cipherstash-client 0.37.1

//! Types for representing EQL payloads, and encryption/decryption functions.
//!
//! The serialization shapes here match the EQL v2.3 schema at
//! `cipherstash/encrypt-query-language/docs/reference/schema/eql-payload-v2.3.schema.json`.

#![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;

/// 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,
{
    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?,
    );

    let encrypted = targets
        .into_iter()
        .map(|target| -> Result<EqlOutput, 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"),
                    )?;

                    Ok(EqlOutput::Store(to_eql_ciphertext(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)?;

                    Ok(EqlOutput::Query(to_eql_query_payload(
                        index_term, identifier,
                    )?))
                }
            }
        })
        .collect::<Result<Vec<_>, _>>()?;

    Ok(encrypted)
}

/// 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`.
#[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`). Per the v2.3 schema, `c` is required and the
/// orderable term at this scope is Block ORE (`ob`); CLLW ORE (`oc`) lives
/// only on STE vector 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. 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>>,
}

/// 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`].
///
/// Per the v2.3 schema, every element carries `s` (selector), `c`
/// (ciphertext), and exactly one of [`SteVecEntryTerm::Hmac`] or
/// [`SteVecEntryTerm::OreCllw`].
#[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` or `oc`.
    #[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. `OreCllw` (`oc`) covers string and number leaves.
/// Domain separation between numeric and string `oc` 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` ciphertext alone.
#[derive(Clone, Debug, Deserialize, Serialize)]
#[serde(untagged)]
pub enum SteVecEntryTerm {
    Hmac {
        #[serde(rename = "hm")]
        hmac_256: String,
    },
    OreCllw {
        #[serde(rename = "oc")]
        ore_cllw_8: 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`).
    #[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`, or `ob`.
#[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>,
    },
}

/// 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`, 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.
#[derive(Debug, Deserialize, Serialize)]
#[serde(untagged)]
pub enum SteVecQueryTerm {
    Selector {
        #[serde(rename = "s")]
        selector: String,
    },
    Hmac {
        #[serde(rename = "hm")]
        hmac_256: String,
    },
    OreCllw {
        #[serde(rename = "oc")]
        ore_cllw_8: 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>,
    },
}

/// 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,
            };

            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`].
///
/// Terms with no v2.3 root-scope representation (e.g. `BinaryVec`, OPE) are
/// skipped silently — the encryption pipeline is responsible for not
/// configuring index types whose terms are unsupported under v2.3.
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());
        }
        // No v2.3 root-scope field for these. OPE was removed; STE-vec terms
        // belong on `sv` elements; the rest are unused by EQL.
        IndexTerm::OpeFixed(_)
        | IndexTerm::OpeVariable(_)
        | IndexTerm::BinaryVec(_)
        | IndexTerm::SteVecSelector(_)
        | IndexTerm::SteVecTerm(_)
        | IndexTerm::SteQueryVec(_)
        | IndexTerm::Null => {}
    }
}

/// Maps an [`EncryptedSteVecTerm`] onto an [`SteVecEntryTerm`].
///
/// Compat-mode OPE bytes are emitted under the `oc` field. The v2.3 schema
/// defines `oc` as CLLW ORE, but both ORE and OPE are CLLW variants and the
/// EQL layer carries the chosen variant's bytes through unchanged; consumers
/// interpret them based on the column's configured `SteVecMode`.
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::OreCllw {
                ore_cllw_8: hex::encode(ope.as_ref()),
            }
        }
    }
}

/// Builds an [`EqlQueryPayload`] from a single query-mode [`IndexTerm`].
///
/// Returns [`EqlError::InvalidIndexTerm`] for terms with no v2.3
/// representation (OPE root terms, accumulator-style 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) => {
            let query_term = match ste_vec_term_from_encrypted(term) {
                SteVecEntryTerm::Hmac { hmac_256 } => SteVecQueryTerm::Hmac { hmac_256 },
                SteVecEntryTerm::OreCllw { ore_cllw_8 } => SteVecQueryTerm::OreCllw { ore_cllw_8 },
            };
            Ok(EqlQueryPayload::SteVec(SteVecQueryPayload {
                version: EQL_SCHEMA_VERSION,
                identifier,
                term: query_term,
            }))
        }
        IndexTerm::SteQueryVec(query_vec) => Ok(EqlQueryPayload::SteVec(SteVecQueryPayload {
            version: EQL_SCHEMA_VERSION,
            identifier,
            term: SteVecQueryTerm::Containment { query_vec },
        })),
        IndexTerm::OpeFixed(_)
        | IndexTerm::OpeVariable(_)
        | IndexTerm::BinaryVec(_)
        | IndexTerm::Null => Err(EqlError::InvalidIndexTerm),
    }
}

/// 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,

    /// 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,
        });

        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());
    }

    #[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:?}"),
        }
    }

    #[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,
        }));

        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,
        }));

        // `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",
            })
        );
    }
}