tds_protocol/

token.rs

//! TDS token stream definitions.
//!
//! Tokens are the fundamental units of TDS response data. The server sends
//! a stream of tokens that describe metadata, rows, errors, and other information.
//!
//! ## Token Structure
//!
//! Each token begins with a 1-byte token type identifier, followed by
//! token-specific data. Some tokens have fixed lengths, while others
//! have length prefixes.
//!
//! ## Usage
//!
//! ```rust,no_run
//! use tds_protocol::token::{Token, TokenParser};
//! use bytes::Bytes;
//!
//! fn parse(data: Bytes) -> Result<(), tds_protocol::ProtocolError> {
//!     let mut parser = TokenParser::new(data);
//!
//!     while let Some(token) = parser.next_token()? {
//!         match token {
//!             Token::Done(done) => println!("Rows affected: {}", done.row_count),
//!             Token::Error(err) => eprintln!("Error {}: {}", err.number, err.message),
//!             _ => {}
//!         }
//!     }
//!     Ok(())
//! }
//! ```

use bytes::{Buf, BufMut, Bytes};

use crate::codec::{read_b_varchar, read_us_varchar};
use crate::error::ProtocolError;
use crate::prelude::*;
use crate::types::TypeId;

/// Token type identifier.
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
#[repr(u8)]
#[non_exhaustive]
pub enum TokenType {
    /// Column metadata (COLMETADATA).
    ColMetaData = 0x81,
    /// Error message (ERROR).
    Error = 0xAA,
    /// Informational message (INFO).
    Info = 0xAB,
    /// Login acknowledgment (LOGINACK).
    LoginAck = 0xAD,
    /// Row data (ROW).
    Row = 0xD1,
    /// Null bitmap compressed row (NBCROW).
    NbcRow = 0xD2,
    /// Environment change (ENVCHANGE).
    EnvChange = 0xE3,
    /// SSPI authentication (SSPI).
    Sspi = 0xED,
    /// Done (DONE).
    Done = 0xFD,
    /// Done in procedure (DONEINPROC).
    DoneInProc = 0xFF,
    /// Done procedure (DONEPROC).
    DoneProc = 0xFE,
    /// Return status (RETURNSTATUS).
    ReturnStatus = 0x79,
    /// Return value (RETURNVALUE).
    ReturnValue = 0xAC,
    /// Order (ORDER).
    Order = 0xA9,
    /// Feature extension acknowledgment (FEATUREEXTACK).
    FeatureExtAck = 0xAE,
    /// Session state (SESSIONSTATE).
    SessionState = 0xE4,
    /// Federated authentication info (FEDAUTHINFO).
    FedAuthInfo = 0xEE,
    /// Column info (COLINFO).
    ColInfo = 0xA5,
    /// Table name (TABNAME).
    TabName = 0xA4,
    /// Offset (OFFSET).
    Offset = 0x78,
    /// Compute (COMPUTE BY) column metadata (ALTMETADATA). Deprecated in TDS 7.4.
    AltMetaData = 0x88,
    /// Compute (COMPUTE BY) row data (ALTROW). Deprecated in TDS 7.4.
    AltRow = 0xD3,
}

impl TokenType {
    /// Create a token type from a raw byte.
    pub fn from_u8(value: u8) -> Option<Self> {
        match value {
            0x81 => Some(Self::ColMetaData),
            0xAA => Some(Self::Error),
            0xAB => Some(Self::Info),
            0xAD => Some(Self::LoginAck),
            0xD1 => Some(Self::Row),
            0xD2 => Some(Self::NbcRow),
            0xE3 => Some(Self::EnvChange),
            0xED => Some(Self::Sspi),
            0xFD => Some(Self::Done),
            0xFF => Some(Self::DoneInProc),
            0xFE => Some(Self::DoneProc),
            0x79 => Some(Self::ReturnStatus),
            0xAC => Some(Self::ReturnValue),
            0xA9 => Some(Self::Order),
            0xAE => Some(Self::FeatureExtAck),
            0xE4 => Some(Self::SessionState),
            0xEE => Some(Self::FedAuthInfo),
            0xA5 => Some(Self::ColInfo),
            0xA4 => Some(Self::TabName),
            0x78 => Some(Self::Offset),
            0x88 => Some(Self::AltMetaData),
            0xD3 => Some(Self::AltRow),
            _ => None,
        }
    }
}

/// Parsed TDS token.
///
/// This enum represents all possible tokens that can be received from SQL Server.
/// Each variant contains the parsed token data.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum Token {
    /// Column metadata describing result set structure.
    ColMetaData(ColMetaData),
    /// Row data.
    Row(RawRow),
    /// Null bitmap compressed row.
    NbcRow(NbcRow),
    /// Completion of a SQL statement.
    Done(Done),
    /// Completion of a stored procedure.
    DoneProc(DoneProc),
    /// Completion within a stored procedure.
    DoneInProc(DoneInProc),
    /// Return status from stored procedure.
    ReturnStatus(i32),
    /// Return value from stored procedure.
    ReturnValue(ReturnValue),
    /// Error message from server.
    Error(ServerError),
    /// Informational message from server.
    Info(ServerInfo),
    /// Login acknowledgment.
    LoginAck(LoginAck),
    /// Environment change notification.
    EnvChange(EnvChange),
    /// Column ordering information.
    Order(Order),
    /// Feature extension acknowledgment.
    FeatureExtAck(FeatureExtAck),
    /// SSPI authentication data.
    Sspi(SspiToken),
    /// Session state information.
    SessionState(SessionState),
    /// Federated authentication info.
    FedAuthInfo(FedAuthInfo),
}

/// Column metadata token.
#[derive(Debug, Clone, Default)]
pub struct ColMetaData {
    /// Column definitions.
    pub columns: Vec<ColumnData>,
    /// CEK table for Always Encrypted result sets.
    /// Present only when the server sends encrypted column metadata.
    pub cek_table: Option<crate::crypto::CekTable>,
}

/// Column definition within metadata.
#[derive(Debug, Clone)]
pub struct ColumnData {
    /// Column name.
    pub name: String,
    /// Column data type ID.
    pub type_id: TypeId,
    /// Column data type raw byte (for unknown types).
    pub col_type: u8,
    /// Column flags.
    pub flags: u16,
    /// User type ID.
    pub user_type: u32,
    /// Type-specific metadata.
    pub type_info: TypeInfo,
    /// Per-column encryption metadata (Always Encrypted).
    /// Present only for columns with the encrypted flag (0x0800) set.
    pub crypto_metadata: Option<crate::crypto::CryptoMetadata>,
}

/// Type-specific metadata.
#[derive(Debug, Clone, Default)]
pub struct TypeInfo {
    /// Maximum length for variable-length types.
    pub max_length: Option<u32>,
    /// Precision for numeric types.
    pub precision: Option<u8>,
    /// Scale for numeric types.
    pub scale: Option<u8>,
    /// Collation for string types.
    pub collation: Option<Collation>,
}

/// SQL Server collation.
///
/// Collations in SQL Server define the character encoding and sorting rules
/// for string data. For `VARCHAR` columns, the collation determines which
/// code page (character encoding) is used to store the data.
///
/// # Encoding Support
///
/// When the `encoding` feature is enabled, the [`Collation::encoding()`] method
/// returns the appropriate [`encoding_rs::Encoding`] for decoding `VARCHAR` data.
///
/// # Example
///
/// ```rust,ignore
/// use tds_protocol::token::Collation;
///
/// let collation = Collation { lcid: 0x0804, sort_id: 0 }; // Chinese (PRC)
/// if let Some(encoding) = collation.encoding() {
///     let (decoded, _, _) = encoding.decode(raw_bytes);
///     // decoded is now proper Chinese text
/// }
/// ```
#[derive(Debug, Clone, Copy, Default)]
pub struct Collation {
    /// Locale ID (LCID).
    ///
    /// The LCID encodes both the language and region. The lower 16 bits
    /// contain the primary language ID, and bits 16-19 contain the sort ID
    /// for some collations.
    ///
    /// For UTF-8 collations (SQL Server 2019+), fUTF8 (bit 26, 0x0400_0000) is set.
    pub lcid: u32,
    /// Sort ID.
    ///
    /// Used with certain collations to specify sorting behavior.
    pub sort_id: u8,
}

impl Collation {
    /// Create a `Collation` from the 5-byte TDS wire format.
    ///
    /// Format: 4 bytes LCID (little-endian u32) + 1 byte sort ID.
    pub fn from_bytes(bytes: &[u8; 5]) -> Self {
        Self {
            lcid: u32::from_le_bytes([bytes[0], bytes[1], bytes[2], bytes[3]]),
            sort_id: bytes[4],
        }
    }

    /// Serialize to the 5-byte TDS wire format.
    ///
    /// Format: 4 bytes LCID (little-endian u32) + 1 byte sort ID.
    pub fn to_bytes(&self) -> [u8; 5] {
        let b = self.lcid.to_le_bytes();
        [b[0], b[1], b[2], b[3], self.sort_id]
    }

    /// Returns the character encoding for this collation.
    ///
    /// This method maps the collation's LCID to the appropriate character
    /// encoding from the `encoding_rs` crate.
    ///
    /// # Returns
    ///
    /// - `Some(&Encoding)` - The encoding to use for decoding `VARCHAR` data
    /// - `None` - If the collation uses UTF-8 (no transcoding needed) or
    ///   the LCID is not recognized (caller should use Windows-1252 fallback)
    ///
    /// # UTF-8 Collations
    ///
    /// SQL Server 2019+ supports UTF-8 collations (identified by the `_UTF8`
    /// suffix). These return `None` because no transcoding is needed.
    ///
    /// # Example
    ///
    /// ```rust,ignore
    /// let collation = Collation { lcid: 0x0419, sort_id: 0 }; // Russian
    /// if let Some(encoding) = collation.encoding() {
    ///     // encoding is Windows-1251 for Cyrillic
    ///     let (text, _, had_errors) = encoding.decode(&raw_bytes);
    /// }
    /// ```
    #[cfg(feature = "encoding")]
    pub fn encoding(&self) -> Option<&'static encoding_rs::Encoding> {
        // A non-zero SortId means a SQL collation, whose code page derives
        // from the SortId, not the LCID (MS-TDS). Consulting only the LCID
        // silently decoded these as windows-1252 (issue #158).
        if self.sort_id != 0 {
            return crate::collation::encoding_for_sort_id(self.sort_id);
        }
        crate::collation::encoding_for_lcid(self.lcid)
    }

    /// Returns whether this collation uses UTF-8 encoding.
    ///
    /// UTF-8 collations were introduced in SQL Server 2019 and are
    /// identified by the `_UTF8` suffix in the collation name.
    #[cfg(feature = "encoding")]
    pub fn is_utf8(&self) -> bool {
        crate::collation::is_utf8_collation(self.lcid)
    }

    /// Returns the Windows code page number for this collation.
    ///
    /// Useful for error messages and debugging.
    ///
    /// # Returns
    ///
    /// The code page number (e.g., 1252 for Western European, 932 for Japanese).
    #[cfg(feature = "encoding")]
    pub fn code_page(&self) -> Option<u16> {
        // SQL collations (non-zero SortId) derive their code page from the
        // SortId, not the LCID (issue #158).
        if self.sort_id != 0 {
            return crate::collation::code_page_for_sort_id(self.sort_id);
        }
        crate::collation::code_page_for_lcid(self.lcid)
    }

    /// Returns the encoding name for this collation.
    ///
    /// Useful for error messages and debugging.
    #[cfg(feature = "encoding")]
    pub fn encoding_name(&self) -> &'static str {
        if self.sort_id != 0 {
            return match crate::collation::encoding_for_sort_id(self.sort_id) {
                Some(enc) => enc.name(),
                None => "unsupported",
            };
        }
        crate::collation::encoding_name_for_lcid(self.lcid)
    }
}

/// Raw row data (not yet decoded).
#[derive(Debug, Clone)]
pub struct RawRow {
    /// Raw column values.
    pub data: bytes::Bytes,
}

/// Null bitmap compressed row.
#[derive(Debug, Clone)]
pub struct NbcRow {
    /// Null bitmap.
    pub null_bitmap: Vec<u8>,
    /// Raw non-null column values.
    pub data: bytes::Bytes,
}

/// Done token indicating statement completion.
#[derive(Debug, Clone, Copy)]
pub struct Done {
    /// Status flags.
    pub status: DoneStatus,
    /// Current command.
    pub cur_cmd: u16,
    /// Row count (if applicable).
    pub row_count: u64,
}

/// Done status flags.
#[derive(Debug, Clone, Copy, Default)]
#[non_exhaustive]
pub struct DoneStatus {
    /// More results follow.
    pub more: bool,
    /// Error occurred.
    pub error: bool,
    /// Transaction in progress.
    pub in_xact: bool,
    /// Row count is valid.
    pub count: bool,
    /// Attention acknowledgment.
    pub attn: bool,
    /// Server error caused statement termination.
    pub srverror: bool,
}

/// Done in procedure token.
#[derive(Debug, Clone, Copy)]
pub struct DoneInProc {
    /// Status flags.
    pub status: DoneStatus,
    /// Current command.
    pub cur_cmd: u16,
    /// Row count.
    pub row_count: u64,
}

/// Done procedure token.
#[derive(Debug, Clone, Copy)]
pub struct DoneProc {
    /// Status flags.
    pub status: DoneStatus,
    /// Current command.
    pub cur_cmd: u16,
    /// Row count.
    pub row_count: u64,
}

/// Return value from stored procedure.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub struct ReturnValue {
    /// Parameter ordinal.
    pub param_ordinal: u16,
    /// Parameter name.
    pub param_name: String,
    /// Status flags.
    pub status: u8,
    /// User type.
    pub user_type: u32,
    /// Type flags.
    pub flags: u16,
    /// Raw column type byte from the wire.
    pub col_type: u8,
    /// Type info.
    pub type_info: TypeInfo,
    /// Value data.
    pub value: bytes::Bytes,
}

/// Server error message.
#[derive(Debug, Clone)]
pub struct ServerError {
    /// Error number.
    pub number: i32,
    /// Error state.
    pub state: u8,
    /// Error severity class.
    pub class: u8,
    /// Error message text.
    pub message: String,
    /// Server name.
    pub server: String,
    /// Procedure name.
    pub procedure: String,
    /// Line number.
    pub line: i32,
}

/// Server informational message.
#[derive(Debug, Clone)]
pub struct ServerInfo {
    /// Info number.
    pub number: i32,
    /// Info state.
    pub state: u8,
    /// Info class (severity).
    pub class: u8,
    /// Info message text.
    pub message: String,
    /// Server name.
    pub server: String,
    /// Procedure name.
    pub procedure: String,
    /// Line number.
    pub line: i32,
}

/// Login acknowledgment token.
#[derive(Debug, Clone)]
pub struct LoginAck {
    /// Interface type.
    pub interface: u8,
    /// TDS version.
    pub tds_version: u32,
    /// Program name.
    pub prog_name: String,
    /// Program version.
    pub prog_version: u32,
}

/// Environment change token.
#[derive(Debug, Clone)]
pub struct EnvChange {
    /// Type of environment change.
    pub env_type: EnvChangeType,
    /// New value.
    pub new_value: EnvChangeValue,
    /// Old value.
    pub old_value: EnvChangeValue,
}

/// Environment change type.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
#[non_exhaustive]
pub enum EnvChangeType {
    /// Database changed.
    Database = 1,
    /// Language changed.
    Language = 2,
    /// Character set changed.
    CharacterSet = 3,
    /// Packet size changed.
    PacketSize = 4,
    /// Unicode data sorting locale ID.
    UnicodeSortingLocalId = 5,
    /// Unicode comparison flags.
    UnicodeComparisonFlags = 6,
    /// SQL collation.
    SqlCollation = 7,
    /// Begin transaction.
    BeginTransaction = 8,
    /// Commit transaction.
    CommitTransaction = 9,
    /// Rollback transaction.
    RollbackTransaction = 10,
    /// Enlist DTC transaction.
    EnlistDtcTransaction = 11,
    /// Defect DTC transaction.
    DefectTransaction = 12,
    /// Real-time log shipping.
    RealTimeLogShipping = 13,
    /// Promote transaction.
    PromoteTransaction = 15,
    /// Transaction manager address.
    TransactionManagerAddress = 16,
    /// Transaction ended.
    TransactionEnded = 17,
    /// Reset connection completion acknowledgment.
    ResetConnectionCompletionAck = 18,
    /// User instance started.
    UserInstanceStarted = 19,
    /// Routing information.
    Routing = 20,
}

/// Environment change value.
#[derive(Debug, Clone)]
#[non_exhaustive]
pub enum EnvChangeValue {
    /// String value.
    String(String),
    /// Binary value.
    Binary(bytes::Bytes),
    /// Routing information.
    Routing {
        /// Host name.
        host: String,
        /// Port number.
        port: u16,
    },
}

/// Column ordering information.
#[derive(Debug, Clone)]
pub struct Order {
    /// Ordered column indices.
    pub columns: Vec<u16>,
}

/// Feature extension acknowledgment.
#[derive(Debug, Clone)]
pub struct FeatureExtAck {
    /// Acknowledged features.
    pub features: Vec<FeatureAck>,
}

/// Individual feature acknowledgment.
#[derive(Debug, Clone)]
pub struct FeatureAck {
    /// Feature ID.
    pub feature_id: u8,
    /// Feature data.
    pub data: bytes::Bytes,
}

/// SSPI authentication token.
#[derive(Debug, Clone)]
pub struct SspiToken {
    /// SSPI data.
    pub data: bytes::Bytes,
}

/// Session state token.
#[derive(Debug, Clone)]
pub struct SessionState {
    /// Session state data.
    pub data: bytes::Bytes,
}

/// Federated authentication info.
#[derive(Debug, Clone)]
pub struct FedAuthInfo {
    /// STS URL.
    pub sts_url: String,
    /// Service principal name.
    pub spn: String,
}

// =============================================================================
// ColMetaData and Row Parsing Implementation
// =============================================================================

/// Decode collation information (5 bytes).
///
/// Shared by ColMetaData column parsing and CryptoMetadata base type parsing.
pub(crate) fn decode_collation(src: &mut impl Buf) -> Result<Collation, ProtocolError> {
    if src.remaining() < 5 {
        return Err(ProtocolError::UnexpectedEof);
    }
    // Collation: LCID (4 bytes) + Sort ID (1 byte)
    let lcid = src.get_u32_le();
    let sort_id = src.get_u8();
    Ok(Collation { lcid, sort_id })
}

/// Decode type-specific metadata for a column based on its TypeId.
///
/// Shared by ColMetaData column parsing and CryptoMetadata base type parsing.
pub(crate) fn decode_type_info(
    src: &mut impl Buf,
    type_id: TypeId,
    col_type: u8,
) -> Result<TypeInfo, ProtocolError> {
    match type_id {
        // Fixed-length types have no additional metadata
        TypeId::Null => Ok(TypeInfo::default()),
        TypeId::Int1 | TypeId::Bit => Ok(TypeInfo::default()),
        TypeId::Int2 => Ok(TypeInfo::default()),
        TypeId::Int4 => Ok(TypeInfo::default()),
        TypeId::Int8 => Ok(TypeInfo::default()),
        TypeId::Float4 => Ok(TypeInfo::default()),
        TypeId::Float8 => Ok(TypeInfo::default()),
        TypeId::Money => Ok(TypeInfo::default()),
        TypeId::Money4 => Ok(TypeInfo::default()),
        TypeId::DateTime => Ok(TypeInfo::default()),
        TypeId::DateTime4 => Ok(TypeInfo::default()),

        // Variable length integer/float/money (1-byte max length)
        TypeId::IntN | TypeId::BitN | TypeId::FloatN | TypeId::MoneyN | TypeId::DateTimeN => {
            if src.remaining() < 1 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let max_length = src.get_u8() as u32;
            Ok(TypeInfo {
                max_length: Some(max_length),
                ..Default::default()
            })
        }

        // GUID has 1-byte length
        TypeId::Guid => {
            if src.remaining() < 1 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let max_length = src.get_u8() as u32;
            Ok(TypeInfo {
                max_length: Some(max_length),
                ..Default::default()
            })
        }

        // Decimal/Numeric types (1-byte length + precision + scale)
        TypeId::Decimal | TypeId::Numeric | TypeId::DecimalN | TypeId::NumericN => {
            if src.remaining() < 3 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let max_length = src.get_u8() as u32;
            let precision = src.get_u8();
            let scale = src.get_u8();
            Ok(TypeInfo {
                max_length: Some(max_length),
                precision: Some(precision),
                scale: Some(scale),
                ..Default::default()
            })
        }

        // Old-style byte-length strings (Char, VarChar, Binary, VarBinary)
        TypeId::Char | TypeId::VarChar | TypeId::Binary | TypeId::VarBinary => {
            if src.remaining() < 1 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let max_length = src.get_u8() as u32;
            Ok(TypeInfo {
                max_length: Some(max_length),
                ..Default::default()
            })
        }

        // Big varchar/binary with 2-byte length + collation for strings
        TypeId::BigVarChar | TypeId::BigChar => {
            if src.remaining() < 7 {
                // 2 (length) + 5 (collation)
                return Err(ProtocolError::UnexpectedEof);
            }
            let max_length = src.get_u16_le() as u32;
            let collation = decode_collation(src)?;
            Ok(TypeInfo {
                max_length: Some(max_length),
                collation: Some(collation),
                ..Default::default()
            })
        }

        // Big binary (2-byte length, no collation)
        TypeId::BigVarBinary | TypeId::BigBinary => {
            if src.remaining() < 2 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let max_length = src.get_u16_le() as u32;
            Ok(TypeInfo {
                max_length: Some(max_length),
                ..Default::default()
            })
        }

        // Unicode strings (NChar, NVarChar) - 2-byte length + collation
        TypeId::NChar | TypeId::NVarChar => {
            if src.remaining() < 7 {
                // 2 (length) + 5 (collation)
                return Err(ProtocolError::UnexpectedEof);
            }
            let max_length = src.get_u16_le() as u32;
            let collation = decode_collation(src)?;
            Ok(TypeInfo {
                max_length: Some(max_length),
                collation: Some(collation),
                ..Default::default()
            })
        }

        // Date type (no additional metadata)
        TypeId::Date => Ok(TypeInfo::default()),

        // Time, DateTime2, DateTimeOffset have scale
        TypeId::Time | TypeId::DateTime2 | TypeId::DateTimeOffset => {
            if src.remaining() < 1 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let scale = src.get_u8();
            Ok(TypeInfo {
                scale: Some(scale),
                ..Default::default()
            })
        }

        // Text/NText/Image (deprecated LOB types)
        TypeId::Text | TypeId::NText | TypeId::Image => {
            // These have complex metadata: length (4) + collation (5) + table name parts
            if src.remaining() < 4 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let max_length = src.get_u32_le();

            // For Text/NText, read collation
            let collation = if type_id == TypeId::Text || type_id == TypeId::NText {
                if src.remaining() < 5 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                Some(decode_collation(src)?)
            } else {
                None
            };

            // Skip table name parts (variable length)
            // Format: numParts (1 byte) followed by us_varchar for each part
            if src.remaining() < 1 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let num_parts = src.get_u8();
            for _ in 0..num_parts {
                // Read and discard table name part
                let _ = read_us_varchar(src).ok_or(ProtocolError::UnexpectedEof)?;
            }

            Ok(TypeInfo {
                max_length: Some(max_length),
                collation,
                ..Default::default()
            })
        }

        // XML type
        TypeId::Xml => {
            if src.remaining() < 1 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let schema_present = src.get_u8();

            if schema_present != 0 {
                // XML_INFO per MS-TDS §2.2.5.5.3: DBNAME and OWNING_SCHEMA are
                // B_VARCHAR (1-byte length); only XML_SCHEMA_COLLECTION is
                // US_VARCHAR (2-byte length).
                let _ = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?; // db name
                let _ = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?; // owning schema
                let _ = read_us_varchar(src).ok_or(ProtocolError::UnexpectedEof)?; // xml schema collection
            }

            Ok(TypeInfo::default())
        }

        // UDT (User-defined type) - complex metadata
        TypeId::Udt => {
            // Max length (2 bytes)
            if src.remaining() < 2 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let max_length = src.get_u16_le() as u32;

            // UDT_INFO per MS-TDS: DB_NAME, SCHEMA_NAME, and TYPE_NAME are
            // B_VARCHAR (1-byte length); only ASSEMBLY_QUALIFIED_NAME is
            // US_VARCHAR (2-byte length).
            let _ = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?; // db name
            let _ = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?; // schema name
            let _ = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?; // type name
            let _ = read_us_varchar(src).ok_or(ProtocolError::UnexpectedEof)?; // assembly qualified name

            Ok(TypeInfo {
                max_length: Some(max_length),
                ..Default::default()
            })
        }

        // Table-valued parameter - complex metadata (skip for now)
        TypeId::Tvp => {
            // TVP has very complex metadata, not commonly used
            // For now, we can't properly parse this
            Err(ProtocolError::InvalidTokenType(col_type))
        }

        // SQL Variant - 4-byte length
        TypeId::Variant => {
            if src.remaining() < 4 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let max_length = src.get_u32_le();
            Ok(TypeInfo {
                max_length: Some(max_length),
                ..Default::default()
            })
        }
    }
}

impl ColMetaData {
    /// Special value indicating no metadata.
    pub const NO_METADATA: u16 = 0xFFFF;

    /// Decode a COLMETADATA token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        if src.remaining() < 2 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let column_count = src.get_u16_le();

        // 0xFFFF means no metadata present
        if column_count == Self::NO_METADATA {
            return Ok(Self {
                columns: Vec::new(),
                cek_table: None,
            });
        }

        let mut columns = Vec::with_capacity(column_count as usize);

        for _ in 0..column_count {
            let column = Self::decode_column(src)?;
            columns.push(column);
        }

        Ok(Self {
            columns,
            cek_table: None,
        })
    }

    /// Decode an ALTMETADATA (COMPUTE BY) token body, returning the compute Id
    /// and the column metadata describing the matching ALTROW values.
    ///
    /// The token type byte (0x88) must already be consumed. ALTMETADATA was
    /// deprecated in TDS 7.4, but legacy `COMPUTE BY` queries still emit it. The
    /// driver parses it only to learn the ALTROW value layout so those rows can
    /// be consumed and dropped, keeping the token stream byte-aligned.
    ///
    /// Wire format (MS-TDS 2.2.7.1):
    /// `Count(USHORT) Id(USHORT) ByCols(UCHAR) ByCols*ColNum(USHORT) Count*ComputeData`,
    /// where each `ComputeData` is `Op(BYTE) Operand(USHORT)` followed by a
    /// column definition identical to COLMETADATA (TableName is never sent for
    /// COMPUTE, which excludes text/ntext/image columns).
    fn decode_alt(src: &mut impl Buf) -> Result<(u16, Self), ProtocolError> {
        // Count (USHORT) + Id (USHORT) + ByCols (UCHAR)
        if src.remaining() < 5 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let count = src.get_u16_le();
        let id = src.get_u16_le();
        let by_cols = src.get_u8() as usize;

        // ColNum: USHORT repeated ByCols times (the grouping columns). Not needed
        // to parse ALTROW values, but must be consumed to stay byte-aligned.
        if src.remaining() < by_cols * 2 {
            return Err(ProtocolError::UnexpectedEof);
        }
        for _ in 0..by_cols {
            let _ = src.get_u16_le();
        }

        let mut columns = Vec::with_capacity(count as usize);
        for _ in 0..count {
            // ComputeData prefix: Op (BYTE) + Operand (USHORT).
            if src.remaining() < 3 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let _op = src.get_u8();
            let _operand = src.get_u16_le();
            columns.push(Self::decode_column(src)?);
        }

        Ok((
            id,
            Self {
                columns,
                cek_table: None,
            },
        ))
    }

    /// Decode a single column from the metadata.
    fn decode_column(src: &mut impl Buf) -> Result<ColumnData, ProtocolError> {
        // UserType (4 bytes) + Flags (2 bytes) + TypeId (1 byte)
        if src.remaining() < 7 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let user_type = src.get_u32_le();
        let flags = src.get_u16_le();
        let col_type = src.get_u8();

        // An unknown type byte must be a hard error: treating it as Null (a
        // zero-length column) misaligns every subsequent column and row,
        // producing plausible garbage values (issue #157).
        let type_id = TypeId::from_u8(col_type).ok_or(ProtocolError::InvalidDataType(col_type))?;

        // Parse type-specific metadata
        let type_info = decode_type_info(src, type_id, col_type)?;

        // Read column name (B_VARCHAR format - 1 byte length in characters)
        let name = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?;

        Ok(ColumnData {
            name,
            type_id,
            col_type,
            flags,
            user_type,
            type_info,
            crypto_metadata: None,
        })
    }

    /// Decode a COLMETADATA token with Always Encrypted support.
    ///
    /// When column encryption was negotiated in Login7, the server sends a CekTable
    /// before column definitions and per-column CryptoMetadata for encrypted columns.
    ///
    /// # Wire Format (with encryption)
    ///
    /// ```text
    /// column_count: USHORT
    /// cek_table: CekTable (always present when encryption negotiated)
    /// columns: ColumnData[column_count] (with CryptoMetadata for encrypted columns)
    /// ```
    pub fn decode_encrypted(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        if src.remaining() < 2 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let column_count = src.get_u16_le();

        if column_count == Self::NO_METADATA {
            return Ok(Self {
                columns: Vec::new(),
                cek_table: None,
            });
        }

        // Parse CEK table (always present when encryption was negotiated)
        let cek_table = crate::crypto::CekTable::decode(src)?;

        let mut columns = Vec::with_capacity(column_count as usize);

        for _ in 0..column_count {
            let column = Self::decode_column_encrypted(src)?;
            columns.push(column);
        }

        Ok(Self {
            columns,
            cek_table: Some(cek_table),
        })
    }

    /// Decode a single column definition with Always Encrypted support.
    ///
    /// For encrypted columns (flags & 0x0800), parses CryptoMetadata after the type info.
    fn decode_column_encrypted(src: &mut impl Buf) -> Result<ColumnData, ProtocolError> {
        if src.remaining() < 7 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let user_type = src.get_u32_le();
        let flags = src.get_u16_le();
        let col_type = src.get_u8();

        let type_id = TypeId::from_u8(col_type).ok_or(ProtocolError::InvalidDataType(col_type))?;

        // Parse type-specific metadata (for encrypted columns, this is the transport type)
        let type_info = decode_type_info(src, type_id, col_type)?;

        // Parse CryptoMetadata if the column is encrypted
        let crypto_metadata = if crate::crypto::is_column_encrypted(flags) {
            Some(crate::crypto::CryptoMetadata::decode(src)?)
        } else {
            None
        };

        // Read column name
        let name = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?;

        Ok(ColumnData {
            name,
            type_id,
            col_type,
            flags,
            user_type,
            type_info,
            crypto_metadata,
        })
    }

    /// Get the number of columns.
    #[must_use]
    pub fn column_count(&self) -> usize {
        self.columns.len()
    }

    /// Check if this represents no metadata.
    #[must_use]
    pub fn is_empty(&self) -> bool {
        self.columns.is_empty()
    }
}

impl ColumnData {
    /// Check if this column is nullable.
    #[must_use]
    pub fn is_nullable(&self) -> bool {
        (self.flags & 0x0001) != 0
    }

    /// Get the fixed size in bytes for this column, if applicable.
    ///
    /// Returns `None` for variable-length types.
    #[must_use]
    pub fn fixed_size(&self) -> Option<usize> {
        match self.type_id {
            TypeId::Null => Some(0),
            TypeId::Int1 | TypeId::Bit => Some(1),
            TypeId::Int2 => Some(2),
            TypeId::Int4 => Some(4),
            TypeId::Int8 => Some(8),
            TypeId::Float4 => Some(4),
            TypeId::Float8 => Some(8),
            TypeId::Money => Some(8),
            TypeId::Money4 => Some(4),
            TypeId::DateTime => Some(8),
            TypeId::DateTime4 => Some(4),
            TypeId::Date => Some(3),
            _ => None,
        }
    }
}

// =============================================================================
// Row Parsing Implementation
// =============================================================================

impl RawRow {
    /// Decode a ROW token from bytes.
    ///
    /// This function requires the column metadata to know how to parse the row.
    /// The row data is stored as raw bytes for later parsing.
    pub fn decode(src: &mut impl Buf, metadata: &ColMetaData) -> Result<Self, ProtocolError> {
        let mut data = bytes::BytesMut::new();

        for col in &metadata.columns {
            Self::decode_column_value(src, col, &mut data)?;
        }

        Ok(Self {
            data: data.freeze(),
        })
    }

    /// Decode only the first `prefix_len` columns of a ROW token, leaving `src`
    /// positioned at the start of column `prefix_len`.
    ///
    /// Used by the BLOB streaming path to decode the leading scalar columns of a
    /// row and stop at a trailing MAX column, whose PLP value is then streamed
    /// directly from the socket rather than buffered.
    pub fn decode_prefix(
        src: &mut impl Buf,
        metadata: &ColMetaData,
        prefix_len: usize,
    ) -> Result<Self, ProtocolError> {
        let mut data = bytes::BytesMut::new();
        for col in metadata.columns.iter().take(prefix_len) {
            Self::decode_column_value(src, col, &mut data)?;
        }
        Ok(Self {
            data: data.freeze(),
        })
    }

    /// Decode a single column value and append to the output buffer.
    fn decode_column_value(
        src: &mut impl Buf,
        col: &ColumnData,
        dst: &mut bytes::BytesMut,
    ) -> Result<(), ProtocolError> {
        match col.type_id {
            // Fixed-length types
            TypeId::Null => {
                // No data
            }
            TypeId::Int1 | TypeId::Bit => {
                if src.remaining() < 1 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                dst.extend_from_slice(&[src.get_u8()]);
            }
            TypeId::Int2 => {
                if src.remaining() < 2 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                dst.extend_from_slice(&src.get_u16_le().to_le_bytes());
            }
            TypeId::Int4 => {
                if src.remaining() < 4 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                dst.extend_from_slice(&src.get_u32_le().to_le_bytes());
            }
            TypeId::Int8 => {
                if src.remaining() < 8 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                dst.extend_from_slice(&src.get_u64_le().to_le_bytes());
            }
            TypeId::Float4 => {
                if src.remaining() < 4 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                dst.extend_from_slice(&src.get_u32_le().to_le_bytes());
            }
            TypeId::Float8 => {
                if src.remaining() < 8 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                dst.extend_from_slice(&src.get_u64_le().to_le_bytes());
            }
            TypeId::Money => {
                if src.remaining() < 8 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                let hi = src.get_u32_le();
                let lo = src.get_u32_le();
                dst.extend_from_slice(&hi.to_le_bytes());
                dst.extend_from_slice(&lo.to_le_bytes());
            }
            TypeId::Money4 => {
                if src.remaining() < 4 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                dst.extend_from_slice(&src.get_u32_le().to_le_bytes());
            }
            TypeId::DateTime => {
                if src.remaining() < 8 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                let days = src.get_u32_le();
                let time = src.get_u32_le();
                dst.extend_from_slice(&days.to_le_bytes());
                dst.extend_from_slice(&time.to_le_bytes());
            }
            TypeId::DateTime4 => {
                if src.remaining() < 4 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                dst.extend_from_slice(&src.get_u32_le().to_le_bytes());
            }
            // DATE type uses 1-byte length prefix (can be NULL)
            TypeId::Date => {
                Self::decode_bytelen_type(src, dst)?;
            }

            // Variable-length nullable types (length-prefixed)
            TypeId::IntN | TypeId::BitN | TypeId::FloatN | TypeId::MoneyN | TypeId::DateTimeN => {
                Self::decode_bytelen_type(src, dst)?;
            }

            TypeId::Guid => {
                Self::decode_bytelen_type(src, dst)?;
            }

            TypeId::Decimal | TypeId::Numeric | TypeId::DecimalN | TypeId::NumericN => {
                Self::decode_bytelen_type(src, dst)?;
            }

            // Old-style byte-length strings
            TypeId::Char | TypeId::VarChar | TypeId::Binary | TypeId::VarBinary => {
                Self::decode_bytelen_type(src, dst)?;
            }

            // 2-byte length strings (or PLP for MAX types)
            TypeId::BigVarChar | TypeId::BigVarBinary => {
                // max_length == 0xFFFF indicates VARCHAR(MAX) or VARBINARY(MAX), which uses PLP
                if col.type_info.max_length == Some(0xFFFF) {
                    Self::decode_plp_type(src, dst)?;
                } else {
                    Self::decode_ushortlen_type(src, dst)?;
                }
            }

            // Fixed-length types that don't have MAX variants
            TypeId::BigChar | TypeId::BigBinary => {
                Self::decode_ushortlen_type(src, dst)?;
            }

            // Unicode strings (2-byte length in bytes, or PLP for NVARCHAR(MAX))
            TypeId::NVarChar => {
                // max_length == 0xFFFF indicates NVARCHAR(MAX), which uses PLP
                if col.type_info.max_length == Some(0xFFFF) {
                    Self::decode_plp_type(src, dst)?;
                } else {
                    Self::decode_ushortlen_type(src, dst)?;
                }
            }

            // Fixed-length NCHAR doesn't have MAX variant
            TypeId::NChar => {
                Self::decode_ushortlen_type(src, dst)?;
            }

            // Time types with scale
            TypeId::Time | TypeId::DateTime2 | TypeId::DateTimeOffset => {
                Self::decode_bytelen_type(src, dst)?;
            }

            // TEXT/NTEXT/IMAGE - deprecated LOB types using textptr format
            TypeId::Text | TypeId::NText | TypeId::Image => {
                Self::decode_textptr_type(src, dst)?;
            }

            // XML - uses actual PLP format
            TypeId::Xml => {
                Self::decode_plp_type(src, dst)?;
            }

            // Complex types
            TypeId::Variant => {
                Self::decode_intlen_type(src, dst)?;
            }

            TypeId::Udt => {
                // UDT uses PLP encoding
                Self::decode_plp_type(src, dst)?;
            }

            TypeId::Tvp => {
                // TVP not supported in row data
                return Err(ProtocolError::InvalidTokenType(col.col_type));
            }
        }

        Ok(())
    }

    /// Decode a 1-byte length-prefixed value.
    fn decode_bytelen_type(
        src: &mut impl Buf,
        dst: &mut bytes::BytesMut,
    ) -> Result<(), ProtocolError> {
        if src.remaining() < 1 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let len = src.get_u8() as usize;
        if len == 0xFF {
            // NULL value - store as zero-length with NULL marker
            dst.extend_from_slice(&[0xFF]);
        } else if len == 0 {
            // Empty value
            dst.extend_from_slice(&[0x00]);
        } else {
            if src.remaining() < len {
                return Err(ProtocolError::UnexpectedEof);
            }
            dst.extend_from_slice(&[len as u8]);
            for _ in 0..len {
                dst.extend_from_slice(&[src.get_u8()]);
            }
        }
        Ok(())
    }

    /// Decode a 2-byte length-prefixed value.
    fn decode_ushortlen_type(
        src: &mut impl Buf,
        dst: &mut bytes::BytesMut,
    ) -> Result<(), ProtocolError> {
        if src.remaining() < 2 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let len = src.get_u16_le() as usize;
        if len == 0xFFFF {
            // NULL value
            dst.extend_from_slice(&0xFFFFu16.to_le_bytes());
        } else if len == 0 {
            // Empty value
            dst.extend_from_slice(&0u16.to_le_bytes());
        } else {
            if src.remaining() < len {
                return Err(ProtocolError::UnexpectedEof);
            }
            dst.extend_from_slice(&(len as u16).to_le_bytes());
            for _ in 0..len {
                dst.extend_from_slice(&[src.get_u8()]);
            }
        }
        Ok(())
    }

    /// Decode a 4-byte length-prefixed value.
    fn decode_intlen_type(
        src: &mut impl Buf,
        dst: &mut bytes::BytesMut,
    ) -> Result<(), ProtocolError> {
        if src.remaining() < 4 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let len = src.get_u32_le() as usize;
        if len == 0xFFFFFFFF {
            // NULL value
            dst.extend_from_slice(&0xFFFFFFFFu32.to_le_bytes());
        } else if len == 0 {
            // Empty value
            dst.extend_from_slice(&0u32.to_le_bytes());
        } else {
            if src.remaining() < len {
                return Err(ProtocolError::UnexpectedEof);
            }
            dst.extend_from_slice(&(len as u32).to_le_bytes());
            for _ in 0..len {
                dst.extend_from_slice(&[src.get_u8()]);
            }
        }
        Ok(())
    }

    /// Decode a TEXT/NTEXT/IMAGE type (textptr format).
    ///
    /// These deprecated LOB types use a special format:
    /// - 1 byte: textptr_len (0 = NULL)
    /// - textptr_len bytes: textptr (if not NULL)
    /// - 8 bytes: timestamp (if not NULL)
    /// - 4 bytes: data length (if not NULL)
    /// - data_len bytes: the actual data (if not NULL)
    ///
    /// We convert this to PLP format for the client to parse:
    /// - 8 bytes: total length (0xFFFFFFFFFFFFFFFF = NULL)
    /// - 4 bytes: chunk length (= data length)
    /// - chunk data
    /// - 4 bytes: 0 (terminator)
    fn decode_textptr_type(
        src: &mut impl Buf,
        dst: &mut bytes::BytesMut,
    ) -> Result<(), ProtocolError> {
        if src.remaining() < 1 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let textptr_len = src.get_u8() as usize;

        if textptr_len == 0 {
            // NULL value - write PLP NULL marker
            dst.extend_from_slice(&0xFFFFFFFFFFFFFFFFu64.to_le_bytes());
            return Ok(());
        }

        // Skip textptr bytes
        if src.remaining() < textptr_len {
            return Err(ProtocolError::UnexpectedEof);
        }
        src.advance(textptr_len);

        // Skip 8-byte timestamp
        if src.remaining() < 8 {
            return Err(ProtocolError::UnexpectedEof);
        }
        src.advance(8);

        // Read data length
        if src.remaining() < 4 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let data_len = src.get_u32_le() as usize;

        if src.remaining() < data_len {
            return Err(ProtocolError::UnexpectedEof);
        }

        // Write in PLP format for client parsing:
        // - 8 bytes: total length
        // - 4 bytes: chunk length
        // - chunk data
        // - 4 bytes: 0 (terminator)
        dst.extend_from_slice(&(data_len as u64).to_le_bytes());
        dst.extend_from_slice(&(data_len as u32).to_le_bytes());
        for _ in 0..data_len {
            dst.extend_from_slice(&[src.get_u8()]);
        }
        dst.extend_from_slice(&0u32.to_le_bytes()); // PLP terminator

        Ok(())
    }

    /// Decode a PLP (Partially Length-Prefixed) value.
    ///
    /// PLP format:
    /// - 8 bytes: total length (0xFFFFFFFFFFFFFFFE = unknown, 0xFFFFFFFFFFFFFFFF = NULL)
    /// - If not NULL: chunks of (4 byte chunk length + data) until chunk length = 0
    fn decode_plp_type(src: &mut impl Buf, dst: &mut bytes::BytesMut) -> Result<(), ProtocolError> {
        if src.remaining() < 8 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let total_len = src.get_u64_le();

        // Store the total length marker
        dst.extend_from_slice(&total_len.to_le_bytes());

        if total_len == 0xFFFFFFFFFFFFFFFF {
            // NULL value - no more data
            return Ok(());
        }

        // Read chunks until terminator
        loop {
            if src.remaining() < 4 {
                return Err(ProtocolError::UnexpectedEof);
            }
            let chunk_len = src.get_u32_le() as usize;
            dst.extend_from_slice(&(chunk_len as u32).to_le_bytes());

            if chunk_len == 0 {
                // End of PLP data
                break;
            }

            if src.remaining() < chunk_len {
                return Err(ProtocolError::UnexpectedEof);
            }

            for _ in 0..chunk_len {
                dst.extend_from_slice(&[src.get_u8()]);
            }
        }

        Ok(())
    }
}

// =============================================================================
// NbcRow Parsing Implementation
// =============================================================================

impl NbcRow {
    /// Decode an NBCROW token from bytes.
    ///
    /// NBCROW (Null Bitmap Compressed Row) stores a bitmap indicating which
    /// columns are NULL, followed by only the non-NULL values.
    pub fn decode(src: &mut impl Buf, metadata: &ColMetaData) -> Result<Self, ProtocolError> {
        let col_count = metadata.columns.len();
        let bitmap_len = col_count.div_ceil(8);

        if src.remaining() < bitmap_len {
            return Err(ProtocolError::UnexpectedEof);
        }

        // Read null bitmap
        let mut null_bitmap = vec![0u8; bitmap_len];
        for byte in &mut null_bitmap {
            *byte = src.get_u8();
        }

        // Read non-null values
        let mut data = bytes::BytesMut::new();

        for (i, col) in metadata.columns.iter().enumerate() {
            let byte_idx = i / 8;
            let bit_idx = i % 8;
            let is_null = (null_bitmap[byte_idx] & (1 << bit_idx)) != 0;

            if !is_null {
                // Read the value - for NBCROW, we read without the length prefix
                // for fixed-length types, and with length prefix for variable types
                RawRow::decode_column_value(src, col, &mut data)?;
            }
        }

        Ok(Self {
            null_bitmap,
            data: data.freeze(),
        })
    }

    /// Decode the null bitmap and the first `prefix_len` columns of an NBCROW,
    /// leaving `src` positioned at column `prefix_len`'s value (when that column
    /// is non-NULL per the bitmap). The returned row carries the full bitmap and
    /// the leading non-NULL values; query the trailing column's nullness with
    /// [`is_null`](Self::is_null).
    pub fn decode_prefix(
        src: &mut impl Buf,
        metadata: &ColMetaData,
        prefix_len: usize,
    ) -> Result<Self, ProtocolError> {
        let col_count = metadata.columns.len();
        let bitmap_len = col_count.div_ceil(8);

        if src.remaining() < bitmap_len {
            return Err(ProtocolError::UnexpectedEof);
        }

        let mut null_bitmap = vec![0u8; bitmap_len];
        for byte in &mut null_bitmap {
            *byte = src.get_u8();
        }

        let mut data = bytes::BytesMut::new();
        for (i, col) in metadata.columns.iter().enumerate().take(prefix_len) {
            let is_null = (null_bitmap[i / 8] & (1 << (i % 8))) != 0;
            if !is_null {
                RawRow::decode_column_value(src, col, &mut data)?;
            }
        }

        Ok(Self {
            null_bitmap,
            data: data.freeze(),
        })
    }

    /// Check if a column at the given index is NULL.
    #[must_use]
    pub fn is_null(&self, column_index: usize) -> bool {
        let byte_idx = column_index / 8;
        let bit_idx = column_index % 8;
        if byte_idx < self.null_bitmap.len() {
            (self.null_bitmap[byte_idx] & (1 << bit_idx)) != 0
        } else {
            true // Out of bounds = NULL
        }
    }
}

// =============================================================================
// ReturnValue Parsing Implementation
// =============================================================================

impl ReturnValue {
    /// Decode a RETURNVALUE token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        // MS-TDS §2.2.7.18: the RETURNVALUE token has no length prefix —
        // it begins directly with the 2-byte ParamOrdinal. The previous
        // spurious 2-byte read consumed the ordinal and shifted every
        // subsequent field, leaving the stream parser two bytes ahead and
        // reading value bytes as the next token type (e.g. `0x74` from a
        // Unicode name fragment was misread as an unknown token).
        if src.remaining() < 2 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let param_ordinal = src.get_u16_le();

        // Parameter name (B_VARCHAR)
        let param_name = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?;

        // Status (1 byte)
        if src.remaining() < 1 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let status = src.get_u8();

        // User type (4 bytes) + flags (2 bytes) + type id (1 byte)
        if src.remaining() < 7 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let user_type = src.get_u32_le();
        let flags = src.get_u16_le();
        let col_type = src.get_u8();

        let type_id = TypeId::from_u8(col_type).ok_or(ProtocolError::InvalidDataType(col_type))?;

        // Parse type info
        let type_info = decode_type_info(src, type_id, col_type)?;

        // Read the value data
        let mut value_buf = bytes::BytesMut::new();

        // Create a temporary column for value parsing
        let temp_col = ColumnData {
            name: String::new(),
            type_id,
            col_type,
            flags,
            user_type,
            type_info: type_info.clone(),
            crypto_metadata: None,
        };

        RawRow::decode_column_value(src, &temp_col, &mut value_buf)?;

        Ok(Self {
            param_ordinal,
            param_name,
            status,
            user_type,
            flags,
            col_type,
            type_info,
            value: value_buf.freeze(),
        })
    }
}

// =============================================================================
// SessionState Parsing Implementation
// =============================================================================

impl SessionState {
    /// Decode a SESSIONSTATE token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        if src.remaining() < 4 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let length = src.get_u32_le() as usize;

        if src.remaining() < length {
            return Err(ProtocolError::IncompletePacket {
                expected: length,
                actual: src.remaining(),
            });
        }

        let data = src.copy_to_bytes(length);

        Ok(Self { data })
    }
}

// =============================================================================
// Token Parsing Implementation
// =============================================================================

/// Done token status flags bit positions.
mod done_status_bits {
    pub const DONE_MORE: u16 = 0x0001;
    pub const DONE_ERROR: u16 = 0x0002;
    pub const DONE_INXACT: u16 = 0x0004;
    pub const DONE_COUNT: u16 = 0x0010;
    pub const DONE_ATTN: u16 = 0x0020;
    pub const DONE_SRVERROR: u16 = 0x0100;
}

impl DoneStatus {
    /// Parse done status from raw bits.
    #[must_use]
    pub fn from_bits(bits: u16) -> Self {
        use done_status_bits::*;
        Self {
            more: (bits & DONE_MORE) != 0,
            error: (bits & DONE_ERROR) != 0,
            in_xact: (bits & DONE_INXACT) != 0,
            count: (bits & DONE_COUNT) != 0,
            attn: (bits & DONE_ATTN) != 0,
            srverror: (bits & DONE_SRVERROR) != 0,
        }
    }

    /// Convert to raw bits.
    #[must_use]
    pub fn to_bits(&self) -> u16 {
        use done_status_bits::*;
        let mut bits = 0u16;
        if self.more {
            bits |= DONE_MORE;
        }
        if self.error {
            bits |= DONE_ERROR;
        }
        if self.in_xact {
            bits |= DONE_INXACT;
        }
        if self.count {
            bits |= DONE_COUNT;
        }
        if self.attn {
            bits |= DONE_ATTN;
        }
        if self.srverror {
            bits |= DONE_SRVERROR;
        }
        bits
    }
}

impl Done {
    /// Size of the DONE token in bytes (excluding token type byte).
    pub const SIZE: usize = 12; // 2 (status) + 2 (curcmd) + 8 (rowcount)

    /// Decode a DONE token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        if src.remaining() < Self::SIZE {
            return Err(ProtocolError::IncompletePacket {
                expected: Self::SIZE,
                actual: src.remaining(),
            });
        }

        let status = DoneStatus::from_bits(src.get_u16_le());
        let cur_cmd = src.get_u16_le();
        let row_count = src.get_u64_le();

        Ok(Self {
            status,
            cur_cmd,
            row_count,
        })
    }

    /// Encode the DONE token to bytes.
    pub fn encode(&self, dst: &mut impl BufMut) {
        dst.put_u8(TokenType::Done as u8);
        dst.put_u16_le(self.status.to_bits());
        dst.put_u16_le(self.cur_cmd);
        dst.put_u64_le(self.row_count);
    }

    /// Check if more results follow this DONE token.
    #[must_use]
    pub const fn has_more(&self) -> bool {
        self.status.more
    }

    /// Check if an error occurred.
    #[must_use]
    pub const fn has_error(&self) -> bool {
        self.status.error
    }

    /// Check if the row count is valid.
    #[must_use]
    pub const fn has_count(&self) -> bool {
        self.status.count
    }
}

impl DoneProc {
    /// Size of the DONEPROC token in bytes (excluding token type byte).
    pub const SIZE: usize = 12;

    /// Decode a DONEPROC token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        if src.remaining() < Self::SIZE {
            return Err(ProtocolError::IncompletePacket {
                expected: Self::SIZE,
                actual: src.remaining(),
            });
        }

        let status = DoneStatus::from_bits(src.get_u16_le());
        let cur_cmd = src.get_u16_le();
        let row_count = src.get_u64_le();

        Ok(Self {
            status,
            cur_cmd,
            row_count,
        })
    }

    /// Encode the DONEPROC token to bytes.
    pub fn encode(&self, dst: &mut impl BufMut) {
        dst.put_u8(TokenType::DoneProc as u8);
        dst.put_u16_le(self.status.to_bits());
        dst.put_u16_le(self.cur_cmd);
        dst.put_u64_le(self.row_count);
    }
}

impl DoneInProc {
    /// Size of the DONEINPROC token in bytes (excluding token type byte).
    pub const SIZE: usize = 12;

    /// Decode a DONEINPROC token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        if src.remaining() < Self::SIZE {
            return Err(ProtocolError::IncompletePacket {
                expected: Self::SIZE,
                actual: src.remaining(),
            });
        }

        let status = DoneStatus::from_bits(src.get_u16_le());
        let cur_cmd = src.get_u16_le();
        let row_count = src.get_u64_le();

        Ok(Self {
            status,
            cur_cmd,
            row_count,
        })
    }

    /// Encode the DONEINPROC token to bytes.
    pub fn encode(&self, dst: &mut impl BufMut) {
        dst.put_u8(TokenType::DoneInProc as u8);
        dst.put_u16_le(self.status.to_bits());
        dst.put_u16_le(self.cur_cmd);
        dst.put_u64_le(self.row_count);
    }
}

impl ServerError {
    /// Decode an ERROR token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        // ERROR token: length (2) + number (4) + state (1) + class (1) +
        //              message (us_varchar) + server (b_varchar) + procedure (b_varchar) + line (4)
        if src.remaining() < 2 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let _length = src.get_u16_le();

        if src.remaining() < 6 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let number = src.get_i32_le();
        let state = src.get_u8();
        let class = src.get_u8();

        let message = read_us_varchar(src).ok_or(ProtocolError::UnexpectedEof)?;
        let server = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?;
        let procedure = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?;

        if src.remaining() < 4 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let line = src.get_i32_le();

        Ok(Self {
            number,
            state,
            class,
            message,
            server,
            procedure,
            line,
        })
    }

    /// Check if this is a fatal error (severity >= 20).
    #[must_use]
    pub const fn is_fatal(&self) -> bool {
        self.class >= 20
    }

    /// Check if this error indicates the batch was aborted (severity >= 16).
    #[must_use]
    pub const fn is_batch_abort(&self) -> bool {
        self.class >= 16
    }
}

impl ServerInfo {
    /// Decode an INFO token from bytes.
    ///
    /// INFO tokens have the same structure as ERROR tokens but with lower severity.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        if src.remaining() < 2 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let _length = src.get_u16_le();

        if src.remaining() < 6 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let number = src.get_i32_le();
        let state = src.get_u8();
        let class = src.get_u8();

        let message = read_us_varchar(src).ok_or(ProtocolError::UnexpectedEof)?;
        let server = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?;
        let procedure = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?;

        if src.remaining() < 4 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let line = src.get_i32_le();

        Ok(Self {
            number,
            state,
            class,
            message,
            server,
            procedure,
            line,
        })
    }
}

impl LoginAck {
    /// Decode a LOGINACK token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        // LOGINACK: length (2) + interface (1) + tds_version (4) + prog_name (b_varchar) + prog_version (4)
        if src.remaining() < 2 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let _length = src.get_u16_le();

        if src.remaining() < 5 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let interface = src.get_u8();
        let tds_version = src.get_u32_le();
        let prog_name = read_b_varchar(src).ok_or(ProtocolError::UnexpectedEof)?;

        if src.remaining() < 4 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let prog_version = src.get_u32_le();

        Ok(Self {
            interface,
            tds_version,
            prog_name,
            prog_version,
        })
    }

    /// Get the TDS version as a `TdsVersion`.
    #[must_use]
    pub fn tds_version(&self) -> crate::version::TdsVersion {
        crate::version::TdsVersion::new(self.tds_version)
    }
}

impl EnvChangeType {
    /// Create from raw byte value.
    pub fn from_u8(value: u8) -> Option<Self> {
        match value {
            1 => Some(Self::Database),
            2 => Some(Self::Language),
            3 => Some(Self::CharacterSet),
            4 => Some(Self::PacketSize),
            5 => Some(Self::UnicodeSortingLocalId),
            6 => Some(Self::UnicodeComparisonFlags),
            7 => Some(Self::SqlCollation),
            8 => Some(Self::BeginTransaction),
            9 => Some(Self::CommitTransaction),
            10 => Some(Self::RollbackTransaction),
            11 => Some(Self::EnlistDtcTransaction),
            12 => Some(Self::DefectTransaction),
            13 => Some(Self::RealTimeLogShipping),
            15 => Some(Self::PromoteTransaction),
            16 => Some(Self::TransactionManagerAddress),
            17 => Some(Self::TransactionEnded),
            18 => Some(Self::ResetConnectionCompletionAck),
            19 => Some(Self::UserInstanceStarted),
            20 => Some(Self::Routing),
            _ => None,
        }
    }
}

impl EnvChange {
    /// Decode an ENVCHANGE token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        if src.remaining() < 3 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let length = src.get_u16_le() as usize;
        if length == 0 {
            // The frame must at least contain the type byte; reading it from
            // outside a zero-length frame would consume the next token.
            return Err(ProtocolError::UnexpectedEof);
        }
        if src.remaining() < length {
            return Err(ProtocolError::IncompletePacket {
                expected: length,
                actual: src.remaining(),
            });
        }

        // Frame-strict decoding (issue #145): the value decoders below only
        // bounds-check against the *buffer*, so on an under-declared frame
        // they could read past the declared length into the next token's
        // bytes. Slice exactly the declared frame and decode from that:
        // over-read attempts now hit frame end and take the lenient
        // empty-value fallbacks (preserving the #140 hostile-input
        // behavior), and the outer buffer always advances by exactly
        // `length`.
        let mut frame = src.copy_to_bytes(length);
        let src = &mut frame;

        let env_type_byte = src.get_u8();
        let env_type = EnvChangeType::from_u8(env_type_byte)
            .ok_or(ProtocolError::InvalidTokenType(env_type_byte))?;

        let (new_value, old_value) = match env_type {
            EnvChangeType::Routing => {
                // Routing has special format
                let new_value = Self::decode_routing_value(src)?;
                let old_value = EnvChangeValue::Binary(Bytes::new());
                (new_value, old_value)
            }
            EnvChangeType::BeginTransaction
            | EnvChangeType::CommitTransaction
            | EnvChangeType::RollbackTransaction
            | EnvChangeType::EnlistDtcTransaction
            | EnvChangeType::SqlCollation => {
                // These use binary format per MS-TDS spec:
                // - Transaction tokens: transaction descriptor (8 bytes)
                // - SqlCollation: collation info (5 bytes: LCID + sort flags)
                // The declared ENVCHANGE `length` can be shorter than this
                // branch needs (e.g. covers only the type byte), so the
                // length-prefix reads must be bounds-checked individually:
                // `get_u8` on an empty buffer panics. Match the branch's
                // existing graceful style — a missing prefix means empty.
                let new_len = if src.has_remaining() {
                    src.get_u8() as usize
                } else {
                    0
                };
                let new_value = if new_len > 0 && src.remaining() >= new_len {
                    EnvChangeValue::Binary(src.copy_to_bytes(new_len))
                } else {
                    EnvChangeValue::Binary(Bytes::new())
                };

                let old_len = if src.has_remaining() {
                    src.get_u8() as usize
                } else {
                    0
                };
                let old_value = if old_len > 0 && src.remaining() >= old_len {
                    EnvChangeValue::Binary(src.copy_to_bytes(old_len))
                } else {
                    EnvChangeValue::Binary(Bytes::new())
                };

                (new_value, old_value)
            }
            _ => {
                // String format for most env changes
                let new_value = read_b_varchar(src)
                    .map(EnvChangeValue::String)
                    .unwrap_or(EnvChangeValue::String(String::new()));

                let old_value = read_b_varchar(src)
                    .map(EnvChangeValue::String)
                    .unwrap_or(EnvChangeValue::String(String::new()));

                (new_value, old_value)
            }
        };

        // No frame-boundary fixup needed: the whole declared frame was
        // consumed from the outer buffer up front, so decoders that
        // under-consume (e.g. Routing's implicit zero-length OldValue) just
        // leave bytes behind in the dropped sub-frame.

        Ok(Self {
            env_type,
            new_value,
            old_value,
        })
    }

    fn decode_routing_value(src: &mut impl Buf) -> Result<EnvChangeValue, ProtocolError> {
        // Routing format: length (2) + protocol (1) + port (2) + server_len (2) + server (utf16)
        if src.remaining() < 2 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let _routing_len = src.get_u16_le();

        if src.remaining() < 5 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let _protocol = src.get_u8();
        let port = src.get_u16_le();
        let server_len = src.get_u16_le() as usize;

        // Read UTF-16LE server name
        if src.remaining() < server_len * 2 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let mut chars = Vec::with_capacity(server_len);
        for _ in 0..server_len {
            chars.push(src.get_u16_le());
        }

        let host = String::from_utf16(&chars).map_err(|_| {
            ProtocolError::StringEncoding(
                #[cfg(feature = "std")]
                "invalid UTF-16 in routing hostname".to_string(),
                #[cfg(not(feature = "std"))]
                "invalid UTF-16 in routing hostname",
            )
        })?;

        Ok(EnvChangeValue::Routing { host, port })
    }

    /// Check if this is a routing redirect.
    #[must_use]
    pub fn is_routing(&self) -> bool {
        self.env_type == EnvChangeType::Routing
    }

    /// Get routing information if this is a routing change.
    #[must_use]
    pub fn routing_info(&self) -> Option<(&str, u16)> {
        if let EnvChangeValue::Routing { host, port } = &self.new_value {
            Some((host, *port))
        } else {
            None
        }
    }

    /// Get the new database name if this is a database change.
    #[must_use]
    pub fn new_database(&self) -> Option<&str> {
        if self.env_type == EnvChangeType::Database {
            if let EnvChangeValue::String(s) = &self.new_value {
                return Some(s);
            }
        }
        None
    }
}

impl Order {
    /// Decode an ORDER token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        if src.remaining() < 2 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let length = src.get_u16_le() as usize;
        let column_count = length / 2;

        if src.remaining() < length {
            return Err(ProtocolError::IncompletePacket {
                expected: length,
                actual: src.remaining(),
            });
        }

        let mut columns = Vec::with_capacity(column_count);
        for _ in 0..column_count {
            columns.push(src.get_u16_le());
        }

        Ok(Self { columns })
    }
}

impl FeatureExtAck {
    /// Feature terminator byte.
    pub const TERMINATOR: u8 = 0xFF;

    /// Decode a FEATUREEXTACK token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        let mut features = Vec::new();

        loop {
            if !src.has_remaining() {
                return Err(ProtocolError::UnexpectedEof);
            }

            let feature_id = src.get_u8();
            if feature_id == Self::TERMINATOR {
                break;
            }

            if src.remaining() < 4 {
                return Err(ProtocolError::UnexpectedEof);
            }

            let data_len = src.get_u32_le() as usize;

            if src.remaining() < data_len {
                return Err(ProtocolError::IncompletePacket {
                    expected: data_len,
                    actual: src.remaining(),
                });
            }

            let data = src.copy_to_bytes(data_len);
            features.push(FeatureAck { feature_id, data });
        }

        Ok(Self { features })
    }
}

impl SspiToken {
    /// Decode an SSPI token from bytes.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        if src.remaining() < 2 {
            return Err(ProtocolError::UnexpectedEof);
        }

        let length = src.get_u16_le() as usize;

        if src.remaining() < length {
            return Err(ProtocolError::IncompletePacket {
                expected: length,
                actual: src.remaining(),
            });
        }

        let data = src.copy_to_bytes(length);
        Ok(Self { data })
    }
}

impl FedAuthInfo {
    /// `FedAuthInfoID` for the STS URL (MS-TDS §2.2.7.12: %0x01 = STSURL).
    const ID_STSURL: u8 = 0x01;
    /// `FedAuthInfoID` for the service principal name (MS-TDS §2.2.7.12:
    /// %0x02 = SPN).
    const ID_SPN: u8 = 0x02;
    /// Size of one `FedAuthInfoOpt` header: ID (1) + DataLen (4) + DataOffset (4).
    const OPT_HEADER_LEN: usize = 9;

    /// Decode a FEDAUTHINFO token from bytes.
    ///
    /// Wire layout per MS-TDS §2.2.7.12 (after the 0xEE token byte):
    /// `TokenLength` (DWORD) covering everything that follows, then
    /// `CountOfInfoIDs` (DWORD), then `CountOfInfoIDs` option headers of
    /// ID (BYTE) + `FedAuthInfoDataLen` (DWORD) + `FedAuthInfoDataOffset`
    /// (DWORD), then the data block. Offsets are relative to the start of
    /// the `CountOfInfoIDs` field, and the option data is UTF-16LE.
    ///
    /// Exactly `TokenLength` bytes are consumed, so tokens that follow
    /// FEDAUTHINFO in the login stream (LOGINACK, DONE) are preserved.
    pub fn decode(src: &mut impl Buf) -> Result<Self, ProtocolError> {
        if src.remaining() < 4 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let token_len = src.get_u32_le() as usize;
        if src.remaining() < token_len {
            return Err(ProtocolError::UnexpectedEof);
        }

        // Offsets in the option headers are relative to the start of this
        // region (the CountOfInfoIDs field), so address into it directly.
        let region = src.copy_to_bytes(token_len);
        if region.len() < 4 {
            return Err(ProtocolError::UnexpectedEof);
        }
        let count = u32::from_le_bytes([region[0], region[1], region[2], region[3]]) as usize;

        // All headers must fit between the count field and the end of the
        // token. The checked math also rejects hostile counts that would
        // overflow the offset arithmetic.
        let headers_end = count
            .checked_mul(Self::OPT_HEADER_LEN)
            .and_then(|n| n.checked_add(4))
            .ok_or(ProtocolError::UnexpectedEof)?;
        if headers_end > region.len() {
            return Err(ProtocolError::UnexpectedEof);
        }

        let mut sts_url = String::new();
        let mut spn = String::new();

        for i in 0..count {
            let h = 4 + i * Self::OPT_HEADER_LEN;
            let info_id = region[h];
            let data_len =
                u32::from_le_bytes([region[h + 1], region[h + 2], region[h + 3], region[h + 4]])
                    as usize;
            let data_off =
                u32::from_le_bytes([region[h + 5], region[h + 6], region[h + 7], region[h + 8]])
                    as usize;

            // Unknown IDs are skipped without validating their data, per the
            // spec's instruction to ignore unrecognized options.
            if info_id != Self::ID_SPN && info_id != Self::ID_STSURL {
                continue;
            }

            let data_end = data_off
                .checked_add(data_len)
                .ok_or(ProtocolError::UnexpectedEof)?;
            if data_end > region.len() {
                return Err(ProtocolError::UnexpectedEof);
            }
            if data_len % 2 != 0 {
                return Err(ProtocolError::StringEncoding(
                    #[cfg(feature = "std")]
                    "FEDAUTHINFO option data has odd length, not UTF-16".to_string(),
                    #[cfg(not(feature = "std"))]
                    "FEDAUTHINFO option data has odd length, not UTF-16",
                ));
            }

            let chars: Vec<u16> = region[data_off..data_end]
                .chunks_exact(2)
                .map(|b| u16::from_le_bytes([b[0], b[1]]))
                .collect();
            let value = String::from_utf16(&chars).map_err(|_| {
                ProtocolError::StringEncoding(
                    #[cfg(feature = "std")]
                    "invalid UTF-16 in FEDAUTHINFO option".to_string(),
                    #[cfg(not(feature = "std"))]
                    "invalid UTF-16 in FEDAUTHINFO option",
                )
            })?;

            if info_id == Self::ID_SPN {
                spn = value;
            } else {
                sts_url = value;
            }
        }

        Ok(Self { sts_url, spn })
    }
}

// =============================================================================
// Token Parser
// =============================================================================

/// Token stream parser.
///
/// Parses a stream of TDS tokens from a byte buffer.
///
/// # Basic vs Context-Aware Parsing
///
/// Some tokens (like `Done`, `Error`, `LoginAck`) can be parsed without context.
/// Use [`next_token()`](TokenParser::next_token) for these.
///
/// Other tokens (like `ColMetaData`, `Row`, `NbcRow`) require column metadata
/// to parse correctly. Use [`next_token_with_metadata()`](TokenParser::next_token_with_metadata)
/// for these.
///
/// # Example
///
/// ```rust,ignore
/// let mut parser = TokenParser::new(data);
/// let mut metadata = None;
///
/// while let Some(token) = parser.next_token_with_metadata(metadata.as_ref())? {
///     match token {
///         Token::ColMetaData(meta) => {
///             metadata = Some(meta);
///         }
///         Token::Row(row) => {
///             // Process row using metadata
///         }
///         Token::Done(done) => {
///             if !done.has_more() {
///                 break;
///             }
///         }
///         _ => {}
///     }
/// }
/// ```
pub struct TokenParser {
    data: Bytes,
    position: usize,
    /// Whether Always Encrypted was negotiated for this connection.
    /// When true, ColMetaData tokens are parsed with CekTable and per-column CryptoMetadata.
    encryption_enabled: bool,
    /// COMPUTE BY (ALTMETADATA) column layouts keyed by compute Id, used to
    /// parse and drop the matching ALTROW tokens. Cleared on each new
    /// ColMetaData (result-set boundary), where compute Ids become unique again.
    alt_metadata: Vec<(u16, ColMetaData)>,
}

impl TokenParser {
    /// Create a new token parser from bytes.
    #[must_use]
    pub fn new(data: Bytes) -> Self {
        Self {
            data,
            position: 0,
            encryption_enabled: false,
            alt_metadata: Vec::new(),
        }
    }

    /// Enable Always Encrypted metadata parsing.
    ///
    /// When enabled, ColMetaData tokens are parsed using the encrypted format
    /// which includes a CekTable and per-column CryptoMetadata.
    #[must_use]
    pub fn with_encryption(mut self, enabled: bool) -> Self {
        self.encryption_enabled = enabled;
        self
    }

    /// Get remaining bytes in the buffer.
    #[must_use]
    pub fn remaining(&self) -> usize {
        self.data.len().saturating_sub(self.position)
    }

    /// Check if there are more bytes to parse.
    #[must_use]
    pub fn has_remaining(&self) -> bool {
        self.position < self.data.len()
    }

    /// Peek at the next token type without consuming it.
    #[must_use]
    pub fn peek_token_type(&self) -> Option<TokenType> {
        if self.position < self.data.len() {
            TokenType::from_u8(self.data[self.position])
        } else {
            None
        }
    }

    /// Parse the next token from the stream.
    ///
    /// This method can only parse context-independent tokens. For tokens that
    /// require column metadata (ColMetaData, Row, NbcRow), use
    /// [`next_token_with_metadata()`](TokenParser::next_token_with_metadata).
    ///
    /// Returns `None` if no more tokens are available.
    pub fn next_token(&mut self) -> Result<Option<Token>, ProtocolError> {
        self.next_token_with_metadata(None)
    }

    /// Parse the next token with optional column metadata context.
    ///
    /// When `metadata` is provided, this method can parse Row and NbcRow tokens.
    /// Without metadata, those tokens will return an error.
    ///
    /// Returns `None` if no more tokens are available.
    pub fn next_token_with_metadata(
        &mut self,
        metadata: Option<&ColMetaData>,
    ) -> Result<Option<Token>, ProtocolError> {
        loop {
            if !self.has_remaining() {
                return Ok(None);
            }

            let mut buf = &self.data[self.position..];
            let start_pos = self.position;

            let token_type_byte = buf.get_u8();
            let token_type = TokenType::from_u8(token_type_byte);

            let token = match token_type {
                Some(TokenType::Done) => {
                    let done = Done::decode(&mut buf)?;
                    Token::Done(done)
                }
                Some(TokenType::DoneProc) => {
                    let done = DoneProc::decode(&mut buf)?;
                    Token::DoneProc(done)
                }
                Some(TokenType::DoneInProc) => {
                    let done = DoneInProc::decode(&mut buf)?;
                    Token::DoneInProc(done)
                }
                Some(TokenType::Error) => {
                    let error = ServerError::decode(&mut buf)?;
                    Token::Error(error)
                }
                Some(TokenType::Info) => {
                    let info = ServerInfo::decode(&mut buf)?;
                    Token::Info(info)
                }
                Some(TokenType::LoginAck) => {
                    let login_ack = LoginAck::decode(&mut buf)?;
                    Token::LoginAck(login_ack)
                }
                Some(TokenType::EnvChange) => {
                    let env_change = EnvChange::decode(&mut buf)?;
                    Token::EnvChange(env_change)
                }
                Some(TokenType::Order) => {
                    let order = Order::decode(&mut buf)?;
                    Token::Order(order)
                }
                Some(TokenType::FeatureExtAck) => {
                    let ack = FeatureExtAck::decode(&mut buf)?;
                    Token::FeatureExtAck(ack)
                }
                Some(TokenType::Sspi) => {
                    let sspi = SspiToken::decode(&mut buf)?;
                    Token::Sspi(sspi)
                }
                Some(TokenType::FedAuthInfo) => {
                    let info = FedAuthInfo::decode(&mut buf)?;
                    Token::FedAuthInfo(info)
                }
                Some(TokenType::ReturnStatus) => {
                    if buf.remaining() < 4 {
                        return Err(ProtocolError::UnexpectedEof);
                    }
                    let status = buf.get_i32_le();
                    Token::ReturnStatus(status)
                }
                Some(TokenType::ColMetaData) => {
                    let col_meta = if self.encryption_enabled {
                        ColMetaData::decode_encrypted(&mut buf)?
                    } else {
                        ColMetaData::decode(&mut buf)?
                    };
                    // New result set: prior COMPUTE BY layouts no longer apply
                    // and their Ids may be reused.
                    self.alt_metadata.clear();
                    Token::ColMetaData(col_meta)
                }
                Some(TokenType::Row) => {
                    let meta = metadata.ok_or_else(|| {
                        ProtocolError::StringEncoding(
                            #[cfg(feature = "std")]
                            "Row token requires column metadata".to_string(),
                            #[cfg(not(feature = "std"))]
                            "Row token requires column metadata",
                        )
                    })?;
                    let row = RawRow::decode(&mut buf, meta)?;
                    Token::Row(row)
                }
                Some(TokenType::NbcRow) => {
                    let meta = metadata.ok_or_else(|| {
                        ProtocolError::StringEncoding(
                            #[cfg(feature = "std")]
                            "NbcRow token requires column metadata".to_string(),
                            #[cfg(not(feature = "std"))]
                            "NbcRow token requires column metadata",
                        )
                    })?;
                    let row = NbcRow::decode(&mut buf, meta)?;
                    Token::NbcRow(row)
                }
                Some(TokenType::ReturnValue) => {
                    let ret_val = ReturnValue::decode(&mut buf)?;
                    Token::ReturnValue(ret_val)
                }
                Some(TokenType::SessionState) => {
                    let session = SessionState::decode(&mut buf)?;
                    Token::SessionState(session)
                }
                Some(TokenType::ColInfo) | Some(TokenType::TabName) | Some(TokenType::Offset) => {
                    // These tokens are rarely used and have complex formats.
                    // Skip them by reading the length and advancing.
                    if buf.remaining() < 2 {
                        return Err(ProtocolError::UnexpectedEof);
                    }
                    let length = buf.get_u16_le() as usize;
                    if buf.remaining() < length {
                        return Err(ProtocolError::IncompletePacket {
                            expected: length,
                            actual: buf.remaining(),
                        });
                    }
                    // Skip the data
                    buf.advance(length);
                    // #273: advance past the skipped token and iterate. The skip
                    // path must NOT recurse — a server-controlled flat run of these
                    // tokens would otherwise add one stack frame per token and
                    // overflow the stack (remote DoS).
                    self.position = start_pos + (self.data.len() - start_pos - buf.remaining());
                    continue;
                }
                Some(TokenType::AltMetaData) => {
                    // COMPUTE BY metadata (#275). Parse to learn the ALTROW value
                    // layout, store it keyed by Id, then drop the token — compute
                    // rows are not surfaced, but their bytes must be consumed so
                    // the base result set stays parseable.
                    let (id, alt_meta) = ColMetaData::decode_alt(&mut buf)?;
                    self.position = start_pos + (self.data.len() - start_pos - buf.remaining());
                    self.alt_metadata.push((id, alt_meta));
                    continue;
                }
                Some(TokenType::AltRow) => {
                    // COMPUTE BY row (#275). Parse using the matching ALTMETADATA
                    // to consume its bytes, then drop it. Like the ColInfo skip
                    // path, this MUST NOT recurse (remote DoS via a flat token run).
                    if buf.remaining() < 2 {
                        return Err(ProtocolError::UnexpectedEof);
                    }
                    let id = buf.get_u16_le();
                    let alt_meta = self
                        .alt_metadata
                        .iter()
                        .find(|(stored_id, _)| *stored_id == id)
                        .map(|(_, meta)| meta)
                        .ok_or_else(|| {
                            ProtocolError::StringEncoding(
                                #[cfg(feature = "std")]
                                "ALTROW token without matching ALTMETADATA".to_string(),
                                #[cfg(not(feature = "std"))]
                                "ALTROW token without matching ALTMETADATA",
                            )
                        })?;
                    let _ = RawRow::decode(&mut buf, alt_meta)?;
                    self.position = start_pos + (self.data.len() - start_pos - buf.remaining());
                    continue;
                }
                None => {
                    return Err(ProtocolError::InvalidTokenType(token_type_byte));
                }
            };

            // Update position based on how much was consumed
            let consumed = self.data.len() - start_pos - buf.remaining();
            self.position = start_pos + consumed;

            return Ok(Some(token));
        }
    }

    /// Skip the current token without fully parsing it.
    ///
    /// This is useful for skipping unknown or uninteresting tokens.
    pub fn skip_token(&mut self) -> Result<(), ProtocolError> {
        if !self.has_remaining() {
            return Ok(());
        }

        let token_type_byte = self.data[self.position];
        let token_type = TokenType::from_u8(token_type_byte);

        // Calculate how many bytes to skip based on token type
        let skip_amount = match token_type {
            // Fixed-size tokens
            Some(TokenType::Done) | Some(TokenType::DoneProc) | Some(TokenType::DoneInProc) => {
                1 + Done::SIZE // token type + 12 bytes
            }
            Some(TokenType::ReturnStatus) => {
                1 + 4 // token type + 4 bytes
            }
            // Variable-length tokens with 2-byte length prefix
            Some(TokenType::Error)
            | Some(TokenType::Info)
            | Some(TokenType::LoginAck)
            | Some(TokenType::EnvChange)
            | Some(TokenType::Order)
            | Some(TokenType::Sspi)
            | Some(TokenType::ColInfo)
            | Some(TokenType::TabName)
            | Some(TokenType::Offset)
            | Some(TokenType::ReturnValue) => {
                if self.remaining() < 3 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                let length = u16::from_le_bytes([
                    self.data[self.position + 1],
                    self.data[self.position + 2],
                ]) as usize;
                1 + 2 + length // token type + length prefix + data
            }
            // Tokens with 4-byte length prefix
            Some(TokenType::SessionState) | Some(TokenType::FedAuthInfo) => {
                if self.remaining() < 5 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                let length = u32::from_le_bytes([
                    self.data[self.position + 1],
                    self.data[self.position + 2],
                    self.data[self.position + 3],
                    self.data[self.position + 4],
                ]) as usize;
                1 + 4 + length
            }
            // FeatureExtAck has no length prefix - must parse
            Some(TokenType::FeatureExtAck) => {
                // Parse to find end
                let mut buf = &self.data[self.position + 1..];
                let _ = FeatureExtAck::decode(&mut buf)?;
                self.data.len() - self.position - buf.remaining()
            }
            // COMPUTE BY metadata (#275) has no length prefix; parse to find its
            // end, recording the layout so a following ALTROW can be skipped.
            Some(TokenType::AltMetaData) => {
                let mut buf = &self.data[self.position + 1..];
                let (id, alt_meta) = ColMetaData::decode_alt(&mut buf)?;
                let consumed = self.data.len() - self.position - buf.remaining();
                self.alt_metadata.push((id, alt_meta));
                consumed
            }
            // COMPUTE BY row (#275); length depends on the matching ALTMETADATA.
            Some(TokenType::AltRow) => {
                let mut buf = &self.data[self.position + 1..];
                if buf.remaining() < 2 {
                    return Err(ProtocolError::UnexpectedEof);
                }
                let id = buf.get_u16_le();
                let alt_meta = self
                    .alt_metadata
                    .iter()
                    .find(|(stored_id, _)| *stored_id == id)
                    .map(|(_, meta)| meta)
                    .ok_or(ProtocolError::InvalidTokenType(token_type_byte))?;
                let _ = RawRow::decode(&mut buf, alt_meta)?;
                self.data.len() - self.position - buf.remaining()
            }
            // ColMetaData, Row, NbcRow require context and can't be easily skipped
            Some(TokenType::ColMetaData) | Some(TokenType::Row) | Some(TokenType::NbcRow) => {
                return Err(ProtocolError::InvalidTokenType(token_type_byte));
            }
            None => {
                return Err(ProtocolError::InvalidTokenType(token_type_byte));
            }
        };

        if self.remaining() < skip_amount {
            return Err(ProtocolError::UnexpectedEof);
        }

        self.position += skip_amount;
        Ok(())
    }

    /// Get the current position in the buffer.
    #[must_use]
    pub fn position(&self) -> usize {
        self.position
    }

    /// Reset the parser to the beginning.
    pub fn reset(&mut self) {
        self.position = 0;
    }
}

// =============================================================================
// Tests
// =============================================================================

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

    #[test]
    fn test_done_roundtrip() {
        let done = Done {
            status: DoneStatus {
                more: false,
                error: false,
                in_xact: false,
                count: true,
                attn: false,
                srverror: false,
            },
            cur_cmd: 193, // SELECT
            row_count: 42,
        };

        let mut buf = BytesMut::new();
        done.encode(&mut buf);

        // Skip the token type byte
        let mut cursor = &buf[1..];
        let decoded = Done::decode(&mut cursor).unwrap();

        assert_eq!(decoded.status.count, done.status.count);
        assert_eq!(decoded.cur_cmd, done.cur_cmd);
        assert_eq!(decoded.row_count, done.row_count);
    }

    #[test]
    fn test_done_status_bits() {
        let status = DoneStatus {
            more: true,
            error: true,
            in_xact: true,
            count: true,
            attn: false,
            srverror: false,
        };

        let bits = status.to_bits();
        let restored = DoneStatus::from_bits(bits);

        assert_eq!(status.more, restored.more);
        assert_eq!(status.error, restored.error);
        assert_eq!(status.in_xact, restored.in_xact);
        assert_eq!(status.count, restored.count);
    }

    #[test]
    fn test_token_parser_done() {
        // DONE token: type (1) + status (2) + curcmd (2) + rowcount (8)
        let data = Bytes::from_static(&[
            0xFD, // DONE token type
            0x10, 0x00, // status: DONE_COUNT
            0xC1, 0x00, // cur_cmd: 193 (SELECT)
            0x05, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // row_count: 5
        ]);

        let mut parser = TokenParser::new(data);
        let token = parser.next_token().unwrap().unwrap();

        match token {
            Token::Done(done) => {
                assert!(done.status.count);
                assert!(!done.status.more);
                assert_eq!(done.cur_cmd, 193);
                assert_eq!(done.row_count, 5);
            }
            _ => panic!("Expected Done token"),
        }

        // No more tokens
        assert!(parser.next_token().unwrap().is_none());
    }

    #[test]
    fn test_env_change_type_from_u8() {
        assert_eq!(EnvChangeType::from_u8(1), Some(EnvChangeType::Database));
        assert_eq!(EnvChangeType::from_u8(20), Some(EnvChangeType::Routing));
        assert_eq!(EnvChangeType::from_u8(100), None);
    }

    /// A spec-faithful Routing ENVCHANGE (MS-TDS 2.2.7.9) carries a
    /// zero-length OldValue (two bytes) after the routing data. The decoder
    /// reads only the NewValue, so it must skip to the declared frame
    /// boundary — otherwise the leftover `00 00` is misparsed as the next
    /// token type and the rest of the login response is garbage. Azure SQL
    /// Gateway redirects send exactly this shape.
    #[test]
    fn test_env_change_routing_consumes_declared_length() {
        let host = "redirect.example";
        let host_utf16: Vec<u16> = host.encode_utf16().collect();

        let mut data = BytesMut::new();
        // RoutingDataValue: length + protocol + port + server_len + server
        let routing_len = 1 + 2 + 2 + host_utf16.len() * 2;
        // ENVCHANGE length: type byte + routing length prefix + routing data
        // + zero-length OldValue
        let env_len = 1 + 2 + routing_len + 2;
        data.put_u16_le(env_len as u16);
        data.put_u8(20); // Routing
        data.put_u16_le(routing_len as u16);
        data.put_u8(0); // protocol: TCP
        data.put_u16_le(11000); // port
        data.put_u16_le(host_utf16.len() as u16);
        for c in &host_utf16 {
            data.put_u16_le(*c);
        }
        data.put_u16_le(0); // OldValue: zero-length US_VARBYTE
        // A trailing DONE token type byte that must remain for the next read.
        data.put_u8(0xFD);

        let mut buf: &[u8] = &data;
        let env = EnvChange::decode(&mut buf).unwrap();
        assert_eq!(env.routing_info(), Some((host, 11000)));
        assert_eq!(
            buf,
            &[0xFD],
            "decode must consume exactly the declared ENVCHANGE frame"
        );
    }

    fn put_b_varchar(buf: &mut BytesMut, s: &str) {
        let utf16: Vec<u16> = s.encode_utf16().collect();
        buf.put_u8(utf16.len() as u8);
        for c in utf16 {
            buf.put_u16_le(c);
        }
    }

    fn put_us_varchar(buf: &mut BytesMut, s: &str) {
        let utf16: Vec<u16> = s.encode_utf16().collect();
        buf.put_u16_le(utf16.len() as u16);
        for c in utf16 {
            buf.put_u16_le(c);
        }
    }

    /// UDT_INFO regression (issue #154): per MS-TDS, DB_NAME, SCHEMA_NAME,
    /// and TYPE_NAME are B_VARCHAR (1-byte length); only
    /// ASSEMBLY_QUALIFIED_NAME is US_VARCHAR. Reading all four as US_VARCHAR
    /// misaligned the stream, so every query selecting a UDT column
    /// (geography, geometry, hierarchyid, CLR UDTs) failed with
    /// UnexpectedEof.
    #[test]
    fn test_udt_info_metadata_uses_b_varchar_names() {
        let mut data = BytesMut::new();
        data.put_u16_le(0xFFFF); // MAX_BYTE_SIZE
        put_b_varchar(&mut data, "master");
        put_b_varchar(&mut data, "dbo");
        put_b_varchar(&mut data, "hierarchyid");
        put_us_varchar(
            &mut data,
            "Microsoft.SqlServer.Types.SqlHierarchyId, Microsoft.SqlServer.Types",
        );
        // A trailing token type byte that must remain for the next read.
        data.put_u8(0xFD);

        let mut buf: &[u8] = &data;
        let info = decode_type_info(&mut buf, TypeId::Udt, TypeId::Udt as u8).unwrap();
        assert_eq!(info.max_length, Some(0xFFFF));
        assert_eq!(
            buf,
            &[0xFD],
            "decode must consume exactly the UDT_INFO frame"
        );
    }

    /// XML_INFO regression (issue #154): per MS-TDS §2.2.5.5.3, DBNAME and
    /// OWNING_SCHEMA are B_VARCHAR; only XML_SCHEMA_COLLECTION is US_VARCHAR.
    /// Schema-bound xml columns (SCHEMA_PRESENT=1) previously misparsed.
    #[test]
    fn test_xml_info_schema_bound_uses_b_varchar_names() {
        let mut data = BytesMut::new();
        data.put_u8(1); // SCHEMA_PRESENT
        put_b_varchar(&mut data, "master");
        put_b_varchar(&mut data, "dbo");
        put_us_varchar(&mut data, "MyXmlSchemaCollection");
        data.put_u8(0xFD);

        let mut buf: &[u8] = &data;
        decode_type_info(&mut buf, TypeId::Xml, TypeId::Xml as u8).unwrap();
        assert_eq!(
            buf,
            &[0xFD],
            "decode must consume exactly the XML_INFO frame"
        );
    }

    #[test]
    fn hostile_env_change_binary_truncated_is_not_panic() {
        // length=1 covers only the type byte (0x08 = BeginTransaction, a
        // binary-format type); the new_len/old_len prefix reads then hit an
        // empty buffer. Must decode gracefully, never panic (found by the
        // parse_env_change and parse_token fuzz targets).
        let data = [0x01, 0x00, 0x08];
        let mut buf: &[u8] = &data;
        let env = EnvChange::decode(&mut buf).unwrap();
        assert_eq!(env.env_type, EnvChangeType::BeginTransaction);
    }

    /// Issue #145: an under-declared frame must not let the value decoders
    /// read past the declared length into the next token's bytes.
    #[test]
    fn hostile_env_change_under_declared_cannot_steal_following_bytes() {
        // length=1 covers only the type byte; the bytes after the frame are
        // shaped exactly like the transaction-descriptor payload the old
        // buffer-bounded decoder would have consumed (new_len=8, descriptor,
        // old_len=0).
        let mut data = BytesMut::new();
        data.put_u16_le(1); // declared frame: type byte only
        data.put_u8(0x08); // BeginTransaction
        let following: &[u8] = &[0x08, 1, 2, 3, 4, 5, 6, 7, 8, 0x00];
        data.extend_from_slice(following);

        let mut buf: &[u8] = &data;
        let env = EnvChange::decode(&mut buf).unwrap();
        assert_eq!(env.env_type, EnvChangeType::BeginTransaction);
        match &env.new_value {
            EnvChangeValue::Binary(b) => {
                assert!(
                    b.is_empty(),
                    "under-declared frame yields the lenient empty value"
                );
            }
            other => panic!("expected empty Binary value, got {other:?}"),
        }
        assert_eq!(
            buf, following,
            "bytes beyond the declared frame belong to the next token"
        );
    }

    /// Issue #145: a zero-length frame cannot supply a type byte; reading
    /// one from beyond the frame would consume the next token.
    #[test]
    fn hostile_env_change_zero_length_frame_errors() {
        let data = [0x00, 0x00, 0xFD];
        let mut buf: &[u8] = &data;
        assert!(EnvChange::decode(&mut buf).is_err());
    }

    #[test]
    fn test_colmetadata_no_columns() {
        // No metadata marker (0xFFFF)
        let data = Bytes::from_static(&[0xFF, 0xFF]);
        let mut cursor: &[u8] = &data;
        let meta = ColMetaData::decode(&mut cursor).unwrap();
        assert!(meta.is_empty());
        assert_eq!(meta.column_count(), 0);
    }

    #[test]
    fn test_colmetadata_single_int_column() {
        // COLMETADATA with 1 INT column
        // Format: column_count (2) + [user_type (4) + flags (2) + type_id (1) + name (b_varchar)]
        let mut data = BytesMut::new();
        data.extend_from_slice(&[0x01, 0x00]); // 1 column
        data.extend_from_slice(&[0x00, 0x00, 0x00, 0x00]); // user_type = 0
        data.extend_from_slice(&[0x01, 0x00]); // flags (nullable)
        data.extend_from_slice(&[0x38]); // TypeId::Int4
        // Column name "id" as B_VARCHAR (1 byte length + UTF-16LE)
        data.extend_from_slice(&[0x02]); // 2 characters
        data.extend_from_slice(&[b'i', 0x00, b'd', 0x00]); // "id" in UTF-16LE

        let mut cursor: &[u8] = &data;
        let meta = ColMetaData::decode(&mut cursor).unwrap();

        assert_eq!(meta.column_count(), 1);
        assert_eq!(meta.columns[0].name, "id");
        assert_eq!(meta.columns[0].type_id, TypeId::Int4);
        assert!(meta.columns[0].is_nullable());
    }

    #[test]
    fn test_colmetadata_nvarchar_column() {
        // COLMETADATA with 1 NVARCHAR(50) column
        let mut data = BytesMut::new();
        data.extend_from_slice(&[0x01, 0x00]); // 1 column
        data.extend_from_slice(&[0x00, 0x00, 0x00, 0x00]); // user_type = 0
        data.extend_from_slice(&[0x01, 0x00]); // flags (nullable)
        data.extend_from_slice(&[0xE7]); // TypeId::NVarChar
        // Type info: max_length (2 bytes) + collation (5 bytes)
        data.extend_from_slice(&[0x64, 0x00]); // max_length = 100 (50 chars * 2)
        data.extend_from_slice(&[0x09, 0x04, 0xD0, 0x00, 0x34]); // collation
        // Column name "name"
        data.extend_from_slice(&[0x04]); // 4 characters
        data.extend_from_slice(&[b'n', 0x00, b'a', 0x00, b'm', 0x00, b'e', 0x00]);

        let mut cursor: &[u8] = &data;
        let meta = ColMetaData::decode(&mut cursor).unwrap();

        assert_eq!(meta.column_count(), 1);
        assert_eq!(meta.columns[0].name, "name");
        assert_eq!(meta.columns[0].type_id, TypeId::NVarChar);
        assert_eq!(meta.columns[0].type_info.max_length, Some(100));
        assert!(meta.columns[0].type_info.collation.is_some());
    }

    #[test]
    fn test_raw_row_decode_int() {
        // Create metadata for a single INT column
        let metadata = ColMetaData {
            cek_table: None,
            columns: vec![ColumnData {
                name: "id".to_string(),
                type_id: TypeId::Int4,
                col_type: 0x38,
                flags: 0,
                user_type: 0,
                type_info: TypeInfo::default(),
                crypto_metadata: None,
            }],
        };

        // Row data: just 4 bytes for the int value 42
        let data = Bytes::from_static(&[0x2A, 0x00, 0x00, 0x00]); // 42 in little-endian
        let mut cursor: &[u8] = &data;
        let row = RawRow::decode(&mut cursor, &metadata).unwrap();

        // The raw data should contain the 4 bytes
        assert_eq!(row.data.len(), 4);
        assert_eq!(&row.data[..], &[0x2A, 0x00, 0x00, 0x00]);
    }

    #[test]
    fn test_raw_row_decode_nullable_int() {
        // Create metadata for a nullable INT column (IntN)
        let metadata = ColMetaData {
            cek_table: None,
            columns: vec![ColumnData {
                name: "id".to_string(),
                type_id: TypeId::IntN,
                col_type: 0x26,
                flags: 0x01, // nullable
                user_type: 0,
                type_info: TypeInfo {
                    max_length: Some(4),
                    ..Default::default()
                },
                crypto_metadata: None,
            }],
        };

        // Row data with value: 1 byte length + 4 bytes value
        let data = Bytes::from_static(&[0x04, 0x2A, 0x00, 0x00, 0x00]); // length=4, value=42
        let mut cursor: &[u8] = &data;
        let row = RawRow::decode(&mut cursor, &metadata).unwrap();

        assert_eq!(row.data.len(), 5);
        assert_eq!(row.data[0], 4); // length
        assert_eq!(&row.data[1..], &[0x2A, 0x00, 0x00, 0x00]);
    }

    #[test]
    fn test_raw_row_decode_null_value() {
        // Create metadata for a nullable INT column (IntN)
        let metadata = ColMetaData {
            cek_table: None,
            columns: vec![ColumnData {
                name: "id".to_string(),
                type_id: TypeId::IntN,
                col_type: 0x26,
                flags: 0x01, // nullable
                user_type: 0,
                type_info: TypeInfo {
                    max_length: Some(4),
                    ..Default::default()
                },
                crypto_metadata: None,
            }],
        };

        // NULL value: length = 0xFF (for bytelen types)
        let data = Bytes::from_static(&[0xFF]);
        let mut cursor: &[u8] = &data;
        let row = RawRow::decode(&mut cursor, &metadata).unwrap();

        assert_eq!(row.data.len(), 1);
        assert_eq!(row.data[0], 0xFF); // NULL marker
    }

    #[test]
    fn test_nbcrow_null_bitmap() {
        let row = NbcRow {
            null_bitmap: vec![0b00000101], // columns 0 and 2 are NULL
            data: Bytes::new(),
        };

        assert!(row.is_null(0));
        assert!(!row.is_null(1));
        assert!(row.is_null(2));
        assert!(!row.is_null(3));
    }

    #[test]
    fn test_token_parser_colmetadata() {
        // Build a COLMETADATA token with 1 INT column
        let mut data = BytesMut::new();
        data.extend_from_slice(&[0x81]); // COLMETADATA token type
        data.extend_from_slice(&[0x01, 0x00]); // 1 column
        data.extend_from_slice(&[0x00, 0x00, 0x00, 0x00]); // user_type = 0
        data.extend_from_slice(&[0x01, 0x00]); // flags (nullable)
        data.extend_from_slice(&[0x38]); // TypeId::Int4
        data.extend_from_slice(&[0x02]); // column name length
        data.extend_from_slice(&[b'i', 0x00, b'd', 0x00]); // "id"

        let mut parser = TokenParser::new(data.freeze());
        let token = parser.next_token().unwrap().unwrap();

        match token {
            Token::ColMetaData(meta) => {
                assert_eq!(meta.column_count(), 1);
                assert_eq!(meta.columns[0].name, "id");
                assert_eq!(meta.columns[0].type_id, TypeId::Int4);
            }
            _ => panic!("Expected ColMetaData token"),
        }
    }

    #[test]
    fn test_token_parser_row_with_metadata() {
        // Build metadata
        let metadata = ColMetaData {
            cek_table: None,
            columns: vec![ColumnData {
                name: "id".to_string(),
                type_id: TypeId::Int4,
                col_type: 0x38,
                flags: 0,
                user_type: 0,
                type_info: TypeInfo::default(),
                crypto_metadata: None,
            }],
        };

        // Build ROW token
        let mut data = BytesMut::new();
        data.extend_from_slice(&[0xD1]); // ROW token type
        data.extend_from_slice(&[0x2A, 0x00, 0x00, 0x00]); // value = 42

        let mut parser = TokenParser::new(data.freeze());
        let token = parser
            .next_token_with_metadata(Some(&metadata))
            .unwrap()
            .unwrap();

        match token {
            Token::Row(row) => {
                assert_eq!(row.data.len(), 4);
            }
            _ => panic!("Expected Row token"),
        }
    }

    #[test]
    fn test_token_parser_row_without_metadata_fails() {
        // Build ROW token
        let mut data = BytesMut::new();
        data.extend_from_slice(&[0xD1]); // ROW token type
        data.extend_from_slice(&[0x2A, 0x00, 0x00, 0x00]); // value = 42

        let mut parser = TokenParser::new(data.freeze());
        let result = parser.next_token(); // No metadata provided

        assert!(result.is_err());
    }

    #[test]
    fn test_token_parser_peek() {
        let data = Bytes::from_static(&[
            0xFD, // DONE token type
            0x10, 0x00, // status
            0xC1, 0x00, // cur_cmd
            0x05, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // row_count
        ]);

        let parser = TokenParser::new(data);
        assert_eq!(parser.peek_token_type(), Some(TokenType::Done));
    }

    #[test]
    fn test_column_data_fixed_size() {
        let col = ColumnData {
            name: String::new(),
            type_id: TypeId::Int4,
            col_type: 0x38,
            flags: 0,
            user_type: 0,
            type_info: TypeInfo::default(),
            crypto_metadata: None,
        };
        assert_eq!(col.fixed_size(), Some(4));

        let col2 = ColumnData {
            name: String::new(),
            type_id: TypeId::NVarChar,
            col_type: 0xE7,
            flags: 0,
            user_type: 0,
            type_info: TypeInfo::default(),
            crypto_metadata: None,
        };
        assert_eq!(col2.fixed_size(), None);
    }

    // ========================================================================
    // End-to-End Decode Tests (Wire → Stored → Verification)
    // ========================================================================
    //
    // These tests verify that RawRow::decode_column_value correctly stores
    // column values in a format that can be parsed back.

    #[test]
    fn test_decode_nvarchar_then_intn_roundtrip() {
        // Simulate wire data for: "World" (NVarChar), 42 (IntN)
        // This tests the scenario from the MCP parameterized query

        // Build wire data (what the server sends)
        let mut wire_data = BytesMut::new();

        // Column 0: NVarChar "World" - 2-byte length prefix in bytes
        // "World" in UTF-16LE: W=0x0057, o=0x006F, r=0x0072, l=0x006C, d=0x0064
        let word = "World";
        let utf16: Vec<u16> = word.encode_utf16().collect();
        wire_data.put_u16_le((utf16.len() * 2) as u16); // byte length = 10
        for code_unit in &utf16 {
            wire_data.put_u16_le(*code_unit);
        }

        // Column 1: IntN 42 - 1-byte length prefix
        wire_data.put_u8(4); // 4 bytes for INT
        wire_data.put_i32_le(42);

        // Build column metadata
        let metadata = ColMetaData {
            cek_table: None,
            columns: vec![
                ColumnData {
                    name: "greeting".to_string(),
                    type_id: TypeId::NVarChar,
                    col_type: 0xE7,
                    flags: 0x01,
                    user_type: 0,
                    type_info: TypeInfo {
                        max_length: Some(10), // non-MAX
                        precision: None,
                        scale: None,
                        collation: None,
                    },
                    crypto_metadata: None,
                },
                ColumnData {
                    name: "number".to_string(),
                    type_id: TypeId::IntN,
                    col_type: 0x26,
                    flags: 0x01,
                    user_type: 0,
                    type_info: TypeInfo {
                        max_length: Some(4),
                        precision: None,
                        scale: None,
                        collation: None,
                    },
                    crypto_metadata: None,
                },
            ],
        };

        // Decode the wire data into stored format
        let mut wire_cursor = wire_data.freeze();
        let raw_row = RawRow::decode(&mut wire_cursor, &metadata).unwrap();

        // Verify wire data was fully consumed
        assert_eq!(
            wire_cursor.remaining(),
            0,
            "wire data should be fully consumed"
        );

        // Now parse the stored data
        let mut stored_cursor: &[u8] = &raw_row.data;

        // Parse column 0 (NVarChar)
        // Stored format for non-MAX NVarChar: [2-byte len][data]
        assert!(
            stored_cursor.remaining() >= 2,
            "need at least 2 bytes for length"
        );
        let len0 = stored_cursor.get_u16_le() as usize;
        assert_eq!(len0, 10, "NVarChar length should be 10 bytes");
        assert!(
            stored_cursor.remaining() >= len0,
            "need {len0} bytes for data"
        );

        // Read UTF-16LE and convert to string
        let mut utf16_read = Vec::new();
        for _ in 0..(len0 / 2) {
            utf16_read.push(stored_cursor.get_u16_le());
        }
        let string0 = String::from_utf16(&utf16_read).unwrap();
        assert_eq!(string0, "World", "column 0 should be 'World'");

        // Parse column 1 (IntN)
        // Stored format for IntN: [1-byte len][data]
        assert!(
            stored_cursor.remaining() >= 1,
            "need at least 1 byte for length"
        );
        let len1 = stored_cursor.get_u8();
        assert_eq!(len1, 4, "IntN length should be 4");
        assert!(stored_cursor.remaining() >= 4, "need 4 bytes for INT data");
        let int1 = stored_cursor.get_i32_le();
        assert_eq!(int1, 42, "column 1 should be 42");

        // Verify stored data was fully consumed
        assert_eq!(
            stored_cursor.remaining(),
            0,
            "stored data should be fully consumed"
        );
    }

    #[test]
    fn test_decode_nvarchar_max_then_intn_roundtrip() {
        // Test NVARCHAR(MAX) followed by IntN - uses PLP encoding

        // Build wire data for PLP NVARCHAR(MAX) + IntN
        let mut wire_data = BytesMut::new();

        // Column 0: NVARCHAR(MAX) "Hello" - PLP format
        // PLP: 8-byte total length, then chunks
        let word = "Hello";
        let utf16: Vec<u16> = word.encode_utf16().collect();
        let byte_len = (utf16.len() * 2) as u64;

        wire_data.put_u64_le(byte_len); // total length = 10
        wire_data.put_u32_le(byte_len as u32); // chunk length = 10
        for code_unit in &utf16 {
            wire_data.put_u16_le(*code_unit);
        }
        wire_data.put_u32_le(0); // terminating zero-length chunk

        // Column 1: IntN 99
        wire_data.put_u8(4);
        wire_data.put_i32_le(99);

        // Build metadata with MAX type
        let metadata = ColMetaData {
            cek_table: None,
            columns: vec![
                ColumnData {
                    name: "text".to_string(),
                    type_id: TypeId::NVarChar,
                    col_type: 0xE7,
                    flags: 0x01,
                    user_type: 0,
                    type_info: TypeInfo {
                        max_length: Some(0xFFFF), // MAX indicator
                        precision: None,
                        scale: None,
                        collation: None,
                    },
                    crypto_metadata: None,
                },
                ColumnData {
                    name: "num".to_string(),
                    type_id: TypeId::IntN,
                    col_type: 0x26,
                    flags: 0x01,
                    user_type: 0,
                    type_info: TypeInfo {
                        max_length: Some(4),
                        precision: None,
                        scale: None,
                        collation: None,
                    },
                    crypto_metadata: None,
                },
            ],
        };

        // Decode wire data
        let mut wire_cursor = wire_data.freeze();
        let raw_row = RawRow::decode(&mut wire_cursor, &metadata).unwrap();

        // Verify wire data was fully consumed
        assert_eq!(
            wire_cursor.remaining(),
            0,
            "wire data should be fully consumed"
        );

        // Parse stored PLP data for column 0
        let mut stored_cursor: &[u8] = &raw_row.data;

        // PLP stored format: [8-byte total][chunks...][4-byte 0]
        let total_len = stored_cursor.get_u64_le();
        assert_eq!(total_len, 10, "PLP total length should be 10");

        let chunk_len = stored_cursor.get_u32_le();
        assert_eq!(chunk_len, 10, "PLP chunk length should be 10");

        let mut utf16_read = Vec::new();
        for _ in 0..(chunk_len / 2) {
            utf16_read.push(stored_cursor.get_u16_le());
        }
        let string0 = String::from_utf16(&utf16_read).unwrap();
        assert_eq!(string0, "Hello", "column 0 should be 'Hello'");

        let terminator = stored_cursor.get_u32_le();
        assert_eq!(terminator, 0, "PLP should end with 0");

        // Parse IntN
        let len1 = stored_cursor.get_u8();
        assert_eq!(len1, 4);
        let int1 = stored_cursor.get_i32_le();
        assert_eq!(int1, 99, "column 1 should be 99");

        // Verify fully consumed
        assert_eq!(
            stored_cursor.remaining(),
            0,
            "stored data should be fully consumed"
        );
    }

    // ========================================================================
    // ReturnStatus Token Tests
    // ========================================================================

    #[test]
    fn test_return_status_via_parser() {
        // RETURNSTATUS token: type (0x79) + value (i32 LE)
        let data = Bytes::from_static(&[
            0x79, // RETURNSTATUS token type
            0x00, 0x00, 0x00, 0x00, // return value = 0 (success)
        ]);

        let mut parser = TokenParser::new(data);
        let token = parser.next_token().unwrap().unwrap();

        match token {
            Token::ReturnStatus(status) => {
                assert_eq!(status, 0);
            }
            _ => panic!("Expected ReturnStatus token, got {token:?}"),
        }

        assert!(parser.next_token().unwrap().is_none());
    }

    #[test]
    fn test_return_status_nonzero() {
        // Return value = -6 (common for error returns)
        let mut buf = BytesMut::new();
        buf.put_u8(0x79); // RETURNSTATUS
        buf.put_i32_le(-6);

        let mut parser = TokenParser::new(buf.freeze());
        let token = parser.next_token().unwrap().unwrap();

        match token {
            Token::ReturnStatus(status) => {
                assert_eq!(status, -6);
            }
            _ => panic!("Expected ReturnStatus token"),
        }
    }

    // ========================================================================
    // DoneProc Token Tests
    // ========================================================================

    #[test]
    fn test_done_proc_roundtrip() {
        let done = DoneProc {
            status: DoneStatus {
                more: false,
                error: false,
                in_xact: false,
                count: true,
                attn: false,
                srverror: false,
            },
            cur_cmd: 0x00C6, // EXECUTE (198)
            row_count: 100,
        };

        let mut buf = BytesMut::new();
        done.encode(&mut buf);

        // Verify token type byte
        assert_eq!(buf[0], 0xFE);

        // Skip token type byte and decode
        let mut cursor = &buf[1..];
        let decoded = DoneProc::decode(&mut cursor).unwrap();

        assert!(decoded.status.count);
        assert!(!decoded.status.more);
        assert!(!decoded.status.error);
        assert_eq!(decoded.cur_cmd, 0x00C6);
        assert_eq!(decoded.row_count, 100);
    }

    #[test]
    fn test_done_proc_via_parser() {
        let data = Bytes::from_static(&[
            0xFE, // DONEPROC token type
            0x00, 0x00, // status: no flags
            0xC6, 0x00, // cur_cmd: EXECUTE (198)
            0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // row_count: 0
        ]);

        let mut parser = TokenParser::new(data);
        let token = parser.next_token().unwrap().unwrap();

        match token {
            Token::DoneProc(done) => {
                assert!(!done.status.count);
                assert!(!done.status.more);
                assert_eq!(done.cur_cmd, 198);
                assert_eq!(done.row_count, 0);
            }
            _ => panic!("Expected DoneProc token"),
        }
    }

    #[test]
    fn test_done_proc_with_error_flag() {
        let mut buf = BytesMut::new();
        buf.put_u8(0xFE); // DONEPROC
        buf.put_u16_le(0x0002); // status: DONE_ERROR
        buf.put_u16_le(0x00C6); // cur_cmd: EXECUTE
        buf.put_u64_le(0); // row_count

        let mut parser = TokenParser::new(buf.freeze());
        let token = parser.next_token().unwrap().unwrap();

        match token {
            Token::DoneProc(done) => {
                assert!(done.status.error);
                assert!(!done.status.count);
                assert!(!done.status.more);
            }
            _ => panic!("Expected DoneProc token"),
        }
    }

    // ========================================================================
    // DoneInProc Token Tests
    // ========================================================================

    #[test]
    fn test_done_in_proc_roundtrip() {
        let done = DoneInProc {
            status: DoneStatus {
                more: true,
                error: false,
                in_xact: false,
                count: true,
                attn: false,
                srverror: false,
            },
            cur_cmd: 193, // SELECT
            row_count: 7,
        };

        let mut buf = BytesMut::new();
        done.encode(&mut buf);

        assert_eq!(buf[0], 0xFF);

        let mut cursor = &buf[1..];
        let decoded = DoneInProc::decode(&mut cursor).unwrap();

        assert!(decoded.status.more);
        assert!(decoded.status.count);
        assert!(!decoded.status.error);
        assert_eq!(decoded.cur_cmd, 193);
        assert_eq!(decoded.row_count, 7);
    }

    #[test]
    fn test_done_in_proc_via_parser() {
        let data = Bytes::from_static(&[
            0xFF, // DONEINPROC token type
            0x11, 0x00, // status: MORE | COUNT
            0xC1, 0x00, // cur_cmd: SELECT (193)
            0x03, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, // row_count: 3
        ]);

        let mut parser = TokenParser::new(data);
        let token = parser.next_token().unwrap().unwrap();

        match token {
            Token::DoneInProc(done) => {
                assert!(done.status.more);
                assert!(done.status.count);
                assert_eq!(done.cur_cmd, 193);
                assert_eq!(done.row_count, 3);
            }
            _ => panic!("Expected DoneInProc token"),
        }
    }

    // ========================================================================
    // ServerError Token Tests
    // ========================================================================

    #[test]
    fn test_server_error_decode() {
        // Build a realistic ERROR token (without the 0xAA type byte,
        // since decode() is called after the parser strips it).
        let mut buf = BytesMut::new();

        // Construct the message fields first to compute length
        let msg_utf16: Vec<u16> = "Invalid column name 'foo'.".encode_utf16().collect();
        let srv_utf16: Vec<u16> = "SQLDB01".encode_utf16().collect();
        let proc_utf16: Vec<u16> = "".encode_utf16().collect();

        // Length = number(4) + state(1) + class(1)
        //        + us_varchar(message): 2 + msg_utf16.len()*2
        //        + b_varchar(server): 1 + srv_utf16.len()*2
        //        + b_varchar(procedure): 1 + proc_utf16.len()*2
        //        + line(4)
        let length: u16 = (4
            + 1
            + 1
            + 2
            + (msg_utf16.len() * 2)
            + 1
            + (srv_utf16.len() * 2)
            + 1
            + (proc_utf16.len() * 2)
            + 4) as u16;

        buf.put_u16_le(length);
        buf.put_i32_le(207); // error number: Invalid column
        buf.put_u8(1); // state
        buf.put_u8(16); // class (severity 16)

        // Message (US_VARCHAR: 2-byte char count + UTF-16LE)
        buf.put_u16_le(msg_utf16.len() as u16);
        for &c in &msg_utf16 {
            buf.put_u16_le(c);
        }

        // Server (B_VARCHAR: 1-byte char count + UTF-16LE)
        buf.put_u8(srv_utf16.len() as u8);
        for &c in &srv_utf16 {
            buf.put_u16_le(c);
        }

        // Procedure (B_VARCHAR: empty)
        buf.put_u8(proc_utf16.len() as u8);

        // Line number
        buf.put_i32_le(42);

        let mut cursor = buf.freeze();
        let error = ServerError::decode(&mut cursor).unwrap();

        assert_eq!(error.number, 207);
        assert_eq!(error.state, 1);
        assert_eq!(error.class, 16);
        assert_eq!(error.message, "Invalid column name 'foo'.");
        assert_eq!(error.server, "SQLDB01");
        assert_eq!(error.procedure, "");
        assert_eq!(error.line, 42);
    }

    #[test]
    fn test_server_error_severity_helpers() {
        let fatal = ServerError {
            number: 4014,
            state: 1,
            class: 20,
            message: "Fatal error".to_string(),
            server: String::new(),
            procedure: String::new(),
            line: 0,
        };
        assert!(fatal.is_fatal());
        assert!(fatal.is_batch_abort());

        let batch_abort = ServerError {
            number: 547,
            state: 0,
            class: 16,
            message: "Constraint violation".to_string(),
            server: String::new(),
            procedure: String::new(),
            line: 1,
        };
        assert!(!batch_abort.is_fatal());
        assert!(batch_abort.is_batch_abort());

        let informational = ServerError {
            number: 5701,
            state: 2,
            class: 10,
            message: "Changed db context".to_string(),
            server: String::new(),
            procedure: String::new(),
            line: 0,
        };
        assert!(!informational.is_fatal());
        assert!(!informational.is_batch_abort());
    }

    #[test]
    fn test_server_error_via_parser() {
        // Build an ERROR token with the 0xAA type byte for the parser
        let mut buf = BytesMut::new();
        buf.put_u8(0xAA); // ERROR token type

        let msg_utf16: Vec<u16> = "Syntax error".encode_utf16().collect();
        let srv_utf16: Vec<u16> = "SRV".encode_utf16().collect();
        let proc_utf16: Vec<u16> = "sp_test".encode_utf16().collect();

        let length: u16 = (4
            + 1
            + 1
            + 2
            + (msg_utf16.len() * 2)
            + 1
            + (srv_utf16.len() * 2)
            + 1
            + (proc_utf16.len() * 2)
            + 4) as u16;

        buf.put_u16_le(length);
        buf.put_i32_le(102); // Syntax error
        buf.put_u8(1);
        buf.put_u8(15);

        buf.put_u16_le(msg_utf16.len() as u16);
        for &c in &msg_utf16 {
            buf.put_u16_le(c);
        }
        buf.put_u8(srv_utf16.len() as u8);
        for &c in &srv_utf16 {
            buf.put_u16_le(c);
        }
        buf.put_u8(proc_utf16.len() as u8);
        for &c in &proc_utf16 {
            buf.put_u16_le(c);
        }
        buf.put_i32_le(5);

        let mut parser = TokenParser::new(buf.freeze());
        let token = parser.next_token().unwrap().unwrap();

        match token {
            Token::Error(err) => {
                assert_eq!(err.number, 102);
                assert_eq!(err.class, 15);
                assert_eq!(err.message, "Syntax error");
                assert_eq!(err.server, "SRV");
                assert_eq!(err.procedure, "sp_test");
                assert_eq!(err.line, 5);
            }
            _ => panic!("Expected Error token"),
        }
    }

    // ========================================================================
    // ReturnValue Token Tests
    // ========================================================================

    /// Helper: build a ReturnValue token (without the 0xAC type byte)
    /// for an IntN output parameter.
    fn build_return_value_intn(
        ordinal: u16,
        name: &str,
        status: u8,
        value: Option<i32>,
    ) -> BytesMut {
        let mut inner = BytesMut::new();

        // param_ordinal
        inner.put_u16_le(ordinal);

        // param_name (B_VARCHAR)
        let name_utf16: Vec<u16> = name.encode_utf16().collect();
        inner.put_u8(name_utf16.len() as u8);
        for &c in &name_utf16 {
            inner.put_u16_le(c);
        }

        // status
        inner.put_u8(status);

        // user_type (4 bytes)
        inner.put_u32_le(0);

        // flags (2 bytes)
        inner.put_u16_le(0x0001); // nullable

        // type_id: IntN = 0x26
        inner.put_u8(0x26);

        // type_info for IntN: 1-byte max_length
        inner.put_u8(4);

        // value (TYPE_VARBYTE for IntN: 1-byte length + data)
        match value {
            Some(v) => {
                inner.put_u8(4); // length = 4
                inner.put_i32_le(v);
            }
            None => {
                inner.put_u8(0); // length = 0 means NULL
            }
        }

        // RETURNVALUE has no outer length prefix (MS-TDS §2.2.7.18) — the
        // decoder walks the inner fields directly after the 0xAC token byte.
        inner
    }

    #[test]
    fn test_return_value_int_output() {
        let buf = build_return_value_intn(1, "@result", 0x01, Some(42));
        let mut cursor = buf.freeze();
        let rv = ReturnValue::decode(&mut cursor).unwrap();

        assert_eq!(rv.param_ordinal, 1);
        assert_eq!(rv.param_name, "@result");
        assert_eq!(rv.status, 0x01); // OUTPUT
        assert_eq!(rv.col_type, 0x26); // IntN
        assert_eq!(rv.type_info.max_length, Some(4));
        // Value should contain: length byte (4) + i32 LE (42)
        assert_eq!(rv.value.len(), 5);
        assert_eq!(rv.value[0], 4);
        assert_eq!(
            i32::from_le_bytes([rv.value[1], rv.value[2], rv.value[3], rv.value[4]]),
            42
        );
    }

    #[test]
    fn test_return_value_null_output() {
        let buf = build_return_value_intn(2, "@count", 0x01, None);
        let mut cursor = buf.freeze();
        let rv = ReturnValue::decode(&mut cursor).unwrap();

        assert_eq!(rv.param_ordinal, 2);
        assert_eq!(rv.param_name, "@count");
        assert_eq!(rv.status, 0x01);
        assert_eq!(rv.col_type, 0x26);
        // NULL value: length byte = 0
        assert_eq!(rv.value.len(), 1);
        assert_eq!(rv.value[0], 0);
    }

    #[test]
    fn test_return_value_udf_status() {
        // UDF return value has status = 0x02
        let buf = build_return_value_intn(0, "@RETURN_VALUE", 0x02, Some(-1));
        let mut cursor = buf.freeze();
        let rv = ReturnValue::decode(&mut cursor).unwrap();

        assert_eq!(rv.param_ordinal, 0);
        assert_eq!(rv.param_name, "@RETURN_VALUE");
        assert_eq!(rv.status, 0x02); // UDF return value
        assert_eq!(rv.value[0], 4);
        assert_eq!(
            i32::from_le_bytes([rv.value[1], rv.value[2], rv.value[3], rv.value[4]]),
            -1
        );
    }

    #[test]
    fn test_return_value_nvarchar_output() {
        // Build a ReturnValue for NVARCHAR(100) output parameter
        let mut inner = BytesMut::new();

        // param_ordinal
        inner.put_u16_le(1);

        // param_name "@name"
        let name_utf16: Vec<u16> = "@name".encode_utf16().collect();
        inner.put_u8(name_utf16.len() as u8);
        for &c in &name_utf16 {
            inner.put_u16_le(c);
        }

        // status = OUTPUT
        inner.put_u8(0x01);
        // user_type
        inner.put_u32_le(0);
        // flags (nullable)
        inner.put_u16_le(0x0001);
        // type_id: NVarChar = 0xE7
        inner.put_u8(0xE7);
        // type_info for NVarChar: 2-byte max_length + 5-byte collation
        inner.put_u16_le(200); // max 100 chars * 2 bytes
        inner.put_u32_le(0x0904D000); // collation LCID
        inner.put_u8(0x34); // collation sort_id

        // value: "Hello" in UTF-16LE with 2-byte length prefix
        let val_utf16: Vec<u16> = "Hello".encode_utf16().collect();
        let byte_len = (val_utf16.len() * 2) as u16;
        inner.put_u16_le(byte_len);
        for &c in &val_utf16 {
            inner.put_u16_le(c);
        }

        let mut cursor = inner.freeze();
        let rv = ReturnValue::decode(&mut cursor).unwrap();

        assert_eq!(rv.param_ordinal, 1);
        assert_eq!(rv.param_name, "@name");
        assert_eq!(rv.status, 0x01);
        assert_eq!(rv.col_type, 0xE7); // NVarChar
        assert_eq!(rv.type_info.max_length, Some(200));
        assert!(rv.type_info.collation.is_some());

        // Value: 2-byte length (10) + "Hello" in UTF-16LE
        assert_eq!(rv.value.len(), 12); // 2 + 10
        let val_len = u16::from_le_bytes([rv.value[0], rv.value[1]]);
        assert_eq!(val_len, 10);
    }

    #[test]
    fn test_return_value_via_parser() {
        // Build a full ReturnValue token with the 0xAC type byte
        let mut data = BytesMut::new();
        data.put_u8(0xAC); // RETURNVALUE token type
        data.extend_from_slice(&build_return_value_intn(0, "@out", 0x01, Some(99)));

        let mut parser = TokenParser::new(data.freeze());
        let token = parser.next_token().unwrap().unwrap();

        match token {
            Token::ReturnValue(rv) => {
                assert_eq!(rv.param_name, "@out");
                assert_eq!(rv.param_ordinal, 0);
                assert_eq!(rv.status, 0x01);
                assert_eq!(rv.col_type, 0x26);
            }
            _ => panic!("Expected ReturnValue token"),
        }
    }

    // ========================================================================
    // Multi-Token Stream Tests
    // ========================================================================

    #[test]
    fn test_multi_token_stored_proc_response() {
        // Simulate a stored procedure response:
        // DoneInProc (result set done) → ReturnStatus → DoneProc
        let mut data = BytesMut::new();

        // Token 1: DONEINPROC — result set with 3 rows
        data.put_u8(0xFF); // DONEINPROC
        data.put_u16_le(0x0010); // status: COUNT
        data.put_u16_le(0x00C1); // cur_cmd: SELECT
        data.put_u64_le(3); // row_count

        // Token 2: RETURNSTATUS — procedure returned 0
        data.put_u8(0x79); // RETURNSTATUS
        data.put_i32_le(0);

        // Token 3: DONEPROC — final
        data.put_u8(0xFE); // DONEPROC
        data.put_u16_le(0x0000); // status: no flags
        data.put_u16_le(0x00C6); // cur_cmd: EXECUTE
        data.put_u64_le(0);

        let mut parser = TokenParser::new(data.freeze());

        // Token 1: DoneInProc
        let t1 = parser.next_token().unwrap().unwrap();
        match t1 {
            Token::DoneInProc(done) => {
                assert!(done.status.count);
                assert_eq!(done.row_count, 3);
                assert_eq!(done.cur_cmd, 193);
            }
            _ => panic!("Expected DoneInProc, got {t1:?}"),
        }

        // Token 2: ReturnStatus
        let t2 = parser.next_token().unwrap().unwrap();
        match t2 {
            Token::ReturnStatus(status) => {
                assert_eq!(status, 0);
            }
            _ => panic!("Expected ReturnStatus, got {t2:?}"),
        }

        // Token 3: DoneProc
        let t3 = parser.next_token().unwrap().unwrap();
        match t3 {
            Token::DoneProc(done) => {
                assert!(!done.status.count);
                assert!(!done.status.more);
                assert_eq!(done.cur_cmd, 198);
            }
            _ => panic!("Expected DoneProc, got {t3:?}"),
        }

        // No more tokens
        assert!(parser.next_token().unwrap().is_none());
    }

    #[test]
    fn test_multi_token_error_in_stream() {
        // Simulate: ERROR → DONE (error during query)
        let mut data = BytesMut::new();

        // Token 1: ERROR
        data.put_u8(0xAA);

        let msg_utf16: Vec<u16> = "Deadlock".encode_utf16().collect();
        let srv_utf16: Vec<u16> = "DB1".encode_utf16().collect();

        let length: u16 = (4 + 1 + 1
            + 2 + (msg_utf16.len() * 2)
            + 1 + (srv_utf16.len() * 2)
            + 1  // empty procedure
            + 4) as u16;

        data.put_u16_le(length);
        data.put_i32_le(1205); // deadlock
        data.put_u8(51); // state
        data.put_u8(13); // class

        data.put_u16_le(msg_utf16.len() as u16);
        for &c in &msg_utf16 {
            data.put_u16_le(c);
        }
        data.put_u8(srv_utf16.len() as u8);
        for &c in &srv_utf16 {
            data.put_u16_le(c);
        }
        data.put_u8(0); // empty procedure
        data.put_i32_le(0);

        // Token 2: DONE with error flag
        data.put_u8(0xFD);
        data.put_u16_le(0x0002); // DONE_ERROR
        data.put_u16_le(0x00C1); // SELECT
        data.put_u64_le(0);

        let mut parser = TokenParser::new(data.freeze());

        // Token 1: Error
        let t1 = parser.next_token().unwrap().unwrap();
        match t1 {
            Token::Error(err) => {
                assert_eq!(err.number, 1205);
                assert_eq!(err.class, 13);
                assert_eq!(err.message, "Deadlock");
                assert_eq!(err.server, "DB1");
            }
            _ => panic!("Expected Error token, got {t1:?}"),
        }

        // Token 2: Done with error
        let t2 = parser.next_token().unwrap().unwrap();
        match t2 {
            Token::Done(done) => {
                assert!(done.status.error);
                assert!(!done.status.count);
            }
            _ => panic!("Expected Done token, got {t2:?}"),
        }

        assert!(parser.next_token().unwrap().is_none());
    }

    #[test]
    fn test_multi_token_proc_with_return_value() {
        // Simulate stored proc: ReturnValue → ReturnStatus → DoneProc
        let mut data = BytesMut::new();

        // Token 1: ReturnValue (@result = 42)
        data.put_u8(0xAC);
        data.extend_from_slice(&build_return_value_intn(1, "@result", 0x01, Some(42)));

        // Token 2: ReturnStatus = 0
        data.put_u8(0x79);
        data.put_i32_le(0);

        // Token 3: DoneProc
        data.put_u8(0xFE);
        data.put_u16_le(0x0000);
        data.put_u16_le(0x00C6);
        data.put_u64_le(0);

        let mut parser = TokenParser::new(data.freeze());

        let t1 = parser.next_token().unwrap().unwrap();
        match t1 {
            Token::ReturnValue(rv) => {
                assert_eq!(rv.param_name, "@result");
                assert_eq!(rv.param_ordinal, 1);
            }
            _ => panic!("Expected ReturnValue, got {t1:?}"),
        }

        let t2 = parser.next_token().unwrap().unwrap();
        assert!(matches!(t2, Token::ReturnStatus(0)));

        let t3 = parser.next_token().unwrap().unwrap();
        assert!(matches!(t3, Token::DoneProc(_)));

        assert!(parser.next_token().unwrap().is_none());
    }

    // ========================================================================
    // EOF / Truncation Edge Cases
    // ========================================================================

    #[test]
    fn test_return_status_truncated() {
        // Only 3 bytes instead of 4 for i32
        let data = Bytes::from_static(&[0x79, 0x01, 0x02, 0x03]);
        let mut parser = TokenParser::new(data);
        assert!(parser.next_token().is_err());
    }

    #[test]
    fn test_done_proc_truncated() {
        // Only 8 bytes instead of 12
        let data = Bytes::from_static(&[0xFE, 0x00, 0x00, 0xC1, 0x00, 0x01, 0x00, 0x00, 0x00]);
        let mut parser = TokenParser::new(data);
        assert!(parser.next_token().is_err());
    }

    #[test]
    fn test_server_error_truncated() {
        // ERROR token with only the length field (body truncated)
        let data = Bytes::from_static(&[0xAA, 0x20, 0x00]);
        let mut parser = TokenParser::new(data);
        assert!(parser.next_token().is_err());
    }

    // ========================================================================
    // FEDAUTHINFO (issue #189: parser must follow MS-TDS §2.2.7.12)
    // ========================================================================

    /// Build a spec-exact FEDAUTHINFO token (including the 0xEE type byte):
    /// DWORD TokenLength, DWORD CountOfInfoIDs, option headers of
    /// ID/DataLen/DataOffset, then UTF-16LE data addressed by the offsets
    /// (relative to the start of the count field).
    fn build_fed_auth_info_token(options: &[(u8, &str)]) -> Vec<u8> {
        let headers_end = 4 + options.len() * 9;
        let mut data_block = Vec::new();
        let mut headers = Vec::new();
        for (id, value) in options {
            let encoded: Vec<u8> = value.encode_utf16().flat_map(u16::to_le_bytes).collect();
            let offset = headers_end + data_block.len();
            headers.push(*id);
            headers.extend_from_slice(&u32::try_from(encoded.len()).unwrap().to_le_bytes());
            headers.extend_from_slice(&u32::try_from(offset).unwrap().to_le_bytes());
            data_block.extend_from_slice(&encoded);
        }

        let token_len = 4 + headers.len() + data_block.len();
        let mut out = vec![0xEE];
        out.extend_from_slice(&u32::try_from(token_len).unwrap().to_le_bytes());
        out.extend_from_slice(&u32::try_from(options.len()).unwrap().to_le_bytes());
        out.extend_from_slice(&headers);
        out.extend_from_slice(&data_block);
        out
    }

    #[test]
    fn test_fed_auth_info_decodes_spec_layout() {
        const STS: &str = "https://login.microsoftonline.com/common";
        const SPN: &str = "https://database.windows.net/";
        // STSURL (0x01) listed first, SPN (0x02) second. Real Azure servers
        // list SPN first (see the captured-token test below); decoding must
        // not depend on option order.
        let token = build_fed_auth_info_token(&[(0x01, STS), (0x02, SPN)]);

        let mut parser = TokenParser::new(Bytes::from(token));
        let parsed = parser.next_token().unwrap().unwrap();
        let Token::FedAuthInfo(info) = parsed else {
            panic!("expected FedAuthInfo, got {parsed:?}");
        };
        assert_eq!(info.sts_url, STS);
        assert_eq!(info.spn, SPN);
        assert!(parser.next_token().unwrap().is_none(), "exact consumption");
    }

    #[test]
    fn test_fed_auth_info_preserves_following_tokens() {
        // The old parser looped over the whole remaining stream, swallowing
        // the tokens that follow FEDAUTHINFO during login. A DONE token
        // appended after it must survive.
        let mut stream = build_fed_auth_info_token(&[
            (0x01, "https://sts.example/"),
            (0x02, "https://db.example/"),
        ]);
        stream.push(0xFD); // DONE
        stream.extend_from_slice(&0u16.to_le_bytes()); // status
        stream.extend_from_slice(&0u16.to_le_bytes()); // curcmd
        stream.extend_from_slice(&0u64.to_le_bytes()); // rowcount

        let mut parser = TokenParser::new(Bytes::from(stream));
        assert!(matches!(
            parser.next_token().unwrap(),
            Some(Token::FedAuthInfo(_))
        ));
        assert!(
            matches!(parser.next_token().unwrap(), Some(Token::Done(_))),
            "DONE after FEDAUTHINFO must not be swallowed"
        );
        assert!(parser.next_token().unwrap().is_none());
    }

    #[test]
    fn test_fed_auth_info_unknown_ids_ignored() {
        // Spec: unrecognized FedAuthInfoIDs must be ignored.
        let token =
            build_fed_auth_info_token(&[(0x7F, "ignore-me"), (0x01, "https://sts.example/")]);
        let mut parser = TokenParser::new(Bytes::from(token));
        let Some(Token::FedAuthInfo(info)) = parser.next_token().unwrap() else {
            panic!("expected FedAuthInfo");
        };
        assert_eq!(info.sts_url, "https://sts.example/");
        assert_eq!(info.spn, "");
    }

    #[test]
    fn test_fed_auth_info_hostile_inputs_error() {
        // TokenLength longer than the buffer.
        let mut truncated = build_fed_auth_info_token(&[(0x02, "https://sts.example/")]);
        truncated.truncate(truncated.len() - 4);
        assert!(
            TokenParser::new(Bytes::from(truncated))
                .next_token()
                .is_err()
        );

        // CountOfInfoIDs claims more headers than the token holds
        // (also covers hostile counts whose header math would overflow).
        let mut bad_count = build_fed_auth_info_token(&[(0x02, "https://sts.example/")]);
        bad_count[5..9].copy_from_slice(&u32::MAX.to_le_bytes());
        assert!(
            TokenParser::new(Bytes::from(bad_count))
                .next_token()
                .is_err()
        );

        // Data offset pointing past the end of the token.
        let mut bad_offset = build_fed_auth_info_token(&[(0x02, "https://sts.example/")]);
        bad_offset[14..18].copy_from_slice(&u32::MAX.to_le_bytes());
        assert!(
            TokenParser::new(Bytes::from(bad_offset))
                .next_token()
                .is_err()
        );

        // Odd data length cannot be UTF-16.
        let mut odd_len = build_fed_auth_info_token(&[(0x02, "https://sts.example/")]);
        odd_len[10..14].copy_from_slice(&3u32.to_le_bytes());
        assert!(TokenParser::new(Bytes::from(odd_len)).next_token().is_err());
    }

    #[test]
    fn test_fed_auth_info_parse_and_skip_agree() {
        // Issue #189: decode() and skip_token() must consume the same bytes
        // (the old decode ran past the token while skip honored the length).
        let token = build_fed_auth_info_token(&[(0x02, "https://sts.example/")]);
        let total = token.len();

        let mut parser = TokenParser::new(Bytes::from(token.clone()));
        parser.next_token().unwrap();
        assert_eq!(parser.position(), total, "decode consumption");

        let mut skipper = TokenParser::new(Bytes::from(token));
        skipper.skip_token().unwrap();
        assert_eq!(skipper.position(), total, "skip consumption");
    }

    /// A FEDAUTHINFO token captured from a live Azure SQL Database login on
    /// 2026-06-12 (the client declared the ADAL library in LOGIN7; the server
    /// responded with this token). The tenant GUID inside the STS URL is
    /// replaced with an all-zero GUID of identical length, so every offset
    /// and length is byte-identical to the wire capture.
    ///
    /// This is the regression test deferred from PR #193, and it earns its
    /// keep: the real token proves FedAuthInfoID 0x01 = STSURL and
    /// 0x02 = SPN (Azure lists SPN first), which the synthetic tests
    /// originally had swapped.
    #[test]
    fn test_fed_auth_info_captured_from_azure() {
        const CAPTURED: &[u8] = &[
            0xEE, 0xCC, 0x00, 0x00, 0x00, 0x02, 0x00, 0x00, 0x00, 0x02, 0x3A, 0x00, 0x00, 0x00,
            0x16, 0x00, 0x00, 0x00, 0x01, 0x7C, 0x00, 0x00, 0x00, 0x50, 0x00, 0x00, 0x00, 0x68,
            0x00, 0x74, 0x00, 0x74, 0x00, 0x70, 0x00, 0x73, 0x00, 0x3A, 0x00, 0x2F, 0x00, 0x2F,
            0x00, 0x64, 0x00, 0x61, 0x00, 0x74, 0x00, 0x61, 0x00, 0x62, 0x00, 0x61, 0x00, 0x73,
            0x00, 0x65, 0x00, 0x2E, 0x00, 0x77, 0x00, 0x69, 0x00, 0x6E, 0x00, 0x64, 0x00, 0x6F,
            0x00, 0x77, 0x00, 0x73, 0x00, 0x2E, 0x00, 0x6E, 0x00, 0x65, 0x00, 0x74, 0x00, 0x2F,
            0x00, 0x68, 0x00, 0x74, 0x00, 0x74, 0x00, 0x70, 0x00, 0x73, 0x00, 0x3A, 0x00, 0x2F,
            0x00, 0x2F, 0x00, 0x6C, 0x00, 0x6F, 0x00, 0x67, 0x00, 0x69, 0x00, 0x6E, 0x00, 0x2E,
            0x00, 0x77, 0x00, 0x69, 0x00, 0x6E, 0x00, 0x64, 0x00, 0x6F, 0x00, 0x77, 0x00, 0x73,
            0x00, 0x2E, 0x00, 0x6E, 0x00, 0x65, 0x00, 0x74, 0x00, 0x2F, 0x00, 0x30, 0x00, 0x30,
            0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x2D,
            0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x2D, 0x00, 0x30, 0x00, 0x30,
            0x00, 0x30, 0x00, 0x30, 0x00, 0x2D, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30,
            0x00, 0x2D, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30,
            0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00, 0x30, 0x00,
        ];

        let mut parser = TokenParser::new(Bytes::from_static(CAPTURED));
        let Some(Token::FedAuthInfo(info)) = parser.next_token().unwrap() else {
            panic!("expected FedAuthInfo");
        };
        assert_eq!(
            info.sts_url,
            "https://login.windows.net/00000000-0000-0000-0000-000000000000"
        );
        assert_eq!(info.spn, "https://database.windows.net/");
        assert!(
            parser.next_token().unwrap().is_none(),
            "the captured token must be consumed exactly"
        );
    }

    /// Regression test for #273 (remote DoS via unbounded token-skip recursion).
    ///
    /// COLINFO/TABNAME/OFFSET tokens are skipped, and the skip path used to
    /// self-recurse to fetch the next token — one real stack frame per skipped
    /// token (Rust guarantees no tail-call elimination). Because
    /// `Connection::read_message` reassembles every packet into one `Bytes`
    /// before tokenizing, a server-controlled multi-MB message of these 3-byte
    /// tokens drove ~10^5–10^6 frames and aborted the process.
    ///
    /// This feeds a flat run of 200_000 skip-tokens — far deeper than the
    /// (~2 MB) test-thread stack could survive by recursion — followed by a
    /// real DONE. Pre-fix, the recursive parser aborted (SIGABRT) on this
    /// input. Post-fix it must return that DONE with the whole buffer consumed,
    /// proving the skip path is bounded-stack and that the input was
    /// non-degenerate (all 200_000 tokens were actually traversed).
    #[test]
    fn skip_tokens_iterate_not_recurse_273() {
        const SKIP_COUNT: usize = 200_000;
        let mut buf = BytesMut::with_capacity(SKIP_COUNT * 3 + 13);
        for _ in 0..SKIP_COUNT {
            buf.put_u8(TokenType::ColInfo as u8);
            buf.put_u16_le(0); // zero-length body: 3 bytes total per skip-token
        }
        let done = Done {
            status: DoneStatus {
                more: false,
                error: false,
                in_xact: false,
                count: true,
                attn: false,
                srverror: false,
            },
            cur_cmd: 0xABCD,
            row_count: 99,
        };
        done.encode(&mut buf);
        let total_len = buf.len();

        let mut parser = TokenParser::new(buf.freeze());

        // A specific outcome — the DONE that follows the skip run — not a
        // generic is_err an unrelated parse failure could also satisfy.
        let Some(Token::Done(decoded)) = parser.next_token().unwrap() else {
            panic!("expected the DONE token after the skip run");
        };
        assert_eq!(decoded.cur_cmd, 0xABCD);
        assert_eq!(decoded.row_count, 99);

        // Every skipped token was traversed: proof the input was non-trivial.
        assert_eq!(parser.position(), total_len);
        assert!(parser.next_token().unwrap().is_none());
    }

    /// Build an ALTMETADATA token describing one `SUM` aggregate over an `int`
    /// column (empty name), followed by `rows` ALTROW tokens each carrying that
    /// int value. Mirrors the wire shape of a `COMPUTE BY` result set.
    fn compute_by_tokens(id: u16, rows: &[i32]) -> BytesMut {
        let mut buf = BytesMut::new();
        // ALTMETADATA
        buf.put_u8(TokenType::AltMetaData as u8);
        buf.put_u16_le(1); // Count = 1 ComputeData
        buf.put_u16_le(id); // Id
        buf.put_u8(1); // ByCols = 1
        buf.put_u16_le(1); // ColNum[0]
        buf.put_u8(0x4D); // Op = AOPSUM
        buf.put_u16_le(1); // Operand (column 1)
        buf.put_u32_le(0); // UserType (ULONG)
        buf.put_u16_le(0); // Flags
        buf.put_u8(TypeId::Int4 as u8); // TYPE_INFO (fixed-length int4)
        buf.put_u8(0); // ColName: B_VARCHAR length 0
        // ALTROW(s)
        for &v in rows {
            buf.put_u8(TokenType::AltRow as u8);
            buf.put_u16_le(id); // Id matching the ALTMETADATA
            buf.put_i32_le(v); // int4 value
        }
        buf
    }

    fn trailing_done(buf: &mut BytesMut) {
        let done = Done {
            status: DoneStatus {
                more: false,
                error: false,
                in_xact: false,
                count: true,
                attn: false,
                srverror: false,
            },
            cur_cmd: 0xBEEF,
            row_count: 3,
        };
        done.encode(buf);
    }

    /// #275: a COMPUTE BY result set (ALTMETADATA 0x88 + ALTROW 0xD3) must be
    /// silently consumed, not hard-error. The DONE following the compute tokens
    /// must still be delivered, and every byte traversed.
    #[test]
    fn compute_by_alt_tokens_skipped_275() {
        let mut buf = compute_by_tokens(7, &[42, -1]);
        trailing_done(&mut buf);
        let total_len = buf.len();

        let mut parser = TokenParser::new(buf.freeze());

        // The first token surfaced is the DONE — both ALTROWs and the
        // ALTMETADATA were dropped, not returned.
        let Some(Token::Done(done)) = parser.next_token().unwrap() else {
            panic!("expected DONE after the COMPUTE BY tokens were skipped");
        };
        assert_eq!(done.cur_cmd, 0xBEEF);

        // All alt-token bytes were consumed to reach the DONE.
        assert_eq!(parser.position(), total_len);
        assert!(parser.next_token().unwrap().is_none());
    }

    /// #275: ALTMETADATA must be remembered across calls so a following
    /// ColMetaData (new result set) clears stale compute Ids.
    #[test]
    fn compute_by_alt_metadata_cleared_on_new_colmetadata_275() {
        let mut buf = compute_by_tokens(7, &[42]);
        // New result set: a COLMETADATA token clears alt_metadata. Encode an
        // empty (NO_METADATA) ColMetaData, then reuse Id 7 for a fresh compute.
        buf.put_u8(TokenType::ColMetaData as u8);
        buf.put_u16_le(ColMetaData::NO_METADATA);
        buf.extend_from_slice(&compute_by_tokens(7, &[99]));
        trailing_done(&mut buf);
        let total_len = buf.len();

        let mut parser = TokenParser::new(buf.freeze());

        // First surfaced token: the empty ColMetaData (compute tokens dropped).
        let Some(Token::ColMetaData(_)) = parser.next_token().unwrap() else {
            panic!("expected ColMetaData between the two compute groups");
        };
        // Then the DONE, after the second compute group is also dropped.
        let Some(Token::Done(_)) = parser.next_token().unwrap() else {
            panic!("expected DONE after the second COMPUTE BY group");
        };
        assert_eq!(parser.position(), total_len);
    }

    /// #275: an ALTROW with no matching ALTMETADATA cannot be length-decoded;
    /// it must error cleanly rather than panic or desync.
    #[test]
    fn altrow_without_altmetadata_errors_275() {
        let mut buf = BytesMut::new();
        buf.put_u8(TokenType::AltRow as u8);
        buf.put_u16_le(7); // Id with no preceding ALTMETADATA
        buf.put_i32_le(42);

        let mut parser = TokenParser::new(buf.freeze());
        assert!(parser.next_token().is_err());
    }
}