//! CQL Type Coverage Write-Read Roundtrip Tests
//!
//! Tests that verify all Stage 0 supported CQL types round-trip correctly
//! through the write and read path.
//!
//! ## Supported Types (Stage 0)
//!
//! - Text (varchar)
//! - Int (32-bit signed)
//! - BigInt (64-bit signed)
//! - Boolean
//! - Timestamp (milliseconds since epoch)
//! - UUID
//!
//! ## Dependencies
//!
//! - Writer: `cqlite_core::storage::write_engine::WriteEngine`
//! - Statistics: `cqlite_core::parser::enhanced_statistics_parser`

#![cfg(feature = "write-support")]

use cqlite_core::schema::{Column, KeyColumn, TableSchema, UdtRegistry};
use cqlite_core::storage::write_engine::{CellOperation, Mutation, PartitionKey, TableId};
use cqlite_core::types::Value;
use std::collections::HashMap;
use tempfile::TempDir;

/// Create a schema for testing a specific column type
fn create_type_test_schema(col_name: &str, col_type: &str) -> TableSchema {
    // Sanitize table name: strip angle-bracket parameterization
    // e.g., "frozen<list<int>>" → "frozen", "tuple<int, text>" → "tuple"
    let base_type = col_type.split('<').next().unwrap_or(col_type);
    TableSchema {
        keyspace: "test_types".to_string(),
        table: format!("test_{}", base_type.to_lowercase()),
        partition_keys: vec![KeyColumn {
            name: "pk".to_string(),
            data_type: "int".to_string(),
            position: 0,
        }],
        clustering_keys: vec![],
        columns: vec![
            Column {
                name: "pk".to_string(),
                data_type: "int".to_string(),
                nullable: false,
                default: None,
                is_static: false,
            },
            Column {
                name: col_name.to_string(),
                data_type: col_type.to_string(),
                nullable: true,
                default: None,
                is_static: false,
            },
        ],
        comments: HashMap::new(),
    }
}

/// Helper to write and flush a single value
async fn write_single_value(
    temp_dir: &TempDir,
    schema: &TableSchema,
    col_name: &str,
    value: Value,
) -> cqlite_core::storage::sstable::writer::SSTableInfo {
    let mut engine = super::create_test_engine(temp_dir, schema.clone())
        .expect("Engine creation should succeed");

    let table_id = TableId::new(&schema.keyspace, &schema.table);
    let pk = PartitionKey::single("pk", Value::Integer(1));
    let ops = vec![CellOperation::Write {
        column: col_name.to_string(),
        value,
    }];
    let mutation = Mutation::new(table_id, pk, None, ops, 1000000, None);

    engine
        .write_async(mutation)
        .await
        .expect("Write should succeed");

    engine
        .flush()
        .await
        .expect("Flush should succeed")
        .expect("Should return SSTableInfo")
}

/// Assert that a single partition was written successfully
fn assert_single_partition_written(info: &cqlite_core::storage::sstable::writer::SSTableInfo) {
    super::assert_file_exists_and_nonempty(&info.data_path, "Data.db");
    assert_eq!(info.partition_count, 1, "Should have 1 partition");
}

/// Test Text type roundtrip
#[tokio::test]
async fn test_type_text_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("text_col", "text");
    let original = Value::Text("Hello, CQLite! 你好世界 🎉".to_string());

    let info = write_single_value(&temp_dir, &schema, "text_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "text_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Text type with empty string
#[tokio::test]
async fn test_type_text_empty() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("text_col", "text");
    let original = Value::Text(String::new());

    let info = write_single_value(&temp_dir, &schema, "text_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "text_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Text type with long string
#[tokio::test]
async fn test_type_text_long() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("text_col", "text");

    // Create a 10KB string
    let original = Value::Text("A".repeat(10 * 1024));

    let info = write_single_value(&temp_dir, &schema, "text_col", original.clone()).await;

    assert_single_partition_written(&info);
    let data_size = std::fs::metadata(&info.data_path).unwrap().len();
    assert!(
        data_size > 10000,
        "Data.db should be > 10KB for long text (got {} bytes)",
        data_size
    );
    let read_back = super::read_back_column(&temp_dir, &schema, "text_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Int type roundtrip
#[tokio::test]
async fn test_type_int_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("int_col", "int");
    let original = Value::Integer(42);

    let info = write_single_value(&temp_dir, &schema, "int_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "int_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Int type with min value
#[tokio::test]
async fn test_type_int_min() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("int_col", "int");
    let original = Value::Integer(i32::MIN);

    let info = write_single_value(&temp_dir, &schema, "int_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "int_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Int type with max value
#[tokio::test]
async fn test_type_int_max() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("int_col", "int");
    let original = Value::Integer(i32::MAX);

    let info = write_single_value(&temp_dir, &schema, "int_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "int_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Int type with zero
#[tokio::test]
async fn test_type_int_zero() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("int_col", "int");
    let original = Value::Integer(0);

    let info = write_single_value(&temp_dir, &schema, "int_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "int_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test BigInt type roundtrip
#[tokio::test]
async fn test_type_bigint_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("bigint_col", "bigint");
    let original = Value::BigInt(1_000_000_000_000i64);

    let info = write_single_value(&temp_dir, &schema, "bigint_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "bigint_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test BigInt type with min value
#[tokio::test]
async fn test_type_bigint_min() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("bigint_col", "bigint");
    let original = Value::BigInt(i64::MIN);

    let info = write_single_value(&temp_dir, &schema, "bigint_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "bigint_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test BigInt type with max value
#[tokio::test]
async fn test_type_bigint_max() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("bigint_col", "bigint");
    let original = Value::BigInt(i64::MAX);

    let info = write_single_value(&temp_dir, &schema, "bigint_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "bigint_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Boolean type roundtrip (true)
#[tokio::test]
async fn test_type_boolean_true() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("bool_col", "boolean");
    let original = Value::Boolean(true);

    let info = write_single_value(&temp_dir, &schema, "bool_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "bool_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Boolean type roundtrip (false)
#[tokio::test]
async fn test_type_boolean_false() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("bool_col", "boolean");
    let original = Value::Boolean(false);

    let info = write_single_value(&temp_dir, &schema, "bool_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "bool_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Timestamp type roundtrip
#[tokio::test]
async fn test_type_timestamp_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("ts_col", "timestamp");

    // 2024-01-01 00:00:00 UTC in milliseconds
    let original = Value::Timestamp(1704067200000i64);

    let info = write_single_value(&temp_dir, &schema, "ts_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "ts_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Timestamp type with epoch
#[tokio::test]
async fn test_type_timestamp_epoch() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("ts_col", "timestamp");
    let original = Value::Timestamp(0);

    let info = write_single_value(&temp_dir, &schema, "ts_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "ts_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Timestamp type with far future
#[tokio::test]
async fn test_type_timestamp_future() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("ts_col", "timestamp");

    // Year 3000 (in milliseconds)
    let original = Value::Timestamp(32503680000000i64);

    let info = write_single_value(&temp_dir, &schema, "ts_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "ts_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test UUID type roundtrip
#[tokio::test]
async fn test_type_uuid_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("uuid_col", "uuid");

    let uuid_bytes = *uuid::Uuid::new_v4().as_bytes();
    let original = Value::Uuid(uuid_bytes);
    let info = write_single_value(&temp_dir, &schema, "uuid_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "uuid_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test UUID type with known value
#[tokio::test]
async fn test_type_uuid_known() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("uuid_col", "uuid");

    let known_uuid = uuid::Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap();
    let original = Value::Uuid(*known_uuid.as_bytes());
    let info = write_single_value(&temp_dir, &schema, "uuid_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "uuid_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test UUID type with nil UUID
#[tokio::test]
async fn test_type_uuid_nil() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("uuid_col", "uuid");

    let original = Value::Uuid(*uuid::Uuid::nil().as_bytes());
    let info = write_single_value(&temp_dir, &schema, "uuid_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "uuid_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test TinyInt type roundtrip
#[tokio::test]
async fn test_type_tinyint_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("tinyint_col", "tinyint");
    let original = Value::TinyInt(42);
    let info = write_single_value(&temp_dir, &schema, "tinyint_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "tinyint_col").await;
    assert_eq!(read_back, original, "TinyInt roundtrip failed");
}

/// Test TinyInt type with min value
#[tokio::test]
async fn test_type_tinyint_min() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("tinyint_col", "tinyint");
    let original = Value::TinyInt(i8::MIN);
    let info = write_single_value(&temp_dir, &schema, "tinyint_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "tinyint_col").await;
    assert_eq!(read_back, original, "TinyInt(MIN) roundtrip failed");
}

/// Test TinyInt type with max value
#[tokio::test]
async fn test_type_tinyint_max() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("tinyint_col", "tinyint");
    let original = Value::TinyInt(i8::MAX);
    let info = write_single_value(&temp_dir, &schema, "tinyint_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "tinyint_col").await;
    assert_eq!(read_back, original, "TinyInt(MAX) roundtrip failed");
}

/// Test SmallInt type roundtrip
#[tokio::test]
async fn test_type_smallint_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("smallint_col", "smallint");
    let original = Value::SmallInt(1000);
    let info = write_single_value(&temp_dir, &schema, "smallint_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "smallint_col").await;
    assert_eq!(read_back, original, "SmallInt roundtrip failed");
}

/// Test SmallInt type with min value
#[tokio::test]
async fn test_type_smallint_min() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("smallint_col", "smallint");
    let original = Value::SmallInt(i16::MIN);
    let info = write_single_value(&temp_dir, &schema, "smallint_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "smallint_col").await;
    assert_eq!(read_back, original, "SmallInt(MIN) roundtrip failed");
}

/// Test SmallInt type with max value
#[tokio::test]
async fn test_type_smallint_max() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("smallint_col", "smallint");
    let original = Value::SmallInt(i16::MAX);
    let info = write_single_value(&temp_dir, &schema, "smallint_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "smallint_col").await;
    assert_eq!(read_back, original, "SmallInt(MAX) roundtrip failed");
}

/// The reader widens Float32 → Float(f64) during read-back.
/// IEEE 754 bits are preserved through the widening.
fn widen_float32(v: Value) -> Value {
    if let Value::Float32(f) = v {
        Value::Float(f as f64)
    } else {
        v
    }
}

/// Test Float32 type roundtrip
///
/// The reader widens f32 to f64 (Value::Float) during read-back.
/// IEEE 754 bits are preserved; we compare against the widened value.
#[tokio::test]
async fn test_type_float32_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("float_col", "float");
    let original = Value::Float32(1.234_567);
    let info = write_single_value(&temp_dir, &schema, "float_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "float_col").await;
    assert_eq!(
        read_back,
        widen_float32(original),
        "Float32 roundtrip failed"
    );
}

/// Test Float32 type with special value
///
/// The reader widens f32 to f64 (Value::Float) during read-back.
#[tokio::test]
async fn test_type_float32_special() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("float_col", "float");
    let original = Value::Float32(0.0);
    let info = write_single_value(&temp_dir, &schema, "float_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "float_col").await;
    assert_eq!(
        read_back,
        widen_float32(original),
        "Float32 roundtrip failed"
    );
}

/// Test Float32 type with min value
///
/// The reader widens f32 to f64 (Value::Float) during read-back.
#[tokio::test]
async fn test_type_float32_min() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("float_col", "float");
    let original = Value::Float32(f32::MIN);
    let info = write_single_value(&temp_dir, &schema, "float_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "float_col").await;
    assert_eq!(
        read_back,
        widen_float32(original),
        "Float32 roundtrip failed"
    );
}

/// Test Double type roundtrip
#[tokio::test]
async fn test_type_double_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("double_col", "double");
    let original = Value::Float(9.876_543_210_123_456);
    let info = write_single_value(&temp_dir, &schema, "double_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "double_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Double type with special value
#[tokio::test]
async fn test_type_double_special() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("double_col", "double");
    let original = Value::Float(f64::INFINITY);
    let info = write_single_value(&temp_dir, &schema, "double_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "double_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Double type with min/max value
#[tokio::test]
async fn test_type_double_min_max() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("double_col", "double");
    let original = Value::Float(f64::MIN);
    let info = write_single_value(&temp_dir, &schema, "double_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "double_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Blob type roundtrip
#[tokio::test]
async fn test_type_blob_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("blob_col", "blob");
    let original = Value::Blob(vec![0xDE, 0xAD, 0xBE, 0xEF]);
    let info = write_single_value(&temp_dir, &schema, "blob_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "blob_col").await;
    assert_eq!(read_back, original, "Blob roundtrip failed");
}

/// Test Blob type with empty value
#[tokio::test]
async fn test_type_blob_empty() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("blob_col", "blob");
    let original = Value::Blob(vec![]);
    let info = write_single_value(&temp_dir, &schema, "blob_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "blob_col").await;
    assert_eq!(read_back, original, "Blob(empty) roundtrip failed");
}

/// Test Blob type with large value
#[tokio::test]
async fn test_type_blob_large() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("blob_col", "blob");
    let original = Value::Blob(vec![0xAB; 10240]);

    let info = write_single_value(&temp_dir, &schema, "blob_col", original.clone()).await;

    assert_single_partition_written(&info);
    let data_size = std::fs::metadata(&info.data_path).unwrap().len();
    assert!(
        data_size > 10000,
        "Data.db should be > 10000 bytes for large blob (got {} bytes)",
        data_size
    );
    let read_back = super::read_back_column(&temp_dir, &schema, "blob_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Date type roundtrip
#[tokio::test]
async fn test_type_date_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("date_col", "date");
    // 2024-01-01
    let original = Value::Date(19723);
    let info = write_single_value(&temp_dir, &schema, "date_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "date_col").await;
    assert_eq!(read_back, original, "Date roundtrip failed");
}

/// Test Date type with epoch value
#[tokio::test]
async fn test_type_date_epoch() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("date_col", "date");
    let original = Value::Date(0);
    let info = write_single_value(&temp_dir, &schema, "date_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "date_col").await;
    assert_eq!(read_back, original, "Date(epoch) roundtrip failed");
}

/// Test Date type with negative value
#[tokio::test]
async fn test_type_date_negative() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("date_col", "date");
    let original = Value::Date(-1);

    let info = write_single_value(&temp_dir, &schema, "date_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "date_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Time type roundtrip
#[tokio::test]
async fn test_type_time_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("time_col", "time");
    // noon: 43200 seconds in nanoseconds
    let original = Value::Time(43_200_000_000_000);
    let info = write_single_value(&temp_dir, &schema, "time_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "time_col").await;
    assert_eq!(read_back, original, "Time roundtrip failed");
}

/// Test Time type with midnight value
#[tokio::test]
async fn test_type_time_midnight() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("time_col", "time");
    let original = Value::Time(0);
    let info = write_single_value(&temp_dir, &schema, "time_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "time_col").await;
    assert_eq!(read_back, original, "Time(midnight) roundtrip failed");
}

/// Test Time type with max value
#[tokio::test]
async fn test_type_time_max() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("time_col", "time");
    let original = Value::Time(86_399_999_999_999);
    let info = write_single_value(&temp_dir, &schema, "time_col", original.clone()).await;
    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "time_col").await;
    assert_eq!(read_back, original, "Time(max) roundtrip failed");
}

/// Counter columns require server-side distributed increment semantics
/// (counter UPDATE `SET col = col + n`) that cannot be expressed as a
/// last-write-wins `Mutation`. The engine rejects such mutations eagerly
/// with `Error::InvalidOperation` so callers receive an actionable error
/// rather than silently writing semantically incorrect data.
#[tokio::test]
async fn test_counter_write_returns_typed_error() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("counter_col", "counter");
    let mut engine = super::create_test_engine(&temp_dir, schema.clone())
        .expect("Engine creation should succeed");

    let table_id = TableId::new(&schema.keyspace, &schema.table);
    let pk = PartitionKey::single("pk", Value::Integer(1));
    let ops = vec![CellOperation::Write {
        column: "counter_col".to_string(),
        value: Value::Counter(100),
    }];
    let mutation = Mutation::new(table_id, pk, None, ops, 1_000_000, None);

    let result = engine.write_async(mutation).await;

    assert!(
        result.is_err(),
        "Counter write via WriteEngine must return an error, but it succeeded"
    );
    let err = result.unwrap_err();
    assert!(
        matches!(err, cqlite_core::error::Error::InvalidOperation(_)),
        "Counter write must return Error::InvalidOperation, got: {:?}",
        err
    );
    assert!(
        err.to_string().contains("counter"),
        "Error message should mention 'counter', got: {}",
        err
    );
}

/// Both `write()` and `write_async()` must enforce the counter guard
/// consistently so callers cannot bypass it by choosing the sync path.
#[test]
fn test_counter_write_sync_returns_typed_error() {
    use cqlite_core::storage::write_engine::{WriteEngine, WriteEngineConfig};

    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("counter_col", "counter");

    let config = WriteEngineConfig::new(
        temp_dir.path().join("data"),
        temp_dir.path().join("wal"),
        schema.clone(),
    );
    let mut engine = WriteEngine::new(config).expect("Engine creation should succeed");

    let table_id = TableId::new(&schema.keyspace, &schema.table);
    let pk = PartitionKey::single("pk", Value::Integer(1));
    let ops = vec![CellOperation::Write {
        column: "counter_col".to_string(),
        value: Value::Counter(42),
    }];
    let mutation = Mutation::new(table_id, pk, None, ops, 1_000_000, None);

    let result = engine.write(mutation);

    assert!(
        result.is_err(),
        "Counter write via sync WriteEngine::write() must return an error, but it succeeded"
    );
    let err = result.unwrap_err();
    assert!(
        matches!(err, cqlite_core::error::Error::InvalidOperation(_)),
        "Counter write must return Error::InvalidOperation, got: {:?}",
        err
    );
    assert!(
        err.to_string().contains("counter"),
        "Error message should mention 'counter', got: {}",
        err
    );
}

/// Test Inet type with IPv4 address
#[tokio::test]
async fn test_type_inet_ipv4() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("inet_col", "inet");
    let original = Value::Inet(vec![192, 168, 1, 1]);

    let info = write_single_value(&temp_dir, &schema, "inet_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "inet_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Inet type with IPv6 address
#[tokio::test]
async fn test_type_inet_ipv6() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("inet_col", "inet");
    let original = Value::Inet(vec![0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 1]);

    let info = write_single_value(&temp_dir, &schema, "inet_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "inet_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Inet type with loopback address
#[tokio::test]
async fn test_type_inet_loopback() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("inet_col", "inet");
    let original = Value::Inet(vec![127, 0, 0, 1]);

    let info = write_single_value(&temp_dir, &schema, "inet_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "inet_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Varint type with small value
#[tokio::test]
async fn test_type_varint_small() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("varint_col", "varint");
    let original = Value::Varint(vec![0x2A]);

    let info = write_single_value(&temp_dir, &schema, "varint_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "varint_col").await;
    // Reader returns Blob with the same bytes (varint serialized as raw bytes)
    let expected_bytes = vec![0x2A];
    match &read_back {
        Value::Varint(b) => assert_eq!(b, &expected_bytes, "Varint bytes mismatch"),
        Value::Blob(b) => assert_eq!(
            b, &expected_bytes,
            "Varint read back as Blob: bytes mismatch"
        ),
        other => panic!("Expected Varint or Blob, got {:?}", other),
    }
}

/// Test Varint type with large value
#[tokio::test]
async fn test_type_varint_large() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("varint_col", "varint");
    let original = Value::Varint(vec![0x00, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF]);

    let info = write_single_value(&temp_dir, &schema, "varint_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "varint_col").await;
    // Reader returns Blob with the same bytes (varint serialized as raw bytes)
    let expected_bytes = vec![0x00u8, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF, 0xFF];
    match &read_back {
        Value::Varint(b) => assert_eq!(b, &expected_bytes, "Varint bytes mismatch"),
        Value::Blob(b) => assert_eq!(
            b, &expected_bytes,
            "Varint read back as Blob: bytes mismatch"
        ),
        other => panic!("Expected Varint or Blob, got {:?}", other),
    }
}

/// Test Varint type with negative value
#[tokio::test]
async fn test_type_varint_negative() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("varint_col", "varint");
    let original = Value::Varint(vec![0xFF]);

    let info = write_single_value(&temp_dir, &schema, "varint_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "varint_col").await;
    // Reader returns Blob with the same bytes (varint serialized as raw bytes)
    let expected_bytes = vec![0xFFu8];
    match &read_back {
        Value::Varint(b) => assert_eq!(b, &expected_bytes, "Varint bytes mismatch"),
        Value::Blob(b) => assert_eq!(
            b, &expected_bytes,
            "Varint read back as Blob: bytes mismatch"
        ),
        other => panic!("Expected Varint or Blob, got {:?}", other),
    }
}

/// Test Decimal type roundtrip
#[tokio::test]
async fn test_type_decimal_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("decimal_col", "decimal");
    let original = Value::Decimal {
        scale: 2,
        unscaled: vec![0x30, 0x39],
    };

    let info = write_single_value(&temp_dir, &schema, "decimal_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "decimal_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Decimal type with zero value
#[tokio::test]
async fn test_type_decimal_zero() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("decimal_col", "decimal");
    let original = Value::Decimal {
        scale: 0,
        unscaled: vec![0],
    };

    let info = write_single_value(&temp_dir, &schema, "decimal_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "decimal_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Decimal type with negative scale
#[tokio::test]
async fn test_type_decimal_neg_scale() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("decimal_col", "decimal");
    let original = Value::Decimal {
        scale: -2,
        unscaled: vec![1],
    };

    let info = write_single_value(&temp_dir, &schema, "decimal_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "decimal_col").await;
    assert_eq!(read_back, original, "Type roundtrip failed");
}

/// Test Duration type roundtrip
#[tokio::test]
async fn test_type_duration_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("duration_col", "duration");
    let original = Value::Duration {
        months: 1,
        days: 15,
        nanos: 3_600_000_000_000,
    };

    let info = write_single_value(&temp_dir, &schema, "duration_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "duration_col").await;
    // The reader may return Duration directly or as Blob (raw bytes) if the type isn't
    // fully schema-decoded. Accept either form.
    match &read_back {
        Value::Duration {
            months,
            days,
            nanos,
        } => {
            assert_eq!(*months, 1, "Duration months mismatch");
            assert_eq!(*days, 15, "Duration days mismatch");
            assert_eq!(*nanos, 3_600_000_000_000, "Duration nanos mismatch");
        }
        Value::Blob(_) => {
            // Reader returns raw bytes for duration — document this behavior
            // until full schema-aware duration decoding is implemented.
            eprintln!(
                "note: duration read back as Blob (schema-aware decoding not yet implemented)"
            );
        }
        other => panic!("Duration roundtrip failed: got {:?}", other),
    }
}

/// Test Duration type with zero value
#[tokio::test]
async fn test_type_duration_zero() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("duration_col", "duration");
    let original = Value::Duration {
        months: 0,
        days: 0,
        nanos: 0,
    };

    let info = write_single_value(&temp_dir, &schema, "duration_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "duration_col").await;
    match &read_back {
        Value::Duration {
            months,
            days,
            nanos,
        } => {
            assert_eq!(*months, 0, "Duration(zero) months mismatch");
            assert_eq!(*days, 0, "Duration(zero) days mismatch");
            assert_eq!(*nanos, 0, "Duration(zero) nanos mismatch");
        }
        Value::Blob(_) => {
            eprintln!("note: duration(zero) read back as Blob (schema-aware decoding not yet implemented)");
        }
        other => panic!("Duration(zero) roundtrip failed: got {:?}", other),
    }
}

/// Test Duration type with negative value
#[tokio::test]
async fn test_type_duration_negative() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("duration_col", "duration");
    let original = Value::Duration {
        months: -1,
        days: -5,
        nanos: -1_000_000_000,
    };

    let info = write_single_value(&temp_dir, &schema, "duration_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "duration_col").await;
    match &read_back {
        Value::Duration {
            months,
            days,
            nanos,
        } => {
            assert_eq!(*months, -1, "Duration(negative) months mismatch");
            assert_eq!(*days, -5, "Duration(negative) days mismatch");
            assert_eq!(*nanos, -1_000_000_000, "Duration(negative) nanos mismatch");
        }
        Value::Blob(_) => {
            eprintln!("note: duration(negative) read back as Blob (schema-aware decoding not yet implemented)");
        }
        other => panic!("Duration(negative) roundtrip failed: got {:?}", other),
    }
}

/// Test Tuple type roundtrip — schema-aware element decoding (Issue #501).
#[tokio::test]
async fn test_type_tuple_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("tuple_col", "tuple<int, text>");
    let original = Value::Tuple(vec![Value::Integer(42), Value::Text("hello".to_string())]);

    let info = write_single_value(&temp_dir, &schema, "tuple_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "tuple_col").await;
    assert_eq!(read_back, original, "Tuple<int,text> roundtrip failed");
}

/// Test Tuple type with null element — schema-aware element decoding (Issue #501).
#[tokio::test]
async fn test_type_tuple_with_null() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("tuple_col", "tuple<int, text>");
    let original = Value::Tuple(vec![Value::Integer(42), Value::Null]);

    let info = write_single_value(&temp_dir, &schema, "tuple_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "tuple_col").await;
    assert_eq!(
        read_back, original,
        "Tuple<int,text>(with_null) roundtrip failed"
    );
}

/// Test Tuple type nested — schema-aware element decoding (Issue #501).
#[tokio::test]
async fn test_type_tuple_nested() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("tuple_col", "tuple<int, tuple<int, text>>");
    let original = Value::Tuple(vec![
        Value::Integer(1),
        Value::Tuple(vec![Value::Integer(2), Value::Text("nested".to_string())]),
    ]);

    let info = write_single_value(&temp_dir, &schema, "tuple_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "tuple_col").await;
    assert_eq!(
        read_back, original,
        "Tuple<int,tuple<int,text>>(nested) roundtrip failed"
    );
}

/// Test Frozen list type roundtrip
#[tokio::test]
async fn test_type_frozen_list() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("frozen_col", "frozen<list<int>>");
    let inner = Value::List(vec![
        Value::Integer(1),
        Value::Integer(2),
        Value::Integer(3),
    ]);
    let original = Value::Frozen(Box::new(inner.clone()));

    let info = write_single_value(&temp_dir, &schema, "frozen_col", original.clone()).await;

    assert_single_partition_written(&info);
    let col_value = super::read_back_column(&temp_dir, &schema, "frozen_col").await;
    // Accept either Value::Frozen(List) or Value::List directly
    assert!(
        col_value == original || col_value == inner,
        "Frozen list roundtrip failed: expected {:?} or {:?}, got {:?}",
        original,
        inner,
        col_value
    );
}

/// Test Frozen map type roundtrip
#[tokio::test]
async fn test_type_frozen_map() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("frozen_col", "frozen<map<text, int>>");
    let inner = Value::Map(vec![(Value::Text("key".to_string()), Value::Integer(42))]);
    let original = Value::Frozen(Box::new(inner.clone()));

    let info = write_single_value(&temp_dir, &schema, "frozen_col", original.clone()).await;

    assert_single_partition_written(&info);
    let col_value = super::read_back_column(&temp_dir, &schema, "frozen_col").await;
    assert!(
        col_value == original || col_value == inner,
        "Frozen map roundtrip failed: expected {:?} or {:?}, got {:?}",
        original,
        inner,
        col_value
    );
}

/// Test Frozen empty list type
#[tokio::test]
async fn test_type_frozen_empty() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("frozen_col", "frozen<list<int>>");
    let inner = Value::List(vec![]);
    let original = Value::Frozen(Box::new(inner.clone()));

    let info = write_single_value(&temp_dir, &schema, "frozen_col", original.clone()).await;

    assert_single_partition_written(&info);
    let col_value = super::read_back_column(&temp_dir, &schema, "frozen_col").await;
    assert!(
        col_value == original || col_value == inner,
        "Frozen(empty) list roundtrip failed: expected {:?} or {:?}, got {:?}",
        original,
        inner,
        col_value
    );
}

/// Test all types in single partition
#[tokio::test]
async fn test_all_types_single_partition() {
    use super::{
        create_comprehensive_mutation, create_comprehensive_schema, WriteEngine, WriteEngineConfig,
    };

    let temp_dir = TempDir::new().unwrap();
    let schema = create_comprehensive_schema();

    let config = WriteEngineConfig::new(
        temp_dir.path().join("data"),
        temp_dir.path().join("wal"),
        schema.clone(),
    );

    let mut engine = WriteEngine::new(config).expect("Engine creation should succeed");

    // Write mutation with all types
    let mutation = create_comprehensive_mutation(1, "row1", 1000000);
    engine
        .write_async(mutation)
        .await
        .expect("Write should succeed");

    let info = engine
        .flush()
        .await
        .expect("Flush should succeed")
        .expect("Should return SSTableInfo");

    assert!(
        info.data_path.exists(),
        "Data.db should exist for all types"
    );
    assert_eq!(info.partition_count, 1, "Should have 1 partition");

    // Verify data has substantial size (all types present)
    let data_size = std::fs::metadata(&info.data_path).unwrap().len();
    assert!(
        data_size > 50,
        "Data.db with all types should have > 50 bytes (got {})",
        data_size
    );
}

/// Test multiple rows with varying types
#[tokio::test]
async fn test_types_multiple_rows() {
    use super::{
        create_comprehensive_mutation, create_comprehensive_schema, WriteEngine, WriteEngineConfig,
    };

    let temp_dir = TempDir::new().unwrap();
    let schema = create_comprehensive_schema();

    let config = WriteEngineConfig::new(
        temp_dir.path().join("data"),
        temp_dir.path().join("wal"),
        schema.clone(),
    );

    let mut engine = WriteEngine::new(config).expect("Engine creation should succeed");

    // Write multiple partitions with all types
    for i in 0..10 {
        let mutation = create_comprehensive_mutation(i, &format!("row_{i}"), 1000000 + i as i64);
        engine
            .write_async(mutation)
            .await
            .expect("Write should succeed");
    }

    let info = engine
        .flush()
        .await
        .expect("Flush should succeed")
        .expect("Should return SSTableInfo");

    assert_eq!(
        info.partition_count, 10,
        "Should have 10 partitions with all types"
    );
}

/// Test Ascii type roundtrip (handled identically to Text)
#[tokio::test]
async fn test_type_ascii_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("ascii_col", "ascii");
    let original = Value::Text("hello_ascii".to_string());

    let info = write_single_value(&temp_dir, &schema, "ascii_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "ascii_col").await;
    assert_eq!(read_back, original, "Ascii type roundtrip failed");
}

/// Test Varchar type roundtrip (handled identically to Text)
#[tokio::test]
async fn test_type_varchar_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("varchar_col", "varchar");
    let original = Value::Text("hello_varchar".to_string());

    let info = write_single_value(&temp_dir, &schema, "varchar_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "varchar_col").await;
    assert_eq!(read_back, original, "Varchar type roundtrip failed");
}

/// Test Timeuuid type roundtrip (uses same Uuid([u8; 16]) path as uuid)
#[tokio::test]
async fn test_type_timeuuid_roundtrip() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("timeuuid_col", "timeuuid");

    // Valid v1 UUID bytes
    let v1_uuid_bytes: [u8; 16] = [
        0x01, 0xb2, 0x1d, 0xd2, 0x13, 0x81, 0x11, 0xe1, 0x85, 0x5a, 0x00, 0x02, 0xa5, 0xd5, 0xc5,
        0x1b,
    ];
    let original = Value::Uuid(v1_uuid_bytes);

    let info = write_single_value(&temp_dir, &schema, "timeuuid_col", original.clone()).await;

    assert_single_partition_written(&info);
    let read_back = super::read_back_column(&temp_dir, &schema, "timeuuid_col").await;
    assert_eq!(read_back, original, "Timeuuid type roundtrip failed");
}

/// Test Tuple<int,text,uuid> — three-element tuple with schema-aware decoding (Issue #501).
#[tokio::test]
async fn test_type_tuple_int_text_uuid() {
    let temp_dir = TempDir::new().unwrap();
    let schema = create_type_test_schema("tuple_col", "tuple<int, text, uuid>");

    let known_uuid = uuid::Uuid::parse_str("550e8400-e29b-41d4-a716-446655440000").unwrap();
    let original = Value::Tuple(vec![
        Value::Integer(99),
        Value::Text("hello".to_string()),
        Value::Uuid(*known_uuid.as_bytes()),
    ]);

    let info = write_single_value(&temp_dir, &schema, "tuple_col", original.clone()).await;

    assert_single_partition_written(&info);

    let read_back = super::read_back_column(&temp_dir, &schema, "tuple_col").await;
    assert_eq!(read_back, original, "Tuple<int,text,uuid> roundtrip failed");
}

/// Verify that `frozen<person>` round-trips correctly when the reader is given a
/// `UdtRegistry` that carries the "person" type definition (Issue #502).
///
/// The concrete UDT name ("person") is placed in the column type string so the
/// reader can look it up in the registry.  A `UdtRegistry` with the matching
/// field definitions is injected via `read_back_column_with_udt_registry`.
#[cfg(feature = "state_machine")]
#[tokio::test]
async fn test_type_frozen_udt() {
    use cqlite_core::schema::CqlType;
    use cqlite_core::types::{UdtField, UdtTypeDef, UdtValue};

    let temp_dir = TempDir::new().unwrap();
    // Use a concrete UDT name so the reader can resolve it from the registry.
    let schema = create_type_test_schema("frozen_col", "frozen<person>");

    // Build the matching UDT definition (field names + types).
    let udt_def = UdtTypeDef::new("test_types".to_string(), "person".to_string())
        .with_field("name".to_string(), CqlType::Text, true)
        .with_field("age".to_string(), CqlType::Int, true);
    let mut udt_registry = UdtRegistry::new();
    udt_registry.register_udt(udt_def);

    // Build a simple two-field UDT value.
    let inner_udt = Value::Udt(UdtValue {
        type_name: "person".to_string(),
        keyspace: "test_types".to_string(),
        fields: vec![
            UdtField {
                name: "name".to_string(),
                value: Some(Value::Text("Alice".to_string())),
            },
            UdtField {
                name: "age".to_string(),
                value: Some(Value::Integer(30)),
            },
        ],
    });
    let original = Value::Frozen(Box::new(inner_udt.clone()));

    let info = write_single_value(&temp_dir, &schema, "frozen_col", original.clone()).await;
    assert_single_partition_written(&info);

    // Use the UDT-registry-aware scan helper so the reader can resolve "person".
    let col_value =
        super::read_back_column_with_udt_registry(&temp_dir, &schema, "frozen_col", udt_registry)
            .await;

    // The reader must return the fully decoded UDT — either wrapped in Frozen or
    // unwrapped.  Frozen(Null) and Blob are no longer acceptable outcomes.
    assert!(
        col_value == original || col_value == inner_udt,
        "frozen<person> roundtrip: expected {:?} or {:?}, got {:?}",
        original,
        inner_udt,
        col_value
    );
}
