orion-server 0.3.0

//! Elasticsearch rendering.
//!
//! Walks a [`Cond`] + [`QuerySpec`] into an ES Query DSL search body over the same
//! IR. Every predicate is emitted in **filter context** (`bool.filter` /
//! `bool.must_not`, `should` + `minimum_should_match` for `or`) so results are
//! set-equivalent to SQL/Mongo, never relevance-ranked (§5.2). Relations render as
//! `nested` / `has_child` (§4.2). `all` and deep pagination are capability-gated
//! (§5.6, §5.8).
//!
//! Also renders the write dialect: [`render_write`] turns a [`ResolvedWrite`] into
//! an [`EsWrite`] (`_bulk` / `_update_by_query` / `_delete_by_query` / `_update` /
//! `_doc?op_type=create`), reusing the same condition rendering for the
//! update/delete filter.

use serde_json::{Value as Json, json};

use crate::query::error::QueryError;
use crate::query::ir::{CmpOp, Cond, EsStorage, FieldRef, Quant, TextOp, Value};
use crate::query::spec::{QuerySpec, SortDir};
use crate::query::write::{ConflictAction, ResolvedWrite, WriteError, WriteOp};

/// ES bounds `from + size` by `index.max_result_window` (default 10k). Beyond it
/// we raise a capability error rather than return a truncated page (§5.8).
const MAX_RESULT_WINDOW: u64 = 10_000;

/// A rendered Elasticsearch search: the index plus the request body.
#[derive(Debug, Clone, PartialEq)]
pub struct EsQuery {
    pub index: String,
    pub body: Json,
}

/// Build an `EsQuery` from the envelope and lowered condition, enforcing the
/// page-size bounds and the deep-pagination cap.
pub fn render(
    spec: &QuerySpec,
    cond: &Cond,
    index: &str,
    default_limit: u64,
    max_limit: u64,
) -> Result<EsQuery, QueryError> {
    let size = match spec.limit {
        Some(l) if l > max_limit => {
            return Err(QueryError::LimitExceeded {
                requested: l,
                max: max_limit,
            });
        }
        Some(l) => l,
        None => default_limit.min(max_limit),
    };
    let from = spec.skip.unwrap_or(0);
    if from.saturating_add(size) > MAX_RESULT_WINDOW {
        return Err(QueryError::FeatureUnsupportedByTarget {
            feature: format!(
                "deep pagination (from {from} + size {size} exceeds max_result_window {MAX_RESULT_WINDOW})"
            ),
            target: "elasticsearch".to_string(),
        });
    }

    let mut body = json!({
        "query": query_json(cond, "")?,
        "size": size,
        "from": from,
    });

    if !spec.sort.is_empty() {
        let sort: Vec<Json> = spec
            .sort
            .iter()
            .map(|k| {
                let (order, missing) = match k.dir {
                    // nulls last on asc, first on desc (§5.7).
                    SortDir::Asc => ("asc", "_last"),
                    SortDir::Desc => ("desc", "_first"),
                };
                json!({ &k.field: { "order": order, "missing": missing } })
            })
            .collect();
        body["sort"] = Json::Array(sort);
    }

    if !spec.fields.is_empty() {
        body["_source"] = json!(spec.fields);
    }

    Ok(EsQuery {
        index: index.to_string(),
        body,
    })
}

/// Render a `Cond` into an ES query clause (filter context). `prefix` qualifies
/// bare field names inside a `nested` query (e.g. `"orders."`).
fn query_json(cond: &Cond, prefix: &str) -> Result<Json, QueryError> {
    Ok(match cond {
        Cond::True => json!({ "match_all": {} }),
        Cond::False => json!({ "match_none": {} }),
        Cond::And(cs) => json!({ "bool": { "filter": clauses(cs, prefix)? } }),
        Cond::Or(cs) => {
            json!({ "bool": { "should": clauses(cs, prefix)?, "minimum_should_match": 1 } })
        }
        Cond::Not(inner) => json!({ "bool": { "must_not": [query_json(inner, prefix)?] } }),
        Cond::Compare { field, op, value } => {
            let f = fname(field, prefix);
            let v = to_json(value);
            match op {
                CmpOp::Eq => json!({ "term": { f: v } }),
                CmpOp::Ne => json!({ "bool": { "must_not": [{ "term": { f: v } }] } }),
                CmpOp::Lt => json!({ "range": { f: { "lt": v } } }),
                CmpOp::Le => json!({ "range": { f: { "lte": v } } }),
                CmpOp::Gt => json!({ "range": { f: { "gt": v } } }),
                CmpOp::Ge => json!({ "range": { f: { "gte": v } } }),
            }
        }
        Cond::In {
            field,
            values,
            negated,
        } => {
            let f = fname(field, prefix);
            let vs: Vec<Json> = values.iter().map(to_json).collect();
            let terms = json!({ "terms": { f: vs } });
            if *negated {
                json!({ "bool": { "must_not": [terms] } })
            } else {
                terms
            }
        }
        Cond::IsNull { field, negated } => {
            let exists = json!({ "exists": { "field": fname(field, prefix) } });
            if *negated {
                exists
            } else {
                json!({ "bool": { "must_not": [exists] } })
            }
        }
        Cond::Between {
            field,
            low,
            high,
            low_incl,
            high_incl,
            negated,
        } => {
            let f = fname(field, prefix);
            let mut range = serde_json::Map::new();
            range.insert(
                if *low_incl { "gte" } else { "gt" }.to_string(),
                to_json(low),
            );
            range.insert(
                if *high_incl { "lte" } else { "lt" }.to_string(),
                to_json(high),
            );
            let clause = json!({ "range": { f: Json::Object(range) } });
            if *negated {
                json!({ "bool": { "must_not": [clause] } })
            } else {
                clause
            }
        }
        Cond::Text {
            field,
            op,
            pattern,
            ci,
        } => {
            let f = fname(field, prefix);
            match op {
                TextOp::StartsWith => {
                    let mut v = json!({ "value": pattern });
                    if *ci {
                        v["case_insensitive"] = Json::Bool(true);
                    }
                    json!({ "prefix": { f: v } })
                }
                TextOp::EndsWith => wildcard(&f, format!("*{}", wildcard_escape(pattern)), *ci),
                TextOp::Contains => wildcard(&f, format!("*{}*", wildcard_escape(pattern)), *ci),
            }
        }
        Cond::Rel { quant, rel, cond } => rel_json(*quant, &rel.name, rel.es, cond)?,
    })
}

/// Render a relation predicate as a `nested` or `has_child` query.
fn rel_json(
    quant: Quant,
    name: &str,
    storage: EsStorage,
    inner: &Cond,
) -> Result<Json, QueryError> {
    if quant == Quant::All {
        // `all` over ES nesting needs must_not-of-negation with empty-relation
        // caveats; not set-equivalent without an explicit opt-in (§5.6).
        return Err(QueryError::FeatureUnsupportedByTarget {
            feature: format!("`all` over relation '{name}'"),
            target: "elasticsearch".to_string(),
        });
    }
    let positive = match storage {
        EsStorage::Nested => {
            // Inside a nested query, fields are qualified with the path.
            let q = query_json(inner, &format!("{name}."))?;
            json!({ "nested": { "path": name, "query": q } })
        }
        EsStorage::Child => {
            let q = query_json(inner, "")?;
            json!({ "has_child": { "type": name, "query": q } })
        }
    };
    Ok(match quant {
        Quant::Any => positive,
        Quant::None => json!({ "bool": { "must_not": [positive] } }),
        Quant::All => unreachable!("handled above"),
    })
}

fn clauses(cs: &[Cond], prefix: &str) -> Result<Vec<Json>, QueryError> {
    cs.iter().map(|c| query_json(c, prefix)).collect()
}

fn wildcard(field: &str, value: String, ci: bool) -> Json {
    let mut v = json!({ "value": value });
    if ci {
        v["case_insensitive"] = Json::Bool(true);
    }
    json!({ "wildcard": { field: v } })
}

fn fname(field: &FieldRef, prefix: &str) -> String {
    if prefix.is_empty() {
        field.physical.clone()
    } else {
        format!("{prefix}{}", field.physical)
    }
}

fn to_json(v: &Value) -> Json {
    match v {
        Value::Null => Json::Null,
        Value::Bool(b) => Json::Bool(*b),
        Value::Int(i) => json!(i),
        Value::Float(f) => json!(f),
        Value::Str(s) => Json::String(s.clone()),
        Value::List(items) => Json::Array(items.iter().map(to_json).collect()),
    }
}

/// Escape wildcard metacharacters so user text matches literally in a `wildcard`.
fn wildcard_escape(s: &str) -> String {
    let mut out = String::with_capacity(s.len());
    for c in s.chars() {
        if matches!(c, '*' | '?' | '\\') {
            out.push('\\');
        }
        out.push(c);
    }
    out
}

// ---- Write rendering (insert / update / delete / upsert) ----

/// A rendered Elasticsearch write, ready for the HTTP call the handler makes.
/// Each variant implies the method, path, and refresh parameter; bodies are plain
/// JSON (NDJSON for `_bulk`, assembled by [`bulk_ndjson`]).
#[derive(Debug, Clone, PartialEq)]
pub enum EsWrite {
    /// `POST {index}/_bulk?refresh=wait_for` — one `index` action + source line
    /// per row. A `Some` id becomes the action's `_id`; `None` auto-generates.
    BulkInsert {
        index: String,
        docs: Vec<(Option<String>, Json)>,
    },
    /// `POST {index}/_update_by_query?refresh=true` — query + painless script.
    UpdateByQuery { index: String, body: Json },
    /// `POST {index}/_delete_by_query?refresh=true` — query only.
    DeleteByQuery { index: String, body: Json },
    /// `POST {index}/_update/{id}?refresh=wait_for` — single-document upsert
    /// (`doc` / `doc_as_upsert` / `upsert`).
    UpdateDoc {
        index: String,
        id: String,
        body: Json,
    },
    /// `PUT {index}/_doc/{id}?op_type=create&refresh=wait_for` — insert-if-absent;
    /// an HTTP 409 is the "conflict → do nothing" no-op.
    CreateDoc {
        index: String,
        id: String,
        doc: Json,
    },
}

/// Render a resolved mutation into an [`EsWrite`]. The `filter` of an
/// update/delete reuses the query dialect's [`query_json`]; values become JSON via
/// the same [`to_json`] the read path uses.
///
/// Unlike Mongo there is **no** implicit `id` → `_id` mapping: `_id` is metadata
/// outside `_source`, and a genuine `id` field inside `_source` is legal, so the
/// rename is an explicit schema decision (`{"columns": {"id": {"name": "_id"}}}`).
/// A physical `_id` column is lifted out of the source into the action/path.
pub fn render_write(w: &ResolvedWrite) -> Result<EsWrite, WriteError> {
    if !w.returning.is_empty() {
        return Err(WriteError::FeatureUnsupportedByTarget {
            feature: "returning".to_string(),
            target: "elasticsearch".to_string(),
        });
    }
    Ok(match w.op {
        WriteOp::Insert => {
            let id_idx = w.columns.iter().position(|c| c == "_id");
            let mut docs = Vec::with_capacity(w.rows.len());
            for row in &w.rows {
                let id = match id_idx {
                    Some(i) => id_string(&row[i], "values")?,
                    None => None,
                };
                docs.push((id, source_doc(&w.columns, row, "_id")));
            }
            EsWrite::BulkInsert {
                index: w.table.clone(),
                docs,
            }
        }
        WriteOp::Update => {
            reject_id_in_set(&w.set)?;
            // Field names AND values travel as painless params — nothing
            // user-controlled is spliced into the script source.
            let mut source = String::new();
            let mut params = serde_json::Map::new();
            for (i, (col, v)) in w.set.iter().enumerate() {
                source.push_str(&format!("ctx._source[params.f{i}] = params.v{i};"));
                params.insert(format!("f{i}"), Json::String(col.clone()));
                params.insert(format!("v{i}"), to_json(v));
            }
            EsWrite::UpdateByQuery {
                index: w.table.clone(),
                body: json!({
                    "query": cond_to_query(&w.cond)?,
                    "script": { "lang": "painless", "source": source, "params": params },
                }),
            }
        }
        WriteOp::Delete => EsWrite::DeleteByQuery {
            index: w.table.clone(),
            body: json!({ "query": cond_to_query(&w.cond)? }),
        },
        WriteOp::Upsert => render_es_upsert(w)?,
    })
}

fn render_es_upsert(w: &ResolvedWrite) -> Result<EsWrite, WriteError> {
    let conflict = w
        .conflict
        .as_ref()
        .ok_or_else(|| WriteError::MissingField {
            field: "on_conflict".to_string(),
            op: "upsert".to_string(),
        })?;
    // A single-document upsert keyed on `_id`. Bulk upsert would need one
    // `_update` call per row; deferred (fail loudly, don't guess).
    if w.rows.len() != 1 {
        return Err(WriteError::FeatureUnsupportedByTarget {
            feature: "bulk upsert".to_string(),
            target: "elasticsearch".to_string(),
        });
    }
    // ES has no unique constraints; the only conflict key it can express is the
    // document `_id`.
    if conflict.targets.len() != 1 || conflict.targets[0] != "_id" {
        return Err(WriteError::FeatureUnsupportedByTarget {
            feature: format!(
                "upsert on conflict target [{}] (Elasticsearch keys upserts on the document `_id`; declare a schema rename to \"_id\")",
                conflict.targets.join(", ")
            ),
            target: "elasticsearch".to_string(),
        });
    }
    let row = &w.rows[0];
    let idx = w.columns.iter().position(|c| c == "_id").ok_or_else(|| {
        WriteError::InvalidEnvelope(
            "on_conflict target '_id' must be one of the inserted columns".to_string(),
        )
    })?;
    let id = id_string(&row[idx], "values")?.ok_or_else(|| WriteError::NotRepresentable {
        what: "a null `_id` in an upsert".to_string(),
        at: "values".to_string(),
    })?;
    let doc = source_doc(&w.columns, row, "_id");

    Ok(match conflict.action {
        ConflictAction::Update => {
            reject_id_in_set(&w.set)?;
            if w.set.is_empty() {
                // Overwrite every non-`_id` column on conflict; index the row
                // when absent.
                EsWrite::UpdateDoc {
                    index: w.table.clone(),
                    id,
                    body: json!({ "doc": doc, "doc_as_upsert": true }),
                }
            } else {
                // On conflict apply `set`; on insert index the row overlaid with
                // `set` (Mongo's `$set` + `$setOnInsert` split).
                let mut set_doc = serde_json::Map::new();
                for (col, v) in &w.set {
                    set_doc.insert(col.clone(), to_json(v));
                }
                let mut merged = match &doc {
                    Json::Object(m) => m.clone(),
                    _ => serde_json::Map::new(),
                };
                for (k, v) in &set_doc {
                    merged.insert(k.clone(), v.clone());
                }
                EsWrite::UpdateDoc {
                    index: w.table.clone(),
                    id,
                    body: json!({ "doc": Json::Object(set_doc), "upsert": Json::Object(merged) }),
                }
            }
        }
        ConflictAction::Nothing => EsWrite::CreateDoc {
            index: w.table.clone(),
            id,
            doc,
        },
    })
}

/// Serialise a rendered bulk insert into the `_bulk` NDJSON body. The trailing
/// newline is mandatory — ES rejects bulk bodies without it.
pub fn bulk_ndjson(docs: &[(Option<String>, Json)]) -> String {
    let mut out = String::new();
    for (id, source) in docs {
        let action = match id {
            Some(id) => json!({ "index": { "_id": id } }),
            None => json!({ "index": {} }),
        };
        out.push_str(&action.to_string());
        out.push('\n');
        out.push_str(&source.to_string());
        out.push('\n');
    }
    out
}

/// Lower an optional filter to a query clause (`None` — an acknowledged
/// unfiltered mutation — matches everything).
fn cond_to_query(cond: &Option<Cond>) -> Result<Json, WriteError> {
    Ok(match cond {
        None => json!({ "match_all": {} }),
        Some(c) => query_json(c, "").map_err(WriteError::from)?,
    })
}

/// `_id` is immutable metadata; a `set` touching it cannot be expressed.
fn reject_id_in_set(set: &[(String, Value)]) -> Result<(), WriteError> {
    if set.iter().any(|(c, _)| c == "_id") {
        return Err(WriteError::FeatureUnsupportedByTarget {
            feature: "updating `_id`".to_string(),
            target: "elasticsearch".to_string(),
        });
    }
    Ok(())
}

/// The row as a `_source` document, excluding the `skip` column (`_id` lives in
/// the action/path — ES rejects it inside `_source`).
fn source_doc(columns: &[String], row: &[Value], skip: &str) -> Json {
    let mut m = serde_json::Map::new();
    for (col, v) in columns.iter().zip(row) {
        if col != skip {
            m.insert(col.clone(), to_json(v));
        }
    }
    Json::Object(m)
}

/// Coerce an `_id` value to the string form ES document ids use. `Null` means
/// "let ES auto-generate" (insert only; upsert rejects it).
fn id_string(v: &Value, at: &str) -> Result<Option<String>, WriteError> {
    Ok(match v {
        Value::Null => None,
        Value::Str(s) => Some(s.clone()),
        Value::Int(i) => Some(i.to_string()),
        _ => {
            return Err(WriteError::NotRepresentable {
                what: "a non-string/integer `_id` value".to_string(),
                at: at.to_string(),
            });
        }
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::query::{EntityRegistry, QueryError};
    use serde_json::json;

    fn es(query: Json) -> EsQuery {
        translate(&query, &EntityRegistry::default())
    }

    fn es_schema(query: Json, schema: Json) -> EsQuery {
        translate(&query, &EntityRegistry::from_json(&schema).expect("schema"))
    }

    /// Local translate helper (mirrors crate::query::translate_es).
    fn translate(query: &Json, reg: &EntityRegistry) -> EsQuery {
        let spec = crate::query::spec::parse(query).expect("spec");
        let cond = match &spec.filter {
            Some(f) => {
                crate::query::lower::lower_with(f, &serde_json::Map::new(), reg, &spec.source)
                    .expect("lower")
            }
            None => crate::query::ir::Cond::True,
        };
        let index = reg.physical_table(&spec.source);
        render(&spec, &cond, &index, 100, 1000).expect("render")
    }

    #[test]
    fn test_bool_filter_and_term() {
        let q = es(json!({
            "source": "users",
            "filter": { "and": [
                { "==": [{"field": "status"}, "active"] },
                { ">": [{"field": "age"}, 18] }
            ] }
        }));
        assert_eq!(q.index, "users");
        assert_eq!(
            q.body["query"],
            json!({ "bool": { "filter": [
                { "term": { "status": "active" } },
                { "range": { "age": { "gt": 18 } } }
            ] } })
        );
        assert_eq!(q.body["size"], json!(100));
        assert_eq!(q.body["from"], json!(0));
    }

    #[test]
    fn test_or_is_should() {
        let q = es(json!({
            "source": "t",
            "filter": { "or": [
                { "==": [{"field": "a"}, 1] },
                { "==": [{"field": "b"}, 2] }
            ] }
        }));
        assert_eq!(
            q.body["query"],
            json!({ "bool": { "should": [
                { "term": { "a": 1 } },
                { "term": { "b": 2 } }
            ], "minimum_should_match": 1 } })
        );
    }

    #[test]
    fn test_membership_terms() {
        let q = es(json!({ "source": "t", "filter": { "in": [{"field": "status"}, ["a", "b"]] } }));
        assert_eq!(
            q.body["query"],
            json!({ "terms": { "status": ["a", "b"] } })
        );
    }

    #[test]
    fn test_is_null_is_must_not_exists() {
        let q = es(json!({ "source": "t", "filter": { "==": [{"field": "email"}, null] } }));
        assert_eq!(
            q.body["query"],
            json!({ "bool": { "must_not": [{ "exists": { "field": "email" } }] } })
        );
    }

    #[test]
    fn test_range_inclusive() {
        let q = es(json!({ "source": "t", "filter": { "<=": [1, {"field": "x"}, 10] } }));
        assert_eq!(
            q.body["query"],
            json!({ "range": { "x": { "gte": 1, "lte": 10 } } })
        );
    }

    #[test]
    fn test_text_prefix_and_wildcard() {
        let starts =
            es(json!({ "source": "t", "filter": { "starts_with": [{"field": "name"}, "sm"] } }));
        assert_eq!(
            starts.body["query"],
            json!({ "prefix": { "name": { "value": "sm" } } })
        );
        let contains = es(json!({ "source": "t", "filter": { "in": ["a*b", {"field": "name"}] } }));
        assert_eq!(
            contains.body["query"],
            json!({ "wildcard": { "name": { "value": "*a\\*b*" } } })
        );
    }

    #[test]
    fn test_nested_relation() {
        let q = es_schema(
            json!({
                "source": "users",
                "filter": { "some": [{"field": "orders"}, {">": [{"field": "total"}, 100]}] }
            }),
            json!({ "entities": { "users": { "relations": {
                "orders": { "to": "orders", "kind": "has_many", "local": "id", "foreign": "user_id", "es": "nested" }
            } } } }),
        );
        assert_eq!(
            q.body["query"],
            json!({ "nested": { "path": "orders", "query": {
                "range": { "orders.total": { "gt": 100 } }
            } } })
        );
    }

    #[test]
    fn test_has_child_relation() {
        let q = es_schema(
            json!({
                "source": "users",
                "filter": { "some": [{"field": "orders"}, {">": [{"field": "total"}, 100]}] }
            }),
            json!({ "entities": { "users": { "relations": {
                "orders": { "to": "orders", "kind": "has_many", "local": "id", "foreign": "user_id", "es": "child" }
            } } } }),
        );
        assert_eq!(
            q.body["query"],
            json!({ "has_child": { "type": "orders", "query": {
                "range": { "total": { "gt": 100 } }
            } } })
        );
    }

    #[test]
    fn test_all_over_relation_is_capability_error() {
        let reg = EntityRegistry::from_json(&json!({ "entities": { "users": { "relations": {
            "orders": { "to": "orders", "kind": "has_many", "local": "id", "foreign": "user_id", "es": "nested" }
        } } } }))
        .expect("schema");
        let spec = crate::query::spec::parse(&json!({
            "source": "users",
            "filter": { "all": [{"field": "orders"}, {">": [{"field": "total"}, 100]}] }
        }))
        .expect("spec");
        let cond = crate::query::lower::lower_with(
            spec.filter.as_ref().expect("filter"),
            &serde_json::Map::new(),
            &reg,
            "users",
        )
        .expect("lower");
        let err = render(&spec, &cond, "users", 100, 1000).expect_err("all is gated on ES");
        assert!(matches!(err, QueryError::FeatureUnsupportedByTarget { .. }));
    }

    #[test]
    fn test_deep_pagination_rejected() {
        let spec = crate::query::spec::parse(&json!({ "source": "t", "skip": 9999, "limit": 100 }))
            .expect("spec");
        let err =
            render(&spec, &crate::query::ir::Cond::True, "t", 100, 1000).expect_err("deep paging");
        assert!(matches!(err, QueryError::FeatureUnsupportedByTarget { .. }));
    }

    #[test]
    fn test_envelope_source_and_sort() {
        let q = es(json!({
            "source": "t",
            "fields": ["id", "name"],
            "sort": [{ "created_at": "desc" }],
            "limit": 20
        }));
        assert_eq!(q.body["_source"], json!(["id", "name"]));
        assert_eq!(
            q.body["sort"],
            json!([{ "created_at": { "order": "desc", "missing": "_first" } }])
        );
        assert_eq!(q.body["size"], json!(20));
    }

    // ---- Write rendering ----

    fn resolve(input: Json) -> crate::query::write::ResolvedWrite {
        crate::query::write::resolve_write(
            &input,
            &serde_json::Map::new(),
            &EntityRegistry::default(),
        )
        .expect("resolve_write should succeed")
    }

    fn resolve_schema(input: Json, schema: Json) -> crate::query::write::ResolvedWrite {
        crate::query::write::resolve_write(
            &input,
            &serde_json::Map::new(),
            &EntityRegistry::from_json(&schema).expect("schema"),
        )
        .expect("resolve_write should succeed")
    }

    /// The schema declaring that the logical `id` keys the ES document.
    fn id_schema() -> Json {
        json!({ "entities": { "users": { "columns": { "id": { "name": "_id" } } } } })
    }

    #[test]
    fn test_es_insert_bulk_lifts_id() {
        let ew = render_write(&resolve_schema(
            json!({
                "op": "insert", "target": "users",
                "values": [ { "id": "u1", "name": "Ada" }, { "id": "u2", "name": "Bob" } ]
            }),
            id_schema(),
        ))
        .expect("render");
        // The physical `_id` column is lifted out of the source into the action.
        assert_eq!(
            ew,
            EsWrite::BulkInsert {
                index: "users".to_string(),
                docs: vec![
                    (Some("u1".to_string()), json!({ "name": "Ada" })),
                    (Some("u2".to_string()), json!({ "name": "Bob" })),
                ],
            }
        );
    }

    #[test]
    fn test_es_insert_id_stays_in_source_without_schema() {
        // No implicit `id` → `_id` mapping: without the schema rename, `id` is an
        // ordinary source field and ES auto-generates the document id.
        let ew = render_write(&resolve(json!({
            "op": "insert", "target": "users",
            "values": { "id": "u1", "name": "Ada" }
        })))
        .expect("render");
        assert_eq!(
            ew,
            EsWrite::BulkInsert {
                index: "users".to_string(),
                docs: vec![(None, json!({ "id": "u1", "name": "Ada" }))],
            }
        );
    }

    #[test]
    fn test_es_insert_null_id_autogenerates() {
        let ew = render_write(&resolve_schema(
            json!({
                "op": "insert", "target": "users",
                "values": { "id": null, "name": "Ada" }
            }),
            id_schema(),
        ))
        .expect("render");
        assert_eq!(
            ew,
            EsWrite::BulkInsert {
                index: "users".to_string(),
                docs: vec![(None, json!({ "name": "Ada" }))],
            }
        );
    }

    #[test]
    fn test_es_bulk_ndjson_shape() {
        let body = bulk_ndjson(&[
            (Some("u1".to_string()), json!({ "name": "Ada" })),
            (None, json!({ "name": "Bob" })),
        ]);
        assert_eq!(
            body,
            "{\"index\":{\"_id\":\"u1\"}}\n{\"name\":\"Ada\"}\n{\"index\":{}}\n{\"name\":\"Bob\"}\n"
        );
    }

    #[test]
    fn test_es_update_by_query_script_params() {
        let ew = render_write(&resolve(json!({
            "op": "update", "target": "users",
            "set": { "status": "inactive", "age": null },
            "filter": { ">": [{ "field": "age" }, 25] }
        })))
        .expect("render");
        // `set` keeps the envelope's key order (serde_json preserve_order):
        // status first, then age.
        assert_eq!(
            ew,
            EsWrite::UpdateByQuery {
                index: "users".to_string(),
                body: json!({
                    "query": { "range": { "age": { "gt": 25 } } },
                    "script": {
                        "lang": "painless",
                        "source": "ctx._source[params.f0] = params.v0;ctx._source[params.f1] = params.v1;",
                        "params": { "f0": "status", "v0": "inactive", "f1": "age", "v1": null }
                    }
                }),
            }
        );
    }

    #[test]
    fn test_es_update_all_true_is_match_all() {
        let ew = render_write(&resolve(json!({
            "op": "update", "target": "users",
            "set": { "status": "x" },
            "all": true
        })))
        .expect("render");
        assert_eq!(
            ew,
            EsWrite::UpdateByQuery {
                index: "users".to_string(),
                body: json!({
                    "query": { "match_all": {} },
                    "script": {
                        "lang": "painless",
                        "source": "ctx._source[params.f0] = params.v0;",
                        "params": { "f0": "status", "v0": "x" }
                    }
                }),
            }
        );
    }

    #[test]
    fn test_es_delete_by_query() {
        let ew = render_write(&resolve(json!({
            "op": "delete", "target": "sessions",
            "filter": { "<": [{ "field": "age" }, 0] }
        })))
        .expect("render");
        assert_eq!(
            ew,
            EsWrite::DeleteByQuery {
                index: "sessions".to_string(),
                body: json!({ "query": { "range": { "age": { "lt": 0 } } } }),
            }
        );
    }

    #[test]
    fn test_es_upsert_update_is_doc_as_upsert() {
        let ew = render_write(&resolve_schema(
            json!({
                "op": "upsert", "target": "users",
                "values": { "id": "u1", "name": "Alice2", "age": 31 },
                "on_conflict": { "target": ["id"], "action": "update" }
            }),
            id_schema(),
        ))
        .expect("render");
        assert_eq!(
            ew,
            EsWrite::UpdateDoc {
                index: "users".to_string(),
                id: "u1".to_string(),
                body: json!({ "doc": { "age": 31, "name": "Alice2" }, "doc_as_upsert": true }),
            }
        );
    }

    #[test]
    fn test_es_upsert_with_set_uses_doc_plus_upsert() {
        let ew = render_write(&resolve_schema(
            json!({
                "op": "upsert", "target": "users",
                "values": { "id": "u1", "name": "Alice2", "age": 31 },
                "set": { "status": "active" },
                "on_conflict": { "target": ["id"], "action": "update" }
            }),
            id_schema(),
        ))
        .expect("render");
        // On conflict apply `set`; on insert the row overlaid with `set`.
        assert_eq!(
            ew,
            EsWrite::UpdateDoc {
                index: "users".to_string(),
                id: "u1".to_string(),
                body: json!({
                    "doc": { "status": "active" },
                    "upsert": { "age": 31, "name": "Alice2", "status": "active" }
                }),
            }
        );
    }

    #[test]
    fn test_es_upsert_nothing_is_create() {
        let ew = render_write(&resolve_schema(
            json!({
                "op": "upsert", "target": "users",
                "values": { "id": "u1", "name": "Ada" },
                "on_conflict": { "target": ["id"], "action": "nothing" }
            }),
            id_schema(),
        ))
        .expect("render");
        assert_eq!(
            ew,
            EsWrite::CreateDoc {
                index: "users".to_string(),
                id: "u1".to_string(),
                doc: json!({ "name": "Ada" }),
            }
        );
    }

    #[test]
    fn test_es_bulk_upsert_rejected() {
        let err = render_write(&resolve_schema(
            json!({
                "op": "upsert", "target": "users",
                "values": [ { "id": "u1", "name": "A" }, { "id": "u2", "name": "B" } ],
                "on_conflict": { "target": ["id"], "action": "update" }
            }),
            id_schema(),
        ))
        .expect_err("bulk upsert is gated on ES");
        assert!(matches!(err, WriteError::FeatureUnsupportedByTarget { .. }));
    }

    #[test]
    fn test_es_upsert_non_id_target_rejected() {
        let err = render_write(&resolve(json!({
            "op": "upsert", "target": "users",
            "values": { "email": "a@x.io", "name": "Ada" },
            "on_conflict": { "target": ["email"], "action": "update" }
        })))
        .expect_err("non-_id conflict target is gated on ES");
        assert!(matches!(err, WriteError::FeatureUnsupportedByTarget { .. }));
    }

    #[test]
    fn test_es_returning_rejected() {
        let err = render_write(&resolve(json!({
            "op": "insert", "target": "users",
            "values": { "name": "Ada" },
            "returning": ["id"]
        })))
        .expect_err("returning is gated on ES");
        assert!(matches!(err, WriteError::FeatureUnsupportedByTarget { .. }));
    }

    #[test]
    fn test_es_update_set_id_rejected() {
        let err = render_write(&resolve_schema(
            json!({
                "op": "update", "target": "users",
                "set": { "id": "u9" },
                "filter": { "==": [{ "field": "name" }, "Ada"] }
            }),
            id_schema(),
        ))
        .expect_err("updating _id is gated on ES");
        assert!(matches!(err, WriteError::FeatureUnsupportedByTarget { .. }));
    }
}