graphddb_runtime 0.7.5

//! Declarative transaction expansion — a port of
//! `python/graphddb_runtime/transactions.py`.
//!
//! Expands a serializable transaction spec into `TransactWriteItems`: resolves
//! each templated item (`{param}` / `{item.field}`) against the caller params (and,
//! for `forEach` items, each element of the named array param), derives PK/SK/GSI
//! from the manifest, applies conditions, builds put/update/delete/conditionCheck,
//! and collapses same-key items (swap no-op, self-row de-dup, symmetric ADD merge)
//! with the SAME op-aware rules the TS/Python runtimes apply.

use std::collections::HashMap;

use aws_sdk_dynamodb::types::{
    AttributeValue, ConditionCheck, Delete, Put, TransactWriteItem, Update,
};
use serde_json::Value as Json;

use crate::attribute::{serialize_int, serialize_json};
use crate::entity::add_key_attributes;
use crate::errors::{GraphDDBError, Result};
use crate::filters::compile_filter;
use crate::runtime::GraphDDBRuntime;
use crate::templates::Params;

const MAX_TRANSACT_ITEMS: usize = 25;

/// Placeholder-name iterator over a template (regex `\{[^{}]+\}`).
fn each_ph(template: &str) -> Vec<String> {
    crate::templates::each_placeholder_pub(template)
}

/// Resolve `{param}` / `{item.field}` placeholders to a string.
fn resolve_template(template: &str, params: &Params, element: Option<&Json>) -> Result<String> {
    let mut out = String::new();
    let bytes = template.as_bytes();
    let mut i = 0;
    while i < bytes.len() {
        if bytes[i] == b'{' {
            if let Some(close) = template[i + 1..].find('}') {
                let name = &template[i + 1..i + 1 + close];
                if !name.is_empty() && !name.contains('{') {
                    let value = resolve_name(name, params, element)?;
                    out.push_str(&value);
                    i += 1 + close + 1;
                    continue;
                }
            }
        }
        out.push(bytes[i] as char);
        i += 1;
    }
    Ok(out)
}

fn resolve_name(name: &str, params: &Params, element: Option<&Json>) -> Result<String> {
    if let Some(field) = name.strip_prefix("item.") {
        let el = element
            .and_then(Json::as_object)
            .and_then(|o| o.get(field))
            .filter(|v| !v.is_null());
        return el
            .map(crate::templates::json_to_template_string)
            .ok_or_else(|| {
                GraphDDBError::new(format!("element field '{field}' not bound for template"))
            });
    }
    match params.get(name) {
        Some(v) if !v.is_null() => Ok(crate::templates::json_to_template_string(v)),
        _ => Err(GraphDDBError::new(format!(
            "template references unbound param '{name}'"
        ))),
    }
}

/// Resolve a value template, keeping the typed value for a WHOLE placeholder.
fn resolve_value(template: &str, params: &Params, element: Option<&Json>) -> Result<Json> {
    if is_whole_placeholder(template) {
        let name = &template[1..template.len() - 1];
        if let Some(field) = name.strip_prefix("item.") {
            if let Some(v) = element.and_then(Json::as_object).and_then(|o| o.get(field)) {
                if !v.is_null() {
                    return Ok(v.clone());
                }
            }
        } else if let Some(v) = params.get(name) {
            if !v.is_null() {
                return Ok(v.clone());
            }
        }
    }
    Ok(Json::String(resolve_template(template, params, element)?))
}

fn is_whole_placeholder(t: &str) -> bool {
    t.starts_with('{')
        && t.ends_with('}')
        && t.len() >= 3
        && !t[1..t.len() - 1].contains('{')
        && !t[1..t.len() - 1].contains('}')
}

/// `{"$param": name}` marker → typed value (from params / element), else literal.
fn resolve_condition_leaf(value: &Json, params: &Params, element: Option<&Json>) -> Result<Json> {
    if let Some(name) = value.get("$param").and_then(Json::as_str) {
        return resolve_value(&format!("{{{name}}}"), params, element);
    }
    Ok(value.clone())
}

fn resolve_condition_tree(node: &Json, params: &Params, element: Option<&Json>) -> Result<Json> {
    let obj = node.as_object().cloned().unwrap_or_default();
    let mut out = serde_json::Map::new();
    for (key, value) in obj {
        match key.as_str() {
            "and" | "or" => {
                let arr = value.as_array().cloned().unwrap_or_default();
                let mut parts = Vec::new();
                for s in &arr {
                    parts.push(resolve_condition_tree(s, params, element)?);
                }
                out.insert(key, Json::Array(parts));
            }
            "not" => {
                out.insert(key, resolve_condition_tree(&value, params, element)?);
            }
            _ => {
                let ops_obj = value.as_object().cloned().unwrap_or_default();
                let mut ops = serde_json::Map::new();
                for (op, op_val) in ops_obj {
                    let resolved = match op.as_str() {
                        "between" => {
                            let arr =
                                op_val.as_array().filter(|a| a.len() == 2).ok_or_else(|| {
                                    GraphDDBError::new(
                                        "`between` condition expects a [lo, hi] array of length 2",
                                    )
                                })?;
                            Json::Array(vec![
                                resolve_condition_leaf(&arr[0], params, element)?,
                                resolve_condition_leaf(&arr[1], params, element)?,
                            ])
                        }
                        "in" => {
                            let arr = op_val.as_array().cloned().unwrap_or_default();
                            let mut vals = Vec::new();
                            for v in &arr {
                                vals.push(resolve_condition_leaf(v, params, element)?);
                            }
                            Json::Array(vals)
                        }
                        "attributeExists" => op_val.clone(),
                        _ => resolve_condition_leaf(&op_val, params, element)?,
                    };
                    ops.insert(op, resolved);
                }
                out.insert(key, Json::Object(ops));
            }
        }
    }
    Ok(Json::Object(out))
}

fn when_holds(when: &Json, params: &Params, element: Option<&Json>) -> Result<bool> {
    let left = resolve_template(when["left"].as_str().unwrap_or(""), params, element)?;
    let right = resolve_template(when["right"].as_str().unwrap_or(""), params, element)?;
    Ok(if when["op"].as_str() == Some("eq") {
        left == right
    } else {
        left != right
    })
}

// ── expression / value maps carried through building ──────────────────────────

/// A composed transact item plus its (table, PK, SK) signature for the collapse.
struct Built {
    kind: BuiltKind,
    table: String,
    pk: String,
    sk: String,
}

enum BuiltKind {
    Put {
        item: HashMap<String, AttributeValue>,
        condition: Option<Expr>,
    },
    Delete {
        key: HashMap<String, AttributeValue>,
        condition: Option<Expr>,
    },
    ConditionCheck {
        key: HashMap<String, AttributeValue>,
        condition: Expr,
    },
    Update {
        key: HashMap<String, AttributeValue>,
        update_expression: Option<String>,
        names: HashMap<String, String>,
        values: HashMap<String, AttributeValue>,
        condition: Option<Expr>,
    },
}

#[derive(Clone)]
struct Expr {
    expression: String,
    names: HashMap<String, String>,
    values: HashMap<String, AttributeValue>,
}

fn attr_str(av: Option<&AttributeValue>) -> String {
    match av {
        Some(AttributeValue::S(s)) => s.clone(),
        Some(AttributeValue::N(n)) => n.clone(),
        _ => String::new(),
    }
}

/// Apply a serialized condition onto an in-progress expr/names/values set.
fn apply_condition(
    rt: &GraphDDBRuntime,
    condition: Option<&Json>,
    params: &Params,
    element: Option<&Json>,
    names: &mut HashMap<String, String>,
    values: &mut HashMap<String, AttributeValue>,
) -> Result<Option<String>> {
    let _ = rt;
    let condition = match condition {
        Some(c) if !c.is_null() => c,
        _ => return Ok(None),
    };
    let kind = condition["kind"].as_str().unwrap_or("");
    match kind {
        "notExists" => {
            names.insert("#pk".to_string(), "PK".to_string());
            Ok(Some("attribute_not_exists(#pk)".to_string()))
        }
        "attributeExists" | "attributeNotExists" => {
            let fname = if kind == "attributeExists" {
                "attribute_exists"
            } else {
                "attribute_not_exists"
            };
            names.insert(
                "#ce".to_string(),
                condition["field"].as_str().unwrap_or("").to_string(),
            );
            Ok(Some(format!("{fname}(#ce)")))
        }
        "expr" => {
            let concrete = resolve_condition_tree(&condition["declarative"], params, element)?;
            match compile_filter(&concrete)? {
                None => Ok(None),
                Some(c) => {
                    for (a, col) in c.names.iter() {
                        names.insert(a.clone(), col.clone());
                    }
                    for (a, v) in c.values.iter() {
                        values.insert(a.clone(), v.clone());
                    }
                    Ok(Some(c.expression))
                }
            }
        }
        "raw" => {
            if let Some(n) = condition.get("names").and_then(Json::as_object) {
                for (a, col) in n {
                    names.insert(a.clone(), col.as_str().unwrap_or("").to_string());
                }
            }
            if let Some(vals) = condition.get("values").and_then(Json::as_object) {
                for (alias, raw_val) in vals {
                    let resolved = resolve_condition_leaf(raw_val, params, element)?;
                    values.insert(alias.clone(), serialize_json(&resolved)?);
                }
            }
            Ok(Some(
                condition["expression"].as_str().unwrap_or("").to_string(),
            ))
        }
        "equals" => {
            let mut clauses = Vec::new();
            if let Some(fields) = condition.get("fields").and_then(Json::as_object) {
                for (i, (field, tmpl)) in fields.iter().enumerate() {
                    let n = format!("#e{i}");
                    let v = format!(":e{i}");
                    names.insert(n.clone(), field.clone());
                    let resolved = resolve_value(tmpl.as_str().unwrap_or(""), params, element)?;
                    values.insert(v.clone(), serialize_json(&resolved)?);
                    clauses.push(format!("{n} = {v}"));
                }
            }
            Ok(Some(clauses.join(" AND ")))
        }
        _ => Ok(None),
    }
}

fn build_item(
    rt: &GraphDDBRuntime,
    spec: &Json,
    params: &Params,
    element: Option<&Json>,
) -> Result<Built> {
    let entity_name = spec["entity"].as_str().unwrap_or("");
    let table = rt.physical_table(spec["tableName"].as_str().unwrap_or(""));
    let item_type = spec["type"].as_str().unwrap_or("");

    if item_type == "Put" {
        // Resolve the record (typed values kept for whole placeholders).
        let mut plain: Vec<(String, Json)> = Vec::new();
        if let Some(rec) = spec.get("item").and_then(Json::as_object) {
            for (field, tmpl) in rec {
                plain.push((
                    field.clone(),
                    resolve_value(tmpl.as_str().unwrap_or(""), params, element)?,
                ));
            }
        }
        if !spec
            .get("literalKey")
            .and_then(Json::as_bool)
            .unwrap_or(false)
        {
            add_key_attributes(rt.manifest_ref(), entity_name, &mut plain);
        }
        let mut item: HashMap<String, AttributeValue> = HashMap::new();
        for (k, v) in &plain {
            item.insert(k.clone(), serialize_json(v)?);
        }
        let mut names = HashMap::new();
        let mut values = HashMap::new();
        let expr = apply_condition(
            rt,
            spec.get("condition"),
            params,
            element,
            &mut names,
            &mut values,
        )?;
        let condition = expr.map(|e| Expr {
            expression: e,
            names,
            values,
        });
        let pk = attr_str(item.get("PK"));
        let sk = attr_str(item.get("SK"));
        return Ok(Built {
            kind: BuiltKind::Put { item, condition },
            table,
            pk,
            sk,
        });
    }

    // key from keyCondition
    let mut key_plain: Vec<(String, Json)> = Vec::new();
    if let Some(kc) = spec.get("keyCondition").and_then(Json::as_object) {
        for (field, tmpl) in kc {
            key_plain.push((
                field.clone(),
                resolve_value(tmpl.as_str().unwrap_or(""), params, element)?,
            ));
        }
    }
    let mut key: HashMap<String, AttributeValue> = HashMap::new();
    for (k, v) in &key_plain {
        key.insert(k.clone(), serialize_json(v)?);
    }
    let pk = attr_str(key.get("PK"));
    let sk = attr_str(key.get("SK"));

    if item_type == "Delete" {
        let mut names = HashMap::new();
        let mut values = HashMap::new();
        let expr = apply_condition(
            rt,
            spec.get("condition"),
            params,
            element,
            &mut names,
            &mut values,
        )?;
        let condition = expr.map(|e| Expr {
            expression: e,
            names,
            values,
        });
        return Ok(Built {
            kind: BuiltKind::Delete { key, condition },
            table,
            pk,
            sk,
        });
    }

    if item_type == "ConditionCheck" {
        let cond_spec = spec.get("condition");
        if cond_spec.map(Json::is_null).unwrap_or(true) {
            return Err(GraphDDBError::new(format!(
                "ConditionCheck on '{entity_name}' has no condition"
            )));
        }
        let mut names = HashMap::new();
        let mut values = HashMap::new();
        let expr = apply_condition(rt, cond_spec, params, element, &mut names, &mut values)?
            .ok_or_else(|| GraphDDBError::new("ConditionCheck condition compiled to nothing"))?;
        return Ok(Built {
            kind: BuiltKind::ConditionCheck {
                key,
                condition: Expr {
                    expression: expr,
                    names,
                    values,
                },
            },
            table,
            pk,
            sk,
        });
    }

    // maintain update?
    if let Some(maintain) = spec.get("maintain").filter(|m| !m.is_null()) {
        return build_maintain_update(rt, spec, maintain, table, key, pk, sk, params, element);
    }

    // Update (SET changes + ADD add).
    let mut names: HashMap<String, String> = HashMap::new();
    let mut values: HashMap<String, AttributeValue> = HashMap::new();
    let mut sets: Vec<String> = Vec::new();
    let mut add_clauses: Vec<String> = Vec::new();
    if let Some(changes) = spec.get("changes").and_then(Json::as_object) {
        for (i, (field, tmpl)) in changes.iter().enumerate() {
            let n = format!("#c{i}");
            let v = format!(":c{i}");
            names.insert(n.clone(), field.clone());
            let val = resolve_value(tmpl.as_str().unwrap_or(""), params, element)?;
            values.insert(v.clone(), serialize_json(&val)?);
            sets.push(format!("{n} = {v}"));
        }
    }
    if let Some(adds) = spec.get("add").and_then(Json::as_object) {
        for (i, (field, tmpl)) in adds.iter().enumerate() {
            let n = format!("#a{i}");
            let v = format!(":a{i}");
            names.insert(n.clone(), field.clone());
            let val = resolve_value(tmpl.as_str().unwrap_or(""), params, element)?;
            let av = match &val {
                Json::String(s) => serialize_int(s.parse::<i64>().map_err(|_| {
                    GraphDDBError::new(format!("ADD delta '{s}' is not an integer"))
                })?),
                other => serialize_json(other)?,
            };
            values.insert(v.clone(), av);
            add_clauses.push(format!("{n} {v}"));
        }
    }
    let mut expr_parts: Vec<String> = Vec::new();
    if !sets.is_empty() {
        expr_parts.push(format!("SET {}", sets.join(", ")));
    }
    if !add_clauses.is_empty() {
        expr_parts.push(format!("ADD {}", add_clauses.join(", ")));
    }
    let update_expression = if expr_parts.is_empty() {
        None
    } else {
        Some(expr_parts.join(" "))
    };
    // Names/values only attach when there's an update expression (parity).
    let (u_names, u_values) = if update_expression.is_some() {
        (names, values)
    } else {
        (HashMap::new(), HashMap::new())
    };
    let mut c_names = u_names.clone();
    let mut c_values = u_values.clone();
    let cond_expr = apply_condition(
        rt,
        spec.get("condition"),
        params,
        element,
        &mut c_names,
        &mut c_values,
    )?;
    // The condition adds its own aliases; merge them into the update's maps.
    let (final_names, final_values, condition) = match cond_expr {
        Some(e) => (
            c_names,
            c_values,
            Some(Expr {
                expression: e,
                names: HashMap::new(),
                values: HashMap::new(),
            }),
        ),
        None => (u_names, u_values, None),
    };
    Ok(Built {
        kind: BuiltKind::Update {
            key,
            update_expression,
            names: final_names,
            values: final_values,
            condition,
        },
        table,
        pk,
        sk,
    })
}

#[allow(clippy::too_many_arguments)]
fn build_maintain_update(
    _rt: &GraphDDBRuntime,
    _spec: &Json,
    maintain: &Json,
    table: String,
    key: HashMap<String, AttributeValue>,
    pk: String,
    sk: String,
    params: &Params,
    element: Option<&Json>,
) -> Result<Built> {
    let kind = maintain["kind"].as_str().unwrap_or("");
    let mut names: HashMap<String, String> = HashMap::new();
    let mut values: HashMap<String, AttributeValue> = HashMap::new();
    let update_expression;

    if kind == "counter" {
        let counter = maintain
            .get("counter")
            .filter(|c| !c.is_null())
            .ok_or_else(|| {
                GraphDDBError::new("a 'counter' maintenance write has no `counter` payload.")
            })?;
        let delta = counter["delta"].clone();
        let av = match &delta {
            Json::String(s) => serialize_int(s.parse::<i64>().map_err(|_| {
                GraphDDBError::new(format!("counter delta '{s}' is not an integer"))
            })?),
            other => serialize_json(other)?,
        };
        names.insert(
            "#a0".to_string(),
            counter["attribute"].as_str().unwrap_or("").to_string(),
        );
        values.insert(":a0".to_string(), av);
        update_expression = Some("ADD #a0 :a0".to_string());
        return Ok(Built {
            kind: BuiltKind::Update {
                key,
                update_expression,
                names,
                values,
                condition: None,
            },
            table,
            pk,
            sk,
        });
    }

    let projection = build_maintain_projection(maintain, params, element)?;
    if kind == "collection" {
        let collection = maintain.get("collection").and_then(Json::as_object);
        let field = collection
            .and_then(|c| c.get("field"))
            .and_then(Json::as_str)
            .ok_or_else(|| {
                GraphDDBError::new("a 'collection' maintenance write has no `collection.field`.")
            })?;
        names.insert("#c".to_string(), field.to_string());
        values.insert(":empty".to_string(), AttributeValue::L(vec![]));
        // The projected map, as a single-element L.
        let mut proj_map: HashMap<String, AttributeValue> = HashMap::new();
        for (attr, v) in &projection {
            proj_map.insert(attr.clone(), serialize_json(v)?);
        }
        values.insert(
            ":item".to_string(),
            AttributeValue::L(vec![AttributeValue::M(proj_map)]),
        );
        update_expression =
            Some("SET #c = list_append(if_not_exists(#c, :empty), :item)".to_string());
    } else {
        // snapshot
        let mut sets = Vec::new();
        for (i, (attr, v)) in projection.iter().enumerate() {
            names.insert(format!("#m{i}"), attr.clone());
            values.insert(format!(":m{i}"), serialize_json(v)?);
            sets.push(format!("#m{i} = :m{i}"));
        }
        if sets.is_empty() {
            return Err(GraphDDBError::new(
                "a 'snapshot' maintenance write has an empty projection.",
            ));
        }
        update_expression = Some(format!("SET {}", sets.join(", ")));
    }
    Ok(Built {
        kind: BuiltKind::Update {
            key,
            update_expression,
            names,
            values,
            condition: None,
        },
        table,
        pk,
        sk,
    })
}

fn build_maintain_projection(
    maintain: &Json,
    params: &Params,
    element: Option<&Json>,
) -> Result<Vec<(String, Json)>> {
    let mut out = Vec::new();
    if let Some(proj) = maintain.get("projection").and_then(Json::as_object) {
        for (attr, transform) in proj {
            let input_field = transform["inputField"].as_str().unwrap_or("");
            let value = resolve_value(&format!("{{{input_field}}}"), params, element)?;
            let op = transform["op"].as_str().unwrap_or("");
            let args = transform
                .get("args")
                .and_then(Json::as_array)
                .cloned()
                .unwrap_or_default();
            out.push((attr.clone(), apply_maintain_transform(op, &args, value)?));
        }
    }
    Ok(out)
}

fn apply_maintain_transform(op: &str, args: &[Json], value: Json) -> Result<Json> {
    match op {
        "identity" => Ok(value),
        "preview" => {
            let n = args.first().and_then(Json::as_i64);
            let n = match n {
                Some(n) if n > 0 => n as usize,
                _ => return Err(GraphDDBError::new(
                    "a maintenance `preview` projection has a non-positive-integer length bound.",
                )),
            };
            if value.is_null() {
                return Ok(value);
            }
            let s = crate::templates::json_to_template_string(&value);
            Ok(Json::String(s.chars().take(n).collect()))
        }
        other => Err(GraphDDBError::new(format!(
            "unknown maintenance projection op '{other}' (expected 'identity' / 'preview')."
        ))),
    }
}

// ── expansion + collapse ──────────────────────────────────────────────────────

/// Expand a transaction spec into serialized `TransactWriteItem`s (collapsed).
pub fn expand_transaction(
    rt: &GraphDDBRuntime,
    spec: &Json,
    params: &Params,
) -> Result<Vec<TransactWriteItem>> {
    let mut expanded: Vec<Built> = Vec::new();
    for item in spec
        .get("items")
        .and_then(Json::as_array)
        .cloned()
        .unwrap_or_default()
    {
        let for_each = item.get("forEach").filter(|v| !v.is_null());
        let when = item.get("when").filter(|v| !v.is_null());
        if let Some(fe) = for_each {
            let source = params.get(fe["source"].as_str().unwrap_or(""));
            let arr = match source.and_then(Json::as_array) {
                Some(a) => a.clone(),
                None => {
                    return Err(GraphDDBError::new(format!(
                        "forEach source '{}' must be an array param",
                        fe["source"].as_str().unwrap_or("")
                    )))
                }
            };
            for element in &arr {
                if let Some(w) = when {
                    if !when_holds(w, params, Some(element))? {
                        continue;
                    }
                }
                expanded.push(build_item(rt, &item, params, Some(element))?);
            }
        } else {
            if let Some(w) = when {
                if !when_holds(w, params, None)? {
                    continue;
                }
            }
            expanded.push(build_item(rt, &item, params, None)?);
        }
    }
    let collapsed = collapse_same_key_items(expanded)?;
    Ok(collapsed.into_iter().map(to_transact_item).collect())
}

/// Yield `(kind, op_spec, op_params)` for each expanded logical write — parity with
/// the Python `_transaction_logical_ops` (surfaced to W1; ConditionCheck skipped).
/// Returned as JSON tuples the caller can inspect; unused by the conformance
/// surface beyond existence, but kept for completeness / future W1 wiring.
pub fn transaction_logical_ops(
    spec: &Json,
    params: &Params,
) -> Result<Vec<(String, Json, Params)>> {
    let mut out = Vec::new();
    for item in spec
        .get("items")
        .and_then(Json::as_array)
        .cloned()
        .unwrap_or_default()
    {
        let kind = match item.get("type").and_then(Json::as_str) {
            Some("Put") => "put",
            Some("Update") => "update",
            Some("Delete") => "delete",
            _ => continue,
        };
        let for_each = item.get("forEach").filter(|v| !v.is_null());
        let when = item.get("when").filter(|v| !v.is_null());
        if let Some(fe) = for_each {
            let source = params.get(fe["source"].as_str().unwrap_or(""));
            let arr = match source.and_then(Json::as_array) {
                Some(a) => a.clone(),
                None => continue,
            };
            for element in &arr {
                if let Some(w) = when {
                    if !when_holds(w, params, Some(element))? {
                        continue;
                    }
                }
                let mut op_params = params.clone();
                if let Some(obj) = element.as_object() {
                    for (k, v) in obj {
                        op_params.insert(format!("item.{k}"), v.clone());
                    }
                    for (k, v) in obj {
                        op_params.insert(k.clone(), v.clone());
                    }
                }
                out.push((kind.to_string(), item.clone(), op_params));
            }
        } else {
            if let Some(w) = when {
                if !when_holds(w, params, None)? {
                    continue;
                }
            }
            out.push((kind.to_string(), item.clone(), params.clone()));
        }
    }
    Ok(out)
}

fn built_kind_name(b: &Built) -> &'static str {
    match b.kind {
        BuiltKind::Put { .. } => "Put",
        BuiltKind::Delete { .. } => "Delete",
        BuiltKind::ConditionCheck { .. } => "ConditionCheck",
        BuiltKind::Update { .. } => "Update",
    }
}

fn signature(b: &Built) -> String {
    format!("{}#{}#{}", b.table, b.pk, b.sk)
}

fn collapse_same_key_items(items: Vec<Built>) -> Result<Vec<Built>> {
    use std::collections::HashMap as Map;
    let mut groups: Map<String, Vec<usize>> = Map::new();
    for (i, it) in items.iter().enumerate() {
        groups.entry(signature(it)).or_default().push(i);
    }
    let mut out: Vec<Built> = Vec::new();
    // Iterate in original order; track which indices are consumed.
    for (i, it) in items.iter().enumerate() {
        let sig = signature(it);
        let bucket = &groups[&sig];
        if bucket.len() == 1 {
            out.push(clone_built(it));
            continue;
        }
        let mut kinds: Vec<&str> = bucket.iter().map(|&j| built_kind_name(&items[j])).collect();
        kinds.sort();
        kinds.dedup();
        if kinds == ["Delete", "Put"] {
            // swap pair no-op → drop every member.
            continue;
        }
        if kinds == ["Put", "Update"] {
            return Err(GraphDDBError::new(
                "a derived counter (`increment`) resolves to the SAME item as the entity write \
                 at runtime. A Put+Update on one transaction item is unsupported.",
            ));
        }
        // Non-first members are folded into / dropped by the first member's handling.
        if i != bucket[0] {
            continue;
        }
        if kinds == ["Update"] {
            out.push(merge_add_updates(bucket, &items)?);
        } else if kinds == ["Put"] || kinds == ["Delete"] {
            out.push(clone_built(it));
        } else {
            return Err(GraphDDBError::new(format!(
                "{} transaction items resolve to the same physical row with an unsupported op \
                 combination. A TransactWriteItems may not touch one key twice.",
                bucket.len()
            )));
        }
    }
    Ok(out)
}

fn merge_add_updates(bucket: &[usize], items: &[Built]) -> Result<Built> {
    let first = &items[bucket[0]];
    let (table, pk, sk, first_key) = match &first.kind {
        BuiltKind::Update { key, .. } => (
            first.table.clone(),
            first.pk.clone(),
            first.sk.clone(),
            key.clone(),
        ),
        _ => return Err(GraphDDBError::new("merge_add_updates on non-update")),
    };
    // Sum deltas per attribute (order = first appearance).
    let mut order: Vec<String> = Vec::new();
    let mut deltas: HashMap<String, i64> = HashMap::new();
    for &j in bucket {
        if let BuiltKind::Update { names, values, .. } = &items[j].kind {
            for (name_ph, attr) in names {
                let value_ph = format!(":{}", &name_ph[1..]);
                if let Some(AttributeValue::N(n)) = values.get(&value_ph) {
                    let d: i64 = n.parse().unwrap_or(0);
                    if !deltas.contains_key(attr) {
                        order.push(attr.clone());
                    }
                    *deltas.entry(attr.clone()).or_insert(0) += d;
                }
            }
        }
    }
    let mut names_out = HashMap::new();
    let mut values_out = HashMap::new();
    let mut clauses = Vec::new();
    for (i, attr) in order.iter().enumerate() {
        names_out.insert(format!("#a{i}"), attr.clone());
        values_out.insert(
            format!(":a{i}"),
            AttributeValue::N(deltas[attr].to_string()),
        );
        clauses.push(format!("#a{i} :a{i}"));
    }
    Ok(Built {
        kind: BuiltKind::Update {
            key: first_key,
            update_expression: Some(format!("ADD {}", clauses.join(", "))),
            names: names_out,
            values: values_out,
            condition: None,
        },
        table,
        pk,
        sk,
    })
}

fn clone_built(b: &Built) -> Built {
    Built {
        kind: match &b.kind {
            BuiltKind::Put { item, condition } => BuiltKind::Put {
                item: item.clone(),
                condition: condition.clone().map(|e| Expr {
                    expression: e.expression,
                    names: e.names,
                    values: e.values,
                }),
            },
            BuiltKind::Delete { key, condition } => BuiltKind::Delete {
                key: key.clone(),
                condition: condition.clone(),
            },
            BuiltKind::ConditionCheck { key, condition } => BuiltKind::ConditionCheck {
                key: key.clone(),
                condition: condition.clone(),
            },
            BuiltKind::Update {
                key,
                update_expression,
                names,
                values,
                condition,
            } => BuiltKind::Update {
                key: key.clone(),
                update_expression: update_expression.clone(),
                names: names.clone(),
                values: values.clone(),
                condition: condition.clone(),
            },
        },
        table: b.table.clone(),
        pk: b.pk.clone(),
        sk: b.sk.clone(),
    }
}

fn to_transact_item(b: Built) -> TransactWriteItem {
    match b.kind {
        BuiltKind::Put { item, condition } => {
            let mut put = Put::builder().table_name(b.table).set_item(Some(item));
            if let Some(e) = condition {
                put = put.condition_expression(e.expression);
                if !e.names.is_empty() {
                    put = put.set_expression_attribute_names(Some(e.names));
                }
                if !e.values.is_empty() {
                    put = put.set_expression_attribute_values(Some(e.values));
                }
            }
            TransactWriteItem::builder()
                .put(put.build().expect("Put build"))
                .build()
        }
        BuiltKind::Delete { key, condition } => {
            let mut del = Delete::builder().table_name(b.table).set_key(Some(key));
            if let Some(e) = condition {
                del = del.condition_expression(e.expression);
                if !e.names.is_empty() {
                    del = del.set_expression_attribute_names(Some(e.names));
                }
                if !e.values.is_empty() {
                    del = del.set_expression_attribute_values(Some(e.values));
                }
            }
            TransactWriteItem::builder()
                .delete(del.build().expect("Delete build"))
                .build()
        }
        BuiltKind::ConditionCheck { key, condition } => {
            let mut cc = ConditionCheck::builder()
                .table_name(b.table)
                .set_key(Some(key))
                .condition_expression(condition.expression);
            if !condition.names.is_empty() {
                cc = cc.set_expression_attribute_names(Some(condition.names));
            }
            if !condition.values.is_empty() {
                cc = cc.set_expression_attribute_values(Some(condition.values));
            }
            TransactWriteItem::builder()
                .condition_check(cc.build().expect("ConditionCheck build"))
                .build()
        }
        BuiltKind::Update {
            key,
            update_expression,
            names,
            values,
            condition,
        } => {
            let mut upd = Update::builder().table_name(b.table).set_key(Some(key));
            let mut all_names = names;
            let mut all_values = values;
            let mut cond_expr = None;
            if let Some(e) = condition {
                cond_expr = Some(e.expression);
                for (k, v) in e.names {
                    all_names.insert(k, v);
                }
                for (k, v) in e.values {
                    all_values.insert(k, v);
                }
            }
            if let Some(ue) = update_expression {
                upd = upd.update_expression(ue);
            }
            if let Some(ce) = cond_expr {
                upd = upd.condition_expression(ce);
            }
            if !all_names.is_empty() {
                upd = upd.set_expression_attribute_names(Some(all_names));
            }
            if !all_values.is_empty() {
                upd = upd.set_expression_attribute_values(Some(all_values));
            }
            TransactWriteItem::builder()
                .update(upd.build().expect("Update build"))
                .build()
        }
    }
}

/// The DynamoDB TransactWriteItems limit (exposed for the runtime's guard).
pub const MAX_TRANSACT_ITEMS_LIMIT: usize = MAX_TRANSACT_ITEMS;

// keep each_ph referenced (used indirectly by resolvers elsewhere in the future)
#[allow(dead_code)]
fn _keep_each_ph(t: &str) -> Vec<String> {
    each_ph(t)
}