mssql_types/

decode.rs

//! TDS binary decoding for SQL values.
//!
//! This module provides decoding of TDS wire format data into Rust values.

// Allow expect() for chrono date construction with known-valid constant dates
// (e.g., from_ymd_opt(1, 1, 1) for SQL Server epoch)
#![allow(clippy::expect_used)]

use bytes::{Buf, Bytes};

use crate::error::TypeError;
use crate::value::SqlValue;

/// Trait for decoding values from TDS binary format.
pub trait TdsDecode: Sized {
    /// Decode a value from the buffer.
    fn decode(buf: &mut Bytes, type_info: &TypeInfo) -> Result<Self, TypeError>;
}

/// TDS type information for decoding.
#[derive(Debug, Clone)]
pub struct TypeInfo {
    /// The TDS type ID.
    pub type_id: u8,
    /// Length/precision for variable-length types.
    pub length: Option<u32>,
    /// Scale for decimal/time types.
    pub scale: Option<u8>,
    /// Precision for decimal types.
    pub precision: Option<u8>,
    /// Collation for string types.
    pub collation: Option<Collation>,
}

/// SQL Server collation information.
#[derive(Debug, Clone, Copy)]
pub struct Collation {
    /// Locale ID.
    pub lcid: u32,
    /// Collation flags.
    pub flags: u8,
}

impl Collation {
    /// Check if this collation uses UTF-8 encoding (SQL Server 2019+).
    ///
    /// UTF-8 collations have fUTF8, bit 26 (0x0400_0000), set in the
    /// collation info field (bit 27 is FRESERVEDBIT per MS-TDS).
    #[must_use]
    pub fn is_utf8(&self) -> bool {
        (self.lcid & 0x0400_0000) != 0
    }

    /// Get the encoding for this collation.
    ///
    /// Returns the appropriate `encoding_rs::Encoding` for the collation's LCID,
    /// or `None` if the encoding is not supported.
    #[cfg(feature = "encoding")]
    #[must_use]
    pub fn encoding(&self) -> Option<&'static encoding_rs::Encoding> {
        encoding_for_lcid(self.lcid)
    }
}

/// UTF-8 collation flag bit — fUTF8, bit 26 per the MS-TDS Collation Rule
/// Definition (bit 27 is FRESERVEDBIT).
#[cfg(feature = "encoding")]
const UTF8_COLLATION_FLAG: u32 = 0x0400_0000;

/// Get the encoding for an LCID value.
#[cfg(feature = "encoding")]
fn encoding_for_lcid(lcid: u32) -> Option<&'static encoding_rs::Encoding> {
    // Check for UTF-8 collation first (SQL Server 2019+)
    if (lcid & UTF8_COLLATION_FLAG) != 0 {
        return Some(encoding_rs::UTF_8);
    }

    // Get code page from LCID
    let code_page = code_page_for_lcid(lcid)?;

    // Map code page to encoding
    match code_page {
        874 => Some(encoding_rs::WINDOWS_874),
        932 => Some(encoding_rs::SHIFT_JIS),
        936 => Some(encoding_rs::GB18030),
        949 => Some(encoding_rs::EUC_KR),
        950 => Some(encoding_rs::BIG5),
        1250 => Some(encoding_rs::WINDOWS_1250),
        1251 => Some(encoding_rs::WINDOWS_1251),
        1252 => Some(encoding_rs::WINDOWS_1252),
        1253 => Some(encoding_rs::WINDOWS_1253),
        1254 => Some(encoding_rs::WINDOWS_1254),
        1255 => Some(encoding_rs::WINDOWS_1255),
        1256 => Some(encoding_rs::WINDOWS_1256),
        1257 => Some(encoding_rs::WINDOWS_1257),
        1258 => Some(encoding_rs::WINDOWS_1258),
        _ => None,
    }
}

/// Get the Windows code page for an LCID value.
#[cfg(feature = "encoding")]
fn code_page_for_lcid(lcid: u32) -> Option<u16> {
    // Mask for the locale ID (lower 16 bits); the high bits carry sort/version
    // flags, not the locale, so a 10-bit mask collapsed every full LCID and the
    // arms below (full LCIDs like 0x0411) never matched. Mirrors
    // `tds_protocol::collation::PRIMARY_LANGUAGE_MASK`.
    const PRIMARY_LANGUAGE_MASK: u32 = 0xFFFF;
    let primary_lang = lcid & PRIMARY_LANGUAGE_MASK;

    match primary_lang {
        0x0411 => Some(932),                   // Japanese - Shift_JIS
        0x0804 | 0x1004 => Some(936),          // Chinese Simplified - GBK
        0x0404 | 0x0C04 | 0x1404 => Some(950), // Chinese Traditional - Big5
        0x0412 => Some(949),                   // Korean - EUC-KR
        0x041E => Some(874),                   // Thai
        0x042A => Some(1258),                  // Vietnamese

        // Code Page 1250 - Central European
        0x0405 | 0x0415 | 0x040E | 0x041A | 0x081A | 0x141A | 0x101A | 0x041B | 0x0424 | 0x0418
        | 0x041C => Some(1250),

        // Code Page 1251 - Cyrillic
        0x0419 | 0x0422 | 0x0423 | 0x0402 | 0x042F | 0x0C1A | 0x201A | 0x0440 | 0x0843 | 0x0444
        | 0x0450 | 0x0485 => Some(1251),

        0x0408 => Some(1253),          // Greek
        0x041F | 0x042C => Some(1254), // Turkish, Azerbaijani
        0x040D => Some(1255),          // Hebrew

        // Code Page 1256 - Arabic
        0x0401 | 0x0801 | 0x0C01 | 0x1001 | 0x1401 | 0x1801 | 0x1C01 | 0x2001 | 0x2401 | 0x2801
        | 0x2C01 | 0x3001 | 0x3401 | 0x3801 | 0x3C01 | 0x4001 | 0x0429 | 0x0420 | 0x048C
        | 0x0463 => Some(1256),

        // Code Page 1257 - Baltic
        0x0425..=0x0427 => Some(1257),

        // Default to 1252 (Western European) for English and related languages
        0x0409 | 0x0809 | 0x0C09 | 0x1009 | 0x1409 | 0x1809 | 0x1C09 | 0x2009 | 0x2409 | 0x2809
        | 0x2C09 | 0x3009 | 0x3409 | 0x0407 | 0x0807 | 0x0C07 | 0x1007 | 0x1407 | 0x040C
        | 0x080C | 0x0C0C | 0x100C | 0x140C | 0x180C | 0x0410 | 0x0810 | 0x0413 | 0x0813
        | 0x0416 | 0x0816 | 0x040A | 0x080A | 0x0C0A | 0x100A | 0x140A | 0x180A | 0x1C0A
        | 0x200A | 0x240A | 0x280A | 0x2C0A | 0x300A | 0x340A | 0x380A | 0x3C0A | 0x400A
        | 0x440A | 0x480A | 0x4C0A | 0x500A => Some(1252),

        _ => Some(1252), // Default fallback
    }
}

impl TypeInfo {
    /// Create type info for a fixed-length integer type.
    #[must_use]
    pub fn int(type_id: u8) -> Self {
        Self {
            type_id,
            length: None,
            scale: None,
            precision: None,
            collation: None,
        }
    }

    /// Create type info for a variable-length type.
    #[must_use]
    pub fn varchar(length: u32) -> Self {
        Self {
            type_id: 0xE7, // NVARCHARTYPE
            length: Some(length),
            scale: None,
            precision: None,
            collation: None,
        }
    }

    /// Create type info for a decimal type.
    #[must_use]
    pub fn decimal(precision: u8, scale: u8) -> Self {
        Self {
            type_id: 0x6C,
            length: None,
            scale: Some(scale),
            precision: Some(precision),
            collation: None,
        }
    }

    /// Create type info for a datetime type with scale.
    #[must_use]
    pub fn datetime_with_scale(type_id: u8, scale: u8) -> Self {
        Self {
            type_id,
            length: None,
            scale: Some(scale),
            precision: None,
            collation: None,
        }
    }
}

/// Low-level TDS value decoder shared across the workspace crates.
///
/// Internal plumbing reached cross-crate only via [`crate::__private`]; not
/// public API and exempt from semver guarantees (see #242). The per-type
/// helpers it dispatches to remain module-private below.
pub(crate) mod sealed {
    use super::*;

    /// Decode a DECIMAL/NUMERIC value (1-byte length prefix, sign byte,
    /// little-endian mantissa).
    ///
    /// Shared with `mssql-client`'s `column_parser` so the two decode stacks
    /// cannot drift on the overflow policy (the divergence that caused #188).
    /// Generic over [`bytes::Buf`] so both `&[u8]` and [`Bytes`] callers reuse it.
    pub fn decode_decimal<B: Buf>(
        buf: &mut B,
        type_info: &TypeInfo,
    ) -> Result<SqlValue, TypeError> {
        super::decode_decimal(buf, type_info)
    }

    /// Number of TDS wire bytes a scale-`scale` TIME/DATETIME2/DATETIMEOFFSET
    /// time component occupies.
    ///
    /// Shared with `mssql-client`'s `column_parser` so the two decode stacks
    /// agree on the scale→width mapping (see #204). Pure scale math, so it is
    /// available without the `chrono` feature (column_parser validates frame
    /// lengths against it even when `chrono` is off).
    pub fn time_bytes_for_scale(scale: u8) -> usize {
        super::time_bytes_for_scale(scale)
    }

    /// Convert wire 100ns-interval-derived `intervals` (at scale `scale`) into a
    /// `NaiveTime`, shared with `column_parser` so the scale conversion cannot
    /// drift (see #204).
    #[cfg(feature = "chrono")]
    pub fn intervals_to_time(intervals: u64, scale: u8) -> chrono::NaiveTime {
        super::intervals_to_time(intervals, scale)
    }

    /// Decode a SQL value based on type information.
    pub fn decode_value(buf: &mut Bytes, type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
        match type_info.type_id {
            // Fixed-length types
            0x1F => Ok(SqlValue::Null),   // NULLTYPE
            0x32 => decode_bit(buf),      // BITTYPE
            0x30 => decode_tinyint(buf),  // INT1TYPE
            0x34 => decode_smallint(buf), // INT2TYPE
            0x38 => decode_int(buf),      // INT4TYPE
            0x7F => decode_bigint(buf),   // INT8TYPE
            0x3B => decode_float(buf),    // FLT4TYPE
            0x3E => decode_double(buf),   // FLT8TYPE

            // Nullable integer types (INTNTYPE)
            0x26 => decode_intn(buf, type_info),

            // Variable-length string types
            0xE7 => decode_nvarchar(buf, type_info), // NVARCHARTYPE
            0xAF => decode_varchar(buf, type_info),  // BIGCHARTYPE
            0xA7 => decode_varchar(buf, type_info),  // BIGVARCHARTYPE

            // Binary types
            0xA5 => decode_varbinary(buf, type_info), // BIGVARBINTYPE
            0xAD => decode_varbinary(buf, type_info), // BIGBINARYTYPE

            // GUID
            0x24 => decode_guid(buf),

            // Decimal/Numeric (0x6A/0x6C are the N-types; 0x3F is legacy NUMERICTYPE)
            0x6C | 0x6A | 0x3F => decode_decimal(buf, type_info),

            // Date/Time types
            0x28 => decode_date(buf),                      // DATETYPE
            0x29 => decode_time(buf, type_info),           // TIMETYPE
            0x2A => decode_datetime2(buf, type_info),      // DATETIME2TYPE
            0x2B => decode_datetimeoffset(buf, type_info), // DATETIMEOFFSETTYPE
            0x3A => decode_smalldatetime(buf),             // DATETIM4TYPE (smalldatetime)
            0x3D => decode_datetime(buf),                  // DATETIMETYPE

            // XML
            0xF1 => decode_xml(buf),

            _ => Err(TypeError::UnsupportedConversion {
                from: format!("TDS type 0x{:02X}", type_info.type_id),
                to: "SqlValue",
            }),
        }
    }
}

fn decode_bit(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }
    Ok(SqlValue::Bool(buf.get_u8() != 0))
}

fn decode_tinyint(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }
    Ok(SqlValue::TinyInt(buf.get_u8()))
}

fn decode_smallint(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 2 {
        return Err(TypeError::BufferTooSmall {
            needed: 2,
            available: buf.remaining(),
        });
    }
    Ok(SqlValue::SmallInt(buf.get_i16_le()))
}

fn decode_int(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 4 {
        return Err(TypeError::BufferTooSmall {
            needed: 4,
            available: buf.remaining(),
        });
    }
    Ok(SqlValue::Int(buf.get_i32_le()))
}

fn decode_bigint(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 8 {
        return Err(TypeError::BufferTooSmall {
            needed: 8,
            available: buf.remaining(),
        });
    }
    Ok(SqlValue::BigInt(buf.get_i64_le()))
}

fn decode_float(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 4 {
        return Err(TypeError::BufferTooSmall {
            needed: 4,
            available: buf.remaining(),
        });
    }
    Ok(SqlValue::Float(buf.get_f32_le()))
}

fn decode_double(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 8 {
        return Err(TypeError::BufferTooSmall {
            needed: 8,
            available: buf.remaining(),
        });
    }
    Ok(SqlValue::Double(buf.get_f64_le()))
}

fn decode_intn(buf: &mut Bytes, _type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let actual_len = buf.get_u8() as usize;
    if actual_len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < actual_len {
        return Err(TypeError::BufferTooSmall {
            needed: actual_len,
            available: buf.remaining(),
        });
    }

    match actual_len {
        1 => Ok(SqlValue::TinyInt(buf.get_u8())),
        2 => Ok(SqlValue::SmallInt(buf.get_i16_le())),
        4 => Ok(SqlValue::Int(buf.get_i32_le())),
        8 => Ok(SqlValue::BigInt(buf.get_i64_le())),
        _ => Err(TypeError::InvalidBinary(format!(
            "invalid INTN length: {actual_len}"
        ))),
    }
}

fn decode_nvarchar(buf: &mut Bytes, _type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 2 {
        return Err(TypeError::BufferTooSmall {
            needed: 2,
            available: buf.remaining(),
        });
    }

    let byte_len = buf.get_u16_le() as usize;

    // 0xFFFF indicates NULL
    if byte_len == 0xFFFF {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < byte_len {
        return Err(TypeError::BufferTooSmall {
            needed: byte_len,
            available: buf.remaining(),
        });
    }

    let utf16_data = buf.copy_to_bytes(byte_len);
    let s = decode_utf16_string(&utf16_data)?;
    Ok(SqlValue::String(s))
}

fn decode_varchar(buf: &mut Bytes, type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 2 {
        return Err(TypeError::BufferTooSmall {
            needed: 2,
            available: buf.remaining(),
        });
    }

    let byte_len = buf.get_u16_le() as usize;

    // 0xFFFF indicates NULL
    if byte_len == 0xFFFF {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < byte_len {
        return Err(TypeError::BufferTooSmall {
            needed: byte_len,
            available: buf.remaining(),
        });
    }

    let data = buf.copy_to_bytes(byte_len);

    // Try UTF-8 first (most common case and zero-cost for ASCII)
    if let Ok(s) = String::from_utf8(data.to_vec()) {
        return Ok(SqlValue::String(s));
    }

    // If UTF-8 fails, try collation-aware decoding
    #[cfg(feature = "encoding")]
    if let Some(ref collation) = type_info.collation {
        if let Some(encoding) = collation.encoding() {
            let (decoded, _, had_errors) = encoding.decode(&data);
            if !had_errors {
                return Ok(SqlValue::String(decoded.into_owned()));
            }
        }
    }

    // Suppress unused warning when encoding feature is disabled
    #[cfg(not(feature = "encoding"))]
    let _ = type_info;

    // Fallback: lossy UTF-8 conversion
    Ok(SqlValue::String(
        String::from_utf8_lossy(&data).into_owned(),
    ))
}

fn decode_varbinary(buf: &mut Bytes, _type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 2 {
        return Err(TypeError::BufferTooSmall {
            needed: 2,
            available: buf.remaining(),
        });
    }

    let byte_len = buf.get_u16_le() as usize;

    // 0xFFFF indicates NULL
    if byte_len == 0xFFFF {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < byte_len {
        return Err(TypeError::BufferTooSmall {
            needed: byte_len,
            available: buf.remaining(),
        });
    }

    let data = buf.copy_to_bytes(byte_len);
    Ok(SqlValue::Binary(data))
}

#[cfg(feature = "uuid")]
fn decode_guid(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if len != 16 {
        return Err(TypeError::InvalidBinary(format!(
            "invalid GUID length: {len}"
        )));
    }

    if buf.remaining() < 16 {
        return Err(TypeError::BufferTooSmall {
            needed: 16,
            available: buf.remaining(),
        });
    }

    // SQL Server stores UUIDs in mixed-endian format
    let mut bytes = [0u8; 16];

    // First 4 bytes - little-endian (reverse)
    bytes[3] = buf.get_u8();
    bytes[2] = buf.get_u8();
    bytes[1] = buf.get_u8();
    bytes[0] = buf.get_u8();

    // Next 2 bytes - little-endian (reverse)
    bytes[5] = buf.get_u8();
    bytes[4] = buf.get_u8();

    // Next 2 bytes - little-endian (reverse)
    bytes[7] = buf.get_u8();
    bytes[6] = buf.get_u8();

    // Last 8 bytes - big-endian (keep as-is)
    for byte in &mut bytes[8..16] {
        *byte = buf.get_u8();
    }

    Ok(SqlValue::Uuid(uuid::Uuid::from_bytes(bytes)))
}

#[cfg(not(feature = "uuid"))]
fn decode_guid(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    // Skip the GUID data
    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < len {
        return Err(TypeError::BufferTooSmall {
            needed: len,
            available: buf.remaining(),
        });
    }

    let data = buf.copy_to_bytes(len);
    Ok(SqlValue::Binary(data))
}

#[cfg(feature = "decimal")]
fn decode_decimal<B: Buf>(buf: &mut B, type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    use rust_decimal::Decimal;

    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < len {
        return Err(TypeError::BufferTooSmall {
            needed: len,
            available: buf.remaining(),
        });
    }

    // First byte is sign (0 = negative, 1 = positive)
    let sign = buf.get_u8();
    let remaining = len - 1;

    // Read mantissa (little-endian)
    let mut mantissa_bytes = [0u8; 16];
    for byte in mantissa_bytes.iter_mut().take(remaining.min(16)) {
        *byte = buf.get_u8();
    }
    // Consume any excess bytes so the frame stays aligned (shouldn't happen
    // with valid data; matches the column_parser decoder).
    for _ in 16..remaining {
        buf.get_u8();
    }

    let mantissa = u128::from_le_bytes(mantissa_bytes);
    let scale = type_info.scale.unwrap_or(0) as u32;

    // rust_decimal holds 96-bit mantissas with scale <= 28; SQL Server
    // NUMERIC goes to 38 digits, so legitimate wire values can exceed it.
    // That must be an error, not a silent fall back to f64 (~15-16
    // significant digits): a lossy value read, written back, or compared
    // downstream corrupts data. This matches the column_parser decoder —
    // the two policies diverged after #157 (issue #188).
    let decimal = i128::try_from(mantissa)
        .ok()
        .and_then(|m| Decimal::try_from_i128_with_scale(m, scale).ok());
    match decimal {
        Some(mut decimal) => {
            if sign == 0 {
                decimal.set_sign_negative(true);
            }
            Ok(SqlValue::Decimal(decimal))
        }
        None => Err(TypeError::InvalidDecimal(format!(
            "NUMERIC value (mantissa {mantissa}, scale {scale}) exceeds \
             rust_decimal's 96-bit/scale-28 range; CAST the column to a \
             narrower NUMERIC, FLOAT, or VARCHAR in the query"
        ))),
    }
}

#[cfg(not(feature = "decimal"))]
fn decode_decimal<B: Buf>(buf: &mut B, _type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    // Skip decimal data and return as string
    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < len {
        return Err(TypeError::BufferTooSmall {
            needed: len,
            available: buf.remaining(),
        });
    }

    buf.advance(len);
    Ok(SqlValue::String("DECIMAL (feature disabled)".to_string()))
}

#[cfg(feature = "chrono")]
fn decode_date(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if len != 3 {
        return Err(TypeError::InvalidDateTime(format!(
            "invalid DATE length: {len}"
        )));
    }

    if buf.remaining() < 3 {
        return Err(TypeError::BufferTooSmall {
            needed: 3,
            available: buf.remaining(),
        });
    }

    // 3 bytes little-endian representing days since 0001-01-01
    let days = buf.get_u8() as u32 | ((buf.get_u8() as u32) << 8) | ((buf.get_u8() as u32) << 16);

    let base = chrono::NaiveDate::from_ymd_opt(1, 1, 1).expect("valid date");
    let date = base + chrono::Duration::days(days as i64);

    Ok(SqlValue::Date(date))
}

#[cfg(not(feature = "chrono"))]
fn decode_date(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < len {
        return Err(TypeError::BufferTooSmall {
            needed: len,
            available: buf.remaining(),
        });
    }

    buf.advance(len);
    Ok(SqlValue::String("DATE (feature disabled)".to_string()))
}

#[cfg(feature = "chrono")]
fn decode_time(buf: &mut Bytes, type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    let scale = type_info.scale.unwrap_or(7);
    let time_len = time_bytes_for_scale(scale);

    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < len {
        return Err(TypeError::BufferTooSmall {
            needed: len,
            available: buf.remaining(),
        });
    }
    // Reads below are driven by scale metadata, not by `len`: a short
    // declared length must be an error, not a panic.
    if len < time_len {
        return Err(TypeError::InvalidDateTime(format!(
            "TIME length {len} too short for scale {scale}"
        )));
    }

    // Read time bytes (variable length based on scale)
    let mut time_bytes = [0u8; 8];
    for byte in time_bytes.iter_mut().take(time_len) {
        *byte = buf.get_u8();
    }

    let intervals = u64::from_le_bytes(time_bytes);
    let time = intervals_to_time(intervals, scale);

    Ok(SqlValue::Time(time))
}

#[cfg(not(feature = "chrono"))]
fn decode_time(buf: &mut Bytes, _type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < len {
        return Err(TypeError::BufferTooSmall {
            needed: len,
            available: buf.remaining(),
        });
    }

    buf.advance(len);
    Ok(SqlValue::String("TIME (feature disabled)".to_string()))
}

#[cfg(feature = "chrono")]
fn decode_datetime2(buf: &mut Bytes, type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    let scale = type_info.scale.unwrap_or(7);
    let time_len = time_bytes_for_scale(scale);

    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < len {
        return Err(TypeError::BufferTooSmall {
            needed: len,
            available: buf.remaining(),
        });
    }
    // Reads below are driven by scale metadata, not by `len`: a short
    // declared length must be an error, not a panic.
    if len < time_len + 3 {
        return Err(TypeError::InvalidDateTime(format!(
            "DATETIME2 length {len} too short for scale {scale}"
        )));
    }

    // Decode time
    let mut time_bytes = [0u8; 8];
    for byte in time_bytes.iter_mut().take(time_len) {
        *byte = buf.get_u8();
    }
    let intervals = u64::from_le_bytes(time_bytes);
    let time = intervals_to_time(intervals, scale);

    // Decode date
    let days = buf.get_u8() as u32 | ((buf.get_u8() as u32) << 8) | ((buf.get_u8() as u32) << 16);
    let base = chrono::NaiveDate::from_ymd_opt(1, 1, 1).expect("valid date");
    let date = base + chrono::Duration::days(days as i64);

    Ok(SqlValue::DateTime(date.and_time(time)))
}

#[cfg(not(feature = "chrono"))]
fn decode_datetime2(buf: &mut Bytes, _type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < len {
        return Err(TypeError::BufferTooSmall {
            needed: len,
            available: buf.remaining(),
        });
    }

    buf.advance(len);
    Ok(SqlValue::String("DATETIME2 (feature disabled)".to_string()))
}

#[cfg(feature = "chrono")]
fn decode_datetimeoffset(buf: &mut Bytes, type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    use chrono::TimeZone;

    let scale = type_info.scale.unwrap_or(7);
    let time_len = time_bytes_for_scale(scale);

    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < len {
        return Err(TypeError::BufferTooSmall {
            needed: len,
            available: buf.remaining(),
        });
    }
    // Reads below are driven by scale metadata, not by `len`: a short
    // declared length must be an error, not a panic.
    if len < time_len + 5 {
        return Err(TypeError::InvalidDateTime(format!(
            "DATETIMEOFFSET length {len} too short for scale {scale}"
        )));
    }

    // Decode time
    let mut time_bytes = [0u8; 8];
    for byte in time_bytes.iter_mut().take(time_len) {
        *byte = buf.get_u8();
    }
    let intervals = u64::from_le_bytes(time_bytes);
    let time = intervals_to_time(intervals, scale);

    // Decode date
    let days = buf.get_u8() as u32 | ((buf.get_u8() as u32) << 8) | ((buf.get_u8() as u32) << 16);
    let base = chrono::NaiveDate::from_ymd_opt(1, 1, 1).expect("valid date");
    let date = base + chrono::Duration::days(days as i64);

    // Decode timezone offset in minutes
    let offset_minutes = buf.get_i16_le();
    let offset = chrono::FixedOffset::east_opt((offset_minutes as i32) * 60)
        .ok_or_else(|| TypeError::InvalidDateTime(format!("invalid offset: {offset_minutes}")))?;

    // The wire date/time portion is UTC per MS-TDS §2.2.5.5.1.9; attach the
    // offset without shifting the instant.
    let datetime = offset.from_utc_datetime(&date.and_time(time));

    Ok(SqlValue::DateTimeOffset(datetime))
}

#[cfg(not(feature = "chrono"))]
fn decode_datetimeoffset(buf: &mut Bytes, _type_info: &TypeInfo) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 1 {
        return Err(TypeError::BufferTooSmall {
            needed: 1,
            available: buf.remaining(),
        });
    }

    let len = buf.get_u8() as usize;
    if len == 0 {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < len {
        return Err(TypeError::BufferTooSmall {
            needed: len,
            available: buf.remaining(),
        });
    }

    buf.advance(len);
    Ok(SqlValue::String(
        "DATETIMEOFFSET (feature disabled)".to_string(),
    ))
}

#[cfg(feature = "chrono")]
fn decode_datetime(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    // DATETIME is 8 bytes: 4 bytes days since 1900-01-01 + 4 bytes 300ths of second
    if buf.remaining() < 8 {
        return Err(TypeError::BufferTooSmall {
            needed: 8,
            available: buf.remaining(),
        });
    }

    let days = buf.get_i32_le();
    let time_300ths = buf.get_u32_le();

    let base = chrono::NaiveDate::from_ymd_opt(1900, 1, 1).expect("valid date");
    // days comes from the wire: out-of-range is an error, never a panic.
    let date = base
        .checked_add_signed(chrono::Duration::days(days as i64))
        .ok_or_else(|| TypeError::InvalidDateTime(format!("DATETIME days out of range: {days}")))?;

    // Convert 300ths of second to time
    let total_ms = (time_300ths as u64 * 1000) / 300;
    let secs = (total_ms / 1000) as u32;
    let nanos = ((total_ms % 1000) * 1_000_000) as u32;

    let time = chrono::NaiveTime::from_num_seconds_from_midnight_opt(secs, nanos)
        .ok_or_else(|| TypeError::InvalidDateTime("invalid DATETIME time".to_string()))?;

    Ok(SqlValue::DateTime(date.and_time(time)))
}

#[cfg(not(feature = "chrono"))]
fn decode_datetime(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 8 {
        return Err(TypeError::BufferTooSmall {
            needed: 8,
            available: buf.remaining(),
        });
    }

    buf.advance(8);
    Ok(SqlValue::String("DATETIME (feature disabled)".to_string()))
}

#[cfg(feature = "chrono")]
fn decode_smalldatetime(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    // SMALLDATETIME is 4 bytes: 2 bytes days since 1900-01-01 + 2 bytes minutes
    if buf.remaining() < 4 {
        return Err(TypeError::BufferTooSmall {
            needed: 4,
            available: buf.remaining(),
        });
    }

    let days = buf.get_u16_le();
    let minutes = buf.get_u16_le();

    let base = chrono::NaiveDate::from_ymd_opt(1900, 1, 1).expect("valid date");
    let date = base + chrono::Duration::days(days as i64);

    let time = chrono::NaiveTime::from_num_seconds_from_midnight_opt((minutes as u32) * 60, 0)
        .ok_or_else(|| TypeError::InvalidDateTime("invalid SMALLDATETIME time".to_string()))?;

    Ok(SqlValue::DateTime(date.and_time(time)))
}

#[cfg(not(feature = "chrono"))]
fn decode_smalldatetime(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    if buf.remaining() < 4 {
        return Err(TypeError::BufferTooSmall {
            needed: 4,
            available: buf.remaining(),
        });
    }

    buf.advance(4);
    Ok(SqlValue::String(
        "SMALLDATETIME (feature disabled)".to_string(),
    ))
}

fn decode_xml(buf: &mut Bytes) -> Result<SqlValue, TypeError> {
    // XML is sent as UTF-16LE string with length prefix
    if buf.remaining() < 2 {
        return Err(TypeError::BufferTooSmall {
            needed: 2,
            available: buf.remaining(),
        });
    }

    let byte_len = buf.get_u16_le() as usize;

    if byte_len == 0xFFFF {
        return Ok(SqlValue::Null);
    }

    if buf.remaining() < byte_len {
        return Err(TypeError::BufferTooSmall {
            needed: byte_len,
            available: buf.remaining(),
        });
    }

    let utf16_data = buf.copy_to_bytes(byte_len);
    let s = decode_utf16_string(&utf16_data)?;
    Ok(SqlValue::Xml(s))
}

/// Decode a UTF-16LE string from bytes.
pub fn decode_utf16_string(data: &[u8]) -> Result<String, TypeError> {
    if data.len() % 2 != 0 {
        return Err(TypeError::InvalidEncoding(
            "UTF-16 data must have even length".to_string(),
        ));
    }

    let utf16: Vec<u16> = data
        .chunks_exact(2)
        .map(|chunk| u16::from_le_bytes([chunk[0], chunk[1]]))
        .collect();

    String::from_utf16(&utf16).map_err(|e| TypeError::InvalidEncoding(e.to_string()))
}

/// Calculate number of bytes needed for TIME based on scale.
fn time_bytes_for_scale(scale: u8) -> usize {
    match scale {
        0..=2 => 3,
        3..=4 => 4,
        5..=7 => 5,
        _ => 5, // Default to max precision
    }
}

/// Convert 100-nanosecond intervals to NaiveTime.
#[cfg(feature = "chrono")]
fn intervals_to_time(intervals: u64, scale: u8) -> chrono::NaiveTime {
    // Scale determines the unit:
    // scale 0: seconds
    // scale 1: 100ms
    // scale 2: 10ms
    // scale 3: 1ms
    // scale 4: 100us
    // scale 5: 10us
    // scale 6: 1us
    // scale 7: 100ns

    // Saturating: `intervals` comes from the wire, and a hostile value must
    // not overflow-panic in debug builds (saturation lands in the
    // out-of-range fallback below).
    let nanos = match scale {
        0 => intervals.saturating_mul(1_000_000_000),
        1 => intervals.saturating_mul(100_000_000),
        2 => intervals.saturating_mul(10_000_000),
        3 => intervals.saturating_mul(1_000_000),
        4 => intervals.saturating_mul(100_000),
        5 => intervals.saturating_mul(10_000),
        6 => intervals.saturating_mul(1_000),
        7 => intervals.saturating_mul(100),
        _ => intervals.saturating_mul(100),
    };

    let secs = (nanos / 1_000_000_000) as u32;
    let nano_part = (nanos % 1_000_000_000) as u32;

    chrono::NaiveTime::from_num_seconds_from_midnight_opt(secs, nano_part)
        .unwrap_or_else(|| chrono::NaiveTime::from_hms_opt(0, 0, 0).expect("valid time"))
}

#[cfg(test)]
#[allow(clippy::unwrap_used, clippy::panic)]
mod tests {
    use super::sealed::decode_value;
    use super::*;

    #[test]
    fn test_decode_int() {
        let mut buf = Bytes::from_static(&[42, 0, 0, 0]);
        let type_info = TypeInfo::int(0x38);
        let result = decode_value(&mut buf, &type_info).unwrap();
        assert_eq!(result, SqlValue::Int(42));
    }

    // ---- #204: secondary-decode-stack regressions ----

    /// #204 bug 1: type 0x3F is legacy NUMERICTYPE, not SMALLDATETIME. It was
    /// wrongly dispatched to `decode_smalldatetime`; it must decode as a decimal.
    #[cfg(feature = "decimal")]
    #[test]
    fn legacy_numeric_0x3f_decodes_as_decimal() {
        // Decimal wire: [len][sign=1 (+)][magnitude LE]; 12345 = 0x3039, scale 2.
        let mut buf = Bytes::from_static(&[0x03, 0x01, 0x39, 0x30]);
        let type_info = TypeInfo {
            type_id: 0x3F,
            length: None,
            scale: Some(2),
            precision: Some(5),
            collation: None,
        };
        let result = decode_value(&mut buf, &type_info).unwrap();
        assert_eq!(result, SqlValue::Decimal("123.45".parse().unwrap()));
    }

    /// #204 bug 1 (other half): 0x3A is DATETIM4 (SMALLDATETIME) and previously
    /// had no arm at all — the smalldatetime decoder was wired to 0x3F instead.
    #[cfg(feature = "chrono")]
    #[test]
    fn smalldatetime_0x3a_decodes() {
        // Raw 4 bytes: u16 days since 1900-01-01 + u16 minutes. 100 days, 600 min.
        let mut buf = Bytes::from_static(&[100, 0, 0x58, 0x02]);
        let type_info = TypeInfo {
            type_id: 0x3A,
            length: None,
            scale: None,
            precision: None,
            collation: None,
        };
        let result = decode_value(&mut buf, &type_info).unwrap();
        let expected = chrono::NaiveDate::from_ymd_opt(1900, 1, 1)
            .unwrap()
            .checked_add_signed(chrono::Duration::days(100))
            .unwrap()
            .and_hms_opt(10, 0, 0)
            .unwrap();
        assert_eq!(result, SqlValue::DateTime(expected));
    }

    /// #204 bug 2: the secondary stack masked the LCID to 10 bits before matching
    /// full 16-bit LCIDs, so every non-Latin collation collapsed to codepage 1252.
    /// A Japanese collation must decode cp932. Uses the real `Japanese_CI_AS` lcid
    /// (sort/version bits set: 0x00D00411) so the mask width genuinely matters —
    /// pre-fix this decoded the bytes as Windows-1252 garbage.
    #[cfg(feature = "encoding")]
    #[test]
    fn varchar_japanese_collation_decodes_cp932() {
        use bytes::BufMut;
        let (cp932, _, had_err) = encoding_rs::SHIFT_JIS.encode("あい");
        assert!(!had_err, "sample must be representable in cp932");
        let mut wire = bytes::BytesMut::new();
        wire.put_u16_le(cp932.len() as u16);
        wire.put_slice(&cp932);
        let type_info = TypeInfo {
            type_id: 0xA7, // BIGVARCHARTYPE
            length: None,
            scale: None,
            precision: None,
            collation: Some(Collation {
                lcid: 13632529, // Japanese_CI_AS (0x00D00411)
                flags: 0,
            }),
        };
        let result = decode_value(&mut wire.freeze(), &type_info).unwrap();
        assert_eq!(result, SqlValue::String("あい".to_string()));
    }

    /// The mixed-endian GUID encoding must round-trip: encode then decode
    /// returns the original UUID. A UUID with all-distinct bytes catches an
    /// asymmetric swap. `decode_guid` previously had no unit coverage — only
    /// live round-trips exercised it.
    #[cfg(feature = "uuid")]
    #[test]
    fn test_guid_encode_decode_roundtrip() {
        use crate::encode::encode_uuid;
        use bytes::BufMut;

        let original = uuid::Uuid::parse_str("00112233-4455-6677-8899-aabbccddeeff").unwrap();
        let mut encoded = bytes::BytesMut::new();
        encode_uuid(original, &mut encoded);
        assert_eq!(encoded.len(), 16);

        // `decode_guid` expects a length-prefixed buffer (1-byte length = 16).
        let mut framed = bytes::BytesMut::new();
        framed.put_u8(16);
        framed.put_slice(&encoded);
        let decoded = decode_guid(&mut framed.freeze()).unwrap();
        assert_eq!(decoded, SqlValue::Uuid(original));
    }

    #[cfg(feature = "chrono")]
    #[test]
    fn hostile_datetime_days_overflow_is_error_not_panic() {
        // days=i32::MAX from the wire overflows chrono's date range; must be
        // a TypeError, never a panic.
        let mut data = Vec::new();
        data.extend_from_slice(&i32::MAX.to_le_bytes());
        data.extend_from_slice(&0u32.to_le_bytes());
        let mut buf = Bytes::from(data);
        assert!(decode_datetime(&mut buf).is_err());
    }

    /// Issue #152 regression: the DATETIMEOFFSET wire date/time portion is
    /// the UTC instant per MS-TDS §2.2.5.5.1.9; decoding must attach the
    /// offset without shifting it. Wire 10:00 UTC with +02:00 → 12:00+02:00.
    /// The previous from_local_datetime decode produced 10:00+02:00 (a
    /// different instant), which round-tripped only against our own equally
    /// inverted encoder.
    #[cfg(feature = "chrono")]
    #[test]
    fn test_datetimeoffset_decodes_wire_as_utc() {
        use chrono::TimeZone;

        let mut data = Vec::new();
        data.push(10u8); // BYTELEN: 5 time + 3 date + 2 offset
        let intervals: u64 = 10 * 3600 * 10_000_000; // 10:00:00 UTC, scale 7
        for i in 0..5 {
            data.push(((intervals >> (8 * i)) & 0xFF) as u8);
        }
        let base = chrono::NaiveDate::from_ymd_opt(1, 1, 1).unwrap();
        let days = (chrono::NaiveDate::from_ymd_opt(2024, 3, 15).unwrap() - base).num_days() as u32;
        data.push((days & 0xFF) as u8);
        data.push(((days >> 8) & 0xFF) as u8);
        data.push(((days >> 16) & 0xFF) as u8);
        data.extend_from_slice(&120i16.to_le_bytes()); // +02:00

        let type_info = TypeInfo {
            type_id: 0x2B,
            length: None,
            scale: Some(7),
            precision: None,
            collation: None,
        };
        let mut buf = Bytes::from(data);
        let value = decode_datetimeoffset(&mut buf, &type_info).unwrap();

        let offset = chrono::FixedOffset::east_opt(2 * 3600).unwrap();
        let expected = offset.with_ymd_and_hms(2024, 3, 15, 12, 0, 0).unwrap();
        match value {
            SqlValue::DateTimeOffset(dt) => {
                assert_eq!(dt, expected);
                assert_eq!(dt.offset().local_minus_utc(), 7200);
                assert_eq!(
                    dt.naive_utc(),
                    chrono::NaiveDate::from_ymd_opt(2024, 3, 15)
                        .unwrap()
                        .and_hms_opt(10, 0, 0)
                        .unwrap()
                );
            }
            other => panic!("expected DateTimeOffset, got {other:?}"),
        }
    }

    #[test]
    fn test_decode_utf16_string() {
        // "AB" in UTF-16LE
        let data = [0x41, 0x00, 0x42, 0x00];
        let result = decode_utf16_string(&data).unwrap();
        assert_eq!(result, "AB");
    }

    #[test]
    fn test_decode_nvarchar() {
        // Length (4 bytes for "AB") + "AB" in UTF-16LE
        let mut buf = Bytes::from_static(&[4, 0, 0x41, 0x00, 0x42, 0x00]);
        let type_info = TypeInfo::varchar(100);
        let type_info = TypeInfo {
            type_id: 0xE7,
            ..type_info
        };
        let result = decode_value(&mut buf, &type_info).unwrap();
        assert_eq!(result, SqlValue::String("AB".to_string()));
    }

    #[test]
    fn test_decode_null_nvarchar() {
        // 0xFFFF indicates NULL
        let mut buf = Bytes::from_static(&[0xFF, 0xFF]);
        let type_info = TypeInfo {
            type_id: 0xE7,
            length: Some(100),
            scale: None,
            precision: None,
            collation: None,
        };
        let result = decode_value(&mut buf, &type_info).unwrap();
        assert_eq!(result, SqlValue::Null);
    }

    // ========================================================================
    // Targeted round-trip tests for negative decimals (work-item 1.3)
    // ========================================================================

    #[cfg(feature = "decimal")]
    mod decimal_roundtrip {
        use super::*;
        use bytes::{BufMut, BytesMut};
        use rust_decimal::Decimal;

        /// Encode a Decimal, prepend TDS length byte, then decode — verifying round-trip.
        fn roundtrip_decimal(value: Decimal, precision: u8, scale: u8) -> Decimal {
            // Encode
            let mut encode_buf = BytesMut::new();
            crate::encode::encode_decimal(value, &mut encode_buf);
            let encoded_len = encode_buf.len() as u8; // 17 bytes (1 sign + 16 mantissa)

            // Build decode buffer: length prefix + encoded data
            let mut decode_buf = BytesMut::with_capacity(1 + encoded_len as usize);
            decode_buf.put_u8(encoded_len);
            decode_buf.extend_from_slice(&encode_buf);

            let mut bytes = decode_buf.freeze();
            let type_info = TypeInfo::decimal(precision, scale);
            match decode_value(&mut bytes, &type_info).unwrap() {
                SqlValue::Decimal(d) => d,
                other => panic!("expected Decimal, got {other:?}"),
            }
        }

        #[test]
        fn test_negative_decimal_17_80() {
            let d = Decimal::new(-1780, 2); // -17.80
            let result = roundtrip_decimal(d, 18, 2);
            assert_eq!(result, d, "round-trip of -17.80 must be exact");
        }

        #[test]
        fn test_negative_decimal_0_01() {
            let d = Decimal::new(-1, 2); // -0.01
            let result = roundtrip_decimal(d, 18, 2);
            assert_eq!(result, d, "round-trip of -0.01 must be exact");
        }

        #[test]
        fn test_negative_decimal_large() {
            let d = Decimal::new(-9999999999, 2); // -99999999.99
            let result = roundtrip_decimal(d, 18, 2);
            assert_eq!(result, d, "round-trip of -99999999.99 must be exact");
        }

        #[test]
        fn test_positive_decimal() {
            let d = Decimal::new(1780, 2); // 17.80
            let result = roundtrip_decimal(d, 18, 2);
            assert_eq!(result, d, "round-trip of 17.80 must be exact");
        }

        #[test]
        fn test_decimal_zero() {
            let d = Decimal::ZERO;
            let result = roundtrip_decimal(d, 18, 0);
            assert_eq!(result, d, "round-trip of 0 must be exact");
        }

        #[test]
        fn test_decimal_max_precision() {
            // Large value that fits in 38-digit precision
            let d = Decimal::new(i64::MAX, 0);
            let result = roundtrip_decimal(d, 38, 0);
            assert_eq!(result, d, "round-trip of large positive must be exact");
        }

        #[test]
        fn test_decimal_min_precision() {
            let d = Decimal::new(i64::MIN + 1, 0);
            let result = roundtrip_decimal(d, 38, 0);
            assert_eq!(result, d, "round-trip of large negative must be exact");
        }

        /// Issue #188: a NUMERIC beyond rust_decimal's range must be a hard
        /// error, matching the column_parser policy from #157 — not a silent
        /// fall back to f64 (~15-16 significant digits of a 38-digit value).
        #[test]
        fn test_decimal_out_of_range_errors_instead_of_f64() {
            // 16 mantissa bytes of 0xFF = u128::MAX, far past 96 bits.
            let mut buf = BytesMut::new();
            buf.put_u8(17); // length: sign + 16 mantissa bytes
            buf.put_u8(1); // sign: positive
            buf.extend_from_slice(&[0xFF; 16]);

            let mut bytes = buf.freeze();
            let type_info = TypeInfo::decimal(38, 0);
            match decode_value(&mut bytes, &type_info) {
                Err(TypeError::InvalidDecimal(_)) => {}
                other => panic!("expected InvalidDecimal error, got {other:?}"),
            }
        }

        /// Issue #188: mantissa bytes beyond the 16 we can hold must still be
        /// consumed, or every following column in the row misparses.
        #[test]
        fn test_decimal_oversized_mantissa_keeps_frame_aligned() {
            let mut buf = BytesMut::new();
            buf.put_u8(18); // length: sign + 17 mantissa bytes (one excess)
            buf.put_u8(1); // sign: positive
            buf.put_u8(42); // mantissa = 42
            buf.extend_from_slice(&[0u8; 16]); // 15 in-range zeros + 1 excess
            buf.put_u8(0xAB); // sentinel: the next value in the frame

            let mut bytes = buf.freeze();
            let type_info = TypeInfo::decimal(38, 0);
            let value = decode_value(&mut bytes, &type_info).unwrap();
            assert_eq!(value, SqlValue::Decimal(Decimal::new(42, 0)));
            assert_eq!(
                bytes.remaining(),
                1,
                "excess mantissa bytes must be consumed, leaving the sentinel"
            );
            assert_eq!(bytes.get_u8(), 0xAB);
        }
    }

    // ========================================================================
    // Date encoding tests (work-item 3.7)
    // ========================================================================

    #[cfg(feature = "chrono")]
    mod date_tests {
        use bytes::{BufMut, BytesMut};
        use chrono::NaiveDate;

        #[test]
        fn test_encode_date_pre_1900() {
            // This is the scenario where Tiberius panics
            let mut buf = BytesMut::new();
            let date = NaiveDate::from_ymd_opt(1753, 1, 1).unwrap();
            crate::encode::encode_date(date, &mut buf).unwrap();
            assert_eq!(buf.len(), 3, "DATE encoding is always 3 bytes");
        }

        #[test]
        fn test_encode_date_epoch() {
            // The DATE epoch: 0001-01-01
            let mut buf = BytesMut::new();
            let date = NaiveDate::from_ymd_opt(1, 1, 1).unwrap();
            crate::encode::encode_date(date, &mut buf).unwrap();
            // Days since 0001-01-01 = 0
            assert_eq!(&buf[..], &[0, 0, 0]);
        }

        #[test]
        fn test_encode_date_max() {
            // SQL Server max DATE: 9999-12-31
            let mut buf = BytesMut::new();
            let date = NaiveDate::from_ymd_opt(9999, 12, 31).unwrap();
            crate::encode::encode_date(date, &mut buf).unwrap();
            assert_eq!(buf.len(), 3, "DATE encoding is always 3 bytes");
            // 3652058 days from 0001-01-01 — fits in 3 bytes (max ~16M)
            let days = buf[0] as u32 | ((buf[1] as u32) << 8) | ((buf[2] as u32) << 16);
            assert_eq!(days, 3_652_058);
        }

        #[test]
        fn test_decode_datetime_pre_1900() {
            // DATETIME uses i32 days from 1900-01-01 epoch.
            // 1753-01-01 is ~53690 days before 1900-01-01.
            use super::*;

            let base = NaiveDate::from_ymd_opt(1900, 1, 1).unwrap();
            let target = NaiveDate::from_ymd_opt(1753, 1, 1).unwrap();
            let days = target.signed_duration_since(base).num_days() as i32;

            // Build DATETIME buffer: i32 days + u32 time_300ths
            let mut raw = BytesMut::new();
            raw.put_i32_le(days);
            raw.put_u32_le(0); // midnight

            let mut buf = raw.freeze();
            let result = decode_datetime(&mut buf).unwrap();

            match result {
                SqlValue::DateTime(dt) => {
                    assert_eq!(dt.date(), target);
                }
                other => panic!("expected DateTime, got {other:?}"),
            }
        }

        #[test]
        fn test_decode_smalldatetime_1900() {
            // SMALLDATETIME: u16 days from 1900-01-01 + u16 minutes
            use super::*;

            // Day 0, minute 0 = 1900-01-01 00:00:00
            let mut raw = BytesMut::new();
            raw.put_u16_le(0);
            raw.put_u16_le(0);

            let mut buf = raw.freeze();
            let result = decode_smalldatetime(&mut buf).unwrap();

            match result {
                SqlValue::DateTime(dt) => {
                    assert_eq!(
                        dt,
                        NaiveDate::from_ymd_opt(1900, 1, 1)
                            .unwrap()
                            .and_hms_opt(0, 0, 0)
                            .unwrap()
                    );
                }
                other => panic!("expected DateTime, got {other:?}"),
            }
        }
    }

    // ========================================================================
    // Property-based tests (work-item 5.2)
    // ========================================================================

    #[cfg(feature = "decimal")]
    mod proptest_decimal {
        use super::*;
        use bytes::{BufMut, BytesMut};
        use proptest::prelude::*;
        use rust_decimal::Decimal;

        /// Encode a Decimal, prepend TDS length byte, then decode.
        fn roundtrip_decimal(value: Decimal, scale: u8) -> Decimal {
            let mut encode_buf = BytesMut::new();
            crate::encode::encode_decimal(value, &mut encode_buf);
            let encoded_len = encode_buf.len() as u8;

            let mut decode_buf = BytesMut::with_capacity(1 + encoded_len as usize);
            decode_buf.put_u8(encoded_len);
            decode_buf.extend_from_slice(&encode_buf);

            let mut bytes = decode_buf.freeze();
            let type_info = TypeInfo::decimal(38, scale);
            match decode_value(&mut bytes, &type_info).unwrap() {
                SqlValue::Decimal(d) => d,
                other => panic!("expected Decimal, got {other:?}"),
            }
        }

        proptest! {
            #[test]
            fn decimal_roundtrip_scale0(mantissa in -999_999_999_999i64..=999_999_999_999i64) {
                let d = Decimal::new(mantissa, 0);
                let result = roundtrip_decimal(d, 0);
                prop_assert_eq!(result, d);
            }

            #[test]
            fn decimal_roundtrip_scale2(mantissa in -999_999_999_999i64..=999_999_999_999i64) {
                let d = Decimal::new(mantissa, 2);
                let result = roundtrip_decimal(d, 2);
                prop_assert_eq!(result, d);
            }

            #[test]
            fn decimal_roundtrip_various_scales(
                mantissa in -999_999_999i64..=999_999_999i64,
                scale in 0u8..=10u8,
            ) {
                let d = Decimal::new(mantissa, scale as u32);
                let result = roundtrip_decimal(d, scale);
                prop_assert_eq!(result, d);
            }
        }
    }

    #[cfg(feature = "chrono")]
    mod proptest_date {
        use bytes::BytesMut;
        use chrono::NaiveDate;
        use proptest::prelude::*;

        proptest! {
            #[test]
            fn date_encode_never_panics(
                year in 1i32..=9999i32,
                month in 1u32..=12u32,
                day in 1u32..=28u32, // 28 is always valid
            ) {
                let date = NaiveDate::from_ymd_opt(year, month, day).unwrap();
                let mut buf = BytesMut::new();
                crate::encode::encode_date(date, &mut buf).unwrap();
                prop_assert_eq!(buf.len(), 3);
            }
        }
    }
}