package executor

import (
	"encoding/binary"
	"testing"
	"time"

	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/parser"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/systemtables"
)

// TestCQLProtocolFrameFormat tests that the CQL protocol frame format is correct
// This specifically tests the fix for the "unknown type bootstr" error
func TestCQLProtocolFrameFormat(t *testing.T) {
	translator := NewQueryResultTranslator()

	// Create a QueryResult that mimics system.local
	result := &QueryResult{
		Columns: []ColumnInfo{
			{Keyspace: "system", Table: "local", Name: "key", Type: "text"},
			{Keyspace: "system", Table: "local", Name: "bootstrapped", Type: "text"},
			{Keyspace: "system", Table: "local", Name: "broadcast_address", Type: "inet"},
			{Keyspace: "system", Table: "local", Name: "cluster_name", Type: "text"},
			{Keyspace: "system", Table: "local", Name: "cql_version", Type: "text"},
			{Keyspace: "system", Table: "local", Name: "data_center", Type: "text"},
			{Keyspace: "system", Table: "local", Name: "gossip_generation", Type: "int"},
			{Keyspace: "system", Table: "local", Name: "host_id", Type: "uuid"},
			{Keyspace: "system", Table: "local", Name: "listen_address", Type: "inet"},
			{Keyspace: "system", Table: "local", Name: "native_protocol_version", Type: "text"},
			{Keyspace: "system", Table: "local", Name: "partitioner", Type: "text"},
			{Keyspace: "system", Table: "local", Name: "rack", Type: "text"},
			{Keyspace: "system", Table: "local", Name: "release_version", Type: "text"},
			{Keyspace: "system", Table: "local", Name: "rpc_address", Type: "inet"},
			{Keyspace: "system", Table: "local", Name: "schema_version", Type: "uuid"},
			{Keyspace: "system", Table: "local", Name: "storage_port", Type: "int"},
			{Keyspace: "system", Table: "local", Name: "tokens", Type: "set<text>"},
		},
		Rows: []Row{
			{
				Values: []interface{}{
					"local",
					"COMPLETED",
					"127.0.0.1",
					"Cassandra",
					"3.4.7",
					"datacenter1",
					1,
					"00000000-0000-0000-0000-000000000001",
					"127.0.0.1",
					"4",
					"org.apache.cassandra.dht.Murmur3Partitioner",
					"rack1",
					"4.0",
					"127.0.0.1",
					"00000000-0000-0000-0000-000000000002",
					7000,
					[]string{"-9223372036854775808"},
				},
			},
		},
	}

	frame, err := translator.TranslateToResultFrame(result, 0)
	if err != nil {
		t.Fatalf("TranslateToResultFrame failed: %v", err)
	}

	// Verify frame structure
	if frame.Opcode != 0x08 {
		t.Errorf("Expected opcode 0x08 (RESULT), got 0x%02x", frame.Opcode)
	}

	if frame.Version != 0x84 {
		t.Errorf("Expected version 0x84 (proto v4), got 0x%02x", frame.Version)
	}

	// Parse the frame body
	body := frame.Body
	pos := 0

	// 1. Result kind (4 bytes)
	resultKind := binary.BigEndian.Uint32(body[pos : pos+4])
	pos += 4
	if resultKind != 0x0002 {
		t.Errorf("Expected result kind 0x0002 (ROWS), got 0x%08x", resultKind)
	}

	// 2. Flags (4 bytes)
	flags := binary.BigEndian.Uint32(body[pos : pos+4])
	pos += 4
	if flags&resultMetadataGlobalSpec == 0 {
		t.Fatalf("Expected GLOBAL_TABLES_SPEC flag set, got %d", flags)
	}

	// 3. Column count (4 bytes)
	colCount := binary.BigEndian.Uint32(body[pos : pos+4])
	pos += 4
	if colCount != uint32(len(result.Columns)) {
		t.Errorf("Expected %d columns, got %d", len(result.Columns), colCount)
	}

	// 4. Column specifications
	if flags&resultMetadataHasMore != 0 {
		pageLen := binary.BigEndian.Uint32(body[pos : pos+4])
		pos += 4 + int(pageLen)
	}

	if flags&resultMetadataGlobalSpec != 0 {
		ksLen := binary.BigEndian.Uint16(body[pos : pos+2])
		pos += 2 + int(ksLen)
		tbLen := binary.BigEndian.Uint16(body[pos : pos+2])
		pos += 2 + int(tbLen)
	}

	for i, col := range result.Columns {
		// Column name (2-byte length + string)
		colNameLen := binary.BigEndian.Uint16(body[pos : pos+2])
		pos += 2
		colName := string(body[pos : pos+int(colNameLen)])
		pos += int(colNameLen)

		if colName != col.Name {
			t.Errorf("Column %d: expected name %s, got %s", i, col.Name, colName)
		}

		// Column type (TypeInfo structure)
		// Read type code
		typeCode := binary.BigEndian.Uint16(body[pos : pos+2])
		pos += 2

		// Verify the type code is correct
		expectedType := translator.mysqlTypeToCQLType(col.Type)
		if typeCode != uint16(expectedType) {
			t.Errorf("Column %s: expected type code 0x%04x, got 0x%04x", col.Name, expectedType, typeCode)
		}

		// For collection types, skip the nested type info
		// set<type> and list<type> have element type after the type code
		// map<keyType, valueType> has key type and value type
		if typeCode == uint16(CQLTypeSet) || typeCode == uint16(CQLTypeList) {
			// Skip the element type (recursively, but for our test cases it's just 2 bytes)
			pos += 2
		} else if typeCode == uint16(CQLTypeMap) {
			// Skip key type and value type
			pos += 4
		}
	}

	// 5. Row count (4 bytes)
	rowCount := binary.BigEndian.Uint32(body[pos : pos+4])
	pos += 4
	if rowCount != uint32(len(result.Rows)) {
		t.Errorf("Expected %d rows, got %d", len(result.Rows), rowCount)
	}

	// 6. Row data
	for rowIdx, row := range result.Rows {
		for colIdx, value := range row.Values {
			// Value length (4 bytes)
			valLen := binary.BigEndian.Uint32(body[pos : pos+4])
			pos += 4

			if value == nil {
				// NULL value has length 0xFFFFFFFF
				if valLen != 0xFFFFFFFF {
					t.Errorf("Row %d, Col %d: expected NULL (0xFFFFFFFF), got 0x%08x", rowIdx, colIdx, valLen)
				}
				continue
			}

			// Read the value bytes
			valBytes := body[pos : pos+int(valLen)]
			pos += int(valLen)

			// Verify the value can be read back correctly
			// This is a basic sanity check
			if len(valBytes) == 0 && valLen != 0 {
				t.Errorf("Row %d, Col %d: expected %d bytes, got 0", rowIdx, colIdx, valLen)
			}
		}
	}

	t.Logf("Successfully parsed frame with %d columns and %d rows", colCount, rowCount)
}

// TestColumnSpecFormat tests that column specs are written correctly
func TestColumnSpecFormat(t *testing.T) {
	translator := NewQueryResultTranslator()

	testCases := []struct {
		name         string
		col          ColumnInfo
		expectedType CQLType
	}{
		{"text column", ColumnInfo{Name: "name", Type: "text"}, CQLTypeVarchar},
		{"int column", ColumnInfo{Name: "count", Type: "int"}, CQLTypeInt},
		{"bigint column", ColumnInfo{Name: "big", Type: "bigint"}, CQLTypeBigInt},
		{"uuid column", ColumnInfo{Name: "id", Type: "uuid"}, CQLTypeUUID},
		{"inet column", ColumnInfo{Name: "addr", Type: "inet"}, CQLTypeInet},
		{"boolean column", ColumnInfo{Name: "flag", Type: "boolean"}, CQLTypeBoolean},
		{"timestamp column", ColumnInfo{Name: "ts", Type: "timestamp"}, CQLTypeTimestamp},
		{"set<text> column", ColumnInfo{Name: "tags", Type: "set<text>"}, CQLTypeSet},
		{"map<text,text> column", ColumnInfo{Name: "props", Type: "map<text, text>"}, CQLTypeMap},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			spec := translator.translateColumnSpec(tc.col)

			// Should be at least: 2 (ks len) + 2 (table len) + 2 (col name len) + col name + 2 (type)
			if len(spec) < 6 {
				t.Fatalf("Spec too short: %d bytes", len(spec))
			}

			// Skip keyspace, table, column name
			pos := 0
			pos += 2 + int(binary.BigEndian.Uint16(spec[pos:pos+2])) // ks
			pos += 2 + int(binary.BigEndian.Uint16(spec[pos:pos+2])) // table
			pos += 2 + int(binary.BigEndian.Uint16(spec[pos:pos+2])) // col name

			// Read type code (2 bytes)
			if pos+2 > len(spec) {
				t.Fatalf("Spec too short for type code")
			}
			typeCode := binary.BigEndian.Uint16(spec[pos : pos+2])

			if typeCode != uint16(tc.expectedType) {
				t.Errorf("Expected type 0x%04x, got 0x%04x", tc.expectedType, typeCode)
			}
		})
	}
}

func TestRowsResultGlobalSpec(t *testing.T) {
	translator := NewQueryResultTranslator()

	result := &QueryResult{
		Columns: []ColumnInfo{
			{Keyspace: "system", Table: "describe", Name: "keyspace_name", Type: "text"},
			{Keyspace: "system", Table: "describe", Name: "type", Type: "text"},
			{Keyspace: "system", Table: "describe", Name: "name", Type: "text"},
		},
		Rows: []Row{
			{Values: []interface{}{"system", "keyspace", "system"}},
		},
		UseGlobalSpec: true,
	}

	frame, err := translator.TranslateToResultFrame(result, 0)
	if err != nil {
		t.Fatalf("TranslateToResultFrame failed: %v", err)
	}

	body := frame.Body
	pos := 0

	resultKind := binary.BigEndian.Uint32(body[pos : pos+4])
	pos += 4
	if resultKind != 0x0002 {
		t.Fatalf("expected result kind rows, got 0x%08x", resultKind)
	}

	flags := binary.BigEndian.Uint32(body[pos : pos+4])
	pos += 4
	if flags&resultMetadataGlobalSpec == 0 {
		t.Fatalf("expected global spec flag set, got %d", flags)
	}

	colCount := binary.BigEndian.Uint32(body[pos : pos+4])
	pos += 4
	if colCount != uint32(len(result.Columns)) {
		t.Fatalf("expected %d columns, got %d", len(result.Columns), colCount)
	}

	ksLen := binary.BigEndian.Uint16(body[pos : pos+2])
	pos += 2
	ks := string(body[pos : pos+int(ksLen)])
	pos += int(ksLen)
	if ks != "system" {
		t.Fatalf("expected keyspace system, got %q", ks)
	}

	tbLen := binary.BigEndian.Uint16(body[pos : pos+2])
	pos += 2
	tb := string(body[pos : pos+int(tbLen)])
	pos += int(tbLen)
	if tb != "describe" {
		t.Fatalf("expected table describe, got %q", tb)
	}

	for _, col := range result.Columns {
		nameLen := binary.BigEndian.Uint16(body[pos : pos+2])
		pos += 2
		name := string(body[pos : pos+int(nameLen)])
		pos += int(nameLen)
		if name != col.Name {
			t.Fatalf("expected column name %q, got %q", col.Name, name)
		}

		typeCode := binary.BigEndian.Uint16(body[pos : pos+2])
		pos += 2
		if typeCode != uint16(CQLTypeVarchar) {
			t.Fatalf("expected text type code 0x%04x, got 0x%04x", CQLTypeVarchar, typeCode)
		}
	}
}

func TestSetKeyspaceResult(t *testing.T) {
	translator := NewQueryResultTranslator()
	keyspace := "system"

	result := &QueryResult{
		SetKeyspace: keyspace,
	}

	frame, err := translator.TranslateToResultFrame(result, 0)
	if err != nil {
		t.Fatalf("TranslateToResultFrame failed: %v", err)
	}

	if frame.Opcode != 0x08 {
		t.Errorf("Expected opcode 0x08 (RESULT), got 0x%02x", frame.Opcode)
	}

	if frame.Version != 0x84 {
		t.Errorf("Expected version 0x84 (proto v4), got 0x%02x", frame.Version)
	}

	body := frame.Body
	if len(body) < 6 {
		t.Fatalf("Expected SET_KEYSPACE body length >= 6, got %d", len(body))
	}

	resultKind := binary.BigEndian.Uint32(body[0:4])
	if resultKind != uint32(ResultKindSetKeyspace) {
		t.Errorf("Expected result kind 0x%08x, got 0x%08x", ResultKindSetKeyspace, resultKind)
	}

	ksLen := binary.BigEndian.Uint16(body[4:6])
	if int(ksLen) != len(keyspace) {
		t.Fatalf("Expected keyspace length %d, got %d", len(keyspace), ksLen)
	}

	ksEnd := 6 + int(ksLen)
	if len(body) < ksEnd {
		t.Fatalf("Expected keyspace bytes length %d, got %d", ksEnd, len(body))
	}

	gotKeyspace := string(body[6:ksEnd])
	if gotKeyspace != keyspace {
		t.Errorf("Expected keyspace %q, got %q", keyspace, gotKeyspace)
	}
}

func TestSchemaChangeResultTable(t *testing.T) {
	translator := NewQueryResultTranslator()

	result := &QueryResult{
		SchemaChange: &SchemaChange{
			ChangeType: "CREATED",
			Target:     "TABLE",
			Keyspace:   "test",
			Name:       "tbl",
		},
	}

	frame, err := translator.TranslateToResultFrame(result, 0)
	if err != nil {
		t.Fatalf("TranslateToResultFrame failed: %v", err)
	}

	body := frame.Body
	if len(body) < 4 {
		t.Fatalf("Expected schema change body length >= 4, got %d", len(body))
	}
	resultKind := binary.BigEndian.Uint32(body[0:4])
	if resultKind != uint32(ResultKindSchemaChange) {
		t.Fatalf("Expected result kind 0x%08x, got 0x%08x", ResultKindSchemaChange, resultKind)
	}

	readShortString := func(buf []byte, pos int) (string, int) {
		if len(buf) < pos+2 {
			t.Fatalf("short string truncated at %d", pos)
		}
		l := int(binary.BigEndian.Uint16(buf[pos : pos+2]))
		pos += 2
		if len(buf) < pos+l {
			t.Fatalf("short string data truncated at %d", pos)
		}
		return string(buf[pos : pos+l]), pos + l
	}

	pos := 4
	changeType, pos := readShortString(body, pos)
	target, pos := readShortString(body, pos)
	keyspace, pos := readShortString(body, pos)
	table, _ := readShortString(body, pos)

	if changeType != "CREATED" {
		t.Fatalf("expected change type CREATED, got %q", changeType)
	}
	if target != "TABLE" {
		t.Fatalf("expected target TABLE, got %q", target)
	}
	if keyspace != "test" {
		t.Fatalf("expected keyspace test, got %q", keyspace)
	}
	if table != "tbl" {
		t.Fatalf("expected table tbl, got %q", table)
	}
}

// TestValueSerialization tests that values are serialized correctly
func TestValueSerialization(t *testing.T) {
	translator := NewQueryResultTranslator()

	testCases := []struct {
		name    string
		value   interface{}
		cqlType string
		minLen  int
		checkFn func([]byte) bool
	}{
		{
			name:    "string value",
			value:   "test",
			cqlType: "text",
			minLen:  4,
			checkFn: func(b []byte) bool { return string(b) == "test" },
		},
		{
			name:    "int32 value",
			value:   int32(42),
			cqlType: "int",
			minLen:  4,
			checkFn: func(b []byte) bool { return len(b) == 4 },
		},
		{
			name:    "mysql INT value",
			value:   int64(2),
			cqlType: "INT",
			minLen:  4,
			checkFn: func(b []byte) bool {
				return len(b) == 4 && binary.BigEndian.Uint32(b) == 2
			},
		},
		{
			name:    "int64 value",
			value:   int64(12345),
			cqlType: "bigint",
			minLen:  8,
			checkFn: func(b []byte) bool { return len(b) == 8 },
		},
		{
			name:    "boolean true",
			value:   true,
			cqlType: "boolean",
			minLen:  1,
			checkFn: func(b []byte) bool { return len(b) == 1 && b[0] == 0x01 },
		},
		{
			name:    "boolean false",
			value:   false,
			cqlType: "boolean",
			minLen:  1,
			checkFn: func(b []byte) bool { return len(b) == 1 && b[0] == 0x00 },
		},
		{
			name:    "inet IPv4",
			value:   "127.0.0.1",
			cqlType: "inet",
			minLen:  4,
			checkFn: func(b []byte) bool { return len(b) == 4 },
		},
		{
			name:    "uuid string",
			value:   "00000000-0000-0000-0000-000000000001",
			cqlType: "uuid",
			minLen:  16,
			checkFn: func(b []byte) bool { return len(b) == 16 },
		},
		{
			name:    "timeuuid string",
			value:   "d2177dd0-eaa2-11de-a572-000000000001",
			cqlType: "timeuuid",
			minLen:  16,
			checkFn: func(b []byte) bool { return len(b) == 16 },
		},
		{
			name:    "uuid bytes (already binary)",
			value:   []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
			cqlType: "uuid",
			minLen:  16,
			checkFn: func(b []byte) bool { return len(b) == 16 },
		},
		{
			name:    "set<text>",
			value:   []string{"a", "b", "c"},
			cqlType: "set<text>",
			minLen:  4, // At least element count
			checkFn: func(b []byte) bool { return len(b) >= 4 },
		},
		{
			name:    "map<text,text>",
			value:   map[string]string{"key": "value"},
			cqlType: "map<text, text>",
			minLen:  4, // At least pair count
			checkFn: func(b []byte) bool { return len(b) >= 4 },
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			result := translator.translateValue(tc.value, tc.cqlType, nil)

			if len(result) < tc.minLen {
				t.Errorf("Serialized value too short: expected >= %d, got %d", tc.minLen, len(result))
			}

			if tc.checkFn != nil && !tc.checkFn(result) {
				t.Errorf("Serialized value check failed")
			}
		})
	}
}

// TestUUIDWireFormat tests that UUIDs are correctly converted to 16-byte binary format
func TestUUIDWireFormat(t *testing.T) {
	translator := NewQueryResultTranslator()

	testCases := []struct {
		uuidStr  string
		expected []byte
	}{
		{
			uuidStr:  "00000000-0000-0000-0000-000000000000",
			expected: []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00},
		},
		{
			uuidStr:  "00000000-0000-0000-0000-000000000001",
			expected: []byte{0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
		},
		{
			uuidStr:  "d2177dd0-eaa2-11de-a572-000000000001",
			expected: []byte{0xd2, 0x17, 0x7d, 0xd0, 0xea, 0xa2, 0x11, 0xde, 0xa5, 0x72, 0x00, 0x00, 0x00, 0x00, 0x00, 0x01},
		},
		{
			uuidStr:  "FFFFFFFF-FFFF-FFFF-FFFF-FFFFFFFFFFFF",
			expected: []byte{0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff},
		},
	}

	for _, tc := range testCases {
		t.Run(tc.uuidStr, func(t *testing.T) {
			result := translator.writeUUID(tc.uuidStr)

			if len(result) != 16 {
				t.Errorf("Expected 16 bytes, got %d", len(result))
			}

			for i, b := range tc.expected {
				if result[i] != b {
					t.Errorf("Byte %d: expected 0x%02x, got 0x%02x", i, b, result[i])
				}
			}
		})
	}
}

// TestInvalidUUID tests that invalid UUIDs return zero UUID
func TestInvalidUUID(t *testing.T) {
	translator := NewQueryResultTranslator()

	testCases := []string{
		"",                                    // Empty
		"invalid",                             // Not hex
		"00000000-0000",                       // Too short
		"00000000-0000-0000-0000-00000000000", // 31 chars
	}

	for _, uuidStr := range testCases {
		t.Run(uuidStr, func(t *testing.T) {
			result := translator.writeUUID(uuidStr)

			if len(result) != 16 {
				t.Errorf("Expected 16 bytes for invalid UUID, got %d", len(result))
			}

			// Should be all zeros
			for i, b := range result {
				if b != 0 {
					t.Errorf("Expected zero byte at position %d, got 0x%02x", i, b)
				}
			}
		})
	}
}

// TestWriteInet tests the writeInet function specifically
func TestWriteInet(t *testing.T) {
	translator := NewQueryResultTranslator()

	testCases := []struct {
		input    string
		expected []byte
	}{
		{"127.0.0.1", []byte{127, 0, 0, 1}},
		{"192.168.1.1", []byte{192, 168, 1, 1}},
		{"10.0.0.1", []byte{10, 0, 0, 1}},
	}

	for _, tc := range testCases {
		t.Run(tc.input, func(t *testing.T) {
			result := translator.writeInet(tc.input)

			if len(result) != len(tc.expected) {
				t.Errorf("Expected %d bytes, got %d", len(tc.expected), len(result))
			}

			for i, b := range tc.expected {
				if result[i] != b {
					t.Errorf("Byte %d: expected %d, got %d", i, b, result[i])
				}
			}
		})
	}
}

// TestTypeMapping tests the mysqlTypeToCQLType function
func TestTypeMapping(t *testing.T) {
	translator := NewQueryResultTranslator()

	testCases := []struct {
		input    string
		expected CQLType
	}{
		{"text", CQLTypeVarchar},
		{"varchar", CQLTypeVarchar},
		{"ascii", CQLTypeVarchar},
		{"int", CQLTypeInt},
		{"bigint", CQLTypeBigInt},
		{"uuid", CQLTypeUUID},
		{"timeuuid", CQLTypeTimeUUID},
		{"inet", CQLTypeInet},
		{"boolean", CQLTypeBoolean},
		{"timestamp", CQLTypeTimestamp},
		{"date", CQLTypeDate},
		{"time", CQLTypeTime},
		{"double", CQLTypeDouble},
		{"float", CQLTypeFloat},
		{"blob", CQLTypeBlob},
		{"smallint", CQLTypeSmallInt},
		{"tinyint", CQLTypeTinyInt},
		{"decimal", CQLTypeDecimal},
		{"varint", CQLTypeVarint},
		{"duration", CQLTypeDuration},
		{"set<text>", CQLTypeSet},
		{"set<inet>", CQLTypeSet},
		{"list<text>", CQLTypeList},
		{"map<text, text>", CQLTypeMap},
		{"map<text, bigint>", CQLTypeMap},
		{"frozen<set<text>>", CQLTypeSet},
		{"TINYINT", CQLTypeTinyInt},
		{"INT", CQLTypeInt},
		{"BIGINT", CQLTypeBigInt},
		{"VARCHAR", CQLTypeVarchar},
		{"BOOLEAN", CQLTypeBoolean},
		{"TIMESTAMP", CQLTypeTimestamp},
		{"BLOB", CQLTypeBlob},
		{"JSON", CQLTypeVarchar},
	}

	for _, tc := range testCases {
		t.Run(tc.input, func(t *testing.T) {
			result := translator.mysqlTypeToCQLType(tc.input)

			if result != tc.expected {
				t.Errorf("Input %s: expected 0x%04x, got 0x%04x", tc.input, tc.expected, result)
			}
		})
	}
}

func TestTranslateValueBooleanNumeric(t *testing.T) {
	translator := NewQueryResultTranslator()

	testCases := []struct {
		name     string
		value    interface{}
		expected byte
	}{
		{"int64-true", int64(1), 0x01},
		{"int64-false", int64(0), 0x00},
		{"int8-true", int8(1), 0x01},
		{"uint8-false", uint8(0), 0x00},
		{"string-true", "true", 0x01},
		{"string-false", "false", 0x00},
		{"string-1", "1", 0x01},
		{"string-0", "0", 0x00},
		{"bytes-1", []byte("1"), 0x01},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			data := translator.translateValue(tc.value, "boolean", nil)
			if len(data) != 1 {
				t.Fatalf("expected 1 byte, got %d", len(data))
			}
			if data[0] != tc.expected {
				t.Fatalf("unexpected boolean encoding: 0x%02x", data[0])
			}
		})
	}
}

func TestTranslateValueDecimalEncoding(t *testing.T) {
	translator := NewQueryResultTranslator()

	testCases := []struct {
		value    interface{}
		expected string
	}{
		{value: "12.34", expected: "12.34"},
		{value: "12.30", expected: "12.30"},
		{value: int64(42), expected: "42"},
		{value: "-0.50", expected: "-0.50"},
		{value: "1e3", expected: "1000"},
		{value: "1e-2", expected: "0.01"},
		{value: "123456789012345678901234567890.1234", expected: "123456789012345678901234567890.1234"},
	}

	for _, tc := range testCases {
		data := translator.translateValue(tc.value, "decimal", nil)
		if data == nil || len(data) < 4 {
			t.Fatalf("expected decimal bytes for %v, got %v", tc.value, data)
		}

		got, err := decodeDecimal(data)
		if err != nil {
			t.Fatalf("decodeDecimal failed for %v: %v", tc.value, err)
		}
		if got != tc.expected {
			t.Fatalf("expected %q, got %q", tc.expected, got)
		}
	}
}

func TestTranslateValueDateTimeEncoding(t *testing.T) {
	translator := NewQueryResultTranslator()

	dateVal := time.Date(2024, 1, 2, 0, 0, 0, 0, time.UTC)
	dateBytes := translator.translateValue(dateVal, "date", nil)
	if len(dateBytes) != 4 {
		t.Fatalf("expected date to encode to 4 bytes, got %d", len(dateBytes))
	}
	dateDays := dateVal.UTC().UnixNano()/int64(time.Millisecond)/millisecondsInADay + (1 << 31)
	if got := int32(binary.BigEndian.Uint32(dateBytes)); got != int32(dateDays) {
		t.Fatalf("unexpected date encoding: %d", got)
	}

	timeVal := 12*time.Hour + 34*time.Minute + 56*time.Second + 789*time.Millisecond
	timeBytes := translator.translateValue(timeVal, "time", nil)
	if len(timeBytes) != 8 {
		t.Fatalf("expected time to encode to 8 bytes, got %d", len(timeBytes))
	}
	if got := int64(binary.BigEndian.Uint64(timeBytes)); got != timeVal.Nanoseconds() {
		t.Fatalf("unexpected time encoding: %d", got)
	}
}

func TestTranslateValueTimestampMicroseconds(t *testing.T) {
	translator := NewQueryResultTranslator()
	raw := "2024-06-01 12:34:56.123456"
	data := translator.translateValue(raw, "timestamp", nil)
	if len(data) != 8 {
		t.Fatalf("expected timestamp to encode to 8 bytes, got %d", len(data))
	}
	parsed, err := time.ParseInLocation("2006-01-02 15:04:05.000000", raw, time.UTC)
	if err != nil {
		t.Fatalf("parse failed: %v", err)
	}
	want := parsed.UnixNano() / int64(time.Millisecond)
	got := int64(binary.BigEndian.Uint64(data))
	if got != want {
		t.Fatalf("unexpected timestamp encoding: %d", got)
	}
}

func TestWriteTimestampTruncatesMicroseconds(t *testing.T) {
	translator := NewQueryResultTranslator()
	ts := time.Date(2024, 6, 1, 12, 34, 56, 123456*int(time.Microsecond), time.UTC)
	data := translator.writeTimestamp(ts)
	if len(data) != 8 {
		t.Fatalf("expected timestamp to encode to 8 bytes, got %d", len(data))
	}
	got := int64(binary.BigEndian.Uint64(data))
	want := ts.UnixNano() / int64(time.Millisecond)
	if got != want {
		t.Fatalf("unexpected timestamp encoding: %d", got)
	}
}

// TestPreparedResultFrameFormat tests that PREPARED result frames have proper column metadata
// This tests that prepared metadata contains bind markers (Parameters) and result metadata contains result columns (Columns)
func TestPreparedResultFrameFormat(t *testing.T) {
	translator := NewQueryResultTranslator()

	testCases := []struct {
		name            string
		prepared        *PreparedStatement
		expPrepColCount int
		expResColCount  int
		prepColSpecs    []ColumnInfo
		resColSpecs     []ColumnInfo
	}{
		{
			name: "No parameters (SELECT release_version FROM system.local)",
			prepared: &PreparedStatement{
				ID:         "test_stmt_id_1",
				Query:      "SELECT release_version FROM system.local",
				SQL:        "SELECT release_version FROM system.local",
				ParamCount: 0,
				Parameters: []ColumnInfo{},
				Columns: []ColumnInfo{
					{Keyspace: "system", Table: "local", Name: "release_version", Type: "text"},
				},
			},
			expPrepColCount: 0, // No bind markers
			expResColCount:  1, // 1 result column
			prepColSpecs:    []ColumnInfo{},
			resColSpecs: []ColumnInfo{
				{Keyspace: "system", Table: "local", Name: "release_version", Type: "text"},
			},
		},
		{
			name: "With parameters (SELECT * FROM system.local WHERE key = ?)",
			prepared: &PreparedStatement{
				ID:         "test_stmt_id_2",
				Query:      "SELECT * FROM system.local WHERE key = ?",
				SQL:        "SELECT * FROM system.local WHERE key = ?",
				ParamCount: 1,
				Parameters: []ColumnInfo{
					{Keyspace: "", Table: "", Name: "param_1", Type: "unknown", Nullable: true},
				},
				Columns: []ColumnInfo{
					{Keyspace: "system", Table: "local", Name: "key", Type: "text"},
					{Keyspace: "system", Table: "local", Name: "release_version", Type: "text"},
				},
			},
			expPrepColCount: 1, // 1 bind marker
			expResColCount:  2, // 2 result columns
			prepColSpecs: []ColumnInfo{
				{Keyspace: "", Table: "", Name: "param_1", Type: "unknown", Nullable: true},
			},
			resColSpecs: []ColumnInfo{
				{Keyspace: "system", Table: "local", Name: "key", Type: "text"},
				{Keyspace: "system", Table: "local", Name: "release_version", Type: "text"},
			},
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			buf := translator.BuildPreparedResult(tc.prepared, 4)

			// Now parse and verify the frame
			pos := 0

			// 1. Result kind (4 bytes)
			resultKind := binary.BigEndian.Uint32(buf[pos : pos+4])
			pos += 4
			if resultKind != 0x0004 {
				t.Errorf("Expected result kind 0x0004 (PREPARED), got 0x%08x", resultKind)
			}

			// 2. Prepared ID
			idLen := binary.BigEndian.Uint16(buf[pos : pos+2])
			pos += 2
			stmtID := string(buf[pos : pos+int(idLen)])
			pos += int(idLen)
			if stmtID != tc.prepared.ID {
				t.Errorf("Expected prepared ID %s, got %s", tc.prepared.ID, stmtID)
			}

			// 3. Prepared metadata flags
			prepFlags := binary.BigEndian.Uint32(buf[pos : pos+4])
			pos += 4
			if prepFlags != 0 {
				t.Errorf("Expected prepared flags 0, got %d", prepFlags)
			}

			// 4. Prepared metadata column count
			prepColCount := binary.BigEndian.Uint32(buf[pos : pos+4])
			pos += 4
			if prepColCount != uint32(tc.expPrepColCount) {
				t.Errorf("Expected %d parameters in prepared metadata, got %d", tc.expPrepColCount, prepColCount)
			}

			// 5. Prepared metadata pkey_count
			pkeyCount := binary.BigEndian.Uint32(buf[pos : pos+4])
			pos += 4
			if pkeyCount != 0 {
				t.Errorf("Expected pkey_count 0, got %d", pkeyCount)
			}

			// 6. Prepared metadata column specs (parameters)
			for i, col := range tc.prepColSpecs {
				// Skip keyspace
				ksLen := binary.BigEndian.Uint16(buf[pos : pos+2])
				pos += 2 + int(ksLen)

				// Skip table
				tbLen := binary.BigEndian.Uint16(buf[pos : pos+2])
				pos += 2 + int(tbLen)

				// Read column name
				colNameLen := binary.BigEndian.Uint16(buf[pos : pos+2])
				pos += 2
				colName := string(buf[pos : pos+int(colNameLen)])
				pos += int(colNameLen)

				if colName != col.Name {
					t.Errorf("Prepared param %d: expected name %s, got %s", i, col.Name, colName)
				}

				// Read column type
				typeCode := binary.BigEndian.Uint16(buf[pos : pos+2])
				pos += 2
				expectedType := translator.mysqlTypeToCQLType(col.Type)
				if typeCode != uint16(expectedType) {
					t.Errorf("Prepared param %s: expected type 0x%04x, got 0x%04x", col.Name, expectedType, typeCode)
				}
			}

			// 7. Result metadata flags
			resFlags := binary.BigEndian.Uint32(buf[pos : pos+4])
			pos += 4
			if resFlags != 0 {
				t.Errorf("Expected result flags 0, got %d", resFlags)
			}

			// 8. Result metadata column count
			resColCount := binary.BigEndian.Uint32(buf[pos : pos+4])
			pos += 4
			if resColCount != uint32(tc.expResColCount) {
				t.Errorf("Expected %d columns in result metadata, got %d", tc.expResColCount, resColCount)
			}

			// 9. Result metadata column specs
			for i, col := range tc.resColSpecs {
				// Skip keyspace
				ksLen := binary.BigEndian.Uint16(buf[pos : pos+2])
				pos += 2 + int(ksLen)

				// Skip table
				tbLen := binary.BigEndian.Uint16(buf[pos : pos+2])
				pos += 2 + int(tbLen)

				// Read column name
				colNameLen := binary.BigEndian.Uint16(buf[pos : pos+2])
				pos += 2
				colName := string(buf[pos : pos+int(colNameLen)])
				pos += int(colNameLen)

				if colName != col.Name {
					t.Errorf("Result column %d: expected name %s, got %s", i, col.Name, colName)
				}

				// Read column type
				typeCode := binary.BigEndian.Uint16(buf[pos : pos+2])
				pos += 2
				expectedType := translator.mysqlTypeToCQLType(col.Type)
				if typeCode != uint16(expectedType) {
					t.Errorf("Result column %s: expected type 0x%04x, got 0x%04x", col.Name, expectedType, typeCode)
				}
			}

			t.Logf("Successfully verified PREPARED frame: %d params, %d result cols", tc.expPrepColCount, tc.expResColCount)
		})
	}
}

// writeShortBytes is a helper to write string as [short_bytes] format
func writeShortBytes(data []byte) []byte {
	buf := make([]byte, 2+len(data))
	binary.BigEndian.PutUint16(buf[:2], uint16(len(data)))
	copy(buf[2:], data)
	return buf
}

// TestExtractSelectedColumns tests the column extraction helper function
// This specifically tests the fix for "SELECT release_version FROM system.local" column filtering
func TestExtractSelectedColumns(t *testing.T) {
	testCases := []struct {
		name           string
		cql            string
		expectedCols   []string
		expectedIsStar bool
	}{
		{
			name:           "Single column select",
			cql:            "SELECT release_version FROM system.local",
			expectedCols:   []string{"release_version"},
			expectedIsStar: false,
		},
		{
			name:           "Multiple column select",
			cql:            "SELECT key, release_version, cluster_name FROM system.local",
			expectedCols:   []string{"key", "release_version", "cluster_name"},
			expectedIsStar: false,
		},
		{
			name:           "Wildcard select",
			cql:            "SELECT * FROM system.local",
			expectedCols:   nil,
			expectedIsStar: true,
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			// Parse the CQL statement
			stmt, err := parser.NewParser(tc.cql).Parse()
			if err != nil {
				t.Fatalf("Failed to parse CQL: %v", err)
			}

			selectStmt, ok := stmt.(*parser.SelectStatement)
			if !ok {
				t.Fatalf("Expected SelectStatement, got %T", stmt)
			}

			// Extract columns
			cols, isStar := extractSelectedColumns(selectStmt)

			// Verify wildcard flag
			if isStar != tc.expectedIsStar {
				t.Errorf("Expected isStar=%v, got %v", tc.expectedIsStar, isStar)
			}

			// Verify column count
			if !isStar && len(cols) != len(tc.expectedCols) {
				t.Errorf("Expected %d columns, got %d", len(tc.expectedCols), len(cols))
			}

			// Verify column names
			if !isStar {
				for i, expected := range tc.expectedCols {
					if i >= len(cols) {
						t.Errorf("Missing column %d: expected %s", i, expected)
						continue
					}
					if cols[i] != expected {
						t.Errorf("Column %d: expected %s, got %s", i, expected, cols[i])
					}
				}
			}
		})
	}
}

// TestSystemTableColumnFiltering tests that system table queries return only selected columns
// This is the main test for the bug fix
func TestSystemTableColumnFiltering(t *testing.T) {
	testCases := []struct {
		name         string
		cql          string
		expectedCols int
	}{
		{
			name:         "Single column - release_version",
			cql:          "SELECT release_version FROM system.local",
			expectedCols: 1,
		},
		{
			name:         "Two columns",
			cql:          "SELECT release_version, cluster_name FROM system.local",
			expectedCols: 2,
		},
		{
			name:         "Wildcard - all columns",
			cql:          "SELECT * FROM system.local",
			expectedCols: 20,
		},
	}

	for _, tc := range testCases {
		t.Run(tc.name, func(t *testing.T) {
			stmt, err := parser.NewParser(tc.cql).Parse()
			if err != nil {
				t.Fatalf("Failed to parse CQL: %v", err)
			}

			selectStmt, ok := stmt.(*parser.SelectStatement)
			if !ok {
				t.Fatalf("Expected SelectStatement, got %T", stmt)
			}

			if !systemtables.IsSystemTableQuery(selectStmt) {
				t.Skip("Not a system table query")
			}

			tableName := selectStmt.From.String()
			selectedCols, isWildcard := extractSelectedColumns(selectStmt)

			stRegistry := systemtables.NewRegistry()
			st, ok := stRegistry.Lookup(tableName)
			if !ok {
				t.Fatalf("System table %s not found in registry", tableName)
			}

			var columns []ColumnInfo
			if isWildcard {
				columns = make([]ColumnInfo, len(st.Columns))
				for i, col := range st.Columns {
					columns[i] = ColumnInfo{Keyspace: col.Keyspace, Table: col.Table, Name: col.Name, Type: col.Type, Nullable: col.Nullable}
				}
			} else {
				for _, colName := range selectedCols {
					for _, stCol := range st.Columns {
						if stCol.Name == colName {
							columns = append(columns, ColumnInfo{Keyspace: stCol.Keyspace, Table: stCol.Table, Name: stCol.Name, Type: stCol.Type, Nullable: stCol.Nullable})
							break
						}
					}
				}
			}

			if len(columns) != tc.expectedCols {
				t.Errorf("Expected %d columns, got %d", tc.expectedCols, len(columns))
			}
		})
	}
}