graphddb_runtime 0.7.5

//! Declarative filter -> DynamoDB `FilterExpression` compiler — a port of
//! `python/graphddb_runtime/filters.py` (itself a port of the TS
//! `compileFilterExpression`).
//!
//! Names are `#`-aliased columns (reused per distinct column); values are
//! `:`-aliased parameters — no literal interpolation. Attribute values are
//! serialized `AttributeValue`s (the boto3 *client* shape). Alias allocation order
//! matches the Python reference (name aliases reused per column in first-seen
//! order; value aliases allocated sequentially), so the compiled expression and
//! its name/value maps are byte-identical across runtimes.

use aws_sdk_dynamodb::types::AttributeValue;
use serde_json::Value as Json;

use crate::attribute::serialize_json;
use crate::errors::GraphDDBError;
use crate::value::indexmap_shim::IndexMap;

const OPERATOR_KEYS: &[&str] = &[
    "eq",
    "ne",
    "gt",
    "ge",
    "lt",
    "le",
    "between",
    "in",
    "beginsWith",
    "contains",
    "notContains",
    "attributeExists",
    "attributeType",
    "size",
];

const LOGICAL_KEYS: &[&str] = &["and", "or", "not"];

/// The compiled filter (client shape).
#[derive(Debug, Clone)]
pub struct CompiledFilter {
    /// The `FilterExpression` / `ConditionExpression` string.
    pub expression: String,
    /// `ExpressionAttributeNames` (insertion-ordered).
    pub names: IndexMap<String>,
    /// `ExpressionAttributeValues` (insertion-ordered).
    pub values: IndexMap<AttributeValue>,
}

struct Ctx {
    names: IndexMap<String>,
    values: IndexMap<AttributeValue>,
    name_n: usize,
    value_n: usize,
}

impl Ctx {
    fn new() -> Self {
        Self {
            names: IndexMap::new(),
            values: IndexMap::new(),
            name_n: 0,
            value_n: 0,
        }
    }

    fn name_alias(&mut self, column: &str) -> String {
        for (alias, col) in self.names.iter() {
            if col == column {
                return alias.clone();
            }
        }
        let alias = format!("#f{}", self.name_n);
        self.name_n += 1;
        self.names.insert(alias.clone(), column.to_string());
        alias
    }

    fn value_alias(&mut self, raw: &Json) -> Result<String, GraphDDBError> {
        let alias = format!(":vf{}", self.value_n);
        self.value_n += 1;
        self.values.insert(alias.clone(), serialize_json(raw)?);
        Ok(alias)
    }
}

fn is_operator_object(value: &Json) -> bool {
    match value.as_object() {
        Some(obj) if !obj.is_empty() => obj.keys().all(|k| OPERATOR_KEYS.contains(&k.as_str())),
        _ => false,
    }
}

fn is_already_wrapped(expr: &str) -> bool {
    let bytes = expr.as_bytes();
    if !(expr.starts_with('(') && expr.ends_with(')')) {
        return false;
    }
    let mut depth = 0i32;
    for (i, &ch) in bytes.iter().enumerate() {
        if ch == b'(' {
            depth += 1;
        } else if ch == b')' {
            depth -= 1;
            if depth == 0 && i < bytes.len() - 1 {
                return false;
            }
        }
    }
    depth == 0
}

fn wrap(expr: &str) -> String {
    if is_already_wrapped(expr) {
        return expr.to_string();
    }
    if expr.contains(" AND ") || expr.contains(" OR ") {
        format!("({expr})")
    } else {
        expr.to_string()
    }
}

fn join_and(clauses: &[String]) -> String {
    if clauses.len() == 1 {
        return clauses[0].clone();
    }
    clauses
        .iter()
        .map(|c| wrap(c))
        .collect::<Vec<_>>()
        .join(" AND ")
}

fn compile_field(ctx: &mut Ctx, field: &str, condition: &Json) -> Result<String, GraphDDBError> {
    let n = ctx.name_alias(field);
    if !is_operator_object(condition) {
        let v = ctx.value_alias(condition)?;
        return Ok(format!("{n} = {v}"));
    }
    let obj = condition.as_object().unwrap();
    let mut clauses: Vec<String> = Vec::new();
    for (op, value) in obj {
        let clause = match op.as_str() {
            "eq" => format!("{n} = {}", ctx.value_alias(value)?),
            "ne" => format!("{n} <> {}", ctx.value_alias(value)?),
            "gt" => format!("{n} > {}", ctx.value_alias(value)?),
            "ge" => format!("{n} >= {}", ctx.value_alias(value)?),
            "lt" => format!("{n} < {}", ctx.value_alias(value)?),
            "le" => format!("{n} <= {}", ctx.value_alias(value)?),
            "between" => {
                let arr = value.as_array().filter(|a| a.len() == 2).ok_or_else(|| {
                    GraphDDBError::new(format!(
                        "between operator on field '{field}' expects a [lo, hi] array of length 2"
                    ))
                })?;
                let lo = ctx.value_alias(&arr[0])?;
                let hi = ctx.value_alias(&arr[1])?;
                format!("{n} BETWEEN {lo} AND {hi}")
            }
            "in" => {
                let arr = value
                    .as_array()
                    .ok_or_else(|| GraphDDBError::new("in operator expects an array"))?;
                let mut aliases = Vec::with_capacity(arr.len());
                for v in arr {
                    aliases.push(ctx.value_alias(v)?);
                }
                format!("{n} IN ({})", aliases.join(", "))
            }
            "beginsWith" => format!("begins_with({n}, {})", ctx.value_alias(value)?),
            "contains" => format!("contains({n}, {})", ctx.value_alias(value)?),
            "notContains" => format!("NOT contains({n}, {})", ctx.value_alias(value)?),
            "attributeExists" => {
                if value == &Json::Bool(false) {
                    format!("attribute_not_exists({n})")
                } else {
                    format!("attribute_exists({n})")
                }
            }
            "attributeType" => format!("attribute_type({n}, {})", ctx.value_alias(value)?),
            "size" => format!("size({n}) = {}", ctx.value_alias(value)?),
            other => {
                return Err(GraphDDBError::new(format!(
                    "Unknown filter operator '{other}' on field '{field}'"
                )))
            }
        };
        clauses.push(clause);
    }
    Ok(join_and(&clauses))
}

fn compile_node(ctx: &mut Ctx, node: &Json) -> Result<String, GraphDDBError> {
    let obj = match node.as_object() {
        Some(o) => o,
        None => return Ok(String::new()),
    };
    let mut clauses: Vec<String> = Vec::new();
    for (key, value) in obj {
        if value.is_null() {
            continue;
        }
        if LOGICAL_KEYS.contains(&key.as_str()) {
            if key == "and" || key == "or" {
                let arr = value.as_array().cloned().unwrap_or_default();
                let mut parts: Vec<String> = Vec::new();
                for s in &arr {
                    let p = compile_node(ctx, s)?;
                    if !p.is_empty() {
                        parts.push(p);
                    }
                }
                if parts.is_empty() {
                    continue;
                }
                if parts.len() == 1 {
                    clauses.push(parts.remove(0));
                } else {
                    let sep = if key == "and" { " AND " } else { " OR " };
                    let joined = parts.iter().map(|p| wrap(p)).collect::<Vec<_>>().join(sep);
                    clauses.push(format!("({joined})"));
                }
            } else {
                // not
                let inner = compile_node(ctx, value)?;
                if !inner.is_empty() {
                    clauses.push(format!("NOT {}", wrap(&inner)));
                }
            }
            continue;
        }
        let clause = compile_field(ctx, key, value)?;
        if !clause.is_empty() {
            clauses.push(clause);
        }
    }
    Ok(join_and(&clauses))
}

/// Compile a declarative filter tree into a client-shape filter. Returns `None`
/// for an empty / no-op filter so callers can skip attaching it.
pub fn compile_filter(declarative: &Json) -> Result<Option<CompiledFilter>, GraphDDBError> {
    let obj = match declarative.as_object() {
        Some(o) if !o.is_empty() => o,
        _ => return Ok(None),
    };
    let _ = obj;
    let mut ctx = Ctx::new();
    let expr = compile_node(&mut ctx, declarative)?;
    if expr.is_empty() {
        return Ok(None);
    }
    Ok(Some(CompiledFilter {
        expression: expr,
        names: ctx.names,
        values: ctx.values,
    }))
}

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

    #[test]
    fn simple_eq() {
        let c = compile_filter(&json!({"role": "admin"})).unwrap().unwrap();
        assert_eq!(c.expression, "#f0 = :vf0");
        assert_eq!(c.names.get("#f0").unwrap(), "role");
    }

    #[test]
    fn and_of_ops_wraps_correctly() {
        let f = json!({"and": [{"role": {"in": ["a", "b"]}}, {"role": {"beginsWith": "x"}}]});
        let c = compile_filter(&f).unwrap().unwrap();
        // Same column reuses #f0; two ops -> parenthesized AND.
        assert_eq!(
            c.expression,
            "(#f0 IN (:vf0, :vf1) AND begins_with(#f0, :vf2))"
        );
    }

    #[test]
    fn not_and_between_and_size() {
        let c = compile_filter(&json!({"role": {"size": 3}}))
            .unwrap()
            .unwrap();
        assert_eq!(c.expression, "size(#f0) = :vf0");
        let c2 = compile_filter(&json!({"not": {"role": "admin"}}))
            .unwrap()
            .unwrap();
        assert_eq!(c2.expression, "NOT #f0 = :vf0");
        let c3 = compile_filter(&json!({"n": {"between": [1, 5]}}))
            .unwrap()
            .unwrap();
        assert_eq!(c3.expression, "#f0 BETWEEN :vf0 AND :vf1");
    }

    #[test]
    fn empty_is_none() {
        assert!(compile_filter(&json!({})).unwrap().is_none());
    }
}