/// Structured input for creating a {{ model_name }} record.
///
/// All fields are optional; unset fields are omitted from the INSERT.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}CreateInput {
{%- for field in create_fields %}
    /// {{ field.name }} column.
    pub {{ field.name }}: Option<{{ field.rust_type }}>,
{%- endfor %}
}

impl {{ model_name }}CreateInput {
    fn to_engine_data(&self) -> serde_json::Value {
        let mut data = serde_json::Map::new();
{%- for field in create_fields %}
        if let Some(value) = &self.{{ field.name }} {
            data.insert(
                "{{ field.db_name }}".to_string(),
{%- if field.is_optional %}
                match value {
                    Some(inner) => nautilus_core::Value::from(inner.clone()).to_json_plain(),
                    None => serde_json::Value::Null,
                },
{%- else %}
                nautilus_core::Value::from(value.clone()).to_json_plain(),
{%- endif %}
            );
        }
{%- endfor %}
        serde_json::Value::Object(data)
    }
}

/// Structured input for updating {{ model_name }} records.
///
/// All fields are optional; only set fields are included in the UPDATE.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}UpdateInput {
{%- for field in create_fields %}
    /// {{ field.name }} column.
    pub {{ field.name }}: Option<{{ field.rust_type }}>,
{%- endfor %}
}

impl {{ model_name }}UpdateInput {
    fn to_engine_data(&self) -> serde_json::Value {
        let mut data = serde_json::Map::new();
{%- for field in create_fields %}
        if let Some(value) = &self.{{ field.name }} {
            data.insert(
                "{{ field.db_name }}".to_string(),
{%- if field.is_optional %}
                match value {
                    Some(inner) => nautilus_core::Value::from(inner.clone()).to_json_plain(),
                    None => serde_json::Value::Null,
                },
{%- else %}
                nautilus_core::Value::from(value.clone()).to_json_plain(),
{%- endif %}
            );
        }
{%- endfor %}
        serde_json::Value::Object(data)
    }

    fn has_assignments(&self) -> bool {
{%- for field in create_fields %}
        self.{{ field.name }}.is_some()
{%- if not loop.last %} ||{% endif %}
{%- endfor %}
    }
}

/// Structured arguments for updating {{ model_name }} records.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}UpdateArgs {
    /// Optional WHERE filter — if `None`, all rows are updated.
    pub where_: Option<nautilus_core::Expr>,
    /// Fields to update.
    pub data: {{ model_name }}UpdateInput,
}

/// Structured arguments for deleting {{ model_name }} records.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}DeleteArgs {
    /// Optional WHERE filter — if `None`, all rows are deleted.
    pub where_: Option<nautilus_core::Expr>,
}

/// Structured arguments for upserting a {{ model_name }} record.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}UpsertArgs {
    /// WHERE filter used to look up the existing record (should target a unique field).
    pub where_: Option<nautilus_core::Expr>,
    /// Fields used when creating a new record (record not found).
    pub create: {{ model_name }}CreateInput,
    /// Fields applied when updating an existing record.
    pub update: {{ model_name }}UpdateInput,
    /// Whether to return the created/updated record (default: true).
    pub return_data: bool,
}

pub type {{ model_name }}CreateEventContext =
    crate::CrudEventContext<{{ model_name }}CreateInput, {{ model_name }}>;
pub type {{ model_name }}CreateManyEventContext =
    crate::CrudEventContext<Vec<{{ entry_name }}>, Vec<{{ model_name }}>>;
pub type {{ model_name }}UpdateEventContext =
    crate::CrudEventContext<{{ model_name }}UpdateArgs, Vec<{{ model_name }}>>;
pub type {{ model_name }}DeleteEventContext =
    crate::CrudEventContext<{{ model_name }}DeleteArgs, Option<{{ model_name }}>>;
pub type {{ model_name }}DeleteManyEventContext =
    crate::CrudEventContext<{{ model_name }}DeleteArgs, Vec<{{ model_name }}>>;

/// Scalar fields on {{ model_name }} that can be referenced in generated aggregate APIs.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum {{ model_name }}ScalarField {
{%- for field in scalar_fields %}
    {{ field.variant_name }},
{%- endfor %}
}

impl {{ model_name }}ScalarField {
    fn as_str(&self) -> &'static str {
        match self {
{%- for field in scalar_fields %}
            Self::{{ field.variant_name }} => "{{ field.logical_name }}",
{%- endfor %}
        }
    }
}

/// Sort direction used by {{ model_name }} `group_by()` ordering.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum {{ model_name }}SortOrder {
    Asc,
    Desc,
}

impl {{ model_name }}SortOrder {
    fn as_str(&self) -> &'static str {
        match self {
            Self::Asc => "asc",
            Self::Desc => "desc",
        }
    }
}

/// Structured arguments for counting {{ model_name }} records.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}CountArgs {
    /// Optional WHERE filter.
    pub where_: Option<nautilus_core::Expr>,
    /// Limit the counted window.
    pub take: Option<i32>,
    /// Offset the counted window.
    pub skip: Option<u32>,
    /// Optional keyset cursor for paginated counts.
    pub cursor: Option<std::collections::HashMap<String, nautilus_core::Value>>,
}

impl {{ model_name }}CountArgs {
    fn to_engine_args(&self) -> nautilus_core::Result<Option<serde_json::Value>> {
        let mut args = serde_json::Map::new();

        if let Some(filter) = self.where_.as_ref() {
            args.insert(
                "where".to_string(),
                nautilus_core::where_expr_to_protocol_json(filter)?,
            );
        }
        if let Some(take) = self.take {
            args.insert("take".to_string(), serde_json::Value::from(take));
        }
        if let Some(skip) = self.skip {
            args.insert("skip".to_string(), serde_json::Value::from(skip));
        }
        if let Some(cursor) = self.cursor.as_ref() {
            args.insert("cursor".to_string(), serialize_cursor_for_engine(cursor));
        }

        Ok((!args.is_empty()).then_some(serde_json::Value::Object(args)))
    }
}

/// Select which COUNT aggregates to return from `{{ delegate_name }}::group_by`.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}CountAggregateInput {
    pub _all: bool,
{%- for field in scalar_fields %}
    pub {{ field.name }}: bool,
{%- endfor %}
}

impl {{ model_name }}CountAggregateInput {
    fn to_json(&self) -> Option<serde_json::Value> {
        let mut obj = serde_json::Map::new();
        if self._all {
            obj.insert("_all".to_string(), serde_json::Value::Bool(true));
        }
{%- for field in scalar_fields %}
        if self.{{ field.name }} {
            obj.insert("{{ field.logical_name }}".to_string(), serde_json::Value::Bool(true));
        }
{%- endfor %}
        (!obj.is_empty()).then_some(serde_json::Value::Object(obj))
    }
}

{%- if has_numeric_fields %}
/// Select which AVG aggregates to return from `{{ delegate_name }}::group_by`.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}AvgAggregateInput {
{%- for field in numeric_fields %}
    pub {{ field.name }}: bool,
{%- endfor %}
}

impl {{ model_name }}AvgAggregateInput {
    fn to_json(&self) -> Option<serde_json::Value> {
        let mut obj = serde_json::Map::new();
{%- for field in numeric_fields %}
        if self.{{ field.name }} {
            obj.insert("{{ field.logical_name }}".to_string(), serde_json::Value::Bool(true));
        }
{%- endfor %}
        (!obj.is_empty()).then_some(serde_json::Value::Object(obj))
    }
}

/// Select which SUM aggregates to return from `{{ delegate_name }}::group_by`.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}SumAggregateInput {
{%- for field in numeric_fields %}
    pub {{ field.name }}: bool,
{%- endfor %}
}

impl {{ model_name }}SumAggregateInput {
    fn to_json(&self) -> Option<serde_json::Value> {
        let mut obj = serde_json::Map::new();
{%- for field in numeric_fields %}
        if self.{{ field.name }} {
            obj.insert("{{ field.logical_name }}".to_string(), serde_json::Value::Bool(true));
        }
{%- endfor %}
        (!obj.is_empty()).then_some(serde_json::Value::Object(obj))
    }
}
{%- endif %}

/// Select which MIN aggregates to return from `{{ delegate_name }}::group_by`.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}MinAggregateInput {
{%- for field in orderable_fields %}
    pub {{ field.name }}: bool,
{%- endfor %}
}

impl {{ model_name }}MinAggregateInput {
    fn to_json(&self) -> Option<serde_json::Value> {
        let mut obj = serde_json::Map::new();
{%- for field in orderable_fields %}
        if self.{{ field.name }} {
            obj.insert("{{ field.logical_name }}".to_string(), serde_json::Value::Bool(true));
        }
{%- endfor %}
        (!obj.is_empty()).then_some(serde_json::Value::Object(obj))
    }
}

/// Select which MAX aggregates to return from `{{ delegate_name }}::group_by`.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}MaxAggregateInput {
{%- for field in orderable_fields %}
    pub {{ field.name }}: bool,
{%- endfor %}
}

impl {{ model_name }}MaxAggregateInput {
    fn to_json(&self) -> Option<serde_json::Value> {
        let mut obj = serde_json::Map::new();
{%- for field in orderable_fields %}
        if self.{{ field.name }} {
            obj.insert("{{ field.logical_name }}".to_string(), serde_json::Value::Bool(true));
        }
{%- endfor %}
        (!obj.is_empty()).then_some(serde_json::Value::Object(obj))
    }
}

/// One ORDER BY item for `{{ delegate_name }}::group_by()`.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum {{ model_name }}GroupByOrderBy {
    Field {
        field: {{ model_name }}ScalarField,
        direction: {{ model_name }}SortOrder,
    },
    CountAll({{ model_name }}SortOrder),
    Count {
        field: {{ model_name }}ScalarField,
        direction: {{ model_name }}SortOrder,
    },
{%- if has_numeric_fields %}
    Avg {
        field: {{ model_name }}ScalarField,
        direction: {{ model_name }}SortOrder,
    },
    Sum {
        field: {{ model_name }}ScalarField,
        direction: {{ model_name }}SortOrder,
    },
{%- endif %}
{%- if has_orderable_fields %}
    Min {
        field: {{ model_name }}ScalarField,
        direction: {{ model_name }}SortOrder,
    },
    Max {
        field: {{ model_name }}ScalarField,
        direction: {{ model_name }}SortOrder,
    },
{%- endif %}
}

impl {{ model_name }}GroupByOrderBy {
    fn to_json(&self) -> serde_json::Value {
        match self {
            Self::Field { field, direction } => {
                let mut obj = serde_json::Map::new();
                obj.insert(
                    field.as_str().to_string(),
                    serde_json::Value::String(direction.as_str().to_string()),
                );
                serde_json::Value::Object(obj)
            }
            Self::CountAll(direction) => {
                let mut inner = serde_json::Map::new();
                inner.insert(
                    "_all".to_string(),
                    serde_json::Value::String(direction.as_str().to_string()),
                );
                let mut outer = serde_json::Map::new();
                outer.insert("_count".to_string(), serde_json::Value::Object(inner));
                serde_json::Value::Object(outer)
            }
            Self::Count { field, direction } => {
                let mut inner = serde_json::Map::new();
                inner.insert(
                    field.as_str().to_string(),
                    serde_json::Value::String(direction.as_str().to_string()),
                );
                let mut outer = serde_json::Map::new();
                outer.insert("_count".to_string(), serde_json::Value::Object(inner));
                serde_json::Value::Object(outer)
            }
{%- if has_numeric_fields %}
            Self::Avg { field, direction } => {
                let mut inner = serde_json::Map::new();
                inner.insert(
                    field.as_str().to_string(),
                    serde_json::Value::String(direction.as_str().to_string()),
                );
                let mut outer = serde_json::Map::new();
                outer.insert("_avg".to_string(), serde_json::Value::Object(inner));
                serde_json::Value::Object(outer)
            }
            Self::Sum { field, direction } => {
                let mut inner = serde_json::Map::new();
                inner.insert(
                    field.as_str().to_string(),
                    serde_json::Value::String(direction.as_str().to_string()),
                );
                let mut outer = serde_json::Map::new();
                outer.insert("_sum".to_string(), serde_json::Value::Object(inner));
                serde_json::Value::Object(outer)
            }
{%- endif %}
{%- if has_orderable_fields %}
            Self::Min { field, direction } => {
                let mut inner = serde_json::Map::new();
                inner.insert(
                    field.as_str().to_string(),
                    serde_json::Value::String(direction.as_str().to_string()),
                );
                let mut outer = serde_json::Map::new();
                outer.insert("_min".to_string(), serde_json::Value::Object(inner));
                serde_json::Value::Object(outer)
            }
            Self::Max { field, direction } => {
                let mut inner = serde_json::Map::new();
                inner.insert(
                    field.as_str().to_string(),
                    serde_json::Value::String(direction.as_str().to_string()),
                );
                let mut outer = serde_json::Map::new();
                outer.insert("_max".to_string(), serde_json::Value::Object(inner));
                serde_json::Value::Object(outer)
            }
{%- endif %}
        }
    }
}

/// Structured arguments for grouping {{ model_name }} rows and computing aggregates.
#[derive(Debug, Default, Clone)]
pub struct {{ model_name }}GroupByArgs {
    /// Logical field names to group by.
    pub by: Vec<{{ model_name }}ScalarField>,
    /// Optional pre-group WHERE filter.
    pub where_: Option<nautilus_core::Expr>,
    /// Optional HAVING JSON object using the protocol aggregate shape.
    pub having: Option<serde_json::Value>,
    /// Limit the number of returned groups.
    pub take: Option<i32>,
    /// Offset the returned groups.
    pub skip: Option<u32>,
    /// COUNT aggregation selection.
    pub count: Option<{{ model_name }}CountAggregateInput>,
{%- if has_numeric_fields %}
    /// AVG aggregation selection.
    pub avg: Option<{{ model_name }}AvgAggregateInput>,
    /// SUM aggregation selection.
    pub sum: Option<{{ model_name }}SumAggregateInput>,
{%- endif %}
    /// MIN aggregation selection.
    pub min: Option<{{ model_name }}MinAggregateInput>,
    /// MAX aggregation selection.
    pub max: Option<{{ model_name }}MaxAggregateInput>,
    /// ORDER BY items applied after grouping.
    pub order_by: Vec<{{ model_name }}GroupByOrderBy>,
}

impl {{ model_name }}GroupByArgs {
    fn to_engine_args(&self) -> nautilus_core::Result<serde_json::Value> {
        let mut args = serde_json::Map::new();
        args.insert(
            "by".to_string(),
            serde_json::Value::Array(
                self.by
                    .iter()
                    .map(|field| serde_json::Value::String(field.as_str().to_string()))
                    .collect(),
            ),
        );

        if let Some(filter) = self.where_.as_ref() {
            args.insert(
                "where".to_string(),
                nautilus_core::where_expr_to_protocol_json(filter)?,
            );
        }
        if let Some(having) = self.having.as_ref() {
            args.insert("having".to_string(), having.clone());
        }
        if let Some(take) = self.take {
            args.insert("take".to_string(), serde_json::Value::from(take));
        }
        if let Some(skip) = self.skip {
            args.insert("skip".to_string(), serde_json::Value::from(skip));
        }
        if let Some(count) = self.count.as_ref().and_then({{ model_name }}CountAggregateInput::to_json) {
            args.insert("count".to_string(), count);
        }
{%- if has_numeric_fields %}
        if let Some(avg) = self.avg.as_ref().and_then({{ model_name }}AvgAggregateInput::to_json) {
            args.insert("avg".to_string(), avg);
        }
        if let Some(sum) = self.sum.as_ref().and_then({{ model_name }}SumAggregateInput::to_json) {
            args.insert("sum".to_string(), sum);
        }
{%- endif %}
        if let Some(min) = self.min.as_ref().and_then({{ model_name }}MinAggregateInput::to_json) {
            args.insert("min".to_string(), min);
        }
        if let Some(max) = self.max.as_ref().and_then({{ model_name }}MaxAggregateInput::to_json) {
            args.insert("max".to_string(), max);
        }
        if !self.order_by.is_empty() {
            args.insert(
                "orderBy".to_string(),
                serde_json::Value::Array(
                    self.order_by
                        .iter()
                        .map({{ model_name }}GroupByOrderBy::to_json)
                        .collect(),
                ),
            );
        }

        Ok(serde_json::Value::Object(args))
    }
}

/// Typed COUNT aggregates returned by `{{ delegate_name }}::group_by()`.
#[derive(Debug, Default, Clone, PartialEq)]
pub struct {{ model_name }}CountAggregateOutput {
    pub _all: Option<i64>,
{%- for field in scalar_fields %}
    pub {{ field.name }}: Option<i64>,
{%- endfor %}
}

impl {{ model_name }}CountAggregateOutput {
    fn from_json_object(
        obj: &serde_json::Map<String, serde_json::Value>,
    ) -> nautilus_core::Result<Self> {
        Ok(Self {
            _all: decode_optional_group_by_json_value(obj, "_all")?,
{%- for field in scalar_fields %}
            {{ field.name }}: decode_optional_group_by_json_value(obj, "{{ field.logical_name }}")?,
{%- endfor %}
        })
    }
}

{%- if has_numeric_fields %}
/// Typed AVG aggregates returned by `{{ delegate_name }}::group_by()`.
#[derive(Debug, Default, Clone, PartialEq)]
pub struct {{ model_name }}AvgAggregateOutput {
{%- for field in numeric_fields %}
    pub {{ field.name }}: Option<{{ field.avg_rust_type }}>,
{%- endfor %}
}

impl {{ model_name }}AvgAggregateOutput {
    fn from_json_object(
        obj: &serde_json::Map<String, serde_json::Value>,
    ) -> nautilus_core::Result<Self> {
        Ok(Self {
{%- for field in numeric_fields %}
            {{ field.name }}: decode_optional_group_by_json_value(obj, "{{ field.logical_name }}")?,
{%- endfor %}
        })
    }
}

/// Typed SUM aggregates returned by `{{ delegate_name }}::group_by()`.
#[derive(Debug, Default, Clone, PartialEq)]
pub struct {{ model_name }}SumAggregateOutput {
{%- for field in numeric_fields %}
    pub {{ field.name }}: Option<{{ field.sum_rust_type }}>,
{%- endfor %}
}

impl {{ model_name }}SumAggregateOutput {
    fn from_json_object(
        obj: &serde_json::Map<String, serde_json::Value>,
    ) -> nautilus_core::Result<Self> {
        Ok(Self {
{%- for field in numeric_fields %}
            {{ field.name }}: decode_optional_group_by_json_value(obj, "{{ field.logical_name }}")?,
{%- endfor %}
        })
    }
}
{%- endif %}

/// Typed MIN aggregates returned by `{{ delegate_name }}::group_by()`.
#[derive(Debug, Default, Clone, PartialEq)]
pub struct {{ model_name }}MinAggregateOutput {
{%- for field in orderable_fields %}
    pub {{ field.name }}: Option<{{ field.rust_type }}>,
{%- endfor %}
}

impl {{ model_name }}MinAggregateOutput {
    fn from_json_object(
        obj: &serde_json::Map<String, serde_json::Value>,
    ) -> nautilus_core::Result<Self> {
        Ok(Self {
{%- for field in orderable_fields %}
            {{ field.name }}: decode_optional_group_by_json_value(obj, "{{ field.logical_name }}")?,
{%- endfor %}
        })
    }
}

/// Typed MAX aggregates returned by `{{ delegate_name }}::group_by()`.
#[derive(Debug, Default, Clone, PartialEq)]
pub struct {{ model_name }}MaxAggregateOutput {
{%- for field in orderable_fields %}
    pub {{ field.name }}: Option<{{ field.rust_type }}>,
{%- endfor %}
}

impl {{ model_name }}MaxAggregateOutput {
    fn from_json_object(
        obj: &serde_json::Map<String, serde_json::Value>,
    ) -> nautilus_core::Result<Self> {
        Ok(Self {
{%- for field in orderable_fields %}
            {{ field.name }}: decode_optional_group_by_json_value(obj, "{{ field.logical_name }}")?,
{%- endfor %}
        })
    }
}

/// One typed row returned by `{{ delegate_name }}::group_by()`.
#[derive(Debug, Default, Clone, PartialEq)]
pub struct {{ model_name }}GroupByOutput {
{%- for field in scalar_fields %}
    pub {{ field.name }}: Option<{{ field.base_rust_type }}>,
{%- endfor %}
    pub _count: Option<{{ model_name }}CountAggregateOutput>,
{%- if has_numeric_fields %}
    pub _avg: Option<{{ model_name }}AvgAggregateOutput>,
    pub _sum: Option<{{ model_name }}SumAggregateOutput>,
{%- endif %}
    pub _min: Option<{{ model_name }}MinAggregateOutput>,
    pub _max: Option<{{ model_name }}MaxAggregateOutput>,
}

impl {{ model_name }}GroupByOutput {
    fn from_row(row: &crate::Row) -> nautilus_core::Result<Self> {
        Ok(Self {
{%- for field in scalar_fields %}
            {{ field.name }}: decode_optional_group_by_row_value(row, "{{ field.logical_name }}")?,
{%- endfor %}
            _count: decode_optional_group_by_object(row, "_count")?
                .map({{ model_name }}CountAggregateOutput::from_json_object)
                .transpose()?,
{%- if has_numeric_fields %}
            _avg: decode_optional_group_by_object(row, "_avg")?
                .map({{ model_name }}AvgAggregateOutput::from_json_object)
                .transpose()?,
            _sum: decode_optional_group_by_object(row, "_sum")?
                .map({{ model_name }}SumAggregateOutput::from_json_object)
                .transpose()?,
{%- endif %}
            _min: decode_optional_group_by_object(row, "_min")?
                .map({{ model_name }}MinAggregateOutput::from_json_object)
                .transpose()?,
            _max: decode_optional_group_by_object(row, "_max")?
                .map({{ model_name }}MaxAggregateOutput::from_json_object)
                .transpose()?,
        })
    }
}

fn serialize_cursor_for_engine(
    cursor: &std::collections::HashMap<String, nautilus_core::Value>,
) -> serde_json::Value {
    serde_json::Value::Object(
        cursor
            .iter()
            .map(|(key, value)| (key.clone(), value.to_json_plain()))
            .collect(),
    )
}

fn decode_optional_group_by_row_value<T>(
    row: &crate::Row,
    key: &str,
) -> nautilus_core::Result<Option<T>>
where
    T: nautilus_core::FromValue,
{
    match row.get(key) {
        Some(nautilus_core::Value::Null) | None => Ok(None),
        Some(value) => <T as nautilus_core::FromValue>::from_value(value).map(Some),
    }
}

fn decode_optional_group_by_object<'a>(
    row: &'a crate::Row,
    key: &str,
) -> nautilus_core::Result<Option<&'a serde_json::Map<String, serde_json::Value>>> {
    match row.get(key) {
        Some(nautilus_core::Value::Json(serde_json::Value::Object(obj))) => Ok(Some(obj)),
        Some(nautilus_core::Value::Null) | None => Ok(None),
        Some(other) => Err(nautilus_core::Error::TypeError(format!(
            "expected object-valued aggregate '{}' in groupBy output, got {:?}",
            key, other
        ))),
    }
}

fn decode_optional_group_by_json_value<T>(
    obj: &serde_json::Map<String, serde_json::Value>,
    key: &str,
) -> nautilus_core::Result<Option<T>>
where
    T: nautilus_core::FromValue,
{
    match obj.get(key) {
        Some(serde_json::Value::Null) | None => Ok(None),
        Some(value) => <T as nautilus_core::FromValue>::from_value_owned(
            crate::runtime::wire_value_to_core_value(key, value),
        )
        .map(Some),
    }
}

fn serialize_update_filter_for_engine(
    filter: Option<&nautilus_core::Expr>,
) -> nautilus_core::Result<Option<serde_json::Value>> {
    let filter_json = match filter {
        Some(expr) => match nautilus_core::where_expr_to_protocol_json(expr) {
            Ok(value) => value,
            Err(nautilus_core::Error::InvalidQuery(_)) => return Ok(None),
            Err(err) => return Err(err),
        },
        None => return Ok(None),
    };

    Ok(Some(filter_json))
}

fn single_record_filter_field_name(column: &str) -> Option<&'static str> {
{%- for field in scalar_fields %}
    if column == "{{ table_name }}__{{ field.db_name }}"
        || column == "{{ field.db_name }}"
{%- if field.logical_name != field.db_name %}
        || column == "{{ field.logical_name }}"
{%- endif %}
    {
        return Some("{{ field.name }}");
    }
{%- endfor %}
    None
}

fn collect_single_record_filter_fields(
    expr: &nautilus_core::Expr,
    fields: &mut std::collections::HashSet<&'static str>,
) -> bool {
    use nautilus_core::{BinaryOp, Expr};

    match expr {
        Expr::Binary {
            left,
            op: BinaryOp::And,
            right,
        } => {
            collect_single_record_filter_fields(left, fields)
                && collect_single_record_filter_fields(right, fields)
        }
        Expr::Binary {
            left,
            op: BinaryOp::Eq,
            right,
        } => match (&**left, &**right) {
            (Expr::Column(column), Expr::Param(_)) | (Expr::Param(_), Expr::Column(column)) => {
                let Some(field_name) = single_record_filter_field_name(column) else {
                    return false;
                };
                fields.insert(field_name)
            }
            _ => false,
        },
        _ => false,
    }
}

fn matches_single_record_constraint(
    fields: &std::collections::HashSet<&'static str>,
    constraint: &[&str],
) -> bool {
    fields.len() == constraint.len() && constraint.iter().all(|field| fields.contains(field))
}

fn is_single_record_filter(filter: &nautilus_core::Expr) -> bool {
    const SINGLE_RECORD_CONSTRAINTS: &[&[&str]] = &[
{%- for constraint in single_record_constraints %}
        &[{% for field in constraint %}"{{ field.name }}"{% if not loop.last %}, {% endif %}{% endfor %}],
{%- endfor %}
    ];

    let mut fields = std::collections::HashSet::new();
    if !collect_single_record_filter_fields(filter, &mut fields) {
        return false;
    }

    SINGLE_RECORD_CONSTRAINTS
        .iter()
        .any(|constraint| matches_single_record_constraint(&fields, constraint))
}

/// Delegate for {{ model_name }} queries.
///
/// Obtained via `client.{{ table_name }}` (or the model accessor). Each method
/// accepts a structured args object and executes the query immediately — no
/// extra `.exec()` call is needed.
pub struct {{ delegate_name }}<E: crate::Executor> {
    client: crate::Client<E>,
}

impl<E> {{ delegate_name }}<E>
where
    E: crate::Executor,
    for<'a> E::Row<'a>: Into<crate::Row>,
{
    pub(crate) fn new(client: crate::Client<E>) -> Self {
        {{ delegate_name }} { client }
    }

    fn build_find_many_builder(
        client: crate::Client<E>,
        args: nautilus_core::FindManyArgs,
    ) -> nautilus_core::Result<{{ find_many_name }}<E>> {
        Self::reject_select_for_model_result(&args.select)?;
        if !args.include.is_empty() {
            return Err(nautilus_core::Error::InvalidQuery(
                "include queries require the embedded engine path in the generated Rust client"
                    .to_string(),
            ));
        }

        let mut builder = {{ find_many_name }}::new(client);
        if let Some(filter) = args.where_ {
            builder = builder.where_(filter);
        }
        for order in args.order_by {
            builder = builder.order_by(order);
        }
        if let Some(take) = args.take {
            builder = builder.take(take);
        }
        if let Some(skip) = args.skip {
            builder = builder.skip(skip);
        }
        if let Some(cursor) = args.cursor {
            builder = builder.cursor(cursor);
        }
        if !args.distinct.is_empty() {
            builder = builder.distinct(args.distinct);
        }

        Ok(builder)
    }

    fn reject_select_for_model_result(
        select: &std::collections::HashMap<String, bool>,
    ) -> nautilus_core::Result<()> {
        if !select.is_empty() {
            return Err(nautilus_core::Error::InvalidQuery(
                "select returns partial rows and cannot be decoded as a full {{ model_name }}; use find_many_select, find_first_select, or find_unique_select instead"
                    .to_string(),
            ));
        }
        Ok(())
    }

    fn engine_decode_plan_for_find_many_args(
        args: &nautilus_core::FindManyArgs,
    ) -> (Vec<usize>, Vec<Option<crate::ValueHint>>) {
        Self::engine_decode_plan_for_select(&args.select)
    }

    fn engine_decode_plan_for_find_unique_args(
        args: &nautilus_core::FindUniqueArgs,
    ) -> (Vec<usize>, Vec<Option<crate::ValueHint>>) {
        Self::engine_decode_plan_for_select(&args.select)
    }

    fn engine_decode_plan_for_select(
        select: &std::collections::HashMap<String, bool>,
    ) -> (Vec<usize>, Vec<Option<crate::ValueHint>>) {
        let selected_fields: std::collections::HashSet<String> = select
            .iter()
            .filter_map(|(key, enabled)| enabled.then_some(key.clone()))
            .collect();

        (
            projected_scalar_decode_slots(&selected_fields),
            projected_scalar_value_hints(&selected_fields),
        )
    }

    /// Find all {{ model_name }} records matching `args`.
    ///
    /// Executes immediately and returns a `Vec<{{ model_name }}>`.
    /// Prefer this when the caller naturally wants the whole result set in
    /// memory, such as small pages, eager-loaded relation graphs, or code that
    /// does further in-memory sorting/grouping.
    pub {% if is_async %}async {% endif %}fn find_many(
        &self,
        args: nautilus_core::FindManyArgs,
    ) -> nautilus_core::Result<Vec<{{ model_name }}>> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        Self::reject_select_for_model_result(&args.select)?;
        let (engine_decode_slots, engine_row_hints) =
            Self::engine_decode_plan_for_find_many_args(&args);
        if let Some(rows) =
            crate::runtime::try_find_many_via_engine::<_, {{ model_name }}>(
                {% if is_async %}&self.client{% else %}&_c{% endif %},
                "{{ model_name }}",
                &args,
                move |row| {
                    decode_model_row_with_hints(row, &engine_decode_slots, &engine_row_hints)
                },
            )
            .await?
        {
            return Ok(rows);
        }

        let builder = Self::build_find_many_builder(
            {% if is_async %}self.client.clone(){% else %}_c{% endif %},
            args,
        )?;
        builder.exec().await
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Find the first {{ model_name }} record matching `args`, or `None`.
    pub {% if is_async %}async {% endif %}fn find_first(
        &self,
        args: nautilus_core::FindManyArgs,
    ) -> nautilus_core::Result<Option<{{ model_name }}>> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        Self::reject_select_for_model_result(&args.select)?;
        let args = nautilus_core::FindManyArgs { take: Some(1), ..args };
        let (engine_decode_slots, engine_row_hints) =
            Self::engine_decode_plan_for_find_many_args(&args);
        if let Some(rows) =
            crate::runtime::try_find_many_via_engine::<_, {{ model_name }}>(
                {% if is_async %}&self.client{% else %}&_c{% endif %},
                "{{ model_name }}",
                &args,
                move |row| {
                    decode_model_row_with_hints(row, &engine_decode_slots, &engine_row_hints)
                },
            )
            .await?
        {
            return Ok(rows.into_iter().next());
        }

        let builder = Self::build_find_many_builder(
            {% if is_async %}self.client.clone(){% else %}_c{% endif %},
            args,
        )?;
        builder.exec_optional().await
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Find a single {{ model_name }} record by a unique filter, or `None` if not found.
    ///
    /// `args.where_` should reference a `@id` or `@unique` field for correct semantics.
    pub {% if is_async %}async {% endif %}fn find_unique(
        &self,
        args: nautilus_core::FindUniqueArgs,
    ) -> nautilus_core::Result<Option<{{ model_name }}>> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        Self::reject_select_for_model_result(&args.select)?;
        let (engine_decode_slots, engine_row_hints) =
            Self::engine_decode_plan_for_find_unique_args(&args);
        if let Some(record) =
            crate::runtime::try_find_unique_via_engine::<_, {{ model_name }}>(
                {% if is_async %}&self.client{% else %}&_c{% endif %},
                "{{ model_name }}",
                &args,
                move |row| {
                    decode_model_row_with_hints(row, &engine_decode_slots, &engine_row_hints)
                },
            )
            .await?
        {
            return Ok(Some(record));
        }

        let find_args = nautilus_core::FindManyArgs {
            where_: Some(args.where_),
            take: Some(1),
            select: args.select,
            include: args.include,
            ..Default::default()
        };
        let builder = Self::build_find_many_builder(
            {% if is_async %}self.client.clone(){% else %}_c{% endif %},
            find_args,
        )?;
        builder.exec_optional().await
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Find a single {{ model_name }} record by a unique filter, or return `Error::NotFound`.
    pub {% if is_async %}async {% endif %}fn find_unique_or_throw(
        &self,
        args: nautilus_core::FindUniqueArgs,
    ) -> nautilus_core::Result<{{ model_name }}> {
        self.find_unique(args){% if is_async %}.await{% endif %}?.ok_or_else(|| {
            nautilus_core::Error::NotFound(
                format!("No {} record found matching the given unique filter", "{{ model_name }}")
            )
        })
    }

    /// Find the first {{ model_name }} record matching `args`, or return `Error::NotFound`.
    pub {% if is_async %}async {% endif %}fn find_first_or_throw(
        &self,
        args: nautilus_core::FindManyArgs,
    ) -> nautilus_core::Result<{{ model_name }}> {
        self.find_first(args){% if is_async %}.await{% endif %}?.ok_or_else(|| {
            nautilus_core::Error::NotFound(
                format!("No {} record found matching the given filter", "{{ model_name }}")
            )
        })
    }

    /// Find all {{ model_name }} records with a typed scalar projection.
    pub {% if is_async %}async {% endif %}fn find_many_select<C, F>(
        &self,
        args: nautilus_core::FindManyArgs,
        columns: F,
    ) -> nautilus_core::Result<Vec<C::Output>>
    where
        C: nautilus_core::SelectColumns,
        F: FnOnce({{ columns_name }}) -> C,
    {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        if !args.include.is_empty() {
            return Err(nautilus_core::Error::InvalidQuery(
                "include is not supported by typed select projections".to_string(),
            ));
        }
        if !args.select.is_empty() {
            return Err(nautilus_core::Error::InvalidQuery(
                "FindManyArgs.select is ignored by typed select projections; pass projected columns to find_many_select instead"
                    .to_string(),
            ));
        }
        let selection = columns({{ columns_name }});
        let builder = Self::build_find_many_builder(
            {% if is_async %}self.client.clone(){% else %}_c{% endif %},
            nautilus_core::FindManyArgs {
                select: Default::default(),
                include: Default::default(),
                ..args
            },
        )?;
        builder.exec_select(selection).await
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Find the first {{ model_name }} record with a typed scalar projection.
    pub {% if is_async %}async {% endif %}fn find_first_select<C, F>(
        &self,
        args: nautilus_core::FindManyArgs,
        columns: F,
    ) -> nautilus_core::Result<Option<C::Output>>
    where
        C: nautilus_core::SelectColumns,
        F: FnOnce({{ columns_name }}) -> C,
    {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        if !args.include.is_empty() {
            return Err(nautilus_core::Error::InvalidQuery(
                "include is not supported by typed select projections".to_string(),
            ));
        }
        if !args.select.is_empty() {
            return Err(nautilus_core::Error::InvalidQuery(
                "FindManyArgs.select is ignored by typed select projections; pass projected columns to find_first_select instead"
                    .to_string(),
            ));
        }
        let selection = columns({{ columns_name }});
        let args = nautilus_core::FindManyArgs {
            take: Some(1),
            select: Default::default(),
            include: Default::default(),
            ..args
        };
        let builder = Self::build_find_many_builder(
            {% if is_async %}self.client.clone(){% else %}_c{% endif %},
            args,
        )?;
        builder.exec_select_optional(selection).await
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Find a unique {{ model_name }} record with a typed scalar projection.
    pub {% if is_async %}async {% endif %}fn find_unique_select<C, F>(
        &self,
        args: nautilus_core::FindUniqueArgs,
        columns: F,
    ) -> nautilus_core::Result<Option<C::Output>>
    where
        C: nautilus_core::SelectColumns,
        F: FnOnce({{ columns_name }}) -> C,
    {
        if !args.select.is_empty() {
            return Err(nautilus_core::Error::InvalidQuery(
                "FindUniqueArgs.select is ignored by typed select projections; pass projected columns to find_unique_select instead"
                    .to_string(),
            ));
        }
        if !args.include.is_empty() {
            return Err(nautilus_core::Error::InvalidQuery(
                "include is not supported by typed select projections".to_string(),
            ));
        }
        let find_args = nautilus_core::FindManyArgs {
            where_: Some(args.where_),
            take: Some(1),
            select: Default::default(),
            include: Default::default(),
            ..Default::default()
        };
        self.find_first_select(find_args, columns){% if is_async %}.await{% endif %}
    }
{% if is_async %}
    /// Stream {{ model_name }} records matching `args` row-by-row.
    ///
    /// Unlike [`Self::find_many`], which materializes the full result set
    /// into a `Vec<{{ model_name }}>`, `stream_many` returns an async stream
    /// that yields each decoded record as it arrives from the database. This
    /// keeps client memory bounded for large reads (exports, scans) and
    /// reduces time-to-first-row.
    ///
    /// Prefer [`Self::find_many`] for small result sets or when the caller
    /// needs a complete `Vec<{{ model_name }}>` before proceeding.
    ///
    /// The stream owns the database connection for the duration of the
    /// iteration; dropping the stream early releases the connection cleanly
    /// because the underlying worker drains the remaining rows before
    /// returning the connection to the pool.
    ///
    /// Backward pagination (`take` < 0) is not supported because the engine
    /// would have to buffer the full window before reversing it; use
    /// [`Self::find_many`] for that case.
    ///
    /// Note: `stream_many` always runs against the connector directly. The
    /// embedded-engine path used by `find_many` is not available here.
    pub fn stream_many(
        &self,
        args: nautilus_core::FindManyArgs,
    ) -> nautilus_core::Result<
        impl futures::stream::Stream<Item = nautilus_core::Result<{{ model_name }}>>
            + Send
            + 'static,
    >
    where
        E: 'static,
    {
        let builder = Self::build_find_many_builder(self.client.clone(), args)?;
        let (client, sql, row_hints, decode_slots, backward) = builder.into_execution_parts()?;
        if backward {
            return Err(nautilus_core::Error::InvalidQuery(
                "stream_many does not support backward pagination — use find_many for negative take values"
                    .to_string(),
            ));
        }
        Ok(async_stream::stream! {
            use futures::stream::StreamExt;
            let mut rows = client.executor().execute_owned(sql);
            while let Some(item) = rows.next().await {
                match item {
                    Ok(raw) => {
                        let row: crate::Row = raw.into();
                        yield decode_model_row_with_hints(row, &decode_slots, &row_hints);
                    }
                    Err(e) => {
                        yield Err(nautilus_core::Error::Other(e.to_string()));
                        break;
                    }
                }
            }
        })
    }
{% endif %}

    /// Create a new {{ model_name }} record.
    ///
    /// Executes immediately and returns the created `{{ model_name }}`.
    pub {% if is_async %}async {% endif %}fn create(
        &self,
        data: {{ model_name }}CreateInput,
    ) -> nautilus_core::Result<{{ model_name }}> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        let _events = {% if is_async %}self.client.events().clone(){% else %}_c.events().clone(){% endif %};
        let _transaction_id = {% if is_async %}self.client.transaction_id(){% else %}_c.transaction_id(){% endif %};
        let _event_args = data.clone();
        let _event_payload = data.to_engine_data();
        let mut _event_state = std::collections::HashMap::new();
        let mut _before_ctx = {{ model_name }}CreateEventContext::new(
            "{{ model_name }}",
            crate::CrudOperation::Create,
            crate::EventPhase::Before,
            _event_args.clone(),
            _event_payload.clone(),
            _event_state,
        )
        .with_transaction_id(_transaction_id.clone());
        if let Some(_stop) = _events.run::<{{ model_name }}CreateEventContext, {{ model_name }}>(
            "{{ model_name }}",
            crate::CrudOperation::Create,
            crate::EventPhase::Before,
            &mut _before_ctx,
        ).await? {
            return Ok(_stop);
        }
        _event_state = _before_ctx.state;

        let _operation_result: nautilus_core::Result<{{ model_name }}> = async {
            if let Some(record) =
                crate::runtime::try_create_via_engine::<_, {{ model_name }}>(
                    {% if is_async %}&self.client{% else %}&_c{% endif %},
                    "{{ model_name }}",
                    _event_payload.clone(),
                    |row| {
                        decode_model_row_with_hints(
                            row,
                            all_scalar_decode_slots(),
                            all_scalar_value_hints(),
                        )
                    },
                )
                .await?
            {
                Ok(record)
            } else {
                let mut builder = {{ create_name }}::new({% if is_async %}self.client.clone(){% else %}_c{% endif %});
{%- for field in create_fields %}
                if let Some(value) = data.{{ field.name }} {
                    builder = builder.{{ field.name }}(value);
                }
{%- endfor %}
                builder.exec().await
            }
        }.await;

        match _operation_result {
            Ok(_decoded) => {
                let mut _after_ctx = {{ model_name }}CreateEventContext::new(
                    "{{ model_name }}",
                    crate::CrudOperation::Create,
                    crate::EventPhase::After,
                    _event_args,
                    _event_payload,
                    _event_state,
                )
                .with_result(_decoded.clone())
                .with_transaction_id(_transaction_id);
                if let Some(_stop) = _events.run::<{{ model_name }}CreateEventContext, {{ model_name }}>(
                    "{{ model_name }}",
                    crate::CrudOperation::Create,
                    crate::EventPhase::After,
                    &mut _after_ctx,
                ).await? {
                    return Ok(_stop);
                }
                Ok(_decoded)
            }
            Err(_error) => {
                let _error_text = _error.to_string();
                let mut _error_ctx = {{ model_name }}CreateEventContext::new(
                    "{{ model_name }}",
                    crate::CrudOperation::Create,
                    crate::EventPhase::Error,
                    _event_args,
                    _event_payload,
                    _event_state,
                )
                .with_error(_error_text)
                .with_transaction_id(_transaction_id);
                let _ = _events.run::<{{ model_name }}CreateEventContext, {{ model_name }}>(
                    "{{ model_name }}",
                    crate::CrudOperation::Create,
                    crate::EventPhase::Error,
                    &mut _error_ctx,
                ).await?;
                Err(_error)
            }
        }
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Create multiple {{ model_name }} records in a single batch.
    ///
    /// Executes immediately and returns the created records.
    pub {% if is_async %}async {% endif %}fn create_many(
        &self,
        entries: Vec<{{ entry_name }}>,
    ) -> nautilus_core::Result<Vec<{{ model_name }}>> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        let _events = {% if is_async %}self.client.events().clone(){% else %}_c.events().clone(){% endif %};
        let _transaction_id = {% if is_async %}self.client.transaction_id(){% else %}_c.transaction_id(){% endif %};
        let _event_args = entries.clone();
        let _event_payload = serde_json::Value::Array(
            entries.iter().map({{ entry_name }}::to_engine_data).collect(),
        );
        let mut _event_state = std::collections::HashMap::new();
        let mut _before_ctx = {{ model_name }}CreateManyEventContext::new(
            "{{ model_name }}",
            crate::CrudOperation::CreateMany,
            crate::EventPhase::Before,
            _event_args.clone(),
            _event_payload.clone(),
            _event_state,
        )
        .with_transaction_id(_transaction_id.clone());
        if let Some(_stop) = _events.run::<{{ model_name }}CreateManyEventContext, Vec<{{ model_name }}>>(
            "{{ model_name }}",
            crate::CrudOperation::CreateMany,
            crate::EventPhase::Before,
            &mut _before_ctx,
        ).await? {
            return Ok(_stop);
        }
        _event_state = _before_ctx.state;

        let _operation_result: nautilus_core::Result<Vec<{{ model_name }}>> = async {
            if entries.is_empty() {
                return Ok(vec![]);
            }

            if let Some(rows) =
                crate::runtime::try_create_many_via_engine::<_, {{ model_name }}>(
                    {% if is_async %}&self.client{% else %}&_c{% endif %},
                    "{{ model_name }}",
                    entries.iter().map({{ entry_name }}::to_engine_data).collect(),
                    |row| {
                        decode_model_row_with_hints(
                            row,
                            all_scalar_decode_slots(),
                            all_scalar_value_hints(),
                        )
                    },
                )
                .await?
            {
                return Ok(rows);
            }

            let mut builder = {{ create_many_name }}::new({% if is_async %}self.client.clone(){% else %}_c{% endif %});
            for entry in entries {
                builder = builder.entry(entry);
            }
            builder.exec().await
        }.await;

        match _operation_result {
            Ok(_decoded) => {
                let mut _after_ctx = {{ model_name }}CreateManyEventContext::new(
                    "{{ model_name }}",
                    crate::CrudOperation::CreateMany,
                    crate::EventPhase::After,
                    _event_args,
                    _event_payload,
                    _event_state,
                )
                .with_result(_decoded.clone())
                .with_transaction_id(_transaction_id);
                if let Some(_stop) = _events.run::<{{ model_name }}CreateManyEventContext, Vec<{{ model_name }}>>(
                    "{{ model_name }}",
                    crate::CrudOperation::CreateMany,
                    crate::EventPhase::After,
                    &mut _after_ctx,
                ).await? {
                    return Ok(_stop);
                }
                Ok(_decoded)
            }
            Err(_error) => {
                let _error_text = _error.to_string();
                let mut _error_ctx = {{ model_name }}CreateManyEventContext::new(
                    "{{ model_name }}",
                    crate::CrudOperation::CreateMany,
                    crate::EventPhase::Error,
                    _event_args,
                    _event_payload,
                    _event_state,
                )
                .with_error(_error_text)
                .with_transaction_id(_transaction_id);
                let _ = _events.run::<{{ model_name }}CreateManyEventContext, Vec<{{ model_name }}>>(
                    "{{ model_name }}",
                    crate::CrudOperation::CreateMany,
                    crate::EventPhase::Error,
                    &mut _error_ctx,
                ).await?;
                Err(_error)
            }
        }
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Update {{ model_name }} records matching `args.where_`.
    ///
    /// If `args.where_` is `None`, all rows are updated.
    /// Executes immediately and returns the updated records.
    pub {% if is_async %}async {% endif %}fn update(
        &self,
        args: {{ model_name }}UpdateArgs,
    ) -> nautilus_core::Result<Vec<{{ model_name }}>> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        let _events = {% if is_async %}self.client.events().clone(){% else %}_c.events().clone(){% endif %};
        let _transaction_id = {% if is_async %}self.client.transaction_id(){% else %}_c.transaction_id(){% endif %};
        let _event_args = args.clone();
        let _engine_data = args.data.to_engine_data();
        let _event_payload = {
            let mut _payload = serde_json::Map::new();
            _payload.insert(
                "filter".to_string(),
                match serialize_update_filter_for_engine(args.where_.as_ref())? {
                    Some(value) => value,
                    None => serde_json::Value::Object(serde_json::Map::new()),
                },
            );
            _payload.insert("data".to_string(), _engine_data.clone());
            serde_json::Value::Object(_payload)
        };
        let mut _event_state = std::collections::HashMap::new();
        let mut _before_ctx = {{ model_name }}UpdateEventContext::new(
            "{{ model_name }}",
            crate::CrudOperation::Update,
            crate::EventPhase::Before,
            _event_args.clone(),
            _event_payload.clone(),
            _event_state,
        )
        .with_transaction_id(_transaction_id.clone());
        if let Some(_stop) = _events.run::<{{ model_name }}UpdateEventContext, Vec<{{ model_name }}>>(
            "{{ model_name }}",
            crate::CrudOperation::Update,
            crate::EventPhase::Before,
            &mut _before_ctx,
        ).await? {
            return Ok(_stop);
        }
        _event_state = _before_ctx.state;

        let _operation_result: nautilus_core::Result<Vec<{{ model_name }}>> = async {
            let no_assignments = _engine_data
                .as_object()
                .is_some_and(|data| data.is_empty());
            if no_assignments {
                return Ok(vec![]);
            }

            if let Some(filter_json) = serialize_update_filter_for_engine(args.where_.as_ref())? {
                if let Some(rows) =
                    crate::runtime::try_update_via_engine::<_, {{ model_name }}>(
                        {% if is_async %}&self.client{% else %}&_c{% endif %},
                        "{{ model_name }}",
                        filter_json,
                        _engine_data.clone(),
                        |row| {
                            decode_model_row_with_hints(
                                row,
                                all_scalar_decode_slots(),
                                all_scalar_value_hints(),
                            )
                        },
                    )
                    .await?
                {
                    return Ok(rows);
                }
            }

            let mut builder = {{ update_name }}::new({% if is_async %}self.client.clone(){% else %}_c{% endif %}).set_data(args.data);
            if let Some(filter) = args.where_ {
                builder = builder.where_(filter);
            } else {
                builder = builder.all();
            }
            builder.exec().await
        }.await;

        match _operation_result {
            Ok(_decoded) => {
                let mut _after_ctx = {{ model_name }}UpdateEventContext::new(
                    "{{ model_name }}",
                    crate::CrudOperation::Update,
                    crate::EventPhase::After,
                    _event_args,
                    _event_payload,
                    _event_state,
                )
                .with_result(_decoded.clone())
                .with_transaction_id(_transaction_id);
                if let Some(_stop) = _events.run::<{{ model_name }}UpdateEventContext, Vec<{{ model_name }}>>(
                    "{{ model_name }}",
                    crate::CrudOperation::Update,
                    crate::EventPhase::After,
                    &mut _after_ctx,
                ).await? {
                    return Ok(_stop);
                }
                Ok(_decoded)
            }
            Err(_error) => {
                let _error_text = _error.to_string();
                let mut _error_ctx = {{ model_name }}UpdateEventContext::new(
                    "{{ model_name }}",
                    crate::CrudOperation::Update,
                    crate::EventPhase::Error,
                    _event_args,
                    _event_payload,
                    _event_state,
                )
                .with_error(_error_text)
                .with_transaction_id(_transaction_id);
                let _ = _events.run::<{{ model_name }}UpdateEventContext, Vec<{{ model_name }}>>(
                    "{{ model_name }}",
                    crate::CrudOperation::Update,
                    crate::EventPhase::Error,
                    &mut _error_ctx,
                ).await?;
                Err(_error)
            }
        }
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Delete the first {{ model_name }} record matching `args.where_`.
    ///
    /// finds the first matching record and
    /// deletes it by its primary key. Returns `Ok(None)` if no record matches.
    pub {% if is_async %}async {% endif %}fn delete(
        &self,
        args: {{ model_name }}DeleteArgs,
    ) -> nautilus_core::Result<Option<{{ model_name }}>> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        let _events = {% if is_async %}self.client.events().clone(){% else %}_c.events().clone(){% endif %};
        let _transaction_id = {% if is_async %}self.client.transaction_id(){% else %}_c.transaction_id(){% endif %};
        let _event_args = args.clone();
        let _event_payload = {
            let mut _payload = serde_json::Map::new();
            let _filter = match args.where_.as_ref() {
                Some(filter) => serialize_update_filter_for_engine(Some(filter))?
                    .unwrap_or_else(|| serde_json::Value::Object(serde_json::Map::new())),
                None => serde_json::Value::Object(serde_json::Map::new()),
            };
            _payload.insert("filter".to_string(), _filter);
            serde_json::Value::Object(_payload)
        };
        let mut _event_state = std::collections::HashMap::new();
        let mut _before_ctx = {{ model_name }}DeleteEventContext::new(
            "{{ model_name }}",
            crate::CrudOperation::Delete,
            crate::EventPhase::Before,
            _event_args.clone(),
            _event_payload.clone(),
            _event_state,
        )
        .with_transaction_id(_transaction_id.clone());
        if let Some(_stop) = _events.run::<{{ model_name }}DeleteEventContext, Option<{{ model_name }}>>(
            "{{ model_name }}",
            crate::CrudOperation::Delete,
            crate::EventPhase::Before,
            &mut _before_ctx,
        ).await? {
            return Ok(_stop);
        }
        _event_state = _before_ctx.state;

        let _operation_result: nautilus_core::Result<Option<{{ model_name }}>> = async {
            if let Some(filter) = args.where_.clone() {
                if {% if is_async %}self.client{% else %}_c{% endif %}.dialect().supports_returning() && is_single_record_filter(&filter) {
                    let deleted = {{ delete_name }}::new({% if is_async %}self.client.clone(){% else %}_c.clone(){% endif %}).where_(filter).exec().await?;
                    return match deleted.len() {
                        0 => Ok(None),
                        1 => Ok(deleted.into_iter().next()),
                        count => Err(nautilus_core::Error::Other(format!(
                            "expected at most one deleted {{ model_name }} row, got {}",
                            count
                        ))),
                    };
                }
            }

            // Find the first matching record to get its primary key
            let record = self.find_first(nautilus_core::FindManyArgs {
                where_: args.where_,
                ..Default::default()
            }){% if is_async %}.await{% endif %}?;

            let Some(record) = record else {
                return Ok(None);
            };

            // Build a primary-key filter to delete exactly that one record
            {%- set pk_fields = scalar_fields | filter(attribute="is_pk", value=true) %}
            let pk_filter = {% for f in pk_fields %}{% if not loop.first %}.and({% endif %}nautilus_core::Expr::column("{{ table_name }}__{{ f.db_name }}").eq(nautilus_core::Expr::param(record.{{ f.name }}.clone())){% if not loop.first %}){% endif %}{% endfor %};
            {{ delete_name }}::new({% if is_async %}self.client.clone(){% else %}_c.clone(){% endif %}).where_(pk_filter).exec().await?;
            Ok(Some(record))
        }.await;

        match _operation_result {
            Ok(_decoded) => {
                let mut _after_ctx = {{ model_name }}DeleteEventContext::new(
                    "{{ model_name }}",
                    crate::CrudOperation::Delete,
                    crate::EventPhase::After,
                    _event_args,
                    _event_payload,
                    _event_state,
                )
                .with_result(_decoded.clone())
                .with_transaction_id(_transaction_id);
                if let Some(_stop) = _events.run::<{{ model_name }}DeleteEventContext, Option<{{ model_name }}>>(
                    "{{ model_name }}",
                    crate::CrudOperation::Delete,
                    crate::EventPhase::After,
                    &mut _after_ctx,
                ).await? {
                    return Ok(_stop);
                }
                Ok(_decoded)
            }
            Err(_error) => {
                let _error_text = _error.to_string();
                let mut _error_ctx = {{ model_name }}DeleteEventContext::new(
                    "{{ model_name }}",
                    crate::CrudOperation::Delete,
                    crate::EventPhase::Error,
                    _event_args,
                    _event_payload,
                    _event_state,
                )
                .with_error(_error_text)
                .with_transaction_id(_transaction_id);
                let _ = _events.run::<{{ model_name }}DeleteEventContext, Option<{{ model_name }}>>(
                    "{{ model_name }}",
                    crate::CrudOperation::Delete,
                    crate::EventPhase::Error,
                    &mut _error_ctx,
                ).await?;
                Err(_error)
            }
        }
{%- if not is_async %}
        }))
{%- else %}
{%- endif %}
    }

    /// Delete all {{ model_name }} records matching `args.where_`.
    ///
    /// If `args.where_` is `None`, all rows are deleted.
    /// Executes immediately and returns the deleted records.
    pub {% if is_async %}async {% endif %}fn delete_many(
        &self,
        args: {{ model_name }}DeleteArgs,
    ) -> nautilus_core::Result<Vec<{{ model_name }}>> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        let _events = {% if is_async %}self.client.events().clone(){% else %}_c.events().clone(){% endif %};
        let _transaction_id = {% if is_async %}self.client.transaction_id(){% else %}_c.transaction_id(){% endif %};
        let _event_args = args.clone();
        let _event_payload = {
            let mut _payload = serde_json::Map::new();
            let _filter = match args.where_.as_ref() {
                Some(filter) => serialize_update_filter_for_engine(Some(filter))?
                    .unwrap_or_else(|| serde_json::Value::Object(serde_json::Map::new())),
                None => serde_json::Value::Object(serde_json::Map::new()),
            };
            _payload.insert("filter".to_string(), _filter);
            serde_json::Value::Object(_payload)
        };
        let mut _event_state = std::collections::HashMap::new();
        let mut _before_ctx = {{ model_name }}DeleteManyEventContext::new(
            "{{ model_name }}",
            crate::CrudOperation::DeleteMany,
            crate::EventPhase::Before,
            _event_args.clone(),
            _event_payload.clone(),
            _event_state,
        )
        .with_transaction_id(_transaction_id.clone());
        if let Some(_stop) = _events.run::<{{ model_name }}DeleteManyEventContext, Vec<{{ model_name }}>>(
            "{{ model_name }}",
            crate::CrudOperation::DeleteMany,
            crate::EventPhase::Before,
            &mut _before_ctx,
        ).await? {
            return Ok(_stop);
        }
        _event_state = _before_ctx.state;

        let _operation_result: nautilus_core::Result<Vec<{{ model_name }}>> = async {
            let mut builder = {{ delete_name }}::new({% if is_async %}self.client.clone(){% else %}_c{% endif %});
            if let Some(expr) = args.where_ {
                builder = builder.where_(expr);
            } else {
                builder = builder.all();
            }
            builder.exec().await
        }.await;

        match _operation_result {
            Ok(_decoded) => {
                let mut _after_ctx = {{ model_name }}DeleteManyEventContext::new(
                    "{{ model_name }}",
                    crate::CrudOperation::DeleteMany,
                    crate::EventPhase::After,
                    _event_args,
                    _event_payload,
                    _event_state,
                )
                .with_result(_decoded.clone())
                .with_transaction_id(_transaction_id);
                if let Some(_stop) = _events.run::<{{ model_name }}DeleteManyEventContext, Vec<{{ model_name }}>>(
                    "{{ model_name }}",
                    crate::CrudOperation::DeleteMany,
                    crate::EventPhase::After,
                    &mut _after_ctx,
                ).await? {
                    return Ok(_stop);
                }
                Ok(_decoded)
            }
            Err(_error) => {
                let _error_text = _error.to_string();
                let mut _error_ctx = {{ model_name }}DeleteManyEventContext::new(
                    "{{ model_name }}",
                    crate::CrudOperation::DeleteMany,
                    crate::EventPhase::Error,
                    _event_args,
                    _event_payload,
                    _event_state,
                )
                .with_error(_error_text)
                .with_transaction_id(_transaction_id);
                let _ = _events.run::<{{ model_name }}DeleteManyEventContext, Vec<{{ model_name }}>>(
                    "{{ model_name }}",
                    crate::CrudOperation::DeleteMany,
                    crate::EventPhase::Error,
                    &mut _error_ctx,
                ).await?;
                Err(_error)
            }
        }
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Count {{ model_name }} records matching `args`.
    pub {% if is_async %}async {% endif %}fn count(
        &self,
        args: {{ model_name }}CountArgs,
    ) -> nautilus_core::Result<i64> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        if let Some(count) =
            crate::runtime::try_count_via_engine(
                {% if is_async %}&self.client{% else %}&_c{% endif %},
                "{{ model_name }}",
                args.to_engine_args()?,
            )
            .await?
        {
            return Ok(count);
        }

        Err(nautilus_core::Error::InvalidQuery(
            "count queries require the embedded engine path in the generated Rust client"
                .to_string(),
        ))
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Group {{ model_name }} rows and compute aggregates.
    pub {% if is_async %}async {% endif %}fn group_by(
        &self,
        args: {{ model_name }}GroupByArgs,
    ) -> nautilus_core::Result<Vec<{{ model_name }}GroupByOutput>> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        let engine_args = args.to_engine_args()?;
        if let Some(rows) =
            crate::runtime::try_group_by_rows_via_engine(
                {% if is_async %}&self.client{% else %}&_c{% endif %},
                "{{ model_name }}",
                engine_args,
            )
            .await?
        {
            let mut decoded = Vec::with_capacity(rows.len());
            for row in &rows {
                decoded.push({{ model_name }}GroupByOutput::from_row(row)?);
            }
            return Ok(decoded);
        }

        Err(nautilus_core::Error::InvalidQuery(
            "groupBy queries require the embedded engine path in the generated Rust client"
                .to_string(),
        ))
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Execute a raw SQL string and return the result rows as generic maps.
    ///
    /// The SQL is sent to the database as-is with no parameter binding.
    /// Prefer `raw_stmt_query` when user-supplied values are involved to avoid
    /// SQL injection.
    pub {% if is_async %}async {% endif %}fn raw_query(
        &self,
        sql: impl Into<String>,
    ) -> nautilus_core::Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        let sql_obj = nautilus_dialect::Sql { text: sql.into(), params: vec![] };
        let stream = {% if is_async %}self.client{% else %}_c{% endif %}.executor().execute(&sql_obj);
        futures::pin_mut!(stream);
        let mut results: Vec<std::collections::HashMap<String, serde_json::Value>> = Vec::new();
        while let Some(result) = stream.next().await {
            let row: crate::Row = result
                .map_err(|e| nautilus_core::Error::Other(e.to_string()))?
                .into();
            let map = row
                .iter()
                .map(|(name, val)| {
                    let json_val = val.to_json_plain();
                    (name.to_string(), json_val)
                })
                .collect();
            results.push(map);
        }
        Ok(results)
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Execute a raw prepared-statement query with bound parameters.
    ///
    /// Use `$1`, `$2`, … (PostgreSQL) or `?` (MySQL / SQLite) as placeholders.
    /// Parameters are bound in the order they appear in the `params` slice.
    pub {% if is_async %}async {% endif %}fn raw_stmt_query(
        &self,
        sql: impl Into<String>,
        params: Vec<nautilus_core::Value>,
    ) -> nautilus_core::Result<Vec<std::collections::HashMap<String, serde_json::Value>>> {
{%- if not is_async %}
        let _c = self.client.clone();
        tokio::task::block_in_place(|| tokio::runtime::Handle::current().block_on(async move {
{%- endif %}
        let sql_obj = nautilus_dialect::Sql { text: sql.into(), params };
        let stream = {% if is_async %}self.client{% else %}_c{% endif %}.executor().execute(&sql_obj);
        futures::pin_mut!(stream);
        let mut results: Vec<std::collections::HashMap<String, serde_json::Value>> = Vec::new();
        while let Some(result) = stream.next().await {
            let row: crate::Row = result
                .map_err(|e| nautilus_core::Error::Other(e.to_string()))?
                .into();
            let map = row
                .iter()
                .map(|(name, val)| {
                    let json_val = val.to_json_plain();
                    (name.to_string(), json_val)
                })
                .collect();
            results.push(map);
        }
        Ok(results)
{%- if not is_async %}
        }))
{%- endif %}
    }

    /// Create or update a {{ model_name }} record.
    ///
    /// Looks up an existing record using `args.where_`. If found, updates it
    /// with `args.data`; otherwise creates a new record with `args.data`.
    /// Executes immediately and returns the created or updated `{{ model_name }}`
    /// (or `None` when `args.return_data` is `false`).
    pub {% if is_async %}async {% endif %}fn upsert(
        &self,
        args: {{ model_name }}UpsertArgs,
    ) -> nautilus_core::Result<Option<{{ model_name }}>> {
        let has_update_assignments = args.update.has_assignments();

        if self.client.dialect().supports_returning() && has_update_assignments {
            let updated = self.update({{ model_name }}UpdateArgs {
                where_: args.where_.clone(),
                data: args.update.clone(),
            }){% if is_async %}.await{% endif %}?;

            if let Some(record) = updated.into_iter().next() {
                if !args.return_data {
                    return Ok(None);
                }
                return Ok(Some(record));
            }
        }

        let existing = self.find_first(nautilus_core::FindManyArgs {
            where_: args.where_.clone(),
            ..Default::default()
        }){% if is_async %}.await{% endif %}?;

        if let Some(record) = existing {
            if !has_update_assignments {
                if !args.return_data {
                    return Ok(None);
                }
                return Ok(Some(record));
            }

            let updated = self.update({{ model_name }}UpdateArgs {
                where_: args.where_,
                data: args.update,
            }){% if is_async %}.await{% endif %}?;
            if !args.return_data {
                return Ok(None);
            }
            Ok(Some(updated.into_iter().next().unwrap_or(record)))
        } else {
            let created = self.create(args.create){% if is_async %}.await{% endif %}?;
            if !args.return_data {
                return Ok(None);
            }
            Ok(Some(created))
        }
    }
}
