//! Parameter conversion for SQL Server RPC calls.
//!
//! This module converts Rust types (via `ToSql`) into TDS wire-format
//! RPC parameters, including Table-Valued Parameter (TVP) encoding.

use bytes::BytesMut;
#[cfg(feature = "decimal")]
use tds_protocol::__private::encode_tvp_decimal;
use tds_protocol::__private::{
    encode_tvp_bit, encode_tvp_float, encode_tvp_int, encode_tvp_null, encode_tvp_nvarchar,
    encode_tvp_varbinary, encode_tvp_varchar_with_collation,
};
#[cfg(feature = "chrono")]
use tds_protocol::__private::{encode_tvp_datetime, encode_tvp_smalldatetime};
#[cfg(feature = "decimal")]
use tds_protocol::__private::{encode_tvp_money, encode_tvp_smallmoney};
use tds_protocol::rpc::{RpcParam, TypeInfo as RpcTypeInfo};
use tds_protocol::tvp::{TvpColumnDef as TvpWireColumnDef, TvpEncoder, TvpWireType};

use crate::error::{Error, Result};
use crate::state::ConnectionState;
use crate::validation::{validate_identifier, validate_qualified_identifier};

use super::Client;

impl<S: ConnectionState> Client<S> {
    /// Convert a `SqlValue` into an `RpcParam` with the given name.
    ///
    /// This is the core conversion logic shared by positional parameters
    /// (`convert_params`), named procedure parameters (`ProcedureBuilder::input`),
    /// and named query parameters (`convert_named_params`).
    ///
    /// When `send_unicode` is `false`, `SqlValue::String` values are encoded
    /// as VARCHAR (single-byte) instead of NVARCHAR (UTF-16), which allows
    /// SQL Server to use index seeks on VARCHAR columns.
    pub(crate) fn sql_value_to_rpc_param(
        name: &str,
        sql_value: &mssql_types::SqlValue,
        send_unicode: bool,
        collation: Option<&tds_protocol::token::Collation>,
    ) -> Result<RpcParam> {
        use bytes::{BufMut, BytesMut};
        use mssql_types::SqlValue;

        Ok(match sql_value {
            SqlValue::Null => RpcParam::null(name, RpcTypeInfo::nvarchar(1)),
            SqlValue::Bool(v) => {
                let mut buf = BytesMut::with_capacity(1);
                buf.put_u8(if *v { 1 } else { 0 });
                RpcParam::new(name, RpcTypeInfo::bit(), buf.freeze())
            }
            SqlValue::TinyInt(v) => {
                let mut buf = BytesMut::with_capacity(1);
                buf.put_u8(*v);
                RpcParam::new(name, RpcTypeInfo::tinyint(), buf.freeze())
            }
            SqlValue::SmallInt(v) => {
                let mut buf = BytesMut::with_capacity(2);
                buf.put_i16_le(*v);
                RpcParam::new(name, RpcTypeInfo::smallint(), buf.freeze())
            }
            SqlValue::Int(v) => RpcParam::int(name, *v),
            SqlValue::BigInt(v) => RpcParam::bigint(name, *v),
            SqlValue::Float(v) => {
                let mut buf = BytesMut::with_capacity(4);
                buf.put_f32_le(*v);
                RpcParam::new(name, RpcTypeInfo::real(), buf.freeze())
            }
            SqlValue::Double(v) => {
                let mut buf = BytesMut::with_capacity(8);
                buf.put_f64_le(*v);
                RpcParam::new(name, RpcTypeInfo::float(), buf.freeze())
            }
            SqlValue::String(s) => {
                if send_unicode {
                    RpcParam::nvarchar(name, s)
                } else if let Some(c) = collation {
                    RpcParam::varchar_with_collation(name, s, c)
                } else {
                    RpcParam::varchar(name, s)
                }
            }
            SqlValue::Binary(b) => {
                // VARBINARY(n) accepts 1..=8000 — `varbinary(0)` is rejected by the
                // server ("Data type 0xA5 has an invalid data length or metadata
                // length") and anything larger must be VARBINARY(MAX) / PLP. Mirrors
                // the NVARCHAR / VARCHAR size handling in `RpcParam::{nvarchar,varchar}`.
                let type_info = if b.len() > 8000 {
                    RpcTypeInfo::varbinary_max()
                } else {
                    RpcTypeInfo::varbinary(b.len().max(1) as u16)
                };
                RpcParam::new(name, type_info, b.clone())
            }
            SqlValue::Xml(s) => RpcParam::nvarchar(name, s),
            #[cfg(feature = "uuid")]
            SqlValue::Uuid(u) => {
                let bytes = u.as_bytes();
                let mut buf = BytesMut::with_capacity(16);
                // SQL Server stores GUIDs in mixed-endian format
                buf.put_u32_le(u32::from_be_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]));
                buf.put_u16_le(u16::from_be_bytes([bytes[4], bytes[5]]));
                buf.put_u16_le(u16::from_be_bytes([bytes[6], bytes[7]]));
                buf.put_slice(&bytes[8..16]);
                RpcParam::new(name, RpcTypeInfo::uniqueidentifier(), buf.freeze())
            }
            #[cfg(feature = "decimal")]
            SqlValue::Decimal(d) => {
                let mut buf = BytesMut::with_capacity(17);
                mssql_types::__private::encode_decimal(*d, &mut buf);
                let scale = d.scale() as u8;
                RpcParam::new(name, RpcTypeInfo::decimal(38, scale), buf.freeze())
            }
            #[cfg(feature = "decimal")]
            SqlValue::Money(d) => {
                let mut buf = BytesMut::with_capacity(8);
                mssql_types::__private::encode_money(*d, &mut buf)?;
                RpcParam::new(name, RpcTypeInfo::money(), buf.freeze())
            }
            #[cfg(feature = "decimal")]
            SqlValue::SmallMoney(d) => {
                let mut buf = BytesMut::with_capacity(4);
                mssql_types::__private::encode_smallmoney(*d, &mut buf)?;
                RpcParam::new(name, RpcTypeInfo::smallmoney(), buf.freeze())
            }
            #[cfg(feature = "chrono")]
            SqlValue::Date(d) => {
                let mut buf = BytesMut::with_capacity(3);
                mssql_types::__private::encode_date(*d, &mut buf)?;
                RpcParam::new(name, RpcTypeInfo::date(), buf.freeze())
            }
            #[cfg(feature = "chrono")]
            SqlValue::Time(t) => {
                let mut buf = BytesMut::with_capacity(5);
                mssql_types::__private::encode_time(*t, &mut buf);
                RpcParam::new(name, RpcTypeInfo::time(7), buf.freeze())
            }
            #[cfg(feature = "chrono")]
            SqlValue::DateTime(dt) => {
                let mut buf = BytesMut::with_capacity(8);
                mssql_types::__private::encode_datetime2(*dt, &mut buf)?;
                RpcParam::new(name, RpcTypeInfo::datetime2(7), buf.freeze())
            }
            #[cfg(feature = "chrono")]
            SqlValue::SmallDateTime(dt) => {
                let mut buf = BytesMut::with_capacity(4);
                mssql_types::__private::encode_smalldatetime(*dt, &mut buf)?;
                RpcParam::new(name, RpcTypeInfo::smalldatetime(), buf.freeze())
            }
            #[cfg(feature = "chrono")]
            SqlValue::DateTimeOffset(dto) => {
                let mut buf = BytesMut::with_capacity(10);
                mssql_types::__private::encode_datetimeoffset(*dto, &mut buf)?;
                RpcParam::new(name, RpcTypeInfo::datetimeoffset(7), buf.freeze())
            }
            #[cfg(feature = "json")]
            SqlValue::Json(j) => RpcParam::nvarchar(name, &j.to_string()),
            SqlValue::Tvp(tvp_data) => Self::encode_tvp_param(name, tvp_data, collation)?,
            _ => {
                return Err(Error::Type(mssql_types::TypeError::UnsupportedConversion {
                    from: sql_value.type_name().to_string(),
                    to: "RPC parameter",
                }));
            }
        })
    }

    /// Convert a single `ToSql` value into an `RpcParam` with the given name.
    ///
    /// This is the shared conversion logic used by both `convert_params()`
    /// (for positional query parameters) and `ProcedureBuilder::input()`
    /// (for named procedure parameters).
    pub(crate) fn convert_single_param(
        name: &str,
        value: &(dyn crate::ToSql + Sync),
        send_unicode: bool,
        collation: Option<&tds_protocol::token::Collation>,
    ) -> Result<RpcParam> {
        let sql_value = value.to_sql()?;
        Self::sql_value_to_rpc_param(name, &sql_value, send_unicode, collation)
    }

    /// Convert ToSql parameters to RPC parameters with auto-generated names.
    pub(crate) fn convert_params(
        params: &[&(dyn crate::ToSql + Sync)],
        send_unicode: bool,
        collation: Option<&tds_protocol::token::Collation>,
    ) -> Result<Vec<RpcParam>> {
        params
            .iter()
            .enumerate()
            .map(|(i, p)| {
                let name = format!("@p{}", i + 1);
                Self::convert_single_param(&name, *p, send_unicode, collation)
            })
            .collect()
    }

    /// Convert ToSql parameters to RPC parameters with empty names for positional binding.
    ///
    /// Used by [`Client::call_procedure`](super::Client::call_procedure) because
    /// SQL Server's RPC layer binds parameters by name when names are present and
    /// by position when names are empty. The previous behavior of emitting
    /// `@p1`, `@p2`, … rejected any procedure whose parameters weren't literally
    /// named `@p1`, `@p2`, … with "Procedure or function expects parameter '@…',
    /// which was not supplied."
    pub(crate) fn convert_params_positional(
        params: &[&(dyn crate::ToSql + Sync)],
        send_unicode: bool,
        collation: Option<&tds_protocol::token::Collation>,
    ) -> Result<Vec<RpcParam>> {
        params
            .iter()
            .map(|p| Self::convert_single_param("", *p, send_unicode, collation))
            .collect()
    }

    /// Convert named parameters to RPC parameters.
    ///
    /// Ensures each parameter name has an `@` prefix as required by
    /// `sp_executesql` parameter declarations.
    pub(crate) fn convert_named_params(
        params: &[crate::to_params::NamedParam],
        send_unicode: bool,
        collation: Option<&tds_protocol::token::Collation>,
    ) -> Result<Vec<RpcParam>> {
        params
            .iter()
            .map(|p| {
                let name = if p.name.starts_with('@') {
                    p.name.clone()
                } else {
                    format!("@{}", p.name)
                };
                // The name is interpolated verbatim into the `sp_executesql`
                // `@params` declaration (`build_param_declarations`); validate it
                // to prevent SQL injection via a crafted parameter name (#277).
                validate_identifier(&name)?;
                Self::sql_value_to_rpc_param(&name, &p.value, send_unicode, collation)
            })
            .collect()
    }

    /// Encode a TVP parameter for RPC.
    ///
    /// This encodes the complete TVP structure including metadata and row data
    /// into the TDS wire format. `collation` (captured from the login
    /// ENVCHANGE) determines the codepage VARCHAR cells are encoded with and
    /// declared as; without it they fall back to Windows-1252.
    fn encode_tvp_param(
        name: &str,
        tvp_data: &mssql_types::TvpData,
        collation: Option<&tds_protocol::token::Collation>,
    ) -> Result<RpcParam> {
        // The TVP type name is interpolated verbatim into the `sp_executesql`
        // `@params` declaration as `{name} READONLY` (`build_param_declarations`);
        // validate it (schema-qualified `schema.Type` allowed) to prevent SQL
        // injection via a crafted type name (#277). Delimited/bracketed names
        // (`[My Type]`) are rejected, consistent with the rest of the driver's
        // identifier handling.
        let full_type_name = if tvp_data.schema.is_empty() {
            tvp_data.type_name.clone()
        } else {
            format!("{}.{}", tvp_data.schema, tvp_data.type_name)
        };
        validate_qualified_identifier(&full_type_name)?;

        // Convert mssql-types column definitions to wire format
        let wire_columns: Vec<TvpWireColumnDef> = tvp_data
            .columns
            .iter()
            .map(|col| {
                let wire_type = Self::convert_tvp_column_type(&col.column_type)?;
                Ok(if col.nullable {
                    TvpWireColumnDef::nullable(wire_type)
                } else {
                    TvpWireColumnDef::new(wire_type)
                })
            })
            .collect::<Result<Vec<_>>>()?;

        // Create encoder
        let encoder = TvpEncoder::new(&tvp_data.schema, &tvp_data.type_name, &wire_columns);

        // Encode to buffer
        let mut buf = BytesMut::with_capacity(256);

        // Encode metadata
        encoder.encode_metadata_with_collation(&mut buf, collation);

        // Encode each row. encode_row takes an infallible closure, so capture
        // the first cell error and abort the whole conversion — the partially
        // written buffer is discarded with it.
        let mut cell_error: Option<Error> = None;
        for row in &tvp_data.rows {
            encoder.encode_row(&mut buf, |row_buf| {
                for (col_idx, value) in row.iter().enumerate() {
                    let wire_type = &wire_columns[col_idx].wire_type;
                    if let Err(e) = Self::encode_tvp_value(value, wire_type, collation, row_buf) {
                        if cell_error.is_none() {
                            cell_error = Some(e);
                        }
                        break;
                    }
                }
            });
            if let Some(e) = cell_error.take() {
                return Err(e);
            }
        }

        // Encode end marker
        encoder.encode_end(&mut buf);

        // Create RPC param with TVP type info. The type info carries the
        // (already-validated) TVP type name for the parameter declaration.
        let type_info = RpcTypeInfo::tvp(&full_type_name);

        Ok(RpcParam {
            name: name.to_string(),
            flags: tds_protocol::rpc::ParamFlags::default(),
            type_info,
            value: Some(buf.freeze()),
            crypto_metadata: None,
        })
    }

    /// Convert mssql-types TvpColumnType to wire TvpWireType.
    fn convert_tvp_column_type(col_type: &mssql_types::TvpColumnType) -> Result<TvpWireType> {
        // TvpColumnType is #[non_exhaustive], so the wildcard arm is required
        // for forward compatibility even though all current variants are covered.
        #[allow(unreachable_patterns)]
        Ok(match col_type {
            mssql_types::TvpColumnType::Bit => TvpWireType::Bit,
            mssql_types::TvpColumnType::TinyInt => TvpWireType::Int { size: 1 },
            mssql_types::TvpColumnType::SmallInt => TvpWireType::Int { size: 2 },
            mssql_types::TvpColumnType::Int => TvpWireType::Int { size: 4 },
            mssql_types::TvpColumnType::BigInt => TvpWireType::Int { size: 8 },
            mssql_types::TvpColumnType::Real => TvpWireType::Float { size: 4 },
            mssql_types::TvpColumnType::Float => TvpWireType::Float { size: 8 },
            mssql_types::TvpColumnType::Decimal { precision, scale } => TvpWireType::Decimal {
                precision: *precision,
                scale: *scale,
            },
            mssql_types::TvpColumnType::NVarChar { max_length } => TvpWireType::NVarChar {
                max_length: *max_length,
            },
            mssql_types::TvpColumnType::VarChar { max_length } => TvpWireType::VarChar {
                max_length: *max_length,
            },
            mssql_types::TvpColumnType::VarBinary { max_length } => TvpWireType::VarBinary {
                max_length: *max_length,
            },
            mssql_types::TvpColumnType::UniqueIdentifier => TvpWireType::Guid,
            mssql_types::TvpColumnType::Date => TvpWireType::Date,
            mssql_types::TvpColumnType::Time { scale } => TvpWireType::Time { scale: *scale },
            mssql_types::TvpColumnType::DateTime2 { scale } => {
                TvpWireType::DateTime2 { scale: *scale }
            }
            mssql_types::TvpColumnType::DateTimeOffset { scale } => {
                TvpWireType::DateTimeOffset { scale: *scale }
            }
            mssql_types::TvpColumnType::Money => TvpWireType::Money,
            mssql_types::TvpColumnType::SmallMoney => TvpWireType::SmallMoney,
            mssql_types::TvpColumnType::DateTime => TvpWireType::DateTime,
            mssql_types::TvpColumnType::SmallDateTime => TvpWireType::SmallDateTime,
            mssql_types::TvpColumnType::Xml => TvpWireType::Xml,
            _ => {
                return Err(Error::Type(mssql_types::TypeError::UnsupportedConversion {
                    from: format!("{col_type:?}"),
                    to: "TVP wire type",
                }));
            }
        })
    }

    /// Encode a single TVP column value.
    /// Encode one TVP cell.
    ///
    /// # Errors
    ///
    /// Returns an error for values the wire type cannot represent
    /// (out-of-range MONEY/SMALLMONEY/SMALLDATETIME) and for unsupported
    /// variants (nested TVPs) — the cell must NOT be silently written as
    /// NULL: the server cannot tell a substituted NULL from an intentional
    /// one, so the row would insert wrong data (issue #157).
    fn encode_tvp_value(
        value: &mssql_types::SqlValue,
        wire_type: &TvpWireType,
        collation: Option<&tds_protocol::token::Collation>,
        buf: &mut BytesMut,
    ) -> Result<()> {
        use mssql_types::SqlValue;

        match value {
            SqlValue::Null => {
                encode_tvp_null(wire_type, buf);
            }
            SqlValue::Bool(v) => {
                encode_tvp_bit(*v, buf);
            }
            SqlValue::TinyInt(v) => {
                encode_tvp_int(*v as i64, 1, buf);
            }
            SqlValue::SmallInt(v) => {
                encode_tvp_int(*v as i64, 2, buf);
            }
            SqlValue::Int(v) => {
                encode_tvp_int(*v as i64, 4, buf);
            }
            SqlValue::BigInt(v) => {
                encode_tvp_int(*v, 8, buf);
            }
            SqlValue::Float(v) => {
                encode_tvp_float(*v as f64, 4, buf);
            }
            SqlValue::Double(v) => {
                encode_tvp_float(*v, 8, buf);
            }
            SqlValue::String(s) => match wire_type {
                TvpWireType::NVarChar { max_length } => {
                    encode_tvp_nvarchar(s, *max_length, buf);
                }
                TvpWireType::VarChar { max_length } => {
                    encode_tvp_varchar_with_collation(s, *max_length, collation, buf);
                }
                _ => {
                    encode_tvp_nvarchar(s, 4000, buf);
                }
            },
            SqlValue::Binary(b) => {
                let max_len = match wire_type {
                    TvpWireType::VarBinary { max_length } => *max_length,
                    _ => 8000,
                };
                encode_tvp_varbinary(b, max_len, buf);
            }
            #[cfg(feature = "decimal")]
            SqlValue::Decimal(d) | SqlValue::Money(d) | SqlValue::SmallMoney(d) => {
                // Dispatch on the column wire type so that MONEY columns receive
                // the scaled-integer MONEYN format (not DECIMAL), DATETIME2-backed
                // columns get DECIMAL, etc. `SqlValue::Money` coming into a
                // DECIMAL column (or vice versa) still needs to match the column
                // metadata, so `wire_type` — not the variant tag — is the
                // authoritative dispatch key.
                match wire_type {
                    TvpWireType::Money => {
                        let scaled = mssql_types::__private::decimal_to_money_cents_i64(*d)?;
                        encode_tvp_money(scaled, buf);
                    }
                    TvpWireType::SmallMoney => {
                        let scaled = mssql_types::__private::decimal_to_smallmoney_cents_i32(*d)?;
                        encode_tvp_smallmoney(scaled, buf);
                    }
                    _ => {
                        let sign = if d.is_sign_negative() { 0u8 } else { 1u8 };
                        let mantissa = d.mantissa().unsigned_abs();
                        encode_tvp_decimal(sign, mantissa, buf);
                    }
                }
            }
            #[cfg(feature = "uuid")]
            SqlValue::Uuid(u) => {
                let bytes = u.as_bytes();
                tds_protocol::__private::encode_tvp_guid(bytes, buf);
            }
            #[cfg(feature = "chrono")]
            SqlValue::Date(d) => {
                // Calculate days since 0001-01-01
                let base =
                    chrono::NaiveDate::from_ymd_opt(1, 1, 1).expect("epoch 0001-01-01 is valid");
                let days = d.signed_duration_since(base).num_days() as u32;
                tds_protocol::__private::encode_tvp_date(days, buf);
            }
            #[cfg(feature = "chrono")]
            SqlValue::Time(t) => {
                use chrono::Timelike;
                let nanos =
                    t.num_seconds_from_midnight() as u64 * 1_000_000_000 + t.nanosecond() as u64;
                let intervals = nanos / 100;
                let scale = match wire_type {
                    TvpWireType::Time { scale } => *scale,
                    _ => 7,
                };
                tds_protocol::__private::encode_tvp_time(intervals, scale, buf);
            }
            #[cfg(feature = "chrono")]
            SqlValue::DateTime(dt) | SqlValue::SmallDateTime(dt) => {
                // Dispatch on the column wire type. DATETIME columns use the
                // legacy days+ticks wire format; SMALLDATETIME uses the
                // 4-byte days+minutes format; anything else (DATETIME2 column)
                // gets the DATETIME2 time+date format.
                match wire_type {
                    TvpWireType::DateTime => {
                        let (days, ticks) =
                            mssql_types::__private::datetime_to_legacy_days_ticks(*dt);
                        encode_tvp_datetime(days, ticks, buf);
                    }
                    TvpWireType::SmallDateTime => {
                        let (days, minutes) =
                            mssql_types::__private::datetime_to_smalldatetime_days_minutes(*dt)?;
                        encode_tvp_smalldatetime(days, minutes, buf);
                    }
                    _ => {
                        use chrono::Timelike;
                        // Time component
                        let nanos = dt.time().num_seconds_from_midnight() as u64 * 1_000_000_000
                            + dt.time().nanosecond() as u64;
                        let intervals = nanos / 100;
                        // Date component
                        let base = chrono::NaiveDate::from_ymd_opt(1, 1, 1)
                            .expect("epoch 0001-01-01 is valid");
                        let days = dt.date().signed_duration_since(base).num_days() as u32;
                        let scale = match wire_type {
                            TvpWireType::DateTime2 { scale } => *scale,
                            _ => 7,
                        };
                        tds_protocol::__private::encode_tvp_datetime2(intervals, days, scale, buf);
                    }
                }
            }
            #[cfg(feature = "chrono")]
            SqlValue::DateTimeOffset(dto) => {
                use chrono::{Offset, Timelike};
                // The wire date/time portion is UTC per MS-TDS §2.2.5.5.1.9,
                // not the local wall-clock.
                let utc = dto.naive_utc();
                // Time component (in 100-nanosecond intervals)
                let nanos = utc.time().num_seconds_from_midnight() as u64 * 1_000_000_000
                    + utc.time().nanosecond() as u64;
                let intervals = nanos / 100;
                // Date component (days since 0001-01-01)
                let base =
                    chrono::NaiveDate::from_ymd_opt(1, 1, 1).expect("epoch 0001-01-01 is valid");
                let days = utc.date().signed_duration_since(base).num_days() as u32;
                // Timezone offset in minutes
                let offset_minutes = (dto.offset().fix().local_minus_utc() / 60) as i16;
                let scale = match wire_type {
                    TvpWireType::DateTimeOffset { scale } => *scale,
                    _ => 7,
                };
                tds_protocol::__private::encode_tvp_datetimeoffset(
                    intervals,
                    days,
                    offset_minutes,
                    scale,
                    buf,
                );
            }
            #[cfg(feature = "json")]
            SqlValue::Json(j) => {
                // JSON is encoded as NVARCHAR
                encode_tvp_nvarchar(&j.to_string(), 0xFFFF, buf);
            }
            SqlValue::Xml(s) => {
                // XML is encoded as NVARCHAR for TVP
                encode_tvp_nvarchar(s, 0xFFFF, buf);
            }
            SqlValue::Tvp(_) => {
                return Err(mssql_types::TypeError::UnsupportedConversion {
                    from: "TVP".to_string(),
                    to: "TVP cell (nested TVPs are not supported)",
                }
                .into());
            }
            // Future SqlValue variants must error, not silently become NULL.
            other => {
                return Err(mssql_types::TypeError::UnsupportedConversion {
                    from: other.type_name().to_string(),
                    to: "TVP cell",
                }
                .into());
            }
        }
        Ok(())
    }
}

#[cfg(test)]
#[allow(clippy::unwrap_used)]
mod tests {
    use super::*;
    use crate::state::Ready;
    use mssql_types::SqlValue;
    use tds_protocol::tvp::TvpWireType;

    /// Issue #157 regression: TVP cells whose value the wire type cannot
    /// represent must error — previously they were silently written as NULL,
    /// inserting wrong data the server cannot distinguish from intentional
    /// NULLs.
    #[cfg(feature = "decimal")]
    #[test]
    fn tvp_out_of_range_money_cell_errors_instead_of_null() {
        let mut buf = BytesMut::new();
        let err = Client::<Ready>::encode_tvp_value(
            &SqlValue::Money(rust_decimal::Decimal::MAX),
            &TvpWireType::Money,
            None,
            &mut buf,
        )
        .expect_err("out-of-range MONEY must error");
        assert!(
            !err.to_string().is_empty(),
            "error must be descriptive: {err}"
        );

        let mut buf = BytesMut::new();
        Client::<Ready>::encode_tvp_value(
            &SqlValue::SmallMoney(rust_decimal::Decimal::MAX),
            &TvpWireType::SmallMoney,
            None,
            &mut buf,
        )
        .expect_err("out-of-range SMALLMONEY must error");
    }

    /// Issue #157 regression: unsupported variants (nested TVPs) must error,
    /// not silently encode NULL.
    #[test]
    fn tvp_nested_tvp_cell_errors_instead_of_null() {
        let nested = mssql_types::TvpData::new("dbo", "Inner");
        let mut buf = BytesMut::new();
        Client::<Ready>::encode_tvp_value(
            &SqlValue::Tvp(Box::new(nested)),
            &TvpWireType::Int { size: 4 },
            None,
            &mut buf,
        )
        .expect_err("nested TVP cell must error");
    }

    /// #277: a crafted named-parameter name must be rejected before it is
    /// interpolated into the `sp_executesql` `@params` declaration. A normal
    /// name still converts.
    #[test]
    fn named_param_name_injection_is_rejected() {
        let malicious = [crate::to_params::NamedParam::new(
            "p1 int); DROP TABLE users; --",
            SqlValue::Int(1),
        )];
        Client::<Ready>::convert_named_params(&malicious, true, None)
            .expect_err("a parameter name carrying injection must be rejected");

        let ok = [crate::to_params::NamedParam::new("p1", SqlValue::Int(1))];
        Client::<Ready>::convert_named_params(&ok, true, None)
            .expect("a normal parameter name must convert");
    }

    /// #277: a crafted TVP type name must be rejected before it is interpolated
    /// as `{name} READONLY`. A schema-qualified name still converts (validation
    /// runs before encoding, so the empty column set is irrelevant here).
    #[test]
    fn tvp_type_name_injection_is_rejected() {
        let bad = mssql_types::TvpData::new("dbo", "T READONLY); DROP TABLE users; --");
        Client::<Ready>::encode_tvp_param("@tvp", &bad, None)
            .expect_err("a TVP type name carrying injection must be rejected");

        let good = mssql_types::TvpData::new("dbo", "IntList");
        Client::<Ready>::encode_tvp_param("@tvp", &good, None)
            .expect("a schema-qualified TVP type name must convert");
    }
}
