package cql

import (
	"context"
	"encoding/binary"
	"errors"
	"fmt"
	"strings"
	"time"

	"github.com/everything-on-mysql/cassandra-on-mysql/internal/config"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/cqlerrors"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/executor"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/parser"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/systemtables"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/tenancy"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/tenantmanager"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/translator"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/virtualtables"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/metrics"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/mysqlbackend"
	"go.uber.org/zap"
)

// connContextKey is used to store/retrieve connection from context
type connContextKey struct{}

// MessageHandler is the main handler for CQL messages
type MessageHandler struct {
	logger              *zap.Logger
	handlers            map[byte]MessageHandlerFunc
	authenticator       Authenticator
	executor            *executor.Executor
	tenantManager       *tenantmanager.TenantManager
	tenantResolver      *tenancy.Resolver
	translator          *translator.Translator
	executePinMySQLConn bool
}

// connFromContext extracts the connection from context
func connFromContext(ctx context.Context) *Conn {
	if v := ctx.Value(connContextKey{}); v != nil {
		if conn, ok := v.(*Conn); ok {
			return conn
		}
	}
	return nil
}

type overrideKeyspaceState struct {
	base     executor.KeyspaceState
	keyspace string
}

// GetKeyspace returns the override keyspace value.
func (o *overrideKeyspaceState) GetKeyspace() string {
	return o.keyspace
}

// SetKeyspace updates the keyspace and propagates to the base state when present.
func (o *overrideKeyspaceState) SetKeyspace(keyspace string) {
	if o.base != nil {
		o.base.SetKeyspace(keyspace)
	}
}

// contextWithConn adds the connection to context
func contextWithConn(ctx context.Context, conn *Conn) context.Context {
	ctx = context.WithValue(ctx, connContextKey{}, conn)
	if conn != nil {
		ctx = systemtables.WithConnInfo(ctx, systemtables.ConnInfo{LocalAddr: conn.LocalAddr()})
		if tenantID := conn.GetTenantID(); tenantID != "" {
			ctx = tenancy.WithTenantID(ctx, tenantID)
		}
		if tenantCtx := conn.GetTenantContext(); tenantCtx != nil {
			ctx = mysqlbackend.WithConnLeaseProvider(ctx, tenantCtx)
		}
		ctx = virtualtables.WithClientInfoProvider(ctx, func() []virtualtables.ClientInfo {
			if conn.server == nil {
				return nil
			}
			return buildClientInfo(conn.server.GetActiveConnections())
		})
	}
	return executor.WithKeyspaceState(ctx, conn)
}

func buildClientInfo(connections []ConnectionInfo) []virtualtables.ClientInfo {
	if len(connections) == 0 {
		return nil
	}
	out := make([]virtualtables.ClientInfo, 0, len(connections))
	for _, info := range connections {
		out = append(out, virtualtables.ClientInfo{
			TenantID:        info.TenantID,
			Address:         info.Address,
			Port:            info.Port,
			ProtocolVersion: info.ProtocolVersion,
			Username:        info.Username,
			Hostname:        info.Hostname,
			DriverName:      info.DriverName,
			DriverVersion:   info.DriverVersion,
			RequestCount:    info.RequestCount,
			SSLEnabled:      info.SSL,
			SSLProtocol:     info.SSLProtocol,
			SSLCipherSuite:  info.SSLCipherSuite,
			Keyspace:        info.Keyspace,
			ClientOptions:   info.ClientOptions,
		})
	}
	return out
}

// MessageHandlerFunc handles a single CQL frame and returns a response frame.
type MessageHandlerFunc func(ctx context.Context, frame *Frame) (*Frame, error)

// NewMessageHandler creates a new message handler
func NewMessageHandler(logger *zap.Logger, authenticator Authenticator) *MessageHandler {
	cfg := config.GetConfig()
	h := &MessageHandler{
		logger:              logger,
		authenticator:       authenticator,
		executePinMySQLConn: cfg.Executor.ExecutePinMySQLConn,
	}

	// Register message handlers
	h.handlers = map[byte]MessageHandlerFunc{
		OpcodeStartup:      h.handleStartup,
		OpcodeOptions:      h.handleOptions,
		OpcodeQuery:        h.handleQuery,
		OpcodePrepare:      h.handlePrepare,
		OpcodeExecute:      h.handleExecute,
		OpcodeAuthResponse: h.handleAuthResponse,
		// Add handlers for other opcodes as needed
	}

	return h
}

// NewMessageHandlerWithExecutor creates a new message handler with executor.
func NewMessageHandlerWithExecutor(logger *zap.Logger, authenticator Authenticator, exec *executor.Executor) *MessageHandler {
	cfg := config.GetConfig()
	h := &MessageHandler{
		logger:              logger,
		authenticator:       authenticator,
		executor:            exec,
		translator:          translator.NewTranslator(),
		executePinMySQLConn: cfg.Executor.ExecutePinMySQLConn,
	}

	// Register message handlers
	h.handlers = map[byte]MessageHandlerFunc{
		OpcodeStartup:       h.handleStartup,
		OpcodeOptions:       h.handleOptions,
		OpcodeQuery:         h.handleQuery,
		OpcodePrepare:       h.handlePrepare,
		OpcodeExecute:       h.handleExecute,
		OpcodeBatch:         h.handleBatch,
		OpcodeRegister:      h.handleRegister,
		OpcodeAuthResponse:  h.handleAuthResponse,
		OpcodeAuthenticate:  h.handleAuthenticate,
		OpcodeAuthChallenge: h.handleAuthChallenge,
		OpcodeAuthSuccess:   h.handleAuthSuccess,
	}

	return h
}

// NewMessageHandlerWithTenancy creates a new message handler with tenant management.
func NewMessageHandlerWithTenancy(logger *zap.Logger, authenticator Authenticator, tenantManager *tenantmanager.TenantManager, resolver *tenancy.Resolver) *MessageHandler {
	cfg := config.GetConfig()
	h := &MessageHandler{
		logger:              logger,
		authenticator:       authenticator,
		tenantManager:       tenantManager,
		tenantResolver:      resolver,
		translator:          translator.NewTranslator(),
		executePinMySQLConn: cfg.Executor.ExecutePinMySQLConn,
	}

	// Register message handlers
	h.handlers = map[byte]MessageHandlerFunc{
		OpcodeStartup:       h.handleStartup,
		OpcodeOptions:       h.handleOptions,
		OpcodeQuery:         h.handleQuery,
		OpcodePrepare:       h.handlePrepare,
		OpcodeExecute:       h.handleExecute,
		OpcodeBatch:         h.handleBatch,
		OpcodeRegister:      h.handleRegister,
		OpcodeAuthResponse:  h.handleAuthResponse,
		OpcodeAuthenticate:  h.handleAuthenticate,
		OpcodeAuthChallenge: h.handleAuthChallenge,
		OpcodeAuthSuccess:   h.handleAuthSuccess,
	}

	return h
}

// HandleMessage dispatches to the appropriate handler
func (h *MessageHandler) HandleMessage(ctx context.Context, frame *Frame) (*Frame, error) {
	handler := h.handlers[frame.Opcode]
	if handler == nil {
		return nil, fmt.Errorf("unsupported opcode: 0x%02x", frame.Opcode)
	}

	return handler(ctx, frame)
}

func (h *MessageHandler) executorForContext(ctx context.Context) (*executor.Executor, error) {
	if h.tenantManager != nil {
		conn := connFromContext(ctx)
		if conn == nil {
			return nil, fmt.Errorf("connection not available")
		}
		tenantCtx := conn.GetTenantContext()
		if tenantCtx == nil {
			return nil, cqlerrors.Unauthorized("tenant not resolved")
		}
		exec := tenantCtx.Executor()
		if exec == nil {
			return nil, fmt.Errorf("executor not configured")
		}
		return exec, nil
	}
	if h.executor == nil {
		return nil, fmt.Errorf("executor not configured")
	}
	return h.executor, nil
}

func (h *MessageHandler) leaseBackend(ctx context.Context, stream int16, username, password string) (context.Context, error) {
	if h.tenantManager == nil {
		return ctx, nil
	}
	conn := connFromContext(ctx)
	if conn == nil {
		return ctx, fmt.Errorf("connection not available")
	}
	tenantCtx := conn.GetTenantContext()
	if tenantCtx == nil {
		return ctx, cqlerrors.Unauthorized("tenant not resolved")
	}
	backendConn, release, err := tenantCtx.LeaseMySQLConn(ctx, username, password)
	if err != nil {
		if errors.Is(err, tenantmanager.ErrMySQLLimitExceeded) {
			return ctx, cqlerrors.Overloaded("tenant MySQL connection limit exceeded")
		}
		return ctx, err
	}
	lease := &streamLease{conn: backendConn, release: release}
	if !conn.attachStreamLease(stream, lease) {
		release()
		return ctx, cqlerrors.Protocol("stream lease registration failed")
	}
	ctx = mysqlbackend.WithConn(ctx, backendConn)
	return ctx, nil
}

func (h *MessageHandler) leaseBackendSlot(ctx context.Context, stream int16) (context.Context, error) {
	if h.tenantManager == nil {
		return ctx, nil
	}
	conn := connFromContext(ctx)
	if conn == nil {
		return ctx, fmt.Errorf("connection not available")
	}
	tenantCtx := conn.GetTenantContext()
	if tenantCtx == nil {
		return ctx, cqlerrors.Unauthorized("tenant not resolved")
	}
	release, err := tenantCtx.LeaseMySQLSlot()
	if err != nil {
		if errors.Is(err, tenantmanager.ErrMySQLLimitExceeded) {
			return ctx, cqlerrors.Overloaded("tenant MySQL connection limit exceeded")
		}
		return ctx, err
	}
	lease := &streamLease{release: release}
	if !conn.attachStreamLease(stream, lease) {
		release()
		return ctx, cqlerrors.Protocol("stream lease registration failed")
	}
	ctx = mysqlbackend.WithHeldConnLease(ctx, mysqlbackend.ConnLeaseKindStandard)
	return ctx, nil
}

// handleStartup handles the STARTUP message
func (h *MessageHandler) handleStartup(ctx context.Context, frame *Frame) (*Frame, error) {
	logger := NewFrameLogger(h.logger)

	// Parse STARTUP options from body
	options := make(map[string]string)
	if len(frame.Body) >= 2 {
		parsedOptions, err := readStringMap(frame.Body)
		if err != nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeProtocol, fmt.Sprintf("Invalid STARTUP options: %s", err.Error())), nil
		}
		options = parsedOptions
	}

	// Extract protocol version using reusable component
	protoVer := ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion())
	logger.LogProtocolVersion(frame.Version, protoVer, ResponseVersion(protoVer))

	// Check if authentication is required using reusable checker
	authChecker := NewAuthChecker(h.authenticator)
	if authChecker.ShouldAuthenticateStartup() {
		conn := connFromContext(ctx)
		if conn != nil {
			if err := conn.ApplyStartupOptions(options); err != nil {
				errorBuilder := NewErrorFrameBuilder(protoVer)
				return errorBuilder.Build(frame.Stream, ErrorCodeProtocol, err.Error()), nil
			}
		}
		// Build AUTHENTICATE response using reusable builder
		responseBuilder := NewResponseBuilder(protoVer)
		return responseBuilder.BuildAuthenticate(frame.Stream, "org.apache.cassandra.auth.PasswordAuthenticator"), nil
	}

	conn := connFromContext(ctx)
	if conn != nil {
		if err := conn.ApplyStartupOptions(options); err != nil {
			errorBuilder := NewErrorFrameBuilder(protoVer)
			return errorBuilder.Build(frame.Stream, ErrorCodeProtocol, err.Error()), nil
		}
	}

	// Return READY for all protocol versions
	responseBuilder := NewResponseBuilder(protoVer)
	return responseBuilder.BuildReady(frame.Stream), nil
}

// handleAuthenticate handles the AUTHENTICATE message
func (h *MessageHandler) handleAuthenticate(ctx context.Context, frame *Frame) (*Frame, error) {
	// Extract protocol version using reusable component
	protoVer := ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion())

	// Build AUTH_SUCCESS response using reusable builder
	responseBuilder := NewResponseBuilder(protoVer)
	return responseBuilder.BuildAuthSuccess(frame.Stream), nil
}

// handleOptions handles the OPTIONS message
func (h *MessageHandler) handleOptions(ctx context.Context, frame *Frame) (*Frame, error) {
	// Extract protocol version using reusable component
	protoVer := ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion())

	// Build SUPPORTED response using reusable builder
	compressionOptions := []string{"lz4", "snappy"}
	if protoVer >= ProtoVersion5 {
		compressionOptions = []string{"lz4"}
	}
	options := map[string][]string{
		"CQL_VERSION": {"3.4.0", "3.4.1", "4.0.0", "4.0.1", "5.0.0"},
		"COMPRESSION": compressionOptions,
	}
	responseBuilder := NewResponseBuilder(protoVer)
	return responseBuilder.BuildSupported(frame.Stream, options), nil
}

// handleQuery handles the QUERY message
func (h *MessageHandler) handleQuery(ctx context.Context, frame *Frame) (*Frame, error) {
	// Parse query from body using reusable parser
	bodyParser := NewFrameBodyParser(frame.Body)
	query, err := bodyParser.ReadLongString()
	if err != nil {
		return nil, fmt.Errorf("invalid QUERY message: %w", err)
	}
	if query == "" {
		return nil, fmt.Errorf("invalid QUERY message: empty query")
	}

	exec, err := h.executorForContext(ctx)
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}

	// Extract credentials using reusable component
	credExtractor := NewCredentialExtractor()
	username, password := credExtractor.Extract(ctx)

	// If no authenticated user, return error
	if username == "" {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeUnauthorized, "Not authenticated"), nil
	}

	// Parse consistency and flags (skip values; only keyspace flag is used).
	protoVer := frame.GetProtocolVersion()
	if _, err := bodyParser.ReadConsistency(); err != nil {
		return nil, fmt.Errorf("invalid QUERY message: %w", err)
	}
	flags, err := bodyParser.ReadFlags(protoVer)
	if err != nil {
		return nil, fmt.Errorf("invalid QUERY message: %w", err)
	}

	values, namedValues, err := bodyParser.ReadValueList(flags)
	if err != nil {
		return nil, fmt.Errorf("invalid QUERY message: %w", err)
	}

	var pageSize int
	if flags&QueryFlagPageSize != 0 {
		size, err := bodyParser.ReadInt()
		if err != nil {
			return nil, fmt.Errorf("invalid QUERY message: %w", err)
		}
		pageSize = int(size)
	}
	var pagingState []byte
	if flags&QueryFlagWithPagingState != 0 {
		state, err := bodyParser.ReadBytes()
		if err != nil {
			return nil, fmt.Errorf("invalid QUERY message: %w", err)
		}
		pagingState = state
	}
	if flags&QueryFlagWithSerialConsistency != 0 {
		if _, err := bodyParser.ReadConsistency(); err != nil {
			return nil, fmt.Errorf("invalid QUERY message: %w", err)
		}
	}
	if flags&QueryFlagWithDefaultTimestamp != 0 {
		if err := bodyParser.Skip(8); err != nil {
			return nil, fmt.Errorf("invalid QUERY message: %w", err)
		}
	}
	var queryKeyspace string
	if flags&QueryFlagWithKeyspace != 0 {
		keyspace, err := bodyParser.ReadString()
		if err != nil {
			return nil, fmt.Errorf("invalid QUERY message: %w", err)
		}
		queryKeyspace = keyspace
	}
	if flags&QueryFlagNowInSeconds != 0 {
		if err := bodyParser.Skip(4); err != nil {
			return nil, fmt.Errorf("invalid QUERY message: %w", err)
		}
	}

	if queryKeyspace != "" {
		override := &overrideKeyspaceState{
			base:     connFromContext(ctx),
			keyspace: queryKeyspace,
		}
		ctx = executor.WithKeyspaceState(ctx, override)
	}

	stmt, err := parser.NewParser(query).Parse()
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Failed to parse query: %s", err.Error())), nil
	}

	if metrics.FromContext(ctx) != nil {
		keyspace, table := statementTarget(stmt)
		if keyspace == "" {
			if queryKeyspace != "" {
				keyspace = queryKeyspace
			} else if conn := connFromContext(ctx); conn != nil {
				keyspace = conn.GetKeyspace()
			}
		}
		metrics.SetRequestTarget(ctx, keyspace, table)
	}

	params, err := mapQueryValuesToParams(stmt, values, namedValues)
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, err.Error()), nil
	}

	var paging *executor.PagingOptions
	if pageSize <= 0 && len(pagingState) > 0 {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Paging state requires a positive page size"), nil
	}
	if pageSize > 0 || len(pagingState) > 0 {
		if pageSize <= 0 {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Page size must be positive"), nil
		}
		paging = &executor.PagingOptions{PageSize: pageSize, PagingState: pagingState, Query: query}
	}

	leaseCtx, err := h.leaseContextForStatement(ctx, frame.Stream, username, password, exec, stmt)
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}

	// Execute query with client credentials (will create pool on-the-fly)
	result, err := exec.ExecuteWithParams(leaseCtx, username, password, stmt, params, paging)
	if err != nil {
		h.logger.Error("query execution failed", zap.Stringer("cql", stmt), zap.Error(err))
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(leaseCtx, connFromContext(leaseCtx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}

	broadcastSchemaChange(leaseCtx, result)

	// Translate result to frame
	resultTranslator := executor.NewQueryResultTranslator()
	tracker := metrics.FromContext(leaseCtx)
	var encodeStart time.Time
	if tracker != nil {
		encodeStart = time.Now()
	}
	resultFrame, err := resultTranslator.TranslateToResultFrame(result, frame.Stream)
	if tracker != nil {
		metrics.ObserveStage(metrics.StageEncode, time.Since(encodeStart), tracker)
	}
	if err != nil {
		h.logger.Error("translate result failed", zap.Error(err))
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(leaseCtx, connFromContext(leaseCtx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, err.Error()), nil
	}

	// Get protocol version for response
	protoVer = ExtractProtocolVersion(leaseCtx, connFromContext(leaseCtx), frame.GetProtocolVersion())

	return &Frame{
		Version:  0,
		Flags:    0,
		Stream:   frame.Stream,
		Opcode:   resultFrame.Opcode,
		Length:   int32(len(resultFrame.Body)),
		Body:     resultFrame.Body,
		Warnings: result.Warnings,
	}, nil
}

// handlePrepare handles the PREPARE message
func (h *MessageHandler) handlePrepare(ctx context.Context, frame *Frame) (*Frame, error) {
	h.logger.Debug("handlePrepare called",
		zap.Int16("stream", frame.Stream),
		zap.Int("body_len", len(frame.Body)),
	)

	exec, err := h.executorForContext(ctx)
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}

	// Extract credentials using reusable component
	credExtractor := NewCredentialExtractor()
	username, password := credExtractor.Extract(ctx)

	// If no authenticated user, return error
	if username == "" {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeUnauthorized, "Not authenticated"), nil
	}

	// Parse PREPARE message body using reusable parser
	bodyParser := NewFrameBodyParser(frame.Body)
	if len(frame.Body) < 4 {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid PREPARE message"), nil
	}

	// Read statement length
	stmtLen, err := bodyParser.ReadInt()
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Invalid PREPARE message: %s", err.Error())), nil
	}

	if stmtLen < 0 {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid PREPARE message: truncated"), nil
	}

	// Read statement string
	stmtBytes, err := bodyParser.ReadBytesN(int(stmtLen))
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid PREPARE message: truncated"), nil
	}
	stmt := string(stmtBytes)
	h.logger.Debug("preparing statement", zap.String("stmt", stmt))

	// Parse PREPARE flags and keyspace for protocol v5+
	reqProtoVer := frame.GetProtocolVersion()
	var prepareKeyspace string
	if reqProtoVer >= ProtoVersion5 {
		if bodyParser.Pos()+4 > len(frame.Body) {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid PREPARE message: missing flags"), nil
		}
		flags, err := bodyParser.ReadFlags(reqProtoVer)
		if err != nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Invalid PREPARE message: %s", err.Error())), nil
		}
		if flags&PrepareFlagWithKeyspace != 0 {
			keyspace, err := bodyParser.ReadString()
			if err != nil {
				errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
				return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Invalid PREPARE message: %s", err.Error())), nil
			}
			prepareKeyspace = keyspace
		}
	}

	if reqProtoVer >= ProtoVersion5 {
		ctx = executor.WithAllowMissingKeyspace(ctx)
	}

	if prepareKeyspace != "" {
		ctx = executor.WithKeyspaceState(ctx, &overrideKeyspaceState{keyspace: prepareKeyspace})
	}

	leaseCtx, err := h.leaseBackend(ctx, frame.Stream, username, password)
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}

	// Execute prepare through executor
	prepared, err := exec.Prepare(leaseCtx, username, password, stmt)
	if err != nil {
		h.logger.Error("prepare failed", zap.Error(err))
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(leaseCtx, connFromContext(leaseCtx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}

	// Get protocol version for response
	protoVer := ExtractProtocolVersion(leaseCtx, connFromContext(leaseCtx), frame.GetProtocolVersion())

	// Create PREPARED result frame with column metadata
	body := h.createPreparedResult(prepared, protoVer)

	return &Frame{
		Version: 0,
		Flags:   0,
		Stream:  frame.Stream,
		Opcode:  OpcodeResult,
		Length:  int32(len(body)),
		Body:    body,
	}, nil
}

// createPreparedResult creates a PREPARED result body with column metadata
func (h *MessageHandler) createPreparedResult(prepared *executor.PreparedStatement, protoVer byte) []byte {
	translator := executor.NewQueryResultTranslator()
	return translator.BuildPreparedResult(prepared, protoVer)
}

// handleExecute handles the EXECUTE message
func (h *MessageHandler) handleExecute(ctx context.Context, frame *Frame) (*Frame, error) {
	exec, err := h.executorForContext(ctx)
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}

	// Extract credentials using reusable component
	credExtractor := NewCredentialExtractor()
	username, password := credExtractor.Extract(ctx)

	// If no authenticated user, return error
	if username == "" {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeUnauthorized, "Not authenticated"), nil
	}

	// Parse EXECUTE message body using reusable parser
	bodyParser := NewFrameBodyParser(frame.Body)
	if len(frame.Body) < 2 {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid EXECUTE message"), nil
	}

	// Read prepared ID (short_bytes)
	stmtIDBytes, err := bodyParser.ReadShortBytes()
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Failed to read prepared ID: %s", err.Error())), nil
	}
	stmtID := string(stmtIDBytes)

	h.logger.Debug("executing prepared statement", zap.String("stmtID", stmtID))

	// Skip result metadata ID if protocol v5+
	protoVer := frame.GetProtocolVersion()
	if protoVer >= ProtoVersion5 {
		if bodyParser.Pos()+2 > len(frame.Body) {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid EXECUTE message: missing result metadata ID"), nil
		}
		// Read and skip result metadata ID
		_, err := bodyParser.ReadShortBytes()
		if err != nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Failed to read result metadata ID: %s", err.Error())), nil
		}
	}

	// Parse query parameters
	// [consistency][flags][values...]
	if bodyParser.Pos()+2 > len(frame.Body) {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid EXECUTE message: missing consistency"), nil
	}

	// Read consistency (skip for now)
	_, err = bodyParser.ReadConsistency()
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Failed to read consistency: %s", err.Error())), nil
	}

	// Read flags
	flags, err := bodyParser.ReadFlags(protoVer)
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid EXECUTE message: missing flags"), nil
	}

	values, namedValues, err := bodyParser.ReadValueList(flags)
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Failed to read values: %s", err.Error())), nil
	}

	// Skip or read optional fields and capture keyspace override (v5+).
	var execKeyspace string
	var pageSize int
	if flags&QueryFlagPageSize != 0 {
		size, err := bodyParser.ReadInt()
		if err != nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid EXECUTE message: missing page size"), nil
		}
		pageSize = int(size)
	}
	var pagingState []byte
	if flags&QueryFlagWithPagingState != 0 {
		state, err := bodyParser.ReadBytes()
		if err != nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid EXECUTE message: missing paging state"), nil
		}
		pagingState = state
	}
	if flags&QueryFlagWithSerialConsistency != 0 {
		if _, err := bodyParser.ReadConsistency(); err != nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid EXECUTE message: missing serial consistency"), nil
		}
	}
	if flags&QueryFlagWithDefaultTimestamp != 0 {
		if err := bodyParser.Skip(8); err != nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid EXECUTE message: missing default timestamp"), nil
		}
	}
	if flags&QueryFlagWithKeyspace != 0 {
		keyspace, err := bodyParser.ReadString()
		if err != nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid EXECUTE message: missing keyspace"), nil
		}
		execKeyspace = keyspace
	}
	if flags&QueryFlagNowInSeconds != 0 {
		if err := bodyParser.Skip(4); err != nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid EXECUTE message: missing now in seconds"), nil
		}
	}

	if execKeyspace != "" {
		ctx = executor.WithKeyspaceState(ctx, &overrideKeyspaceState{keyspace: execKeyspace})
	}

	var prepared *executor.PreparedStatement
	if namedValues || pageSize > 0 || len(pagingState) > 0 {
		var ok bool
		prepared, ok = exec.GetPreparedStatement(stmtID)
		if !ok {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.BuildUnprepared(frame.Stream, fmt.Sprintf("Prepared statement not found: %s", stmtID), stmtID), nil
		}
	}

	params, err := mapExecuteValuesToParams(prepared, stmtID, values, namedValues)
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, err.Error()), nil
	}

	if metrics.FromContext(ctx) != nil {
		keyspace := execKeyspace
		table := ""
		if prepared != nil && prepared.Statement != nil {
			stmtKeyspace, stmtTable := statementTarget(prepared.Statement)
			if stmtKeyspace != "" {
				keyspace = stmtKeyspace
			}
			if stmtTable != "" {
				table = stmtTable
			}
		}
		if keyspace == "" {
			if conn := connFromContext(ctx); conn != nil {
				keyspace = conn.GetKeyspace()
			}
		}
		metrics.SetRequestTarget(ctx, keyspace, table)
	}

	var paging *executor.PagingOptions
	if pageSize <= 0 && len(pagingState) > 0 {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Paging state requires a positive page size"), nil
	}
	if pageSize > 0 || len(pagingState) > 0 {
		if pageSize <= 0 {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Page size must be positive"), nil
		}
		query := ""
		if prepared != nil {
			query = prepared.Query
		}
		paging = &executor.PagingOptions{PageSize: pageSize, PagingState: pagingState, Query: query}
	}
	if prepared == nil {
		prepared, _ = exec.GetPreparedStatement(stmtID)
	}

	// Execute prepared statement
	var result *executor.QueryResult
	var leaseCtx context.Context
	if h.executePinMySQLConn {
		leaseCtx, err = h.leaseBackend(ctx, frame.Stream, username, password)
	} else if prepared != nil && prepared.Statement != nil {
		leaseCtx, err = h.leaseContextForStatement(ctx, frame.Stream, username, password, exec, prepared.Statement)
	} else {
		leaseCtx, err = h.leaseBackendSlot(ctx, frame.Stream)
	}
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}

	if paging != nil {
		result, err = exec.ExecutePreparedWithPaging(leaseCtx, username, password, stmtID, params, paging)
	} else {
		result, err = exec.ExecutePrepared(leaseCtx, username, password, stmtID, params)
	}
	if err != nil {
		h.logger.Error("execute prepared failed", zap.Error(err))
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(leaseCtx, connFromContext(leaseCtx), frame.GetProtocolVersion()))
		if errorCodeFromErr(err) == ErrorCodeUnprepared {
			return errorBuilder.BuildUnprepared(frame.Stream, err.Error(), stmtID), nil
		}
		return errorBuilder.Build(frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}

	broadcastSchemaChange(leaseCtx, result)

	// Translate result to frame
	resultTranslator := executor.NewQueryResultTranslator()
	tracker := metrics.FromContext(leaseCtx)
	var encodeStart time.Time
	if tracker != nil {
		encodeStart = time.Now()
	}
	resultFrame, err := resultTranslator.TranslateToResultFrame(result, frame.Stream)
	if tracker != nil {
		metrics.ObserveStage(metrics.StageEncode, time.Since(encodeStart), tracker)
	}
	if err != nil {
		h.logger.Error("translate result failed", zap.Error(err))
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(leaseCtx, connFromContext(leaseCtx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, err.Error()), nil
	}

	return &Frame{
		Version:  0,
		Flags:    0,
		Stream:   frame.Stream,
		Opcode:   resultFrame.Opcode,
		Length:   int32(len(resultFrame.Body)),
		Body:     resultFrame.Body,
		Warnings: result.Warnings,
	}, nil
}

// handleAuthResponse handles the AUTH_RESPONSE message
func (h *MessageHandler) handleAuthResponse(ctx context.Context, frame *Frame) (*Frame, error) {
	// Parse credentials from AUTH_RESPONSE body using reusable parser
	bodyParser := NewFrameBodyParser(frame.Body)
	if len(frame.Body) < 4 {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeCredentials, "Invalid credentials format"), nil
	}

	// Read credentials as bytes
	credentials, err := bodyParser.ReadBytes()
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeCredentials, fmt.Sprintf("Failed to read credentials: %s", err.Error())), nil
	}

	h.logger.Debug("credentials read", zap.Int("len", len(credentials)))

	if len(credentials) < 2 {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeCredentials, "Invalid credentials format"), nil
	}

	// Skip the first byte (should be 0)
	credentials = credentials[1:]

	// Find the first NULL terminator (end of username)
	nullIndex := -1
	for i, b := range credentials {
		if b == 0 {
			nullIndex = i
			break
		}
	}

	if nullIndex == -1 || nullIndex == len(credentials)-1 {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeCredentials, "Invalid credentials format"), nil
	}

	// Extract username (from start to first NULL)
	username := string(credentials[0:nullIndex])

	// Extract password (from after first NULL to end)
	password := string(credentials[nullIndex+1:])
	ctx = metrics.WithUsername(ctx, username)

	h.logger.Debug("parsed credentials", zap.String("username", username))

	// Authenticate using the authenticator (validates credentials are non-empty)
	_, err = h.authenticator.Authenticate(username, password)
	if err != nil {
		h.logger.Warn("authentication failed", zap.String("username", username), zap.Error(err))
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeCredentials, fmt.Sprintf("Authentication failed: %s", err.Error())), nil
	}

	var tenantID string
	var tenantCtx *tenantmanager.TenantContext
	if h.tenantManager != nil {
		if h.tenantResolver == nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeServer, "Tenant resolver not configured"), nil
		}
		tenantID, err = h.tenantResolver.Resolve(username)
		if err != nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeCredentials, fmt.Sprintf("Tenant resolution failed: %s", err.Error())), nil
		}
		tenantCtx, err = h.tenantManager.GetTenant(tenantID)
		if err != nil {
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeServer, fmt.Sprintf("Tenant lookup failed: %s", err.Error())), nil
		}
		if tenantCtx != nil && !tenantCtx.AcquireProxySlot() {
			if conn := connFromContext(ctx); conn != nil {
				// Reject this connection: send OVERLOADED below, then close it
				// cleanly. beginDrain stops the serve loop from reading further
				// requests and closes once this response is flushed.
				conn.beginDrain()
			}
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeOverloaded, "Tenant proxy connection limit exceeded"), nil
		}
	}

	authExec := h.executor
	if tenantCtx != nil {
		authExec = tenantCtx.Executor()
	}
	if authExec != nil {
		if err := authExec.ValidateCredentials(ctx, username, password); err != nil {
			if tenantCtx != nil {
				tenantCtx.ReleaseProxySlot()
			}
			h.logger.Warn("authentication failed against TiDB", zap.String("username", username), zap.Error(err))
			errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
			return errorBuilder.Build(frame.Stream, ErrorCodeCredentials, fmt.Sprintf("Authentication failed: %s", err.Error())), nil
		}
	}

	// Set credentials on connection for use in queries
	conn := connFromContext(ctx)
	if conn != nil {
		conn.SetCredentials(username, password)
		if tenantCtx != nil {
			conn.SetTenant(tenantID, tenantCtx)
			conn.SetProxyRelease(tenantCtx.ReleaseProxySlot)
		}
	} else if tenantCtx != nil {
		tenantCtx.ReleaseProxySlot()
	}
	if tenantCtx != nil {
		if exec := tenantCtx.Executor(); exec != nil {
			exec.SetCredentials(username, password)
		}
	} else if h.executor != nil {
		h.executor.SetCredentials(username, password)
	}

	h.logger.Info("user authenticated successfully", zap.String("username", username))

	// Get protocol version for response
	protoVer := ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion())

	// Build AUTH_SUCCESS response using reusable builder
	responseBuilder := NewResponseBuilder(protoVer)
	response := responseBuilder.BuildAuthSuccess(frame.Stream)

	h.logger.Debug("returning AUTH_SUCCESS",
		zap.Uint8("version", response.Version),
		zap.Int32("length", response.Length),
		zap.Int("body_len", len(response.Body)),
	)

	return response, nil
}

func mapQueryValuesToParams(stmt parser.Statement, values []QueryValue, named bool) ([]interface{}, error) {
	expected := executor.BindMarkerCount(stmt)
	if expected == 0 {
		if len(values) == 0 {
			return nil, nil
		}
		return nil, fmt.Errorf("expected no values, got %d", len(values))
	}
	if len(values) != expected {
		return nil, fmt.Errorf("expected %d values, got %d", expected, len(values))
	}
	if named {
		names, err := executor.BindMarkerNames(stmt)
		if err != nil {
			return nil, err
		}
		return mapNamedValuesToParams(names, values)
	}
	return mapPositionalValuesToParams(values), nil
}

func mapExecuteValuesToParams(prepared *executor.PreparedStatement, stmtID string, values []QueryValue, named bool) ([]interface{}, error) {
	if !named {
		return mapPositionalValuesToParams(values), nil
	}
	if prepared == nil {
		return nil, cqlerrors.Unprepared(fmt.Sprintf("Prepared statement not found: %s", stmtID))
	}
	stmt := prepared.Statement
	if stmt == nil {
		parsed, err := parser.NewParser(prepared.Query).Parse()
		if err != nil {
			return nil, fmt.Errorf("failed to parse prepared statement: %w", err)
		}
		stmt = parsed
	}
	names, err := executor.BindMarkerNames(stmt)
	if err != nil {
		return nil, err
	}
	if len(values) != prepared.ParamCount {
		return nil, fmt.Errorf("expected %d values, got %d", prepared.ParamCount, len(values))
	}
	return mapNamedValuesToParams(names, values)
}

func statementTarget(stmt parser.Statement) (string, string) {
	switch s := stmt.(type) {
	case *parser.SelectStatement:
		if s.From != nil {
			return s.From.Schema, s.From.Name
		}
	case *parser.InsertStatement:
		if s.Table != nil {
			return s.Table.Schema, s.Table.Name
		}
	case *parser.UpdateStatement:
		if s.Table != nil {
			return s.Table.Schema, s.Table.Name
		}
	case *parser.DeleteStatement:
		if s.From != nil {
			return s.From.Schema, s.From.Name
		}
	case *parser.CreateTableStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.CreateTableLikeStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.AlterTableStatement:
		if s.Table != nil {
			return s.Table.Schema, s.Table.Name
		}
	case *parser.DropTableStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.TruncateStatement:
		if s.Table != nil {
			return s.Table.Schema, s.Table.Name
		}
	case *parser.CreateIndexStatement:
		if s.Table != nil {
			return s.Table.Schema, s.Table.Name
		}
	case *parser.CreateTypeStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.AlterTypeStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.DropTypeStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.CreateMaterializedViewStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.AlterMaterializedViewStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.DropMaterializedViewStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.CreateTriggerStatement:
		if s.Table != nil {
			return s.Table.Schema, s.Table.Name
		}
	case *parser.DropTriggerStatement:
		if s.Table != nil {
			return s.Table.Schema, s.Table.Name
		}
	case *parser.CreateFunctionStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.DropFunctionStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.CreateAggregateStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.DropAggregateStatement:
		if s.Name != nil {
			return s.Name.Schema, s.Name.Name
		}
	case *parser.UseStatement:
		return s.Keyspace, ""
	case *parser.CreateKeyspaceStatement:
		return s.Name, ""
	case *parser.AlterKeyspaceStatement:
		return s.Name, ""
	case *parser.DropKeyspaceStatement:
		return s.Name, ""
	}
	return "", ""
}

func mapNamedValuesToParams(order []string, values []QueryValue) ([]interface{}, error) {
	if len(order) == 0 {
		if len(values) == 0 {
			return nil, nil
		}
		return nil, fmt.Errorf("expected no values, got %d", len(values))
	}
	if len(values) != len(order) {
		return nil, fmt.Errorf("expected %d values, got %d", len(order), len(values))
	}
	lookup := make(map[string]QueryValue, len(values))
	for _, value := range values {
		if value.Name == "" {
			return nil, fmt.Errorf("named value missing name")
		}
		if _, exists := lookup[value.Name]; exists {
			return nil, fmt.Errorf("duplicate named value %q", value.Name)
		}
		lookup[value.Name] = value
	}
	params := make([]interface{}, len(order))
	for i, name := range order {
		val, ok := lookup[name]
		if !ok {
			return nil, fmt.Errorf("missing named value for %q", name)
		}
		if val.Unset {
			params[i] = executor.UnsetValue
		} else if val.Value == nil {
			params[i] = nil
		} else {
			params[i] = val.Value
		}
		delete(lookup, name)
	}
	if len(lookup) != 0 {
		return nil, fmt.Errorf("unexpected named values")
	}
	return params, nil
}

func mapPositionalValuesToParams(values []QueryValue) []interface{} {
	if len(values) == 0 {
		return nil
	}
	params := make([]interface{}, len(values))
	for i, value := range values {
		if value.Unset {
			params[i] = executor.UnsetValue
		} else if value.Value == nil {
			params[i] = nil
		} else {
			params[i] = value.Value
		}
	}
	return params
}

type cqlErrorCoder interface {
	Code() int
}

func errorCodeFromErr(err error) int {
	var coder cqlErrorCoder
	if errors.As(err, &coder) {
		return coder.Code()
	}
	return ErrorCodeInvalid
}

// createErrorFrame creates an error response frame
// Deprecated: Use NewErrorFrameBuilder instead
func (h *MessageHandler) createErrorFrame(ctx context.Context, stream int16, errorCode int, message string) *Frame {
	// Use the reusable error frame builder
	version := ExtractProtocolVersion(ctx, connFromContext(ctx), 0)
	builder := NewErrorFrameBuilder(version)
	return builder.Build(stream, errorCode, message)
}

// AllowAllAuthenticator implements a no-op authenticator
type AllowAllAuthenticator struct{}

// Authenticate accepts any credentials and returns an empty auth token.
func (a *AllowAllAuthenticator) Authenticate(username, password string) (string, error) {
	return "", nil
}

// RequiresAuth reports whether authentication is required.
func (a *AllowAllAuthenticator) RequiresAuth() bool {
	return false
}

// PasswordAuthenticator requires authentication and validates credentials are non-empty
// Actual credential validation happens when connecting to MySQL backend
type PasswordAuthenticator struct{}

// NewPasswordAuthenticator creates a new password authenticator
func NewPasswordAuthenticator() *PasswordAuthenticator {
	return &PasswordAuthenticator{}
}

// Authenticate validates that both username and password are provided.
func (a *PasswordAuthenticator) Authenticate(username, password string) (string, error) {
	// Check that credentials are provided
	// Actual validation will happen when connecting to MySQL
	if username == "" {
		return "", fmt.Errorf("username is required")
	}
	if password == "" {
		return "", fmt.Errorf("password is required")
	}
	return "", nil
}

// RequiresAuth reports whether authentication is required.
func (a *PasswordAuthenticator) RequiresAuth() bool {
	return true
}

// handleBatch handles the BATCH message
func (h *MessageHandler) handleBatch(ctx context.Context, frame *Frame) (*Frame, error) {
	start := time.Now()
	success := false
	defer func() {
		metrics.ObserveCQLQuery(ctx, "batch", success, time.Since(start))
	}()

	exec, err := h.executorForContext(ctx)
	if err != nil {
		return h.createErrorFrame(ctx, frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}

	// Get credentials from connection
	conn := connFromContext(ctx)
	var username, password string
	if conn != nil {
		username, password = conn.GetCredentials()
	}
	ctx = metrics.WithUsername(ctx, username)

	// If no authenticated user, return error
	if username == "" {
		return h.createErrorFrame(ctx, frame.Stream, ErrorCodeUnauthorized, "Not authenticated"), nil
	}

	// Parse BATCH message body
	// Format: [type][statement_count][statements...][consistency][flags][optional fields...]
	if len(frame.Body) < 3 {
		return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message"), nil
	}
	batchParseStart := time.Now()
	bodyParser := NewFrameBodyParser(frame.Body)

	// Read batch type (byte)
	batchType, err := bodyParser.ReadByte()
	if err != nil {
		return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message"), nil
	}

	// Read statement count (short)
	if bodyParser.Pos()+2 > len(frame.Body) {
		return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing statement count"), nil
	}
	stmtCount, err := bodyParser.ReadShort()
	if err != nil {
		return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing statement count"), nil
	}

	type batchItem struct {
		statement      parser.Statement
		rawParams      []interface{}
		paramTypes     []string
		expectedParams int
		kind           byte
	}

	// Process each statement and collect raw batch entries.
	items := make([]batchItem, 0, stmtCount)
	for i := uint16(0); i < stmtCount; i++ {
		if bodyParser.Pos()+1 > len(frame.Body) {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing statement kind"), nil
		}

		// Read statement kind (byte)
		kind, err := bodyParser.ReadByte()
		if err != nil {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing statement kind"), nil
		}

		var parsedStmt parser.Statement
		var paramTypes []string
		expectedParams := 0

		if kind == 0 {
			// Query string
			if bodyParser.Pos()+4 > len(frame.Body) {
				return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing query length"), nil
			}
			queryLen, err := bodyParser.ReadInt()
			if err != nil {
				return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing query length"), nil
			}
			if queryLen < 0 {
				return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: truncated query"), nil
			}
			if bodyParser.Pos()+int(queryLen) > len(frame.Body) {
				return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: truncated query"), nil
			}
			queryBytes, err := bodyParser.ReadBytesN(int(queryLen))
			if err != nil {
				return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: truncated query"), nil
			}
			query := string(queryBytes)

			parsedStmtLocal, err := parser.NewParser(query).Parse()
			if err != nil {
				return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Failed to parse query: %s", err.Error())), nil
			}
			parsedStmt = parsedStmtLocal

		} else if kind == 1 {
			// Prepared statement ID
			if bodyParser.Pos()+2 > len(frame.Body) {
				return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing prepared ID length"), nil
			}
			stmtIDBytes, err := bodyParser.ReadShortBytes()
			if err != nil {
				return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Failed to read prepared ID: %s", err.Error())), nil
			}
			stmtID := string(stmtIDBytes)

			// Get prepared statement
			prepared, ok := exec.GetPreparedStatement(stmtID)
			if !ok {
				errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
				return errorBuilder.BuildUnprepared(frame.Stream, fmt.Sprintf("Prepared statement not found: %s", stmtID), stmtID), nil
			}

			expectedParams = prepared.ParamCount
			if prepared.Statement != nil {
				parsedStmt = prepared.Statement
			} else {
				parsedStmtLocal, err := parser.NewParser(prepared.Query).Parse()
				if err != nil {
					return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Failed to parse prepared statement: %s", err.Error())), nil
				}
				parsedStmt = parsedStmtLocal
			}

			if expectedParams > 0 {
				paramTypes = make([]string, expectedParams)
				for i, param := range prepared.Parameters {
					if i >= len(paramTypes) {
						break
					}
					paramTypes[i] = param.Type
				}
			}

		} else {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Invalid statement kind: %d", kind)), nil
		}

		// Read values for this statement
		if bodyParser.Pos()+2 > len(frame.Body) {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing value count"), nil
		}
		valueCount, err := bodyParser.ReadShort()
		if err != nil {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing value count"), nil
		}

		rawParams := make([]interface{}, valueCount)
		for j := uint16(0); j < valueCount; j++ {
			value, _, isUnset, err := bodyParser.ReadValue()
			if err != nil {
				return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Failed to read value: %s", err.Error())), nil
			}
			if isUnset {
				rawParams[j] = executor.UnsetValue
			} else {
				rawParams[j] = value
			}
		}

		if kind == 1 && expectedParams != int(valueCount) {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Prepared statement parameter count mismatch: expected %d, got %d", expectedParams, valueCount)), nil
		}

		items = append(items, batchItem{
			statement:      parsedStmt,
			rawParams:      rawParams,
			paramTypes:     paramTypes,
			expectedParams: expectedParams,
			kind:           kind,
		})
	}

	// Read consistency (skip)
	if bodyParser.Pos()+2 > len(frame.Body) {
		return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing consistency"), nil
	}
	if _, err := bodyParser.ReadConsistency(); err != nil {
		return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing consistency"), nil
	}

	// Read flags
	protoVer := frame.GetProtocolVersion()
	var flags uint32
	if protoVer >= ProtoVersion5 {
		// Protocol v5+: flags is uint
		if bodyParser.Pos()+4 > len(frame.Body) {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing flags"), nil
		}
	} else {
		// Protocol v3-v4: flags is byte
		if bodyParser.Pos()+1 > len(frame.Body) {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing flags"), nil
		}
	}
	flags, err = bodyParser.ReadFlags(protoVer)
	if err != nil {
		return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing flags"), nil
	}

	// Skip optional fields based on flags
	// Serial consistency
	if flags&QueryFlagWithSerialConsistency != 0 {
		if bodyParser.Pos()+2 > len(frame.Body) {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing serial consistency"), nil
		}
		if _, err := bodyParser.ReadConsistency(); err != nil {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing serial consistency"), nil
		}
	}

	// Default timestamp
	if flags&QueryFlagWithDefaultTimestamp != 0 {
		if bodyParser.Pos()+8 > len(frame.Body) {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing timestamp"), nil
		}
		if err := bodyParser.Skip(8); err != nil {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing timestamp"), nil
		}
	}

	// Keyspace (v5+)
	var batchKeyspace string
	if flags&QueryFlagWithKeyspace != 0 {
		if bodyParser.Pos()+2 > len(frame.Body) {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing keyspace"), nil
		}
		keyspaceLen, err := bodyParser.ReadShort()
		if err != nil {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing keyspace"), nil
		}
		if bodyParser.Pos()+int(keyspaceLen) > len(frame.Body) {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: truncated keyspace"), nil
		}
		keyspaceBytes, err := bodyParser.ReadBytesN(int(keyspaceLen))
		if err != nil {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: truncated keyspace"), nil
		}
		batchKeyspace = string(keyspaceBytes)
	}

	// Now in seconds (v5+)
	if flags&QueryFlagNowInSeconds != 0 {
		if bodyParser.Pos()+4 > len(frame.Body) {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing now in seconds"), nil
		}
		if err := bodyParser.Skip(4); err != nil {
			return h.createErrorFrame(ctx, frame.Stream, ErrorCodeInvalid, "Invalid BATCH message: missing now in seconds"), nil
		}
	}

	defaultKeyspace := ""
	if batchKeyspace != "" {
		defaultKeyspace = batchKeyspace
	} else if conn != nil {
		defaultKeyspace = conn.GetKeyspace()
	}
	metrics.ObserveStageFromContext(ctx, metrics.StageBatchParse, time.Since(batchParseStart))

	leaseCtx, err := h.leaseContextForBatch(ctx, frame.Stream, username, password, exec, batchType)
	if err != nil {
		return h.createErrorFrame(ctx, frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}

	batchDecodeStart := time.Now()
	entries := make([]executor.BatchEntry, 0, len(items))
	for _, item := range items {
		if item.statement == nil {
			return h.createErrorFrame(leaseCtx, frame.Stream, ErrorCodeInvalid, "Failed to parse batch statement"), nil
		}
		if defaultKeyspace != "" {
			executor.ApplyDefaultKeyspace(item.statement, defaultKeyspace)
		}
		paramTypes := item.paramTypes
		if item.kind == 0 && len(item.rawParams) > 0 {
			inferred, err := exec.InferParamTypes(leaseCtx, username, password, item.statement)
			if err != nil {
				return h.createErrorFrame(leaseCtx, frame.Stream, errorCodeFromErr(err), fmt.Sprintf("Failed to infer param types: %s", err.Error())), nil
			}
			if len(inferred) != len(item.rawParams) {
				return h.createErrorFrame(leaseCtx, frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Parameter count mismatch: expected %d, got %d", len(inferred), len(item.rawParams))), nil
			}
			paramTypes = inferred
		}
		var decoded []interface{}
		if len(item.rawParams) > 0 {
			var err error
			decoded, err = exec.DecodeParams(leaseCtx, username, password, item.rawParams, paramTypes)
			if err != nil {
				return h.createErrorFrame(leaseCtx, frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Failed to decode params: %s", err.Error())), nil
			}
		}
		entries = append(entries, executor.BatchEntry{
			Statement: item.statement,
			Params:    decoded,
		})
	}
	if len(entries) > 0 {
		if err := exec.ValidateBatchCounterUsageEntries(leaseCtx, username, password, batchType == BatchTypeCounter, entries); err != nil {
			return h.createErrorFrame(leaseCtx, frame.Stream, errorCodeFromErr(err), err.Error()), nil
		}
	}
	metrics.ObserveStageFromContext(leaseCtx, metrics.StageBatchDecode, time.Since(batchDecodeStart))

	result, err := exec.ExecuteBatch(leaseCtx, username, password, protocolBatchTypeName(batchType), entries)
	if err != nil {
		fields := []zap.Field{zap.Error(err)}
		fields = appendBatchLogFields(fields, metrics.FromContext(leaseCtx))
		h.logger.Error("batch execution failed", fields...)
		return h.createErrorFrame(leaseCtx, frame.Stream, errorCodeFromErr(err), err.Error()), nil
	}
	metrics.ObserveCQLWriteRows(leaseCtx, "batch", result.CQLWriteRows)

	broadcastSchemaChange(leaseCtx, result)

	// Return VOID result
	body := make([]byte, 4)
	binary.BigEndian.PutUint32(body, uint32(ResultKindVoid))
	success = true

	return &Frame{
		Version:  0,
		Flags:    0,
		Stream:   frame.Stream,
		Opcode:   OpcodeResult,
		Length:   int32(len(body)),
		Body:     body,
		Warnings: result.Warnings,
	}, nil
}

// handleRegister handles the REGISTER message
func (h *MessageHandler) handleRegister(ctx context.Context, frame *Frame) (*Frame, error) {
	// Parse event types to register for using reusable codec
	if len(frame.Body) < 2 {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, "Invalid REGISTER message"), nil
	}

	events, err := readStringListFromBytes(frame.Body)
	if err != nil {
		errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
		return errorBuilder.Build(frame.Stream, ErrorCodeInvalid, fmt.Sprintf("Failed to parse events: %s", err.Error())), nil
	}

	h.logger.Debug("Client registered for events", zap.Any("events", events))

	if conn := connFromContext(ctx); conn != nil {
		conn.AddEventSubscriptions(events)
	}

	// Get protocol version for response
	protoVer := ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion())

	// Build READY response using reusable builder
	responseBuilder := NewResponseBuilder(protoVer)
	return responseBuilder.BuildReady(frame.Stream), nil
}

func broadcastSchemaChange(ctx context.Context, result *executor.QueryResult) {
	if result == nil || result.SchemaChange == nil {
		return
	}
	conn := connFromContext(ctx)
	if conn == nil || conn.server == nil {
		return
	}
	conn.server.BroadcastSchemaChangeForTenant(result.SchemaChange, conn.GetTenantID())
}

func protocolBatchTypeName(batchType byte) string {
	switch batchType {
	case BatchTypeUnlogged:
		return "UNLOGGED"
	case BatchTypeCounter:
		return "COUNTER"
	default:
		return "LOGGED"
	}
}

func (h *MessageHandler) leaseContextForStatement(ctx context.Context, stream int16, username, password string, exec *executor.Executor, stmt parser.Statement) (context.Context, error) {
	if !h.executePinMySQLConn {
		if batchStmt, ok := stmt.(*parser.BatchStatement); ok && strings.EqualFold(batchStmt.BatchType, "UNLOGGED") && exec != nil && exec.UnloggedBatchParallelEnabled() {
			return mysqlbackend.WithPreferredConnLeaseKind(ctx, mysqlbackend.ConnLeaseKindBulk), nil
		}
	}
	return h.leaseBackend(ctx, stream, username, password)
}

func (h *MessageHandler) leaseContextForBatch(ctx context.Context, stream int16, username, password string, exec *executor.Executor, batchType byte) (context.Context, error) {
	if !h.executePinMySQLConn && batchType == BatchTypeUnlogged && exec != nil && exec.UnloggedBatchParallelEnabled() {
		return mysqlbackend.WithPreferredConnLeaseKind(ctx, mysqlbackend.ConnLeaseKindBulk), nil
	}
	return h.leaseBackend(ctx, stream, username, password)
}

// handleAuthChallenge handles the AUTH_CHALLENGE message (client-side)
func (h *MessageHandler) handleAuthChallenge(ctx context.Context, frame *Frame) (*Frame, error) {
	// This is a client-side handler, should not receive this on server
	errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
	return errorBuilder.Build(frame.Stream, ErrorCodeProtocol, "Unexpected AUTH_CHALLENGE"), nil
}

// handleAuthSuccess handles the AUTH_SUCCESS message (client-side)
func (h *MessageHandler) handleAuthSuccess(ctx context.Context, frame *Frame) (*Frame, error) {
	// This is a client-side handler, should not receive this on server
	errorBuilder := NewErrorFrameBuilder(ExtractProtocolVersion(ctx, connFromContext(ctx), frame.GetProtocolVersion()))
	return errorBuilder.Build(frame.Stream, ErrorCodeProtocol, "Unexpected AUTH_SUCCESS"), nil
}