package cql

import (
	"bytes"
	"encoding/binary"
	"fmt"
	"io"
	"sync"

	gocql "github.com/apache/cassandra-gocql-driver/v2"
)

// Protocol versions (using constants from opcodes.go)
const (
	protoDirectionMask = 0x80
	protoVersionMask   = 0x7F
)

// Frame flags
const (
	FlagCompression   = 0x01
	FlagTracing       = 0x02
	FlagCustomPayload = 0x04
	FlagWarning       = 0x08
	FlagBeta          = 0x10
)

var frameHeaderPool = sync.Pool{
	New: func() interface{} {
		return make([]byte, FrameHeaderSize)
	},
}

// ProtocolVersion represents a protocol version with helper methods
type ProtocolVersion byte

// GetVersion returns the numeric version (without direction bit)
func (v ProtocolVersion) GetVersion() byte {
	return byte(v) & protoVersionMask
}

// IsResponse returns true if this is a response version
func (v ProtocolVersion) IsResponse() bool {
	return byte(v)&protoDirectionMask != 0
}

// IsRequest returns true if this is a request version
func (v ProtocolVersion) IsRequest() bool {
	return byte(v)&protoDirectionMask == 0
}

// SupportsTracing returns true if protocol version supports tracing (v2+)
func (v ProtocolVersion) SupportsTracing() bool {
	return v.GetVersion() >= ProtoVersion2
}

// SupportsPreparedStatements returns true if protocol version supports prepared statements (v3+)
func (v ProtocolVersion) SupportsPreparedStatements() bool {
	return v.GetVersion() >= ProtoVersion3
}

// SupportsBatchQueries returns true if protocol version supports batch queries (v3+)
func (v ProtocolVersion) SupportsBatchQueries() bool {
	return v.GetVersion() >= ProtoVersion3
}

// SupportsCompressionFlags returns true if protocol version uses compression flags (v3+)
func (v ProtocolVersion) SupportsCompressionFlags() bool {
	return v.GetVersion() >= ProtoVersion3
}

// SupportsResultMetadataID returns true if protocol version supports result metadata ID (v4+)
func (v ProtocolVersion) SupportsResultMetadataID() bool {
	return v.GetVersion() >= ProtoVersion4
}

// SupportsCustomPayload returns true if protocol version supports custom payload (v4+)
func (v ProtocolVersion) SupportsCustomPayload() bool {
	return v.GetVersion() >= ProtoVersion4
}

// SupportsWarnings returns true if protocol version supports warnings (v4+)
func (v ProtocolVersion) SupportsWarnings() bool {
	return v.GetVersion() >= ProtoVersion4
}

// SupportsUnset returns true if protocol version supports unset values (v4+)
func (v ProtocolVersion) SupportsUnset() bool {
	return v.GetVersion() >= ProtoVersion4
}

// SupportsSegmentFormat returns true if protocol version uses segment format (v5+)
func (v ProtocolVersion) SupportsSegmentFormat() bool {
	return v.GetVersion() >= ProtoVersion5
}

// SupportsLargePayloads returns true if protocol version supports large payloads via segments (v5+)
func (v ProtocolVersion) SupportsLargePayloads() bool {
	return v.GetVersion() >= ProtoVersion5
}

// SupportsBatchFlagsV5 returns true if protocol version uses uint batch flags (v5+)
func (v ProtocolVersion) SupportsBatchFlagsV5() bool {
	return v.GetVersion() >= ProtoVersion5
}

// SupportsQueryFlagsV5 returns true if protocol version uses uint query flags (v5+)
func (v ProtocolVersion) SupportsQueryFlagsV5() bool {
	return v.GetVersion() >= ProtoVersion5
}

// SupportsNowInSeconds returns true if protocol version supports now in seconds (v5+)
func (v ProtocolVersion) SupportsNowInSeconds() bool {
	return v.GetVersion() >= ProtoVersion5
}

// SupportsKeyspaceInQuery returns true if protocol version supports keyspace in query (v5+)
func (v ProtocolVersion) SupportsKeyspaceInQuery() bool {
	return v.GetVersion() >= ProtoVersion5
}

// SupportsVectorType returns true if protocol version supports vector type (v5+)
func (v ProtocolVersion) SupportsVectorType() bool {
	return v.GetVersion() >= ProtoVersion5
}

// GetMaxStreamID returns the maximum stream ID for this version
func (v ProtocolVersion) GetMaxStreamID() int16 {
	if v.GetVersion() <= ProtoVersion2 {
		return MaxStreamIDv1v2
	}
	return MaxStreamIDv3p
}

// GetHeaderSize returns the frame header size for this version
func (v ProtocolVersion) GetHeaderSize() int {
	if v.SupportsSegmentFormat() {
		return FrameHeaderSize // Still 9 bytes for frames inside segments
	}
	return FrameHeaderSize
}

// GetFlagsSize returns the size of flags field (byte for v3-v4, uint for v5+ in some contexts)
func (v ProtocolVersion) GetFlagsSize() int {
	if v.SupportsQueryFlagsV5() {
		return 4 // uint32
	}
	return 1 // byte
}

// ResponseVersion returns the response version for a given request version
func ResponseVersion(requestVersion byte) byte {
	version := requestVersion & protoVersionMask
	return version | protoDirectionMask
}

// RequestVersion returns the request version for a given response version
func RequestVersion(responseVersion byte) byte {
	return responseVersion & protoVersionMask
}

// CQL message opcodes
const (
	OpcodeError         = 0x00
	OpcodeStartup       = 0x01
	OpcodeReady         = 0x02
	OpcodeAuthenticate  = 0x03
	OpcodeOptions       = 0x05
	OpcodeSupported     = 0x06
	OpcodeQuery         = 0x07
	OpcodeResult        = 0x08
	OpcodePrepare       = 0x09
	OpcodeExecute       = 0x0A
	OpcodeRegister      = 0x0B
	OpcodeEvent         = 0x0C
	OpcodeBatch         = 0x0D
	OpcodeAuthChallenge = 0x0E
	OpcodeAuthResponse  = 0x0F
	OpcodeAuthSuccess   = 0x10
)

// Result kinds
const (
	ResultKindVoid         = 0x0001
	ResultKindRows         = 0x0002
	ResultKindSetKeyspace  = 0x0003
	ResultKindPrepared     = 0x0004
	ResultKindSchemaChange = 0x0005
)

// Consistency levels - matching GoCQL
const (
	ConsistencyAny         = 0x0000
	ConsistencyOne         = 0x0001
	ConsistencyTwo         = 0x0002
	ConsistencyThree       = 0x0003
	ConsistencyQuorum      = 0x0004
	ConsistencyAll         = 0x0005
	ConsistencyLocalQuorum = 0x0006
	ConsistencyEachQuorum  = 0x0007
	ConsistencySerial      = 0x0008
	ConsistencyLocalSerial = 0x0009
	ConsistencyLocalOne    = 0x000A
)

// Error codes
const (
	ErrorCodeServer          = 0x0000
	ErrorCodeProtocol        = 0x000A
	ErrorCodeCredentials     = 0x0100
	ErrorCodeUnavailable     = 0x1000
	ErrorCodeOverloaded      = 0x1001
	ErrorCodeBootstrapping   = 0x1002
	ErrorCodeTruncate        = 0x1003
	ErrorCodeWriteTimeout    = 0x1100
	ErrorCodeReadTimeout     = 0x1200
	ErrorCodeReadFailure     = 0x1300
	ErrorCodeFunctionFailure = 0x1400
	ErrorCodeWriteFailure    = 0x1500
	ErrorCodeCDCWriteFailure = 0x1600
	ErrorCodeCasWriteUnknown = 0x1700
	ErrorCodeInvalid         = 0x2200
	ErrorCodeConfig          = 0x2201
	ErrorCodeAlreadyExists   = 0x2300
	ErrorCodeUnauthorized    = 0x2400
	ErrorCodeUnprepared      = 0x2500
)

// Frame represents a CQL protocol frame
type Frame struct {
	Version       byte
	Flags         byte
	Stream        int16
	Opcode        byte
	Length        int32
	Body          []byte
	TracingID     *[16]byte     // Present if FlagTracing is set
	CustomPayload CustomPayload // Present if FlagCustomPayload is set
	Warnings      []string      // Present if FlagWarning is set
}

func (f *Frame) wireLength() int {
	flags := f.effectiveFlags()
	return f.wireLengthWithFlags(flags)
}

func (f *Frame) wireLengthWithFlags(flags byte) int {
	length := len(f.Body)
	if f.Body == nil {
		length = int(f.Length)
	}
	if f.IsResponse() && flags&FlagTracing != 0 && f.TracingID != nil {
		length += 16
	}
	if flags&FlagCustomPayload != 0 && len(f.CustomPayload) > 0 {
		length += customPayloadLength(f.CustomPayload)
	}
	if flags&FlagWarning != 0 && len(f.Warnings) > 0 {
		length += warningsLength(f.Warnings)
	}
	return length
}

func (f *Frame) effectiveFlags() byte {
	flags := f.Flags
	if len(f.CustomPayload) > 0 {
		flags |= FlagCustomPayload
	} else {
		flags &^= FlagCustomPayload
	}
	if len(f.Warnings) > 0 {
		flags |= FlagWarning
	} else {
		flags &^= FlagWarning
	}
	return flags
}

func customPayloadLength(payload CustomPayload) int {
	if len(payload) == 0 {
		return 0
	}
	length := 2 // entry count
	for key, value := range payload {
		length += 2 + len(key)
		length += 4
		if value != nil {
			length += len(value)
		}
	}
	return length
}

func warningsLength(warnings []string) int {
	if len(warnings) == 0 {
		return 0
	}
	length := 2 // entry count
	for _, warning := range warnings {
		length += 2 + len(warning)
	}
	return length
}

// HeaderSize returns the size of the frame header
func (f *Frame) HeaderSize() int {
	size := 9 // version(1) + flags(1) + stream(2) + opcode(1) + length(4)
	if f.Flags&FlagTracing != 0 && f.TracingID != nil {
		size += 16 // tracing ID
	}
	return size
}

// FullSize returns the total frame size including header and body
func (f *Frame) FullSize() int {
	return f.HeaderSize() + int(f.Length)
}

// ReadFrame reads a CQL frame from the reader using GoCQL's protocol
// Supports protocol versions v3-v5
func ReadFrame(r io.Reader) (*Frame, error) {
	// Read fixed header (9 bytes)
	header := make([]byte, 9)
	if _, err := io.ReadFull(r, header); err != nil {
		return nil, fmt.Errorf("failed to read frame header: %w", err)
	}

	frame := &Frame{}
	frame.Version = header[0]
	frame.Flags = header[1]
	frame.Stream = int16(binary.BigEndian.Uint16(header[2:4]))
	frame.Opcode = header[4]
	frame.Length = int32(binary.BigEndian.Uint32(header[5:9]))

	// Validate protocol version
	protoVersion := normalizeProtoVersionByte(frame.Version)
	if protoVersion < ProtoVersion1 || protoVersion > ProtoVersion5 {
		return nil, fmt.Errorf("unsupported protocol version: %d", protoVersion)
	}

	// v1-v2 use 7-bit stream IDs, need to mask
	if protoVersion <= ProtoVersion2 {
		frame.Stream = frame.Stream & 0x007F
	}

	totalLength := frame.Length

	// Read full body (includes any tracing/custom payload/warnings prefixes).
	if totalLength > 0 {
		// Check for max frame size from GoCQL
		if totalLength > 256*1024*1024 {
			return nil, gocql.ErrFrameTooBig
		}
		frame.Body = make([]byte, totalLength)
		if _, err := io.ReadFull(r, frame.Body); err != nil {
			return nil, fmt.Errorf("failed to read frame body: %w", err)
		}
	}

	if err := parseFrameBody(frame); err != nil {
		return nil, err
	}

	return frame, nil
}

// ReadFrameVersion reads a frame with specific protocol version handling
func ReadFrameVersion(r io.Reader, version byte) (*Frame, error) {
	protoVer := version & protoVersionMask

	// For v5+, frames may be wrapped in segments
	if protoVer >= ProtoVersion5 {
		// Read segment first, then extract frame from segment payload
		segment, err := ReadSegment(r, nil)
		if err != nil {
			return nil, fmt.Errorf("failed to read segment: %w", err)
		}

		// Parse frame from segment payload
		if len(segment.Payload) < 9 {
			return nil, fmt.Errorf("segment payload too small for frame header")
		}

		frame := &Frame{}
		frame.Version = segment.Payload[0]
		frame.Flags = segment.Payload[1]
		frame.Stream = int16(binary.BigEndian.Uint16(segment.Payload[2:4]))
		frame.Opcode = segment.Payload[4]
		frame.Length = int32(binary.BigEndian.Uint32(segment.Payload[5:9]))

		// Read body from segment payload
		if frame.Length > 0 {
			if len(segment.Payload) < 9+int(frame.Length) {
				return nil, fmt.Errorf("segment payload too small for frame body")
			}
			frame.Body = segment.Payload[9 : 9+frame.Length]
		}

		// Parse warnings if present
		if err := parseFrameBody(frame); err != nil {
			return nil, err
		}

		return frame, nil
	}

	// For v1-v4, use classic frame format
	return ReadFrame(r)
}

// WriteFrameVersion writes a frame with protocol version-specific handling
func WriteFrameVersion(w io.Writer, frame *Frame, version byte) error {
	protoVer := version & protoVersionMask

	// For v5+, wrap frame in segment
	if protoVer >= ProtoVersion5 {
		// Build frame payload
		var frameBuf bytes.Buffer
		if err := frame.WriteFrame(&frameBuf); err != nil {
			return err
		}

		// Wrap in segment
		segment := &Segment{
			Payload:       frameBuf.Bytes(),
			SelfContained: true,
			Compressed:    false,
		}

		return segment.WriteSegment(w, nil)
	}

	// For v1-v4, write classic frame
	return frame.WriteFrame(w)
}

// WriteFrame writes a CQL frame to the writer
func (f *Frame) WriteFrame(w io.Writer) error {
	// Build header
	flags := f.effectiveFlags()
	header, _ := frameHeaderPool.Get().([]byte)
	if cap(header) < FrameHeaderSize {
		header = make([]byte, FrameHeaderSize)
	}
	header = header[:FrameHeaderSize]
	defer frameHeaderPool.Put(header)
	header[0] = f.Version
	header[1] = flags
	binary.BigEndian.PutUint16(header[2:4], uint16(f.Stream))
	header[4] = f.Opcode
	binary.BigEndian.PutUint32(header[5:9], uint32(f.wireLengthWithFlags(flags)))

	if _, err := w.Write(header); err != nil {
		return fmt.Errorf("failed to write frame header: %w", err)
	}

	// Write tracing ID if present
	if f.IsResponse() && flags&FlagTracing != 0 && f.TracingID != nil {
		if _, err := w.Write(f.TracingID[:]); err != nil {
			return fmt.Errorf("failed to write tracing ID: %w", err)
		}
	}

	// Write custom payload if present
	if flags&FlagCustomPayload != 0 && len(f.CustomPayload) > 0 {
		if _, err := w.Write(writeBytesMap(f.CustomPayload)); err != nil {
			return fmt.Errorf("failed to write custom payload: %w", err)
		}
	}

	// Write warnings if present
	if flags&FlagWarning != 0 && len(f.Warnings) > 0 {
		if _, err := w.Write(writeStringListFromBytes(f.Warnings)); err != nil {
			return fmt.Errorf("failed to write warnings: %w", err)
		}
	}

	// Write body
	if f.Body != nil && len(f.Body) > 0 {
		if _, err := w.Write(f.Body); err != nil {
			return fmt.Errorf("failed to write frame body: %w", err)
		}
	}

	return nil
}

// IsRequest returns true if this frame is a request (version high bit not set)
func (f *Frame) IsRequest() bool {
	return f.Version&protoDirectionMask == 0x00
}

// IsResponse returns true if this frame is a response (version high bit set)
func (f *Frame) IsResponse() bool {
	return f.Version&protoDirectionMask == protoDirectionMask
}

func normalizeProtoVersionByte(version byte) byte {
	v := version & protoVersionMask
	if v <= ProtoVersion5 {
		return v
	}
	// Accept vendor-extended versions (e.g., DSE) where 0x40 is set.
	if v&0x40 != 0 {
		base := v & 0x3F
		if base >= ProtoVersion3 && base <= ProtoVersion5 {
			return base
		}
		// Older vendor versions map to v4 for compatibility.
		if base >= ProtoVersion1 && base <= ProtoVersion2 {
			return ProtoVersion4
		}
	}
	return v
}

// GetProtocolVersion extracts the protocol version from the version byte
func (f *Frame) GetProtocolVersion() byte {
	return normalizeProtoVersionByte(f.Version)
}

// String returns a string representation of the frame
// String returns a human-readable frame summary.
func (f *Frame) String() string {
	return fmt.Sprintf("Frame{Version: %d, Flags: 0x%02x, Stream: %d, Opcode: 0x%02x, Length: %d}",
		f.GetProtocolVersion(), f.Flags, f.Stream, f.Opcode, f.Length)
}

// MessageType returns the string representation of the opcode
type MessageType byte

// String returns the opcode name.
func (m MessageType) String() string {
	switch m {
	case OpcodeError:
		return "ERROR"
	case OpcodeStartup:
		return "STARTUP"
	case OpcodeReady:
		return "READY"
	case OpcodeAuthenticate:
		return "AUTHENTICATE"
	case OpcodeOptions:
		return "OPTIONS"
	case OpcodeSupported:
		return "SUPPORTED"
	case OpcodeQuery:
		return "QUERY"
	case OpcodeResult:
		return "RESULT"
	case OpcodePrepare:
		return "PREPARE"
	case OpcodeExecute:
		return "EXECUTE"
	case OpcodeRegister:
		return "REGISTER"
	case OpcodeEvent:
		return "EVENT"
	case OpcodeBatch:
		return "BATCH"
	case OpcodeAuthChallenge:
		return "AUTH_CHALLENGE"
	case OpcodeAuthResponse:
		return "AUTH_RESPONSE"
	case OpcodeAuthSuccess:
		return "AUTH_SUCCESS"
	default:
		return fmt.Sprintf("UNKNOWN_OP_%d", m)
	}
}

// Helper functions for encoding/decoding using GoCQL's marshaling

// encodeStringMultimap encodes a string multimap to CQL protocol format
func encodeStringMultimap(options map[string][]string) []byte {
	var buf bytes.Buffer

	// Number of entries
	binary.Write(&buf, binary.BigEndian, uint16(len(options)))

	for key, values := range options {
		// Key length and key
		keyBytes := []byte(key)
		binary.Write(&buf, binary.BigEndian, uint16(len(keyBytes)))
		buf.Write(keyBytes)

		// Number of values
		binary.Write(&buf, binary.BigEndian, uint16(len(values)))

		// Each value
		for _, value := range values {
			valueBytes := []byte(value)
			binary.Write(&buf, binary.BigEndian, uint16(len(valueBytes)))
			buf.Write(valueBytes)
		}
	}

	return buf.Bytes()
}

// decodeStringMultimap decodes a CQL protocol string multimap
func decodeStringMultimap(data []byte) (map[string][]string, error) {
	if len(data) < 2 {
		return make(map[string][]string), nil
	}

	result := make(map[string][]string)
	buf := bytes.NewReader(data)

	// Read number of entries
	var numEntries uint16
	if err := binary.Read(buf, binary.BigEndian, &numEntries); err != nil {
		return nil, err
	}

	for i := 0; i < int(numEntries); i++ {
		// Read key
		var keyLen uint16
		if err := binary.Read(buf, binary.BigEndian, &keyLen); err != nil {
			return nil, err
		}
		keyBytes := make([]byte, keyLen)
		if _, err := io.ReadFull(buf, keyBytes); err != nil {
			return nil, err
		}

		// Read number of values
		var numValues uint16
		if err := binary.Read(buf, binary.BigEndian, &numValues); err != nil {
			return nil, err
		}

		values := make([]string, numValues)
		for j := 0; j < int(numValues); j++ {
			// Read value length
			var valueLen uint16
			if err := binary.Read(buf, binary.BigEndian, &valueLen); err != nil {
				return nil, err
			}
			// Read value
			valueBytes := make([]byte, valueLen)
			if _, err := io.ReadFull(buf, valueBytes); err != nil {
				return nil, err
			}
			values[j] = string(valueBytes)
		}

		result[string(keyBytes)] = values
	}

	return result, nil
}

// readStringList reads a string list from a byte buffer
func readStringList(data []byte) []string {
	if len(data) < 2 {
		return nil
	}

	buf := bytes.NewReader(data)
	var numStrings uint16
	if err := binary.Read(buf, binary.BigEndian, &numStrings); err != nil {
		return nil
	}

	result := make([]string, numStrings)
	for i := 0; i < int(numStrings); i++ {
		var strLen uint16
		if err := binary.Read(buf, binary.BigEndian, &strLen); err != nil {
			return result
		}
		strBytes := make([]byte, strLen)
		if _, err := io.ReadFull(buf, strBytes); err != nil {
			return result
		}
		result[i] = string(strBytes)
	}

	return result
}

// writeStringList writes a string list to bytes
func writeStringList(strings []string) []byte {
	total := 2
	for _, s := range strings {
		total += 2 + len(s)
	}
	buf := make([]byte, total)
	binary.BigEndian.PutUint16(buf[0:2], uint16(len(strings)))
	offset := 2
	for _, s := range strings {
		strBytes := []byte(s)
		binary.BigEndian.PutUint16(buf[offset:offset+2], uint16(len(strBytes)))
		offset += 2
		copy(buf[offset:], strBytes)
		offset += len(strBytes)
	}
	return buf
}

// writeInt writes an int32 to bytes in big endian format
func writeInt(v int32) []byte {
	buf := make([]byte, 4)
	binary.BigEndian.PutUint32(buf, uint32(v))
	return buf
}

// writeShort writes a uint16 to bytes in big endian format
func writeShort(v uint16) []byte {
	buf := make([]byte, 2)
	binary.BigEndian.PutUint16(buf, v)
	return buf
}

// writeString writes a string to bytes in CQL format (length prefixed)
func writeString(s string) []byte {
	strBytes := []byte(s)
	buf := make([]byte, 2+len(strBytes))
	binary.BigEndian.PutUint16(buf[0:2], uint16(len(strBytes)))
	copy(buf[2:], strBytes)
	return buf
}

// writeLongString writes a long string to bytes in CQL format (length prefixed with int32)
func writeLongString(s string) []byte {
	strBytes := []byte(s)
	buf := make([]byte, 4+len(strBytes))
	binary.BigEndian.PutUint32(buf[0:4], uint32(len(strBytes)))
	copy(buf[4:], strBytes)
	return buf
}

// readString reads a string from bytes in CQL format (length prefixed)
func readString(data []byte) (string, int, error) {
	if len(data) < 2 {
		return "", 0, fmt.Errorf("not enough bytes to read string header")
	}

	length := int(binary.BigEndian.Uint16(data[0:2]))
	if len(data) < 2+length {
		return "", 0, fmt.Errorf("not enough bytes to read string value")
	}

	return string(data[2 : 2+length]), 2 + length, nil
}

// readBytes reads bytes from bytes in CQL format (length prefixed)
func readBytes(data []byte) ([]byte, int, error) {
	if len(data) < 4 {
		return nil, 0, fmt.Errorf("not enough bytes to read bytes header")
	}

	length := int32(binary.BigEndian.Uint32(data[0:4]))
	if length == NullValue || length == UnsetValue {
		return nil, 4, nil
	}
	if length < 0 {
		return nil, 0, fmt.Errorf("invalid bytes length %d", length)
	}
	if len(data) < 4+int(length) {
		return nil, 0, fmt.Errorf("not enough bytes to read bytes value")
	}

	return data[4 : 4+int(length)], 4 + int(length), nil
}

// readShortBytes reads short bytes from bytes in CQL format (length prefixed with uint16)
func readShortBytes(data []byte) ([]byte, int, error) {
	if len(data) < 2 {
		return nil, 0, fmt.Errorf("not enough bytes to read short bytes header")
	}

	length := int(binary.BigEndian.Uint16(data[0:2]))
	if len(data) < 2+length {
		return nil, 0, fmt.Errorf("not enough bytes to read short bytes value")
	}

	return data[2 : 2+length], 2 + length, nil
}

// readLongString reads a long string from bytes in CQL format (length prefixed with int32)
func readLongString(data []byte) (string, int, error) {
	if len(data) < 4 {
		return "", 0, fmt.Errorf("not enough bytes to read long string header")
	}

	length := int(binary.BigEndian.Uint32(data[0:4]))
	if len(data) < 4+length {
		return "", 0, fmt.Errorf("not enough bytes to read long string value")
	}

	return string(data[4 : 4+length]), 4 + length, nil
}

// readUUID reads a UUID from bytes in CQL format
func readUUID(data []byte) ([]byte, int, error) {
	if len(data) < 16 {
		return nil, 0, fmt.Errorf("not enough bytes to read UUID")
	}

	return data[0:16], 16, nil
}

// readInet reads an INET address from bytes in CQL format
func readInet(data []byte) ([]byte, int, error) {
	if len(data) < 1 {
		return nil, 0, fmt.Errorf("not enough bytes to read INET header")
	}

	length := int(data[0])
	if len(data) < 1+length {
		return nil, 0, fmt.Errorf("not enough bytes to read INET value")
	}

	return data[1 : 1+length], 1 + length, nil
}

// readList reads a list from bytes in CQL format
func readList(data []byte) ([]byte, int, error) {
	if len(data) < 4 {
		return nil, 0, fmt.Errorf("not enough bytes to read list header")
	}

	length := int(binary.BigEndian.Uint32(data[0:4]))
	if len(data) < 4+length {
		return nil, 0, fmt.Errorf("not enough bytes to read list value")
	}

	return data[4 : 4+length], 4 + length, nil
}

// readMap reads a map from bytes in CQL format
func readMap(data []byte) ([]byte, int, error) {
	if len(data) < 4 {
		return nil, 0, fmt.Errorf("not enough bytes to read map header")
	}

	length := int(binary.BigEndian.Uint32(data[0:4]))
	if len(data) < 4+length {
		return nil, 0, fmt.Errorf("not enough bytes to read map value")
	}

	return data[4 : 4+length], 4 + length, nil
}

// readSet reads a set from bytes in CQL format
func readSet(data []byte) ([]byte, int, error) {
	if len(data) < 4 {
		return nil, 0, fmt.Errorf("not enough bytes to read set header")
	}

	length := int(binary.BigEndian.Uint32(data[0:4]))
	if len(data) < 4+length {
		return nil, 0, fmt.Errorf("not enough bytes to read set value")
	}

	return data[4 : 4+length], 4 + length, nil
}

// readTuple reads a tuple from bytes in CQL format
func readTuple(data []byte) ([]byte, int, error) {
	if len(data) < 4 {
		return nil, 0, fmt.Errorf("not enough bytes to read tuple header")
	}

	length := int(binary.BigEndian.Uint32(data[0:4]))
	if len(data) < 4+length {
		return nil, 0, fmt.Errorf("not enough bytes to read tuple value")
	}

	return data[4 : 4+length], 4 + length, nil
}

// readUDT reads a user-defined type from bytes in CQL format
func readUDT(data []byte) ([]byte, int, error) {
	if len(data) < 4 {
		return nil, 0, fmt.Errorf("not enough bytes to read UDT header")
	}

	length := int(binary.BigEndian.Uint32(data[0:4]))
	if len(data) < 4+length {
		return nil, 0, fmt.Errorf("not enough bytes to read UDT value")
	}

	return data[4 : 4+length], 4 + length, nil
}

// readCustom reads a custom type from bytes in CQL format
func readCustom(data []byte) ([]byte, int, error) {
	if len(data) < 4 {
		return nil, 0, fmt.Errorf("not enough bytes to read custom header")
	}

	length := int(binary.BigEndian.Uint32(data[0:4]))
	if len(data) < 4+length {
		return nil, 0, fmt.Errorf("not enough bytes to read custom value")
	}

	return data[4 : 4+length], 4 + length, nil
}

// writeBytes writes bytes in CQL format (length prefixed with int32, -1 for nil)
func writeBytes(data []byte) []byte {
	var buf bytes.Buffer
	if data == nil {
		binary.Write(&buf, binary.BigEndian, int32(-1))
	} else {
		binary.Write(&buf, binary.BigEndian, int32(len(data)))
		buf.Write(data)
	}
	return buf.Bytes()
}

// writeConsistency writes a consistency level
func writeConsistency(consistency uint16) []byte {
	buf := make([]byte, 2)
	binary.BigEndian.PutUint16(buf, consistency)
	return buf
}

// writeShortBytes writes short bytes in CQL format (length prefixed with uint16)
func writeShortBytes(data []byte) []byte {
	var buf bytes.Buffer
	if data == nil {
		binary.Write(&buf, binary.BigEndian, uint16(0xFFFF)) // -1 for nil
	} else {
		binary.Write(&buf, binary.BigEndian, uint16(len(data)))
		buf.Write(data)
	}
	return buf.Bytes()
}

// writeUUID writes a UUID to bytes in CQL format
func writeUUID(uuid []byte) []byte {
	return uuid
}

// writeList writes a list to bytes in CQL format
func writeList(list []byte) []byte {
	var buf bytes.Buffer
	if list == nil {
		binary.Write(&buf, binary.BigEndian, int32(-1)) // -1 for nil
	} else {
		binary.Write(&buf, binary.BigEndian, int32(len(list)))
		buf.Write(list)
	}
	return buf.Bytes()
}

// writeMap writes a map to bytes in CQL format
func writeMap(m map[string][]byte) []byte {
	var buf bytes.Buffer
	if m == nil {
		binary.Write(&buf, binary.BigEndian, int32(-1)) // -1 for nil
	} else {
		binary.Write(&buf, binary.BigEndian, int32(len(m)))
		for key, value := range m {
			keyBytes := []byte(key)
			binary.Write(&buf, binary.BigEndian, uint16(len(keyBytes)))
			buf.Write(keyBytes)
			binary.Write(&buf, binary.BigEndian, uint16(len(value)))
			buf.Write(value)
		}
	}
	return buf.Bytes()
}

// writeSet writes a set to bytes in CQL format
func writeSet(set []byte) []byte {
	var buf bytes.Buffer
	if set == nil {
		binary.Write(&buf, binary.BigEndian, int32(-1)) // -1 for nil
	} else {
		binary.Write(&buf, binary.BigEndian, int32(len(set)))
		buf.Write(set)
	}
	return buf.Bytes()
}

// writeTuple writes a tuple to bytes in CQL format
func writeTuple(tuple []byte) []byte {
	var buf bytes.Buffer
	if tuple == nil {
		binary.Write(&buf, binary.BigEndian, int32(-1)) // -1 for nil
	} else {
		binary.Write(&buf, binary.BigEndian, int32(len(tuple)))
		buf.Write(tuple)
	}
	return buf.Bytes()
}

// writeUDT writes a user-defined type to bytes in CQL format
func writeUDT(udt []byte) []byte {
	var buf bytes.Buffer
	if udt == nil {
		binary.Write(&buf, binary.BigEndian, int32(-1)) // -1 for nil
	} else {
		binary.Write(&buf, binary.BigEndian, int32(len(udt)))
		buf.Write(udt)
	}
	return buf.Bytes()
}

// writeCustom writes a custom type to bytes in CQL format
func writeCustom(custom []byte) []byte {
	var buf bytes.Buffer
	if custom == nil {
		binary.Write(&buf, binary.BigEndian, int32(-1)) // -1 for nil
	} else {
		binary.Write(&buf, binary.BigEndian, int32(len(custom)))
		buf.Write(custom)
	}
	return buf.Bytes()
}

// readStringMap reads a string map from bytes in CQL format
func readStringMap(data []byte) (map[string]string, error) {
	if len(data) < 2 {
		return make(map[string]string), nil
	}

	result := make(map[string]string)
	buf := bytes.NewReader(data)

	var numEntries uint16
	if err := binary.Read(buf, binary.BigEndian, &numEntries); err != nil {
		return nil, err
	}

	for i := 0; i < int(numEntries); i++ {
		var keyLen uint16
		if err := binary.Read(buf, binary.BigEndian, &keyLen); err != nil {
			return nil, err
		}
		keyBytes := make([]byte, keyLen)
		if _, err := io.ReadFull(buf, keyBytes); err != nil {
			return nil, err
		}

		var valueLen uint16
		if err := binary.Read(buf, binary.BigEndian, &valueLen); err != nil {
			return nil, err
		}
		valueBytes := make([]byte, valueLen)
		if _, err := io.ReadFull(buf, valueBytes); err != nil {
			return nil, err
		}

		result[string(keyBytes)] = string(valueBytes)
	}

	return result, nil
}

// writeStringMap writes a string map to bytes in CQL format
func writeStringMap(m map[string]string) []byte {
	var buf bytes.Buffer
	binary.Write(&buf, binary.BigEndian, uint16(len(m)))
	for k, v := range m {
		kb := []byte(k)
		vb := []byte(v)
		binary.Write(&buf, binary.BigEndian, uint16(len(kb)))
		buf.Write(kb)
		binary.Write(&buf, binary.BigEndian, uint16(len(vb)))
		buf.Write(vb)
	}
	return buf.Bytes()
}

// readBytesMap reads a bytes map from bytes in CQL format.
func readBytesMap(data []byte) (map[string][]byte, int, error) {
	if len(data) < 2 {
		return nil, 0, fmt.Errorf("not enough bytes to read bytes map")
	}

	numEntries := int(binary.BigEndian.Uint16(data[0:2]))
	offset := 2
	result := make(map[string][]byte, numEntries)

	for i := 0; i < numEntries; i++ {
		if len(data) < offset+2 {
			return nil, 0, fmt.Errorf("not enough bytes to read bytes map key length")
		}
		keyLen := int(binary.BigEndian.Uint16(data[offset : offset+2]))
		offset += 2
		if len(data) < offset+keyLen {
			return nil, 0, fmt.Errorf("not enough bytes to read bytes map key")
		}
		key := string(data[offset : offset+keyLen])
		offset += keyLen

		if len(data) < offset+4 {
			return nil, 0, fmt.Errorf("not enough bytes to read bytes map value length")
		}
		valueLen := int32(binary.BigEndian.Uint32(data[offset : offset+4]))
		offset += 4

		switch valueLen {
		case NullValue, UnsetValue:
			result[key] = nil
			continue
		default:
			if valueLen < 0 {
				return nil, 0, fmt.Errorf("invalid bytes map value length %d", valueLen)
			}
			if len(data) < offset+int(valueLen) {
				return nil, 0, fmt.Errorf("not enough bytes to read bytes map value")
			}
			result[key] = data[offset : offset+int(valueLen)]
			offset += int(valueLen)
		}
	}

	return result, offset, nil
}

// writeBytesMap writes a bytes map to bytes in CQL format
func writeBytesMap(m map[string][]byte) []byte {
	total := 2
	for k, v := range m {
		total += 2 + len(k) + 4
		if v != nil {
			total += len(v)
		}
	}
	buf := make([]byte, total)
	binary.BigEndian.PutUint16(buf[0:2], uint16(len(m)))
	offset := 2
	for k, v := range m {
		kb := []byte(k)
		binary.BigEndian.PutUint16(buf[offset:offset+2], uint16(len(kb)))
		offset += 2
		copy(buf[offset:], kb)
		offset += len(kb)
		if v == nil {
			binary.BigEndian.PutUint32(buf[offset:offset+4], 0xFFFFFFFF)
			offset += 4
			continue
		}
		binary.BigEndian.PutUint32(buf[offset:offset+4], uint32(len(v)))
		offset += 4
		copy(buf[offset:], v)
		offset += len(v)
	}
	return buf
}

// readLong reads a long (int64) from bytes
func readLong(data []byte) (int64, int, error) {
	if len(data) < 8 {
		return 0, 0, fmt.Errorf("not enough bytes to read long")
	}
	val := int64(binary.BigEndian.Uint64(data[0:8]))
	return val, 8, nil
}

// writeLong writes a long (int64) to bytes
func writeLong(v int64) []byte {
	buf := make([]byte, 8)
	binary.BigEndian.PutUint64(buf, uint64(v))
	return buf
}

// readShort reads a short (uint16) from bytes
func readShort(data []byte) (uint16, int, error) {
	if len(data) < 2 {
		return 0, 0, fmt.Errorf("not enough bytes to read short")
	}
	val := binary.BigEndian.Uint16(data[0:2])
	return val, 2, nil
}

// readInt reads an int (int32) from bytes
func readInt(data []byte) (int32, int, error) {
	if len(data) < 4 {
		return 0, 0, fmt.Errorf("not enough bytes to read int")
	}
	val := int32(binary.BigEndian.Uint32(data[0:4]))
	return val, 4, nil
}

// readConsistency reads a consistency level from bytes
func readConsistency(data []byte) (uint16, int, error) {
	return readShort(data)
}

// readByte reads a single byte
func readByte(data []byte) (byte, int, error) {
	if len(data) < 1 {
		return 0, 0, fmt.Errorf("not enough bytes to read byte")
	}
	return data[0], 1, nil
}

// readInetAddress reads an IP address (without port)
func readInetAddress(data []byte) ([]byte, int, error) {
	if len(data) < 1 {
		return nil, 0, fmt.Errorf("not enough bytes to read inet address")
	}
	length := int(data[0])
	if len(data) < 1+length {
		return nil, 0, fmt.Errorf("not enough bytes to read inet address value")
	}
	return data[1 : 1+length], 1 + length, nil
}

// readInet reads an IP address and port
func readInetFull(data []byte) ([]byte, int, int, error) {
	addr, n, err := readInetAddress(data)
	if err != nil {
		return nil, 0, 0, err
	}
	port, m, err := readShort(data[n:])
	if err != nil {
		return nil, 0, 0, err
	}
	return addr, n + m, int(port), nil
}

// writeInetAddress writes an IP address and port
func writeInetAddress(addr []byte, port int) []byte {
	var buf bytes.Buffer
	if addr == nil {
		buf.WriteByte(0)
	} else {
		buf.WriteByte(byte(len(addr)))
		buf.Write(addr)
	}
	binary.Write(&buf, binary.BigEndian, uint16(port))
	return buf.Bytes()
}

// readErrorMap reads an error map (v5+)
func readErrorMap(data []byte) (map[string]uint16, error) {
	if len(data) < 4 {
		return nil, fmt.Errorf("not enough bytes for error map")
	}
	numErrs := int32(binary.BigEndian.Uint32(data[0:4]))
	result := make(map[string]uint16, numErrs)
	offset := 4
	for i := 0; i < int(numErrs); i++ {
		addr, n, err := readInetAddress(data[offset:])
		if err != nil {
			return nil, err
		}
		offset += n
		port, n, err := readShort(data[offset:])
		if err != nil {
			return nil, err
		}
		offset += n
		// Store as "ip:port" -> error code
		key := fmt.Sprintf("%s:%d", string(addr), port)
		result[key] = port
	}
	return result, nil
}

// writeErrorMap writes an error map (v5+)
func writeErrorMap(m map[string]uint16) []byte {
	var buf bytes.Buffer
	binary.Write(&buf, binary.BigEndian, int32(len(m)))
	for k, v := range m {
		// Parse "ip:port" format
		// For simplicity, we'll just write the value
		// In real implementation, would need to parse IP and port
		buf.Write(writeInetAddress([]byte(k), int(v)))
	}
	return buf.Bytes()
}

// readStringList reads a string list from bytes
func readStringListWithLength(data []byte) ([]string, int, error) {
	if len(data) < 2 {
		return nil, 0, fmt.Errorf("not enough bytes to read string list")
	}
	count := int(binary.BigEndian.Uint16(data[0:2]))
	offset := 2
	result := make([]string, count)
	for i := 0; i < count; i++ {
		str, n, err := readString(data[offset:])
		if err != nil {
			return nil, 0, err
		}
		result[i] = str
		offset += n
	}
	return result, offset, nil
}

func readStringListFromBytes(data []byte) ([]string, error) {
	if len(data) < 2 {
		return nil, nil
	}
	count := int(binary.BigEndian.Uint16(data[0:2]))
	result := make([]string, count)
	offset := 2
	for i := 0; i < count; i++ {
		str, n, err := readString(data[offset:])
		if err != nil {
			return nil, err
		}
		result[i] = str
		offset += n
	}
	return result, nil
}

// writeStringList writes a string list to bytes
func writeStringListFromBytes(list []string) []byte {
	return writeStringList(list)
}

// readTypeInfo reads type information
func readTypeInfo(data []byte) (TypeInfo, int, error) {
	if len(data) < 2 {
		return nil, 0, fmt.Errorf("not enough bytes for type info")
	}
	typeCode := binary.BigEndian.Uint16(data[0:2])
	offset := 2

	switch Type(typeCode) {
	case TypeCustom:
		typeName, n, err := readString(data[offset:])
		if err != nil {
			return nil, 0, err
		}
		offset += n
		return &CustomTypeInfo{Name: typeName}, offset, nil
	case TypeList:
		elemType, n, err := readTypeInfo(data[offset:])
		if err != nil {
			return nil, 0, err
		}
		offset += n
		return &ListTypeInfo{Elem: elemType}, offset, nil
	case TypeSet:
		elemType, n, err := readTypeInfo(data[offset:])
		if err != nil {
			return nil, 0, err
		}
		offset += n
		return &SetTypeInfo{Elem: elemType}, offset, nil
	case TypeMap:
		keyType, n1, err := readTypeInfo(data[offset:])
		if err != nil {
			return nil, 0, err
		}
		offset += n1
		valueType, n2, err := readTypeInfo(data[offset:])
		if err != nil {
			return nil, 0, err
		}
		offset += n2
		return &MapTypeInfo{Key: keyType, Value: valueType}, offset, nil
	case TypeUDT:
		keyspace, n1, err := readString(data[offset:])
		if err != nil {
			return nil, 0, err
		}
		offset += n1
		typeName, n2, err := readString(data[offset:])
		if err != nil {
			return nil, 0, err
		}
		offset += n2
		var numFields uint16
		if len(data[offset:]) < 2 {
			return nil, 0, fmt.Errorf("not enough bytes for UDT fields")
		}
		numFields = binary.BigEndian.Uint16(data[offset : offset+2])
		offset += 2
		fields := make([]UDTField, numFields)
		for i := 0; i < int(numFields); i++ {
			fieldName, n, err := readString(data[offset:])
			if err != nil {
				return nil, 0, err
			}
			offset += n
			fieldType, n, err := readTypeInfo(data[offset:])
			if err != nil {
				return nil, 0, err
			}
			offset += n
			fields[i] = UDTField{Name: fieldName, Type: fieldType}
		}
		return &UDTTypeInfo{Keyspace: keyspace, Name: typeName, Fields: fields}, offset, nil
	case TypeTuple:
		var numElems uint16
		if len(data[offset:]) < 2 {
			return nil, 0, fmt.Errorf("not enough bytes for tuple elements")
		}
		numElems = binary.BigEndian.Uint16(data[offset : offset+2])
		offset += 2
		elems := make([]TypeInfo, numElems)
		for i := 0; i < int(numElems); i++ {
			elemType, n, err := readTypeInfo(data[offset:])
			if err != nil {
				return nil, 0, err
			}
			offset += n
			elems[i] = elemType
		}
		return &TupleTypeInfo{Elems: elems}, offset, nil
	case TypeVector:
		elemType, n, err := readTypeInfo(data[offset:])
		if err != nil {
			return nil, 0, err
		}
		offset += n
		if len(data[offset:]) < 4 {
			return nil, 0, fmt.Errorf("not enough bytes for vector")
		}
		size := binary.BigEndian.Uint32(data[offset : offset+4])
		offset += 4
		return &VectorTypeInfo{Size: size, Elem: elemType}, offset, nil
	default:
		return &NativeTypeInfo{TypeCode: Type(typeCode)}, 2, nil
	}
}

// writeTypeInfo writes type information
func writeTypeInfo(ti TypeInfo) []byte {
	var buf bytes.Buffer
	code := ti.Code()
	binary.Write(&buf, binary.BigEndian, uint16(code))

	switch t := ti.(type) {
	case *CustomTypeInfo:
		buf.Write(writeString(t.Name))
	case *ListTypeInfo:
		buf.Write(writeTypeInfo(t.Elem))
	case *SetTypeInfo:
		buf.Write(writeTypeInfo(t.Elem))
	case *MapTypeInfo:
		buf.Write(writeTypeInfo(t.Key))
		buf.Write(writeTypeInfo(t.Value))
	case *UDTTypeInfo:
		buf.Write(writeString(t.Keyspace))
		buf.Write(writeString(t.Name))
		binary.Write(&buf, binary.BigEndian, uint16(len(t.Fields)))
		for _, field := range t.Fields {
			buf.Write(writeString(field.Name))
			buf.Write(writeTypeInfo(field.Type))
		}
	case *TupleTypeInfo:
		binary.Write(&buf, binary.BigEndian, uint16(len(t.Elems)))
		for _, elem := range t.Elems {
			buf.Write(writeTypeInfo(elem))
		}
	case *VectorTypeInfo:
		buf.Write(writeTypeInfo(t.Elem))
		binary.Write(&buf, binary.BigEndian, uint32(t.Size))
	}

	return buf.Bytes()
}

// TypeInfo represents CQL type information
type TypeInfo interface {
	Code() Type
}

// NativeTypeInfo represents a native CQL type
type NativeTypeInfo struct {
	TypeCode Type
}

// Code returns the CQL type code.
func (t *NativeTypeInfo) Code() Type {
	return t.TypeCode
}

// CustomTypeInfo represents a custom type
type CustomTypeInfo struct {
	Name string
}

// Code returns the CQL type code.
func (t *CustomTypeInfo) Code() Type {
	return TypeCustom
}

// ListTypeInfo represents a list type
type ListTypeInfo struct {
	Elem TypeInfo
}

// Code returns the CQL type code.
func (t *ListTypeInfo) Code() Type {
	return TypeList
}

// SetTypeInfo represents a set type
type SetTypeInfo struct {
	Elem TypeInfo
}

// Code returns the CQL type code.
func (t *SetTypeInfo) Code() Type {
	return TypeSet
}

// MapTypeInfo represents a map type
type MapTypeInfo struct {
	Key   TypeInfo
	Value TypeInfo
}

// Code returns the CQL type code.
func (t *MapTypeInfo) Code() Type {
	return TypeMap
}

// UDTField represents a field in a UDT
type UDTField struct {
	Name string
	Type TypeInfo
}

// UDTTypeInfo represents a user-defined type
type UDTTypeInfo struct {
	Keyspace string
	Name     string
	Fields   []UDTField
}

// Code returns the CQL type code.
func (t *UDTTypeInfo) Code() Type {
	return TypeUDT
}

// TupleTypeInfo represents a tuple type
type TupleTypeInfo struct {
	Elems []TypeInfo
}

// Code returns the CQL type code.
func (t *TupleTypeInfo) Code() Type {
	return TypeTuple
}

// VectorTypeInfo represents a vector type (v5+)
type VectorTypeInfo struct {
	Size uint32
	Elem TypeInfo
}

// Code returns the CQL type code.
func (t *VectorTypeInfo) Code() Type {
	return TypeVector
}

// ColumnInfo represents column metadata
type ColumnInfo struct {
	Keyspace string
	Table    string
	Name     string
	TypeInfo TypeInfo
}

// CustomPayload represents custom payload for frames
type CustomPayload map[string][]byte

// FrameBuilder helps build frames
type FrameBuilder struct {
	version byte
	flags   byte
	stream  int16
	opcode  byte
	body    []byte
}

// NewFrameBuilder creates a new frame builder
func NewFrameBuilder() *FrameBuilder {
	return &FrameBuilder{
		version: ProtoVersion4,
		stream:  0,
	}
}

// Version sets the frame version.
func (fb *FrameBuilder) Version(v byte) *FrameBuilder {
	fb.version = v
	return fb
}

// Flags sets the frame flags.
func (fb *FrameBuilder) Flags(f byte) *FrameBuilder {
	fb.flags = f
	return fb
}

// Stream sets the frame stream ID.
func (fb *FrameBuilder) Stream(s int16) *FrameBuilder {
	fb.stream = s
	return fb
}

// Opcode sets the frame opcode.
func (fb *FrameBuilder) Opcode(o byte) *FrameBuilder {
	fb.opcode = o
	return fb
}

// Body sets the frame body.
func (fb *FrameBuilder) Body(b []byte) *FrameBuilder {
	fb.body = b
	return fb
}

// Build constructs the frame using the configured fields.
func (fb *FrameBuilder) Build() *Frame {
	return &Frame{
		Version: fb.version,
		Flags:   fb.flags,
		Stream:  fb.stream,
		Opcode:  fb.opcode,
		Length:  int32(len(fb.body)),
		Body:    fb.body,
	}
}

// BuildStartupFrame creates a STARTUP frame
func BuildStartupFrame(version string, driverName string, driverVersion string, compression string) *Frame {
	options := make(map[string]string)
	options["CQL_VERSION"] = version
	if driverName != "" {
		options["DRIVER_NAME"] = driverName
	}
	if driverVersion != "" {
		options["DRIVER_VERSION"] = driverVersion
	}
	if compression != "" {
		options["COMPRESSION"] = compression
	}

	body := writeStringMap(options)
	return NewFrameBuilder().Opcode(OpcodeStartup).Body(body).Build()
}

// BuildOptionsFrame creates an OPTIONS frame
func BuildOptionsFrame() *Frame {
	return NewFrameBuilder().Opcode(OpcodeOptions).Body([]byte{}).Build()
}

// BuildQueryFrame creates a QUERY frame
func BuildQueryFrame(query string, consistency uint16, flags uint32, values []QueryValue, pageSize int, pagingState []byte, serialConsistency uint16, timestamp int64, keyspace string, nowInSeconds *int32) *Frame {
	var buf bytes.Buffer

	// Query string
	buf.Write(writeLongString(query))

	// Consistency
	buf.Write(writeConsistency(consistency))

	// Flags (v5+ uses uint32, v3-v4 uses byte)
	if flags > 0xFF {
		buf.Write([]byte{byte(flags), byte(flags >> 8), byte(flags >> 16), byte(flags >> 24)})
	} else {
		buf.Write([]byte{byte(flags)})
	}

	// Values
	if flags&QueryFlagValues != 0 {
		if flags&QueryFlagWithNameValues != 0 {
			// Named values
			buf.Write(writeShort(uint16(len(values))))
			for _, v := range values {
				buf.Write(writeString(v.Name))
				writeQueryValue(&buf, v)
			}
		} else {
			// Positional values
			buf.Write(writeShort(uint16(len(values))))
			for _, v := range values {
				writeQueryValue(&buf, v)
			}
		}
	}

	// Page size
	if flags&QueryFlagPageSize != 0 {
		buf.Write(writeInt(int32(pageSize)))
	}

	// Paging state
	if flags&QueryFlagWithPagingState != 0 {
		buf.Write(writeBytes(pagingState))
	}

	// Serial consistency
	if flags&QueryFlagWithSerialConsistency != 0 {
		buf.Write(writeConsistency(serialConsistency))
	}

	// Default timestamp
	if flags&QueryFlagWithDefaultTimestamp != 0 {
		buf.Write(writeLong(timestamp))
	}

	// Keyspace (v5+)
	if flags&QueryFlagWithKeyspace != 0 {
		buf.Write(writeString(keyspace))
	}

	// Now in seconds (v5+)
	if flags&QueryFlagNowInSeconds != 0 && nowInSeconds != nil {
		buf.Write(writeInt(*nowInSeconds))
	}

	return NewFrameBuilder().Opcode(OpcodeQuery).Body(buf.Bytes()).Build()
}

// QueryValue represents a query parameter value
type QueryValue struct {
	Name  string
	Value []byte
	Unset bool
}

func writeQueryValue(buf *bytes.Buffer, value QueryValue) {
	if value.Unset {
		buf.Write(writeInt(int32(UnsetValue)))
		return
	}
	buf.Write(writeBytes(value.Value))
}

// BuildAuthResponseFrame creates an AUTH_RESPONSE frame
func BuildAuthResponseFrame(username, password string) *Frame {
	// Format: [0][username][0][password]
	payload := make([]byte, 1+len(username)+1+len(password))
	payload[0] = 0
	copy(payload[1:], username)
	payload[1+len(username)] = 0
	copy(payload[2+len(username):], password)

	body := writeBytes(payload)
	return NewFrameBuilder().Opcode(OpcodeAuthResponse).Body(body).Build()
}

// BuildRegisterFrame creates a REGISTER frame
func BuildRegisterFrame(events []string) *Frame {
	body := writeStringListFromBytes(events)
	return NewFrameBuilder().Opcode(OpcodeRegister).Body(body).Build()
}

// BuildPrepareFrame creates a PREPARE frame
func BuildPrepareFrame(statement string, keyspace string) *Frame {
	var buf bytes.Buffer
	buf.Write(writeLongString(statement))

	if keyspace != "" {
		// Protocol v5+ with keyspace
		buf.Write([]byte{PrepareFlagWithKeyspace})
		buf.Write(writeString(keyspace))
	}

	return NewFrameBuilder().Opcode(OpcodePrepare).Body(buf.Bytes()).Build()
}

// BuildExecuteFrame creates an EXECUTE frame
func BuildExecuteFrame(preparedID []byte, resultMetadataID []byte, consistency uint16, flags uint32, values []QueryValue, pageSize int, pagingState []byte, serialConsistency uint16, timestamp int64, keyspace string, nowInSeconds *int32) *Frame {
	var buf bytes.Buffer

	// Prepared ID
	buf.Write(writeShortBytes(preparedID))

	// Result metadata ID (v5+)
	if resultMetadataID != nil {
		buf.Write(writeShortBytes(resultMetadataID))
	}

	// Consistency
	buf.Write(writeConsistency(consistency))

	// Flags
	if flags > 0xFF {
		buf.Write([]byte{byte(flags), byte(flags >> 8), byte(flags >> 16), byte(flags >> 24)})
	} else {
		buf.Write([]byte{byte(flags)})
	}

	// Values
	if flags&QueryFlagValues != 0 {
		if flags&QueryFlagWithNameValues != 0 {
			buf.Write(writeShort(uint16(len(values))))
			for _, v := range values {
				buf.Write(writeString(v.Name))
				writeQueryValue(&buf, v)
			}
		} else {
			buf.Write(writeShort(uint16(len(values))))
			for _, v := range values {
				writeQueryValue(&buf, v)
			}
		}
	}

	// Page size
	if flags&QueryFlagPageSize != 0 {
		buf.Write(writeInt(int32(pageSize)))
	}

	// Paging state
	if flags&QueryFlagWithPagingState != 0 {
		buf.Write(writeBytes(pagingState))
	}

	// Serial consistency
	if flags&QueryFlagWithSerialConsistency != 0 {
		buf.Write(writeConsistency(serialConsistency))
	}

	// Default timestamp
	if flags&QueryFlagWithDefaultTimestamp != 0 {
		buf.Write(writeLong(timestamp))
	}

	// Keyspace (v5+)
	if flags&QueryFlagWithKeyspace != 0 {
		buf.Write(writeString(keyspace))
	}

	// Now in seconds (v5+)
	if flags&QueryFlagNowInSeconds != 0 && nowInSeconds != nil {
		buf.Write(writeInt(*nowInSeconds))
	}

	return NewFrameBuilder().Opcode(OpcodeExecute).Body(buf.Bytes()).Build()
}

// BatchStatement represents a statement in a batch
type BatchStatement struct {
	Kind       byte // 0=query string, 1=prepared ID
	Query      string
	PreparedID []byte
	Values     []QueryValue
}

// BuildBatchFrame creates a BATCH frame
func BuildBatchFrame(batchType byte, statements []BatchStatement, consistency uint16, flags uint32, serialConsistency uint16, timestamp int64, keyspace string, nowInSeconds *int32) *Frame {
	var buf bytes.Buffer

	// Batch type
	buf.WriteByte(batchType)

	// Number of statements
	buf.Write(writeShort(uint16(len(statements))))

	// Statements
	for _, stmt := range statements {
		buf.WriteByte(stmt.Kind)
		if stmt.Kind == 0 {
			// Query string
			buf.Write(writeLongString(stmt.Query))
		} else {
			// Prepared ID
			buf.Write(writeShortBytes(stmt.PreparedID))
		}
		// Values
		buf.Write(writeShort(uint16(len(stmt.Values))))
		for _, v := range stmt.Values {
			writeQueryValue(&buf, v)
		}
	}

	// Consistency
	buf.Write(writeConsistency(consistency))

	// Flags
	if flags > 0xFF {
		buf.Write([]byte{byte(flags), byte(flags >> 8), byte(flags >> 16), byte(flags >> 24)})
	} else {
		buf.Write([]byte{byte(flags)})
	}

	// Serial consistency
	if flags&BatchFlagWithSerialConsistency != 0 {
		buf.Write(writeConsistency(serialConsistency))
	}

	// Default timestamp
	if flags&BatchFlagWithDefaultTimestamp != 0 {
		buf.Write(writeLong(timestamp))
	}

	// Keyspace (v5+)
	if flags&BatchFlagWithKeyspace != 0 {
		buf.Write(writeString(keyspace))
	}

	// Now in seconds (v5+)
	if flags&BatchFlagNowInSeconds != 0 && nowInSeconds != nil {
		buf.Write(writeInt(*nowInSeconds))
	}

	return NewFrameBuilder().Opcode(OpcodeBatch).Body(buf.Bytes()).Build()
}

// FrameReader reads frames with support for protocol v5+ segments
type FrameReader struct {
	reader          io.Reader
	compressor      gocql.Compressor
	protocolVersion byte
	segmentReader   *SegmentReader
}

// NewFrameReader creates a new frame reader
func NewFrameReader(r io.Reader, compressor gocql.Compressor, protocolVersion byte) *FrameReader {
	return &FrameReader{
		reader:          r,
		compressor:      compressor,
		protocolVersion: protocolVersion,
		segmentReader:   NewSegmentReader(r, compressor),
	}
}

// ReadFrame reads a complete frame
func (fr *FrameReader) ReadFrame() (*Frame, error) {
	var frameData []byte
	var err error

	if fr.protocolVersion >= 5 {
		// Use segment reader for v5+
		frameData, err = fr.segmentReader.ReadFrame()
		if err != nil {
			return nil, err
		}
	} else {
		// Use classic frame reading for v3-v4
		header := make([]byte, FrameHeaderSize)
		if _, err := io.ReadFull(fr.reader, header); err != nil {
			return nil, err
		}

		length := int32(binary.BigEndian.Uint32(header[5:9]))
		body := make([]byte, length)
		if _, err := io.ReadFull(fr.reader, body); err != nil {
			return nil, err
		}

		frameData = append(header, body...)
	}

	// Parse frame
	return ParseFrame(frameData)
}

// ReadFrameWithReader reads a frame using an alternate reader while preserving
// FrameReader/SegmentReader internal buffering state.
func (fr *FrameReader) ReadFrameWithReader(r io.Reader) (*Frame, error) {
	if fr == nil {
		return nil, fmt.Errorf("frame reader is nil")
	}
	if r == nil {
		return fr.ReadFrame()
	}

	oldReader := fr.reader
	fr.reader = r
	defer func() {
		fr.reader = oldReader
	}()

	var oldSegReader io.Reader
	if fr.segmentReader != nil {
		oldSegReader = fr.segmentReader.reader
		fr.segmentReader.reader = r
		defer func() {
			fr.segmentReader.reader = oldSegReader
		}()
	}

	return fr.ReadFrame()
}

// ParseFrame parses frame data into a Frame struct
func ParseFrame(data []byte) (*Frame, error) {
	if len(data) < FrameHeaderSize {
		return nil, fmt.Errorf("frame data too short")
	}

	frame := &Frame{}
	frame.Version = data[0]
	frame.Flags = data[1]
	frame.Stream = int16(binary.BigEndian.Uint16(data[2:4]))
	frame.Opcode = data[4]
	frame.Length = int32(binary.BigEndian.Uint32(data[5:9]))

	// Validate protocol version
	protoVersion := normalizeProtoVersionByte(frame.Version)
	if protoVersion < 3 || protoVersion > 5 {
		return nil, fmt.Errorf("unsupported protocol version: %d", protoVersion)
	}

	// Calculate total size
	totalSize := FrameHeaderSize + int(frame.Length)
	if len(data) < totalSize {
		return nil, fmt.Errorf("incomplete frame: need %d bytes, have %d", totalSize, len(data))
	}

	// Extract body
	frame.Body = data[FrameHeaderSize:totalSize]
	if err := parseFrameBody(frame); err != nil {
		return nil, err
	}

	return frame, nil
}

func parseFrameBody(frame *Frame) error {
	if frame == nil {
		return fmt.Errorf("frame is nil")
	}
	protoVersion := frame.GetProtocolVersion()
	body := frame.Body

	// Handle tracing ID (comes after header, before body in response).
	if frame.IsResponse() && frame.Flags&FlagTracing != 0 && protoVersion >= ProtoVersion2 {
		if len(body) < 16 {
			return fmt.Errorf("tracing ID requires 16 bytes")
		}
		var id [16]byte
		copy(id[:], body[:16])
		frame.TracingID = &id
		body = body[16:]
	}

	// Handle custom payload (comes after tracing, before warnings/body).
	if frame.Flags&FlagCustomPayload != 0 {
		if protoVersion < ProtoVersion4 {
			return fmt.Errorf("custom payload is not supported in protocol version %d", protoVersion)
		}
		payload, n, err := readBytesMap(body)
		if err != nil {
			return fmt.Errorf("failed to read custom payload: %w", err)
		}
		frame.CustomPayload = payload
		body = body[n:]
	}

	// Handle warnings (comes after custom payload, before body).
	if frame.Flags&FlagWarning != 0 {
		if protoVersion < ProtoVersion4 {
			return fmt.Errorf("warnings are not supported in protocol version %d", protoVersion)
		}
		warnings, n, err := readStringListWithLength(body)
		if err != nil {
			return fmt.Errorf("failed to parse warnings: %w", err)
		}
		frame.Warnings = warnings
		body = body[n:]
	}

	frame.Body = body
	frame.Length = int32(len(body))
	return nil
}

// FrameWriter writes frames with support for protocol v5+ segments
type FrameWriter struct {
	writer             io.Writer
	compressor         gocql.Compressor
	protocolVersion    byte
	segmentWriter      *SegmentWriter
	compressionEnabled bool
}

// NewFrameWriter creates a new frame writer
func NewFrameWriter(w io.Writer, compressor gocql.Compressor, protocolVersion byte) *FrameWriter {
	return &FrameWriter{
		writer:          w,
		compressor:      compressor,
		protocolVersion: protocolVersion,
		segmentWriter:   NewSegmentWriter(w, compressor),
	}
}

// SetCompressionEnabled controls segment compression for protocol v5+.
func (fw *FrameWriter) SetCompressionEnabled(enabled bool) {
	fw.compressionEnabled = enabled
}

// WriteFrame writes a frame
func (fw *FrameWriter) WriteFrame(frame *Frame) error {
	// Build frame data
	var buf bytes.Buffer
	if err := frame.WriteFrame(&buf); err != nil {
		return err
	}

	frameData := buf.Bytes()

	if fw.protocolVersion >= 5 {
		// Use segment writer for v5+
		return fw.segmentWriter.WriteFrame(frameData, fw.compressionEnabled)
	}

	// Use classic writing for v3-v4
	_, err := fw.writer.Write(frameData)
	return err
}