package cql

import (
	"bufio"
	"context"
	"crypto/tls"
	"encoding/binary"
	"errors"
	"fmt"
	"io"
	"net"
	"strings"
	"sync"
	"sync/atomic"
	"time"

	gocql "github.com/apache/cassandra-gocql-driver/v2"
	"github.com/apache/cassandra-gocql-driver/v2/lz4"
	"github.com/apache/cassandra-gocql-driver/v2/snappy"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/cql/tenantmanager"
	"github.com/everything-on-mysql/cassandra-on-mysql/pkg/metrics"
	"go.uber.org/zap"
)

const (
	defaultMaxInFlightFrames = 64
	defaultEventBuffer       = 16
	// drainInFlightPoll is how often drainInFlight re-checks the in-flight count.
	drainInFlightPoll = 20 * time.Millisecond
	// maxDrainInFlightWait caps how long a single connection waits for its
	// in-flight requests to finish before falling through to close. The
	// server-wide wait_timeout + ForceCloseConnections is the real backstop;
	// this only prevents a wedged request from hanging a goroutine forever.
	maxDrainInFlightWait = 30 * time.Second
)

// Conn represents a single CQL client connection
type Conn struct {
	conn               net.Conn
	server             *Server
	logger             *zap.Logger
	frameInfo          bool
	reader             *bufio.Reader
	writer             *bufio.Writer
	stateMu            sync.Mutex
	writeMu            sync.Mutex
	closed             bool
	streamMgr          *StreamManager
	username           string // Authenticated username
	password           string // Authenticated password
	credentialKey      string
	tenantID           string
	tenantContext      *tenantmanager.TenantContext
	proxyRelease       func()
	currentKeyspace    string
	protocolVersion    byte // Protocol version (1-5)
	tlsEnabled         bool
	tlsProtocol        string
	tlsCipherSuite     string
	compressor         gocql.Compressor
	compressionEnabled bool
	startupComplete    bool
	startupOptions     map[string]string
	frameReader        *FrameReader
	frameWriter        *FrameWriter
	framePermitMu      sync.Mutex
	framePermitCond    *sync.Cond
	maxInFlightFrames  int
	maxBatchInFlight   int
	inFlightFrames     int
	batchInFlight      int
	waitingNonBatch    int
	streamMu           sync.Mutex
	streamLeases       map[int16]*streamLease
	requestCount       int64
	draining           atomic.Bool
	eventSubs          map[string]struct{}
	eventCh            chan *Frame
	eventDone          chan struct{}
}

// GetProtocolVersion returns the current protocol version
func (c *Conn) GetProtocolVersion() byte {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.protocolVersion
}

// SetProtocolVersion sets the protocol version
func (c *Conn) SetProtocolVersion(version byte) {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	c.protocolVersion = version
}

// GetKeyspace returns the current session keyspace.
func (c *Conn) GetKeyspace() string {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.currentKeyspace
}

// GetTenantID returns the tenant ID associated with this connection.
func (c *Conn) GetTenantID() string {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.tenantID
}

// GetTenantContext returns the tenant context associated with this connection.
func (c *Conn) GetTenantContext() *tenantmanager.TenantContext {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.tenantContext
}

// SetKeyspace sets the current session keyspace.
func (c *Conn) SetKeyspace(keyspace string) {
	c.stateMu.Lock()
	c.currentKeyspace = keyspace
	c.stateMu.Unlock()
}

// SetTenant sets the tenant identity and context for this connection.
func (c *Conn) SetTenant(tenantID string, tenantCtx *tenantmanager.TenantContext) {
	c.stateMu.Lock()
	c.tenantID = tenantID
	c.tenantContext = tenantCtx
	c.stateMu.Unlock()
}

// SetProxyRelease stores a release function for tenant proxy limits.
func (c *Conn) SetProxyRelease(release func()) {
	c.stateMu.Lock()
	c.proxyRelease = release
	c.stateMu.Unlock()
}

// GetRequestCount returns the total number of processed frames for this connection.
func (c *Conn) GetRequestCount() int64 {
	return atomic.LoadInt64(&c.requestCount)
}

func (c *Conn) incrementRequestCount() {
	atomic.AddInt64(&c.requestCount, 1)
}

// IsStartupComplete reports whether STARTUP has completed.
func (c *Conn) IsStartupComplete() bool {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.startupComplete
}

func (c *Conn) isStartupComplete() bool {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.startupComplete
}

// LocalAddr returns the local network address.
func (c *Conn) LocalAddr() net.Addr {
	if c == nil || c.conn == nil {
		return nil
	}
	return c.conn.LocalAddr()
}

// SetTLSState stores TLS details for the connection.
func (c *Conn) SetTLSState(state tls.ConnectionState) {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()

	if !state.HandshakeComplete {
		return
	}
	c.tlsEnabled = true
	if state.Version != 0 {
		if name := tls.VersionName(state.Version); name != "" {
			c.tlsProtocol = name
		} else {
			c.tlsProtocol = fmt.Sprintf("0x%04x", state.Version)
		}
	}
	if state.CipherSuite != 0 {
		if name := tls.CipherSuiteName(state.CipherSuite); name != "" {
			c.tlsCipherSuite = name
		} else {
			c.tlsCipherSuite = fmt.Sprintf("0x%04x", state.CipherSuite)
		}
	}
}

// TLSInfo returns TLS status and negotiated parameters.
func (c *Conn) TLSInfo() (bool, string, string) {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.tlsEnabled, c.tlsProtocol, c.tlsCipherSuite
}

// SupportsSegmentFormat returns true if connection uses v5+ segment format
func (c *Conn) SupportsSegmentFormat() bool {
	version := c.GetProtocolVersion()
	return version >= ProtoVersion5
}

func (c *Conn) bootstrapSegmentLikely() bool {
	if c.reader == nil {
		return false
	}
	// Peek enough bytes to validate uncompressed or compressed segment headers.
	buf, err := c.reader.Peek(8)
	if err != nil && len(buf) < 6 {
		return false
	}
	if len(buf) >= 6 {
		headerCRC24 := uint32(buf[3]) | uint32(buf[4])<<8 | uint32(buf[5])<<16
		if Crc24(buf[:3]) == headerCRC24 {
			return true
		}
	}
	if len(buf) >= 8 {
		headerCRC24 := uint32(buf[5]) | uint32(buf[6])<<8 | uint32(buf[7])<<16
		if Crc24(buf[:5]) == headerCRC24 {
			return true
		}
	}
	return false
}

// GetMaxStreamID returns the maximum stream ID for the current protocol version
func (c *Conn) GetMaxStreamID() int16 {
	version := c.GetProtocolVersion()
	if version <= ProtoVersion2 {
		return MaxStreamIDv1v2
	}
	return MaxStreamIDv3p
}

// StreamManager manages stream IDs for a connection
type StreamManager struct {
	mu          sync.Mutex
	nextID      int16
	inUse       map[int16]bool
	maxStreamID int16 // Maximum stream ID based on protocol version
}

// NewStreamManager creates a stream manager with default limits.
func NewStreamManager() *StreamManager {
	return &StreamManager{
		nextID:      1, // 0 is reserved for events
		inUse:       make(map[int16]bool),
		maxStreamID: MaxStreamIDv3p, // Default to v3+
	}
}

// SetMaxStreamID sets the maximum stream ID based on protocol version
func (sm *StreamManager) SetMaxStreamID(maxID int16) {
	sm.mu.Lock()
	defer sm.mu.Unlock()
	sm.maxStreamID = maxID
}

// Acquire reserves and returns the next available stream ID.
func (sm *StreamManager) Acquire() int16 {
	sm.mu.Lock()
	defer sm.mu.Unlock()

	// Find next available stream ID
	startID := sm.nextID
	for {
		if !sm.inUse[sm.nextID] {
			id := sm.nextID
			sm.inUse[id] = true
			sm.nextID++
			if sm.nextID > sm.maxStreamID {
				sm.nextID = 1
			}
			return id
		}
		sm.nextID++
		if sm.nextID > sm.maxStreamID {
			sm.nextID = 1
		}
		// Prevent infinite loop if all streams are in use
		if sm.nextID == startID {
			// Should not happen in practice, but return -1 to indicate error
			return -1
		}
	}
}

// Release releases a previously acquired stream ID.
func (sm *StreamManager) Release(id int16) {
	sm.mu.Lock()
	defer sm.mu.Unlock()
	delete(sm.inUse, id)
}

// NewConn creates a new connection wrapper
func NewConn(conn net.Conn, server *Server) *Conn {
	maxInFlight := defaultMaxInFlightFrames
	frameInfo := false
	if server != nil && server.options != nil {
		if server.options.MaxInFlightFrames != 0 {
			maxInFlight = server.options.MaxInFlightFrames
		}
		if server.options.FrameInfo {
			frameInfo = true
		}
	}
	if maxInFlight > int(MaxStreamIDv3p) {
		maxInFlight = int(MaxStreamIDv3p)
	}
	if maxInFlight < 1 {
		maxInFlight = 1
	}
	maxBatchInFlight := maxInFlight
	if maxInFlight > 1 {
		reserve := maxInFlight / 8
		if reserve < 1 {
			reserve = 1
		}
		maxBatchInFlight = maxInFlight - reserve
		if maxBatchInFlight < 1 {
			maxBatchInFlight = 1
		}
	}

	c := &Conn{
		conn:              conn,
		server:            server,
		logger:            server.logger.With(zap.String("remote", conn.RemoteAddr().String())),
		frameInfo:         frameInfo,
		reader:            bufio.NewReader(conn),
		writer:            bufio.NewWriter(conn),
		streamMgr:         NewStreamManager(),
		protocolVersion:   ProtoVersion4, // Default to v4 until negotiated
		maxInFlightFrames: maxInFlight,
		maxBatchInFlight:  maxBatchInFlight,
		eventSubs:         make(map[string]struct{}),
		eventCh:           make(chan *Frame, defaultEventBuffer),
		eventDone:         make(chan struct{}),
	}
	c.framePermitCond = sync.NewCond(&c.framePermitMu)

	return c
}

func detectSegmentHeader(r *bufio.Reader) (bool, error) {
	header, err := r.Peek(8)
	if err != nil && len(header) == 0 {
		return false, err
	}
	if len(header) < 6 {
		if err != nil {
			return false, err
		}
		return false, fmt.Errorf("insufficient data for segment header: %d bytes", len(header))
	}

	// Uncompressed segment header CRC24 is stored in bytes 3-5.
	crc24 := Crc24(header[:3])
	headerCRC24 := uint32(header[3]) | uint32(header[4])<<8 | uint32(header[5])<<16
	if crc24 == headerCRC24 {
		return true, nil
	}

	// Compressed segment header CRC24 is stored in bytes 5-7.
	if len(header) < 8 {
		if err != nil {
			return false, err
		}
		return false, fmt.Errorf("insufficient data for compressed segment header: %d bytes", len(header))
	}
	crc24 = Crc24(header[:5])
	headerCRC24 = uint32(header[5]) | uint32(header[6])<<8 | uint32(header[7])<<16
	if crc24 == headerCRC24 {
		return true, nil
	}

	return false, nil
}

type frameReadObserver struct {
	reader io.Reader

	bytesRead     int
	totalRead     time.Duration
	firstByteWait time.Duration
	sawFirstByte  bool
}

func newFrameReadObserver(r io.Reader) *frameReadObserver {
	return &frameReadObserver{reader: r}
}

func (o *frameReadObserver) Read(p []byte) (int, error) {
	start := time.Now()
	n, err := o.reader.Read(p)
	elapsed := time.Since(start)
	o.totalRead += elapsed
	if n > 0 {
		o.bytesRead += n
		if !o.sawFirstByte {
			o.firstByteWait = elapsed
			o.sawFirstByte = true
		}
	}
	return n, err
}

func (o *frameReadObserver) Breakdown(extraFirstByteWait time.Duration) (requestBytes int, firstByteWait, bodyRead time.Duration) {
	if o == nil {
		return 0, extraFirstByteWait, 0
	}
	requestBytes = o.bytesRead
	if extraFirstByteWait < 0 {
		extraFirstByteWait = 0
	}
	firstByteWait = o.firstByteWait + extraFirstByteWait
	bodyRead = o.totalRead - o.firstByteWait
	if bodyRead < 0 {
		bodyRead = 0
	}
	return requestBytes, firstByteWait, bodyRead
}

// Serve handles the connection loop
func (c *Conn) Serve() error {
	defer c.Close()
	c.startEventLoop()

	// Track if we've done protocol negotiation
	negotiated := false

	for {
		// Check if connection is closed
		c.stateMu.Lock()
		if c.closed {
			c.stateMu.Unlock()
			return nil
		}
		c.stateMu.Unlock()

		// Graceful drain: stop reading new requests, let already-dispatched
		// in-flight requests finish (responses flushed), then close cleanly.
		if c.isDraining() {
			c.drainInFlight(maxDrainInFlightWait)
			return nil
		}

		var frame *Frame
		var err error
		var tracker *metrics.RequestMetrics
		tracking := metrics.RequestTrackingEnabled()
		var readStart time.Time
		var readSource io.Reader = c.reader
		var readObs *frameReadObserver
		var extraFirstByteWait time.Duration
		if tracking {
			readStart = time.Now()
			readObs = newFrameReadObserver(c.reader)
			readSource = readObs
		}

		if !negotiated {
			detectStart := time.Now()
			useSegment, detectErr := detectSegmentHeader(c.reader)
			if tracking {
				extraFirstByteWait += time.Since(detectStart)
			}
			if detectErr != nil {
				if errors.Is(detectErr, io.EOF) {
					return nil // Client closed connection before sending the first frame.
				}
				return fmt.Errorf("failed to detect segment header: %w", detectErr)
			}
			if useSegment {
				c.stateMu.Lock()
				if c.frameReader == nil {
					c.frameReader = NewFrameReader(c.reader, c.compressor, ProtoVersion5)
					c.frameWriter = NewFrameWriter(c.writer, c.compressor, ProtoVersion5)
					c.frameWriter.SetCompressionEnabled(c.compressionEnabled)
				}
				frameReader := c.frameReader
				c.stateMu.Unlock()
				frame, err = frameReader.ReadFrameWithReader(readSource)
			} else {
				// Use classic frame reader (v1-v4, or initial handshake for v5)
				frame, err = ReadFrame(readSource)
			}
		} else {
			// For negotiated v5 connections, prefer segment mode only when actually detected.
			supportsSegment := c.SupportsSegmentFormat()
			var frameReader *FrameReader
			if supportsSegment {
				c.stateMu.Lock()
				frameReader = c.frameReader
				c.stateMu.Unlock()
			}
			if supportsSegment && frameReader != nil {
				frame, err = frameReader.ReadFrameWithReader(readSource)
			} else if supportsSegment {
				likelyStart := time.Now()
				likely := c.bootstrapSegmentLikely()
				if tracking {
					extraFirstByteWait += time.Since(likelyStart)
				}
				if !likely {
					// Use classic frame reader (v1-v4 and v5 classic framing).
					frame, err = ReadFrame(readSource)
				} else {
					c.stateMu.Lock()
					if c.frameReader == nil {
						c.frameReader = NewFrameReader(c.reader, c.compressor, ProtoVersion5)
						c.frameWriter = NewFrameWriter(c.writer, c.compressor, ProtoVersion5)
						c.frameWriter.SetCompressionEnabled(c.compressionEnabled)
					}
					frameReader = c.frameReader
					c.stateMu.Unlock()
					frame, err = frameReader.ReadFrameWithReader(readSource)
				}
			} else {
				// Use classic frame reader (v1-v4 and v5 classic framing).
				frame, err = ReadFrame(readSource)
			}
		}

		if err != nil {
			if c.isDraining() {
				// Read was interrupted by drain (SetReadDeadline). Finish
				// in-flight work, then close cleanly so the driver reconnects
				// without seeing a torn request.
				c.drainInFlight(maxDrainInFlightWait)
				return nil
			}
			if errors.Is(err, io.EOF) {
				return nil // Client closed connection
			}
			return fmt.Errorf("failed to read frame: %w", err)
		}

		if frame == nil {
			continue
		}

		if tracking {
			tracker = metrics.NewRequestMetricsWithDiagnostics(frame.Opcode, frame.Stream, readStart, metrics.SlowThreshold() > 0)
			requestBytes, firstByteWait, bodyRead := readObs.Breakdown(extraFirstByteWait)
			// Exported frame_read latency should describe actual socket read work for this frame,
			// not the idle gap while waiting for the next request to arrive on a keepalive connection.
			metrics.ObserveStage(metrics.StageFrameRead, bodyRead, tracker)
			if requestBytes == 0 {
				requestBytes = frame.FullSize()
			}
			tracker.SetFrameReadBreakdown(requestBytes, firstByteWait, bodyRead)
		}

		if !negotiated {
			c.logger.Info("protocol version first frame",
				zap.Uint8("version_byte", frame.Version),
				zap.Uint8("protocol_version", frame.GetProtocolVersion()),
				zap.Bool("is_request", frame.IsRequest()),
			)
		}

		// Protocol version negotiation on first frame
		if !negotiated {
			protoVer := frame.GetProtocolVersion()
			c.stateMu.Lock()
			c.protocolVersion = protoVer

			// Get max stream ID based on the protocol version we just set
			// Use the protoVer value directly instead of calling GetMaxStreamID() which would deadlock
			var maxStreamID int16
			if protoVer <= ProtoVersion2 {
				maxStreamID = MaxStreamIDv1v2
			} else {
				maxStreamID = MaxStreamIDv3p
			}
			c.streamMgr.SetMaxStreamID(maxStreamID)

			// For v5+, initialize segment-based reader/writer after STARTUP completes
			// The initial handshake uses classic frame format
			if protoVer >= ProtoVersion5 {
				c.logger.Info("client using protocol v5+", zap.Int("version", int(protoVer)))
			} else {
				c.logger.Info("client using protocol v3-v4", zap.Int("version", int(protoVer)))
			}
			c.stateMu.Unlock()
			negotiated = true
		}

		// Handle handshake frames inline to avoid racing v5 segment switch.
		ctx := context.Background()
		if !c.isStartupComplete() {
			c.processFrame(ctx, frame, tracker)
			continue
		}

		releasePermit := c.acquireFramePermit(frame, tracker)
		if releasePermit == nil {
			// Connection closed or draining while waiting for a permit: stop
			// reading new requests, drain in-flight work, then close cleanly.
			c.drainInFlight(maxDrainInFlightWait)
			return nil
		}
		go func(frame *Frame, tracker *metrics.RequestMetrics) {
			defer releasePermit()
			c.processFrame(ctx, frame, tracker)
		}(frame, tracker)
	}
}

func (c *Conn) acquireFramePermit(frame *Frame, tracker *metrics.RequestMetrics) func() {
	waitStart := time.Now()
	isBatch := frame != nil && frame.Opcode == OpcodeBatch
	c.framePermitMu.Lock()
	if !isBatch {
		c.waitingNonBatch++
	}
	for !c.canAcquireFramePermit(isBatch) {
		// Abort the wait if the connection is closing or draining; otherwise a
		// saturated connection whose in-flight handler is wedged would park the
		// serve goroutine here forever (beginDrain/Close cannot make progress
		// through the permit cap). Returning nil tells Serve to stop dispatching.
		if c.isClosed() || c.isDraining() {
			if !isBatch {
				c.waitingNonBatch--
			}
			c.framePermitMu.Unlock()
			return nil
		}
		c.framePermitCond.Wait()
	}
	if !isBatch {
		c.waitingNonBatch--
	}
	c.inFlightFrames++
	if isBatch {
		c.batchInFlight++
	}
	c.framePermitMu.Unlock()
	if tracker != nil {
		metrics.ObserveStage(metrics.StageQueueWait, time.Since(waitStart), tracker)
	}
	var once sync.Once
	return func() {
		once.Do(func() {
			c.framePermitMu.Lock()
			if c.inFlightFrames > 0 {
				c.inFlightFrames--
			}
			if isBatch && c.batchInFlight > 0 {
				c.batchInFlight--
			}
			c.framePermitMu.Unlock()
			c.framePermitCond.Broadcast()
		})
	}
}

func (c *Conn) canAcquireFramePermit(isBatch bool) bool {
	if c.maxInFlightFrames <= 0 {
		return true
	}
	if c.inFlightFrames >= c.maxInFlightFrames {
		return false
	}
	if !isBatch {
		return true
	}
	if c.maxBatchInFlight > 0 && c.batchInFlight >= c.maxBatchInFlight {
		return false
	}
	// When non-batch work is queued, keep the reserved headroom free for it.
	if c.waitingNonBatch > 0 && c.maxBatchInFlight > 0 && c.inFlightFrames >= c.maxBatchInFlight {
		return false
	}
	return true
}

// processFrame processes a single frame
func (c *Conn) processFrame(ctx context.Context, frame *Frame, tracker *metrics.RequestMetrics) {
	// Use frame logger for consistent logging
	logger := NewFrameLogger(c.logger)
	logger.LogReceivedFrame(frame)
	c.incrementRequestCount()
	if tracker != nil {
		ctx = metrics.WithRequestMetrics(ctx, tracker)
		if username := c.GetUsername(); username != "" {
			ctx = metrics.WithUsername(ctx, username)
		}
		if len(frame.CustomPayload) > 0 {
			metrics.SetRequestClientPayload(ctx, formatClientCustomPayload(frame.CustomPayload))
		}
	}
	metrics.IncRequests(frame.Opcode, tracker)
	metrics.IncInFlightRequests(tracker)
	defer metrics.DecInFlightRequests(tracker)

	var requestErr error
	requestFailed := false
	var response *Frame // Declared here so it's visible in defer
	if tracker != nil {
		defer func() {
			total := time.Since(tracker.Start())
			requestBytes, firstByteWait, bodyRead := tracker.FrameReadBreakdown()
			frameRead := computeFrameReadDuration(firstByteWait, bodyRead)
			// Keep slow-log diagnostics on the full read path, but export request latency as
			// processing time so Prometheus aligns with slow-request logging semantics.
			processingTime := computeProcessingTime(total, frameRead)
			metrics.ObserveStage(metrics.StageTotal, processingTime, tracker)
			metrics.ObserveRequestCompleted(frame.Opcode, !requestFailed, processingTime, tracker)

			threshold := metrics.SlowThreshold()
			if threshold > 0 && processingTime >= threshold {
				queueWait := tracker.StageDuration(metrics.StageQueueWait)
				handler := tracker.StageDuration(metrics.StageHandler)
				batchParse := tracker.StageDuration(metrics.StageBatchParse)
				batchDecode := tracker.StageDuration(metrics.StageBatchDecode)
				batchMergePlan := tracker.StageDuration(metrics.StageBatchMergePlan)
				batchExecuteWait := tracker.StageDuration(metrics.StageBatchExecuteWait)
				schemaSync := tracker.StageDuration(metrics.StageSchemaSync)
				schemaCache := tracker.StageDuration(metrics.StageSchemaCache)
				cqlMetaCache := tracker.StageDuration(metrics.StageCQLMetaCache)
				mysqlPoolWait := tracker.StageDuration(metrics.StageMySQLWait)
				mysqlQuery := tracker.StageDuration(metrics.StageMySQLQuery)
				encode := tracker.StageDuration(metrics.StageEncode)
				writeLockWait := tracker.StageDuration(metrics.StageWriteLockWait)
				writeFlush := tracker.StageDuration(metrics.StageWrite)
				sqlCount, sqlBytes, dataSizeBytes, sqlOmittedCount, executedSQLs := tracker.SQLSummary()
				handlerOther := computeHandlerOtherDuration(
					handler,
					schemaSync,
					schemaCache,
					cqlMetaCache,
					mysqlPoolWait,
					mysqlQuery,
					encode,
				)

				fields := []zap.Field{
					zap.Duration("total", total),
					zap.Duration("processing_time", processingTime),
					zap.Duration("frame_read", frameRead),
					zap.Int("request_bytes", requestBytes),
					zap.Duration("first_byte_wait", firstByteWait),
					zap.Duration("body_read", bodyRead),
					zap.Duration("queue_wait", queueWait),
					zap.Duration("handler", handler),
					zap.Duration("batch_parse", batchParse),
					zap.Duration("batch_decode", batchDecode),
					zap.Duration("batch_merge_plan", batchMergePlan),
					zap.Duration("batch_execute_wait", batchExecuteWait),
					zap.Duration("handler_other", handlerOther),
					zap.Duration("schema_sync", schemaSync),
					zap.Duration("schema_cache_refresh", schemaCache),
					zap.Duration("cql_meta_cache_refresh", cqlMetaCache),
					zap.Duration("mysql_pool_wait", mysqlPoolWait),
					zap.Duration("mysql_query", mysqlQuery),
					zap.Duration("result_encode", encode),
					zap.Duration("write_lock_wait", writeLockWait),
					zap.Duration("write_flush", writeFlush),
					zap.Int("executed_sql_count", sqlCount),
					zap.Int("sql_bytes", sqlBytes),
					zap.Int("data_size_bytes", dataSizeBytes),
					zap.Int("executed_sql_omitted_count", sqlOmittedCount),
					zap.Int16("stream", tracker.Stream()),
					zap.String("opcode", MessageType(frame.Opcode).String()),
				}
				if len(executedSQLs) > 0 {
					fields = append(fields, zap.Strings("executed_sqls", executedSQLs))
				}
				fields = appendBatchLogFields(fields, tracker)
				if keyspace, table := tracker.Target(); keyspace != "" || table != "" {
					if keyspace != "" {
						fields = append(fields, zap.String("keyspace", keyspace))
					}
					if table != "" {
						fields = append(fields, zap.String("table", table))
					}
				}
				if requestErr != nil {
					fields = append(fields, zap.Error(requestErr))
				}
				c.logger.Warn("slow request", fields...)
			}

			// Log failed requests for observability
			if requestFailed {
				fields := []zap.Field{
					zap.Int16("stream", frame.Stream),
					zap.String("opcode", MessageType(frame.Opcode).String()),
				}
				if keyspace, table := tracker.Target(); keyspace != "" || table != "" {
					if keyspace != "" {
						fields = append(fields, zap.String("keyspace", keyspace))
					}
					if table != "" {
						fields = append(fields, zap.String("table", table))
					}
				}
				if requestErr != nil {
					fields = append(fields, zap.Error(requestErr))
				} else if response != nil && response.Opcode == OpcodeError {
					// CQL protocol-level error response (e.g., syntax error, permission denied)
					errMsg := extractErrorMessage(response.Body)
					fields = append(fields, zap.String("cql_error", errMsg))
				}
				c.logger.Error("request failed", fields...)
			}
		}()
	}

	// Handle protocol version using reusable component
	version := frame.GetProtocolVersion()
	if version < ProtoVersion1 || version > ProtoVersion5 {
		// Send ERROR response using reusable builder
		errorBuilder := NewErrorFrameBuilder(c.GetProtocolVersion())
		errFrame := errorBuilder.Build(frame.Stream, ErrorCodeProtocol, fmt.Sprintf("Unsupported protocol version: %d", version))
		if err := c.writeFrame(errFrame, tracker); err != nil {
			c.logger.Error("failed to write error frame", zap.Error(err))
			requestErr = err
		}
		requestFailed = true
		return
	}

	// Validate stream ID using reusable validator
	validator := NewStreamIDValidator(c.GetProtocolVersion())
	if err := validator.Validate(frame.Stream); err != nil {
		// Send ERROR response using reusable builder
		errorBuilder := NewErrorFrameBuilder(c.GetProtocolVersion())
		errFrame := errorBuilder.Build(frame.Stream, ErrorCodeProtocol, err.Error())
		if err := c.writeFrame(errFrame, tracker); err != nil {
			c.logger.Error("failed to write error frame", zap.Error(err))
			requestErr = err
		}
		requestFailed = true
		return
	}

	if err := c.beginStream(frame.Stream); err != nil {
		errorBuilder := NewErrorFrameBuilder(c.GetProtocolVersion())
		errFrame := errorBuilder.Build(frame.Stream, ErrorCodeProtocol, err.Error())
		if err := c.writeFrame(errFrame, tracker); err != nil {
			c.logger.Error("failed to write stream error frame", zap.Error(err))
			requestErr = err
		}
		requestFailed = true
		return
	}
	streamActive := true
	releaseStream := func() {
		if streamActive {
			c.releaseStream(frame.Stream)
			streamActive = false
		}
	}
	defer releaseStream()

	if err := c.maybeDecompressFrame(frame); err != nil {
		errorBuilder := NewErrorFrameBuilder(c.GetProtocolVersion())
		errFrame := errorBuilder.Build(frame.Stream, ErrorCodeProtocol, err.Error())
		releaseStream()
		if err := c.writeFrame(errFrame, tracker); err != nil {
			c.logger.Error("failed to write compression error frame", zap.Error(err))
			requestErr = err
		}
		requestFailed = true
		return
	}

	// Check if authentication is required using reusable checker
	authChecker := NewAuthChecker(c.server.options.Authenticator)
	if authChecker.IsAuthenticationRequired(frame.Opcode, c.username) {
		// Send ERROR response using reusable builder
		errorBuilder := NewErrorFrameBuilder(c.GetProtocolVersion())
		errFrame := errorBuilder.Build(frame.Stream, ErrorCodeUnauthorized, "Authentication required")
		releaseStream()
		if err := c.writeFrame(errFrame, tracker); err != nil {
			c.logger.Error("failed to write auth error frame", zap.Error(err))
			requestErr = err
		}
		requestFailed = true
		return
	}

	// Pass to handler with connection in context
	ctx = contextWithConn(ctx, c)
	var handlerStart time.Time
	if tracker != nil {
		handlerStart = time.Now()
	}
	var err error
	response, err = c.server.handler.HandleMessage(ctx, frame)
	if tracker != nil {
		metrics.ObserveStage(metrics.StageHandler, time.Since(handlerStart), tracker)
	}
	if err != nil {
		c.logger.Error("failed to handle message", zap.Error(err))
		requestErr = err
		requestFailed = true
		// Send ERROR response using reusable builder
		errorBuilder := NewErrorFrameBuilder(c.GetProtocolVersion())
		errFrame := errorBuilder.Build(frame.Stream, ErrorCodeServer, err.Error())
		releaseStream()
		if err := c.writeFrame(errFrame, tracker); err != nil {
			c.logger.Error("failed to write error frame", zap.Error(err))
			requestErr = err
		}
		return
	}

	if response != nil {
		if response.Opcode == OpcodeError {
			requestFailed = true
		}
		response.Stream = frame.Stream
		if response.Version == 0 {
			response.Version = ResponseVersion(frame.Version)
		}
		if len(response.CustomPayload) == 0 && len(frame.CustomPayload) > 0 {
			response.CustomPayload = frame.CustomPayload
		}
		if len(response.Warnings) == 0 && len(frame.Warnings) > 0 {
			response.Warnings = frame.Warnings
		}
		// Stream capacity tracks active request processing. Release it before
		// socket writes so a slow client cannot hold a completed stream slot.
		releaseStream()
		if err := c.writeFrame(response, tracker); err != nil {
			c.logger.Error("failed to write response frame", zap.Error(err))
			requestErr = err
			requestFailed = true
		}
		if frame.Opcode == OpcodeStartup && response.Opcode == OpcodeReady {
			c.stateMu.Lock()
			c.startupComplete = true
			c.stateMu.Unlock()
		}
		if frame.Opcode == OpcodeAuthResponse && response.Opcode == OpcodeAuthSuccess {
			c.stateMu.Lock()
			c.startupComplete = true
			c.stateMu.Unlock()
		}
	}
}

func computeProcessingTime(total, frameRead time.Duration) time.Duration {
	if total <= frameRead {
		return 0
	}
	return total - frameRead
}

func computeFrameReadDuration(firstByteWait, bodyRead time.Duration) time.Duration {
	if firstByteWait < 0 {
		firstByteWait = 0
	}
	if bodyRead < 0 {
		bodyRead = 0
	}
	return firstByteWait + bodyRead
}

func computeHandlerOtherDuration(handler, schemaSync, schemaCache, cqlMetaCache, mysqlWait, mysqlQuery, encode time.Duration) time.Duration {
	known := schemaSync + schemaCache + cqlMetaCache + mysqlWait + mysqlQuery + encode
	if handler <= known {
		return 0
	}
	return handler - known
}

// extractErrorMessage extracts the error message from an ERROR frame body.
// ERROR frame body format: int32(error_code) + [short]length + string(message)
func extractErrorMessage(body []byte) string {
	if len(body) < 6 { // 4 bytes for error code + 2 bytes for length prefix
		return "malformed error frame"
	}
	length := int(binary.BigEndian.Uint16(body[4:6]))
	if len(body) < 6+length {
		return "truncated error message"
	}
	return string(body[6 : 6+length])
}

// writeFrame writes a frame to the connection
func (c *Conn) writeFrame(frame *Frame, tracker *metrics.RequestMetrics) error {
	state := c.snapshotWriteState()
	if state.closed {
		return fmt.Errorf("connection closed")
	}
	var lockStart time.Time
	if tracker != nil {
		lockStart = time.Now()
	}
	c.writeMu.Lock()
	if tracker != nil {
		metrics.ObserveStage(metrics.StageWriteLockWait, time.Since(lockStart), tracker)
	}
	defer c.writeMu.Unlock()

	// Verbose frame logging is opt-in because it runs on every response.
	if state.frameInfo {
		c.logger.Info("writeFrame - frame details",
			zap.Uint8("version", frame.Version),
			zap.Int16("stream", frame.Stream),
			zap.String("opcode", MessageType(frame.Opcode).String()),
			zap.Int32("length", frame.Length),
			zap.Int("body_len", len(frame.Body)),
		)
	}

	wireFrame := *frame
	var writeStart time.Time
	if tracker != nil {
		writeStart = time.Now()
	}
	if shouldCompressFrame(&wireFrame, state) {
		compressed, err := state.compressor.AppendCompressedWithLength(nil, wireFrame.Body)
		if err != nil {
			c.logger.Error("failed to compress frame", zap.Error(err))
			if tracker != nil {
				metrics.ObserveStage(metrics.StageWrite, time.Since(writeStart), tracker)
			}
			return err
		}
		wireFrame.Body = compressed
		wireFrame.Length = int32(len(compressed))
		wireFrame.Flags |= FlagCompression
	}

	// Use frameWriter for v5+, classic writer for v3-v4
	if state.frameWriter != nil {
		if err := state.frameWriter.WriteFrame(&wireFrame); err != nil {
			c.logger.Error("frameWriter.WriteFrame failed", zap.Error(err))
			if tracker != nil {
				metrics.ObserveStage(metrics.StageWrite, time.Since(writeStart), tracker)
			}
			return err
		}
	} else {
		if err := wireFrame.WriteFrame(c.writer); err != nil {
			c.logger.Error("frame.WriteFrame failed", zap.Error(err))
			if tracker != nil {
				metrics.ObserveStage(metrics.StageWrite, time.Since(writeStart), tracker)
			}
			return err
		}
	}

	if err := c.writer.Flush(); err != nil {
		c.logger.Error("writer.Flush failed", zap.Error(err))
		if tracker != nil {
			metrics.ObserveStage(metrics.StageWrite, time.Since(writeStart), tracker)
		}
		return err
	}

	if tracker != nil {
		metrics.ObserveStage(metrics.StageWrite, time.Since(writeStart), tracker)
	}

	return nil
}

func (c *Conn) startEventLoop() {
	if c.eventCh == nil || c.eventDone == nil {
		return
	}
	go func() {
		for {
			select {
			case frame := <-c.eventCh:
				if frame == nil {
					continue
				}
				if err := c.writeFrame(frame, nil); err != nil {
					c.logger.Error("failed to write event frame", zap.Error(err))
				}
			case <-c.eventDone:
				return
			}
		}
	}()
}

// AddEventSubscriptions registers event types for this connection.
func (c *Conn) AddEventSubscriptions(events []string) {
	if len(events) == 0 {
		return
	}
	c.stateMu.Lock()
	if c.eventSubs == nil {
		c.eventSubs = make(map[string]struct{}, len(events))
	}
	for _, event := range events {
		event = normalizeEventType(event)
		if event == "" {
			continue
		}
		c.eventSubs[event] = struct{}{}
	}
	c.stateMu.Unlock()
}

// HasEventSubscription reports whether the connection subscribed to the event type.
func (c *Conn) HasEventSubscription(eventType string) bool {
	eventType = normalizeEventType(eventType)
	if eventType == "" {
		return false
	}
	c.stateMu.Lock()
	_, ok := c.eventSubs[eventType]
	c.stateMu.Unlock()
	return ok
}

func (c *Conn) enqueueEvent(frame *Frame) {
	if frame == nil {
		return
	}
	select {
	case <-c.eventDone:
		return
	default:
	}
	select {
	case c.eventCh <- frame:
	default:
	}
}

func normalizeEventType(eventType string) string {
	return strings.ToUpper(strings.TrimSpace(eventType))
}

// ApplyStartupOptions applies STARTUP options such as compression settings.
func (c *Conn) ApplyStartupOptions(options map[string]string) error {
	normalized := make(map[string]string, len(options))
	for key, value := range options {
		normalized[strings.ToUpper(key)] = value
	}

	keyspace := strings.TrimSpace(normalized["KEYSPACE"])
	compression := strings.ToLower(normalized["COMPRESSION"])
	if compression == "" {
		c.stateMu.Lock()
		c.startupOptions = normalized
		if keyspace != "" {
			c.currentKeyspace = keyspace
		}
		c.compressionEnabled = false
		c.compressor = nil
		c.stateMu.Unlock()
		return nil
	}

	var compressor gocql.Compressor
	switch compression {
	case "lz4":
		compressor = &lz4.LZ4Compressor{}
	case "snappy":
		compressor = &snappy.SnappyCompressor{}
	default:
		return fmt.Errorf("unsupported compression %q", compression)
	}

	c.stateMu.Lock()
	c.startupOptions = normalized
	if keyspace != "" {
		c.currentKeyspace = keyspace
	}
	c.compressor = compressor
	c.compressionEnabled = true
	c.stateMu.Unlock()

	return nil
}

// GetStartupOptions returns a copy of the normalized STARTUP options.
func (c *Conn) GetStartupOptions() map[string]string {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	if len(c.startupOptions) == 0 {
		return nil
	}
	out := make(map[string]string, len(c.startupOptions))
	for key, value := range c.startupOptions {
		out[key] = value
	}
	return out
}

func (c *Conn) maybeDecompressFrame(frame *Frame) error {
	if frame.Flags&FlagCompression == 0 {
		return nil
	}

	c.stateMu.Lock()
	compressionEnabled := c.compressionEnabled
	compressor := c.compressor
	startupComplete := c.startupComplete
	protocolVersion := c.protocolVersion
	c.stateMu.Unlock()

	if protocolVersion >= ProtoVersion5 {
		// v5 compression is handled at the segment layer; ignore the frame flag.
		frame.Flags &^= FlagCompression
		return nil
	}

	if !compressionEnabled || compressor == nil {
		return fmt.Errorf("compressed frame received without negotiated compression")
	}
	if !startupComplete && frame.Opcode != OpcodeAuthResponse {
		return fmt.Errorf("compressed frame received before STARTUP completed")
	}

	decompressed, err := compressor.AppendDecompressedWithLength(nil, frame.Body)
	if err != nil {
		return fmt.Errorf("failed to decompress frame: %w", err)
	}
	frame.Body = decompressed
	frame.Length = int32(len(decompressed))
	frame.Flags &^= FlagCompression
	return nil
}

type writeState struct {
	frameWriter        *FrameWriter
	compressor         gocql.Compressor
	compressionEnabled bool
	startupComplete    bool
	protocolVersion    byte
	frameInfo          bool
	closed             bool
}

func (c *Conn) snapshotWriteState() writeState {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return writeState{
		frameWriter:        c.frameWriter,
		compressor:         c.compressor,
		compressionEnabled: c.compressionEnabled,
		startupComplete:    c.startupComplete,
		protocolVersion:    c.protocolVersion,
		frameInfo:          c.frameInfo,
		closed:             c.closed,
	}
}

func shouldCompressFrame(frame *Frame, state writeState) bool {
	if !state.compressionEnabled || state.compressor == nil {
		return false
	}
	if state.protocolVersion >= ProtoVersion5 {
		return false
	}
	if !state.startupComplete {
		return false
	}
	if frame.Flags&FlagCompression != 0 {
		return false
	}
	return len(frame.Body) > 0
}

// beginDrain marks the connection as draining and unblocks any in-progress
// blocking read so Serve() can transition into the drain path. Requests already
// dispatched are allowed to complete (their responses flushed) before close.
func (c *Conn) beginDrain() {
	if c == nil {
		return
	}
	if !c.draining.CompareAndSwap(false, true) {
		return
	}
	c.stateMu.Lock()
	conn := c.conn
	closed := c.closed
	c.stateMu.Unlock()
	if conn != nil && !closed {
		// Force any blocked Read to return immediately so the serve loop stops
		// reading new requests and starts draining in-flight work.
		_ = conn.SetReadDeadline(time.Now())
	}
	// Wake a serve goroutine parked waiting for an in-flight permit so it can
	// observe the drain and stop instead of blocking until a wedged handler
	// returns. The broadcast must hold framePermitMu: otherwise it could land in
	// the window after the waiter checked the abort predicate but before it
	// parked in Wait(), losing the wakeup.
	c.framePermitMu.Lock()
	c.framePermitCond.Broadcast()
	c.framePermitMu.Unlock()
}

// isDraining reports whether the connection has entered drain mode.
func (c *Conn) isDraining() bool {
	return c != nil && c.draining.Load()
}

// isClosed reports whether the connection has been closed.
func (c *Conn) isClosed() bool {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.closed
}

// drainInFlight blocks until every in-flight request goroutine has finished
// (its response written and flushed) or the connection is closed, bounded by
// timeout. The socket must not be closed while responses are still owed.
func (c *Conn) drainInFlight(timeout time.Duration) {
	deadline := time.Now().Add(timeout)
	c.framePermitMu.Lock()
	for c.inFlightFrames > 0 {
		if time.Now().After(deadline) {
			break
		}
		c.framePermitMu.Unlock()
		if c.isClosed() {
			return
		}
		time.Sleep(drainInFlightPoll)
		c.framePermitMu.Lock()
	}
	c.framePermitMu.Unlock()
}

// Close closes the connection
func (c *Conn) Close() error {
	var eventDone chan struct{}
	var conn net.Conn
	var release func()

	c.stateMu.Lock()
	if c.closed {
		c.stateMu.Unlock()
		return nil
	}
	c.closed = true
	eventDone = c.eventDone
	conn = c.conn
	release = c.proxyRelease
	c.proxyRelease = nil
	c.stateMu.Unlock()

	// Wake a serve goroutine parked waiting for an in-flight permit so it can
	// observe the close and stop instead of blocking until a wedged handler
	// returns (e.g. force-close via server.Stop without a prior drain). Hold
	// framePermitMu so the wakeup cannot be lost in the window between the
	// waiter's abort-predicate check and its Wait().
	c.framePermitMu.Lock()
	c.framePermitCond.Broadcast()
	c.framePermitMu.Unlock()

	if eventDone != nil {
		close(eventDone)
	}
	if release != nil {
		release()
	}
	c.releaseAllStreamLeases()
	if conn != nil {
		return conn.Close()
	}
	return nil
}

// SetUsername sets the authenticated username for this connection
func (c *Conn) SetUsername(username string) {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	c.username = username
}

// GetUsername returns the authenticated username
func (c *Conn) GetUsername() string {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.username
}

// SetPassword sets the authenticated password for this connection
func (c *Conn) SetPassword(password string) {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	c.password = password
}

// GetPassword returns the authenticated password
func (c *Conn) GetPassword() string {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.password
}

// SetCredentials sets both username and password for this connection
func (c *Conn) SetCredentials(username, password string) {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	c.username = username
	c.password = password
	c.credentialKey = fmt.Sprintf("%s:%s", username, password)
}

// GetCredentials returns both username and password for this connection
func (c *Conn) GetCredentials() (string, string) {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.username, c.password
}

// GetCredentialKey returns the credential cache key for this connection.
func (c *Conn) GetCredentialKey() string {
	c.stateMu.Lock()
	defer c.stateMu.Unlock()
	return c.credentialKey
}