tds_protocol/

rpc.rs

//! RPC (Remote Procedure Call) request encoding.
//!
//! This module provides encoding for RPC requests (packet type 0x03).
//! RPC is used for calling stored procedures and sp_executesql for parameterized queries.
//!
//! ## sp_executesql
//!
//! The primary use case is `sp_executesql` for parameterized queries, which prevents
//! SQL injection and enables query plan caching.
//!
//! ## Wire Format
//!
//! ```text
//! RPC Request:
//! +-------------------+
//! | ALL_HEADERS       | (TDS 7.2+, optional)
//! +-------------------+
//! | ProcName/ProcID   | (procedure identifier)
//! +-------------------+
//! | Option Flags      | (2 bytes)
//! +-------------------+
//! | Parameters        | (repeated)
//! +-------------------+
//! ```

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

use crate::codec::write_utf16_string;
use crate::crypto::EncryptionTypeWire;
use crate::prelude::*;
use crate::token::Collation;

/// Well-known stored procedure IDs.
///
/// These are special procedure IDs that SQL Server recognizes
/// without requiring the procedure name.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u16)]
#[non_exhaustive]
pub enum ProcId {
    /// sp_cursor (0x0001)
    Cursor = 0x0001,
    /// sp_cursoropen (0x0002)
    CursorOpen = 0x0002,
    /// sp_cursorprepare (0x0003)
    CursorPrepare = 0x0003,
    /// sp_cursorexecute (0x0004)
    CursorExecute = 0x0004,
    /// sp_cursorprepexec (0x0005)
    CursorPrepExec = 0x0005,
    /// sp_cursorunprepare (0x0006)
    CursorUnprepare = 0x0006,
    /// sp_cursorfetch (0x0007)
    CursorFetch = 0x0007,
    /// sp_cursoroption (0x0008)
    CursorOption = 0x0008,
    /// sp_cursorclose (0x0009)
    CursorClose = 0x0009,
    /// sp_executesql (0x000A) - Primary method for parameterized queries
    ExecuteSql = 0x000A,
    /// sp_prepare (0x000B)
    Prepare = 0x000B,
    /// sp_execute (0x000C)
    Execute = 0x000C,
    /// sp_prepexec (0x000D) - Prepare and execute in one call
    PrepExec = 0x000D,
    /// sp_prepexecrpc (0x000E)
    PrepExecRpc = 0x000E,
    /// sp_unprepare (0x000F)
    Unprepare = 0x000F,
}

/// RPC option flags.
#[derive(Debug, Clone, Copy, Default)]
pub struct RpcOptionFlags {
    /// Recompile the procedure.
    pub with_recompile: bool,
    /// No metadata in response.
    pub no_metadata: bool,
    /// Reuse metadata from previous call.
    pub reuse_metadata: bool,
}

impl RpcOptionFlags {
    /// Create new empty flags.
    pub fn new() -> Self {
        Self::default()
    }

    /// Set with recompile flag.
    #[must_use]
    pub fn with_recompile(mut self, value: bool) -> Self {
        self.with_recompile = value;
        self
    }

    /// Encode to wire format (2 bytes).
    pub fn encode(&self) -> u16 {
        let mut flags = 0u16;
        if self.with_recompile {
            flags |= 0x0001;
        }
        if self.no_metadata {
            flags |= 0x0002;
        }
        if self.reuse_metadata {
            flags |= 0x0004;
        }
        flags
    }
}

/// RPC parameter status flags.
#[derive(Debug, Clone, Copy, Default)]
pub struct ParamFlags {
    /// Parameter is passed by reference (OUTPUT parameter).
    pub by_ref: bool,
    /// Parameter has a default value.
    pub default: bool,
    /// Parameter is encrypted (Always Encrypted).
    pub encrypted: bool,
}

impl ParamFlags {
    /// Create new empty flags.
    pub fn new() -> Self {
        Self::default()
    }

    /// Set as output parameter.
    #[must_use]
    pub fn output(mut self) -> Self {
        self.by_ref = true;
        self
    }

    /// Encode to wire format (1 byte).
    pub fn encode(&self) -> u8 {
        let mut flags = 0u8;
        if self.by_ref {
            flags |= 0x01;
        }
        if self.default {
            flags |= 0x02;
        }
        if self.encrypted {
            flags |= 0x08;
        }
        flags
    }
}

/// TDS type information for RPC parameters.
#[derive(Debug, Clone)]
pub struct TypeInfo {
    /// Type ID.
    pub type_id: u8,
    /// Maximum length for variable-length types.
    pub max_length: Option<u16>,
    /// Precision for numeric types.
    pub precision: Option<u8>,
    /// Scale for numeric types.
    pub scale: Option<u8>,
    /// Collation for string types.
    pub collation: Option<[u8; 5]>,
    /// TVP type name (e.g., "dbo.IntIdList") for Table-Valued Parameters.
    pub tvp_type_name: Option<String>,
}

impl TypeInfo {
    /// Create type info for INT.
    pub fn int() -> Self {
        Self {
            type_id: 0x26, // INTNTYPE (variable-length int)
            max_length: Some(4),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for BIGINT.
    pub fn bigint() -> Self {
        Self {
            type_id: 0x26, // INTNTYPE
            max_length: Some(8),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for SMALLINT.
    pub fn smallint() -> Self {
        Self {
            type_id: 0x26, // INTNTYPE
            max_length: Some(2),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for TINYINT.
    pub fn tinyint() -> Self {
        Self {
            type_id: 0x26, // INTNTYPE
            max_length: Some(1),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for BIT.
    pub fn bit() -> Self {
        Self {
            type_id: 0x68, // BITNTYPE
            max_length: Some(1),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for FLOAT.
    pub fn float() -> Self {
        Self {
            type_id: 0x6D, // FLTNTYPE
            max_length: Some(8),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for REAL.
    pub fn real() -> Self {
        Self {
            type_id: 0x6D, // FLTNTYPE
            max_length: Some(4),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for NVARCHAR with max length.
    pub fn nvarchar(max_len: u16) -> Self {
        Self {
            type_id: 0xE7,                 // NVARCHARTYPE
            max_length: Some(max_len * 2), // UTF-16, so double the char count
            precision: None,
            scale: None,
            // Default collation (Latin1_General_CI_AS equivalent)
            collation: Some([0x09, 0x04, 0xD0, 0x00, 0x34]),
            tvp_type_name: None,
        }
    }

    /// Create type info for NVARCHAR(MAX).
    pub fn nvarchar_max() -> Self {
        Self {
            type_id: 0xE7,            // NVARCHARTYPE
            max_length: Some(0xFFFF), // MAX indicator
            precision: None,
            scale: None,
            collation: Some([0x09, 0x04, 0xD0, 0x00, 0x34]),
            tvp_type_name: None,
        }
    }

    /// Default collation bytes: Latin1_General_CI_AS (LCID 0x0409, sort ID 0x34).
    const DEFAULT_COLLATION: [u8; 5] = [0x09, 0x04, 0xD0, 0x00, 0x34];

    /// Create type info for VARCHAR with max length (in bytes).
    pub fn varchar(max_len: u16) -> Self {
        Self::varchar_with_collation(max_len, Self::DEFAULT_COLLATION)
    }

    /// Create type info for VARCHAR with max length and explicit collation.
    pub fn varchar_with_collation(max_len: u16, collation: [u8; 5]) -> Self {
        Self {
            type_id: 0xA7, // BIGVARCHARTYPE
            max_length: Some(max_len),
            precision: None,
            scale: None,
            collation: Some(collation),
            tvp_type_name: None,
        }
    }

    /// Create type info for VARCHAR(MAX).
    pub fn varchar_max() -> Self {
        Self::varchar_max_with_collation(Self::DEFAULT_COLLATION)
    }

    /// Create type info for VARCHAR(MAX) with explicit collation.
    pub fn varchar_max_with_collation(collation: [u8; 5]) -> Self {
        Self {
            type_id: 0xA7,            // BIGVARCHARTYPE
            max_length: Some(0xFFFF), // MAX indicator
            precision: None,
            scale: None,
            collation: Some(collation),
            tvp_type_name: None,
        }
    }

    /// Create type info for VARBINARY with max length.
    pub fn varbinary(max_len: u16) -> Self {
        Self {
            type_id: 0xA5, // BIGVARBINTYPE
            max_length: Some(max_len),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for VARBINARY(MAX).
    pub fn varbinary_max() -> Self {
        Self {
            type_id: 0xA5,            // BIGVARBINTYPE
            max_length: Some(0xFFFF), // MAX indicator — triggers PLP encoding
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for UNIQUEIDENTIFIER.
    pub fn uniqueidentifier() -> Self {
        Self {
            type_id: 0x24, // GUIDTYPE
            max_length: Some(16),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for UNIQUEIDENTIFIER.
    pub fn uuid() -> Self {
        Self {
            type_id: 0x24, // GUIDTYPE
            max_length: Some(16),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for DATE.
    pub fn date() -> Self {
        Self {
            type_id: 0x28, // DATETYPE
            max_length: None,
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for TIME.
    pub fn time(scale: u8) -> Self {
        Self {
            type_id: 0x29, // TIMETYPE
            max_length: None,
            precision: None,
            scale: Some(scale),
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for DATETIME2.
    pub fn datetime2(scale: u8) -> Self {
        Self {
            type_id: 0x2A, // DATETIME2TYPE
            max_length: None,
            precision: None,
            scale: Some(scale),
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for DATETIMEOFFSET.
    pub fn datetimeoffset(scale: u8) -> Self {
        Self {
            type_id: 0x2B, // DATETIMEOFFSETTYPE
            max_length: None,
            precision: None,
            scale: Some(scale),
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for DECIMAL.
    pub fn decimal(precision: u8, scale: u8) -> Self {
        Self {
            type_id: 0x6C,        // DECIMALNTYPE
            max_length: Some(17), // Max decimal size
            precision: Some(precision),
            scale: Some(scale),
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for MONEY (8-byte scaled integer via MONEYN / 0x6E).
    pub fn money() -> Self {
        Self {
            type_id: 0x6E, // MONEYNTYPE
            max_length: Some(8),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for SMALLMONEY (4-byte scaled integer via MONEYN / 0x6E).
    pub fn smallmoney() -> Self {
        Self {
            type_id: 0x6E, // MONEYNTYPE
            max_length: Some(4),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for SMALLDATETIME (4-byte days+minutes via DATETIMEN / 0x6F).
    pub fn smalldatetime() -> Self {
        Self {
            type_id: 0x6F, // DATETIMENTYPE
            max_length: Some(4),
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: None,
        }
    }

    /// Create type info for a Table-Valued Parameter.
    ///
    /// # Arguments
    /// * `type_name` - The fully qualified table type name (e.g., "dbo.IntIdList")
    pub fn tvp(type_name: impl Into<String>) -> Self {
        Self {
            type_id: 0xF3, // TVP type
            max_length: None,
            precision: None,
            scale: None,
            collation: None,
            tvp_type_name: Some(type_name.into()),
        }
    }

    /// Encode type info to buffer.
    pub fn encode(&self, buf: &mut BytesMut) {
        // TVP (0xF3) has type_id embedded in the value data itself
        // (written by TvpEncoder::encode_metadata), so don't write it here
        if self.type_id != 0xF3 {
            buf.put_u8(self.type_id);
        }

        // Variable-length types need max length
        match self.type_id {
            0x26 | 0x68 | 0x6D | 0x6E | 0x6F => {
                // INTNTYPE, BITNTYPE, FLTNTYPE, MONEYNTYPE, DATETIMENTYPE
                if let Some(len) = self.max_length {
                    buf.put_u8(len as u8);
                }
            }
            0xE7 | 0xA7 | 0xA5 | 0xEF => {
                // NVARCHARTYPE, BIGVARCHARTYPE, BIGVARBINTYPE, NCHARTYPE
                if let Some(len) = self.max_length {
                    buf.put_u16_le(len);
                }
                // Collation for string types
                if let Some(collation) = self.collation {
                    buf.put_slice(&collation);
                }
            }
            0x24 => {
                // GUIDTYPE
                if let Some(len) = self.max_length {
                    buf.put_u8(len as u8);
                }
            }
            0x29..=0x2B => {
                // DATETIME2TYPE, TIMETYPE, DATETIMEOFFSETTYPE
                if let Some(scale) = self.scale {
                    buf.put_u8(scale);
                }
            }
            0x6C | 0x6A => {
                // DECIMALNTYPE, NUMERICNTYPE
                if let Some(len) = self.max_length {
                    buf.put_u8(len as u8);
                }
                if let Some(precision) = self.precision {
                    buf.put_u8(precision);
                }
                if let Some(scale) = self.scale {
                    buf.put_u8(scale);
                }
            }
            _ => {}
        }
    }
}

/// Always Encrypted cipher metadata, written after an encrypted parameter's
/// ciphertext value (MS-TDS 2.2.6.6 `CryptoMetadata`).
///
/// It tells the server how to validate and route the encrypted value: the
/// plaintext column type (`BaseTypeInfo`), the AEAD algorithm and encryption
/// mode, which column encryption key, and the normalization rule version.
#[derive(Debug, Clone)]
pub struct EncryptedParamMetadata {
    /// Type info of the plaintext column (the `BaseTypeInfo`).
    pub base_type_info: TypeInfo,
    /// Encryption algorithm ID (2 = AEAD_AES_256_CBC_HMAC_SHA256).
    pub algorithm_id: u8,
    /// Deterministic or randomized encryption.
    pub encryption_type: EncryptionTypeWire,
    /// Database ID of the column encryption key.
    pub database_id: u32,
    /// Column encryption key ID.
    pub cek_id: u32,
    /// Column encryption key version.
    pub cek_version: u32,
    /// Column encryption key metadata version.
    pub cek_md_version: u64,
    /// Normalization rule version applied to the plaintext.
    pub normalization_rule_version: u8,
}

impl EncryptedParamMetadata {
    /// Encode the cipher-metadata trailer to the buffer, in the order the
    /// server expects it after the ciphertext value.
    pub fn encode(&self, buf: &mut BytesMut) {
        self.base_type_info.encode(buf);
        buf.put_u8(self.algorithm_id);
        buf.put_u8(self.encryption_type.to_u8());
        buf.put_u32_le(self.database_id);
        buf.put_u32_le(self.cek_id);
        buf.put_u32_le(self.cek_version);
        buf.put_u64_le(self.cek_md_version);
        buf.put_u8(self.normalization_rule_version);
    }
}

/// An RPC parameter.
#[derive(Debug, Clone)]
pub struct RpcParam {
    /// Parameter name (can be empty for positional params).
    pub name: String,
    /// Status flags.
    pub flags: ParamFlags,
    /// Type information.
    pub type_info: TypeInfo,
    /// Parameter value (raw bytes).
    pub value: Option<Bytes>,
    /// Always Encrypted cipher metadata, written after the value when the
    /// parameter is encrypted. `None` for ordinary parameters.
    pub crypto_metadata: Option<EncryptedParamMetadata>,
}

impl RpcParam {
    /// Create a new parameter with a value.
    pub fn new(name: impl Into<String>, type_info: TypeInfo, value: Bytes) -> Self {
        Self {
            name: name.into(),
            flags: ParamFlags::default(),
            type_info,
            value: Some(value),
            crypto_metadata: None,
        }
    }

    /// Create a NULL parameter.
    pub fn null(name: impl Into<String>, type_info: TypeInfo) -> Self {
        Self {
            name: name.into(),
            flags: ParamFlags::default(),
            type_info,
            value: None,
            crypto_metadata: None,
        }
    }

    /// Create an Always Encrypted parameter.
    ///
    /// `ciphertext` is the AEAD-encrypted, normalized value; `metadata` is the
    /// cipher info the server needs to validate and route it. On the wire the
    /// value is carried as `BIGVARBINARY(max)` with the `fEncrypted` status bit
    /// set and the [`EncryptedParamMetadata`] trailer after the value.
    pub fn encrypted(
        name: impl Into<String>,
        ciphertext: Bytes,
        metadata: EncryptedParamMetadata,
    ) -> Self {
        Self {
            name: name.into(),
            flags: ParamFlags {
                encrypted: true,
                ..ParamFlags::default()
            },
            type_info: TypeInfo::varbinary_max(),
            value: Some(ciphertext),
            crypto_metadata: Some(metadata),
        }
    }

    /// Create a NULL Always Encrypted parameter.
    ///
    /// The server rejects a plaintext parameter bound to an encrypted column,
    /// even for NULL, so a NULL value is still sent encrypted: `BIGVARBINARY(max)`
    /// with the `fEncrypted` status bit, a NULL value, and the cipher metadata.
    pub fn encrypted_null(name: impl Into<String>, metadata: EncryptedParamMetadata) -> Self {
        Self {
            name: name.into(),
            flags: ParamFlags {
                encrypted: true,
                ..ParamFlags::default()
            },
            type_info: TypeInfo::varbinary_max(),
            value: None,
            crypto_metadata: Some(metadata),
        }
    }

    /// Create an INT parameter.
    pub fn int(name: impl Into<String>, value: i32) -> Self {
        let mut buf = BytesMut::with_capacity(4);
        buf.put_i32_le(value);
        Self::new(name, TypeInfo::int(), buf.freeze())
    }

    /// Create a BIGINT parameter.
    pub fn bigint(name: impl Into<String>, value: i64) -> Self {
        let mut buf = BytesMut::with_capacity(8);
        buf.put_i64_le(value);
        Self::new(name, TypeInfo::bigint(), buf.freeze())
    }

    /// Create an NVARCHAR parameter.
    pub fn nvarchar(name: impl Into<String>, value: &str) -> Self {
        let mut buf = BytesMut::new();
        let mut code_units: usize = 0;
        for code_unit in value.encode_utf16() {
            buf.put_u16_le(code_unit);
            code_units += 1;
        }
        // NVARCHAR length is measured in UTF-16 code units, not Rust chars —
        // supplementary characters (emoji, CJK extension B) encode to a surrogate
        // pair (2 code units) but count as 1 `char`. Using chars().count() here
        // under-reports the buffer length and the server rejects the RPC with
        // "Data type 0xE7 has an invalid data length or metadata length."
        let type_info = if code_units > 4000 {
            TypeInfo::nvarchar_max()
        } else {
            TypeInfo::nvarchar(code_units.max(1) as u16)
        };
        Self::new(name, type_info, buf.freeze())
    }

    /// Create a VARCHAR parameter.
    ///
    /// Encodes the string as single-byte characters using Windows-1252 encoding
    /// (when the `encoding` feature is enabled) or Latin-1 fallback. Characters
    /// not representable in the target encoding are replaced with `?`.
    ///
    /// Use this instead of [`nvarchar`](Self::nvarchar) when
    /// `SendStringParametersAsUnicode=false` to allow SQL Server to use
    /// index seeks on VARCHAR columns.
    pub fn varchar(name: impl Into<String>, value: &str) -> Self {
        let encoded = Self::encode_varchar_bytes(value);
        let byte_len = encoded.len();
        let type_info = if byte_len > 8000 {
            TypeInfo::varchar_max()
        } else {
            TypeInfo::varchar(byte_len.max(1) as u16)
        };
        Self::new(name, type_info, Bytes::from(encoded))
    }

    /// Encode a string as single-byte VARCHAR data using the default
    /// Windows-1252 encoding (or Latin-1 fallback without the `encoding` feature).
    fn encode_varchar_bytes(value: &str) -> Vec<u8> {
        crate::collation::encode_str_for_collation(value, None)
    }

    /// Create a VARCHAR parameter using the server's collation for encoding.
    ///
    /// Uses the collation's character encoding instead of the default Windows-1252.
    /// For UTF-8 collations (SQL Server 2019+), the string bytes are used directly.
    pub fn varchar_with_collation(
        name: impl Into<String>,
        value: &str,
        collation: &Collation,
    ) -> Self {
        let collation_bytes = collation.to_bytes();
        let encoded = Self::encode_varchar_bytes_for_collation(value, collation);
        let byte_len = encoded.len();
        let type_info = if byte_len > 8000 {
            TypeInfo::varchar_max_with_collation(collation_bytes)
        } else {
            TypeInfo::varchar_with_collation(byte_len.max(1) as u16, collation_bytes)
        };
        Self::new(name, type_info, Bytes::from(encoded))
    }

    /// Encode a string using the collation's character encoding.
    fn encode_varchar_bytes_for_collation(value: &str, collation: &Collation) -> Vec<u8> {
        crate::collation::encode_str_for_collation(value, Some(collation))
    }

    /// Mark as output parameter.
    #[must_use]
    pub fn as_output(mut self) -> Self {
        self.flags = self.flags.output();
        self
    }

    /// Encode the parameter to buffer.
    pub fn encode(&self, buf: &mut BytesMut) {
        // Parameter name (B_VARCHAR - length-prefixed)
        let name_len = self.name.encode_utf16().count() as u8;
        buf.put_u8(name_len);
        if name_len > 0 {
            for code_unit in self.name.encode_utf16() {
                buf.put_u16_le(code_unit);
            }
        }

        // Status flags
        buf.put_u8(self.flags.encode());

        // Type info
        self.type_info.encode(buf);

        // Value
        if let Some(ref value) = self.value {
            // Length prefix based on type
            match self.type_info.type_id {
                0x26 => {
                    // INTNTYPE
                    buf.put_u8(value.len() as u8);
                    buf.put_slice(value);
                }
                0x68 | 0x6D | 0x6E | 0x6F => {
                    // BITNTYPE, FLTNTYPE, MONEYNTYPE, DATETIMENTYPE
                    buf.put_u8(value.len() as u8);
                    buf.put_slice(value);
                }
                0xE7 | 0xA7 | 0xA5 => {
                    // NVARCHARTYPE, BIGVARCHARTYPE, BIGVARBINTYPE
                    if self.type_info.max_length == Some(0xFFFF) {
                        // MAX type - use PLP format
                        // For simplicity, send as single chunk
                        let total_len = value.len() as u64;
                        buf.put_u64_le(total_len);
                        buf.put_u32_le(value.len() as u32);
                        buf.put_slice(value);
                        buf.put_u32_le(0); // Terminator
                    } else {
                        buf.put_u16_le(value.len() as u16);
                        buf.put_slice(value);
                    }
                }
                0x24 => {
                    // GUIDTYPE
                    buf.put_u8(value.len() as u8);
                    buf.put_slice(value);
                }
                0x28..=0x2B => {
                    // DATE, TIME, DATETIME2, DATETIMEOFFSET
                    buf.put_u8(value.len() as u8);
                    buf.put_slice(value);
                }
                0x6C => {
                    // DECIMALNTYPE
                    buf.put_u8(value.len() as u8);
                    buf.put_slice(value);
                }
                0xF3 => {
                    // TVP (Table-Valued Parameter)
                    // TVP values are self-delimiting: they contain complete metadata,
                    // row data, and end token (TVP_END_TOKEN = 0x00). No length prefix.
                    buf.put_slice(value);
                }
                _ => {
                    // Generic: assume length-prefixed
                    buf.put_u8(value.len() as u8);
                    buf.put_slice(value);
                }
            }
        } else {
            // NULL value
            match self.type_info.type_id {
                0xE7 | 0xA7 | 0xA5 => {
                    // Variable-length types use 0xFFFF for NULL
                    if self.type_info.max_length == Some(0xFFFF) {
                        buf.put_u64_le(0xFFFFFFFFFFFFFFFF); // PLP NULL
                    } else {
                        buf.put_u16_le(0xFFFF);
                    }
                }
                _ => {
                    buf.put_u8(0); // Zero-length for NULL
                }
            }
        }

        // Always Encrypted: the CryptoMetadata trailer follows the value.
        if let Some(ref metadata) = self.crypto_metadata {
            metadata.encode(buf);
        }
    }
}

/// RPC request builder.
#[derive(Debug, Clone)]
pub struct RpcRequest {
    /// Procedure name (if using named procedure).
    proc_name: Option<String>,
    /// Procedure ID (if using well-known procedure).
    proc_id: Option<ProcId>,
    /// Option flags.
    options: RpcOptionFlags,
    /// Parameters.
    params: Vec<RpcParam>,
}

impl RpcRequest {
    /// Create a new RPC request for a named procedure.
    pub fn named(proc_name: impl Into<String>) -> Self {
        Self {
            proc_name: Some(proc_name.into()),
            proc_id: None,
            options: RpcOptionFlags::default(),
            params: Vec::new(),
        }
    }

    /// Create a new RPC request for a well-known procedure.
    pub fn by_id(proc_id: ProcId) -> Self {
        Self {
            proc_name: None,
            proc_id: Some(proc_id),
            options: RpcOptionFlags::default(),
            params: Vec::new(),
        }
    }

    /// Create an sp_executesql request.
    ///
    /// This is the primary method for parameterized queries.
    ///
    /// # Example
    ///
    /// ```
    /// use tds_protocol::rpc::{RpcRequest, RpcParam};
    ///
    /// let rpc = RpcRequest::execute_sql(
    ///     "SELECT * FROM users WHERE id = @p1 AND name = @p2",
    ///     vec![
    ///         RpcParam::int("@p1", 42),
    ///         RpcParam::nvarchar("@p2", "Alice"),
    ///     ],
    /// );
    /// ```
    pub fn execute_sql(sql: &str, params: Vec<RpcParam>) -> Self {
        let mut request = Self::by_id(ProcId::ExecuteSql);

        // First parameter: the SQL statement (NVARCHAR(MAX))
        request.params.push(RpcParam::nvarchar("", sql));

        // Second parameter: parameter declarations
        if !params.is_empty() {
            let declarations = Self::build_param_declarations(&params);
            request.params.push(RpcParam::nvarchar("", &declarations));
        }

        // Add the actual parameters
        request.params.extend(params);

        request
    }

    /// Build parameter declaration string for sp_executesql.
    /// Build the `sp_executesql` `@params` declaration string for `params`.
    ///
    /// An Always Encrypted parameter declares its plaintext column type (its
    /// [`EncryptedParamMetadata::base_type_info`]), not the `BIGVARBINARY`
    /// transport type its value is carried as, so the declaration matches what
    /// `sp_describe_parameter_encryption` was asked about.
    pub fn build_param_declarations(params: &[RpcParam]) -> String {
        params
            .iter()
            .map(|p| {
                let name = if p.name.starts_with('@') {
                    p.name.clone()
                } else if p.name.is_empty() {
                    // Generate positional name
                    format!(
                        "@p{}",
                        params.iter().position(|x| x.name == p.name).unwrap_or(0) + 1
                    )
                } else {
                    format!("@{}", p.name)
                };

                // Encrypted parameters declare their plaintext type, not the
                // BIGVARBINARY their ciphertext rides in.
                let ti = p
                    .crypto_metadata
                    .as_ref()
                    .map(|m| &m.base_type_info)
                    .unwrap_or(&p.type_info);

                let type_name: String = match ti.type_id {
                    0x26 => match ti.max_length {
                        Some(1) => "tinyint".to_string(),
                        Some(2) => "smallint".to_string(),
                        Some(4) => "int".to_string(),
                        Some(8) => "bigint".to_string(),
                        _ => "int".to_string(),
                    },
                    0x68 => "bit".to_string(),
                    0x6D => match ti.max_length {
                        Some(4) => "real".to_string(),
                        _ => "float".to_string(),
                    },
                    0xE7 => {
                        if ti.max_length == Some(0xFFFF) {
                            "nvarchar(max)".to_string()
                        } else {
                            let len = ti.max_length.unwrap_or(4000) / 2;
                            format!("nvarchar({len})")
                        }
                    }
                    0xA7 => {
                        if ti.max_length == Some(0xFFFF) {
                            "varchar(max)".to_string()
                        } else {
                            let len = ti.max_length.unwrap_or(8000);
                            format!("varchar({len})")
                        }
                    }
                    0xA5 => {
                        if ti.max_length == Some(0xFFFF) {
                            "varbinary(max)".to_string()
                        } else {
                            let len = ti.max_length.unwrap_or(8000);
                            format!("varbinary({len})")
                        }
                    }
                    0x24 => "uniqueidentifier".to_string(),
                    0x28 => "date".to_string(),
                    0x29 => {
                        let scale = ti.scale.unwrap_or(7);
                        format!("time({scale})")
                    }
                    0x2A => {
                        let scale = ti.scale.unwrap_or(7);
                        format!("datetime2({scale})")
                    }
                    0x2B => {
                        let scale = ti.scale.unwrap_or(7);
                        format!("datetimeoffset({scale})")
                    }
                    0x6C => {
                        let precision = ti.precision.unwrap_or(18);
                        let scale = ti.scale.unwrap_or(0);
                        format!("decimal({precision}, {scale})")
                    }
                    0x6E => match ti.max_length {
                        Some(4) => "smallmoney".to_string(),
                        _ => "money".to_string(),
                    },
                    0x6F => match ti.max_length {
                        Some(4) => "smalldatetime".to_string(),
                        _ => "datetime".to_string(),
                    },
                    0xF3 => {
                        // TVP - Table-Valued Parameter
                        // Must be declared with the table type name and READONLY
                        if let Some(ref tvp_name) = ti.tvp_type_name {
                            format!("{tvp_name} READONLY")
                        } else {
                            // Fallback if type name is missing (shouldn't happen)
                            "sql_variant".to_string()
                        }
                    }
                    _ => "sql_variant".to_string(),
                };

                format!("{name} {type_name}")
            })
            .collect::<Vec<_>>()
            .join(", ")
    }

    /// Create an sp_prepare request.
    pub fn prepare(sql: &str, params: &[RpcParam]) -> Self {
        let mut request = Self::by_id(ProcId::Prepare);

        // OUT: handle (INT)
        request
            .params
            .push(RpcParam::null("@handle", TypeInfo::int()).as_output());

        // Param declarations
        let declarations = Self::build_param_declarations(params);
        request
            .params
            .push(RpcParam::nvarchar("@params", &declarations));

        // SQL statement
        request.params.push(RpcParam::nvarchar("@stmt", sql));

        // Options (1 = WITH RECOMPILE)
        request.params.push(RpcParam::int("@options", 1));

        request
    }

    /// Create an sp_execute request.
    pub fn execute(handle: i32, params: Vec<RpcParam>) -> Self {
        let mut request = Self::by_id(ProcId::Execute);

        // Handle from sp_prepare
        request.params.push(RpcParam::int("@handle", handle));

        // Add parameters
        request.params.extend(params);

        request
    }

    /// Create an sp_unprepare request.
    pub fn unprepare(handle: i32) -> Self {
        let mut request = Self::by_id(ProcId::Unprepare);
        request.params.push(RpcParam::int("@handle", handle));
        request
    }

    /// Set option flags.
    #[must_use]
    pub fn with_options(mut self, options: RpcOptionFlags) -> Self {
        self.options = options;
        self
    }

    /// Add a parameter.
    #[must_use]
    pub fn param(mut self, param: RpcParam) -> Self {
        self.params.push(param);
        self
    }

    /// Encode the RPC request to bytes (auto-commit mode).
    ///
    /// For requests within an explicit transaction, use [`Self::encode_with_transaction`].
    #[must_use]
    pub fn encode(&self) -> Bytes {
        self.encode_with_transaction(0)
    }

    /// Encode the RPC request with a transaction descriptor.
    ///
    /// Per MS-TDS spec, when executing within an explicit transaction:
    /// - The `transaction_descriptor` MUST be the value returned by the server
    ///   in the BeginTransaction EnvChange token.
    /// - For auto-commit mode (no explicit transaction), use 0.
    ///
    /// # Arguments
    ///
    /// * `transaction_descriptor` - The transaction descriptor from BeginTransaction EnvChange,
    ///   or 0 for auto-commit mode.
    #[must_use]
    pub fn encode_with_transaction(&self, transaction_descriptor: u64) -> Bytes {
        let mut buf = BytesMut::with_capacity(256);

        // ALL_HEADERS - TDS 7.2+ requires this section
        // Total length placeholder (will be filled in)
        let all_headers_start = buf.len();
        buf.put_u32_le(0); // Total length placeholder

        // Transaction descriptor header (required for RPC)
        // Per MS-TDS 2.2.5.3: HeaderLength (4) + HeaderType (2) + TransactionDescriptor (8) + OutstandingRequestCount (4)
        buf.put_u32_le(18); // Header length
        buf.put_u16_le(0x0002); // Header type: transaction descriptor
        buf.put_u64_le(transaction_descriptor); // Transaction descriptor from BeginTransaction EnvChange
        buf.put_u32_le(1); // Outstanding request count (1 for non-MARS connections)

        // Fill in ALL_HEADERS total length
        let all_headers_len = buf.len() - all_headers_start;
        let len_bytes = (all_headers_len as u32).to_le_bytes();
        buf[all_headers_start..all_headers_start + 4].copy_from_slice(&len_bytes);

        // Procedure name or ID
        if let Some(proc_id) = self.proc_id {
            // Use PROCID format
            buf.put_u16_le(0xFFFF); // Name length = 0xFFFF indicates PROCID follows
            buf.put_u16_le(proc_id as u16);
        } else if let Some(ref proc_name) = self.proc_name {
            // Use procedure name
            let name_len = proc_name.encode_utf16().count() as u16;
            buf.put_u16_le(name_len);
            write_utf16_string(&mut buf, proc_name);
        }

        // Option flags
        buf.put_u16_le(self.options.encode());

        // Parameters
        for param in &self.params {
            param.encode(&mut buf);
        }

        buf.freeze()
    }
}

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

    /// The encrypted-RPC-param wire format must match `Microsoft.Data.SqlClient`
    /// byte-for-byte. Goldens were captured from a live deterministic Always
    /// Encrypted INSERT (`.tmp/ae-wire-encrypted-param.md`): one INT, one
    /// NVARCHAR(50), and one VARBINARY(50) parameter, all bound to a single CEK
    /// (db_id=6, cek_id=2, cek_version=1, md_version=0x0000b469002f7223). The
    /// ciphertext is sliced out of each golden and fed back through the encoder,
    /// so a match proves the full framing (fEncrypted status, BIGVARBINARY(max)
    /// PLP value, and the CryptoMetadata trailer) reproduces the real client.
    #[test]
    fn encrypted_param_encode_matches_captured_dotnet_wire() {
        fn unhex(s: &str) -> Vec<u8> {
            (0..s.len())
                .step_by(2)
                .map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap())
                .collect()
        }

        // @s carries the encrypted column's BIN2 collation in its BaseTypeInfo.
        let mut nvarchar_50 = TypeInfo::nvarchar(50);
        nvarchar_50.collation = Some([0x09, 0x04, 0xd0, 0x00, 0x34]);

        let cases = [
            (
                "@i",
                TypeInfo::int(),
                "024000690008a5ffff41000000000000004100000001ed7b8c6030870d92358f12acb0d0c69c00bc3aa3ba578ecb0ea5f514b5045912a2b1ae52ed834f6bac49520956e4a574c30d573590fb3785556c8fe42f87c5b4000000002604020106000000020000000100000023722f0069b4000001",
            ),
            (
                "@s",
                nvarchar_50,
                "024000730008a5ffff4100000000000000410000000150c0a7dec4d4241c7a4a617007d32d97e7131f8c57a5ad212487891170f12ecb9957fce16389f4728d1c3c65813beeea085ae3fd516d29f84298df3e97f0d05d00000000e764000904d00034020106000000020000000100000023722f0069b4000001",
            ),
            (
                "@b",
                TypeInfo::varbinary(50),
                "024000620008a5ffff41000000000000004100000001d17165aa6df0155be6b78c6712d3b03870ea394cfed10956cf07fbfa204c4b82cddfa5e2f4fc03335f579e2767657e3067cd9da7d62a07427106b91f747b97da00000000a53200020106000000020000000100000023722f0069b4000001",
            ),
        ];

        for (name, base_type_info, golden_hex) in cases {
            let golden = unhex(golden_hex);
            // Slice the ciphertext out of the golden: B_VARCHAR name
            // (1 + 2*chars) + status(1) + A5FFFF(3) + PLP total(8) + chunk(4).
            let cipher_off = 1 + name.encode_utf16().count() * 2 + 1 + 3 + 8 + 4;
            let cipher = Bytes::copy_from_slice(&golden[cipher_off..cipher_off + 65]);

            let param = RpcParam::encrypted(
                name,
                cipher,
                EncryptedParamMetadata {
                    base_type_info,
                    algorithm_id: 2,
                    encryption_type: EncryptionTypeWire::Deterministic,
                    database_id: 6,
                    cek_id: 2,
                    cek_version: 1,
                    cek_md_version: 0x0000_b469_002f_7223,
                    normalization_rule_version: 1,
                },
            );

            let mut buf = BytesMut::new();
            param.encode(&mut buf);
            assert_eq!(
                buf.to_vec(),
                golden,
                "encrypted {name} param must match the captured Microsoft.Data.SqlClient bytes"
            );
        }
    }

    #[test]
    fn test_proc_id_values() {
        assert_eq!(ProcId::ExecuteSql as u16, 0x000A);
        assert_eq!(ProcId::Prepare as u16, 0x000B);
        assert_eq!(ProcId::Execute as u16, 0x000C);
        assert_eq!(ProcId::Unprepare as u16, 0x000F);
    }

    #[test]
    fn test_option_flags_encode() {
        let flags = RpcOptionFlags::new().with_recompile(true);
        assert_eq!(flags.encode(), 0x0001);
    }

    #[test]
    fn test_param_flags_encode() {
        let flags = ParamFlags::new().output();
        assert_eq!(flags.encode(), 0x01);
    }

    #[test]
    fn test_int_param() {
        let param = RpcParam::int("@p1", 42);
        assert_eq!(param.name, "@p1");
        assert_eq!(param.type_info.type_id, 0x26);
        assert!(param.value.is_some());
    }

    #[test]
    fn test_nvarchar_param() {
        let param = RpcParam::nvarchar("@name", "Alice");
        assert_eq!(param.name, "@name");
        assert_eq!(param.type_info.type_id, 0xE7);
        // UTF-16 encoded "Alice" = 10 bytes
        assert_eq!(param.value.as_ref().unwrap().len(), 10);
    }

    #[test]
    fn test_nvarchar_param_surrogate_pair_length() {
        // 🌍 is a supplementary character — 1 Rust char but 2 UTF-16 code units
        // (4 bytes). TypeInfo.max_length is stored doubled internally, so
        // the metadata must declare 2 code units for the buffer to match.
        let param = RpcParam::nvarchar("@p", "🌍");
        assert_eq!(param.value.as_ref().unwrap().len(), 4);
        // TypeInfo::nvarchar(n) stores max_length as n*2 bytes.
        assert_eq!(param.type_info.max_length, Some(4));

        let param = RpcParam::nvarchar("@p", "Hello 世界 🌍");
        // "Hello 世界 " = 9 BMP code units + 🌍 = 2 surrogate units → 11 code units, 22 bytes
        assert_eq!(param.value.as_ref().unwrap().len(), 22);
        assert_eq!(param.type_info.max_length, Some(22));
    }

    #[test]
    fn test_execute_sql_request() {
        let rpc = RpcRequest::execute_sql(
            "SELECT * FROM users WHERE id = @p1",
            vec![RpcParam::int("@p1", 42)],
        );

        assert_eq!(rpc.proc_id, Some(ProcId::ExecuteSql));
        // SQL statement + param declarations + actual params
        assert_eq!(rpc.params.len(), 3);
    }

    #[test]
    fn test_param_declarations() {
        let params = vec![
            RpcParam::int("@p1", 42),
            RpcParam::nvarchar("@name", "Alice"),
        ];

        let decls = RpcRequest::build_param_declarations(&params);
        assert!(decls.contains("@p1 int"));
        assert!(decls.contains("@name nvarchar"));
    }

    #[test]
    fn test_rpc_encode_not_empty() {
        let rpc = RpcRequest::execute_sql("SELECT 1", vec![]);
        let encoded = rpc.encode();
        assert!(!encoded.is_empty());
    }

    #[test]
    fn test_prepare_request() {
        let rpc = RpcRequest::prepare(
            "SELECT * FROM users WHERE id = @p1",
            &[RpcParam::int("@p1", 0)],
        );

        assert_eq!(rpc.proc_id, Some(ProcId::Prepare));
        // handle (output), params, stmt, options
        assert_eq!(rpc.params.len(), 4);
        assert!(rpc.params[0].flags.by_ref); // handle is OUTPUT
    }

    #[test]
    fn test_execute_request() {
        let rpc = RpcRequest::execute(123, vec![RpcParam::int("@p1", 42)]);

        assert_eq!(rpc.proc_id, Some(ProcId::Execute));
        assert_eq!(rpc.params.len(), 2); // handle + param
    }

    #[test]
    fn test_unprepare_request() {
        let rpc = RpcRequest::unprepare(123);

        assert_eq!(rpc.proc_id, Some(ProcId::Unprepare));
        assert_eq!(rpc.params.len(), 1); // just the handle
    }

    #[test]
    fn test_varchar_param() {
        let param = RpcParam::varchar("@name", "Alice");
        assert_eq!(param.name, "@name");
        assert_eq!(param.type_info.type_id, 0xA7);
        // Single-byte encoded "Alice" = 5 bytes
        assert_eq!(param.value.as_ref().unwrap().len(), 5);
        assert_eq!(&param.value.as_ref().unwrap()[..], b"Alice");
    }

    #[test]
    fn test_varchar_param_max() {
        // String > 8000 bytes should use VARCHAR(MAX)
        let long_str = "a".repeat(9000);
        let param = RpcParam::varchar("@big", &long_str);
        assert_eq!(param.type_info.type_id, 0xA7);
        assert_eq!(param.type_info.max_length, Some(0xFFFF));
        assert_eq!(param.value.as_ref().unwrap().len(), 9000);
    }

    #[test]
    fn test_varchar_param_declarations() {
        let params = vec![
            RpcParam::int("@p1", 42),
            RpcParam::varchar("@name", "Alice"),
        ];

        let decls = RpcRequest::build_param_declarations(&params);
        assert!(decls.contains("@p1 int"));
        assert!(decls.contains("@name varchar(5)"));
    }

    #[test]
    fn test_varchar_type_info_has_collation() {
        let ti = TypeInfo::varchar(100);
        assert_eq!(ti.type_id, 0xA7);
        assert_eq!(ti.max_length, Some(100));
        assert!(ti.collation.is_some());
    }

    #[test]
    fn test_varchar_encode_round_trip() {
        // Verify the encoded param can be serialized without panics
        let param = RpcParam::varchar("@val", "test value");
        let mut buf = bytes::BytesMut::new();
        param.encode(&mut buf);
        assert!(!buf.is_empty());
    }

    #[test]
    fn test_collation_round_trip() {
        let collation = Collation {
            lcid: 0x00D0_0409,
            sort_id: 0x34,
        };
        let bytes = collation.to_bytes();
        assert_eq!(bytes, [0x09, 0x04, 0xD0, 0x00, 0x34]);

        let restored = Collation::from_bytes(&bytes);
        assert_eq!(restored.lcid, collation.lcid);
        assert_eq!(restored.sort_id, collation.sort_id);
    }

    #[test]
    fn test_varchar_with_collation_uses_custom_collation_bytes() {
        // Chinese_PRC_CI_AS collation (LCID 0x0804)
        let collation = Collation {
            lcid: 0x0804,
            sort_id: 0,
        };
        let param = RpcParam::varchar_with_collation("@val", "test", &collation);
        assert_eq!(param.type_info.type_id, 0xA7);
        // Collation bytes should match the custom collation, not default Latin1
        assert_eq!(param.type_info.collation, Some(collation.to_bytes()));
    }

    #[test]
    fn test_money_type_info() {
        let ti = TypeInfo::money();
        assert_eq!(ti.type_id, 0x6E);
        assert_eq!(ti.max_length, Some(8));
    }

    #[test]
    fn test_smallmoney_type_info() {
        let ti = TypeInfo::smallmoney();
        assert_eq!(ti.type_id, 0x6E);
        assert_eq!(ti.max_length, Some(4));
    }

    #[test]
    fn test_smalldatetime_type_info() {
        let ti = TypeInfo::smalldatetime();
        assert_eq!(ti.type_id, 0x6F);
        assert_eq!(ti.max_length, Some(4));
    }

    #[test]
    fn test_money_param_declarations() {
        let decls = RpcRequest::build_param_declarations(&[
            RpcParam::new("@m", TypeInfo::money(), Bytes::from_static(&[0u8; 8])),
            RpcParam::new("@sm", TypeInfo::smallmoney(), Bytes::from_static(&[0u8; 4])),
            RpcParam::new(
                "@sdt",
                TypeInfo::smalldatetime(),
                Bytes::from_static(&[0u8; 4]),
            ),
        ]);
        assert!(decls.contains("@m money"), "got: {decls}");
        assert!(decls.contains("@sm smallmoney"), "got: {decls}");
        assert!(decls.contains("@sdt smalldatetime"), "got: {decls}");
    }

    #[test]
    fn test_money_typeinfo_encodes_max_length_byte() {
        let mut buf = bytes::BytesMut::new();
        TypeInfo::money().encode(&mut buf);
        // type_id 0x6E + max_length byte 0x08
        assert_eq!(&buf[..], &[0x6E, 0x08]);

        let mut buf = bytes::BytesMut::new();
        TypeInfo::smallmoney().encode(&mut buf);
        assert_eq!(&buf[..], &[0x6E, 0x04]);

        let mut buf = bytes::BytesMut::new();
        TypeInfo::smalldatetime().encode(&mut buf);
        assert_eq!(&buf[..], &[0x6F, 0x04]);
    }

    #[test]
    fn test_varchar_with_collation_default_vs_custom_differ() {
        let default_param = RpcParam::varchar("@val", "test");
        let custom_collation = Collation {
            lcid: 0x0419, // Russian
            sort_id: 0,
        };
        let custom_param = RpcParam::varchar_with_collation("@val", "test", &custom_collation);
        // The collation bytes should differ
        assert_ne!(
            default_param.type_info.collation,
            custom_param.type_info.collation
        );
    }
}